feat(agent): survive + auto-recover when curator is unreachable

For redeploying curator while away with nobody to restart the agent:

- _process now distinguishes a TRANSPORT error (curator down/redeploying, 5xx,
  401/403/408/409/429, or our lease reclaimed mid-flight) from a genuine job
  fault. On a transport error it hands the job back (best effort) and signals
  the loop to back off — instead of calling fail(), which would burn the job's
  server-side attempt budget (MAX_ATTEMPTS=3) and permanently error good jobs
  across a redeploy. Job-specific 4xx (404 image gone) still fail so they don't
  re-lease forever.
- lease loop retries with capped exponential backoff (poll_idle → 60s) and
  resets on the first successful lease, so a long outage is gentle and recovery
  is automatic within ≤60s of curator returning. Sleeps are interruptible so
  Stop / pool-shrink stays responsive.
- AUTO_START env (default on in compose) resumes the worker on container start,
  so a host reboot / crash-restart (restart: unless-stopped) self-heals with
  nobody at the desktop.
- control UI shows a "waited out" counter + an "curator unreachable, holding
  work" banner so the recovering state reads as recovery, not failure.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa

.forgejo/workflows/build.yml

@@ -329,3 +329,41 @@ jobs:
           file: Dockerfile.ml
           push: true
           tags: ${{ steps.tag.outputs.tags }}
+
+  # The desktop GPU agent (#114) — published so the operator pulls + runs it on
+  # the GPU machine instead of building locally. Independent of web/ml (its own
+  # CUDA + onnxruntime-gpu image, context = agent/). Same tag cadence.
+  build-agent:
+    runs-on: python-ci
+    container:
+      image: git.fabledsword.com/bvandeusen/ci-python:3.14
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Determine tag
+        id: tag
+        run: |
+          SHORT_SHA=$(printf '%s' "$GITHUB_SHA" | cut -c1-7)
+          if [ "${GITHUB_REF#refs/tags/}" != "${GITHUB_REF}" ]; then
+            TAG_NAME="${GITHUB_REF#refs/tags/}"
+            echo "tags=git.fabledsword.com/bvandeusen/fabledcurator-agent:${TAG_NAME}" >> "$GITHUB_OUTPUT"
+          elif [ "${GITHUB_REF##*/}" = "main" ]; then
+            echo "tags=git.fabledsword.com/bvandeusen/fabledcurator-agent:main,git.fabledsword.com/bvandeusen/fabledcurator-agent:latest,git.fabledsword.com/bvandeusen/fabledcurator-agent:c-${SHORT_SHA}" >> "$GITHUB_OUTPUT"
+          else
+            echo "tags=git.fabledsword.com/bvandeusen/fabledcurator-agent:dev" >> "$GITHUB_OUTPUT"
+          fi
+
+      - name: Login to Forgejo registry
+        uses: docker/login-action@v3
+        with:
+          registry: git.fabledsword.com
+          username: ${{ github.actor }}
+          password: ${{ secrets.RELEASE_TOKEN }}
+
+      - name: Build and push agent image
+        uses: docker/build-push-action@v5
+        with:
+          context: agent
+          file: agent/Dockerfile
+          push: true
+          tags: ${{ steps.tag.outputs.tags }}

agent/Dockerfile

@@ -1,6 +1,8 @@
 # FabledCurator GPU agent — runs on the desktop with the GPU.
-# CUDA runtime so onnxruntime-gpu can use the card; ffmpeg for video frames.
+# CUDA + cuDNN runtime so onnxruntime-gpu can use the card (it needs cuDNN 9 —
-FROM nvidia/cuda:12.4.1-runtime-ubuntu22.04
+# 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
 
 ENV DEBIAN_FRONTEND=noninteractive PYTHONUNBUFFERED=1
 RUN apt-get update \
