Merge pull request 'Agent: server-side throughput rates + killable-on-stop ffmpeg' (#183) from dev into main

Two follow-ups from live debugging of "work/min never populates" and
"stopped never reached".

1) jobs/min + downloads/min are now computed in the BACKEND on a fixed
   cadence (_rate_loop, EWMA) and reported ready-to-show. The rates were
   derived client-side from poll deltas with a dt<30s guard — but a
   backgrounded/unfocused browser tab throttles its timers to ~1/min, so
   every delta exceeded 30s and the guard blanked the rates forever. A
   server-side rate is independent of how often the tab polls. Frontend just
   displays s.jobs_per_min / s.downloads_per_min. VERSION → .9.

2) ffmpeg video sampling is now killable on Stop. A downloader stuck in a
   slow/reconnecting decode (observed: 47s, 230s for one video) couldn't see
   the stop signal until ffmpeg returned, so Stop detached still-running
   threads and work kept flowing long after — "stopped" that wasn't really
   stopped. sample_frames_from_url now runs ffmpeg via Popen and polls a
   `should_stop` callback every 0.5s, terminating (then killing) the process
   at once on Stop or the per-video timeout. A stop-killed job is handed back
   (transient), not failed.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.forgejo/workflows/ci.yml

@@ -27,7 +27,14 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - name: Ruff lint
-        run: ruff check backend/ tests/ alembic/
+        # agent/ included so the GPU-agent is linted before its image is built
+        # (build.yml only `docker build`s it — this is where it gets checked).
+        run: ruff check backend/ tests/ alembic/ agent/
+      - name: Agent syntax check
+        # The agent's runtime deps (torch/transformers/ultralytics) aren't in the
+        # CI image, so we can't import it — but compileall parses every module,
+        # catching syntax errors before the image build.
+        run: python -m compileall -q agent/fc_agent
 
   backend-lint-and-test:
     runs-on: python-ci

agent/Dockerfile

@@ -1,18 +1,24 @@
 # FabledCurator GPU agent — runs on the desktop with the GPU.
-# CUDA + cuDNN runtime so onnxruntime-gpu can use the card (it needs cuDNN 9 —
-# the plain -runtime image lacks it: "libcudnn.so.9: cannot open shared object
-# file"); ffmpeg for video frames.
-FROM nvidia/cuda:12.4.1-cudnn-runtime-ubuntu22.04
+# CUDA 12.9 + cuDNN 9 runtime so onnxruntime-gpu can use the card (it needs
+# cuDNN 9 — the plain -runtime image lacks it: "libcudnn.so.9: cannot open
+# shared object file"); ffmpeg for video frames. Ubuntu 24.04 → Python 3.12.
+# Stays on the CUDA-12 / cuDNN-9 line the default onnxruntime-gpu + torch are
+# built against (CUDA 13 has only nascent ONNX Runtime support).
+FROM nvidia/cuda:12.9.2-cudnn-runtime-ubuntu24.04
 
-ENV DEBIAN_FRONTEND=noninteractive PYTHONUNBUFFERED=1
+# PIP_BREAK_SYSTEM_PACKAGES: Ubuntu 24.04 marks its system Python as externally
+# managed (PEP 668), so a global `pip install` errors without this. It's a
+# single-purpose container — we own the whole environment, so installing into
+# the system site-packages is fine (and simplest — no venv on PATH to manage).
+ENV DEBIAN_FRONTEND=noninteractive PYTHONUNBUFFERED=1 PIP_BREAK_SYSTEM_PACKAGES=1
 RUN apt-get update \
     && apt-get install -y --no-install-recommends python3 python3-pip ffmpeg \
     && rm -rf /var/lib/apt/lists/*
 
 WORKDIR /app
-# torch from the CUDA-12.4 wheel index (matches the base image); its wheels
-# bundle their own CUDA + cuDNN and coexist with onnxruntime-gpu. Installed
-# first + separately so the GPU build of torch is deterministic and layer-cached.
+# torch from the CUDA-12.4 wheel index; its wheels bundle their own CUDA + cuDNN
+# so they run on the 12.9 base and coexist with onnxruntime-gpu. Installed first
+# + separately so the GPU build of torch is deterministic and layer-cached.
 RUN pip3 install --no-cache-dir torch==2.6.0 --index-url https://download.pytorch.org/whl/cu124
 COPY requirements.txt .
 RUN pip3 install --no-cache-dir -r requirements.txt

agent/docker-compose.yml

@@ -34,9 +34,25 @@ services:
       # Resume the worker automatically on container start (survive a reboot /
       # crash-restart while you're away). Set to 0 to require a manual Start.
       AUTO_START: ${AUTO_START:-1}
+      # Autoscale the worker count (throughput hill-climb that finds the sweet
+      # 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}
       # 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}
+      # Crop PROPOSERS (extra YOLO detectors → more/better concept crops). Each
+      # downloads its weights once (cached on the models volume) and self-disables
+      # if the download/load fails. Blank any one to turn it off.
+      #   PERSON_WEIGHTS: general COCO person detector (Western/realistic figures),
+      #     merged with the anime detector. yolo11n.pt (~6 MB, auto-downloaded).
+      #   ANATOMY_WEIGHTS: booru_yolo anime/furry/NSFW components (~40 MB). NB the
+      #     repo states no license — fine for private use. yolov8n_as01.pt is the
+      #     6 MB nano if you want lighter than yolov11m_aa22.pt.
+      #   PANEL_WEIGHTS: mosesb comic-panel detector (Apache-2.0), "hf_repo::file".
+      PERSON_WEIGHTS: ${PERSON_WEIGHTS:-yolo11n.pt}
+      ANATOMY_WEIGHTS: ${ANATOMY_WEIGHTS:-https://github.com/aperveyev/booru_yolo/raw/main/models/yolov11m_aa22.pt}
+      PANEL_WEIGHTS: ${PANEL_WEIGHTS:-mosesb/best-comic-panel-detection::best.pt}
     volumes:
       # Persist the downloaded ONNX models so restarts are fast.
       - fc-agent-models:/models

agent/fc_agent/app.py

@@ -1,21 +1,45 @@
 """FastAPI control surface for the agent (served on localhost).
 
-Start / stop the worker pool, tune the worker count live (trades desktop
-responsiveness for throughput), and watch GPU load + progress + the server-side
-queue. Config is env-seeded; the worker count is adjustable here on the fly.
+Start / stop the download→GPU pipeline, tune the downloader count live (the
+workload is download-bound, so downloaders are the dial that trades desktop
+bandwidth for throughput), and watch GPU load + buffer occupancy + progress +
+the server-side queue. Config is env-seeded; the downloader count is adjustable
+here on the fly (GPU consumers autoscale between 1 and 2 on their own).
 """
+import logging
+
 from fastapi import FastAPI, Request
 from fastapi.responses import HTMLResponse, JSONResponse
 
+from . import logbuf
 from .config import Config
 from .gpu import read_gpu
 from .worker import Worker
 
+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-01.9 · server-computed jobs/min + downloads/min (poll-rate independent)"
+
+logbuf.install()
 cfg = Config.from_env()
 worker = Worker(cfg)
 app = FastAPI(title="FabledCurator GPU agent")
 
 
+@app.middleware("http")
+async def _no_store(request, call_next):
+    # The control page is a static string and the status/gpu/logs polls are
+    # live data — never let the browser cache either, or a freshly-pulled agent
+    # image still shows the OLD UI until a hard refresh (operator-flagged
+    # 2026-06-30).
+    resp = await call_next(request)
+    resp.headers["Cache-Control"] = "no-store"
+    return resp
+
+
 @app.on_event("startup")
 def _maybe_autostart() -> None:
     # With AUTO_START set, a container restart (host reboot, or `restart:
@@ -28,17 +52,19 @@ def _maybe_autostart() -> None:
 
 @app.get("/", response_class=HTMLResponse)
 def index() -> str:
-    return _PAGE
+    return _PAGE.replace("__BUILD__", VERSION)
 
 
 @app.post("/start")
 def start():
+    log.info("UI: Start button pressed")   # the press; worker logs the transition
     worker.start()
     return JSONResponse(worker.status())
 
 
 @app.post("/stop")
 def stop():
+    log.info("UI: Stop button pressed")
     worker.stop()
     return JSONResponse(worker.status())
 
@@ -50,85 +76,304 @@ async def concurrency(request: Request):
     return JSONResponse(worker.status())
 
 
+@app.post("/auto")
+async def auto(request: Request):
+    body = await request.json()
+    worker.set_auto(bool(body.get("value", True)))
+    return JSONResponse(worker.status())
+
+
+@app.get("/gpu")
+def gpu():
+    # GPU meters poll this on their own fast cadence. It only reads local
+    # nvidia-smi — no curator round-trip — so the util/VRAM bars stay live even
+    # when /status is slow waiting on the (sometimes busy) curator queue call.
+    g = read_gpu() or {}
+    us = worker.util_smooth()
+    if us is not None:
+        g["util_smooth"] = round(us, 1)   # autoscaler's EWMA — the UI bar tracks this
+    return JSONResponse(g)
+
+
+@app.get("/logs")
+def logs():
+    return JSONResponse({"lines": list(logbuf.LINES)})
+
+
 @app.get("/status")
 def status():
+    # Pure in-memory read: worker.status() is lock-free and the queue snapshot is
+    # kept fresh by a background poller — NO inline curator call, so this can't
+    # stall the status view when curator is buried under a big backlog.
+    worker.note_ui()          # a browser is watching → keep the queue snapshot warm
     s = worker.status()
     s["fc_url"] = cfg.fc_url
     s["configured"] = bool(cfg.token)
-    s["gpu"] = read_gpu()
-    try:
-        s["queue"] = worker.client.queue_status()
-    except Exception:
-        s["queue"] = None
+    s["queue"] = worker.latest_queue()
+    s["build"] = VERSION
     return JSONResponse(s)
 
 
 _PAGE = """<!doctype html><html><head><meta charset=utf-8>
-<title>FabledCurator GPU agent</title>
+<meta name=viewport content="width=device-width,initial-scale=1">
+<title>FabledCurator · GPU agent</title>
 <style>
- body{font:14px system-ui;margin:2rem;max-width:680px;background:#14171a;color:#e8e8e8}
- h1{font-size:18px} button{font:14px system-ui;padding:.5rem 1rem;border:0;border-radius:6px;
- margin-right:.5rem;cursor:pointer;color:#fff} .start{background:#2e7d32}.stop{background:#b3261e}
- .step{background:#33373b;padding:.4rem .7rem;font-weight:700}
- .stat{display:inline-block;margin-right:1.5rem;vertical-align:top}
- .n{font-size:22px;font-weight:700} code{background:#222;padding:2px 6px;border-radius:4px}
- .q,.gpu{margin-top:1rem;color:#9aa} .bar{height:8px;border-radius:4px;background:#222;overflow:hidden;
- max-width:320px;margin-top:4px} .bar>i{display:block;height:100%;background:#3f7d3f}
- .row{margin:.8rem 0}
+ :root{--bg:#0f1216;--panel:#181c22;--panel2:#1e232b;--bd:#2a313b;--fg:#e9edf2;
+  --mut:#8b97a6;--acc:#e8923a;--grn:#46c46a;--red:#e8584d;--amb:#e8b23a}
+ *{box-sizing:border-box}
+ body{font:14px/1.5 system-ui,-apple-system,Segoe UI,Roboto,sans-serif;margin:0;
+  background:radial-gradient(1200px 600px at 50% -10%,#1a2029,#0f1216);color:var(--fg)}
+ .wrap{max-width:820px;margin:0 auto;padding:28px 20px 28px;height:100vh;
+  box-sizing:border-box;overflow:hidden;display:flex;flex-direction:column}
+ header{display:flex;align-items:center;justify-content:space-between;margin-bottom:4px}
+ .brand{display:flex;align-items:center;gap:10px;font-size:19px;font-weight:700;letter-spacing:.2px}
+ .logo{color:var(--acc);font-size:20px}
+ .brand .sub{color:var(--mut);font-weight:600;font-size:13px;text-transform:uppercase;letter-spacing:.12em}
+ .conn{display:flex;align-items:center;gap:8px;color:var(--mut);font-size:13px;font-weight:600}
+ .dot{width:9px;height:9px;border-radius:50%;background:var(--mut);box-shadow:0 0 0 0 rgba(0,0,0,0)}
+ .dot.green{background:var(--grn);box-shadow:0 0 10px 1px rgba(70,196,106,.5)}
+ .dot.amber{background:var(--amb)} .dot.red{background:var(--red)}
+ .meta{color:var(--mut);margin:0 0 18px;font-size:13px}
+ code{background:#11151a;border:1px solid var(--bd);padding:2px 7px;border-radius:6px;
+  font:12px ui-monospace,SFMono-Regular,Menlo,monospace;color:#cdd6e0}
+ .card{background:linear-gradient(180deg,var(--panel),var(--panel2));border:1px solid var(--bd);
+  border-radius:14px;padding:16px 18px;margin-bottom:14px;box-shadow:0 1px 0 rgba(255,255,255,.02) inset}
+ .card-h{font-size:11px;font-weight:800;letter-spacing:.12em;text-transform:uppercase;
+  color:var(--mut);margin-bottom:14px}
+ .controls{display:flex;align-items:center;gap:10px;flex-wrap:wrap}
+ .spacer{flex:1}
+ .btn{font:600 14px system-ui;padding:.5rem 1rem;border:1px solid transparent;border-radius:9px;
+  cursor:pointer;color:#fff;transition:.12s}
+ .btn:hover{transform:translateY(-1px)}
+ .btn[disabled]{opacity:.45;pointer-events:none;transform:none}
+ @keyframes pulse{0%,100%{opacity:1}50%{opacity:.4}}
+ .tile .n.busy{color:var(--acc);animation:pulse 1s ease-in-out infinite}
+ .btn.start{background:linear-gradient(180deg,#2f9c4c,#247a3c)}
+ .btn.stop{background:linear-gradient(180deg,#3a3f48,#2a2f37);color:#e9edf2;border-color:var(--bd)}
+ .switch{display:inline-flex;align-items:center;gap:8px;cursor:pointer;font-weight:600;user-select:none}
+ .switch input{display:none}
+ .switch .track{width:38px;height:22px;border-radius:11px;background:#2a313b;position:relative;transition:.15s}
+ .switch .track:after{content:"";position:absolute;top:2px;left:2px;width:18px;height:18px;border-radius:50%;
+  background:#cdd6e0;transition:.15s}
+ .switch input:checked+.track{background:var(--acc)}
+ .switch input:checked+.track:after{transform:translateX(16px);background:#fff}
+ .stepper{display:inline-flex;align-items:center;gap:6px}
+ .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;
+  color:var(--fg);border:1px solid var(--bd);border-radius:8px}
+ .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}
+ .tile .n{font:800 22px ui-monospace,monospace;line-height:1.1}
+ .tile .n.warn{color:var(--red)} .tile .n.ok{color:var(--grn)}
+ .tile .l{font-size:10px;text-transform:uppercase;letter-spacing:.06em;color:var(--mut);margin-top:4px}
+ .meters{display:flex;flex-direction:column;gap:10px;margin-bottom:14px}
+ .meter-h{display:flex;justify-content:space-between;font-size:12px;color:var(--mut);margin-bottom:4px}
+ .meter-h b{color:var(--fg);font-variant-numeric:tabular-nums}
+ .bar{height:9px;border-radius:5px;background:#11151a;border:1px solid var(--bd);overflow:hidden}
+ .bar>i{display:block;height:100%;width:0;background:linear-gradient(90deg,#3a7d57,var(--grn));transition:width .4s}
+ #utilbar{background:linear-gradient(90deg,#9a5a1f,var(--acc))}
+ #bufbar{background:linear-gradient(90deg,#2f5a9a,#4a86d8)}
+ .queue{font:13px ui-monospace,monospace;color:var(--mut)}
+ .banner{margin:0 0 14px;padding:.7rem .9rem;border-radius:10px;background:#3a2f12;
+  border:1px solid #5a4a17;color:#ffd98a;font-size:13px}
+ .logs-h{display:flex;align-items:center;justify-content:space-between}
+ .grow{flex:1;display:flex;flex-direction:column;min-height:0}
+ .grow .logs{flex:1;min-height:0}
+ .copybtn{font:600 11px system-ui;letter-spacing:.04em;text-transform:uppercase;
+  background:#262c34;color:var(--fg);border:1px solid var(--bd);border-radius:7px;
+  padding:5px 11px;cursor:pointer}
+ .copybtn:hover{border-color:var(--acc)}
+ .logs{margin:0;background:#0b0e12;border:1px solid var(--bd);border-radius:10px;padding:12px;
+  overflow:auto;font:12px/1.55 ui-monospace,SFMono-Regular,Menlo,monospace;
+  color:#b9c4d0;white-space:pre-wrap;word-break:break-word}
 </style></head><body>
-<h1>FabledCurator GPU agent</h1>
-<p>FC: <code id=fc>—</code> · token <code id=cfg>—</code></p>
-<div class=row>
- <button class=start onclick=act('start')>Start</button>
- <button class=stop onclick=act('stop')>Stop</button>
+<div class=wrap>
+ <header>
+  <div class=brand><span class=logo>◆</span> FabledCurator <span class=sub>GPU agent</span></div>
+  <div class=conn><span class="dot" id=dot></span><span id=connlbl>—</span></div>
+ </header>
+ <p class=meta>Server <code id=fc>—</code> · token <code id=cfg>—</code> · build <code id=build>__BUILD__</code></p>
+
+ <div id=verbanner class=banner style="display:none;background:#3a1212;border-color:#5a1717;color:#ffb3b3">
+  a newer agent version is running — reload this page (Ctrl+Shift+R) to update the controls
+ </div>
+ <div id=banner class=banner style=display:none>
+  curator unreachable — holding work + retrying, resumes on its own (no restart needed)
+ </div>
+
+ <section class=card>
+  <div class=card-h>Control</div>
+  <div class=controls>
+   <button class="btn start" id=startbtn onclick=act('start')>▶ Start</button>
+   <button class="btn stop" id=stopbtn onclick=act('stop')>■ Stop</button>
+   <div class=spacer></div>
+   <label class=switch><input type=checkbox id=autochk onchange="setauto(this.checked)"><span class=track></span>Auto</label>
+   <div class=stepper>
+    <button class=step onclick=setc(-1)>−</button>
+    <input id=conc type=number min=1 value=1 onchange="setv(this.value)">
+    <button class=step onclick=setc(1)>+</button>
+   </div>
+  </div>
+  <div class=hint id=conchint>auto-tuning downloaders to keep the GPU fed · max <b id=capn>8</b></div>
+ </section>
+
+ <section class=card>
+  <div class=card-h>Status</div>
+  <div class=tiles>
+   <div class=tile><div class=n id=state>—</div><div class=l>state</div></div>
+   <div class=tile><div class=n id=jpm>—</div><div class=l>jobs / min</div></div>
+   <div class=tile><div class=n id=dpm>—</div><div class=l>downloads / min</div></div>
+   <div class=tile><div class="n ok" id=done>0</div><div class=l>processed</div></div>
+   <div class=tile><div class=n id=err>0</div><div class=l>errors</div></div>
+   <div class=tile><div class=n id=waited>0</div><div class=l>waited out</div></div>
+  </div>
+  <div class=meters>
+   <div class=meter><div class=meter-h><span>GPU util</span><b id=utillbl>—</b></div>
+    <div class=bar><i id=utilbar></i></div></div>
+   <div class=meter><div class=meter-h><span>VRAM</span><b id=vramlbl>—</b></div>
+    <div class=bar><i id=gpubar></i></div></div>
+   <div class=meter><div class=meter-h><span>buffer occupancy</span><b id=buflbl>—</b></div>
+    <div class=bar><i id=bufbar></i></div></div>
+  </div>
+  <div class=queue id=pipe>downloaders — · consumers — · on GPU 0</div>
+  <div class=queue id=queue>queue —</div>
+ </section>
+
+ <section class="card grow">
+  <div class="card-h logs-h">Logs
+   <button class=copybtn id=copybtn onclick=copyLogs()>Copy</button>
+  </div>
+  <pre class=logs id=logs>waiting for activity…</pre>
+ </section>
 </div>
-<div class=row>
- workers
- <button class=step onclick=setc(-1)>−</button>
- <input id=conc type=number min=1 value=1
-  style="width:3.5rem;font:700 16px system-ui;text-align:center;background:#222;color:#e8e8e8;border:1px solid #444;border-radius:6px;padding:.3rem"
-  onchange="setv(this.value)">
- <button class=step onclick=setc(1)>+</button>
- <span class=cap style=color:#9aa>(more = overlap I/O, fill the GPU) max <b id=capn>8</b></span>
-</div>
-<div class=row>
- <span class=stat><span class=n id=state>stopped</span><br>state</span>
- <span class=stat><span class=n id=active>0</span><br>active now</span>
- <span class=stat><span class=n id=done>0</span><br>processed</span>
- <span class=stat><span class=n id=err>0</span><br>errors</span>
- <span class=stat><span class=n id=wait>0</span><br>waited out</span>
-</div>
-<div id=banner style="display:none;margin:.6rem 0;padding:.5rem .8rem;border-radius:6px;background:#5a4a17;color:#ffe28a">
- curator unreachable — holding work + retrying, will resume on its own (no restart needed)
-</div>
-<div class=gpu id=gpu>GPU — …</div>
-<div class=bar><i id=gpubar style=width:0%></i></div>
-<div class=q id=queue></div>
 <script>
+ const PAGE_BUILD="__BUILD__"
  let CAP=8
- async function act(p){await fetch('/'+p,{method:'POST'});refresh()}
- function setc(d){ setv((parseInt(conc.value||'1'))+d) }
+ // Optimistic transitional state on click, then apply the POST's own status
+ // response (it returns worker.status()) for instant feedback — don't wait on the
+ // separate /status poll, which can lag behind the curator queue call.
+ async function act(p){
+  pending(p==='start'?'starting':'stopping')
+  // Abort a slow POST after 8s so the buttons never stay stuck — the periodic
+  // /status refresh (now always fast) recovers the true state either way.
+  const ac=new AbortController(); const to=setTimeout(()=>ac.abort(),8000)
+  try{ applyStatus(await (await fetch('/'+p,{method:'POST',signal:ac.signal})).json()) }
+  catch{ refresh() /* on abort/error, repaint the real state from /status */ }
+  finally{ clearTimeout(to) }
+ }
+ function pending(label){
+  // Instant optimistic feedback on click; applyStatus (POST response, then the
+  // periodic poll) then owns the real state + which buttons are enabled.
+  state.textContent=label; state.className='n busy'
+  dot.className='dot amber'
+  startbtn.disabled=true; stopbtn.disabled=true
+ }
+ function setc(d){ if(conc.disabled)return; setv((parseInt(conc.value||'1'))+d) }
  async function setv(v){
   v=Math.max(1,Math.min(CAP,parseInt(v)||1)); conc.value=v
   await fetch('/concurrency',{method:'POST',headers:{'Content-Type':'application/json'},
    body:JSON.stringify({value:v})});refresh()
  }
+ async function setauto(on){
+  await fetch('/auto',{method:'POST',headers:{'Content-Type':'application/json'},
+   body:JSON.stringify({value:on})});refresh()
+ }
  async function refresh(){
-  const s=await (await fetch('/status')).json()
+  let s; try{ s=await (await fetch('/status')).json() }catch{ return }
+  applyStatus(s)
+ }
+ function applyStatus(s){
   CAP=s.max_concurrency||8; capn.textContent=CAP
-  state.textContent=s.state; active.textContent=s.active; done.textContent=s.processed
-  err.textContent=s.errors; fc.textContent=s.fc_url; wait.textContent=s.transient||0
-  // Running but the queue read failed → curator is unreachable; show we're
-  // riding it out rather than erroring.
-  banner.style.display=(s.state==='running' && !s.queue)?'block':'none'
+  // The backend owns the state now (stopped|starting|running|stopping) and drives
+  // every transition, so the pill is always truthful — no client-side guessing
+  // from active>0, which used to wedge on "stopping" forever.
+  const st=s.state||'stopped'
+  const running=st==='running'
+  const busy=(st==='starting'||st==='stopping')
+  // Stale-page guard: if the server is a newer build than this page, the cached
+  // controls may misbehave — tell the operator to reload.
+  if(s.build && s.build!==PAGE_BUILD) verbanner.style.display='block'
+  state.textContent=st
+  state.className='n'+(busy?' busy':'')
+  // Buttons follow the real state so you can't fight a transition: Start only
+  // from stopped; Stop only while up; both disabled through "stopping" until the
+  // backend truthfully lands on "stopped".
+  startbtn.disabled=(st!=='stopped')
+  stopbtn.disabled=!(running||st==='starting')
+  // Throughput rates arrive READY from the backend (jobs/min ≈ GPU throughput,
+  // dl/min ≈ fetch throughput), computed there on a fixed cadence — so they show
+  // a real number no matter how often this tab polls (a backgrounded tab throttles
+  // its timers, which used to leave a client-side delta-rate blank forever).
+  jpm.textContent=(s.jobs_per_min!=null)?Math.round(s.jobs_per_min):'—'
+  dpm.textContent=(s.downloads_per_min!=null)?Math.round(s.downloads_per_min):'—'
+  done.textContent=s.processed
+  err.textContent=s.errors; err.className='n'+(s.errors>0?' warn':'')
+  waited.textContent=s.transient||0
+  // 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)
+  // 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+'%' }
+  // Auto on → dial reflects the auto-chosen count (read-only); off → manual.
+  if(document.activeElement!==autochk) autochk.checked=!!s.auto
+  conc.disabled=!!s.auto; conc.style.opacity=s.auto?0.55:1
+  conchint.textContent=s.auto?('auto-tuning downloaders to keep the GPU fed · max '+CAP):('manual downloaders · max '+CAP)
   if(document.activeElement!==conc) conc.value=s.concurrency
   conc.max=CAP
-  cfg.textContent=s.configured?'set':'MISSING'
-  if(s.gpu){
-   gpu.textContent=`GPU — ${s.gpu.util_pct}% util · VRAM ${s.gpu.mem_used_mb}/${s.gpu.mem_total_mb} MB · ${s.gpu.temp_c}°C`
-   gpubar.style.width=Math.round(100*s.gpu.mem_used_mb/s.gpu.mem_total_mb)+'%'
-  } else { gpu.textContent='GPU — n/a (CPU fallback?)'; gpubar.style.width='0%' }
-  queue.textContent=s.queue?`queue — pending ${s.queue.pending} · in flight ${s.queue.leased} · done ${s.queue.done} · errored ${s.queue.error}`:'queue — unreachable'
+  // Connection pill + queue come only from the /status poll (the Start/Stop POST
+  // responses skip the slow curator call to stay snappy) — guard so an action
+  // response doesn't blank them.
+  if('configured' in s){
+   const ok=s.configured
+   fc.textContent=s.fc_url; cfg.textContent=ok?'set':'MISSING'
+   // Pill colour + label track the real state: green only when running AND
+   // curator is answering; amber for the transient states + a running-but-
+   // unreachable curator; grey when stopped; red with no token.
+   let dc='dot', lbl='stopped'
+   if(!ok){ dc='dot red'; lbl='no token' }
+   else if(st==='running'){ dc='dot '+(s.queue?'green':'amber'); lbl=s.queue?'running':'running · curator unreachable' }
+   else if(st==='starting'){ dc='dot amber'; lbl='starting…' }
+   else if(st==='stopping'){ dc='dot amber'; lbl='stopping…' }
+   dot.className=dc; connlbl.textContent=lbl
+   banner.style.display=(st==='running' && !s.queue)?'block':'none'
+   queue.textContent=s.queue?('queue · pending '+s.queue.pending+' · in flight '+s.queue.leased+' · done '+s.queue.done+' · errored '+s.queue.error):'queue · unreachable'
+  }
  }
- refresh(); setInterval(refresh,3000)
+ // GPU meters poll their OWN endpoint on a fast cadence — kept off /status so a
+ // slow curator queue call can't freeze the bars (they only stale on refresh).
+ let UAVG=null   // smoothed util for the bar (raw util swings 0↔99; show the trend)
+ async function refreshGpu(){
+  let g; try{ g=await (await fetch('/gpu')).json() }catch{ return }
+  if(g && g.util_pct!=null){
+   // Prefer the agent's own EWMA (util_smooth) when running; otherwise smooth
+   // the raw reading here so a stopped agent's bar still glides, not jumps.
+   const raw=g.util_pct
+   UAVG = (g.util_smooth!=null) ? g.util_smooth
+         : (UAVG==null ? raw : 0.25*raw + 0.75*UAVG)
+   const used=g.mem_used_mb, tot=g.mem_total_mb||1
+   utillbl.textContent=Math.round(UAVG)+'% · '+g.temp_c+'°C'; utilbar.style.width=Math.round(UAVG)+'%'
+   vramlbl.textContent=used+' / '+tot+' MB'; gpubar.style.width=Math.round(100*used/tot)+'%'
+  } else { UAVG=null; utillbl.textContent='n/a'; vramlbl.textContent='n/a (CPU?)'; utilbar.style.width='0%'; gpubar.style.width='0%' }
+ }
+ async function refreshLogs(){
+  try{
+   const r=await (await fetch('/logs')).json()
+   const el=logs, atBottom=el.scrollHeight-el.scrollTop-el.clientHeight<40
+   el.textContent=(r.lines&&r.lines.length)?r.lines.join('\\n'):'waiting for activity…'
+   if(atBottom) el.scrollTop=el.scrollHeight
+  }catch{}
+ }
+ async function copyLogs(){
+  const txt=logs.textContent||''
+  try{ await navigator.clipboard.writeText(txt) }
+  catch{ const t=document.createElement('textarea'); t.value=txt; document.body.appendChild(t);
+   t.select(); try{document.execCommand('copy')}catch{}; t.remove() }
+  copybtn.textContent='Copied'; setTimeout(()=>{copybtn.textContent='Copy'},1200)
+ }
+ refresh(); refreshGpu(); refreshLogs()
+ setInterval(refresh,3000); setInterval(refreshGpu,1500); setInterval(refreshLogs,2500)
 </script></body></html>"""

agent/fc_agent/client.py

@@ -5,19 +5,69 @@ bytes, all over HTTP with the bearer token. No DB/Redis.
 """
 import requests
 from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
 
 
 class FcClient:
     def __init__(self, base_url: str, token: str, agent_id: str):
         self.base = base_url.rstrip("/")
         self.agent_id = agent_id
-        self.s = requests.Session()
-        self.s.headers["Authorization"] = f"Bearer {token}"
-        # Many worker threads share this Session; the default pool (10) would
+        # Main session: NO in-request retry — lease/fetch are cheap to redo and
+        # the worker loop already backs off + re-leases on failure. (Auto-retrying
+        # a lease could double-claim a batch if a response is lost.)
+        self.s = self._session(token)
+        # Submit session: retry in-place, because by submit time the GPU work is
+        # already DONE — a momentary blip (dropped connection, gateway 5xx during
+        # a curator redeploy) must not throw that work away and force a full
+        # re-download + recompute on another agent. A duplicate submit after a
+        # lost response is harmless: the job is already closed, so it just returns
+        # 409 lease_invalid (a no-op). Idempotent enough to retry POST safely.
+        retry = Retry(
+            total=3, connect=3, read=3, status=3,
+            backoff_factor=0.5,                       # ~0.5s, 1s, 2s between tries
+            status_forcelist=(500, 502, 503, 504),    # transient server/gateway
+            allowed_methods=frozenset({"POST"}),
+            raise_on_status=False,                     # let raise_for_status decide
+        )
+        self._submit_s = self._session(token, retry)
+
+    @staticmethod
+    def _session(token: str, retry: Retry | None = None) -> requests.Session:
+        s = requests.Session()
+        s.headers["Authorization"] = f"Bearer {token}"
+        # Many worker threads share a Session; the default pool (10) would
         # throttle them + spam "connection pool is full". Size it for the cap.
