Compare commits

...

7 Commits

Author SHA1 Message Date
bvandeusen 55fa4656ff 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
2026-06-30 08:33:33 -04:00
bvandeusen c6f38b0dac feat(tagging): SigLIP concept crops + max-over-bag scoring (#114)
Lift recall on small/local concepts (glasses, cum, stomach-bulge, xray,
lactation) that the whole-image SigLIP vector washes out: the GPU agent now
embeds figure crops with SigLIP too, stored as kind='concept' regions, and the
suggestion rail scores each image as a BAG (whole-image + every concept crop),
taking each head's MAX over the bag. The whole-image vector is always in the
bag, so this can never score lower than before.

Model-agnostic by construction: the server ANNOUNCES the embedding model
(HF name + version) in the lease, so the agent loads whatever the heads were
trained in and stays in lock-step — a model swap is a server setting + a
re-embed migration, never an agent change.

- agent: model-agnostic CropEmbedder (torch/transformers get_image_features,
  fp16 on CUDA, inference-locked); worker branches on job.task — 'ccip' emits
  figure(CCIP)+concept(SigLIP) in one pass, 'siglip' emits concept-only so the
  back-catalogue backfill never churns figure/CCIP regions; torch cu124 +
  transformers in the image.
- server: lease announces embed_model_name/embed_version; score_image is
  max-over-bag (version-filtered region embeddings); enqueue_gpu_backfill
  'siglip' gates on a missing concept region (drains the back-catalogue,
  retries failures, no double-enqueue); daily siglip-backfill beat; UI button;
  /api/ccip/overview reports images_with_concept_siglip.
- v1 scope: suggestion rail only — auto-apply stays whole-image (conservative;
  heads' thresholds were calibrated on whole-image). Bulk-apply bag = follow-up.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa
2026-06-30 08:17:47 -04:00
bvandeusen b91a230f12 feat(ccip): automation + reference quality — keep identity flowing hands-free (#114)
CI / lint (push) Successful in 2s
CI / frontend-build (push) Successful in 19s
CI / backend-lint-and-test (push) Successful in 26s
CI / integration (push) Successful in 3m32s
Works through the optional CCIP ideas + the "keep moving even if I forget" ask:

AUTOMATION (no button needed):
- Hourly beat auto-enqueues CCIP backfill — new images get embedded (and errored
  ones retried) on their own; the queue never goes idle waiting for a click.
- CCIP auto-apply: a daily sweep tags confident matches (source='ccip_auto') so
  identity tags keep flowing. ON by default (opt-out, like head auto-apply);
  ml_settings.ccip_auto_apply_enabled + _threshold (0.92, above the suggest cut),
  migration 0064. Vectorized (one matmul + reduceat per image), reversible, skips
  already-applied/rejected. Switch + threshold in the GPU agent card; GET/PATCH
  /api/ml/settings; auto_applied count in /api/ccip/overview.

REFERENCE QUALITY (the over-fire root cause):
- character_references now draws ONLY from single-character images — on a
  multi-character image the tag is image-level, so every figure would otherwise
  pollute each character's prototypes (a 2-char image tagged 'Velma' made
  Daphne's figure a Velma reference). This is the contamination behind residual
  over-firing.
- Cached on a cheap signature (char-tag count + ccip-region count/max-id) so the
  reference load isn't redone on every modal open.

Tests: multi-character image not used as a reference; auto-apply tags a confident
match as ccip_auto.

NEXT (not done, confirmed): comic-panel cropping + SigLIP concept crops ("spot
interesting content").

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa
2026-06-29 22:25:40 -04:00
bvandeusen 74b7ceaf47 fix(tags): return focus to the tag input after reject/un-reject too
CI / lint (push) Successful in 2s
CI / frontend-build (push) Successful in 19s
CI / backend-lint-and-test (push) Successful in 27s
CI / integration (push) Successful in 3m25s
Accept already re-focused the tag input (so you keep typing without re-clicking);
reject (✗) and un-reject (↶) went straight to the store and skipped it. Route
them through onDismiss/onUndismiss which emit 'dismissed', and wire that to
focusTagInput in TagPanel — same return-to-input behaviour as accept. TagPanel is
shared, so this covers both the image modal and the Explore workspace. The
field's mobile-focus guard is preserved.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa
2026-06-29 21:06:34 -04:00
bvandeusen 301f2de989 fix(explore): variance + no loop-back on → navigation (#94)
CI / lint (push) Successful in 2s
CI / frontend-build (push) Successful in 21s
CI / backend-lint-and-test (push) Successful in 26s
CI / integration (push) Successful in 3m26s
Two reports: → sometimes "loops back", and the walk gets stuck on near-identical
images. Cause: forwardTarget picked a uniformly-random neighbour from the 24
NEAREST, so it (a) often landed on an image already in the trail — which snaps
the cursor back into history and makes → bounce between visited nodes — and (b)
only ever offered near-duplicates.

forwardTarget now: excludes already-visited neighbours (→ opens something new,
no snap-back), and skips the closest third of the (similarity-sorted) pool so the
jump favours the more-varied remainder instead of lookalikes. Neighbour pool
widened 24→40 for more variety to browse + jump into. The post-← browser-forward
walk through visited crumbs is unchanged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa
2026-06-29 20:44:16 -04:00
bvandeusen 625336b6b4 feat(ccip): tunable match threshold, default 0.85 (#114)
CI / lint (push) Successful in 3s
CI / frontend-build (push) Successful in 20s
CI / backend-lint-and-test (push) Successful in 25s
CI / integration (push) Successful in 3m28s
Live data showed the v1 flat 0.75 cosine over-fired — ~64% of matched images got
3-10 character guesses dominated by the most-referenced characters (a 27-ref
character clears a low bar on many images). A sweep showed 0.85 collapses the
noise (noisy multi-matches 47→3) while keeping the confident single-character
matches.

- ml_settings.ccip_match_threshold (migration 0063, default 0.85); match_image
  reads it (override still accepted). DEFAULT_SIM_THRESHOLD fallback 0.75→0.85.
- Exposed in GET/PATCH /api/ml/settings (validated 0.5–0.999).
- Slider in the GPU agent card ("Character-match strictness") — tune live, no
  redeploy, same observe-and-tune loop as auto-apply.

Test: a ~0.9-cosine figure matches at 0.85, dropped at 0.95.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa
2026-06-29 20:41:09 -04:00
bvandeusen b7fd69815e feat(agent): raise worker cap to 32 + size the HTTP pool for it (#114)
CI / lint (push) Successful in 3s
CI / frontend-build (push) Successful in 19s
CI / backend-lint-and-test (push) Successful in 28s
CI / integration (push) Successful in 3m35s
At 8 workers the GPU sat at ~5% util / <5GB VRAM — the pipeline is I/O-bound
(downloading + decoding images over HTTP), so the GPU starves until many workers
overlap that I/O. Raise MAX_CONCURRENCY 8→32 and make the UI worker control a
number input (reaching 32 by ±1 was tedious); the cap is reported via /status so
the UI clamps to it. Also size the shared requests pool (pool_maxsize=64) — the
default 10 would have throttled 32 workers + spammed "connection pool is full".

Verified by running; watch GPU util/VRAM climb as you dial up.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa
2026-06-29 19:41:52 -04:00
25 changed files with 927 additions and 67 deletions
+6 -1
View File
@@ -10,11 +10,16 @@ RUN apt-get update \
&& rm -rf /var/lib/apt/lists/* && rm -rf /var/lib/apt/lists/*
WORKDIR /app 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 . COPY requirements.txt .
RUN pip3 install --no-cache-dir -r requirements.txt RUN pip3 install --no-cache-dir -r requirements.txt
COPY fc_agent ./fc_agent 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 ENV HF_HOME=/models
EXPOSE 8770 EXPOSE 8770
+13
View File
@@ -10,6 +10,13 @@
# 4. Open http://localhost:8770 → Start. Pause/Stop hands the GPU back. # 4. Open http://localhost:8770 → Start. Pause/Stop hands the GPU back.
# docker compose down to stop the container entirely. # 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. # Needs the NVIDIA Container Toolkit installed on the host for --gpus.
services: services:
@@ -24,6 +31,12 @@ services:
CCIP_MODEL: ${CCIP_MODEL:-} CCIP_MODEL: ${CCIP_MODEL:-}
DETECTOR_LEVEL: ${DETECTOR_LEVEL:-m} DETECTOR_LEVEL: ${DETECTOR_LEVEL:-m}
BATCH_SIZE: ${BATCH_SIZE:-4} 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: volumes:
# Persist the downloaded ONNX models so restarts are fast. # Persist the downloaded ONNX models so restarts are fast.
- fc-agent-models:/models - fc-agent-models:/models
+29 -5
View File
@@ -16,6 +16,16 @@ worker = Worker(cfg)
app = FastAPI(title="FabledCurator GPU agent") 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) @app.get("/", response_class=HTMLResponse)
def index() -> str: def index() -> str:
return _PAGE return _PAGE
@@ -75,30 +85,44 @@ _PAGE = """<!doctype html><html><head><meta charset=utf-8>
<div class=row> <div class=row>
workers workers
<button class=step onclick=setc(-1)></button> <button class=step onclick=setc(-1)></button>
<b id=conc style=margin:0+.5rem>1</b> <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> <button class=step onclick=setc(1)>+</button>
<span class=cap style=color:#9aa>(more = faster + more GPU)</span> <span class=cap style=color:#9aa>(more = overlap I/O, fill the GPU) max <b id=capn>8</b></span>
</div> </div>
<div class=row> <div class=row>
<span class=stat><span class=n id=state>stopped</span><br>state</span> <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=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=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=err>0</span><br>errors</span>
<span class=stat><span class=n id=wait>0</span><br>waited out</span>
</div>
<div id=banner style="display:none;margin:.6rem 0;padding:.5rem .8rem;border-radius:6px;background:#5a4a17;color:#ffe28a">
curator unreachable — holding work + retrying, will resume on its own (no restart needed)
</div> </div>
<div class=gpu id=gpu>GPU — …</div> <div class=gpu id=gpu>GPU — …</div>
<div class=bar><i id=gpubar style=width:0%></i></div> <div class=bar><i id=gpubar style=width:0%></i></div>
<div class=q id=queue></div> <div class=q id=queue></div>
<script> <script>
let CAP=8
async function act(p){await fetch('/'+p,{method:'POST'});refresh()} async function act(p){await fetch('/'+p,{method:'POST'});refresh()}
async function setc(d){ function setc(d){ setv((parseInt(conc.value||'1'))+d) }
const v=Math.max(1,Math.min(8,parseInt(conc.textContent||'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'}, await fetch('/concurrency',{method:'POST',headers:{'Content-Type':'application/json'},
body:JSON.stringify({value:v})});refresh() body:JSON.stringify({value:v})});refresh()
} }
async function refresh(){ async function refresh(){
const s=await (await fetch('/status')).json() const s=await (await fetch('/status')).json()
CAP=s.max_concurrency||8; capn.textContent=CAP
state.textContent=s.state; active.textContent=s.active; done.textContent=s.processed state.textContent=s.state; active.textContent=s.active; done.textContent=s.processed
err.textContent=s.errors; conc.textContent=s.concurrency; 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' cfg.textContent=s.configured?'set':'MISSING'
if(s.gpu){ 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` 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`
+6
View File
@@ -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. bytes, all over HTTP with the bearer token. No DB/Redis.
""" """
import requests import requests
from requests.adapters import HTTPAdapter
class FcClient: class FcClient:
@@ -12,6 +13,11 @@ class FcClient:
self.agent_id = agent_id self.agent_id = agent_id
self.s = requests.Session() self.s = requests.Session()
self.s.headers["Authorization"] = f"Bearer {token}" 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]: def lease(self, batch_size: int) -> list[dict]:
r = self.s.post( r = self.s.post(
+8
View File
@@ -13,6 +13,11 @@ class Config:
ccip_model: str # imgutils CCIP model name ("" → imgutils default) ccip_model: str # imgutils CCIP model name ("" → imgutils default)
detector_level: str # imgutils person-detector level: n|s|m|x detector_level: str # imgutils person-detector level: n|s|m|x
poll_idle_seconds: float # wait between empty leases 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 @classmethod
def from_env(cls) -> "Config": def from_env(cls) -> "Config":
@@ -25,4 +30,7 @@ class Config:
ccip_model=os.environ.get("CCIP_MODEL", ""), ccip_model=os.environ.get("CCIP_MODEL", ""),
detector_level=os.environ.get("DETECTOR_LEVEL", "m"), detector_level=os.environ.get("DETECTOR_LEVEL", "m"),
poll_idle_seconds=float(os.environ.get("POLL_IDLE_SECONDS", "10")), 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"),
) )
+69
View File
@@ -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()
+142 -20
View File
@@ -10,14 +10,42 @@ 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. orphaned work is re-picked at once rather than waiting out the lease.
""" """
import threading import threading
import time
import requests
from . import media, models from . import media, models
from .client import FcClient from .client import FcClient
from .config import Config from .config import Config
from .crops import crop_region from .crops import crop_region
MAX_CONCURRENCY = 8 # 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: class _Slot:
@@ -40,7 +68,13 @@ class Worker:
self._slots: list[_Slot] = [] self._slots: list[_Slot] = []
self.processed = 0 self.processed = 0
self.errors = 0 self.errors = 0
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 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 ----------------------------------------------------------- # --- control -----------------------------------------------------------
def start(self): def start(self):
@@ -74,35 +108,53 @@ class Worker:
return { return {
"state": "running" if self._running else "stopped", "state": "running" if self._running else "stopped",
"concurrency": self._target, "concurrency": self._target,
"max_concurrency": MAX_CONCURRENCY,
"workers": len(self._slots), "workers": len(self._slots),
"active": self._active, "active": self._active,
"processed": self.processed, "processed": self.processed,
"errors": self.errors, "errors": self.errors,
"transient": self.transient,
} }
def _bump(self, *, processed=0, errors=0, active=0): def _bump(self, *, processed=0, errors=0, active=0, transient=0):
with self._lock: with self._lock:
self.processed += processed self.processed += processed
self.errors += errors self.errors += errors
self.transient += transient
self._active += active self._active += active
# --- per-slot loop ----------------------------------------------------- # --- per-slot loop -----------------------------------------------------
def _loop(self, slot: _Slot): def _loop(self, slot: _Slot):
backoff = self.cfg.poll_idle_seconds
while not slot.stop.is_set() and self._running: while not slot.stop.is_set() and self._running:
try: try:
jobs = self.client.lease(self.cfg.batch_size) jobs = self.client.lease(self.cfg.batch_size)
backoff = self.cfg.poll_idle_seconds # server answered → reset
except Exception: 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 continue
if not jobs: if not jobs:
time.sleep(self.cfg.poll_idle_seconds) self._interruptible_sleep(slot, self.cfg.poll_idle_seconds)
continue continue
slot.inflight = [j["job_id"] for j in jobs] slot.inflight = [j["job_id"] for j in jobs]
for job in jobs: for job in jobs:
if slot.stop.is_set() or not self._running: if slot.stop.is_set() or not self._running:
break break
self._process(job) ok = self._process(job)
slot.inflight = [i for i in slot.inflight if i != job["job_id"]] slot.inflight = [i for i in slot.inflight if i != job["job_id"]]
if not ok:
# 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: if slot.inflight:
self.client.heartbeat(slot.inflight) self.client.heartbeat(slot.inflight)
# Graceful hand-back of anything leased but not processed. # Graceful hand-back of anything leased but not processed.
@@ -110,7 +162,26 @@ class Worker:
self.client.release(slot.inflight) self.client.release(slot.inflight)
slot.inflight = [] slot.inflight = []
def _process(self, job: dict): def _interruptible_sleep(self, slot: _Slot, seconds: float):
"""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) self._bump(active=1)
try: try:
data = self.client.fetch_image(job["image_url"]) data = self.client.fetch_image(job["image_url"])
@@ -122,8 +193,31 @@ class Worker:
else: else:
frames = [(None, media.load_image(data))] 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 = [] 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}" dv = f"person-{self.cfg.detector_level}"
for t, frame in frames: for t, frame in frames:
figs = models.detect_figures(frame, self.cfg.detector_level) figs = models.detect_figures(frame, self.cfg.detector_level)
@@ -133,20 +227,48 @@ class Worker:
crop = crop_region(frame, bbox) crop = crop_region(frame, bbox)
if crop is None: if crop is None:
continue continue
vec = models.ccip_vector(crop, self.cfg.ccip_model or None) if want_ccip:
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": vec, "ccip_embedding": models.ccip_vector(
"embedding_version": ev, crop, self.cfg.ccip_model or None
"detector_version": dv, ),
}) "embedding_version": ccip_ev,
self.client.submit(job["job_id"], regions, ["figure", "face"]) "detector_version": dv,
})
if want_siglip:
regions.append({
"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) self._bump(processed=1)
except Exception as exc: # noqa: BLE001 — report + move on 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._bump(errors=1)
self.client.fail(job["job_id"], str(exc)[:500]) 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: finally:
self._bump(active=-1) self._bump(active=-1)
+4
View File
@@ -3,6 +3,10 @@ dghs-imgutils>=0.4
# GPU inference for the ONNX models. Swap to onnxruntime (CPU) for a slow # GPU inference for the ONNX models. Swap to onnxruntime (CPU) for a slow
# server-side fallback run. # server-side fallback run.
onnxruntime-gpu 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. # Control surface + HTTP.
fastapi fastapi
uvicorn[standard] uvicorn[standard]
@@ -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")
+42
View File
@@ -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")
+18
View File
@@ -37,6 +37,15 @@ async def overview():
.where(ImageRegion.ccip_embedding.is_not(None)) .where(ImageRegion.ccip_embedding.is_not(None))
) )
).scalar_one() ).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 # Per-character reference counts (no vectors loaded) — which characters
# have enough examples to match on. # have enough examples to match on.
ref_rows = ( ref_rows = (
@@ -62,14 +71,23 @@ async def overview():
) )
).all() if v ).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({ return jsonify({
"regions_by_kind": by_kind, "regions_by_kind": by_kind,
"images_with_figure_ccip": images_with_figure_ccip, "images_with_figure_ccip": images_with_figure_ccip,
"images_with_concept_siglip": images_with_concept_siglip,
"characters_with_references": len(ref_rows), "characters_with_references": len(ref_rows),
"character_references": [ "character_references": [
{"tag_id": t, "name": n, "n_refs": c} for (t, n, c) in ref_rows {"tag_id": t, "name": n, "n_refs": c} for (t, n, c) in ref_rows
], ],
"embedding_versions": versions, "embedding_versions": versions,
"auto_applied": auto_applied,
}) })
+7
View File
@@ -17,6 +17,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert
from ..extensions import get_session from ..extensions import get_session
from ..models import AppSetting, GpuJob, ImageRecord, MLSettings from ..models import AppSetting, GpuJob, ImageRecord, MLSettings
from ..services.gallery_service import image_url 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.gpu_jobs import GpuJobService
from ..services.ml.regions import RegionService from ..services.ml.regions import RegionService
@@ -137,6 +138,12 @@ async def lease():
# For video/animated: the agent samples at this cadence. # For video/animated: the agent samples at this cadence.
"frame_interval_seconds": ml.video_frame_interval_seconds, "frame_interval_seconds": ml.video_frame_interval_seconds,
"max_frames": ml.video_max_frames, "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}) return jsonify({"jobs": out})
+10
View File
@@ -21,6 +21,9 @@ _EDITABLE = (
"head_auto_apply_precision", "head_auto_apply_precision",
"head_auto_apply_enabled", "head_auto_apply_enabled",
"head_auto_apply_min_positives", "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_precision": s.head_auto_apply_precision,
"head_auto_apply_enabled": s.head_auto_apply_enabled, "head_auto_apply_enabled": s.head_auto_apply_enabled,
"head_auto_apply_min_positives": s.head_auto_apply_min_positives, "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" return "head_auto_apply_precision must be between 0.5 and 0.999"
if int(p["head_auto_apply_min_positives"]) < 1: if int(p["head_auto_apply_min_positives"]) < 1:
return "head_auto_apply_min_positives must be >= 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 return None
+14
View File
@@ -121,6 +121,20 @@ def make_celery() -> Celery:
"task": "backend.app.tasks.ml.recover_orphaned_gpu_jobs", "task": "backend.app.tasks.ml.recover_orphaned_gpu_jobs",
"schedule": 60.0, # quick pickup of work a dead agent orphaned "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": { "snapshot-head-metrics-daily": {
"task": "backend.app.tasks.maintenance.snapshot_head_metrics", "task": "backend.app.tasks.maintenance.snapshot_head_metrics",
"schedule": 86400.0, "schedule": 86400.0,
+15
View File
@@ -86,6 +86,21 @@ class MLSettings(Base):
head_auto_apply_min_positives: Mapped[int] = mapped_column( head_auto_apply_min_positives: Mapped[int] = mapped_column(
Integer, nullable=False, default=30 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( tagger_model_version: Mapped[str] = mapped_column(
String(128), nullable=False, default="camie-tagger-v2" String(128), nullable=False, default="camie-tagger-v2"
) )
+69 -9
View File
@@ -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). 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 sqlalchemy.ext.asyncio import AsyncSession
from ...models import ImageRegion, Tag, TagKind from ...models import ImageRegion, MLSettings, Tag, TagKind
from ...models.tag import image_tag 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") _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): def _l2norm(mat, np):
n = np.linalg.norm(mat, axis=1, keepdims=True) n = np.linalg.norm(mat, axis=1, keepdims=True)
n[n == 0] = 1.0 n[n == 0] = 1.0
return mat / n 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]: async def character_references(session: AsyncSession) -> dict[int, list]:
"""Per character-tag CCIP reference vectors: figure/face-region CCIP """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 = ( rows = (
await session.execute( await session.execute(
select(image_tag.c.tag_id, ImageRegion.ccip_embedding) 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(Tag.kind == TagKind.character)
.where(ImageRegion.kind.in_(_FIGURE_KINDS)) .where(ImageRegion.kind.in_(_FIGURE_KINDS))
.where(ImageRegion.ccip_embedding.is_not(None)) .where(ImageRegion.ccip_embedding.is_not(None))
.where(ImageRegion.image_record_id.in_(_single_character_images()))
) )
).all() ).all()
refs: dict[int, list] = {} refs: dict[int, list] = {}
for tag_id, vec in rows: for tag_id, vec in rows:
refs.setdefault(tag_id, []).append(vec) refs.setdefault(tag_id, []).append(vec)
_REF_CACHE.update(sig=sig, refs=refs)
return refs return refs
@@ -68,14 +124,18 @@ async def _tag_names(session: AsyncSession, tag_ids: list[int]) -> dict[int, str
async def match_image( 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]: ) -> list[dict]:
"""Character suggestions for one image from its figure-region CCIP vectors: """Character suggestions for one image from its figure-region CCIP vectors:
[{tag_id, name, category:'character', score, source:'ccip'}], ranked. [{tag_id, name, category:'character', score, source:'ccip'}], ranked.
Already-applied character tags are excluded. Empty if the image has no figure 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 import numpy as np
if threshold is None:
threshold = await _settings_threshold(session)
qvecs = ( qvecs = (
await session.execute( await session.execute(
select(ImageRegion.ccip_embedding).where( select(ImageRegion.ccip_embedding).where(
+29 -6
View File
@@ -29,6 +29,7 @@ from ...models import (
HeadAutoApplyRun, HeadAutoApplyRun,
HeadTrainingRun, HeadTrainingRun,
ImageRecord, ImageRecord,
ImageRegion,
MLSettings, MLSettings,
Tag, Tag,
TagHead, TagHead,
@@ -296,7 +297,14 @@ async def score_image(
category, score}], ranked. A concept surfaces when its score clears the category, score}], ranked. A concept surfaces when its score clears the
head's own suggest_threshold — or, when threshold_override is given (the head's own suggest_threshold — or, when threshold_override is given (the
typed-dropdown "show everything" mode), that flat floor instead (0 → every 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 import numpy as np
img = await session.get(ImageRecord, image_id) img = await session.get(ImageRecord, image_id)
@@ -306,11 +314,26 @@ async def score_image(
heads = await _current_heads(session, settings.embedder_model_version) heads = await _current_heads(session, settings.embedder_model_version)
if heads["W"] is None: if heads["W"] is None:
return [] 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 = [] out = []
for i, p in enumerate(probs): for i, p in enumerate(probs):
cut = threshold_override if threshold_override is not None else heads["thr"][i] cut = threshold_override if threshold_override is not None else heads["thr"][i]
+152 -12
View File
@@ -742,24 +742,43 @@ def scheduled_apply_head_tags() -> str:
@celery.task(name="backend.app.tasks.ml.enqueue_gpu_backfill") @celery.task(name="backend.app.tasks.ml.enqueue_gpu_backfill")
def enqueue_gpu_backfill(task_name: str) -> int: 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 exists, insert, literal
from sqlalchemy import select as sa_select from sqlalchemy import select as sa_select
from ..models import GpuJob, ImageRecord from ..models import GpuJob, ImageRecord, ImageRegion
SessionLocal = _sync_session_factory() SessionLocal = _sync_session_factory()
with SessionLocal() as session: with SessionLocal() as session:
already = exists().where( if task_name == "siglip":
GpuJob.image_record_id == ImageRecord.id, has_concept = exists().where(
GpuJob.task == task_name, ImageRegion.image_record_id == ImageRecord.id,
GpuJob.status.in_(["pending", "leased", "done"]), ImageRegion.kind == "concept",
) )
sel = sa_select( queued = exists().where(
ImageRecord.id, literal(task_name), literal("pending") GpuJob.image_record_id == ImageRecord.id,
).where(~already) 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,
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. # RETURNING + count: result.rowcount is unreliable for INSERT…SELECT.
rows = session.execute( rows = session.execute(
insert(GpuJob) insert(GpuJob)
@@ -795,3 +814,124 @@ def recover_orphaned_gpu_jobs() -> int:
) )
session.commit() session.commit()
return res.rowcount or 0 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" v-show="store.byCategory[cat] && store.byCategory[cat].length"
:label="labelFor(cat)" :items="store.byCategory[cat] || []" :label="labelFor(cat)" :items="store.byCategory[cat] || []"
@accept="onAccept" @alias="onAlias" @remove-alias="onRemoveAlias" @accept="onAccept" @alias="onAlias" @remove-alias="onRemoveAlias"
@dismiss="store.dismiss" @undismiss="store.undismiss" @dismiss="onDismiss" @undismiss="onUndismiss"
/> />
<SuggestionsCategoryGroup <SuggestionsCategoryGroup
v-if="store.byCategory.general && store.byCategory.general.length" v-if="store.byCategory.general && store.byCategory.general.length"
label="General" :items="store.byCategory.general" label="General" :items="store.byCategory.general"
collapsible :default-open="true" collapsible :default-open="true"
@accept="onAccept" @alias="onAlias" @remove-alias="onRemoveAlias" @accept="onAccept" @alias="onAlias" @remove-alias="onRemoveAlias"
@dismiss="store.dismiss" @undismiss="store.undismiss" @dismiss="onDismiss" @undismiss="onUndismiss"
/> />
</div> </div>
@@ -57,9 +57,15 @@ const props = defineProps({
// so the same panel refreshes the right surface. See TagPanel. // so the same panel refreshes the right surface. See TagPanel.
host: { type: Object, default: null }, 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 store = useSuggestionsStore()
const modalStore = useModalStore() const modalStore = useModalStore()
const host = props.host || modalStore const host = props.host || modalStore
@@ -28,6 +28,7 @@
:image-id="host.currentImageId" :image-id="host.currentImageId"
:host="host" :host="host"
@accepted="focusTagInput" @accepted="focusTagInput"
@dismissed="focusTagInput"
/> />
<!-- @after-leave: when either dialog finishes closing (apply OR cancel), <!-- @after-leave: when either dialog finishes closing (apply OR cancel),
@@ -60,6 +60,52 @@
Enqueues every image that doesn't have a CCIP embedding yet. Nothing Enqueues every image that doesn't have a CCIP embedding yet. Nothing
processes until the agent is running. processes until the agent is running.
</p> </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> </MaintenanceTile>
</template> </template>
@@ -69,14 +115,22 @@ import { computed, onMounted, onUnmounted, ref } from 'vue'
import MaintenanceTile from '../common/MaintenanceTile.vue' import MaintenanceTile from '../common/MaintenanceTile.vue'
import { useGpuStore } from '../../stores/gpu.js' import { useGpuStore } from '../../stores/gpu.js'
import { useMLStore } from '../../stores/ml.js'
import { copyText } from '../../utils/clipboard.js' import { copyText } from '../../utils/clipboard.js'
const store = useGpuStore() const store = useGpuStore()
const ml = useMLStore()
const loading = ref(true) const loading = ref(true)
const tokenValue = ref(null) const tokenValue = ref(null)
const masked = ref(true) const masked = ref(true)
const rotating = ref(false) const rotating = ref(false)
const backfilling = 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 }) const queue = ref({ pending: 0, leased: 0, done: 0, error: 0 })
let pollTimer = null let pollTimer = null
@@ -94,9 +148,46 @@ onMounted(async () => {
} }
await refreshQueue() await refreshQueue()
pollTimer = setInterval(() => { if (!document.hidden) refreshQueue() }, 5000) 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) }) 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() { async function refreshQueue() {
try { queue.value = await store.status() } catch { /* non-fatal */ } try { queue.value = await store.status() } catch { /* non-fatal */ }
} }
@@ -135,6 +226,19 @@ async function onBackfill() {
backfilling.value = false 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> </script>
<style scoped> <style scoped>
+17 -7
View File
@@ -11,7 +11,7 @@ import { toast } from '../utils/toast.js'
// trail. The store ALSO acts as a TagPanel "host" (current/currentImageId + // 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 // tag CRUD over the anchor) so the Explore workspace reuses the modal's tag
// rail verbatim for modal-parity tagging while rabbit-holing. // 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', () => { export const useExploreStore = defineStore('explore', () => {
const api = useApi() const api = useApi()
@@ -81,16 +81,26 @@ export const useExploreStore = defineStore('explore', () => {
return cursor.value > 0 ? breadcrumb.value[cursor.value - 1].id : null 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 () { function forwardTarget () {
if (cursor.value >= 0 && cursor.value < breadcrumb.value.length - 1) { if (cursor.value >= 0 && cursor.value < breadcrumb.value.length - 1) {
return breadcrumb.value[cursor.value + 1].id 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 () { function reset () {
+55
View File
@@ -1,6 +1,7 @@
"""CCIP few-shot character matcher (#114). numpy cosine on stored vectors — no """CCIP few-shot character matcher (#114). numpy cosine on stored vectors — no
model needed, so it runs in CI with synthetic CCIP vectors.""" model needed, so it runs in CI with synthetic CCIP vectors."""
import pytest import pytest
from sqlalchemy import select
from backend.app.models import ImageRecord, ImageRegion, TagKind from backend.app.models import ImageRecord, ImageRegion, TagKind
from backend.app.models.tag import image_tag 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) query = await _img(db, "g" * 64)
await db.commit() await db.commit()
assert await match_image(db, query.id) == [] 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]
+39 -2
View File
@@ -2,9 +2,9 @@
from datetime import UTC, datetime, timedelta from datetime import UTC, datetime, timedelta
import pytest 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 from backend.app.services.ml.gpu_jobs import GpuJobService
pytestmark = pytest.mark.integration pytestmark = pytest.mark.integration
@@ -20,6 +20,43 @@ async def _img(db, sha) -> ImageRecord:
return img 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 @pytest.mark.asyncio
async def test_enqueue_dedupes_same_pair(db): async def test_enqueue_dedupes_same_pair(db):
img = await _img(db, "a" * 64) img = await _img(db, "a" * 64)
+34
View File
@@ -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"]) 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 @pytest.mark.asyncio
async def test_rejected_tag_surfaced_flagged_then_reversible(db): async def test_rejected_tag_surfaced_flagged_then_reversible(db):
# A dismissed suggestion is NOT dropped: it stays flagged rejected so the # A dismissed suggestion is NOT dropped: it stays flagged rejected so the