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

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)

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