diff --git a/agent/fc_agent/app.py b/agent/fc_agent/app.py index 93795d4..f743e84 100644 --- a/agent/fc_agent/app.py +++ b/agent/fc_agent/app.py @@ -21,7 +21,7 @@ log = logging.getLogger("fc_agent.app") # Bump on every agent change. The page embeds this and /status reports it; the UI # warns to reload when they differ — so a stale browser-cached page can't be # mistaken for "the new image didn't deploy". (Belt-and-braces with no-store.) -VERSION = "2026-07-02.4 · bandwidth governor: aggregate download cap (MB/s dial) so the agent can't saturate the desktop's network" +VERSION = "2026-07-02.5 · cap-aware autoscaler: downloaders stop growing (and shed) when the bandwidth cap — not concurrency — is the bottleneck" logbuf.install() cfg = Config.from_env() @@ -344,7 +344,8 @@ _PAGE = """ // Auto on → dial reflects the auto-chosen count (read-only); off → manual. if(document.activeElement!==autochk) autochk.checked=!!s.auto conc.disabled=!!s.auto; conc.style.opacity=s.auto?0.55:1 - conchint.textContent=s.auto?('auto-tuning downloaders to keep the GPU fed · max '+CAP):('manual downloaders · max '+CAP) + conchint.textContent=(s.auto?('auto-tuning downloaders to keep the GPU fed · max '+CAP):('manual downloaders · max '+CAP)) + +(s.bw_capped?' · holding at the bandwidth cap (more downloaders would not go faster)':'') if(document.activeElement!==conc) conc.value=s.concurrency conc.max=CAP // Connection pill + queue come only from the /status poll (the Start/Stop POST diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py index 4554703..8c65668 100644 --- a/agent/fc_agent/worker.py +++ b/agent/fc_agent/worker.py @@ -124,6 +124,17 @@ OCC_LOW = 0.25 # below this = buffer starving → add a downloade OCC_HIGH = 0.80 # above this = downloaders outpace the GPU UTIL_ALPHA = 0.25 # GPU-util EWMA weight UTIL_START = 85 # GPU has headroom below this (gate a 2nd consumer) +# Bandwidth-cap awareness (operator 2026-07-02): with the aggregate governor in +# place, the occupancy signal alone would peg downloaders at DL_MAX while the +# CAP — not concurrency — is the real constraint: 8 streams sharing 8 MB/s move +# no more data than 4, they just hold more leases + RAM and stretch every job's +# latency. So growth is gated on the pipe having headroom, and a pipe pinned at +# the cap sheds streams down to BW_MIN_DL (enough overlap to keep the cap +# filled through TTFB + decode gaps). The dead band between the two thresholds +# prevents add/trim flapping. +BW_ADD_HEADROOM = 0.85 # add a downloader only while net < 85% of the cap +BW_TRIM_AT = 0.95 # net ≥ 95% of the cap → shed toward BW_MIN_DL +BW_MIN_DL = 3 VRAM_HI = 0.90 # memory pressure → shed a consumer VRAM_GROW_MAX = 0.82 # don't add a consumer past this VRAM TPUT_ALPHA = 0.5 # throughput EWMA weight @@ -225,6 +236,7 @@ class Worker: self._jpm = 0.0 self._dpm = 0.0 self._net_mb_s = 0.0 # smoothed aggregate download rate (UI readout) + self._bw_capped = False # autoscaler is holding/shedding at the cap (UI) self._util_smooth: float | None = None # EWMA GPU util (set by control loop) # Curator queue snapshot, refreshed by a background poller so the UI # /status read is instant — never an inline curator HTTP call (which @@ -584,6 +596,7 @@ class Worker: "transient": self.transient, "bandwidth_limit_mb_s": round(self.throttle.rate / 1_048_576, 1), "net_mb_s": round(self._net_mb_s, 1), # observed aggregate rate + "bw_capped": self._bw_capped, # autoscaler holding at the cap (UI hint) } def _bump(self, *, processed=0, downloaded=0, errors=0, active=0, transient=0): @@ -982,10 +995,19 @@ class Worker: tick = 0 con_grew = False self._util_smooth = None + self._bw_capped = False continue occ = self._buffer.qsize() / BUFFER_MAX occ_ewma = _ewma(occ_ewma, occ, OCC_ALPHA) + # Bandwidth-cap position: compare the observed aggregate (the same + # EWMA the UI shows) against the governor's cap. `soft` gates + # growth; `hard` sheds streams (see BW_* rationale above). + bw_rate = self.throttle.rate + net_bytes = self._net_mb_s * 1_048_576 + bw_soft = bw_rate > 0 and net_bytes >= BW_ADD_HEADROOM * bw_rate + bw_hard = bw_rate > 0 and net_bytes >= BW_TRIM_AT * bw_rate + self._bw_capped = bw_soft g = gpumod.read_gpu() or {} mt = g.get("mem_total_mb") or 0 vram = (g.get("mem_used_mb", 0) / mt) if mt else 0.0 @@ -1022,8 +1044,15 @@ class Worker: self._apply_downloaders(-1) con_grew = False elif occ_ewma < OCC_LOW: - # Buffer starving → GPU idle waiting on downloads → add a feeder. - self._apply_downloaders(+1) + # Buffer starving → downloads are the bottleneck. WHICH kind + # decides the move: a pipe pinned at the bandwidth cap gains + # nothing from more streams (they'd split the same budget and + # stretch per-job latency) — shed toward BW_MIN_DL; with cap + # headroom, concurrency is genuinely short — add a feeder. + if bw_hard and self._dl_target > BW_MIN_DL: + self._apply_downloaders(-1) + elif not bw_soft: + self._apply_downloaders(+1) con_grew = False elif occ_ewma > OCC_HIGH: # Downloaders outpace the GPU. Prefer helping the GPU (add a 2nd @@ -1049,7 +1078,9 @@ class Worker: if self._dl_target != d0 or self._consumer_target != c0: log.info( "autoscale: dl %d→%d · consumers %d→%d " - "(buf %d%% · util~%d%% · %.2f j/s · vram %d%%)", + "(buf %d%% · util~%d%% · %.2f j/s · vram %d%% · " + "net %.1f MB/s%s)", d0, self._dl_target, c0, self._consumer_target, round(occ_ewma * 100), round(util_ewma), tput_ewma, - round(vram * 100)) + round(vram * 100), self._net_mb_s, + " — at cap" if bw_soft else "") 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/admin.py b/backend/app/api/admin.py index 4bbb12e..317265e 100644 --- a/backend/app/api/admin.py +++ b/backend/app/api/admin.py @@ -276,18 +276,48 @@ async def posts_reconcile_duplicates(): return await _run_dry_run_op(reconcile_duplicate_posts, source_id=source_id) +def _reset_content_confirm_token(projection: dict) -> str: + """Stable 8-hex token derived from the live counts (mirrors the Tier-C + bulk-delete token): it changes whenever the data changes, so the apply can + only ever run against numbers the operator just previewed.""" + canon = f"reset-content:{projection.get('count')}:{projection.get('applications')}" + return hashlib.sha256(canon.encode("utf-8")).hexdigest()[:8] + + @admin_bp.route("/tags/reset-content", methods=["POST"]) async def tags_reset_content(): - """Tier-A: delete ALL general + character tags (the Camie-suggestable - content vocabulary) so the operator can re-tag from scratch via - auto-suggest. fandom + series tags + series_page ordering are preserved, - and image_prediction rows are untouched so suggestions repopulate. - dry-run preview returns per-kind counts + applications + a sample so the - UI shows exactly what'll go before the operator confirms (dry_run=false). - Irreversible except via DB backup restore.""" + """Full-instance reset of the CONTENT vocabulary: deletes ALL general + + character tags and their image applications — INCLUDING the examples the + tagging heads learned from. Suggestions do NOT repopulate on their own + (the Camie predictions that once did are long retired): the operator + re-tags from scratch and the heads retrain from the new signal. fandom + + series tags + series_page ordering are preserved. + + Deliberately Tier-C-gated despite the Tier-A shape (operator 2026-07-02: + the full reset stays, but behind extra steps): dry_run returns the + projection + a `confirm` token derived from the live counts; the apply + must echo that token back or it is rejected.""" from ..services.cleanup_service import reset_content_tagging - return await _run_dry_run_op(reset_content_tagging) + body = await request.get_json(silent=True) or {} + dry_run = bool(body.get("dry_run", False)) + async with get_session() as session: + projection = await session.run_sync( + lambda s: reset_content_tagging(s, dry_run=True) + ) + token = _reset_content_confirm_token(projection) + if dry_run: + projection["confirm"] = token + return jsonify(projection) + if str(body.get("confirm", "")) != token: + return _bad( + "confirm_mismatch", + detail="run a fresh preview and echo its confirm token", + ) + result = await session.run_sync( + lambda s: reset_content_tagging(s, dry_run=False) + ) + return jsonify(result) @admin_bp.route("/tags/normalize", methods=["POST"]) 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..74e2403 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -7,7 +7,7 @@ Queues: download — gallery-dl tasks (FC-3) scan — periodic source checks (FC-3) — kept separate so long imports don't starve the scheduler - maintenance — pHash recomputation, centroid rebuild, etc. (FC-2/FC-3) + maintenance — recovery sweeps, pHash backfill, GPU-queue coordination, etc. default — anything not explicitly routed """ @@ -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/audits/single_color.py b/backend/app/services/audits/single_color.py index 167e078..6bddf07 100644 --- a/backend/app/services/audits/single_color.py +++ b/backend/app/services/audits/single_color.py @@ -1,8 +1,9 @@ """Single-color audit: matches images where one color dominates beyond -the threshold (within the given Euclidean RGB tolerance). The first -canonical implementation — the import-side filter (SkipReason.single_color) -was never wired; FC-Cleanup's audit module is the source of truth and a -future spec can adopt it on the import path too. +the threshold (within the given Euclidean RGB tolerance). The canonical +predicate for BOTH surfaces: FC-Cleanup's retroactive audit and — since +2026-07-02 — the import-side filter (Importer._single_color_hit / +SkipReason.single_color), so what the audit flags and what the import +skips can never disagree. """ from PIL import Image diff --git a/backend/app/services/cleanup_service.py b/backend/app/services/cleanup_service.py index 5e36898..8708e28 100644 --- a/backend/app/services/cleanup_service.py +++ b/backend/app/services/cleanup_service.py @@ -726,7 +726,10 @@ RESETTABLE_TAG_KINDS = ("general", "character") def reset_content_tagging(session: Session, *, dry_run: bool = False) -> dict: """Count (dry_run) or DELETE every general + character tag so the operator - can re-tag from scratch (heads/CCIP repopulate suggestions). + can re-tag from scratch. NB: the deleted applications include the tagging + heads' training positives — suggestions do NOT repopulate on their own; the + heads retrain from whatever the operator re-tags. (The API route gates the + live run behind a preview-derived confirm token for exactly this reason.) PRESERVED: fandom + series tags and their series_page ordering. CASCADE on image_tag / tag_alias / tag_suggestion_rejection clears each deleted tag's @@ -1005,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 @@ -1086,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/services/importer.py b/backend/app/services/importer.py index 8afd219..2459836 100644 --- a/backend/app/services/importer.py +++ b/backend/app/services/importer.py @@ -44,6 +44,7 @@ from ..utils.sidecar import find_sidecar, parse_sidecar from ..utils.slug import slugify from .archive_extractor import extract_archive, is_archive from .attachment_store import AttachmentStore +from .audits import single_color from .link_extract import extract_external_links from .thumbnailer import Thumbnailer @@ -790,6 +791,13 @@ class Importer: error=f"{pct:.2%} transparent", ) + if self.settings.skip_single_color and self._single_color_hit(source): + return ImportResult( + status="skipped", skip_reason=SkipReason.single_color, + error=(f"one color dominates >" + f"{self.settings.single_color_threshold:.0%}"), + ) + # Artist anchored to the attribution path (folder→artist), resolved # UP-FRONT so the enrich-on-duplicate branches link provenance with the # right artist even when the sidecar carries none — which is now the norm @@ -1123,6 +1131,13 @@ class Importer: status="skipped", skip_reason=SkipReason.too_transparent, error=f"{pct:.2%} transparent", ) + + if self.settings.skip_single_color and self._single_color_hit(path): + return ImportResult( + status="skipped", skip_reason=SkipReason.single_color, + error=(f"one color dominates >" + f"{self.settings.single_color_threshold:.0%}"), + ) else: # Best-effort probe for dims + duration so downloaded videos can dedup # (#871). LENIENT: unlike _import_media this path does not reject on a @@ -1538,6 +1553,33 @@ class Importer: # Benign orphan; the DB swap already committed. Don't undo it. pass + # Matches the Cleanup audit card's default tolerance: the import-side + # filter and the retroactive audit must agree on what "single color" MEANS + # (Euclidean RGB distance to the dominant color); only the match threshold + # is operator-tunable per surface. + _SINGLE_COLOR_TOLERANCE = 30 + + def _single_color_hit(self, source: Path) -> bool: + """True when one color dominates beyond the configured threshold — the + same canonical predicate the Cleanup audit runs (audits.single_color, + whose docstring anticipated this adoption; the skip_single_color + setting existed but was never wired until 2026-07-02). Never raises: + unreadable files were already rejected by verify() upstream, and a + residual decode error just declines to match (the import proceeds).""" + try: + with Image.open(source) as im: + if getattr(im, "is_animated", False): + # Frame 0 only would misjudge animations; skip like the + # transparency check does. + return False + return single_color.evaluate( + im, + threshold=self.settings.single_color_threshold, + tolerance=self._SINGLE_COLOR_TOLERANCE, + ) + except Exception: + return False + def _transparency_pct(self, source: Path) -> float: """Fraction of fully-transparent pixels in the image. 0.0 if no alpha. diff --git a/backend/app/services/ml/__init__.py b/backend/app/services/ml/__init__.py index 82ab712..723ddb4 100644 --- a/backend/app/services/ml/__init__.py +++ b/backend/app/services/ml/__init__.py @@ -1 +1,3 @@ -"""ML pipeline services: tagger, embedder, suggestions, centroids, allowlist, aliases.""" +"""ML pipeline services: embedders, heads (the learning suggester), suggestions, +GPU-job queue + failure triage, CCIP characters, crops/regions, allowlist and +aliases.""" 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..f9549d9 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 @@ -139,10 +139,9 @@ QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = { "download": 30, # Audit 2026-06-02 — maintenance/scan queues run tasks that # legitimately exceed the 5-min default (verify_integrity at 70m - # hard, scan_directory at 70m hard, apply_allowlist_tags / - # recompute_centroids / backfill_phash at 35m hard). 75 min lives - # above the longest of those and the per-task overrides below - # cover the outliers (backups, library audit). + # hard, scan_directory at 70m hard, backfill_phash at 35m hard). + # 75 min lives above the longest of those and the per-task + # overrides below cover the outliers (backups, library audit). "maintenance": 75, "scan": 75, } 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/DangerZoneCard.vue b/frontend/src/components/settings/DangerZoneCard.vue new file mode 100644 index 0000000..a876a7d --- /dev/null +++ b/frontend/src/components/settings/DangerZoneCard.vue @@ -0,0 +1,122 @@ + + + + + diff --git a/frontend/src/components/settings/DownloadsActivityPanel.vue b/frontend/src/components/settings/DownloadsActivityPanel.vue new file mode 100644 index 0000000..a3d379f --- /dev/null +++ b/frontend/src/components/settings/DownloadsActivityPanel.vue @@ -0,0 +1,77 @@ + + + + + diff --git a/frontend/src/components/settings/GpuActivityPanel.vue b/frontend/src/components/settings/GpuActivityPanel.vue new file mode 100644 index 0000000..7378309 --- /dev/null +++ b/frontend/src/components/settings/GpuActivityPanel.vue @@ -0,0 +1,110 @@ + + + + + diff --git a/frontend/src/components/settings/HeadsCard.vue b/frontend/src/components/settings/HeadsCard.vue index b9e684b..7117b0d 100644 --- a/frontend/src/components/settings/HeadsCard.vue +++ b/frontend/src/components/settings/HeadsCard.vue @@ -2,7 +2,7 @@

diff --git a/frontend/src/components/settings/ImportFiltersForm.vue b/frontend/src/components/settings/ImportFiltersForm.vue index c538b35..3210d9e 100644 --- a/frontend/src/components/settings/ImportFiltersForm.vue +++ b/frontend/src/components/settings/ImportFiltersForm.vue @@ -90,10 +90,14 @@ - - diff --git a/frontend/src/components/settings/ImportTriggerPanel.vue b/frontend/src/components/settings/ImportTriggerPanel.vue deleted file mode 100644 index 8873d92..0000000 --- a/frontend/src/components/settings/ImportTriggerPanel.vue +++ /dev/null @@ -1,97 +0,0 @@ - - - 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 @@