@@ -8,11 +10,16 @@ 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 caches downloaded ONNX models here; mount a volume to persist them.
+# imgutils ONNX models + the transformers SigLIP weights both cache here; mount
+# a volume to persist them across restarts (the SigLIP download is ~3.5 GB once).
 ENV HF_HOME=/models
 EXPOSE 8770
 

agent/README.md

@@ -7,14 +7,25 @@ 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.
 
-## 2. Build
+## 2. Pull (CI publishes it alongside the web/ml images)
 ```sh
-cd agent
+docker pull git.fabledsword.com/bvandeusen/fabledcurator-agent:latest
-docker build -t fc-gpu-agent .
 ```
+> Local build for development instead: `docker build -t fc-gpu-agent agent/`
 
 ## 3. Run (on the machine with the GPU)
 ```sh
@@ -22,7 +33,7 @@ docker run --rm --gpus all -p 8770:8770 \
   -e FC_URL=http://curator.traefik.internal \
   -e FC_TOKEN=<paste-the-token> \
   -v fc-agent-models:/models \
-  fc-gpu-agent
+  git.fabledsword.com/bvandeusen/fabledcurator-agent:latest
 ```
 Then open <http://localhost:8770> — the control page. Click **Start** to begin
 draining the queue; **Pause**/**Stop** to yield the GPU. The `-v fc-agent-models`

agent/docker-compose.yml

@@ -0,0 +1,53 @@
+# FabledCurator GPU agent — desktop run via docker compose.
+#
+# Usage:
+#   1. Generate a token: FC → Settings → Tagging → GPU agent → Generate token.
+#   2. Create a .env next to this file:
+#        FC_URL=http://curator.traefik.internal
+#        FC_TOKEN=<paste-the-token>
+#        # optional: CCIP_MODEL=ccip-caformer_b36-24   (the F1-0.94 variant)
+#   3. docker compose up -d     (pulls the published image)
+#   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:
+  fc-gpu-agent:
+    image: git.fabledsword.com/bvandeusen/fabledcurator-agent:latest
+    pull_policy: always
+    ports:
+      - "8770:8770"
+    environment:
+      FC_URL: ${FC_URL:-http://curator.traefik.internal}
+      FC_TOKEN: ${FC_TOKEN:?set FC_TOKEN in .env (FC → GPU agent → Generate token)}
+      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
+    restart: unless-stopped
+    deploy:
+      resources:
+        reservations:
+          devices:
+            - driver: nvidia
+              count: all
+              capabilities: [gpu]
+
+volumes:
+  fc-agent-models:

agent/fc_agent/app.py

@@ -1,13 +1,14 @@
 """FastAPI control surface for the agent (served on localhost).
 
-Start / pause / resume / stop the worker, set nothing else here (config is env),
+Start / stop the worker pool, tune the worker count live (trades desktop
-and watch progress + the server-side queue. The container exposes this on a
+responsiveness for throughput), and watch GPU load + progress + the server-side
-localhost port; stopping the worker frees the GPU.
+queue. Config is env-seeded; the worker count is adjustable here on the fly.
 """
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
 from fastapi.responses import HTMLResponse, JSONResponse
 
 from .config import Config
+from .gpu import read_gpu
 from .worker import Worker
 
 cfg = Config.from_env()
@@ -15,6 +16,16 @@ 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
@@ -26,29 +37,25 @@ def start():
     return JSONResponse(worker.status())
 
 
-@app.post("/pause")
-def pause():
-    worker.pause()
-    return JSONResponse(worker.status())
-
-
-@app.post("/resume")
-def resume():
-    worker.resume()
-    return JSONResponse(worker.status())
-
-
 @app.post("/stop")
 def stop():
     worker.stop()
     return JSONResponse(worker.status())
 
 
+@app.post("/concurrency")
+async def concurrency(request: Request):
+    body = await request.json()
+    worker.set_concurrency(int(body.get("value", 1)))
+    return JSONResponse(worker.status())
+
+
 @app.get("/status")
 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:
@@ -59,35 +66,68 @@ 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:640px;background:#14171a;color:#e8e8e8}
+ 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}.pause{background:#b26a00}
+ margin-right:.5rem;cursor:pointer;color:#fff} .start{background:#2e7d32}.stop{background:#b3261e}
- .stop{background:#b3261e} .stat{display:inline-block;margin-right:1.5rem}
+ .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{margin-top:1rem;color:#9aa}
+ .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}
 </style></head><body>
 <h1>FabledCurator GPU agent</h1>
 <p>FC: <code id=fc>—</code> · token <code id=cfg>—</code></p>
-<p>
+<div class=row>
  <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>
-</p>
+</div>
-<p>
+<div class=row>
- <span class=stat><span class=n id=state>idle</span><br>state</span>
+ 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=cur>—</span><br>current image</span>
+ <span class=stat><span class=n id=wait>0</span><br>waited out</span>
-</p>
+</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>
+ 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()
-  state.textContent=s.state; done.textContent=s.processed; err.textContent=s.errors
+  CAP=s.max_concurrency||8; capn.textContent=CAP
-  cur.textContent=s.current??'—'; fc.textContent=s.fc_url
+  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'
+  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)

agent/fc_agent/client.py

@@ -4,6 +4,7 @@ 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:
@@ -12,6 +13,11 @@ 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(
@@ -54,6 +60,19 @@ 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)

agent/fc_agent/config.py

@@ -8,10 +8,16 @@ 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 leased per round (concurrency is still 1)
+    batch_size: int      # jobs a worker leases per round
+    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":
@@ -20,7 +26,11 @@ 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"),
         )

agent/fc_agent/embedder.py

@@ -0,0 +1,69 @@
+"""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()

agent/fc_agent/gpu.py

@@ -0,0 +1,30 @@
+"""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

agent/fc_agent/media.py

@@ -13,8 +13,23 @@ 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 Image.open(io.BytesIO(data)).convert("RGB")
+    return to_rgb(Image.open(io.BytesIO(data)))
 
 
 def sample_frames(
@@ -44,5 +59,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), im.convert("RGB")))
+                out.append((round(i * interval, 2), to_rgb(im)))
         return out

agent/fc_agent/worker.py

@@ -1,92 +1,188 @@
-"""The lease → fetch → detect+embed → submit loop, with start/pause/stop control.
+"""The lease → fetch → detect+embed → submit loop, run by a pool of worker
+slots whose count is tunable live from the UI.
 
-Concurrency is 1 (one image at a time) so the GPU footprint stays small and a
+Each slot is an independent loop (its own leases; the server's SKIP-LOCKED lease
-stop frees the card promptly. Stop halts leasing + finishes the current item;
+keeps them from colliding). More slots = more GPU load + throughput; the model is
-unprocessed leases expire and the server re-queues them — nothing is lost.
+loaded once and shared, so slots add concurrent inference, not N× model VRAM.
+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._thread: threading.Thread | None = None
+        self._running = False
+        self._target = max(1, min(MAX_CONCURRENCY, cfg.concurrency))
+        self._slots: list[_Slot] = []
         self.processed = 0
         self.errors = 0
-        self.current = None
+        self.transient = 0        # jobs handed back due to a server outage (NOT
+        #                           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:
-            if self._state in ("running", "paused"):
+            self._running = True
-                self._state = "running"
+            self._reconcile_locked()
-                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:
-            if self._state in ("running", "paused"):
+            self._running = False
-                self._state = "stopping"
+            slots, self._slots = self._slots, []
+        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:
-            state = self._state
             return {
-            "state": state, "processed": self.processed,
+                "state": "running" if self._running else "stopped",
-            "errors": self.errors, "current": self.current,
+                "concurrency": self._target,
+                "max_concurrency": MAX_CONCURRENCY,
+                "workers": len(self._slots),
+                "active": self._active,
+                "processed": self.processed,
+                "errors": self.errors,
+                "transient": self.transient,
             }
 
-    # --- loop --------------------------------------------------------------
+    def _bump(self, *, processed=0, errors=0, active=0, transient=0):
-    def _run(self):
-        while True:
         with self._lock:
-                st = self._state
+            self.processed += processed
-            if st == "stopping":
+            self.errors += errors
-                break
+            self.transient += transient
-            if st == "paused":
+            self._active += active
-                time.sleep(1)
+
-                continue
+    # --- per-slot loop -----------------------------------------------------
+    def _loop(self, slot: _Slot):
+        backoff = self.cfg.poll_idle_seconds
+        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:
-                time.sleep(self.cfg.poll_idle_seconds)
+                # curator unreachable (redeploy, network drop): wait it out with
+                # 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:
-                time.sleep(self.cfg.poll_idle_seconds)
+                self._interruptible_sleep(slot, self.cfg.poll_idle_seconds)
                 continue
-            ids = [j["job_id"] for j in jobs]
+            slot.inflight = [j["job_id"] for j in jobs]
             for job in jobs:
-                with self._lock:
+                if slot.stop.is_set() or not self._running:
-                    if self._state == "stopping":
                     break
-                self._process(job)
+                ok = self._process(job)
-                self.client.heartbeat(ids)   # keep the rest of the batch alive
+                slot.inflight = [i for i in slot.inflight if i != job["job_id"]]
-        with self._lock:
+                if not ok:
-            self._state = "idle"
+                    # Server went away mid-batch: hand the rest back (best effort)
+                    # and back off instead of hammering a recovering server or
+                    # 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 _process(self, job: dict):
+    def _interruptible_sleep(self, slot: _Slot, seconds: float):
-        self.current = job.get("image_id")
+        """Sleep, but wake immediately if the slot is told to stop — so a Stop or
+        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", "")):
@@ -97,8 +193,31 @@ 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 = []
-            ev = self.cfg.ccip_model or "ccip-default"
+            ccip_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)
@@ -108,20 +227,48 @@ class Worker:
                     crop = crop_region(frame, bbox)
                     if crop is None:
                         continue
-                    vec = models.ccip_vector(crop, self.cfg.ccip_model or None)
+                    if want_ccip:
                         regions.append({
                             "kind": "figure",
                             "bbox": list(bbox),
                             "frame_time": t,
                             "score": score,
-                        "ccip_embedding": vec,
+                            "ccip_embedding": models.ccip_vector(
-                        "embedding_version": ev,
+                                crop, self.cfg.ccip_model or None
+                            ),
+                            "embedding_version": ccip_ev,
                             "detector_version": dv,
                         })
-            self.client.submit(job["job_id"], regions, ["figure", "face"])
+                    if want_siglip:
-            self.processed += 1
+                        regions.append({
-        except Exception as exc:   # noqa: BLE001 — report + move on
+                            "kind": "concept",
-            self.errors += 1
+                            "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.current = None
+            self._bump(active=-1)

agent/requirements.txt

@@ -3,6 +3,10 @@ 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]

alembic/versions/0063_ccip_match_threshold.py

@@ -0,0 +1,33 @@
+"""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")

alembic/versions/0064_ccip_auto_apply.py

@@ -0,0 +1,42 @@
+"""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")

backend/app/api/ccip.py

@@ -37,6 +37,15 @@ 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 = (
@@ -62,14 +71,23 @@ 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,
     })
 
 

backend/app/api/gpu.py

@@ -17,6 +17,7 @@ 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
 
@@ -137,6 +138,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,
+                "embed_version": ml.embedder_model_version,
             })
     return jsonify({"jobs": out})
 
@@ -196,3 +203,18 @@ 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})

backend/app/api/ml_admin.py

@@ -21,6 +21,9 @@ _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",
 )
 
 
@@ -48,6 +51,9 @@ 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,
         }
     )
 
@@ -115,6 +121,10 @@ 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
 
 

backend/app/celery_app.py

@@ -117,6 +117,24 @@ 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,

backend/app/models/ml_settings.py

@@ -86,6 +86,21 @@ 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"
     )

backend/app/services/ml/ccip.py

@@ -13,28 +13,82 @@ 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 select
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from ...models import ImageRegion, Tag, TagKind
+from ...models import ImageRegion, MLSettings, Tag, TagKind
 from ...models.tag import image_tag
 
-# Cosine-similarity floor to call a figure the same character. Conservative
+# Cosine-similarity floor to call a figure the same character. The live setting
-# default; tune from real matches (CCIP same-char clusters tightly).
+# (ml_settings.ccip_match_threshold) drives it; this is only the fallback when no
-DEFAULT_SIM_THRESHOLD = 0.75
+# threshold is supplied AND no settings row exists.
+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 images that carry that character tag (the operator's examples).
+    embeddings on UNAMBIGUOUS (single-character) images carrying that tag.
-    Multi-prototype — several vectors per character."""
+    Multi-prototype — several vectors per character. Cached on a cheap signature."""
+    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)
@@ -47,11 +101,13 @@ 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
 
 
@@ -68,14 +124,18 @@ async def _tag_names(session: AsyncSession, tag_ids: list[int]) -> dict[int, str
 
 
 async def match_image(
-    session: AsyncSession, image_id: int, threshold: float = DEFAULT_SIM_THRESHOLD
+    session: AsyncSession, image_id: int, threshold: float | None = None
 ) -> 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."""
+    CCIP vectors or no character references exist yet. threshold defaults to the
+    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(

backend/app/services/ml/gpu_jobs.py

@@ -1,10 +1,13 @@
-"""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 (or a
+OR expired-leased jobs with FOR UPDATE SKIP LOCKED, so concurrent agents/workers
-retry after an agent died) never grab the same job and the queue self-heals
+never grab the same job. Orphan recovery is three-layered: a graceful agent stop
-without a separate recovery sweep. Result-writing (regions) is done by the API
+calls release() to hand its in-flight jobs back instantly; a hard crash is caught
-handler via RegionService; complete() just closes the job.
+by recover_orphaned() (a 60s beat sweep) which resets expired leases to pending;
+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
@@ -14,7 +17,10 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from ...models import GpuJob
 
-DEFAULT_LEASE_TTL = 300     # seconds an agent holds a job before it can be re-leased
+# Lease window. Kept comfortably above any single job (a capped-frame video embed
+# 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
 
@@ -132,3 +138,40 @@ 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

backend/app/services/ml/heads.py

@@ -29,6 +29,7 @@ from ...models import (
     HeadAutoApplyRun,
     HeadTrainingRun,
     ImageRecord,
+    ImageRegion,
     MLSettings,
     Tag,
     TagHead,
@@ -296,7 +297,14 @@ 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)
@@ -306,11 +314,26 @@ 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)
+
-    n = float(np.linalg.norm(x)) or 1.0
+    bag = [np.asarray(img.siglip_embedding, dtype=np.float32)]
-    xn = x / n
+    region_vecs = (
-    z = heads["W"] @ xn + heads["b"]
+        await session.execute(
-    probs = 1.0 / (1.0 + np.exp(-z))
+            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)
+        )
+    ).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]

backend/app/tasks/ml.py

@@ -742,16 +742,35 @@ 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 doesn't already have one for
+    """Enqueue a gpu_job for every image that still needs `task_name` (one
-    `task_name` (one INSERT…SELECT, so it scales to a full library). The desktop
+    INSERT…SELECT, so it scales to a full library). The desktop agent drains the
-    agent drains the queue over HTTP. Returns the number enqueued."""
+    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
+    from ..models import GpuJob, ImageRecord, ImageRegion
 
     SessionLocal = _sync_session_factory()
     with SessionLocal() as session:
+        if task_name == "siglip":
+            has_concept = exists().where(
+                ImageRegion.image_record_id == ImageRecord.id,
+                ImageRegion.kind == "concept",
+            )
+            queued = exists().where(
+                GpuJob.image_record_id == ImageRecord.id,
+                GpuJob.task == "siglip",
+                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,
@@ -768,3 +787,151 @@ 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}"

frontend/src/components/modal/SuggestionsPanel.vue

@@ -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="store.dismiss" @undismiss="store.undismiss"
+        @dismiss="onDismiss" @undismiss="onUndismiss"
       />
       <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="store.dismiss" @undismiss="store.undismiss"
+        @dismiss="onDismiss" @undismiss="onUndismiss"
       />
     </div>
 
@@ -57,9 +57,15 @@ const props = defineProps({
   // so the same panel refreshes the right surface. See TagPanel.
   host: { type: Object, default: null },
 })
-// 'accepted' lets the parent return focus to the tag input after a suggestion is
+// 'accepted'/'dismissed' let the parent return focus to the tag input after a
-// applied (operator-asked 2026-06-08).
+// suggestion is accepted OR rejected, so the operator keeps the keyboard flow on
-const emit = defineEmits(['accepted'])
+// the input without re-clicking (operator-asked 2026-06-08, 2026-06-30).
+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

frontend/src/components/modal/TagPanel.vue

@@ -28,6 +28,7 @@
       :image-id="host.currentImageId"
       :host="host"
       @accepted="focusTagInput"
+      @dismissed="focusTagInput"
     />
 
     <!-- @after-leave: when either dialog finishes closing (apply OR cancel),

frontend/src/components/settings/GpuAgentCard.vue

@@ -60,6 +60,52 @@
       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>
 
@@ -69,14 +115,22 @@ 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
 
@@ -94,9 +148,46 @@ 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 */ }
 }
