Merge pull request 'fix(agent): cuDNN base image so onnxruntime-gpu loads' (#146 ) from dev into main

Merge pull request 'ci: publish the GPU agent image' (#145 ) from dev into main
Merge pull request 'CCIP characters + crop/region pipeline + desktop GPU agent (#114 )' (#144 ) from dev into main
2026-06-29 18:48:24 -04:00 · 2026-06-29 14:26:51 -04:00 · 2026-06-29 14:18:57 -04:00 · 2026-06-29 07:30:41 -04:00 · 2026-06-28 11:55:49 -04:00 · 2026-06-28 01:36:26 -04:00
30 changed files with 150 additions and 1265 deletions
@@ -10,16 +10,11 @@ RUN apt-get update \
    && 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.
 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
 COPY fc_agent ./fc_agent
-# imgutils ONNX models + the transformers SigLIP weights both cache here; mount
+# imgutils caches downloaded ONNX models here; mount a volume to persist them.
 # a volume to persist them across restarts (the SigLIP download is ~3.5 GB once).
 ENV HF_HOME=/models
 EXPOSE 8770
@@ -7,17 +7,6 @@ Redis stay private; the agent never touches them.
 You run it when you want a burst and stop it to reclaim the card.
 ## 0. Host prerequisite — NVIDIA Container Toolkit
 Docker needs the toolkit to hand the GPU to a container (else: *"could not select
 device driver nvidia with capabilities [[gpu]]"*). On Arch/CachyOS:
 ```sh
 sudo pacman -S nvidia-container-toolkit
 sudo nvidia-ctk runtime configure --runtime=docker
 sudo systemctl restart docker
 # verify:
 docker run --rm --gpus all nvidia/cuda:12.4.1-base-ubuntu22.04 nvidia-smi
 ```
 ## 1. Get a token
 In FC: **Settings → Tagging → GPU agent → Generate token** (or Rotate). Copy it.
@@ -10,13 +10,6 @@
 #   4. Open http://localhost:8770 → Start. Pause/Stop hands the GPU back.
 #      docker compose down       to stop the container entirely.
 #
 # Surviving a curator redeploy (you're away, can't touch the agent):
 #   - A running agent rides out curator being unreachable on its own — it retries
 #     leasing with capped backoff and resumes when the server is back. In-flight
 #     work is handed back (not failed), so a redeploy never poisons good jobs.
 #   - AUTO_START=1 (below) also resumes the worker if the AGENT container itself
 #     restarts (host reboot / crash via `restart: unless-stopped`) — no click.
 #
 # Needs the NVIDIA Container Toolkit installed on the host for --gpus.
 services:
@@ -31,12 +24,6 @@ services:
      CCIP_MODEL: ${CCIP_MODEL:-}
      DETECTOR_LEVEL: ${DETECTOR_LEVEL:-m}
      BATCH_SIZE: ${BATCH_SIZE:-4}
      # 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}
      # 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}
    volumes:
      # Persist the downloaded ONNX models so restarts are fast.
      - fc-agent-models:/models
@@ -1,14 +1,13 @@
 """FastAPI control surface for the agent (served on localhost).
-Start / stop the worker pool, tune the worker count live (trades desktop
+Start / pause / resume / stop the worker, set nothing else here (config is env),
-responsiveness for throughput), and watch GPU load + progress + the server-side
+and watch progress + the server-side queue. The container exposes this on a
-queue. Config is env-seeded; the worker count is adjustable here on the fly.
+localhost port; stopping the worker frees the GPU.
 """
-from fastapi import FastAPI, Request
+from fastapi import FastAPI
 from fastapi.responses import HTMLResponse, JSONResponse
 from .config import Config
 from .gpu import read_gpu
 from .worker import Worker
 cfg = Config.from_env()
@@ -16,16 +15,6 @@ worker = Worker(cfg)
 app = FastAPI(title="FabledCurator GPU agent")
@app.on_event("startup")
 def _maybe_autostart() -> None:
    # With AUTO_START set, a container restart (host reboot, or `restart:
    # unless-stopped` after a crash) resumes the worker on its own — the slots
    # then ride out a still-down curator via lease backoff. Lets the agent
    # survive a redeploy with nobody at the desktop to click Start.
    if cfg.auto_start and cfg.token:
        worker.start()
@app.get("/", response_class=HTMLResponse)
 def index() -> str:
    return _PAGE
@@ -37,16 +26,21 @@ def start():
    return JSONResponse(worker.status())
-@app.post("/stop")
+@app.post("/pause")
-def stop():
+def pause():
-    worker.stop()
+    worker.pause()
    return JSONResponse(worker.status())
-@app.post("/concurrency")
+@app.post("/resume")
-async def concurrency(request: Request):
+def resume():
-    body = await request.json()
+    worker.resume()
-    worker.set_concurrency(int(body.get("value", 1)))
+    return JSONResponse(worker.status())
@app.post("/stop")
 def stop():
    worker.stop()
    return JSONResponse(worker.status())
@@ -55,7 +49,6 @@ def status():
    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:
@@ -66,68 +59,35 @@ def status():
 _PAGE = """<!doctype html><html><head><meta charset=utf-8>
 <title>FabledCurator GPU agent</title>
 <style>
- body{font:14px system-ui;margin:2rem;max-width:680px;background:#14171a;color:#e8e8e8}
+ body{font:14px system-ui;margin:2rem;max-width:640px;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}
+ margin-right:.5rem;cursor:pointer;color:#fff} .start{background:#2e7d32}.pause{background:#b26a00}
- .step{background:#33373b;padding:.4rem .7rem;font-weight:700}
+ .stop{background:#b3261e} .stat{display:inline-block;margin-right:1.5rem}
 .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;
+ .q{margin-top:1rem;color:#9aa}
 max-width:320px;margin-top:4px} .bar>i{display:block;height:100%;background:#3f7d3f}
 .row{margin:.8rem 0}
 </style></head><body>
 <h1>FabledCurator GPU agent</h1>
 <p>FC: <code id=fc>—</code> · token <code id=cfg>—</code></p>
-<div class=row>
+<p>
 <button class=start onclick=act('start')>Start</button>
 <button class=pause onclick=act('pause')>Pause</button>
 <button class=pause onclick=act('resume')>Resume</button>
 <button class=stop onclick=act('stop')>Stop</button>
-</div>
+</p>
-<div class=row>
+<p>
- workers
+ <span class=stat><span class=n id=state>idle</span><br>state</span>
 <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>
+ <span class=stat><span class=n id=cur>—</span><br>current image</span>
-</div>
+</p>
 <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>
 let CAP=8
 async function act(p){await fetch('/'+p,{method:'POST'});refresh()}
 function setc(d){ 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 refresh(){
  const s=await (await fetch('/status')).json()
-  CAP=s.max_concurrency||8; capn.textContent=CAP
+  state.textContent=s.state; done.textContent=s.processed; err.textContent=s.errors
-  state.textContent=s.state; active.textContent=s.active; done.textContent=s.processed
+  cur.textContent=s.current??'—'; fc.textContent=s.fc_url
  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'
  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'
 }
 refresh(); setInterval(refresh,3000)
@@ -4,7 +4,6 @@ The agent's ONLY contact with FC — lease/submit/heartbeat/fail + fetch image
 bytes, all over HTTP with the bearer token. No DB/Redis.
 """
 import requests
 from requests.adapters import HTTPAdapter
 class FcClient:
@@ -13,11 +12,6 @@ class FcClient:
        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
        # 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)
    def lease(self, batch_size: int) -> list[dict]:
        r = self.s.post(
@@ -60,19 +54,6 @@ class FcClient:
        except requests.RequestException:
            pass
    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
    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)
@@ -8,16 +8,10 @@ class Config:
    fc_url: str          # base URL of the FabledCurator web service
    token: str           # the bearer token from Settings → Tagging → GPU agent
    agent_id: str        # identifies this agent's leases
-    batch_size: int      # jobs a worker leases per round
+    batch_size: int      # jobs leased per round (concurrency is still 1)
    concurrency: int     # INITIAL parallel workers (tunable live from the UI)
    ccip_model: str      # imgutils CCIP model name ("" → imgutils default)
    detector_level: str  # imgutils person-detector level: n|s|m|x
    poll_idle_seconds: float   # wait between empty leases
    embed_dtype: str     # torch dtype for the crop embedder: float16|float32
    embed_model_override: str  # force a SigLIP-family model ("" → use the one
    #                            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)
    @classmethod
    def from_env(cls) -> "Config":
@@ -26,11 +20,7 @@ class Config:
            token=os.environ.get("FC_TOKEN", ""),
            agent_id=os.environ.get("AGENT_ID", "desktop-agent"),
            batch_size=int(os.environ.get("BATCH_SIZE", "4")),
            concurrency=int(os.environ.get("CONCURRENCY", "1")),
            ccip_model=os.environ.get("CCIP_MODEL", ""),
            detector_level=os.environ.get("DETECTOR_LEVEL", "m"),
            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"),
        )
@@ -1,69 +0,0 @@
 """Crop EMBEDDER for the concept bag — model-agnostic (CLIP/SigLIP-family).
 The server trains its per-concept heads in the embedding space of whatever model
 its `embedder_model_version` names; a crop must be embedded with the SAME model
 or its vector lands in a different coordinate system and every head misfires. So
 the model identity (HF name + version) is ANNOUNCED BY THE SERVER in the lease —
 nothing here is hardcoded to SigLIP. Whatever name the server sends is loaded via
 transformers `get_image_features` (the CLIP/SigLIP-family image-tower call); a
 non-CLIP backbone (e.g. a DINO encoder) would need its own pooling adapter.
 torch on CUDA, fp16 by default to keep VRAM low on a shared desktop GPU — the
 tiny fp16-vs-fp32 difference is negligible for the linear heads (cosine ~0.999).
 A single inference lock serializes the forward pass: the pipeline is I/O-bound,
 so the GPU isn't the bottleneck, and one model shared across worker threads is
 safest behind a lock.
 """
 import threading
 import numpy as np
 from PIL import Image
 class CropEmbedder:
    def __init__(self, model_name: str, dtype: str = "float16"):
        self._name = model_name
        self._dtype_name = dtype
        self._model = None
        self._processor = None
        self._torch = None
        self._device = None
        self._dt = None
        self._load_lock = threading.Lock()
        self._infer_lock = threading.Lock()
    @property
    def model_name(self) -> str:
        return self._name
    def load(self) -> None:
        if self._model is not None:
            return
        with self._load_lock:
            if self._model is not None:
                return
            import torch
            from transformers import AutoImageProcessor, AutoModel
            self._torch = torch
            self._device = "cuda" if torch.cuda.is_available() else "cpu"
            dt = getattr(torch, self._dtype_name, torch.float16)
            if self._device == "cpu":
                dt = torch.float32   # fp16 matmul is unsupported/slow on CPU
            self._dt = dt
            self._processor = AutoImageProcessor.from_pretrained(self._name)
            model = AutoModel.from_pretrained(self._name, torch_dtype=dt)
            model.eval().to(self._device)
            self._model = model
    def embed(self, image: Image.Image) -> list[float]:
        """A crop → its embedding as a plain float list, ready to POST."""
        self.load()
        torch = self._torch
        enc = self._processor(images=image, 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()
@@ -1,30 +0,0 @@
 """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)."""
 import subprocess
 def read_gpu() -> dict | None:
    try:
        out = subprocess.run(
            [
                "nvidia-smi",
                "--query-gpu=utilization.gpu,memory.used,memory.total,temperature.gpu",
                "--format=csv,noheader,nounits",
            ],
            capture_output=True, text=True, timeout=5, check=True,
        ).stdout.strip().splitlines()
    except (OSError, subprocess.SubprocessError):
        return None
    if not out:
        return None
    parts = [p.strip() for p in out[0].split(",")]
    try:
        return {
            "util_pct": int(float(parts[0])),
            "mem_used_mb": int(float(parts[1])),
            "mem_total_mb": int(float(parts[2])),
            "temp_c": int(float(parts[3])),
        }
    except (ValueError, IndexError):
        return None
@@ -13,23 +13,8 @@ def is_video(mime: str) -> bool:
    return bool(mime) and (mime.startswith("video/") or mime in {"image/gif"})
 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
    background) lets PIL guess the transparent pixels — usually black artifacts
    that bleed into the crop + the embedding (and the "should be converted to
    RGBA" warning). Compositing over white gives a clean, consistent background."""
    if img.mode in ("RGBA", "LA", "PA") or (
        img.mode == "P" and "transparency" in img.info
    ):
        img = img.convert("RGBA")
        bg = Image.new("RGBA", img.size, (255, 255, 255, 255))
        return Image.alpha_composite(bg, img).convert("RGB")
    return img.convert("RGB")
 def load_image(data: bytes) -> Image.Image:
-    return to_rgb(Image.open(io.BytesIO(data)))
+    return Image.open(io.BytesIO(data)).convert("RGB")
 def sample_frames(
@@ -59,5 +44,5 @@ def sample_frames(
        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)))
+                out.append((round(i * interval, 2), im.convert("RGB")))
        return out
@@ -1,188 +1,92 @@
-"""The lease → fetch → detect+embed → submit loop, run by a pool of worker
+"""The lease → fetch → detect+embed → submit loop, with start/pause/stop control.
 slots whose count is tunable live from the UI.
-Each slot is an independent loop (its own leases; the server's SKIP-LOCKED lease
+Concurrency is 1 (one image at a time) so the GPU footprint stays small and a
-keeps them from colliding). More slots = more GPU load + throughput; the model is
+stop frees the card promptly. Stop halts leasing + finishes the current item;
-loaded once and shared, so slots add concurrent inference, not N× model VRAM.
+unprocessed leases expire and the server re-queues them — nothing is lost.
 That's the dial the operator turns to trade desktop responsiveness for speed.
 Stop (or shrinking the pool) RELEASES a slot's still-leased jobs immediately so
 orphaned work is re-picked at once rather than waiting out the lease.
 """
 import threading
-
+import time
 import requests
 from . import media, models
 from .client import FcClient
 from .config import Config
 from .crops import crop_region
 # Cap on the lease-retry backoff: when curator is unreachable (e.g. you redeploy
 # it while away), each slot retries leasing with exponential backoff up to this
 # many seconds, then resumes within this window once the server is back — no
 # restart needed.
 MAX_BACKOFF_SECONDS = 60.0
 def _is_transient(exc: "requests.RequestException") -> bool:
    """A server/transport problem (wait it out) vs a job-specific fault (fail it).
    No response → connection refused/timeout → curator is down → transient. With
    a response: 5xx, auth (401/403, e.g. a token blip on redeploy), 408/409/429
    (timeout / our lease reclaimed / rate-limited) are all 'not this job's fault'.
    A specific 4xx like 404 (image gone) / 400 IS the job's fault → fail it."""
    resp = getattr(exc, "response", None)
    if resp is None:
        return True
    return resp.status_code >= 500 or resp.status_code in (401, 403, 408, 409, 429)
 # Generous cap: the pipeline is usually I/O-bound (downloading + decoding images
 # over HTTP), so the GPU stays underused until many workers overlap that I/O.
 # Push it up while watching the GPU util + VRAM in the UI.
 MAX_CONCURRENCY = 32
 # Fallbacks only — the server ANNOUNCES the embedding model (name + version) in
 # the lease so the agent stays model-agnostic and in lock-step with the space
 # the heads were trained in. These cover an older server that doesn't send them.
 DEFAULT_EMBED_MODEL = "google/siglip-so400m-patch14-384"
 DEFAULT_EMBED_VERSION = "siglip-so400m-patch14-384"
 class _Slot:
    """One worker loop. `inflight` = jobs leased but not yet processed, so a
    graceful stop can hand them back."""
    __slots__ = ("stop", "inflight")
    def __init__(self):
        self.stop = threading.Event()
        self.inflight: list[int] = []
 class Worker:
    def __init__(self, cfg: Config):
        self.cfg = cfg
        self.client = FcClient(cfg.fc_url, cfg.token, cfg.agent_id)
        self._state = "idle"        # idle | running | paused | stopping
        self._lock = threading.Lock()
-        self._running = False
+        self._thread: threading.Thread | None = None
        self._target = max(1, min(MAX_CONCURRENCY, cfg.concurrency))
        self._slots: list[_Slot] = []
        self.processed = 0
        self.errors = 0
-        self.transient = 0        # jobs handed back due to a server outage (NOT
+        self.current = None
        #                           failed) — the "waiting out curator" counter
        self._active = 0          # slots currently mid-image
        # The crop embedder (SigLIP-family) is built lazily on the first job that
        # needs it, from the model the server announces — one shared instance.
        self._embedder = None
        self._embedder_lock = threading.Lock()
    # --- control -----------------------------------------------------------
    def start(self):
        with self._lock:
-            self._running = True
+            if self._state in ("running", "paused"):
-            self._reconcile_locked()
+                self._state = "running"
                return
            self._state = "running"
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()
    def pause(self):
        with self._lock:
            if self._state == "running":
                self._state = "paused"
    def resume(self):
        with self._lock:
            if self._state == "paused":
                self._state = "running"
    def stop(self):
        with self._lock:
-            self._running = False
+            if self._state in ("running", "paused"):
-            slots, self._slots = self._slots, []
+                self._state = "stopping"
        for s in slots:
            s.stop.set()          # each slot releases its inflight on exit
    def set_concurrency(self, n: int):
        with self._lock:
            self._target = max(1, min(MAX_CONCURRENCY, int(n)))
            if self._running:
                self._reconcile_locked()
    def _reconcile_locked(self):
        while len(self._slots) < self._target:
            slot = _Slot()
            self._slots.append(slot)
            threading.Thread(target=self._loop, args=(slot,), daemon=True).start()
        while len(self._slots) > self._target:
            self._slots.pop().stop.set()
    def status(self) -> dict:
        with self._lock:
-            return {
+            state = self._state
-                "state": "running" if self._running else "stopped",
+        return {
-                "concurrency": self._target,
+            "state": state, "processed": self.processed,
-                "max_concurrency": MAX_CONCURRENCY,
+            "errors": self.errors, "current": self.current,
-                "workers": len(self._slots),
+        }
                "active": self._active,
                "processed": self.processed,
                "errors": self.errors,
                "transient": self.transient,
            }
-    def _bump(self, *, processed=0, errors=0, active=0, transient=0):
+    # --- loop --------------------------------------------------------------
-        with self._lock:
+    def _run(self):
-            self.processed += processed
+        while True:
-            self.errors += errors
+            with self._lock:
-            self.transient += transient
+                st = self._state
-            self._active += active
+            if st == "stopping":
-
+                break
-    # --- per-slot loop -----------------------------------------------------
+            if st == "paused":
-    def _loop(self, slot: _Slot):
+                time.sleep(1)
-        backoff = self.cfg.poll_idle_seconds
+                continue
        while not slot.stop.is_set() and self._running:
            try:
                jobs = self.client.lease(self.cfg.batch_size)
                backoff = self.cfg.poll_idle_seconds      # server answered → reset
            except Exception:
-                # curator unreachable (redeploy, network drop): wait it out with
+                time.sleep(self.cfg.poll_idle_seconds)
                # exponential backoff, capped — resume on our own when it returns.
                self._interruptible_sleep(slot, backoff)
                backoff = min(backoff * 2, MAX_BACKOFF_SECONDS)
                continue
            if not jobs:
-                self._interruptible_sleep(slot, self.cfg.poll_idle_seconds)
+                time.sleep(self.cfg.poll_idle_seconds)
                continue
-            slot.inflight = [j["job_id"] for j in jobs]
+            ids = [j["job_id"] for j in jobs]
            for job in jobs:
-                if slot.stop.is_set() or not self._running:
+                with self._lock:
-                    break
+                    if self._state == "stopping":
-                ok = self._process(job)
+                        break
-                slot.inflight = [i for i in slot.inflight if i != job["job_id"]]
+                self._process(job)
-                if not ok:
+                self.client.heartbeat(ids)   # keep the rest of the batch alive
-                    # Server went away mid-batch: hand the rest back (best effort)
+        with self._lock:
-                    # and back off instead of hammering a recovering server or
+            self._state = "idle"
                    # burning the jobs' attempt budgets on fail().
                    if slot.inflight:
                        self.client.release(slot.inflight)
                        slot.inflight = []
                    self._interruptible_sleep(slot, backoff)
                    backoff = min(backoff * 2, MAX_BACKOFF_SECONDS)
                    break
                if slot.inflight:
                    self.client.heartbeat(slot.inflight)
        # Graceful hand-back of anything leased but not processed.
        if slot.inflight:
            self.client.release(slot.inflight)
            slot.inflight = []
-    def _interruptible_sleep(self, slot: _Slot, seconds: float):
+    def _process(self, job: dict):
-        """Sleep, but wake immediately if the slot is told to stop — so a Stop or
+        self.current = job.get("image_id")
        a pool-shrink doesn't hang for a full backoff window."""
        slot.stop.wait(timeout=seconds)
    def _ensure_embedder(self, model_name: str):
        if self._embedder is not None:
            return self._embedder
        with self._embedder_lock:
            if self._embedder is None:
                from .embedder import CropEmbedder
                self._embedder = CropEmbedder(model_name, self.cfg.embed_dtype)
        return self._embedder
    def _process(self, job: dict) -> bool:
        """Process one job. Returns True when handled (completed, or hard-failed
        because the job itself is bad) and False on a TRANSPORT error (curator
        unreachable / 5xx / our lease was reclaimed mid-flight) — which is not
        the job's fault, so the caller backs off and the job is left to be
        re-leased rather than fail()ed into its attempt budget."""
        self._bump(active=1)
        try:
            data = self.client.fetch_image(job["image_url"])
            if media.is_video(job.get("mime", "")):
@@ -193,31 +97,8 @@ class Worker:
            else:
                frames = [(None, media.load_image(data))]
            # task picks what to produce per crop:
            #   'siglip' (backfill existing images) → concept (SigLIP) regions
            #     ONLY, so it never churns their figure/CCIP regions or the
            #     character-reference cache.
            #   'ccip' / 'both' (a new image's first pass) → figure (CCIP) AND
            #     concept (SigLIP) in one go, off the same crop.
            task = job.get("task") or "ccip"
            want_ccip = task in ("ccip", "both")
            want_siglip = task in ("ccip", "siglip", "both")
            replace_kinds = (
                ["concept"] if task == "siglip" else ["figure", "face", "concept"]
            )
            embed_version = job.get("embed_version") or DEFAULT_EMBED_VERSION
            embedder = None
            if want_siglip:
                model_name = (
                    self.cfg.embed_model_override
                    or job.get("embed_model_name")
                    or DEFAULT_EMBED_MODEL
                )
                embedder = self._ensure_embedder(model_name)
            regions = []
-            ccip_ev = self.cfg.ccip_model or "ccip-default"
+            ev = self.cfg.ccip_model or "ccip-default"
            dv = f"person-{self.cfg.detector_level}"
            for t, frame in frames:
                figs = models.detect_figures(frame, self.cfg.detector_level)
@@ -227,48 +108,20 @@ class Worker:
                    crop = crop_region(frame, bbox)
                    if crop is None:
                        continue
-                    if want_ccip:
+                    vec = models.ccip_vector(crop, self.cfg.ccip_model or None)
-                        regions.append({
+                    regions.append({
-                            "kind": "figure",
+                        "kind": "figure",
-                            "bbox": list(bbox),
+                        "bbox": list(bbox),
-                            "frame_time": t,
+                        "frame_time": t,
-                            "score": score,
+                        "score": score,
-                            "ccip_embedding": models.ccip_vector(
+                        "ccip_embedding": vec,
-                                crop, self.cfg.ccip_model or None
+                        "embedding_version": ev,
-                            ),
+                        "detector_version": dv,
-                            "embedding_version": ccip_ev,
+                    })
-                            "detector_version": dv,
+            self.client.submit(job["job_id"], regions, ["figure", "face"])
-                        })
+            self.processed += 1
-                    if want_siglip:
+        except Exception as exc:   # noqa: BLE001 — report + move on
-                        regions.append({
+            self.errors += 1
                            "kind": "concept",
                            "bbox": list(bbox),
                            "frame_time": t,
                            "score": score,
                            "siglip_embedding": embedder.embed(crop),
                            "embedding_version": embed_version,
                            "detector_version": dv,
                        })
            self.client.submit(job["job_id"], regions, replace_kinds)
            self._bump(processed=1)
            return True
        except requests.RequestException as exc:
            if _is_transient(exc):
                # curator down/redeploying, a 5xx, or our lease was reclaimed
                # while we worked. NOT the job's fault — hand it back (best
                # effort; no-ops if the server is still down, then the server's
                # orphan-recovery reclaims it) and signal the loop to wait.
                self._bump(transient=1)
                self.client.release([job["job_id"]])
                return False
            # A job-specific HTTP fault (404 image gone, 400) → fail it so it
            # doesn't re-lease forever.
            self._bump(errors=1)
            self.client.fail(job["job_id"], str(exc)[:500])
            return True
        except Exception as exc:   # noqa: BLE001 — a genuine job fault: report it
            self._bump(errors=1)
            self.client.fail(job["job_id"], str(exc)[:500])
            return True
        finally:
-            self._bump(active=-1)
+            self.current = None
@@ -3,10 +3,6 @@ dghs-imgutils>=0.4
 # GPU inference for the ONNX models. Swap to onnxruntime (CPU) for a slow
 # server-side fallback run.
 onnxruntime-gpu
 # The crop EMBEDDER (concept bag). torch is installed separately in the
 # 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
 # Control surface + HTTP.
 fastapi
 uvicorn[standard]
@@ -1,33 +0,0 @@
 """ml_settings.ccip_match_threshold — tunable CCIP character-match cut (#114)
 The v1 matcher used a flat 0.75 cosine; live data showed that over-fires (a
 high-reference character matched a scatter of images). 0.85 keeps the confident
 single-character matches and drops the noise. Tunable from the GPU agent card.
 Revision ID: 0063
 Revises: 0062
 Create Date: 2026-06-29
 """
 from typing import Sequence, Union
 import sqlalchemy as sa
 from alembic import op
 revision: str = "0063"
 down_revision: Union[str, None] = "0062"
 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(
            "ccip_match_threshold", sa.Float(), nullable=False,
            server_default="0.85",
        ),
    )
 def downgrade() -> None:
    op.drop_column("ml_settings", "ccip_match_threshold")
@@ -1,42 +0,0 @@
 """ml_settings: CCIP auto-apply switch + threshold (#114)
 Confident CCIP character matches auto-tag (source='ccip_auto') on a daily sweep,
 so identity tags keep flowing without pressing a button. ON by default (opt-out,
 like head auto-apply); the high threshold (0.92, above the 0.85 suggest cut) +
 single-character references keep it safe, and every auto-tag is reversible.
 Revision ID: 0064
 Revises: 0063
 Create Date: 2026-06-30
 """
 from typing import Sequence, Union
 import sqlalchemy as sa
 from alembic import op
 revision: str = "0064"
 down_revision: Union[str, None] = "0063"
 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(
            "ccip_auto_apply_enabled", sa.Boolean(), nullable=False,
            server_default=sa.true(),
        ),
    )
    op.add_column(
        "ml_settings",
        sa.Column(
            "ccip_auto_apply_threshold", sa.Float(), nullable=False,
            server_default="0.92",
        ),
    )
 def downgrade() -> None:
    op.drop_column("ml_settings", "ccip_auto_apply_threshold")
    op.drop_column("ml_settings", "ccip_auto_apply_enabled")
@@ -37,15 +37,6 @@ async def overview():
                .where(ImageRegion.ccip_embedding.is_not(None))
            )
        ).scalar_one()
        # Concept-crop (SigLIP bag) coverage — how far the back-catalogue embed
        # has progressed, so the max-over-bag scorer's reach is checkable.
        images_with_concept_siglip = (
            await session.execute(
                select(func.count(distinct(ImageRegion.image_record_id)))
                .where(ImageRegion.kind == "concept")
                .where(ImageRegion.siglip_embedding.is_not(None))
            )
        ).scalar_one()
        # Per-character reference counts (no vectors loaded) — which characters
        # have enough examples to match on.
        ref_rows = (
@@ -71,23 +62,14 @@ async def overview():
                )
            ).all() if v
        ]
        auto_applied = (
            await session.execute(
                select(func.count()).select_from(image_tag).where(
                    image_tag.c.source == "ccip_auto"
                )
            )
        ).scalar_one()
    return jsonify({
        "regions_by_kind": by_kind,
        "images_with_figure_ccip": images_with_figure_ccip,
        "images_with_concept_siglip": images_with_concept_siglip,
        "characters_with_references": len(ref_rows),
        "character_references": [
            {"tag_id": t, "name": n, "n_refs": c} for (t, n, c) in ref_rows
        ],
        "embedding_versions": versions,
        "auto_applied": auto_applied,
    })
