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