fix(audit-g3): lifecycle batch — recovery sweeps, retention, timeouts

Plugs the FC long-running-entity discipline gaps the 2026-06-02 audit
flagged: every entity that can get stuck now has recovery + retention +
timeout, and the long-runners no longer collide with the FC-3i sweep.

Recovery sweeps (every 5 min):
- recover_stalled_backup_runs — flips BackupRun stuck in
  running/restoring past 7h (covers the 6.5h images-backup hard
  limit) to error. prune_backups docstring corrected — the FC-3i
  TaskRun sweep never touched BackupRun rows.
- recover_stalled_library_audit_runs — flips LibraryAuditRun stuck
  past 135 min (10-min buffer above scan_library_for_rule's 2h5m
  hard limit) to error. Previously a SIGKILL'd row blocked all
  future audits until manual DB surgery.
- recover_stalled_import_batches — finalizes ImportBatch rows
  stuck running >2h whose child tasks are all terminal (orphan case
  where the orchestrator crashed before the closing UPDATE). Uses
  the same EXISTS predicate /api/system/stats already had.

Retention (daily):
- prune_library_audit_runs — 30-day window. Audit rows carry
  matched_ids JSONB blobs that can hold tens of thousands of ids.
- prune_import_batches — 30-day window. Cascades to ImportTask via
  the model relationship.

time_limits on five long-runners that previously had none (the
audit's headline finding — every one of these collided with the
recover_stalled_task_runs 5-min default and could be marked
'error' mid-flight):
- scan_directory: 60m soft / 70m hard
- verify_integrity: 60m / 70m
- backfill_phash: 30m / 35m
- apply_allowlist_tags: 30m / 35m
- recompute_centroids: 30m / 35m

QUEUE_STUCK_THRESHOLD_MINUTES now covers maintenance (75) and scan
(75) — above the longest task on each — with per-task overrides
for the outliers (backup_images_task 420, restore_images_task 420,
scan_library_for_rule 130).

start_audit_run guard is now age-aware: a 'running' row older than
the audit hard limit doesn't block a new run (the sweep will catch
it within 5 min). Previously a SIGKILL'd row blocked forever.

/api/import/status now uses the same EXISTS predicate
/api/system/stats does, so the two endpoints no longer disagree on
the active-batch question.

DownloadEvent.started_at resets on pending→running so a freshly-
promoted event from a busy queue isn't measured against its
original enqueue time (was racing recover_stalled_download_events
on heavy-queue days).

backend/app/api/import_admin.py

@@ -35,10 +35,26 @@ async def trigger_scan():
 @import_admin_bp.route("/status", methods=["GET"])
 async def status():
     async with get_session() as session:
+        # Active batch = running batch that still has outstanding work.
+        # Plain "most recent running" picks freshly-created scans that
+        # enqueued zero new files and hides the older batch that's
+        # actually being processed. Mirrors the EXISTS predicate
+        # /api/system/stats already uses (api/settings.py:145-160).
+        # Audit 2026-06-02 — /api/import/status and /api/system/stats
+        # used to disagree on the active-batch predicate; the UI banner
+        # said "Scanning…" indefinitely while the stats card said idle.
         active = (
             await session.execute(
                 select(ImportBatch)
-                .where(ImportBatch.status == "running")
+                .where(
+                    ImportBatch.status == "running",
+                    select(ImportTask.id)
+                    .where(
+                        ImportTask.batch_id == ImportBatch.id,
+                        ImportTask.status.in_(["pending", "queued", "processing"]),
+                    )
+                    .exists(),
+                )
                 .order_by(ImportBatch.started_at.desc())
                 .limit(1)
             )

backend/app/celery_app.py

@@ -105,6 +105,31 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.backup.prune_backups",
                 "schedule": 86400.0,  # daily
             },
+            # Audit 2026-06-02 — three new per-entity recovery sweeps.
+            # Each runs every 5 min like the other recover_stalled_*
+            # sweeps; each is a no-op when nothing is stuck.
+            "recover-stalled-backup-runs": {
+                "task": "backend.app.tasks.maintenance.recover_stalled_backup_runs",
+                "schedule": 300.0,
+            },
+            "recover-stalled-library-audit-runs": {
+                "task": "backend.app.tasks.maintenance.recover_stalled_library_audit_runs",
+                "schedule": 300.0,
+            },
+            "recover-stalled-import-batches": {
+                "task": "backend.app.tasks.maintenance.recover_stalled_import_batches",
+                "schedule": 300.0,
+            },
+            # Audit 2026-06-02 — daily retention for two entities
+            # whose terminal rows otherwise accumulate forever.
+            "prune-library-audit-runs": {
+                "task": "backend.app.tasks.maintenance.prune_library_audit_runs",
+                "schedule": 86400.0,
+            },
+            "prune-import-batches": {
+                "task": "backend.app.tasks.maintenance.prune_import_batches",
+                "schedule": 86400.0,
+            },
         },
         timezone="UTC",
     )

