Merge pull request 'Settings defaults, GPU error tombstones + failure triage/recovery, agent bandwidth cap, approved retirements' (#186) from dev into main
Build images / sign-extension (push) Successful in 3s
CI / lint (push) Successful in 4s
Build images / build-agent (push) Successful in 9s
Build images / build-ml (push) Successful in 10s
Build images / build-web (push) Successful in 11s
CI / frontend-build (push) Successful in 20s
CI / backend-lint-and-test (push) Successful in 38s
CI / integration (push) Successful in 3m33s

This commit was merged in pull request #186.
This commit is contained in:
2026-07-02 15:48:48 -04:00
43 changed files with 1628 additions and 1382 deletions
+4
View File
@@ -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}
+22 -2
View File
@@ -21,7 +21,7 @@ log = logging.getLogger("fc_agent.app")
# Bump on every agent change. The page embeds this and /status reports it; the UI
# warns to reload when they differ — so a stale browser-cached page can't be
# mistaken for "the new image didn't deploy". (Belt-and-braces with no-store.)
VERSION = "2026-07-02.1 · short videos sample again (select, not fps) + ffmpeg errors logged"
VERSION = "2026-07-02.4 · bandwidth governor: aggregate download cap (MB/s dial) so the agent can't saturate the desktop's network"
logbuf.install()
cfg = Config.from_env()
@@ -83,6 +83,13 @@ async def auto(request: Request):
return JSONResponse(worker.status())
@app.post("/bandwidth")
async def bandwidth(request: Request):
body = await request.json()
worker.set_bandwidth(float(body.get("value", 0)))
return JSONResponse(worker.status())
@app.get("/gpu")
def gpu():
# GPU meters poll this on their own fast cadence. It only reads local
@@ -161,8 +168,9 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
.step{background:#262c34;color:var(--fg);border:1px solid var(--bd);border-radius:8px;
width:30px;height:32px;font:700 16px system-ui;cursor:pointer}
.step:hover{border-color:var(--acc)}
#conc{width:3.4rem;height:32px;text-align:center;font:700 16px system-ui;background:#11151a;
#conc,#bw{width:3.4rem;height:32px;text-align:center;font:700 16px system-ui;background:#11151a;
color:var(--fg);border:1px solid var(--bd);border-radius:8px}
.unit{color:var(--mut);font-size:12px;font-weight:600}
.hint{color:var(--mut);font-size:12px;margin-top:12px}
.tiles{display:grid;grid-template-columns:repeat(6,1fr);gap:8px;margin-bottom:16px}
.tile{background:#13171d;border:1px solid var(--bd);border-radius:10px;padding:12px 8px;text-align:center}
@@ -216,6 +224,10 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
<input id=conc type=number min=1 value=1 onchange="setv(this.value)">
<button class=step onclick=setc(1)>+</button>
</div>
<div class=stepper title="aggregate download cap, downloads + video streams combined — 0 = unlimited">
<input id=bw type=number min=0 step=1 value=8 onchange="setbw(this.value)">
<span class=unit>MB/s</span>
</div>
</div>
<div class=hint id=conchint>auto-tuning downloaders to keep the GPU fed · max 8</div>
</section>
@@ -281,6 +293,11 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
await fetch('/auto',{method:'POST',headers:{'Content-Type':'application/json'},
body:JSON.stringify({value:on})});refresh()
}
async function setbw(v){
v=Math.max(0,parseFloat(v)||0); bw.value=v
await fetch('/bandwidth',{method:'POST',headers:{'Content-Type':'application/json'},
body:JSON.stringify({value:v})});refresh()
}
async function refresh(){
let s; try{ s=await (await fetch('/status')).json() }catch{ return }
applyStatus(s)
@@ -318,6 +335,9 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
// Instantaneous pool state → demoted to the sub-line, where its jumpiness reads
// as live churn rather than a "broken" headline metric.
pipe.textContent='downloaders '+(s.downloaders!=null?s.downloaders:'')+' · consumers '+(s.consumers!=null?s.consumers:'')+' · on GPU '+(s.active||0)
+' · net '+(s.net_mb_s!=null?s.net_mb_s.toFixed(1):'')+' MB/s'
+(s.bandwidth_limit_mb_s>0?(' / cap '+s.bandwidth_limit_mb_s):'')
if(document.activeElement!==bw && s.bandwidth_limit_mb_s!=null) bw.value=s.bandwidth_limit_mb_s
// Buffer occupancy bar (also driven here so it tracks the /status cadence).
if(s.buffer!=null && s.buffer_max){ const p=Math.round(100*s.buffer/s.buffer_max)
buflbl.textContent=s.buffer+' / '+s.buffer_max; bufbar.style.width=p+'%' }
+15 -4
View File
@@ -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,
+10
View File
@@ -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")),
)
+66 -11
View File
@@ -4,12 +4,15 @@ instances, each with a timestamp."""
import io
import logging
import os
import signal
import subprocess
import tempfile
import time
from PIL import Image, ImageFile
from .throttle import PidReadMeter
log = logging.getLogger("fc_agent.media")
# Load slightly-truncated images (a few missing trailing bytes) instead of
@@ -111,6 +114,12 @@ def _collect_frames(
def _terminate(proc: subprocess.Popen) -> None:
"""Stop an ffmpeg cleanly, then hard-kill if it ignores SIGTERM."""
try:
# A bandwidth-paused (SIGSTOPped) process can't receive SIGTERM until it
# resumes — always CONT first so termination is prompt, not queued.
proc.send_signal(signal.SIGCONT)
except OSError:
pass
proc.terminate()
try:
proc.wait(timeout=2)
@@ -122,10 +131,34 @@ def _terminate(proc: subprocess.Popen) -> None:
pass
def _pause(proc: subprocess.Popen, seconds: float, should_stop) -> bool:
"""SIGSTOP ffmpeg for ~`seconds` of bandwidth debt, staying responsive to
Stop. While paused, the kernel socket buffer fills and TCP flow control
stalls curator's send side — that's the throttle. SIGCONT is ALWAYS sent
before returning. False = a Stop arrived mid-pause."""
try:
proc.send_signal(signal.SIGSTOP)
except OSError:
return True # already exited — nothing to pause
try:
end = time.monotonic() + seconds
while (left := end - time.monotonic()) > 0:
if should_stop and should_stop():
return False
time.sleep(min(0.5, left))
return True
finally:
try:
proc.send_signal(signal.SIGCONT)
except OSError:
pass
def sample_frames_from_url(
url: str, interval_seconds: float, max_frames: int,
*, headers: str = "", timeout: float = 1200.0, should_stop=None,
) -> list[tuple[float, Image.Image]]:
governor=None,
) -> tuple[list[tuple[float, Image.Image]], str | None]:
"""Sample frames by pointing ffmpeg STRAIGHT at the media URL — it Range-reads
only the video index + up to max_frames worth of content, so the agent never
downloads the whole file (VR/4K originals run 800MB+ and would buffer ~1GB in
@@ -133,7 +166,15 @@ def sample_frames_from_url(
the timeout is the per-video ceiling (a slow/reconnecting stream can otherwise
run for minutes). `should_stop` is polled while ffmpeg runs so a Stop KILLS the
subprocess at once — otherwise a downloader stuck in a long decode keeps the
agent "working" long after Stop. Empty on failure / stop / timeout."""
agent "working" long after Stop. `governor` (the worker's shared TokenBucket)
meters ffmpeg's network reads from outside via /proc/<pid>/io and SIGSTOPs
the process into budget, so video streaming honors the same aggregate
bandwidth cap as still downloads.
Returns (frames, reason): frames is empty on failure/stop/timeout, and
`reason` then carries the SPECIFIC cause (ffmpeg's stderr tail / timeout) so
the caller can put it in the job's error — a bare "no frames" hid a filter
bug as "unprocessable" for weeks. None reason on success."""
interval = max(0.5, float(interval_seconds or 4.0))
cap = max(1, int(max_frames or 64))
hdr = ["-headers", headers] if headers else []
@@ -163,6 +204,7 @@ def sample_frames_from_url(
cmd, stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL, stderr=errf,
)
meter = PidReadMeter(proc.pid) if governor is not None else None
# Poll rather than block, so a Stop (or the per-video timeout) can
# kill a slow/wedged ffmpeg promptly instead of waiting it out.
start = time.monotonic()
@@ -174,17 +216,30 @@ def sample_frames_from_url(
stopped = should_stop and should_stop()
if stopped or (time.monotonic() - start > timeout):
_terminate(proc)
if not stopped:
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:
+111
View File
@@ -0,0 +1,111 @@
"""Global download-bandwidth governor (one token bucket for the whole agent).
The agent lives on someone's desktop and shares that desktop's network —
typically WiFi, where saturating the link doesn't just slow other apps: it
bufferbloats the airtime (RTT 21→45ms) and collapses EVERY connection,
the operator's browser included. Measured 2026-07-02: the idle link moved
~38 MB/s single-stream, but under the 8-downloader sweep every stream on the
machine crawled at ~1-1.5 MB/s. So the cap is on the AGGREGATE, not per
stream: still downloads pump their chunks through take(), and ffmpeg video
streams — whose sockets live in a subprocess we can't wrap — are metered from
outside via /proc/<pid>/io and paused (SIGSTOP) into budget using charge()'s
debt signal; TCP flow control then stalls the sender while ffmpeg sleeps.
Accounting is post-paid (charge the bytes first, then wait out any debt): the
bytes have already crossed the network by the time we count them, and it means
a chunk larger than one second of budget can never deadlock the bucket.
Stdlib-only on purpose — unit-tested in CI, where the agent's ML deps
don't exist.
"""
import threading
import time
class TokenBucket:
"""Thread-safe token bucket in bytes/second. rate 0 = unlimited.
`consumed` is the monotonic total of bytes charged (throttled or not) —
the worker's rate loop derives the UI's "net MB/s" readout from it.
"""
def __init__(self, rate_bytes_per_s: float = 0.0):
self._cond = threading.Condition()
self._rate = max(0.0, float(rate_bytes_per_s))
# Burst = one second of budget: enough that chunked reads stay smooth,
# small enough that a burst can't meaningfully lift the average.
self._level = self._rate
self._stamp = time.monotonic()
self.consumed = 0
@property
def rate(self) -> float:
return self._rate
def set_rate(self, rate_bytes_per_s: float) -> None:
"""Retune live (the UI dial). Waiters re-check immediately, so raising
the cap (or lifting it with 0) unblocks a mid-download wait at once."""
with self._cond:
self._refill_locked() # settle elapsed time at the OLD rate
self._rate = max(0.0, float(rate_bytes_per_s))
self._level = min(self._level, self._rate)
self._cond.notify_all()
def _refill_locked(self) -> None:
now = time.monotonic()
self._level = min(self._rate, self._level + (now - self._stamp) * self._rate)
self._stamp = now
def take(self, n: int) -> None:
"""Charge n bytes and block until the budget recovers (stills path)."""
with self._cond:
self.consumed += n
if self._rate <= 0:
return
self._refill_locked()
self._level -= n
while self._level < 0:
# Wake early on set_rate; cap the wait so a big debt is paid in
# re-checked slices rather than one long uninterruptible sleep.
self._cond.wait(min(-self._level / self._rate, 0.5))
if self._rate <= 0:
return
self._refill_locked()
def charge(self, n: int) -> float:
"""Charge n bytes WITHOUT blocking; return seconds of debt (0 = within
budget). The ffmpeg governor can't block the subprocess's own reads, so
it SIGSTOPs the process for (about) the returned debt instead."""
with self._cond:
self.consumed += n
if self._rate <= 0:
return 0.0
self._refill_locked()
self._level -= n
return max(0.0, -self._level / self._rate)
class PidReadMeter:
"""Cumulative read-bytes meter for a subprocess, via /proc/<pid>/io.
`rchar` counts every read() syscall's bytes — for a streaming ffmpeg the
network reads dominate, so the delta is a good-enough aggregate-bandwidth
signal (it's a governor, not a billing meter). Returns None when /proc is
unavailable (process exited, or a non-Linux host): the caller then simply
doesn't govern — degrade to unthrottled rather than break video sampling.
"""
def __init__(self, pid: int):
self._path = f"/proc/{pid}/io"
self._last = 0
def delta(self) -> int | None:
try:
with open(self._path, "rb") as f:
for line in f:
if line.startswith(b"rchar:"):
total = int(line.split()[1])
d, self._last = total - self._last, total
return max(0, d)
except (OSError, ValueError):
return None
return None
+85 -6
View File
@@ -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).
@@ -0,0 +1,32 @@
"""gpu_job.triage_status — the probe's verdict on an errored job's FILE
Failure triage (#125): a periodic sweep probes each errored image's file
(sha256 + decode, verify_integrity's machinery) exactly once and stores the
verdict here — 'defect' (the file is bad: recovery material, excluded from
/retry_errors) or 'file_ok' (failure was operational, safe to retry). NULL
means not yet probed; selecting on NULL is what makes the sweep resumable.
No index: the errored slice the sweep scans is tiny by design (tombstones).
Revision ID: 0072
Revises: 0071
Create Date: 2026-07-02
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
revision: str = "0072"
down_revision: Union[str, None] = "0071"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.add_column(
"gpu_job", sa.Column("triage_status", sa.String(16), nullable=True)
)
def downgrade() -> None:
op.drop_column("gpu_job", "triage_status")
@@ -0,0 +1,46 @@
"""drop tag_eval_run — the head-vs-centroid eval harness is retired
The eval (#1130) existed to prove the heads tagging spine on the operator's own
data. It did; the operator accepted the system and retired the harness
(2026-07-02) — card, API, task, model and this table all go. The eval's data
loaders + metric helpers live on in services/ml/training_data.py, where the
production heads trainer uses them nightly.
Revision ID: 0073
Revises: 0072
Create Date: 2026-07-02
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
revision: str = "0073"
down_revision: Union[str, None] = "0072"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.drop_index("ix_tag_eval_run_status", table_name="tag_eval_run")
op.drop_table("tag_eval_run")
def downgrade() -> None:
# Recreates the shape from 0056 (data is not restorable).
op.create_table(
"tag_eval_run",
sa.Column("id", sa.Integer(), primary_key=True),
sa.Column("params", postgresql.JSONB(), nullable=False),
sa.Column("status", sa.String(length=16), nullable=False,
server_default="running"),
sa.Column("started_at", sa.DateTime(timezone=True), nullable=False,
server_default=sa.func.now()),
sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("report", postgresql.JSONB(), nullable=True),
sa.Column("error", sa.Text(), nullable=True),
sa.Column("last_progress_at", sa.DateTime(timezone=True),
nullable=True),
)
op.create_index("ix_tag_eval_run_status", "tag_eval_run", ["status"])
-2
View File
@@ -38,7 +38,6 @@ def all_blueprints() -> list[Blueprint]:
from .suggestions import suggestions_bp
from .system_activity import system_activity_bp
from .system_backup import system_backup_bp
from .tag_eval import tag_eval_bp
from .tags import tags_bp
from .thumbnails import thumbnails_bp
return [
@@ -58,7 +57,6 @@ def all_blueprints() -> list[Blueprint]:
import_admin_bp,
suggestions_bp,
aliases_bp,
tag_eval_bp,
heads_bp,
gpu_bp,
ccip_bp,
+1 -15
View File
@@ -1,13 +1,12 @@
"""FC-3k: /api/admin — destructive admin actions.
Five action surfaces:
Action surfaces:
POST /api/admin/artists/<slug>/cascade-delete (Tier C)
POST /api/admin/images/bulk-delete (Tier C)
DELETE /api/admin/tags/<int:tag_id> (Tier B)
POST /api/admin/tags/<int:dest_id>/merge (Tier B)
POST /api/admin/tags/prune-unused (Tier A)
POST /api/admin/posts/prune-bare (Tier A)
POST /api/admin/tags/purge-legacy (Tier A)
GET /api/admin/tags/<int:tag_id>/usage-count (helper)
Tier-C ops take a dry_run body flag (returns projection inline,
@@ -277,19 +276,6 @@ async def posts_reconcile_duplicates():
return await _run_dry_run_op(reconcile_duplicate_posts, source_id=source_id)
@admin_bp.route("/tags/purge-legacy", methods=["POST"])
async def tags_purge_legacy():
"""Tier-A: delete legacy IR-migration tags — archive/post/artist
kinds (e.g. `BlenderKnight:Hannah_BJ_Loops`) PLUS general tags with
a legacy name prefix (`source:*`, from IR's source kind that fell
back to general). dry-run preview returns per-kind + per-prefix
counts + a sample so the UI shows exactly what'll go before the
operator confirms with dry_run=false."""
from ..services.cleanup_service import purge_legacy_tags
return await _run_dry_run_op(purge_legacy_tags)
@admin_bp.route("/tags/reset-content", methods=["POST"])
async def tags_reset_content():
"""Tier-A: delete ALL general + character tags (the Camie-suggestable
+119 -7
View File
@@ -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/<id>/recover instead.
or_(GpuJob.triage_status.is_(None),
GpuJob.triage_status != "defect"),
)
.values(
status="pending", attempts=0, error=None, lease_token=None,
leased_at=None, lease_expires_at=None, updated_at=func.now(),
leased_at=None, lease_expires_at=None, triage_status=None,
updated_at=func.now(),
)
)
kept = (
await session.execute(
select(func.count()).select_from(GpuJob)
.where(GpuJob.status == "error")
)
).scalar_one()
await session.commit()
return jsonify({"requeued": res.rowcount or 0})
return jsonify({
"requeued": res.rowcount or 0, "pruned": pruned, "defects_kept": kept,
})
# --- Failure triage + recovery (#125) ------------------------------------
@gpu_bp.route("/errors", methods=["GET"])
async def errors():
"""The triage view of the error tombstones: every errored job joined with
its image's integrity verdict, bucketed by reason for the overview. The
probe sweep (triage_gpu_errors, 15-min beat) fills triage_status; 'defect'
rows are the recovery surface's list."""
async with get_session() as session:
rows = (
await session.execute(
select(
GpuJob.id, GpuJob.image_record_id, GpuJob.task,
GpuJob.error, GpuJob.triage_status, GpuJob.updated_at,
ImageRecord.integrity_status, ImageRecord.mime,
ImageRecord.path, ImageRecord.thumbnail_path,
)
.join(ImageRecord, ImageRecord.id == GpuJob.image_record_id)
.where(GpuJob.status == "error")
.order_by(GpuJob.updated_at.desc())
.limit(500)
)
).all()
total = (
await session.execute(
select(func.count()).select_from(GpuJob)
.where(GpuJob.status == "error")
)
).scalar_one()
by_class: dict[str, int] = {}
triage = {"defect": 0, "file_ok": 0, "unclassified": 0}
items = []
for r in rows:
cls = classify_reason(r.error)
by_class[cls] = by_class.get(cls, 0) + 1
bucket = r.triage_status or "unclassified"
triage[bucket] = triage.get(bucket, 0) + 1
items.append({
"job_id": r.id,
"image_id": r.image_record_id,
"task": r.task,
"error": r.error,
"reason_class": cls,
"triage_status": r.triage_status,
"integrity_status": r.integrity_status,
"mime": r.mime,
"image_url": image_url(r.path),
"thumbnail_url": (
image_url(r.thumbnail_path) if r.thumbnail_path else None
),
"updated_at": r.updated_at.isoformat() if r.updated_at else None,
})
return jsonify({
"total": total, "by_class": by_class, "triage": triage, "items": items,
})
@gpu_bp.route("/errors/triage", methods=["POST"])
async def errors_triage():
"""Run the probe sweep NOW (the card's button) instead of waiting out the
15-minute beat cadence."""
from ..tasks.maintenance import triage_gpu_errors
r = triage_gpu_errors.delay()
return jsonify({"celery_task_id": r.id}), 202
@gpu_bp.route("/errors/<int:image_id>/recover", methods=["POST"])
async def errors_recover(image_id: int):
"""Recover a defect-triaged original: delete the bad copy + record and
re-poll its subscription Source (a fresh fetch re-imports the file, which
re-enters the GPU pipeline). Returns status 'no_source' when nothing
pollable resolves — the file needs manual replacement there."""
async with get_session() as session:
result = await session.run_sync(
lambda s: recover_defective_image(
s, image_id, images_root=_IMAGES_ROOT,
)
)
return jsonify(result)
# --- Agent (bearer token): lease / submit / heartbeat / fail ------------
-70
View File
@@ -1,70 +0,0 @@
"""Tag-eval API (#1130): trigger + revisit the head-vs-centroid eval.
The run + full report live in the tag_eval_run row, so the admin card rehydrates
from GET (history / detail) on mount — the report survives navigation rather than
living in transient frontend state.
"""
from quart import Blueprint, jsonify, request
from sqlalchemy import select
from ..extensions import get_session
from ..models import TagEvalRun
from ..services.ml.tag_eval import EvalAlreadyRunning, start_tag_eval_run
tag_eval_bp = Blueprint("tag_eval", __name__, url_prefix="/api/tag-eval")
def _serialize(run: TagEvalRun, *, include_report: bool) -> dict:
out = {
"id": run.id,
"params": run.params,
"status": run.status,
"started_at": run.started_at.isoformat() if run.started_at else None,
"finished_at": run.finished_at.isoformat() if run.finished_at else None,
"error": run.error,
}
if include_report:
out["report"] = run.report
return out
@tag_eval_bp.route("", methods=["POST"])
async def create():
body = await request.get_json(silent=True) or {}
params = body.get("params") or body or {}
async with get_session() as session:
try:
run_id = await session.run_sync(
lambda s: start_tag_eval_run(s, params)
)
except EvalAlreadyRunning as running:
return jsonify({
"error": "eval_already_running",
"running_id": int(running.args[0]),
}), 409
await session.commit()
return jsonify({"run_id": run_id, "status": "running"}), 202
@tag_eval_bp.route("", methods=["GET"])
async def history():
try:
limit = min(int(request.args.get("limit", "20")), 100)
except ValueError:
return jsonify({"error": "invalid_limit"}), 400
async with get_session() as session:
rows = (await session.execute(
select(TagEvalRun).order_by(TagEvalRun.id.desc()).limit(limit)
)).scalars().all()
# List is light — no full report (the detail endpoint carries it).
return jsonify({"runs": [_serialize(r, include_report=False) for r in rows]})
@tag_eval_bp.route("/<int:run_id>", methods=["GET"])
async def detail(run_id: int):
async with get_session() as session:
run = await session.get(TagEvalRun, run_id)
if run is None:
return jsonify({"error": "not_found"}), 404
return jsonify(_serialize(run, include_report=True))
+8 -12
View File
@@ -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,
-2
View File
@@ -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",
+5
View File
@@ -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()
)
+4 -4
View File
@@ -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
)
-45
View File
@@ -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,
)
-65
View File
@@ -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.
+111 -14
View File
@@ -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"]
+156
View File
@@ -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}
+5 -4
View File
@@ -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,
-430
View File
@@ -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
+121
View File
@@ -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),
}
+22 -44
View File
@@ -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
+45 -70
View File
@@ -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")
@@ -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 <button> with aria-expanded + focus ring.
(e.g. keep a running task's tile expanded). Manual expand/collapse persists per
tile in localStorage, so the page reloads the way the operator left it.
Keyboard accessible: the header is a real <button> with aria-expanded + focus
ring.
-->
<template>
<v-card class="fc-tile" :class="{ 'fc-tile--open': isOpen }">
@@ -53,12 +55,21 @@ const props = defineProps({
open: { type: Boolean, default: false },
})
const local = ref(props.open)
watch(() => props.open, (v) => { local.value = v })
// Only MANUAL toggles are saved (keyed by tile title): a forced `open` while a
// task is mid-run is transient state, not a preference — persisting it would
// resurrect the "several tiles open by default" bug this replaces. When the
// force clears, the tile falls back to the operator's saved choice.
const storeKey = `fc.tile.${props.title}`
function savedOpen() {
try { return localStorage.getItem(storeKey) === '1' } catch { return false }
}
const local = ref(props.open || savedOpen())
watch(() => props.open, (v) => { local.value = v || savedOpen() })
const isOpen = computed(() => local.value)
function toggle() {
local.value = !local.value
try { localStorage.setItem(storeKey, local.value ? '1' : '0') } catch { /* non-fatal */ }
}
</script>
@@ -132,8 +132,8 @@ const hasMenu = computed(() =>
color: rgb(var(--v-theme-on-surface-variant, var(--v-theme-on-surface)));
font-family: 'JetBrains Mono', monospace;
}
/* Green ✓ / red ✗ verdict pair — same circular language as the eval card
(TagEvalCard .fc-act) so accept/reject read identically across surfaces. */
/* Green ✓ / red ✗ verdict pair — circular buttons so accept/reject read
identically across surfaces. */
.fc-suggestion__acts {
flex: 0 0 auto; display: flex; gap: 4px;
}
@@ -3,7 +3,6 @@
icon="mdi-expansion-card"
title="GPU agent (CCIP + crops)"
blurb="Connect a desktop-GPU agent to embed characters (CCIP) and crops. It pulls work over HTTP — your database and Redis stay private."
:open="true"
>
<p class="fc-muted text-body-2 mb-3">
The agent is a container you run on the machine with the GPU. It
@@ -338,8 +337,12 @@ async function onBackfillSiglip() {
async function onRetryErrors() {
retrying.value = true
try {
const { requeued } = await store.retryErrors()
toast({ text: `Requeued ${requeued} errored job${requeued === 1 ? '' : 's'} — run the agent to process them`, type: 'success' })
const { requeued, pruned, defects_kept: kept } = await store.retryErrors()
const extras = []
if (pruned) extras.push(`${pruned} stale duplicate${pruned === 1 ? '' : 's'} pruned`)
if (kept) extras.push(`${kept} known-bad file${kept === 1 ? '' : 's'} kept for recovery`)
const extra = extras.length ? ` (${extras.join(', ')})` : ''
toast({ text: `Requeued ${requeued} errored job${requeued === 1 ? '' : 's'}${extra} — run the agent to process them`, type: 'success' })
await refreshQueue()
} catch (e) {
toast({ text: `Could not retry errored jobs: ${e.message}`, type: 'error' })
@@ -0,0 +1,189 @@
<template>
<MaintenanceTile
icon="mdi-file-alert"
title="Failed processing"
blurb="Triage originals that failed GPU processing — probe the files, flag defects, recover them."
>
<p class="fc-muted text-body-2 mb-3">
A job that keeps failing parks as an error with its reason. A background
probe then checks the FILE itself (checksum + decode) and splits the
errors: <b>defective files</b> (truncated/corrupt originals listed below
for recovery) vs <b>file&nbsp;OK</b> (the failure was operational; requeue
those with <i>Retry errored jobs</i> on the GPU agent card).
</p>
<div v-if="loading" class="fc-muted text-body-2">Loading</div>
<template v-else-if="overview">
<div class="fc-queue mb-3">
<div class="fc-q"><div class="fc-q__n">{{ overview.total }}</div><div class="fc-q__l">errored</div></div>
<div class="fc-q"><div class="fc-q__n" :class="overview.triage.defect ? 'fc-weak' : ''">{{ overview.triage.defect }}</div><div class="fc-q__l">defective</div></div>
<div class="fc-q"><div class="fc-q__n fc-good">{{ overview.triage.file_ok }}</div><div class="fc-q__l">file ok</div></div>
<div class="fc-q"><div class="fc-q__n">{{ overview.triage.unclassified }}</div><div class="fc-q__l">unprobed</div></div>
</div>
<p v-if="classSummary" class="fc-muted text-caption mb-3">
Reasons: {{ classSummary }}
</p>
<div class="d-flex mb-4" style="gap:8px">
<v-btn
size="small" color="accent" variant="tonal" rounded="pill"
prepend-icon="mdi-magnify-scan" :loading="probing"
:disabled="!overview.triage.unclassified" @click="onProbe"
>Probe unclassified now</v-btn>
<v-btn
size="small" variant="text" rounded="pill"
prepend-icon="mdi-refresh" @click="refresh"
>Refresh</v-btn>
</div>
<template v-if="defects.length">
<div class="fc-section-h mb-2">Defective originals</div>
<div v-for="it in defects" :key="it.job_id" class="fc-defect mb-2">
<a :href="it.image_url" target="_blank" rel="noopener" class="fc-defect__thumb">
<img v-if="it.thumbnail_url" :src="it.thumbnail_url" alt="">
<v-icon v-else icon="mdi-file-question" size="28" />
</a>
<div class="fc-defect__meta">
<div class="text-body-2">
image <b>{{ it.image_id }}</b> · {{ it.task }} ·
<span class="fc-weak">{{ it.integrity_status }}</span>
</div>
<div class="fc-muted text-caption fc-defect__err" :title="it.error || ''">
{{ it.error || 'no stored reason' }}
</div>
</div>
<v-btn
v-if="recovered[it.image_id] !== 'no_source'"
size="small" color="accent" variant="tonal" rounded="pill"
prepend-icon="mdi-cloud-download" :loading="recovering === it.image_id"
@click="onRecover(it)"
>Recover</v-btn>
<span v-else class="fc-muted text-caption">
no pollable source replace the file manually
</span>
</div>
<p class="fc-muted text-caption mt-2 mb-0">
Recover deletes the bad copy (and its record) and re-checks its
subscription source, so a fresh download re-imports it and re-enters
processing. Files without a pollable source need manual replacement.
</p>
</template>
<p v-else-if="!overview.total" class="fc-muted text-body-2 mb-0">
No failed jobs the pipeline is clean.
</p>
<p v-else-if="!overview.triage.unclassified" class="fc-muted text-body-2 mb-0">
No defective files every probed failure was operational
(file&nbsp;OK). Requeue them from the GPU agent card.
</p>
</template>
</MaintenanceTile>
</template>
<script setup>
import { computed, onMounted, ref } from 'vue'
import { toast } from '../../utils/toast.js'
import MaintenanceTile from '../common/MaintenanceTile.vue'
import { useGpuStore } from '../../stores/gpu.js'
const store = useGpuStore()
const loading = ref(true)
const overview = ref(null)
const probing = ref(false)
const recovering = ref(null)
// image_id -> 'no_source' for rows recovery already declined; keeps the
// verdict visible instead of a button that fails the same way again.
const recovered = ref({})
const defects = computed(() =>
(overview.value?.items || []).filter((i) => i.triage_status === 'defect'))
const classSummary = computed(() => {
const bc = overview.value?.by_class || {}
return Object.entries(bc)
.sort((a, b) => b[1] - a[1])
.map(([k, n]) => `${k.replaceAll('_', ' ')} ${n}`)
.join(' · ')
})
onMounted(refresh)
async function refresh() {
loading.value = true
try {
overview.value = await store.errors()
} catch (e) {
toast({ text: `Could not load failed jobs: ${e.message}`, type: 'error' })
} finally {
loading.value = false
}
}
async function onProbe() {
probing.value = true
try {
await store.triageErrors()
toast({ text: 'Probe queued — verdicts appear here as files are checked (large videos take a while)', type: 'success' })
} catch (e) {
toast({ text: `Could not start the probe: ${e.message}`, type: 'error' })
} finally {
probing.value = false
}
}
async function onRecover(it) {
recovering.value = it.image_id
try {
const res = await store.recoverImage(it.image_id)
if (res.status === 'refetch_queued') {
toast({ text: `Deleted the bad copy and queued a re-check of source #${res.source_id} — it re-imports on the next fetch`, type: 'success' })
await refresh()
} else if (res.status === 'no_source') {
recovered.value = { ...recovered.value, [it.image_id]: 'no_source' }
toast({ text: 'No enabled subscription source covers this file — replace it manually', type: 'warning' })
} else {
toast({ text: 'Image record no longer exists — refreshing', type: 'warning' })
await refresh()
}
} catch (e) {
toast({ text: `Recovery failed: ${e.message}`, type: 'error' })
} finally {
recovering.value = null
}
}
</script>
<style scoped>
.fc-muted { color: rgb(var(--v-theme-on-surface-variant)); }
.fc-section-h {
font-size: 13px; font-weight: 700; letter-spacing: 0.03em;
text-transform: uppercase; color: rgb(var(--v-theme-on-surface));
}
.fc-queue { display: flex; gap: 24px; }
.fc-q__n {
font-size: 20px; font-weight: 700; line-height: 1.1;
font-family: 'JetBrains Mono', monospace;
}
.fc-q__l {
font-size: 11px; text-transform: uppercase; letter-spacing: 0.04em;
color: rgb(var(--v-theme-on-surface-variant));
}
.fc-good { color: rgb(var(--v-theme-success)); }
.fc-weak { color: rgb(var(--v-theme-error)); }
.fc-defect {
display: flex; align-items: center; gap: 12px;
background: rgb(var(--v-theme-surface-light)); border-radius: 8px;
padding: 6px 10px;
}
.fc-defect__thumb {
flex: 0 0 44px; width: 44px; height: 44px; border-radius: 6px;
overflow: hidden; display: flex; align-items: center; justify-content: center;
background: rgba(0, 0, 0, 0.25);
}
.fc-defect__thumb img { width: 100%; height: 100%; object-fit: cover; }
.fc-defect__meta { flex: 1; min-width: 0; }
.fc-defect__err {
overflow: hidden; text-overflow: ellipsis; white-space: nowrap;
}
</style>
@@ -3,7 +3,7 @@
icon="mdi-brain"
title="Concept heads (the learning suggester)"
blurb="Train the per-concept heads that turn your tags into suggestions — they replace Camie and sharpen every time you accept or reject."
:open="headCount > 0 || running"
:open="running"
>
<p class="fc-muted text-body-2 mb-3">
A <strong>head</strong> is a tiny classifier trained on the SigLIP
@@ -13,6 +13,7 @@
<ThumbnailBackfillCard />
<ArchiveReextractCard />
<MissingFileRepairCard />
<GpuTriageCard />
<DbMaintenanceCard />
</div>
</section>
@@ -27,7 +28,6 @@
<HeadsCard />
<GpuAgentCard />
<AliasTable />
<TagEvalCard />
</div>
</section>
@@ -48,12 +48,12 @@ import MLBackfillCard from './MLBackfillCard.vue'
import ThumbnailBackfillCard from './ThumbnailBackfillCard.vue'
import ArchiveReextractCard from './ArchiveReextractCard.vue'
import MissingFileRepairCard from './MissingFileRepairCard.vue'
import GpuTriageCard from './GpuTriageCard.vue'
import DbMaintenanceCard from './DbMaintenanceCard.vue'
import MLThresholdSliders from './MLThresholdSliders.vue'
import HeadsCard from './HeadsCard.vue'
import GpuAgentCard from './GpuAgentCard.vue'
import AliasTable from './AliasTable.vue'
import TagEvalCard from './TagEvalCard.vue'
import BackupCard from './BackupCard.vue'
import { useSystemActivityStore } from '../../stores/systemActivity.js'
@@ -1,303 +0,0 @@
<template>
<MaintenanceTile
icon="mdi-flask-outline"
title="Tagging eval (heads vs centroid)"
blurb="Measure whether a trained head beats the old centroid on your own tags — and whether tagging more sharpens it."
:open="!!run"
>
<p class="fc-muted text-body-2 mb-3">
Reuses the SigLIP embeddings already stored on your images (no re-embed, no
GPU). For each concept it trains a logistic-regression <strong>head</strong>
on your positives + negatives and compares it to the old single
<strong>centroid</strong>, with cross-validated AP/F1 and a learning curve.
Runs as a background task; the result is saved and reloads here.
</p>
<v-textarea
v-model="conceptsText" label="Concepts (comma-separated)"
rows="2" auto-grow density="compact" hide-details class="mb-3"
:disabled="running"
/>
<div class="d-flex mb-3" style="gap: 12px;">
<v-text-field
v-model.number="autoTopN" label="+ auto-add top-N concepts"
type="number" min="0" max="200" density="compact" hide-details
:disabled="running" style="max-width: 220px;"
/>
<v-text-field
v-model.number="precisionTarget" label="Auto-apply precision target"
type="number" min="0.5" max="0.999" step="0.01" density="compact" hide-details
:disabled="running" style="max-width: 220px;"
/>
</div>
<v-btn
v-if="!running"
color="accent" variant="flat" rounded="pill"
prepend-icon="mdi-play" :loading="busy" @click="onStart"
>Run eval</v-btn>
<div v-if="running" class="mt-3">
<v-progress-linear indeterminate color="accent" />
<div class="text-body-2 mt-2 fc-muted">Running (started {{ startedAgo }})</div>
</div>
<v-alert
v-if="run && run.status === 'error'"
type="error" variant="tonal" density="compact" class="mt-3"
>Eval failed: {{ run.error }}</v-alert>
<div v-if="report" class="mt-4">
<div class="fc-muted text-caption mb-2">
Ran {{ formatTime(report.generated_at) }} ·
{{ report.concepts.length }} concept(s) ·
neg ratio {{ report.params.neg_ratio }}, {{ report.params.cv_folds }}-fold CV
</div>
<div v-for="c in report.concepts" :key="c.name" class="fc-cc">
<div class="fc-cc__head">
<span class="fc-cc__name">{{ c.name }}</span>
<span v-if="c.skipped" class="fc-muted text-caption"> skipped: {{ c.skipped }}</span>
<span v-else class="fc-muted text-caption">
{{ c.n_pos }} pos · {{ c.n_neg }} neg<span v-if="c.n_rejected"> ({{ c.n_rejected }} rejected)</span>
</span>
</div>
<template v-if="!c.skipped">
<table class="fc-metrics">
<thead>
<tr><th></th><th>AP</th><th>F1</th><th>Prec</th><th>Rec</th></tr>
</thead>
<tbody>
<tr>
<td class="fc-metrics__lbl">Head</td>
<td class="fc-num fc-win">{{ c.head.ap }}</td>
<td class="fc-num">{{ c.head.f1 }}</td>
<td class="fc-num">{{ c.head.precision }}</td>
<td class="fc-num">{{ c.head.recall }}</td>
</tr>
<tr>
<td class="fc-metrics__lbl fc-muted">Centroid</td>
<td class="fc-num fc-muted">{{ c.centroid.ap }}</td>
<td class="fc-num fc-muted">{{ c.centroid.f1 }}</td>
<td class="fc-num fc-muted">{{ c.centroid.precision }}</td>
<td class="fc-num fc-muted">{{ c.centroid.recall }}</td>
</tr>
</tbody>
</table>
<div class="text-caption mb-2" :class="apDelta(c) >= 0 ? 'fc-up' : 'fc-down'">
Δ AP {{ apDelta(c) >= 0 ? '+' : '' }}{{ apDelta(c).toFixed(3) }}
(head centroid)
</div>
<div class="text-caption mb-2">
<span class="fc-muted">Auto-apply:</span>
<template v-if="c.head.auto_apply">
<span class="fc-up">ready</span> at P{{ c.head.auto_apply.target }}
catches recall <strong>{{ c.head.auto_apply.recall }}</strong>
(thr {{ c.head.auto_apply.threshold }})
</template>
<span v-else class="fc-down">not reachable at P{{ report.params.precision_target }}</span>
</div>
<div v-if="c.curve && c.curve.length" class="fc-curve">
<span class="fc-muted text-caption">Learning curve (AP @ N positives):</span>
<span v-for="p in c.curve" :key="p.n_pos" class="fc-curve__pt">
{{ p.n_pos }}<strong>{{ p.ap }}</strong>
</span>
</div>
<div v-if="c.examples" class="fc-ex">
<div
v-for="grp in [
{ dir: 'suggest', items: c.examples.head_would_suggest,
label: `Head would suggest — ✓ tag it, ✗ not ${c.name}` },
{ dir: 'doubts', items: c.examples.head_doubts_positive,
label: `Head doubts your tag — ✓ keep, ✗ remove (not ${c.name})` },
]" :key="grp.dir" class="fc-ex__row"
>
<div class="fc-muted text-caption mb-1">{{ grp.label }}</div>
<div class="fc-ex__thumbs">
<div
v-for="it in grp.items" :key="`${grp.dir}${it.id}`"
class="fc-ex__item"
:class="actedLabel(c, grp.dir, it) ? 'fc-ex__item--acted' : ''"
>
<button
type="button" class="fc-ex__thumb"
:title="`#${it.id} — click to enlarge`" @click="modal.open(it.id)"
>
<img :src="it.thumbnail_url" loading="lazy" />
</button>
<div v-if="actedLabel(c, grp.dir, it)" class="fc-ex__badge">
{{ actedLabel(c, grp.dir, it) }}
</div>
<div v-else class="fc-ex__acts">
<button
class="fc-act fc-act--yes" type="button"
:title="`Yes — it is ${c.name}`" @click="act(c, it, grp.dir, 'yes')"
><v-icon size="15">mdi-check</v-icon></button>
<button
class="fc-act fc-act--no" type="button"
:title="`No — not ${c.name}`" @click="act(c, it, grp.dir, 'no')"
><v-icon size="15">mdi-close</v-icon></button>
</div>
</div>
</div>
</div>
</div>
</template>
</div>
</div>
</MaintenanceTile>
</template>
<script setup>
import { toast } from '../../utils/toast.js'
import { computed, onMounted, onUnmounted, ref } from 'vue'
import MaintenanceTile from '../common/MaintenanceTile.vue'
import { useTagEvalStore } from '../../stores/tagEval.js'
import { useModalStore } from '../../stores/modal.js'
const DEFAULT_CONCEPTS =
'glasses, cat, dog, horse, goblin, cum, lactation, fellatio, xray, stomach bulge'
const store = useTagEvalStore()
const modal = useModalStore()
const run = ref(null)
const conceptsText = ref(DEFAULT_CONCEPTS)
const autoTopN = ref(0)
const precisionTarget = ref(0.97)
const busy = ref(false)
let pollTimer = null
const running = computed(() => run.value?.status === 'running')
const report = computed(() => (run.value?.status === 'ready' ? run.value.report : null))
const startedAgo = computed(() =>
run.value?.started_at ? formatTime(run.value.started_at) : '')
// Rehydrate the persisted run on mount so the report survives navigation — the
// task runs backend-side regardless; we just reconnect to its row.
onMounted(async () => {
try {
const latest = await store.latest()
if (latest) {
run.value = await store.getRun(latest.id)
if (run.value.status === 'running') startPoll(latest.id)
}
} catch { /* non-fatal — card still works for a fresh run */ }
})
onUnmounted(stopPoll)
function startPoll(id) {
stopPoll()
pollTimer = setInterval(async () => {
try {
run.value = await store.getRun(id)
if (run.value.status !== 'running') stopPoll()
} catch (e) {
stopPoll()
toast({ text: `Eval poll failed: ${e.message}`, type: 'error' })
}
}, 5000)
}
function stopPoll() {
if (pollTimer) { clearInterval(pollTimer); pollTimer = null }
}
async function onStart() {
busy.value = true
try {
const concepts = conceptsText.value.split(',').map(s => s.trim()).filter(Boolean)
const res = await store.start({
concepts,
auto_top_n: Number(autoTopN.value) || 0,
precision_target: Number(precisionTarget.value) || 0.97,
})
run.value = await store.getRun(res.run_id)
startPoll(res.run_id)
} catch (e) {
const msg = e.body?.running_id
? 'An eval is already running.'
: e.message
toast({ text: `Could not start eval: ${msg}`, type: 'error' })
} finally {
busy.value = false
}
}
function apDelta(c) { return (c.head?.ap ?? 0) - (c.centroid?.ap ?? 0) }
function formatTime(iso) {
if (!iso) return ''
try { return new Date(iso).toLocaleString() } catch { return iso }
}
// Acting on an example writes the SAME tables the head trains on, so a re-run
// reflects the correction. Keyed per (concept, list, image); the report ids are
// frozen at run time, so we just grey out what's been handled in this view.
const acted = ref({})
const actedKey = (c, dir, it) => `${c.tag_id}:${dir}:${it.id}`
const actedLabel = (c, dir, it) => acted.value[actedKey(c, dir, it)] || ''
async function act(c, it, dir, verdict) {
const key = actedKey(c, dir, it)
let call, label
if (dir === 'suggest' && verdict === 'yes') { call = store.applyTag(it.id, c.tag_id); label = 'tagged' }
else if (dir === 'suggest' && verdict === 'no') { call = store.rejectTag(it.id, c.tag_id); label = 'rejected' }
else if (dir === 'doubts' && verdict === 'no') { call = store.removeTag(it.id, c.tag_id); label = 'removed' }
else { call = store.confirmTag(it.id, c.tag_id); label = 'kept' } // doubt + yes = keep (confirm)
try {
await call
acted.value[key] = label
} catch (e) {
toast({ text: `Action failed: ${e.message}`, type: 'error' })
}
}
</script>
<style scoped>
.fc-muted { color: rgb(var(--v-theme-on-surface-variant)); }
.fc-cc {
padding: 12px 0;
border-top: 1px solid rgb(var(--v-theme-surface-light));
}
.fc-cc__head { display: flex; align-items: baseline; gap: 8px; margin-bottom: 6px; }
.fc-cc__name { font-weight: 600; }
.fc-metrics { width: 100%; max-width: 360px; border-collapse: collapse; font-size: 13px; }
.fc-metrics th { text-align: right; font-weight: 600; color: rgb(var(--v-theme-on-surface-variant)); padding: 0 8px; }
.fc-metrics__lbl { text-align: left; }
.fc-num { text-align: right; font-variant-numeric: tabular-nums; padding: 1px 8px; }
.fc-win { color: rgb(var(--v-theme-accent)); font-weight: 600; }
.fc-up { color: rgb(var(--v-theme-success)); }
.fc-down { color: rgb(var(--v-theme-error)); }
.fc-curve { margin-bottom: 8px; }
.fc-curve__pt { margin-left: 10px; font-size: 13px; font-variant-numeric: tabular-nums; }
.fc-ex__row { margin-top: 8px; }
.fc-ex__thumbs { display: flex; flex-wrap: wrap; gap: 6px; }
.fc-ex__item { position: relative; width: 120px; height: 120px; }
.fc-ex__item--acted { opacity: 0.45; }
.fc-ex__thumb {
display: block; width: 100%; height: 100%; border-radius: 6px;
overflow: hidden; background: rgb(var(--v-theme-surface-light));
outline: 1px solid transparent; transition: outline-color 0.12s;
border: none; padding: 0; cursor: pointer;
}
.fc-ex__thumb:hover { outline-color: rgb(var(--v-theme-accent)); }
.fc-ex__thumb img { width: 100%; height: 100%; object-fit: cover; display: block; }
.fc-ex__acts { position: absolute; top: 4px; right: 4px; display: flex; gap: 4px; }
.fc-act {
width: 26px; height: 26px; border-radius: 50%; border: none; cursor: pointer;
display: flex; align-items: center; justify-content: center; color: #fff;
opacity: 0.9; box-shadow: 0 1px 3px rgba(0, 0, 0, 0.4); transition: transform 0.1s;
}
.fc-act:hover { opacity: 1; transform: scale(1.1); }
.fc-act--yes { background: rgb(var(--v-theme-success)); }
.fc-act--no { background: rgb(var(--v-theme-error)); }
.fc-ex__badge {
position: absolute; bottom: 4px; left: 4px; right: 4px; text-align: center;
font-size: 10px; text-transform: uppercase; letter-spacing: 0.05em;
background: rgba(0, 0, 0, 0.65); color: #fff; border-radius: 3px; padding: 1px 0;
}
</style>
@@ -42,53 +42,6 @@
</div>
</MaintenanceTile>
<MaintenanceTile
icon="mdi-tag-off"
title="Legacy migration tags"
blurb="Purge retired archive/post/artist + source:* tags."
destructive
>
<p class="fc-muted text-body-2 mb-3">
Purge legacy IR-migration tags FC no longer uses: retired/system
kinds (<code>archive</code>, <code>post</code>, <code>artist</code> e.g.
<code>BlenderKnight:Hannah_BJ_Loops</code>) plus <code>source:*</code> tags
(ImageRepo's old <code>source</code> kind, migrated to <code>general</code>).
Provenance and artists are their own systems now, so these are pure noise.
Removes them from every image.
</p>
<v-btn
color="accent" variant="flat" rounded="pill"
prepend-icon="mdi-magnify"
:loading="loadingKindPreview"
class="mb-3"
@click="onKindPreview"
>Preview legacy tags</v-btn>
<div v-if="kindPreview">
<p class="text-body-2 mb-2">
<strong>{{ kindPreview.count }}</strong> legacy tag(s).
<span v-for="(n, k) in kindPreview.by_kind" :key="k" class="fc-muted">
{{ k }}: {{ n }}&nbsp;&nbsp;
</span>
<span v-for="(n, p) in kindPreview.by_prefix" :key="p" class="fc-muted">
{{ p }}: {{ n }}&nbsp;&nbsp;
</span>
</p>
<SampleNameGrid
v-if="kindPreview.sample_names?.length"
:names="kindPreview.sample_names" class="mb-3"
/>
<v-btn
color="error" variant="flat" rounded="pill"
prepend-icon="mdi-delete-sweep"
:disabled="!kindPreview.count"
:loading="kindCommitting"
@click="onKindCommit"
>Delete {{ kindPreview.count }} legacy tag(s)</v-btn>
</div>
</MaintenanceTile>
<MaintenanceTile
icon="mdi-tag-multiple"
title="Reset content tagging"
@@ -216,16 +169,6 @@ const {
emptyPreview: (r) => ({ count: 0, sample_names: r.sample_names || [] }),
})
// Legacy migration-tag purge.
const {
previewData: kindPreview, previewing: loadingKindPreview,
committing: kindCommitting, runPreview: onKindPreview, runCommit: onKindCommit,
} = usePreviewCommit({
preview: () => store.purgeLegacyTags({ dryRun: true }),
commit: () => store.purgeLegacyTags({ dryRun: false }),
emptyPreview: { count: 0, by_kind: {}, by_prefix: {}, sample_names: [] },
})
// Reset content tagging (general + character).
const {
previewData: resetPreview, previewing: loadingResetPreview,
-5
View File
@@ -101,10 +101,6 @@ export const useAdminStore = defineStore('admin', () => {
})
}
function purgeLegacyTags(opts = {}) {
return _dryRunPost('/api/admin/tags/purge-legacy', opts)
}
// Destructive: deletes ALL general + character tags so the operator can
// re-tag from scratch via auto-suggest. fandom + series preserved.
function resetContentTagging(opts = {}) {
@@ -154,7 +150,6 @@ export const useAdminStore = defineStore('admin', () => {
pruneUnusedTags,
pruneBarePosts,
reconcileDuplicatePosts,
purgeLegacyTags,
resetContentTagging,
normalizeTags,
pollTaskUntilDone,
+24 -2
View File
@@ -36,10 +36,32 @@ export const useGpuStore = defineStore('gpu', () => {
}
// Requeue ONLY the errored jobs (all task types) — the scoped recovery after
// an agent fix, without re-running the done library. Returns { requeued }.
// an agent fix, without re-running the done library. Triage-confirmed
// defects stay put (they recover via recoverImage instead).
// Returns { requeued, pruned, defects_kept }.
async function retryErrors() {
return await api.post('/api/gpu/retry_errors')
}
return { token, rotateToken, status, backfill, reprocess, retryErrors }
// Failure-triage view (#125): errored jobs joined with integrity verdicts.
// Returns { total, by_class, triage, items }.
async function errors() {
return await api.get('/api/gpu/errors')
}
// Run the file-probe sweep now instead of waiting out the 15-min beat.
async function triageErrors() {
return await api.post('/api/gpu/errors/triage')
}
// Recover a defective original: delete the bad copy + record, re-poll its
// Source. Returns { status: 'refetch_queued'|'no_source'|'not_found', ... }.
async function recoverImage(imageId) {
return await api.post(`/api/gpu/errors/${imageId}/recover`)
}
return {
token, rotateToken, status, backfill, reprocess, retryErrors,
errors, triageErrors, recoverImage,
}
})
-57
View File
@@ -1,57 +0,0 @@
import { defineStore } from 'pinia'
import { useApi } from '../composables/useApi.js'
// Tag-eval (#1130): trigger + revisit the head-vs-centroid learning-curve eval.
// The run + full report live server-side (tag_eval_run), so the card rehydrates
// from getRun() on mount — the report survives navigation.
export const useTagEvalStore = defineStore('tagEval', () => {
const api = useApi()
async function start(params) {
return await api.post('/api/tag-eval', { body: { params } })
}
async function getRun(id) {
return await api.get(`/api/tag-eval/${id}`) // includes the full report
}
// The most recent run (light row, no report) — the card calls getRun() with
// its id to pull the persisted report on mount.
async function latest() {
const body = await api.get('/api/tag-eval', { params: { limit: 1 } })
return (body.runs && body.runs[0]) || null
}
// --- Acting on the head's example lists (closes the learn-from-tags loop).
// These write the SAME tables the head trains on: image_tag (positives) and
// tag_suggestion_rejection (negatives, via the dismiss endpoint).
// "Yes, it is this" — apply the tag (new positive).
async function applyTag(imageId, tagId) {
return await api.post(`/api/images/${imageId}/tags`,
{ body: { tag_id: tagId, source: 'manual' } })
}
// "No, it's not" on an UNtagged suggestion — record a rejection (hard negative).
async function rejectTag(imageId, tagId) {
return await api.post(`/api/images/${imageId}/suggestions/dismiss`,
{ body: { tag_id: tagId } })
}
// "Not it" on one of YOUR positives the head doubts — remove the tag AND
// record the rejection (kills the bad positive, leaves a hard negative).
async function removeTag(imageId, tagId) {
await api.delete(`/api/images/${imageId}/tags/${tagId}`)
return await api.post(`/api/images/${imageId}/suggestions/dismiss`,
{ body: { tag_id: tagId } })
}
// "Keep" — affirm a doubted positive is correct. Records a confirmation so it
// stops resurfacing in the doubts list (it stays a positive either way).
async function confirmTag(imageId, tagId) {
return await api.post(`/api/images/${imageId}/tags/${tagId}/confirm`)
}
return { start, getRun, latest, applyTag, rejectTag, removeTag, confirmTag }
})
+71
View File
@@ -0,0 +1,71 @@
"""Agent bandwidth governor (TokenBucket + PidReadMeter) — pure stdlib, so it
runs in the unit lane even though the rest of the agent (torch/ultralytics)
can't be imported in CI. Timing asserts use generous margins: they check the
throttle's ORDER of magnitude, not scheduler precision."""
import os
import threading
import time
from agent.fc_agent.throttle import PidReadMeter, TokenBucket
def test_unlimited_never_blocks_and_counts():
b = TokenBucket(0)
t0 = time.monotonic()
for _ in range(100):
b.take(10_000_000)
assert time.monotonic() - t0 < 0.2
assert b.consumed == 1_000_000_000
def test_take_enforces_average_rate():
# 400 KB/s with a 1s burst: the first 400KB is free (burst), the next
# 200KB is 0.5s of debt the caller must wait out.
b = TokenBucket(400_000)
b.take(400_000)
t0 = time.monotonic()
b.take(200_000)
elapsed = time.monotonic() - t0
assert 0.35 <= elapsed < 3.0
def test_set_rate_zero_unblocks_a_waiter():
b = TokenBucket(1_000) # 1 KB/s → a 1MB take would wait ~17 min
started = threading.Event()
def blocked_take():
started.set()
b.take(1_000_000)
th = threading.Thread(target=blocked_take, daemon=True)
th.start()
started.wait(1.0)
time.sleep(0.1) # let it enter the wait loop
b.set_rate(0) # lift the cap live (the UI dial)
th.join(timeout=2.0)
assert not th.is_alive()
def test_charge_reports_debt_without_blocking():
b = TokenBucket(1_000)
t0 = time.monotonic()
assert b.charge(1_000) == 0.0 # covered by the 1s burst
debt = b.charge(2_000) # 2s over budget → ~2s of debt
assert time.monotonic() - t0 < 0.2
assert 1.5 <= debt <= 2.1
assert b.consumed == 3_000
def test_pid_read_meter_tracks_own_reads():
m = PidReadMeter(os.getpid())
first = m.delta() # cumulative rchar since process start
assert first is not None and first > 0
with open("/dev/zero", "rb") as f:
f.read(1_048_576)
grown = m.delta()
assert grown is not None and grown >= 1_048_576
def test_pid_read_meter_missing_pid_degrades_to_none():
# PID 2^22+ is above the default pid_max — guaranteed absent.
assert PidReadMeter(2**22 + 12345).delta() is None
-57
View File
@@ -390,63 +390,6 @@ async def test_prune_unused_commit_deletes_and_returns_count(client, db):
assert "byebye" in body["sample_names"]
@pytest.mark.asyncio
async def test_purge_legacy_dry_run_counts_by_kind_and_prefix(client, db):
db.add_all([
Tag(name="BlenderKnight:Hannah_BJ_Loops", kind=TagKind.archive),
Tag(name="BlenderKnight:May Animation", kind=TagKind.post),
Tag(name="SomeArtist", kind=TagKind.artist),
# IR `source` kind fell back to general during migration —
# caught by the source:* name prefix, not by kind.
Tag(name="source:patreon", kind=TagKind.general),
Tag(name="blonde hair", kind=TagKind.general), # must survive
])
await db.commit()
resp = await client.post(
"/api/admin/tags/purge-legacy", json={"dry_run": True},
)
body = await resp.get_json()
assert body["count"] == 4
assert body["by_kind"] == {"archive": 1, "post": 1, "artist": 1}
assert body["by_prefix"] == {"source:*": 1}
assert "blonde hair" not in body["sample_names"]
assert "source:patreon" in body["sample_names"]
@pytest.mark.asyncio
async def test_purge_legacy_commit_deletes_only_legacy(client, db):
from sqlalchemy import func, select
db.add_all([
Tag(name="arch1", kind=TagKind.archive),
Tag(name="source:fanbox", kind=TagKind.general),
Tag(name="keepme", kind=TagKind.general),
Tag(name="char1", kind=TagKind.character),
])
await db.commit()
resp = await client.post(
"/api/admin/tags/purge-legacy", json={"dry_run": False},
)
body = await resp.get_json()
assert body["deleted"] == 2 # arch1 + source:fanbox
# The plain general + character tags survive.
keep = (await db.execute(
select(Tag.name).where(Tag.name.in_(["keepme", "char1"]))
)).scalars().all()
assert set(keep) == {"keepme", "char1"}
# No archive/post/artist or source:* left.
gone = (await db.execute(
select(func.count()).select_from(Tag).where(
(Tag.kind.in_([TagKind.archive, TagKind.post, TagKind.artist]))
| (Tag.name.like("source:%"))
)
)).scalar_one()
assert gone == 0
# --- Tier-A: POST /posts/prune-bare + /posts/reconcile-duplicates ---
# These two routes share _run_dry_run_op with the tag prunes (DRY pass,
# task #753); cover their apply path + reconcile's source_id passthrough so
+79 -4
View File
@@ -1,6 +1,8 @@
"""GPU-job HTTP API (#114): bearer auth + lease/submit round-trip + backfill."""
from datetime import UTC, datetime, timedelta
import pytest
from sqlalchemy import select
from sqlalchemy import func, select
from backend.app.models import GpuJob, ImageRecord
from backend.app.services.ml.gpu_jobs import GpuJobService
@@ -153,8 +155,11 @@ async def test_release_hands_job_back_to_pending(client, db):
@pytest.mark.asyncio
async def test_retry_errors_requeues_only_errored(client, db):
"""/retry_errors resets ERRORED jobs (any task) to pending with a fresh
retry budget — and leaves done work untouched (it is NOT /reprocess)."""
"""/retry_errors prunes stale tombstones first (older duplicates + rows a
later success made moot), then resets the SURVIVING errored jobs to pending
with a fresh retry budget — and leaves done work untouched (NOT /reprocess).
The prune is what stops one failing file fanning out into duplicate pending
jobs (the 2026-07-02 tombstone loop minted one error row per hour)."""
img1 = await _img(db, "1" * 64)
img2 = await _img(db, "2" * 64)
svc = GpuJobService(db)
@@ -165,12 +170,30 @@ async def test_retry_errors_requeues_only_errored(client, db):
j_err.status = "error"
j_err.attempts = 3
j_err.error = "no frames sampled from video (unprocessable)"
j_err.updated_at = datetime.now(UTC)
j_done.status = "done"
# Loop-era leftovers: an OLDER duplicate error row for img1's ccip, and a
# tombstone img2's done row makes moot — both pruned, never requeued.
dup = GpuJob(
image_record_id=img1.id, task="ccip", status="error",
error="older duplicate", updated_at=datetime.now(UTC) - timedelta(hours=1),
)
moot = GpuJob(
image_record_id=img2.id, task="siglip", status="error",
error="superseded by the done row",
)
db.add(dup)
db.add(moot)
await db.flush()
dup_id = dup.id
moot_id = moot.id
await db.commit()
resp = await client.post("/api/gpu/retry_errors")
assert resp.status_code == 200
assert (await resp.get_json())["requeued"] == 1
body = await resp.get_json()
assert body["requeued"] == 1
assert body["pruned"] == 2
# Column selects, not ORM refresh — the route wrote via Core DML.
row = (await db.execute(
@@ -182,6 +205,58 @@ async def test_retry_errors_requeues_only_errored(client, db):
select(GpuJob.status).where(GpuJob.id == done_id)
)
assert done_status == "done"
survivors = (await db.execute(
select(func.count()).select_from(GpuJob)
.where(GpuJob.id.in_([dup_id, moot_id]))
)).scalar_one()
assert survivors == 0
st = await (await client.get("/api/gpu/status")).get_json()
assert st["pending"] == 1 and st["error"] == 0
@pytest.mark.asyncio
async def test_retry_errors_keeps_triaged_defects(client, db):
"""A probe-confirmed DEFECT is a bad FILE — requeueing it just burns agent
time re-minting the tombstone, so /retry_errors leaves it for the recovery
surface and reports it as defects_kept."""
img1 = await _img(db, "4" * 64)
img2 = await _img(db, "5" * 64)
db.add(GpuJob(image_record_id=img1.id, task="ccip", status="error",
attempts=3, error="moov atom not found",
triage_status="defect"))
db.add(GpuJob(image_record_id=img2.id, task="ccip", status="error",
attempts=3, error="ffmpeg timed out after 1200s"))
await db.commit()
body = await (await client.post("/api/gpu/retry_errors")).get_json()
assert body["requeued"] == 1
assert body["defects_kept"] == 1
rows = dict((await db.execute(
select(GpuJob.image_record_id, GpuJob.status)
)).all())
assert rows[img1.id] == "error" # defect stays tombstoned
assert rows[img2.id] == "pending" # operational failure requeued
@pytest.mark.asyncio
async def test_errors_endpoint_reports_triage_view(client, db):
img = await _img(db, "6" * 64)
db.add(GpuJob(image_record_id=img.id, task="ccip", status="error",
attempts=3,
error="no frames sampled from video — moov atom not found"))
await db.commit()
resp = await client.get("/api/gpu/errors")
assert resp.status_code == 200
body = await resp.get_json()
assert body["total"] == 1
assert body["by_class"] == {"truncated_or_corrupt": 1}
assert body["triage"]["unclassified"] == 1
item = body["items"][0]
assert item["image_id"] == img.id
assert item["task"] == "ccip"
assert item["reason_class"] == "truncated_or_corrupt"
assert item["triage_status"] is None
assert item["image_url"].startswith("/images/")
-77
View File
@@ -1,77 +0,0 @@
import pytest
from backend.app.models import TagEvalRun
from backend.app.services.ml.tag_eval import (
DEFAULT_CONCEPTS,
_normalize_params,
)
pytestmark = pytest.mark.integration
def test_normalize_params_defaults_and_overrides():
d = _normalize_params(None)
assert d["concepts"] == DEFAULT_CONCEPTS
assert d["neg_ratio"] >= 1 and d["cv_folds"] >= 2
over = _normalize_params(
{"concepts": ["glasses", " ", "cat"], "neg_ratio": "4",
"cv_folds": "1", "curve_points": [30, 10, 10]}
)
assert over["concepts"] == ["glasses", "cat"] # blanks dropped
assert over["neg_ratio"] == 4
assert over["cv_folds"] == 2 # clamped to >=2
assert over["curve_points"] == [10, 30] # deduped + sorted
@pytest.mark.asyncio
async def test_history_and_detail_rehydrate(client, db):
# A finished run with a report — the persisted row IS the survives-navigation
# source: history is light (no report), detail carries it.
run = TagEvalRun(
params={"concepts": ["glasses"]},
status="ready",
report={"concepts": [{"name": "glasses", "head": {"ap": 0.9}}]},
)
db.add(run)
await db.flush()
await db.commit()
rid = run.id
h = await client.get("/api/tag-eval?limit=10")
assert h.status_code == 200
hbody = await h.get_json()
row = next(r for r in hbody["runs"] if r["id"] == rid)
assert row["status"] == "ready"
assert "report" not in row # list stays light
d = await client.get(f"/api/tag-eval/{rid}")
assert d.status_code == 200
dbody = await d.get_json()
assert dbody["report"]["concepts"][0]["name"] == "glasses"
@pytest.mark.asyncio
async def test_create_enqueues_running(client, db, monkeypatch):
monkeypatch.setattr(
"backend.app.tasks.ml.tag_eval_run.delay", lambda *a, **k: None
)
resp = await client.post("/api/tag-eval", json={"params": {"concepts": ["cat"]}})
assert resp.status_code == 202
body = await resp.get_json()
assert body["status"] == "running"
got = await db.get(TagEvalRun, body["run_id"])
assert got is not None and got.status == "running"
@pytest.mark.asyncio
async def test_create_conflicts_when_one_running(client, db, monkeypatch):
monkeypatch.setattr(
"backend.app.tasks.ml.tag_eval_run.delay", lambda *a, **k: None
)
db.add(TagEvalRun(params={}, status="running"))
await db.flush()
await db.commit()
resp = await client.post("/api/tag-eval", json={"params": {}})
assert resp.status_code == 409
body = await resp.get_json()
assert body["error"] == "eval_already_running"
+86 -1
View File
@@ -5,7 +5,11 @@ import pytest
from sqlalchemy import func, select
from backend.app.models import GpuJob, ImageRecord, ImageRegion
from backend.app.services.ml.gpu_jobs import GpuJobService
from backend.app.services.ml.gpu_jobs import (
EXPIRED_POISON_CAP,
PENDING_POISON_CAP,
GpuJobService,
)
pytestmark = pytest.mark.integration
@@ -263,3 +267,84 @@ async def test_recover_orphaned_resets_only_expired(db):
await db.commit()
assert (await db.get(GpuJob, expired.id)).status == "pending"
assert (await db.get(GpuJob, fresh.id)).status == "leased" # untouched
@pytest.mark.asyncio
async def test_backfill_skips_errored_images(db):
# An errored job is a TOMBSTONE for its (image, task): no backfill variant
# re-enqueues it — retry is deliberate-only (/retry_errors). Pre-fix, the
# hourly ccip run minted a fresh doomed job per bad file forever.
from backend.app.tasks.ml import enqueue_gpu_backfill
img = await _img(db, "f1" * 32)
svc = GpuJobService(db)
for task in ("ccip", "siglip", "embed"):
job = await svc.enqueue(img.id, task)
job.status = "error"
job.error = "no frames sampled from video"
await db.commit()
assert enqueue_gpu_backfill("ccip") == 0
assert enqueue_gpu_backfill("siglip") == 0
assert enqueue_gpu_backfill("embed") == 0
@pytest.mark.asyncio
async def test_backfill_prunes_moot_error_tombstones(db):
# Loop-era duplicates: several error rows for one (image, task), all made
# moot by a later done row. The backfill's dedupe pass removes them, and
# the done row still blocks re-enqueue.
from backend.app.tasks.ml import enqueue_gpu_backfill
img = await _img(db, "f2" * 32)
for i in range(3):
db.add(GpuJob(
image_record_id=img.id, task="ccip", status="error",
error=f"boom {i}",
))
db.add(GpuJob(image_record_id=img.id, task="ccip", status="done"))
await db.commit()
assert enqueue_gpu_backfill("ccip") == 0
statuses = (await db.execute(
select(GpuJob.status).where(
GpuJob.image_record_id == img.id, GpuJob.task == "ccip"
)
)).scalars().all()
assert statuses == ["done"]
@pytest.mark.asyncio
async def test_recover_poisons_runaway_jobs(db):
# release/expiry loops never reach fail()'s attempt cap — the sweep converts
# them to 'error': an expired lease after EXPIRED_POISON_CAP grants (job
# crashes/wedges the agent every time) and a pending job after
# PENDING_POISON_CAP grants that never completed (transfer stalls forever).
img1 = await _img(db, "06" + "a" * 62)
img2 = await _img(db, "07" + "a" * 62)
svc = GpuJobService(db)
j1 = await svc.enqueue(img1.id, "ccip")
j2 = await svc.enqueue(img2.id, "ccip")
j1.status = "leased"
j1.lease_token = "dead"
j1.lease_expires_at = datetime.now(UTC) - timedelta(minutes=10)
j1.attempts = EXPIRED_POISON_CAP
j2.attempts = PENDING_POISON_CAP
j1_id = j1.id
j2_id = j2.id
await db.commit()
assert await svc.recover_orphaned() == 0 # nothing recovered — both poisoned
await db.commit()
# Column selects, not ORM refresh — the sweep wrote via Core DML.
rows = (await db.execute(
select(GpuJob.id, GpuJob.status, GpuJob.error)
.where(GpuJob.id.in_([j1_id, j2_id]))
)).all()
by_id = {r.id: r for r in rows}
assert by_id[j1_id].status == "error"
assert "poisoned" in by_id[j1_id].error
assert by_id[j2_id].status == "error"
assert "poisoned" in by_id[j2_id].error
+165
View File
@@ -0,0 +1,165 @@
"""Failure triage (#125): probe errored jobs' files, flag verdicts, recover.
The probe is the arbiter: reason strings only bucket the overview. A file that
passes checksum+decode is 'file_ok' (operational failure); anything else is a
'defect' — surfaced for recovery and excluded from /retry_errors.
"""
import hashlib
import pytest
from PIL import Image as PILImage
from sqlalchemy import select
from backend.app.models import (
Artist,
GpuJob,
ImageProvenance,
ImageRecord,
Post,
Source,
)
from backend.app.services.ml.gpu_triage import (
classify_reason,
recover_defective_image,
triage_errored_jobs,
)
pytestmark = pytest.mark.integration
def test_classify_reason_buckets():
assert classify_reason(
"no frames sampled from video — moov atom not found"
) == "truncated_or_corrupt"
assert classify_reason("ffmpeg timed out after 1200s") == "timeout"
assert classify_reason(
"gave up after repeated transient failures: HTTPConnectionPool read timed out"
) == "transient"
assert classify_reason(
"poisoned: 10+ lease attempts without ever completing"
) == "poisoned"
assert classify_reason("cannot identify image file") == "decode"
assert classify_reason("something novel") == "other"
assert classify_reason(None) == "other"
async def _errored_image(db, tmp_path, *, name, sha, content: bytes | None,
error="no frames sampled from video — moov atom not found"):
"""An ImageRecord (file written iff content is not None) + an errored job."""
path = tmp_path / name
if content is not None:
path.write_bytes(content)
img = ImageRecord(
path=str(path), sha256=sha, size_bytes=1, mime="image/png",
width=1, height=1, origin="imported_filesystem",
integrity_status="unknown",
)
db.add(img)
await db.flush()
db.add(GpuJob(image_record_id=img.id, task="ccip", status="error",
error=error, attempts=3))
await db.flush()
return img
@pytest.mark.asyncio
async def test_triage_probes_and_splits_defect_vs_file_ok(db, tmp_path):
# Healthy: a real PNG whose recorded sha matches its bytes → file_ok.
ok_path = tmp_path / "fine.png"
PILImage.new("RGB", (4, 4), (200, 30, 30)).save(ok_path)
ok_sha = hashlib.sha256(ok_path.read_bytes()).hexdigest()
ok = await _errored_image(db, tmp_path, name="fine.png", sha=ok_sha,
content=None, error="ffmpeg timed out after 1200s")
# Corrupt: bytes don't match the recorded sha → defect.
bad = await _errored_image(db, tmp_path, name="bad.png", sha="0" * 64,
content=b"not a real png")
# Missing: no file on disk at all → defect (failed_verification).
gone = await _errored_image(db, tmp_path, name="gone.png", sha="1" * 64,
content=None)
await db.commit()
summary = await db.run_sync(lambda s: triage_errored_jobs(s))
assert summary["probed"] == 3
assert summary["defect"] == 2
assert summary["file_ok"] == 1
assert summary["partial"] is False
# Column selects, not ORM refresh — the sweep wrote via Core DML.
rows = dict((await db.execute(
select(GpuJob.image_record_id, GpuJob.triage_status)
.where(GpuJob.status == "error")
)).all())
assert rows[ok.id] == "file_ok"
assert rows[bad.id] == "defect"
assert rows[gone.id] == "defect"
verdicts = dict((await db.execute(
select(ImageRecord.id, ImageRecord.integrity_status)
.where(ImageRecord.id.in_([ok.id, bad.id, gone.id]))
)).all())
assert verdicts[ok.id] == "ok"
assert verdicts[bad.id] == "corrupt"
assert verdicts[gone.id] == "failed_verification"
# Idempotent: everything already triaged → no re-probe.
again = await db.run_sync(lambda s: triage_errored_jobs(s))
assert again["probed"] == 0
@pytest.mark.asyncio
async def test_recover_without_pollable_source_reports_no_source(db, tmp_path):
img = await _errored_image(db, tmp_path, name="orphan.png", sha="2" * 64,
content=b"x")
await db.commit()
res = await db.run_sync(
lambda s: recover_defective_image(s, img.id, images_root=tmp_path)
)
assert res["status"] == "no_source"
still_there = (await db.execute(
select(ImageRecord.id).where(ImageRecord.id == img.id)
)).scalar_one_or_none()
assert still_there == img.id
@pytest.mark.asyncio
async def test_recover_deletes_record_and_requeues_source(
client, db, tmp_path, monkeypatch,
):
img = await _errored_image(db, tmp_path, name="fixme.png", sha="3" * 64,
content=b"x")
artist = Artist(name="Recov", slug="recov")
db.add(artist)
await db.flush()
src = Source(artist_id=artist.id, platform="patreon",
url="https://www.patreon.com/recov", enabled=True)
db.add(src)
await db.flush()
post = Post(artist_id=artist.id, source_id=src.id, external_post_id="p1")
db.add(post)
await db.flush()
db.add(ImageProvenance(image_record_id=img.id, post_id=post.id,
source_id=src.id))
img_id, src_id = img.id, src.id
await db.commit()
queued = []
monkeypatch.setattr(
"backend.app.tasks.download.download_source.delay",
lambda sid: queued.append(sid),
)
resp = await client.post(f"/api/gpu/errors/{img_id}/recover")
assert resp.status_code == 200
body = await resp.get_json()
assert body["status"] == "refetch_queued"
assert body["source_id"] == src_id
assert queued == [src_id]
# Record gone — the error tombstones cascade away with it.
remaining = (await db.execute(
select(ImageRecord.id).where(ImageRecord.id == img_id)
)).scalar_one_or_none()
assert remaining is None
jobs_left = (await db.execute(
select(GpuJob.id).where(GpuJob.image_record_id == img_id)
)).scalars().all()
assert jobs_left == []