fix: task runs record the lane Celery really routes them to, and no healthy long job is swept as stalled (#4432)

celery_signals._queue_for was a hand-kept copy of task_routes and had
drifted:

- backup, admin and library_audit jobs, and backfill_phash, run on
  maintenance_long but were recorded as `maintenance`;
- translation and gpu_queue jobs were recorded as `default`, where the
  5-minute stall sweep failed healthy 35-minute translation runs.

It now asks the router, cached per task name.

A new guard test checks every registered task's hard time limit against
the stall threshold the sweep would use for it. It also caught these
sweeps, which failed healthy runs mid-flight and are fixed here:

- ml's scheduled sweeps (35 min, previously swept at 25);
- train_heads and apply_head_tags (65 min);
- import_media_file (6 min, previously swept at 5);
- the long lane, which now has its own 45-minute threshold.

UI changes:

- the admin job poller follows a job by celery_task_id, via a new filter
  on /runs, instead of by lane;
- the archive re-extract and missing-file repair cards show the long
  lane's backlog, where their jobs actually wait;
- the queue table lists maintenance_long.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR

backend/app/api/system_activity.py

@@ -152,6 +152,8 @@ async def list_runs():
       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
+      celery_task_id=<id>  exactly one run — how a page follows a job it
+                           started without having to know its lane
       limit=<int>      default 50, max 200
       before_id=<int>  cursor for keyset pagination
 
@@ -167,6 +169,7 @@ async def list_runs():
     queue = request.args.get("queue")
     status = request.args.get("status")
     task = request.args.get("task")
+    celery_task_id = request.args.get("celery_task_id")
     before_id_raw = request.args.get("before_id")
     before_id = int(before_id_raw) if before_id_raw else None
 
@@ -176,6 +179,8 @@ async def list_runs():
             stmt = stmt.where(TaskRun.queue == queue)
         if status:
             stmt = stmt.where(TaskRun.status == status)
+        if celery_task_id:
+            stmt = stmt.where(TaskRun.celery_task_id == celery_task_id)
         if task:
             # Task names contain literal underscores (download_source,
             # vacuum_analyze) — escape LIKE wildcards so a search for

backend/app/celery_signals.py

@@ -18,6 +18,7 @@ dark for that interval. Monitoring NEVER breaks the thing it's
 monitoring.
 """
 
+import functools
 import logging
 from datetime import UTC, datetime
 
@@ -53,42 +54,29 @@ _INT32_MIN = -2_147_483_648
 
 
 def _queue_for(task) -> str:
-    """Reverse the task→queue routing from celery_app.task_routes.
-    Keep in sync if task_routes is reordered.
+    """The queue Celery routes this task to — asked of the router itself.
 
-    Audit 2026-06-02: backup/admin/library_audit prefixes were
-    missing here even though task_routes sent all three to
-    'maintenance'. The TaskRun.queue column then lied for those
-    rows (claimed 'default') so per-queue dashboard filters and
-    per-queue threshold overrides silently missed them.
+    This was a hand-kept copy of `celery_app.task_routes`, and it drifted
+    twice (the 2026-06-02 audit, then #4432). Long-lane jobs were recorded as
+    `maintenance`, and translation and gpu_queue runs as `default`, where the
+    5-minute stall sweep failed healthy 35-minute translation runs. The router
+    answers from the same table the broker uses, so the two cannot disagree.
     """
     name = getattr(task, "name", "") or ""
-    if name.startswith("backend.app.tasks.import_file."):
-        return "import"
-    if name.startswith("backend.app.tasks.ml."):
-        return "ml"
-    if name.startswith("backend.app.tasks.thumbnail."):
-        return "thumbnail"
-    if name.startswith((
-        "backend.app.tasks.download.",
-        # External file-host fetches share the download lane (celery_app
-        # routes external.* → download). Mirror it here or TaskRun.queue
-        # lies 'default' for them, so per-queue dashboard filters and the
-        # per-queue threshold override miss them — the same gap the
-        # 2026-06-02 audit fixed for backup/admin/library_audit.
-        "backend.app.tasks.external.",
-    )):
-        return "download"
-    if name.startswith("backend.app.tasks.scan."):
-        return "scan"
-    if name.startswith((
-        "backend.app.tasks.maintenance.",
-        "backend.app.tasks.backup.",
-        "backend.app.tasks.admin.",
-        "backend.app.tasks.library_audit.",
-    )):
-        return "maintenance"
+    app = getattr(task, "app", None)
+    if app is None:
+        from .celery_app import celery as app
+    return _routed_queue(app, name)
+
+
+@functools.lru_cache(maxsize=1024)
+def _routed_queue(app, name: str) -> str:
+    try:
+        queue = app.amqp.router.route({}, name).get("queue")
+    except Exception:  # noqa: BLE001 — monitoring never breaks the task
+        log.warning("task_run: could not resolve the queue for %s", name)
         return "default"
+    return getattr(queue, "name", None) or (queue if isinstance(queue, str) else "default")
 
 
 def _target_id_from_args(args) -> int | None:

backend/app/tasks/maintenance.py

@@ -137,7 +137,13 @@ IMPORT_BATCH_KEEP_DAYS = 30
 #   (the import queue itself stays at the 5-min default for single
 #   files); time_limit=2100.
 QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
-    "ml": 25,
+    # ml: the scheduled auto-apply sweeps and refresh_character_prototypes run
+    # to a 35-min hard limit (2100s); 25 swept them mid-run (#4432). The two
+    # 65-min jobs have their own entries below.
+    "ml": 40,
+    # import: import_media_file's hard limit is 6 min (360s), one past the
+    # 5-min default this queue fell to (#4432).
+    "import": 10,
     # 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
@@ -154,6 +160,12 @@ QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
     # overrides below cover the outliers (backups, library audit).
     "maintenance": 75,
     "scan": 75,
+    # The long lane (#4432). Until TaskRun.queue asked the router, nothing was
+    # recorded here: these runs read as `maintenance` (75) or, for
+    # translation, `default` (5 — which failed healthy 35-min runs). The
+    # longest task without its own entry below is the admin family at a
+    # 40-min hard limit; 45 = 40 + 5.
+    "maintenance_long": 45,
 }
 TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
     "backend.app.tasks.import_file.import_archive_file": 40,
