From 7c19ad91ed2ae5a4a2b1f5996a96c9173ef37c8c Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 2 Jul 2026 16:14:48 -0400 Subject: [PATCH 1/6] feat: cap-aware autoscaler + token-gated whole-instance tag reset (operator feedback) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Autoscaler (agent 2026-07-02.5): the buffer-occupancy signal alone would peg downloaders at DL_MAX while the bandwidth CAP — not concurrency — is the real constraint (8 streams sharing 8 MB/s move no more data than 4). Growth is now gated on the pipe having headroom (net < 85% of cap) and a pipe pinned at the cap (>= 95%) sheds streams down to 3; dead band prevents flapping. The UI hint says 'holding at the bandwidth cap' and /status reports bw_capped, so the behavior is legible without tests that need the ML stack. Reset content tagging: stays a FULL-instance reset (operator's call), but now lives in a fenced 'Danger zone' section on Cleanup and the apply is gated by a preview-derived confirm token (mirrors the Tier-C bulk-delete pattern — stale counts are rejected server-side). Copy no longer claims suggestions repopulate: it says plainly the heads' training examples are deleted and re-tagging starts fresh. Moved out of TagMaintenanceCard into DangerZoneCard. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM --- agent/fc_agent/app.py | 5 +- agent/fc_agent/worker.py | 39 +++++- backend/app/api/admin.py | 46 +++++-- backend/app/services/cleanup_service.py | 5 +- .../components/settings/DangerZoneCard.vue | 122 ++++++++++++++++++ .../settings/TagMaintenanceCard.vue | 60 --------- frontend/src/stores/admin.js | 6 +- frontend/src/views/CleanupView.vue | 20 +++ tests/test_api_admin.py | 32 ++++- 9 files changed, 257 insertions(+), 78 deletions(-) create mode 100644 frontend/src/components/settings/DangerZoneCard.vue 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/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/services/cleanup_service.py b/backend/app/services/cleanup_service.py index 5e36898..08f513c 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 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/TagMaintenanceCard.vue b/frontend/src/components/settings/TagMaintenanceCard.vue index 81db5fa..c4869ef 100644 --- a/frontend/src/components/settings/TagMaintenanceCard.vue +++ b/frontend/src/components/settings/TagMaintenanceCard.vue @@ -42,56 +42,6 @@ - -

- Deletes every general and character tag and - removes them from every image, so you can re-tag from scratch with the - auto-suggest. Fandoms and series (with their page order) are - kept, and each image's saved predictions are untouched — open - an image and its suggestions reappear. -

- - Irreversible — there's no undo except restoring a DB backup. - Back one up first (Settings → Maintenance → Backup). - - - Preview content-tag reset - -
-

- {{ resetPreview.count }} content tag(s) - - ({{ k }}: {{ n }})  - - across {{ resetPreview.applications }} image - application(s). -

- - Delete {{ resetPreview.count }} content tag(s) + - {{ resetPreview.applications }} application(s) -
- - ({ count: 0, sample_names: r.sample_names || [] }), }) -// Reset content tagging (general + character). -const { - previewData: resetPreview, previewing: loadingResetPreview, - committing: resetCommitting, runPreview: onResetPreview, runCommit: onResetCommit, -} = usePreviewCommit({ - preview: () => store.resetContentTagging({ dryRun: true }), - commit: () => store.resetContentTagging({ dryRun: false }), - emptyPreview: { count: 0, by_kind: {}, applications: 0, sample_names: [] }, -}) - // Standardize casing. The apply DISPATCHES a self-resuming background task (no // poll-until-done — that would falsely report complete after the first chunk), // so there's no emptyPreview: leave the projection up; a truthy normResult means diff --git a/frontend/src/stores/admin.js b/frontend/src/stores/admin.js index 94b8270..99743cd 100644 --- a/frontend/src/stores/admin.js +++ b/frontend/src/stores/admin.js @@ -101,8 +101,10 @@ export const useAdminStore = defineStore('admin', () => { }) } - // Destructive: deletes ALL general + character tags so the operator can - // re-tag from scratch via auto-suggest. fandom + series preserved. + // Destructive whole-instance reset: deletes ALL general + character tags AND + // their applications (the heads' training data included) — fandom + series + // preserved. dry-run returns a `confirm` token; the apply must pass it back + // ({ dryRun: false, confirm }) or the server rejects it. function resetContentTagging(opts = {}) { return _dryRunPost('/api/admin/tags/reset-content', opts) } diff --git a/frontend/src/views/CleanupView.vue b/frontend/src/views/CleanupView.vue index e45a4b5..999d5fe 100644 --- a/frontend/src/views/CleanupView.vue +++ b/frontend/src/views/CleanupView.vue @@ -39,6 +39,17 @@ + +
+

