alembic/versions/0044_ml_settings_tagger_store_floor.py

@@ -0,0 +1,37 @@
+"""ml_settings.tagger_store_floor
+
+The ingest confidence floor below which tagger predictions are not stored,
+promoted from the TAGGER_STORE_FLOOR env var to a DB-backed, UI-tunable
+setting. Default 0.70 (was an env default of 0.05): the suggestion path
+already filters at 0.70 and the centroid/learned path covers low-confidence
+preferred tags, so the sub-0.70 tail was redundant weight — it had grown
+image_record's TOAST to ~100 GB. See plan-task #764.
+
+Revision ID: 0044
+Revises: 0043
+Create Date: 2026-06-10
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0044"
+down_revision: Union[str, None] = "0043"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "ml_settings",
+        sa.Column(
+            "tagger_store_floor", sa.Float(),
+            nullable=False, server_default="0.7",
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("ml_settings", "tagger_store_floor")

backend/app/api/admin.py

@@ -348,3 +348,15 @@ async def trigger_reextract_archives():
 
     async_result = reextract_archive_attachments_task.delay()
     return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+
+
+@admin_bp.route("/maintenance/prune-predictions", methods=["POST"])
+async def trigger_prune_predictions():
+    """Operator-triggered #764 backfill: drop stored tagger predictions below
+    the current ml_settings.tagger_store_floor and clamp allowlist thresholds
+    up to it. Shrinks image_record's TOAST (~100 GB of sub-0.70 scores).
+    Idempotent + self-resuming; runs on the maintenance_long lane."""
+    from ..tasks.admin import prune_low_confidence_predictions_task
+
+    async_result = prune_low_confidence_predictions_task.delay()
+    return jsonify({"task_id": async_result.id, "status": "queued"}), 202

backend/app/api/ml_admin.py

@@ -13,6 +13,7 @@ _EDITABLE = (
     "suggestion_threshold_general",
     "centroid_similarity_threshold",
     "min_reference_images",
+    "tagger_store_floor",
 )
 
 
@@ -30,6 +31,7 @@ async def get_settings():
             "suggestion_threshold_general": s.suggestion_threshold_general,
             "centroid_similarity_threshold": s.centroid_similarity_threshold,
             "min_reference_images": s.min_reference_images,
+            "tagger_store_floor": s.tagger_store_floor,
             "tagger_model_version": s.tagger_model_version,
             "embedder_model_version": s.embedder_model_version,
         }
@@ -47,13 +49,45 @@ async def patch_settings():
         s = (
             await session.execute(select(MLSettings).where(MLSettings.id == 1))
         ).scalar_one()
+
+        # Merge the patch over current values, then validate the result as a
+        # whole — the store-floor invariant couples three fields, so they
+        # can't be checked one at a time.
+        proposed = {f: getattr(s, f) for f in _EDITABLE}
         for field in _EDITABLE:
             if field in body:
-                setattr(s, field, body[field])
+                proposed[field] = body[field]
+
+        err = _validate(proposed)
+        if err is not None:
+            return jsonify({"error": err}), 400
+
+        for field in _EDITABLE:
+            setattr(s, field, proposed[field])
         await session.commit()
     return await get_settings()
 
 
+def _validate(p: dict) -> str | None:
+    """Returns an error string if the proposed settings are invalid, else None.
+
+    Invariant (plan-task #764): the per-category suggestion thresholds can't
+    drop below tagger_store_floor — nothing below the floor is stored, so a
+    lower threshold would silently surface nothing in that gap. The UI clamps
+    the sliders to the floor; this is the server-side backstop.
+    """
+    floor = p["tagger_store_floor"]
+    if not (0.0 <= floor <= 1.0):
+        return "tagger_store_floor must be between 0 and 1"
+    for cat in ("character", "general"):
+        if p[f"suggestion_threshold_{cat}"] < floor:
+            return (
+                f"suggestion_threshold_{cat} cannot be below tagger_store_floor "
+                f"({floor}) — predictions below the floor are not stored"
+            )
+    return None
+
+
 @ml_admin_bp.route("/backfill", methods=["POST"])
 async def trigger_backfill():
     from ..tasks.ml import backfill

backend/app/api/system_activity.py

@@ -147,6 +147,7 @@ async def list_runs():
     """Paginated task_run history. Query params:
       queue=<name>     filter to one queue
       status=<status>  filter to one status (running/ok/error/timeout/retry)
+      task=<substr>    case-insensitive substring match on task_name
       limit=<int>      default 50, max 200
       before_id=<int>  cursor for keyset pagination
 
@@ -161,6 +162,7 @@ async def list_runs():
 
     queue = request.args.get("queue")
     status = request.args.get("status")
+    task = request.args.get("task")
     before_id_raw = request.args.get("before_id")
     before_id = int(before_id_raw) if before_id_raw else None
 
@@ -170,6 +172,11 @@ async def list_runs():
             stmt = stmt.where(TaskRun.queue == queue)
         if status:
             stmt = stmt.where(TaskRun.status == status)
+        if task:
+            # Task names contain literal underscores (download_source,
+            # vacuum_analyze) — escape LIKE wildcards so a search for
+            # "vacuum_analyze" doesn't treat "_" as a single-char match.
+            stmt = stmt.where(TaskRun.task_name.ilike(f"%{_escape_like(task)}%", escape="\\"))
         if before_id is not None:
             stmt = stmt.where(TaskRun.id < before_id)
         stmt = stmt.limit(limit + 1)
@@ -225,6 +232,12 @@ async def list_failures():
     })
 
 