backend/app/services/cleanup_service.py

@@ -11,7 +11,7 @@ the one-and-done GS/IR migration tooling.)
 """
 from __future__ import annotations
 
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any
 
@@ -517,6 +517,9 @@ class ConfirmTokenMismatch(Exception):
 _VALID_RULES = ("transparency", "single_color")
 
 
+_AUDIT_GUARD_THRESHOLD_MINUTES = 135  # matches LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES
+
+
 def start_audit_run(
     session: Session, *, rule: str, params: dict[str, Any],
 ) -> int:
@@ -524,11 +527,21 @@ def start_audit_run(
     scan_library_for_rule Celery task. Returns the new audit_id.
 
     Concurrent-runs guard: raises AuditAlreadyRunning if any audit_run
-    has status='running'. Operator must cancel or wait."""
+    has status='running' AND started recently. Audit 2026-06-02 made
+    the guard age-aware: a SIGKILL'd run leaves a row in 'running'
+    that the recovery sweep flips on its next pass (~5 min), but a
+    fresh start_audit_run between the SIGKILL and the sweep would
+    previously block forever. Past the threshold, treat the running
+    row as stale and let the sweep clean it up — the new run still
+    gets to start.
+    """
     if rule not in _VALID_RULES:
         raise ValueError(f"unknown rule {rule!r}; expected one of {_VALID_RULES}")
+    cutoff = datetime.now(UTC) - timedelta(minutes=_AUDIT_GUARD_THRESHOLD_MINUTES)
     existing = session.execute(
-        select(LibraryAuditRun.id).where(LibraryAuditRun.status == "running")
+        select(LibraryAuditRun.id)
+        .where(LibraryAuditRun.status == "running")
+        .where(LibraryAuditRun.started_at >= cutoff)
     ).scalar_one_or_none()
     if existing is not None:
         raise AuditAlreadyRunning(existing)

backend/app/services/download_service.py

@@ -177,6 +177,13 @@ class DownloadService:
             return {"status": "in_flight", "event_id": existing.id}
         if existing and existing.status == "pending":
             existing.status = "running"
+            # Reset started_at on the pending→running transition so the
+            # recovery sweep (DOWNLOAD_STALL_THRESHOLD_MINUTES, 30 min)
+            # measures from real start, not from enqueue. On heavy-queue
+            # days a freshly-promoted event whose original started_at
+            # predated the cutoff would otherwise get swept mid-flight,
+            # racing phase3's commit. Audit 2026-06-02.
+            existing.started_at = datetime.now(UTC)
             await self.async_session.commit()
             event_id = existing.id
         else:

backend/app/tasks/backup.py

@@ -240,8 +240,11 @@ def prune_backups() -> dict:
 
     Returns {"db_deleted": N, "images_deleted": M, "files_unlinked": K}.
     Tagged rows (tag IS NOT NULL) are never pruned.
-    Status='running' / 'restoring' rows are never pruned (recovery
-    sweep from FC-3i handles those via task_run).
+    Status='running' / 'restoring' rows are never pruned — the
+    recover_stalled_backup_runs sweep flips truly-stuck ones to
+    'error' first. (Earlier docstring claimed the FC-3i TaskRun sweep
+    handled those, but TaskRun cleanup never touched BackupRun rows.
+    Audit 2026-06-02 added the dedicated sweep.)
     """
     SessionLocal = _sync_session_factory()
     counts = {"db_deleted": 0, "images_deleted": 0, "files_unlinked": 0}

backend/app/tasks/maintenance.py

@@ -11,10 +11,13 @@ from sqlalchemy import Integer, and_, cast, delete, func, or_, select, update
 
 from ..celery_app import celery
 from ..models import (
+    BackupRun,
     DownloadEvent,
     ImageRecord,
+    ImportBatch,
     ImportSettings,
     ImportTask,
+    LibraryAuditRun,
     Source,
     TaskRun,
 )