@@ -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
@@ -138,12 +137,6 @@ 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,
                "embed_version": ml.embedder_model_version,
            })
    return jsonify({"jobs": out})
@@ -203,18 +196,3 @@ async def fail():
        )
        await session.commit()
    return jsonify({"ok": ok})
@gpu_bp.route("/jobs/release", methods=["POST"])
 async def release():
    """Graceful stop: the agent hands its still-leased jobs back to pending so
    they're picked up immediately instead of waiting out the lease."""
    body = await request.get_json(silent=True) or {}
    agent_id = str(body.get("agent_id") or "agent")
    job_ids = [int(x) for x in (body.get("job_ids") or [])]
    async with get_session() as session:
        if not await _agent_authed(session):
            return jsonify({"error": "unauthorized"}), 401
        n = await GpuJobService(session).release(agent_id, job_ids)
        await session.commit()
    return jsonify({"released": n})
@@ -21,9 +21,6 @@ _EDITABLE = (
    "head_auto_apply_precision",
    "head_auto_apply_enabled",
    "head_auto_apply_min_positives",
    "ccip_match_threshold",
    "ccip_auto_apply_enabled",
    "ccip_auto_apply_threshold",
 )
@@ -51,9 +48,6 @@ async def get_settings():
            "head_auto_apply_precision": s.head_auto_apply_precision,
            "head_auto_apply_enabled": s.head_auto_apply_enabled,
            "head_auto_apply_min_positives": s.head_auto_apply_min_positives,
            "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,
        }
    )