@@ -179,6 +191,10 @@ TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
     # external-fetch entry above — without an override a healthy in-flight walk
     # is swept 'RecoverySweep' at the bare 5-min default. 30 = 25 + 5.
     "backend.app.tasks.admin.reclaim_orphaned_attachments_task": 30,
+    # Head training and the manual head apply run to 65 min (3900s) — past the
+    # ml queue's threshold (#4432). 70 = 65 + 5.
+    "backend.app.tasks.ml.train_heads": 70,
+    "backend.app.tasks.ml.apply_head_tags": 70,
 }
 
 

frontend/src/components/settings/ArchiveReextractCard.vue

@@ -17,7 +17,7 @@
         <v-icon start>mdi-folder-zip-outline</v-icon> Re-extract archives now
       </v-btn>
       <span v-if="queued" class="ml-3 text-caption text-success">Queued ✓</span>
-      <QueueStatusBar queue="maintenance" queue-label="Maintenance" />
+      <QueueStatusBar queue="maintenance_long" queue-label="Long maintenance" />
   </MaintenanceTile>
 </template>
 

frontend/src/components/settings/MissingFileRepairCard.vue

@@ -18,7 +18,7 @@
         <v-icon start>mdi-file-remove-outline</v-icon> Repair missing-file records
       </v-btn>
       <span v-if="queued" class="ml-3 text-caption text-success">Queued ✓</span>
-      <QueueStatusBar queue="maintenance" queue-label="Maintenance" />
+      <QueueStatusBar queue="maintenance_long" queue-label="Long maintenance" />
   </MaintenanceTile>
 </template>
 

frontend/src/components/settings/QueuesTable.vue

