From 8f6547f8d7d8ecfefeee9c6e873310106a97c2b2 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 6 Jul 2026 14:15:19 -0400 Subject: [PATCH 1/8] refactor(subscriptions): remove the dead backfill dry-run preview (#1281) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PreviewDialog.vue was orphaned — nothing mounted it — so the dry-run preview was unreachable. Rather than wire it up, remove the whole chain: its unique value (a capped 3-page count of what a backfill would grab) is low, and its adjacent needs are already covered — auth validation by verify_source_credential, and actually fetching recent posts by "Check now". Operator decision 2026-07-06. Removed: - frontend: PreviewDialog.vue + sources store previewSource() - backend: POST /api/sources//preview route, download_backends.preview_source, IngestCore.preview() + its now-unused NativeIngestError import - tests: the 3 ingester preview tests Nothing else referenced the chain (verified). Shared campaign-resolution and ledger helpers stay — they're used by run()/verify_source_credential. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM --- backend/app/api/sources.py | 46 ------- backend/app/services/download_backends.py | 44 ------- backend/app/services/ingest_core.py | 67 ---------- .../subscriptions/PreviewDialog.vue | 114 ------------------ frontend/src/stores/sources.js | 7 -- tests/test_patreon_ingester.py | 51 -------- 6 files changed, 329 deletions(-) delete mode 100644 frontend/src/components/subscriptions/PreviewDialog.vue diff --git a/backend/app/api/sources.py b/backend/app/api/sources.py index 44227cb..2463ea2 100644 --- a/backend/app/api/sources.py +++ b/backend/app/api/sources.py @@ -230,52 +230,6 @@ async def set_backfill(source_id: int): return jsonify(record.to_dict()) -@sources_bp.route("//preview", methods=["POST"]) -async def preview_source_endpoint(source_id: int): - """Plan #708 B4: dry-run — count what a backfill WOULD download for a native - platform (Patreon today), without downloading. Walks the first few feed pages - and counts media not already in the seen/dead ledgers. Returns - {total_new, posts_scanned, pages_scanned, has_more, sample[]} or 409 + reason - (unresolvable campaign id / auth / drift). 400 for gallery-dl platforms (no - cheap dry-run — their verify is a slow --simulate).""" - from pathlib import Path - - from ..services.credential_service import CredentialService - from ..services.download_backends import preview_source, uses_native_ingester - from ..tasks._sync_engine import sync_session_factory - from .credentials import _get_crypto - - async with get_session() as session: - rec = await SourceService(session).get(source_id) - if rec is None: - return _bad("not_found", status=404) - if not uses_native_ingester(rec.platform): - return _bad( - "unsupported", - detail="Preview is only available for native-ingester platforms.", - status=400, - ) - cred = CredentialService(session, _get_crypto()) - cookies_path = await cred.get_cookies_path(rec.platform) - auth_token = await cred.get_token(rec.platform) - - # The walk + ledger reads are sync (run off the request loop); the process - # sync engine is the same one the download task uses. - result = await preview_source( - platform=rec.platform, - url=rec.url, - source_id=source_id, - config_overrides=rec.config_overrides or {}, - cookies_path=str(cookies_path) if cookies_path else None, - images_root=Path("/images"), - sync_session_factory=sync_session_factory(), - auth_token=auth_token, - ) - if "error" in result: - return _bad("preview_failed", detail=result["error"], status=409) - return jsonify(result) - - @sources_bp.route("//check", methods=["POST"]) async def check_source(source_id: int): """FC-3c: enqueue a download for this source. diff --git a/backend/app/services/download_backends.py b/backend/app/services/download_backends.py index 9d799ef..c6e1d2e 100644 --- a/backend/app/services/download_backends.py +++ b/backend/app/services/download_backends.py @@ -24,7 +24,6 @@ import asyncio from pathlib import Path from .gallery_dl import DownloadResult, ErrorType -from .native_ingest_common import NativeIngestError from .patreon_ingester import PatreonIngester from .patreon_resolver import extract_vanity, resolve_campaign_id_for_source from .pixiv_client import user_id_from_url @@ -204,49 +203,6 @@ async def _run_native_ingester( return dl_result, resolved_campaign_id -async def preview_source( - *, - platform: str, - url: str, - source_id: int, - config_overrides: dict | None, - cookies_path: str | None, - images_root: Path, - sync_session_factory, - auth_token: str | None = None, - page_limit: int = 3, -) -> dict: - """Dry-run preview for a native platform (plan #708 B4): resolve the campaign - id, then walk a few pages counting media not already seen/dead — no download. - - Returns the preview dict (total_new / posts_scanned / pages_scanned / - has_more / sample), or `{"error": msg}` on a resolve / auth / drift failure. - Native-only — the caller gates on `uses_native_ingester`. - """ - import asyncio - - campaign_id, _ = await _resolve_native_campaign_id( - platform, url, cookies_path, config_overrides or {} - ) - if not campaign_id: - return {"error": _campaign_resolution_error(platform, url)} - ingester = _native_ingester_cls(platform)( - images_root=images_root, - cookies_path=cookies_path, - session_factory=sync_session_factory, - auth_token=auth_token, - ) - loop = asyncio.get_running_loop() - try: - result = await loop.run_in_executor( - None, - lambda: ingester.preview(source_id, campaign_id, page_limit=page_limit), - ) - except NativeIngestError as exc: - return {"error": f"Couldn't preview: {exc}"} - return result - - async def verify_source_credential( *, platform: str, diff --git a/backend/app/services/ingest_core.py b/backend/app/services/ingest_core.py index 9b5d391..38f7eb7 100644 --- a/backend/app/services/ingest_core.py +++ b/backend/app/services/ingest_core.py @@ -577,73 +577,6 @@ class Ingester: error_type=None, error_message=None, ) - # -- preview (dry-run) ------------------------------------------------- - - def preview( - self, - source_id: int, - campaign_id: str, - *, - page_limit: int = 3, - sample_size: int = 10, - ) -> dict: - """Dry-run (plan #708 B4): walk up to `page_limit` pages and count media - NOT already in the seen/dead ledgers, WITHOUT downloading anything. - - Read-only — only the seen/dead SELECTs touch the DB (short sessions). Lets - an operator gauge "is this source worth a backfill?" cheaply. Returns: - {total_new, posts_scanned, pages_scanned, has_more, - sample: [{title, date, new}, ...]} # sample = posts with new media - A client-level failure (auth/drift) propagates to the caller. - """ - total_new = 0 - posts_scanned = 0 - pages_scanned = 0 - has_more = False - sample: list[dict] = [] - unset = object() - last_page: object = unset - # #874: same gated-post gate as run() — the preview must not count - # blurred locked-preview media as "new", or it would overstate a gated - # source's backlog (preview/apply parity, rule 93). - post_is_gated = getattr(self.client, "post_is_gated", None) - for post, included, page_cursor in self.client.iter_posts( - campaign_id, cursor=None - ): - if page_cursor != last_page: - last_page = page_cursor - pages_scanned += 1 - if pages_scanned > page_limit: - has_more = True - pages_scanned = page_limit - break - posts_scanned += 1 - if post_is_gated and post_is_gated(post): - continue - media = self.client.extract_media(post, included) - if not media: - continue - keys = [self._ledger_key(m) for m in media] - skip = self._seen_keys(source_id, keys) | self._dead_keys(source_id, keys) - new_count = sum(1 for m in media if self._ledger_key(m) not in skip) - total_new += new_count - if new_count > 0 and len(sample) < sample_size: - meta = self.client.post_meta(post) - sample.append( - { - "title": meta.get("title") or "(untitled)", - "date": meta.get("date"), - "new": new_count, - } - ) - return { - "total_new": total_new, - "posts_scanned": posts_scanned, - "pages_scanned": pages_scanned, - "has_more": has_more, - "sample": sample, - } - # -- failure mapping (adapter overrides) ------------------------------- def _failure_result(self, exc: Exception, _result) -> DownloadResult: diff --git a/frontend/src/components/subscriptions/PreviewDialog.vue b/frontend/src/components/subscriptions/PreviewDialog.vue deleted file mode 100644 index 0dac268..0000000 --- a/frontend/src/components/subscriptions/PreviewDialog.vue +++ /dev/null @@ -1,114 +0,0 @@ - - - - - diff --git a/frontend/src/stores/sources.js b/frontend/src/stores/sources.js index d169873..b1f760e 100644 --- a/frontend/src/stores/sources.js +++ b/frontend/src/stores/sources.js @@ -149,12 +149,6 @@ export const useSourcesStore = defineStore('sources', () => { return body } - // Plan #708 B4: dry-run preview — count what a backfill WOULD download - // (native platforms), without downloading. Read-only, so no cache invalidate. - async function previewSource(id) { - return await api.post(`/api/sources/${id}/preview`) - } - function sourcesByArtistGrouped() { // returns [{artist: {id,name,slug}, sources: [...]}, ...] const arr = byArtist.value.get(null) ?? [] @@ -185,7 +179,6 @@ export const useSourcesStore = defineStore('sources', () => { stopBackfill, recoverSource, recaptureSource, - previewSource, findOrCreateArtist, autocompleteArtist, reassign, loadScheduleStatus, sourcesByArtistGrouped, diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py index 9e75462..b711c02 100644 --- a/tests/test_patreon_ingester.py +++ b/tests/test_patreon_ingester.py @@ -448,42 +448,6 @@ async def test_backfill_budget_cut_returns_partial_with_progress( assert result.cursor == "CUR2" -@pytest.mark.asyncio -async def test_preview_counts_new_media_without_downloading( - source_id, sync_engine, tmp_path, -): - """plan #708 B4: preview walks + counts media not already seen/dead, downloads - nothing, and samples only posts that have new media.""" - m1, m2, m3 = _media("p1", 1), _media("p2", 1), _media("p2", 2) - # Seed m1 as already-seen → only p2's two media are "new". - factory = sessionmaker(sync_engine, expire_on_commit=False) - with factory() as s: - s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m1), post_id="p1")) - s.commit() - client = _FakeClient([(None, [("p1", [m1]), ("p2", [m2, m3])])]) - downloader = _FakeDownloader(tmp_path) - ing = _ingester(sync_engine, tmp_path, client, downloader) - - result = ing.preview(source_id, "c1") - assert result["total_new"] == 2 # p2's m2 + m3 (m1 already seen) - assert result["posts_scanned"] == 2 - assert result["has_more"] is False - assert downloader.download_calls == 0 # dry-run — nothing downloaded - # The sample lists only posts WITH new media (p2), not the all-seen p1. - assert [row["new"] for row in result["sample"]] == [2] - - -@pytest.mark.asyncio -async def test_preview_page_limit_caps_walk(source_id, sync_engine, tmp_path): - """plan #708 B4: preview stops after page_limit pages and flags has_more.""" - pages = [(f"CUR{i}", [(f"p{i}", [_media(f"p{i}", 1)])]) for i in range(6)] - ing = _ingester(sync_engine, tmp_path, _FakeClient(pages), _FakeDownloader(tmp_path)) - result = ing.preview(source_id, "c1", page_limit=2) - assert result["pages_scanned"] == 2 - assert result["posts_scanned"] == 2 # one post per page, 2 pages - assert result["has_more"] is True - - @pytest.mark.asyncio async def test_write_live_progress_updates_running_event( source_id, sync_engine, tmp_path, db, @@ -714,21 +678,6 @@ async def test_gated_post_skipped_entirely_no_media_no_record( assert _count_ledger(sync_engine, source_id) == 2 -@pytest.mark.asyncio -async def test_preview_excludes_gated_posts(source_id, sync_engine, tmp_path): - """#874 preview/apply parity (rule 93): the dry-run must not count a gated - post's blurred preview media as new, or it overstates a gated source's - backlog.""" - gm = _media("gated", 1) - am = _media("open", 1) - client = _FakeClient([(None, [("gated", [gm]), ("open", [am])])], gated={"gated"}) - ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path)) - - preview = ing.preview(source_id, "c1") - assert preview["total_new"] == 1 # only the open post - assert [s for s in preview["sample"] if s["title"] == "gated"] == [] - - @pytest.mark.asyncio async def test_recapture_does_not_refetch_seen_media_missing_from_disk( source_id, sync_engine, tmp_path, From 7939dba9ed88d9d35b4a8fdd7fd76bd033256ca8 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 6 Jul 2026 15:11:10 -0400 Subject: [PATCH 2/8] feat(ui): grounding overlay leads with the hovered tag, crop origin as subtext (#1206) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The overlay label showed only the crop origin (booru:head / panel / person-m). Put the hovered TAG on top as the headline and demote the crop origin to a small, dimmed, monospace subline — the tag is what you're evaluating; the origin is just provenance. Threads the tag name through the fcSuggestionHover payload ({g, tag}) from both setters (SuggestionItem for suggestions, TagChip for applied chips). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM --- frontend/src/components/modal/ImageCanvas.vue | 41 ++++++++++++++----- .../src/components/modal/SuggestionItem.vue | 2 +- frontend/src/components/modal/TagChip.vue | 2 +- 3 files changed, 33 insertions(+), 12 deletions(-) diff --git a/frontend/src/components/modal/ImageCanvas.vue b/frontend/src/components/modal/ImageCanvas.vue index efe2e52..f3af01e 100644 --- a/frontend/src/components/modal/ImageCanvas.vue +++ b/frontend/src/components/modal/ImageCanvas.vue @@ -35,7 +35,10 @@ :class="{ 'fc-canvas__region--whole': !regionStyle }" :style="regionStyle || {}" > - {{ regionLabel }} + + {{ regionTag }} + {{ regionOrigin }} + @@ -48,10 +51,12 @@ import { usePanZoom } from '../../composables/usePanZoom.js' const props = defineProps({ src: String, alt: String, - // Hovered-suggestion grounding, or null when nothing is hovered: - // null → no overlay - // { g: {bbox,kind,detector} } → highlight that crop region - // { g: null } → whole-image frame (not localized) + // Hovered-suggestion state, or null when nothing is hovered: + // null → no overlay + // { g: {bbox,kind,detector}, tag } → highlight that crop, label with the tag + // { g: null, tag } → whole-image frame (not localized) + // `tag` is the hovered concept's name (primary label line); `g` is where the + // crop came from (secondary provenance line). hover: { type: Object, default: null }, }) const emit = defineEmits(['close-request']) @@ -86,7 +91,12 @@ const regionStyle = computed(() => { width: `${rw * 100}%`, height: `${rh * 100}%`, } }) -const regionLabel = computed(() => { +// The hovered tag/suggestion name — the PRIMARY label line (what you care about). +const regionTag = computed(() => props.hover?.tag || '') +// Where the winning crop came from (the region's detector_version, else kind) — +// shown small + dimmed beneath the tag as provenance, not the headline. "whole +// image" = null grounding (the global vector won, not a crop). +const regionOrigin = computed(() => { const g = props.hover?.g return g ? (g.detector || g.kind || 'region') : 'whole image' }) @@ -176,15 +186,26 @@ function onCanvasClick(ev) { position: absolute; top: 0; left: 0; transform: translateY(-100%); - font-size: 11px; - font-family: 'JetBrains Mono', monospace; - line-height: 1.4; - padding: 1px 6px; + display: flex; flex-direction: column; + line-height: 1.25; + padding: 2px 6px; white-space: nowrap; background: rgb(var(--v-theme-accent)); color: rgb(var(--v-theme-on-surface)); border-radius: 3px 3px 0 0; } +/* The hovered tag is the headline. */ +.fc-canvas__region-tag { + font-size: 12px; + font-weight: 600; +} +/* Crop origin is provenance, not the point — subordinate: smaller, dimmed, and + monospace (it's a detector token like booru:head, not prose). */ +.fc-canvas__region-origin { + font-size: 10px; + font-family: 'JetBrains Mono', monospace; + color: rgb(var(--v-theme-on-surface), 0.72); +} .fc-canvas__region--whole .fc-canvas__region-label { background: rgb(var(--v-theme-on-surface-variant)); } diff --git a/frontend/src/components/modal/SuggestionItem.vue b/frontend/src/components/modal/SuggestionItem.vue index 4d00b70..7a47e3c 100644 --- a/frontend/src/components/modal/SuggestionItem.vue +++ b/frontend/src/components/modal/SuggestionItem.vue @@ -85,7 +85,7 @@ defineEmits(['accept', 'alias', 'remove-alias', 'dismiss', 'undismiss']) // it highlights that region. Provided by ImageViewer/Explore; a no-op elsewhere. const hover = inject('fcSuggestionHover', null) function onEnter () { - if (hover) hover.value = { g: props.suggestion.grounding ?? null } + if (hover) hover.value = { g: props.suggestion.grounding ?? null, tag: props.suggestion.display_name } } function onLeave () { if (hover) hover.value = null diff --git a/frontend/src/components/modal/TagChip.vue b/frontend/src/components/modal/TagChip.vue index bcb1cdf..6437255 100644 --- a/frontend/src/components/modal/TagChip.vue +++ b/frontend/src/components/modal/TagChip.vue @@ -83,7 +83,7 @@ async function onEnter () { if (seq !== hoverSeq) return // pointer already left / moved to another chip // No head → nothing to localize; don't draw an overlay at all. With a head, // null grounding still draws the whole-image frame ("the global vector won"). - if (res.has_head) hover.value = { g: res.grounding ?? null } + if (res.has_head) hover.value = { g: res.grounding ?? null, tag: props.tag.name } } function onLeave () { hoverSeq++ From 48be921b8d1a440d88b94f7ce94cc6472a4f27e2 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 6 Jul 2026 15:25:04 -0400 Subject: [PATCH 3/8] fix(ui): tag-rail no longer scroll-jumps on accept/reject; keep suggestions clear of the toast MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - TagAutocomplete: focus the inner input with preventScroll so handing focus back after an accept/reject stops yanking the rail up to the field (which sits above the suggestions list — you had to scroll back down every time). Applies to both the image modal and the Explore rail. - ExploreView: pad the right rail's scroll bottom (88px) so the bottom-right snackbar floats over empty space instead of covering the last suggestions and their accept/reject controls. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM --- frontend/src/components/modal/TagAutocomplete.vue | 11 ++++++++++- frontend/src/views/ExploreView.vue | 6 +++++- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/frontend/src/components/modal/TagAutocomplete.vue b/frontend/src/components/modal/TagAutocomplete.vue index 8d368fc..7888bba 100644 --- a/frontend/src/components/modal/TagAutocomplete.vue +++ b/frontend/src/components/modal/TagAutocomplete.vue @@ -127,7 +127,16 @@ const suggestions = useSuggestionsStore() const inputRef = ref(null) function focusInput () { if (window.matchMedia?.('(max-width: 900px)')?.matches) return - nextTick(() => inputRef.value?.focus?.()) + // Focus the inner with preventScroll: handing focus back after an + // accept/reject must NOT yank the rail's scroll up to this field, which sits + // above the suggestions list (operator-flagged 2026-07-06 — "I have to scroll + // back down every time I apply/reject"). Vuetify's own .focus() forwards no + // options, so reach the element directly; fall back to it if it's not there. + nextTick(() => { + const el = inputRef.value?.$el?.querySelector?.('input') + if (el) el.focus({ preventScroll: true }) + else inputRef.value?.focus?.() + }) } onMounted(focusInput) // Exposed so the parent (TagPanel) can hand focus back to this field after an diff --git a/frontend/src/views/ExploreView.vue b/frontend/src/views/ExploreView.vue index 17ea79b..bb893d5 100644 --- a/frontend/src/views/ExploreView.vue +++ b/frontend/src/views/ExploreView.vue @@ -350,10 +350,14 @@ onUnmounted(() => { } .fc-ex__artist { padding: 10px 16px 0; font-weight: 600; } -/* Right rail = the modal's tag panel, hosted on the anchor. */ +/* Right rail = the modal's tag panel, hosted on the anchor. The bottom padding + keeps the scrollable content from reaching the viewport floor, so the + bottom-right snackbar floats over empty space instead of covering the last + suggestions / their accept-reject controls (operator-flagged 2026-07-06). */ .fc-ex__rail { background: rgb(var(--v-theme-surface)); overflow-y: auto; + padding-bottom: 88px; } .fc-ex__spinner { From f24dc8176424ff7e806fd51eb86c0261a4d29e12 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 6 Jul 2026 15:58:11 -0400 Subject: [PATCH 4/8] feat(ccip): schema for precomputed incremental character prototypes (#1317, m138 step 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Foundation for making CCIP character references a precomputed, INCREMENTAL artifact instead of a request-path rebuild (kills the per-accept ~4s suggestions stall; cost will scale with change, not library size): - character_prototype: a character's reference CCIP vectors, capped to MLSettings.ccip_prototype_cap so match cost doesn't grow with popularity. - ccip_prototype_state: per-character fingerprint (ref count + max region id) + updated_at → drives per-character incremental rebuilds and the matcher cache's reload-only-what-advanced. - MLSettings.ccip_ref_signature (cheap global change gate) + ccip_prototype_cap. Migration 0079. Schema + models only — the builder service, refresh task/beat, and matcher rewrite land in the following steps. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM --- alembic/versions/0079_character_prototypes.py | 77 +++++++++++++++++++ backend/app/models/__init__.py | 3 + backend/app/models/character_prototype.py | 62 +++++++++++++++ backend/app/models/ml_settings.py | 12 +++ 4 files changed, 154 insertions(+) create mode 100644 alembic/versions/0079_character_prototypes.py create mode 100644 backend/app/models/character_prototype.py diff --git a/alembic/versions/0079_character_prototypes.py b/alembic/versions/0079_character_prototypes.py new file mode 100644 index 0000000..8ada2f4 --- /dev/null +++ b/alembic/versions/0079_character_prototypes.py @@ -0,0 +1,77 @@ +"""character prototype store (#1317) — precomputed, incremental CCIP references + +New tables character_prototype + ccip_prototype_state, plus MLSettings columns +ccip_ref_signature (cheap global change gate) + ccip_prototype_cap (per-character +reference cap). The reference set the CCIP matcher uses becomes a precomputed +artifact refreshed incrementally off the request path. See milestone 138 / +backend.app.services.ml.character_prototypes. + +Revision ID: 0079 +Revises: 0078 +Create Date: 2026-07-06 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from pgvector.sqlalchemy import Vector + +revision: str = "0079" +down_revision: Union[str, None] = "0078" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +# Matches models.image_region.CCIP_DIM (the CCIP figure-embedding width). +_CCIP_DIM = 768 + + +def upgrade() -> None: + op.create_table( + "character_prototype", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column( + "tag_id", sa.Integer(), + sa.ForeignKey("tag.id", ondelete="CASCADE"), nullable=False, + ), + sa.Column("ccip_embedding", Vector(_CCIP_DIM), nullable=False), + sa.Column( + "region_id", sa.Integer(), + sa.ForeignKey("image_region.id", ondelete="SET NULL"), nullable=True, + ), + ) + op.create_index( + "ix_character_prototype_tag_id", "character_prototype", ["tag_id"] + ) + op.create_table( + "ccip_prototype_state", + sa.Column( + "tag_id", sa.Integer(), + sa.ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True, + ), + sa.Column("fingerprint", sa.String(64), nullable=False), + sa.Column( + "updated_at", sa.DateTime(timezone=True), nullable=False, + server_default=sa.func.now(), + ), + ) + op.add_column( + "ml_settings", + sa.Column("ccip_ref_signature", sa.String(128), nullable=True), + ) + op.add_column( + "ml_settings", + sa.Column( + "ccip_prototype_cap", sa.Integer(), nullable=False, + server_default=sa.text("64"), + ), + ) + + +def downgrade() -> None: + op.drop_column("ml_settings", "ccip_prototype_cap") + op.drop_column("ml_settings", "ccip_ref_signature") + op.drop_table("ccip_prototype_state") + op.drop_index( + "ix_character_prototype_tag_id", table_name="character_prototype" + ) + op.drop_table("character_prototype") diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 99ebaa6..0da2555 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -5,6 +5,7 @@ from .artist import Artist from .artist_visit import ArtistVisit from .backup_run import BackupRun from .base import Base +from .character_prototype import CcipPrototypeState, CharacterPrototype from .credential import Credential from .download_event import DownloadEvent from .external_link import ExternalLink @@ -79,6 +80,8 @@ __all__ = [ "HeadTrainingRun", "TagAlias", "TagHead", + "CharacterPrototype", + "CcipPrototypeState", "TagPositiveConfirmation", "TagSuggestionRejection", "TaskRun", diff --git a/backend/app/models/character_prototype.py b/backend/app/models/character_prototype.py new file mode 100644 index 0000000..191a29d --- /dev/null +++ b/backend/app/models/character_prototype.py @@ -0,0 +1,62 @@ +"""Precomputed CCIP character prototypes (#1317, milestone 138). + +The live matcher (ccip.match_image) needs each character's reference figure +vectors. Building that on the request path reloaded EVERY figure CCIP vector in +the library on any change (~4s, invalidated by every character accept). These +tables make the references a PRECOMPUTED, INCREMENTAL artifact refreshed off the +request path (services.ml.character_prototypes): + +- CharacterPrototype: a character's reference vectors, capped to + MLSettings.ccip_prototype_cap so MATCH cost doesn't grow with a character's + popularity. The async matcher only READS these. +- CcipPrototypeState: a per-character fingerprint (reference count + max region + id) so a refresh rebuilds ONLY the characters whose references changed, and + its updated_at lets the matcher's cache reload just the advanced characters. +""" + +from datetime import datetime + +from pgvector.sqlalchemy import Vector +from sqlalchemy import DateTime, ForeignKey, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from .base import Base +from .image_region import CCIP_DIM + + +class CharacterPrototype(Base): + __tablename__ = "character_prototype" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + # The character tag these vectors identify. CASCADE: deleting the tag drops + # its prototypes. + tag_id: Mapped[int] = mapped_column( + ForeignKey("tag.id", ondelete="CASCADE"), nullable=False, index=True + ) + # A reference figure/face CCIP vector (same space as + # ImageRegion.ccip_embedding). + ccip_embedding: Mapped[list[float]] = mapped_column( + Vector(CCIP_DIM), nullable=False + ) + # Provenance: the region this vector was copied from. SET NULL so pruning a + # region doesn't delete the prototype mid-cycle (the next refresh reconciles). + region_id: Mapped[int | None] = mapped_column( + ForeignKey("image_region.id", ondelete="SET NULL"), nullable=True + ) + + +class CcipPrototypeState(Base): + __tablename__ = "ccip_prototype_state" + + # One row per character that currently has prototypes. + tag_id: Mapped[int] = mapped_column( + ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True + ) + # count(reference regions) + max(region id) at last build — the cheap + # per-character change detector that drives incremental rebuilds. + fingerprint: Mapped[str] = mapped_column(String(64), nullable=False) + # Bumped when this character's prototypes are rebuilt; the matcher cache + # reloads only characters whose updated_at advanced. + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) diff --git a/backend/app/models/ml_settings.py b/backend/app/models/ml_settings.py index dfb97d5..c03075e 100644 --- a/backend/app/models/ml_settings.py +++ b/backend/app/models/ml_settings.py @@ -155,6 +155,18 @@ class MLSettings(Base): detector_dedupe_iou: Mapped[float] = mapped_column( Float, nullable=False, default=0.85 ) + # -- CCIP character prototypes (#1317) --------------------------------- + # The per-character reference set is precomputed + refreshed INCREMENTALLY + # (services.ml.character_prototypes) instead of rebuilt on the request path. + # ccip_ref_signature is the cheap GLOBAL gate — when it's unchanged the + # refresh no-ops; ccip_prototype_cap bounds the reference vectors kept per + # character so MATCH cost doesn't grow with a character's popularity. + ccip_ref_signature: Mapped[str | None] = mapped_column( + String(128), nullable=True + ) + ccip_prototype_cap: Mapped[int] = mapped_column( + Integer, nullable=False, default=64 + ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) From 9504870c9ad529f23ff8c696642d141d15fe319e Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 6 Jul 2026 16:07:07 -0400 Subject: [PATCH 5/8] feat(ccip): incremental character-prototype builder (#1317, m138 step 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit refresh_character_prototypes (sync, celery ml worker): - Cheap GLOBAL gate (a few COUNTs) → no-op when nothing that affects references changed since the last refresh (the operator's "only recompute if something was tagged" trigger). - Else a per-character fingerprint diff (one GROUP BY: ref count + max region id) rebuilds ONLY the characters whose references moved — each capped to MLSettings.ccip_prototype_cap — and drops characters that lost all refs. Cost scales with WHAT changed, not library size. Reuses ccip's reference predicate (single-character, non-hygiene, figure CCIP) so prototypes match the legacy matcher exactly. The async matcher (next step) will READ the table. Tests: gate no-op when idle, only-changed-character rebuild, capping, single-character exclusion, lost-reference cleanup. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM --- .../app/services/ml/character_prototypes.py | 175 ++++++++++++++++++ tests/test_character_prototypes.py | 163 ++++++++++++++++ 2 files changed, 338 insertions(+) create mode 100644 backend/app/services/ml/character_prototypes.py create mode 100644 tests/test_character_prototypes.py diff --git a/backend/app/services/ml/character_prototypes.py b/backend/app/services/ml/character_prototypes.py new file mode 100644 index 0000000..8f8fb72 --- /dev/null +++ b/backend/app/services/ml/character_prototypes.py @@ -0,0 +1,175 @@ +"""Precomputed CCIP character prototypes — incremental builder (#1317, m138). + +Turns the per-character CCIP reference set into a precomputed artifact so the +matcher never rebuilds it on the request path. Sync (runs in the celery ml +worker via tasks.ml.refresh_character_prototypes); the async matcher only READS +the character_prototype table. + +`refresh_character_prototypes`: + 1. Cheap GLOBAL gate — a few COUNTs (`_global_signature`). Unchanged + not + `full` → no-op (nothing that affects references changed since last refresh). + 2. Per-character fingerprint diff (one GROUP BY) → rebuild ONLY the characters + whose references changed (or are new); drop characters that lost all refs. +Each rebuild loads just ONE character's reference vectors, caps them to +MLSettings.ccip_prototype_cap, and replaces that character's prototype rows — so +cost scales with WHAT changed, not with the library size. + +The reference PREDICATE (single-character, non-hygiene, figure CCIP) is imported +from ccip so the prototypes match exactly what the legacy matcher selected. +""" + +import random +from datetime import UTC, datetime + +from sqlalchemy import delete, func, select +from sqlalchemy.orm import Session + +from ...models import ( + CcipPrototypeState, + CharacterPrototype, + ImageRegion, + MLSettings, + Tag, + TagKind, +) +from ...models.tag import image_tag +from .ccip import _FIGURE_KINDS, _hygiene_tagged_images, _single_character_images + +# Deterministic per-tag capping so a rebuild of an UNCHANGED reference set +# resamples identically (stable prototypes, no churn between refreshes). +_SAMPLE_SEED = 1317 + + +def _global_signature(session: Session) -> str: + """Cheap 'could any references have changed' gate: character-tag count, + figure-CCIP region count + max id, hygiene-tag count. A few COUNTs — the same + quantities the legacy per-request signature used, now computed once per + refresh instead of on every /suggestions call.""" + n_tags = session.execute( + select(func.count()) + .select_from(image_tag) + .join(Tag, Tag.id == image_tag.c.tag_id) + .where(Tag.kind == TagKind.character) + ).scalar_one() + n_regs, max_id = session.execute( + select(func.count(), func.max(ImageRegion.id)).where( + ImageRegion.kind.in_(_FIGURE_KINDS), + ImageRegion.ccip_embedding.is_not(None), + ) + ).one() + n_hygiene = session.execute( + select(func.count()) + .select_from(image_tag) + .join(Tag, Tag.id == image_tag.c.tag_id) + .where(Tag.is_system.is_(True)) + ).scalar_one() + return f"{n_tags}:{n_regs}:{max_id or 0}:{n_hygiene}" + + +def _current_fingerprints(session: Session) -> dict[int, str]: + """Per-character (reference count, max reference region id) over the SAME + predicate the matcher's references use. One GROUP BY → the change detector: + a character whose fingerprint moved (gained/lost a reference) needs a + rebuild; everyone else is left untouched.""" + rows = session.execute( + select( + image_tag.c.tag_id, + func.count(ImageRegion.id), + func.max(ImageRegion.id), + ) + .select_from(ImageRegion) + .join( + image_tag, + image_tag.c.image_record_id == ImageRegion.image_record_id, + ) + .join(Tag, Tag.id == image_tag.c.tag_id) + .where(Tag.kind == TagKind.character) + .where(ImageRegion.kind.in_(_FIGURE_KINDS)) + .where(ImageRegion.ccip_embedding.is_not(None)) + .where(ImageRegion.image_record_id.in_(_single_character_images())) + .where(ImageRegion.image_record_id.not_in(_hygiene_tagged_images())) + .group_by(image_tag.c.tag_id) + ).all() + return {tag_id: f"{cnt}:{mx}" for tag_id, cnt, mx in rows} + + +def _rebuild_one(session: Session, tag_id: int, cap: int) -> int: + """Replace ONE character's prototype rows from its current references, capped + to `cap`. Loads only this character's vectors (bounded by its popularity).""" + rows = session.execute( + select(ImageRegion.id, ImageRegion.ccip_embedding) + .select_from(ImageRegion) + .join( + image_tag, + image_tag.c.image_record_id == ImageRegion.image_record_id, + ) + .where(image_tag.c.tag_id == tag_id) + .where(ImageRegion.kind.in_(_FIGURE_KINDS)) + .where(ImageRegion.ccip_embedding.is_not(None)) + .where(ImageRegion.image_record_id.in_(_single_character_images())) + .where(ImageRegion.image_record_id.not_in(_hygiene_tagged_images())) + ).all() + # Cap for bounded MATCH cost. A random sample (not most-recent) keeps the + # prototypes representative of the whole reference set; a fixed per-tag seed + # makes an unchanged set resample identically. + if cap > 0 and len(rows) > cap: + rows = random.Random(f"{_SAMPLE_SEED}:{tag_id}").sample(rows, cap) + session.execute( + delete(CharacterPrototype).where(CharacterPrototype.tag_id == tag_id) + ) + for region_id, vec in rows: + session.add( + CharacterPrototype( + tag_id=tag_id, region_id=region_id, ccip_embedding=vec + ) + ) + return len(rows) + + +def refresh_character_prototypes( + session: Session, *, full: bool = False +) -> dict[str, int | bool]: + """Incrementally refresh the prototype store. `full=True` rebuilds every + character regardless of the gate/fingerprints (nightly reconcile). Returns + {skipped, rebuilt, removed}; commits.""" + settings = session.execute( + select(MLSettings).where(MLSettings.id == 1) + ).scalar_one() + sig = _global_signature(session) + if not full and settings.ccip_ref_signature == sig: + return {"skipped": True, "rebuilt": 0, "removed": 0} + + cap = settings.ccip_prototype_cap + current = _current_fingerprints(session) + stored = dict( + session.execute( + select(CcipPrototypeState.tag_id, CcipPrototypeState.fingerprint) + ).all() + ) + now = datetime.now(UTC) + rebuilt = 0 + for tag_id, fp in current.items(): + if full or stored.get(tag_id) != fp: + _rebuild_one(session, tag_id, cap) + state = session.get(CcipPrototypeState, tag_id) + if state is None: + state = CcipPrototypeState(tag_id=tag_id) + session.add(state) + state.fingerprint = fp + state.updated_at = now + rebuilt += 1 + # Characters that lost every reference (refs removed / re-kinded / image now + # multi-character) → drop their prototypes + state so they stop matching. + removed = 0 + for tag_id in set(stored) - set(current): + session.execute( + delete(CharacterPrototype).where(CharacterPrototype.tag_id == tag_id) + ) + session.execute( + delete(CcipPrototypeState).where(CcipPrototypeState.tag_id == tag_id) + ) + removed += 1 + + settings.ccip_ref_signature = sig + session.commit() + return {"skipped": False, "rebuilt": rebuilt, "removed": removed} diff --git a/tests/test_character_prototypes.py b/tests/test_character_prototypes.py new file mode 100644 index 0000000..6842053 --- /dev/null +++ b/tests/test_character_prototypes.py @@ -0,0 +1,163 @@ +"""Incremental CCIP prototype builder (#1317, m138). Sync (celery ml worker), +so tested directly via db_sync — the global gate, per-character fingerprint diff, +capping, single-character exclusion, and lost-reference cleanup.""" +import pytest +from sqlalchemy import func, select + +from backend.app.models import ( + CcipPrototypeState, + CharacterPrototype, + ImageRecord, + ImageRegion, + MLSettings, + Tag, + TagKind, +) +from backend.app.models.tag import image_tag +from backend.app.services.ml.character_prototypes import ( + refresh_character_prototypes, +) + +pytestmark = pytest.mark.integration + + +def _ccip(slot: int = 0) -> list[float]: + v = [0.0] * 768 + v[slot] = 1.0 + return v + + +def _img(db, sha: str) -> ImageRecord: + img = ImageRecord( + path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg", + width=1, height=1, origin="imported_filesystem", + integrity_status="unknown", + ) + db.add(img) + db.flush() + return img + + +def _figure(db, image_id: int, ccip=None) -> None: + db.add(ImageRegion( + image_record_id=image_id, kind="figure", + rx=0.0, ry=0.0, rw=1.0, rh=1.0, + ccip_embedding=ccip if ccip is not None else _ccip(), + embedding_version="ccip-test", + )) + db.flush() + + +def _char(db, name: str) -> Tag: + t = Tag(name=name, kind=TagKind.character) + db.add(t) + db.flush() + return t + + +def _tag(db, image_id: int, tag_id: int) -> None: + db.execute(image_tag.insert().values( + image_record_id=image_id, tag_id=tag_id, source="manual", + )) + + +def _proto_count(db, tag_id: int) -> int: + return db.execute( + select(func.count()).select_from(CharacterPrototype) + .where(CharacterPrototype.tag_id == tag_id) + ).scalar_one() + + +def _set_cap(db, cap: int) -> None: + s = db.execute(select(MLSettings).where(MLSettings.id == 1)).scalar_one() + s.ccip_prototype_cap = cap + db.flush() + + +def test_refresh_builds_then_gate_no_ops(db_sync): + raven = _char(db_sync, "Raven") + img = _img(db_sync, "a" * 64) + _figure(db_sync, img.id) + _tag(db_sync, img.id, raven.id) + db_sync.commit() + + r1 = refresh_character_prototypes(db_sync) + assert r1["skipped"] is False and r1["rebuilt"] == 1 + assert _proto_count(db_sync, raven.id) == 1 + assert db_sync.get(CcipPrototypeState, raven.id) is not None + + # Nothing changed → the global gate short-circuits (no per-character work). + r2 = refresh_character_prototypes(db_sync) + assert r2["skipped"] is True and r2["rebuilt"] == 0 + + +def test_only_changed_character_rebuilt(db_sync): + raven = _char(db_sync, "Raven") + daphne = _char(db_sync, "Daphne") + ri = _img(db_sync, "b" * 64) + _figure(db_sync, ri.id) + _tag(db_sync, ri.id, raven.id) + di = _img(db_sync, "c" * 64) + _figure(db_sync, di.id) + _tag(db_sync, di.id, daphne.id) + db_sync.commit() + assert refresh_character_prototypes(db_sync)["rebuilt"] == 2 + + # A new figure on Raven's image changes only Raven's fingerprint. + _figure(db_sync, ri.id, _ccip(1)) + db_sync.commit() + r = refresh_character_prototypes(db_sync) + assert r["rebuilt"] == 1 # Daphne untouched + assert _proto_count(db_sync, raven.id) == 2 + assert _proto_count(db_sync, daphne.id) == 1 + + +def test_cap_bounds_prototypes(db_sync): + _set_cap(db_sync, 2) + raven = _char(db_sync, "Raven") + for i in range(3): # 3 single-character images + img = _img(db_sync, f"{i}" * 64) + _figure(db_sync, img.id) + _tag(db_sync, img.id, raven.id) + db_sync.commit() + + refresh_character_prototypes(db_sync) + assert _proto_count(db_sync, raven.id) == 2 # capped from 3 → 2 + + +def test_multi_character_image_not_referenced(db_sync): + # A figure on a 2-character image is ambiguous (tag is image-level), so it + # must NOT become a reference for either — parity with the legacy matcher. + raven = _char(db_sync, "Raven") + daphne = _char(db_sync, "Daphne") + img = _img(db_sync, "d" * 64) + _figure(db_sync, img.id) + _tag(db_sync, img.id, raven.id) + _tag(db_sync, img.id, daphne.id) + db_sync.commit() + + refresh_character_prototypes(db_sync) + assert _proto_count(db_sync, raven.id) == 0 + assert _proto_count(db_sync, daphne.id) == 0 + + +def test_lost_references_are_removed(db_sync): + raven = _char(db_sync, "Raven") + img = _img(db_sync, "e" * 64) + _figure(db_sync, img.id) + _tag(db_sync, img.id, raven.id) + db_sync.commit() + refresh_character_prototypes(db_sync) + assert _proto_count(db_sync, raven.id) == 1 + + # Untag Raven → the image is no longer a reference → prototypes + state drop. + db_sync.execute( + image_tag.delete() + .where(image_tag.c.image_record_id == img.id) + .where(image_tag.c.tag_id == raven.id) + ) + db_sync.commit() + r = refresh_character_prototypes(db_sync) + assert r["removed"] == 1 + assert _proto_count(db_sync, raven.id) == 0 + assert db_sync.get(CcipPrototypeState, raven.id) is None From a1ed53136e90cce269f395c5b5859faf8007ce16 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 6 Jul 2026 16:13:22 -0400 Subject: [PATCH 6/8] feat(ccip): refresh task + beat + retrain hook for character prototypes (#1317, m138 step 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - refresh_character_prototypes celery task wraps the incremental builder (sync ml worker); returns skipped / rebuilt=N removed=N. - Beat: every ~15 min (cheap global-gate no-op when idle) + a nightly full=True reconcile as belt-and-suspenders. - train_heads enqueues it on success, so the Retrain button AND the nightly head retrain refresh CCIP on the SAME trigger — unified lifecycle, as asked. The initial (cold) full build loads the whole reference set once in the background, never on a /suggestions request. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM --- backend/app/celery_app.py | 9 +++++++++ backend/app/tasks/ml.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+) diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 74e2403..f42db3b 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -107,6 +107,15 @@ def make_celery() -> Celery: "task": "backend.app.tasks.ml.scheduled_train_heads", "schedule": 86400.0, # passive cadence; manual retrain stays available }, + "refresh-character-prototypes": { + "task": "backend.app.tasks.ml.refresh_character_prototypes", + "schedule": 900.0, # ~15 min; cheap global-gate no-op when idle (#1317) + }, + "reconcile-character-prototypes-nightly": { + "task": "backend.app.tasks.ml.refresh_character_prototypes", + "schedule": 86400.0, # nightly FULL reconcile (belt-and-suspenders) + "args": (True,), # full=True + }, "apply-head-tags-daily": { "task": "backend.app.tasks.ml.scheduled_apply_head_tags", "schedule": 86400.0, # no-op unless head_auto_apply_enabled diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index b1650e2..bfb8d0c 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -328,6 +328,11 @@ def train_heads(self, run_id: int) -> str: run.status = "ready" run.finished_at = datetime.now(UTC) session.commit() + # A retrain folds the day's accepts/rejects into the heads; refresh the + # CCIP character prototypes on the SAME trigger (#1317) so the Retrain + # button + nightly run keep character matching current too. Cheap no-op + # when references didn't change. + refresh_character_prototypes.delay() return "ready" @@ -361,6 +366,31 @@ def scheduled_train_heads() -> str: return "dispatched" +@celery.task( + name="backend.app.tasks.ml.refresh_character_prototypes", + # Global-gate no-op when idle; a rebuild touches only changed characters. The + # initial (cold) build loads the whole reference set once, in the BACKGROUND + # — never on a /suggestions request. + soft_time_limit=1800, time_limit=2100, +) +def refresh_character_prototypes(full: bool = False) -> str: + """Incrementally refresh the CCIP character-prototype store (#1317, m138): + cheap global-gate skip when nothing changed, else rebuild only the characters + whose references moved. Triggered by a ~15-min beat, after every head retrain + (train_heads), and nightly with full=True. Returns a short status.""" + from ..services.ml.character_prototypes import ( + refresh_character_prototypes as _refresh, + ) + + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + result = _refresh(session, full=full) + return ( + "skipped" if result["skipped"] + else f"rebuilt={result['rebuilt']} removed={result['removed']}" + ) + + @celery.task( name="backend.app.tasks.ml.apply_head_tags", bind=True, From a94f6a2789542479b35c155c4d2176a9358920b6 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 6 Jul 2026 16:21:38 -0400 Subject: [PATCH 7/8] feat(ccip): matcher reads the incremental prototype store (#1317, m138 step 4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit match_image now sources character references from character_prototype via a per-character in-process cache (_load_prototypes) that reloads ONLY the characters whose ccip_prototype_state.updated_at advanced — no request-path rebuild, so the per-accept ~4s stall is gone once the store is populated. Cold start (store empty pre-first-refresh) falls back to the legacy on-the-fly reference build, so character suggestions work immediately post-deploy and the background refresh populates the store within ~15 min. Match math + grounding are unchanged; existing tests exercise the legacy fallback, and a new test covers matching from the populated prototype store. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM --- backend/app/services/ml/ccip.py | 77 +++++++++++++++++++++++++++++++-- tests/test_ccip.py | 26 ++++++++++- 2 files changed, 99 insertions(+), 4 deletions(-) diff --git a/backend/app/services/ml/ccip.py b/backend/app/services/ml/ccip.py index 0a30def..bf09a09 100644 --- a/backend/app/services/ml/ccip.py +++ b/backend/app/services/ml/ccip.py @@ -16,7 +16,14 @@ the hands-on eval. numpy is imported lazily (API worker has it via pgvector). from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession -from ...models import ImageRegion, MLSettings, Tag, TagKind +from ...models import ( + CcipPrototypeState, + CharacterPrototype, + ImageRegion, + MLSettings, + Tag, + TagKind, +) from ...models.tag import image_tag # Cosine-similarity floor to call a figure the same character. The live setting @@ -148,6 +155,60 @@ async def _tag_names(session: AsyncSession, tag_ids: list[int]) -> dict[int, str ) +# Per-character normalized prototype matrices, cached per process and refreshed +# INCREMENTALLY: only characters whose ccip_prototype_state.updated_at advanced +# are reloaded. This replaces the request-path rebuild of the ENTIRE reference +# blob (the ~4s stall, #1317) — the prototypes are precomputed off the request +# path by services.ml.character_prototypes (a beat + after each retrain). +_PROTO_CACHE: dict = {"mats": {}, "ver": {}} + + +async def _load_prototypes(session: AsyncSession) -> dict: + """{tag_id: (P, D) L2-normalized prototype matrix} from character_prototype, + served from the in-process cache and reloading ONLY the characters whose + updated_at changed. Empty dict when the store isn't populated yet (cold start + → match_image falls back to the legacy on-the-fly reference build).""" + import numpy as np + + versions = dict( + ( + await session.execute( + select(CcipPrototypeState.tag_id, CcipPrototypeState.updated_at) + ) + ).all() + ) + mats = _PROTO_CACHE["mats"] + ver = _PROTO_CACHE["ver"] + # Forget characters that no longer have prototypes. + for tag_id in [t for t in mats if t not in versions]: + mats.pop(tag_id, None) + ver.pop(tag_id, None) + # Reload only the characters whose prototypes changed since we cached them. + stale = [t for t, u in versions.items() if ver.get(t) != u] + if stale: + rows = ( + await session.execute( + select( + CharacterPrototype.tag_id, CharacterPrototype.ccip_embedding + ).where(CharacterPrototype.tag_id.in_(stale)) + ) + ).all() + by_tag: dict[int, list] = {} + for tag_id, vec in rows: + by_tag.setdefault(tag_id, []).append( + np.asarray(vec, dtype=np.float32) + ) + for tag_id in stale: + vecs = by_tag.get(tag_id) + if vecs: + mats[tag_id] = _l2norm(np.vstack(vecs), np) + ver[tag_id] = versions[tag_id] + else: + mats.pop(tag_id, None) + ver.pop(tag_id, None) + return mats + + async def match_image( session: AsyncSession, image_id: int, threshold: float | None = None ) -> list[dict]: @@ -178,7 +239,13 @@ async def match_image( ).all() if not fig_rows: return [] - refs = await character_references(session) + # Prefer the precomputed prototype store (fast, incremental). On a cold start + # (store not yet populated post-deploy) fall back to the legacy on-the-fly + # reference build so character suggestions work immediately — the background + # refresh populates the store within ~15 min, after which this path is used + # and the per-accept ~4s rebuild is gone (#1317). + protos = await _load_prototypes(session) + refs = protos if protos else await character_references(session) if not refs: return [] applied = set( @@ -202,7 +269,11 @@ async def match_image( for tag_id, vecs in refs.items(): if tag_id in applied: continue - R = _l2norm(np.vstack([np.asarray(v, dtype=np.float32) for v in vecs]), np) + # Prototype matrices are already L2-normalized; legacy refs are raw + # vector lists that still need stacking + normalizing. + R = vecs if protos else _l2norm( + np.vstack([np.asarray(v, dtype=np.float32) for v in vecs]), np + ) sims = Q @ R.T # (n_query_figures, n_references) per_figure = sims.max(axis=1) # best reference cosine per figure best_figure = int(per_figure.argmax()) diff --git a/tests/test_ccip.py b/tests/test_ccip.py index 8231c4b..df52378 100644 --- a/tests/test_ccip.py +++ b/tests/test_ccip.py @@ -3,7 +3,13 @@ model needed, so it runs in CI with synthetic CCIP vectors.""" import pytest from sqlalchemy import select -from backend.app.models import ImageRecord, ImageRegion, TagKind +from backend.app.models import ( + CcipPrototypeState, + CharacterPrototype, + ImageRecord, + ImageRegion, + TagKind, +) from backend.app.models.tag import image_tag from backend.app.services.ml.ccip import match_image from backend.app.services.tag_service import TagService @@ -61,6 +67,24 @@ async def test_matches_same_character_across_images(db): assert m["grounding"]["kind"] == "figure" +@pytest.mark.asyncio +async def test_matches_from_prototype_store(db): + # Once the prototype store is populated (#1317), match_image reads it (the + # fast incremental path) instead of rebuilding references on the request — + # same match + grounding as the legacy build. + raven = await TagService(db).find_or_create("Raven", TagKind.character) + db.add(CharacterPrototype(tag_id=raven.id, ccip_embedding=_ccip(0))) + db.add(CcipPrototypeState(tag_id=raven.id, fingerprint="1:1")) + query = await _img(db, "n" * 64) + await _figure(db, query.id, _ccip(0)) # query figure matches the prototype + await db.commit() + + matches = await match_image(db, query.id) + m = next(x for x in matches if x["tag_id"] == raven.id) + assert m["source"] == "ccip" and m["score"] > 0.9 + assert m["grounding"]["kind"] == "figure" + + @pytest.mark.asyncio async def test_no_match_for_different_character(db): raven = await TagService(db).find_or_create("Raven", TagKind.character) From 2cfbb284d5a5a38ffd02abca6ba3ff48c1476fbd Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 6 Jul 2026 16:36:30 -0400 Subject: [PATCH 8/8] =?UTF-8?q?feat(heads):=20incremental=20retraining=20?= =?UTF-8?q?=E2=80=94=20refit=20only=20changed=20tags=20(#1317=20phase=202,?= =?UTF-8?q?=20m138)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit train_all_heads is now incremental by default: a per-tag training-data fingerprint (positive + rejection count/latest-timestamp, stored on tag_head.train_fingerprint) means a manual Retrain refits ONLY the tags whose data changed — O(what you touched), not O(all heads). The nightly scheduled_train_heads passes full=True to reconcile sampled-negative + hygiene drift across every head. First incremental run after deploy still refits everyone (NULL fingerprints), stamping them, then it's incremental. The refit decision + fingerprint are split into sklearn-free helpers (_head_fingerprints, _heads_needing_retrain) so the incremental logic is unit-tested directly (train_head itself needs scikit-learn). Migration 0080. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM --- .../0080_tag_head_train_fingerprint.py | 31 ++++ backend/app/models/tag_head.py | 7 + backend/app/services/ml/heads.py | 101 ++++++++++++- backend/app/tasks/ml.py | 5 +- tests/test_head_incremental.py | 136 ++++++++++++++++++ 5 files changed, 273 insertions(+), 7 deletions(-) create mode 100644 alembic/versions/0080_tag_head_train_fingerprint.py create mode 100644 tests/test_head_incremental.py diff --git a/alembic/versions/0080_tag_head_train_fingerprint.py b/alembic/versions/0080_tag_head_train_fingerprint.py new file mode 100644 index 0000000..b4bd224 --- /dev/null +++ b/alembic/versions/0080_tag_head_train_fingerprint.py @@ -0,0 +1,31 @@ +"""tag_head.train_fingerprint (#1317 phase 2) — incremental head retraining + +A per-head training-data fingerprint (positive + rejection count/latest-timestamp) +so a manual Retrain refits only the tags whose data changed; the nightly run +ignores it (full reconcile). Nullable — a NULL fingerprint (existing heads) forces +a refit on the first incremental run, then it's stamped. + +Revision ID: 0080 +Revises: 0079 +Create Date: 2026-07-06 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0080" +down_revision: Union[str, None] = "0079" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "tag_head", + sa.Column("train_fingerprint", sa.String(128), nullable=True), + ) + + +def downgrade() -> None: + op.drop_column("tag_head", "train_fingerprint") diff --git a/backend/app/models/tag_head.py b/backend/app/models/tag_head.py index 34a5bcb..0390810 100644 --- a/backend/app/models/tag_head.py +++ b/backend/app/models/tag_head.py @@ -73,5 +73,12 @@ class TagHead(Base): trained_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) + # Training-data fingerprint (positives + rejections) at last fit — the + # incremental-retrain change detector (#1317 p2). A manual Retrain refits only + # heads whose fingerprint moved; the nightly run ignores it (full reconcile). + # NULL forces a refit (pre-fingerprint heads). + train_fingerprint: Mapped[str | None] = mapped_column( + String(128), nullable=True + ) # Extra detail (auto-apply operating point, F1, etc.) — non-load-bearing. metrics: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) diff --git a/backend/app/services/ml/heads.py b/backend/app/services/ml/heads.py index bb18c25..41059cd 100644 --- a/backend/app/services/ml/heads.py +++ b/backend/app/services/ml/heads.py @@ -150,24 +150,103 @@ def _eligible_tag_ids(session: Session, min_pos: int) -> list[int]: return [r[0] for r in rows] +def _head_fingerprints(session: Session, tag_ids: list[int]) -> dict[int, str]: + """Per-tag training-data fingerprint: (positive count, latest positive + created_at) + (rejection count, latest rejected_at). It moves whenever a tag + gains/loses a positive or a rejection — the incremental-retrain change + detector (#1317 p2). A newly-added positive/rejection always has the latest + timestamp, so even a remove-one-add-one (unchanged count) is caught. The + sampled-unlabeled negative pool + the hygiene set drift GLOBALLY and are + reconciled by the nightly full run, not captured here.""" + if not tag_ids: + return {} + pos = session.execute( + select( + image_tag.c.tag_id, + func.count(image_tag.c.image_record_id), + func.max(image_tag.c.created_at), + ) + .where(image_tag.c.tag_id.in_(tag_ids)) + .group_by(image_tag.c.tag_id) + ).all() + pos_map = {t: (c, m) for t, c, m in pos} + rej = session.execute( + select( + TagSuggestionRejection.tag_id, + func.count(), + func.max(TagSuggestionRejection.rejected_at), + ) + .where(TagSuggestionRejection.tag_id.in_(tag_ids)) + .group_by(TagSuggestionRejection.tag_id) + ).all() + rej_map = {t: (c, m) for t, c, m in rej} + out = {} + for t in tag_ids: + pc, pm = pos_map.get(t, (0, None)) + rc, rm = rej_map.get(t, (0, None)) + out[t] = f"{pc}:{pm}:{rc}:{rm}" + return out + + +def _heads_needing_retrain( + session: Session, eligible: list[int], embedding_version: str, + fps: dict[int, str], full: bool, +) -> list[int]: + """The eligible tag_ids to (re)fit: no head yet, a head trained in a DIFFERENT + embedding space (a model swap), or a changed training-data fingerprint. + full=True forces every eligible tag. sklearn-free (train_head itself needs + scikit-learn) so the incremental decision is unit-testable on its own.""" + if full: + return list(eligible) + existing = { + tag_id: (fp, ev) + for tag_id, fp, ev in session.execute( + select( + TagHead.tag_id, TagHead.train_fingerprint, + TagHead.embedding_version, + ) + ).all() + } + out = [] + for tag_id in eligible: + prev = existing.get(tag_id) + if ( + prev is None + or prev[1] != embedding_version + or prev[0] != fps.get(tag_id) + ): + out.append(tag_id) + return out + + def train_all_heads( session: Session, params: dict[str, Any], run: HeadTrainingRun | None = None ) -> dict[str, int]: - """(Re)train a head for every eligible concept; prune heads whose tag is no - longer eligible. Commits per head so a SIGKILL leaves trained heads durable - (training is idempotent). Returns {n_trained, n_skipped}.""" + """(Re)train eligible concept heads, INCREMENTALLY by default (#1317 p2): + refit only the tags whose training data changed since last fit, so a manual + Retrain click is fast. `params["full"]=True` (the nightly run) refits every + head to reconcile sampled-negative + hygiene drift. Prunes heads whose tag is + no longer eligible. Commits per head so a SIGKILL leaves trained heads durable. + Returns {n_trained, n_skipped} (n_skipped = unchanged + too-few-examples).""" import numpy as np cfg = _normalize_params(session, params) embedding_version = _embedder_version(session) + full = bool((params or {}).get("full")) eligible = _eligible_tag_ids(session, cfg["min_positives"]) eligible_set = set(eligible) # Computed once per run, not per head — the hygiene set is identical for # every non-system concept. hygiene = _hygiene_excluded_ids(session) + fps = _head_fingerprints(session, eligible) + to_train = set( + _heads_needing_retrain(session, eligible, embedding_version, fps, full) + ) trained = 0 - skipped = 0 + failed = 0 for i, tag_id in enumerate(eligible): + if tag_id not in to_train: + continue try: ok = train_head( session, tag_id, embedding_version, cfg, np, hygiene=hygiene @@ -175,9 +254,15 @@ def train_all_heads( except Exception: log.exception("train_head failed for tag %d", tag_id) ok = False + if ok: + # Stamp the fingerprint we trained against so an unchanged tag is + # skipped on the next incremental run. + head = session.get(TagHead, tag_id) + if head is not None: + head.train_fingerprint = fps.get(tag_id) session.commit() trained += int(ok) - skipped += int(not ok) + failed += int(not ok) if run is not None and i % 10 == 0: run.last_progress_at = datetime.now(UTC) session.commit() @@ -188,7 +273,11 @@ def train_all_heads( else: session.execute(delete(TagHead)) session.commit() - return {"n_trained": trained, "n_skipped": skipped} + # n_skipped = unchanged (not attempted) + failed-to-fit (too few examples). + return { + "n_trained": trained, + "n_skipped": (len(eligible) - len(to_train)) + failed, + } def head_training_ids( diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index bfb8d0c..4959099 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -356,7 +356,10 @@ def scheduled_train_heads() -> str: if running is not None: return "already running" run = HeadTrainingRun( - params={"source": "scheduled"}, status="running", + # Nightly = FULL reconcile (refit every head) so sampled-negative + + # hygiene drift is folded in; the manual Retrain button stays + # incremental (#1317 p2). + params={"source": "scheduled", "full": True}, status="running", last_progress_at=datetime.now(UTC), ) session.add(run) diff --git a/tests/test_head_incremental.py b/tests/test_head_incremental.py new file mode 100644 index 0000000..b16b9dd --- /dev/null +++ b/tests/test_head_incremental.py @@ -0,0 +1,136 @@ +"""Incremental head retraining (#1317 phase 2). The refit-decision + fingerprint +are split out sklearn-free (train_head itself needs scikit-learn), so they're +tested directly via db_sync.""" +import pytest +from sqlalchemy import select + +from backend.app.models import ( + ImageRecord, + MLSettings, + Tag, + TagHead, + TagKind, + TagSuggestionRejection, +) +from backend.app.models.tag import image_tag +from backend.app.services.ml.heads import ( + _head_fingerprints, + _heads_needing_retrain, +) + +pytestmark = pytest.mark.integration + + +def _img(db, sha: str) -> ImageRecord: + img = ImageRecord( + path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg", + width=1, height=1, origin="imported_filesystem", + integrity_status="unknown", + ) + db.add(img) + db.flush() + return img + + +def _tag(db, name: str) -> Tag: + t = Tag(name=name, kind=TagKind.general) + db.add(t) + db.flush() + return t + + +def _apply(db, image_id: int, tag_id: int) -> None: + db.execute(image_tag.insert().values( + image_record_id=image_id, tag_id=tag_id, source="manual", + )) + + +def _reject(db, image_id: int, tag_id: int) -> None: + db.add(TagSuggestionRejection(image_record_id=image_id, tag_id=tag_id)) + db.flush() + + +def _version(db) -> str: + return db.execute( + select(MLSettings.embedder_model_version).where(MLSettings.id == 1) + ).scalar_one() + + +def _head(db, tag_id: int, fp: str | None, version: str) -> None: + db.add(TagHead( + tag_id=tag_id, embedding_version=version, + weights=[0.0] * 1152, bias=0.0, suggest_threshold=0.5, + auto_apply_threshold=None, n_pos=10, n_neg=30, + ap=0.8, precision_cv=0.9, recall=0.6, train_fingerprint=fp, + )) + db.flush() + + +def test_fingerprint_changes_on_new_positive(db_sync): + tag = _tag(db_sync, "glasses") + i1 = _img(db_sync, "a" * 64) + _apply(db_sync, i1.id, tag.id) + db_sync.commit() + fp1 = _head_fingerprints(db_sync, [tag.id])[tag.id] + + i2 = _img(db_sync, "b" * 64) + _apply(db_sync, i2.id, tag.id) + db_sync.commit() + assert _head_fingerprints(db_sync, [tag.id])[tag.id] != fp1 + + +def test_fingerprint_changes_on_new_rejection(db_sync): + tag = _tag(db_sync, "glasses") + i1 = _img(db_sync, "c" * 64) + _apply(db_sync, i1.id, tag.id) + db_sync.commit() + fp1 = _head_fingerprints(db_sync, [tag.id])[tag.id] + + _reject(db_sync, i1.id, tag.id) + db_sync.commit() + assert _head_fingerprints(db_sync, [tag.id])[tag.id] != fp1 + + +def test_needing_retrain_selects_only_changed(db_sync): + ver = _version(db_sync) + a = _tag(db_sync, "A") + _apply(db_sync, _img(db_sync, "d" * 64).id, a.id) + b = _tag(db_sync, "B") + _apply(db_sync, _img(db_sync, "e" * 64).id, b.id) + c = _tag(db_sync, "C") + _apply(db_sync, _img(db_sync, "f" * 64).id, c.id) + db_sync.commit() + + ids = [a.id, b.id, c.id] + fps = _head_fingerprints(db_sync, ids) + _head(db_sync, a.id, fps[a.id], ver) # A: head with CURRENT fp → skip + _head(db_sync, b.id, "stale", ver) # B: head with STALE fp → retrain + db_sync.commit() # C: no head → retrain + + need = set(_heads_needing_retrain(db_sync, ids, ver, fps, full=False)) + assert a.id not in need + assert {b.id, c.id} <= need + + +def test_stale_embedding_version_forces_retrain(db_sync): + ver = _version(db_sync) + a = _tag(db_sync, "A") + _apply(db_sync, _img(db_sync, "g" * 64).id, a.id) + db_sync.commit() + fps = _head_fingerprints(db_sync, [a.id]) + # Matching fingerprint but a DIFFERENT embedding space → must refit. + _head(db_sync, a.id, fps[a.id], "old-model-v0") + db_sync.commit() + assert a.id in set(_heads_needing_retrain(db_sync, [a.id], ver, fps, full=False)) + + +def test_full_forces_all(db_sync): + ver = _version(db_sync) + a = _tag(db_sync, "A") + _apply(db_sync, _img(db_sync, "h" * 64).id, a.id) + db_sync.commit() + fps = _head_fingerprints(db_sync, [a.id]) + _head(db_sync, a.id, fps[a.id], ver) # current fp → would be skipped + db_sync.commit() + # full=True ignores the fingerprint (nightly reconcile). + assert a.id in set(_heads_needing_retrain(db_sync, [a.id], ver, fps, full=True))