@@ -121,10 +115,6 @@ def _validate(p: dict) -> str | None:
        return "head_auto_apply_precision must be between 0.5 and 0.999"
    if int(p["head_auto_apply_min_positives"]) < 1:
        return "head_auto_apply_min_positives must be >= 1"
    if not (0.5 <= float(p["ccip_match_threshold"]) <= 0.999):
        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"
    return None
@@ -117,24 +117,6 @@ def make_celery() -> Celery:
                "task": "backend.app.tasks.ml.scheduled_apply_head_tags",
                "schedule": 86400.0,  # no-op unless head_auto_apply_enabled
            },
            "recover-orphaned-gpu-jobs": {
                "task": "backend.app.tasks.ml.recover_orphaned_gpu_jobs",
                "schedule": 60.0,  # quick pickup of work a dead agent orphaned
            },
            "enqueue-ccip-backfill-hourly": {
                "task": "backend.app.tasks.ml.enqueue_gpu_backfill",
                "schedule": 3600.0,  # auto-feed new images (+ retry errored) so
                "args": ("ccip",),   # the queue keeps moving without the button
            },
            "enqueue-siglip-backfill-daily": {
                "task": "backend.app.tasks.ml.enqueue_gpu_backfill",
                "schedule": 86400.0,  # drain the concept-crop back-catalogue +
                "args": ("siglip",),  # retry failed embeds, no button needed
            },
            "ccip-auto-apply-daily": {
                "task": "backend.app.tasks.ml.scheduled_ccip_auto_apply",
                "schedule": 86400.0,  # no-op unless ccip_auto_apply_enabled
            },
            "snapshot-head-metrics-daily": {
                "task": "backend.app.tasks.maintenance.snapshot_head_metrics",
                "schedule": 86400.0,
@@ -86,21 +86,6 @@ class MLSettings(Base):
    head_auto_apply_min_positives: Mapped[int] = mapped_column(
        Integer, nullable=False, default=30
    )
    # CCIP character-match cosine cut (#114). 0.85 default — the v1 flat 0.75
    # over-fired (high-reference characters matched a scatter of images); 0.85
    # keeps the confident single-character matches. Tunable from the agent card.
    ccip_match_threshold: Mapped[float] = mapped_column(
        Float, nullable=False, default=0.85
    )
    # CCIP auto-apply (#114). Confident matches (>= ccip_auto_apply_threshold,
    # above the suggest cut) auto-tag on a daily sweep. ON by default (opt-out);
    # single-character references + the high bar keep it safe, every tag reversible.
    ccip_auto_apply_enabled: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True
    )
    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"
    )