-        adapter = HTTPAdapter(pool_connections=64, pool_maxsize=64)
-        self.s.mount("http://", adapter)
-        self.s.mount("https://", adapter)
+        adapter = HTTPAdapter(
+            pool_connections=64, pool_maxsize=64, max_retries=retry or 0
+        )
+        s.mount("http://", adapter)
+        s.mount("https://", adapter)
+        return s
+
+    def _submit(self, path: str, payload: dict) -> dict:
+        """POST to a submit endpoint on the RETRYING session (by submit time the
+        GPU work is done — a blip must not throw it away), raise on a hard error,
+        and return the parsed JSON. `agent_id` is added to every body."""
+        r = self._submit_s.post(
+            f"{self.base}{path}",
+            json={"agent_id": self.agent_id, **payload},
+            timeout=120,
+        )
+        r.raise_for_status()
+        return r.json()
+
+    def _post_quiet(self, path: str, payload: dict) -> None:
+        """Fire-and-forget POST on the main session — heartbeat/fail/release are
+        best-effort, so a transport error is swallowed (the worker's own retry and
+        the server's orphan-recovery cover a lost call). `agent_id` is added."""
+        try:
+            self.s.post(
+                f"{self.base}{path}",
+                json={"agent_id": self.agent_id, **payload},
+                timeout=30,
+            )
+        except requests.RequestException:
+            pass
 
     def lease(self, batch_size: int) -> list[dict]:
         r = self.s.post(
@@ -29,57 +79,51 @@ class FcClient:
         return r.json().get("jobs", [])
 
     def submit(self, job_id: int, regions: list[dict], replace_kinds: list[str]) -> dict:
-        r = self.s.post(
-            f"{self.base}/api/gpu/jobs/submit",
-            json={
-                "agent_id": self.agent_id, "job_id": job_id,
-                "regions": regions, "replace_kinds": replace_kinds,
-            },
-            timeout=120,
-        )
-        r.raise_for_status()
-        return r.json()
+        return self._submit("/api/gpu/jobs/submit", {
+            "job_id": job_id, "regions": regions, "replace_kinds": replace_kinds,
+        })
+
+    def submit_embedding(self, job_id: int, embedding: list, version: str) -> dict:
+        """Post a whole-image SigLIP embedding (the 'embed' task) → image_record."""
+        return self._submit("/api/gpu/jobs/submit_embedding", {
+            "job_id": job_id, "embedding": embedding, "embedding_version": version,
+        })
 
     def heartbeat(self, job_ids: list[int]) -> None:
-        try:
-            self.s.post(
-                f"{self.base}/api/gpu/jobs/heartbeat",
-                json={"agent_id": self.agent_id, "job_ids": job_ids},
-                timeout=30,
-            )
-        except requests.RequestException:
-            pass
+        self._post_quiet("/api/gpu/jobs/heartbeat", {"job_ids": job_ids})
 
     def fail(self, job_id: int, error: str) -> None:
-        try:
-            self.s.post(
-                f"{self.base}/api/gpu/jobs/fail",
-                json={"agent_id": self.agent_id, "job_id": job_id, "error": error},
-                timeout=30,
-            )
-        except requests.RequestException:
-            pass
+        self._post_quiet("/api/gpu/jobs/fail", {"job_id": job_id, "error": error})
 
     def release(self, job_ids: list[int]) -> None:
         # Graceful hand-back on stop so orphaned work is re-leased at once.
         if not job_ids:
             return
-        try:
-            self.s.post(
-                f"{self.base}/api/gpu/jobs/release",
-                json={"agent_id": self.agent_id, "job_ids": job_ids},
-                timeout=30,
-            )
-        except requests.RequestException:
-            pass
+        self._post_quiet("/api/gpu/jobs/release", {"job_ids": job_ids})
 
     def fetch_image(self, image_url: str) -> bytes:
         # image_url is a server-relative path ("/images/...").
-        r = self.s.get(f"{self.base}{image_url}", timeout=180)
+        # 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
 
+    def is_reachable(self) -> bool:
+        """Cheap 'is curator responding at all right now?' check. Used to decide,
+        when a video can't be sampled, between a transient outage (keep retrying —
+        survives a redeploy) and an unprocessable file (fail it, don't loop)."""
+        try:
+            r = self.s.get(f"{self.base}/api/gpu/status", timeout=5)
+            return r.status_code < 500
+        except requests.RequestException:
+            return False
+
     def queue_status(self) -> dict:
-        r = self.s.get(f"{self.base}/api/gpu/status", timeout=15)
+        # Short timeout: this backs the UI /status poll, so a busy curator must
+        # not hang the page for long (the GPU meters poll /gpu separately).
+        r = self.s.get(f"{self.base}/api/gpu/status", timeout=5)
         r.raise_for_status()
         return r.json()

agent/fc_agent/config.py

@@ -1,8 +1,18 @@
 """Agent config, all from env (the control container is configured at run)."""
+# Lazy annotations so the `from_env(cls) -> Config` self-reference is a string,
+# not evaluated at class-definition time — otherwise it NameErrors on the agent's
+# Python 3.10 (CI lints on 3.14, where PEP 649 hides this).
+from __future__ import annotations
+
 import os
 from dataclasses import dataclass
 
 
+def _bool_env(name: str, default: str = "") -> bool:
+    """A boolean env var — present + truthy ('1'/'true'/'yes') → True."""
+    return os.environ.get(name, default).lower() in ("1", "true", "yes")
+
+
 @dataclass
 class Config:
     fc_url: str          # base URL of the FabledCurator web service
@@ -18,9 +28,29 @@ class Config:
     #                            the server announces in the lease)
     auto_start: bool     # start the worker pool on boot (so a container restart
     #                      resumes processing without anyone clicking Start)
+    auto_scale: bool     # autoscale the worker count (throughput hill-climb)
+    # Crop PROPOSERS (extra YOLO detectors that say where to crop). Each weight
+    # spec is an ultralytics name | http(s) URL | "hf_repo::file" ("" = off).
+    person_weights: str  # general COCO person detector (Western/realistic figs)
+    person_conf: float
+    anatomy_weights: str  # booru_yolo anime/furry/NSFW components
+    anatomy_conf: float
+    panel_weights: str    # comic-panel detector
+    panel_conf: float
+    max_components: int   # cap anatomy component crops per frame
+    max_panels: int       # cap panel crops per frame
+    max_figures: int      # cap figure boxes per frame (each = a CCIP call + crop)
+    max_regions: int      # hard cap on total regions per JOB (submit-size backstop)
+    dedupe_iou: float     # crops overlapping >= this (same kind) are near-dupes,
+    #                       dropped before the embed; >=1.0 disables it
+    frame_dedupe_distance: int  # video frames whose dHash differs by < this many
+    #                             bits are near-duplicates, dropped before detect;
+    #                             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
 
     @classmethod
-    def from_env(cls) -> "Config":
+    def from_env(cls) -> Config:
         return cls(
             fc_url=os.environ.get("FC_URL", "http://localhost:8000").rstrip("/"),
             token=os.environ.get("FC_TOKEN", ""),
@@ -32,5 +62,19 @@ class Config:
             poll_idle_seconds=float(os.environ.get("POLL_IDLE_SECONDS", "10")),
             embed_dtype=os.environ.get("SIGLIP_DTYPE", "float16"),
             embed_model_override=os.environ.get("EMBED_MODEL_NAME", ""),
-            auto_start=os.environ.get("AUTO_START", "").lower() in ("1", "true", "yes"),
+            auto_start=_bool_env("AUTO_START"),
+            auto_scale=_bool_env("AUTO_SCALE", "true"),
+            person_weights=os.environ.get("PERSON_WEIGHTS", "yolo11n.pt"),
+            person_conf=float(os.environ.get("PERSON_CONF", "0.35")),
+            anatomy_weights=os.environ.get("ANATOMY_WEIGHTS", ""),
+            anatomy_conf=float(os.environ.get("ANATOMY_CONF", "0.30")),
+            panel_weights=os.environ.get("PANEL_WEIGHTS", ""),
+            panel_conf=float(os.environ.get("PANEL_CONF", "0.30")),
+            max_components=int(os.environ.get("MAX_COMPONENTS", "8")),
+            max_panels=int(os.environ.get("MAX_PANELS", "8")),
+            max_figures=int(os.environ.get("MAX_FIGURES", "8")),
+            max_regions=int(os.environ.get("MAX_REGIONS", "128")),
+            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")),
         )

agent/fc_agent/detectors.py

@@ -0,0 +1,218 @@
+"""Region PROPOSERS — small YOLO detectors that decide WHERE to crop. They run
+on the agent GPU and their boxes feed the crop → SigLIP → max-over-bag pipeline:
+
+  - person (general COCO yolo11n): full-figure boxes for realistic / Western art
+    the anime person-detector misses; NMS-merged with imgutils detect_person and
+    fed to CCIP (identity) + a concept crop.
+  - anatomy (booru_yolo): anime / furry / NSFW torso components (head, cat-head,
+    boob, hip, …) — concept crops aligned to the operator's tag vocabulary.
+  - panel (mosesb): a comic page → panel regions → concept crops.
+
+Each proposer is INDEPENDENTLY optional + guarded: a bad weight path or an
+inference error disables just that proposer (logged) and never breaks the
+worker, which still falls back to imgutils detection. Weights resolve from an
+ultralytics builtin name ("yolo11n.pt"), an http(s) URL, or "hf_repo::file" —
+cached under HF_HOME so the download happens once.
+"""
+import logging
+import os
+import threading
+import types
+from pathlib import Path
+
+log = logging.getLogger("fc_agent.detectors")
+_CACHE = Path(os.environ.get("HF_HOME", "/models")) / "yolo"
+
+
+def _resolve(spec: str) -> str | None:
+    """A local weights path (downloading if needed) or an ultralytics builtin
+    name. None if the spec is empty/unresolvable."""
+    if not spec:
+        return None
+    if "::" in spec:                                    # hf_repo::filename
+        repo, _, fname = spec.partition("::")
+        from huggingface_hub import hf_hub_download
+        return hf_hub_download(
+            repo_id=repo, filename=fname, cache_dir=str(_CACHE)
+        )
+    if spec.startswith(("http://", "https://")):
+        _CACHE.mkdir(parents=True, exist_ok=True)
+        dest = _CACHE / spec.rsplit("/", 1)[-1]
+        if not dest.is_file():
+            import requests
+            r = requests.get(spec, timeout=300)
+            r.raise_for_status()
+            dest.write_bytes(r.content)
+        return str(dest)
+    return spec                                         # ultralytics builtin name
+
+
+def _iou(a, b) -> float:
+    ax, ay, aw, ah = a
+    bx, by, bw, bh = b
+    ix = max(0.0, min(ax + aw, bx + bw) - max(ax, bx))
+    iy = max(0.0, min(ay + ah, by + bh) - max(ay, by))
+    inter = ix * iy
+    union = aw * ah + bw * bh - inter
+    return inter / union if union > 0 else 0.0
+
+
+def nms_merge(boxes, iou_thresh: float = 0.6):
+    """Greedy NMS over (bbox_norm, score, label) from possibly several detectors,
+    so the same figure found by two of them collapses to one (higher-score) box."""
+    kept = []
+    for bb, sc, lb in sorted(boxes, key=lambda b: b[1], reverse=True):
+        if all(_iou(bb, k[0]) < iou_thresh for k in kept):
+            kept.append((bb, sc, lb))
+    return kept
+
+
+def dedupe_crops(pending, iou_thresh: float = 0.85):
+    """Greedy high-IoU dedupe over a list of (crop, region_template) pairs, run
+    just before the batched SigLIP embed so we never embed the same region twice.
+
+    Figure boxes are already NMS-merged and each YOLO self-NMSes, but the combined
+    per-frame pile (figure→concept ∪ anatomy component→concept ∪ panel) can still
+    carry genuine near-duplicates across proposers — e.g. a figure box that nearly
+    coincides with an anatomy component on a solo bust, or overlapping booru head
+    classes on one head. Those embed the same region twice, wasting GPU and a slot
+    against max_regions.
+
+    Boxes are compared ONLY within the same output kind and dropped when they
+    overlap at >= iou_thresh, keeping the highest-scoring one. The HIGH default
+    threshold is deliberate: it collapses only true near-identical boxes while
+    preserving intentional nested crops across scopes (a whole figure vs a small
+    head component sit well below it) and distinct kinds (concept vs panel). A
+    value >= 1.0 effectively disables it (nothing but an exact box matches)."""
+    kept = []
+    kept_boxes: dict = {}   # kind -> [bbox, ...] already kept
+    for crop, tmpl in sorted(
+        pending, key=lambda p: p[1].get("score") or 0.0, reverse=True
+    ):
+        bb = tmpl.get("bbox")
+        prior = kept_boxes.setdefault(tmpl.get("kind"), [])
+        if bb is not None and any(_iou(bb, kb) >= iou_thresh for kb in prior):
+            continue
+        prior.append(bb)
+        kept.append((crop, tmpl))
+    return kept
+
+
+class YoloProposer:
+    """One lazily-loaded ultralytics YOLO. detect(image) → [(bbox_norm, score,
+    label)] with bbox normalized (x, y, w, h) in [0,1]. Self-disables on any
+    load/inference failure."""
+
+    def __init__(self, name, weights, conf=0.25, keep_labels=None):
+        self.name = name
+        self._spec = weights
+        self._conf = conf
+        self._keep = [k.lower() for k in keep_labels] if keep_labels else None
+        self._model = None
+        self._ok = True
+        self._lock = threading.Lock()
+
+    def _load(self):
+        if self._model is not None or not self._ok:
+            return
+        with self._lock:
+            if self._model is not None or not self._ok:
+                return
+            try:
+                from ultralytics import YOLO
+                path = _resolve(self._spec)
+                if path is None:
+                    self._ok = False
+                    return
+                self._model = YOLO(path)
+                # Disable ultralytics' load-time Conv+BN fusion. AutoBackend fuses
+                # the graph on the first predict; some checkpoints (yolo11n, the
+                # comic-panel model) crash that step with "'Conv' object has no
+                # attribute 'bn'" (a partially-fused / version-mismatched graph),
+                # which silently disabled those proposers (operator-flagged
+                # 2026-07-01). Unfused inference is correct — only marginally
+                # slower — and this is robust across ultralytics versions; if a
+                # future version ignores the override, the detect() guard below
+                # still self-disables the proposer instead of spamming per image.
+                inner = getattr(self._model, "model", None)
+                if inner is not None:
+                    inner.fuse = types.MethodType(lambda self, *a, **k: self, inner)
+                log.info("detector %s loaded (%s)", self.name, path)
+            except Exception as exc:   # noqa: BLE001
+                log.warning("detector %s disabled (load failed): %s", self.name, exc)
+                self._ok = False
+
+    def detect(self, image):
+        self._load()
+        if self._model is None:
+            return []
+        try:
+            res = self._model.predict(image, conf=self._conf, verbose=False)[0]
+        except Exception as exc:   # noqa: BLE001
+            # Permanently self-disable on the FIRST inference failure rather than
+            # re-throwing (and re-logging) on every image forever — an unfixable
+            # model fault degrades to "this proposer is off", logged once.
+            log.warning("detector %s disabled (inference failed): %s", self.name, exc)
+            self._ok = False
+            self._model = None
+            return []
+        iw, ih = image.size
+        names = getattr(res, "names", None) or {}
+        out = []
+        for b in res.boxes:
+            label = str(names.get(int(b.cls), int(b.cls))).lower()
+            if self._keep is not None and not any(k in label for k in self._keep):
+                continue
+            x0, y0, x1, y1 = (float(v) for v in b.xyxy[0].tolist())
+            out.append((
+                (x0 / iw, y0 / ih, (x1 - x0) / iw, (y1 - y0) / ih),
+                float(b.conf), label,
+            ))
+        return out
+
+
+class Proposers:
+    """The agent's proposer set, built from config. Each detector is optional —
+    an empty weight spec leaves that proposer off."""
+
+    def __init__(self, cfg):
+        self.cfg = cfg
+        self._person = (
+            YoloProposer("person-coco", cfg.person_weights,
+                         conf=cfg.person_conf, keep_labels=["person"])
+            if cfg.person_weights else None
+        )
+        self._anatomy = (
+            YoloProposer("anatomy", cfg.anatomy_weights, conf=cfg.anatomy_conf)
+            if cfg.anatomy_weights else None
+        )
+        self._panel = (
+            YoloProposer("panel", cfg.panel_weights, conf=cfg.panel_conf)
+            if cfg.panel_weights else None
+        )
+
+    def figures(self, image, base_boxes):
+        """Merge imgutils person boxes (base_boxes: [(bbox, score)]) with the
+        general COCO person detector → NMS'd figure boxes [(bbox, score, label)],
+        capped to the highest-scoring max_figures. Uncapped, a busy/huge image
+        (many characters) yields hundreds of boxes → hundreds of per-figure CCIP
+        calls + crops → a 30s+ job and an oversized submit (operator-flagged)."""
+        boxes = [(bb, sc if sc is not None else 1.0, "person") for bb, sc in base_boxes]
+        if self._person is not None:
+            boxes += self._person.detect(image)
+        return nms_merge(boxes)[: self.cfg.max_figures]   # nms_merge is score-desc
+
+    @staticmethod
+    def _top(detector, image, cap: int):
+        """Top-`cap` detections by score from an optional proposer (None → the
+        proposer is off → []). Shared by the anatomy + panel proposers, which
+        differ only in which detector and which cap."""
+        if detector is None:
+            return []
+        return sorted(detector.detect(image), key=lambda b: b[1], reverse=True)[:cap]
+
+    def components(self, image):
+        return self._top(self._anatomy, image, self.cfg.max_components)
+
+    def panels(self, image):
+        return self._top(self._panel, image, self.cfg.max_panels)

agent/fc_agent/embedder.py

@@ -58,12 +58,20 @@ class CropEmbedder:
 
     def embed(self, image: Image.Image) -> list[float]:
         """A crop → its embedding as a plain float list, ready to POST."""
+        return self.embed_batch([image])[0]
+
+    def embed_batch(self, images: list) -> list[list[float]]:
+        """Embed many crops in ONE forward pass — far better GPU utilisation +
+        only one lock acquisition than embedding each crop separately (which
+        starved the GPU and serialised the whole pool)."""
+        if not images:
+            return []
         self.load()
         torch = self._torch
-        enc = self._processor(images=image, return_tensors="pt")
+        enc = self._processor(images=images, return_tensors="pt")
         pixel_values = enc["pixel_values"].to(self._device, self._dt)
         with self._infer_lock, torch.no_grad():
             out = self._model.get_image_features(pixel_values=pixel_values)
         pooled = out.pooler_output if hasattr(out, "pooler_output") else out
-        vec = pooled[0].float().cpu().numpy().astype(np.float32).reshape(-1)
-        return vec.tolist()
+        arr = pooled.float().cpu().numpy().astype(np.float32)
+        return [row.reshape(-1).tolist() for row in arr]

agent/fc_agent/gpu.py

@@ -1,10 +1,24 @@
 """GPU load readout via nvidia-smi (present in the container thanks to the
 NVIDIA Container Toolkit's `utility` capability). Returns None if unavailable —
-the UI just shows n/a (e.g. CPU-fallback run)."""
+the UI just shows n/a (e.g. CPU-fallback run).
+
+Reads are CACHED and de-duplicated: the UI meter polls fast, /status reads it,
+and the autoscaler samples it — if each spawned its own `nvidia-smi` (slow on a
+busy GPU) those blocking subprocesses would pile up in the server's thread pool
+and make the Start/Stop buttons feel dead. So a short TTL serves recent callers
+from cache, and only ONE probe runs at a time (others get the last value)."""
 import subprocess
+import threading
+import time
+
+_TTL = 1.0                       # seconds a sample is reused before re-probing
+_lock = threading.Lock()
+_cache: dict | None = None
+_cache_t = 0.0
+_probing = False
 
 
-def read_gpu() -> dict | None:
+def _probe() -> dict | None:
     try:
         out = subprocess.run(
             [
@@ -28,3 +42,24 @@ def read_gpu() -> dict | None:
         }
     except (ValueError, IndexError):
         return None
+
+
+def read_gpu(max_age: float = _TTL) -> dict | None:
+    """Latest GPU reading, cached. Serves from cache when fresh; when stale,
+    exactly one caller re-probes while the rest get the last value — so request
+    threads never block behind more than one `nvidia-smi`."""
+    global _cache, _cache_t, _probing
+    now = time.monotonic()
+    with _lock:
+        fresh = _cache is not None and (now - _cache_t) < max_age
+        if fresh or _probing:        # fresh, or a probe is already running
+            return _cache
+        _probing = True
+    try:
+        val = _probe()
+    finally:
+        with _lock:
+            _cache = val
+            _cache_t = time.monotonic()
+            _probing = False
+    return val

agent/fc_agent/logbuf.py

@@ -0,0 +1,44 @@
+"""In-memory log ring buffer so the control UI can show recent agent logs
+(detector loads, job errors, autoscaler decisions, outage back-offs) without
+needing `docker logs`. A bounded deque holds the last N formatted lines; a
+logging.Handler appends to it; the UI polls /logs."""
+import logging
+from collections import deque
+
+LINES: deque[str] = deque(maxlen=400)
+
+
+class RingHandler(logging.Handler):
+    def emit(self, record: logging.LogRecord) -> None:
+        try:
+            LINES.append(self.format(record))
+        except Exception:
+            pass
+
+
+_installed = False
+
+
+def install(level: int = logging.INFO) -> None:
+    """Attach the ring handler to the root logger once. fc_agent module loggers
+    propagate to root, so their records land here."""
+    global _installed
+    if _installed:
+        return
+    _installed = True
+    h = RingHandler()
+    h.setFormatter(
+        logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s", "%H:%M:%S")
+    )
+    root = logging.getLogger()
+    root.addHandler(h)
+    if root.level == logging.NOTSET or root.level > level:
+        root.setLevel(level)
+    # Keep the buffer signal-rich: silence the chatty HTTP/download libs (every
+    # HF model fetch logs per-request) so the console shows agent activity —
+    # detector loads, job errors, autoscale moves — not request spam.
+    for noisy in (
+        "uvicorn.access", "ultralytics", "httpx", "httpcore",
+        "huggingface_hub", "urllib3", "filelock",
+    ):
+        logging.getLogger(noisy).setLevel(logging.WARNING)

agent/fc_agent/media.py

@@ -5,14 +5,68 @@ import io
 import os
 import subprocess
 import tempfile
+import time
 
-from PIL import Image
+from PIL import Image, ImageFile
+
+# Load slightly-truncated images (a few missing trailing bytes) instead of
+# raising — matches the server embedder. These are common in scraped libraries
+# and would otherwise fail the job 3× then error (operator-flagged 2026-06-30).
+ImageFile.LOAD_TRUNCATED_IMAGES = True
+
+# Disable PIL's decompression-bomb guard: this is a TRUSTED local library, not an
+# untrusted upload surface, so a legitimately huge image (high-res scans/prints,
+# 90M+ pixels) must load. The default 89M-pixel limit only WARNS, but PIL raises
+# DecompressionBombError at 2× (~179M px) — which would fail those jobs outright
+# (operator-flagged 2026-06-30, images of 90–95M px).
+Image.MAX_IMAGE_PIXELS = None
 
 
 def is_video(mime: str) -> bool:
     return bool(mime) and (mime.startswith("video/") or mime in {"image/gif"})
 
 
+def _dhash(img: Image.Image, size: int = 8) -> int:
+    """Difference hash: compare adjacent pixels of a (size+1 × size) grayscale
+    thumbnail → a `size*size`-bit fingerprint. Cheap (64 comparisons on a 72-px
+    thumbnail) and robust to scaling/compression noise — near-identical frames
+    hash within a few bits, a real scene change moves many."""
+    small = img.convert("L").resize((size + 1, size))
+    px = list(small.getdata())
+    bits = 0
+    for row in range(size):
+        base = row * (size + 1)
+        for col in range(size):
+            bits = (bits << 1) | int(px[base + col] > px[base + col + 1])
+    return bits
+
+
+def dedupe_frames(
+    frames: list[tuple[float, Image.Image]], min_distance: int
+) -> list[tuple[float, Image.Image]]:
+    """Drop visually near-duplicate frames. A near-static video sampled into many
+    frames re-runs the WHOLE detect→CCIP→SigLIP chain on ~identical frames — the
+    dominant video load. Greedy perceptual-hash dedup: keep a frame only if its
+    dHash differs from every already-kept frame by >= min_distance bits (Hamming),
+    so a static run collapses to one frame while genuinely distinct scenes all
+    survive. Order + timestamps preserved. CPU-only (64-bit int XORs), so it runs
+    in the decode stage and spares the GPU the skipped frames entirely.
+
+    min_distance is the coarseness dial: higher keeps more frames (safer for brief
+    localized changes an 8×8 hash can miss), 0 disables. The first frame is always
+    kept (nothing to compare against)."""
+    if min_distance <= 0 or len(frames) <= 1:
+        return frames
+    kept: list[tuple[float, Image.Image]] = []
+    hashes: list[int] = []
+    for t, frame in frames:
+        h = _dhash(frame)
+        if all(bin(h ^ k).count("1") >= min_distance for k in hashes):
+            hashes.append(h)
+            kept.append((t, frame))
+    return kept
+
+
 def to_rgb(img: Image.Image) -> Image.Image:
     """RGB, flattening any transparency onto white first. A naive convert('RGB')
     on a palette-with-transparency image (common for character PNGs on a clear
@@ -32,32 +86,77 @@ def load_image(data: bytes) -> Image.Image:
     return to_rgb(Image.open(io.BytesIO(data)))
 
 
-def sample_frames(
-    data: bytes, interval_seconds: float, max_frames: int
+# ffmpeg reconnect flags — resume a dropped HTTP transfer (a slow/contended media
+# store can cut a long stream) instead of failing the whole job. Relies only on
+# HTTP + Range, which every FC deployment serves → environment-agnostic.
+_RECONNECT = [
+    "-reconnect", "1", "-reconnect_streamed", "1",
+    "-reconnect_on_network_error", "1", "-reconnect_delay_max", "5",
+]
+
+
+def _collect_frames(
+    tmp: str, interval: float, cap: int
 ) -> list[tuple[float, Image.Image]]:
-    """Extract up to max_frames frames at one-every-interval_seconds via ffmpeg.
-    Returns [(timestamp_seconds, frame)]. Empty on failure (caller falls back)."""
+    out: list[tuple[float, Image.Image]] = []
+    names = sorted(n for n in os.listdir(tmp) if n.startswith("f_"))
+    for i, name in enumerate(names[:cap]):
+        with Image.open(os.path.join(tmp, name)) as im:
+            out.append((round(i * interval, 2), to_rgb(im)))
+    return out
+
+
+def _terminate(proc: subprocess.Popen) -> None:
+    """Stop an ffmpeg cleanly, then hard-kill if it ignores SIGTERM."""
+    proc.terminate()
+    try:
+        proc.wait(timeout=2)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        try:
+            proc.wait(timeout=2)
+        except subprocess.TimeoutExpired:
+            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]]:
+    """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
+    RAM and get cut off mid-download). Reconnect flags resume a dropped transfer;
+    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."""
     interval = max(0.5, float(interval_seconds or 4.0))
     cap = max(1, int(max_frames or 64))
+    hdr = ["-headers", headers] if headers else []
     with tempfile.TemporaryDirectory() as tmp:
-        src = os.path.join(tmp, "in")
-        with open(src, "wb") as fh:
-            fh.write(data)
         pattern = os.path.join(tmp, "f_%05d.jpg")
+        cmd = ["ffmpeg", "-nostdin", "-loglevel", "error", *_RECONNECT, *hdr,
+               "-i", url, "-vf", f"fps=1/{interval}", "-frames:v", str(cap),
+               "-q:v", "3", pattern]
         try:
-            subprocess.run(
-                [
-                    "ffmpeg", "-nostdin", "-loglevel", "error", "-i", src,
-                    "-vf", f"fps=1/{interval}", "-frames:v", str(cap),
-                    "-q:v", "3", pattern,
-                ],
-                check=True, timeout=600,
+            proc = subprocess.Popen(
+                cmd, stdin=subprocess.DEVNULL,
+                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
             )
-        except (subprocess.SubprocessError, FileNotFoundError):
+        except (OSError, ValueError):
             return []
-        out: list[tuple[float, Image.Image]] = []
-        names = sorted(n for n in os.listdir(tmp) if n.startswith("f_"))
-        for i, name in enumerate(names[:cap]):
-            with Image.open(os.path.join(tmp, name)) as im:
-                out.append((round(i * interval, 2), to_rgb(im)))
-        return out
+        # 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()
+        while True:
+            try:
+                proc.wait(timeout=0.5)
+                break
+            except subprocess.TimeoutExpired:
+                if (should_stop and should_stop()) or (time.monotonic() - start > timeout):
+                    _terminate(proc)
+                    return []
+        if proc.returncode != 0:
+            return []
+        return _collect_frames(tmp, interval, cap)

agent/requirements.txt

@@ -7,6 +7,9 @@ onnxruntime-gpu
 # Dockerfile from the CUDA-12.4 wheel index so the GPU build is deterministic;
 # transformers loads whatever SigLIP-family model the server announces.
 transformers>=4.45
+# Crop PROPOSERS — small YOLO detectors (booru_yolo anatomy, COCO person, comic
+# panel) that decide where to crop. Uses the torch already installed above.
+ultralytics>=8.3
 # Control surface + HTTP.
 fastapi
 uvicorn[standard]

agent/ruff.toml

@@ -0,0 +1,7 @@
+# The agent runs on the CUDA base image's Python 3.12 (Ubuntu 24.04) — NOT the
+# 3.14 that CI's ci-python image and the repo-root ruff.toml target. Pin the
+# agent to py312 so ruff enforces 3.12 compatibility and never auto-applies a
+# 3.14-only fix (e.g. unquoting a self-referential annotation, which PEP 649
+# makes safe on 3.14 but NameErrors on 3.12). Inherit the root lint rules.
+extend = "../ruff.toml"
+target-version = "py312"

alembic/versions/0065_embedder_model_name.py

@@ -0,0 +1,35 @@
+"""ml_settings: embedder_model_name (#1190 operator model swap)
+
+The embedder MODEL VERSION was already a setting (and stamps image_record.
+siglip_model_version); the HF model NAME was env-only, so an operator couldn't
+actually point the pipeline at a different embedder. Storing the name as a
+setting makes the model an operator choice: set name + version → re-embed (the
+GPU agent) → retrain heads. Default = the current SigLIP so400m.
+
+Revision ID: 0065
+Revises: 0064
+Create Date: 2026-06-30
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0065"
+down_revision: Union[str, None] = "0064"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "ml_settings",
+        sa.Column(
+            "embedder_model_name", sa.String(length=128), nullable=False,
+            server_default="google/siglip-so400m-patch14-384",
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("ml_settings", "embedder_model_name")

alembic/versions/0066_drop_centroids.py

@@ -0,0 +1,57 @@
+"""drop the dead per-tag centroid subsystem (#1189 cleanup)
+
+The v2 pivot replaced per-tag SigLIP centroids with learned heads + CCIP.
+Nothing read the centroids anymore — they were recomputed (on merge + a daily
+beat) but never consumed for suggestions or auto-apply. Remove the storage +
+its two now-unused settings columns. (The recompute tasks, beat, endpoint,
+service, and UI card are removed in the same change.)
+
+Revision ID: 0066
+Revises: 0065
+Create Date: 2026-06-30
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0066"
+down_revision: Union[str, None] = "0065"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.drop_table("tag_reference_embedding")
+    op.drop_column("ml_settings", "centroid_similarity_threshold")
+    op.drop_column("ml_settings", "min_reference_images")
+
+
+def downgrade() -> None:
+    op.add_column(
+        "ml_settings",
+        sa.Column(
+            "min_reference_images", sa.Integer(), nullable=False,
+            server_default="5",
+        ),
+    )
+    op.add_column(
+        "ml_settings",
+        sa.Column(
+            "centroid_similarity_threshold", sa.Float(), nullable=False,
+            server_default="0.55",
+        ),
+    )
+    op.create_table(
+        "tag_reference_embedding",
+        sa.Column("tag_id", sa.Integer(), nullable=False),
+        sa.Column("embedding", sa.LargeBinary(), nullable=False),
+        sa.Column("reference_count", sa.Integer(), nullable=False),
+        sa.Column("model_version", sa.String(length=128), nullable=False),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True),
+            server_default=sa.func.now(), nullable=False,
+        ),
+        sa.ForeignKeyConstraint(["tag_id"], ["tag.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("tag_id"),
+    )

alembic/versions/0067_retire_camie_allowlist.py

@@ -0,0 +1,66 @@
+"""retire the Camie tagger + allowlist bulk-apply (#1189)
+
+The v2 pivot made heads + CCIP the tag source and head auto-apply the earned
+propagation. The Camie tagger ran only to feed the allowlist bulk-apply (its
+predictions had no other consumer), and the allowlist was a second, un-earned
+auto-apply path parallel to heads. Both are retired — drop their storage.
+
+(image_prediction = Camie's per-image predictions; tag_allowlist = the bulk-
+apply allowlist. Nothing references INTO these tables, so the drop is clean.)
+
+Revision ID: 0067
+Revises: 0066
+Create Date: 2026-06-30
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0067"
+down_revision: Union[str, None] = "0066"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.drop_table("image_prediction")
+    op.drop_table("tag_allowlist")
+
+
+def downgrade() -> None:
+    op.create_table(
+        "tag_allowlist",
+        sa.Column("tag_id", sa.Integer(), nullable=False),
+        sa.Column(
+            "min_confidence", sa.Float(), nullable=False, server_default="0.9"
+        ),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True),
+            server_default=sa.func.now(), nullable=False,
+        ),
+        sa.ForeignKeyConstraint(["tag_id"], ["tag.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("tag_id"),
+        sa.CheckConstraint(
+            "min_confidence >= 0 AND min_confidence <= 1",
+            name="ck_tag_allowlist_confidence_range",
+        ),
+    )
+    op.create_table(
+        "image_prediction",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("image_record_id", sa.Integer(), nullable=False),
+        sa.Column("raw_name", sa.String(length=255), nullable=False),
+        sa.Column("category", sa.String(length=32), nullable=False),
+        sa.Column("score", sa.Float(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["image_record_id"], ["image_record.id"], ondelete="CASCADE"
+        ),
+    )
+    op.create_index(
+        "ix_image_prediction_image", "image_prediction", ["image_record_id"]
+    )
+    op.create_index(
+        "ix_image_prediction_name_score", "image_prediction",
+        ["raw_name", "score"],
+    )

alembic/versions/0068_drop_dead_tagger_settings.py

@@ -0,0 +1,80 @@
+"""drop dead tagger/suggestion settings + columns left after Camie retirement (#1199)
+
+Hygiene follow-up to #1189. These were left inert to bound that change; nothing
+reads them now:
+- ml_settings: tagger_store_floor + tagger_model_version (only the deleted Camie
+  tagger used them), suggestion_threshold_character/general (already dead pre-
+  retirement — scoring uses per-head thresholds), video_min_tag_frames (only the
+  deleted video-prediction aggregator used it).
+- image_record: tagger_model_version (no writer now), centroid_scores (long-dead
+  JSON cache, no reader).
+
+Revision ID: 0068
+Revises: 0067
+Create Date: 2026-06-30
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0068"
+down_revision: Union[str, None] = "0067"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.drop_column("ml_settings", "suggestion_threshold_character")
+    op.drop_column("ml_settings", "suggestion_threshold_general")
+    op.drop_column("ml_settings", "tagger_store_floor")
+    op.drop_column("ml_settings", "video_min_tag_frames")
+    op.drop_column("ml_settings", "tagger_model_version")
+    op.drop_column("image_record", "tagger_model_version")
+    op.drop_column("image_record", "centroid_scores")
+
+
+def downgrade() -> None:
+    op.add_column(
+        "image_record",
+        sa.Column("centroid_scores", sa.JSON(), nullable=True),
+    )
+    op.add_column(
+        "image_record",
+        sa.Column("tagger_model_version", sa.String(length=128), nullable=True),
+    )
+    op.add_column(
+        "ml_settings",
+        sa.Column(
+            "tagger_model_version", sa.String(length=128), nullable=False,
+            server_default="camie-tagger-v2",
+        ),
+    )
+    op.add_column(
+        "ml_settings",
+        sa.Column(
+            "video_min_tag_frames", sa.Integer(), nullable=False,
+            server_default="3",
+        ),
+    )
+    op.add_column(
+        "ml_settings",
+        sa.Column(
+            "tagger_store_floor", sa.Float(), nullable=False,
+            server_default="0.7",
+        ),
+    )
+    op.add_column(
+        "ml_settings",
+        sa.Column(
+            "suggestion_threshold_general", sa.Float(), nullable=False,
+            server_default="0.7",
+        ),
+    )
+    op.add_column(
+        "ml_settings",
+        sa.Column(
+            "suggestion_threshold_character", sa.Float(), nullable=False,
+            server_default="0.7",
+        ),
+    )

alembic/versions/0069_default_siglip2.py

@@ -0,0 +1,51 @@
+"""default the embedder to SigLIP 2 — for FRESH installs only (#1203)
+
+Make SigLIP 2 (so400m, 512px; a 1152-d drop-in) the default embedder. New
+installs start on it. An EXISTING library is NOT touched: flipping its stored
+embedder version would mark every embedding stale (the scorer is version-gated)
+and kill suggestions until a full re-embed+retrain — so an existing instance
+switches deliberately via Settings → GPU agent → Embedding model → Re-embed →
+Retrain. We detect "fresh" by the absence of any embedded image.
+
+Revision ID: 0069
+Revises: 0068
+Create Date: 2026-06-30
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "0069"
+down_revision: Union[str, None] = "0068"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+_NEW_NAME = "google/siglip2-so400m-patch16-512"
+_NEW_VERSION = "siglip2-so400m-patch16-512"
+_OLD_NAME = "google/siglip-so400m-patch14-384"
+_OLD_VERSION = "siglip-so400m-patch14-384"
+
+
+def upgrade() -> None:
+    # Fresh install (nothing embedded yet) → adopt SigLIP 2.
+    op.execute(
+        f"""
+        UPDATE ml_settings SET
+            embedder_model_name = '{_NEW_NAME}',
+            embedder_model_version = '{_NEW_VERSION}'
+        WHERE NOT EXISTS (
+            SELECT 1 FROM image_record WHERE siglip_embedding IS NOT NULL
+        )
+        """
+    )
+    op.alter_column("ml_settings", "embedder_model_name", server_default=_NEW_NAME)
+    op.alter_column(
+        "ml_settings", "embedder_model_version", server_default=_NEW_VERSION
+    )
+
+
+def downgrade() -> None:
+    op.alter_column("ml_settings", "embedder_model_name", server_default=_OLD_NAME)
+    op.alter_column(
+        "ml_settings", "embedder_model_version", server_default=_OLD_VERSION
+    )

alembic/versions/0070_gpu_job_lease_indexes.py

@@ -0,0 +1,44 @@
+"""partial indexes so GPU-job leasing stays O(batch), not O(completed)
+
+The lease claims the lowest-id pending (or expired-leased) jobs. With only a
+plain `status` index, `... ORDER BY id LIMIT n` walked the primary-key index from
+the start, skipping the entire prefix of already-done/error rows before reaching
+pending ones — so leasing slowed to a crawl as `done` piled up (the whole reason
+throughput fell off a cliff mid-run and /status stalled). Two partial indexes fix
+it: the pending one is id-ordered so the hot path reads just the first n entries,
+and the leased-expiry one keeps the crash-recovery reclaim + the orphan sweep
+cheap. They cover only the small live slice of the table, so they stay tiny even
+as the done/error history grows to millions.
+
+Revision ID: 0070
+Revises: 0069
+Create Date: 2026-06-30
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0070"
+down_revision: Union[str, None] = "0069"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Hot path: lowest-id pending jobs. Index on id, restricted to pending, so
+    # `WHERE status='pending' ORDER BY id LIMIT n` is a short index-order scan.
+    op.create_index(
+        "ix_gpu_job_pending", "gpu_job", ["id"],
+        postgresql_where=sa.text("status = 'pending'"),
+    )
+    # Crash-recovery: expired leases, for the lease backstop + recover_orphaned.
+    op.create_index(
+        "ix_gpu_job_leased_expires", "gpu_job", ["lease_expires_at"],
+        postgresql_where=sa.text("status = 'leased'"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_gpu_job_leased_expires", table_name="gpu_job")
+    op.drop_index("ix_gpu_job_pending", table_name="gpu_job")

alembic/versions/0071_image_record_earliest_post_date.py

@@ -0,0 +1,80 @@
+"""image_record.earliest_post_date: original-publish gallery sort key + index
+
+Revision ID: 0071
+Revises: 0070
+Create Date: 2026-07-01
+
+effective_date (0035) keys off the PRIMARY post — which is often the repost /
+download the file actually came from — and falls back to created_at, so the
+gallery's default order surfaces download dates rather than when content was
+first posted (operator-flagged 2026-07-01). Materialize a second sort key,
+earliest_post_date = MIN(post_date) across ALL of an image's provenance posts
+(every post it appears in), falling back to created_at only when no linked post
+carries a date. Indexed (DESC, id DESC) so the "post date" gallery sort is an
+index range scan just like effective_date.
+
+Backfill mirrors 0035: created_at baseline, then override with the MIN over
+image_provenance ⋈ post. New rows get the created_at-equivalent server default;
+services/importer.py recomputes it whenever a dated post is linked.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0071"
+down_revision: Union[str, None] = "0070"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Add nullable first so the backfill can populate before NOT NULL.
+    op.add_column(
+        "image_record",
+        sa.Column("earliest_post_date", sa.DateTime(timezone=True), nullable=True),
+    )
+    # Baseline: download date. Set-based (no per-row binds) → immune to the
+    # 65535 bind-parameter ceiling regardless of library size.
+    op.execute(
+        """
+        UPDATE image_record
+        SET earliest_post_date = created_at
+        """
+    )
+    # Override with the earliest post_date across EVERY post the image appears
+    # in (image_provenance is the many-to-many edge; ignore posts with no date).
+    op.execute(
+        """
+        UPDATE image_record AS ir
+        SET earliest_post_date = sub.min_date
+        FROM (
+            SELECT ip.image_record_id AS iid, MIN(p.post_date) AS min_date
+            FROM image_provenance AS ip
+            JOIN post AS p ON p.id = ip.post_id
+            WHERE p.post_date IS NOT NULL
+            GROUP BY ip.image_record_id
+        ) AS sub
+        WHERE ir.id = sub.iid
+        """
+    )
+    op.alter_column(
+        "image_record",
+        "earliest_post_date",
+        nullable=False,
+        server_default=sa.text("now()"),
+    )
+    # DESC/DESC matches the gallery's ORDER BY earliest_post_date DESC, id DESC
+    # so the "post date" scroll is a forward index scan; raw SQL because
+    # alembic's column list doesn't express per-column DESC cleanly.
+    op.execute(
+        "CREATE INDEX ix_image_record_earliest_post_date "
+        "ON image_record (earliest_post_date DESC, id DESC)"
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_image_record_earliest_post_date", table_name="image_record"
+    )
+    op.drop_column("image_record", "earliest_post_date")

backend/app/__init__.py

@@ -34,6 +34,23 @@ def create_app() -> Quart:
     app = Quart(__name__)
     app.secret_key = cfg.secret_key
 
+    # Stream files in 4 MiB chunks instead of Quart's 8 KiB default. The image
+    # library lives on a CIFS/SMB share (mounted rsize=4 MiB), so 8 KiB reads
+    # meant ~19k network round-trips for one large original — 30–58s downloads
+    # that starved both the GPU agent and the browser (operator-flagged
+    # 2026-07-01). 4 MiB matches the mount's read size → one round-trip per read,
+    # ~500× fewer. buffer_size is the MAX read, so small thumbnails still read in
+    # a single gulp, and Range/mime/ETag/conditional handling lives on Response,
+    # so this keeps all of it. Guarded so a future Quart-internal change can't
+    # break boot — worst case we fall back to the slow default.
+    try:
+        from quart.wrappers.response import FileBody
+        FileBody.buffer_size = 4 * 1024 * 1024
+    except Exception:
+        logging.getLogger(__name__).warning(
+            "could not raise FileBody.buffer_size — file serving stays on 8 KiB chunks"
+        )
+
     for bp in all_blueprints():
         app.register_blueprint(bp)
     # Registered last so /api/* routes win over the SPA catch-all.

backend/app/api/__init__.py

@@ -16,7 +16,6 @@ api_bp.add_url_rule("/health", view_func=health.get_health, methods=["GET"])
 def all_blueprints() -> list[Blueprint]:
     from .admin import admin_bp
     from .aliases import aliases_bp
-    from .allowlist import allowlist_bp
     from .artist import artist_bp
     from .artists import artists_bp
     from .attachments import attachments_bp
@@ -58,7 +57,6 @@ def all_blueprints() -> list[Blueprint]:
         cleanup_bp,
         import_admin_bp,
         suggestions_bp,
-        allowlist_bp,
         aliases_bp,
         tag_eval_bp,
         heads_bp,

backend/app/api/allowlist.py

@@ -1,84 +0,0 @@
-"""Allowlist API: list, adjust threshold, remove."""
-
-from quart import Blueprint, jsonify, request
-
-from ..extensions import get_session
-from ..models import TagAllowlist
-from ..services.ml.allowlist import AllowlistService
-
-allowlist_bp = Blueprint("allowlist", __name__, url_prefix="/api")
-
-
-@allowlist_bp.route("/allowlist", methods=["GET"])
-async def list_allowlist():
-    async with get_session() as session:
-        rows = await AllowlistService(session).list_all()
-    return jsonify(
-        [
-            {
-                "tag_id": r.tag_id,
-                "tag_name": r.tag_name,
-                "tag_kind": r.tag_kind,
-                "min_confidence": r.min_confidence,
-                "applied_count": r.applied_count,
-                "coverage_count": r.coverage_count,
-            }
-            for r in rows
-        ]
-    )
-
-
-@allowlist_bp.route("/tags/<int:tag_id>/allowlist/coverage", methods=["GET"])
-async def coverage(tag_id: int):
-    """Live "at threshold T, a sweep would cover ~N images" projection for the
-    allowlist tuning dashboard. Defaults to the tag's stored threshold."""
-    raw = request.args.get("threshold")
-    async with get_session() as session:
-        svc = AllowlistService(session)
-        if raw is not None:
-            try:
-                threshold = float(raw)
-            except ValueError:
-                return jsonify({"error": "threshold must be a float"}), 400
-            if not (0 < threshold <= 1):
-                return jsonify({"error": "threshold must be in (0, 1]"}), 400
-        else:
-            row = await session.get(TagAllowlist, tag_id)
-            if row is None:
-                return jsonify({"error": "not on allowlist"}), 404
-            threshold = row.min_confidence
-        count = await svc.coverage(tag_id, threshold)
-    return jsonify({"count": count, "threshold": threshold})
-
-
-@allowlist_bp.route("/tags/<int:tag_id>/allowlist", methods=["GET"])
-async def get_one(tag_id: int):
-    async with get_session() as session:
-        row = await session.get(TagAllowlist, tag_id)
-    if row is None:
-        return jsonify({"error": "not on allowlist"}), 404
-    return jsonify(
-        {"min_confidence": row.min_confidence, "added_at": row.added_at.isoformat()}
-    )
-
-
-@allowlist_bp.route("/tags/<int:tag_id>/allowlist", methods=["PATCH"])
-async def patch_threshold(tag_id: int):
-    body = await request.get_json()
-    if not body or "min_confidence" not in body:
-        return jsonify({"error": "min_confidence required"}), 400
-    mc = float(body["min_confidence"])
-    if not (0 < mc <= 1):
-        return jsonify({"error": "min_confidence must be in (0, 1]"}), 400
-    async with get_session() as session:
-        await AllowlistService(session).update_threshold(tag_id, mc)
-        await session.commit()
-    return "", 204
-
-
-@allowlist_bp.route("/tags/<int:tag_id>/allowlist", methods=["DELETE"])
-async def remove(tag_id: int):
-    async with get_session() as session:
-        await AllowlistService(session).remove(tag_id)
-        await session.commit()
-    return "", 204

backend/app/api/gallery.py

@@ -44,7 +44,8 @@ def _parse_filters():
       the image must match at least one tag from EACH group (groups ANDed).
     - `tag_not` is a comma-separated exclude list (image must carry none).
 
-    `media` is image|video; `sort` is newest|oldest; `platform` selects one
+    `media` is image|video; `sort` is newest|oldest|posted_new|posted_old
+    (default posted_new); `platform` selects one
     platform (or the UNSOURCED_PLATFORM sentinel); `untagged`/`no_artist` are
     boolean flags; `date_from`/`date_to` are inclusive calendar-day bounds
     (date_to is widened by a day so the whole day is covered by the service's
@@ -67,8 +68,12 @@ def _parse_filters():
     artist_id = int(artist_id_raw) if artist_id_raw else None
     media = request.args.get("media")
     media_type = media if media in ("image", "video") else None
+    # newest/oldest key off effective_date (primary post / download); posted_new/
+    # posted_old off earliest_post_date (original publish across all posts). The
+    # default is posted_new so the grid leads with original publish date, not the
+    # download/repost the primary post points at (operator-flagged 2026-07-01).
     sort = request.args.get("sort")
-    sort = sort if sort in ("newest", "oldest") else "newest"
+    sort = sort if sort in ("newest", "oldest", "posted_new", "posted_old") else "posted_new"
     platform = request.args.get("platform") or None
     untagged = request.args.get("untagged") in ("1", "true", "yes")
     no_artist = request.args.get("no_artist") in ("1", "true", "yes")

backend/app/api/gpu.py

@@ -17,7 +17,6 @@ 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.embedder import MODEL_NAME as EMBED_MODEL_NAME
 from ..services.ml.gpu_jobs import GpuJobService
 from ..services.ml.regions import RegionService
 
@@ -97,6 +96,19 @@ async def backfill():
     return jsonify({"celery_task_id": r.id, "task": task}), 202
 
 
+@gpu_bp.route("/reprocess", methods=["POST"])
+async def reprocess():
+    """Reset every done/error job of `task` back to pending so the agent re-runs
+    the WHOLE library under the current pipeline (e.g. after adding crop
+    detectors). Heavy — the back-catalogue is otherwise skipped by the backfills."""
+    body = await request.get_json(silent=True) or {}
+    task = str(body.get("task") or "ccip")
+    from ..tasks.ml import reprocess_gpu_jobs
+
+    r = reprocess_gpu_jobs.delay(task)
+    return jsonify({"celery_task_id": r.id, "task": task}), 202
+
+
 # --- Agent (bearer token): lease / submit / heartbeat / fail ------------
 
 @gpu_bp.route("/jobs/lease", methods=["POST"])
@@ -138,11 +150,12 @@ async def lease():
                 # For video/animated: the agent samples at this cadence.
                 "frame_interval_seconds": ml.video_frame_interval_seconds,
                 "max_frames": ml.video_max_frames,
-                # The embedding model the agent must use for concept crops, so
-                # its region vectors land in the SAME space the heads trained in.
-                # Server-announced → the agent stays model-agnostic; a swap is a
-                # server setting + a re-embed migration, never an agent change.
-                "embed_model_name": EMBED_MODEL_NAME,
+                # The embedding model the agent must use for concept crops + the
+                # whole-image 'embed' task, so its vectors land in the SAME space
+                # the heads trained in. Server-announced FROM THE SETTING → the
+                # agent stays model-agnostic; an operator swap is a setting + a
+                # re-embed, never an agent change.
+                "embed_model_name": ml.embedder_model_name,
                 "embed_version": ml.embedder_model_version,
             })
     return jsonify({"jobs": out})
@@ -188,6 +201,34 @@ async def submit():
     return jsonify({"ok": True, "stored": len(regions)})
 
 
+@gpu_bp.route("/jobs/submit_embedding", methods=["POST"])
+async def submit_embedding():
+    """Store a whole-image SigLIP embedding (the 'embed' task) on image_record +
+    close the job. Body: {agent_id, job_id, embedding:[...], embedding_version}.
+    This is how the GPU agent re-embeds the library under a new model (#1190) —
+    much faster than the CPU ml-worker at higher resolutions."""
+    body = await request.get_json(silent=True) or {}
+    agent_id = str(body.get("agent_id") or "agent")
+    job_id = body.get("job_id")
+    embedding = body.get("embedding")
+    version = body.get("embedding_version")
+    if job_id is None or not embedding or not version:
+        return jsonify({"error": "job_id, embedding, embedding_version required"}), 400
+    async with get_session() as session:
+        if not await _agent_authed(session):
+            return jsonify({"error": "unauthorized"}), 401
+        job = await session.get(GpuJob, int(job_id))
+        if job is None or job.status != "leased" or job.lease_token != agent_id:
+            return jsonify({"error": "lease_invalid"}), 409
+        img = await session.get(ImageRecord, job.image_record_id)
+        if img is not None:
+            img.siglip_embedding = embedding
+            img.siglip_model_version = version
+        await GpuJobService(session).complete(agent_id, int(job_id))
+        await session.commit()
+    return jsonify({"ok": True})
+
+
 @gpu_bp.route("/jobs/fail", methods=["POST"])
 async def fail():
     body = await request.get_json(silent=True) or {}

backend/app/api/ml_admin.py

@@ -1,4 +1,4 @@
-"""ML admin API: settings, backfill trigger, centroid recompute trigger."""
+"""ML admin API: settings + backfill trigger."""
 
 from quart import Blueprint, jsonify, request
 
@@ -9,14 +9,8 @@ ml_admin_bp = Blueprint("ml_admin", __name__, url_prefix="/api/ml")
 
 
 _EDITABLE = (
-    "suggestion_threshold_character",
-    "suggestion_threshold_general",
-    "centroid_similarity_threshold",
-    "min_reference_images",
-    "tagger_store_floor",
     "video_frame_interval_seconds",
     "video_max_frames",
-    "video_min_tag_frames",
     "head_min_positives",
     "head_auto_apply_precision",
     "head_auto_apply_enabled",
@@ -24,9 +18,41 @@ _EDITABLE = (
     "ccip_match_threshold",
     "ccip_auto_apply_enabled",
     "ccip_auto_apply_threshold",
+    "embedder_model_name",
+    "embedder_model_version",
 )
 
 
+# Supported embedders for the Settings dropdown — all 1152-d so a swap is a
+# drop-in (re-embed + retrain, no schema change). Server-authoritative so the UI
+# never free-types a model name.
+SUPPORTED_EMBEDDERS = (
+    {
+        "name": "google/siglip2-so400m-patch16-512",
+        "version": "siglip2-so400m-patch16-512",
+        "label": "SigLIP 2 · so400m · 512px (recommended)",
+        "dim": 1152,
+    },
+    {
+        "name": "google/siglip2-so400m-patch16-384",
+        "version": "siglip2-so400m-patch16-384",
+        "label": "SigLIP 2 · so400m · 384px (faster)",
+        "dim": 1152,
+    },
+    {
+        "name": "google/siglip-so400m-patch14-384",
+        "version": "siglip-so400m-patch14-384",
+        "label": "SigLIP 1 · so400m · 384px (original)",
+        "dim": 1152,
+    },
+)
+
+
+@ml_admin_bp.route("/embedder-models", methods=["GET"])
+async def embedder_models():
+    return jsonify({"models": list(SUPPORTED_EMBEDDERS)})
+
+
 @ml_admin_bp.route("/settings", methods=["GET"])
 async def get_settings():
     from sqlalchemy import select
@@ -37,15 +63,8 @@ async def get_settings():
         ).scalar_one()
     return jsonify(
         {
-            "suggestion_threshold_character": s.suggestion_threshold_character,
-            "suggestion_threshold_general": s.suggestion_threshold_general,
-            "centroid_similarity_threshold": s.centroid_similarity_threshold,
-            "min_reference_images": s.min_reference_images,
-            "tagger_store_floor": s.tagger_store_floor,
             "video_frame_interval_seconds": s.video_frame_interval_seconds,
             "video_max_frames": s.video_max_frames,
-            "video_min_tag_frames": s.video_min_tag_frames,
-            "tagger_model_version": s.tagger_model_version,
             "embedder_model_version": s.embedder_model_version,
             "head_min_positives": s.head_min_positives,
             "head_auto_apply_precision": s.head_auto_apply_precision,
@@ -54,6 +73,7 @@ async def get_settings():
             "ccip_match_threshold": s.ccip_match_threshold,
             "ccip_auto_apply_enabled": s.ccip_auto_apply_enabled,
             "ccip_auto_apply_threshold": s.ccip_auto_apply_threshold,
+            "embedder_model_name": s.embedder_model_name,
         }
     )
 
@@ -89,31 +109,12 @@ async def patch_settings():
 
 
 def _validate(p: dict) -> str | None:
-    """Returns an error string if the proposed settings are invalid, else None.
-
-    Invariant (plan-task #764): the per-category suggestion thresholds can't
-    drop below tagger_store_floor — nothing below the floor is stored, so a
-    lower threshold would silently surface nothing in that gap. The UI clamps
-    the sliders to the floor; this is the server-side backstop.
-    """
-    floor = p["tagger_store_floor"]
-    if not (0.0 <= floor <= 1.0):
-        return "tagger_store_floor must be between 0 and 1"
-    for cat in ("character", "general"):
-        if p[f"suggestion_threshold_{cat}"] < floor:
-            return (
-                f"suggestion_threshold_{cat} cannot be below tagger_store_floor "
-                f"({floor}) — predictions below the floor are not stored"
-            )
-    # Video tagging (#747).
+    """Returns an error string if the proposed settings are invalid, else None."""
+    # Video embedding (#747).
     if p["video_frame_interval_seconds"] <= 0:
         return "video_frame_interval_seconds must be > 0"
     if p["video_max_frames"] < 1:
         return "video_max_frames must be >= 1"
-    if p["video_min_tag_frames"] < 1:
-        return "video_min_tag_frames must be >= 1"
-    if p["video_min_tag_frames"] > p["video_max_frames"]:
-        return "video_min_tag_frames cannot exceed video_max_frames"
     # Head training (#114).
     if int(p["head_min_positives"]) < 1:
         return "head_min_positives must be >= 1"
@@ -125,6 +126,11 @@ def _validate(p: dict) -> str | None:
         return "ccip_match_threshold must be between 0.5 and 0.999"
     if not (0.5 <= float(p["ccip_auto_apply_threshold"]) <= 0.999):
         return "ccip_auto_apply_threshold must be between 0.5 and 0.999"
+    # Embedder model swap (#1190): both must be non-empty. Changing them means a
+    # different embedding space — the operator must re-embed + retrain after.
+    for key in ("embedder_model_name", "embedder_model_version"):
+        if not str(p[key]).strip():
+            return f"{key} must not be empty"
     return None
 
 
@@ -134,11 +140,3 @@ async def trigger_backfill():
 
     r = backfill.delay()
     return jsonify({"celery_task_id": r.id}), 202
-
-
-@ml_admin_bp.route("/recompute-centroids", methods=["POST"])
-async def trigger_recompute():
-    from ..tasks.ml import recompute_centroids
-
-    r = recompute_centroids.delay()
-    return jsonify({"celery_task_id": r.id}), 202

backend/app/api/suggestions.py

@@ -3,31 +3,12 @@
 from quart import Blueprint, jsonify, request
 
 from ..extensions import get_session
-from ..models import Tag, TagAllowlist
 from ..services.ml.allowlist import AllowlistService
 from ..services.ml.suggestions import SuggestionService
 
 suggestions_bp = Blueprint("suggestions", __name__, url_prefix="/api")
 
 
-async def _accept_payload(session, svc, newly_added: bool, tag_id: int) -> dict:
-    """Shape the accept/alias response. When accepting newly allowlists a tag,
-    include the coverage PROJECTION (at the tag's threshold) so the UI can show
-    a non-blocking "auto-applying to ~N images" toast — the actual apply runs
-    async via apply_allowlist_tags, so this is an estimate, not a post-hoc
-    count (#7)."""
-    payload = {"allowlisted": newly_added}
-    if newly_added:
-        tag = await session.get(Tag, tag_id)
-        row = await session.get(TagAllowlist, tag_id)
-        payload["tag_id"] = tag_id
-        payload["tag_name"] = tag.name if tag is not None else None
-        payload["projected_count"] = await svc.coverage(
-            tag_id, row.min_confidence if row is not None else 0.90,
-        )
-    return payload
-
-
 @suggestions_bp.route("/images/<int:image_id>/suggestions", methods=["GET"])
 async def get_suggestions(image_id: int):
     # ?min=<float> overrides the configured per-category thresholds so the typed
@@ -83,15 +64,9 @@ async def accept_suggestion(image_id: int):
         return jsonify({"error": "tag_id required"}), 400
     tag_id = body["tag_id"]
     async with get_session() as session:
-        svc = AllowlistService(session)
-        newly_added = await svc.accept(image_id, tag_id)
-        payload = await _accept_payload(session, svc, newly_added, tag_id)
+        await AllowlistService(session).accept(image_id, tag_id)
         await session.commit()
-    if newly_added:
-        from ..tasks.ml import apply_allowlist_tags
-
-        apply_allowlist_tags.delay(tag_id=tag_id)
-    return jsonify(payload)
+    return jsonify({"accepted": True, "tag_id": tag_id})
 
 
 @suggestions_bp.route(
@@ -104,22 +79,14 @@ async def alias_suggestion(image_id: int):
         return jsonify({"error": f"required: {sorted(required)}"}), 400
     canonical_tag_id = body["canonical_tag_id"]
     async with get_session() as session:
-        svc = AllowlistService(session)
-        newly_added = await svc.add_alias_and_accept(
+        await AllowlistService(session).add_alias_and_accept(
             image_id,
             body["alias_string"],
             body["alias_category"],
             canonical_tag_id,
         )
-        payload = await _accept_payload(
-            session, svc, newly_added, canonical_tag_id,
-        )
         await session.commit()
-    if newly_added:
-        from ..tasks.ml import apply_allowlist_tags
-
-        apply_allowlist_tags.delay(tag_id=canonical_tag_id)
-    return jsonify(payload)
+    return jsonify({"accepted": True, "tag_id": canonical_tag_id})
 
 
 @suggestions_bp.route(

backend/app/api/tags.py

@@ -1,13 +1,14 @@
 """Tags API: autocomplete, create, list/add/remove for an image."""
 
 from quart import Blueprint, jsonify, request
-from sqlalchemy import exists, select
+from sqlalchemy import func, select
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.exc import IntegrityError
 
 from ..extensions import get_session
-from ..models import Tag, TagKind, TagPositiveConfirmation
-from ..models.tag_allowlist import TagAllowlist
+from ..models import Tag, TagHead, TagKind, TagPositiveConfirmation
+from ..models.tag import image_tag
+from ..models.tag_suggestion_rejection import TagSuggestionRejection
 from ..services.bulk_tag_service import BulkTagService
 from ..services.ml.aliases import AliasService
 from ..services.series_match_service import SeriesMatchService
@@ -61,6 +62,117 @@ def _parse_bulk_ids(
     return ids, None
 
 
+# Application-source groupings (image_tag.source). HUMAN = operator signal;
+# AUTO = machine-applied (heads/CCIP, + legacy Camie ml_auto).
+_SOURCE_GROUPS = {
+    "human": ("manual", "ml_accepted"),
+    "manual": ("manual",),
+    "accepted": ("ml_accepted",),
+    "auto": ("head_auto", "ccip_auto", "ml_auto"),
+}
+
+
+@tags_bp.route("/tags/top", methods=["GET"])
+async def tags_top():
+    """Top tags by image count — a fast indexed aggregate for ANALYSIS (not the
+    paged UI directory, which is alphabetical + builds previews). Params:
+    ?kind=general|character|fandom|… ?source=all|human|manual|accepted|auto
+    ?limit=50 (cap 500) ?min_count=N. → {tags:[{tag_id,name,kind,count}]} desc."""
+    kind = _coerce_kind(request.args.get("kind"))
+    try:
+        limit = min(max(int(request.args.get("limit", "50")), 1), 500)
+    except ValueError:
+        return jsonify({"error": "limit must be an integer"}), 400
+    min_count = None
+    if "min_count" in request.args:
+        try:
+            min_count = int(request.args["min_count"])
+        except ValueError:
+            return jsonify({"error": "min_count must be an integer"}), 400
+    src_vals = _SOURCE_GROUPS.get((request.args.get("source") or "all").lower())
+
+    cnt = func.count(image_tag.c.image_record_id)
+    stmt = (
+        select(Tag.id, Tag.name, Tag.kind, cnt.label("count"))
+        .select_from(Tag)
+        .join(image_tag, image_tag.c.tag_id == Tag.id)
+        .group_by(Tag.id, Tag.name, Tag.kind)
+        .order_by(cnt.desc(), Tag.name.asc())
+        .limit(limit)
+    )
+    if kind is not None:
+        stmt = stmt.where(Tag.kind == kind)
+    if src_vals is not None:
+        stmt = stmt.where(image_tag.c.source.in_(src_vals))
+    if min_count is not None:
+        stmt = stmt.having(cnt >= min_count)
+    async with get_session() as session:
+        rows = (await session.execute(stmt)).all()
+    return jsonify({"tags": [
+        {
+            "tag_id": r.id, "name": r.name,
+            "kind": r.kind.value if hasattr(r.kind, "value") else str(r.kind),
+            "count": r.count,
+        }
+        for r in rows
+    ]})
+
+
+@tags_bp.route("/tags/<int:tag_id>/stats", methods=["GET"])
+async def tag_stats(tag_id: int):
+    """Per-tag dataset health: total + per-source application counts (human vs
+    machine), rejection count, and whether a trained head exists. Read-only,
+    analysis-shaped — backs concept-readiness + source-split decisions."""
+    async with get_session() as session:
+        tag = await session.get(Tag, tag_id)
+        if tag is None:
+            return jsonify({"error": "not found"}), 404
+        by_source = dict(
+            (
+                await session.execute(
+                    select(image_tag.c.source, func.count())
+                    .where(image_tag.c.tag_id == tag_id)
+                    .group_by(image_tag.c.source)
+                )
+            ).all()
+        )
+        rejected = (
+            await session.execute(
+                select(func.count())
+                .select_from(TagSuggestionRejection)
+                .where(TagSuggestionRejection.tag_id == tag_id)
+            )
+        ).scalar_one()
+        has_head = (
+            await session.execute(
+                select(func.count())
+                .select_from(TagHead)
+                .where(TagHead.tag_id == tag_id)
+            )
+        ).scalar_one() > 0
+    human = by_source.get("manual", 0) + by_source.get("ml_accepted", 0)
+    auto = (
+        by_source.get("head_auto", 0)
+        + by_source.get("ccip_auto", 0)
+        + by_source.get("ml_auto", 0)
+    )
+    return jsonify({
+        "tag_id": tag_id,
+        "name": tag.name,
+        "kind": tag.kind.value if hasattr(tag.kind, "value") else str(tag.kind),
+        "count_total": sum(by_source.values()),
+        "count_human": human,
+        "count_manual": by_source.get("manual", 0),
+        "count_accepted": by_source.get("ml_accepted", 0),
+        "count_auto": auto,
+        "count_head_auto": by_source.get("head_auto", 0),
+        "count_ccip_auto": by_source.get("ccip_auto", 0),
+        "count_rejected": rejected,
+        "by_source": by_source,
+        "has_head": has_head,
+    })
+
+
 @tags_bp.route("/tags/autocomplete", methods=["GET"])
 async def autocomplete():
     q = request.args.get("q", "")
@@ -297,19 +409,6 @@ async def merge_tag(source_id: int):
             status = 404 if "not found" in msg else 400
             return jsonify({"error": msg}), status
         await session.commit()
-        target_allowlisted = await session.scalar(
-            select(exists().where(TagAllowlist.tag_id == result.target_id))
-        )
-    if target_allowlisted:
-        from ..tasks.ml import apply_allowlist_tags
-
-        apply_allowlist_tags.delay(tag_id=result.target_id)
-    # Tag merge invalidates the target's centroid (the merged-in source
-    # tag's images now contribute to it). Daily list_drifted catches it
-    # within 24h, but eager recompute closes the suggestion-quality dip
-    # in the meantime. Audit 2026-06-02.
-    from ..tasks.ml import recompute_centroid
-    recompute_centroid.delay(result.target_id)
     return jsonify(
         {
             "target": {

backend/app/celery_app.py

@@ -101,14 +101,6 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.ml.backfill",
                 "schedule": 86400.0,
             },
-            "recompute-centroids-daily": {
-                "task": "backend.app.tasks.ml.recompute_centroids",
-                "schedule": 86400.0,
-            },
-            "apply-allowlist-sweep-daily": {
-                "task": "backend.app.tasks.ml.apply_allowlist_tags",
-                "schedule": 86400.0,
-            },
             "train-heads-nightly": {
                 "task": "backend.app.tasks.ml.scheduled_train_heads",
                 "schedule": 86400.0,  # passive cadence; manual retrain stays available
@@ -131,6 +123,11 @@ def make_celery() -> Celery:
                 "schedule": 86400.0,  # drain the concept-crop back-catalogue +
                 "args": ("siglip",),  # retry failed embeds, no button needed
             },
+            "enqueue-embed-backfill-daily": {
+                "task": "backend.app.tasks.ml.enqueue_gpu_backfill",
+                "schedule": 86400.0,  # whole-image re-embed under the current
+                "args": ("embed",),   # model (an operator swap) drains via agent
+            },
             "ccip-auto-apply-daily": {
                 "task": "backend.app.tasks.ml.scheduled_ccip_auto_apply",
                 "schedule": 86400.0,  # no-op unless ccip_auto_apply_enabled

backend/app/models/__init__.py

@@ -13,7 +13,6 @@ from .head_auto_apply_run import HeadAutoApplyRun
 from .head_metric import HeadMetric
 from .head_metrics_snapshot import HeadMetricsSnapshot
 from .head_training_run import HeadTrainingRun
-from .image_prediction import ImagePrediction
 from .image_provenance import ImageProvenance
 from .image_record import ImageRecord
 from .image_region import ImageRegion
@@ -34,11 +33,9 @@ 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_allowlist import TagAllowlist
 from .tag_eval_run import TagEvalRun
 from .tag_head import TagHead
 from .tag_positive_confirmation import TagPositiveConfirmation
-from .tag_reference_embedding import TagReferenceEmbedding
 from .tag_suggestion_rejection import TagSuggestionRejection
 from .task_run import TaskRun
 
@@ -60,7 +57,6 @@ __all__ = [
     "SeriesPage",
     "SeriesSuggestion",
     "ImageRecord",
-    "ImagePrediction",
     "ImageProvenance",
     "ImageRegion",
     "Tag",
@@ -79,11 +75,9 @@ __all__ = [
     "HeadMetricsSnapshot",
     "HeadTrainingRun",
     "TagAlias",
-    "TagAllowlist",
     "TagEvalRun",
     "TagHead",
     "TagPositiveConfirmation",
-    "TagReferenceEmbedding",
     "TagSuggestionRejection",
     "TaskRun",
 ]

backend/app/models/gpu_job.py

@@ -14,7 +14,16 @@ pending for another agent).
 
 from datetime import datetime
 
-from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, func
+from sqlalchemy import (
+    DateTime,
+    ForeignKey,
+    Index,
+    Integer,
+    String,
+    Text,
+    func,
+    text,
+)
 from sqlalchemy.orm import Mapped, mapped_column
 
 from .base import Base
@@ -23,6 +32,17 @@ from .base import Base
 class GpuJob(Base):
     __tablename__ = "gpu_job"
 
+    # Partial indexes over just the live slice (see migration 0070): the lease
+    # reads the lowest-id pending jobs on the hot path, and reclaims expired
+    # leases as a backstop — both stay O(batch) as done/error history grows.
+    __table_args__ = (
+        Index("ix_gpu_job_pending", "id", postgresql_where=text("status = 'pending'")),
+        Index(
+            "ix_gpu_job_leased_expires", "lease_expires_at",
+            postgresql_where=text("status = 'leased'"),
+        ),
+    )
+
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
     image_record_id: Mapped[int] = mapped_column(
         ForeignKey("image_record.id", ondelete="CASCADE"), index=True

backend/app/models/image_prediction.py

@@ -1,37 +0,0 @@
-"""ImagePrediction — one row per (image, tagger vocab prediction).
-
-Replaces the image_record.tagger_predictions JSON blob (#768). Storing the
-raw Camie/booru vocab name (not a tag_id) preserves the suggestion read
-path's semantics: raw_name → canonical Tag resolution happens at read time
-via the alias map, and accepting a prediction can CREATE the Tag. The store
-floor (ml_settings.tagger_store_floor) is applied at WRITE time, so only
-predictions >= the floor land here.
-"""
-
-from sqlalchemy import Float, ForeignKey, Index, String, UniqueConstraint
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class ImagePrediction(Base):
-    __tablename__ = "image_prediction"
-    __table_args__ = (
-        UniqueConstraint(
-            "image_record_id", "raw_name", name="image_raw_name",
-        ),
-        # Per-image read (suggestion build) and the "images with tag X above
-        # Y" query the JSON blob never allowed.
-        Index("ix_image_prediction_image", "image_record_id"),
-        Index("ix_image_prediction_name_score", "raw_name", "score"),
-    )
-
-    id: Mapped[int] = mapped_column(primary_key=True)
-    image_record_id: Mapped[int] = mapped_column(
-        ForeignKey("image_record.id", ondelete="CASCADE"), nullable=False,
-    )
-    # The raw tagger vocab key (booru form) — NOT a tag_id. Resolved to a
-    # canonical Tag at read time, exactly as the old JSON keys were.
-    raw_name: Mapped[str] = mapped_column(String(255), nullable=False)
-    category: Mapped[str] = mapped_column(String(64), nullable=False)
-    score: Mapped[float] = mapped_column(Float, nullable=False)

backend/app/models/image_record.py

@@ -9,7 +9,6 @@ from datetime import datetime
 
 from pgvector.sqlalchemy import Vector
 from sqlalchemy import (
-    JSON,
     BigInteger,
     DateTime,
     Enum,
@@ -77,19 +76,13 @@ class ImageRecord(Base):
         ForeignKey("artist.id", ondelete="SET NULL"), nullable=True, index=True
     )
 
-    # ML fields (populated by FC-2's ml-worker). Per-tag predictions live in the
-    # normalized image_prediction table (#768) — the tagger_predictions JSON
-    # column was dropped in migration 0046. tagger_model_version stays as the
-    # "has this been tagged / is it current?" signal the backfill sweep reads.
-    tagger_model_version: Mapped[str | None] = mapped_column(String(128), nullable=True)
-    # 1152 = SigLIP-so400m embedding dim. Swapping models in FC-2 may require
-    # a column-width migration.
+    # ML fields (populated by the ml-worker / GPU agent). 1152 = SigLIP-so400m
+    # embedding dim; siglip_model_version stamps which model produced it (so an
+    # operator model swap, #1190, can re-embed the stale rows). A different-dim
+    # model would need a column-width migration.
     siglip_embedding: Mapped[list[float] | None] = mapped_column(Vector(1152), nullable=True)
     siglip_model_version: Mapped[str | None] = mapped_column(String(128), nullable=True)
 
-    # Centroid score cache (populated post-tagging)
-    centroid_scores: Mapped[dict | None] = mapped_column(JSON, nullable=True)
-
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()
     )
@@ -104,6 +97,16 @@ class ImageRecord(Base):
     effective_date: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()
     )
+    # Denormalized ORIGINAL-publish sort key (alembic 0071) = MIN(post_date)
+    # across ALL of the image's provenance posts, else created_at. effective_date
+    # above keys off the PRIMARY post (often the repost/download the file came
+    # from); this keys off the earliest publish across EVERY post the image
+    # appears in, so the gallery can sort by when content was first posted rather
+    # than when it was downloaded (operator-flagged 2026-07-01). Maintained by
+    # services/importer.py, recomputed whenever a dated post is linked.
+    earliest_post_date: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
         nullable=False,

backend/app/models/image_region.py

@@ -31,7 +31,10 @@ class ImageRegion(Base):
         ForeignKey("image_record.id", ondelete="CASCADE"), index=True
     )
     # 'frame' (a whole video frame → SigLIP bag) | 'face' | 'figure' (→ CCIP
-    # character id) | 'concept' (→ SigLIP head bag).
+    # character id) | 'concept' (→ SigLIP head bag) | 'panel' (a comic panel crop,
+    # also SigLIP → the bag). Free String, not an enum — proposers can add kinds
+    # without a migration; the bag scorer keys on a non-null siglip_embedding, not
+    # the kind, so any SigLIP-embedded region joins the bag.
     kind: Mapped[str] = mapped_column(String(16), nullable=False)
     # For video/animated media: the source frame's timestamp in SECONDS. NULL for
     # static images. Lets a video be a BAG of per-frame instances (fixes the

backend/app/models/ml_settings.py

@@ -23,46 +23,16 @@ class MLSettings(Base):
     __table_args__ = (CheckConstraint("id = 1", name="singleton"),)
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    suggestion_threshold_character: Mapped[float] = mapped_column(
-        Float, nullable=False, default=0.70
-    )
-    # Default raised 0.50 → 0.70 on 2026-06-02 — operator-flagged 0.50
-    # surfaced too many low-confidence picks; 0.70 keeps the rail
-    # signal-rich while still surfacing more than the original 0.95
-    # which hid almost everything. Operator-tunable via Settings → ML.
-    suggestion_threshold_general: Mapped[float] = mapped_column(
-        Float, nullable=False, default=0.70
-    )
-    centroid_similarity_threshold: Mapped[float] = mapped_column(
-        Float, nullable=False, default=0.55
-    )
-    # Ingest floor: tagger predictions below this confidence are not stored
-    # (tagger.Tagger.infer). Default 0.70 — the suggestion path already
-    # filters at 0.70 and the centroid/learned path covers low-confidence
-    # preferred tags, so the sub-0.70 tail is redundant weight (it had
-    # bloated image_record's TOAST to ~100 GB; plan-task #764). Operator-
-    # tunable via Settings → ML; must stay ≤ the suggestion thresholds.
-    tagger_store_floor: Mapped[float] = mapped_column(
-        Float, nullable=False, default=0.70
-    )
-    min_reference_images: Mapped[int] = mapped_column(
-        Integer, nullable=False, default=5
-    )
-    # Video tagging (#747). Sample one frame every N seconds (fixed CADENCE, not a
-    # fixed count) so a tag's frame-presence reflects real screen time regardless
-    # of video length; cap the total so a long video can't explode into hundreds
-    # of inferences (the cadence stretches past the cap). A tag is kept only if it
-    # appears in >= video_min_tag_frames sampled frames (≈ that many × interval
-    # seconds on screen) — duration-independent noise rejection. Operator-tunable.
+    # Video embedding (#747). Sample one frame every N seconds (fixed CADENCE, not
+    # a fixed count) so coverage reflects real screen time regardless of length;
+    # cap the total so a long video can't explode into hundreds of embeds. The
+    # per-frame SigLIP embeddings are mean-pooled. Operator-tunable.
     video_frame_interval_seconds: Mapped[float] = mapped_column(
         Float, nullable=False, default=4.0
     )
     video_max_frames: Mapped[int] = mapped_column(
         Integer, nullable=False, default=64
     )
-    video_min_tag_frames: Mapped[int] = mapped_column(
-        Integer, nullable=False, default=3
-    )
     # Tagging-v2 head training (#114). The head is the suggestion source that
     # LEARNS from the operator's tags (replacing Camie + centroid). A concept
     # needs >= head_min_positives labelled images before a head is trained;
@@ -101,11 +71,16 @@ class MLSettings(Base):
     ccip_auto_apply_threshold: Mapped[float] = mapped_column(
         Float, nullable=False, default=0.92
     )
-    tagger_model_version: Mapped[str] = mapped_column(
-        String(128), nullable=False, default="camie-tagger-v2"
-    )
+    # Default = SigLIP 2 (so400m, 512px) for new installs (migration 0069);
+    # existing libraries keep their stored value until the operator re-embeds.
     embedder_model_version: Mapped[str] = mapped_column(
-        String(128), nullable=False, default="siglip-so400m-patch14-384"
+        String(128), nullable=False, default="siglip2-so400m-patch16-512"
+    )
+    # The HF model NAME the embedder loads (server CPU embed + announced to the
+    # GPU agent in the lease). Operator-settable so the embedder is a choice, not
+    # a hardcode (#1190): set name + version together, then re-embed + retrain.
+    embedder_model_name: Mapped[str] = mapped_column(
+        String(128), nullable=False, default="google/siglip2-so400m-patch16-512"
     )
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()

backend/app/models/tag_allowlist.py

@@ -1,32 +0,0 @@
-"""TagAllowlist — tags the operator opted-in to auto-apply via maintenance."""
-
-from datetime import datetime
-
-from sqlalchemy import CheckConstraint, DateTime, Float, ForeignKey, func
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class TagAllowlist(Base):
-    __tablename__ = "tag_allowlist"
-    # Bare name — Base.metadata's naming convention prepends ck_<table>_,
-    # producing the final ck_tag_allowlist_confidence_range (matches migration 0003).
-    __table_args__ = (
-        CheckConstraint(
-            "min_confidence > 0 AND min_confidence <= 1",
-            name="confidence_range",
-        ),
-    )
-
-    tag_id: Mapped[int] = mapped_column(
-        ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True
-    )
-    # Default auto-apply threshold for a newly-accepted tag. 0.90 (lowered from
-    # 0.95 on operator evidence 2026-06-07: 0.95 was too strict and skipped
-    # confident-enough applications). Per-tag value is still tunable in the
-    # allowlist table; existing rows keep whatever they were stored with.
-    min_confidence: Mapped[float] = mapped_column(Float, nullable=False, default=0.90)
-    added_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )

backend/app/models/tag_reference_embedding.py

@@ -1,23 +0,0 @@
-"""TagReferenceEmbedding — per-tag centroid (mean SigLIP embedding of members)."""
-
-from datetime import datetime
-
-from pgvector.sqlalchemy import Vector
-from sqlalchemy import DateTime, ForeignKey, Integer, String, func
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class TagReferenceEmbedding(Base):
-    __tablename__ = "tag_reference_embedding"
-
-    tag_id: Mapped[int] = mapped_column(
-        ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True
-    )
-    embedding: Mapped[list[float]] = mapped_column(Vector(1152), nullable=False)
-    reference_count: Mapped[int] = mapped_column(Integer, nullable=False)
-    model_version: Mapped[str] = mapped_column(String(128), nullable=False)
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )

backend/app/scripts/download_models.py

@@ -7,7 +7,6 @@ import sys
 from pathlib import Path
 
 MODEL_ROOT = Path(os.environ.get("ML_MODEL_DIR", "/models"))
-CAMIE_REPO = os.environ.get("CAMIE_HF_REPO", "Camais03/camie-tagger-v2")
 SIGLIP_REPO = os.environ.get(
     "SIGLIP_HF_REPO", "google/siglip-so400m-patch14-384"
 )
@@ -24,34 +23,6 @@ def _snapshot(repo_id: str, dest: Path, allow_patterns: list[str] | None) -> Non
     )
 
 
-def ensure_camie() -> None:
-    """Fetch Camie v2 weights + metadata.
-
-    v2 layout (HuggingFace Camais03/camie-tagger-v2): the ONNX file is
-    named camie-tagger-v2.onnx (not model.onnx) and tags ship inside
-    camie-tagger-v2-metadata.json (not selected_tags.csv). Both at root.
-    The repo also contains app/, game/, training/, images/ subdirs full
-    of setup/demo files we don't need — allow_patterns scopes the fetch
-    to just the inference essentials (~790 MB instead of ~2 GB).
-    """
-    dest = MODEL_ROOT / "camie"
-    model_file = dest / "camie-tagger-v2.onnx"
-    meta_file = dest / "camie-tagger-v2-metadata.json"
-    if model_file.is_file() and meta_file.is_file():
-        print(f"[download_models] Camie present at {dest}")
-        return
-    print(f"[download_models] Fetching {CAMIE_REPO} -> {dest}")
-    _snapshot(
-        CAMIE_REPO, dest,
-        [
-            "camie-tagger-v2.onnx",
-            "camie-tagger-v2-metadata.json",
-            "config.json",
-            "config.yaml",
-        ],
-    )
-
-
 def ensure_siglip() -> None:
     dest = MODEL_ROOT / "siglip"
     if (dest / "config.json").is_file() and any(dest.glob("*.safetensors")):
@@ -62,7 +33,6 @@ def ensure_siglip() -> None:
 
 
 def main() -> int:
-    ensure_camie()
     ensure_siglip()
     print("[download_models] Done.")
     return 0

backend/app/services/cleanup_service.py

@@ -395,9 +395,8 @@ def delete_images(
 def delete_tag(session: Session, *, tag_id: int) -> dict:
     """Simple DELETE FROM tag WHERE id=?.
 
-    Postgres cascades the rest (image_tag, tag_alias, tag_allowlist,
-    tag_reference_embedding, tag_suggestion_rejection, series_page).
-    Returns counts BEFORE delete so the caller can surface them.
+    Postgres cascades the rest (image_tag, tag_alias, tag_suggestion_rejection,
+    series_page). Returns counts BEFORE delete so the caller can surface them.
     Raises LookupError if tag_id not found.
     """
     tag = session.get(Tag, tag_id)
@@ -742,8 +741,7 @@ def purge_legacy_tags(session: Session, *, dry_run: bool = False) -> dict:
     artist-kind tags PLUS general tags whose name matches a legacy
     prefix (source:*).
 
-    CASCADE on image_tag / tag_alias / tag_allowlist /
-    tag_reference_embedding / tag_suggestion_rejection / series_page
+    CASCADE on image_tag / tag_alias / tag_suggestion_rejection / series_page
     clears the related rows on the parent DELETE.
 
     Returns:
@@ -785,23 +783,21 @@ def purge_legacy_tags(session: Session, *, dry_run: bool = False) -> dict:
     return result
 
 
-# The Camie-suggestable CONTENT vocabulary. "Reset content tagging" wipes
-# these so the operator can re-tag from scratch via auto-suggest. fandom +
-# series (and series_page ordering) are deliberately NOT here — they're kept.
+# 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.
 RESETTABLE_TAG_KINDS = ("general", "character")
 
 
 def reset_content_tagging(session: Session, *, dry_run: bool = False) -> dict:
     """Count (dry_run) or DELETE every general + character tag so the operator
-    can re-tag from scratch via the Camie auto-suggest.
+    can re-tag from scratch (heads/CCIP repopulate suggestions).
 
-    PRESERVED: fandom + series tags and their series_page ordering, plus every
-    image's image_prediction rows (untouched) so suggestions
-    repopulate immediately. CASCADE on image_tag / tag_alias / tag_allowlist /
-    tag_reference_embedding / tag_suggestion_rejection clears each deleted
-    tag's applications + metadata. Tag.fandom_id is SET NULL, so deleting
-    character tags never touches the fandom rows. Irreversible except via DB
-    backup restore.
+    PRESERVED: fandom + series tags and their series_page ordering. CASCADE on
+    image_tag / tag_alias / tag_suggestion_rejection clears each deleted tag's
+    applications + metadata. Tag.fandom_id is SET NULL, so deleting character
+    tags never touches the fandom rows. Irreversible except via DB backup
+    restore.
 
     Returns:
       {"by_kind": {"general": N, "character": M},

backend/app/services/gallery_service.py

@@ -55,6 +55,25 @@ def _effective_date_col():
     return ImageRecord.effective_date
 
 
+# Sort key -> the materialized column the gallery orders + cursors on. Both are
+# indexed (DESC, id DESC), so every sort is a forward index range scan.
+#   newest/oldest    → effective_date    (primary post's date, else download)
+#   posted_new/_old  → earliest_post_date (earliest publish across ALL posts)
+_SORT_COLUMNS = {
+    "newest": ImageRecord.effective_date,
+    "oldest": ImageRecord.effective_date,
+    "posted_new": ImageRecord.earliest_post_date,
+    "posted_old": ImageRecord.earliest_post_date,
+}
+_ASCENDING_SORTS = {"oldest", "posted_old"}
+
+
+def _sort_column(sort: str):
+    """The materialized date column a gallery sort orders/cursors on (falls back
+    to effective_date for any unknown sort)."""
+    return _SORT_COLUMNS.get(sort, ImageRecord.effective_date)
+
+
 def _outer_join_primary_post(stmt: Select) -> Select:
     """LEFT JOIN Post on ImageRecord.primary_post_id so the COALESCE
     above sees Post.post_date when available. Images without a post
@@ -289,6 +308,80 @@ def _gallery_images(rows, artists: dict[int, dict]) -> list[GalleryImage]:
     ]
 
 
+def _diversify_similar(src, rows, limit, *, dup_threshold=8, lam=0.40):
+    """Trim a nearest-cosine candidate pool down to `limit` diverse picks.
+
+    1. pHash collapse: drop any candidate whose perceptual hash is within
+       `dup_threshold` Hamming bits of the anchor or an already-kept candidate —
+       so a reposted banner (and the anchor's own clones) appears at most once.
+    2. MMR (Maximal Marginal Relevance): greedily pick the candidate maximising
+       `lam * sim_to_anchor - (1 - lam) * max_sim_to_already_picked`. This keeps
+       the most relevant up top but pushes the selection to SPAN clusters
+       instead of returning 40 variations of one image.
+
+    `lam` is the variance dial: lower = weight the diversity penalty harder, so
+    the rail reaches further across clusters (operator wanted MORE variance,
+    2026-07-01 — dropped 0.55→0.40, dup 6→8, paired with a wider pool in
+    `similar()`).
+
+    Falls back to nearest-order (`rows[:limit]`) on any failure or a small pool.
+    """
+    if len(rows) <= 1:
+        return rows[:limit]
+    try:
+        import imagehash
+        import numpy as np
+    except Exception:
+        return rows[:limit]
+
+    # --- 1. pHash near-duplicate collapse (videos/NULL phash pass through) ---
+    kept = []
+    seen = []
+    if src.phash:
+        try:
+            seen.append(imagehash.hex_to_hash(src.phash))
+        except Exception:
+            pass
+    for row in rows:
+        ph = row[0].phash
+        if ph:
+            try:
+                h = imagehash.hex_to_hash(ph)
+                if any((h - k) <= dup_threshold for k in seen):
+                    continue
+                seen.append(h)
+            except Exception:
+                pass
+        kept.append(row)
+    if len(kept) <= limit:
+        return kept
+
+    # --- 2. MMR re-rank on the L2-normalised SigLIP embeddings ---
+    try:
+        a = np.asarray(src.siglip_embedding, dtype=np.float32)
+        a = a / (np.linalg.norm(a) or 1.0)
+        V = np.vstack([
+            np.asarray(row[0].siglip_embedding, dtype=np.float32) for row in kept
+        ])
+        V = V / np.clip(np.linalg.norm(V, axis=1, keepdims=True), 1e-8, None)
+    except Exception:
+        return kept[:limit]
+
+    rel = V @ a                                  # (N,) cosine to the anchor
+    n = len(kept)
+    picked_mask = np.zeros(n, dtype=bool)
+    max_sim = np.zeros(n, dtype=np.float32)      # max sim to anything picked yet
+    order = []
+    for _ in range(min(limit, n)):
+        scores = lam * rel - (1.0 - lam) * max_sim
+        scores[picked_mask] = -np.inf
+        i = int(np.argmax(scores))
+        order.append(i)
+        picked_mask[i] = True
+        max_sim = np.maximum(max_sim, V @ V[i])
+    return [kept[i] for i in order]
+
+
 async def _artists_for(session, image_ids: list[int]) -> dict[int, dict]:
     """Map image_id -> {"name","slug"} via the canonical
     image_record.artist_id (FC-2d-vii-c). Bounded by page size."""
@@ -332,7 +425,10 @@ class GalleryService:
             tag_ids, post_id, artist_id, tag_or_groups, tag_exclude,
         )
 
-        eff = _effective_date_col()
+        # eff is the ACTIVE sort column (effective_date or earliest_post_date);
+        # the cursor, ordering and year/month grouping all key off it, so the
+        # 'post date' sort paginates + buckets by original publish transparently.
+        eff = _sort_column(sort)
         stmt = select(ImageRecord, Post.post_date, eff.label("eff"))
         stmt = _outer_join_primary_post(stmt)
         stmt = _apply_scope(
@@ -343,7 +439,7 @@ class GalleryService:
             date_from=date_from, date_to=date_to,
         )
 
-        descending = sort != "oldest"
+        descending = sort not in _ASCENDING_SORTS
         if cursor:
             cur_ts, cur_id = decode_cursor(cursor)
             # The cursor is just (last eff, last id); the request's sort
@@ -565,14 +661,20 @@ class GalleryService:
         untagged: bool = False, no_artist: bool = False,
         date_from: datetime | None = None, date_to: datetime | None = None,
     ) -> list[GalleryImage] | None:
-        """Visual "more like this": images ranked by cosine distance to
-        `image_id`'s SigLIP embedding (pgvector, HNSW-indexed — alembic 0036).
-        No ML inference here; the embedding was computed at import.
+        """Visual "more like this": images near `image_id`'s SigLIP embedding
+        (pgvector, HNSW-indexed — alembic 0036), then DIVERSIFIED so the result
+        doesn't collapse into one cluster. No ML inference here.
 
-        Returns None if the source image doesn't exist (→ 404), [] if it has
-        no embedding (a video / not-yet-embedded). Composes with the Phase-1/2
-        scope filters (AND) but REPLACES the date sort — always nearest-first,
-        bounded to `limit` (no cursor; distance-ranking has no date cursor).
+        Pure nearest-cosine piles up near-identical images — a reposted banner
+        fills the whole grid, and once you wander into a B&W / comic-panel
+        cluster every neighbour is more of the same with no way back to colour
+        (operator-reported 2026-06-30). So we pull a WIDER candidate pool, then:
+          1. collapse near-duplicate pHashes (and drop clones of the anchor),
+          2. MMR re-rank — pick for closeness-to-anchor but penalise similarity
+             to what's already picked, so the result SPANS clusters.
+
+        Returns None if the source doesn't exist (→ 404), [] if it has no
+        embedding. Composes with the scope filters (AND); REPLACES the date sort.
         """
         if limit < 1 or limit > 200:
             raise ValueError("limit must be between 1 and 200")
@@ -582,6 +684,11 @@ class GalleryService:
         if src.siglip_embedding is None:
             return []
 
+        # Over-fetch so diversification has clusters to spread across — without a
+        # wide pool there's nothing but the near-dupes to choose from. Widened
+        # (5×→8×, cap 200→400) so the stronger MMR has genuinely distinct
+        # neighbourhoods to reach into for more variance (operator, 2026-07-01).
+        pool_n = min(400, max(limit * 8, 100))
         distance = ImageRecord.siglip_embedding.cosine_distance(src.siglip_embedding)
         eff = _effective_date_col()
         stmt = select(ImageRecord, Post.post_date, eff.label("eff"))
@@ -597,8 +704,9 @@ class GalleryService:
             platform=platform, untagged=untagged, no_artist=no_artist,
             date_from=date_from, date_to=date_to,
         )
-        stmt = stmt.order_by(distance.asc()).limit(limit)
+        stmt = stmt.order_by(distance.asc()).limit(pool_n)
         rows = (await self.session.execute(stmt)).all()
+        rows = _diversify_similar(src, rows, limit)
         artists = await _artists_for(self.session, [r[0].id for r in rows])
         return _gallery_images(rows, artists)
 

backend/app/services/importer.py

@@ -17,7 +17,7 @@ from enum import StrEnum
 from pathlib import Path
 
 from PIL import Image
-from sqlalchemy import select, update
+from sqlalchemy import func, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -1411,6 +1411,24 @@ class Importer:
         # default (matches the old COALESCE(post_date, created_at) fallback).
         if record.primary_post_id == post.id and post.post_date is not None:
             record.effective_date = post.post_date
+        # earliest_post_date (alembic 0071) = MIN(post_date) across ALL of this
+        # image's provenance posts, not just the primary — so the gallery can
+        # sort by original publish rather than the download/repost the primary
+        # points at. Recompute from provenance whenever a dated post is linked;
+        # the provenance row for THIS post was committed above, so the MIN
+        # includes it. Leaves the created_at default when no linked post is dated.
+        if post.post_date is not None:
+            earliest = self.session.execute(
+                select(func.min(Post.post_date))
+                .select_from(ImageProvenance)
+                .join(Post, Post.id == ImageProvenance.post_id)
+                .where(
+                    ImageProvenance.image_record_id == record.id,
+                    Post.post_date.is_not(None),
+                )
+            ).scalar_one_or_none()
+            if earliest is not None:
+                record.earliest_post_date = earliest
         self.session.flush()
 
     def _copy_to_library(
@@ -1475,20 +1493,8 @@ class Importer:
         existing.duration_seconds = duration  # #871: keep the kept copy's duration
         existing.thumbnail_path = None
         existing.integrity_status = "unknown"
-        existing.tagger_model_version = None
         existing.siglip_embedding = None
         existing.siglip_model_version = None
-        existing.centroid_scores = None
-        # #768: predictions also live in the normalized image_prediction table
-        # now — clear them so a re-imported file re-derives a fresh set.
-        from sqlalchemy import delete as _delete
-
-        from ..models import ImagePrediction as _ImagePrediction
-        self.session.execute(
-            _delete(_ImagePrediction).where(
-                _ImagePrediction.image_record_id == existing.id
-            )
-        )
         # created_at intentionally preserved; updated_at auto-bumps.
         self.session.flush()
         self.session.commit()

backend/app/services/ml/allowlist.py

@@ -1,36 +1,20 @@
-"""Allowlist semantics: accepting a suggestion adds the canonical tag to
-image_tag AND to tag_allowlist; per-image removal/dismiss writes a rejection.
+"""Suggestion actions: accept applies the canonical tag to an image (which
+feeds head training); dismiss / reject record a per-image rejection.
+
+(The Camie allowlist bulk-apply was retired #1189 — heads + CCIP are the tag
+source, and head auto-apply is the earned propagation. Accept no longer
+allowlists or fans a tag out across the library.)
 """
 
-from collections.abc import Sequence
-from dataclasses import dataclass
-
-from sqlalchemy import and_, delete, distinct, func, or_, select
+from sqlalchemy import delete
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from ...models import (
-    ImagePrediction,
-    MLSettings,
-    Tag,
-    TagAlias,
-    TagAllowlist,
-    TagSuggestionRejection,
-)
+from ...models import TagSuggestionRejection
 from ...models.tag import image_tag
 from .aliases import AliasService
 
 
-@dataclass(frozen=True)
-class AllowlistRow:
-    tag_id: int
-    tag_name: str
-    tag_kind: str
-    min_confidence: float
-    applied_count: int      # image_tag rows currently carrying this tag
-    coverage_count: int     # images a sweep WOULD cover at min_confidence
-
-
 class AllowlistService:
     def __init__(self, session: AsyncSession):
         self.session = session
@@ -39,21 +23,11 @@ class AllowlistService:
     async def _apply_image_tag(self, image_id: int, tag_id: int, source: str):
         stmt = insert(image_tag).values(
             image_record_id=image_id, tag_id=tag_id, source=source
-        )
-        stmt = stmt.on_conflict_do_nothing(
+        ).on_conflict_do_nothing(
             index_elements=["image_record_id", "tag_id"]
         )
         await self.session.execute(stmt)
 
-    async def _add_to_allowlist(self, tag_id: int) -> bool:
-        """Returns True if newly added (caller should kick off retro-apply)."""
-        exists = await self.session.get(TagAllowlist, tag_id)
-        if exists is not None:
-            return False
-        self.session.add(TagAllowlist(tag_id=tag_id))
-        await self.session.flush()
-        return True
-
     async def _clear_rejection(self, image_id: int, tag_id: int):
         await self.session.execute(
             delete(TagSuggestionRejection)
@@ -61,12 +35,11 @@ class AllowlistService:
             .where(TagSuggestionRejection.tag_id == tag_id)
         )
 
-    async def accept(self, image_id: int, tag_id: int) -> bool:
-        """Accept a suggestion. Returns True if the tag was newly added to
-        the allowlist (the API layer enqueues apply_allowlist_tags then)."""
+    async def accept(self, image_id: int, tag_id: int) -> None:
+        """Apply the accepted tag to this image (source='ml_accepted', a head
+        training positive) and clear any prior rejection."""
         await self._apply_image_tag(image_id, tag_id, source="ml_accepted")
         await self._clear_rejection(image_id, tag_id)
-        return await self._add_to_allowlist(tag_id)
 
     async def add_alias_and_accept(
         self,
@@ -74,17 +47,16 @@ class AllowlistService:
         alias_string: str,
         alias_category: str,
         canonical_tag_id: int,
-    ) -> bool:
+    ) -> None:
         await self.aliases.create(
             alias_string, alias_category, canonical_tag_id
         )
-        return await self.accept(image_id, canonical_tag_id)
+        await self.accept(image_id, canonical_tag_id)
 
     async def dismiss(self, image_id: int, tag_id: int) -> None:
         stmt = insert(TagSuggestionRejection).values(
             image_record_id=image_id, tag_id=tag_id
-        )
-        stmt = stmt.on_conflict_do_nothing(
+        ).on_conflict_do_nothing(
             index_elements=["image_record_id", "tag_id"]
         )
         await self.session.execute(stmt)
@@ -96,118 +68,11 @@ class AllowlistService:
         await self._clear_rejection(image_id, tag_id)
 
     async def reject_applied_tag(self, image_id: int, tag_id: int) -> None:
-        """Operator removed an applied tag from an image. Remove the
-        image_tag row AND record a rejection so the allowlist won't
-        re-apply it on the next maintenance sweep."""
+        """Operator removed an applied tag from an image. Remove the image_tag
+        row AND record a rejection so head auto-apply won't re-apply it."""
         await self.session.execute(
             image_tag.delete()
             .where(image_tag.c.image_record_id == image_id)
             .where(image_tag.c.tag_id == tag_id)
         )
         await self.dismiss(image_id, tag_id)
-
-    async def _store_floor(self) -> float:
-        return (
-            await self.session.execute(
-                select(MLSettings.tagger_store_floor).where(MLSettings.id == 1)
-            )
-        ).scalar_one()
-
-    async def update_threshold(
-        self, tag_id: int, min_confidence: float
-    ) -> None:
-        row = await self.session.get(TagAllowlist, tag_id)
-        if row is not None:
-            # An allowlist tag can't auto-apply more permissively than the
-            # ingest store floor — predictions below tagger_store_floor aren't
-            # stored, so a lower min_confidence would behave identically to the
-            # floor. Clamp so the stored threshold matches actual behavior
-            # (#764).
-            floor = await self._store_floor()
-            row.min_confidence = max(min_confidence, floor)
-
-    async def remove(self, tag_id: int) -> None:
-        await self.session.execute(
-            delete(TagAllowlist).where(TagAllowlist.tag_id == tag_id)
-        )
-
-    async def _coverage_match(self, tag: Tag):
-        """The predicate over image_prediction rows that resolve to `tag`,
-        mirroring tasks.ml._confidence_for_tag's resolution: a prediction whose
-        raw_name equals the tag name (any category), OR an alias maps
-        (raw_name, category) -> this tag. Returns a SQLAlchemy boolean clause.
-        """
-        alias_rows = (
-            await self.session.execute(
-                select(TagAlias.alias_string, TagAlias.alias_category).where(
-                    TagAlias.canonical_tag_id == tag.id
-                )
-            )
-        ).all()
-        name_clause = ImagePrediction.raw_name == tag.name
-        alias_clauses = [
-            and_(
-                ImagePrediction.raw_name == a,
-                ImagePrediction.category == c,
-            )
-            for a, c in alias_rows
-        ]
-        return or_(name_clause, *alias_clauses) if alias_clauses else name_clause
-
-    async def coverage(self, tag_id: int, threshold: float) -> int:
-        """How many distinct images a sweep WOULD cover for this tag at
-        `threshold`: images with a resolving prediction scoring >= threshold.
-        The gross candidate pool (NOT minus already-applied/rejected) — it's
-        the tuning signal for "lower the threshold and ~N more images qualify".
-        """
-        tag = await self.session.get(Tag, tag_id)
-        if tag is None:
-            return 0
-        match = await self._coverage_match(tag)
-        stmt = select(
-            func.count(distinct(ImagePrediction.image_record_id))
-        ).where(ImagePrediction.score >= threshold, match)
-        return (await self.session.execute(stmt)).scalar_one()
-
-    async def list_all(self) -> Sequence[AllowlistRow]:
-        stmt = (
-            select(
-                TagAllowlist.tag_id,
-                Tag.name,
-                Tag.kind,
-                TagAllowlist.min_confidence,
-            )
-            .join(Tag, Tag.id == TagAllowlist.tag_id)
-            .order_by(Tag.name.asc())
-        )
-        rows = (await self.session.execute(stmt)).all()
-        tag_ids = [r[0] for r in rows]
-
-        # Applied counts in ONE grouped query (vs N per-row counts).
-        applied: dict[int, int] = {}
-        if tag_ids:
-            applied = dict(
-                (
-                    await self.session.execute(
-                        select(image_tag.c.tag_id, func.count())
-                        .where(image_tag.c.tag_id.in_(tag_ids))
-                        .group_by(image_tag.c.tag_id)
-                    )
-                ).all()
-            )
-
-        result = []
-        for r in rows:
-            # Coverage is per-tag (alias set differs); allowlist is small.
-            cov = await self.coverage(r[0], r[3])
-            result.append(
-                AllowlistRow(
-                    tag_id=r[0],
-                    tag_name=r[1],
-                    tag_kind=r[2].value if hasattr(r[2], "value") else str(r[2]),
-                    min_confidence=r[3],
-                    applied_count=applied.get(r[0], 0),
-                    coverage_count=cov,
-                )
-            )
-        return result

backend/app/services/ml/centroids.py

@@ -1,163 +0,0 @@
-"""Tag centroids: the mean SigLIP embedding of a tag's member images.
-
-Powers centroid-augmented suggestions (a tag whose centroid is close to an
-image's embedding becomes a suggestion even if Camie didn't predict it).
-"""
-
-from dataclasses import dataclass
-
-import numpy as np
-from sqlalchemy import func, select
-from sqlalchemy.dialects.postgresql import insert
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from ...models import (
-    ImageRecord,
-    MLSettings,
-    Tag,
-    TagKind,
-    TagReferenceEmbedding,
-)
-from ...models.tag import image_tag
-
-ELIGIBLE_KINDS = {
-    TagKind.character,
-    TagKind.fandom,
-    TagKind.general,
-    TagKind.series,
-}
-
-
-@dataclass(frozen=True)
-class CentroidHit:
-    tag_id: int
-    similarity: float
-
-
-class CentroidService:
-    def __init__(self, session: AsyncSession):
-        self.session = session
-
-    async def _min_reference_images(self) -> int:
-        return (
-            await self.session.execute(
-                select(MLSettings.min_reference_images).where(MLSettings.id == 1)
-            )
-        ).scalar_one()
-
-    async def _model_version(self) -> str:
-        """Audit 2026-06-02: SigLIP model-version stamp comes from the
-        DB row, not the env constant. tag_and_embed (tasks/ml.py:110)
-        already reads from MLSettings.embedder_model_version, so by
-        sourcing centroid stamps + drift checks from the same row, we
-        eliminate the silent-drift case the audit flagged. env
-        SIGLIP_MODEL_VERSION still drives which model embedder.py
-        loads at runtime; the version stamp is purely the operator-
-        controlled identifier."""
-        return (
-            await self.session.execute(
-                select(MLSettings.embedder_model_version).where(MLSettings.id == 1)
-            )
-        ).scalar_one()
-
-    async def recompute_for_tag(self, tag_id: int) -> bool:
-        """Recompute one tag's centroid. Returns True if a centroid was
-        written, False if skipped (ineligible kind or too few members)."""
-        tag = await self.session.get(Tag, tag_id)
-        if tag is None or tag.kind not in ELIGIBLE_KINDS:
-            return False
-
-        min_refs = await self._min_reference_images()
-
-        stmt = (
-            select(ImageRecord.siglip_embedding)
-            .join(image_tag, image_tag.c.image_record_id == ImageRecord.id)
-            .where(image_tag.c.tag_id == tag_id)
-            .where(ImageRecord.siglip_embedding.is_not(None))
-        )
-        embeddings = [
-            np.array(e, dtype=np.float32)
-            for e in (await self.session.execute(stmt)).scalars().all()
-        ]
-        if len(embeddings) < min_refs:
-            return False
-
-        centroid = np.mean(np.stack(embeddings), axis=0).astype(np.float32)
-        model_version = await self._model_version()
-
-        stmt = insert(TagReferenceEmbedding).values(
-            tag_id=tag_id,
-            embedding=centroid.tolist(),
-            reference_count=len(embeddings),
-            model_version=model_version,
-        )
-        stmt = stmt.on_conflict_do_update(
-            index_elements=["tag_id"],
-            set_={
-                "embedding": centroid.tolist(),
-                "reference_count": len(embeddings),
-                "model_version": model_version,
-                "updated_at": func.now(),
-            },
-        )
-        await self.session.execute(stmt)
-        return True
-
-    async def list_drifted(self) -> list[int]:
-        """Tag ids whose centroid is stale: member count != reference_count,
-        OR no centroid row, OR centroid built on a different SigLIP version.
-        Only considers eligible-kind tags with embeddings present."""
-        current_model_version = await self._model_version()
-        member_counts = (
-            select(
-                image_tag.c.tag_id.label("tag_id"),
-                func.count(image_tag.c.image_record_id).label("members"),
-            )
-            .join(ImageRecord, ImageRecord.id == image_tag.c.image_record_id)
-            .where(ImageRecord.siglip_embedding.is_not(None))
-            .group_by(image_tag.c.tag_id)
-            .subquery()
-        )
-        stmt = (
-            select(Tag.id)
-            .join(member_counts, member_counts.c.tag_id == Tag.id)
-            .outerjoin(
-                TagReferenceEmbedding,
-                TagReferenceEmbedding.tag_id == Tag.id,
-            )
-            .where(Tag.kind.in_(ELIGIBLE_KINDS))
-            .where(
-                (TagReferenceEmbedding.tag_id.is_(None))
-                | (
-                    TagReferenceEmbedding.reference_count
-                    != member_counts.c.members
-                )
-                | (TagReferenceEmbedding.model_version != current_model_version)
-            )
-        )
-        return list((await self.session.execute(stmt)).scalars().all())
-
-    async def find_similar_tags(
-        self, image_id: int, limit: int = 20
-    ) -> list[CentroidHit]:
-        """Cosine similarity between an image's embedding and stored
-        centroids. Returns top-`limit` by similarity DESC. pgvector's
-        cosine_distance gives 1 - cosine_similarity."""
-        img = await self.session.get(ImageRecord, image_id)
-        if img is None or img.siglip_embedding is None:
-            return []
-        emb = img.siglip_embedding
-        distance = TagReferenceEmbedding.embedding.cosine_distance(emb)
-        stmt = (
-            select(
-                TagReferenceEmbedding.tag_id,
-                (1 - distance).label("similarity"),
-            )
-            .order_by(distance.asc())
-            .limit(limit)
-        )
-        rows = (await self.session.execute(stmt)).all()
-        return [
-            CentroidHit(tag_id=r.tag_id, similarity=float(r.similarity))
-            for r in rows
-        ]

backend/app/services/ml/embedder.py

@@ -18,9 +18,11 @@ ImageFile.LOAD_TRUNCATED_IMAGES = True
 # N_replicas × this within the cores allotted to ML to avoid oversubscription.
 _INTRA_OP_THREADS = 4
 
-MODEL_NAME = os.environ.get(
+DEFAULT_MODEL_NAME = os.environ.get(
     "SIGLIP_MODEL_NAME", "google/siglip-so400m-patch14-384"
 )
+# Back-compat alias (api/gpu imported this name as the fallback embedder id).
+MODEL_NAME = DEFAULT_MODEL_NAME
 MODEL_VERSION = os.environ.get(
     "SIGLIP_MODEL_VERSION", "siglip-so400m-patch14-384"
 )
@@ -29,35 +31,42 @@ _LOCAL_DIR = Path(os.environ.get("ML_MODEL_DIR", "/models")) / "siglip"
 
 
 class Embedder:
-    def __init__(self, model_dir: Path | None = None):
-        self._model_dir = model_dir or _LOCAL_DIR
+    """Loads whatever SigLIP-family model it's given by HF NAME. For the default
+    model it prefers the pre-downloaded local dir (no re-download on existing
+    deploys); any other name resolves as an HF repo id (downloaded + cached on
+    first use), so an operator model swap (#1190) just works server-side."""
+
+    def __init__(self, model_name: str | None = None, model_dir: Path | None = None):
+        self.model_name = model_name or DEFAULT_MODEL_NAME
+        self._explicit_dir = model_dir
         self._model = None
         self._processor = None
         self._torch = None
 
+    def _source(self) -> str:
+        if self._explicit_dir is not None:
+            return str(self._explicit_dir)
+        if self.model_name == DEFAULT_MODEL_NAME and _LOCAL_DIR.exists():
+            return str(_LOCAL_DIR)
+        return self.model_name
+
     def load(self) -> None:
         if self._model is not None:
             return
         import torch
-        from transformers import AutoModel, SiglipImageProcessor
+        from transformers import AutoImageProcessor, AutoModel
 
         self._torch = torch
         # Bound torch's CPU thread pool (see _INTRA_OP_THREADS) so each replica
         # stays a predictable core consumer on a shared node.
         torch.set_num_threads(_INTRA_OP_THREADS)
-        # FC's embedder only does IMAGE inference — never text. AutoProcessor
-        # loads the full processor including SiglipTokenizer, which requires
-        # the sentencepiece library at import time even if we never call it.
-        # SiglipImageProcessor loads ONLY preprocessor_config.json (image
-        # side) and skips the tokenizer config entirely. Operator hit the
-        # ImportError 2026-05-25 once the ml-worker started actually running
-        # tag_and_embed; switching to the image-only loader avoids the
-        # tokenizer dep without adding ~30 MB of unused C++ build to the
-        # lean ml-worker image.
-        self._processor = SiglipImageProcessor.from_pretrained(
-            str(self._model_dir)
-        )
-        self._model = AutoModel.from_pretrained(str(self._model_dir))
+        # IMAGE inference only — AutoImageProcessor loads just the image side
+        # (preprocessor_config.json), skipping the SigLIP tokenizer + its
+        # sentencepiece dep (operator hit that ImportError 2026-05-25). Works
+        # for any SigLIP-family model, keeping the embedder model-agnostic.
+        src = self._source()
+        self._processor = AutoImageProcessor.from_pretrained(src)
+        self._model = AutoModel.from_pretrained(src)
         self._model.eval()
 
     def infer(self, image_path: Path) -> np.ndarray:
@@ -74,8 +83,12 @@ class Embedder:
 _default_embedder: Embedder | None = None
 
 
-def get_embedder() -> Embedder:
+def get_embedder(model_name: str | None = None) -> Embedder:
+    """Cached embedder for `model_name` (default if None). Rebuilds the singleton
+    when the requested name changes, so an operator model swap takes effect
+    without restarting the worker."""
     global _default_embedder
-    if _default_embedder is None:
-        _default_embedder = Embedder()
+    name = model_name or DEFAULT_MODEL_NAME
+    if _default_embedder is None or _default_embedder.model_name != name:
+        _default_embedder = Embedder(model_name=name)
     return _default_embedder

backend/app/services/ml/gpu_jobs.py

@@ -12,7 +12,7 @@ and the lease itself reclaims expired leases as a final backstop. Result-writing
 
 from datetime import UTC, datetime, timedelta
 
-from sqlalchemy import and_, or_, select, update
+from sqlalchemy import and_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ...models import GpuJob
@@ -51,25 +51,33 @@ class GpuJobService:
     async def lease(
         self, token: str, batch_size: int = DEFAULT_BATCH, ttl: int = DEFAULT_LEASE_TTL
     ) -> list[GpuJob]:
-        """Claim up to batch_size pending (or expired-leased) jobs for `token`."""
+        """Claim up to batch_size pending (or expired-leased) jobs for `token`.
+
+        Two phases so each hits a partial index (0070) and stays O(batch) no
+        matter how many done/error rows have accumulated: the pending pool is the
+        hot path; expired leases are reclaimed only when pending can't fill the
+        batch (a crashed agent's work — rare). The old single OR-query walked the
+        primary key past the whole done-prefix in id order → O(done), which is
+        why leasing crawled — and the DB saturated — as the run progressed."""
         now = datetime.now(UTC)
-        picked = (
-            await self.session.execute(
-                select(GpuJob.id)
-                .where(
-                    or_(
-                        GpuJob.status == "pending",
-                        and_(
-                            GpuJob.status == "leased",
-                            GpuJob.lease_expires_at < now,
-                        ),
+
+        async def _claim(condition, limit: int) -> list[int]:
+            return list(
+                (
+                    await self.session.execute(
+                        select(GpuJob.id).where(condition)
+                        .order_by(GpuJob.id).limit(limit)
+                        .with_for_update(skip_locked=True)
                     )
-                )
-                .order_by(GpuJob.id)
-                .limit(batch_size)
-                .with_for_update(skip_locked=True)
+                ).scalars().all()
+            )
+
+        picked = await _claim(GpuJob.status == "pending", batch_size)
+        if len(picked) < batch_size:     # pending exhausted → reclaim expired leases
+            picked += await _claim(
+                and_(GpuJob.status == "leased", GpuJob.lease_expires_at < now),
+                batch_size - len(picked),
             )
-        ).scalars().all()
         if not picked:
             return []
         await self.session.execute(

backend/app/services/ml/heads.py

@@ -308,25 +308,36 @@ async def score_image(
     import numpy as np
 
     img = await session.get(ImageRecord, image_id)
-    if img is None or img.siglip_embedding is None:
+    if img is None:
         return []
     settings = await _settings_async(session)
-    heads = await _current_heads(session, settings.embedder_model_version)
+    cur_version = settings.embedder_model_version
+    heads = await _current_heads(session, cur_version)
     if heads["W"] is None:
         return []
 
-    bag = [np.asarray(img.siglip_embedding, dtype=np.float32)]
+    # Only embeddings in the CURRENT model's space enter the bag. Mid model-swap
+    # (#1190), an image still carrying the OLD-version whole-image vector is
+    # skipped rather than scored by heads trained in a different space; a legacy
+    # NULL version is treated as current (those predate per-row stamping).
+    bag = []
+    if img.siglip_embedding is not None and img.siglip_model_version in (
+        cur_version, None,
+    ):
+        bag.append(np.asarray(img.siglip_embedding, dtype=np.float32))
     region_vecs = (
         await session.execute(
             select(ImageRegion.siglip_embedding)
             .where(ImageRegion.image_record_id == image_id)
             .where(ImageRegion.siglip_embedding.is_not(None))
-            .where(ImageRegion.embedding_version == settings.embedder_model_version)
+            .where(ImageRegion.embedding_version == cur_version)
         )
     ).all()
     for (vec,) in region_vecs:
         if vec is not None:
             bag.append(np.asarray(vec, dtype=np.float32))
+    if not bag:
+        return []
 
     X = np.vstack(bag)                                  # (B, D)
     norms = np.linalg.norm(X, axis=1, keepdims=True)

backend/app/services/ml/tagger.py

@@ -1,210 +0,0 @@
-"""Camie-tagger-v2 ONNX wrapper (CPU).
-
-Single-image at a time. Loaded lazily inside the ml-worker process; NOT
-thread-safe — the ml queue worker runs --concurrency=1 per process (scale ML by
-running multiple worker replicas, not threads).
-
-v2 layout reference: HuggingFace Camais03/camie-tagger-v2 root has
-camie-tagger-v2.onnx (789 MB) + camie-tagger-v2-metadata.json (7.77 MB)
-+ config.json. Tags ship as nested JSON, not CSV. Preprocessing and
-output handling follow the published onnx_inference.py reference:
-ImageNet normalize, NCHW layout, sigmoid on refined logits (output[1]).
-"""
-
-import json
-import os
-from dataclasses import dataclass
-from pathlib import Path
-
-import numpy as np
-from PIL import Image, ImageFile
-
-# Cap inference threads (see Tagger.load) so each ml-worker replica is a bounded
-# core consumer on a shared node — keep N_replicas × this within the cores
-# allotted to ML so replicas don't oversubscribe the box / starve the DB.
-_INTRA_OP_THREADS = 4
-
-# onnxruntime lives in requirements-ml.txt only — it is NOT installed in the
-# lean web image or in CI. Imported lazily inside Tagger.load() so this module
-# imports fine without it (the suggestion service imports SURFACED_CATEGORIES
-# from here in the web container, and CI collects the pure-logic tests).
-
-# Tolerate minutely-truncated source images (same rationale as IR's wd14.py:
-# a few missing bytes at the JPEG EOI shouldn't block tagging the whole image).
-ImageFile.LOAD_TRUNCATED_IMAGES = True
-
-MODEL_NAME = os.environ.get("CAMIE_MODEL_NAME", "camie-tagger-v2")
-_MODEL_DIR = Path(os.environ.get("ML_MODEL_DIR", "/models")) / "camie"
-_MODEL_FILE = f"{MODEL_NAME}.onnx"
-_METADATA_FILE = f"{MODEL_NAME}-metadata.json"
-
-# Ingest floor below which predictions aren't stored (keeps the JSON compact).
-# DEFAULT/fallback only — the live value is DB-backed
-# (ml_settings.tagger_store_floor) and passed into infer() per call by the ml
-# task. 0.70: the suggestion path already filters there and the centroid path
-# covers lower-confidence preferred tags, so the sub-0.70 tail is redundant
-# (it had bloated image_record's TOAST to ~100 GB; plan-task #764).
-DEFAULT_STORE_FLOOR = 0.70
-
-# The categories FC-2b surfaces in the UI. Others (meta/rating/year) are
-# still stored but the suggestion service filters them out.
-# 'artist' retired in FC-2d-vii-c — artist identity is acquisition-derived
-# (image_record.artist_id), never ML-inferred. 'copyright' retired
-# 2026-06-01 — operator doesn't use the copyright tag-kind; fandom is
-# this app's franchise/series concept (per TagsView.vue's doc comment).
-# Raw predictions for both categories still get stored at STORE_FLOOR but
-# don't surface in suggestions.
-SURFACED_CATEGORIES = {"character", "general"}
-
-# ImageNet preprocessing constants (per Camie v2 onnx_inference.py).
-_IMAGENET_MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32)
-_IMAGENET_STD = np.array([0.229, 0.224, 0.225], dtype=np.float32)
-# Square-pad color ≈ ImageNet mean × 255 (matches reference inference).
-_PAD_COLOR = (124, 116, 104)
-
-
-@dataclass(frozen=True)
-class TagPrediction:
-    name: str
-    category: str
-    confidence: float
-
-
-class Tagger:
-    def __init__(self, model_dir: Path | None = None):
-        self._model_dir = model_dir or _MODEL_DIR
-        self._session = None  # onnxruntime.InferenceSession once load()ed
-        self._tag_names: list[str] | None = None
-        self._tag_categories: list[str] | None = None
-        self._input_name: str | None = None
-        self._input_size: int = 512
-
-    def load(self) -> None:
-        if self._session is not None:
-            return
-        model_path = self._model_dir / _MODEL_FILE
-        meta_path = self._model_dir / _METADATA_FILE
-        if not model_path.is_file():
-            raise RuntimeError(
-                f"Camie {_MODEL_FILE} missing at {model_path}. "
-                f"Populate /models via the ml-worker downloader."
-            )
-        if not meta_path.is_file():
-            raise RuntimeError(
-                f"Camie {_METADATA_FILE} missing at {meta_path}. "
-                f"Populate /models via the ml-worker downloader."
-            )
-
-        with open(meta_path) as f:
-            metadata = json.load(f)
-
-        # Per Camie v2 onnx_inference.py: idx_to_tag is keyed by str(idx);
-        # tag_to_category maps tag_name -> category. Project to two parallel
-        # lists indexed by output position for O(1) lookup in the hot path.
-        ds = metadata["dataset_info"]
-        idx_to_tag = ds["tag_mapping"]["idx_to_tag"]
-        tag_to_category = ds["tag_mapping"]["tag_to_category"]
-        total = ds["total_tags"]
-        names: list[str] = []
-        cats: list[str] = []
-        for i in range(total):
-            name = idx_to_tag.get(str(i), f"unknown-{i}")
-            names.append(name)
-            cats.append(tag_to_category.get(name, "general"))
-
-        # Input size from metadata; fall back to 512 (the v2 default).
-        self._input_size = int(
-            metadata.get("model_info", {}).get("img_size", 512)
-        )
-
-        # Lazy import — kept after the file-existence checks so the
-        # missing-model RuntimeError still fires first in environments
-        # without onnxruntime (CI / lean web image).
-        import onnxruntime as ort
-
-        # Cap the intra-op thread pool. ONNX Runtime otherwise sizes it to ALL
-        # host cores, so on a shared node each ml-worker replica would grab every
-        # core and oversubscribe (and starve the co-located DB/web). Bounding it
-        # makes each replica a predictable core consumer — run N replicas where
-        # N × _INTRA_OP_THREADS stays within the cores you allot to ML.
-        opts = ort.SessionOptions()
-        opts.intra_op_num_threads = _INTRA_OP_THREADS
-        session = ort.InferenceSession(
-            str(model_path), sess_options=opts, providers=["CPUExecutionProvider"],
-        )
-        self._input_name = session.get_inputs()[0].name
-        # Assign sentinels last so a partial load isn't observable.
-        self._tag_names = names
-        self._tag_categories = cats
-        self._session = session
-
-    def _preprocess(self, image_path: Path) -> np.ndarray:
-        img = Image.open(image_path)
-        # Composite RGBA onto neutral so transparency doesn't bias the model.
-        if img.mode == "RGBA":
-            bg = Image.new("RGBA", img.size, (255, 255, 255, 255))
-            bg.paste(img, mask=img.split()[3])
-            img = bg.convert("RGB")
-        elif img.mode != "RGB":
-            img = img.convert("RGB")
-
-        # Pad to square with ImageNet-mean color, then bicubic resize.
-        w, h = img.size
-        side = max(w, h)
-        square = Image.new("RGB", (side, side), _PAD_COLOR)
-        square.paste(img, ((side - w) // 2, (side - h) // 2))
-        square = square.resize(
-            (self._input_size, self._input_size), Image.BICUBIC
-        )
-
-        arr = np.array(square, dtype=np.float32) / 255.0   # HWC, [0,1]
-        arr = (arr - _IMAGENET_MEAN) / _IMAGENET_STD        # ImageNet normalize
-        arr = arr.transpose(2, 0, 1)                        # HWC -> CHW
-        return arr[np.newaxis, :, :, :]                     # NCHW
-
-    def infer(
-        self, image_path: Path, *, store_floor: float = DEFAULT_STORE_FLOOR,
-    ) -> dict[str, TagPrediction]:
-        """Run Camie v2 on one image. Returns {name: TagPrediction} with
-        confidence >= store_floor (across all categories — the suggestion
-        service does category filtering later). store_floor is the DB-backed
-        ml_settings.tagger_store_floor, passed in by the ml task.
-
-        v2 emits multiple outputs; we use the refined predictions
-        (output[1] per onnx_inference.py). Sigmoid is applied to raw
-        logits to produce [0,1] confidence scores.
-        """
-        self.load()
-        x = self._preprocess(image_path)
-        outputs = self._session.run(None, {self._input_name: x})
-        # Refined predictions if present (v2 emits initial + refined),
-        # fall back to initial for single-output forks.
-        logits = outputs[1] if len(outputs) > 1 else outputs[0]
-        # Squeeze batch dim, apply sigmoid.
-        probs = 1.0 / (1.0 + np.exp(-logits[0]))
-        results: dict[str, TagPrediction] = {}
-        names = self._tag_names
-        cats = self._tag_categories
-        for idx, score in enumerate(probs):
-            conf = float(score)
-            if conf < store_floor:
-                continue
-            if idx >= len(names):
-                # Output longer than metadata declared — shouldn't happen but
-                # don't crash the import pipeline if v2 metadata desynchronizes.
-                continue
-            results[names[idx]] = TagPrediction(
-                name=names[idx], category=cats[idx], confidence=conf
-            )
-        return results
-
-
-_default_tagger: Tagger | None = None
-
-
-def get_tagger() -> Tagger:
-    """Process-level singleton so the ONNX session loads once per worker."""
-    global _default_tagger
-    if _default_tagger is None:
-        _default_tagger = Tagger()
-    return _default_tagger

backend/app/services/tag_service.py

@@ -10,8 +10,6 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import HeadMetric, Tag, TagHead, TagKind, image_tag
-from ..models.tag_allowlist import TagAllowlist
-from ..models.tag_reference_embedding import TagReferenceEmbedding
 from .db_helpers import get_or_create
 from .tag_query import fandom_join_alias, tag_columns
 
@@ -304,35 +302,22 @@ class TagService:
 
     async def _keep_as_alias(self, tag_id: int) -> bool:
         """A merged-away tag's old name must survive as an alias iff the ML
-        pipeline has ever applied it OR could re-emit it (allowlisted / has
-        a centroid) — otherwise the proactive apply_allowlist_tags worker
-        would silently regenerate it. Purely-manual, ML-unknown tags are
-        deleted outright (no DB bloat)."""
+        pipeline has ever applied it (manual accept or head auto-apply) — so a
+        re-application or an alias remap resolves the canonical name. Purely-
+        manual, ML-unknown tags are deleted outright (no DB bloat)."""
         is_machine = await self.session.scalar(
             select(
                 exists().where(
                     and_(
                         image_tag.c.tag_id == tag_id,
                         image_tag.c.source.in_(
-                            ("ml_auto", "ml_accepted", "auto")
+                            ("ml_auto", "ml_accepted", "head_auto", "auto")
                         ),
                     )
                 )
             )
         )
-        if is_machine:
-            return True
-        allowlisted = await self.session.scalar(
-            select(exists().where(TagAllowlist.tag_id == tag_id))
-        )
-        if allowlisted:
-            return True
-        has_centroid = await self.session.scalar(
-            select(
-                exists().where(TagReferenceEmbedding.tag_id == tag_id)
-            )
-        )
-        return bool(has_centroid)
+        return bool(is_machine)
 
     async def rename(self, tag_id: int, new_name: str) -> Tag:
         """Rename a tag. Raises TagMergeConflict if the new name collides
@@ -572,8 +557,6 @@ class TagService:
 
         merged_count = await self._repoint_image_tags(source_id, target_id)
         await self._repoint_rejections(source_id, target_id)
-        await self._repoint_allowlist(source_id, target_id)
-        await self._repoint_embedding(source_id)
         await self._repoint_aliases(source_id, target_id)
         await self._repoint_fandom_children(
             source_id, target_id, source_kind
@@ -639,30 +622,6 @@ class TagService:
             .values(tag_id=tgt)
         )
 
-    async def _repoint_allowlist(self, src: int, tgt: int) -> None:
-        tgt_has = await self.session.scalar(
-            select(exists().where(TagAllowlist.tag_id == tgt))
-        )
-        if tgt_has:
-            await self.session.execute(
-                text("DELETE FROM tag_allowlist WHERE tag_id = :src"),
-                {"src": src},
-            )
-        else:
-            await self.session.execute(
-                update(TagAllowlist)
-                .where(TagAllowlist.tag_id == src)
-                .values(tag_id=tgt)
-            )
-
-    async def _repoint_embedding(self, src: int) -> None:
-        await self.session.execute(
-            text(
-                "DELETE FROM tag_reference_embedding WHERE tag_id = :src"
-            ),
-            {"src": src},
-        )
-
     async def _repoint_aliases(self, src: int, tgt: int) -> None:
         from ..models.tag_alias import TagAlias
 

backend/app/tasks/ml.py

@@ -1,20 +1,19 @@
-"""ML Celery tasks: per-image inference, backfill discovery, centroid
-recompute, allowlist auto-apply, model self-heal.
+"""ML Celery tasks: per-image embedding, backfill discovery, head training,
+model self-heal.
 
-All run on the ml-worker (queue 'ml') except recompute_centroids and
-apply_allowlist_tags sweeps which are 'maintenance' lane. Sync sessions
-(Celery workers are sync processes), same pattern as FC-2a tasks.
+All run on the ml-worker (queue 'ml'). Sync sessions (Celery workers are sync
+processes), same pattern as FC-2a tasks.
 """
 
 import logging
 from pathlib import Path
 
 from celery.exceptions import SoftTimeLimitExceeded
-from sqlalchemy import delete, select
+from sqlalchemy import select
 from sqlalchemy.exc import DBAPIError, OperationalError
 
 from ..celery_app import celery
-from ..models import ImagePrediction, ImageRecord, MLSettings
+from ..models import ImageRecord, MLSettings
 from ._sync_engine import sync_session_factory as _sync_session_factory
 
 log = logging.getLogger(__name__)
@@ -46,19 +45,16 @@ def _is_video(path: Path) -> bool:
     time_limit=1200,       # 20 min hard
 )
 def tag_and_embed(self, image_id: int) -> dict:
-    """Run Camie + SigLIP on one image; store predictions + embedding;
-    then enqueue per-image allowlist application.
+    """Compute + store one image's SigLIP embedding.
 
     Video (#747): sample frames at a fixed cadence (ml_settings
-    video_frame_interval_seconds, capped at video_max_frames), keep a tag only if
-    it appears in >= video_min_tag_frames frames and average its confidence over
-    those frames (mean-pool, not max — kills one-frame noise); mean-pool the
-    SigLIP embeddings. On no-frames returns status='no_frames' (not an error).
+    video_frame_interval_seconds, capped at video_max_frames) and mean-pool the
+    per-frame SigLIP embeddings. On no-frames returns status='no_frames' (not an
+    error). (Camie tagging was retired #1189 — heads + CCIP are the tag source.)
     """
     import time
 
     from ..services.ml.embedder import get_embedder
-    from ..services.ml.tagger import get_tagger
 
     # Phase + file context, so a timeout/crash names WHICH file and WHERE it
     # died instead of a bare SoftTimeLimitExceeded() (operator-flagged 2026-06-08:
@@ -94,15 +90,13 @@ def tag_and_embed(self, image_id: int) -> dict:
                 return {"status": "file_missing", "image_id": image_id}
 
             phase = "load_models"
-            tagger = get_tagger()
-            embedder = get_embedder()
+            embedder = get_embedder(settings.embedder_model_name)
 
             if is_vid:
                 # Layer-3 isolation: ffprobe (a separate process) validates
-                # the container before we burn ~20 GPU ops sampling frames
-                # from it. A corrupt video that would crash the frame
-                # decoder is rejected cleanly here instead of taking down
-                # the ml-worker. Operator-flagged 2026-05-28.
+                # the container before we burn GPU ops sampling frames from it.
+                # A corrupt video that would crash the frame decoder is rejected
+                # cleanly here instead of taking down the ml-worker.
                 phase = "video_probe"
                 from ..utils import safe_probe
                 vprobe = safe_probe.probe_video(src)
@@ -115,48 +109,23 @@ def tag_and_embed(self, image_id: int) -> dict:
                         "reason": vprobe.reason,
                     }
                 phase = "video_sample_frames"
-                t0 = time.monotonic()
                 frames = _sample_video_frames(
                     src,
                     interval=settings.video_frame_interval_seconds,
                     max_frames=settings.video_max_frames,
                 )
-                log.info(
-                    "tag_and_embed sampled %d frame(s) in %.1fs: %s",
-                    len(frames), time.monotonic() - t0, ctx,
-                )
                 if not frames:
                     return {"status": "no_frames", "image_id": image_id}
-                phase = "video_infer"
+                phase = "video_embed"
                 import numpy as np
 
-                preds = _aggregate_video_predictions(
-                    [tagger.infer(f, store_floor=settings.tagger_store_floor)
-                     for f in frames],
-                    min_frames=settings.video_min_tag_frames,
-                )
+                # Mean-pool the per-frame SigLIP embeddings into one vector.
                 embedding = np.mean(
                     [embedder.infer(f) for f in frames], axis=0
                 ).astype("float32")
-                log.info(
-                    "tag_and_embed video aggregated %d tag(s) from %d frame(s) "
-                    "(min_frames=%d): %s",
-                    len(preds), len(frames), settings.video_min_tag_frames, ctx,
-                )
                 for f in frames:
                     f.unlink(missing_ok=True)
             else:
-                phase = "tag"
-                t0 = time.monotonic()
-                raw = tagger.infer(src, store_floor=settings.tagger_store_floor)
-                log.info(
-                    "tag_and_embed tagged in %.1fs (%d tags): %s",
-                    time.monotonic() - t0, len(raw), ctx,
-                )
-                preds = {
-                    name: {"category": p.category, "confidence": p.confidence}
-                    for name, p in raw.items()
-                }
                 phase = "embed"
                 t0 = time.monotonic()
                 embedding = embedder.infer(src)
@@ -166,28 +135,9 @@ def tag_and_embed(self, image_id: int) -> dict:
                 )
 
             phase = "persist"
-            record.tagger_model_version = settings.tagger_model_version
             record.siglip_embedding = embedding.tolist()
             record.siglip_model_version = settings.embedder_model_version
             session.add(record)
-            # Write the normalized image_prediction rows (#768) — the sole home
-            # for predictions now (image_record.tagger_predictions was dropped in
-            # migration 0046). Delete-then-insert keeps a re-tag idempotent;
-            # tagger_store_floor was already applied in tagger.infer, so preds is
-            # the >=floor set.
-            session.execute(
-                delete(ImagePrediction).where(
-                    ImagePrediction.image_record_id == image_id
-                )
-            )
-            session.add_all([
-                ImagePrediction(
-                    image_record_id=image_id, raw_name=name,
-                    category=p.get("category", "general"),
-                    score=float(p.get("confidence", 0.0)),
-                )
-                for name, p in preds.items()
-            ])
             session.commit()
     except SoftTimeLimitExceeded:
         log.error(
@@ -210,11 +160,8 @@ def tag_and_embed(self, image_id: int) -> dict:
         )
         raise
 
-    log.info(
-        "tag_and_embed ok in %.1fs (%d tags): %s", _elapsed(), len(preds), ctx
-    )
-    apply_allowlist_tags.delay(image_id=image_id)
-    return {"status": "ok", "image_id": image_id, "tags": len(preds)}
+    log.info("tag_and_embed ok in %.1fs: %s", _elapsed(), ctx)
+    return {"status": "ok", "image_id": image_id}
 
 
 def _sample_video_frames(
@@ -273,68 +220,24 @@ def _sample_video_frames(
     return out
 
 
-def _aggregate_video_predictions(per_frame: list[dict], *, min_frames: int) -> dict:
-    """Aggregate per-frame {name: TagPrediction} into one prediction set (#747).
-
-    A tag is kept only if it appears (≥ the tagger store floor, already applied)
-    in at least `min_frames` of the sampled frames — because sampling is at a
-    fixed cadence, that means it was on screen for roughly min_frames×interval
-    seconds, so a single-frame flicker / scene-transition artifact is dropped
-    while a genuine scene-local tag in a long video survives. Confidence is the
-    MEAN over the frames where the tag appears (not max — max re-inflated the
-    one-frame noise this whole change exists to remove).
-
-    `min_frames` is clamped to the number of frames actually sampled so a very
-    short video (1–2 frames) still tags instead of dropping everything.
-    """
-    n = len(per_frame)
-    if n == 0:
-        return {}
-    threshold = max(1, min(min_frames, n))
-    agg: dict[str, dict] = {}
-    for frame_preds in per_frame:
-        for name, p in frame_preds.items():
-            cur = agg.get(name)
-            if cur is None:
-                agg[name] = {"category": p.category, "sum": p.confidence, "count": 1}
-            else:
-                cur["sum"] += p.confidence
-                cur["count"] += 1
-    return {
-        name: {"category": v["category"], "confidence": v["sum"] / v["count"]}
-        for name, v in agg.items()
-        if v["count"] >= threshold
-    }
-
-
 @celery.task(name="backend.app.tasks.ml.backfill", bind=True)
 def backfill(self) -> int:
-    """Enqueue tag_and_embed for images missing predictions/embeddings for
-    the current model versions. Keyset pagination by id ASC (restart-safe).
+    """Enqueue tag_and_embed (embed-only) for images with no SigLIP embedding.
+    Keyset pagination by id ASC (restart-safe).
+
+    NB: a siglip MODEL-VERSION mismatch (an operator model swap, #1190) is NOT
+    re-embedded here — the CPU ml-worker can't churn the library at 384/512px;
+    the GPU agent owns version re-embeds via the 'embed' job.
     """
     SessionLocal = _sync_session_factory()
     enqueued = 0
     last_id = 0
     with SessionLocal() as session:
-        settings = session.execute(
-            select(MLSettings).where(MLSettings.id == 1)
-        ).scalar_one()
         while True:
             rows = session.execute(
                 select(ImageRecord.id)
                 .where(ImageRecord.id > last_id)
-                .where(
-                    (ImageRecord.tagger_model_version.is_(None))
-                    | (
-                        ImageRecord.tagger_model_version
-                        != settings.tagger_model_version
-                    )
-                    | (ImageRecord.siglip_embedding.is_(None))
-                    | (
-                        ImageRecord.siglip_model_version
-                        != settings.embedder_model_version
-                    )
-                )
+                .where(ImageRecord.siglip_embedding.is_(None))
                 .order_by(ImageRecord.id.asc())
                 .limit(500)
             ).scalars().all()
@@ -347,199 +250,6 @@ def backfill(self) -> int:
     return enqueued
 
 
-@celery.task(
-    name="backend.app.tasks.ml.apply_allowlist_tags",
-    bind=True,
-    # Audit 2026-06-02 — the full-sweep mode (neither tag_id nor image_id)
-    # is O(images × allowlist) and legitimately runs >5 min on large
-    # libraries. Cap matches the maintenance queue's recovery threshold.
-    soft_time_limit=1800, time_limit=2100,
-)
-def apply_allowlist_tags(self, tag_id: int | None = None,
-                         image_id: int | None = None) -> int:
-    """Retroactively apply allowlisted tags.
-
-    Modes:
-    - tag_id only      : scan all images for this tag.
-    - image_id only    : scan all allowlisted tags for this image.
-    - both             : just the (image, tag) pair.
-    - neither          : full sweep (daily beat).
-
-    Skips: already-applied, rejected (tag_suggestion_rejection), or
-    confidence below the tag's allowlist min_confidence. Applied with
-    source='ml_auto'.
-    """
-    from sqlalchemy import and_
-    from sqlalchemy import select as sa_select
-    from sqlalchemy.dialects.postgresql import insert as pg_insert
-
-    from ..models import TagAllowlist, TagSuggestionRejection
-    from ..models.tag import image_tag
-
-    SessionLocal = _sync_session_factory()
-    applied = 0
-    with SessionLocal() as session:
-        allow_rows = session.execute(
-            sa_select(TagAllowlist.tag_id, TagAllowlist.min_confidence)
-            if tag_id is None
-            else sa_select(
-                TagAllowlist.tag_id, TagAllowlist.min_confidence
-            ).where(TagAllowlist.tag_id == tag_id)
-        ).all()
-        allow = {r[0]: r[1] for r in allow_rows}
-        if not allow:
-            return 0
-
-        # Images that have any predictions (#768: from image_prediction, not
-        # the old JSON column), optionally narrowed to one image.
-        img_ids_query = sa_select(ImagePrediction.image_record_id).distinct()
-        if image_id is not None:
-            img_ids_query = img_ids_query.where(
-                ImagePrediction.image_record_id == image_id
-            )
-
-        for (img_id,) in session.execute(img_ids_query).all():
-            preds = _load_predictions_sync(session, img_id)
-            for a_tag_id, min_conf in allow.items():
-                exists = session.execute(
-                    sa_select(image_tag.c.tag_id).where(
-                        and_(
-                            image_tag.c.image_record_id == img_id,
-                            image_tag.c.tag_id == a_tag_id,
-                        )
-                    )
-                ).scalar_one_or_none()
-                if exists is not None:
-                    continue
-                rej = session.get(
-                    TagSuggestionRejection, (img_id, a_tag_id)
-                )
-                if rej is not None:
-                    continue
-                from ..models import Tag
-
-                tag = session.get(Tag, a_tag_id)
-                if tag is None:
-                    continue
-                conf = _confidence_for_tag(session, tag, preds)
-                if conf is None or conf < min_conf:
-                    continue
-                stmt = pg_insert(image_tag).values(
-                    image_record_id=img_id,
-                    tag_id=a_tag_id,
-                    source="ml_auto",
-                )
-                stmt = stmt.on_conflict_do_nothing(
-                    index_elements=["image_record_id", "tag_id"]
-                )
-                session.execute(stmt)
-                applied += 1
-        session.commit()
-    return applied
-
-
-def _load_predictions_sync(session, image_id: int) -> dict:
-    """Predictions for one image from image_prediction (#768), in the
-    {raw_name: {category, confidence}} shape _confidence_for_tag consumes —
-    keeps the allowlist resolution logic unchanged."""
-    from sqlalchemy import select as sa_select
-
-    rows = session.execute(
-        sa_select(
-            ImagePrediction.raw_name,
-            ImagePrediction.category,
-            ImagePrediction.score,
-        ).where(ImagePrediction.image_record_id == image_id)
-    ).all()
-    return {
-        r.raw_name: {"category": r.category, "confidence": r.score}
-        for r in rows
-    }
-
-
-def _confidence_for_tag(session, tag, preds: dict) -> float | None:
-    """Highest confidence among predictions that resolve to `tag` —
-    either the prediction name equals the tag name, or an alias maps
-    (prediction name, category) -> tag.id.
-    """
-    from sqlalchemy import select as sa_select
-
-    from ..models import TagAlias
-
-    best: float | None = None
-    direct = preds.get(tag.name)
-    if direct is not None:
-        best = float(direct.get("confidence", 0.0))
-    alias_rows = session.execute(
-        sa_select(TagAlias.alias_string, TagAlias.alias_category).where(
-            TagAlias.canonical_tag_id == tag.id
-        )
-    ).all()
-    for alias_string, alias_category in alias_rows:
-        p = preds.get(alias_string)
-        if p is None:
-            continue
-        if p.get("category") != alias_category:
-            continue
-        c = float(p.get("confidence", 0.0))
-        if best is None or c > best:
-            best = c
-    return best
-
-
-@celery.task(name="backend.app.tasks.ml.recompute_centroid", bind=True)
-def recompute_centroid(self, tag_id: int) -> bool:
-    import asyncio
-
-    from ..services.ml.centroids import CentroidService
-    from ._async_session import async_session_factory
-
-    async def _run() -> bool:
-        # Per-task NullPool engine bound to THIS asyncio.run loop — the shared
-        # process-wide engine reuses connections across loops and raises
-        # "Future attached to a different loop" on every call after the first.
-        async_factory, async_engine = async_session_factory()
-        try:
-            async with async_factory() as session:
-                svc = CentroidService(session)
-                result = await svc.recompute_for_tag(tag_id)
-                await session.commit()
-                return result
-        finally:
-            await async_engine.dispose()
-
-    return asyncio.run(_run())
-
-
-@celery.task(
-    name="backend.app.tasks.ml.recompute_centroids",
-    bind=True,
-    # Audit 2026-06-02 — drifted-centroid rebuild over potentially
-    # hundreds of tags.
-    soft_time_limit=1800, time_limit=2100,
-)
-def recompute_centroids(self) -> int:
-    """Daily: find drifted centroids, enqueue recompute_centroid for each."""
-    import asyncio
-
-    from ..services.ml.centroids import CentroidService
-    from ._async_session import async_session_factory
-
-    async def _list() -> list[int]:
-        # Per-task NullPool engine bound to this loop (see recompute_centroid).
-        async_factory, async_engine = async_session_factory()
-        try:
-            async with async_factory() as session:
-                return await CentroidService(session).list_drifted()
-        finally:
-            await async_engine.dispose()
-
-    drifted = asyncio.run(_list())
-    for tid in drifted:
-        recompute_centroid.delay(tid)
-    return len(drifted)
-
-
 @celery.task(
     name="backend.app.tasks.ml.tag_eval_run",
     bind=True,
@@ -750,17 +460,40 @@ def enqueue_gpu_backfill(task_name: str) -> int:
     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."""
-    from sqlalchemy import exists, insert, literal
+    from sqlalchemy import exists, insert, literal, or_
     from sqlalchemy import select as sa_select
 
-    from ..models import GpuJob, ImageRecord, ImageRegion
+    from ..models import GpuJob, ImageRecord, ImageRegion, MLSettings
 
     SessionLocal = _sync_session_factory()
     with SessionLocal() as session:
-        if task_name == "siglip":
-            has_concept = exists().where(
+        cur_version = session.execute(
+            select(MLSettings.embedder_model_version).where(MLSettings.id == 1)
+        ).scalar_one()
+        if task_name == "embed":
+            # Whole-image GPU re-embed (#1190): images with no embedding, or one
+            # stamped under a DIFFERENT model version (an operator model swap).
+            stale = or_(
+                ImageRecord.siglip_embedding.is_(None),
+                ImageRecord.siglip_model_version.is_(None),
+                ImageRecord.siglip_model_version != cur_version,
+            )
+            queued = exists().where(
+                GpuJob.image_record_id == ImageRecord.id,
+                GpuJob.task == "embed",
+                GpuJob.status.in_(["pending", "leased"]),
+            )
+            sel = sa_select(
+                ImageRecord.id, literal("embed"), literal("pending")
+            ).where(stale).where(~queued)
+        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
+            # only the never-embedded back-catalogue.
+            has_current_concept = exists().where(
                 ImageRegion.image_record_id == ImageRecord.id,
                 ImageRegion.kind == "concept",
+                ImageRegion.embedding_version == cur_version,
             )
             queued = exists().where(
                 GpuJob.image_record_id == ImageRecord.id,
@@ -769,7 +502,7 @@ def enqueue_gpu_backfill(task_name: str) -> int:
             )
             sel = sa_select(
                 ImageRecord.id, literal("siglip"), literal("pending")
-            ).where(~has_concept).where(~queued)
+            ).where(~has_current_concept).where(~queued)
         else:
             already = exists().where(
                 GpuJob.image_record_id == ImageRecord.id,
@@ -816,6 +549,37 @@ def recover_orphaned_gpu_jobs() -> int:
         return res.rowcount or 0
 
 
+@celery.task(name="backend.app.tasks.ml.reprocess_gpu_jobs")
+def reprocess_gpu_jobs(task_name: str = "ccip") -> int:
+    """Reset every done/error job of `task_name` back to pending so the agent
+    re-runs the WHOLE library under the CURRENT pipeline — e.g. after adding crop
+    detectors (#1202), re-cropping existing images. Heavy + operator-triggered;
+    the back-catalogue won't otherwise re-process (the backfills skip images that
+    already have current-version regions). Returns the number reset."""
+    from datetime import UTC, datetime
+
+    from sqlalchemy import update
+
+    from ..models import GpuJob
+
+    SessionLocal = _sync_session_factory()
+    with SessionLocal() as session:
+        now = datetime.now(UTC)
+        res = session.execute(
+            update(GpuJob)
+            .where(
+                GpuJob.task == task_name,
+                GpuJob.status.in_(["done", "error"]),
+            )
+            .values(
+                status="pending", attempts=0, lease_token=None, leased_at=None,
+                lease_expires_at=None, updated_at=now,
+            )
+        )
+        session.commit()
+        return res.rowcount or 0
+
+
 @celery.task(
     name="backend.app.tasks.ml.scheduled_ccip_auto_apply",
     soft_time_limit=1800, time_limit=2100,

entrypoint.sh

@@ -12,9 +12,13 @@ case "$ROLE" in
     # create_app is a factory — the `()` tells hypercorn to call it once
     # and serve the returned Quart (ASGI) app, rather than treating the
     # function itself as the application (which it then mis-invokes as WSGI).
+    # Default 4 workers (was 2): each worker is one asyncio loop, and a large
+    # file download occupies its worker for the transfer — 2 was too few once the
+    # GPU agent + the browser's thumbnail grid hit /images concurrently (they
+    # queued behind each other). Env-tunable via HYPERCORN_WORKERS.
     exec hypercorn \
       --bind 0.0.0.0:8080 \
-      --workers "${HYPERCORN_WORKERS:-2}" \
+      --workers "${HYPERCORN_WORKERS:-4}" \
       --access-logfile - \
       "backend.app:create_app()"
     ;;

frontend/src/components/gallery/GalleryFilterBar.vue

@@ -142,9 +142,13 @@ const tagStore = useTagStore()
 const api = useApi()
 const router = useRouter()
 
+// posted_* sort by earliest publish across ALL of an image's posts (original
+// post date); newest/oldest sort by the primary post's date, else download date.
 const SORTS = [
-  { title: 'Newest first', value: 'newest' },
-  { title: 'Oldest first', value: 'oldest' },
+  { title: 'Newest post date', value: 'posted_new' },
+  { title: 'Oldest post date', value: 'posted_old' },
+  { title: 'Newest added', value: 'newest' },
+  { title: 'Oldest added', value: 'oldest' },
 ]
 
 const selected = ref(null)
@@ -175,7 +179,7 @@ const hasActiveFilters = computed(() =>
   store.filter.tag_or.length > 0 ||
   store.filter.artist_id != null ||
   store.filter.media_type != null ||
-  store.filter.sort !== 'newest' ||
+  store.filter.sort !== 'posted_new' ||
   store.filter.similar_to != null ||
   hasRefineFilters.value
 )

frontend/src/components/settings/AllowlistTable.vue

@@ -1,120 +0,0 @@
-<template>
-  <MaintenanceTile
-    icon="mdi-playlist-check"
-    :title="`Allowlisted tags (${store.rows.length})`"
-    blurb="Tags auto-applied to images that score above their threshold. Tune the
-           threshold and see how many images it would cover."
-  >
-    <v-data-table-virtual
-      :headers="headers" :items="store.rows" :loading="store.loading"
-      height="360" density="compact" fixed-header
-      no-data-text="No tags on the allowlist yet — accept a suggestion to add one."
-    >
-      <template #item.applied_count="{ item }">
-        <span class="fc-num">{{ item.applied_count ?? '—' }}</span>
-      </template>
-
-      <template #item.min_confidence="{ item }">
-        <div class="fc-thr">
-          <v-text-field
-            :model-value="item.min_confidence" type="number"
-            density="compact" hide-details style="max-width: 100px;"
-            :min="floor" max="1" step="0.05"
-            :aria-label="`Auto-apply threshold for ${item.tag_name}`"
-            @update:model-value="(v) => onThreshold(item, v)"
-          />
-          <span
-            v-if="proj[item.tag_id]"
-            class="fc-thr__proj"
-            :class="{ 'fc-thr__proj--loading': proj[item.tag_id].loading }"
-            :title="`At ${proj[item.tag_id].threshold}, a sweep would cover this many images`"
-          >≈ {{ proj[item.tag_id].count }} at {{ proj[item.tag_id].threshold }}</span>
-        </div>
-      </template>
-
-      <template #item.coverage_count="{ item }">
-        <span class="fc-num" :title="`Images a sweep covers at ${item.min_confidence}`">
-          {{ item.coverage_count ?? '—' }}
-        </span>
-      </template>
-
-      <template #item.actions="{ item }">
-        <v-btn
-          icon="mdi-delete" size="x-small" variant="text" color="error"
-          :aria-label="`Remove ${item.tag_name} from the allowlist`"
-          @click="store.remove(item.tag_id)"
-        />
-      </template>
-    </v-data-table-virtual>
-    <p class="fc-muted text-caption mt-2">
-      <strong>Applied</strong> = images currently carrying the tag.
-      <strong>Covers</strong> = images a sweep would auto-apply it to at the
-      current threshold. Lower the threshold to cover more (less certain) images.
-    </p>
-  </MaintenanceTile>
-</template>
-
-<script setup>
-import { computed, onMounted, reactive } from 'vue'
-import { useAllowlistStore } from '../../stores/allowlist.js'
-import { useMLStore } from '../../stores/ml.js'
-import MaintenanceTile from '../common/MaintenanceTile.vue'
-
-const store = useAllowlistStore()
-const ml = useMLStore()
-// min_confidence can't be set below the tagger store floor — predictions
-// below it aren't stored, so a lower threshold would behave identically to
-// the floor. The backend clamps too (#764).
-const floor = computed(() => ml.settings?.tagger_store_floor ?? 0.70)
-const headers = [
-  { title: 'Tag', key: 'tag_name', sortable: true },
-  { title: 'Kind', key: 'tag_kind', sortable: true, width: 100 },
-  { title: 'Applied', key: 'applied_count', sortable: true, width: 90 },
-  { title: 'Min confidence', key: 'min_confidence', sortable: false, width: 220 },
-  { title: 'Covers', key: 'coverage_count', sortable: true, width: 90 },
-  { title: '', key: 'actions', sortable: false, width: 56 }
-]
-
-// Per-row live projection while the operator drags a threshold:
-//   proj[tagId] = { threshold, count, loading }
-const proj = reactive({})
-
-onMounted(() => {
-  store.load()
-  if (!ml.settings) ml.loadSettings()
-})
-
-const debounces = {}
-function onThreshold(item, value) {
-  const tagId = item.tag_id
-  const v = Math.max(parseFloat(value), floor.value)
-  if (!(v > 0 && v <= 1)) return
-  const shown = Number(v.toFixed(2))
-  // Optimistic live projection box (loading until the count returns).
-  proj[tagId] = { threshold: shown, count: proj[tagId]?.count ?? '…', loading: true }
-  if (debounces[tagId]) clearTimeout(debounces[tagId])
-  debounces[tagId] = setTimeout(async () => {
-    try {
-      const { count } = await store.coverage(tagId, v)
-      proj[tagId] = { threshold: shown, count, loading: false }
-    } catch {
-      delete proj[tagId]  // drop the projection rather than show a wrong number
-    }
-    // Commit the new threshold (also refreshes the row's stored coverage_count).
-    store.updateThreshold(tagId, v)
-  }, 500)
-}
-</script>
-
-<style scoped>
-.fc-num { font-variant-numeric: tabular-nums; }
-.fc-thr { display: flex; align-items: center; gap: 10px; }
-.fc-thr__proj {
-  font-size: 12px;
-  font-variant-numeric: tabular-nums;
-  color: rgb(var(--v-theme-accent));
-  white-space: nowrap;
-}
-.fc-thr__proj--loading { color: rgb(var(--v-theme-on-surface-variant)); }
-.fc-muted { color: rgb(var(--v-theme-on-surface-variant)); }
-</style>

frontend/src/components/settings/CentroidRecomputeCard.vue

@@ -1,36 +0,0 @@
-<template>
-  <MaintenanceTile
-    icon="mdi-vector-triangle"
-    title="Tag centroids"
-    blurb="Rebuild SigLIP centroids for similarity suggestions."
-    :open="busy"
-  >
-      <p class="text-body-2 mb-3">
-        Rebuild the per-tag SigLIP centroids that power similarity-based
-        suggestions. Runs nightly automatically; trigger manually after a
-        large tagging session.
-      </p>
-      <v-btn color="primary" rounded="pill" :loading="busy" @click="run">
-        <v-icon start>mdi-vector-triangle</v-icon> Recompute centroids
-      </v-btn>
-      <span v-if="done" class="ml-3 text-caption">Enqueued.</span>
-      <QueueStatusBar queue="ml" queue-label="ML" />
-  </MaintenanceTile>
-</template>
-
-<script setup>
-import { toast } from '../../utils/toast.js'
-import { ref } from 'vue'
-import { useMLStore } from '../../stores/ml.js'
-import MaintenanceTile from '../common/MaintenanceTile.vue'
-import QueueStatusBar from './QueueStatusBar.vue'
-const store = useMLStore()
-const busy = ref(false)
-const done = ref(false)
-async function run() {
-  busy.value = true
-  try { await store.triggerRecomputeCentroids(); done.value = true }
-  catch (e) { toast({ text: e.message, type: 'error' }) }
-  finally { busy.value = false }
-}
-</script>

frontend/src/components/settings/GpuAgentCard.vue

@@ -71,6 +71,16 @@
       images get these automatically; this catches the back-catalogue.
     </p>
 
+    <v-btn
+      class="mt-3" color="warning" variant="tonal" rounded="pill" size="small"
+      prepend-icon="mdi-backup-restore" :loading="reprocessing" @click="onReprocess"
+    >Re-process library (re-detect + re-crop)</v-btn>
+    <p class="fc-muted text-caption mt-2 mb-0">
+      Re-runs the FULL pipeline (figure detection + CCIP + concept/panel crops) on
+      <b>every</b> image — use after changing crop detectors so the back-catalogue
+      gets re-cropped, not just new images. Heavy: re-processes the whole library.
+    </p>
+
     <!-- Match strictness -->
     <div class="fc-section-h mt-5 mb-1">Character-match strictness</div>
     <div v-if="ml.settings" class="d-flex align-center" style="gap:12px">
@@ -106,6 +116,33 @@
       reversible) — so identity tags keep flowing without review. Stricter than
       the suggest cut; 0.92 recommended.
     </p>
+
+    <!-- Embedding model -->
+    <div v-if="ml.settings" class="fc-section-h mt-5 mb-1">Embedding model</div>
+    <div v-if="ml.settings">
+      <v-select
+        v-model="selectedModel" :items="modelItems" item-title="label"
+        item-value="name" label="Model" density="compact" hide-details
+        variant="outlined"
+      />
+      <div class="d-flex mt-3" style="gap:8px">
+        <v-btn
+          size="small" variant="tonal" rounded="pill" :loading="savingModel"
+          prepend-icon="mdi-content-save" @click="onSaveModel"
+        >Save model</v-btn>
+        <v-btn
+          size="small" color="accent" variant="flat" rounded="pill"
+          :loading="reembedding" prepend-icon="mdi-backup-restore" @click="onReembed"
+        >Re-embed library (GPU)</v-btn>
+      </div>
+      <p class="fc-muted text-caption mt-2 mb-0">
+        Switching the model is a DIFFERENT embedding space. After <b>Save model</b>,
+        run <b>Re-embed library</b> (the GPU agent re-embeds whole images + concept
+        crops), then <b>Retrain heads</b> — suggestions degrade until both finish.
+        SigLIP 2 (512px) is a 1152-d drop-in over SigLIP 1; new installs default to
+        it. Your existing library stays on its current model until you re-embed.
+      </p>
+    </div>
   </MaintenanceTile>
 </template>
 
@@ -126,11 +163,16 @@ const masked = ref(true)
 const rotating = ref(false)
 const backfilling = ref(false)
 const backfillingSiglip = ref(false)
+const reprocessing = ref(false)
 const threshold = ref(0.85)
 const savingThreshold = ref(false)
 const autoApply = ref(true)
 const autoThreshold = ref(0.92)
 const savingAuto = ref(false)
+const modelItems = ref([])
+const selectedModel = ref(null)
+const savingModel = ref(false)
+const reembedding = ref(false)
 const queue = ref({ pending: 0, leased: 0, done: 0, error: 0 })
 let pollTimer = null
 
@@ -157,9 +199,50 @@ onMounted(async () => {
       autoApply.value = ml.settings.ccip_auto_apply_enabled
       autoThreshold.value = ml.settings.ccip_auto_apply_threshold
     }
+    if (ml.settings?.embedder_model_name != null) {
+      const items = await ml.embedderModels()
+      // Make sure the current model is selectable even if it's not in the list.
+      const cur = ml.settings.embedder_model_name
+      if (!items.some((m) => m.name === cur)) {
+        items.push({ name: cur, version: ml.settings.embedder_model_version, label: `${cur} (current)` })
+      }
+      modelItems.value = items
+      selectedModel.value = cur
+    }
   } catch { /* non-fatal */ }
 })
 
+async function onSaveModel() {
+  const opt = modelItems.value.find((m) => m.name === selectedModel.value)
+  if (!opt) return
+  savingModel.value = true
+  try {
+    await ml.patchSettings({
+      embedder_model_name: opt.name,
+      embedder_model_version: opt.version,
+    })
+    toast({ text: 'Embedding model saved — now Re-embed library, then Retrain heads', type: 'success' })
+  } catch (e) {
+    toast({ text: `Could not save model: ${e.message}`, type: 'error' })
+  } finally {
+    savingModel.value = false
+  }
+}
+
+async function onReembed() {
+  reembedding.value = true
+  try {
+    await store.backfill('embed')
+    await store.backfill('siglip')
+    toast({ text: 'Queued whole-image + concept re-embed — run the agent, then Retrain heads', type: 'success' })
+    await refreshQueue()
+  } catch (e) {
+    toast({ text: `Could not queue re-embed: ${e.message}`, type: 'error' })
+  } finally {
+    reembedding.value = false
+  }
+}
+
 async function onSaveAuto() {
   savingAuto.value = true
   try {
@@ -239,6 +322,20 @@ async function onBackfillSiglip() {
     backfillingSiglip.value = false
   }
 }
+
+async function onReprocess() {
+  if (!window.confirm('Re-process the ENTIRE library (re-detect + re-crop every image)? This is heavy and runs on the GPU agent.')) return
+  reprocessing.value = true
+  try {
+    await store.reprocess('ccip')
+    toast({ text: 'Library queued for re-processing — run the agent to drain it', type: 'success' })
+    await refreshQueue()
+  } catch (e) {
+    toast({ text: `Could not start re-process: ${e.message}`, type: 'error' })
+  } finally {
+    reprocessing.value = false
+  }
+}
 </script>
 
 <style scoped>

frontend/src/components/settings/MLBackfillCard.vue

@@ -2,12 +2,13 @@
   <MaintenanceTile
     icon="mdi-refresh"
     title="ML backfill"
-    blurb="Re-run tagging + embeddings on images missing them."
+    blurb="Compute SigLIP embeddings on images missing them."
     :open="busy"
   >
       <p class="text-body-2 mb-3">
-        Re-run Camie + SigLIP on images missing predictions or embeddings
-        for the current model versions. Safe to re-run.
+        Compute the SigLIP embedding for any image that doesn't have one yet
+        (CPU). Safe to re-run. To re-embed under a NEW model, use the GPU
+        agent's "Re-embed library" instead.
       </p>
       <v-btn color="primary" rounded="pill" :loading="busy" @click="run">
         <v-icon start>mdi-refresh</v-icon> Run backfill now

frontend/src/components/settings/MLThresholdSliders.vue

@@ -1,70 +1,30 @@
 <template>
   <MaintenanceTile
-    icon="mdi-tune"
-    title="Suggestion thresholds"
-    blurb="Confidence cutoffs that gate auto-suggested tags + video sampling."
+    icon="mdi-filmstrip"
+    title="Video embedding"
+    blurb="How videos are sampled into frames before embedding."
   >
     <div v-if="store.settings">
-      <v-row v-for="f in fields" :key="f.key">
-        <v-col cols="12">
-          <v-slider
-            v-model="local[f.key]" :label="f.label"
-            :min="f.floorMin ? local.tagger_store_floor : 0" max="1" step="0.05"
-            thumb-label hide-details
-            color="accent" @end="save"
-          />
-        </v-col>
-      </v-row>
-
-      <v-divider class="my-4" />
-
-      <v-row>
-        <v-col cols="12">
-          <v-slider
-            v-model="local.tagger_store_floor" label="Tagger store floor"
-            min="0" max="1" step="0.05" thumb-label hide-details
-            color="accent" @end="save"
-          />
-          <div class="text-caption fc-muted mt-1">
-            Tagger predictions below this confidence aren't stored — raising it
-            keeps the image library lean. Suggestions can't be shown below the
-            floor; lower-confidence tags you actually want still surface through
-            the learned centroid path.
-          </div>
-        </v-col>
-      </v-row>
-
-      <v-divider class="my-4" />
-
-      <div class="text-subtitle-2 mb-1">Video tagging</div>
       <div class="text-caption fc-muted mb-3">
-        Videos are tagged by sampling frames at a fixed cadence. A tag is kept
-        only if it shows up in enough frames (≈ that many × the interval in
-        seconds of screen time), which filters one-frame noise without losing
-        tags that only appear in part of a longer video.
+        Videos are embedded by sampling frames at a fixed cadence and mean-pooling
+        their SigLIP embeddings. The interval sets the cadence; the cap bounds how
+        many frames a long video samples.
       </div>
       <v-row>
-        <v-col cols="12" sm="4">
+        <v-col cols="12" sm="6">
           <v-text-field
             v-model.number="local.video_frame_interval_seconds"
             label="Frame interval (s)" type="number" min="0.5" step="0.5"
             density="comfortable" hide-details @change="save"
           />
         </v-col>
-        <v-col cols="12" sm="4">
+        <v-col cols="12" sm="6">
           <v-text-field
             v-model.number="local.video_max_frames"
             label="Max frames" type="number" min="1" step="1"
             density="comfortable" hide-details @change="save"
           />
         </v-col>
-        <v-col cols="12" sm="4">
-          <v-text-field
-            v-model.number="local.video_min_tag_frames"
-            label="Min frames per tag" type="number" min="1" step="1"
-            density="comfortable" hide-details @change="save"
-          />
-        </v-col>
       </v-row>
     </div>
     <div v-else><v-skeleton-loader type="paragraph" /></div>
@@ -78,32 +38,14 @@ import { useMLStore } from '../../stores/ml.js'
 import MaintenanceTile from '../common/MaintenanceTile.vue'
 
 const store = useMLStore()
-// 'artist' (FC-2d-vii-c) and 'copyright' (2026-06-01) retired as
-// suggestion categories; their threshold rows are gone.
-// floorMin: the per-category suggestion thresholds can't drop below the
-// tagger store floor (nothing below the floor is stored to surface).
-const fields = [
-  { key: 'suggestion_threshold_character', label: 'Character', floorMin: true },
-  { key: 'suggestion_threshold_general', label: 'General', floorMin: true },
-  { key: 'centroid_similarity_threshold', label: 'Centroid similarity' }
-]
 const local = reactive({})
 watch(() => store.settings, (s) => { if (s) Object.assign(local, s) }, { immediate: true })
 
 async function save() {
-  // Mirror the server invariant: keep the category thresholds at or above the
-  // store floor so a raised floor doesn't leave a threshold stranded below it.
-  const floor = local.tagger_store_floor
-  local.suggestion_threshold_character = Math.max(local.suggestion_threshold_character, floor)
-  local.suggestion_threshold_general = Math.max(local.suggestion_threshold_general, floor)
-  // Mirror the server invariant: a tag can't require more frames than are sampled.
-  local.video_min_tag_frames = Math.min(local.video_min_tag_frames, local.video_max_frames)
-  const patch = {}
-  for (const f of fields) patch[f.key] = local[f.key]
-  patch.tagger_store_floor = local.tagger_store_floor
-  patch.video_frame_interval_seconds = local.video_frame_interval_seconds
-  patch.video_max_frames = local.video_max_frames
-  patch.video_min_tag_frames = local.video_min_tag_frames
+  const patch = {
+    video_frame_interval_seconds: local.video_frame_interval_seconds,
+    video_max_frames: local.video_max_frames
+  }
   try { await store.patchSettings(patch) }
   catch (e) { toast({ text: e.message, type: 'error' }) }
 }

frontend/src/components/settings/MaintenancePanel.vue

@@ -1,9 +1,8 @@
 <template>
   <div class="fc-maint">
     <p class="fc-muted text-body-2 mb-5">
-      One-off backfills, tagging config and storage tools. The ML backfill and
-      centroid recompute also run nightly; the allowlist auto-applies accepted
-      tags. Click a tile to open it.
+      One-off backfills, tagging config and storage tools. Heads train nightly
+      and auto-apply earned tags. Click a tile to open it.
     </p>
 
     <section class="fc-section">
@@ -11,7 +10,6 @@
       <p class="fc-section__hint">Re-run tagging, thumbnails, extraction and DB upkeep.</p>
       <div class="fc-tile-grid">
         <MLBackfillCard />
-        <CentroidRecomputeCard />
         <ThumbnailBackfillCard />
         <ArchiveReextractCard />
         <MissingFileRepairCard />
@@ -28,7 +26,6 @@
         <MLThresholdSliders />
         <HeadsCard />
         <GpuAgentCard />
-        <AllowlistTable />
         <AliasTable />
         <TagEvalCard />
       </div>
@@ -48,7 +45,6 @@
 import { onMounted, onUnmounted } from 'vue'
 
 import MLBackfillCard from './MLBackfillCard.vue'
-import CentroidRecomputeCard from './CentroidRecomputeCard.vue'
 import ThumbnailBackfillCard from './ThumbnailBackfillCard.vue'
 import ArchiveReextractCard from './ArchiveReextractCard.vue'
 import MissingFileRepairCard from './MissingFileRepairCard.vue'
@@ -56,7 +52,6 @@ import DbMaintenanceCard from './DbMaintenanceCard.vue'
 import MLThresholdSliders from './MLThresholdSliders.vue'
 import HeadsCard from './HeadsCard.vue'
 import GpuAgentCard from './GpuAgentCard.vue'
-import AllowlistTable from './AllowlistTable.vue'
 import AliasTable from './AliasTable.vue'
 import TagEvalCard from './TagEvalCard.vue'
 import BackupCard from './BackupCard.vue'

frontend/src/stores/allowlist.js

@@ -1,44 +0,0 @@
-import { defineStore } from 'pinia'
-import { ref } from 'vue'
-import { useApi } from '../composables/useApi.js'
-
-export const useAllowlistStore = defineStore('allowlist', () => {
-  const api = useApi()
-  const rows = ref([])
-  const loading = ref(false)
-
-  async function load() {
-    loading.value = true
-    try { rows.value = await api.get('/api/allowlist') }
-    finally { loading.value = false }
-  }
-
-  async function updateThreshold(tagId, minConfidence) {
-    await api.patch(`/api/tags/${tagId}/allowlist`, {
-      body: { min_confidence: minConfidence }
-    })
-    const r = rows.value.find(x => x.tag_id === tagId)
-    if (r) {
-      r.min_confidence = minConfidence
-      // The committed threshold changed the covered pool — refresh the row's
-      // coverage so the table stays truthful after a save.
-      try { r.coverage_count = (await coverage(tagId, minConfidence)).count }
-      catch { /* leave the stale count rather than blank it */ }
-    }
-  }
-
-  // Live "at threshold T, a sweep would cover ~N images" projection for the
-  // tuning dashboard. Returns { count, threshold }.
-  async function coverage(tagId, threshold) {
-    return api.get(`/api/tags/${tagId}/allowlist/coverage`, {
-      params: { threshold }
-    })
-  }
-
-  async function remove(tagId) {
-    await api.delete(`/api/tags/${tagId}/allowlist`)
-    rows.value = rows.value.filter(x => x.tag_id !== tagId)
-  }
-
-  return { rows, loading, load, updateThreshold, coverage, remove }
-})

frontend/src/stores/explore.js

@@ -93,13 +93,11 @@ export const useExploreStore = defineStore('explore', () => {
     // a crumb (which snaps the cursor back into the trail — the "loops back"
     // report). Fall back to the full set only if every neighbour's been seen.
     const seen = new Set(breadcrumb.value.map((c) => c.id))
-    let pool = neighbors.value.filter((n) => !seen.has(n.id))
-    if (!pool.length) pool = neighbors.value
-    // neighbors come similarity-sorted (nearest first). Skip the closest slice —
-    // those near-duplicates are exactly what you get stuck cycling through — and
-    // pick from the more-varied remainder, for real variance in the walk.
-    const skip = pool.length >= 6 ? Math.floor(pool.length / 3) : 0
-    const cands = pool.slice(skip)
+    const pool = neighbors.value.filter((n) => !seen.has(n.id))
+    const cands = pool.length ? pool : neighbors.value
+    // The list is already pHash-deduped + MMR-diversified server-side (it spans
+    // clusters, not 40 near-dupes), so a plain random pick gives real variance —
+    // no need to skip the nearest slice the way the raw nearest-list required.
     return cands[Math.floor(Math.random() * cands.length)].id
   }
 

frontend/src/stores/gallery.js

@@ -23,7 +23,7 @@ export const useGalleryStore = defineStore('gallery', () => {
   const error = ref(null)
   const filter = ref({
     tag_ids: [], artist_id: null, media_type: null,
-    sort: 'newest', post_id: null,
+    sort: 'posted_new', post_id: null,
     // #6 structured tag filter (AND-of-OR + exclude). tag_ids are the AND
     // "include" singletons (light editor + back-compat); tag_or is a list of
     // OR-groups (each group ORs, groups AND); tag_exclude is the NOT set.
@@ -154,7 +154,7 @@ export const useGalleryStore = defineStore('gallery', () => {
     if (filter.value.tag_exclude.length) p.tag_not = filter.value.tag_exclude.join(',')
     if (filter.value.artist_id) p.artist_id = filter.value.artist_id
     if (filter.value.media_type) p.media = filter.value.media_type
-    if (filter.value.sort && filter.value.sort !== 'newest') p.sort = filter.value.sort
+    if (filter.value.sort && filter.value.sort !== 'posted_new') p.sort = filter.value.sort
     if (filter.value.platform) p.platform = filter.value.platform
     if (filter.value.untagged) p.untagged = '1'
     if (filter.value.no_artist) p.no_artist = '1'
@@ -191,7 +191,7 @@ export const useGalleryStore = defineStore('gallery', () => {
       tag_exclude: _parseIds(q.tag_not),
       artist_id: _toId(q.artist_id),
       media_type: ['image', 'video'].includes(q.media) ? q.media : null,
-      sort: q.sort === 'oldest' ? 'oldest' : 'newest',
+      sort: ['newest', 'oldest', 'posted_new', 'posted_old'].includes(q.sort) ? q.sort : 'posted_new',
       post_id: _toId(q.post_id),
       platform: q.platform || null,
       untagged: _truthy(q.untagged),
@@ -297,7 +297,7 @@ export function filterToQuery(f) {
   if (f.tag_exclude?.length) q.tag_not = f.tag_exclude.join(',')
   if (f.artist_id) q.artist_id = String(f.artist_id)
   if (f.media_type) q.media = f.media_type
-  if (f.sort && f.sort !== 'newest') q.sort = f.sort
+  if (f.sort && f.sort !== 'posted_new') q.sort = f.sort
   if (f.platform) q.platform = f.platform
   if (f.untagged) q.untagged = '1'
   if (f.no_artist) q.no_artist = '1'

frontend/src/stores/gpu.js

@@ -29,5 +29,11 @@ export const useGpuStore = defineStore('gpu', () => {
     return await api.post('/api/gpu/backfill', { body: { task } })
   }
 
-  return { token, rotateToken, status, backfill }
+  // Reset every done/error `task` job to pending → re-run the WHOLE library
+  // under the current pipeline (e.g. after adding crop detectors).
+  async function reprocess(task = 'ccip') {
+    return await api.post('/api/gpu/reprocess', { body: { task } })
+  }
+
+  return { token, rotateToken, status, backfill, reprocess }
 })

frontend/src/stores/ml.js

@@ -22,12 +22,14 @@ export const useMLStore = defineStore('ml', () => {
     await api.post('/api/ml/backfill')
   }
 
-  async function triggerRecomputeCentroids() {
-    await api.post('/api/ml/recompute-centroids')
+  // Server-authoritative list of supported embedders for the model dropdown.
+  async function embedderModels() {
+    const r = await api.get('/api/ml/embedder-models')
+    return r.models || []
   }
 
   return {
     settings, loading, error,
-    loadSettings, patchSettings, triggerBackfill, triggerRecomputeCentroids
+    loadSettings, patchSettings, triggerBackfill, embedderModels
   }
 })

frontend/src/stores/suggestions.js

@@ -113,7 +113,7 @@ export const useSuggestionsStore = defineStore('suggestions', () => {
       })
       tagId = created.id
     }
-    const res = await api.post(`/api/images/${imageId}/suggestions/accept`, {
+    await api.post(`/api/images/${imageId}/suggestions/accept`, {
       body: { tag_id: tagId }
     })
     // Only drop from THIS image's category list — if the user navigated,
@@ -121,23 +121,14 @@ export const useSuggestionsStore = defineStore('suggestions', () => {
     if (currentImageId === imageId) {
       _dropEverywhere(suggestion)
     }
-    _acceptToast('Tagged', suggestion.display_name, res)
+    _acceptToast('Tagged', suggestion.display_name)
   }
 
-  // One non-blocking toast for accept/alias. When the accept newly allowlisted
-  // the tag, surface the coverage PROJECTION (#7) so the operator sees the
-  // auto-apply reach without a blocking pre-accept preview — the apply itself
-  // runs async, hence "~N".
-  function _acceptToast(verb, displayName, res) {
-    if (res?.allowlisted) {
-      const n = res.projected_count
-      toast({
-        text: `${verb}: ${displayName} — allowlisted, auto-applying to ~${n} image${n === 1 ? '' : 's'}`,
-        type: 'success'
-      })
-    } else {
-      toast({ text: `${verb}: ${displayName}`, type: 'success' })
-    }
+  // One non-blocking toast for accept/alias. The accepted tag is applied to this
+  // image and feeds head training; head auto-apply handles propagation (earned),
+  // so there's no instant fan-out to project.
+  function _acceptToast(verb, displayName) {
+    toast({ text: `${verb}: ${displayName}`, type: 'success' })
   }
 
   async function aliasAccept(suggestion, canonicalTagId) {
@@ -149,7 +140,7 @@ export const useSuggestionsStore = defineStore('suggestions', () => {
     // reappearing unaliased. raw_name is null only for centroid hits, which
     // can't be aliased (the UI hides the action for them).
     const aliasString = suggestion.raw_name ?? suggestion.display_name
-    const res = await api.post(`/api/images/${imageId}/suggestions/alias`, {
+    await api.post(`/api/images/${imageId}/suggestions/alias`, {
       body: {
         alias_string: aliasString,
         alias_category: suggestion.category,
@@ -159,7 +150,7 @@ export const useSuggestionsStore = defineStore('suggestions', () => {
     if (currentImageId === imageId) {
       _dropEverywhere(suggestion)
     }
-    _acceptToast('Aliased & tagged', suggestion.display_name, res)
+    _acceptToast('Aliased & tagged', suggestion.display_name)
   }
 
   // Remove the alias behind an aliased suggestion (the raw prediction reverts to

frontend/test/gallery.spec.js

@@ -43,7 +43,7 @@ describe('gallery store: composable filter', () => {
     expect(scroll).toContain('sort=oldest')
   })
 
-  it('omits sort=newest and sends no filter params when empty', async () => {
+  it('omits the default sort and sends no filter params when empty', async () => {
     const s = useGalleryStore()
     const urls = []
     stubFetch((url) => { urls.push(url); return { status: 200, body: EMPTY } })
@@ -197,7 +197,7 @@ describe('gallery store: composable filter', () => {
 describe('filterToQuery / cloneFilter', () => {
   it('serializes faceted params incl. platform sentinel and flags', () => {
     const q = filterToQuery({
-      tag_ids: [3], artist_id: null, media_type: null, sort: 'newest',
+      tag_ids: [3], artist_id: null, media_type: null, sort: 'posted_new',
       platform: '__unsourced__', untagged: true, no_artist: false,
       date_from: '2024-01-01', date_to: null,
     })
@@ -207,7 +207,7 @@ describe('filterToQuery / cloneFilter', () => {
     expect(q.no_artist).toBeUndefined()
     expect(q.date_from).toBe('2024-01-01')
     expect(q.date_to).toBeUndefined()
-    expect(q.sort).toBeUndefined()  // newest is the default → omitted
+    expect(q.sort).toBeUndefined()  // posted_new is the default → omitted
   })
 
   it('serializes similar_to', () => {

tests/_prediction_helpers.py

@@ -1,21 +0,0 @@
-"""#768 test helper: seed image_prediction rows.
-
-Read-path tests used to seed ImageRecord(tagger_predictions={...}); predictions
-now live in the normalized image_prediction table, so seed there instead.
-"""
-from backend.app.models import ImagePrediction
-
-
-async def seed_predictions(session, image_id: int, predictions: dict) -> None:
-    """Insert image_prediction rows from a {raw_name: {category, confidence}}
-    dict (the old JSON shape). Caller commits/flushes as needed; this flushes."""
-    session.add_all([
-        ImagePrediction(
-            image_record_id=image_id,
-            raw_name=name,
-            category=p.get("category", "general"),
-            score=float(p.get("confidence", 0.0)),
-        )
-        for name, p in predictions.items()
-    ])
-    await session.flush()

tests/test_api_allowlist.py

@@ -1,88 +0,0 @@
-import pytest
-
-from backend.app.models import ImagePrediction, ImageRecord, TagAllowlist, TagKind
-from backend.app.services.tag_service import TagService
-
-pytestmark = pytest.mark.integration
-
-
-@pytest.mark.asyncio
-async def test_list_and_patch_and_delete(client, db):
-    tag = await TagService(db).find_or_create("AL", TagKind.character)
-    db.add(TagAllowlist(tag_id=tag.id, min_confidence=0.95))
-    await db.commit()
-
-    resp = await client.get("/api/allowlist")
-    assert resp.status_code == 200
-    assert any(r["tag_id"] == tag.id for r in await resp.get_json())
-
-    resp = await client.patch(
-        f"/api/tags/{tag.id}/allowlist", json={"min_confidence": 0.80}
-    )
-    assert resp.status_code == 204
-
-    resp = await client.get(f"/api/tags/{tag.id}/allowlist")
-    assert (await resp.get_json())["min_confidence"] == pytest.approx(0.80)
-
-    resp = await client.delete(f"/api/tags/{tag.id}/allowlist")
-    assert resp.status_code == 204
-    resp = await client.get(f"/api/tags/{tag.id}/allowlist")
-    assert resp.status_code == 404
-
-
-@pytest.mark.asyncio
-async def test_patch_rejects_out_of_range(client, db):
-    tag = await TagService(db).find_or_create("AL2", TagKind.character)
-    db.add(TagAllowlist(tag_id=tag.id))
-    await db.commit()
-    resp = await client.patch(
-        f"/api/tags/{tag.id}/allowlist", json={"min_confidence": 1.5}
-    )
-    assert resp.status_code == 400
-
-
-@pytest.mark.asyncio
-async def test_coverage_endpoint(client, db):
-    tag = await TagService(db).find_or_create("Cover", TagKind.general)
-    db.add(TagAllowlist(tag_id=tag.id, min_confidence=0.90))
-    for i, score in enumerate((0.95, 0.60)):
-        img = ImageRecord(
-            path=f"/images/cov{i}.jpg", sha256=f"cv{i:062d}", size_bytes=1,
-            mime="image/jpeg", width=1, height=1,
-            origin="imported_filesystem", integrity_status="unknown",
-        )
-        db.add(img)
-        await db.flush()
-        db.add(ImagePrediction(
-            image_record_id=img.id, raw_name="Cover",
-            category="general", score=score,
-        ))
-    await db.commit()
-
-    # Explicit threshold.
-    resp = await client.get(
-        f"/api/tags/{tag.id}/allowlist/coverage?threshold=0.90"
-    )
-    assert resp.status_code == 200
-    assert (await resp.get_json())["count"] == 1
-    # Lower what-if threshold widens coverage.
-    resp = await client.get(
-        f"/api/tags/{tag.id}/allowlist/coverage?threshold=0.50"
-    )
-    assert (await resp.get_json())["count"] == 2
-    # No threshold → uses the stored min_confidence (0.90).
-    resp = await client.get(f"/api/tags/{tag.id}/allowlist/coverage")
-    body = await resp.get_json()
-    assert body["count"] == 1
-    assert body["threshold"] == pytest.approx(0.90)
-
-
-@pytest.mark.asyncio
-async def test_coverage_rejects_bad_threshold(client, db):
-    tag = await TagService(db).find_or_create("Cover2", TagKind.general)
-    db.add(TagAllowlist(tag_id=tag.id))
-    await db.commit()
-    resp = await client.get(
-        f"/api/tags/{tag.id}/allowlist/coverage?threshold=2.0"
-    )
-    assert resp.status_code == 400

tests/test_api_gpu.py

@@ -69,6 +69,39 @@ async def test_lease_submit_round_trip(client, db):
     assert len(regs) == 1 and len(list(regs[0].ccip_embedding)) == 768
 
 
+@pytest.mark.asyncio
+async def test_lease_announces_embed_model_then_submit_embedding(client, db):
+    # Whole-image GPU re-embed (#1190): the lease announces the embedder model so
+    # the agent loads the right one, and submit_embedding writes it back onto
+    # image_record with its version stamp.
+    img = await _img(db, "b" * 64)
+    await GpuJobService(db).enqueue(img.id, "embed")
+    await db.commit()
+
+    token = (await (await client.post("/api/gpu/token/rotate")).get_json())["token"]
+    hdr = {"Authorization": f"Bearer {token}"}
+
+    leased = await client.post(
+        "/api/gpu/jobs/lease", json={"agent_id": "a1", "batch_size": 5}, headers=hdr,
+    )
+    j = (await leased.get_json())["jobs"][0]
+    assert j["task"] == "embed"
+    assert j["embed_model_name"] and j["embed_version"]   # server-announced model
+
+    submitted = await client.post("/api/gpu/jobs/submit_embedding", json={
+        "agent_id": "a1", "job_id": j["job_id"],
+        "embedding": [0.2] * 1152, "embedding_version": "siglip2-test-v9",
+    }, headers=hdr)
+    assert submitted.status_code == 200
+
+    st = await (await client.get("/api/gpu/status")).get_json()
+    assert st["done"] == 1 and st["leased"] == 0
+
+    await db.refresh(img)
+    assert img.siglip_model_version == "siglip2-test-v9"
+    assert img.siglip_embedding is not None and len(list(img.siglip_embedding)) == 1152
+
+
 @pytest.mark.asyncio
 async def test_submit_with_stale_lease_is_409(client, db):
     img = await _img(db, "b" * 64)

tests/test_api_ml_admin.py

@@ -19,79 +19,79 @@ async def test_get_and_patch_settings(client):
     resp = await client.get("/api/ml/settings")
     assert resp.status_code == 200
     body = await resp.get_json()
-    # Default raised 0.50 → 0.70 on 2026-06-02 (alembic 0033) — 0.50
-    # was too noisy in practice. The 0.70 default keeps the rail
-    # signal-rich without hiding everything like the original 0.95.
-    assert body["suggestion_threshold_general"] == pytest.approx(0.70)
-    # Retired threshold columns must not appear in the payload.
-    assert "suggestion_threshold_artist" not in body
-    assert "suggestion_threshold_copyright" not in body
+    assert body["head_min_positives"] == 8
+    # Retired tagger/suggestion-threshold columns are gone from the payload
+    # (Camie retirement #1189/#1199).
+    assert "suggestion_threshold_general" not in body
+    assert "tagger_store_floor" not in body
+    assert "tagger_model_version" not in body
 
     resp = await client.patch(
-        "/api/ml/settings", json={"suggestion_threshold_general": 0.90}
+        "/api/ml/settings", json={"head_min_positives": 12}
     )
     assert resp.status_code == 200
-    assert (await resp.get_json())["suggestion_threshold_general"] == pytest.approx(0.90)
+    assert (await resp.get_json())["head_min_positives"] == 12
 
 
 @pytest.mark.asyncio
-async def test_tagger_store_floor_default_and_patch(client):
+async def test_embedder_model_default_settable_and_empty_rejected(client):
+    # #1203: fresh installs default to SigLIP 2 (migration 0069, no embeddings).
+    # The model is operator-settable (a swap) and neither field may be blanked.
     body = await (await client.get("/api/ml/settings")).get_json()
-    assert body["tagger_store_floor"] == pytest.approx(0.70)
+    assert body["embedder_model_name"] == "google/siglip2-so400m-patch16-512"
 
-    resp = await client.patch("/api/ml/settings", json={"tagger_store_floor": 0.6})
-    assert resp.status_code == 200
-    assert (await resp.get_json())["tagger_store_floor"] == pytest.approx(0.6)
+    ok = await client.patch("/api/ml/settings", json={
+        "embedder_model_name": "google/siglip-so400m-patch14-384",
+        "embedder_model_version": "siglip-so400m-patch14-384",
+    })
+    assert ok.status_code == 200
+    out = await ok.get_json()
+    assert out["embedder_model_name"] == "google/siglip-so400m-patch14-384"
+
+    bad = await client.patch("/api/ml/settings", json={"embedder_model_name": "  "})
+    assert bad.status_code == 400
 
 
 @pytest.mark.asyncio
-async def test_suggestion_threshold_below_store_floor_rejected(client):
-    # Invariant (#764): a category threshold can't sit below the store floor —
-    # nothing below the floor is stored, so the gap would surface nothing.
-    # Floor defaults to 0.70; pushing general down to 0.50 must 400.
-    resp = await client.patch(
-        "/api/ml/settings", json={"suggestion_threshold_general": 0.50}
+async def test_embedder_models_list(client):
+    # #1203: the dropdown reads the supported-model list from the server.
+    body = await (await client.get("/api/ml/embedder-models")).get_json()
+    assert any(
+        m["name"] == "google/siglip2-so400m-patch16-512" for m in body["models"]
     )
-    assert resp.status_code == 400
-    assert "tagger_store_floor" in (await resp.get_json())["error"]
+    assert all({"name", "version", "label"} <= set(m) for m in body["models"])
 
 
 @pytest.mark.asyncio
-async def test_video_tagging_settings_default_and_patch(client):
-    """#747: video cadence/noise knobs are exposed + patchable."""
+async def test_video_settings_default_and_patch(client):
+    """#747: video frame-sampling knobs are exposed + patchable."""
     body = await (await client.get("/api/ml/settings")).get_json()
     assert body["video_frame_interval_seconds"] == pytest.approx(4.0)
     assert body["video_max_frames"] == 64
-    assert body["video_min_tag_frames"] == 3
 
     resp = await client.patch(
         "/api/ml/settings",
-        json={"video_frame_interval_seconds": 5, "video_max_frames": 40,
-              "video_min_tag_frames": 4},
+        json={"video_frame_interval_seconds": 5, "video_max_frames": 40},
     )
     assert resp.status_code == 200
     out = await resp.get_json()
     assert out["video_frame_interval_seconds"] == pytest.approx(5.0)
     assert out["video_max_frames"] == 40
-    assert out["video_min_tag_frames"] == 4
 
 
 @pytest.mark.asyncio
-async def test_video_min_tag_frames_above_max_rejected(client):
+async def test_video_max_frames_below_one_rejected(client):
     resp = await client.patch(
-        "/api/ml/settings",
-        json={"video_max_frames": 10, "video_min_tag_frames": 20},
+        "/api/ml/settings", json={"video_max_frames": 0},
     )
     assert resp.status_code == 400
-    assert "video_min_tag_frames" in (await resp.get_json())["error"]
+    assert "video_max_frames" in (await resp.get_json())["error"]
 
 
 @pytest.mark.asyncio
-async def test_backfill_and_recompute_trigger(client):
+async def test_backfill_trigger(client):
     r1 = await client.post("/api/ml/backfill")
     assert r1.status_code == 202
-    r2 = await client.post("/api/ml/recompute-centroids")
-    assert r2.status_code == 202
 
 
 @pytest.mark.asyncio

tests/test_api_suggestions.py

@@ -15,9 +15,7 @@ def eager():
     celery.conf.task_always_eager = False
 
 
-async def _img(db, preds, sha="s" * 64):
-    from tests._prediction_helpers import seed_predictions
-
+async def _img(db, sha="s" * 64):
     img = ImageRecord(
         path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1,
         mime="image/jpeg", width=1, height=1,
@@ -25,8 +23,6 @@ async def _img(db, preds, sha="s" * 64):
     )
     db.add(img)
     await db.commit()
-    await seed_predictions(db, img.id, preds)
-    await db.commit()
     return img
 
 
@@ -60,7 +56,7 @@ async def test_get_suggestions(client, db):
 
 @pytest.mark.asyncio
 async def test_accept_requires_tag_id(client, db):
-    img = await _img(db, {})
+    img = await _img(db)
     resp = await client.post(
         f"/api/images/{img.id}/suggestions/accept", json={}
     )
@@ -68,43 +64,31 @@ async def test_accept_requires_tag_id(client, db):
 
 
 @pytest.mark.asyncio
-async def test_accept_then_applied(client, db):
-    img = await _img(db, {})
+async def test_accept_applies_tag_to_image(client, db):
+    # Camie/allowlist retired (#1189): accept applies the tag to THIS image
+    # (source='ml_accepted', a head-training positive) — no bulk allowlist
+    # fan-out anymore.
+    from backend.app.models.tag import image_tag
+
+    img = await _img(db)
     tag = await TagService(db).find_or_create("AcceptMe", TagKind.character)
     await db.commit()
     resp = await client.post(
         f"/api/images/{img.id}/suggestions/accept", json={"tag_id": tag.id}
     )
     assert resp.status_code == 200
-    body = await resp.get_json()
-    # #7b: a fresh accept newly-allowlists → projection payload for the toast.
-    assert body["allowlisted"] is True
-    assert body["tag_id"] == tag.id
-    assert body["tag_name"] == "AcceptMe"
-    assert "projected_count" in body
-
-
-@pytest.mark.asyncio
-async def test_accept_already_allowlisted_reports_not_new(client, db):
-    img1 = await _img(db, {}, sha="c" * 64)
-    img2 = await _img(db, {}, sha="d" * 64)
-    tag = await TagService(db).find_or_create("Twice", TagKind.character)
-    await db.commit()
-    first = await client.post(
-        f"/api/images/{img1.id}/suggestions/accept", json={"tag_id": tag.id}
-    )
-    assert (await first.get_json())["allowlisted"] is True
-    second = await client.post(
-        f"/api/images/{img2.id}/suggestions/accept", json={"tag_id": tag.id}
-    )
-    body = await second.get_json()
-    assert body["allowlisted"] is False   # already on the allowlist
-    assert "projected_count" not in body
+    assert (await resp.get_json())["accepted"] is True
+    src = (await db.execute(
+        select(image_tag.c.source)
+        .where(image_tag.c.image_record_id == img.id)
+        .where(image_tag.c.tag_id == tag.id)
+    )).scalar_one()
+    assert src == "ml_accepted"
 
 
 @pytest.mark.asyncio
 async def test_dismiss(client, db):
-    img = await _img(db, {})
+    img = await _img(db)
     tag = await TagService(db).find_or_create("DismissMe", TagKind.general)
     await db.commit()
     resp = await client.post(
@@ -115,7 +99,7 @@ async def test_dismiss(client, db):
 
 @pytest.mark.asyncio
 async def test_undismiss_reverses_rejection(client, db):
-    img = await _img(db, {})
+    img = await _img(db)
     tag = await TagService(db).find_or_create("UndismissMe", TagKind.general)
     await db.commit()
     await client.post(
@@ -134,7 +118,7 @@ async def test_undismiss_reverses_rejection(client, db):
 
 @pytest.mark.asyncio
 async def test_alias_requires_fields(client, db):
-    img = await _img(db, {})
+    img = await _img(db)
     resp = await client.post(
         f"/api/images/{img.id}/suggestions/alias", json={"alias_string": "x"}
     )

tests/test_api_tag_merge.py

@@ -68,15 +68,7 @@ async def test_rename_collision_returns_rich_409(client):
 
 
 @pytest.mark.asyncio
-async def test_merge_endpoint_moves_and_deletes(client, monkeypatch):
-    calls = []
-    from backend.app.tasks import ml as ml_tasks
-
-    monkeypatch.setattr(
-        ml_tasks.apply_allowlist_tags,
-        "delay",
-        lambda **kw: calls.append(kw),
-    )
+async def test_merge_endpoint_moves_and_deletes(client):
     tgt = await _mk(client, "Keep", "general")
     src = await _mk(client, "Gone", "general")
     resp = await client.post(
@@ -92,36 +84,6 @@ async def test_merge_endpoint_moves_and_deletes(client, monkeypatch):
     assert r2.status_code == 200
 
 
-@pytest.mark.asyncio
-async def test_merge_enqueues_backfill_when_target_allowlisted(
-    client, monkeypatch
-):
-    calls = []
-    from backend.app.tasks import ml as ml_tasks
-
-    monkeypatch.setattr(
-        ml_tasks.apply_allowlist_tags,
-        "delay",
-        lambda **kw: calls.append(kw),
-    )
-    tgt = await _mk(client, "AllowTgt", "general")
-    src = await _mk(client, "AllowSrc", "general")
-    # No public route adds a tag to the allowlist (it happens via
-    # accept-suggestion); set the row directly through the app session.
-    from backend.app.extensions import get_session
-    from backend.app.models.tag_allowlist import TagAllowlist
-
-    async with get_session() as s:
-        s.add(TagAllowlist(tag_id=tgt))
-        await s.commit()
-
-    resp = await client.post(
-        f"/api/tags/{src}/merge", json={"target_id": tgt}
-    )
-    assert resp.status_code == 200
-    assert calls == [{"tag_id": tgt}]
-
-
 @pytest.mark.asyncio
 async def test_merge_self_is_400(client):
     t = await _mk(client, "Selfie", "general")

tests/test_api_tag_stats.py

@@ -0,0 +1,84 @@
+"""Agent-friendly tag analysis endpoints (#1136): /api/tags/top + /tags/<id>/stats."""
+import pytest
+
+from backend.app.models import ImageRecord, TagHead, TagKind
+from backend.app.models.tag import image_tag
+from backend.app.models.tag_suggestion_rejection import TagSuggestionRejection
+from backend.app.services.tag_service import TagService
+
+pytestmark = pytest.mark.integration
+
+
+async def _img(db, sha) -> ImageRecord:
+    img = ImageRecord(
+        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
+        width=1, height=1, origin="imported_filesystem", integrity_status="unknown",
+    )
+    db.add(img)
+    await db.flush()
+    return img
+
+
+async def _apply(db, image_id, tag_id, source):
+    await db.execute(image_tag.insert().values(
+        image_record_id=image_id, tag_id=tag_id, source=source,
+    ))
+
+
+@pytest.mark.asyncio
+async def test_tags_top_ranks_by_count_and_filters(client, db):
+    svc = TagService(db)
+    common = await svc.find_or_create("Common", TagKind.general)
+    rare = await svc.find_or_create("Rare", TagKind.general)
+    imgs = [await _img(db, f"{i:064d}") for i in range(3)]
+    await _apply(db, imgs[0].id, common.id, "manual")
+    await _apply(db, imgs[1].id, common.id, "manual")
+    await _apply(db, imgs[2].id, common.id, "head_auto")   # 3 total, 2 human
+    await _apply(db, imgs[0].id, rare.id, "manual")         # 1
+    await db.commit()
+
+    top = await (await client.get("/api/tags/top?kind=general&limit=10")).get_json()
+    counts = {t["name"]: t["count"] for t in top["tags"]}
+    assert counts["Common"] == 3 and counts["Rare"] == 1
+    assert [t["name"] for t in top["tags"]][0] == "Common"   # count desc
+
+    # source=human drops the head_auto application
+    human = await (await client.get("/api/tags/top?source=human&kind=general")).get_json()
+    assert {t["name"]: t["count"] for t in human["tags"]}["Common"] == 2
+
+    # min_count filters out the rare tag
+    mc = await (await client.get("/api/tags/top?min_count=2&kind=general")).get_json()
+    assert "Rare" not in [t["name"] for t in mc["tags"]]
+
+
+@pytest.mark.asyncio
+async def test_tag_stats_source_breakdown(client, db):
+    svc = TagService(db)
+    tag = await svc.find_or_create("Hero", TagKind.character)
+    i1, i2, i3, i4 = [await _img(db, c * 64) for c in "abcd"]
+    await _apply(db, i1.id, tag.id, "manual")
+    await _apply(db, i2.id, tag.id, "ml_accepted")
+    await _apply(db, i3.id, tag.id, "ccip_auto")
+    db.add(TagSuggestionRejection(image_record_id=i4.id, tag_id=tag.id))
+    db.add(TagHead(
+        tag_id=tag.id, embedding_version="v", weights=[0.0] * 1152, bias=0.0,
+        suggest_threshold=0.5, auto_apply_threshold=None, n_pos=10, n_neg=30,
+        ap=0.8, precision_cv=0.9, recall=0.6,
+    ))
+    await db.commit()
+
+    body = await (await client.get(f"/api/tags/{tag.id}/stats")).get_json()
+    assert body["count_total"] == 3
+    assert body["count_human"] == 2          # manual + ml_accepted
+    assert body["count_manual"] == 1
+    assert body["count_accepted"] == 1
+    assert body["count_ccip_auto"] == 1
+    assert body["count_auto"] == 1
+    assert body["count_rejected"] == 1
+    assert body["has_head"] is True
+
+
+@pytest.mark.asyncio
+async def test_tag_stats_404(client):
+    resp = await client.get("/api/tags/99999/stats")
+    assert resp.status_code == 404

tests/test_download_models.py

@@ -1,6 +1,6 @@
 """download_models tests. No network in CI; we test the 'already present →
-skip' short-circuit by faking the expected files, and that main() wires
-both ensure_* calls.
+skip' short-circuit by faking the expected files, and that main() wires the
+SigLIP fetch. (Camie download retired with the tagger, #1199.)
 """
 
 from unittest.mock import patch
@@ -8,29 +8,6 @@ from unittest.mock import patch
 from backend.app.scripts import download_models as dm
 
 
-def test_ensure_camie_skips_when_present(tmp_path, monkeypatch):
-    """v2 layout (HF Camais03/camie-tagger-v2): the ONNX file is named
-    camie-tagger-v2.onnx (not model.onnx) and tags ship inside
-    camie-tagger-v2-metadata.json (not selected_tags.csv). Both at root.
-    Updated 2026-05-25 after the actual repo layout was confirmed via
-    WebFetch — the old assertion pinned the v1 filenames."""
-    monkeypatch.setattr(dm, "MODEL_ROOT", tmp_path)
-    camie = tmp_path / "camie"
-    camie.mkdir(parents=True)
-    (camie / "camie-tagger-v2.onnx").write_bytes(b"x")
-    (camie / "camie-tagger-v2-metadata.json").write_text("{}")
-    with patch.object(dm, "_snapshot") as snap:
-        dm.ensure_camie()
-        snap.assert_not_called()
-
-
-def test_ensure_camie_downloads_when_missing(tmp_path, monkeypatch):
-    monkeypatch.setattr(dm, "MODEL_ROOT", tmp_path)
-    with patch.object(dm, "_snapshot") as snap:
-        dm.ensure_camie()
-        snap.assert_called_once()
-
-
 def test_ensure_siglip_skips_when_present(tmp_path, monkeypatch):
     monkeypatch.setattr(dm, "MODEL_ROOT", tmp_path)
     sig = tmp_path / "siglip"
@@ -42,9 +19,15 @@ def test_ensure_siglip_skips_when_present(tmp_path, monkeypatch):
         snap.assert_not_called()
 
 
-def test_main_calls_both(monkeypatch):
+def test_ensure_siglip_downloads_when_missing(tmp_path, monkeypatch):
+    monkeypatch.setattr(dm, "MODEL_ROOT", tmp_path)
+    with patch.object(dm, "_snapshot") as snap:
+        dm.ensure_siglip()
+        snap.assert_called_once()
+
+
+def test_main_fetches_siglip(monkeypatch):
     calls = []
-    monkeypatch.setattr(dm, "ensure_camie", lambda: calls.append("camie"))
     monkeypatch.setattr(dm, "ensure_siglip", lambda: calls.append("siglip"))
     assert dm.main() == 0
-    assert calls == ["camie", "siglip"]
+    assert calls == ["siglip"]

tests/test_gallery_service.py

@@ -73,6 +73,28 @@ async def test_scroll_returns_newest_first(db):
     assert page.images[0].created_at > page.images[-1].created_at
 
 
+@pytest.mark.asyncio
+async def test_scroll_sorts_by_earliest_post_date(db):
+    """posted_new/posted_old key off earliest_post_date (original publish across
+    all of an image's posts), NOT effective_date/created_at — so an old file that
+    was first posted recently sorts by that recent post date."""
+    imgs = await _seed_images(db, 3)
+    now = _now()
+    # earliest_post_date order deliberately DIFFERS from created_at order (which
+    # _seed_images makes imgs[0] newest → imgs[2] oldest), proving the sort keys
+    # off earliest_post_date.
+    imgs[0].earliest_post_date = now - timedelta(days=1)     # middle
+    imgs[1].earliest_post_date = now - timedelta(days=365)   # oldest post
+    imgs[2].earliest_post_date = now - timedelta(hours=1)    # newest post
+    await db.flush()
+
+    svc = GalleryService(db)
+    newest = await svc.scroll(cursor=None, limit=10, sort="posted_new")
+    assert [i.id for i in newest.images] == [imgs[2].id, imgs[0].id, imgs[1].id]
+    oldest = await svc.scroll(cursor=None, limit=10, sort="posted_old")
+    assert [i.id for i in oldest.images] == [imgs[1].id, imgs[0].id, imgs[2].id]
+
+
 @pytest.mark.asyncio
 async def test_scroll_pagination(db):
     await _seed_images(db, 5)

tests/test_gallery_similar.py

@@ -87,6 +87,29 @@ async def test_similar_composes_with_tag_filter(db):
     assert [i.id for i in res] == [tagged.id]  # scope AND-narrows the ranked set
 
 
+@pytest.mark.asyncio
+async def test_similar_collapses_near_duplicate_phashes(db):
+    # The reported failure: a reposted image fills the whole neighbour grid.
+    # A wall of same-pHash reposts must collapse to at most one, and the
+    # genuinely distinct images must still come through.
+    src = await _img(db, 1, _vec(1, 0))
+    dupes = []
+    for n in range(2, 7):                       # 5 near-identical reposts
+        r = await _img(db, n, _vec(1, 0.01 * n))
+        r.phash = "ffffffffffffffff"            # identical perceptual hash
+        dupes.append(r)
+    distinct_a = await _img(db, 7, _vec(1, 1))
+    distinct_a.phash = "0000000000000000"
+    distinct_b = await _img(db, 8, _vec(0, 1))
+    distinct_b.phash = "0f0f0f0f0f0f0f0f"
+    await db.flush()
+
+    res = await GalleryService(db).similar(src.id, limit=10)
+    ids = [i.id for i in res]
+    assert sum(1 for d in dupes if d.id in ids) <= 1   # repost wall collapsed
+    assert distinct_a.id in ids and distinct_b.id in ids
+
+
 @pytest.mark.asyncio
 async def test_similar_respects_limit(db):
     src = await _img(db, 1, _vec(1, 0))

tests/test_gpu_jobs.py

@@ -25,13 +25,17 @@ async def test_enqueue_siglip_backfill_gates_on_concept_region(db):
     # 'siglip' backfill enqueues images that lack a concept region (the
     # back-catalogue) and skips ones that already have one — and never double-
     # enqueues an image that already has a pending siglip job.
+    from backend.app.models import MLSettings
     from backend.app.tasks.ml import enqueue_gpu_backfill
 
+    cur = (await db.execute(
+        select(MLSettings.embedder_model_version).where(MLSettings.id == 1)
+    )).scalar_one()
     need = await _img(db, "e1" * 32)          # no concept region → wants one
-    have = await _img(db, "e2" * 32)          # already embedded → skip
+    have = await _img(db, "e2" * 32)          # concept @ current version → skip
     db.add(ImageRegion(
         image_record_id=have.id, kind="concept", rx=0.0, ry=0.0, rw=1.0, rh=1.0,
-        siglip_embedding=[0.0] * 1152, embedding_version="siglip-test",
+        siglip_embedding=[0.0] * 1152, embedding_version=cur,
     ))
     await db.commit()
 
@@ -57,6 +61,54 @@ async def test_enqueue_siglip_backfill_gates_on_concept_region(db):
     assert n_for_need == 1
 
 
+@pytest.mark.asyncio
+async def test_enqueue_embed_backfill_selects_stale_and_unembedded(db):
+    # Whole-image GPU re-embed (#1190): enqueue images with no embedding or one
+    # stamped under a DIFFERENT model version (an operator swap); skip ones
+    # already at the current version.
+    from backend.app.models import MLSettings
+    from backend.app.tasks.ml import enqueue_gpu_backfill
+
+    cur = (await db.execute(
+        select(MLSettings.embedder_model_version).where(MLSettings.id == 1)
+    )).scalar_one()
+    current = await _img(db, "f1" * 32)
+    current.siglip_embedding = [0.0] * 1152
+    current.siglip_model_version = cur                 # up to date → skip
+    stale = await _img(db, "f2" * 32)
+    stale.siglip_embedding = [0.0] * 1152
+    stale.siglip_model_version = "old-embedder-v0"      # wrong space → re-embed
+    unembedded = await _img(db, "f3" * 32)             # never embedded → embed
+    await db.commit()
+
+    assert enqueue_gpu_backfill("embed") >= 2
+    queued = {
+        j.image_record_id for j in (
+            await db.execute(select(GpuJob).where(GpuJob.task == "embed"))
+        ).scalars()
+    }
+    assert stale.id in queued and unembedded.id in queued
+    assert current.id not in queued
+
+
+@pytest.mark.asyncio
+async def test_reprocess_resets_done_jobs_to_pending(db):
+    # Re-process (#1202): done/error jobs of a task go back to pending so the
+    # agent re-runs the whole library under the current pipeline.
+    from backend.app.tasks.ml import reprocess_gpu_jobs
+
+    img = await _img(db, "r1" * 32)
+    job = await GpuJobService(db).enqueue(img.id, "ccip")
+    job.status = "done"
+    job.attempts = 2
+    await db.commit()
+
+    assert reprocess_gpu_jobs("ccip") >= 1
+    await db.refresh(job)
+    assert job.status == "pending"
+    assert job.attempts == 0
+
+
 @pytest.mark.asyncio
 async def test_enqueue_dedupes_same_pair(db):
     img = await _img(db, "a" * 64)
@@ -89,6 +141,22 @@ async def test_lease_claims_then_skips_when_held(db):
     assert again == []
 
 
+@pytest.mark.asyncio
+async def test_lease_skips_done_prefix(db):
+    # A long prefix of already-done jobs at the LOW ids must not stop the lease
+    # from reaching the pending ones (the O(done) scan bug this indexing fixes).
+    svc = GpuJobService(db)
+    for i in range(5):
+        j = await svc.enqueue((await _img(db, f"d{i}" * 32)).id, "ccip")
+        j.status = "done"
+    target = await svc.enqueue((await _img(db, "dz" * 32)).id, "ccip")  # highest id
+    await db.commit()
+
+    leased = await svc.lease("agent-1", batch_size=8)
+    await db.commit()
+    assert [j.id for j in leased] == [target.id]
+
+
 @pytest.mark.asyncio
 async def test_expired_lease_is_reclaimed(db):
     img = await _img(db, "c" * 64)

tests/test_image_prediction.py

@@ -1,57 +0,0 @@
-"""#768: image_prediction table — model + constraints round-trip."""
-import pytest
-from sqlalchemy import select
-from sqlalchemy.exc import IntegrityError
-
-from backend.app.models import ImagePrediction, ImageRecord
-
-pytestmark = pytest.mark.integration
-
-
-async def _make_image(db, path="/img/p0.jpg", sha="0"):
-    rec = ImageRecord(
-        path=path, sha256=sha.ljust(64, "0")[:64], size_bytes=10,
-        mime="image/jpeg", origin="imported_filesystem",
-    )
-    db.add(rec)
-    await db.flush()
-    return rec
-
-
-@pytest.mark.asyncio
-async def test_image_prediction_round_trip(db):
-    rec = await _make_image(db)
-    db.add_all([
-        ImagePrediction(
-            image_record_id=rec.id, raw_name="blue_eyes",
-            category="general", score=0.92,
-        ),
-        ImagePrediction(
-            image_record_id=rec.id, raw_name="hatsune_miku",
-            category="character", score=0.88,
-        ),
-    ])
-    await db.commit()
-
-    rows = (await db.execute(
-        select(ImagePrediction.raw_name, ImagePrediction.score)
-        .where(ImagePrediction.image_record_id == rec.id)
-        .order_by(ImagePrediction.score.desc())
-    )).all()
-    assert [r.raw_name for r in rows] == ["blue_eyes", "hatsune_miku"]
-
-
-@pytest.mark.asyncio
-async def test_image_prediction_unique_per_image_name(db):
-    rec = await _make_image(db, path="/img/p1.jpg", sha="1")
-    db.add(ImagePrediction(
-        image_record_id=rec.id, raw_name="dup",
-        category="general", score=0.9,
-    ))
-    await db.commit()
-    db.add(ImagePrediction(
-        image_record_id=rec.id, raw_name="dup",
-        category="general", score=0.7,
-    ))
-    with pytest.raises(IntegrityError):
-        await db.commit()

tests/test_migration_0003.py

@@ -5,18 +5,14 @@ from backend.app.models import (
     ImageRecord,
     MLSettings,
     TagAlias,
-    TagAllowlist,
-    TagReferenceEmbedding,
     TagSuggestionRejection,
 )
 
 
 def test_new_tables_registered():
     expected = {
-        "tag_allowlist",
         "tag_suggestion_rejection",
         "tag_alias",
-        "tag_reference_embedding",
         "ml_settings",
     }
     assert expected.issubset(Base.metadata.tables.keys())
@@ -24,12 +20,15 @@ def test_new_tables_registered():
 
 def test_image_record_columns_renamed():
     cols = {c.name for c in ImageRecord.__table__.columns}
-    # tagger_predictions (the renamed wd14_predictions) was later dropped in
-    # migration 0046 — predictions live in image_prediction now (#768).
-    assert "tagger_model_version" in cols
+    # Legacy tagger columns are all gone: tagger_predictions/wd14_* dropped in
+    # 0046, tagger_model_version + centroid_scores dropped in 0068 (#1199, Camie
+    # retirement). The SigLIP embedding columns are the live ML fields.
+    assert "siglip_embedding" in cols
+    assert "siglip_model_version" in cols
+    assert "tagger_model_version" not in cols
+    assert "centroid_scores" not in cols
     assert "tagger_predictions" not in cols
     assert "wd14_predictions" not in cols
-    assert "wd14_model_version" not in cols
 
 
 def test_tag_alias_composite_pk():
@@ -42,17 +41,6 @@ def test_ml_settings_singleton_constraint():
     assert "ck_ml_settings_singleton" in names
 
 
-def test_tag_reference_embedding_has_vector():
-    cols = {c.name for c in TagReferenceEmbedding.__table__.columns}
-    assert "embedding" in cols
-    assert "reference_count" in cols
-
-
-def test_tag_allowlist_confidence_check():
-    names = {c.name for c in TagAllowlist.__table__.constraints}
-    assert "ck_tag_allowlist_confidence_range" in names
-
-
 def test_tag_suggestion_rejection_pk():
     pk_cols = {c.name for c in TagSuggestionRejection.__table__.primary_key.columns}
     assert pk_cols == {"image_record_id", "tag_id"}

tests/test_ml_allowlist.py

@@ -1,182 +0,0 @@
-import pytest
-from sqlalchemy import select
-
-from backend.app.models import (
-    ImagePrediction,
-    TagAlias,
-    TagAllowlist,
-    TagKind,
-    TagSuggestionRejection,
-)
-from backend.app.models.tag import image_tag
-from backend.app.services.ml.allowlist import AllowlistService
-from backend.app.services.tag_service import TagService
-
-pytestmark = pytest.mark.integration
-
-
-async def _make_image(db, sha: str = "x" * 64):
-    from backend.app.models import ImageRecord
-    img = ImageRecord(
-        # Full sha in the path — the first 8 chars collide for sequential
-        # shas like c{i:063d}, and path is UNIQUE (uq_image_record_path).
-        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1,
-        mime="image/jpeg", width=1, height=1,
-        origin="imported_filesystem", integrity_status="unknown",
-    )
-    db.add(img)
-    await db.flush()
-    return img
-
-
-async def _add_pred(db, image_id, raw_name, score, category="general"):
-    db.add(ImagePrediction(
-        image_record_id=image_id, raw_name=raw_name,
-        category=category, score=score,
-    ))
-    await db.flush()
-
-
-@pytest.mark.asyncio
-async def test_accept_applies_and_allowlists(db):
-    img = await _make_image(db)
-    tag = await TagService(db).find_or_create("Hero", TagKind.character)
-    svc = AllowlistService(db)
-    newly_added = await svc.accept(img.id, tag.id)
-    assert newly_added is True
-
-    applied = (
-        await db.execute(
-            select(image_tag.c.source)
-            .where(image_tag.c.image_record_id == img.id)
-            .where(image_tag.c.tag_id == tag.id)
-        )
-    ).scalar_one()
-    assert applied == "ml_accepted"
-    assert await db.get(TagAllowlist, tag.id) is not None
-
-
-@pytest.mark.asyncio
-async def test_accept_idempotent_allowlist(db):
-    img = await _make_image(db)
-    tag = await TagService(db).find_or_create("Hero2", TagKind.character)
-    svc = AllowlistService(db)
-    assert await svc.accept(img.id, tag.id) is True
-    assert await svc.accept(img.id, tag.id) is False
-
-
-@pytest.mark.asyncio
-async def test_reject_applied_tag_records_rejection(db):
-    img = await _make_image(db)
-    tag = await TagService(db).find_or_create("Removeme", TagKind.general)
-    svc = AllowlistService(db)
-    await svc.accept(img.id, tag.id)
-    await svc.reject_applied_tag(img.id, tag.id)
-
-    still_applied = (
-        await db.execute(
-            select(image_tag.c.tag_id)
-            .where(image_tag.c.image_record_id == img.id)
-            .where(image_tag.c.tag_id == tag.id)
-        )
-    ).scalar_one_or_none()
-    assert still_applied is None
-    rej = await db.get(TagSuggestionRejection, (img.id, tag.id))
-    assert rej is not None
-
-
-@pytest.mark.asyncio
-async def test_dismiss_records_rejection(db):
-    img = await _make_image(db)
-    tag = await TagService(db).find_or_create("Dismissme", TagKind.general)
-    await AllowlistService(db).dismiss(img.id, tag.id)
-    assert await db.get(TagSuggestionRejection, (img.id, tag.id)) is not None
-
-
-@pytest.mark.asyncio
-async def test_add_alias_and_accept(db):
-    img = await _make_image(db)
-    canonical = await TagService(db).find_or_create(
-        "Canonical Char", TagKind.character
-    )
-    svc = AllowlistService(db)
-    await svc.add_alias_and_accept(
-        img.id, "model_char_name", "character", canonical.id
-    )
-    from backend.app.services.ml.aliases import AliasService
-    resolved = await AliasService(db).resolve("model_char_name", "character")
-    assert resolved.id == canonical.id
-    assert await db.get(TagAllowlist, canonical.id) is not None
-
-
-@pytest.mark.asyncio
-async def test_update_threshold_and_remove(db):
-    tag = await TagService(db).find_or_create("Thr", TagKind.general)
-    svc = AllowlistService(db)
-    img = await _make_image(db)
-    await svc.accept(img.id, tag.id)
-    await svc.update_threshold(tag.id, 0.80)
-    row = await db.get(TagAllowlist, tag.id)
-    assert abs(row.min_confidence - 0.80) < 1e-6
-    await svc.remove(tag.id)
-    assert await db.get(TagAllowlist, tag.id) is None
-
-
-@pytest.mark.asyncio
-async def test_coverage_by_threshold_direct_name(db):
-    tag = await TagService(db).find_or_create("Cov", TagKind.general)
-    svc = AllowlistService(db)
-    for i, score in enumerate((0.95, 0.80, 0.60)):
-        img = await _make_image(db, sha=f"c{i:063d}")
-        await _add_pred(db, img.id, "Cov", score)
-    assert await svc.coverage(tag.id, 0.90) == 1
-    assert await svc.coverage(tag.id, 0.70) == 2
-    assert await svc.coverage(tag.id, 0.50) == 3
-
-
-@pytest.mark.asyncio
-async def test_coverage_via_alias_respects_category(db):
-    tag = await TagService(db).find_or_create("Aliased", TagKind.character)
-    db.add(TagAlias(
-        alias_string="model_key", alias_category="character",
-        canonical_tag_id=tag.id,
-    ))
-    await db.flush()
-    svc = AllowlistService(db)
-    hit = await _make_image(db, sha=f"a{0:063d}")
-    await _add_pred(db, hit.id, "model_key", 0.92, category="character")
-    # Same alias string but wrong category must NOT resolve to the tag.
-    miss = await _make_image(db, sha=f"a{1:063d}")
-    await _add_pred(db, miss.id, "model_key", 0.99, category="general")
-    assert await svc.coverage(tag.id, 0.90) == 1
-
-
-@pytest.mark.asyncio
-async def test_list_all_reports_applied_and_coverage(db):
-    tag = await TagService(db).find_or_create("Both", TagKind.general)
-    svc = AllowlistService(db)
-    applied_img = await _make_image(db, sha=f"b{0:063d}")
-    await svc.accept(applied_img.id, tag.id)  # applies + allowlists
-    await _add_pred(db, applied_img.id, "Both", 0.95)
-    # A second image only has a qualifying prediction (covered, not applied).
-    cov_img = await _make_image(db, sha=f"b{1:063d}")
-    await _add_pred(db, cov_img.id, "Both", 0.95)
-
-    rows = await svc.list_all()
-    row = next(r for r in rows if r.tag_id == tag.id)
-    assert row.applied_count == 1                 # only the accepted image
-    assert row.coverage_count == 2                # both have a ≥threshold pred
-
-
-@pytest.mark.asyncio
-async def test_update_threshold_clamped_to_store_floor(db):
-    # A min_confidence below the store floor (default 0.70) is clamped up —
-    # predictions below the floor aren't stored, so a lower threshold can't
-    # apply more permissively than the floor (#764).
-    tag = await TagService(db).find_or_create("Lowthr", TagKind.general)
-    svc = AllowlistService(db)
-    img = await _make_image(db)
-    await svc.accept(img.id, tag.id)
-    await svc.update_threshold(tag.id, 0.30)
-    row = await db.get(TagAllowlist, tag.id)
-    assert abs(row.min_confidence - 0.70) < 1e-6

tests/test_ml_artist_retired.py

@@ -3,17 +3,6 @@ import pytest
 pytestmark = pytest.mark.integration
 
 
-def test_artist_not_surfaced():
-    from backend.app.services.ml.tagger import SURFACED_CATEGORIES
-    assert "artist" not in SURFACED_CATEGORIES
-
-
-def test_artist_not_centroid_eligible():
-    from backend.app.models import TagKind
-    from backend.app.services.ml.centroids import ELIGIBLE_KINDS
-    assert TagKind.artist not in ELIGIBLE_KINDS
-
-
 def test_artist_not_head_eligible():
     # Tagging-v2: suggestions come from heads, and heads are only trained for
     # general/character concepts — so 'artist' (and any other kind) can't surface.

tests/test_ml_centroids.py

@@ -1,112 +0,0 @@
-import numpy as np
-import pytest
-
-from backend.app.models import ImageRecord, TagKind
-from backend.app.models.tag import image_tag
-from backend.app.services.ml.centroids import CentroidService
-from backend.app.services.tag_service import TagService
-
-pytestmark = pytest.mark.integration
-
-
-def _img(sha: str, embedding: list[float] | None) -> ImageRecord:
-    return ImageRecord(
-        path=f"/images/{sha}.jpg",
-        sha256=sha,
-        size_bytes=1,
-        mime="image/jpeg",
-        width=1,
-        height=1,
-        origin="imported_filesystem",
-        integrity_status="unknown",
-        siglip_embedding=embedding,
-    )
-
-
-async def _attach(db, image_id: int, tag_id: int):
-    await db.execute(
-        image_tag.insert().values(
-            image_record_id=image_id, tag_id=tag_id, source="manual"
-        )
-    )
-
-
-@pytest.mark.asyncio
-async def test_recompute_skips_too_few_members(db):
-    tags = TagService(db)
-    tag = await tags.find_or_create("Lonely", TagKind.character)
-    img = _img("a" * 64, [0.1] * 1152)
-    db.add(img)
-    await db.flush()
-    await _attach(db, img.id, tag.id)
-
-    svc = CentroidService(db)
-    assert await svc.recompute_for_tag(tag.id) is False
-
-
-@pytest.mark.asyncio
-async def test_recompute_writes_centroid(db):
-    tags = TagService(db)
-    tag = await tags.find_or_create("Popular", TagKind.character)
-    for i in range(5):
-        img = _img(f"{i:064d}", [float(i)] * 1152)
-        db.add(img)
-        await db.flush()
-        await _attach(db, img.id, tag.id)
-
-    svc = CentroidService(db)
-    assert await svc.recompute_for_tag(tag.id) is True
-
-    from backend.app.models import TagReferenceEmbedding
-    cen = await db.get(TagReferenceEmbedding, tag.id)
-    assert cen is not None
-    assert cen.reference_count == 5
-    assert abs(np.array(cen.embedding)[0] - 2.0) < 1e-4
-
-
-@pytest.mark.asyncio
-async def test_recompute_skips_ineligible_kind(db):
-    tags = TagService(db)
-    tag = await tags.find_or_create("somearchive", TagKind.archive)
-    for i in range(5):
-        img = _img(f"arch{i:060d}", [1.0] * 1152)
-        db.add(img)
-        await db.flush()
-        await _attach(db, img.id, tag.id)
-    svc = CentroidService(db)
-    assert await svc.recompute_for_tag(tag.id) is False
-
-
-@pytest.mark.asyncio
-async def test_list_drifted_includes_uncomputed(db):
-    tags = TagService(db)
-    tag = await tags.find_or_create("Drifty", TagKind.character)
-    for i in range(5):
-        img = _img(f"d{i:063d}", [0.5] * 1152)
-        db.add(img)
-        await db.flush()
-        await _attach(db, img.id, tag.id)
-    svc = CentroidService(db)
-    drifted = await svc.list_drifted()
-    assert tag.id in drifted
-
-
-@pytest.mark.asyncio
-async def test_find_similar_tags(db):
-    tags = TagService(db)
-    tag = await tags.find_or_create("SimTag", TagKind.character)
-    for i in range(5):
-        img = _img(f"s{i:063d}", [1.0] * 1152)
-        db.add(img)
-        await db.flush()
-        await _attach(db, img.id, tag.id)
-    svc = CentroidService(db)
-    await svc.recompute_for_tag(tag.id)
-
-    query_img = _img("q" * 64, [1.0] * 1152)
-    db.add(query_img)
-    await db.flush()
-    hits = await svc.find_similar_tags(query_img.id, limit=10)
-    assert any(h.tag_id == tag.id for h in hits)
-    sim = next(h.similarity for h in hits if h.tag_id == tag.id)
-    assert sim > 0.99

tests/test_ml_suggestions.py

@@ -145,6 +145,20 @@ async def test_concept_region_surfaces_via_max_over_bag(db):
     assert any(s.canonical_tag_id == tag.id and s.score > 0.7 for s in general)
 
 
+@pytest.mark.asyncio
+async def test_stale_embedding_version_excluded_from_scoring(db):
+    # Mid model-swap (#1190): an image still carrying an OLD-version whole-image
+    # embedding must NOT be scored by heads trained in the new model's space —
+    # even though the vector aligns with the head, it's the wrong coordinate
+    # system, so nothing surfaces until it's re-embedded.
+    tag = await TagService(db).find_or_create("glasses", TagKind.general)
+    img = await _img(db, "c1" * 32, _emb(0))
+    img.siglip_model_version = "some-old-model-v0"      # != current embedder
+    await _head(db, tag.id, slot=0, suggest_threshold=0.5)
+    await db.commit()
+    assert not (await SuggestionService(db).for_image(img.id)).by_category.get("general")
+
+
 @pytest.mark.asyncio
 async def test_rejected_tag_surfaced_flagged_then_reversible(db):
     # A dismissed suggestion is NOT dropped: it stays flagged rejected so the

tests/test_ml_tagger.py

@@ -1,54 +0,0 @@
-"""Tagger unit tests. The ONNX model isn't available in CI (it's a 1GB
-download into /models), so these test the pure-logic surface:
-DEFAULT_STORE_FLOOR constant, SURFACED_CATEGORIES set, TagPrediction
-dataclass, and the load()-missing-file error path. Full inference is
-exercised by the local integration suite against a real /models volume.
-"""
-
-import pytest
-
-from backend.app.services.ml.tagger import (
-    DEFAULT_STORE_FLOOR,
-    SURFACED_CATEGORIES,
-    Tagger,
-    TagPrediction,
-    get_tagger,
-)
-
-
-def test_surfaced_categories():
-    # FC-2d-vii-c: 'artist' retired — artist identity is acquisition-
-    # derived (image_record.artist_id), never ML-inferred.
-    # 2026-06-01: 'copyright' retired — fandom serves as the franchise/
-    # copyright concept; operator doesn't use a separate copyright kind.
-    assert SURFACED_CATEGORIES == {"character", "general"}
-    assert "artist" not in SURFACED_CATEGORIES
-    assert "copyright" not in SURFACED_CATEGORIES
-
-
-def test_default_store_floor():
-    # Raised 0.05 → 0.70 (plan-task #764): the suggestion path filters at
-    # 0.70 and the centroid path covers lower-confidence preferred tags, so
-    # storing the sub-0.70 tail was redundant (100 GB of TOAST). The live
-    # value is DB-backed (ml_settings.tagger_store_floor); this is the default.
-    assert DEFAULT_STORE_FLOOR == 0.70
-
-
-def test_tag_prediction_dataclass():
-    p = TagPrediction(name="x", category="general", confidence=0.9)
-    assert p.name == "x"
-    assert p.category == "general"
-    assert p.confidence == 0.9
-
-
-def test_get_tagger_singleton():
-    assert get_tagger() is get_tagger()
-
-
-def test_load_raises_when_model_missing(tmp_path):
-    t = Tagger(model_dir=tmp_path / "nonexistent")
-    # Match the trailing "missing at <path>" rather than the specific
-    # filename, so a future model-version bump (camie-tagger-v3.onnx, etc.)
-    # doesn't bounce this test.
-    with pytest.raises(RuntimeError, match=r"\.onnx missing at "):
-        t.load()

tests/test_phash_dedup.py

@@ -11,7 +11,6 @@ from PIL import Image
 from sqlalchemy import func, select
 
 from backend.app.models import (
-    ImagePrediction,
     ImageProvenance,
     ImageRecord,
     ImportSettings,
@@ -119,11 +118,6 @@ def test_smaller_existing_is_superseded(importer, import_layout):
             image_record_id=eid, tag_id=tag.id, source="manual"
         )
     )
-    importer.session.add(
-        ImagePrediction(
-            image_record_id=eid, raw_name="x", category="general", score=0.9
-        )
-    )
     old.siglip_embedding = [0.0] * 1152
     old.integrity_status = "ok"
     importer.session.commit()
@@ -141,11 +135,6 @@ def test_smaller_existing_is_superseded(importer, import_layout):
     assert row.path != old_path
     assert row.phash is not None
     assert row.integrity_status == "unknown"
-    # #768: re-import clears the normalized predictions too
-    assert importer.session.execute(
-        select(func.count()).select_from(ImagePrediction)
-        .where(ImagePrediction.image_record_id == eid)
-    ).scalar_one() == 0
     assert row.siglip_embedding is None
     linked = importer.session.execute(
         select(image_tag.c.tag_id).where(

tests/test_tag_merge.py

@@ -2,7 +2,6 @@ import pytest
 from sqlalchemy import func, select
 
 from backend.app.models import Tag, TagKind, image_tag
-from backend.app.models.tag_allowlist import TagAllowlist
 from backend.app.services.tag_service import (
     MergeResult,
     TagMergeConflict,
@@ -110,18 +109,6 @@ async def test_will_alias_true_when_machine_sourced(db):
     assert ei.value.will_alias is True
 
 
-@pytest.mark.asyncio
-async def test_will_alias_true_when_allowlisted(db):
-    svc = TagService(db)
-    await svc.find_or_create("Canon2", TagKind.general)
-    source = await svc.find_or_create("Allowed", TagKind.general)
-    db.add(TagAllowlist(tag_id=source.id))
-    await db.flush()
-    with pytest.raises(TagMergeConflict) as ei:
-        await svc.rename(source.id, "Canon2")
-    assert ei.value.will_alias is True
-
-
 @pytest.mark.asyncio
 async def test_merge_rejects_self_merge(db):
     svc = TagService(db)
@@ -250,63 +237,6 @@ async def test_merge_dedups_suggestion_rejections(db):
     ).first() is None
 
 
-@pytest.mark.asyncio
-async def test_merge_allowlist_target_has_keeps_target_threshold(db):
-    svc = TagService(db)
-    a = await svc.find_or_create("SrcAL", TagKind.general)
-    b = await svc.find_or_create("TgtAL", TagKind.general)
-    db.add(TagAllowlist(tag_id=a.id, min_confidence=0.5))
-    db.add(TagAllowlist(tag_id=b.id, min_confidence=0.9))
-    await db.flush()
-    await svc.merge(a.id, b.id)
-    rows = (await db.execute(select(TagAllowlist))).scalars().all()
-    assert len(rows) == 1
-    assert rows[0].tag_id == b.id
-    assert rows[0].min_confidence == 0.9
-
-
-@pytest.mark.asyncio
-async def test_merge_allowlist_source_only_moves_to_target(db):
-    svc = TagService(db)
-    a = await svc.find_or_create("SrcAL2", TagKind.general)
-    b = await svc.find_or_create("TgtAL2", TagKind.general)
-    db.add(TagAllowlist(tag_id=a.id, min_confidence=0.42))
-    await db.flush()
-    await svc.merge(a.id, b.id)
-    rows = (await db.execute(select(TagAllowlist))).scalars().all()
-    assert len(rows) == 1
-    assert rows[0].tag_id == b.id
-    assert rows[0].min_confidence == 0.42
-
-
-@pytest.mark.asyncio
-async def test_merge_deletes_source_embedding(db):
-    from backend.app.models.tag_reference_embedding import (
-        TagReferenceEmbedding,
-    )
-
-    svc = TagService(db)
-    a = await svc.find_or_create("SrcEmb", TagKind.general)
-    b = await svc.find_or_create("TgtEmb", TagKind.general)
-    db.add(
-        TagReferenceEmbedding(
-            tag_id=a.id,
-            embedding=[0.0] * 1152,
-            reference_count=1,
-            model_version="v",
-        )
-    )
-    await db.flush()
-    await svc.merge(a.id, b.id)
-    db.expire_all()  # merge() uses Core DML; drop stale identity-map state
-    remaining = await db.scalar(
-        select(func.count())
-        .select_from(TagReferenceEmbedding)
-        .where(TagReferenceEmbedding.tag_id == a.id)
-    )
-    assert remaining == 0
-
-
 @pytest.mark.asyncio
 async def test_merge_repoints_existing_aliases(db):
     from backend.app.models.tag_alias import TagAlias
@@ -400,7 +330,9 @@ async def test_alias_fallback_to_kind_when_no_predictions(db):
     svc = TagService(db)
     a = await svc.find_or_create("AllowNoPred", TagKind.character)
     b = await svc.find_or_create("CanonF", TagKind.character)
-    db.add(TagAllowlist(tag_id=a.id))
+    # Machine-known via a prior accept (source='ml_accepted') → kept as alias.
+    img = await _img(db)
+    await svc.add_to_image(img, a.id, source="ml_accepted")
     await db.flush()
     result = await svc.merge(a.id, b.id)
     assert result.alias_created is True

tests/test_tasks_ml.py

@@ -1,15 +1,12 @@
-"""tag_and_embed / backfill task tests. Models aren't in CI, so we test
-the pure helpers (_aggregate_video_predictions, _is_video) as unit tests, and
-the DB-touching backfill query as an integration test with monkeypatched
-inference.
-"""
+"""tag_and_embed (embed-only) / backfill task tests. The pure _is_video helper
+is a unit test; the DB-touching backfill query is an integration test with
+monkeypatched dispatch."""
 
 from pathlib import Path
 
 import pytest
 
-from backend.app.services.ml.tagger import TagPrediction
-from backend.app.tasks.ml import _aggregate_video_predictions, _is_video
+from backend.app.tasks.ml import _is_video
 
 
 def test_is_video():
@@ -18,34 +15,6 @@ def test_is_video():
     assert _is_video(Path("a.jpg")) is False
 
 
-def _pred(name, conf, cat="general"):
-    return {name: TagPrediction(name, cat, conf)}
-
-
-def test_aggregate_video_keeps_corroborated_and_means():
-    # #747: 4 frames; "smile" in 3, "sword" in 1 (noise). min_frames=2.
-    per_frame = [
-        {"smile": TagPrediction("smile", "general", 0.6),
-         "sword": TagPrediction("sword", "general", 0.9)},
-        _pred("smile", 0.8),
-        _pred("smile", 0.7),
-        {},
-    ]
-    out = _aggregate_video_predictions(per_frame, min_frames=2)
-    assert "sword" not in out  # one-frame flicker dropped
-    assert abs(out["smile"]["confidence"] - (0.6 + 0.8 + 0.7) / 3) < 1e-9  # mean, not max
-
-
-def test_aggregate_video_clamps_min_frames_to_sample_count():
-    # Short video: 1 frame but min_frames=3 — clamp so it still tags.
-    out = _aggregate_video_predictions([_pred("solo", 0.8)], min_frames=3)
-    assert out["solo"]["confidence"] == 0.8
-
-
-def test_aggregate_video_empty():
-    assert _aggregate_video_predictions([], min_frames=3) == {}
-
-
 @pytest.mark.integration
 @pytest.mark.asyncio
 async def test_backfill_enqueues_missing(db, monkeypatch):
@@ -69,75 +38,3 @@ async def test_backfill_enqueues_missing(db, monkeypatch):
     count = ml_tasks.backfill()
     assert count >= 1
     assert img.id in calls
-
-
-@pytest.mark.integration
-@pytest.mark.asyncio
-async def test_apply_allowlist_applies_above_threshold(db):
-    from sqlalchemy import select
-
-    from backend.app.models import ImageRecord, TagAllowlist, TagKind
-    from backend.app.models.tag import image_tag
-    from backend.app.services.tag_service import TagService
-    from backend.app.tasks import ml as ml_tasks
-    from tests._prediction_helpers import seed_predictions
-
-    tag = await TagService(db).find_or_create("autohero", TagKind.character)
-    db.add(TagAllowlist(tag_id=tag.id, min_confidence=0.95))
-    img = ImageRecord(
-        path="/images/al.jpg", sha256="al" + "0" * 62, size_bytes=1,
-        mime="image/jpeg", width=1, height=1,
-        origin="imported_filesystem", integrity_status="unknown",
-    )
-    db.add(img)
-    await db.commit()
-    await seed_predictions(
-        db, img.id, {"autohero": {"category": "character", "confidence": 0.97}}
-    )
-    await db.commit()
-
-    n = ml_tasks.apply_allowlist_tags(tag_id=tag.id)
-    assert n >= 1
-    src = (
-        await db.execute(
-            select(image_tag.c.source)
-            .where(image_tag.c.image_record_id == img.id)
-            .where(image_tag.c.tag_id == tag.id)
-        )
-    ).scalar_one()
-    assert src == "ml_auto"
-
-
-@pytest.mark.integration
-@pytest.mark.asyncio
-async def test_apply_allowlist_skips_below_threshold(db):
-    from sqlalchemy import select
-
-    from backend.app.models import ImageRecord, TagAllowlist, TagKind
-    from backend.app.models.tag import image_tag
-    from backend.app.services.tag_service import TagService
-    from backend.app.tasks import ml as ml_tasks
-    from tests._prediction_helpers import seed_predictions
-
-    tag = await TagService(db).find_or_create("lowconf", TagKind.character)
-    db.add(TagAllowlist(tag_id=tag.id, min_confidence=0.95))
-    img = ImageRecord(
-        path="/images/lc.jpg", sha256="lc" + "0" * 62, size_bytes=1,
-        mime="image/jpeg", width=1, height=1,
-        origin="imported_filesystem", integrity_status="unknown",
-    )
-    db.add(img)
-    await db.commit()
-    await seed_predictions(
-        db, img.id, {"lowconf": {"category": "character", "confidence": 0.40}}
-    )
-    await db.commit()
-    ml_tasks.apply_allowlist_tags(tag_id=tag.id)
-    applied = (
-        await db.execute(
-            select(image_tag.c.tag_id)
-            .where(image_tag.c.image_record_id == img.id)
-            .where(image_tag.c.tag_id == tag.id)
-        )
-    ).scalar_one_or_none()
-    assert applied is None