2026-07-02 15:48:48 -04:00
43 changed files with 1628 additions and 1382 deletions
@@ -38,6 +38,10 @@ services:
      # spot + backs off under VRAM pressure). On by default; toggle live in the
      # control UI. Set to 0 to start in manual mode.
      AUTO_SCALE: ${AUTO_SCALE:-1}
      # Aggregate download cap in MB/s (stills + video streams combined), so the
      # agent can't saturate the desktop's network and wreck browsing — WiFi
      # especially. 0 = unlimited; tunable live in the control UI.
      BANDWIDTH_LIMIT_MB_S: ${BANDWIDTH_LIMIT_MB_S:-8}
      # Crop embedder (SigLIP concept bag): float16 keeps VRAM low on a shared
      # desktop GPU; the model itself is announced by the server.
      SIGLIP_DTYPE: ${SIGLIP_DTYPE:-float16}
@@ -21,7 +21,7 @@ log = logging.getLogger("fc_agent.app")
 # Bump on every agent change. The page embeds this and /status reports it; the UI
 # warns to reload when they differ — so a stale browser-cached page can't be
 # mistaken for "the new image didn't deploy". (Belt-and-braces with no-store.)
-VERSION = "2026-07-02.1 · short videos sample again (select, not fps) + ffmpeg errors logged"
+VERSION = "2026-07-02.4 · bandwidth governor: aggregate download cap (MB/s dial) so the agent can't saturate the desktop's network"
 logbuf.install()
 cfg = Config.from_env()
@@ -83,6 +83,13 @@ async def auto(request: Request):
    return JSONResponse(worker.status())
@app.post("/bandwidth")
 async def bandwidth(request: Request):
    body = await request.json()
    worker.set_bandwidth(float(body.get("value", 0)))
    return JSONResponse(worker.status())
@app.get("/gpu")
 def gpu():
    # GPU meters poll this on their own fast cadence. It only reads local
@@ -161,8 +168,9 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
 .step{background:#262c34;color:var(--fg);border:1px solid var(--bd);border-radius:8px;
  width:30px;height:32px;font:700 16px system-ui;cursor:pointer}
 .step:hover{border-color:var(--acc)}