@@ -13,82 +13,28 @@ exact CCIP difference metric/threshold gets validated against the model during
 the hands-on eval. numpy is imported lazily (API worker has it via pgvector).
 """
-from sqlalchemy import func, select
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
-from ...models import ImageRegion, MLSettings, Tag, TagKind
+from ...models import ImageRegion, Tag, TagKind
 from ...models.tag import image_tag
-# Cosine-similarity floor to call a figure the same character. The live setting
+# Cosine-similarity floor to call a figure the same character. Conservative
-# (ml_settings.ccip_match_threshold) drives it; this is only the fallback when no
+# default; tune from real matches (CCIP same-char clusters tightly).
-# threshold is supplied AND no settings row exists.
+DEFAULT_SIM_THRESHOLD = 0.75
 DEFAULT_SIM_THRESHOLD = 0.85
 _FIGURE_KINDS = ("face", "figure")
 async def _settings_threshold(session: AsyncSession) -> float:
    val = (
        await session.execute(
            select(MLSettings.ccip_match_threshold).where(MLSettings.id == 1)
        )
    ).scalar_one_or_none()
    return float(val) if val is not None else DEFAULT_SIM_THRESHOLD
 def _l2norm(mat, np):
    n = np.linalg.norm(mat, axis=1, keepdims=True)
    n[n == 0] = 1.0
    return mat / n
 # Single-shot cache of the (expensive) reference load, keyed on a cheap
 # signature that changes exactly when references could: a character tag added/
 # removed (n_char_tags) or a figure embedded (max/ n of ccip regions). Shared by
 # the live matcher (every modal open) and the auto-apply sweep.
 _REF_CACHE: dict = {"sig": None, "refs": None}
 def _single_character_images():
    """Subquery of image ids carrying EXACTLY ONE character tag. References come
    only from these — on a multi-character image the tag is image-level, so every
    figure would otherwise pollute each character's prototype set (a 2-character
    image tagged 'Velma' would make Daphne's figure a Velma reference)."""
    return (
        select(image_tag.c.image_record_id)
        .join(Tag, Tag.id == image_tag.c.tag_id)
        .where(Tag.kind == TagKind.character)
        .group_by(image_tag.c.image_record_id)
        .having(func.count() == 1)
    )
 async def _ref_signature(session: AsyncSession) -> tuple:
    n_tags = (
        await session.execute(
            select(func.count())
            .select_from(image_tag)
            .join(Tag, Tag.id == image_tag.c.tag_id)
            .where(Tag.kind == TagKind.character)
        )
    ).scalar_one()
    n_regs, max_id = (
        await session.execute(
            select(func.count(), func.max(ImageRegion.id)).where(
                ImageRegion.kind.in_(_FIGURE_KINDS),
                ImageRegion.ccip_embedding.is_not(None),
            )
        )
    ).one()
    return (n_tags, n_regs, max_id)
 async def character_references(session: AsyncSession) -> dict[int, list]:
    """Per character-tag CCIP reference vectors: figure/face-region CCIP
-    embeddings on UNAMBIGUOUS (single-character) images carrying that tag.
+    embeddings on images that carry that character tag (the operator's examples).
-    Multi-prototype — several vectors per character. Cached on a cheap signature."""
+    Multi-prototype — several vectors per character."""
    sig = await _ref_signature(session)
    if _REF_CACHE["sig"] == sig and _REF_CACHE["refs"] is not None:
        return _REF_CACHE["refs"]
    rows = (
        await session.execute(
            select(image_tag.c.tag_id, ImageRegion.ccip_embedding)
@@ -101,13 +47,11 @@ async def character_references(session: AsyncSession) -> dict[int, list]:
            .where(Tag.kind == TagKind.character)
            .where(ImageRegion.kind.in_(_FIGURE_KINDS))
            .where(ImageRegion.ccip_embedding.is_not(None))
            .where(ImageRegion.image_record_id.in_(_single_character_images()))
        )
    ).all()
    refs: dict[int, list] = {}
    for tag_id, vec in rows:
        refs.setdefault(tag_id, []).append(vec)
    _REF_CACHE.update(sig=sig, refs=refs)
    return refs
@@ -124,18 +68,14 @@ async def _tag_names(session: AsyncSession, tag_ids: list[int]) -> dict[int, str
 async def match_image(
-    session: AsyncSession, image_id: int, threshold: float | None = None
+    session: AsyncSession, image_id: int, threshold: float = DEFAULT_SIM_THRESHOLD
 ) -> list[dict]:
    """Character suggestions for one image from its figure-region CCIP vectors:
    [{tag_id, name, category:'character', score, source:'ccip'}], ranked.
    Already-applied character tags are excluded. Empty if the image has no figure