@@ -55,6 +58,29 @@ FFPROBE_TIMEOUT_SECONDS = 10
 TASK_RUN_KEEP_OK_SECONDS = 24 * 3600       # 24 h
 TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600  # 7 days
 
+# Audit 2026-06-02: per-entity recovery sweep thresholds. Each must be
+# > the entity's longest legitimate runtime (its task's time_limit + a
+# small buffer) so the sweep never flags in-flight work.
+#
+# Backups: images backup has time_limit=23400s (6.5h). 7h covers it
+# with a 30-min buffer; db backup at 12 min hard limit fits trivially.
+BACKUP_STALL_THRESHOLD_MINUTES = 7 * 60
+# Library audit: scan_library_for_rule has time_limit=7500s (2h5m).
+# 2h15m gives a 10-min buffer.
+LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES = 135
+# Import batches finalize only after every child ImportTask hits a
+# terminal state. The recovery sweep targets the case where every
+# task is done but the batch never got its closing UPDATE
+# (orchestrator crashed at the wrong instant). 2h is well past any
+# realistic single-batch import.
+IMPORT_BATCH_STALL_THRESHOLD_MINUTES = 120
+
+# Retention windows (terminal rows older than these get deleted by
+# the daily prune sweeps). 30 days = operator-flagged "useful for
+# triage for a few weeks, then noise."
+LIBRARY_AUDIT_KEEP_DAYS = 30
+IMPORT_BATCH_KEEP_DAYS = 30
+
 # Overrides for recover_stalled_task_runs (the TaskRun 'running' sweep).
 # Tasks/queues that legitimately run longer than the default 5-min
 # threshold need their own larger value, else the sweep marks in-flight
@@ -69,9 +95,24 @@ TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600  # 7 days
 #   files); time_limit=2100.
 QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
     "ml": 25,
+    # 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 /
+    # recompute_centroids / backfill_phash at 35m hard). 75 min lives
+    # above the longest of those and the per-task overrides below
+    # cover the outliers (backups, library audit).
+    "maintenance": 75,
+    "scan": 75,
 }
 TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
     "backend.app.tasks.import_file.import_archive_file": 40,
+    # Backup images runs hours, not minutes (6.5h hard limit). The
+    # task-name override beats the queue's 75-min default so a
+    # legitimately-running backup isn't flagged.
+    "backend.app.tasks.backup.backup_images_task": 420,
+    "backend.app.tasks.backup.restore_images_task": 420,
+    # Library audit scans the full library — 2h hard limit.
+    "backend.app.tasks.library_audit.scan_library_for_rule": 130,
 }
 
 
@@ -360,7 +401,12 @@ def prune_task_runs() -> dict:
         return {"ok_deleted": ok_deleted, "failures_deleted": fail_deleted}
 
 
-@celery.task(name="backend.app.tasks.maintenance.backfill_phash")
+@celery.task(
+    name="backend.app.tasks.maintenance.backfill_phash",
+    # Audit 2026-06-02 — keyset-paginated phash recompute over the whole
+    # library; legitimately runs >5 min on large libraries.
+    soft_time_limit=1800, time_limit=2100,
+)
 def backfill_phash() -> int:
     """Recompute phash for stored images that have none (imported before
     FC-2d-i+ii). Keyset-paginated by id (restart-safe), NULL-only fill,
@@ -438,7 +484,13 @@ def _verify_one(path: Path, expected_sha: str, mime: str, sha_fn) -> str:
         return "failed_verification"
 
 
-@celery.task(name="backend.app.tasks.maintenance.verify_integrity")
+@celery.task(
+    name="backend.app.tasks.maintenance.verify_integrity",
+    # Audit 2026-06-02 — full library sha256 + decode probe; on 100k-image
+    # libraries this runs an hour or more. Match the maintenance queue's
+    # recovery threshold (75 min) with 30s buffer below.
+    soft_time_limit=3600, time_limit=4200,
+)
 def verify_integrity() -> int:
     """Verify every ImageRecord file: sha256 recompute + decode/probe
     (PIL for images; ffprobe for videos). Writes integrity_status
@@ -534,6 +586,156 @@ def recover_stalled_download_events() -> int:
         return events_recovered
 
 