@@ -41,7 +41,7 @@ const props = defineProps({
 
 const QUEUE_NAMES = [
   'default', 'import', 'thumbnail', 'ml',
-  'download', 'scan', 'maintenance',
+  'download', 'scan', 'maintenance', 'maintenance_long',
 ]
 
 function formatDepth(name) {

frontend/src/stores/admin.js

@@ -119,7 +119,7 @@ export const useAdminStore = defineStore('admin', () => {
   // --- Task progress polling (taps FC-3i activity dashboard) --------
 
   /**
-   * Polls /api/system/activity/runs?queue=maintenance every 3s,
+   * Polls /api/system/activity/runs?celery_task_id=<id> every 3s,
    * resolves when a task_run row with the given celery task_id
    * reaches a terminal status (ok / error / timeout). Returns the
    * row. Times out after 30 min by default.
@@ -129,7 +129,9 @@ export const useAdminStore = defineStore('admin', () => {
     while (Date.now() < deadline) {
       const body = await api.get(
         '/api/system/activity/runs',
-        { params: { queue: 'maintenance', limit: 20 } },
+        // By id, not by lane: these jobs run on `maintenance_long`, and a
+        // lane filter here is one more copy of the routing table (#4432).
+        { params: { celery_task_id: taskId, limit: 1 } },
       )
       const row = (body.runs || []).find(r => r.celery_task_id === taskId)
       if (row && ['ok', 'error', 'timeout'].includes(row.status)) {

tests/test_api_system_activity.py

@@ -295,3 +295,11 @@ async def test_failures_only_within_24h_window(client, _seed_failures):
     body = await resp.get_json()
     ids = {r["error_type"] for r in body["recent"]}
     assert "OldError" not in ids
+
+
+@pytest.mark.asyncio
+async def test_runs_filter_by_celery_task_id(client, _seed_runs):
+    # How a page follows a job it started, without knowing its lane (#4432).
+    resp = await client.get("/api/system/activity/runs?celery_task_id=tid-3")
+    body = await resp.get_json()
+    assert [r["celery_task_id"] for r in body["runs"]] == ["tid-3"]

tests/test_celery_routing.py

@@ -2,6 +2,8 @@
 `maintenance_long` lane so a 30-min backup or a multi-chunk audit can't starve
 the quick recovery sweeps / vacuum on the concurrency-1 `maintenance` lane
 (operator-flagged 2026-06-07)."""
+import pytest
+
 from backend.app.celery_app import celery
 
 
@@ -20,23 +22,64 @@ def test_quick_maintenance_stays_on_maintenance():
     assert routes["backend.app.tasks.maintenance.*"]["queue"] == "maintenance"
 
 
-def test_queue_for_mirrors_external_to_download():
-    """celery_signals._queue_for is a hand-maintained mirror of task_routes
-    that stamps TaskRun.queue. external.* routes to the download lane, so the
-    mirror must agree — else TaskRun.queue lies 'default' for external fetches
-    and per-queue dashboard filters / threshold overrides miss them
-    (operator-flagged 2026-06-17)."""
+@pytest.mark.parametrize(("name", "queue"), [
+    ("backend.app.tasks.external.fetch_external_link", "download"),
+    # The rows #4432 found recorded on the wrong lane:
+    ("backend.app.tasks.translation.translate_posts", "maintenance_long"),
+    ("backend.app.tasks.gpu_queue.enqueue_gpu_backfill", "maintenance"),
+    ("backend.app.tasks.maintenance.backfill_phash", "maintenance_long"),
+    ("backend.app.tasks.admin.normalize_tags_task", "maintenance_long"),
+    ("backend.app.tasks.backup.backup_db_task", "maintenance_long"),
+    ("backend.app.tasks.maintenance.vacuum_analyze", "maintenance"),
+    ("backend.app.tasks.not_routed.anything", "default"),
+])
+def test_task_run_records_the_queue_the_router_sends_to(name, queue):
+    """TaskRun.queue comes from the router, not a copy of task_routes (#4432):
+    the stall sweep's per-queue thresholds and the System activity filters both
+    key off it."""
     from backend.app.celery_signals import _queue_for
 
     class _T:
-        name = "backend.app.tasks.external.fetch_external_link"
+        pass
 
-    assert _queue_for(_T()) == "download"
-    assert (
-        celery.conf.task_routes["backend.app.tasks.external.*"]["queue"]
-        == "download"
+    t = _T()
+    t.name = name
+    t.app = celery
+    assert _queue_for(t) == queue
+
+
+def test_no_task_outlives_its_stall_threshold():
+    """A task whose hard time limit is longer than the stall sweep's threshold
+    for it gets failed 'RecoverySweep' while it is still healthy — the class
+    #4432 found on the ml, import and long-maintenance lanes. The threshold is
+    resolved the way recover_stalled_task_runs resolves it: task-name override,
+    then queue, then the default."""
+    from backend.app.celery_signals import _queue_for
+    from backend.app.tasks.maintenance import (
+        QUEUE_STUCK_THRESHOLD_MINUTES,
+        STUCK_THRESHOLD_MINUTES,
+        TASK_STUCK_THRESHOLD_MINUTES,
     )
 
+    celery.loader.import_default_modules()
+    checked = 0
+    too_short = []
+    for name, task in sorted(celery.tasks.items()):
+        if name.startswith("celery."):
+            continue
+        limit = getattr(task, "time_limit", None)
+        if not limit:
+            continue
+        threshold = TASK_STUCK_THRESHOLD_MINUTES.get(
+            name,
+            QUEUE_STUCK_THRESHOLD_MINUTES.get(_queue_for(task), STUCK_THRESHOLD_MINUTES),
+        )
+        checked += 1
+        if limit / 60 > threshold:
+            too_short.append(f"{name}: limit {limit / 60:.0f} min > sweep {threshold} min")
+    assert checked > 20, "the task registry did not load — this guard checked nothing"
+    assert not too_short, "\n".join(too_short)
+
 
 def test_backfill_phash_runs_on_the_long_lane():
     """It lives in maintenance.py, so the quick-lane glob matches it too —