-    CCIP vectors or no character references exist yet. threshold defaults to the
+    CCIP vectors or no character references exist yet."""
    live ml_settings.ccip_match_threshold."""
    import numpy as np
    if threshold is None:
        threshold = await _settings_threshold(session)
    qvecs = (
        await session.execute(
            select(ImageRegion.ccip_embedding).where(
@@ -1,13 +1,10 @@
-"""GPU-job queue engine (#114): enqueue / lease / heartbeat / complete / fail
+"""GPU-job queue engine (#114): enqueue / lease / heartbeat / complete / fail.
 / release / recover_orphaned.
 Backs the HTTP API the desktop agent pulls work from. The lease claims pending
-OR expired-leased jobs with FOR UPDATE SKIP LOCKED, so concurrent agents/workers
+OR expired-leased jobs with FOR UPDATE SKIP LOCKED, so concurrent agents (or a
-never grab the same job. Orphan recovery is three-layered: a graceful agent stop
+retry after an agent died) never grab the same job and the queue self-heals
-calls release() to hand its in-flight jobs back instantly; a hard crash is caught
+without a separate recovery sweep. Result-writing (regions) is done by the API
-by recover_orphaned() (a 60s beat sweep) which resets expired leases to pending;
+handler via RegionService; complete() just closes the job.
 and the lease itself reclaims expired leases as a final backstop. Result-writing
 (regions) is done by the API handler via RegionService; complete() just closes.
 """
 from datetime import UTC, datetime, timedelta
@@ -17,10 +14,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from ...models import GpuJob
-# Lease window. Kept comfortably above any single job (a capped-frame video embed
+DEFAULT_LEASE_TTL = 300     # seconds an agent holds a job before it can be re-leased
 # is tens of seconds) so a live, heartbeating worker is never falsely expired,
 # but short enough that a hard crash recovers fast once the sweep fires.
 DEFAULT_LEASE_TTL = 180     # seconds an agent holds a job before it can be re-leased
 DEFAULT_BATCH = 8
 MAX_ATTEMPTS = 3
@@ -138,40 +132,3 @@ class GpuJobService:
        job.error = (error or "")[:1000]
        job.updated_at = datetime.now(UTC)
        return True
    async def release(self, token: str, job_ids: list[int]) -> int:
        """Hand the agent's still-leased jobs back to pending NOW (graceful stop),
        so another worker picks them up immediately instead of waiting out the
        lease. Scoped to the token's own leases. Returns rows released."""
        if not job_ids:
            return 0
        now = datetime.now(UTC)
        res = await self.session.execute(
            update(GpuJob)
            .where(
                GpuJob.id.in_(job_ids),
                GpuJob.lease_token == token,
                GpuJob.status == "leased",
            )
            .values(
                status="pending", lease_token=None, leased_at=None,
                lease_expires_at=None, updated_at=now,
            )
        )
        return res.rowcount or 0
    async def recover_orphaned(self) -> int:
        """Reset every expired lease back to pending — catches agents that died
        mid-job (no graceful release). Run on a short beat so the queue recovers
        + reads honestly even when no worker is actively leasing. Returns rows
        recovered."""
        now = datetime.now(UTC)
        res = await self.session.execute(
            update(GpuJob)
            .where(GpuJob.status == "leased", GpuJob.lease_expires_at < now)
            .values(
                status="pending", lease_token=None, leased_at=None,
                lease_expires_at=None, updated_at=now,
            )
        )
        return res.rowcount or 0
@@ -29,7 +29,6 @@ from ...models import (
    HeadAutoApplyRun,
    HeadTrainingRun,
    ImageRecord,
    ImageRegion,
    MLSettings,
    Tag,
    TagHead,
@@ -297,14 +296,7 @@ async def score_image(
    category, score}], ranked. A concept surfaces when its score clears the
    head's own suggest_threshold — or, when threshold_override is given (the
    typed-dropdown "show everything" mode), that flat floor instead (0 → every
-    head). Empty if the image has no embedding or no heads exist yet.
+    head). Empty if the image has no embedding or no heads exist yet."""
    MAX-OVER-BAG: the image is scored as a BAG of embeddings — the whole-image
    vector PLUS every concept-region crop the agent embedded (same model
    version) — and each head takes its MAX score across the bag. A small/local
    concept (glasses, a stomach bulge) that the whole-image vector washes out
    can still surface from the crop where it dominates. The whole-image vector is
    always in the bag, so this can never score lower than whole-image alone."""
    import numpy as np
    img = await session.get(ImageRecord, image_id)
@@ -314,26 +306,11 @@ async def score_image(
    heads = await _current_heads(session, settings.embedder_model_version)
    if heads["W"] is None:
        return []
-
+    x = np.asarray(img.siglip_embedding, dtype=np.float32)
-    bag = [np.asarray(img.siglip_embedding, dtype=np.float32)]
+    n = float(np.linalg.norm(x)) or 1.0
-    region_vecs = (
+    xn = x / n
-        await session.execute(
+    z = heads["W"] @ xn + heads["b"]
-            select(ImageRegion.siglip_embedding)
+    probs = 1.0 / (1.0 + np.exp(-z))
            .where(ImageRegion.image_record_id == image_id)
            .where(ImageRegion.siglip_embedding.is_not(None))
            .where(ImageRegion.embedding_version == settings.embedder_model_version)
        )
    ).all()
    for (vec,) in region_vecs:
        if vec is not None:
            bag.append(np.asarray(vec, dtype=np.float32))
    X = np.vstack(bag)                                  # (B, D)
    norms = np.linalg.norm(X, axis=1, keepdims=True)
    norms[norms == 0] = 1.0
    Xn = X / norms
    Z = Xn @ heads["W"].T + heads["b"]                  # (B, H)
    probs = (1.0 / (1.0 + np.exp(-Z))).max(axis=0)      # (H,) best over the bag
    out = []
    for i, p in enumerate(probs):
        cut = threshold_override if threshold_override is not None else heads["thr"][i]
@@ -742,43 +742,24 @@ def scheduled_apply_head_tags() -> str:
@celery.task(name="backend.app.tasks.ml.enqueue_gpu_backfill")
 def enqueue_gpu_backfill(task_name: str) -> int:
-    """Enqueue a gpu_job for every image that still needs `task_name` (one
+    """Enqueue a gpu_job for every image that doesn't already have one for
-    INSERT…SELECT, so it scales to a full library). The desktop agent drains the
+    `task_name` (one INSERT…SELECT, so it scales to a full library). The desktop
-    queue over HTTP. Returns the number enqueued.
+    agent drains the queue over HTTP. Returns the number enqueued."""
    'siglip' gates on the RESULT (no concept region yet) rather than on a prior
    job, so it picks up the back-catalogue of images that were CCIP-embedded
    before concept crops existed, and retries images whose concept embed failed —
    without re-touching their figure/CCIP regions."""
    from sqlalchemy import exists, insert, literal
    from sqlalchemy import select as sa_select
-    from ..models import GpuJob, ImageRecord, ImageRegion
+    from ..models import GpuJob, ImageRecord
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
-        if task_name == "siglip":
+        already = exists().where(
-            has_concept = exists().where(
+            GpuJob.image_record_id == ImageRecord.id,
-                ImageRegion.image_record_id == ImageRecord.id,
+            GpuJob.task == task_name,
-                ImageRegion.kind == "concept",
+            GpuJob.status.in_(["pending", "leased", "done"]),
-            )
+        )
-            queued = exists().where(
+        sel = sa_select(
-                GpuJob.image_record_id == ImageRecord.id,
+            ImageRecord.id, literal(task_name), literal("pending")
-                GpuJob.task == "siglip",
+        ).where(~already)
                GpuJob.status.in_(["pending", "leased"]),
            )
            sel = sa_select(
                ImageRecord.id, literal("siglip"), literal("pending")
            ).where(~has_concept).where(~queued)
        else:
            already = exists().where(
                GpuJob.image_record_id == ImageRecord.id,
                GpuJob.task == task_name,
                GpuJob.status.in_(["pending", "leased", "done"]),
            )
            sel = sa_select(
                ImageRecord.id, literal(task_name), literal("pending")
            ).where(~already)
        # RETURNING + count: result.rowcount is unreliable for INSERT…SELECT.
        rows = session.execute(
            insert(GpuJob)
@@ -787,151 +768,3 @@ def enqueue_gpu_backfill(task_name: str) -> int:
        ).fetchall()
        session.commit()
        return len(rows)
@celery.task(name="backend.app.tasks.ml.recover_orphaned_gpu_jobs")
 def recover_orphaned_gpu_jobs() -> int:
    """Reset expired GPU-job leases back to pending — recovers work orphaned by an
    agent that died mid-job (no graceful release). Short beat cadence so orphans
    get picked back up quickly + the queue counts read honestly. Returns the
    number recovered."""
    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.status == "leased", GpuJob.lease_expires_at < now)
            .values(
                status="pending", lease_token=None, leased_at=None,
                lease_expires_at=None, updated_at=now,
            )
        )
        session.commit()
        return res.rowcount or 0
