fc3i: recover_stalled_task_runs + prune_task_runs maintenance tasks + Beat entries

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/celery_app.py

@@ -81,6 +81,14 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.maintenance.cleanup_old_download_events",
                 "schedule": 86400.0,  # daily
             },
+            "recover-stalled-task-runs": {
+                "task": "backend.app.tasks.maintenance.recover_stalled_task_runs",
+                "schedule": 300.0,  # every 5 min, matches recover-interrupted-tasks
+            },
+            "prune-task-runs": {
+                "task": "backend.app.tasks.maintenance.prune_task_runs",
+                "schedule": 86400.0,  # daily
+            },
         },
         timezone="UTC",
     )

backend/app/tasks/maintenance.py

@@ -9,7 +9,7 @@ from PIL import Image
 from sqlalchemy import delete, select, update
 
 from ..celery_app import celery
-from ..models import DownloadEvent, ImageRecord, ImportSettings, ImportTask
+from ..models import DownloadEvent, ImageRecord, ImportSettings, ImportTask, TaskRun
 from ..utils.phash import compute_phash
 from ._sync_engine import sync_session_factory as _sync_session_factory
 
@@ -20,6 +20,8 @@ OLD_TASK_DAYS = 7
 PHASH_PAGE = 500
 VERIFY_PAGE = 200
 FFPROBE_TIMEOUT_SECONDS = 10
+TASK_RUN_KEEP_OK_SECONDS = 24 * 3600       # 24 h
+TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600  # 7 days
 
 
 @celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks")
@@ -80,6 +82,71 @@ def cleanup_old_tasks() -> int:
         return result.rowcount or 0
 
 
+@celery.task(name="backend.app.tasks.maintenance.recover_stalled_task_runs")
+def recover_stalled_task_runs() -> int:
+    """Flip task_run rows stuck in 'running' for >STUCK_THRESHOLD_MINUTES
+    to 'error'. FC-3i.
+
+    A row gets stuck when the worker dies without emitting
+    task_postrun / task_failure (e.g. OOM, container restart between
+    signals, signal handler raised+logged). Shares the 5-min threshold
+    with recover_interrupted_tasks for consistency.
+    """
+    SessionLocal = _sync_session_factory()
+    cutoff = datetime.now(UTC) - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
+    with SessionLocal() as session:
+        result = session.execute(
+            update(TaskRun)
+            .where(TaskRun.status == "running")
+            .where(TaskRun.started_at < cutoff)
+            .values(
+                status="error",
+                error_type="RecoverySweep",
+                error_message=(
+                    f"no completion signal received within "
+                    f"{STUCK_THRESHOLD_MINUTES} min"
+                ),
+                finished_at=datetime.now(UTC),
+            )
+        )
+        session.commit()
+        return result.rowcount or 0
+
+
+@celery.task(name="backend.app.tasks.maintenance.prune_task_runs")
+def prune_task_runs() -> dict:
+    """Daily retention for task_run rows. FC-3i.
+
+    - 'ok' rows: deleted after TASK_RUN_KEEP_OK_SECONDS (24h default).
+      Success is high-volume, not interesting after a day.
+    - 'error' / 'timeout' rows: deleted after TASK_RUN_KEEP_FAILURE_SECONDS
+      (7 days default). Failures are operationally interesting longer.
+    - 'running' rows: NEVER deleted by this task. The recovery sweep
+      (recover_stalled_task_runs) is the mechanism that flips them to
+      terminal state; prune doesn't touch in-flight state.
+    - 'retry' rows: treated as failures (>7d).
+
+    Returns dict of how many rows were deleted in each bucket.
+    """
+    SessionLocal = _sync_session_factory()
+    now = datetime.now(UTC)
+    ok_cutoff = now - timedelta(seconds=TASK_RUN_KEEP_OK_SECONDS)
+    fail_cutoff = now - timedelta(seconds=TASK_RUN_KEEP_FAILURE_SECONDS)
+    with SessionLocal() as session:
+        ok_deleted = session.execute(
+            delete(TaskRun)
+            .where(TaskRun.status == "ok")
+            .where(TaskRun.finished_at < ok_cutoff)
+        ).rowcount or 0
+        fail_deleted = session.execute(
+            delete(TaskRun)
+            .where(TaskRun.status.in_(["error", "timeout", "retry"]))
+            .where(TaskRun.finished_at < fail_cutoff)
+        ).rowcount or 0
+        session.commit()
+        return {"ok_deleted": ok_deleted, "failures_deleted": fail_deleted}
+
+
 @celery.task(name="backend.app.tasks.maintenance.backfill_phash")
 def backfill_phash() -> int:
     """Recompute phash for stored images that have none (imported before