Danger zone

+

+ Whole-instance resets. Each needs a fresh preview and a typed + confirmation code. +

+
+ +
+
@@ -50,6 +61,7 @@ import PostMaintenanceCard from '../components/settings/PostMaintenanceCard.vue' import VideoDedupCard from '../components/settings/VideoDedupCard.vue' import GatedPurgeCard from '../components/settings/GatedPurgeCard.vue' import TagMaintenanceCard from '../components/settings/TagMaintenanceCard.vue' +import DangerZoneCard from '../components/settings/DangerZoneCard.vue' diff --git a/tests/test_api_admin.py b/tests/test_api_admin.py index 13e077c..8b9fcb7 100644 --- a/tests/test_api_admin.py +++ b/tests/test_api_admin.py @@ -562,7 +562,7 @@ async def test_trigger_normalize_live_queues(client, monkeypatch): @pytest.mark.asyncio -async def test_reset_content_tagging_dry_run_returns_counts(client, db): +async def test_reset_content_tagging_dry_run_returns_counts_and_token(client, db): db.add_all([ Tag(name="solo", kind=TagKind.general), Tag(name="naruto", kind=TagKind.character), @@ -579,3 +579,33 @@ async def test_reset_content_tagging_dry_run_returns_counts(client, db): assert body["by_kind"] == {"general": 1, "character": 1} # dry-run leaves the rows in place — fandom + series untouched too. assert "deleted" not in body + # The preview arms the apply: an 8-hex confirm token over the live counts. + assert len(body["confirm"]) == 8 + + +@pytest.mark.asyncio +async def test_reset_content_tagging_apply_requires_confirm_token(client, db): + from sqlalchemy import func, select + + db.add(Tag(name="solo", kind=TagKind.general)) + await db.commit() + + # No token / wrong token → rejected, nothing deleted. + for bad in ({"dry_run": False}, {"dry_run": False, "confirm": "deadbeef"}): + resp = await client.post("/api/admin/tags/reset-content", json=bad) + assert resp.status_code == 400 + remaining = (await db.execute( + select(func.count()).select_from(Tag) + )).scalar_one() + assert remaining == 1 + + # The token from a fresh preview unlocks the apply. + token = (await (await client.post( + "/api/admin/tags/reset-content", json={"dry_run": True} + )).get_json())["confirm"] + resp = await client.post( + "/api/admin/tags/reset-content", + json={"dry_run": False, "confirm": token}, + ) + assert resp.status_code == 200 + assert (await resp.get_json())["deleted"] == 1 -- 2.52.0 From 19b962f1a756aa432d3497003154919469c88c9c Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 2 Jul 2026 16:53:08 -0400 Subject: [PATCH 2/6] =?UTF-8?q?feat(b3):=20ml-worker=20becomes=20optional?= =?UTF-8?q?=20=E2=80=94=20embed-only=20role,=20decoupled=20GPU=20coordinat?= =?UTF-8?q?ion,=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 @@ - - 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/MaintenancePanel.vue b/frontend/src/components/settings/MaintenancePanel.vue index 69acbcc..26cae1f 100644 --- a/frontend/src/components/settings/MaintenancePanel.vue +++ b/frontend/src/components/settings/MaintenancePanel.vue @@ -1,36 +1,59 @@