@celery.task(
    name="backend.app.tasks.ml.scheduled_ccip_auto_apply",
    soft_time_limit=1800, time_limit=2100,
 )
 def scheduled_ccip_auto_apply() -> str:
    """Auto-tag confident CCIP character matches (source='ccip_auto') so identity
    tags keep flowing without a button. No-op unless ccip_auto_apply_enabled.
    References come only from single-character images (unambiguous); a tag is
    applied where any figure's best cosine to a character's prototypes clears
    ccip_auto_apply_threshold and it isn't already applied/rejected. Reversible."""
    import numpy as np
    from sqlalchemy import func
    from sqlalchemy import select as sa_select
    from sqlalchemy.dialects.postgresql import insert as pg_insert
    from ..models import ImageRegion, MLSettings, Tag, TagKind, TagSuggestionRejection
    from ..models.tag import image_tag
    fig = ("face", "figure")
    def _l2(m):
        n = np.linalg.norm(m, axis=1, keepdims=True)
        n[n == 0] = 1.0
        return m / n
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        s = session.get(MLSettings, 1)
        if s is None or not s.ccip_auto_apply_enabled:
            return "disabled"
        thr = float(s.ccip_auto_apply_threshold)
        single = (
            sa_select(image_tag.c.image_record_id)
            .join(Tag, Tag.id == image_tag.c.tag_id)
            .where(Tag.kind == TagKind.character)
            .group_by(image_tag.c.image_record_id)
            .having(func.count() == 1)
        )
        ref_rows = session.execute(
            sa_select(image_tag.c.tag_id, ImageRegion.ccip_embedding)
            .select_from(ImageRegion)
            .join(
                image_tag,
                image_tag.c.image_record_id == ImageRegion.image_record_id,
            )
            .join(Tag, Tag.id == image_tag.c.tag_id)
            .where(Tag.kind == TagKind.character)
            .where(ImageRegion.kind.in_(fig))
            .where(ImageRegion.ccip_embedding.is_not(None))
            .where(ImageRegion.image_record_id.in_(single))
        ).all()
        if not ref_rows:
            return "no-references"
        by_char: dict[int, list] = {}
        for tid, vec in ref_rows:
            by_char.setdefault(tid, []).append(vec)
        ref_tags = list(by_char)
        mats = [_l2(np.asarray(by_char[t], dtype=np.float32)) for t in ref_tags]
        allref = np.vstack(mats)                                  # (total, 768)
        seg = np.cumsum([0] + [len(m) for m in mats])[:-1]        # per-char start
        # Per character: images that already carry OR rejected the tag — skip.
        skip = {t: set() for t in ref_tags}
        for t in ref_tags:
            for (iid,) in session.execute(
                sa_select(image_tag.c.image_record_id).where(
                    image_tag.c.tag_id == t
                )
            ):
                skip[t].add(iid)
            for (iid,) in session.execute(
                sa_select(TagSuggestionRejection.image_record_id).where(
                    TagSuggestionRejection.tag_id == t
                )
            ):
                skip[t].add(iid)
        img_ids = list(session.execute(
            sa_select(ImageRegion.image_record_id)
            .where(ImageRegion.kind.in_(fig), ImageRegion.ccip_embedding.is_not(None))
            .distinct()
        ).scalars())
        applied = 0
        chunk_n = 500
        for start in range(0, len(img_ids), chunk_n):
            chunk = img_ids[start:start + chunk_n]
            rows = session.execute(
                sa_select(ImageRegion.image_record_id, ImageRegion.ccip_embedding)
                .where(
                    ImageRegion.image_record_id.in_(chunk),
                    ImageRegion.kind.in_(fig),
                    ImageRegion.ccip_embedding.is_not(None),
                )
            ).all()
            by_img: dict[int, list] = {}
            for iid, vec in rows:
                by_img.setdefault(iid, []).append(vec)
            for iid, vecs in by_img.items():
                q = _l2(np.asarray(vecs, dtype=np.float32))      # (nq, 768)
                colmax = (q @ allref.T).max(axis=0)              # (total,)
                charmax = np.maximum.reduceat(colmax, seg)       # (n_chars,)
                for ci in np.where(charmax >= thr)[0]:
                    t = ref_tags[int(ci)]
                    if iid in skip[t]:
                        continue
                    skip[t].add(iid)
                    session.execute(
                        pg_insert(image_tag)
                        .values(
                            image_record_id=iid, tag_id=t, source="ccip_auto",
                        )
                        .on_conflict_do_nothing()
                    )
                    applied += 1
            session.commit()
        return f"applied={applied}"
@@ -21,14 +21,14 @@
        v-show="store.byCategory[cat] && store.byCategory[cat].length"
        :label="labelFor(cat)" :items="store.byCategory[cat] || []"
        @accept="onAccept" @alias="onAlias" @remove-alias="onRemoveAlias"
-        @dismiss="onDismiss" @undismiss="onUndismiss"
+        @dismiss="store.dismiss" @undismiss="store.undismiss"
      />
      <SuggestionsCategoryGroup
        v-if="store.byCategory.general && store.byCategory.general.length"
        label="General" :items="store.byCategory.general"
        collapsible :default-open="true"
        @accept="onAccept" @alias="onAlias" @remove-alias="onRemoveAlias"
