refactor(ml): DRY pass — shared sweep helpers + table-driven settings (#161)

Consolidate duplication accrued across the ML tagging + settings backend, behavior-preserving (over-DRY guard applied — the three auto-apply sweep BODIES stay separate; only their shared inner helpers are extracted). - _sigmoid / _conflict_scores / _insert_presentation_review (heads.py): the score→prob transform (6 inlined sites), the presentation conflict signal (2 sites), and the ring-loud PresentationReview insert (2 sites, single- sourced so the mode column can't drift on the shared composite PK). - _applied_or_rejected (training_data.py): the per-tag "applied ∪ rejected" skip-set, byte-identical at 3 sweep sites (heads.py x2, tasks/ml.py ccip). - ccip sweep divergence fixes: import ccip._FIGURE_KINDS + training_data._l2norm instead of local copies that silently drift when the canonical changes. - MLSettings.load / .load_sync classmethods (mirror ImportSettings); route all 8 scalar_one singleton reads through them (the session.get None-path stays). - GET serializers for MLSettings + ImportSettings are now table-driven off the same _EDITABLE tuples PATCH writes, so a new field can't be silently absent from GET (the split that historically dropped fields). - AUTO_APPLY_THRESHOLD_MIN/MAX constant single-sources the [0.5,0.999] operating range across the service clamp + the 5 API validators. - test_ml_dry_helpers.py pins _applied_or_rejected + _sigmoid. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NsmJSQxnNxGgtM5Yz4GAqi
2026-07-13 21:41:24 -04:00
parent d80a5255ed
commit 666b3a2ec8
9 changed files with 187 additions and 172 deletions
@@ -256,9 +256,7 @@ async def lease():
        if not await _agent_authed(session):
            return jsonify({"error": "unauthorized"}), 401
        jobs = await GpuJobService(session).lease(agent_id, batch_size=batch)
-        ml = (
-            await session.execute(select(MLSettings).where(MLSettings.id == 1))
-        ).scalar_one()
+        ml = await MLSettings.load(session)
        # image rows for url/mime in one shot
        ids = [j.image_record_id for j in jobs]
        imgs = {
@@ -4,6 +4,7 @@ from quart import Blueprint, jsonify, request

 from ..extensions import get_session
 from ..models import MLSettings
+from ..services.ml.heads import AUTO_APPLY_THRESHOLD_MAX, AUTO_APPLY_THRESHOLD_MIN

 ml_admin_bp = Blueprint("ml_admin", __name__, url_prefix="/api/ml")

@@ -83,48 +84,21 @@ async def embedder_models():

@ml_admin_bp.route("/settings", methods=["GET"])
 async def get_settings():
-    from sqlalchemy import select
-
    async with get_session() as session:
-        s = (
-            await session.execute(select(MLSettings).where(MLSettings.id == 1))
-        ).scalar_one()
-    return jsonify(
-        {
-            "cpu_embed_enabled": s.cpu_embed_enabled,
-            "video_frame_interval_seconds": s.video_frame_interval_seconds,
-            "video_max_frames": s.video_max_frames,
-            "embedder_model_version": s.embedder_model_version,
-            "head_min_positives": s.head_min_positives,
-            "head_auto_apply_precision": s.head_auto_apply_precision,
-            "head_auto_apply_enabled": s.head_auto_apply_enabled,
-            "head_auto_apply_min_positives": s.head_auto_apply_min_positives,
-            "ccip_match_threshold": s.ccip_match_threshold,
-            "ccip_auto_apply_enabled": s.ccip_auto_apply_enabled,
-            "ccip_auto_apply_threshold": s.ccip_auto_apply_threshold,
-            "presentation_auto_apply_enabled": s.presentation_auto_apply_enabled,
-            "presentation_auto_apply_threshold": s.presentation_auto_apply_threshold,
-            "presentation_conflict_threshold": s.presentation_conflict_threshold,
-            "process_auto_apply_enabled": s.process_auto_apply_enabled,
-            "process_auto_apply_threshold": s.process_auto_apply_threshold,
-            "process_conflict_threshold": s.process_conflict_threshold,
-            "embedder_model_name": s.embedder_model_name,
-            **{f: getattr(s, f) for f in _DETECTOR_FIELDS},
-        }
-    )
+        s = await MLSettings.load(session)
+    # Table-driven off _EDITABLE (which PATCH also writes) so a new settings field
+    # can never be silently absent from GET — the split that historically dropped
+    # fields. _EDITABLE already includes *_DETECTOR_FIELDS.
+    return jsonify({f: getattr(s, f) for f in _EDITABLE})


@ml_admin_bp.route("/settings", methods=["PATCH"])
 async def patch_settings():
-    from sqlalchemy import select
-
    body = await request.get_json()
    if not isinstance(body, dict):
        return jsonify({"error": "body must be an object"}), 400
    async with get_session() as session:
-        s = (
-            await session.execute(select(MLSettings).where(MLSettings.id == 1))
-        ).scalar_one()
+        s = await MLSettings.load(session)

        # Merge the patch over current values, then validate the result as a
        # whole — the store-floor invariant couples three fields, so they
@@ -154,24 +128,24 @@ def _validate(p: dict) -> str | None:
    # Head training (#114).
    if int(p["head_min_positives"]) < 1:
        return "head_min_positives must be >= 1"
-    if not (0.5 <= float(p["head_auto_apply_precision"]) <= 0.999):
-        return "head_auto_apply_precision must be between 0.5 and 0.999"
+    if not (AUTO_APPLY_THRESHOLD_MIN <= float(p["head_auto_apply_precision"]) <= AUTO_APPLY_THRESHOLD_MAX):
+        return f"head_auto_apply_precision must be between {AUTO_APPLY_THRESHOLD_MIN} and {AUTO_APPLY_THRESHOLD_MAX}"
    if int(p["head_auto_apply_min_positives"]) < 1:
        return "head_auto_apply_min_positives must be >= 1"
-    if not (0.5 <= float(p["ccip_match_threshold"]) <= 0.999):
-        return "ccip_match_threshold must be between 0.5 and 0.999"
-    if not (0.5 <= float(p["ccip_auto_apply_threshold"]) <= 0.999):
-        return "ccip_auto_apply_threshold must be between 0.5 and 0.999"
+    if not (AUTO_APPLY_THRESHOLD_MIN <= float(p["ccip_match_threshold"]) <= AUTO_APPLY_THRESHOLD_MAX):
+        return f"ccip_match_threshold must be between {AUTO_APPLY_THRESHOLD_MIN} and {AUTO_APPLY_THRESHOLD_MAX}"
+    if not (AUTO_APPLY_THRESHOLD_MIN <= float(p["ccip_auto_apply_threshold"]) <= AUTO_APPLY_THRESHOLD_MAX):
+        return f"ccip_auto_apply_threshold must be between {AUTO_APPLY_THRESHOLD_MIN} and {AUTO_APPLY_THRESHOLD_MAX}"
    # Presentation chrome auto-hide (#141). Auto-apply runs high (hiding is
    # consequential); the conflict cut is a plain probability [0,1].
-    if not (0.5 <= float(p["presentation_auto_apply_threshold"]) <= 0.999):
-        return "presentation_auto_apply_threshold must be between 0.5 and 0.999"
+    if not (AUTO_APPLY_THRESHOLD_MIN <= float(p["presentation_auto_apply_threshold"]) <= AUTO_APPLY_THRESHOLD_MAX):
+        return f"presentation_auto_apply_threshold must be between {AUTO_APPLY_THRESHOLD_MIN} and {AUTO_APPLY_THRESHOLD_MAX}"
    if not (0.0 <= float(p["presentation_conflict_threshold"]) <= 1.0):
        return "presentation_conflict_threshold must be between 0 and 1"
    # Process auto-apply (#1464). wip/editor stay VISIBLE so a false apply is
    # low-harm (excludes-from-training + a review flag), but keep the same bar.
-    if not (0.5 <= float(p["process_auto_apply_threshold"]) <= 0.999):
-        return "process_auto_apply_threshold must be between 0.5 and 0.999"
+    if not (AUTO_APPLY_THRESHOLD_MIN <= float(p["process_auto_apply_threshold"]) <= AUTO_APPLY_THRESHOLD_MAX):
+        return f"process_auto_apply_threshold must be between {AUTO_APPLY_THRESHOLD_MIN} and {AUTO_APPLY_THRESHOLD_MAX}"
    if not (0.0 <= float(p["process_conflict_threshold"]) <= 1.0):
        return "process_conflict_threshold must be between 0 and 1"
    # Embedder model swap (#1190): both must be non-empty. Changing them means a
@@ -66,34 +66,9 @@ _EXTDL_TOGGLE_FIELDS = (
 async def get_import_settings():
    async with get_session() as session:
        row = await ImportSettings.load(session)
-    return jsonify({
-        "min_width": row.min_width,
-        "min_height": row.min_height,
-        "skip_transparent": row.skip_transparent,
-        "transparency_threshold": row.transparency_threshold,
-        "skip_single_color": row.skip_single_color,
-        "single_color_threshold": row.single_color_threshold,
-        "single_color_tolerance": row.single_color_tolerance,
-        "phash_threshold": row.phash_threshold,
-        "download_rate_limit_seconds": row.download_rate_limit_seconds,
-        "download_validate_files": row.download_validate_files,
-        "download_schedule_default_seconds": row.download_schedule_default_seconds,
-        "download_event_retention_days": row.download_event_retention_days,
-        "download_failure_warning_threshold": row.download_failure_warning_threshold,
-        "series_suggest_enabled": row.series_suggest_enabled,
-        "series_suggest_threshold": row.series_suggest_threshold,
-        "extdl_mega_enabled": row.extdl_mega_enabled,
-        "extdl_gdrive_enabled": row.extdl_gdrive_enabled,
-        "extdl_mediafire_enabled": row.extdl_mediafire_enabled,
-        "extdl_dropbox_enabled": row.extdl_dropbox_enabled,
-        "extdl_pixeldrain_enabled": row.extdl_pixeldrain_enabled,
-        "translation_enabled": row.translation_enabled,
-        "interpreter_base_url": row.interpreter_base_url,
-        "translation_target_lang": row.translation_target_lang,
-        "translation_min_confidence": row.translation_min_confidence,
-        "wip_title_tagging_enabled": row.wip_title_tagging_enabled,
-        "wip_soft_title_tagging_enabled": row.wip_soft_title_tagging_enabled,
-    })
+    # Table-driven off _EDITABLE_FIELDS (which PATCH also writes) so a new field
+    # can't be silently absent from GET.
+    return jsonify({f: getattr(row, f) for f in _EDITABLE_FIELDS})


@settings_bp.route("/settings/import", methods=["PATCH"])
@@ -10,6 +10,7 @@ from sqlalchemy import (
    Integer,
    String,
    func,
+    select,
 )
 from sqlalchemy.orm import Mapped, mapped_column

@@ -212,3 +213,14 @@ class MLSettings(Base):
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
+
+    @classmethod
+    async def load(cls, session) -> "MLSettings":
+        """The singleton settings row (id=1), via an async session. Mirrors
+        ImportSettings.load — the shared singleton-loader pattern."""
+        return (await session.execute(select(cls).where(cls.id == 1))).scalar_one()
+
+    @classmethod
+    def load_sync(cls, session) -> "MLSettings":
+        """The singleton settings row (id=1), via a sync session."""
+        return session.execute(select(cls).where(cls.id == 1)).scalar_one()
@@ -150,9 +150,7 @@ def refresh_character_prototypes(
    """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()
+    settings = MLSettings.load_sync(session)
    sig = _global_signature(session)
    if not full and settings.ccip_ref_signature == sig:
        return {"skipped": True, "rebuilt": 0, "removed": 0}
@@ -204,9 +202,7 @@ def retract_auto_applied_ccip(session: Session) -> int:
    n_retracted."""
    import numpy as np

-    settings = session.execute(
-        select(MLSettings).where(MLSettings.id == 1)
-    ).scalar_one()
+    settings = MLSettings.load_sync(session)
    if not settings.ccip_auto_apply_enabled:
        return 0
    thr = float(settings.ccip_auto_apply_threshold)
@@ -23,6 +23,7 @@ from datetime import UTC, datetime
 from typing import Any

 from sqlalchemy import delete, exists, func, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session

@@ -42,6 +43,7 @@ from ...models import (
 from ...models.tag import CHROME_SYSTEM_TAGS, PROCESS_SYSTEM_TAGS, image_tag
 from .training_data import (
    _AUTO_SOURCES,
+    _applied_or_rejected,
    _auto_apply_point,
    _hygiene_excluded_ids,
    _ids_with_tag,
@@ -61,6 +63,14 @@ MIN_POSITIVES_FLOOR = 8     # hard floor; settings.head_min_positives can raise
 _UNLABELED_POOL = 4000
 _EXAMPLES_MIN = 8           # need at least this many embedded +/- to fit a head

+# Auto-apply / match confidence operating range. Every graduated auto-apply or
+# CCIP-match threshold the operator can set lives in this band, and the head
+# precision target is clamped to it: below 0.5 "auto-apply" is meaningless, and
+# 1.0 is unachievable so 0.999 is the ceiling. One source shared by the service
+# clamp (_normalize_params) and the API validator (ml_admin._validate).
+AUTO_APPLY_THRESHOLD_MIN = 0.5
+AUTO_APPLY_THRESHOLD_MAX = 0.999
+
 # Only these tag kinds get heads (the surfaced suggestion categories).
 _HEAD_KINDS = (TagKind.general, TagKind.character)
 # tag.kind -> the suggestion category the rail groups under.
@@ -78,6 +88,38 @@ _CATEGORY = {TagKind.general: "general", TagKind.character: "character"}
 _SYSTEM_TAG_SUGGEST_FLOOR = 0.65


+def _sigmoid(z, np):
+    """Logistic sigmoid 1/(1+e^-z): the head score→probability transform. One home
+    for what was inlined at every scoring site (suggest, both sweeps, retract)."""
+    return 1.0 / (1.0 + np.exp(-z))
+
+
+def _conflict_scores(Xn, Wc, bc, np):
+    """The presentation conflict signal (#141): per row, the MAX content-head
+    probability and WHICH head produced it. Shared by the system-tag sweep's guard-2
+    and the soft-wip audit — both ask "does this ALSO look like real content?"."""
+    cprobs = _sigmoid(Xn @ Wc.T + bc, np)
+    return cprobs.max(axis=1), cprobs.argmax(axis=1)
+
+
+def _insert_presentation_review(
+    session, *, image_record_id, tag_id, conflict_tag_id, conflict_score, mode,
+):
+    """Single-source the ring-loud PresentationReview row shape so the two writers
+    (system-tag sweep guard-2 + soft-wip audit) can't drift on columns or `mode` —
+    they share the (image_record_id, tag_id) composite PK, so a divergent `mode`
+    would be a silent first-writer-wins bug."""
+    session.execute(
+        pg_insert(PresentationReview)
+        .values(
+            image_record_id=image_record_id, tag_id=tag_id,
+            conflict_tag_id=conflict_tag_id, conflict_score=conflict_score,
+            mode=mode,
+        )
+        .on_conflict_do_nothing()
+    )
+
+
 class HeadTrainingAlreadyRunning(Exception):
    """Raised by start_head_training_run when a run is already in flight."""

@@ -103,9 +145,7 @@ def start_head_training_run(session: Session, params: dict[str, Any]) -> int:


 def _settings(session: Session) -> MLSettings:
-    return session.execute(
-        select(MLSettings).where(MLSettings.id == 1)
-    ).scalar_one()
+    return MLSettings.load_sync(session)


 def _normalize_params(session: Session, params: dict[str, Any] | None) -> dict[str, Any]:
@@ -124,7 +164,7 @@ def _normalize_params(session: Session, params: dict[str, Any] | None) -> dict[s
    except (TypeError, ValueError):
        cv_folds = DEFAULT_CV_FOLDS
    try:
-        precision_target = min(max(float(params.get("precision_target", s.head_auto_apply_precision)), 0.5), 0.999)
+        precision_target = min(max(float(params.get("precision_target", s.head_auto_apply_precision)), AUTO_APPLY_THRESHOLD_MIN), AUTO_APPLY_THRESHOLD_MAX)
    except (TypeError, ValueError):
        precision_target = s.head_auto_apply_precision
    return {
@@ -536,7 +576,7 @@ async def score_image(
    norms[norms == 0] = 1.0
    Xn = X / norms
    Z = Xn @ heads["W"].T + heads["b"]                  # (B, H)
-    probs_bag = 1.0 / (1.0 + np.exp(-Z))                # (B, H)
+    probs_bag = _sigmoid(Z, np)                         # (B, H)
    probs = probs_bag.max(axis=0)                       # (H,) best over the bag
    # ARGMAX beside the max: WHICH bag row won each head → the region that grounds
    # the tag (bag_meta[win]); None when the whole-image vector won (#1206).
@@ -614,9 +654,7 @@ async def ground_applied_tag(


 async def _settings_async(session: AsyncSession) -> MLSettings:
-    return (
-        await session.execute(select(MLSettings).where(MLSettings.id == 1))
-    ).scalar_one()
+    return await MLSettings.load(session)


 # --- Earned auto-apply (sync, ml worker) ---------------------------------
@@ -687,7 +725,6 @@ def auto_apply_sweep(
    embeddings in chunks; commits per chunk on a real run. Returns
    {n_applied, concepts:[{tag_id,name,applied,scanned,threshold}]}."""
    import numpy as np
-    from sqlalchemy.dialects.postgresql import insert as pg_insert

    settings = _settings(session)
    rows = _auto_apply_heads(
@@ -704,18 +741,7 @@ def auto_apply_sweep(
    names = [r.name for r in rows]

    # Skip images that already carry, or have rejected, each tag.
-    skip = {tid: set() for tid in tag_ids}
-    for tid in tag_ids:
-        for (iid,) in session.execute(
-            select(image_tag.c.image_record_id).where(image_tag.c.tag_id == tid)
-        ):
-            skip[tid].add(iid)
-        for (iid,) in session.execute(
-            select(TagSuggestionRejection.image_record_id).where(
-                TagSuggestionRejection.tag_id == tid
-            )
-        ):
-            skip[tid].add(iid)
+    skip = _applied_or_rejected(session, tag_ids)

    applied = [0] * len(rows)
    scanned = 0
@@ -729,7 +755,7 @@ def auto_apply_sweep(
        if not cids:
            continue
        Xn = _l2norm(np.vstack([emb[i] for i in cids]).astype(np.float32), np)
-        probs = 1.0 / (1.0 + np.exp(-(Xn @ W.T + b)))   # (N, H)
+        probs = _sigmoid(Xn @ W.T + b, np)              # (N, H)
        scanned += len(cids)
        for h in range(len(rows)):
            tid = tag_ids[h]
@@ -840,7 +866,6 @@ def system_tag_auto_apply_sweep(
    enabled flag is set. numpy-only (no sklearn). Returns {n_applied, n_flagged,
    concepts}."""
    import numpy as np
-    from sqlalchemy.dialects.postgresql import insert as pg_insert

    cfg = _SWEEP_MODES[mode]
    settings = _settings(session)
@@ -869,18 +894,7 @@ def system_tag_auto_apply_sweep(
    valued = _valued_image_ids(session)

    # Skip images that already carry, or have rejected, each presentation tag.
-    skip = {tid: set() for tid in pres_tag_ids}
-    for tid in pres_tag_ids:
-        for (iid,) in session.execute(
-            select(image_tag.c.image_record_id).where(image_tag.c.tag_id == tid)
-        ):
-            skip[tid].add(iid)
-        for (iid,) in session.execute(
-            select(TagSuggestionRejection.image_record_id).where(
-                TagSuggestionRejection.tag_id == tid
-            )
-        ):
-            skip[tid].add(iid)
+    skip = _applied_or_rejected(session, pres_tag_ids)

    applied = [0] * len(pres)
    n_flagged = 0
@@ -895,11 +909,9 @@ def system_tag_auto_apply_sweep(
        if not cids:
            continue
        Xn = _l2norm(np.vstack([emb[i] for i in cids]).astype(np.float32), np)
-        probs = 1.0 / (1.0 + np.exp(-(Xn @ Wp.T + bp)))       # (N, P)
+        probs = _sigmoid(Xn @ Wp.T + bp, np)                  # (N, P)
        if Wc is not None:
-            cprobs = 1.0 / (1.0 + np.exp(-(Xn @ Wc.T + bc)))  # (N, C)
-            max_c = cprobs.max(axis=1)
-            arg_c = cprobs.argmax(axis=1)
+            max_c, arg_c = _conflict_scores(Xn, Wc, bc, np)   # (N,), (N,)
        scanned += len(cids)
        for p in range(len(pres)):
            tid = pres_tag_ids[p]
@@ -924,15 +936,12 @@ def system_tag_auto_apply_sweep(
                if Wc is not None and float(max_c[idx]) >= conflict_thr:
                    n_flagged += 1
                    if not dry_run:
-                        session.execute(
-                            pg_insert(PresentationReview)
-                            .values(
-                                image_record_id=iid, tag_id=tid,
-                                conflict_tag_id=conf_tag_ids[int(arg_c[idx])],
-                                conflict_score=float(max_c[idx]),
-                                mode=mode,
-                            )
-                            .on_conflict_do_nothing()
+                        _insert_presentation_review(
+                            session,
+                            image_record_id=iid, tag_id=tid,
+                            conflict_tag_id=conf_tag_ids[int(arg_c[idx])],
+                            conflict_score=float(max_c[idx]),
+                            mode=mode,
                        )
        if not dry_run:
            session.commit()
@@ -956,7 +965,6 @@ def soft_wip_conflict_audit(session: Session, dry_run: bool = False) -> dict:
    NOT remove the tag; the operator decides. No-op when there are no content heads.
    numpy-only. Returns {n_scanned, n_flagged}."""
    import numpy as np
-    from sqlalchemy.dialects.postgresql import insert as pg_insert

    from ..wip_title import WIP_TITLE_SOFT_SOURCE, resolve_wip_tag_id

@@ -993,22 +1001,17 @@ def soft_wip_conflict_audit(session: Session, dry_run: bool = False) -> dict:
            continue
        scanned += len(cids)
        Xn = _l2norm(np.vstack([emb[i] for i in cids]).astype(np.float32), np)
-        cprobs = 1.0 / (1.0 + np.exp(-(Xn @ Wc.T + bc)))
-        max_c = cprobs.max(axis=1)
-        arg_c = cprobs.argmax(axis=1)
+        max_c, arg_c = _conflict_scores(Xn, Wc, bc, np)
        for k in range(len(cids)):
            if float(max_c[k]) >= conflict_thr:
                n_flagged += 1
                if not dry_run:
-                    session.execute(
-                        pg_insert(PresentationReview)
-                        .values(
-                            image_record_id=cids[k], tag_id=wip_id,
-                            conflict_tag_id=conf_tag_ids[int(arg_c[k])],
-                            conflict_score=float(max_c[k]),
-                            mode="process",
-                        )
-                        .on_conflict_do_nothing()
+                    _insert_presentation_review(
+                        session,
+                        image_record_id=cids[k], tag_id=wip_id,
+                        conflict_tag_id=conf_tag_ids[int(arg_c[k])],
+                        conflict_score=float(max_c[k]),
+                        mode="process",
                    )
        if not dry_run:
            session.commit()
@@ -1062,7 +1065,7 @@ def retract_auto_applied_heads(session: Session) -> int:
            continue
        Xn = _l2norm(np.vstack([emb[i] for i in cids]).astype(np.float32), np)
        w = np.asarray(weights, dtype=np.float32)
-        probs = 1.0 / (1.0 + np.exp(-(Xn @ w + float(bias))))
+        probs = _sigmoid(Xn @ w + float(bias), np)
        below = [cids[k] for k in np.where(probs < float(thr))[0]]
        for iid in below:
            session.execute(
@@ -94,6 +94,24 @@ def _rejected_ids(session: Session, tag_id: int) -> list[int]:
    ]


+def _applied_or_rejected(session: Session, tag_ids) -> dict[int, set[int]]:
+    """Per-tag skip set for the auto-apply sweeps: every image that ALREADY carries
+    the tag (ANY source — not just training positives) OR has rejected it. A sweep
+    never re-applies to these. Shared by auto_apply_sweep + system_tag_auto_apply_sweep
+    (heads.py) and scheduled_ccip_auto_apply (tasks/ml.py). Callers mutate the returned
+    sets in-place to also dedupe within a single run."""
+    skip: dict[int, set[int]] = {}
+    for tid in tag_ids:
+        ids = {
+            r[0] for r in session.execute(
+                select(image_tag.c.image_record_id).where(image_tag.c.tag_id == tid)
+            ).all()
+        }
+        ids.update(_rejected_ids(session, tid))
+        skip[tid] = ids
+    return skip
+
+
 def _sample_unlabeled(session: Session, exclude: set[int], limit: int) -> list[int]:
    """Random image ids (with an embedding) NOT carrying the tag. Concepts are
    sparse, so an untagged image is almost always a true negative."""
@@ -105,9 +105,7 @@ def embed_image(self, image_id: int) -> dict:
            record = session.get(ImageRecord, image_id)
            if record is None:
                return {"status": "missing", "image_id": image_id}
-            settings = session.execute(
-                select(MLSettings).where(MLSettings.id == 1)
-            ).scalar_one()
+            settings = MLSettings.load_sync(session)

            src = Path(record.path)
            is_vid = _is_video(src)
@@ -488,15 +486,10 @@ def scheduled_ccip_auto_apply() -> str:
    from sqlalchemy import select as sa_select
    from sqlalchemy.dialects.postgresql import insert as pg_insert

-    from ..models import ImageRegion, MLSettings, Tag, TagKind, TagSuggestionRejection
+    from ..models import ImageRegion, MLSettings, Tag, TagKind
    from ..models.tag import image_tag
-
-    fig = ("face", "figure")
-
-    def _l2(m):
-        n = np.linalg.norm(m, axis=1, keepdims=True)
-        n[n == 0] = 1.0
-        return m / n
+    from ..services.ml.ccip import _FIGURE_KINDS
+    from ..services.ml.training_data import _applied_or_rejected, _l2norm

    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
@@ -521,7 +514,7 @@ def scheduled_ccip_auto_apply() -> str:
            )
            .join(Tag, Tag.id == image_tag.c.tag_id)
            .where(Tag.kind == TagKind.character)
-            .where(ImageRegion.kind.in_(fig))
+            .where(ImageRegion.kind.in_(_FIGURE_KINDS))
            .where(ImageRegion.ccip_embedding.is_not(None))
            .where(ImageRegion.image_record_id.in_(single))
        ).all()
@@ -532,29 +525,16 @@ def scheduled_ccip_auto_apply() -> str:
        for tid, vec in ref_rows:
            by_char.setdefault(tid, []).append(vec)
        ref_tags = list(by_char)
-        mats = [_l2(np.asarray(by_char[t], dtype=np.float32)) for t in ref_tags]
+        mats = [_l2norm(np.asarray(by_char[t], dtype=np.float32), np) for t in ref_tags]
        allref = np.vstack(mats)                                  # (total, 768)
        seg = np.cumsum([0] + [len(m) for m in mats])[:-1]        # per-char start

        # Per character: images that already carry OR rejected the tag — skip.
-        skip = {t: set() for t in ref_tags}
-        for t in ref_tags:
-            for (iid,) in session.execute(
-                sa_select(image_tag.c.image_record_id).where(
-                    image_tag.c.tag_id == t
-                )
-            ):
-                skip[t].add(iid)
-            for (iid,) in session.execute(
-                sa_select(TagSuggestionRejection.image_record_id).where(
-                    TagSuggestionRejection.tag_id == t
-                )
-            ):
-                skip[t].add(iid)
+        skip = _applied_or_rejected(session, ref_tags)

        img_ids = list(session.execute(
            sa_select(ImageRegion.image_record_id)
-            .where(ImageRegion.kind.in_(fig), ImageRegion.ccip_embedding.is_not(None))
+            .where(ImageRegion.kind.in_(_FIGURE_KINDS), ImageRegion.ccip_embedding.is_not(None))
            .distinct()
        ).scalars())

@@ -566,7 +546,7 @@ def scheduled_ccip_auto_apply() -> str:
                sa_select(ImageRegion.image_record_id, ImageRegion.ccip_embedding)
                .where(
                    ImageRegion.image_record_id.in_(chunk),
-                    ImageRegion.kind.in_(fig),
+                    ImageRegion.kind.in_(_FIGURE_KINDS),
                    ImageRegion.ccip_embedding.is_not(None),
                )
            ).all()
@@ -574,7 +554,7 @@ def scheduled_ccip_auto_apply() -> str:
            for iid, vec in rows:
                by_img.setdefault(iid, []).append(vec)
            for iid, vecs in by_img.items():
-                q = _l2(np.asarray(vecs, dtype=np.float32))      # (nq, 768)
+                q = _l2norm(np.asarray(vecs, dtype=np.float32), np)  # (nq, 768)
                colmax = (q @ allref.T).max(axis=0)              # (total,)
                charmax = np.maximum.reduceat(colmax, seg)       # (n_chars,)
                for ci in np.where(charmax >= thr)[0]: