fix(gpu-jobs): end the error-tombstone loop — deliberate retry semantics + poison-job guards

The hourly ccip backfill's skip-list lacked 'error' (and the daily
siglip/embed variants re-gated failures on their missing results), so every
permanently-bad file got a fresh doomed job each run — ~24 duplicate error
rows/day per file, the perpetual 'unprocessable' flood. An errored job is now
a TOMBSTONE: no backfill re-enqueues it; retry is deliberate-only via
/retry_errors (an errored back-catalogue needs one button press after a
model swap).

One shared set of dedupe DELETEs (services/ml/gpu_jobs.error_dedupe_statements)
runs before every backfill and inside /retry_errors: error rows made moot by a
later pending/leased/done row go first, then older duplicates (newest reason
survives) — so the error count reads as distinct failing files and a retry
can't fan one file out into duplicate pending jobs. /retry_errors now returns
{requeued, pruned} and the toast shows both.

Poison-loop guards (release and lease-expiry burn no attempts, so a job that
stalls its transfer or crashes the agent every time cycled forever —
operator-observed jobs 99044/125288/131594/143131):
- agent: 3 in-session transient bounces (fetch or submit) → fail with the real
  reason instead of another release; strikes never count while stopping, and
  clear on submit success. Agent build 2026-07-02.3.
- server: the 60s orphan sweep (statements shared between the beat task and
  GpuJobService so they can't drift) converts expired leases with >=5 lease
  grants and pending jobs with >=10 to 'error', preserving the last stored
  failure reason. Backstops old agent builds.

Tests: tombstone rule across all three backfill variants, moot-row pruning,
poison conversions, and the extended /retry_errors dedupe contract.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM

backend/app/services/ml/gpu_jobs.py

@@ -12,8 +12,9 @@ and the lease itself reclaims expired leases as a final backstop. Result-writing
 
 from datetime import UTC, datetime, timedelta
 
-from sqlalchemy import and_, select, update
+from sqlalchemy import and_, delete, exists, func, or_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import aliased
 
 from ...models import GpuJob
 
@@ -24,6 +25,107 @@ DEFAULT_LEASE_TTL = 180     # seconds an agent holds a job before it can be re-l
 DEFAULT_BATCH = 8
 MAX_ATTEMPTS = 3
 
+# Poison-loop backstops. `attempts` counts LEASES GRANTED (incremented in
+# lease()), but fail()'s MAX_ATTEMPTS cap only fires when the agent reports a
+# failure — a job that keeps coming back via release() (transient handback) or
+# lease expiry (agent crash/wedge) never gets a verdict and would cycle forever.
+# The orphan sweep converts those to 'error': an expired lease that has already
+# been granted EXPIRED_POISON_CAP leases is presumed to kill/wedge the agent,
+# and a pending job granted PENDING_POISON_CAP leases without ever completing is
+# presumed poisoned (e.g. a transfer that stalls every time). Both stay
+# resurrectable via /retry_errors, which resets attempts.
+EXPIRED_POISON_CAP = MAX_ATTEMPTS + 2
+PENDING_POISON_CAP = 10
+
+
+def error_dedupe_statements():
+    """DELETEs enforcing: at most ONE error row per (image, task), and none that
+    a live or succeeded row makes moot. The 2026-07-02 tombstone loop (backfill
+    skip-lists lacked 'error') minted a duplicate error row per bad file per
+    hour; running these before every backfill and inside /retry_errors keeps the
+    error count reading as "distinct failing files" and stops a retry fanning
+    one file out into several duplicate pending jobs. Shared by the sync beat
+    task and the async API route so both prune by the SAME predicate.
+    Execution order matters: moot rows first, then older duplicates (the newest
+    error — the freshest reason — survives)."""
+    other = aliased(GpuJob)
+    same_pair = and_(
+        other.image_record_id == GpuJob.image_record_id,
+        other.task == GpuJob.task,
+    )
+    moot = (
+        delete(GpuJob)
+        .where(
+            GpuJob.status == "error",
+            exists().where(
+                same_pair, other.status.in_(["pending", "leased", "done"]),
+            ),
+        )
+        .execution_options(synchronize_session=False)
+    )
+    older_dupe = (
+        delete(GpuJob)
+        .where(
+            GpuJob.status == "error",
+            exists().where(
+                same_pair,
+                other.status == "error",
+                or_(
+                    other.updated_at > GpuJob.updated_at,
+                    and_(other.updated_at == GpuJob.updated_at,
+                         other.id > GpuJob.id),
+                ),
+            ),
+        )
+        .execution_options(synchronize_session=False)
+    )
+    return [moot, older_dupe]
+
+
+def recover_statements(now: datetime) -> dict:
+    """UPDATEs for the orphan sweep, keyed by outcome; insertion order IS the
+    required execution order ('recovered' must run after 'poison_expired', which
+    claims the crash-loopers out of the same expired-lease pool)."""
+    expired = and_(GpuJob.status == "leased", GpuJob.lease_expires_at < now)
+    unlease = {"lease_token": None, "leased_at": None, "lease_expires_at": None,
+               "updated_at": now}
+    return {
+        "poison_expired": (
+            update(GpuJob)
+            .where(expired, GpuJob.attempts >= EXPIRED_POISON_CAP)
+            .values(
+                status="error",
+                # Keep the job's last stored failure reason — it's the triage
+                # signal for WHY the loop happened.
+                error=func.concat(
+                    f"poisoned: lease expired after {EXPIRED_POISON_CAP}+ lease "
+                    "attempts (job repeatedly crashes or wedges the agent?); "
+                    "last error: ",
+                    func.coalesce(GpuJob.error, "none"),
+                ),
+                **unlease,
+            )
+        ),
+        "recovered": update(GpuJob).where(expired).values(
+            status="pending", **unlease,
+        ),
+        "poison_pending": (
+            update(GpuJob)
+            .where(GpuJob.status == "pending",
+                   GpuJob.attempts >= PENDING_POISON_CAP)
+            .values(
+                status="error",
+                error=func.concat(
+                    f"poisoned: {PENDING_POISON_CAP}+ lease attempts without "
+                    "ever completing (transfer stalls every time?); "
+                    "last error: ",
+                    func.coalesce(GpuJob.error, "none"),
+                ),
+                updated_at=now,
+            )
+        ),
+    }
+
 
 class GpuJobService:
     def __init__(self, session: AsyncSession):
@@ -170,16 +272,11 @@ class GpuJobService:
 
     async def recover_orphaned(self) -> int:
         """Reset every expired lease back to pending — catches agents that died
-        mid-job (no graceful release). Run on a short beat so the queue recovers
-        + reads honestly even when no worker is actively leasing. Returns rows
-        recovered."""
-        now = datetime.now(UTC)
-        res = await self.session.execute(
-            update(GpuJob)
-            .where(GpuJob.status == "leased", GpuJob.lease_expires_at < now)
-            .values(
-                status="pending", lease_token=None, leased_at=None,
-                lease_expires_at=None, updated_at=now,
-            )
-        )
-        return res.rowcount or 0
+        mid-job (no graceful release) — and convert poison-loopers to 'error'
+        (see the *_POISON_CAP rationale above). Run on a short beat so the queue
+        recovers + reads honestly even when no worker is actively leasing.
+        Returns rows recovered to pending (poison conversions are extra)."""
+        counts = {}
+        for name, stmt in recover_statements(datetime.now(UTC)).items():
+            counts[name] = (await self.session.execute(stmt)).rowcount or 0
+        return counts["recovered"]