-        @dismiss="onDismiss" @undismiss="onUndismiss"
+        @dismiss="store.dismiss" @undismiss="store.undismiss"
      />
    </div>
@@ -57,15 +57,9 @@ const props = defineProps({
  // so the same panel refreshes the right surface. See TagPanel.
  host: { type: Object, default: null },
 })
-// 'accepted'/'dismissed' let the parent return focus to the tag input after a
+// 'accepted' lets the parent return focus to the tag input after a suggestion is
-// suggestion is accepted OR rejected, so the operator keeps the keyboard flow on
+// applied (operator-asked 2026-06-08).
-// the input without re-clicking (operator-asked 2026-06-08, 2026-06-30).
+const emit = defineEmits(['accepted'])
 const emit = defineEmits(['accepted', 'dismissed'])
 // Reject (✗) / un-reject (↶): apply the store change, then signal the parent to
 // re-focus the tag input — same return-to-input behaviour as accept.
 function onDismiss (s) { store.dismiss(s); emit('dismissed') }
 function onUndismiss (s) { store.undismiss(s); emit('dismissed') }
 const store = useSuggestionsStore()
 const modalStore = useModalStore()
 const host = props.host || modalStore
@@ -28,7 +28,6 @@
      :image-id="host.currentImageId"
      :host="host"
      @accepted="focusTagInput"
      @dismissed="focusTagInput"
    />
    <!-- @after-leave: when either dialog finishes closing (apply OR cancel),
@@ -60,52 +60,6 @@
      Enqueues every image that doesn't have a CCIP embedding yet. Nothing
      processes until the agent is running.
    </p>
    <v-btn
      class="mt-3" color="accent" variant="tonal" rounded="pill" size="small"
      prepend-icon="mdi-crop" :loading="backfillingSiglip" @click="onBackfillSiglip"
    >Queue concept crops (SigLIP)</v-btn>
    <p class="fc-muted text-caption mt-2 mb-0">
      Enqueues every image that doesn't have concept-crop embeddings yet — the
      localized vectors that help small/local tags (glasses, etc.) surface. New
      images get these automatically; this catches the back-catalogue.
    </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">
      <v-slider
        v-model="threshold" :min="0.70" :max="0.95" :step="0.01"
        color="accent" hide-details density="compact" class="flex-grow-1"
        :loading="savingThreshold" @end="onSaveThreshold"
      />
      <span class="fc-q__n" style="font-size:16px">{{ threshold.toFixed(2) }}</span>
    </div>
    <p class="fc-muted text-caption mt-1 mb-0">
      How close a figure must be (CCIP cosine) to suggest a character. Higher =
      stricter — fewer but more confident matches. 0.85 recommended; below ~0.80
      a heavily-tagged character starts matching everything.
    </p>
    <!-- Auto-apply -->
    <div v-if="ml.settings" class="d-flex align-center mt-5" style="gap:12px">
      <v-switch
        v-model="autoApply" color="accent" hide-details density="compact"
        :loading="savingAuto" label="Auto-apply confident matches"
        @update:model-value="onSaveAuto"
      />
      <v-text-field
        v-model.number="autoThreshold" type="number" min="0.80" max="0.99"
        step="0.01" density="compact" hide-details variant="outlined"
        style="max-width:96px" :disabled="!autoApply" label="at"
        @change="onSaveAuto"
      />
    </div>
    <p class="fc-muted text-caption mt-1 mb-0">
      When on, a very-confident character match tags the image on its own (daily,
      reversible) — so identity tags keep flowing without review. Stricter than
      the suggest cut; 0.92 recommended.
    </p>
  </MaintenanceTile>
 </template>
@@ -115,22 +69,14 @@ import { computed, onMounted, onUnmounted, ref } from 'vue'
 import MaintenanceTile from '../common/MaintenanceTile.vue'
 import { useGpuStore } from '../../stores/gpu.js'
 import { useMLStore } from '../../stores/ml.js'
 import { copyText } from '../../utils/clipboard.js'
 const store = useGpuStore()
 const ml = useMLStore()
 const loading = ref(true)
 const tokenValue = ref(null)
 const masked = ref(true)
 const rotating = ref(false)
 const backfilling = ref(false)
 const backfillingSiglip = 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 queue = ref({ pending: 0, leased: 0, done: 0, error: 0 })
 let pollTimer = null
@@ -148,46 +94,9 @@ onMounted(async () => {
  }
  await refreshQueue()
  pollTimer = setInterval(() => { if (!document.hidden) refreshQueue() }, 5000)
  try {
    await ml.loadSettings()
    if (ml.settings?.ccip_match_threshold != null) {
      threshold.value = ml.settings.ccip_match_threshold
    }
    if (ml.settings?.ccip_auto_apply_enabled != null) {
      autoApply.value = ml.settings.ccip_auto_apply_enabled
      autoThreshold.value = ml.settings.ccip_auto_apply_threshold
    }
  } catch { /* non-fatal */ }
 })
 async function onSaveAuto() {
  savingAuto.value = true
  try {
    await ml.patchSettings({
      ccip_auto_apply_enabled: autoApply.value,
      ccip_auto_apply_threshold: autoThreshold.value,
    })
    toast({ text: 'Auto-apply settings saved', type: 'success' })
  } catch (e) {
    toast({ text: `Could not save: ${e.message}`, type: 'error' })
  } finally {
    savingAuto.value = false
  }
 }
 onUnmounted(() => { if (pollTimer) clearInterval(pollTimer) })
 async function onSaveThreshold() {
  savingThreshold.value = true
  try {
    await ml.patchSettings({ ccip_match_threshold: threshold.value })
    toast({ text: `Match strictness set to ${threshold.value.toFixed(2)}`, type: 'success' })
  } catch (e) {
    toast({ text: `Could not save: ${e.message}`, type: 'error' })
  } finally {
    savingThreshold.value = false
  }
 }
 async function refreshQueue() {
  try { queue.value = await store.status() } catch { /* non-fatal */ }
 }
@@ -226,19 +135,6 @@ async function onBackfill() {
    backfilling.value = false
  }
 }
 async function onBackfillSiglip() {
  backfillingSiglip.value = true
  try {
    await store.backfill('siglip')
    toast({ text: 'Queued concept crops — run the agent to process them', type: 'success' })
    await refreshQueue()
  } catch (e) {
    toast({ text: `Could not queue backfill: ${e.message}`, type: 'error' })
  } finally {
    backfillingSiglip.value = false
  }
 }
 </script>
 <style scoped>
@@ -11,7 +11,7 @@ import { toast } from '../utils/toast.js'
 // trail. The store ALSO acts as a TagPanel "host" (current/currentImageId +
 // tag CRUD over the anchor) so the Explore workspace reuses the modal's tag
 // rail verbatim for modal-parity tagging while rabbit-holing.
