feat(fc2a): add maintenance tasks (recovery + cleanup) with beat schedule

recover_interrupted_tasks runs every 5 minutes, finds ImportTask rows
stuck in 'processing' for >30 minutes (well above any legitimate import
duration), and re-queues them. cleanup_old_tasks runs daily and deletes
finished tasks older than 7 days so the task table stays an operational
view rather than an archive.

Both thresholds match ImageRepo's precedent. The 30-min stuck threshold
is documented inline so a future reader can adjust it intentionally
rather than mistaking it for a 'magic number'.

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

backend/app/celery_app.py

@@ -44,6 +44,17 @@ def make_celery() -> Celery:
         task_acks_late=True,
         worker_prefetch_multiplier=1,
         broker_connection_retry_on_startup=True,
+        beat_schedule={
+            "recover-interrupted-tasks": {
+                "task": "backend.app.tasks.maintenance.recover_interrupted_tasks",
+                "schedule": 300.0,  # every 5 minutes
+            },
+            "cleanup-old-tasks": {
+                "task": "backend.app.tasks.maintenance.cleanup_old_tasks",
+                "schedule": 86400.0,  # daily
+            },
+        },
+        timezone="UTC",
     )
     return app
 

backend/app/tasks/maintenance.py

@@ -0,0 +1,74 @@
+"""Periodic maintenance: recover stuck import tasks, garbage-collect old finished tasks."""
+
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import create_engine, delete, select, update
+from sqlalchemy.orm import sessionmaker
+
+from ..celery_app import celery
+from ..config import get_config
+from ..models import ImportTask
+
+
+STUCK_THRESHOLD_MINUTES = 30
+OLD_TASK_DAYS = 7
+
+
+def _sync_session_factory():
+    cfg = get_config()
+    engine = create_engine(cfg.database_url_sync, future=True, pool_pre_ping=True)
+    return sessionmaker(engine, expire_on_commit=False)
+
+
+@celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks")
+def recover_interrupted_tasks() -> int:
+    """Find ImportTask rows stuck in 'processing' for >30 min and re-queue them.
+
+    Why 30 min: large videos can legitimately take many minutes to import;
+    30 is a safe gate that catches actual crashes (which leave the row stuck
+    forever) without resetting slow-but-still-running jobs.
+    """
+    SessionLocal = _sync_session_factory()
+    cutoff = datetime.now(timezone.utc) - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
+    with SessionLocal() as session:
+        stuck_ids = session.execute(
+            select(ImportTask.id)
+            .where(ImportTask.status == "processing")
+            .where(ImportTask.started_at < cutoff)
+        ).scalars().all()
+
+        if not stuck_ids:
+            return 0
+
+        session.execute(
+            update(ImportTask)
+            .where(ImportTask.id.in_(stuck_ids))
+            .values(status="queued", started_at=None, error="recovered from stuck state")
+        )
+        session.commit()
+
+        from .import_file import import_media_file
+        for tid in stuck_ids:
+            import_media_file.delay(tid)
+
+        return len(stuck_ids)
+
+
+@celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks")
+def cleanup_old_tasks() -> int:
+    """Delete completed/skipped/failed ImportTask rows older than 7 days.
+
+    Why 7 days: long enough to debug an issue an operator only notices days
+    later; short enough that the task table stays a useful operational view
+    rather than an archive. Matches IR's default.
+    """
+    SessionLocal = _sync_session_factory()
+    cutoff = datetime.now(timezone.utc) - timedelta(days=OLD_TASK_DAYS)
+    with SessionLocal() as session:
+        result = session.execute(
+            delete(ImportTask)
+            .where(ImportTask.status.in_(["complete", "skipped", "failed"]))
+            .where(ImportTask.finished_at < cutoff)
+        )
+        session.commit()
+        return result.rowcount or 0

tests/test_maintenance.py

@@ -0,0 +1,77 @@
+"""Unit tests for the maintenance tasks. Eager-mode Celery so the task
+function runs synchronously inside the same DB transaction as the test.
+"""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from backend.app.celery_app import celery
+from backend.app.models import ImportBatch, ImportTask
+
+
+@pytest.fixture(autouse=True)
+def eager():
+    celery.conf.task_always_eager = True
+    yield
+    celery.conf.task_always_eager = False
+
+
+def _make_batch(session) -> int:
+    batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick")
+    session.add(batch)
+    session.flush()
+    return batch.id
+
+
+def test_recover_interrupted_only_old(db_sync):
+    batch_id = _make_batch(db_sync)
+    now = datetime.now(timezone.utc)
+
+    fresh = ImportTask(
+        batch_id=batch_id, source_path="/import/a.jpg", task_type="media",
+        status="processing", started_at=now - timedelta(minutes=5),
+    )
+    stale = ImportTask(
+        batch_id=batch_id, source_path="/import/b.jpg", task_type="media",
+        status="processing", started_at=now - timedelta(hours=2),
+    )
+    db_sync.add_all([fresh, stale])
+    db_sync.commit()
+
+    from backend.app.tasks.maintenance import recover_interrupted_tasks
+    recovered = recover_interrupted_tasks.apply().get()
+    assert recovered == 1
+
+    db_sync.refresh(fresh)
+    db_sync.refresh(stale)
+    assert fresh.status == "processing"
+    assert stale.status == "queued"
+    assert stale.started_at is None
+
+
+def test_cleanup_old_deletes_finished_old(db_sync):
+    batch_id = _make_batch(db_sync)
+    now = datetime.now(timezone.utc)
+
+    old_complete = ImportTask(
+        batch_id=batch_id, source_path="/import/a.jpg", task_type="media",
+        status="complete", finished_at=now - timedelta(days=10),
+    )
+    recent_complete = ImportTask(
+        batch_id=batch_id, source_path="/import/b.jpg", task_type="media",
+        status="complete", finished_at=now - timedelta(days=2),
+    )
+    old_pending = ImportTask(
+        batch_id=batch_id, source_path="/import/c.jpg", task_type="media",
+        status="pending",
+    )
+    db_sync.add_all([old_complete, recent_complete, old_pending])
+    db_sync.commit()
+
+    from backend.app.tasks.maintenance import cleanup_old_tasks
+    deleted = cleanup_old_tasks.apply().get()
+    assert deleted == 1
+
+    remaining = {t.source_path for t in db_sync.query(ImportTask).all()}
+    assert remaining == {"/import/b.jpg", "/import/c.jpg"}