+def _escape_like(value: str) -> str:
+    """Escape SQL LIKE/ILIKE metacharacters so user search text is matched
+    literally. Pairs with `escape="\\"` on the .ilike() call."""
+    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+
+
 def _row_to_dict(r: TaskRun) -> dict:
     return {
         "id": r.id,

backend/app/models/ml_settings.py

@@ -28,6 +28,15 @@ class MLSettings(Base):
     centroid_similarity_threshold: Mapped[float] = mapped_column(
         Float, nullable=False, default=0.55
     )
+    # Ingest floor: tagger predictions below this confidence are not stored
+    # (tagger.Tagger.infer). Default 0.70 — the suggestion path already
+    # filters at 0.70 and the centroid/learned path covers low-confidence
+    # preferred tags, so the sub-0.70 tail is redundant weight (it had
+    # bloated image_record's TOAST to ~100 GB; plan-task #764). Operator-
+    # tunable via Settings → ML; must stay ≤ the suggestion thresholds.
+    tagger_store_floor: Mapped[float] = mapped_column(
+        Float, nullable=False, default=0.70
+    )
     min_reference_images: Mapped[int] = mapped_column(
         Integer, nullable=False, default=5
     )

backend/app/services/backup_service.py

@@ -24,11 +24,17 @@ from pathlib import Path
 
 _BACKUPS_DIRNAME = "_backups"
 
-# Subprocess-level guardrails BEYOND the Celery soft_time_limit. The
-# Celery soft limit signals the Python process; subprocess.Popen in a
-# blocking syscall ignores that signal. These bound the worst case.
-_DB_SUBPROCESS_TIMEOUT_S = 12 * 60        # 12 min (Celery soft is 10 min)
-_IMAGES_SUBPROCESS_TIMEOUT_S = 7 * 60 * 60  # 7 hr (Celery soft is 6 hr)
+# Subprocess-level guardrails BEYOND the Celery soft_time_limit. The Celery
+# soft limit signals the Python process; subprocess.Popen in a blocking syscall
+# ignores that signal, so these bound the worst case directly. Each sits just
+# UNDER its task's Celery soft_time_limit so the bounded-kill (_run_bounded) is
+# the primary guard and fires cleanly before Celery's soft/hard limits — which
+# matters because an NFS D-state hang defeats even Celery's SIGKILL (the failure
+# that wedged the maintenance lane for hours, #739).
+#   backup_db_task:     soft=1800s / hard=2100s  → 1700s
+#   backup_images_task: soft=21600s / hard=23400s → 21000s
+_DB_SUBPROCESS_TIMEOUT_S = 1700            # ~28 min, under the 30-min DB soft limit
+_IMAGES_SUBPROCESS_TIMEOUT_S = 21000       # ~5.8 hr, under the 6-hr images soft limit
 # Grace after SIGKILL to reap the child. If it can't be reaped in this window
 # (an uninterruptible NFS D-state — the failure mode that wedged the
 # concurrency-1 maintenance lane for hours, operator-flagged 2026-06-07), we
@@ -115,18 +121,21 @@ def backup_db(
     to persist into BackupRun. Raises on subprocess failure."""
     ts = _now_ts()
     out_dir = _backups_dir(images_root)
-    sql_path = out_dir / f"fc_db_{ts}.sql"
+    # Custom format (-Fc): compressed (much smaller on NFS) and restored with
+    # pg_restore. The .dump extension marks it as non-SQL. The BackupRun field
+    # is still named sql_path — it's just "the db artifact path".
+    sql_path = out_dir / f"fc_db_{ts}.dump"
     # Dump to LOCAL disk first, then move the finished file to the (NFS) backups
     # dir. pg_dump's long phase is then a DB-socket wait + local writes — both
     # killable — instead of an NFS write that can hang uninterruptibly. Only the
     # final move touches NFS, and it's a bounded single-file step.
-    fd, tmp_name = tempfile.mkstemp(prefix="fc_db_", suffix=".sql")
+    fd, tmp_name = tempfile.mkstemp(prefix="fc_db_", suffix=".dump")
     os.close(fd)
     tmp_path = Path(tmp_name)
     try:
         _run_bounded(
             [
-                "pg_dump", "--no-owner", "--no-acl",
+                "pg_dump", "--no-owner", "--no-acl", "-Fc",
                 "-f", str(tmp_path), _libpq_url(db_url),
             ],
             _DB_SUBPROCESS_TIMEOUT_S,
@@ -184,8 +193,8 @@ def backup_images(
 
 
 def restore_db(*, db_url: str, sql_path: Path) -> None:
-    """Wipe public schema, then load from .sql. Raises on subprocess
-    failure; partial-restore state is the caller's concern."""
+    """Wipe public schema, then load from the custom-format dump. Raises on
+    subprocess failure; partial-restore state is the caller's concern."""
     libpq = _libpq_url(db_url)
     subprocess.run(
         [
@@ -194,8 +203,9 @@ def restore_db(*, db_url: str, sql_path: Path) -> None:
         ],
         capture_output=True, check=True, timeout=120,
     )
+    # Custom-format (-Fc) dumps are restored with pg_restore, not psql.
     subprocess.run(
-        ["psql", libpq, "-f", str(sql_path)],
+        ["pg_restore", "--no-owner", "--no-acl", "-d", libpq, str(sql_path)],
         capture_output=True, check=True,
         timeout=_DB_SUBPROCESS_TIMEOUT_S,
     )

backend/app/services/ml/allowlist.py

@@ -9,7 +9,7 @@ from sqlalchemy import delete, select
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from ...models import Tag, TagAllowlist, TagSuggestionRejection
+from ...models import MLSettings, Tag, TagAllowlist, TagSuggestionRejection
 from ...models.tag import image_tag
 from .aliases import AliasService
 
@@ -91,12 +91,25 @@ class AllowlistService:
         )
         await self.dismiss(image_id, tag_id)
 
+    async def _store_floor(self) -> float:
+        return (
+            await self.session.execute(
+                select(MLSettings.tagger_store_floor).where(MLSettings.id == 1)
+            )
+        ).scalar_one()
+
     async def update_threshold(
         self, tag_id: int, min_confidence: float
     ) -> None:
         row = await self.session.get(TagAllowlist, tag_id)
         if row is not None:
-            row.min_confidence = min_confidence
+            # An allowlist tag can't auto-apply more permissively than the
+            # ingest store floor — predictions below tagger_store_floor aren't
+            # stored, so a lower min_confidence would behave identically to the
+            # floor. Clamp so the stored threshold matches actual behavior
+            # (#764).
+            floor = await self._store_floor()
+            row.min_confidence = max(min_confidence, floor)
 
     async def remove(self, tag_id: int) -> None:
         await self.session.execute(

backend/app/services/ml/tagger.py

@@ -33,8 +33,13 @@ _MODEL_DIR = Path(os.environ.get("ML_MODEL_DIR", "/models")) / "camie"
 _MODEL_FILE = f"{MODEL_NAME}.onnx"
 _METADATA_FILE = f"{MODEL_NAME}-metadata.json"
 
-# Below this confidence, predictions aren't stored (keeps the JSON compact).
-STORE_FLOOR = float(os.environ.get("TAGGER_STORE_FLOOR", "0.05"))
+# Ingest floor below which predictions aren't stored (keeps the JSON compact).
+# DEFAULT/fallback only — the live value is DB-backed
+# (ml_settings.tagger_store_floor) and passed into infer() per call by the ml
+# task. 0.70: the suggestion path already filters there and the centroid path
+# covers lower-confidence preferred tags, so the sub-0.70 tail is redundant
+# (it had bloated image_record's TOAST to ~100 GB; plan-task #764).
+DEFAULT_STORE_FLOOR = 0.70
 
 # The categories FC-2b surfaces in the UI. Others (meta/rating/year) are
 # still stored but the suggestion service filters them out.
@@ -145,10 +150,13 @@ class Tagger:
         arr = arr.transpose(2, 0, 1)                        # HWC -> CHW
         return arr[np.newaxis, :, :, :]                     # NCHW
 
-    def infer(self, image_path: Path) -> dict[str, TagPrediction]:
+    def infer(
+        self, image_path: Path, *, store_floor: float = DEFAULT_STORE_FLOOR,
+    ) -> dict[str, TagPrediction]:
         """Run Camie v2 on one image. Returns {name: TagPrediction} with
-        confidence >= STORE_FLOOR (across all categories — the suggestion
-        service does category filtering later).
+        confidence >= store_floor (across all categories — the suggestion
+        service does category filtering later). store_floor is the DB-backed
+        ml_settings.tagger_store_floor, passed in by the ml task.
 
         v2 emits multiple outputs; we use the refined predictions
         (output[1] per onnx_inference.py). Sigmoid is applied to raw
@@ -167,7 +175,7 @@ class Tagger:
         cats = self._tag_categories
         for idx, score in enumerate(probs):
             conf = float(score)
-            if conf < STORE_FLOOR:
+            if conf < store_floor:
                 continue
             if idx >= len(names):
                 # Output longer than metadata declared — shouldn't happen but

backend/app/tasks/admin.py

@@ -207,3 +207,99 @@ def rescan_series_suggestions_task(self, after_post_id: int = 0) -> dict:
         )
         rescan_series_suggestions_task.delay(summary["resume_after_id"])
     return summary
+
+
+@celery.task(
+    name="backend.app.tasks.admin.prune_low_confidence_predictions_task",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError),
+    retry_backoff=15, retry_backoff_max=180, max_retries=1,
+    soft_time_limit=3600, time_limit=4200,  # 60 min / 70 min
+)
+def prune_low_confidence_predictions_task(self, after_id: int = 0) -> dict:
+    """One-time #764 backfill: drop tagger_predictions entries below the DB
+    store floor (ml_settings.tagger_store_floor) from existing image_record
+    rows, and clamp any allowlist min_confidence below the floor up to it.
+
+    The Camie tagger emits ~10k tags; the old 0.05 floor stored the entire
+    near-zero tail, bloating image_record's TOAST to ~100 GB. This rewrites
+    each row to the new floor. Keyset by id ASC (restart-safe via after_id);
+    idempotent — already-pruned rows rewrite to themselves and are skipped.
+    Rewriting rows generates bloat, so run VACUUM FULL / pg_repack on
+    image_record afterward to return the disk to the OS.
+
+    The keep predicate (confidence >= floor) mirrors Tagger.infer's store
+    gate so backfilled rows match what new imports store. Self-resumes on the
+    soft time limit (re-enqueues from the last committed id)."""
+    from celery.exceptions import SoftTimeLimitExceeded
+    from sqlalchemy import select, update
+
+    from ..models import ImageRecord, MLSettings, TagAllowlist
+
+    SessionLocal = _sync_session_factory()
+    scanned = 0
+    pruned = 0
+    clamped = 0
+    last_id = after_id
+    try:
+        with SessionLocal() as session:
+            floor = session.execute(
+                select(MLSettings.tagger_store_floor).where(MLSettings.id == 1)
+            ).scalar_one()
+            # Clamp allowlist thresholds below the new floor once, on the
+            # first pass (#764 consumer #4) — a sub-floor min_confidence can't
+            # apply more permissively now that nothing below it is stored.
+            if after_id == 0:
+                clamped = session.execute(
+                    update(TagAllowlist)
+                    .where(TagAllowlist.min_confidence < floor)
+                    .values(min_confidence=floor)
+                ).rowcount or 0
+                session.commit()
+
+            while True:
+                rows = session.execute(
+                    select(ImageRecord.id, ImageRecord.tagger_predictions)
+                    .where(ImageRecord.id > last_id)
+                    .where(ImageRecord.tagger_predictions.is_not(None))
+                    .order_by(ImageRecord.id.asc())
+                    .limit(500)
+                ).all()
+                if not rows:
+                    break
+                for image_id, preds in rows:
+                    scanned += 1
+                    if not preds:
+                        continue
+                    kept = {
+                        name: p for name, p in preds.items()
+                        if float(p.get("confidence", 0.0)) >= floor
+                    }
+                    if len(kept) != len(preds):
+                        session.execute(
+                            update(ImageRecord)
+                            .where(ImageRecord.id == image_id)
+                            .values(tagger_predictions=kept)
+                        )
+                        pruned += 1
+                session.commit()
+                last_id = rows[-1].id  # advance only after commit, for resume
+    except SoftTimeLimitExceeded:
+        log.warning(
+            "prune_low_confidence_predictions soft-limited at id=%s "
+            "(scanned=%d pruned=%d) — re-enqueuing", last_id, scanned, pruned,
+        )
+        prune_low_confidence_predictions_task.delay(last_id)
+        return {
+            "partial": True, "last_id": last_id,
+            "scanned": scanned, "pruned": pruned,
+        }
+
+    log.info(
+        "prune_low_confidence_predictions complete: floor=%s scanned=%d "
+        "pruned=%d allowlist_clamped=%d", floor, scanned, pruned, clamped,
+    )
+    return {
+        "floor": floor, "scanned": scanned, "pruned": pruned,
+        "allowlist_clamped": clamped, "last_id": last_id,
+    }

backend/app/tasks/maintenance.py

@@ -120,6 +120,15 @@ IMPORT_BATCH_KEEP_DAYS = 30
 #   files); time_limit=2100.
 QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
     "ml": 25,
+    # download_source legitimately walks 5-25 min (Patreon/gallery-dl
+    # deep creators); its hard time_limit is DOWNLOAD_HARD_TIME_LIMIT
+    # (1500s = 25m). The 5-min default flagged healthy in-flight walks as
+    # phantom 'RecoverySweep' failures (System Activity showed errors the
+    # Subscriptions view correctly didn't — the download finished ok and
+    # reset the source). 30 clears the 25-min limit with buffer and lines
+    # up with DOWNLOAD_STALL_THRESHOLD_MINUTES (30) so a genuine hard kill
+    # is swept by the task-run AND event sweeps together. Audit 2026-06-10.
+    "download": 30,
     # Audit 2026-06-02 — maintenance/scan queues run tasks that
     # legitimately exceed the 5-min default (verify_integrity at 70m
     # hard, scan_directory at 70m hard, apply_allowlist_tags /

backend/app/tasks/ml.py

@@ -127,7 +127,10 @@ def tag_and_embed(self, image_id: int) -> dict:
                 phase = "video_infer"
                 import numpy as np
 
-                preds = _maxpool_predictions([tagger.infer(f) for f in frames])
+                preds = _maxpool_predictions(
+                    [tagger.infer(f, store_floor=settings.tagger_store_floor)
+                     for f in frames]
+                )
                 embedding = np.mean(
                     [embedder.infer(f) for f in frames], axis=0
                 ).astype("float32")
@@ -136,7 +139,7 @@ def tag_and_embed(self, image_id: int) -> dict:
             else:
                 phase = "tag"
                 t0 = time.monotonic()
-                raw = tagger.infer(src)
+                raw = tagger.infer(src, store_floor=settings.tagger_store_floor)
                 log.info(
                     "tag_and_embed tagged in %.1fs (%d tags): %s",
                     time.monotonic() - t0, len(raw), ctx,

frontend/src/components/settings/AllowlistTable.vue

@@ -10,7 +10,7 @@
         <v-text-field
           :model-value="item.min_confidence" type="number"
           density="compact" hide-details style="max-width: 100px;"
-          min="0.05" max="1" step="0.05"
+          :min="floor" max="1" step="0.05"
           @update:model-value="(v) => onThreshold(item.tag_id, v)"
         />
       </template>
@@ -25,23 +25,32 @@
 </template>
 
 <script setup>
-import { onMounted } from 'vue'
+import { computed, onMounted } from 'vue'
 import { useAllowlistStore } from '../../stores/allowlist.js'
+import { useMLStore } from '../../stores/ml.js'
 
 const store = useAllowlistStore()
+const ml = useMLStore()
+// min_confidence can't be set below the tagger store floor — predictions
+// below it aren't stored, so a lower threshold would behave identically to
+// the floor. The backend clamps too (#764).
+const floor = computed(() => ml.settings?.tagger_store_floor ?? 0.70)
 const headers = [
   { title: 'Tag', key: 'tag_name', sortable: true },
   { title: 'Kind', key: 'tag_kind', sortable: true, width: 110 },
   { title: 'Min confidence', key: 'min_confidence', sortable: false, width: 140 },
   { title: '', key: 'actions', sortable: false, width: 60 }
 ]
-onMounted(() => store.load())
+onMounted(() => {
+  store.load()
+  if (!ml.settings) ml.loadSettings()
+})
 
 let debounce = null
 function onThreshold(tagId, value) {
   if (debounce) clearTimeout(debounce)
   debounce = setTimeout(() => {
-    const v = parseFloat(value)
+    const v = Math.max(parseFloat(value), floor.value)
     if (v > 0 && v <= 1) store.updateThreshold(tagId, v)
   }, 500)
 }

frontend/src/components/settings/MLThresholdSliders.vue

@@ -6,9 +6,28 @@
         <v-col cols="12">
           <v-slider
             v-model="local[f.key]" :label="f.label"
+            :min="f.floorMin ? local.tagger_store_floor : 0" max="1" step="0.05"
+            thumb-label hide-details
+            color="accent" @end="save"
+          />
+        </v-col>
+      </v-row>
+
+      <v-divider class="my-4" />
+
+      <v-row>
+        <v-col cols="12">
+          <v-slider
+            v-model="local.tagger_store_floor" label="Tagger store floor"
             min="0" max="1" step="0.05" thumb-label hide-details
             color="accent" @end="save"
           />
+          <div class="text-caption fc-muted mt-1">
+            Tagger predictions below this confidence aren't stored — raising it
+            keeps the image library lean. Suggestions can't be shown below the
+            floor; lower-confidence tags you actually want still surface through
+            the learned centroid path.
+          </div>
         </v-col>
       </v-row>
     </v-card-text>
@@ -24,17 +43,25 @@ import { useMLStore } from '../../stores/ml.js'
 const store = useMLStore()
 // 'artist' (FC-2d-vii-c) and 'copyright' (2026-06-01) retired as
 // suggestion categories; their threshold rows are gone.
+// floorMin: the per-category suggestion thresholds can't drop below the
+// tagger store floor (nothing below the floor is stored to surface).
 const fields = [
-  { key: 'suggestion_threshold_character', label: 'Character' },
-  { key: 'suggestion_threshold_general', label: 'General' },
+  { key: 'suggestion_threshold_character', label: 'Character', floorMin: true },
+  { key: 'suggestion_threshold_general', label: 'General', floorMin: true },
   { key: 'centroid_similarity_threshold', label: 'Centroid similarity' }
 ]
 const local = reactive({})
 watch(() => store.settings, (s) => { if (s) Object.assign(local, s) }, { immediate: true })
 
 async function save() {
+  // Mirror the server invariant: keep the category thresholds at or above the
+  // store floor so a raised floor doesn't leave a threshold stranded below it.
+  const floor = local.tagger_store_floor
+  local.suggestion_threshold_character = Math.max(local.suggestion_threshold_character, floor)
+  local.suggestion_threshold_general = Math.max(local.suggestion_threshold_general, floor)
   const patch = {}
   for (const f of fields) patch[f.key] = local[f.key]
+  patch.tagger_store_floor = local.tagger_store_floor
   try { await store.patchSettings(patch) }
   catch (e) { toast({ text: e.message, type: 'error' }) }
 }

frontend/src/components/settings/MaintenancePanel.vue

@@ -12,6 +12,7 @@
       <ThumbnailBackfillCard />
     </div>
     <MLThresholdSliders class="mt-4" />
+    <PrunePredictionsCard class="mt-4" />
     <AllowlistTable class="mt-4" />
     <AliasTable class="mt-4" />
     <DbMaintenanceCard class="mt-6" />
@@ -31,6 +32,7 @@ import MLBackfillCard from './MLBackfillCard.vue'
 import CentroidRecomputeCard from './CentroidRecomputeCard.vue'
 import ThumbnailBackfillCard from './ThumbnailBackfillCard.vue'
 import MLThresholdSliders from './MLThresholdSliders.vue'
+import PrunePredictionsCard from './PrunePredictionsCard.vue'
 import AllowlistTable from './AllowlistTable.vue'
 import AliasTable from './AliasTable.vue'
 import DbMaintenanceCard from './DbMaintenanceCard.vue'

frontend/src/components/settings/PrunePredictionsCard.vue

@@ -0,0 +1,58 @@
+<template>
+  <!-- #764: drop stored tagger predictions below the store floor to shrink
+       image_record's TOAST (the sub-0.70 score tail had grown it to ~100 GB). -->
+  <v-card>
+    <v-card-title>Prune low-confidence predictions</v-card-title>
+    <v-card-text>
+      <p class="text-body-2 mb-3">
+        Removes stored tagger predictions below the current store floor
+        (<strong>{{ floorPct }}</strong>) from every image, and clamps any
+        allowlist threshold below the floor up to it. This is what shrinks the
+        database — the low-confidence tail was the bulk of its size. Idempotent
+        and resumable; safe to run more than once. Afterward, reclaim the freed
+        space with <code>VACUUM FULL</code> / <code>pg_repack</code> on
+        <code>image_record</code>.
+      </p>
+      <v-btn color="primary" rounded="pill" :loading="busy" @click="run">
+        <v-icon start>mdi-database-minus-outline</v-icon> Prune predictions now
+      </v-btn>
+      <span v-if="queued" class="ml-3 text-caption text-success">Queued ✓</span>
+      <QueueStatusBar queue="maintenance_long" queue-label="Maintenance (long)" />
+    </v-card-text>
+  </v-card>
+</template>
+
+<script setup>
+import { computed, onMounted, ref } from 'vue'
+
+import { useApi } from '../../composables/useApi.js'
+import { toast } from '../../utils/toast.js'
+import { useMLStore } from '../../stores/ml.js'
+import QueueStatusBar from './QueueStatusBar.vue'
+
+const api = useApi()
+const ml = useMLStore()
+const busy = ref(false)
+const queued = ref(false)
+
+const floorPct = computed(() => {
+  const f = ml.settings?.tagger_store_floor
+  return f == null ? '—' : `${Math.round(f * 100)}%`
+})
+
+onMounted(() => { if (!ml.settings) ml.loadSettings() })
+
+async function run () {
+  busy.value = true
+  queued.value = false
+  try {
+    await api.post('/api/admin/maintenance/prune-predictions')
+    queued.value = true
+    toast({ text: 'Prediction prune queued', type: 'success' })
+  } catch (e) {
+    toast({ text: e?.body?.detail || e?.message || 'Failed to queue', type: 'error' })
+  } finally {
+    busy.value = false
+  }
+}
+</script>

frontend/src/components/settings/SystemActivityTab.vue

@@ -21,8 +21,15 @@
     <v-card class="mb-4">
       <CardHeading icon="mdi-alert-circle-outline" title="Recent failures (last 24h)">
         <v-spacer />
-        <span class="text-caption fc-muted">
-          {{ failureCount }} failures
+        <v-text-field
+          v-model="failureSearch"
+          placeholder="Search task, queue, target, error…"
+          density="compact" hide-details clearable
+          prepend-inner-icon="mdi-magnify"
+          style="max-width: 280px;"
+        />
+        <span class="text-caption fc-muted ml-3">
+          {{ filteredFailures.length }} / {{ failureCount }}
         </span>
       </CardHeading>
       <v-card-text>
@@ -77,6 +84,14 @@
     <v-card>
       <CardHeading icon="mdi-format-list-bulleted" title="All recent activity">
         <v-spacer />
+        <v-text-field
+          v-model="filterTask"
+          placeholder="Search task…"
+          density="compact" hide-details clearable
+          prepend-inner-icon="mdi-magnify"
+          style="max-width: 220px;"
+          @update:model-value="onTaskSearch"
+        />
         <v-select
           v-model="filterQueue"
           :items="queueOptions" density="compact" hide-details
@@ -170,6 +185,8 @@ const store = useSystemActivityStore()
 const filterQueue = ref(null)
 const filterStatus = ref(null)
 const filterErrorType = ref(null)
+const filterTask = ref(null)        // server-side task-name search (All activity)
+const failureSearch = ref('')       // client-side search over loaded failures
 
 const queueOptions = [
   { title: 'All queues', value: null },
@@ -225,9 +242,18 @@ const errorTypes = computed(() => {
 const failureCount = computed(() => (store.failures?.recent || []).length)
 
 const filteredFailures = computed(() => {
-  const all = store.failures?.recent || []
-  if (!filterErrorType.value) return all
-  return all.filter(r => r.error_type === filterErrorType.value)
+  let all = store.failures?.recent || []
+  if (filterErrorType.value) {
+    all = all.filter(r => r.error_type === filterErrorType.value)
+  }
+  const q = (failureSearch.value || '').trim().toLowerCase()
+  if (q) {
+    all = all.filter(r =>
+      [r.task_name, r.queue, r.target_id, r.error_type]
+        .some(v => String(v ?? '').toLowerCase().includes(q)),
+    )
+  }
+  return all
 })
 
 function toggleErrorTypeFilter(name) {
@@ -240,9 +266,23 @@ function toggleErrorTypeFilter(name) {
 }
 
 function onFilterChange() {
-  store.setFilter({ queue: filterQueue.value, status: filterStatus.value })
+  store.setFilter({
+    queue: filterQueue.value,
+    status: filterStatus.value,
+    task: (filterTask.value || '').trim() || null,
+  })
   store.loadRuns({ reset: true })
 }
+
+// Server-side search hits the full task_run history, not just the loaded
+// page — debounce so we don't fire a query per keystroke (matches the
+// 300ms idiom in TagsView/AllowlistTable).
+let taskSearchDebounce = null
+function onTaskSearch() {
+  if (taskSearchDebounce) clearTimeout(taskSearchDebounce)
+  taskSearchDebounce = setTimeout(onFilterChange, 300)
+}
+
 function onLoadMore() { store.loadRuns({ reset: false }) }
 function onRefresh() { store.loadRuns({ reset: true }) }
 

frontend/src/stores/systemActivity.js

@@ -16,7 +16,7 @@ export const useSystemActivityStore = defineStore('systemActivity', () => {
   const runs = ref([])
   const runsCursor = ref(null)
   const runsHasMore = ref(false)
-  const runsFilter = ref({ queue: null, status: null, limit: 50 })
+  const runsFilter = ref({ queue: null, status: null, task: null, limit: 50 })
 
   const loading = ref({ queues: false, workers: false, runs: false, failures: false })
   const lastError = ref(null)
@@ -65,6 +65,7 @@ export const useSystemActivityStore = defineStore('systemActivity', () => {
       const params = { limit: runsFilter.value.limit }
       if (runsFilter.value.queue) params.queue = runsFilter.value.queue
       if (runsFilter.value.status) params.status = runsFilter.value.status
+      if (runsFilter.value.task) params.task = runsFilter.value.task
       if (!reset && runsCursor.value) params.before_id = runsCursor.value
       const body = await api.get('/api/system/activity/runs', { params })
       runs.value = reset ? body.runs : [...runs.value, ...body.runs]

tests/test_api_ml_admin.py

@@ -34,6 +34,28 @@ async def test_get_and_patch_settings(client):
     assert (await resp.get_json())["suggestion_threshold_general"] == pytest.approx(0.90)
 
 
+@pytest.mark.asyncio
+async def test_tagger_store_floor_default_and_patch(client):
+    body = await (await client.get("/api/ml/settings")).get_json()
+    assert body["tagger_store_floor"] == pytest.approx(0.70)
+
+    resp = await client.patch("/api/ml/settings", json={"tagger_store_floor": 0.6})
+    assert resp.status_code == 200
+    assert (await resp.get_json())["tagger_store_floor"] == pytest.approx(0.6)
+
+
+@pytest.mark.asyncio
+async def test_suggestion_threshold_below_store_floor_rejected(client):
+    # Invariant (#764): a category threshold can't sit below the store floor —
+    # nothing below the floor is stored, so the gap would surface nothing.
+    # Floor defaults to 0.70; pushing general down to 0.50 must 400.
+    resp = await client.patch(
+        "/api/ml/settings", json={"suggestion_threshold_general": 0.50}
+    )
+    assert resp.status_code == 400
+    assert "tagger_store_floor" in (await resp.get_json())["error"]
+
+
 @pytest.mark.asyncio
 async def test_backfill_and_recompute_trigger(client):
     r1 = await client.post("/api/ml/backfill")

tests/test_api_system_activity.py

@@ -182,6 +182,25 @@ async def test_runs_filter_by_status(client, _seed_runs):
     assert len(body["runs"]) == 2  # i=3, i=4
 
 
+@pytest.mark.asyncio
+async def test_runs_filter_by_task_substring(client, _seed_runs):
+    # Case-insensitive substring across the full task_name.
+    resp = await client.get("/api/system/activity/runs?task=FAKE")
+    body = await resp.get_json()
+    assert len(body["runs"]) == 5
+    assert all("fake" in r["task_name"] for r in body["runs"])
+
+
+@pytest.mark.asyncio
+async def test_runs_filter_by_task_escapes_underscore(client, _seed_runs):
+    # The literal "_" in "task_3" must match one row, not act as a
+    # single-char wildcard matching every task_N.
+    resp = await client.get("/api/system/activity/runs?task=task_3")
+    body = await resp.get_json()
+    assert len(body["runs"]) == 1
+    assert body["runs"][0]["task_name"].endswith("task_3")
+
+
 @pytest.mark.asyncio
 async def test_runs_keyset_cursor(client, _seed_runs):
     page1 = await (await client.get("/api/system/activity/runs?limit=2")).get_json()

tests/test_backup_service.py

@@ -69,7 +69,7 @@ def test_backup_db_writes_sql_and_manifest(tmp_path, fake_subprocess):
     )
     sql = Path(result["sql_path"])
     manifest = Path(result["manifest_path"])
-    assert sql.is_file() and sql.suffix == ".sql"
+    assert sql.is_file() and sql.suffix == ".dump"
     assert manifest.is_file() and manifest.suffix == ".json"
     assert result["kind"] == "db"
     assert result["tar_path"] is None
@@ -92,6 +92,12 @@ def test_backup_db_strips_sqlalchemy_psycopg_driver(tmp_path, fake_subprocess):
     assert "+psycopg" not in cmd[-1]
 
 
+def test_backup_db_uses_compressed_custom_format(tmp_path, fake_subprocess):
+    # -Fc → pg_restore-loadable, compressed dump (#739 backup polish).
+    backup_service.backup_db(db_url="postgresql://u@h/d", images_root=tmp_path)
+    assert "-Fc" in fake_subprocess[0]
+
+
 def test_backup_db_strips_asyncpg_driver(tmp_path, fake_subprocess):
     backup_service.backup_db(
         db_url="postgresql+asyncpg://u:p@h/d", images_root=tmp_path,
@@ -132,17 +138,19 @@ def test_backup_images_excludes_backups_and_quarantine(tmp_path, fake_subprocess
 
 
 def test_restore_db_drops_schema_then_loads(tmp_path, fake_subprocess):
-    sql_path = tmp_path / "fake.sql"
-    sql_path.write_text("SELECT 1;")
+    dump_path = tmp_path / "fake.dump"
+    dump_path.write_bytes(b"\x00fake custom dump")
     backup_service.restore_db(
-        db_url="postgresql://u@h/d", sql_path=sql_path,
+        db_url="postgresql://u@h/d", sql_path=dump_path,
     )
-    # Two psql calls: one with -c (DROP SCHEMA), one with -f (load).
+    # Two calls: psql -c (DROP SCHEMA), then pg_restore -d (load the dump).
     assert len(fake_subprocess) == 2
+    assert fake_subprocess[0][0] == "psql"
     assert "-c" in fake_subprocess[0]
     assert "DROP SCHEMA IF EXISTS public CASCADE" in fake_subprocess[0][-1]
-    assert "-f" in fake_subprocess[1]
-    assert str(sql_path) in fake_subprocess[1]
+    assert fake_subprocess[1][0] == "pg_restore"
+    assert "-d" in fake_subprocess[1]
+    assert str(dump_path) in fake_subprocess[1]
 
 
 # --- restore_images --------------------------------------------------

tests/test_maintenance.py

@@ -350,6 +350,62 @@ def test_recover_stalled_task_runs_ml_queue_uses_longer_threshold(db_sync):
     assert ml_stale_status == "error"
 
 
+def test_recover_stalled_task_runs_download_queue_uses_longer_threshold(db_sync):
+    """download_source legitimately walks 5-25 min (Patreon/gallery-dl).
+    The 5-min default flagged healthy in-flight walks as phantom
+    'RecoverySweep' failures — visible in System Activity but absent from
+    the Subscriptions view because the download actually finished ok.
+    The 30-min download override (QUEUE_STUCK_THRESHOLD_MINUTES) must
+    protect a 10-min-old download row while still flagging a 35-min one.
+    Audit 2026-06-10."""
+    from sqlalchemy import select
+
+    from backend.app.models import TaskRun
+    from backend.app.tasks.maintenance import recover_stalled_task_runs
+
+    now = datetime.now(UTC)
+    # 10-min-old download row: stale by the default 5-min rule but fresh
+    # by the 30-min download override. Must survive the sweep.
+    dl_fresh_id = _make_task_run(
+        db_sync, status="running", queue="download",
+        task_name="backend.app.tasks.download.download_source",
+        started_at=now - timedelta(minutes=10),
+    )
+    # 35-min-old download row: past even the 30-min override (a genuine
+    # hard kill). Must be flagged.
+    dl_stale_id = _make_task_run(
+        db_sync, status="running", queue="download",
+        task_name="backend.app.tasks.download.download_source",
+        started_at=now - timedelta(minutes=35),
+    )
+    db_sync.commit()
+
+    recovered = recover_stalled_task_runs.apply().get()
+    assert recovered == 1
+
+    db_sync.expire_all()
+    dl_fresh_status = db_sync.execute(
+        select(TaskRun.status).where(TaskRun.id == dl_fresh_id)
+    ).scalar_one()
+    dl_stale_status = db_sync.execute(
+        select(TaskRun.status).where(TaskRun.id == dl_stale_id)
+    ).scalar_one()
+    assert dl_fresh_status == "running"
+    assert dl_stale_status == "error"
+
+
+def test_download_stuck_threshold_exceeds_hard_time_limit():
+    """Invariant guard (maintenance.py:112): every queue override MUST be
+    ≥ the relevant task's hard time_limit, else the sweep flags in-flight
+    work. download_source is the one that regressed — pin it so a future
+    DOWNLOAD_HARD_TIME_LIMIT bump can't silently re-break it."""
+    from backend.app.tasks.download import DOWNLOAD_HARD_TIME_LIMIT
+    from backend.app.tasks.maintenance import QUEUE_STUCK_THRESHOLD_MINUTES
+
+    hard_minutes = DOWNLOAD_HARD_TIME_LIMIT / 60
+    assert QUEUE_STUCK_THRESHOLD_MINUTES["download"] >= hard_minutes
+
+
 def test_recover_stalled_task_runs_archive_task_uses_longer_threshold(db_sync):
     """import_archive_file shares the 'import' queue with fast
     single-file import_media_file, so it gets a per-task-name override

tests/test_ml_allowlist.py

@@ -104,3 +104,17 @@ async def test_update_threshold_and_remove(db):
     assert abs(row.min_confidence - 0.80) < 1e-6
     await svc.remove(tag.id)
     assert await db.get(TagAllowlist, tag.id) is None
+
+
+@pytest.mark.asyncio
+async def test_update_threshold_clamped_to_store_floor(db):
+    # A min_confidence below the store floor (default 0.70) is clamped up —
+    # predictions below the floor aren't stored, so a lower threshold can't
+    # apply more permissively than the floor (#764).
+    tag = await TagService(db).find_or_create("Lowthr", TagKind.general)
+    svc = AllowlistService(db)
+    img = await _make_image(db)
+    await svc.accept(img.id, tag.id)
+    await svc.update_threshold(tag.id, 0.30)
+    row = await db.get(TagAllowlist, tag.id)
+    assert abs(row.min_confidence - 0.70) < 1e-6

tests/test_ml_tagger.py

@@ -1,14 +1,14 @@
 """Tagger unit tests. The ONNX model isn't available in CI (it's a 1GB
-download into /models), so these test the pure-logic surface: STORE_FLOOR
-constant, SURFACED_CATEGORIES set, TagPrediction dataclass, and the
-load()-missing-file error path. Full inference is exercised by the local
-integration suite against a real /models volume.
+download into /models), so these test the pure-logic surface:
+DEFAULT_STORE_FLOOR constant, SURFACED_CATEGORIES set, TagPrediction
+dataclass, and the load()-missing-file error path. Full inference is
+exercised by the local integration suite against a real /models volume.
 """
 
 import pytest
 
 from backend.app.services.ml.tagger import (
-    STORE_FLOOR,
+    DEFAULT_STORE_FLOOR,
     SURFACED_CATEGORIES,
     Tagger,
     TagPrediction,
@@ -26,8 +26,12 @@ def test_surfaced_categories():
     assert "copyright" not in SURFACED_CATEGORIES
 
 
-def test_store_floor_is_low():
-    assert 0 < STORE_FLOOR < 0.2
+def test_default_store_floor():
+    # Raised 0.05 → 0.70 (plan-task #764): the suggestion path filters at
+    # 0.70 and the centroid path covers lower-confidence preferred tags, so
+    # storing the sub-0.70 tail was redundant (100 GB of TOAST). The live
+    # value is DB-backed (ml_settings.tagger_store_floor); this is the default.
+    assert DEFAULT_STORE_FLOOR == 0.70
 
 
 def test_tag_prediction_dataclass():

tests/test_tasks_admin.py

@@ -42,6 +42,68 @@ def test_bulk_delete_images_task_registered():
     )
 
 
+def test_prune_low_confidence_predictions_task_registered():
+    assert (
+        "backend.app.tasks.admin.prune_low_confidence_predictions_task"
+        in celery.tasks
+    )
+
+
+@pytest.mark.asyncio
+async def test_prune_low_confidence_predictions(db_sync, tmp_path):
+    # #764: drop stored tagger predictions below the store floor (default
+    # 0.70) and clamp allowlist thresholds up to it.
+    from backend.app.models import Tag, TagAllowlist, TagKind
+    from backend.app.tasks.admin import prune_low_confidence_predictions_task
+
+    f0 = tmp_path / "p0.jpg"
+    f0.write_bytes(b"x")
+    img0 = ImageRecord(
+        path=str(f0), sha256=f"{0:064x}", size_bytes=10, mime="image/jpeg",
+        origin="imported_filesystem",
+        tagger_predictions={
+            "keep_high": {"category": "general", "confidence": 0.92},
+            "keep_edge": {"category": "general", "confidence": 0.70},
+            "drop_mid": {"category": "general", "confidence": 0.40},
+            "drop_tiny": {"category": "general", "confidence": 0.06},
+        },
+    )
+    db_sync.add(img0)
+    f1 = tmp_path / "p1.jpg"
+    f1.write_bytes(b"x")
+    img1 = ImageRecord(
+        path=str(f1), sha256=f"{1:064x}", size_bytes=10, mime="image/jpeg",
+        origin="imported_filesystem",
+        tagger_predictions={"only": {"category": "general", "confidence": 0.99}},
+    )
+    db_sync.add(img1)
+    tag = Tag(name="lowthr-tag", kind=TagKind.general)
+    db_sync.add(tag)
+    db_sync.flush()
+    db_sync.add(TagAllowlist(tag_id=tag.id, min_confidence=0.30))
+    db_sync.commit()
+    img0_id, img1_id, tag_id = img0.id, img1.id, tag.id
+
+    result = prune_low_confidence_predictions_task.delay().get()
+    assert result["floor"] == pytest.approx(0.70)
+    assert result["pruned"] == 1            # only img0 had sub-floor entries
+    assert result["allowlist_clamped"] == 1
+
+    db_sync.expire_all()
+    p0 = db_sync.execute(
+        select(ImageRecord.tagger_predictions).where(ImageRecord.id == img0_id)
+    ).scalar_one()
+    assert set(p0) == {"keep_high", "keep_edge"}   # >=0.70 kept, <0.70 dropped
+    p1 = db_sync.execute(
+        select(ImageRecord.tagger_predictions).where(ImageRecord.id == img1_id)
+    ).scalar_one()
+    assert set(p1) == {"only"}                      # already clean — untouched
+    clamped = db_sync.execute(
+        select(TagAllowlist.min_confidence).where(TagAllowlist.tag_id == tag_id)
+    ).scalar_one()
+    assert clamped == pytest.approx(0.70)
+
+
 # --- delete_artist_cascade_task -------------------------------------
 
 

tests/test_tasks_backup.py

@@ -93,7 +93,7 @@ async def test_backup_db_task_creates_backup_run_row_status_ok(db_sync):
     ).one()
     assert row.kind == "db"
     assert row.status == "ok"
-    assert row.sql_path and row.sql_path.endswith(".sql")
+    assert row.sql_path and row.sql_path.endswith(".dump")
     assert row.size_bytes is not None and row.size_bytes > 0
     assert row.finished_at is not None
     assert row.error is None