feat(heads): incremental retraining — refit only changed tags (#1317 phase 2, m138)

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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM

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

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