Merge pull request 'Suggestions perf + UX polish: incremental CCIP + head training, hover label, scroll/toast fixes, dead-preview removal' (#196) from dev into main

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")

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")

backend/app/api/sources.py

@@ -230,52 +230,6 @@ async def set_backfill(source_id: int):
     return jsonify(record.to_dict())
 
 
-@sources_bp.route("/<int:source_id>/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("/<int:source_id>/check", methods=["POST"])
 async def check_source(source_id: int):
     """FC-3c: enqueue a download for this source.

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

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",

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()
+    )

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()
     )

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)

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,

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:

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())

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}

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(

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"
 
 
@@ -351,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)
@@ -361,6 +369,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,

frontend/src/components/modal/ImageCanvas.vue

@@ -35,7 +35,10 @@
         :class="{ 'fc-canvas__region--whole': !regionStyle }"
         :style="regionStyle || {}"
       >
-        <span class="fc-canvas__region-label">{{ regionLabel }}</span>
+        <span class="fc-canvas__region-label">
+          <span v-if="regionTag" class="fc-canvas__region-tag">{{ regionTag }}</span>
+          <span class="fc-canvas__region-origin">{{ regionOrigin }}</span>
+        </span>
       </div>
     </div>
   </div>
@@ -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));
 }

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

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 <input> 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

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++

frontend/src/components/subscriptions/PreviewDialog.vue

@@ -1,114 +0,0 @@
-<template>
-  <!-- Plan #708 B4: dry-run preview — what a backfill WOULD download, without
-       downloading. Self-fetches on open; loading / error / empty / result. -->
-  <v-dialog
-    :model-value="modelValue" max-width="520"
-    @update:model-value="$emit('update:modelValue', $event)"
-  >
-    <v-card v-if="source">
-      <CardHeading
-        icon="mdi-eye-outline"
-        :title="`Preview · ${source.artist_name}`"
-      />
-      <v-card-subtitle class="fc-preview__url">{{ source.url }}</v-card-subtitle>
-
-      <v-card-text>
-        <div v-if="loading" class="text-center py-8">
-          <v-progress-circular indeterminate color="accent" />
-          <div class="text-caption mt-3 text-medium-emphasis">Walking the feed…</div>
-        </div>
-
-        <v-alert
-          v-else-if="error" type="error" variant="tonal" density="comfortable"
-        >{{ error }}</v-alert>
-
-        <div v-else-if="result">
-          <div class="text-h5">
-            {{ result.total_new }} new item{{ result.total_new === 1 ? '' : 's' }}
-          </div>
-          <div class="text-body-2 text-medium-emphasis mb-3">
-            in the {{ result.posts_scanned }} most recent
-            post{{ result.posts_scanned === 1 ? '' : 's' }}<span
-              v-if="result.has_more"> · more pages not scanned</span>
-          </div>
-
-          <div
-            v-if="result.total_new === 0"
-            class="text-body-2 text-medium-emphasis"
-          >You’re caught up — nothing new to download.</div>
-
-          <v-list v-else density="compact" class="fc-preview__list" bg-color="transparent">
-            <v-list-item v-for="(row, i) in result.sample" :key="i" class="px-0">
-              <template #prepend>
-                <v-chip
-                  size="x-small" color="info" variant="tonal" label class="mr-3"
-                >+{{ row.new }}</v-chip>
-              </template>
-              <v-list-item-title class="text-body-2">{{ row.title }}</v-list-item-title>
-              <v-list-item-subtitle v-if="row.date" class="text-caption">
-                {{ formatRelative(row.date) }}
-              </v-list-item-subtitle>
-            </v-list-item>
-          </v-list>
-        </div>
-      </v-card-text>
-
-      <v-card-actions>
-        <v-spacer />
-        <v-btn variant="text" @click="$emit('update:modelValue', false)">Close</v-btn>
-        <v-btn
-          v-if="result && result.total_new > 0"
-          color="accent" variant="flat"
-          @click="$emit('backfill', source)"
-        >Start backfill</v-btn>
-      </v-card-actions>
-    </v-card>
-  </v-dialog>
-</template>
-
-<script setup>
-import { ref, watch } from 'vue'
-import { useSourcesStore } from '../../stores/sources.js'
-import { formatRelative } from '../../utils/date.js'
-import CardHeading from '../common/CardHeading.vue'
-
-const props = defineProps({
-  modelValue: { type: Boolean, default: false },
-  source: { type: Object, default: null },
-})
-defineEmits(['update:modelValue', 'backfill'])
-
-const store = useSourcesStore()
-const loading = ref(false)
-const error = ref('')
-const result = ref(null)
-
-watch(
-  () => props.modelValue,
-  async (open) => {
-    if (!open || !props.source) return
-    loading.value = true
-    error.value = ''
-    result.value = null
-    try {
-      result.value = await store.previewSource(props.source.id)
-    } catch (e) {
-      error.value = e?.body?.detail || e?.message || 'Preview failed'
-    } finally {
-      loading.value = false
-    }
-  },
-)
-</script>
-
-<style scoped>
-.fc-preview__url {
-  overflow: hidden;
-  text-overflow: ellipsis;
-  white-space: nowrap;
-}
-.fc-preview__list {
-  max-height: 320px;
-  overflow-y: auto;
-}
-</style>

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,

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 {

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)

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

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))

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,