- #conc{width:3.4rem;height:32px;text-align:center;font:700 16px system-ui;background:#11151a;
+ #conc,#bw{width:3.4rem;height:32px;text-align:center;font:700 16px system-ui;background:#11151a;
  color:var(--fg);border:1px solid var(--bd);border-radius:8px}
 .unit{color:var(--mut);font-size:12px;font-weight:600}
 .hint{color:var(--mut);font-size:12px;margin-top:12px}
 .tiles{display:grid;grid-template-columns:repeat(6,1fr);gap:8px;margin-bottom:16px}
 .tile{background:#13171d;border:1px solid var(--bd);border-radius:10px;padding:12px 8px;text-align:center}
@@ -216,6 +224,10 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
    <input id=conc type=number min=1 value=1 onchange="setv(this.value)">
    <button class=step onclick=setc(1)>+</button>
   </div>
   <div class=stepper title="aggregate download cap, downloads + video streams combined — 0 = unlimited">
    <input id=bw type=number min=0 step=1 value=8 onchange="setbw(this.value)">
    <span class=unit>MB/s</span>
   </div>
  </div>
  <div class=hint id=conchint>auto-tuning downloaders to keep the GPU fed · max 8</div>
 </section>
@@ -281,6 +293,11 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
  await fetch('/auto',{method:'POST',headers:{'Content-Type':'application/json'},
   body:JSON.stringify({value:on})});refresh()
 }
 async function setbw(v){
  v=Math.max(0,parseFloat(v)||0); bw.value=v
  await fetch('/bandwidth',{method:'POST',headers:{'Content-Type':'application/json'},
   body:JSON.stringify({value:v})});refresh()
 }
 async function refresh(){
  let s; try{ s=await (await fetch('/status')).json() }catch{ return }
  applyStatus(s)
@@ -318,6 +335,9 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
  // Instantaneous pool state → demoted to the sub-line, where its jumpiness reads
  // as live churn rather than a "broken" headline metric.
  pipe.textContent='downloaders '+(s.downloaders!=null?s.downloaders:'—')+' · consumers '+(s.consumers!=null?s.consumers:'—')+' · on GPU '+(s.active||0)
   +' · net '+(s.net_mb_s!=null?s.net_mb_s.toFixed(1):'—')+' MB/s'
   +(s.bandwidth_limit_mb_s>0?(' / cap '+s.bandwidth_limit_mb_s):'')
  if(document.activeElement!==bw && s.bandwidth_limit_mb_s!=null) bw.value=s.bandwidth_limit_mb_s
  // Buffer occupancy bar (also driven here so it tracks the /status cadence).
  if(s.buffer!=null && s.buffer_max){ const p=Math.round(100*s.buffer/s.buffer_max)
   buflbl.textContent=s.buffer+' / '+s.buffer_max; bufbar.style.width=p+'%' }
@@ -101,15 +101,26 @@ class FcClient:
            return
        self._post_quiet("/api/gpu/jobs/release", {"job_ids": job_ids})
-    def fetch_image(self, image_url: str) -> bytes:
+    def fetch_image(self, image_url: str, throttle=None) -> bytes:
        # image_url is a server-relative path ("/images/...").
        # timeout=(connect, read): the read timeout is BETWEEN-BYTES, not total,
        # so a large-but-flowing download still completes — but a stuck/dead
        # connection (curator overloaded) fails in 60s instead of hanging a
        # downloader for 180s and piling up concurrent stuck requests on curator.
-        r = self.s.get(f"{self.base}{image_url}", timeout=(10, 60))
+        # With a throttle (the worker's shared TokenBucket), the body is streamed
-        r.raise_for_status()
+        # in chunks and each chunk is charged to the global bandwidth budget —
-        return r.content
+        # pausing between reads lets TCP flow control pace curator's send side.
        with self.s.get(
            f"{self.base}{image_url}", timeout=(10, 60), stream=throttle is not None
        ) as r:
            r.raise_for_status()
            if throttle is None:
                return r.content
            buf = bytearray()
            for chunk in r.iter_content(chunk_size=262_144):
                throttle.take(len(chunk))
                buf.extend(chunk)
            return bytes(buf)
    def is_reachable(self) -> bool:
        """Cheap 'is curator responding at all right now?' check. Used to decide,
@@ -48,6 +48,9 @@ class Config:
    #                             higher keeps more frames, 0 disables
    ffmpeg_timeout: float  # hard ceiling (s) for ffmpeg-from-URL video sampling;
    #                        generous so a SLOW media link still completes
    bandwidth_limit_mb_s: float  # aggregate download cap in MEGABYTES/s across
    #                              all downloaders + video streams (0 = unlimited);
    #                              tunable live from the agent UI
    @classmethod
    def from_env(cls) -> Config:
@@ -77,4 +80,11 @@ class Config:
            dedupe_iou=float(os.environ.get("DEDUPE_IOU", "0.85")),
            frame_dedupe_distance=int(os.environ.get("FRAME_DEDUPE_DISTANCE", "8")),
            ffmpeg_timeout=float(os.environ.get("FFMPEG_TIMEOUT", "1200")),
            # Default 8 MB/s (~64 Mbit/s): ~20% of the measured ~300 Mbit/s home
            # WiFi, so browsing stays snappy while the agent works — yet MORE
            # sweep throughput than the self-inflicted congestion collapse this
            # replaces (2026-07-02: 8 unthrottled downloaders bufferbloated the
            # link to ~1-1.5 MB/s per stream, browser included). Raise it (or 0)
            # from the agent UI on wired/faster networks.
            bandwidth_limit_mb_s=float(os.environ.get("BANDWIDTH_LIMIT_MB_S", "8")),
        )
@@ -4,12 +4,15 @@ instances, each with a timestamp."""
 import io
 import logging
 import os
 import signal
 import subprocess
 import tempfile
 import time
 from PIL import Image, ImageFile
 from .throttle import PidReadMeter
 log = logging.getLogger("fc_agent.media")
 # Load slightly-truncated images (a few missing trailing bytes) instead of
@@ -111,6 +114,12 @@ def _collect_frames(
 def _terminate(proc: subprocess.Popen) -> None:
    """Stop an ffmpeg cleanly, then hard-kill if it ignores SIGTERM."""
    try:
        # A bandwidth-paused (SIGSTOPped) process can't receive SIGTERM until it
        # resumes — always CONT first so termination is prompt, not queued.
        proc.send_signal(signal.SIGCONT)
    except OSError:
        pass
    proc.terminate()
    try:
        proc.wait(timeout=2)
@@ -122,10 +131,34 @@ def _terminate(proc: subprocess.Popen) -> None:
            pass
 def _pause(proc: subprocess.Popen, seconds: float, should_stop) -> bool:
    """SIGSTOP ffmpeg for ~`seconds` of bandwidth debt, staying responsive to
    Stop. While paused, the kernel socket buffer fills and TCP flow control
    stalls curator's send side — that's the throttle. SIGCONT is ALWAYS sent
    before returning. False = a Stop arrived mid-pause."""
    try:
        proc.send_signal(signal.SIGSTOP)
    except OSError:
        return True                     # already exited — nothing to pause
    try:
        end = time.monotonic() + seconds
        while (left := end - time.monotonic()) > 0:
            if should_stop and should_stop():
                return False
            time.sleep(min(0.5, left))
        return True
    finally:
        try:
            proc.send_signal(signal.SIGCONT)
        except OSError:
            pass
 def sample_frames_from_url(
    url: str, interval_seconds: float, max_frames: int,
    *, headers: str = "", timeout: float = 1200.0, should_stop=None,
-) -> list[tuple[float, Image.Image]]:
+    governor=None,
 ) -> tuple[list[tuple[float, Image.Image]], str | None]:
    """Sample frames by pointing ffmpeg STRAIGHT at the media URL — it Range-reads
    only the video index + up to max_frames worth of content, so the agent never
    downloads the whole file (VR/4K originals run 800MB+ and would buffer ~1GB in
@@ -133,7 +166,15 @@ def sample_frames_from_url(
    the timeout is the per-video ceiling (a slow/reconnecting stream can otherwise
    run for minutes). `should_stop` is polled while ffmpeg runs so a Stop KILLS the
    subprocess at once — otherwise a downloader stuck in a long decode keeps the
-    agent "working" long after Stop. Empty on failure / stop / timeout."""
+    agent "working" long after Stop. `governor` (the worker's shared TokenBucket)
    meters ffmpeg's network reads from outside via /proc/<pid>/io and SIGSTOPs
    the process into budget, so video streaming honors the same aggregate
    bandwidth cap as still downloads.
    Returns (frames, reason): frames is empty on failure/stop/timeout, and
    `reason` then carries the SPECIFIC cause (ffmpeg's stderr tail / timeout) so
    the caller can put it in the job's error — a bare "no frames" hid a filter
    bug as "unprocessable" for weeks. None reason on success."""
    interval = max(0.5, float(interval_seconds or 4.0))
    cap = max(1, int(max_frames or 64))
    hdr = ["-headers", headers] if headers else []
@@ -163,6 +204,7 @@ def sample_frames_from_url(
                    cmd, stdin=subprocess.DEVNULL,
                    stdout=subprocess.DEVNULL, stderr=errf,
                )
                meter = PidReadMeter(proc.pid) if governor is not None else None
                # Poll rather than block, so a Stop (or the per-video timeout) can
                # kill a slow/wedged ffmpeg promptly instead of waiting it out.
                start = time.monotonic()
@@ -174,17 +216,30 @@ def sample_frames_from_url(
                        stopped = should_stop and should_stop()
                        if stopped or (time.monotonic() - start > timeout):
                            _terminate(proc)
-                            if not stopped:
+                            if stopped:
-                                log.warning("ffmpeg timed out after %.0fs: %s",
+                                return [], "stopped"
-                                            timeout, url)
+                            log.warning("ffmpeg timed out after %.0fs: %s",
-                            return []
+                                        timeout, url)
-        except (OSError, ValueError):
+                            return [], f"ffmpeg timed out after {timeout:.0f}s"
-            return []
+                        if meter is not None:
                            read = meter.delta()
                            if read is None:      # /proc gone → stop governing
                                meter = None
                            elif (debt := governor.charge(read)) > 0:
                                # Over budget: pause ffmpeg until the bucket
                                # recovers. Pause time counts toward `timeout`
                                # (it stays the wedge backstop either way).
                                if not _pause(proc, debt, should_stop):
                                    _terminate(proc)
                                    return [], "stopped"
        except (OSError, ValueError) as exc:
            return [], f"ffmpeg not runnable: {exc}"
        frames = _collect_frames(tmp, interval, cap)
        if not frames:
-            log.warning("ffmpeg produced no frames (exit %s) for %s — stderr: %s",
+            reason = f"ffmpeg exit {proc.returncode}: {_tail(errpath)}"
-                        proc.returncode, url, _tail(errpath))
+            log.warning("ffmpeg produced no frames for %s — %s", url, reason)
-        return frames
+            return [], reason
        return frames, None
 def _tail(path: str, limit: int = 300) -> str:
@@ -0,0 +1,111 @@
 """Global download-bandwidth governor (one token bucket for the whole agent).
 The agent lives on someone's desktop and shares that desktop's network —
 typically WiFi, where saturating the link doesn't just slow other apps: it
 bufferbloats the airtime (RTT 21→45ms) and collapses EVERY connection,
 the operator's browser included. Measured 2026-07-02: the idle link moved
 ~38 MB/s single-stream, but under the 8-downloader sweep every stream on the
 machine crawled at ~1-1.5 MB/s. So the cap is on the AGGREGATE, not per
 stream: still downloads pump their chunks through take(), and ffmpeg video
 streams — whose sockets live in a subprocess we can't wrap — are metered from
 outside via /proc/<pid>/io and paused (SIGSTOP) into budget using charge()'s
 debt signal; TCP flow control then stalls the sender while ffmpeg sleeps.
 Accounting is post-paid (charge the bytes first, then wait out any debt): the
 bytes have already crossed the network by the time we count them, and it means
 a chunk larger than one second of budget can never deadlock the bucket.
 Stdlib-only on purpose — unit-tested in CI, where the agent's ML deps
 don't exist.
 """
 import threading
 import time
 class TokenBucket:
    """Thread-safe token bucket in bytes/second. rate 0 = unlimited.
    `consumed` is the monotonic total of bytes charged (throttled or not) —
    the worker's rate loop derives the UI's "net MB/s" readout from it.
    """
    def __init__(self, rate_bytes_per_s: float = 0.0):
        self._cond = threading.Condition()
        self._rate = max(0.0, float(rate_bytes_per_s))
        # Burst = one second of budget: enough that chunked reads stay smooth,
        # small enough that a burst can't meaningfully lift the average.
        self._level = self._rate
        self._stamp = time.monotonic()
        self.consumed = 0
    @property
    def rate(self) -> float:
        return self._rate
    def set_rate(self, rate_bytes_per_s: float) -> None:
        """Retune live (the UI dial). Waiters re-check immediately, so raising
        the cap (or lifting it with 0) unblocks a mid-download wait at once."""
        with self._cond:
            self._refill_locked()          # settle elapsed time at the OLD rate
            self._rate = max(0.0, float(rate_bytes_per_s))
            self._level = min(self._level, self._rate)
            self._cond.notify_all()
    def _refill_locked(self) -> None:
        now = time.monotonic()
        self._level = min(self._rate, self._level + (now - self._stamp) * self._rate)
        self._stamp = now
    def take(self, n: int) -> None:
        """Charge n bytes and block until the budget recovers (stills path)."""
        with self._cond:
            self.consumed += n
            if self._rate <= 0:
                return
            self._refill_locked()
            self._level -= n
            while self._level < 0:
                # Wake early on set_rate; cap the wait so a big debt is paid in
                # re-checked slices rather than one long uninterruptible sleep.
                self._cond.wait(min(-self._level / self._rate, 0.5))
                if self._rate <= 0:
                    return
                self._refill_locked()
    def charge(self, n: int) -> float:
        """Charge n bytes WITHOUT blocking; return seconds of debt (0 = within
        budget). The ffmpeg governor can't block the subprocess's own reads, so
        it SIGSTOPs the process for (about) the returned debt instead."""
        with self._cond:
            self.consumed += n
            if self._rate <= 0:
                return 0.0
            self._refill_locked()
            self._level -= n
            return max(0.0, -self._level / self._rate)
 class PidReadMeter:
    """Cumulative read-bytes meter for a subprocess, via /proc/<pid>/io.
    `rchar` counts every read() syscall's bytes — for a streaming ffmpeg the
    network reads dominate, so the delta is a good-enough aggregate-bandwidth
    signal (it's a governor, not a billing meter). Returns None when /proc is
    unavailable (process exited, or a non-Linux host): the caller then simply
    doesn't govern — degrade to unthrottled rather than break video sampling.
    """
    def __init__(self, pid: int):
        self._path = f"/proc/{pid}/io"
        self._last = 0
    def delta(self) -> int | None:
        try:
            with open(self._path, "rb") as f:
                for line in f:
                    if line.startswith(b"rchar:"):
                        total = int(line.split()[1])
                        d, self._last = total - self._last, total
                        return max(0, d)
        except (OSError, ValueError):
            return None
        return None
@@ -42,6 +42,7 @@ from .client import FcClient
 from .config import Config
 from .crops import crop_region
 from .detectors import dedupe_crops
 from .throttle import TokenBucket
 # Cap on the lease-retry backoff: when curator is unreachable (e.g. you redeploy
 # it while away), a downloader retries leasing with exponential backoff up to this
@@ -49,6 +50,16 @@ from .detectors import dedupe_crops
 # restart needed.
 MAX_BACKOFF_SECONDS = 60.0
 # A job whose fetch dies transiently this many times IN ONE SESSION stops being
 # handed back and is failed instead. Transient handbacks (release) burn no
 # attempts on the server, so a poisoned transfer — an original that stalls the
 # download every single time — would otherwise release→re-lease forever, churning
 # bandwidth without ever landing in the error queue (operator-observed jobs
 # 99044/125288/131594/143131, 2026-07-01). Failing it lets curator's attempt cap
 # tombstone it WITH the real reason. A genuine curator outage is unaffected:
 # every job takes at most one strike per outage, and strikes clear on success.
 TRANSIENT_JOB_CAP = 3
 def _is_transient(exc: requests.RequestException) -> bool:
    """A server/transport problem (wait it out) vs a job-specific fault (fail it).
@@ -171,6 +182,10 @@ class Worker:
            f"Authorization: Bearer {cfg.token}\r\n" if cfg.token else ""
        )
        self._lock = threading.Lock()
        # ONE bandwidth budget for everything the agent pulls (still downloads
        # and ffmpeg video streams): the agent shares a desktop's network, so
        # the polite bound is on the aggregate — see throttle.py for why.
        self.throttle = TokenBucket(cfg.bandwidth_limit_mb_s * 1_048_576)
        self._running = False
        # The lifecycle state the UI shows (see the STOPPED/STARTING/... consts).
        # It stays STOPPING — a truthful "winding down" — from the Stop press until
@@ -192,6 +207,11 @@ class Worker:
        # all at once. Add on lease, discard on every terminal client call.
        self._held: set[int] = set()
        self._held_lock = threading.Lock()
        # job_id → in-session transient-handback count (guarded by _held_lock).
        # Cleared on success or terminal fail; KEPT across releases — persisting
        # through the release→re-lease cycle is what lets TRANSIENT_JOB_CAP
        # catch a poisoned transfer.
        self._transient_seen: dict[int, int] = {}
        self.processed = 0
        self.downloaded = 0       # jobs fetched+decoded into the buffer (monotonic);
        #                           feeds the server-side downloads/min rate below.
@@ -204,6 +224,7 @@ class Worker:
        # often the browser polls (see RATE_INTERVAL). Decay to 0 when work stops.
        self._jpm = 0.0
        self._dpm = 0.0
        self._net_mb_s = 0.0      # smoothed aggregate download rate (UI readout)
        self._util_smooth: float | None = None   # EWMA GPU util (set by control loop)
        # Curator queue snapshot, refreshed by a background poller so the UI
        # /status read is instant — never an inline curator HTTP call (which
@@ -235,6 +256,21 @@ class Worker:
        with self._held_lock:
            self._held.discard(job_id)
    def _strike(self, jid: int) -> int:
        """Record a transient bounce for this job; returns the in-session total
        (compared against TRANSIENT_JOB_CAP by both the fetch and submit paths)."""
        with self._held_lock:
            n = self._transient_seen.get(jid, 0) + 1
            self._transient_seen[jid] = n
            return n
    def _forget_strikes(self, jid: int) -> None:
        """Terminal outcome (submitted or failed) → this job's strike count no
        longer matters. Deliberately NOT called on release: strikes surviving
        the release→re-lease cycle is what makes TRANSIENT_JOB_CAP work."""
        with self._held_lock:
            self._transient_seen.pop(jid, None)
    def _release(self, job_ids: list[int]) -> None:
        """Hand still-held leases back to curator and drop them from the held set —
        the single hand-back path for both a downloader exiting (stop/shrink) with
@@ -253,6 +289,7 @@ class Worker:
        log.warning("job %s (image %s) %s: %s", jid, image_id, verb, str(exc)[:200])
        self.client.fail(jid, str(exc)[:500])
        self._unhold(jid)
        self._forget_strikes(jid)
    def _stopped(self, stop_evt: threading.Event) -> bool:
        """The shared 'should I bail now?' check — the worker is stopping (global
@@ -351,6 +388,7 @@ class Worker:
        this here (not in the browser) makes the numbers independent of the poll
        rate, so a throttled/unfocused tab still shows a real rate."""
        prev_p, prev_d, prev_t = self.processed, self.downloaded, time.monotonic()
        prev_b = self.throttle.consumed
        while True:
            time.sleep(RATE_INTERVAL)
            now = time.monotonic()
@@ -358,9 +396,12 @@ class Worker:
            if dt > 0:
                jp = max(0.0, 60.0 * (self.processed - prev_p) / dt)
                dp = max(0.0, 60.0 * (self.downloaded - prev_d) / dt)
                nb = max(0.0, (self.throttle.consumed - prev_b) / dt / 1_048_576)
                self._jpm = RATE_ALPHA * jp + (1 - RATE_ALPHA) * self._jpm
                self._dpm = RATE_ALPHA * dp + (1 - RATE_ALPHA) * self._dpm
                self._net_mb_s = RATE_ALPHA * nb + (1 - RATE_ALPHA) * self._net_mb_s
            prev_p, prev_d, prev_t = self.processed, self.downloaded, now
            prev_b = self.throttle.consumed
    # --- control -----------------------------------------------------------
    def start(self):
@@ -463,6 +504,14 @@ class Worker:
        with self._lock:
            self._auto = bool(on)
    def set_bandwidth(self, mb_s: float):
        # Live-retunes the shared bucket; a downloader blocked mid-wait re-checks
        # immediately (set_rate notifies), so raising the cap takes effect now.
        self.throttle.set_rate(max(0.0, float(mb_s)) * 1_048_576)
        log.info("bandwidth cap set to %s",
                 "unlimited" if self.throttle.rate <= 0
                 else f"{self.throttle.rate / 1_048_576:g} MB/s")
    def set_concurrency(self, n: int):
        # The UI dial tunes the DOWNLOADER count. A manual set is an override →
        # leave Auto so the autoscaler stops fighting the operator.
@@ -533,6 +582,8 @@ class Worker:
            "downloads_per_min": round(self._dpm, 1),
            "errors": self.errors,
            "transient": self.transient,
            "bandwidth_limit_mb_s": round(self.throttle.rate / 1_048_576, 1),
            "net_mb_s": round(self._net_mb_s, 1),       # observed aggregate rate
        }
    def _bump(self, *, processed=0, downloaded=0, errors=0, active=0, transient=0):
@@ -580,6 +631,20 @@ class Worker:
                except requests.RequestException as exc:
                    owned.remove(jid)
                    if _is_transient(exc):
                        # A Stop mid-fetch lands here too (a killed ffmpeg is
                        # not the job's fault) — no strike for that; only count
                        # bounces taken while genuinely running.
                        if (not self._stopped(stop_evt)
                                and self._strike(jid) >= TRANSIENT_JOB_CAP):
                            # THIS job keeps dying while others move: a poisoned
                            # transfer, not a curator outage. Fail it so the
                            # server's attempt cap tombstones it with the real
                            # reason instead of cycling it forever.
                            self._fail(
                                jid, job.get("image_id"), exc,
                                verb="gave up after repeated transient failures",
                            )
                            continue
                        # curator down/redeploying or our lease was reclaimed —
                        # NOT the job's fault. Hand back this job + the rest of the
                        # batch and back the whole loop off.
@@ -632,11 +697,12 @@ class Worker:
            # mid-download was the failure loop). Environment-agnostic + resilient.
            url = f"{self.cfg.fc_url}{job['image_url']}"
            with self._timed("decode"):
-                frames = media.sample_frames_from_url(
+                frames, ffmpeg_err = media.sample_frames_from_url(
                    url, job.get("frame_interval_seconds", 4.0),
                    job.get("max_frames", 64),
                    headers=self._auth_header, timeout=self.cfg.ffmpeg_timeout,
                    should_stop=lambda: self._stopped(stop_evt),
                    governor=self.throttle,
                )
            if not frames:
                # Stop killed ffmpeg → NOT the job's fault; raise transient so the
@@ -644,11 +710,14 @@ class Worker:
                if self._stopped(stop_evt):
                    raise requests.ConnectionError("stopped during video sampling")
                # Else couldn't sample. If curator is up, the file is unprocessable
-                # → a job fault (fail it, don't re-lease forever). If curator is
+                # → a job fault: fail it WITH ffmpeg's reason, so the job's stored
-                # unreachable, it's transient → let the loop back off + retry
+                # error says e.g. "moov atom not found" instead of a bare
-                # (survives a redeploy). ConnectionError is caught as transient.
+                # "unprocessable". If curator is unreachable, it's transient → let
                # the loop back off + retry (ConnectionError is caught as such).
                if self.client.is_reachable():
-                    raise RuntimeError("no frames sampled from video (unprocessable)")
+                    raise RuntimeError(
                        f"no frames sampled from video — {ffmpeg_err or 'unknown reason'}"
                    )
                raise requests.ConnectionError("curator unreachable during video sampling")
            # Temporal dedup: a near-static video re-runs the whole detect+embed
            # chain on ~identical frames — drop near-dups HERE (CPU) pre-GPU.
@@ -661,7 +730,7 @@ class Worker:
            return frames
        # Stills: download the bytes and decode.
        with self._timed("download"):
-            data = self.client.fetch_image(job["image_url"])
+            data = self.client.fetch_image(job["image_url"], throttle=self.throttle)
        with self._timed("decode"):
            frames = [(None, media.load_image(data))]
        return frames
@@ -739,6 +808,7 @@ class Worker:
                with self._timed("submit"):
                    self.client.submit_embedding(jid, vec, embed_version)
                self._unhold(jid)
                self._forget_strikes(jid)
                return True
            # task picks what to produce per crop:
@@ -851,9 +921,18 @@ class Worker:
            with self._timed("submit"):
                self.client.submit(jid, regions, replace_kinds)
            self._unhold(jid)
            self._forget_strikes(jid)
            return True
        except requests.RequestException as exc:
            if _is_transient(exc):
                if (not self._stopped(stop_evt)
                        and self._strike(jid) >= TRANSIENT_JOB_CAP):
                    # Same poison rationale as the fetch path: a job whose
                    # submit keeps dying transiently would otherwise re-lease →
                    # re-download → re-GPU forever.
                    self._fail(jid, job.get("image_id"), exc,
                               verb="gave up after repeated transient failures")
                    return False
                # curator down/redeploying, a 5xx, or our lease was reclaimed
                # while we worked. NOT the job's fault — hand it back (best
                # effort; then the server's orphan-recovery reclaims it if down).
@@ -0,0 +1,32 @@
 """gpu_job.triage_status — the probe's verdict on an errored job's FILE
 Failure triage (#125): a periodic sweep probes each errored image's file
 (sha256 + decode, verify_integrity's machinery) exactly once and stores the
 verdict here — 'defect' (the file is bad: recovery material, excluded from
 /retry_errors) or 'file_ok' (failure was operational, safe to retry). NULL
 means not yet probed; selecting on NULL is what makes the sweep resumable.
 No index: the errored slice the sweep scans is tiny by design (tombstones).
 Revision ID: 0072
 Revises: 0071
 Create Date: 2026-07-02
 """
 from typing import Sequence, Union
 import sqlalchemy as sa
 from alembic import op
 revision: str = "0072"
 down_revision: Union[str, None] = "0071"
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 def upgrade() -> None:
    op.add_column(
        "gpu_job", sa.Column("triage_status", sa.String(16), nullable=True)
    )
 def downgrade() -> None:
    op.drop_column("gpu_job", "triage_status")
@@ -0,0 +1,46 @@
 """drop tag_eval_run — the head-vs-centroid eval harness is retired
 The eval (#1130) existed to prove the heads tagging spine on the operator's own
 data. It did; the operator accepted the system and retired the harness
 (2026-07-02) — card, API, task, model and this table all go. The eval's data
 loaders + metric helpers live on in services/ml/training_data.py, where the
 production heads trainer uses them nightly.
 Revision ID: 0073
 Revises: 0072
 Create Date: 2026-07-02
 """
 from typing import Sequence, Union
 import sqlalchemy as sa
 from alembic import op
 from sqlalchemy.dialects import postgresql
 revision: str = "0073"
 down_revision: Union[str, None] = "0072"
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 def upgrade() -> None:
    op.drop_index("ix_tag_eval_run_status", table_name="tag_eval_run")
    op.drop_table("tag_eval_run")
 def downgrade() -> None:
    # Recreates the shape from 0056 (data is not restorable).
    op.create_table(
        "tag_eval_run",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("params", postgresql.JSONB(), nullable=False),
        sa.Column("status", sa.String(length=16), nullable=False,
                  server_default="running"),
        sa.Column("started_at", sa.DateTime(timezone=True), nullable=False,
                  server_default=sa.func.now()),
        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("report", postgresql.JSONB(), nullable=True),
        sa.Column("error", sa.Text(), nullable=True),
        sa.Column("last_progress_at", sa.DateTime(timezone=True),
                  nullable=True),
    )
    op.create_index("ix_tag_eval_run_status", "tag_eval_run", ["status"])
@@ -38,7 +38,6 @@ def all_blueprints() -> list[Blueprint]:
    from .suggestions import suggestions_bp
    from .system_activity import system_activity_bp
    from .system_backup import system_backup_bp
    from .tag_eval import tag_eval_bp
    from .tags import tags_bp
    from .thumbnails import thumbnails_bp
    return [
@@ -58,7 +57,6 @@ def all_blueprints() -> list[Blueprint]:
        import_admin_bp,
        suggestions_bp,
        aliases_bp,
        tag_eval_bp,
        heads_bp,
        gpu_bp,
        ccip_bp,
@@ -1,13 +1,12 @@
 """FC-3k: /api/admin — destructive admin actions.
-Five action surfaces:
+Action surfaces:
  POST   /api/admin/artists/<slug>/cascade-delete  (Tier C)
  POST   /api/admin/images/bulk-delete             (Tier C)
  DELETE /api/admin/tags/<int:tag_id>              (Tier B)
  POST   /api/admin/tags/<int:dest_id>/merge       (Tier B)
  POST   /api/admin/tags/prune-unused              (Tier A)
  POST   /api/admin/posts/prune-bare               (Tier A)
  POST   /api/admin/tags/purge-legacy              (Tier A)
  GET    /api/admin/tags/<int:tag_id>/usage-count  (helper)
 Tier-C ops take a dry_run body flag (returns projection inline,
@@ -277,19 +276,6 @@ async def posts_reconcile_duplicates():
    return await _run_dry_run_op(reconcile_duplicate_posts, source_id=source_id)
@admin_bp.route("/tags/purge-legacy", methods=["POST"])
 async def tags_purge_legacy():
    """Tier-A: delete legacy IR-migration tags — archive/post/artist
    kinds (e.g. `BlenderKnight:Hannah_BJ_Loops`) PLUS general tags with
    a legacy name prefix (`source:*`, from IR's source kind that fell
    back to general). dry-run preview returns per-kind + per-prefix
    counts + a sample so the UI shows exactly what'll go before the
    operator confirms with dry_run=false."""
    from ..services.cleanup_service import purge_legacy_tags
    return await _run_dry_run_op(purge_legacy_tags)
@admin_bp.route("/tags/reset-content", methods=["POST"])
 async def tags_reset_content():
    """Tier-A: delete ALL general + character tags (the Camie-suggestable
@@ -9,19 +9,25 @@ homelab admin.
 """
 import secrets
 from pathlib import Path
 from quart import Blueprint, jsonify, request
-from sqlalchemy import func, select, update
+from sqlalchemy import func, or_, select, update
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from ..extensions import get_session
 from ..models import AppSetting, GpuJob, ImageRecord, MLSettings
 from ..services.gallery_service import image_url
-from ..services.ml.gpu_jobs import GpuJobService
+from ..services.ml.gpu_jobs import GpuJobService, error_dedupe_statements
 from ..services.ml.gpu_triage import classify_reason, recover_defective_image
 from ..services.ml.regions import RegionService
 gpu_bp = Blueprint("gpu", __name__, url_prefix="/api/gpu")
 # Same container mount the maintenance tasks use (tasks/admin.py) — recovery
 # deletes the defective original + thumbnail under it.
 _IMAGES_ROOT = Path("/images")
 _TOKEN_KEY = "gpu_agent_token"
@@ -115,19 +121,125 @@ async def retry_errors():
    recovery after an agent-side fix (e.g. the short-video sampler), where
    /reprocess would needlessly re-run the whole done library too. Attempts and
    the stored error reset so each job gets its full retry budget under the
-    fixed pipeline. Small row count (errors only) → inline UPDATE, and the
+    fixed pipeline. Stale tombstones are pruned FIRST (loop-era duplicates and
-    response carries the number requeued for the UI toast."""
+    rows a later success made moot — the same statements the backfills run), so
    one failing file requeues as ONE job, never a fan-out of duplicates. Small
    row count (errors only) → inline statements; the response carries the
    counts for the UI toast. Triage-confirmed defects are NOT requeued (see
    the WHERE below) — they stay on the recovery surface."""
    async with get_session() as session:
        pruned = 0
        for stmt in error_dedupe_statements():
            pruned += (await session.execute(stmt)).rowcount or 0
        res = await session.execute(
            update(GpuJob)
-            .where(GpuJob.status == "error")
+            .where(
                GpuJob.status == "error",
                # Triage-confirmed DEFECTS stay errored: the integrity probe
                # already proved the FILE is bad, so re-running the job just
                # burns agent time re-minting the same tombstone — those go
                # through /errors/<id>/recover instead.
                or_(GpuJob.triage_status.is_(None),
                    GpuJob.triage_status != "defect"),
            )
            .values(
                status="pending", attempts=0, error=None, lease_token=None,
-                leased_at=None, lease_expires_at=None, updated_at=func.now(),
+                leased_at=None, lease_expires_at=None, triage_status=None,
                updated_at=func.now(),
            )
        )
        kept = (
            await session.execute(
                select(func.count()).select_from(GpuJob)
                .where(GpuJob.status == "error")
            )
        ).scalar_one()
        await session.commit()
-    return jsonify({"requeued": res.rowcount or 0})
+    return jsonify({
        "requeued": res.rowcount or 0, "pruned": pruned, "defects_kept": kept,
    })
 # --- Failure triage + recovery (#125) ------------------------------------
@gpu_bp.route("/errors", methods=["GET"])
 async def errors():
    """The triage view of the error tombstones: every errored job joined with
    its image's integrity verdict, bucketed by reason for the overview. The
    probe sweep (triage_gpu_errors, 15-min beat) fills triage_status; 'defect'
    rows are the recovery surface's list."""
    async with get_session() as session:
        rows = (
            await session.execute(
                select(
                    GpuJob.id, GpuJob.image_record_id, GpuJob.task,
                    GpuJob.error, GpuJob.triage_status, GpuJob.updated_at,
                    ImageRecord.integrity_status, ImageRecord.mime,
                    ImageRecord.path, ImageRecord.thumbnail_path,
                )
                .join(ImageRecord, ImageRecord.id == GpuJob.image_record_id)
                .where(GpuJob.status == "error")
                .order_by(GpuJob.updated_at.desc())
                .limit(500)
            )
        ).all()
        total = (
            await session.execute(
                select(func.count()).select_from(GpuJob)
                .where(GpuJob.status == "error")
            )
        ).scalar_one()
    by_class: dict[str, int] = {}
    triage = {"defect": 0, "file_ok": 0, "unclassified": 0}
    items = []
    for r in rows:
        cls = classify_reason(r.error)
        by_class[cls] = by_class.get(cls, 0) + 1
        bucket = r.triage_status or "unclassified"
        triage[bucket] = triage.get(bucket, 0) + 1
        items.append({
            "job_id": r.id,
            "image_id": r.image_record_id,
            "task": r.task,
            "error": r.error,
            "reason_class": cls,
            "triage_status": r.triage_status,
            "integrity_status": r.integrity_status,
            "mime": r.mime,
            "image_url": image_url(r.path),
            "thumbnail_url": (
                image_url(r.thumbnail_path) if r.thumbnail_path else None
            ),
            "updated_at": r.updated_at.isoformat() if r.updated_at else None,
        })
    return jsonify({
        "total": total, "by_class": by_class, "triage": triage, "items": items,
    })
@gpu_bp.route("/errors/triage", methods=["POST"])
 async def errors_triage():
    """Run the probe sweep NOW (the card's button) instead of waiting out the
    15-minute beat cadence."""
    from ..tasks.maintenance import triage_gpu_errors
    r = triage_gpu_errors.delay()
    return jsonify({"celery_task_id": r.id}), 202
@gpu_bp.route("/errors/<int:image_id>/recover", methods=["POST"])
 async def errors_recover(image_id: int):
    """Recover a defect-triaged original: delete the bad copy + record and
    re-poll its subscription Source (a fresh fetch re-imports the file, which
    re-enters the GPU pipeline). Returns status 'no_source' when nothing
    pollable resolves — the file needs manual replacement there."""
    async with get_session() as session:
        result = await session.run_sync(
            lambda s: recover_defective_image(
                s, image_id, images_root=_IMAGES_ROOT,
            )
        )
    return jsonify(result)
 # --- Agent (bearer token): lease / submit / heartbeat / fail ------------
@@ -1,70 +0,0 @@
 """Tag-eval API (#1130): trigger + revisit the head-vs-centroid eval.
 The run + full report live in the tag_eval_run row, so the admin card rehydrates
 from GET (history / detail) on mount — the report survives navigation rather than
 living in transient frontend state.
 """
 from quart import Blueprint, jsonify, request
 from sqlalchemy import select
 from ..extensions import get_session
 from ..models import TagEvalRun
 from ..services.ml.tag_eval import EvalAlreadyRunning, start_tag_eval_run
 tag_eval_bp = Blueprint("tag_eval", __name__, url_prefix="/api/tag-eval")
 def _serialize(run: TagEvalRun, *, include_report: bool) -> dict:
    out = {
        "id": run.id,
        "params": run.params,
        "status": run.status,
        "started_at": run.started_at.isoformat() if run.started_at else None,
        "finished_at": run.finished_at.isoformat() if run.finished_at else None,
        "error": run.error,
    }
    if include_report:
        out["report"] = run.report
    return out
@tag_eval_bp.route("", methods=["POST"])
 async def create():
    body = await request.get_json(silent=True) or {}
    params = body.get("params") or body or {}
    async with get_session() as session:
        try:
            run_id = await session.run_sync(
                lambda s: start_tag_eval_run(s, params)
            )
        except EvalAlreadyRunning as running:
            return jsonify({
                "error": "eval_already_running",
                "running_id": int(running.args[0]),
            }), 409
        await session.commit()
    return jsonify({"run_id": run_id, "status": "running"}), 202
@tag_eval_bp.route("", methods=["GET"])
 async def history():
    try:
        limit = min(int(request.args.get("limit", "20")), 100)
    except ValueError:
        return jsonify({"error": "invalid_limit"}), 400
    async with get_session() as session:
        rows = (await session.execute(
            select(TagEvalRun).order_by(TagEvalRun.id.desc()).limit(limit)
        )).scalars().all()
        # List is light — no full report (the detail endpoint carries it).
        return jsonify({"runs": [_serialize(r, include_report=False) for r in rows]})
@tag_eval_bp.route("/<int:run_id>", methods=["GET"])
 async def detail(run_id: int):
    async with get_session() as session:
        run = await session.get(TagEvalRun, run_id)
        if run is None:
            return jsonify({"error": "not_found"}), 404
        return jsonify(_serialize(run, include_report=True))
@@ -97,10 +97,6 @@ def make_celery() -> Celery:
                "task": "backend.app.tasks.maintenance.cleanup_old_tasks",
                "schedule": 86400.0,  # daily
            },
            "ml-backfill-daily": {
                "task": "backend.app.tasks.ml.backfill",
                "schedule": 86400.0,
            },
            "train-heads-nightly": {
                "task": "backend.app.tasks.ml.scheduled_train_heads",
                "schedule": 86400.0,  # passive cadence; manual retrain stays available
@@ -113,15 +109,19 @@ def make_celery() -> Celery:
                "task": "backend.app.tasks.ml.recover_orphaned_gpu_jobs",
                "schedule": 60.0,  # quick pickup of work a dead agent orphaned
            },
            "triage-gpu-errors": {
                "task": "backend.app.tasks.maintenance.triage_gpu_errors",
                "schedule": 900.0,  # probe errored jobs' files → defect/file_ok
            },
            "enqueue-ccip-backfill-hourly": {
                "task": "backend.app.tasks.ml.enqueue_gpu_backfill",
-                "schedule": 3600.0,  # auto-feed new images (+ retry errored) so
+                "schedule": 3600.0,  # auto-feed NEW images; errored are
-                "args": ("ccip",),   # the queue keeps moving without the button
+                "args": ("ccip",),   # tombstoned — retry is the button only
            },
            "enqueue-siglip-backfill-daily": {
                "task": "backend.app.tasks.ml.enqueue_gpu_backfill",
-                "schedule": 86400.0,  # drain the concept-crop back-catalogue +
+                "schedule": 86400.0,  # drain the concept-crop back-catalogue
-                "args": ("siglip",),  # retry failed embeds, no button needed
+                "args": ("siglip",),  # (errored are tombstoned, not retried)
            },
            "enqueue-embed-backfill-daily": {
                "task": "backend.app.tasks.ml.enqueue_gpu_backfill",
@@ -183,10 +183,6 @@ def make_celery() -> Celery:
                "task": "backend.app.tasks.maintenance.recover_stalled_library_audit_runs",
                "schedule": 300.0,
            },
            "recover-stalled-tag-eval-runs": {
                "task": "backend.app.tasks.maintenance.recover_stalled_tag_eval_runs",
                "schedule": 300.0,
            },
            "recover-stalled-head-training-runs": {
                "task": "backend.app.tasks.maintenance.recover_stalled_head_training_runs",
                "schedule": 300.0,
@@ -33,7 +33,6 @@ from .subscribestar_failed_media import SubscribeStarFailedMedia
 from .subscribestar_seen_media import SubscribeStarSeenMedia
 from .tag import Tag, TagKind, image_tag
 from .tag_alias import TagAlias
 from .tag_eval_run import TagEvalRun
 from .tag_head import TagHead
 from .tag_positive_confirmation import TagPositiveConfirmation
 from .tag_suggestion_rejection import TagSuggestionRejection
@@ -75,7 +74,6 @@ __all__ = [
    "HeadMetricsSnapshot",
    "HeadTrainingRun",
    "TagAlias",
    "TagEvalRun",
    "TagHead",
    "TagPositiveConfirmation",
    "TagSuggestionRejection",
@@ -62,6 +62,11 @@ class GpuJob(Base):
    )
    attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    error: Mapped[str | None] = mapped_column(Text, nullable=True)
    # Triage verdict for an ERRORED job (#125): NULL = not yet probed;
    # 'defect' = the integrity probe says the FILE itself is bad (surfaced for
    # recovery, excluded from /retry_errors); 'file_ok' = the file passes —
    # the failure was operational (timeout/transient), safe to retry.
    triage_status: Mapped[str | None] = mapped_column(String(16), nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
@@ -1,7 +1,7 @@
 """HeadTrainingRun — persisted lifecycle of a head-training batch (#114).
-Mirrors TagEvalRun so the run SURVIVES navigation and the admin card can show
+A persisted run row (not transient frontend state) so the run SURVIVES
-live + historical status instead of holding it in transient frontend state.
+navigation and the admin card can show live + historical status.
 Training is idempotent (it upserts tag_head rows), so a SIGKILL'd run is harmless
 — a maintenance recovery sweep flips a stalled `running` row to `error`, and the
 next run re-trains. State machine: running → ready / error.
@@ -37,8 +37,8 @@ class HeadTrainingRun(Base):
    n_trained: Mapped[int | None] = mapped_column(Integer, nullable=True)
    n_skipped: Mapped[int | None] = mapped_column(Integer, nullable=True)
    error: Mapped[str | None] = mapped_column(Text, nullable=True)
-    # Last time the task made progress — the recovery sweep tells a live run from
+    # Last time the task made progress — the recovery sweep tells a live run
-    # a SIGKILL'd one by this (mirrors TagEvalRun).
+    # from a SIGKILL'd one by this.
    last_progress_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
@@ -1,45 +0,0 @@
 """TagEvalRun — persisted lifecycle of a head-vs-centroid tagging eval (#1130).
 Mirrors LibraryAuditRun so the result SURVIVES navigation: the run + its full
 report live in this row, and the admin card rehydrates from it on mount instead
 of holding the report in transient frontend state. State machine:
 running → ready / error. The async ml-queue task writes `report` (JSONB) when
 done; a maintenance recovery sweep flips a stalled `running` row to `error`.
 """
 from datetime import datetime
 from typing import Any
 from sqlalchemy import DateTime, Integer, String, Text, func
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Mapped, mapped_column
 from .base import Base
 class TagEvalRun(Base):
    __tablename__ = "tag_eval_run"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    # The eval parameters: {concepts: [...], curve_points: [...], neg_ratio,
    # cv_folds, ...} — echoed back so the report is self-describing.
    params: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="running", index=True,
    )
    # running | ready | error
    started_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    finished_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True,
    )
    # The full result: per-concept metrics (head vs centroid), learning-curve
    # points, and example image ids. Null until the task finishes.
    report: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True)
    error: Mapped[str | None] = mapped_column(Text, nullable=True)
    # Last time the task made progress — the recovery sweep tells a live run
    # from a SIGKILL'd one by this (mirrors LibraryAuditRun).
    last_progress_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True,
    )
@@ -718,71 +718,6 @@ def reconcile_duplicate_posts(
    return {"groups": len(groups), "merged": losers_total, "sample": sample}
 # Legacy tags FC no longer uses, in two shapes:
 #   (1) kinds the tag input never produces — archive/post/artist.
 #       provenance (post grouping) + archive membership are their own
 #       systems now, and artists are first-class Artist/Source rows.
 #       meta/rating were already hard-deleted by alembic 0023.
 #   (2) name prefixes from IR kinds FC never adopted — `source:*`.
 #       ImageRepo had a `source` kind; FC's enum doesn't, so ir_ingest
 #       fell those back to `general` (kind=general, name="source:patreon"
 #       etc.). They can't be caught by kind, so we match the name prefix.
 PURGEABLE_TAG_KINDS = ("archive", "post", "artist")
 LEGACY_NAME_PREFIXES = ("source:",)
 def _legacy_tag_predicate():
    name_clauses = [Tag.name.like(f"{p}%") for p in LEGACY_NAME_PREFIXES]
    return or_(Tag.kind.in_(PURGEABLE_TAG_KINDS), *name_clauses)
 def purge_legacy_tags(session: Session, *, dry_run: bool = False) -> dict:
    """Count (dry_run) or delete legacy IR-migration tags: archive/post/
    artist-kind tags PLUS general tags whose name matches a legacy
    prefix (source:*).
    CASCADE on image_tag / tag_alias / tag_suggestion_rejection / series_page
    clears the related rows on the parent DELETE.
    Returns:
      {"by_kind": {kind: count, ...},        # kind-matched rows
       "by_prefix": {"source:*": count},     # name-prefix-matched rows
       "count": total, "sample_names": [first 50],
       and on live runs "deleted": total}
    """
    predicate = _legacy_tag_predicate()
    rows = session.execute(
        select(Tag.id, Tag.name, Tag.kind).where(predicate)
    ).all()
    by_kind: dict[str, int] = {}
    by_prefix: dict[str, int] = {}
    for _id, name, kind in rows:
        # Classify by name-prefix first so a source:* row counts once,
        # under the prefix bucket, regardless of its (general) kind.
        matched_prefix = next(
            (p for p in LEGACY_NAME_PREFIXES if name.startswith(p)), None,
        )
        if matched_prefix is not None:
            label = f"{matched_prefix}*"
            by_prefix[label] = by_prefix.get(label, 0) + 1
        else:
            key = kind.value if hasattr(kind, "value") else str(kind)
            by_kind[key] = by_kind.get(key, 0) + 1
    sample = [name for _id, name, _kind in rows[:50]]
    total = len(rows)
    result = {
        "by_kind": by_kind, "by_prefix": by_prefix,
        "count": total, "sample_names": sample,
    }
    if dry_run:
        return result
    if total:
        session.execute(Tag.__table__.delete().where(predicate))
        session.commit()
    result["deleted"] = total
    return result
 # The CONTENT vocabulary. "Reset content tagging" wipes these so the operator
 # can re-tag from scratch. fandom + series (and series_page ordering) are
 # deliberately NOT here — they're kept.
@@ -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
+        mid-job (no graceful release) — and convert poison-loopers to 'error'
-        + reads honestly even when no worker is actively leasing. Returns rows
+        (see the *_POISON_CAP rationale above). Run on a short beat so the queue
-        recovered."""
+        recovers + reads honestly even when no worker is actively leasing.
-        now = datetime.now(UTC)
+        Returns rows recovered to pending (poison conversions are extra)."""
-        res = await self.session.execute(
+        counts = {}
-            update(GpuJob)
+        for name, stmt in recover_statements(datetime.now(UTC)).items():
-            .where(GpuJob.status == "leased", GpuJob.lease_expires_at < now)
+            counts[name] = (await self.session.execute(stmt)).rowcount or 0
-            .values(
+        return counts["recovered"]
                status="pending", lease_token=None, leased_at=None,
                lease_expires_at=None, updated_at=now,
            )
        )
        return res.rowcount or 0
@@ -0,0 +1,156 @@
 """GPU-failure triage (#125): classify errored jobs, PROBE the file, recover.
 An errored GPU job is a tombstone with a stored reason, but the reason alone is
 a suspicion, not a verdict — a timeout can hit a perfectly fine file, and
 "moov atom not found" can mean a truncated download OR a one-off transfer
 fault. So triage EVALUATES: it runs the real integrity probe (sha256 recompute
 + PIL/ffprobe — verify_integrity's own machinery) on each errored image ONCE
 and records both verdicts:
  ImageRecord.integrity_status  <- file-level verdict (ok / corrupt / ...)
  GpuJob.triage_status          <- 'defect'  (file is bad: recovery material,
                                              excluded from /retry_errors)
                                   'file_ok' (file passes: the failure was
                                              operational, safe to retry)
 Recovery reuses established primitives: delete the defective copy + record
 (cleanup_service.delete_images — full cascade) and re-poll the image's
 subscription Source (the Layer-2 refetch pattern: gallery-dl re-fetches the
 now-absent file on the next source check). Images without a pollable Source
 report 'no_source' — manual remediation. Every classification is logged at
 WARNING so the operator notices in Logs / System Activity.
 """
 import logging
 import time
 from datetime import UTC, datetime
 from pathlib import Path
 from sqlalchemy import select, update
 from sqlalchemy.orm import Session
 from ...models import GpuJob, ImageProvenance, ImageRecord, Source
 from ..cleanup_service import delete_images
 log = logging.getLogger(__name__)
 # Reason buckets for the triage overview (reporting only — the PROBE decides
 # 'defect', never the string). Ordered: first match wins.
 _REASON_BUCKETS = (
    ("poisoned", ("poisoned:",)),
    ("transient", ("gave up after repeated transient", "curator unreachable",
                   "connection", "read timed out")),
    ("timeout", ("timed out", "timeout")),
    ("truncated_or_corrupt", ("moov atom", "invalid data", "end of file",
                              "header missing", "error reading header",
                              "truncated", "premature", "corrupt",
                              "no frames sampled")),
    ("decode", ("cannot identify", "decompression", "broken data stream",
                "unrecognized data")),
 )
 def classify_reason(error: str | None) -> str:
    """Bucket a stored job-error string for the overview table."""
    text = (error or "").lower()
    if not text:
        return "other"
    for bucket, needles in _REASON_BUCKETS:
        if any(n in text for n in needles):
            return bucket
    return "other"
 def triage_errored_jobs(
    session: Session, *, time_budget_seconds: float = 300.0,
 ) -> dict:
    """Probe every not-yet-triaged errored image and write both verdicts.
    Time-boxed (sha256 of a large original over NFS can take tens of seconds)
    and inherently resumable: rows are selected by `triage_status IS NULL`, so
    the next sweep continues exactly where a budget cut stopped. Commits per
    image so a mid-run crash keeps completed verdicts."""
    image_ids = session.execute(
        select(GpuJob.image_record_id)
        .where(GpuJob.status == "error", GpuJob.triage_status.is_(None))
        .group_by(GpuJob.image_record_id)
        .order_by(GpuJob.image_record_id)
    ).scalars().all()
    counts = {"probed": 0, "defect": 0, "file_ok": 0, "partial": False}
    if not image_ids:
        return counts
    # Lazy imports: the probe helper lives in the maintenance task module and
    # the hasher in the importer — importing either at module load would pull
    # celery into every service consumer.
    from ...tasks.maintenance import _verify_one
    from ..importer import _sha256_of
    started = time.monotonic()
    for image_id in image_ids:
        if time.monotonic() - started > time_budget_seconds:
            counts["partial"] = True
            break
        rec = session.get(ImageRecord, image_id)
        if rec is None:      # record deleted since the job errored
            continue
        verdict = _verify_one(Path(rec.path), rec.sha256, rec.mime, _sha256_of)
        # 'ok' means the failure was operational; anything else (corrupt /
        # failed_verification = missing/unreadable) makes the file itself the
        # problem — recovery material.
        triage = "file_ok" if verdict == "ok" else "defect"
        reason = session.execute(
            select(GpuJob.error)
            .where(GpuJob.image_record_id == image_id, GpuJob.status == "error")
            .limit(1)
        ).scalar_one_or_none()
        rec.integrity_status = verdict
        session.execute(
            update(GpuJob)
            .where(GpuJob.image_record_id == image_id, GpuJob.status == "error")
            .values(triage_status=triage, updated_at=datetime.now(UTC))
        )
        session.commit()
        counts["probed"] += 1
        counts[triage] += 1
        log.warning(
            "gpu triage: image %s (%s) job error %r -> integrity probe %r -> %s",
            image_id, rec.path, (reason or "")[:120], verdict, triage,
        )
    return counts
 def recover_defective_image(
    session: Session, image_id: int, *, images_root: Path,
 ) -> dict:
    """Delete the defective copy + record and re-poll its subscription Source.
    Mirrors the Layer-2 import refetch: with the bad file gone, the source's
    next gallery-dl run re-fetches a fresh copy, which re-imports as a new
    record and re-enters the GPU pipeline. The record delete cascades the
    error tombstones with it. 'no_source' when no enabled, real-URL Source is
    reachable via the image's provenance — manual remediation there."""
    rec = session.get(ImageRecord, image_id)
    if rec is None:
        return {"status": "not_found"}
    src_id = session.execute(
        select(Source.id)
        .join(ImageProvenance, ImageProvenance.source_id == Source.id)
        .where(
            ImageProvenance.image_record_id == image_id,
            Source.enabled.is_(True),
            ~Source.url.like("sidecar:%"),   # synthetic anchor — not pollable
        )
        .order_by(Source.id.asc())
    ).scalars().first()
    if src_id is None:
        return {"status": "no_source"}
    path = rec.path
    summary = delete_images(session, image_ids=[image_id], images_root=images_root)
    # Lazy import (services -> tasks would cycle at module load).
    from ...tasks.download import download_source
    download_source.delay(src_id)
    log.warning(
        "gpu triage recovery: deleted defective image %s (%s) and queued a "
        "re-check of source %s to re-fetch it", image_id, path, src_id,
    )
    return {"status": "refetch_queued", "source_id": src_id, **summary}
@@ -1,12 +1,13 @@
 """Production heads: train + score the per-concept classifiers (#114).
-The eval (#1130, tag_eval.py) proved the spine; this is its production form.
+The eval harness (#1130) proved the spine, then retired 2026-07-02 once the
 tagging system was accepted; this is the production form.
 - TRAIN (sync, ml worker — needs scikit-learn): for every general/character tag
  with enough labelled positives, fit a logistic-regression head on the FROZEN
  SigLIP embeddings (positives + negatives = rejections + sampled unlabeled),
  derive an honest suggest threshold + earned-auto-apply point from CROSS-
-  VALIDATED scores, and upsert a TagHead row. Reuses tag_eval's proven data
+  VALIDATED scores, and upsert a TagHead row. Uses the eval-proven data loaders
-  loaders + metric helpers so production heads match the eval's measured numbers.
+  + metric helpers (training_data.py) so heads match the measured numbers.
 - SCORE (async, API worker — numpy via pgvector, NO scikit-learn): score one
  image's embedding against all current heads → the suggestions the rail shows,
  REPLACING Camie predictions + per-tag centroids.
@@ -37,7 +38,7 @@ from ...models import (
    TagSuggestionRejection,
 )
 from ...models.tag import image_tag
-from .tag_eval import (
+from .training_data import (
    _auto_apply_point,
    _ids_with_tag,
    _l2norm,
@@ -1,430 +0,0 @@
 """Head-vs-centroid tagging eval (#1130, milestone #114 slice 1).
 Proves the "frozen embedding + small trained head (with negatives)" spine on the
 operator's OWN data, reusing the SigLIP embeddings already stored on
 image_record. For each concept tag it compares:
  - CENTROID baseline (the old approach): cosine to the mean of positive vectors.
  - HEAD (the new approach): logistic regression trained on positives + negatives.
 and reports cross-validated precision/recall/AP for both, a LEARNING CURVE
 (accuracy as the number of tagged positives grows), and example image ids to
 eyeball.
 numpy + scikit-learn are imported LAZILY inside run_eval so the API worker (base
 image, no ML stack) can still import start_tag_eval_run to enqueue the ml-queue
 task — the heavy compute only runs on the ml worker.
 """
 from __future__ import annotations
 import logging
 from datetime import UTC, datetime
 from typing import Any
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 from ...models import (
    ImageRecord,
    Tag,
    TagEvalRun,
    TagKind,
    TagPositiveConfirmation,
    TagSuggestionRejection,
 )
 from ...models.tag import image_tag
 log = logging.getLogger(__name__)
 # The operator's real concept list (mix of whole-ish + small/local cues). The
 # admin trigger can override; this is the default eval set.
 DEFAULT_CONCEPTS = [
    "glasses", "cat", "dog", "horse", "goblin",
    "cum", "lactation", "fellatio", "xray", "stomach bulge",
 ]
 DEFAULT_CURVE_POINTS = [10, 30, 100, 300]
 DEFAULT_NEG_RATIO = 3       # negatives per positive (rejections + sampled unlabeled)
 DEFAULT_CV_FOLDS = 5
 MIN_POSITIVES = 8           # below this, a concept can't be evaluated meaningfully
 _UNLABELED_POOL = 4000      # cap on sampled unlabeled rows pulled per concept
 _EXAMPLES_K = 12
 def start_tag_eval_run(session: Session, params: dict[str, Any]) -> int:
    """Create a TagEvalRun (status='running') and dispatch the ml-queue task.
    Returns the new run id. Light guard: one running eval at a time."""
    existing = session.execute(
        select(TagEvalRun.id).where(TagEvalRun.status == "running")
    ).scalar_one_or_none()
    if existing is not None:
        raise EvalAlreadyRunning(existing)
    norm = _normalize_params(params)
    run = TagEvalRun(params=norm, status="running", last_progress_at=datetime.now(UTC))
    session.add(run)
    session.flush()
    run_id = run.id
    # Same enqueue-by-import pattern api/suggestions.py uses for ml tasks; the
    # commit happens in the API handler so row + dispatch are visible together.
    from ...tasks.ml import tag_eval_run as _task
    _task.delay(run_id)
    return run_id
 class EvalAlreadyRunning(Exception):
    """Raised by start_tag_eval_run when an eval is already in flight."""
 def _normalize_params(params: dict[str, Any] | None) -> dict[str, Any]:
    params = params or {}
    concepts = [str(c).strip() for c in (params.get("concepts") or []) if str(c).strip()]
    try:
        neg_ratio = max(1, int(params.get("neg_ratio", DEFAULT_NEG_RATIO)))
    except (TypeError, ValueError):
        neg_ratio = DEFAULT_NEG_RATIO
    try:
        cv_folds = max(2, int(params.get("cv_folds", DEFAULT_CV_FOLDS)))
    except (TypeError, ValueError):
        cv_folds = DEFAULT_CV_FOLDS
    try:
        auto_top_n = min(max(int(params.get("auto_top_n", 0) or 0), 0), 200)
    except (TypeError, ValueError):
        auto_top_n = 0
    try:
        precision_target = min(max(float(params.get("precision_target", 0.97)), 0.5), 0.999)
    except (TypeError, ValueError):
        precision_target = 0.97
    # No explicit concepts and auto-discovery off → fall back to the hand list.
    if not concepts and not auto_top_n:
        concepts = list(DEFAULT_CONCEPTS)
    curve = params.get("curve_points") or DEFAULT_CURVE_POINTS
    curve = sorted({int(n) for n in curve if int(n) > 0})
    return {
        "concepts": concepts,
        "neg_ratio": neg_ratio,
        "cv_folds": cv_folds,
        "auto_top_n": auto_top_n,
        "precision_target": round(precision_target, 4),
        "curve_points": curve,
    }
 def _top_general_concepts(session: Session, n: int, min_count: int) -> list[str]:
    """The n most-tagged general (concept) tags with >= min_count images — a fast
    server-side way to broaden the eval beyond the hand-picked list (counts all
    sources; source-aware filtering is a separate concern)."""
    rows = session.execute(
        select(Tag.name)
        .join(image_tag, image_tag.c.tag_id == Tag.id)
        .where(Tag.kind == TagKind.general)
        .group_by(Tag.id)
        .having(func.count(image_tag.c.image_record_id) >= min_count)
        .order_by(func.count(image_tag.c.image_record_id).desc())
        .limit(n)
    ).all()
    return [r[0] for r in rows]
 def _resolve_tag_id(session: Session, name: str) -> int | None:
    """Case-insensitive tag-name match; if several share a name, take the one
    applied to the most images (the one the operator actually uses)."""
    rows = session.execute(
        select(Tag.id, func.count(image_tag.c.image_record_id))
        .outerjoin(image_tag, image_tag.c.tag_id == Tag.id)
        .where(func.lower(Tag.name) == name.lower())
        .group_by(Tag.id)
        .order_by(func.count(image_tag.c.image_record_id).desc())
    ).all()
    return rows[0][0] if rows else None
 def _ids_with_tag(session: Session, tag_id: int) -> list[int]:
    return [
        r[0] for r in session.execute(
            select(image_tag.c.image_record_id).where(image_tag.c.tag_id == tag_id)
        ).all()
    ]
 def _rejected_ids(session: Session, tag_id: int) -> list[int]:
    return [
        r[0] for r in session.execute(
            select(TagSuggestionRejection.image_record_id)
            .where(TagSuggestionRejection.tag_id == tag_id)
        ).all()
    ]
 def _confirmed_ids(session: Session, tag_id: int) -> set[int]:
    """Positives the operator explicitly affirmed ('keep') — excluded from the
    doubts list so confirmed-correct images don't resurface every run."""
    return {
        r[0] for r in session.execute(
            select(TagPositiveConfirmation.image_record_id)
            .where(TagPositiveConfirmation.tag_id == tag_id)
        ).all()
    }
 def _sample_unlabeled(session: Session, exclude: set[int], limit: int) -> list[int]:
    """Random image ids (with an embedding) NOT carrying the tag. Concepts are
    sparse, so an untagged image is almost always a true negative."""
    stmt = (
        select(ImageRecord.id)
        .where(ImageRecord.siglip_embedding.is_not(None))
        .order_by(func.random())
        .limit(limit)
    )
    if exclude:
        stmt = stmt.where(ImageRecord.id.not_in(exclude))
    return [r[0] for r in session.execute(stmt).all()]
 def _load_embeddings(session: Session, ids: list[int]) -> dict[int, Any]:
    import numpy as np
    out: dict[int, Any] = {}
    if not ids:
        return out
    # Chunk the IN list to stay well under psycopg's parameter ceiling.
    for i in range(0, len(ids), 2000):
        chunk = ids[i:i + 2000]
        for rid, emb in session.execute(
            select(ImageRecord.id, ImageRecord.siglip_embedding)
            .where(ImageRecord.id.in_(chunk))
            .where(ImageRecord.siglip_embedding.is_not(None))
        ).all():
            out[rid] = np.asarray(emb, dtype=np.float32)
    return out
 def run_eval(session: Session, params: dict[str, Any]) -> dict[str, Any]:
    """Compute the full report. Per-concept failures are captured, not fatal."""
    import numpy as np
    cfg = _normalize_params(params)
    # Auto-discovery: union the explicit concepts with the top-N most-tagged
    # general tags (server-side, fast) so the eval can broaden itself.
    concepts = list(cfg["concepts"])
    if cfg["auto_top_n"]:
        seen = {c.lower() for c in concepts}
        for name in _top_general_concepts(session, cfg["auto_top_n"], MIN_POSITIVES):
            if name.lower() not in seen:
                concepts.append(name)
                seen.add(name.lower())
    cfg["concepts"] = concepts
    concepts_out = []
    for name in cfg["concepts"]:
        try:
            concepts_out.append(_eval_concept(session, name, cfg, np))
        except Exception as exc:  # one bad concept shouldn't kill the run
            log.exception("tag-eval concept %r failed", name)
            concepts_out.append({"name": name, "skipped": f"error: {exc}"})
    return {
        "generated_at": datetime.now(UTC).isoformat(),
        "params": cfg,
        "concepts": concepts_out,
    }
 def _eval_concept(session: Session, name: str, cfg: dict, np) -> dict[str, Any]:
    tag_id = _resolve_tag_id(session, name)
    if tag_id is None:
        return {"name": name, "skipped": "no such tag"}
    pos_ids = _ids_with_tag(session, tag_id)
    if len(pos_ids) < MIN_POSITIVES:
        return {"name": name, "tag_id": tag_id, "n_pos": len(pos_ids),
                "skipped": f"too few positives (<{MIN_POSITIVES})"}
    neg_ratio = cfg["neg_ratio"]
    pos_set = set(pos_ids)
    rejected = [i for i in _rejected_ids(session, tag_id) if i not in pos_set]
    want_neg = max(len(pos_ids) * neg_ratio, _EXAMPLES_K * 4)
    sampled = _sample_unlabeled(session, pos_set | set(rejected),
                                min(_UNLABELED_POOL, want_neg))
    neg_ids = rejected + [i for i in sampled if i not in pos_set]
    emb = _load_embeddings(session, pos_ids + neg_ids)
    pos = [(i, emb[i]) for i in pos_ids if i in emb]
    neg = [(i, emb[i]) for i in neg_ids if i in emb]
    if len(pos) < MIN_POSITIVES or len(neg) < MIN_POSITIVES:
        return {"name": name, "tag_id": tag_id, "n_pos": len(pos),
                "n_neg": len(neg), "skipped": "too few embedded examples"}
    ids = np.array([i for i, _ in pos] + [i for i, _ in neg])
    X = np.vstack([v for _, v in pos] + [v for _, v in neg]).astype(np.float32)
    y = np.array([1] * len(pos) + [0] * len(neg))
    Xn = _l2norm(X, np)
    head = _eval_head(Xn, y, cfg["cv_folds"], cfg["precision_target"], np)
    centroid = _eval_centroid(Xn, y, cfg["cv_folds"], np)
    curve = _learning_curve(Xn, y, cfg["curve_points"], neg_ratio, np)
    confirmed = _confirmed_ids(session, tag_id)
    examples = _examples(session, Xn, y, ids, np, set(rejected), confirmed)
    return {
        "name": name, "tag_id": tag_id,
        "n_pos": len(pos), "n_neg": len(neg),
        "n_rejected": len(rejected),
        "head": head, "centroid": centroid,
        "curve": curve, "examples": examples,
    }
 def _l2norm(X, np):
    n = np.linalg.norm(X, axis=1, keepdims=True)
    n[n == 0] = 1.0
    return X / n
 def _metrics_from_scores(y, scores, np) -> dict[str, float]:
    from sklearn.metrics import average_precision_score, precision_recall_curve
    ap = float(average_precision_score(y, scores))
    prec, rec, thr = precision_recall_curve(y, scores)
    f1 = (2 * prec * rec) / np.clip(prec + rec, 1e-9, None)
    best = int(np.argmax(f1))
    # thr has len = len(prec)-1; map best index safely.
    t = float(thr[min(best, len(thr) - 1)]) if len(thr) else 0.5
    return {
        "ap": round(ap, 4),
        "precision": round(float(prec[best]), 4),
        "recall": round(float(rec[best]), 4),
        "f1": round(float(f1[best]), 4),
        "threshold": round(t, 4),
    }
 def _safe_folds(y, folds, np) -> int:
    minority = int(min(np.bincount(y)))
    return max(2, min(folds, minority))
 def _eval_head(Xn, y, folds, target, np) -> dict[str, float]:
    from sklearn.linear_model import LogisticRegression
    from sklearn.model_selection import StratifiedKFold, cross_val_predict
    clf = LogisticRegression(max_iter=1000, class_weight="balanced")
    cv = StratifiedKFold(n_splits=_safe_folds(y, folds, np), shuffle=True,
                         random_state=0)
    probs = cross_val_predict(clf, Xn, y, cv=cv, method="predict_proba")[:, 1]
    m = _metrics_from_scores(y, probs, np)
    m["auto_apply"] = _auto_apply_point(y, probs, target, np)
    return m
 def _auto_apply_point(y, scores, target, np) -> dict | None:
    """The auto-apply operating point: the threshold that yields the MOST recall
    while holding precision >= target. This answers 'could this concept fire
    without a human, and how much would it catch?' Returns None if no threshold
    reaches the precision target (concept not auto-apply-ready)."""
    from sklearn.metrics import precision_recall_curve
    prec, rec, thr = precision_recall_curve(y, scores)
    best = None  # (threshold, precision, recall) maximizing recall s.t. prec>=target
    for i in range(len(thr)):  # thr[i] corresponds to prec[i], rec[i]
        if prec[i] >= target and (best is None or rec[i] > best[2]):
            best = (float(thr[i]), float(prec[i]), float(rec[i]))
    if best is None:
        return None
    return {
        "target": round(float(target), 4),
        "threshold": round(best[0], 4),
        "precision": round(best[1], 4),
        "recall": round(best[2], 4),
    }
 def _eval_centroid(Xn, y, folds, np) -> dict[str, float]:
    """Cross-validated cosine-to-positive-mean — the OLD method's quality."""
    from sklearn.model_selection import StratifiedKFold
    cv = StratifiedKFold(n_splits=_safe_folds(y, folds, np), shuffle=True,
                         random_state=0)
    scores = np.zeros(len(y), dtype=np.float32)
    for train, test in cv.split(Xn, y):
        c = Xn[train][y[train] == 1].mean(axis=0)
        cn = c / (np.linalg.norm(c) or 1.0)
        scores[test] = Xn[test] @ cn
    return _metrics_from_scores(y, scores, np)
 def _learning_curve(Xn, y, points, neg_ratio, np) -> list[dict[str, float]]:
    """Hold out a fixed test split; train the head on a growing number of
    positives and watch AP/F1 climb — answers 'does tagging more sharpen it?'"""
    from sklearn.linear_model import LogisticRegression
    from sklearn.model_selection import train_test_split
    rng = np.random.default_rng(0)
    idx = np.arange(len(y))
    try:
        tr, te = train_test_split(idx, test_size=0.3, stratify=y, random_state=0)
    except ValueError:
        return []
    tr_pos = tr[y[tr] == 1]
    tr_neg = tr[y[tr] == 0]
    out = []
    for n in points:
        if n > len(tr_pos):
            break
        sp = rng.choice(tr_pos, size=n, replace=False)
        nn = min(len(tr_neg), n * neg_ratio)
        sn = rng.choice(tr_neg, size=nn, replace=False)
        sub = np.concatenate([sp, sn])
        clf = LogisticRegression(max_iter=1000, class_weight="balanced")
        clf.fit(Xn[sub], y[sub])
        prob = clf.predict_proba(Xn[te])[:, 1]
        m = _metrics_from_scores(y[te], prob, np)
        out.append({"n_pos": int(n), "ap": m["ap"], "f1": m["f1"]})
    return out
 def _examples(session, Xn, y, ids, np, rejected_set, confirmed_set) -> dict[str, list[dict]]:
    """Train on all data, then surface: top-scoring negatives the operator has
    NOT already rejected (= fresh suggestions) and lowest-scoring POSITIVES the
    operator has NOT already confirmed (= unreviewed doubts). Excluding rejected
    ids stops an adjudicated near-miss from resurfacing in 'would suggest';
    excluding confirmed ids stops a 'kept' correct positive from resurfacing in
    'head doubts' every run. Resolves thumbnail urls for a self-contained report."""
    from sklearn.linear_model import LogisticRegression
    clf = LogisticRegression(max_iter=1000, class_weight="balanced")
    clf.fit(Xn, y)
    s = clf.predict_proba(Xn)[:, 1]
    neg_idx = np.where(y == 0)[0]
    pos_idx = np.where(y == 1)[0]
    top_neg = []
    for i in neg_idx[np.argsort(s[neg_idx])[::-1]]:  # high score → low
        rid = int(ids[i])
        if rid in rejected_set:
            continue  # already told the head 'no' — don't re-suggest it
        top_neg.append(rid)
        if len(top_neg) >= _EXAMPLES_K:
            break
    low_pos = []
    for i in pos_idx[np.argsort(s[pos_idx])]:  # low score → high
        rid = int(ids[i])
        if rid in confirmed_set:
            continue  # already kept/confirmed — don't re-doubt it
        low_pos.append(rid)
        if len(low_pos) >= _EXAMPLES_K:
            break
    thumbs = _resolve_thumbs(session, top_neg + low_pos)
    return {
        "head_would_suggest": [thumbs[i] for i in top_neg if i in thumbs],
        "head_doubts_positive": [thumbs[i] for i in low_pos if i in thumbs],
    }
 def _resolve_thumbs(session, ids: list[int]) -> dict[int, dict]:
    from ..gallery_service import thumbnail_url
    out: dict[int, dict] = {}
    if not ids:
        return out
    for rid, tp, sha, mime in session.execute(
        select(
            ImageRecord.id, ImageRecord.thumbnail_path,
            ImageRecord.sha256, ImageRecord.mime,
        ).where(ImageRecord.id.in_(ids))
    ).all():
        out[rid] = {"id": rid, "thumbnail_url": thumbnail_url(tp, sha, mime)}
    return out
@@ -0,0 +1,121 @@
 """Shared data-selection + validated-metric helpers for the heads trainer.
 Born in the head-vs-centroid eval harness (#1130, tag_eval.py) that proved the
 "frozen embedding + small trained head (with negatives)" spine; the harness was
 retired 2026-07-02 (operator: the tagging system is proven, the eval isn't
 needed) and these survivors moved here — they ARE the heads' production data
 pipeline (heads.py trains and scores with them nightly).
 numpy/scikit-learn are imported lazily inside the functions that need them so
 the API worker (base image, no ML stack) can import this module.
 """
 from __future__ import annotations
 from typing import Any
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 from ...models import ImageRecord, TagSuggestionRejection
 from ...models.tag import image_tag
 def _ids_with_tag(session: Session, tag_id: int) -> list[int]:
    return [
        r[0] for r in session.execute(
            select(image_tag.c.image_record_id).where(image_tag.c.tag_id == tag_id)
        ).all()
    ]
 def _rejected_ids(session: Session, tag_id: int) -> list[int]:
    return [
        r[0] for r in session.execute(
            select(TagSuggestionRejection.image_record_id)
            .where(TagSuggestionRejection.tag_id == tag_id)
        ).all()
    ]
 def _sample_unlabeled(session: Session, exclude: set[int], limit: int) -> list[int]:
    """Random image ids (with an embedding) NOT carrying the tag. Concepts are
    sparse, so an untagged image is almost always a true negative."""
    stmt = (
        select(ImageRecord.id)
        .where(ImageRecord.siglip_embedding.is_not(None))
        .order_by(func.random())
        .limit(limit)
    )
    if exclude:
        stmt = stmt.where(ImageRecord.id.not_in(exclude))
    return [r[0] for r in session.execute(stmt).all()]
 def _load_embeddings(session: Session, ids: list[int]) -> dict[int, Any]:
    import numpy as np
    out: dict[int, Any] = {}
    if not ids:
        return out
    # Chunk the IN list to stay well under psycopg's parameter ceiling.
    for i in range(0, len(ids), 2000):
        chunk = ids[i:i + 2000]
        for rid, emb in session.execute(
            select(ImageRecord.id, ImageRecord.siglip_embedding)
            .where(ImageRecord.id.in_(chunk))
            .where(ImageRecord.siglip_embedding.is_not(None))
        ).all():
            out[rid] = np.asarray(emb, dtype=np.float32)
    return out
 def _l2norm(X, np):
    n = np.linalg.norm(X, axis=1, keepdims=True)
    n[n == 0] = 1.0
    return X / n
 def _metrics_from_scores(y, scores, np) -> dict[str, float]:
    from sklearn.metrics import average_precision_score, precision_recall_curve
    ap = float(average_precision_score(y, scores))
    prec, rec, thr = precision_recall_curve(y, scores)
    f1 = (2 * prec * rec) / np.clip(prec + rec, 1e-9, None)
    best = int(np.argmax(f1))
    # thr has len = len(prec)-1; map best index safely.
    t = float(thr[min(best, len(thr) - 1)]) if len(thr) else 0.5
    return {
        "ap": round(ap, 4),
        "precision": round(float(prec[best]), 4),
        "recall": round(float(rec[best]), 4),
        "f1": round(float(f1[best]), 4),
        "threshold": round(t, 4),
    }
 def _safe_folds(y, folds, np) -> int:
    minority = int(min(np.bincount(y)))
    return max(2, min(folds, minority))
 def _auto_apply_point(y, scores, target, np) -> dict | None:
    """The auto-apply operating point: the threshold that yields the MOST recall
    while holding precision >= target. This answers 'could this concept fire
    without a human, and how much would it catch?' Returns None if no threshold
    reaches the precision target (concept not auto-apply-ready)."""
    from sklearn.metrics import precision_recall_curve
    prec, rec, thr = precision_recall_curve(y, scores)
    best = None  # (threshold, precision, recall) maximizing recall s.t. prec>=target
    for i in range(len(thr)):  # thr[i] corresponds to prec[i], rec[i]
        if prec[i] >= target and (best is None or rec[i] > best[2]):
            best = (float(thr[i]), float(prec[i]), float(rec[i]))
    if best is None:
        return None
    return {
        "target": round(float(target), 4),
        "threshold": round(best[0], 4),
        "precision": round(best[1], 4),
        "recall": round(best[2], 4),
    }
@@ -21,7 +21,6 @@ from ..models import (
    ImportTask,
    LibraryAuditRun,
    Source,
    TagEvalRun,
    TaskRun,
 )
 from ..utils.phash import compute_phash
@@ -96,9 +95,6 @@ BACKUP_DB_STALL_THRESHOLD_MINUTES = 40
 # Library audit: scan_library_for_rule has time_limit=7500s (2h5m).
 # 2h15m gives a 10-min buffer.
 LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES = 135
 # tag-eval (#1130) has a 30-min soft limit; flag a run with no progress past 40.
 TAG_EVAL_STALL_THRESHOLD_MINUTES = 40
 TAG_EVAL_KEEP_RUNS = 20
 # head training (#114) has a 60-min soft limit; flag no-progress past 75.
 HEAD_TRAINING_STALL_THRESHOLD_MINUTES = 75
 HEAD_TRAINING_KEEP_RUNS = 20
@@ -582,6 +578,28 @@ def verify_integrity() -> int:
    return total
@celery.task(
    name="backend.app.tasks.maintenance.triage_gpu_errors",
    # Bounded small-set probe (only errored images, once each), but a single
    # large original's sha256 over NFS can run tens of seconds — same quick-lane
    # tolerance rationale as verify_integrity above.
    soft_time_limit=600, time_limit=900,
 )
 def triage_gpu_errors() -> dict:
    """Failure triage (#125): probe each errored GPU job's file once and write
    the verdicts (ImageRecord.integrity_status + GpuJob.triage_status) — see
    services/ml/gpu_triage.py. Time-boxed + resumable; no-op when every errored
    job is already triaged."""
    from ..services.ml.gpu_triage import triage_errored_jobs
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        summary = triage_errored_jobs(session, time_budget_seconds=300.0)
    if summary["probed"]:
        log.info("triage_gpu_errors: %s", summary)
    return summary
@celery.task(name="backend.app.tasks.maintenance.recover_stalled_download_events")
 def recover_stalled_download_events() -> int:
    """Recover DownloadEvent rows stuck pending/running past the worker hard kill.
@@ -721,46 +739,6 @@ def recover_stalled_library_audit_runs() -> int:
        return recovered
@celery.task(name="backend.app.tasks.maintenance.recover_stalled_tag_eval_runs")
 def recover_stalled_tag_eval_runs() -> int:
    """Flip TagEvalRun rows stuck in 'running' past the stall threshold to
    'error', and prune old runs to the last TAG_EVAL_KEEP_RUNS (retention,
    rule 89). Runs every 5 min on the maintenance lane; no-op when idle."""
    SessionLocal = _sync_session_factory()
    now = datetime.now(UTC)
    cutoff = now - timedelta(minutes=TAG_EVAL_STALL_THRESHOLD_MINUTES)
    with SessionLocal() as session:
        result = session.execute(
            update(TagEvalRun)
            .where(TagEvalRun.status == "running")
            .where(
                func.coalesce(TagEvalRun.last_progress_at, TagEvalRun.started_at)
                < cutoff
            )
            .values(
                status="error", finished_at=now,
                error=(
                    f"stranded by recovery sweep (no progress for "
                    f"{TAG_EVAL_STALL_THRESHOLD_MINUTES} min)"
                ),
            )
        )
        # Retention: keep only the most recent N runs.
        keep = session.execute(
            select(TagEvalRun.id).order_by(TagEvalRun.id.desc())
            .limit(TAG_EVAL_KEEP_RUNS)
        ).scalars().all()
        if keep:
            session.execute(
                delete(TagEvalRun).where(TagEvalRun.id.not_in(keep))
            )
        session.commit()
        recovered = result.rowcount or 0
        if recovered:
            log.info("recover_stalled_tag_eval_runs: recovered %d rows", recovered)
        return recovered
@celery.task(name="backend.app.tasks.maintenance.recover_stalled_head_training_runs")
 def recover_stalled_head_training_runs() -> int:
    """Flip HeadTrainingRun rows stuck in 'running' past the stall threshold to
@@ -250,51 +250,6 @@ def backfill(self) -> int:
    return enqueued
@celery.task(
    name="backend.app.tasks.ml.tag_eval_run",
    bind=True,
    # The head-vs-centroid eval (#1130) loads embeddings + fits sklearn heads
    # for several concepts — minutes, not seconds. Runs on the ml queue because
    # only that worker has numpy/scikit-learn.
    soft_time_limit=1800, time_limit=2100,
 )
 def tag_eval_run(self, run_id: int) -> str:
    """Compute the eval report into the persisted TagEvalRun row so it survives
    navigation (the admin card rehydrates from the row, not transient state)."""
    from datetime import UTC, datetime
    from ..models import TagEvalRun
    from ..services.ml.tag_eval import run_eval
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        run = session.get(TagEvalRun, run_id)
        if run is None:
            return "missing"
        run.last_progress_at = datetime.now(UTC)
        session.commit()
        try:
            report = run_eval(session, run.params)
        except SoftTimeLimitExceeded:
            run.status = "error"
            run.error = "timed out"
            run.finished_at = datetime.now(UTC)
            session.commit()
            raise
        except Exception as exc:
            log.exception("tag_eval_run %d failed", run_id)
            run.status = "error"
            run.error = str(exc)
            run.finished_at = datetime.now(UTC)
            session.commit()
            return "error"
        run.report = report
        run.status = "ready"
        run.finished_at = datetime.now(UTC)
        session.commit()
        return "ready"
@celery.task(
    name="backend.app.tasks.ml.train_heads",
    bind=True,
@@ -458,15 +413,30 @@ def enqueue_gpu_backfill(task_name: str) -> int:
    'siglip' gates on the RESULT (no concept region yet) rather than on a prior
    job, so it picks up the back-catalogue of images that were CCIP-embedded
-    before concept crops existed, and retries images whose concept embed failed —
+    before concept crops existed — without re-touching their figure/CCIP regions.
-    without re-touching their figure/CCIP regions."""
+
    An ERRORED job is a tombstone for its (image, task): no variant re-enqueues
    it. Retry is deliberate-only (/retry_errors), which also means an errored
    back-catalogue needs one "Retry errored jobs" press after a model swap.
    Before the tombstone rule, this loop re-minted a fresh doomed job for every
    permanently-bad file each run — ~24 duplicate error rows/day per file (the
    2026-07-02 "unprocessable" flood)."""
    from sqlalchemy import exists, insert, literal, or_
    from sqlalchemy import select as sa_select
    from ..models import GpuJob, ImageRecord, ImageRegion, MLSettings
    from ..services.ml.gpu_jobs import error_dedupe_statements
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        # Prune stale tombstones first (loop-era duplicates + rows made moot by
        # a later success), so 'error' reads as one row per distinct failing
        # file and the skip-guards below see a clean picture.
        pruned = sum(
            session.execute(s).rowcount or 0 for s in error_dedupe_statements()
        )
        if pruned:
            log.info("gpu backfill: pruned %d stale/duplicate error rows", pruned)
        cur_version = session.execute(
            select(MLSettings.embedder_model_version).where(MLSettings.id == 1)
        ).scalar_one()
@@ -478,14 +448,15 @@ def enqueue_gpu_backfill(task_name: str) -> int:
                ImageRecord.siglip_model_version.is_(None),
                ImageRecord.siglip_model_version != cur_version,
            )
-            queued = exists().where(
+            # 'error' blocks too — tombstone rule, see docstring.
            blocked = exists().where(
                GpuJob.image_record_id == ImageRecord.id,
                GpuJob.task == "embed",
-                GpuJob.status.in_(["pending", "leased"]),
+                GpuJob.status.in_(["pending", "leased", "error"]),
            )
            sel = sa_select(
                ImageRecord.id, literal("embed"), literal("pending")
-            ).where(stale).where(~queued)
+            ).where(stale).where(~blocked)
        elif task_name == "siglip":
            # Concept-crop re-embed: enqueue when there's no concept region AT THE
            # CURRENT model version — so a model swap re-triggers crops too, not
@@ -495,19 +466,22 @@ def enqueue_gpu_backfill(task_name: str) -> int:
                ImageRegion.kind == "concept",
                ImageRegion.embedding_version == cur_version,
            )
-            queued = exists().where(
+            # 'error' blocks too — tombstone rule, see docstring.
            blocked = exists().where(
                GpuJob.image_record_id == ImageRecord.id,
                GpuJob.task == "siglip",
-                GpuJob.status.in_(["pending", "leased"]),
+                GpuJob.status.in_(["pending", "leased", "error"]),
            )
            sel = sa_select(
                ImageRecord.id, literal("siglip"), literal("pending")
-            ).where(~has_current_concept).where(~queued)
+            ).where(~has_current_concept).where(~blocked)
        else:
            # ANY prior row blocks — including 'error' (tombstone rule, see
            # docstring): pre-fix this branch ran HOURLY and was the loop.
            already = exists().where(
                GpuJob.image_record_id == ImageRecord.id,
                GpuJob.task == task_name,
-                GpuJob.status.in_(["pending", "leased", "done"]),
+                GpuJob.status.in_(["pending", "leased", "done", "error"]),
            )
            sel = sa_select(
                ImageRecord.id, literal(task_name), literal("pending")
@@ -525,28 +499,29 @@ def enqueue_gpu_backfill(task_name: str) -> int:
@celery.task(name="backend.app.tasks.ml.recover_orphaned_gpu_jobs")
 def recover_orphaned_gpu_jobs() -> int:
    """Reset expired GPU-job leases back to pending — recovers work orphaned by an
-    agent that died mid-job (no graceful release). Short beat cadence so orphans
+    agent that died mid-job (no graceful release) — and convert poison-loopers
-    get picked back up quickly + the queue counts read honestly. Returns the
+    (release/expiry cycles that never reach fail()'s attempt cap) to 'error'.
-    number recovered."""
+    Statements are shared with GpuJobService.recover_orphaned so the sweep and
    the service can't drift. Short beat cadence so orphans get picked back up
    quickly + the queue counts read honestly. Returns the number recovered."""
    from datetime import UTC, datetime
-    from sqlalchemy import update
+    from ..services.ml.gpu_jobs import recover_statements
    from ..models import GpuJob
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
-        now = datetime.now(UTC)
+        counts = {
-        res = session.execute(
+            name: session.execute(stmt).rowcount or 0
-            update(GpuJob)
+            for name, stmt in recover_statements(datetime.now(UTC)).items()
-            .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,
            )
        )
        session.commit()
-        return res.rowcount or 0
+        if counts["poison_expired"] or counts["poison_pending"]:
            log.warning(
                "gpu jobs poisoned -> error: %d crash-loop (expired lease), "
                "%d never-complete (pending)",
                counts["poison_expired"], counts["poison_pending"],
            )
        return counts["recovered"]
@celery.task(name="backend.app.tasks.ml.reprocess_gpu_jobs")
@@ -7,8 +7,10 @@
  Usage: wrap a card's action body in the default slot; pass icon/title/blurb.
  `destructive` tints the icon error-red for delete actions. `open` can be forced
-  (e.g. keep a running task's tile expanded). Keyboard accessible: the header is a
+  (e.g. keep a running task's tile expanded). Manual expand/collapse persists per
-  real <button> with aria-expanded + focus ring.
+  tile in localStorage, so the page reloads the way the operator left it.
  Keyboard accessible: the header is a real <button> with aria-expanded + focus
  ring.
 -->
 <template>
  <v-card class="fc-tile" :class="{ 'fc-tile--open': isOpen }">
@@ -53,12 +55,21 @@ const props = defineProps({
  open: { type: Boolean, default: false },
 })
-const local = ref(props.open)
+// Only MANUAL toggles are saved (keyed by tile title): a forced `open` while a
-watch(() => props.open, (v) => { local.value = v })
+// task is mid-run is transient state, not a preference — persisting it would
 // resurrect the "several tiles open by default" bug this replaces. When the
 // force clears, the tile falls back to the operator's saved choice.
 const storeKey = `fc.tile.${props.title}`
 function savedOpen() {
  try { return localStorage.getItem(storeKey) === '1' } catch { return false }
 }
 const local = ref(props.open || savedOpen())
 watch(() => props.open, (v) => { local.value = v || savedOpen() })
 const isOpen = computed(() => local.value)
 function toggle() {
  local.value = !local.value
  try { localStorage.setItem(storeKey, local.value ? '1' : '0') } catch { /* non-fatal */ }
 }
 </script>
@@ -132,8 +132,8 @@ const hasMenu = computed(() =>
  color: rgb(var(--v-theme-on-surface-variant, var(--v-theme-on-surface)));
  font-family: 'JetBrains Mono', monospace;
 }
-/* Green ✓ / red ✗ verdict pair — same circular language as the eval card
+/* Green ✓ / red ✗ verdict pair — circular buttons so accept/reject read
-   (TagEvalCard .fc-act) so accept/reject read identically across surfaces. */
+   identically across surfaces. */
 .fc-suggestion__acts {
  flex: 0 0 auto; display: flex; gap: 4px;
 }
@@ -3,7 +3,6 @@
    icon="mdi-expansion-card"
    title="GPU agent (CCIP + crops)"
    blurb="Connect a desktop-GPU agent to embed characters (CCIP) and crops. It pulls work over HTTP — your database and Redis stay private."
    :open="true"
  >
    <p class="fc-muted text-body-2 mb-3">
      The agent is a container you run on the machine with the GPU. It
@@ -338,8 +337,12 @@ async function onBackfillSiglip() {
 async function onRetryErrors() {
  retrying.value = true
  try {
-    const { requeued } = await store.retryErrors()
+    const { requeued, pruned, defects_kept: kept } = await store.retryErrors()
-    toast({ text: `Requeued ${requeued} errored job${requeued === 1 ? '' : 's'} — run the agent to process them`, type: 'success' })
+    const extras = []
    if (pruned) extras.push(`${pruned} stale duplicate${pruned === 1 ? '' : 's'} pruned`)
    if (kept) extras.push(`${kept} known-bad file${kept === 1 ? '' : 's'} kept for recovery`)
    const extra = extras.length ? ` (${extras.join(', ')})` : ''
    toast({ text: `Requeued ${requeued} errored job${requeued === 1 ? '' : 's'}${extra} — run the agent to process them`, type: 'success' })
    await refreshQueue()
  } catch (e) {
    toast({ text: `Could not retry errored jobs: ${e.message}`, type: 'error' })
@@ -0,0 +1,189 @@
 <template>
  <MaintenanceTile
    icon="mdi-file-alert"
    title="Failed processing"
    blurb="Triage originals that failed GPU processing — probe the files, flag defects, recover them."
  >
    <p class="fc-muted text-body-2 mb-3">
      A job that keeps failing parks as an error with its reason. A background
      probe then checks the FILE itself (checksum + decode) and splits the
      errors: <b>defective files</b> (truncated/corrupt originals — listed below
      for recovery) vs <b>file&nbsp;OK</b> (the failure was operational; requeue
      those with <i>Retry errored jobs</i> on the GPU agent card).
    </p>
    <div v-if="loading" class="fc-muted text-body-2">Loading…</div>
    <template v-else-if="overview">
      <div class="fc-queue mb-3">
        <div class="fc-q"><div class="fc-q__n">{{ overview.total }}</div><div class="fc-q__l">errored</div></div>
        <div class="fc-q"><div class="fc-q__n" :class="overview.triage.defect ? 'fc-weak' : ''">{{ overview.triage.defect }}</div><div class="fc-q__l">defective</div></div>
        <div class="fc-q"><div class="fc-q__n fc-good">{{ overview.triage.file_ok }}</div><div class="fc-q__l">file ok</div></div>
        <div class="fc-q"><div class="fc-q__n">{{ overview.triage.unclassified }}</div><div class="fc-q__l">unprobed</div></div>
      </div>
      <p v-if="classSummary" class="fc-muted text-caption mb-3">
        Reasons: {{ classSummary }}
      </p>
      <div class="d-flex mb-4" style="gap:8px">
        <v-btn
          size="small" color="accent" variant="tonal" rounded="pill"
          prepend-icon="mdi-magnify-scan" :loading="probing"
          :disabled="!overview.triage.unclassified" @click="onProbe"
        >Probe unclassified now</v-btn>
        <v-btn
          size="small" variant="text" rounded="pill"
          prepend-icon="mdi-refresh" @click="refresh"
        >Refresh</v-btn>
      </div>
      <template v-if="defects.length">
        <div class="fc-section-h mb-2">Defective originals</div>
        <div v-for="it in defects" :key="it.job_id" class="fc-defect mb-2">
          <a :href="it.image_url" target="_blank" rel="noopener" class="fc-defect__thumb">
            <img v-if="it.thumbnail_url" :src="it.thumbnail_url" alt="">
            <v-icon v-else icon="mdi-file-question" size="28" />
          </a>
          <div class="fc-defect__meta">
            <div class="text-body-2">
              image <b>{{ it.image_id }}</b> · {{ it.task }} ·
              <span class="fc-weak">{{ it.integrity_status }}</span>
            </div>
            <div class="fc-muted text-caption fc-defect__err" :title="it.error || ''">
              {{ it.error || 'no stored reason' }}
            </div>
          </div>
          <v-btn
            v-if="recovered[it.image_id] !== 'no_source'"
            size="small" color="accent" variant="tonal" rounded="pill"
            prepend-icon="mdi-cloud-download" :loading="recovering === it.image_id"
            @click="onRecover(it)"
          >Recover</v-btn>
          <span v-else class="fc-muted text-caption">
            no pollable source — replace the file manually
          </span>
        </div>
        <p class="fc-muted text-caption mt-2 mb-0">
          Recover deletes the bad copy (and its record) and re-checks its
          subscription source, so a fresh download re-imports it and re-enters
          processing. Files without a pollable source need manual replacement.
        </p>
      </template>
      <p v-else-if="!overview.total" class="fc-muted text-body-2 mb-0">
        No failed jobs — the pipeline is clean.
      </p>
      <p v-else-if="!overview.triage.unclassified" class="fc-muted text-body-2 mb-0">
        No defective files — every probed failure was operational
        (file&nbsp;OK). Requeue them from the GPU agent card.
      </p>
    </template>
  </MaintenanceTile>
 </template>
 <script setup>
 import { computed, onMounted, ref } from 'vue'
 import { toast } from '../../utils/toast.js'
 import MaintenanceTile from '../common/MaintenanceTile.vue'
 import { useGpuStore } from '../../stores/gpu.js'
 const store = useGpuStore()
 const loading = ref(true)
 const overview = ref(null)
 const probing = ref(false)
 const recovering = ref(null)
 // image_id -> 'no_source' for rows recovery already declined; keeps the
 // verdict visible instead of a button that fails the same way again.
 const recovered = ref({})
 const defects = computed(() =>
  (overview.value?.items || []).filter((i) => i.triage_status === 'defect'))
 const classSummary = computed(() => {
  const bc = overview.value?.by_class || {}
  return Object.entries(bc)
    .sort((a, b) => b[1] - a[1])
    .map(([k, n]) => `${k.replaceAll('_', ' ')} ${n}`)
    .join(' · ')
 })
 onMounted(refresh)
 async function refresh() {
  loading.value = true
  try {
    overview.value = await store.errors()
  } catch (e) {
    toast({ text: `Could not load failed jobs: ${e.message}`, type: 'error' })
  } finally {
    loading.value = false
  }
 }
 async function onProbe() {
  probing.value = true
  try {
    await store.triageErrors()
    toast({ text: 'Probe queued — verdicts appear here as files are checked (large videos take a while)', type: 'success' })
  } catch (e) {
    toast({ text: `Could not start the probe: ${e.message}`, type: 'error' })
  } finally {
    probing.value = false
  }
 }
 async function onRecover(it) {
  recovering.value = it.image_id
  try {
    const res = await store.recoverImage(it.image_id)
    if (res.status === 'refetch_queued') {
      toast({ text: `Deleted the bad copy and queued a re-check of source #${res.source_id} — it re-imports on the next fetch`, type: 'success' })
      await refresh()
    } else if (res.status === 'no_source') {
      recovered.value = { ...recovered.value, [it.image_id]: 'no_source' }
      toast({ text: 'No enabled subscription source covers this file — replace it manually', type: 'warning' })
    } else {
      toast({ text: 'Image record no longer exists — refreshing', type: 'warning' })
      await refresh()
    }
  } catch (e) {
    toast({ text: `Recovery failed: ${e.message}`, type: 'error' })
  } finally {
    recovering.value = null
  }
 }
 </script>
 <style scoped>
 .fc-muted { color: rgb(var(--v-theme-on-surface-variant)); }
 .fc-section-h {
  font-size: 13px; font-weight: 700; letter-spacing: 0.03em;
  text-transform: uppercase; color: rgb(var(--v-theme-on-surface));
 }
 .fc-queue { display: flex; gap: 24px; }
 .fc-q__n {
  font-size: 20px; font-weight: 700; line-height: 1.1;
  font-family: 'JetBrains Mono', monospace;
 }
 .fc-q__l {
  font-size: 11px; text-transform: uppercase; letter-spacing: 0.04em;
  color: rgb(var(--v-theme-on-surface-variant));
 }
 .fc-good { color: rgb(var(--v-theme-success)); }
 .fc-weak { color: rgb(var(--v-theme-error)); }
 .fc-defect {
  display: flex; align-items: center; gap: 12px;
  background: rgb(var(--v-theme-surface-light)); border-radius: 8px;
  padding: 6px 10px;
 }
 .fc-defect__thumb {
  flex: 0 0 44px; width: 44px; height: 44px; border-radius: 6px;
  overflow: hidden; display: flex; align-items: center; justify-content: center;
  background: rgba(0, 0, 0, 0.25);
 }
 .fc-defect__thumb img { width: 100%; height: 100%; object-fit: cover; }
 .fc-defect__meta { flex: 1; min-width: 0; }
 .fc-defect__err {
  overflow: hidden; text-overflow: ellipsis; white-space: nowrap;
 }
 </style>
@@ -3,7 +3,7 @@
    icon="mdi-brain"
    title="Concept heads (the learning suggester)"
    blurb="Train the per-concept heads that turn your tags into suggestions — they replace Camie and sharpen every time you accept or reject."
-    :open="headCount > 0 || running"
+    :open="running"
  >
    <p class="fc-muted text-body-2 mb-3">
      A <strong>head</strong> is a tiny classifier trained on the SigLIP
@@ -13,6 +13,7 @@
        <ThumbnailBackfillCard />
        <ArchiveReextractCard />
        <MissingFileRepairCard />
        <GpuTriageCard />
        <DbMaintenanceCard />
      </div>
    </section>
@@ -27,7 +28,6 @@
        <HeadsCard />
        <GpuAgentCard />
        <AliasTable />
        <TagEvalCard />
      </div>
    </section>
@@ -48,12 +48,12 @@ import MLBackfillCard from './MLBackfillCard.vue'
 import ThumbnailBackfillCard from './ThumbnailBackfillCard.vue'
 import ArchiveReextractCard from './ArchiveReextractCard.vue'
 import MissingFileRepairCard from './MissingFileRepairCard.vue'
 import GpuTriageCard from './GpuTriageCard.vue'
 import DbMaintenanceCard from './DbMaintenanceCard.vue'
 import MLThresholdSliders from './MLThresholdSliders.vue'
 import HeadsCard from './HeadsCard.vue'
 import GpuAgentCard from './GpuAgentCard.vue'
 import AliasTable from './AliasTable.vue'
 import TagEvalCard from './TagEvalCard.vue'
 import BackupCard from './BackupCard.vue'
 import { useSystemActivityStore } from '../../stores/systemActivity.js'
@@ -1,303 +0,0 @@
 <template>
  <MaintenanceTile
    icon="mdi-flask-outline"
    title="Tagging eval (heads vs centroid)"
    blurb="Measure whether a trained head beats the old centroid on your own tags — and whether tagging more sharpens it."
    :open="!!run"
  >
    <p class="fc-muted text-body-2 mb-3">
      Reuses the SigLIP embeddings already stored on your images (no re-embed, no
      GPU). For each concept it trains a logistic-regression <strong>head</strong>
      on your positives + negatives and compares it to the old single
      <strong>centroid</strong>, with cross-validated AP/F1 and a learning curve.
      Runs as a background task; the result is saved and reloads here.
    </p>
    <v-textarea
      v-model="conceptsText" label="Concepts (comma-separated)"
      rows="2" auto-grow density="compact" hide-details class="mb-3"
      :disabled="running"
    />
    <div class="d-flex mb-3" style="gap: 12px;">
      <v-text-field
        v-model.number="autoTopN" label="+ auto-add top-N concepts"
        type="number" min="0" max="200" density="compact" hide-details
        :disabled="running" style="max-width: 220px;"
      />
      <v-text-field
        v-model.number="precisionTarget" label="Auto-apply precision target"
        type="number" min="0.5" max="0.999" step="0.01" density="compact" hide-details
        :disabled="running" style="max-width: 220px;"
      />
    </div>
    <v-btn
      v-if="!running"
      color="accent" variant="flat" rounded="pill"
      prepend-icon="mdi-play" :loading="busy" @click="onStart"
    >Run eval</v-btn>
    <div v-if="running" class="mt-3">
      <v-progress-linear indeterminate color="accent" />
      <div class="text-body-2 mt-2 fc-muted">Running… (started {{ startedAgo }})</div>
    </div>
    <v-alert
      v-if="run && run.status === 'error'"
      type="error" variant="tonal" density="compact" class="mt-3"
    >Eval failed: {{ run.error }}</v-alert>
    <div v-if="report" class="mt-4">
      <div class="fc-muted text-caption mb-2">
        Ran {{ formatTime(report.generated_at) }} ·
        {{ report.concepts.length }} concept(s) ·
        neg ratio {{ report.params.neg_ratio }}, {{ report.params.cv_folds }}-fold CV
      </div>
      <div v-for="c in report.concepts" :key="c.name" class="fc-cc">
        <div class="fc-cc__head">
          <span class="fc-cc__name">{{ c.name }}</span>
          <span v-if="c.skipped" class="fc-muted text-caption">— skipped: {{ c.skipped }}</span>
          <span v-else class="fc-muted text-caption">
            {{ c.n_pos }} pos · {{ c.n_neg }} neg<span v-if="c.n_rejected"> ({{ c.n_rejected }} rejected)</span>
          </span>
        </div>
        <template v-if="!c.skipped">
          <table class="fc-metrics">
            <thead>
              <tr><th></th><th>AP</th><th>F1</th><th>Prec</th><th>Rec</th></tr>
            </thead>
            <tbody>
              <tr>
                <td class="fc-metrics__lbl">Head</td>
                <td class="fc-num fc-win">{{ c.head.ap }}</td>
                <td class="fc-num">{{ c.head.f1 }}</td>
                <td class="fc-num">{{ c.head.precision }}</td>
                <td class="fc-num">{{ c.head.recall }}</td>
              </tr>
              <tr>
                <td class="fc-metrics__lbl fc-muted">Centroid</td>
                <td class="fc-num fc-muted">{{ c.centroid.ap }}</td>
                <td class="fc-num fc-muted">{{ c.centroid.f1 }}</td>
                <td class="fc-num fc-muted">{{ c.centroid.precision }}</td>
                <td class="fc-num fc-muted">{{ c.centroid.recall }}</td>
              </tr>
            </tbody>
          </table>
          <div class="text-caption mb-2" :class="apDelta(c) >= 0 ? 'fc-up' : 'fc-down'">
            Δ AP {{ apDelta(c) >= 0 ? '+' : '' }}{{ apDelta(c).toFixed(3) }}
            (head − centroid)
          </div>
          <div class="text-caption mb-2">
            <span class="fc-muted">Auto-apply:</span>
            <template v-if="c.head.auto_apply">
              <span class="fc-up">ready</span> — at P≥{{ c.head.auto_apply.target }}
              catches recall <strong>{{ c.head.auto_apply.recall }}</strong>
              (thr {{ c.head.auto_apply.threshold }})
            </template>
            <span v-else class="fc-down">not reachable at P≥{{ report.params.precision_target }}</span>
          </div>
          <div v-if="c.curve && c.curve.length" class="fc-curve">
            <span class="fc-muted text-caption">Learning curve (AP @ N positives):</span>
            <span v-for="p in c.curve" :key="p.n_pos" class="fc-curve__pt">
              {{ p.n_pos }}→<strong>{{ p.ap }}</strong>
            </span>
          </div>
          <div v-if="c.examples" class="fc-ex">
            <div
              v-for="grp in [
                { dir: 'suggest', items: c.examples.head_would_suggest,
                  label: `Head would suggest — ✓ tag it, ✗ not ${c.name}` },
                { dir: 'doubts', items: c.examples.head_doubts_positive,
                  label: `Head doubts your tag — ✓ keep, ✗ remove (not ${c.name})` },
              ]" :key="grp.dir" class="fc-ex__row"
            >
              <div class="fc-muted text-caption mb-1">{{ grp.label }}</div>
              <div class="fc-ex__thumbs">
                <div
                  v-for="it in grp.items" :key="`${grp.dir}${it.id}`"
                  class="fc-ex__item"
                  :class="actedLabel(c, grp.dir, it) ? 'fc-ex__item--acted' : ''"
                >
                  <button
                    type="button" class="fc-ex__thumb"
                    :title="`#${it.id} — click to enlarge`" @click="modal.open(it.id)"
                  >
                    <img :src="it.thumbnail_url" loading="lazy" />
                  </button>
                  <div v-if="actedLabel(c, grp.dir, it)" class="fc-ex__badge">
                    {{ actedLabel(c, grp.dir, it) }}
                  </div>
                  <div v-else class="fc-ex__acts">
                    <button
                      class="fc-act fc-act--yes" type="button"
                      :title="`Yes — it is ${c.name}`" @click="act(c, it, grp.dir, 'yes')"
                    ><v-icon size="15">mdi-check</v-icon></button>
                    <button
                      class="fc-act fc-act--no" type="button"
                      :title="`No — not ${c.name}`" @click="act(c, it, grp.dir, 'no')"
                    ><v-icon size="15">mdi-close</v-icon></button>
                  </div>
                </div>
              </div>
            </div>
          </div>
        </template>
      </div>
    </div>
  </MaintenanceTile>
 </template>
 <script setup>
 import { toast } from '../../utils/toast.js'
 import { computed, onMounted, onUnmounted, ref } from 'vue'
 import MaintenanceTile from '../common/MaintenanceTile.vue'
 import { useTagEvalStore } from '../../stores/tagEval.js'
 import { useModalStore } from '../../stores/modal.js'
 const DEFAULT_CONCEPTS =
  'glasses, cat, dog, horse, goblin, cum, lactation, fellatio, xray, stomach bulge'
 const store = useTagEvalStore()
 const modal = useModalStore()
 const run = ref(null)
 const conceptsText = ref(DEFAULT_CONCEPTS)
 const autoTopN = ref(0)
 const precisionTarget = ref(0.97)
 const busy = ref(false)
 let pollTimer = null
 const running = computed(() => run.value?.status === 'running')
 const report = computed(() => (run.value?.status === 'ready' ? run.value.report : null))
 const startedAgo = computed(() =>
  run.value?.started_at ? formatTime(run.value.started_at) : '')
 // Rehydrate the persisted run on mount so the report survives navigation — the
 // task runs backend-side regardless; we just reconnect to its row.
 onMounted(async () => {
  try {
    const latest = await store.latest()
    if (latest) {
      run.value = await store.getRun(latest.id)
      if (run.value.status === 'running') startPoll(latest.id)
    }
  } catch { /* non-fatal — card still works for a fresh run */ }
 })
 onUnmounted(stopPoll)
 function startPoll(id) {
  stopPoll()
  pollTimer = setInterval(async () => {
    try {
      run.value = await store.getRun(id)
      if (run.value.status !== 'running') stopPoll()
    } catch (e) {
      stopPoll()
      toast({ text: `Eval poll failed: ${e.message}`, type: 'error' })
    }
  }, 5000)
 }
 function stopPoll() {
  if (pollTimer) { clearInterval(pollTimer); pollTimer = null }
 }
 async function onStart() {
  busy.value = true
  try {
    const concepts = conceptsText.value.split(',').map(s => s.trim()).filter(Boolean)
    const res = await store.start({
      concepts,
      auto_top_n: Number(autoTopN.value) || 0,
      precision_target: Number(precisionTarget.value) || 0.97,
    })
    run.value = await store.getRun(res.run_id)
    startPoll(res.run_id)
  } catch (e) {
    const msg = e.body?.running_id
      ? 'An eval is already running.'
      : e.message
    toast({ text: `Could not start eval: ${msg}`, type: 'error' })
  } finally {
    busy.value = false
  }
 }
 function apDelta(c) { return (c.head?.ap ?? 0) - (c.centroid?.ap ?? 0) }
 function formatTime(iso) {
  if (!iso) return ''
  try { return new Date(iso).toLocaleString() } catch { return iso }
 }
 // Acting on an example writes the SAME tables the head trains on, so a re-run
 // reflects the correction. Keyed per (concept, list, image); the report ids are
 // frozen at run time, so we just grey out what's been handled in this view.
 const acted = ref({})
 const actedKey = (c, dir, it) => `${c.tag_id}:${dir}:${it.id}`
 const actedLabel = (c, dir, it) => acted.value[actedKey(c, dir, it)] || ''
 async function act(c, it, dir, verdict) {
  const key = actedKey(c, dir, it)
  let call, label
  if (dir === 'suggest' && verdict === 'yes') { call = store.applyTag(it.id, c.tag_id); label = 'tagged' }
  else if (dir === 'suggest' && verdict === 'no') { call = store.rejectTag(it.id, c.tag_id); label = 'rejected' }
  else if (dir === 'doubts' && verdict === 'no') { call = store.removeTag(it.id, c.tag_id); label = 'removed' }
  else { call = store.confirmTag(it.id, c.tag_id); label = 'kept' }   // doubt + yes = keep (confirm)
  try {
    await call
    acted.value[key] = label
  } catch (e) {
    toast({ text: `Action failed: ${e.message}`, type: 'error' })
  }
 }
 </script>
 <style scoped>
 .fc-muted { color: rgb(var(--v-theme-on-surface-variant)); }
 .fc-cc {
  padding: 12px 0;
  border-top: 1px solid rgb(var(--v-theme-surface-light));
 }
 .fc-cc__head { display: flex; align-items: baseline; gap: 8px; margin-bottom: 6px; }
 .fc-cc__name { font-weight: 600; }
 .fc-metrics { width: 100%; max-width: 360px; border-collapse: collapse; font-size: 13px; }
 .fc-metrics th { text-align: right; font-weight: 600; color: rgb(var(--v-theme-on-surface-variant)); padding: 0 8px; }
 .fc-metrics__lbl { text-align: left; }
 .fc-num { text-align: right; font-variant-numeric: tabular-nums; padding: 1px 8px; }
 .fc-win { color: rgb(var(--v-theme-accent)); font-weight: 600; }
 .fc-up { color: rgb(var(--v-theme-success)); }
 .fc-down { color: rgb(var(--v-theme-error)); }
 .fc-curve { margin-bottom: 8px; }
 .fc-curve__pt { margin-left: 10px; font-size: 13px; font-variant-numeric: tabular-nums; }
 .fc-ex__row { margin-top: 8px; }
 .fc-ex__thumbs { display: flex; flex-wrap: wrap; gap: 6px; }
 .fc-ex__item { position: relative; width: 120px; height: 120px; }
 .fc-ex__item--acted { opacity: 0.45; }
 .fc-ex__thumb {
  display: block; width: 100%; height: 100%; border-radius: 6px;
  overflow: hidden; background: rgb(var(--v-theme-surface-light));
  outline: 1px solid transparent; transition: outline-color 0.12s;
  border: none; padding: 0; cursor: pointer;
 }
 .fc-ex__thumb:hover { outline-color: rgb(var(--v-theme-accent)); }
 .fc-ex__thumb img { width: 100%; height: 100%; object-fit: cover; display: block; }
 .fc-ex__acts { position: absolute; top: 4px; right: 4px; display: flex; gap: 4px; }
 .fc-act {
  width: 26px; height: 26px; border-radius: 50%; border: none; cursor: pointer;
  display: flex; align-items: center; justify-content: center; color: #fff;
  opacity: 0.9; box-shadow: 0 1px 3px rgba(0, 0, 0, 0.4); transition: transform 0.1s;
 }
 .fc-act:hover { opacity: 1; transform: scale(1.1); }
 .fc-act--yes { background: rgb(var(--v-theme-success)); }
 .fc-act--no { background: rgb(var(--v-theme-error)); }
 .fc-ex__badge {
  position: absolute; bottom: 4px; left: 4px; right: 4px; text-align: center;
  font-size: 10px; text-transform: uppercase; letter-spacing: 0.05em;
  background: rgba(0, 0, 0, 0.65); color: #fff; border-radius: 3px; padding: 1px 0;
 }
 </style>
@@ -42,53 +42,6 @@
    </div>
  </MaintenanceTile>
  <MaintenanceTile
    icon="mdi-tag-off"
    title="Legacy migration tags"
    blurb="Purge retired archive/post/artist + source:* tags."
    destructive
  >
    <p class="fc-muted text-body-2 mb-3">
      Purge legacy IR-migration tags FC no longer uses: retired/system
      kinds (<code>archive</code>, <code>post</code>, <code>artist</code> — e.g.
      <code>BlenderKnight:Hannah_BJ_Loops</code>) plus <code>source:*</code> tags
      (ImageRepo's old <code>source</code> kind, migrated to <code>general</code>).
      Provenance and artists are their own systems now, so these are pure noise.
      Removes them from every image.
    </p>
    <v-btn
      color="accent" variant="flat" rounded="pill"
      prepend-icon="mdi-magnify"
      :loading="loadingKindPreview"
      class="mb-3"
      @click="onKindPreview"
    >Preview legacy tags</v-btn>
    <div v-if="kindPreview">
      <p class="text-body-2 mb-2">
        <strong>{{ kindPreview.count }}</strong> legacy tag(s).
        <span v-for="(n, k) in kindPreview.by_kind" :key="k" class="fc-muted">
          {{ k }}: {{ n }}&nbsp;&nbsp;
        </span>
        <span v-for="(n, p) in kindPreview.by_prefix" :key="p" class="fc-muted">
          {{ p }}: {{ n }}&nbsp;&nbsp;
        </span>
      </p>
      <SampleNameGrid
        v-if="kindPreview.sample_names?.length"
        :names="kindPreview.sample_names" class="mb-3"
      />
      <v-btn
        color="error" variant="flat" rounded="pill"
        prepend-icon="mdi-delete-sweep"
        :disabled="!kindPreview.count"
        :loading="kindCommitting"
        @click="onKindCommit"
      >Delete {{ kindPreview.count }} legacy tag(s)</v-btn>
    </div>
  </MaintenanceTile>
  <MaintenanceTile
    icon="mdi-tag-multiple"
    title="Reset content tagging"
@@ -216,16 +169,6 @@ const {
  emptyPreview: (r) => ({ count: 0, sample_names: r.sample_names || [] }),
 })
 // Legacy migration-tag purge.
 const {
  previewData: kindPreview, previewing: loadingKindPreview,
  committing: kindCommitting, runPreview: onKindPreview, runCommit: onKindCommit,
 } = usePreviewCommit({
  preview: () => store.purgeLegacyTags({ dryRun: true }),
  commit: () => store.purgeLegacyTags({ dryRun: false }),
  emptyPreview: { count: 0, by_kind: {}, by_prefix: {}, sample_names: [] },
 })
 // Reset content tagging (general + character).
 const {
  previewData: resetPreview, previewing: loadingResetPreview,
@@ -101,10 +101,6 @@ export const useAdminStore = defineStore('admin', () => {
    })
  }
  function purgeLegacyTags(opts = {}) {
    return _dryRunPost('/api/admin/tags/purge-legacy', opts)
  }
  // Destructive: deletes ALL general + character tags so the operator can
  // re-tag from scratch via auto-suggest. fandom + series preserved.
  function resetContentTagging(opts = {}) {
@@ -154,7 +150,6 @@ export const useAdminStore = defineStore('admin', () => {
    pruneUnusedTags,
    pruneBarePosts,
    reconcileDuplicatePosts,
    purgeLegacyTags,
    resetContentTagging,
    normalizeTags,
    pollTaskUntilDone,
@@ -36,10 +36,32 @@ export const useGpuStore = defineStore('gpu', () => {
  }
  // Requeue ONLY the errored jobs (all task types) — the scoped recovery after
-  // an agent fix, without re-running the done library. Returns { requeued }.
+  // an agent fix, without re-running the done library. Triage-confirmed
  // defects stay put (they recover via recoverImage instead).
  // Returns { requeued, pruned, defects_kept }.
  async function retryErrors() {
    return await api.post('/api/gpu/retry_errors')
  }
-  return { token, rotateToken, status, backfill, reprocess, retryErrors }
+  // Failure-triage view (#125): errored jobs joined with integrity verdicts.
  // Returns { total, by_class, triage, items }.
  async function errors() {
    return await api.get('/api/gpu/errors')
  }
  // Run the file-probe sweep now instead of waiting out the 15-min beat.
  async function triageErrors() {
    return await api.post('/api/gpu/errors/triage')
  }
  // Recover a defective original: delete the bad copy + record, re-poll its
  // Source. Returns { status: 'refetch_queued'|'no_source'|'not_found', ... }.
  async function recoverImage(imageId) {
    return await api.post(`/api/gpu/errors/${imageId}/recover`)
  }
  return {
    token, rotateToken, status, backfill, reprocess, retryErrors,
    errors, triageErrors, recoverImage,
  }
 })
@@ -1,57 +0,0 @@
 import { defineStore } from 'pinia'
 import { useApi } from '../composables/useApi.js'
 // Tag-eval (#1130): trigger + revisit the head-vs-centroid learning-curve eval.
 // The run + full report live server-side (tag_eval_run), so the card rehydrates
 // from getRun() on mount — the report survives navigation.
 export const useTagEvalStore = defineStore('tagEval', () => {
  const api = useApi()
  async function start(params) {
    return await api.post('/api/tag-eval', { body: { params } })
  }
  async function getRun(id) {
    return await api.get(`/api/tag-eval/${id}`)   // includes the full report
  }
  // The most recent run (light row, no report) — the card calls getRun() with
  // its id to pull the persisted report on mount.
  async function latest() {
    const body = await api.get('/api/tag-eval', { params: { limit: 1 } })
    return (body.runs && body.runs[0]) || null
  }
  // --- Acting on the head's example lists (closes the learn-from-tags loop).
  // These write the SAME tables the head trains on: image_tag (positives) and
  // tag_suggestion_rejection (negatives, via the dismiss endpoint).
  // "Yes, it is this" — apply the tag (new positive).
  async function applyTag(imageId, tagId) {
    return await api.post(`/api/images/${imageId}/tags`,
      { body: { tag_id: tagId, source: 'manual' } })
  }
  // "No, it's not" on an UNtagged suggestion — record a rejection (hard negative).
  async function rejectTag(imageId, tagId) {
    return await api.post(`/api/images/${imageId}/suggestions/dismiss`,
      { body: { tag_id: tagId } })
  }
  // "Not it" on one of YOUR positives the head doubts — remove the tag AND
  // record the rejection (kills the bad positive, leaves a hard negative).
  async function removeTag(imageId, tagId) {
    await api.delete(`/api/images/${imageId}/tags/${tagId}`)
    return await api.post(`/api/images/${imageId}/suggestions/dismiss`,
      { body: { tag_id: tagId } })
  }
  // "Keep" — affirm a doubted positive is correct. Records a confirmation so it
  // stops resurfacing in the doubts list (it stays a positive either way).
  async function confirmTag(imageId, tagId) {
    return await api.post(`/api/images/${imageId}/tags/${tagId}/confirm`)
  }
  return { start, getRun, latest, applyTag, rejectTag, removeTag, confirmTag }
 })
@@ -0,0 +1,71 @@
 """Agent bandwidth governor (TokenBucket + PidReadMeter) — pure stdlib, so it
 runs in the unit lane even though the rest of the agent (torch/ultralytics)
 can't be imported in CI. Timing asserts use generous margins: they check the
 throttle's ORDER of magnitude, not scheduler precision."""
 import os
 import threading
 import time
 from agent.fc_agent.throttle import PidReadMeter, TokenBucket
 def test_unlimited_never_blocks_and_counts():
    b = TokenBucket(0)
    t0 = time.monotonic()
    for _ in range(100):
        b.take(10_000_000)
    assert time.monotonic() - t0 < 0.2
    assert b.consumed == 1_000_000_000
 def test_take_enforces_average_rate():
    # 400 KB/s with a 1s burst: the first 400KB is free (burst), the next
    # 200KB is 0.5s of debt the caller must wait out.
    b = TokenBucket(400_000)
    b.take(400_000)
    t0 = time.monotonic()
    b.take(200_000)
    elapsed = time.monotonic() - t0
    assert 0.35 <= elapsed < 3.0
 def test_set_rate_zero_unblocks_a_waiter():
    b = TokenBucket(1_000)            # 1 KB/s → a 1MB take would wait ~17 min
    started = threading.Event()
    def blocked_take():
        started.set()
        b.take(1_000_000)
    th = threading.Thread(target=blocked_take, daemon=True)
    th.start()
    started.wait(1.0)
    time.sleep(0.1)                   # let it enter the wait loop
    b.set_rate(0)                     # lift the cap live (the UI dial)
    th.join(timeout=2.0)
    assert not th.is_alive()
 def test_charge_reports_debt_without_blocking():
    b = TokenBucket(1_000)
    t0 = time.monotonic()
    assert b.charge(1_000) == 0.0     # covered by the 1s burst
    debt = b.charge(2_000)            # 2s over budget → ~2s of debt
    assert time.monotonic() - t0 < 0.2
    assert 1.5 <= debt <= 2.1
    assert b.consumed == 3_000
 def test_pid_read_meter_tracks_own_reads():
    m = PidReadMeter(os.getpid())
    first = m.delta()                 # cumulative rchar since process start
    assert first is not None and first > 0
    with open("/dev/zero", "rb") as f:
        f.read(1_048_576)
    grown = m.delta()
    assert grown is not None and grown >= 1_048_576
 def test_pid_read_meter_missing_pid_degrades_to_none():
    # PID 2^22+ is above the default pid_max — guaranteed absent.
    assert PidReadMeter(2**22 + 12345).delta() is None
@@ -390,63 +390,6 @@ async def test_prune_unused_commit_deletes_and_returns_count(client, db):
    assert "byebye" in body["sample_names"]
@pytest.mark.asyncio
 async def test_purge_legacy_dry_run_counts_by_kind_and_prefix(client, db):
    db.add_all([
        Tag(name="BlenderKnight:Hannah_BJ_Loops", kind=TagKind.archive),
        Tag(name="BlenderKnight:May Animation", kind=TagKind.post),
        Tag(name="SomeArtist", kind=TagKind.artist),
        # IR `source` kind fell back to general during migration —
        # caught by the source:* name prefix, not by kind.
        Tag(name="source:patreon", kind=TagKind.general),
        Tag(name="blonde hair", kind=TagKind.general),  # must survive
    ])
    await db.commit()
    resp = await client.post(
        "/api/admin/tags/purge-legacy", json={"dry_run": True},
    )
    body = await resp.get_json()
    assert body["count"] == 4
    assert body["by_kind"] == {"archive": 1, "post": 1, "artist": 1}
    assert body["by_prefix"] == {"source:*": 1}
    assert "blonde hair" not in body["sample_names"]
    assert "source:patreon" in body["sample_names"]
@pytest.mark.asyncio
 async def test_purge_legacy_commit_deletes_only_legacy(client, db):
    from sqlalchemy import func, select
    db.add_all([
        Tag(name="arch1", kind=TagKind.archive),
        Tag(name="source:fanbox", kind=TagKind.general),
        Tag(name="keepme", kind=TagKind.general),
        Tag(name="char1", kind=TagKind.character),
    ])
    await db.commit()
    resp = await client.post(
        "/api/admin/tags/purge-legacy", json={"dry_run": False},
    )
    body = await resp.get_json()
    assert body["deleted"] == 2  # arch1 + source:fanbox
    # The plain general + character tags survive.
    keep = (await db.execute(
        select(Tag.name).where(Tag.name.in_(["keepme", "char1"]))
    )).scalars().all()
    assert set(keep) == {"keepme", "char1"}
    # No archive/post/artist or source:* left.
    gone = (await db.execute(
        select(func.count()).select_from(Tag).where(
            (Tag.kind.in_([TagKind.archive, TagKind.post, TagKind.artist]))
            | (Tag.name.like("source:%"))
        )
    )).scalar_one()
    assert gone == 0
 # --- Tier-A: POST /posts/prune-bare + /posts/reconcile-duplicates ---
 # These two routes share _run_dry_run_op with the tag prunes (DRY pass,
 # task #753); cover their apply path + reconcile's source_id passthrough so
@@ -1,6 +1,8 @@
 """GPU-job HTTP API (#114): bearer auth + lease/submit round-trip + backfill."""
 from datetime import UTC, datetime, timedelta
 import pytest
-from sqlalchemy import select
+from sqlalchemy import func, select
 from backend.app.models import GpuJob, ImageRecord
 from backend.app.services.ml.gpu_jobs import GpuJobService
@@ -153,8 +155,11 @@ async def test_release_hands_job_back_to_pending(client, db):
@pytest.mark.asyncio
 async def test_retry_errors_requeues_only_errored(client, db):
-    """/retry_errors resets ERRORED jobs (any task) to pending with a fresh
+    """/retry_errors prunes stale tombstones first (older duplicates + rows a
-    retry budget — and leaves done work untouched (it is NOT /reprocess)."""
+    later success made moot), then resets the SURVIVING errored jobs to pending
    with a fresh retry budget — and leaves done work untouched (NOT /reprocess).
    The prune is what stops one failing file fanning out into duplicate pending
    jobs (the 2026-07-02 tombstone loop minted one error row per hour)."""
    img1 = await _img(db, "1" * 64)
    img2 = await _img(db, "2" * 64)
    svc = GpuJobService(db)
@@ -165,12 +170,30 @@ async def test_retry_errors_requeues_only_errored(client, db):
    j_err.status = "error"
    j_err.attempts = 3
    j_err.error = "no frames sampled from video (unprocessable)"
    j_err.updated_at = datetime.now(UTC)
    j_done.status = "done"
    # Loop-era leftovers: an OLDER duplicate error row for img1's ccip, and a
    # tombstone img2's done row makes moot — both pruned, never requeued.
    dup = GpuJob(
        image_record_id=img1.id, task="ccip", status="error",
        error="older duplicate", updated_at=datetime.now(UTC) - timedelta(hours=1),
    )
    moot = GpuJob(
        image_record_id=img2.id, task="siglip", status="error",
        error="superseded by the done row",
    )
    db.add(dup)
    db.add(moot)
    await db.flush()
    dup_id = dup.id
    moot_id = moot.id
    await db.commit()
    resp = await client.post("/api/gpu/retry_errors")
    assert resp.status_code == 200
-    assert (await resp.get_json())["requeued"] == 1
+    body = await resp.get_json()
    assert body["requeued"] == 1
    assert body["pruned"] == 2
    # Column selects, not ORM refresh — the route wrote via Core DML.
    row = (await db.execute(
@@ -182,6 +205,58 @@ async def test_retry_errors_requeues_only_errored(client, db):
        select(GpuJob.status).where(GpuJob.id == done_id)
    )
    assert done_status == "done"
    survivors = (await db.execute(
        select(func.count()).select_from(GpuJob)
        .where(GpuJob.id.in_([dup_id, moot_id]))
    )).scalar_one()
    assert survivors == 0
    st = await (await client.get("/api/gpu/status")).get_json()
    assert st["pending"] == 1 and st["error"] == 0
@pytest.mark.asyncio
 async def test_retry_errors_keeps_triaged_defects(client, db):
    """A probe-confirmed DEFECT is a bad FILE — requeueing it just burns agent
    time re-minting the tombstone, so /retry_errors leaves it for the recovery
    surface and reports it as defects_kept."""
    img1 = await _img(db, "4" * 64)
    img2 = await _img(db, "5" * 64)
    db.add(GpuJob(image_record_id=img1.id, task="ccip", status="error",
                  attempts=3, error="moov atom not found",
                  triage_status="defect"))
    db.add(GpuJob(image_record_id=img2.id, task="ccip", status="error",
                  attempts=3, error="ffmpeg timed out after 1200s"))
    await db.commit()
    body = await (await client.post("/api/gpu/retry_errors")).get_json()
    assert body["requeued"] == 1
    assert body["defects_kept"] == 1
    rows = dict((await db.execute(
        select(GpuJob.image_record_id, GpuJob.status)
    )).all())
    assert rows[img1.id] == "error"       # defect stays tombstoned
    assert rows[img2.id] == "pending"     # operational failure requeued
@pytest.mark.asyncio
 async def test_errors_endpoint_reports_triage_view(client, db):
    img = await _img(db, "6" * 64)
    db.add(GpuJob(image_record_id=img.id, task="ccip", status="error",
                  attempts=3,
                  error="no frames sampled from video — moov atom not found"))
    await db.commit()
    resp = await client.get("/api/gpu/errors")
    assert resp.status_code == 200
    body = await resp.get_json()
    assert body["total"] == 1
    assert body["by_class"] == {"truncated_or_corrupt": 1}
    assert body["triage"]["unclassified"] == 1
    item = body["items"][0]
    assert item["image_id"] == img.id
    assert item["task"] == "ccip"
    assert item["reason_class"] == "truncated_or_corrupt"
    assert item["triage_status"] is None
    assert item["image_url"].startswith("/images/")
@@ -1,77 +0,0 @@
 import pytest
 from backend.app.models import TagEvalRun
 from backend.app.services.ml.tag_eval import (
    DEFAULT_CONCEPTS,
    _normalize_params,
 )
 pytestmark = pytest.mark.integration
 def test_normalize_params_defaults_and_overrides():
    d = _normalize_params(None)
    assert d["concepts"] == DEFAULT_CONCEPTS
    assert d["neg_ratio"] >= 1 and d["cv_folds"] >= 2
    over = _normalize_params(
        {"concepts": ["glasses", " ", "cat"], "neg_ratio": "4",
         "cv_folds": "1", "curve_points": [30, 10, 10]}
    )
    assert over["concepts"] == ["glasses", "cat"]   # blanks dropped
    assert over["neg_ratio"] == 4
    assert over["cv_folds"] == 2                      # clamped to >=2
    assert over["curve_points"] == [10, 30]           # deduped + sorted
@pytest.mark.asyncio
 async def test_history_and_detail_rehydrate(client, db):
    # A finished run with a report — the persisted row IS the survives-navigation
    # source: history is light (no report), detail carries it.
    run = TagEvalRun(
        params={"concepts": ["glasses"]},
        status="ready",
        report={"concepts": [{"name": "glasses", "head": {"ap": 0.9}}]},
    )
    db.add(run)
    await db.flush()
    await db.commit()
    rid = run.id
    h = await client.get("/api/tag-eval?limit=10")
    assert h.status_code == 200
    hbody = await h.get_json()
    row = next(r for r in hbody["runs"] if r["id"] == rid)
    assert row["status"] == "ready"
    assert "report" not in row                        # list stays light
    d = await client.get(f"/api/tag-eval/{rid}")
    assert d.status_code == 200
    dbody = await d.get_json()
    assert dbody["report"]["concepts"][0]["name"] == "glasses"
@pytest.mark.asyncio
 async def test_create_enqueues_running(client, db, monkeypatch):
    monkeypatch.setattr(
        "backend.app.tasks.ml.tag_eval_run.delay", lambda *a, **k: None
    )
    resp = await client.post("/api/tag-eval", json={"params": {"concepts": ["cat"]}})
    assert resp.status_code == 202
    body = await resp.get_json()
    assert body["status"] == "running"
    got = await db.get(TagEvalRun, body["run_id"])
    assert got is not None and got.status == "running"
@pytest.mark.asyncio
 async def test_create_conflicts_when_one_running(client, db, monkeypatch):
    monkeypatch.setattr(
        "backend.app.tasks.ml.tag_eval_run.delay", lambda *a, **k: None
    )
    db.add(TagEvalRun(params={}, status="running"))
    await db.flush()
    await db.commit()
    resp = await client.post("/api/tag-eval", json={"params": {}})
    assert resp.status_code == 409
    body = await resp.get_json()
    assert body["error"] == "eval_already_running"
@@ -5,7 +5,11 @@ import pytest
 from sqlalchemy import func, select
 from backend.app.models import GpuJob, ImageRecord, ImageRegion
-from backend.app.services.ml.gpu_jobs import GpuJobService
+from backend.app.services.ml.gpu_jobs import (
    EXPIRED_POISON_CAP,
    PENDING_POISON_CAP,
    GpuJobService,
 )
 pytestmark = pytest.mark.integration
@@ -263,3 +267,84 @@ async def test_recover_orphaned_resets_only_expired(db):
    await db.commit()
    assert (await db.get(GpuJob, expired.id)).status == "pending"
    assert (await db.get(GpuJob, fresh.id)).status == "leased"   # untouched
@pytest.mark.asyncio
 async def test_backfill_skips_errored_images(db):
    # An errored job is a TOMBSTONE for its (image, task): no backfill variant
    # re-enqueues it — retry is deliberate-only (/retry_errors). Pre-fix, the
    # hourly ccip run minted a fresh doomed job per bad file forever.
    from backend.app.tasks.ml import enqueue_gpu_backfill
    img = await _img(db, "f1" * 32)
    svc = GpuJobService(db)
    for task in ("ccip", "siglip", "embed"):
        job = await svc.enqueue(img.id, task)
        job.status = "error"
        job.error = "no frames sampled from video"
    await db.commit()
    assert enqueue_gpu_backfill("ccip") == 0
    assert enqueue_gpu_backfill("siglip") == 0
    assert enqueue_gpu_backfill("embed") == 0
@pytest.mark.asyncio
 async def test_backfill_prunes_moot_error_tombstones(db):
    # Loop-era duplicates: several error rows for one (image, task), all made
    # moot by a later done row. The backfill's dedupe pass removes them, and
    # the done row still blocks re-enqueue.
    from backend.app.tasks.ml import enqueue_gpu_backfill
    img = await _img(db, "f2" * 32)
    for i in range(3):
        db.add(GpuJob(
            image_record_id=img.id, task="ccip", status="error",
            error=f"boom {i}",
        ))
    db.add(GpuJob(image_record_id=img.id, task="ccip", status="done"))
    await db.commit()
    assert enqueue_gpu_backfill("ccip") == 0
    statuses = (await db.execute(
        select(GpuJob.status).where(
            GpuJob.image_record_id == img.id, GpuJob.task == "ccip"
        )
    )).scalars().all()
    assert statuses == ["done"]
@pytest.mark.asyncio
 async def test_recover_poisons_runaway_jobs(db):
    # release/expiry loops never reach fail()'s attempt cap — the sweep converts
    # them to 'error': an expired lease after EXPIRED_POISON_CAP grants (job
    # crashes/wedges the agent every time) and a pending job after
    # PENDING_POISON_CAP grants that never completed (transfer stalls forever).
    img1 = await _img(db, "06" + "a" * 62)
    img2 = await _img(db, "07" + "a" * 62)
    svc = GpuJobService(db)
    j1 = await svc.enqueue(img1.id, "ccip")
    j2 = await svc.enqueue(img2.id, "ccip")
    j1.status = "leased"
    j1.lease_token = "dead"
    j1.lease_expires_at = datetime.now(UTC) - timedelta(minutes=10)
    j1.attempts = EXPIRED_POISON_CAP
    j2.attempts = PENDING_POISON_CAP
    j1_id = j1.id
    j2_id = j2.id
    await db.commit()
    assert await svc.recover_orphaned() == 0   # nothing recovered — both poisoned
    await db.commit()
    # Column selects, not ORM refresh — the sweep wrote via Core DML.
    rows = (await db.execute(
        select(GpuJob.id, GpuJob.status, GpuJob.error)
        .where(GpuJob.id.in_([j1_id, j2_id]))
    )).all()
    by_id = {r.id: r for r in rows}
    assert by_id[j1_id].status == "error"
    assert "poisoned" in by_id[j1_id].error
    assert by_id[j2_id].status == "error"
    assert "poisoned" in by_id[j2_id].error
@@ -0,0 +1,165 @@
 """Failure triage (#125): probe errored jobs' files, flag verdicts, recover.
 The probe is the arbiter: reason strings only bucket the overview. A file that
 passes checksum+decode is 'file_ok' (operational failure); anything else is a
 'defect' — surfaced for recovery and excluded from /retry_errors.
 """
 import hashlib
 import pytest
 from PIL import Image as PILImage
 from sqlalchemy import select
 from backend.app.models import (
    Artist,
    GpuJob,
    ImageProvenance,
    ImageRecord,
    Post,
    Source,
 )
 from backend.app.services.ml.gpu_triage import (
    classify_reason,
    recover_defective_image,
    triage_errored_jobs,
 )
 pytestmark = pytest.mark.integration
 def test_classify_reason_buckets():
    assert classify_reason(
        "no frames sampled from video — moov atom not found"
    ) == "truncated_or_corrupt"
    assert classify_reason("ffmpeg timed out after 1200s") == "timeout"
    assert classify_reason(
        "gave up after repeated transient failures: HTTPConnectionPool read timed out"
    ) == "transient"
    assert classify_reason(
        "poisoned: 10+ lease attempts without ever completing"
    ) == "poisoned"
    assert classify_reason("cannot identify image file") == "decode"
    assert classify_reason("something novel") == "other"
    assert classify_reason(None) == "other"
 async def _errored_image(db, tmp_path, *, name, sha, content: bytes | None,
                         error="no frames sampled from video — moov atom not found"):
    """An ImageRecord (file written iff content is not None) + an errored job."""
    path = tmp_path / name
    if content is not None:
        path.write_bytes(content)
    img = ImageRecord(
        path=str(path), sha256=sha, size_bytes=1, mime="image/png",
        width=1, height=1, origin="imported_filesystem",
        integrity_status="unknown",
    )
    db.add(img)
    await db.flush()
    db.add(GpuJob(image_record_id=img.id, task="ccip", status="error",
                  error=error, attempts=3))
    await db.flush()
    return img
@pytest.mark.asyncio
 async def test_triage_probes_and_splits_defect_vs_file_ok(db, tmp_path):
    # Healthy: a real PNG whose recorded sha matches its bytes → file_ok.
    ok_path = tmp_path / "fine.png"
    PILImage.new("RGB", (4, 4), (200, 30, 30)).save(ok_path)
    ok_sha = hashlib.sha256(ok_path.read_bytes()).hexdigest()
    ok = await _errored_image(db, tmp_path, name="fine.png", sha=ok_sha,
                              content=None, error="ffmpeg timed out after 1200s")
    # Corrupt: bytes don't match the recorded sha → defect.
    bad = await _errored_image(db, tmp_path, name="bad.png", sha="0" * 64,
                               content=b"not a real png")
    # Missing: no file on disk at all → defect (failed_verification).
    gone = await _errored_image(db, tmp_path, name="gone.png", sha="1" * 64,
                                content=None)
    await db.commit()
    summary = await db.run_sync(lambda s: triage_errored_jobs(s))
    assert summary["probed"] == 3
    assert summary["defect"] == 2
    assert summary["file_ok"] == 1
    assert summary["partial"] is False
    # Column selects, not ORM refresh — the sweep wrote via Core DML.
    rows = dict((await db.execute(
        select(GpuJob.image_record_id, GpuJob.triage_status)
        .where(GpuJob.status == "error")
    )).all())
    assert rows[ok.id] == "file_ok"
    assert rows[bad.id] == "defect"
    assert rows[gone.id] == "defect"
    verdicts = dict((await db.execute(
        select(ImageRecord.id, ImageRecord.integrity_status)
        .where(ImageRecord.id.in_([ok.id, bad.id, gone.id]))
    )).all())
    assert verdicts[ok.id] == "ok"
    assert verdicts[bad.id] == "corrupt"
    assert verdicts[gone.id] == "failed_verification"
    # Idempotent: everything already triaged → no re-probe.
    again = await db.run_sync(lambda s: triage_errored_jobs(s))
    assert again["probed"] == 0
@pytest.mark.asyncio
 async def test_recover_without_pollable_source_reports_no_source(db, tmp_path):
    img = await _errored_image(db, tmp_path, name="orphan.png", sha="2" * 64,
                               content=b"x")
    await db.commit()
    res = await db.run_sync(
        lambda s: recover_defective_image(s, img.id, images_root=tmp_path)
    )
    assert res["status"] == "no_source"
    still_there = (await db.execute(
        select(ImageRecord.id).where(ImageRecord.id == img.id)
    )).scalar_one_or_none()
    assert still_there == img.id
@pytest.mark.asyncio
 async def test_recover_deletes_record_and_requeues_source(
    client, db, tmp_path, monkeypatch,
 ):
    img = await _errored_image(db, tmp_path, name="fixme.png", sha="3" * 64,
                               content=b"x")
    artist = Artist(name="Recov", slug="recov")
    db.add(artist)
    await db.flush()
    src = Source(artist_id=artist.id, platform="patreon",
                 url="https://www.patreon.com/recov", enabled=True)
    db.add(src)
    await db.flush()
    post = Post(artist_id=artist.id, source_id=src.id, external_post_id="p1")
    db.add(post)
    await db.flush()
    db.add(ImageProvenance(image_record_id=img.id, post_id=post.id,
                           source_id=src.id))
    img_id, src_id = img.id, src.id
    await db.commit()
    queued = []
    monkeypatch.setattr(
        "backend.app.tasks.download.download_source.delay",
        lambda sid: queued.append(sid),
    )
    resp = await client.post(f"/api/gpu/errors/{img_id}/recover")
    assert resp.status_code == 200
    body = await resp.get_json()
    assert body["status"] == "refetch_queued"
    assert body["source_id"] == src_id
    assert queued == [src_id]
    # Record gone — the error tombstones cascade away with it.
    remaining = (await db.execute(
        select(ImageRecord.id).where(ImageRecord.id == img_id)
    )).scalar_one_or_none()
    assert remaining is None
    jobs_left = (await db.execute(
        select(GpuJob.id).where(GpuJob.image_record_id == img_id)
    )).scalars().all()
    assert jobs_left == []