@@ -135,6 +226,19 @@ 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>

frontend/src/stores/explore.js

@@ -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 = 24
+const NEIGHBOR_LIMIT = 40   // a wider pool → more variety to browse + jump into
 
 export const useExploreStore = defineStore('explore', () => {
   const api = useApi()
@@ -81,16 +81,26 @@ export const useExploreStore = defineStore('explore', () => {
     return cursor.value > 0 ? breadcrumb.value[cursor.value - 1].id : null
   }
 
-  // → target: the next already-visited crumb if we'd stepped back, else a
+  // → target: after a ←, walk forward through the already-visited trail
-  // RANDOM neighbour to keep the rabbit-hole going. Null if neither exists.
+  // (browser-style). Otherwise jump to a varied neighbour to keep the
+  // 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) {
+    if (!neighbors.value.length) return null
-      return neighbors.value[Math.floor(Math.random() * neighbors.value.length)].id
+    // Prefer UNVISITED neighbours so → opens something new instead of landing on
-    }
+    // a crumb (which snaps the cursor back into the trail — the "loops back"
-    return null
+    // 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)
+    return cands[Math.floor(Math.random() * cands.length)].id
   }
 
   function reset () {

tests/test_api_gpu.py

@@ -96,3 +96,22 @@ 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

tests/test_ccip.py

@@ -1,6 +1,7 @@
 """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
@@ -86,3 +87,57 @@ 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]

tests/test_gpu_jobs.py

@@ -2,9 +2,9 @@
 from datetime import UTC, datetime, timedelta
 
 import pytest
-from sqlalchemy import select
+from sqlalchemy import func, select
 
-from backend.app.models import GpuJob, ImageRecord
+from backend.app.models import GpuJob, ImageRecord, ImageRegion
 from backend.app.services.ml.gpu_jobs import GpuJobService
 
 pytestmark = pytest.mark.integration
@@ -20,6 +20,43 @@ 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)
@@ -123,3 +160,38 @@ 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

tests/test_ml_suggestions.py

@@ -111,6 +111,40 @@ 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