diff --git a/agent/docker-compose.yml b/agent/docker-compose.yml
index 79538a4..bad164d 100644
--- a/agent/docker-compose.yml
+++ b/agent/docker-compose.yml
@@ -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}
diff --git a/agent/fc_agent/app.py b/agent/fc_agent/app.py
index 51de0b6..93795d4 100644
--- a/agent/fc_agent/app.py
+++ b/agent/fc_agent/app.py
@@ -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 = """
.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 = """
+
+
+ MB/s
+
auto-tuning downloaders to keep the GPU fed · max 8
@@ -281,6 +293,11 @@ _PAGE = """
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 = """
// 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+'%' }
diff --git a/agent/fc_agent/client.py b/agent/fc_agent/client.py
index 1f40150..b2b8cea 100644
--- a/agent/fc_agent/client.py
+++ b/agent/fc_agent/client.py
@@ -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))
- r.raise_for_status()
- return r.content
+ # With a throttle (the worker's shared TokenBucket), the body is streamed
+ # in chunks and each chunk is charged to the global bandwidth budget —
+ # 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,
diff --git a/agent/fc_agent/config.py b/agent/fc_agent/config.py
index 045114b..098d728 100644
--- a/agent/fc_agent/config.py
+++ b/agent/fc_agent/config.py
@@ -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")),
)
diff --git a/agent/fc_agent/media.py b/agent/fc_agent/media.py
index 48c9ed5..e70b2f8 100644
--- a/agent/fc_agent/media.py
+++ b/agent/fc_agent/media.py
@@ -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//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:
- log.warning("ffmpeg timed out after %.0fs: %s",
- timeout, url)
- return []
- except (OSError, ValueError):
- return []
+ if stopped:
+ return [], "stopped"
+ log.warning("ffmpeg timed out after %.0fs: %s",
+ timeout, url)
+ return [], f"ffmpeg timed out after {timeout:.0f}s"
+ 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",
- proc.returncode, url, _tail(errpath))
- return frames
+ reason = f"ffmpeg exit {proc.returncode}: {_tail(errpath)}"
+ log.warning("ffmpeg produced no frames for %s — %s", url, reason)
+ return [], reason
+ return frames, None
def _tail(path: str, limit: int = 300) -> str:
diff --git a/agent/fc_agent/throttle.py b/agent/fc_agent/throttle.py
new file mode 100644
index 0000000..6459105
--- /dev/null
+++ b/agent/fc_agent/throttle.py
@@ -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//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//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
diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py
index 2841845..4554703 100644
--- a/agent/fc_agent/worker.py
+++ b/agent/fc_agent/worker.py
@@ -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
- # unreachable, it's transient → let the loop back off + retry
- # (survives a redeploy). ConnectionError is caught as transient.
+ # → a job fault: fail it WITH ffmpeg's reason, so the job's stored
+ # error says e.g. "moov atom not found" instead of a bare
+ # "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).
diff --git a/alembic/versions/0072_gpu_job_triage_status.py b/alembic/versions/0072_gpu_job_triage_status.py
new file mode 100644
index 0000000..1dce875
--- /dev/null
+++ b/alembic/versions/0072_gpu_job_triage_status.py
@@ -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")
diff --git a/alembic/versions/0073_drop_tag_eval_run.py b/alembic/versions/0073_drop_tag_eval_run.py
new file mode 100644
index 0000000..4aedb38
--- /dev/null
+++ b/alembic/versions/0073_drop_tag_eval_run.py
@@ -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"])
diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py
index 8345f2d..f500acd 100644
--- a/backend/app/api/__init__.py
+++ b/backend/app/api/__init__.py
@@ -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,
diff --git a/backend/app/api/admin.py b/backend/app/api/admin.py
index 53781a9..4bbb12e 100644
--- a/backend/app/api/admin.py
+++ b/backend/app/api/admin.py
@@ -1,13 +1,12 @@
"""FC-3k: /api/admin — destructive admin actions.
-Five action surfaces:
+Action surfaces:
POST /api/admin/artists//cascade-delete (Tier C)
POST /api/admin/images/bulk-delete (Tier C)
DELETE /api/admin/tags/ (Tier B)
POST /api/admin/tags//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//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
diff --git a/backend/app/api/gpu.py b/backend/app/api/gpu.py
index 11605e3..bbbbecd 100644
--- a/backend/app/api/gpu.py
+++ b/backend/app/api/gpu.py
@@ -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
- response carries the number requeued for the UI toast."""
+ fixed pipeline. Stale tombstones are pruned FIRST (loop-era duplicates and
+ 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//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//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 ------------
diff --git a/backend/app/api/tag_eval.py b/backend/app/api/tag_eval.py
deleted file mode 100644
index 31fe10b..0000000
--- a/backend/app/api/tag_eval.py
+++ /dev/null
@@ -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("/", 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))
diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py
index 811b390..0600988 100644
--- a/backend/app/celery_app.py
+++ b/backend/app/celery_app.py
@@ -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
- "args": ("ccip",), # the queue keeps moving without the button
+ "schedule": 3600.0, # auto-feed NEW images; errored are
+ "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 +
- "args": ("siglip",), # retry failed embeds, no button needed
+ "schedule": 86400.0, # drain the concept-crop back-catalogue
+ "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,
diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py
index 8707467..5080b50 100644
--- a/backend/app/models/__init__.py
+++ b/backend/app/models/__init__.py
@@ -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",
diff --git a/backend/app/models/gpu_job.py b/backend/app/models/gpu_job.py
index 5e14e2d..dba5997 100644
--- a/backend/app/models/gpu_job.py
+++ b/backend/app/models/gpu_job.py
@@ -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()
)
diff --git a/backend/app/models/head_training_run.py b/backend/app/models/head_training_run.py
index 150357c..fd5858e 100644
--- a/backend/app/models/head_training_run.py
+++ b/backend/app/models/head_training_run.py
@@ -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
-live + historical status instead of holding it in transient frontend state.
+A persisted run row (not transient frontend state) so the run SURVIVES
+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
- # a SIGKILL'd one by this (mirrors TagEvalRun).
+ # Last time the task made progress — the recovery sweep tells a live run
+ # from a SIGKILL'd one by this.
last_progress_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
diff --git a/backend/app/models/tag_eval_run.py b/backend/app/models/tag_eval_run.py
deleted file mode 100644
index d0775ed..0000000
--- a/backend/app/models/tag_eval_run.py
+++ /dev/null
@@ -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,
- )
diff --git a/backend/app/services/cleanup_service.py b/backend/app/services/cleanup_service.py
index 3916909..5e36898 100644
--- a/backend/app/services/cleanup_service.py
+++ b/backend/app/services/cleanup_service.py
@@ -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.
diff --git a/backend/app/services/ml/gpu_jobs.py b/backend/app/services/ml/gpu_jobs.py
index f41086b..5c20892 100644
--- a/backend/app/services/ml/gpu_jobs.py
+++ b/backend/app/services/ml/gpu_jobs.py
@@ -12,8 +12,9 @@ and the lease itself reclaims expired leases as a final backstop. Result-writing
from datetime import UTC, datetime, timedelta
-from sqlalchemy import and_, select, update
+from sqlalchemy import and_, delete, exists, func, or_, select, update
from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import aliased
from ...models import GpuJob
@@ -24,6 +25,107 @@ DEFAULT_LEASE_TTL = 180 # seconds an agent holds a job before it can be re-l
DEFAULT_BATCH = 8
MAX_ATTEMPTS = 3
+# Poison-loop backstops. `attempts` counts LEASES GRANTED (incremented in
+# lease()), but fail()'s MAX_ATTEMPTS cap only fires when the agent reports a
+# failure — a job that keeps coming back via release() (transient handback) or
+# lease expiry (agent crash/wedge) never gets a verdict and would cycle forever.
+# The orphan sweep converts those to 'error': an expired lease that has already
+# been granted EXPIRED_POISON_CAP leases is presumed to kill/wedge the agent,
+# and a pending job granted PENDING_POISON_CAP leases without ever completing is
+# presumed poisoned (e.g. a transfer that stalls every time). Both stay
+# resurrectable via /retry_errors, which resets attempts.
+EXPIRED_POISON_CAP = MAX_ATTEMPTS + 2
+PENDING_POISON_CAP = 10
+
+
+def error_dedupe_statements():
+ """DELETEs enforcing: at most ONE error row per (image, task), and none that
+ a live or succeeded row makes moot. The 2026-07-02 tombstone loop (backfill
+ skip-lists lacked 'error') minted a duplicate error row per bad file per
+ hour; running these before every backfill and inside /retry_errors keeps the
+ error count reading as "distinct failing files" and stops a retry fanning
+ one file out into several duplicate pending jobs. Shared by the sync beat
+ task and the async API route so both prune by the SAME predicate.
+ Execution order matters: moot rows first, then older duplicates (the newest
+ error — the freshest reason — survives)."""
+ other = aliased(GpuJob)
+ same_pair = and_(
+ other.image_record_id == GpuJob.image_record_id,
+ other.task == GpuJob.task,
+ )
+ moot = (
+ delete(GpuJob)
+ .where(
+ GpuJob.status == "error",
+ exists().where(
+ same_pair, other.status.in_(["pending", "leased", "done"]),
+ ),
+ )
+ .execution_options(synchronize_session=False)
+ )
+ older_dupe = (
+ delete(GpuJob)
+ .where(
+ GpuJob.status == "error",
+ exists().where(
+ same_pair,
+ other.status == "error",
+ or_(
+ other.updated_at > GpuJob.updated_at,
+ and_(other.updated_at == GpuJob.updated_at,
+ other.id > GpuJob.id),
+ ),
+ ),
+ )
+ .execution_options(synchronize_session=False)
+ )
+ return [moot, older_dupe]
+
+
+def recover_statements(now: datetime) -> dict:
+ """UPDATEs for the orphan sweep, keyed by outcome; insertion order IS the
+ required execution order ('recovered' must run after 'poison_expired', which
+ claims the crash-loopers out of the same expired-lease pool)."""
+ expired = and_(GpuJob.status == "leased", GpuJob.lease_expires_at < now)
+ unlease = {"lease_token": None, "leased_at": None, "lease_expires_at": None,
+ "updated_at": now}
+ return {
+ "poison_expired": (
+ update(GpuJob)
+ .where(expired, GpuJob.attempts >= EXPIRED_POISON_CAP)
+ .values(
+ status="error",
+ # Keep the job's last stored failure reason — it's the triage
+ # signal for WHY the loop happened.
+ error=func.concat(
+ f"poisoned: lease expired after {EXPIRED_POISON_CAP}+ lease "
+ "attempts (job repeatedly crashes or wedges the agent?); "
+ "last error: ",
+ func.coalesce(GpuJob.error, "none"),
+ ),
+ **unlease,
+ )
+ ),
+ "recovered": update(GpuJob).where(expired).values(
+ status="pending", **unlease,
+ ),
+ "poison_pending": (
+ update(GpuJob)
+ .where(GpuJob.status == "pending",
+ GpuJob.attempts >= PENDING_POISON_CAP)
+ .values(
+ status="error",
+ error=func.concat(
+ f"poisoned: {PENDING_POISON_CAP}+ lease attempts without "
+ "ever completing (transfer stalls every time?); "
+ "last error: ",
+ func.coalesce(GpuJob.error, "none"),
+ ),
+ updated_at=now,
+ )
+ ),
+ }
+
class GpuJobService:
def __init__(self, session: AsyncSession):
@@ -170,16 +272,11 @@ class GpuJobService:
async def recover_orphaned(self) -> int:
"""Reset every expired lease back to pending — catches agents that died
- mid-job (no graceful release). Run on a short beat so the queue recovers
- + reads honestly even when no worker is actively leasing. Returns rows
- recovered."""
- now = datetime.now(UTC)
- res = await self.session.execute(
- update(GpuJob)
- .where(GpuJob.status == "leased", GpuJob.lease_expires_at < now)
- .values(
- status="pending", lease_token=None, leased_at=None,
- lease_expires_at=None, updated_at=now,
- )
- )
- return res.rowcount or 0
+ mid-job (no graceful release) — and convert poison-loopers to 'error'
+ (see the *_POISON_CAP rationale above). Run on a short beat so the queue
+ recovers + reads honestly even when no worker is actively leasing.
+ Returns rows recovered to pending (poison conversions are extra)."""
+ counts = {}
+ for name, stmt in recover_statements(datetime.now(UTC)).items():
+ counts[name] = (await self.session.execute(stmt)).rowcount or 0
+ return counts["recovered"]
diff --git a/backend/app/services/ml/gpu_triage.py b/backend/app/services/ml/gpu_triage.py
new file mode 100644
index 0000000..dddd095
--- /dev/null
+++ b/backend/app/services/ml/gpu_triage.py
@@ -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}
diff --git a/backend/app/services/ml/heads.py b/backend/app/services/ml/heads.py
index b234897..2f47401 100644
--- a/backend/app/services/ml/heads.py
+++ b/backend/app/services/ml/heads.py
@@ -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
- loaders + metric helpers so production heads match the eval's measured numbers.
+ VALIDATED scores, and upsert a TagHead row. Uses the eval-proven data loaders
+ + 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,
diff --git a/backend/app/services/ml/tag_eval.py b/backend/app/services/ml/tag_eval.py
deleted file mode 100644
index abff374..0000000
--- a/backend/app/services/ml/tag_eval.py
+++ /dev/null
@@ -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
diff --git a/backend/app/services/ml/training_data.py b/backend/app/services/ml/training_data.py
new file mode 100644
index 0000000..d01acf7
--- /dev/null
+++ b/backend/app/services/ml/training_data.py
@@ -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),
+ }
diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py
index e9e334e..e403cf5 100644
--- a/backend/app/tasks/maintenance.py
+++ b/backend/app/tasks/maintenance.py
@@ -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
diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py
index 3501c64..737c36c 100644
--- a/backend/app/tasks/ml.py
+++ b/backend/app/tasks/ml.py
@@ -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 —
- without re-touching their figure/CCIP regions."""
+ before concept crops existed — 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
- get picked back up quickly + the queue counts read honestly. Returns the
- number recovered."""
+ agent that died mid-job (no graceful release) — and convert poison-loopers
+ (release/expiry cycles that never reach fail()'s attempt cap) to 'error'.
+ 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 ..models import GpuJob
+ from ..services.ml.gpu_jobs import recover_statements
SessionLocal = _sync_session_factory()
with SessionLocal() as session:
- now = datetime.now(UTC)
- res = session.execute(
- update(GpuJob)
- .where(GpuJob.status == "leased", GpuJob.lease_expires_at < now)
- .values(
- status="pending", lease_token=None, leased_at=None,
- lease_expires_at=None, updated_at=now,
- )
- )
+ counts = {
+ name: session.execute(stmt).rowcount or 0
+ for name, stmt in recover_statements(datetime.now(UTC)).items()
+ }
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")
diff --git a/frontend/src/components/common/MaintenanceTile.vue b/frontend/src/components/common/MaintenanceTile.vue
index fd0c54b..ad63e45 100644
--- a/frontend/src/components/common/MaintenanceTile.vue
+++ b/frontend/src/components/common/MaintenanceTile.vue
@@ -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
- real