+@celery.task(name="backend.app.tasks.maintenance.recover_stalled_backup_runs")
+def recover_stalled_backup_runs() -> int:
+    """Flip BackupRun rows stuck in running/restoring past the hard limit
+    to error. Audit 2026-06-02.
+
+    prune_backups (FC-3h) used to claim the FC-3i task_run sweep handled
+    these — but that sweep only flips TaskRun rows, not the BackupRun
+    artifact rows. A SIGKILL'd backup left BackupRun stuck forever
+    (dashboard showed phantom in-flight backups, keep_last_n offset
+    arithmetic skewed because zombies sat outside the ok/error window).
+    """
+    SessionLocal = _sync_session_factory()
+    now = datetime.now(UTC)
+    cutoff = now - timedelta(minutes=BACKUP_STALL_THRESHOLD_MINUTES)
+    msg = (
+        f"stranded by recovery sweep (no terminal status after "
+        f"{BACKUP_STALL_THRESHOLD_MINUTES // 60}h)"
+    )
+    with SessionLocal() as session:
+        result = session.execute(
+            update(BackupRun)
+            .where(BackupRun.status.in_(["running", "restoring"]))
+            .where(BackupRun.started_at < cutoff)
+            .values(status="error", finished_at=now, error=msg)
+        )
+        session.commit()
+        recovered = result.rowcount or 0
+        if recovered:
+            log.info("recover_stalled_backup_runs: recovered %d rows", recovered)
+        return recovered
+
+
+@celery.task(name="backend.app.tasks.maintenance.recover_stalled_library_audit_runs")
+def recover_stalled_library_audit_runs() -> int:
+    """Flip LibraryAuditRun rows stuck in running past the hard limit
+    to error. Audit 2026-06-02.
+
+    LibraryAuditRun.status='running' was protected by an exclusive
+    guard in start_audit_run — a SIGKILL'd run would block all future
+    audits until manual DB surgery. (The guard is now age-aware, but
+    this sweep is what makes that work in practice.)
+    """
+    SessionLocal = _sync_session_factory()
+    now = datetime.now(UTC)
+    cutoff = now - timedelta(minutes=LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES)
+    msg = (
+        f"stranded by recovery sweep (no terminal status after "
+        f"{LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES} min)"
+    )
+    with SessionLocal() as session:
+        result = session.execute(
+            update(LibraryAuditRun)
+            .where(LibraryAuditRun.status == "running")
+            .where(LibraryAuditRun.started_at < cutoff)
+            .values(status="error", finished_at=now, error=msg)
+        )
+        session.commit()
+        recovered = result.rowcount or 0
+        if recovered:
+            log.info(
+                "recover_stalled_library_audit_runs: recovered %d rows", recovered,
+            )
+        return recovered
+
+
+@celery.task(name="backend.app.tasks.maintenance.recover_stalled_import_batches")
+def recover_stalled_import_batches() -> int:
+    """Finalize ImportBatch rows stuck in running past the hard limit
+    when NO outstanding ImportTask remains. Audit 2026-06-02.
+
+    A batch row finalizes only after every child task hits a terminal
+    state. The orphan case: scanner crashed between the last task's
+    completion and the batch's closing UPDATE. The
+    `/api/import/status` route then surfaces the batch as 'active'
+    indefinitely while `/api/system/stats` (which uses the same
+    EXISTS predicate we apply below) correctly returns null.
+    """
+    SessionLocal = _sync_session_factory()
+    now = datetime.now(UTC)
+    cutoff = now - timedelta(minutes=IMPORT_BATCH_STALL_THRESHOLD_MINUTES)
+    with SessionLocal() as session:
+        # Batches still 'running' past the cutoff whose tasks are all
+        # terminal — there's no outstanding work, so flip the batch
+        # too. Mirrors the EXISTS predicate the active-batch surfaces use.
+        result = session.execute(
+            update(ImportBatch)
+            .where(ImportBatch.status == "running")
+            .where(ImportBatch.started_at < cutoff)
+            .where(
+                ~select(ImportTask.id)
+                .where(
+                    ImportTask.batch_id == ImportBatch.id,
+                    ImportTask.status.in_(["pending", "queued", "processing"]),
+                )
+                .exists()
+            )
+            .values(status="complete", finished_at=now)
+        )
+        session.commit()
+        recovered = result.rowcount or 0
+        if recovered:
+            log.info(
+                "recover_stalled_import_batches: finalized %d zombie batches",
+                recovered,
+            )
+        return recovered
+
+
+@celery.task(name="backend.app.tasks.maintenance.prune_library_audit_runs")
+def prune_library_audit_runs() -> int:
+    """Daily retention: delete terminal LibraryAuditRun rows older than
+    LIBRARY_AUDIT_KEEP_DAYS. Never touches 'running'. Audit 2026-06-02.
+
+    Audit rows carry matched_ids JSONB blobs that can hold tens of
+    thousands of ids; without retention these accumulate.
+    """
+    SessionLocal = _sync_session_factory()
+    cutoff = datetime.now(UTC) - timedelta(days=LIBRARY_AUDIT_KEEP_DAYS)
+    with SessionLocal() as session:
+        result = session.execute(
+            delete(LibraryAuditRun)
+            .where(LibraryAuditRun.status.in_(["ready", "applied", "cancelled", "error"]))
+            .where(LibraryAuditRun.finished_at < cutoff)
+        )
+        session.commit()
+        return result.rowcount or 0
+
+
+@celery.task(name="backend.app.tasks.maintenance.prune_import_batches")
+def prune_import_batches() -> int:
+    """Daily retention: delete terminal ImportBatch rows older than
+    IMPORT_BATCH_KEEP_DAYS. Cascade-deletes child ImportTask rows via
+    the model relationship. Never touches 'running'. Audit 2026-06-02.
+    """
+    SessionLocal = _sync_session_factory()
+    cutoff = datetime.now(UTC) - timedelta(days=IMPORT_BATCH_KEEP_DAYS)
+    with SessionLocal() as session:
+        # ORM-level delete here (not Core delete) so the
+        # ImportBatch->tasks cascade fires; Core delete would skip it.
+        old_batches = session.execute(
+            select(ImportBatch)
+            .where(ImportBatch.status.in_(["complete", "cancelled"]))
+            .where(ImportBatch.finished_at < cutoff)
+        ).scalars().all()
+        for batch in old_batches:
+            session.delete(batch)
+        session.commit()
+        return len(old_batches)
+
+
 @celery.task(name="backend.app.tasks.maintenance.cleanup_old_download_events")
 def cleanup_old_download_events() -> int:
     """FC-3d: delete terminal DownloadEvent rows older than the configured

backend/app/tasks/ml.py

@@ -212,7 +212,14 @@ def backfill(self) -> int:
     return enqueued
 
 
-@celery.task(name="backend.app.tasks.ml.apply_allowlist_tags", bind=True)
+@celery.task(
+    name="backend.app.tasks.ml.apply_allowlist_tags",
+    bind=True,
+    # Audit 2026-06-02 — the full-sweep mode (neither tag_id nor image_id)
+    # is O(images × allowlist) and legitimately runs >5 min on large
+    # libraries. Cap matches the maintenance queue's recovery threshold.
+    soft_time_limit=1800, time_limit=2100,
+)
 def apply_allowlist_tags(self, tag_id: int | None = None,
                          image_id: int | None = None) -> int:
     """Retroactively apply allowlisted tags.
