From 4daa3f27903838a5be98d8a0bd35ba83d2597f96 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 30 Jun 2026 10:24:30 -0400 Subject: [PATCH] =?UTF-8?q?feat(ml):=20operator=20model=20swap=20=E2=80=94?= =?UTF-8?q?=20GPU=20re-embed=20+=20embedder=20as=20a=20setting=20(#1190)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Make the SigLIP embedder an operator choice (drop-in to SigLIP 2: google/siglip2-so400m-patch16-512 is a verified 1152-d model at 512px → no schema change, better small-cue fidelity). A swap = set model + re-embed + retrain, all operator-driven; the GPU agent does the re-embed so it's fast. - settings: embedder_model_name is now a setting (migration 0065) alongside the existing embedder_model_version; both editable + validated (non-empty) in the ml admin API. The server embedder loads by HF name (AutoImageProcessor/Model, model-agnostic), preferring the pre-downloaded local dir for the default so existing deploys don't re-download; rebuilds on a name change. - agent: new 'embed' job = whole-image SigLIP embedding (mean-pool video frames) under the lease-announced model → POST /jobs/submit_embedding writes image_record.siglip_embedding + siglip_model_version. The lease now announces the model FROM THE SETTING (not a constant). - re-embed routing: enqueue_gpu_backfill('embed') selects unembedded + stale- version images; 'siglip' now re-embeds concept crops whose version != current (so a swap re-triggers crops, not just the never-embedded back-catalogue). The CPU ml-worker backfill no longer re-embeds on a version mismatch (it can't churn the library at 512px) — the GPU agent owns version re-embeds. Daily 'embed' + 'siglip' beats self-heal. - scoring: score_image only bags embeddings in the CURRENT model's space (whole- image gated by siglip_model_version, concept regions by embedding_version) so a mid-swap stale vector isn't scored by new-space heads; legacy NULL = current. - UI: GpuAgentCard "Embedding model (advanced)" — edit name/version, Save, and "Re-embed library (GPU)" (queues embed + siglip); points at SigLIP 2. Tests: lease announces model + submit_embedding round-trip; enqueue 'embed' selects stale/unembedded; stale-version excluded from scoring; embedder model settable + empty rejected; siglip gate updated to current-version concept. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa --- agent/fc_agent/client.py | 13 ++++ agent/fc_agent/worker.py | 37 +++++++--- alembic/versions/0065_embedder_model_name.py | 35 ++++++++++ backend/app/api/gpu.py | 40 +++++++++-- backend/app/api/ml_admin.py | 8 +++ backend/app/celery_app.py | 5 ++ backend/app/models/ml_settings.py | 6 ++ backend/app/services/ml/embedder.py | 53 +++++++++------ backend/app/services/ml/heads.py | 19 ++++-- backend/app/tasks/ml.py | 43 +++++++++--- .../src/components/settings/GpuAgentCard.vue | 68 +++++++++++++++++++ tests/test_api_gpu.py | 33 +++++++++ tests/test_api_ml_admin.py | 20 ++++++ tests/test_gpu_jobs.py | 38 ++++++++++- tests/test_ml_suggestions.py | 14 ++++ 15 files changed, 379 insertions(+), 53 deletions(-) create mode 100644 alembic/versions/0065_embedder_model_name.py diff --git a/agent/fc_agent/client.py b/agent/fc_agent/client.py index 7003bcc..1c297c9 100644 --- a/agent/fc_agent/client.py +++ b/agent/fc_agent/client.py @@ -40,6 +40,19 @@ class FcClient: r.raise_for_status() return r.json() + def submit_embedding(self, job_id: int, embedding: list, version: str) -> dict: + """Post a whole-image SigLIP embedding (the 'embed' task) → image_record.""" + r = self.s.post( + f"{self.base}/api/gpu/jobs/submit_embedding", + json={ + "agent_id": self.agent_id, "job_id": job_id, + "embedding": embedding, "embedding_version": version, + }, + timeout=120, + ) + r.raise_for_status() + return r.json() + def heartbeat(self, job_ids: list[int]) -> None: try: self.s.post( diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py index 8b42277..dd5b206 100644 --- a/agent/fc_agent/worker.py +++ b/agent/fc_agent/worker.py @@ -11,6 +11,7 @@ orphaned work is re-picked at once rather than waiting out the lease. """ import threading +import numpy as np import requests from . import media, models @@ -193,28 +194,42 @@ class Worker: else: frames = [(None, media.load_image(data))] + task = job.get("task") or "ccip" + embed_version = job.get("embed_version") or DEFAULT_EMBED_VERSION + model_name = ( + self.cfg.embed_model_override + or job.get("embed_model_name") + or DEFAULT_EMBED_MODEL + ) + + # 'embed' = WHOLE-IMAGE SigLIP embedding (re-embed the library under a + # new model, #1190) → image_record.siglip_embedding. Mean-pool video + # frames, matching the server's tag_and_embed. No regions. + if task == "embed": + embedder = self._ensure_embedder(model_name) + vecs = [embedder.embed(frame) for _, frame in frames] + if len(vecs) > 1: + vec = np.mean( + np.asarray(vecs, dtype=np.float32), axis=0 + ).tolist() + else: + vec = vecs[0] + self.client.submit_embedding(job["job_id"], vec, embed_version) + self._bump(processed=1) + return True + # 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) + embedder = self._ensure_embedder(model_name) if want_siglip else None regions = [] ccip_ev = self.cfg.ccip_model or "ccip-default" diff --git a/alembic/versions/0065_embedder_model_name.py b/alembic/versions/0065_embedder_model_name.py new file mode 100644 index 0000000..0a986b3 --- /dev/null +++ b/alembic/versions/0065_embedder_model_name.py @@ -0,0 +1,35 @@ +"""ml_settings: embedder_model_name (#1190 operator model swap) + +The embedder MODEL VERSION was already a setting (and stamps image_record. +siglip_model_version); the HF model NAME was env-only, so an operator couldn't +actually point the pipeline at a different embedder. Storing the name as a +setting makes the model an operator choice: set name + version → re-embed (the +GPU agent) → retrain heads. Default = the current SigLIP so400m. + +Revision ID: 0065 +Revises: 0064 +Create Date: 2026-06-30 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0065" +down_revision: Union[str, None] = "0064" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "ml_settings", + sa.Column( + "embedder_model_name", sa.String(length=128), nullable=False, + server_default="google/siglip-so400m-patch14-384", + ), + ) + + +def downgrade() -> None: + op.drop_column("ml_settings", "embedder_model_name") diff --git a/backend/app/api/gpu.py b/backend/app/api/gpu.py index cf8fc73..09a488f 100644 --- a/backend/app/api/gpu.py +++ b/backend/app/api/gpu.py @@ -17,7 +17,6 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert from ..extensions import get_session from ..models import AppSetting, GpuJob, ImageRecord, MLSettings from ..services.gallery_service import image_url -from ..services.ml.embedder import MODEL_NAME as EMBED_MODEL_NAME from ..services.ml.gpu_jobs import GpuJobService from ..services.ml.regions import RegionService @@ -138,11 +137,12 @@ async def lease(): # For video/animated: the agent samples at this cadence. "frame_interval_seconds": ml.video_frame_interval_seconds, "max_frames": ml.video_max_frames, - # The embedding model the agent must use for concept crops, so - # its region vectors land in the SAME space the heads trained in. - # Server-announced → the agent stays model-agnostic; a swap is a - # server setting + a re-embed migration, never an agent change. - "embed_model_name": EMBED_MODEL_NAME, + # The embedding model the agent must use for concept crops + the + # whole-image 'embed' task, so its vectors land in the SAME space + # the heads trained in. Server-announced FROM THE SETTING → the + # agent stays model-agnostic; an operator swap is a setting + a + # re-embed, never an agent change. + "embed_model_name": ml.embedder_model_name, "embed_version": ml.embedder_model_version, }) return jsonify({"jobs": out}) @@ -188,6 +188,34 @@ async def submit(): return jsonify({"ok": True, "stored": len(regions)}) +@gpu_bp.route("/jobs/submit_embedding", methods=["POST"]) +async def submit_embedding(): + """Store a whole-image SigLIP embedding (the 'embed' task) on image_record + + close the job. Body: {agent_id, job_id, embedding:[...], embedding_version}. + This is how the GPU agent re-embeds the library under a new model (#1190) — + much faster than the CPU ml-worker at higher resolutions.""" + body = await request.get_json(silent=True) or {} + agent_id = str(body.get("agent_id") or "agent") + job_id = body.get("job_id") + embedding = body.get("embedding") + version = body.get("embedding_version") + if job_id is None or not embedding or not version: + return jsonify({"error": "job_id, embedding, embedding_version required"}), 400 + async with get_session() as session: + if not await _agent_authed(session): + return jsonify({"error": "unauthorized"}), 401 + job = await session.get(GpuJob, int(job_id)) + if job is None or job.status != "leased" or job.lease_token != agent_id: + return jsonify({"error": "lease_invalid"}), 409 + img = await session.get(ImageRecord, job.image_record_id) + if img is not None: + img.siglip_embedding = embedding + img.siglip_model_version = version + await GpuJobService(session).complete(agent_id, int(job_id)) + await session.commit() + return jsonify({"ok": True}) + + @gpu_bp.route("/jobs/fail", methods=["POST"]) async def fail(): body = await request.get_json(silent=True) or {} diff --git a/backend/app/api/ml_admin.py b/backend/app/api/ml_admin.py index 1472770..102c004 100644 --- a/backend/app/api/ml_admin.py +++ b/backend/app/api/ml_admin.py @@ -24,6 +24,8 @@ _EDITABLE = ( "ccip_match_threshold", "ccip_auto_apply_enabled", "ccip_auto_apply_threshold", + "embedder_model_name", + "embedder_model_version", ) @@ -54,6 +56,7 @@ async def get_settings(): "ccip_match_threshold": s.ccip_match_threshold, "ccip_auto_apply_enabled": s.ccip_auto_apply_enabled, "ccip_auto_apply_threshold": s.ccip_auto_apply_threshold, + "embedder_model_name": s.embedder_model_name, } ) @@ -125,6 +128,11 @@ def _validate(p: dict) -> str | None: return "ccip_match_threshold must be between 0.5 and 0.999" if not (0.5 <= float(p["ccip_auto_apply_threshold"]) <= 0.999): return "ccip_auto_apply_threshold must be between 0.5 and 0.999" + # Embedder model swap (#1190): both must be non-empty. Changing them means a + # different embedding space — the operator must re-embed + retrain after. + for key in ("embedder_model_name", "embedder_model_version"): + if not str(p[key]).strip(): + return f"{key} must not be empty" return None diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 5778a1e..7b7a4e3 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -131,6 +131,11 @@ def make_celery() -> Celery: "schedule": 86400.0, # drain the concept-crop back-catalogue + "args": ("siglip",), # retry failed embeds, no button needed }, + "enqueue-embed-backfill-daily": { + "task": "backend.app.tasks.ml.enqueue_gpu_backfill", + "schedule": 86400.0, # whole-image re-embed under the current + "args": ("embed",), # model (an operator swap) drains via agent + }, "ccip-auto-apply-daily": { "task": "backend.app.tasks.ml.scheduled_ccip_auto_apply", "schedule": 86400.0, # no-op unless ccip_auto_apply_enabled diff --git a/backend/app/models/ml_settings.py b/backend/app/models/ml_settings.py index 9825568..0e84940 100644 --- a/backend/app/models/ml_settings.py +++ b/backend/app/models/ml_settings.py @@ -107,6 +107,12 @@ class MLSettings(Base): embedder_model_version: Mapped[str] = mapped_column( String(128), nullable=False, default="siglip-so400m-patch14-384" ) + # The HF model NAME the embedder loads (server CPU embed + announced to the + # GPU agent in the lease). Operator-settable so the embedder is a choice, not + # a hardcode (#1190): set name + version together, then re-embed + retrain. + embedder_model_name: Mapped[str] = mapped_column( + String(128), nullable=False, default="google/siglip-so400m-patch14-384" + ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) diff --git a/backend/app/services/ml/embedder.py b/backend/app/services/ml/embedder.py index d94f71e..d55646c 100644 --- a/backend/app/services/ml/embedder.py +++ b/backend/app/services/ml/embedder.py @@ -18,9 +18,11 @@ ImageFile.LOAD_TRUNCATED_IMAGES = True # N_replicas × this within the cores allotted to ML to avoid oversubscription. _INTRA_OP_THREADS = 4 -MODEL_NAME = os.environ.get( +DEFAULT_MODEL_NAME = os.environ.get( "SIGLIP_MODEL_NAME", "google/siglip-so400m-patch14-384" ) +# Back-compat alias (api/gpu imported this name as the fallback embedder id). +MODEL_NAME = DEFAULT_MODEL_NAME MODEL_VERSION = os.environ.get( "SIGLIP_MODEL_VERSION", "siglip-so400m-patch14-384" ) @@ -29,35 +31,42 @@ _LOCAL_DIR = Path(os.environ.get("ML_MODEL_DIR", "/models")) / "siglip" class Embedder: - def __init__(self, model_dir: Path | None = None): - self._model_dir = model_dir or _LOCAL_DIR + """Loads whatever SigLIP-family model it's given by HF NAME. For the default + model it prefers the pre-downloaded local dir (no re-download on existing + deploys); any other name resolves as an HF repo id (downloaded + cached on + first use), so an operator model swap (#1190) just works server-side.""" + + def __init__(self, model_name: str | None = None, model_dir: Path | None = None): + self.model_name = model_name or DEFAULT_MODEL_NAME + self._explicit_dir = model_dir self._model = None self._processor = None self._torch = None + def _source(self) -> str: + if self._explicit_dir is not None: + return str(self._explicit_dir) + if self.model_name == DEFAULT_MODEL_NAME and _LOCAL_DIR.exists(): + return str(_LOCAL_DIR) + return self.model_name + def load(self) -> None: if self._model is not None: return import torch - from transformers import AutoModel, SiglipImageProcessor + from transformers import AutoImageProcessor, AutoModel self._torch = torch # Bound torch's CPU thread pool (see _INTRA_OP_THREADS) so each replica # stays a predictable core consumer on a shared node. torch.set_num_threads(_INTRA_OP_THREADS) - # FC's embedder only does IMAGE inference — never text. AutoProcessor - # loads the full processor including SiglipTokenizer, which requires - # the sentencepiece library at import time even if we never call it. - # SiglipImageProcessor loads ONLY preprocessor_config.json (image - # side) and skips the tokenizer config entirely. Operator hit the - # ImportError 2026-05-25 once the ml-worker started actually running - # tag_and_embed; switching to the image-only loader avoids the - # tokenizer dep without adding ~30 MB of unused C++ build to the - # lean ml-worker image. - self._processor = SiglipImageProcessor.from_pretrained( - str(self._model_dir) - ) - self._model = AutoModel.from_pretrained(str(self._model_dir)) + # IMAGE inference only — AutoImageProcessor loads just the image side + # (preprocessor_config.json), skipping the SigLIP tokenizer + its + # sentencepiece dep (operator hit that ImportError 2026-05-25). Works + # for any SigLIP-family model, keeping the embedder model-agnostic. + src = self._source() + self._processor = AutoImageProcessor.from_pretrained(src) + self._model = AutoModel.from_pretrained(src) self._model.eval() def infer(self, image_path: Path) -> np.ndarray: @@ -74,8 +83,12 @@ class Embedder: _default_embedder: Embedder | None = None -def get_embedder() -> Embedder: +def get_embedder(model_name: str | None = None) -> Embedder: + """Cached embedder for `model_name` (default if None). Rebuilds the singleton + when the requested name changes, so an operator model swap takes effect + without restarting the worker.""" global _default_embedder - if _default_embedder is None: - _default_embedder = Embedder() + name = model_name or DEFAULT_MODEL_NAME + if _default_embedder is None or _default_embedder.model_name != name: + _default_embedder = Embedder(model_name=name) return _default_embedder diff --git a/backend/app/services/ml/heads.py b/backend/app/services/ml/heads.py index 879b922..b234897 100644 --- a/backend/app/services/ml/heads.py +++ b/backend/app/services/ml/heads.py @@ -308,25 +308,36 @@ async def score_image( import numpy as np img = await session.get(ImageRecord, image_id) - if img is None or img.siglip_embedding is None: + if img is None: return [] settings = await _settings_async(session) - heads = await _current_heads(session, settings.embedder_model_version) + cur_version = settings.embedder_model_version + heads = await _current_heads(session, cur_version) if heads["W"] is None: return [] - bag = [np.asarray(img.siglip_embedding, dtype=np.float32)] + # Only embeddings in the CURRENT model's space enter the bag. Mid model-swap + # (#1190), an image still carrying the OLD-version whole-image vector is + # skipped rather than scored by heads trained in a different space; a legacy + # NULL version is treated as current (those predate per-row stamping). + bag = [] + if img.siglip_embedding is not None and img.siglip_model_version in ( + cur_version, None, + ): + bag.append(np.asarray(img.siglip_embedding, dtype=np.float32)) region_vecs = ( await session.execute( select(ImageRegion.siglip_embedding) .where(ImageRegion.image_record_id == image_id) .where(ImageRegion.siglip_embedding.is_not(None)) - .where(ImageRegion.embedding_version == settings.embedder_model_version) + .where(ImageRegion.embedding_version == cur_version) ) ).all() for (vec,) in region_vecs: if vec is not None: bag.append(np.asarray(vec, dtype=np.float32)) + if not bag: + return [] X = np.vstack(bag) # (B, D) norms = np.linalg.norm(X, axis=1, keepdims=True) diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index 84b40a2..f2aa6a7 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -95,7 +95,7 @@ def tag_and_embed(self, image_id: int) -> dict: phase = "load_models" tagger = get_tagger() - embedder = get_embedder() + embedder = get_embedder(settings.embedder_model_name) if is_vid: # Layer-3 isolation: ffprobe (a separate process) validates @@ -330,10 +330,10 @@ def backfill(self) -> int: != settings.tagger_model_version ) | (ImageRecord.siglip_embedding.is_(None)) - | ( - ImageRecord.siglip_model_version - != settings.embedder_model_version - ) + # NB: a siglip MODEL-VERSION mismatch (an operator model swap, + # #1190) is intentionally NOT re-embedded here — the CPU + # ml-worker can't churn the whole library at 384/512px. The + # GPU agent owns version re-embeds via the 'embed' job. ) .order_by(ImageRecord.id.asc()) .limit(500) @@ -750,17 +750,40 @@ def enqueue_gpu_backfill(task_name: str) -> int: job, so it picks up the back-catalogue of images that were CCIP-embedded before concept crops existed, and retries images whose concept embed failed — without re-touching their figure/CCIP regions.""" - from sqlalchemy import exists, insert, literal + from sqlalchemy import exists, insert, literal, or_ from sqlalchemy import select as sa_select - from ..models import GpuJob, ImageRecord, ImageRegion + from ..models import GpuJob, ImageRecord, ImageRegion, MLSettings SessionLocal = _sync_session_factory() with SessionLocal() as session: - if task_name == "siglip": - has_concept = exists().where( + cur_version = session.execute( + select(MLSettings.embedder_model_version).where(MLSettings.id == 1) + ).scalar_one() + if task_name == "embed": + # Whole-image GPU re-embed (#1190): images with no embedding, or one + # stamped under a DIFFERENT model version (an operator model swap). + stale = or_( + ImageRecord.siglip_embedding.is_(None), + ImageRecord.siglip_model_version.is_(None), + ImageRecord.siglip_model_version != cur_version, + ) + queued = exists().where( + GpuJob.image_record_id == ImageRecord.id, + GpuJob.task == "embed", + GpuJob.status.in_(["pending", "leased"]), + ) + sel = sa_select( + ImageRecord.id, literal("embed"), literal("pending") + ).where(stale).where(~queued) + elif task_name == "siglip": + # Concept-crop re-embed: enqueue when there's no concept region AT THE + # CURRENT model version — so a model swap re-triggers crops too, not + # only the never-embedded back-catalogue. + has_current_concept = exists().where( ImageRegion.image_record_id == ImageRecord.id, ImageRegion.kind == "concept", + ImageRegion.embedding_version == cur_version, ) queued = exists().where( GpuJob.image_record_id == ImageRecord.id, @@ -769,7 +792,7 @@ def enqueue_gpu_backfill(task_name: str) -> int: ) sel = sa_select( ImageRecord.id, literal("siglip"), literal("pending") - ).where(~has_concept).where(~queued) + ).where(~has_current_concept).where(~queued) else: already = exists().where( GpuJob.image_record_id == ImageRecord.id, diff --git a/frontend/src/components/settings/GpuAgentCard.vue b/frontend/src/components/settings/GpuAgentCard.vue index dc0ed24..b0da818 100644 --- a/frontend/src/components/settings/GpuAgentCard.vue +++ b/frontend/src/components/settings/GpuAgentCard.vue @@ -106,6 +106,37 @@ reversible) — so identity tags keep flowing without review. Stricter than the suggest cut; 0.92 recommended.

+ + +
Embedding model (advanced)
+
+ + +
+ Save model + Re-embed library (GPU) +
+

+ Changing the model means a DIFFERENT embedding space. After saving a new + model + version, run Re-embed library (the GPU agent re-embeds + whole images + concept crops), then Retrain heads. Suggestions + degrade until both finish. SigLIP 2 (google/siglip2-so400m-patch16-512, + version siglip2-so400m-patch16-512) is a 1152-d drop-in at + 512px — no schema change. +

+
@@ -131,6 +162,10 @@ const savingThreshold = ref(false) const autoApply = ref(true) const autoThreshold = ref(0.92) const savingAuto = ref(false) +const modelName = ref('') +const modelVersion = ref('') +const savingModel = ref(false) +const reembedding = ref(false) const queue = ref({ pending: 0, leased: 0, done: 0, error: 0 }) let pollTimer = null @@ -157,9 +192,42 @@ onMounted(async () => { autoApply.value = ml.settings.ccip_auto_apply_enabled autoThreshold.value = ml.settings.ccip_auto_apply_threshold } + if (ml.settings?.embedder_model_name != null) { + modelName.value = ml.settings.embedder_model_name + modelVersion.value = ml.settings.embedder_model_version + } } catch { /* non-fatal */ } }) +async function onSaveModel() { + savingModel.value = true + try { + await ml.patchSettings({ + embedder_model_name: modelName.value.trim(), + embedder_model_version: modelVersion.value.trim(), + }) + toast({ text: 'Embedding model saved — now Re-embed library, then Retrain heads', type: 'success' }) + } catch (e) { + toast({ text: `Could not save model: ${e.message}`, type: 'error' }) + } finally { + savingModel.value = false + } +} + +async function onReembed() { + reembedding.value = true + try { + await store.backfill('embed') + await store.backfill('siglip') + toast({ text: 'Queued whole-image + concept re-embed — run the agent, then Retrain heads', type: 'success' }) + await refreshQueue() + } catch (e) { + toast({ text: `Could not queue re-embed: ${e.message}`, type: 'error' }) + } finally { + reembedding.value = false + } +} + async function onSaveAuto() { savingAuto.value = true try { diff --git a/tests/test_api_gpu.py b/tests/test_api_gpu.py index 40345d8..08fbc5e 100644 --- a/tests/test_api_gpu.py +++ b/tests/test_api_gpu.py @@ -69,6 +69,39 @@ async def test_lease_submit_round_trip(client, db): assert len(regs) == 1 and len(list(regs[0].ccip_embedding)) == 768 +@pytest.mark.asyncio +async def test_lease_announces_embed_model_then_submit_embedding(client, db): + # Whole-image GPU re-embed (#1190): the lease announces the embedder model so + # the agent loads the right one, and submit_embedding writes it back onto + # image_record with its version stamp. + img = await _img(db, "b" * 64) + await GpuJobService(db).enqueue(img.id, "embed") + await db.commit() + + token = (await (await client.post("/api/gpu/token/rotate")).get_json())["token"] + hdr = {"Authorization": f"Bearer {token}"} + + leased = await client.post( + "/api/gpu/jobs/lease", json={"agent_id": "a1", "batch_size": 5}, headers=hdr, + ) + j = (await leased.get_json())["jobs"][0] + assert j["task"] == "embed" + assert j["embed_model_name"] and j["embed_version"] # server-announced model + + submitted = await client.post("/api/gpu/jobs/submit_embedding", json={ + "agent_id": "a1", "job_id": j["job_id"], + "embedding": [0.2] * 1152, "embedding_version": "siglip2-test-v9", + }, headers=hdr) + assert submitted.status_code == 200 + + st = await (await client.get("/api/gpu/status")).get_json() + assert st["done"] == 1 and st["leased"] == 0 + + await db.refresh(img) + assert img.siglip_model_version == "siglip2-test-v9" + assert img.siglip_embedding is not None and len(list(img.siglip_embedding)) == 1152 + + @pytest.mark.asyncio async def test_submit_with_stale_lease_is_409(client, db): img = await _img(db, "b" * 64) diff --git a/tests/test_api_ml_admin.py b/tests/test_api_ml_admin.py index e46be48..ebc944d 100644 --- a/tests/test_api_ml_admin.py +++ b/tests/test_api_ml_admin.py @@ -34,6 +34,26 @@ async def test_get_and_patch_settings(client): assert (await resp.get_json())["suggestion_threshold_general"] == pytest.approx(0.90) +@pytest.mark.asyncio +async def test_embedder_model_settable_and_empty_rejected(client): + # #1190: the embedder model name + version are operator-settable (a swap), + # and neither may be blanked. + body = await (await client.get("/api/ml/settings")).get_json() + assert body["embedder_model_name"] == "google/siglip-so400m-patch14-384" + + ok = await client.patch("/api/ml/settings", json={ + "embedder_model_name": "google/siglip2-so400m-patch16-512", + "embedder_model_version": "siglip2-so400m-patch16-512", + }) + assert ok.status_code == 200 + out = await ok.get_json() + assert out["embedder_model_name"] == "google/siglip2-so400m-patch16-512" + assert out["embedder_model_version"] == "siglip2-so400m-patch16-512" + + bad = await client.patch("/api/ml/settings", json={"embedder_model_name": " "}) + assert bad.status_code == 400 + + @pytest.mark.asyncio async def test_tagger_store_floor_default_and_patch(client): body = await (await client.get("/api/ml/settings")).get_json() diff --git a/tests/test_gpu_jobs.py b/tests/test_gpu_jobs.py index 3606755..38d23ca 100644 --- a/tests/test_gpu_jobs.py +++ b/tests/test_gpu_jobs.py @@ -25,13 +25,17 @@ async def test_enqueue_siglip_backfill_gates_on_concept_region(db): # 'siglip' backfill enqueues images that lack a concept region (the # back-catalogue) and skips ones that already have one — and never double- # enqueues an image that already has a pending siglip job. + from backend.app.models import MLSettings from backend.app.tasks.ml import enqueue_gpu_backfill + cur = (await db.execute( + select(MLSettings.embedder_model_version).where(MLSettings.id == 1) + )).scalar_one() need = await _img(db, "e1" * 32) # no concept region → wants one - have = await _img(db, "e2" * 32) # already embedded → skip + have = await _img(db, "e2" * 32) # concept @ current version → skip db.add(ImageRegion( image_record_id=have.id, kind="concept", rx=0.0, ry=0.0, rw=1.0, rh=1.0, - siglip_embedding=[0.0] * 1152, embedding_version="siglip-test", + siglip_embedding=[0.0] * 1152, embedding_version=cur, )) await db.commit() @@ -57,6 +61,36 @@ async def test_enqueue_siglip_backfill_gates_on_concept_region(db): assert n_for_need == 1 +@pytest.mark.asyncio +async def test_enqueue_embed_backfill_selects_stale_and_unembedded(db): + # Whole-image GPU re-embed (#1190): enqueue images with no embedding or one + # stamped under a DIFFERENT model version (an operator swap); skip ones + # already at the current version. + from backend.app.models import MLSettings + from backend.app.tasks.ml import enqueue_gpu_backfill + + cur = (await db.execute( + select(MLSettings.embedder_model_version).where(MLSettings.id == 1) + )).scalar_one() + current = await _img(db, "f1" * 32) + current.siglip_embedding = [0.0] * 1152 + current.siglip_model_version = cur # up to date → skip + stale = await _img(db, "f2" * 32) + stale.siglip_embedding = [0.0] * 1152 + stale.siglip_model_version = "old-embedder-v0" # wrong space → re-embed + unembedded = await _img(db, "f3" * 32) # never embedded → embed + await db.commit() + + assert enqueue_gpu_backfill("embed") >= 2 + queued = { + j.image_record_id for j in ( + await db.execute(select(GpuJob).where(GpuJob.task == "embed")) + ).scalars() + } + assert stale.id in queued and unembedded.id in queued + assert current.id not in queued + + @pytest.mark.asyncio async def test_enqueue_dedupes_same_pair(db): img = await _img(db, "a" * 64) diff --git a/tests/test_ml_suggestions.py b/tests/test_ml_suggestions.py index 40d938b..a797013 100644 --- a/tests/test_ml_suggestions.py +++ b/tests/test_ml_suggestions.py @@ -145,6 +145,20 @@ async def test_concept_region_surfaces_via_max_over_bag(db): assert any(s.canonical_tag_id == tag.id and s.score > 0.7 for s in general) +@pytest.mark.asyncio +async def test_stale_embedding_version_excluded_from_scoring(db): + # Mid model-swap (#1190): an image still carrying an OLD-version whole-image + # embedding must NOT be scored by heads trained in the new model's space — + # even though the vector aligns with the head, it's the wrong coordinate + # system, so nothing surfaces until it's re-embedded. + tag = await TagService(db).find_or_create("glasses", TagKind.general) + img = await _img(db, "c1" * 32, _emb(0)) + img.siglip_model_version = "some-old-model-v0" # != current embedder + await _head(db, tag.id, slot=0, suggest_threshold=0.5) + await db.commit() + assert not (await SuggestionService(db).for_image(img.id)).by_category.get("general") + + @pytest.mark.asyncio async def test_rejected_tag_surfaced_flagged_then_reversible(db): # A dismissed suggestion is NOT dropped: it stays flagged rejected so the