From 19b962f1a756aa432d3497003154919469c88c9c Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 2 Jul 2026 16:53:08 -0400 Subject: [PATCH] =?UTF-8?q?feat(b3):=20ml-worker=20becomes=20optional=20?= =?UTF-8?q?=E2=80=94=20embed-only=20role,=20decoupled=20GPU=20coordination?= =?UTF-8?q?,=20cpu-embed=20switch?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ml-worker's ONLY processing role is now the CPU whole-image embed fallback (tag_and_embed renamed embed_image — Camie tagging was retired #1189 and the name kept implying otherwise; videos were already handled agent-style: frame sampling + mean-pool). Detection/cropping/CCIP stay GPU-agent-only, and their completion is judged per-pipeline: ccip by gpu_job rows, siglip by concept regions at the current model version — never by image_record.siglip_embedding. A CPU embed therefore can NEVER close crop work for the agent (regression test pins this; only the whole-image 'embed' job, the same artifact, is satisfied). Making removal actually safe (operator will drop the container): - GPU-queue coordination (enqueue_gpu_backfill, recover_orphaned_gpu_jobs, reprocess_gpu_jobs) moved verbatim to tasks/gpu_queue.py on the maintenance quick lane — it lived on the 'ml' queue only by module colocation, which made the ml-worker a hard dependency of the whole agent pipeline. - New ml_settings.cpu_embed_enabled (migration 0074, default ON so agent-less installs keep working): OFF stops the four import hooks queueing embed work nothing will consume and no-ops the manual backfill; switch lives on the renamed 'CPU embedding backfill' card. - NB heads training / auto-apply still run on the ml image (sklearn) — a stack that removes the container gives those up too. Deploy note: in-flight messages under the old task names are dropped by the new workers; the 60s orphan sweep + hourly backfill re-fire under the new names immediately. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM --- .../0074_ml_settings_cpu_embed_enabled.py | 35 +++ backend/app/api/gpu.py | 4 +- backend/app/api/ml_admin.py | 2 + backend/app/celery_app.py | 14 +- backend/app/models/ml_settings.py | 9 + backend/app/services/cleanup_service.py | 6 +- backend/app/services/download_service.py | 6 +- backend/app/tasks/external.py | 6 +- backend/app/tasks/gpu_queue.py | 171 ++++++++++++++ backend/app/tasks/import_file.py | 6 +- backend/app/tasks/maintenance.py | 2 +- backend/app/tasks/ml.py | 217 ++++-------------- .../components/settings/MLBackfillCard.vue | 62 ++++- tests/test_api_gpu.py | 32 ++- tests/test_download_service.py | 2 +- tests/test_external_worker.py | 4 +- tests/test_gpu_jobs.py | 10 +- tests/test_maintenance.py | 2 +- tests/test_reextract_archives.py | 4 +- tests/test_tasks_ml.py | 36 ++- 20 files changed, 428 insertions(+), 202 deletions(-) create mode 100644 alembic/versions/0074_ml_settings_cpu_embed_enabled.py create mode 100644 backend/app/tasks/gpu_queue.py diff --git a/alembic/versions/0074_ml_settings_cpu_embed_enabled.py b/alembic/versions/0074_ml_settings_cpu_embed_enabled.py new file mode 100644 index 0000000..48ff8ea --- /dev/null +++ b/alembic/versions/0074_ml_settings_cpu_embed_enabled.py @@ -0,0 +1,35 @@ +"""ml_settings.cpu_embed_enabled — the CPU embed fallback becomes a switch + +B3 (operator 2026-07-02): the ml-worker's only processing role is the CPU +whole-image embed for stacks without a GPU agent. ON by default (a fresh +install works agent-less); agent-equipped stacks that drop the ml-worker +container turn it off so import hooks stop queueing embed work into a queue +nothing consumes — the daily GPU 'embed' backfill covers those images. + +Revision ID: 0074 +Revises: 0073 +Create Date: 2026-07-02 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0074" +down_revision: Union[str, None] = "0073" +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( + "cpu_embed_enabled", sa.Boolean(), nullable=False, + server_default=sa.true(), + ), + ) + + +def downgrade() -> None: + op.drop_column("ml_settings", "cpu_embed_enabled") diff --git a/backend/app/api/gpu.py b/backend/app/api/gpu.py index bbbbecd..c6e0cd8 100644 --- a/backend/app/api/gpu.py +++ b/backend/app/api/gpu.py @@ -96,7 +96,7 @@ async def backfill(): """Enqueue a job for every image that doesn't already have one for `task`.""" body = await request.get_json(silent=True) or {} task = str(body.get("task") or "ccip") - from ..tasks.ml import enqueue_gpu_backfill + from ..tasks.gpu_queue import enqueue_gpu_backfill r = enqueue_gpu_backfill.delay(task) return jsonify({"celery_task_id": r.id, "task": task}), 202 @@ -109,7 +109,7 @@ async def reprocess(): detectors). Heavy — the back-catalogue is otherwise skipped by the backfills.""" body = await request.get_json(silent=True) or {} task = str(body.get("task") or "ccip") - from ..tasks.ml import reprocess_gpu_jobs + from ..tasks.gpu_queue import reprocess_gpu_jobs r = reprocess_gpu_jobs.delay(task) return jsonify({"celery_task_id": r.id, "task": task}), 202 diff --git a/backend/app/api/ml_admin.py b/backend/app/api/ml_admin.py index 5afeee8..62fbc26 100644 --- a/backend/app/api/ml_admin.py +++ b/backend/app/api/ml_admin.py @@ -9,6 +9,7 @@ ml_admin_bp = Blueprint("ml_admin", __name__, url_prefix="/api/ml") _EDITABLE = ( + "cpu_embed_enabled", "video_frame_interval_seconds", "video_max_frames", "head_min_positives", @@ -63,6 +64,7 @@ async def get_settings(): ).scalar_one() return jsonify( { + "cpu_embed_enabled": s.cpu_embed_enabled, "video_frame_interval_seconds": s.video_frame_interval_seconds, "video_max_frames": s.video_max_frames, "embedder_model_version": s.embedder_model_version, diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 0600988..f6d249e 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -29,6 +29,7 @@ def make_celery() -> Celery: "backend.app.tasks.thumbnail", "backend.app.tasks.maintenance", "backend.app.tasks.ml", + "backend.app.tasks.gpu_queue", "backend.app.tasks.download", "backend.app.tasks.external", "backend.app.tasks.backup", @@ -41,6 +42,11 @@ def make_celery() -> Celery: task_routes={ "backend.app.tasks.import_file.*": {"queue": "import"}, "backend.app.tasks.ml.*": {"queue": "ml"}, + # GPU-queue coordination (backfill enqueues, orphan recovery, + # reprocess) is pure DB work — it rides the maintenance quick lane + # so the GPU agent pipeline works even on stacks that drop the + # (now-optional, B3) ml-worker container entirely. + "backend.app.tasks.gpu_queue.*": {"queue": "maintenance"}, "backend.app.tasks.thumbnail.*": {"queue": "thumbnail"}, "backend.app.tasks.download.*": {"queue": "download"}, # External file-host fetches are downloads — same lane (they can run @@ -106,7 +112,7 @@ def make_celery() -> Celery: "schedule": 86400.0, # no-op unless head_auto_apply_enabled }, "recover-orphaned-gpu-jobs": { - "task": "backend.app.tasks.ml.recover_orphaned_gpu_jobs", + "task": "backend.app.tasks.gpu_queue.recover_orphaned_gpu_jobs", "schedule": 60.0, # quick pickup of work a dead agent orphaned }, "triage-gpu-errors": { @@ -114,17 +120,17 @@ def make_celery() -> Celery: "schedule": 900.0, # probe errored jobs' files → defect/file_ok }, "enqueue-ccip-backfill-hourly": { - "task": "backend.app.tasks.ml.enqueue_gpu_backfill", + "task": "backend.app.tasks.gpu_queue.enqueue_gpu_backfill", "schedule": 3600.0, # auto-feed NEW images; errored are "args": ("ccip",), # tombstoned — retry is the button only }, "enqueue-siglip-backfill-daily": { - "task": "backend.app.tasks.ml.enqueue_gpu_backfill", + "task": "backend.app.tasks.gpu_queue.enqueue_gpu_backfill", "schedule": 86400.0, # drain the concept-crop back-catalogue "args": ("siglip",), # (errored are tombstoned, not retried) }, "enqueue-embed-backfill-daily": { - "task": "backend.app.tasks.ml.enqueue_gpu_backfill", + "task": "backend.app.tasks.gpu_queue.enqueue_gpu_backfill", "schedule": 86400.0, # whole-image re-embed under the current "args": ("embed",), # model (an operator swap) drains via agent }, diff --git a/backend/app/models/ml_settings.py b/backend/app/models/ml_settings.py index e415d9f..a45cc3d 100644 --- a/backend/app/models/ml_settings.py +++ b/backend/app/models/ml_settings.py @@ -23,6 +23,15 @@ class MLSettings(Base): __table_args__ = (CheckConstraint("id = 1", name="singleton"),) id: Mapped[int] = mapped_column(Integer, primary_key=True) + # CPU whole-image embedding (B3, operator 2026-07-02). The ml-worker's ONLY + # processing role is the embed fallback for stacks WITHOUT a GPU agent — ON + # by default so a fresh install works with no agent. Stacks that run the + # agent and drop the ml-worker container turn this OFF so import hooks stop + # queueing embed work nothing will consume (the daily GPU 'embed' backfill + # covers those images instead). + cpu_embed_enabled: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=True + ) # Video embedding (#747). Sample one frame every N seconds (fixed CADENCE, not # a fixed count) so coverage reflects real screen time regardless of length; # cap the total so a long video can't explode into hundreds of embeds. The diff --git a/backend/app/services/cleanup_service.py b/backend/app/services/cleanup_service.py index 08f513c..8708e28 100644 --- a/backend/app/services/cleanup_service.py +++ b/backend/app/services/cleanup_service.py @@ -1008,7 +1008,7 @@ def reextract_archive_attachments( still an archive on disk, so the cursor is what guarantees forward progress. """ from ..models import ImportSettings, Post, PostAttachment, Source - from ..tasks.ml import tag_and_embed + from ..tasks.ml import cpu_embed_enabled, embed_image from ..tasks.thumbnail import generate_thumbnail from .archive_extractor import is_archive from .importer import Importer @@ -1089,10 +1089,12 @@ def reextract_archive_attachments( # Thumbnails + ML for the newly-imported members (best-effort; off the # critical path — a Redis hiccup must not fail the whole re-extract). + do_embed = cpu_embed_enabled() for img_id in enqueue_ids: try: generate_thumbnail.delay(img_id) - tag_and_embed.delay(img_id) + if do_embed: + embed_image.delay(img_id) except Exception as exc: log.warning("re-extract enqueue failed for image %s: %s", img_id, exc) return summary diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index a00b407..772fdc9 100644 --- a/backend/app/services/download_service.py +++ b/backend/app/services/download_service.py @@ -326,14 +326,16 @@ class DownloadService: # for hours after a download landed. Lazy import to avoid # circular-import risk between this service and the # tasks/* modules that import it. - from ..tasks.ml import tag_and_embed + from ..tasks.ml import cpu_embed_enabled, embed_image from ..tasks.thumbnail import generate_thumbnail + do_embed = cpu_embed_enabled() ids = list(result.member_image_ids) if result.image_id is not None and result.image_id not in ids: ids.append(result.image_id) for img_id in ids: generate_thumbnail.delay(img_id) - tag_and_embed.delay(img_id) + if do_embed: + embed_image.delay(img_id) elif result.status == "attached": # Non-media or extracted archive captured as PostAttachment # (FC-2d-iii). The canonical copy lives in the attachments diff --git a/backend/app/tasks/external.py b/backend/app/tasks/external.py index 5f4c2f4..327d89b 100644 --- a/backend/app/tasks/external.py +++ b/backend/app/tasks/external.py @@ -216,11 +216,13 @@ def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict: # Thumbnails + ML for any newly-attached images (mirrors the download # path). Lazy import to dodge a task-module import cycle. if image_ids: - from .ml import tag_and_embed + from .ml import cpu_embed_enabled, embed_image from .thumbnail import generate_thumbnail + do_embed = cpu_embed_enabled() for img_id in image_ids: generate_thumbnail.delay(img_id) - tag_and_embed.delay(img_id) + if do_embed: + embed_image.delay(img_id) return {"link_id": link_id, "files": len(result.files), "images": len(image_ids)} except Exception as exc: # never leave a link stuck in 'downloading' log.exception("external fetch task failed for link %s", link_id) diff --git a/backend/app/tasks/gpu_queue.py b/backend/app/tasks/gpu_queue.py new file mode 100644 index 0000000..3fee447 --- /dev/null +++ b/backend/app/tasks/gpu_queue.py @@ -0,0 +1,171 @@ +"""GPU-job queue coordination: backfill enqueues, orphan recovery, reprocess. + +These are pure-DB sweeps (INSERT…SELECT / UPDATE) — no torch, no sklearn — +that keep the desktop GPU agent's work queue fed and self-healing. They lived +in tasks/ml.py (routed to the 'ml' queue) purely by colocation, which made the +ml-worker container a hard dependency of the GPU pipeline; under B3 the +ml-worker is OPTIONAL (its only processing role is the CPU embed fallback), so +these moved here and route to the 'maintenance' quick lane with the other +recovery sweeps. A stack with no ml-worker keeps a fully-working GPU pipeline. +""" + +import logging + +from sqlalchemy import select + +from ..celery_app import celery +from ._sync_engine import sync_session_factory as _sync_session_factory + +log = logging.getLogger(__name__) + + +@celery.task(name="backend.app.tasks.gpu_queue.enqueue_gpu_backfill") +def enqueue_gpu_backfill(task_name: str) -> int: + """Enqueue a gpu_job for every image that still needs `task_name` (one + INSERT…SELECT, so it scales to a full library). The desktop agent drains the + queue over HTTP. Returns the number enqueued. + + Completion is judged PER PIPELINE, never across them (B3, operator + 2026-07-02): 'ccip' by prior gpu_job rows, 'siglip' by concept regions at + the current model version, and only 'embed' by image_record's whole-image + embedding — the one artifact the CPU fallback also produces. A CPU embed + therefore never closes crop/detect work for the agent. + + An ERRORED job is a tombstone for its (image, task): no variant re-enqueues + it. Retry is deliberate-only (/retry_errors), which also means an errored + back-catalogue needs one "Retry errored jobs" press after a model swap. + Before the tombstone rule, this loop re-minted a fresh doomed job for every + permanently-bad file each run — ~24 duplicate error rows/day per file (the + 2026-07-02 "unprocessable" flood).""" + from sqlalchemy import exists, insert, literal, or_ + from sqlalchemy import select as sa_select + + from ..models import GpuJob, ImageRecord, ImageRegion, MLSettings + from ..services.ml.gpu_jobs import error_dedupe_statements + + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + # Prune stale tombstones first (loop-era duplicates + rows made moot by + # a later success), so 'error' reads as one row per distinct failing + # file and the skip-guards below see a clean picture. + pruned = sum( + session.execute(s).rowcount or 0 for s in error_dedupe_statements() + ) + if pruned: + log.info("gpu backfill: pruned %d stale/duplicate error rows", pruned) + 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, + ) + # 'error' blocks too — tombstone rule, see docstring. + blocked = exists().where( + GpuJob.image_record_id == ImageRecord.id, + GpuJob.task == "embed", + GpuJob.status.in_(["pending", "leased", "error"]), + ) + sel = sa_select( + ImageRecord.id, literal("embed"), literal("pending") + ).where(stale).where(~blocked) + 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, + ) + # 'error' blocks too — tombstone rule, see docstring. + blocked = exists().where( + GpuJob.image_record_id == ImageRecord.id, + GpuJob.task == "siglip", + GpuJob.status.in_(["pending", "leased", "error"]), + ) + sel = sa_select( + ImageRecord.id, literal("siglip"), literal("pending") + ).where(~has_current_concept).where(~blocked) + else: + # ANY prior row blocks — including 'error' (tombstone rule, see + # docstring): pre-fix this branch ran HOURLY and was the loop. + already = exists().where( + GpuJob.image_record_id == ImageRecord.id, + GpuJob.task == task_name, + GpuJob.status.in_(["pending", "leased", "done", "error"]), + ) + sel = sa_select( + ImageRecord.id, literal(task_name), literal("pending") + ).where(~already) + # RETURNING + count: result.rowcount is unreliable for INSERT…SELECT. + rows = session.execute( + insert(GpuJob) + .from_select(["image_record_id", "task", "status"], sel) + .returning(GpuJob.id) + ).fetchall() + session.commit() + return len(rows) + + +@celery.task(name="backend.app.tasks.gpu_queue.recover_orphaned_gpu_jobs") +def recover_orphaned_gpu_jobs() -> int: + """Reset expired GPU-job leases back to pending — recovers work orphaned by an + agent that died mid-job (no graceful release) — and convert poison-loopers + (release/expiry cycles that never reach fail()'s attempt cap) to 'error'. + Statements are shared with GpuJobService.recover_orphaned so the sweep and + the service can't drift. Short beat cadence so orphans get picked back up + quickly + the queue counts read honestly. Returns the number recovered.""" + from datetime import UTC, datetime + + from ..services.ml.gpu_jobs import recover_statements + + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + counts = { + name: session.execute(stmt).rowcount or 0 + for name, stmt in recover_statements(datetime.now(UTC)).items() + } + session.commit() + if counts["poison_expired"] or counts["poison_pending"]: + log.warning( + "gpu jobs poisoned -> error: %d crash-loop (expired lease), " + "%d never-complete (pending)", + counts["poison_expired"], counts["poison_pending"], + ) + return counts["recovered"] + + +@celery.task(name="backend.app.tasks.gpu_queue.reprocess_gpu_jobs") +def reprocess_gpu_jobs(task_name: str = "ccip") -> int: + """Reset every done/error job of `task_name` back to pending so the agent + re-runs the WHOLE library under the CURRENT pipeline — e.g. after adding crop + detectors (#1202), re-cropping existing images. Heavy + operator-triggered; + the back-catalogue won't otherwise re-process (the backfills skip images that + already have current-version regions). Returns the number reset.""" + from datetime import UTC, datetime + + from sqlalchemy import update + + from ..models import GpuJob + + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + now = datetime.now(UTC) + res = session.execute( + update(GpuJob) + .where( + GpuJob.task == task_name, + GpuJob.status.in_(["done", "error"]), + ) + .values( + status="pending", attempts=0, lease_token=None, leased_at=None, + lease_expires_at=None, updated_at=now, + ) + ) + session.commit() + return res.rowcount or 0 diff --git a/backend/app/tasks/import_file.py b/backend/app/tasks/import_file.py index 49c798d..fcdf78e 100644 --- a/backend/app/tasks/import_file.py +++ b/backend/app/tasks/import_file.py @@ -228,15 +228,17 @@ def _do_import(session, task, import_task_id: int) -> dict: # Enqueue thumbnail + ML for newly imported AND superseded images # (a superseded row has cleared ML + no thumbnail). if result.status in ("imported", "superseded"): - from .ml import tag_and_embed + from .ml import cpu_embed_enabled, embed_image from .thumbnail import generate_thumbnail + do_embed = cpu_embed_enabled() ids = list(result.member_image_ids) if result.image_id is not None and result.image_id not in ids: ids.append(result.image_id) for img_id in ids: generate_thumbnail.delay(img_id) - tag_and_embed.delay(img_id) + if do_embed: + embed_image.delay(img_id) # If this was the last task in the batch, mark the batch complete. remaining = session.execute( diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index e403cf5..755cde8 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -121,7 +121,7 @@ IMPORT_BATCH_KEEP_DAYS = 30 # task.time_limit + a small buffer. task_name overrides take precedence # over queue overrides. # -# ml queue: tag_and_embed video branch (≈20 GPU ops); time_limit=1200. +# ml queue: embed_image video branch (≈20 GPU ops); time_limit=1200. # import_archive_file: shares the 'import' queue with the fast # single-file import_media_file, so it needs a task-name override # (the import queue itself stays at the 5-min default for single diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index 737c36c..b1650e2 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -1,8 +1,15 @@ """ML Celery tasks: per-image embedding, backfill discovery, head training, model self-heal. -All run on the ml-worker (queue 'ml'). Sync sessions (Celery workers are sync -processes), same pattern as FC-2a tasks. +All run on the ml-worker (queue 'ml'), which under B3 (2026-07-02) is an +OPTIONAL container: its only processing role is the CPU whole-image embed +fallback (gated by ml_settings.cpu_embed_enabled) for stacks without a GPU +agent — plus head training / auto-apply, which need sklearn/numpy and so +live on this image. GPU-queue coordination (backfill enqueues, orphan +recovery, reprocess) deliberately does NOT live here — see tasks/gpu_queue.py +(maintenance lane), so the agent pipeline works with no ml-worker at all. +Sync sessions (Celery workers are sync processes), same pattern as FC-2a +tasks. """ import logging @@ -26,8 +33,24 @@ def _is_video(path: Path) -> bool: return path.suffix.lower() in VIDEO_EXTS +def cpu_embed_enabled() -> bool: + """Dispatch gate for the CPU embed fallback (B3, operator 2026-07-02): + stacks that run a GPU agent and DROP the (optional) ml-worker container + turn ml_settings.cpu_embed_enabled off, so the import hooks stop queueing + embed work into a queue nothing consumes — the daily GPU 'embed' backfill + covers those images instead. Opens its own short session because the four + dispatch sites sit in different session scopes; defaults ON when the + settings row is missing (a fresh install must work agent-less).""" + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + val = session.execute( + select(MLSettings.cpu_embed_enabled).where(MLSettings.id == 1) + ).scalar_one_or_none() + return True if val is None else bool(val) + + @celery.task( - name="backend.app.tasks.ml.tag_and_embed", + name="backend.app.tasks.ml.embed_image", bind=True, autoretry_for=(OperationalError, DBAPIError, OSError), retry_backoff=5, @@ -44,13 +67,21 @@ def _is_video(path: Path) -> bool: soft_time_limit=900, # 15 min time_limit=1200, # 20 min hard ) -def tag_and_embed(self, image_id: int) -> dict: - """Compute + store one image's SigLIP embedding. +def embed_image(self, image_id: int) -> dict: + """Compute + store one image's whole-image SigLIP embedding — the CPU + fallback path (B3, operator 2026-07-02): this is the ml-worker's ONLY + processing role, keeping search/similarity/head-suggestions alive on + deployments without a GPU agent. Detection, cropping and CCIP are + deliberately agent-only, and their backfill predicates read image_region / + gpu_job state — never image_record.siglip_embedding — so a CPU whole-image + embed can NEVER mark crop work as done. (Renamed from tag_and_embed — + Camie tagging was retired #1189; the old name kept implying a tagging step + that no longer exists.) Video (#747): sample frames at a fixed cadence (ml_settings video_frame_interval_seconds, capped at video_max_frames) and mean-pool the - per-frame SigLIP embeddings. On no-frames returns status='no_frames' (not an - error). (Camie tagging was retired #1189 — heads + CCIP are the tag source.) + per-frame SigLIP embeddings — the same shape as the GPU agent's video + handling. On no-frames returns status='no_frames' (not an error). """ import time @@ -84,9 +115,9 @@ def tag_and_embed(self, image_id: int) -> dict: f"image_id={image_id} path={record.path} mime={record.mime} " f"bytes={record.size_bytes} video={is_vid}" ) - log.info("tag_and_embed start: %s", ctx) + log.info("embed_image start: %s", ctx) if not src.is_file(): - log.warning("tag_and_embed file missing on disk: %s", ctx) + log.warning("embed_image file missing on disk: %s", ctx) return {"status": "file_missing", "image_id": image_id} phase = "load_models" @@ -102,7 +133,7 @@ def tag_and_embed(self, image_id: int) -> dict: vprobe = safe_probe.probe_video(src) if not vprobe.ok: log.warning( - "tag_and_embed bad video (%s): %s", vprobe.reason, ctx + "embed_image bad video (%s): %s", vprobe.reason, ctx ) return { "status": "bad_video", "image_id": image_id, @@ -130,7 +161,7 @@ def tag_and_embed(self, image_id: int) -> dict: t0 = time.monotonic() embedding = embedder.infer(src) log.info( - "tag_and_embed embedded in %.1fs: %s", + "embed_image embedded in %.1fs: %s", time.monotonic() - t0, ctx, ) @@ -141,7 +172,7 @@ def tag_and_embed(self, image_id: int) -> dict: session.commit() except SoftTimeLimitExceeded: log.error( - "tag_and_embed TIMED OUT after %.0fs in phase=%s: %s", + "embed_image TIMED OUT after %.0fs in phase=%s: %s", _elapsed(), phase, ctx, ) # Re-raise as SoftTimeLimitExceeded (preserves the 'timeout' status in @@ -155,12 +186,12 @@ def tag_and_embed(self, image_id: int) -> dict: # ORIGINAL so the type is preserved; just make sure it's logged with # context first. log.exception( - "tag_and_embed FAILED in phase=%s after %.0fs: %s", + "embed_image FAILED in phase=%s after %.0fs: %s", phase, _elapsed(), ctx, ) raise - log.info("tag_and_embed ok in %.1fs: %s", _elapsed(), ctx) + log.info("embed_image ok in %.1fs: %s", _elapsed(), ctx) return {"status": "ok", "image_id": image_id} @@ -222,13 +253,17 @@ def _sample_video_frames( @celery.task(name="backend.app.tasks.ml.backfill", bind=True) def backfill(self) -> int: - """Enqueue tag_and_embed (embed-only) for images with no SigLIP embedding. + """Enqueue embed_image (embed-only) for images with no SigLIP embedding. Keyset pagination by id ASC (restart-safe). NB: a siglip MODEL-VERSION mismatch (an operator model swap, #1190) is NOT re-embedded here — the CPU ml-worker can't churn the library at 384/512px; the GPU agent owns version re-embeds via the 'embed' job. """ + if not cpu_embed_enabled(): + log.info("cpu backfill skipped: cpu_embed_enabled is off (B3 — the " + "GPU 'embed' backfill owns whole-image embeds on this stack)") + return 0 SessionLocal = _sync_session_factory() enqueued = 0 last_id = 0 @@ -244,7 +279,7 @@ def backfill(self) -> int: if not rows: break for image_id in rows: - tag_and_embed.delay(image_id) + embed_image.delay(image_id) enqueued += 1 last_id = rows[-1] return enqueued @@ -405,156 +440,6 @@ def scheduled_apply_head_tags() -> str: return "dispatched" -@celery.task(name="backend.app.tasks.ml.enqueue_gpu_backfill") -def enqueue_gpu_backfill(task_name: str) -> int: - """Enqueue a gpu_job for every image that still needs `task_name` (one - INSERT…SELECT, so it scales to a full library). The desktop agent drains the - queue over HTTP. Returns the number enqueued. - - 'siglip' gates on the RESULT (no concept region yet) rather than on a prior - job, so it picks up the back-catalogue of images that were CCIP-embedded - before concept crops existed — without re-touching their figure/CCIP regions. - - An ERRORED job is a tombstone for its (image, task): no variant re-enqueues - it. Retry is deliberate-only (/retry_errors), which also means an errored - back-catalogue needs one "Retry errored jobs" press after a model swap. - Before the tombstone rule, this loop re-minted a fresh doomed job for every - permanently-bad file each run — ~24 duplicate error rows/day per file (the - 2026-07-02 "unprocessable" flood).""" - from sqlalchemy import exists, insert, literal, or_ - from sqlalchemy import select as sa_select - - from ..models import GpuJob, ImageRecord, ImageRegion, MLSettings - from ..services.ml.gpu_jobs import error_dedupe_statements - - SessionLocal = _sync_session_factory() - with SessionLocal() as session: - # Prune stale tombstones first (loop-era duplicates + rows made moot by - # a later success), so 'error' reads as one row per distinct failing - # file and the skip-guards below see a clean picture. - pruned = sum( - session.execute(s).rowcount or 0 for s in error_dedupe_statements() - ) - if pruned: - log.info("gpu backfill: pruned %d stale/duplicate error rows", pruned) - 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, - ) - # 'error' blocks too — tombstone rule, see docstring. - blocked = exists().where( - GpuJob.image_record_id == ImageRecord.id, - GpuJob.task == "embed", - GpuJob.status.in_(["pending", "leased", "error"]), - ) - sel = sa_select( - ImageRecord.id, literal("embed"), literal("pending") - ).where(stale).where(~blocked) - 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, - ) - # 'error' blocks too — tombstone rule, see docstring. - blocked = exists().where( - GpuJob.image_record_id == ImageRecord.id, - GpuJob.task == "siglip", - GpuJob.status.in_(["pending", "leased", "error"]), - ) - sel = sa_select( - ImageRecord.id, literal("siglip"), literal("pending") - ).where(~has_current_concept).where(~blocked) - else: - # ANY prior row blocks — including 'error' (tombstone rule, see - # docstring): pre-fix this branch ran HOURLY and was the loop. - already = exists().where( - GpuJob.image_record_id == ImageRecord.id, - GpuJob.task == task_name, - GpuJob.status.in_(["pending", "leased", "done", "error"]), - ) - sel = sa_select( - ImageRecord.id, literal(task_name), literal("pending") - ).where(~already) - # RETURNING + count: result.rowcount is unreliable for INSERT…SELECT. - rows = session.execute( - insert(GpuJob) - .from_select(["image_record_id", "task", "status"], sel) - .returning(GpuJob.id) - ).fetchall() - session.commit() - return len(rows) - - -@celery.task(name="backend.app.tasks.ml.recover_orphaned_gpu_jobs") -def recover_orphaned_gpu_jobs() -> int: - """Reset expired GPU-job leases back to pending — recovers work orphaned by an - agent that died mid-job (no graceful release) — and convert poison-loopers - (release/expiry cycles that never reach fail()'s attempt cap) to 'error'. - Statements are shared with GpuJobService.recover_orphaned so the sweep and - the service can't drift. Short beat cadence so orphans get picked back up - quickly + the queue counts read honestly. Returns the number recovered.""" - from datetime import UTC, datetime - - from ..services.ml.gpu_jobs import recover_statements - - SessionLocal = _sync_session_factory() - with SessionLocal() as session: - counts = { - name: session.execute(stmt).rowcount or 0 - for name, stmt in recover_statements(datetime.now(UTC)).items() - } - session.commit() - if counts["poison_expired"] or counts["poison_pending"]: - log.warning( - "gpu jobs poisoned -> error: %d crash-loop (expired lease), " - "%d never-complete (pending)", - counts["poison_expired"], counts["poison_pending"], - ) - return counts["recovered"] - - -@celery.task(name="backend.app.tasks.ml.reprocess_gpu_jobs") -def reprocess_gpu_jobs(task_name: str = "ccip") -> int: - """Reset every done/error job of `task_name` back to pending so the agent - re-runs the WHOLE library under the CURRENT pipeline — e.g. after adding crop - detectors (#1202), re-cropping existing images. Heavy + operator-triggered; - the back-catalogue won't otherwise re-process (the backfills skip images that - already have current-version regions). Returns the number reset.""" - from datetime import UTC, datetime - - from sqlalchemy import update - - from ..models import GpuJob - - SessionLocal = _sync_session_factory() - with SessionLocal() as session: - now = datetime.now(UTC) - res = session.execute( - update(GpuJob) - .where( - GpuJob.task == task_name, - GpuJob.status.in_(["done", "error"]), - ) - .values( - status="pending", attempts=0, lease_token=None, leased_at=None, - lease_expires_at=None, updated_at=now, - ) - ) - session.commit() - return res.rowcount or 0 - - @celery.task( name="backend.app.tasks.ml.scheduled_ccip_auto_apply", soft_time_limit=1800, time_limit=2100, diff --git a/frontend/src/components/settings/MLBackfillCard.vue b/frontend/src/components/settings/MLBackfillCard.vue index d0e9c21..3becc59 100644 --- a/frontend/src/components/settings/MLBackfillCard.vue +++ b/frontend/src/components/settings/MLBackfillCard.vue @@ -1,16 +1,33 @@