feat(ml): training hygiene — system-tagged images are absent from other concepts training

Step 2 of milestone #128. _hygiene_excluded_ids (training_data.py) is the
one shared predicate: images carrying any system tag are dropped from
every OTHER concepts head training — not positives (a rough wip tagged
as a character drags the head toward generic-sketch) and not rejection
or sampled negatives (a wip OF character X is not evidence against X).
A system tags own head trains on them unfiltered; that is what makes
auto-flagging banners work. Selection is split out of train_head as the
sklearn-free head_training_ids so CI (no sklearn) can pin the behavior.

CCIP: reference prototypes skip hygiene-tagged images — a faceless wip
figure region must never become an identity reference — and the ref
cache signature now counts hygiene applications, since tagging an image
wip changes the reference set without touching character/region counts.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM

backend/app/services/ml/ccip.py

@@ -62,6 +62,18 @@ def _single_character_images():
     )
 
 
+def _hygiene_tagged_images():
+    """Subquery of image ids carrying any SYSTEM tag (wip / banner / editor
+    screenshot). Training hygiene (#128): such images never contribute
+    reference prototypes — a faceless wip's figure region would otherwise
+    become an identity reference for the character it's tagged with."""
+    return (
+        select(image_tag.c.image_record_id)
+        .join(Tag, Tag.id == image_tag.c.tag_id)
+        .where(Tag.is_system.is_(True))
+    )
+
+
 async def _ref_signature(session: AsyncSession) -> tuple:
     n_tags = (
         await session.execute(
@@ -79,7 +91,17 @@ async def _ref_signature(session: AsyncSession) -> tuple:
             )
         )
     ).one()
-    return (n_tags, n_regs, max_id)
+    # Hygiene applications must invalidate too: tagging an image `wip` changes
+    # the reference set without touching character-tag or region counts.
+    n_hygiene = (
+        await 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 (n_tags, n_regs, max_id, n_hygiene)
 
 
 async def character_references(session: AsyncSession) -> dict[int, list]:
@@ -102,6 +124,9 @@ async def character_references(session: AsyncSession) -> dict[int, list]:
             .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()
     refs: dict[int, list] = {}

backend/app/services/ml/heads.py

@@ -40,6 +40,7 @@ from ...models import (
 from ...models.tag import image_tag
 from .training_data import (
     _auto_apply_point,
+    _hygiene_excluded_ids,
     _ids_with_tag,
     _l2norm,
     _load_embeddings,
@@ -150,11 +151,16 @@ def train_all_heads(
     embedding_version = _embedder_version(session)
     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)
     trained = 0
     skipped = 0
     for i, tag_id in enumerate(eligible):
         try:
-            ok = train_head(session, tag_id, embedding_version, cfg, np)
+            ok = train_head(
+                session, tag_id, embedding_version, cfg, np, hygiene=hygiene
+            )
         except Exception:
             log.exception("train_head failed for tag %d", tag_id)
             ok = False
@@ -174,27 +180,56 @@ def train_all_heads(
     return {"n_trained": trained, "n_skipped": skipped}
 
 
+def head_training_ids(
+    session: Session, tag_id: int, cfg: dict, hygiene: set[int] | None = None,
+) -> tuple[list[int], list[int]] | None:
+    """Select (pos_ids, neg_ids) for one head. Split out of train_head and
+    kept sklearn-free so the hygiene exclusion is testable in the CI env
+    (sklearn only exists in the ml image). Returns None when the concept has
+    too few usable positives.
+
+    Training hygiene (#128): images carrying a system tag are ABSENT from
+    every other concept's training — dropped as positives AND kept out of
+    the rejection/sampled negative pool (see _hygiene_excluded_ids). A system
+    tag's own head trains on them unfiltered: its positives ARE the hygiene
+    images."""
+    tag = session.get(Tag, tag_id)
+    if tag is not None and tag.is_system:
+        hygiene = set()
+    elif hygiene is None:
+        hygiene = _hygiene_excluded_ids(session)
+
+    pos_ids = [i for i in _ids_with_tag(session, tag_id) if i not in hygiene]
+    if len(pos_ids) < cfg["min_positives"]:
+        return None
+
+    pos_set = set(pos_ids)
+    rejected = [
+        i for i in _rejected_ids(session, tag_id)
+        if i not in pos_set and i not in hygiene
+    ]
+    want_neg = max(len(pos_ids) * cfg["neg_ratio"], _EXAMPLES_MIN * 4)
+    sampled = _sample_unlabeled(
+        session, pos_set | set(rejected) | hygiene,
+        min(_UNLABELED_POOL, want_neg),
+    )
+    return pos_ids, rejected + [i for i in sampled if i not in pos_set]
+
+
 def train_head(
-    session: Session, tag_id: int, embedding_version: str, cfg: dict, np
+    session: Session, tag_id: int, embedding_version: str, cfg: dict, np,
+    hygiene: set[int] | None = None,
 ) -> bool:
     """Fit + upsert one head. Returns True if a head was written, False if the
     concept had too few usable examples to train (the row is then removed)."""
     from sklearn.linear_model import LogisticRegression
     from sklearn.model_selection import StratifiedKFold, cross_val_predict
 
-    pos_ids = _ids_with_tag(session, tag_id)
-    if len(pos_ids) < cfg["min_positives"]:
+    ids = head_training_ids(session, tag_id, cfg, hygiene)
+    if ids is None:
         session.execute(delete(TagHead).where(TagHead.tag_id == tag_id))
         return False
-
-    pos_set = set(pos_ids)
-    rejected = [i for i in _rejected_ids(session, tag_id) if i not in pos_set]
-    want_neg = max(len(pos_ids) * cfg["neg_ratio"], _EXAMPLES_MIN * 4)
-    sampled = _sample_unlabeled(
-        session, pos_set | set(rejected), min(_UNLABELED_POOL, want_neg)
-    )
-    neg_ids = rejected + [i for i in sampled if i not in pos_set]
-
+    pos_ids, neg_ids = ids
     emb = _load_embeddings(session, pos_ids + neg_ids)
     pos = [emb[i] for i in pos_ids if i in emb]
     neg = [emb[i] for i in neg_ids if i in emb]

backend/app/services/ml/training_data.py

@@ -17,10 +17,33 @@ from typing import Any
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
-from ...models import ImageRecord, TagSuggestionRejection
+from ...models import ImageRecord, Tag, TagSuggestionRejection
 from ...models.tag import image_tag
 
 
+def _hygiene_excluded_ids(session: Session) -> set[int]:
+    """Ids of images carrying ANY system tag (wip / banner / editor
+    screenshot — milestone #128). These images are excluded from OTHER
+    concepts' head training entirely: not positives (a rough wip tagged as a
+    character drags that head toward 'generic sketch') and not sampled or
+    rejection negatives (a wip OF character X is not evidence against X) —
+    simply absent. A system tag's OWN head trains on them unchanged; that is
+    what makes auto-flagging banners/editor screenshots work.
+
+    Item-level by design: a wip-tagged process video contributes (or
+    withholds) ALL its sampled frames, though some may show the finished
+    piece. Operator call 2026-07-03: with enough clean data this washes out —
+    no per-frame handling.
+    """
+    return set(
+        session.execute(
+            select(image_tag.c.image_record_id)
+            .join(Tag, Tag.id == image_tag.c.tag_id)
+            .where(Tag.is_system.is_(True))
+        ).scalars().all()
+    )
+
+
 def _ids_with_tag(session: Session, tag_id: int) -> list[int]:
     return [
         r[0] for r in session.execute(