-const NEIGHBOR_LIMIT = 40   // a wider pool → more variety to browse + jump into
+const NEIGHBOR_LIMIT = 24
 export const useExploreStore = defineStore('explore', () => {
  const api = useApi()
@@ -81,26 +81,16 @@ export const useExploreStore = defineStore('explore', () => {
    return cursor.value > 0 ? breadcrumb.value[cursor.value - 1].id : null
  }
-  // → target: after a ←, walk forward through the already-visited trail
+  // → target: the next already-visited crumb if we'd stepped back, else a
-  // (browser-style). Otherwise jump to a varied neighbour to keep the
+  // RANDOM neighbour to keep the rabbit-hole going. Null if neither exists.
  // rabbit-hole going — null if neither exists.
  function forwardTarget () {
    if (cursor.value >= 0 && cursor.value < breadcrumb.value.length - 1) {
      return breadcrumb.value[cursor.value + 1].id
    }
-    if (!neighbors.value.length) return null
+    if (neighbors.value.length) {
-    // Prefer UNVISITED neighbours so → opens something new instead of landing on
+      return neighbors.value[Math.floor(Math.random() * neighbors.value.length)].id
-    // 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.
+    return null
    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)
    return cands[Math.floor(Math.random() * cands.length)].id
  }
  function reset () {
@@ -96,22 +96,3 @@ async def test_backfill_enqueues_then_is_idempotent(db):
    n = enqueue_gpu_backfill("ccip")     # sync task, own session
    assert n >= 2
    assert enqueue_gpu_backfill("ccip") == 0   # all already pending
@pytest.mark.asyncio
 async def test_release_hands_job_back_to_pending(client, db):
    img = await _img(db, "e" * 64)
    await GpuJobService(db).enqueue(img.id, "ccip")
    await db.commit()
    token = (await (await client.post("/api/gpu/token/rotate")).get_json())["token"]
    hdr = {"Authorization": f"Bearer {token}"}
    j = (await (await client.post(
        "/api/gpu/jobs/lease", json={"agent_id": "a1"}, headers=hdr,
    )).get_json())["jobs"][0]
    resp = await client.post("/api/gpu/jobs/release", json={
        "agent_id": "a1", "job_ids": [j["job_id"]],
    }, headers=hdr)
    assert resp.status_code == 200 and (await resp.get_json())["released"] == 1
    st = await (await client.get("/api/gpu/status")).get_json()
    assert st["pending"] == 1 and st["leased"] == 0
@@ -1,7 +1,6 @@
 """CCIP few-shot character matcher (#114). numpy cosine on stored vectors — no
 model needed, so it runs in CI with synthetic CCIP vectors."""
 import pytest
 from sqlalchemy import select
 from backend.app.models import ImageRecord, ImageRegion, TagKind
 from backend.app.models.tag import image_tag
@@ -87,57 +86,3 @@ async def test_no_figure_vectors_means_no_match(db):
    query = await _img(db, "g" * 64)
    await db.commit()
    assert await match_image(db, query.id) == []
@pytest.mark.asyncio
 async def test_threshold_gates_borderline_match(db):
    # A figure ~0.9 cosine from the reference: matched at 0.85, dropped at 0.95.
    raven = await TagService(db).find_or_create("Raven", TagKind.character)
    ref = await _img(db, "h" * 64)
    await _figure(db, ref.id, _ccip(0))                 # e0
    await _tag_image(db, ref.id, raven.id)
    near = [0.0] * 768
    near[0], near[1] = 0.9, 0.4359                       # |·|=1, cos(e0)=0.9
    query = await _img(db, "i" * 64)
    await _figure(db, query.id, near)
    await db.commit()
    assert any(m["tag_id"] == raven.id for m in await match_image(db, query.id, 0.85))
    assert await match_image(db, query.id, 0.95) == []
@pytest.mark.asyncio
 async def test_multi_character_image_not_used_as_reference(db):
    # A figure on a 2-character image is ambiguous (tag is image-level), so it
    # must NOT seed either character's prototypes — else it'd match both.
    raven = await TagService(db).find_or_create("Raven", TagKind.character)
    daphne = await TagService(db).find_or_create("Daphne", TagKind.character)
    multi = await _img(db, "j" * 64)
    await _figure(db, multi.id, _ccip(0))
    await _tag_image(db, multi.id, raven.id)
    await _tag_image(db, multi.id, daphne.id)
    query = await _img(db, "k" * 64)
    await _figure(db, query.id, _ccip(0))            # identical to the ambiguous figure
    await db.commit()
    assert await match_image(db, query.id) == []     # no clean references → nothing
@pytest.mark.asyncio
 async def test_auto_apply_tags_confident_match(db):
    raven = await TagService(db).find_or_create("Raven", TagKind.character)
    ref = await _img(db, "l" * 64)
    await _figure(db, ref.id, _ccip(0))
    await _tag_image(db, ref.id, raven.id)           # single-character reference
    query = await _img(db, "m" * 64)
    await _figure(db, query.id, _ccip(0))            # identical → cosine 1.0
    await db.commit()
    from backend.app.tasks.ml import scheduled_ccip_auto_apply
    assert "applied=" in scheduled_ccip_auto_apply()   # sync task, own session
    rows = (await db.execute(
        select(image_tag.c.tag_id, image_tag.c.source).where(
            image_tag.c.image_record_id == query.id
        )
    )).all()
    assert (raven.id, "ccip_auto") in [(t, s) for t, s in rows]
@@ -2,9 +2,9 @@
 from datetime import UTC, datetime, timedelta
 import pytest
-from sqlalchemy import func, select
+from sqlalchemy import select
-from backend.app.models import GpuJob, ImageRecord, ImageRegion
+from backend.app.models import GpuJob, ImageRecord
 from backend.app.services.ml.gpu_jobs import GpuJobService
 pytestmark = pytest.mark.integration
@@ -20,43 +20,6 @@ async def _img(db, sha) -> ImageRecord:
    return img
@pytest.mark.asyncio
 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.tasks.ml import enqueue_gpu_backfill
    need = await _img(db, "e1" * 32)          # no concept region → wants one
    have = await _img(db, "e2" * 32)          # already embedded → 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",
    ))
    await db.commit()
    assert enqueue_gpu_backfill("siglip") >= 1
    queued = {
        j.image_record_id for j in (
            await db.execute(select(GpuJob).where(GpuJob.task == "siglip"))
        ).scalars()
    }
    assert need.id in queued
    assert have.id not in queued
    # Idempotent: the now-pending job means a second run doesn't re-enqueue it.
    enqueue_gpu_backfill("siglip")
    n_for_need = (
        await db.execute(
            select(func.count()).select_from(GpuJob).where(
                GpuJob.task == "siglip", GpuJob.image_record_id == need.id
            )
        )
    ).scalar_one()
    assert n_for_need == 1
@pytest.mark.asyncio
 async def test_enqueue_dedupes_same_pair(db):
    img = await _img(db, "a" * 64)
@@ -160,38 +123,3 @@ async def test_fail_requeues_until_cap(db):
    assert await svc.fail("agent-1", job.id, "boom again") is True
    await db.commit()
    assert (await db.get(GpuJob, job.id)).status == "error"
@pytest.mark.asyncio
 async def test_release_returns_to_pending(db):
    img = await _img(db, "01" + "a" * 62)
    svc = GpuJobService(db)
    await svc.enqueue(img.id, "ccip")
    await db.commit()
    job = (await svc.lease("agent-1"))[0]
    await db.commit()
    assert await svc.release("other", [job.id]) == 0      # not this token's lease
    assert await svc.release("agent-1", [job.id]) == 1     # graceful hand-back
    await db.commit()
    fresh = await db.get(GpuJob, job.id)
    assert fresh.status == "pending" and fresh.lease_token is None
@pytest.mark.asyncio
 async def test_recover_orphaned_resets_only_expired(db):
    img1 = await _img(db, "02" + "a" * 62)
    img2 = await _img(db, "03" + "a" * 62)
    svc = GpuJobService(db)
    await svc.enqueue(img1.id, "ccip")
    await svc.enqueue(img2.id, "ccip")
    await db.commit()
    expired, fresh = await svc.lease("dead", batch_size=2)
    # One lease is in the past (orphaned), the other still valid.
    expired.lease_expires_at = datetime.now(UTC) - timedelta(minutes=10)
    await db.commit()
    assert await svc.recover_orphaned() == 1
    await db.commit()
    assert (await db.get(GpuJob, expired.id)).status == "pending"
    assert (await db.get(GpuJob, fresh.id)).status == "leased"   # untouched
@@ -111,40 +111,6 @@ async def test_threshold_override_surfaces_below_cut(db):
    assert any(s.canonical_tag_id == tag.id for s in flooded.by_category["general"])
@pytest.mark.asyncio
 async def test_concept_region_surfaces_via_max_over_bag(db):
    # Max-over-bag: the whole-image vector is orthogonal to the head (scores the
    # 0.5 midpoint, under a 0.7 cut → nothing), but a concept CROP that aligns
    # with the head lifts the max over the bag above the cut. A small/local
    # concept surfaces ONLY because of the crop.
    tag = await TagService(db).find_or_create("glasses", TagKind.general)
    img = await _img(db, "b1" * 32, _emb(5))      # whole-image ⟂ head
    await _head(db, tag.id, slot=0, suggest_threshold=0.7)
    await db.commit()
    # Whole-image alone: sigmoid(0)=0.5 < 0.7 → no suggestion.
    assert not (await SuggestionService(db).for_image(img.id)).by_category.get("general")
    # A concept crop aligned with the head, but stamped with a STALE model
    # version → filtered out of the bag, so still nothing.
    db.add(ImageRegion(
        image_record_id=img.id, kind="concept",
        rx=0.1, ry=0.1, rw=0.3, rh=0.3,
        siglip_embedding=_emb(0), embedding_version="stale-embedder-v0",
    ))
    await db.commit()
    assert not (await SuggestionService(db).for_image(img.id)).by_category.get("general")
    # A matching-version concept crop → max-over-bag lifts it over the cut.
    db.add(ImageRegion(
        image_record_id=img.id, kind="concept",
        rx=0.4, ry=0.4, rw=0.3, rh=0.3,
        siglip_embedding=_emb(0), embedding_version=await _embver(db),
    ))
    await db.commit()
    general = (await SuggestionService(db).for_image(img.id)).by_category["general"]
    assert any(s.canonical_tag_id == tag.id and s.score > 0.7 for s in 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