@@ -341,7 +348,13 @@ def recompute_centroid(self, tag_id: int) -> bool:
     return asyncio.run(_run())
 
 
-@celery.task(name="backend.app.tasks.ml.recompute_centroids", bind=True)
+@celery.task(
+    name="backend.app.tasks.ml.recompute_centroids",
+    bind=True,
+    # Audit 2026-06-02 — drifted-centroid rebuild over potentially
+    # hundreds of tags.
+    soft_time_limit=1800, time_limit=2100,
+)
 def recompute_centroids(self) -> int:
     """Daily: find drifted centroids, enqueue recompute_centroid for each."""
     import asyncio

backend/app/tasks/scan.py

@@ -35,7 +35,15 @@ def _iter_import_files(import_root: Path):
         yield entry
 
 
-@celery.task(name="backend.app.tasks.scan.scan_directory", bind=True)
+@celery.task(
+    name="backend.app.tasks.scan.scan_directory",
+    bind=True,
+    # Audit 2026-06-02 — large libraries make the scan legitimately long.
+    # Hard cap at 70 min so the corresponding QUEUE_STUCK_THRESHOLD_MINUTES
+    # ("scan") of 75 min always wins; soft limit gives the task a clean
+    # exit window before SIGKILL.
+    soft_time_limit=3600, time_limit=4200,
+)
 def scan_directory(self, triggered_by: str = "manual",
                    mode: str = "quick") -> int:
     """Walks the import root and creates ImportTasks. `mode` is 'quick'