chore: retire the tag-eval harness — it proved the heads system, job done (operator-approved)

The head-vs-centroid eval (#1130) existed to prove the 'frozen embedding +
trained head' spine; the operator accepted the tagging system and dropped the
harness. Removed per rule 22: TagEvalCard + store, /api/tag_eval blueprint,
tag_eval_run ml task, recover-stalled-tag-eval-runs sweep + beat entry,
TagEvalRun model + table (migration 0073), and its tests.

The eval's data loaders + metric helpers were NOT eval-specific — the nightly
heads trainer runs on them — so they moved verbatim to
services/ml/training_data.py (heads.py import updated; behavior unchanged).

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

backend/app/api/__init__.py

@@ -38,7 +38,6 @@ def all_blueprints() -> list[Blueprint]:
     from .suggestions import suggestions_bp
     from .system_activity import system_activity_bp
     from .system_backup import system_backup_bp
-    from .tag_eval import tag_eval_bp
     from .tags import tags_bp
     from .thumbnails import thumbnails_bp
     return [
@@ -58,7 +57,6 @@ def all_blueprints() -> list[Blueprint]:
         import_admin_bp,
         suggestions_bp,
         aliases_bp,
-        tag_eval_bp,
         heads_bp,
         gpu_bp,
         ccip_bp,

backend/app/api/tag_eval.py

@@ -1,70 +0,0 @@
-"""Tag-eval API (#1130): trigger + revisit the head-vs-centroid eval.
-
-The run + full report live in the tag_eval_run row, so the admin card rehydrates
-from GET (history / detail) on mount — the report survives navigation rather than
-living in transient frontend state.
-"""
-
-from quart import Blueprint, jsonify, request
-from sqlalchemy import select
-
-from ..extensions import get_session
-from ..models import TagEvalRun
-from ..services.ml.tag_eval import EvalAlreadyRunning, start_tag_eval_run
-
-tag_eval_bp = Blueprint("tag_eval", __name__, url_prefix="/api/tag-eval")
-
-
-def _serialize(run: TagEvalRun, *, include_report: bool) -> dict:
-    out = {
-        "id": run.id,
-        "params": run.params,
-        "status": run.status,
-        "started_at": run.started_at.isoformat() if run.started_at else None,
-        "finished_at": run.finished_at.isoformat() if run.finished_at else None,
-        "error": run.error,
-    }
-    if include_report:
-        out["report"] = run.report
-    return out
-
-
-@tag_eval_bp.route("", methods=["POST"])
-async def create():
-    body = await request.get_json(silent=True) or {}
-    params = body.get("params") or body or {}
-    async with get_session() as session:
-        try:
-            run_id = await session.run_sync(
-                lambda s: start_tag_eval_run(s, params)
-            )
-        except EvalAlreadyRunning as running:
-            return jsonify({
-                "error": "eval_already_running",
-                "running_id": int(running.args[0]),
-            }), 409
-        await session.commit()
-    return jsonify({"run_id": run_id, "status": "running"}), 202
-
-
-@tag_eval_bp.route("", methods=["GET"])
-async def history():
-    try:
-        limit = min(int(request.args.get("limit", "20")), 100)
-    except ValueError:
-        return jsonify({"error": "invalid_limit"}), 400
-    async with get_session() as session:
-        rows = (await session.execute(
-            select(TagEvalRun).order_by(TagEvalRun.id.desc()).limit(limit)
-        )).scalars().all()
-        # List is light — no full report (the detail endpoint carries it).
-        return jsonify({"runs": [_serialize(r, include_report=False) for r in rows]})
-
-
-@tag_eval_bp.route("/<int:run_id>", methods=["GET"])
-async def detail(run_id: int):
-    async with get_session() as session:
-        run = await session.get(TagEvalRun, run_id)
-        if run is None:
-            return jsonify({"error": "not_found"}), 404
-        return jsonify(_serialize(run, include_report=True))

backend/app/celery_app.py

@@ -183,10 +183,6 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.maintenance.recover_stalled_library_audit_runs",
                 "schedule": 300.0,
             },
-            "recover-stalled-tag-eval-runs": {
-                "task": "backend.app.tasks.maintenance.recover_stalled_tag_eval_runs",
-                "schedule": 300.0,
-            },
             "recover-stalled-head-training-runs": {
                 "task": "backend.app.tasks.maintenance.recover_stalled_head_training_runs",
                 "schedule": 300.0,

backend/app/models/__init__.py

@@ -33,7 +33,6 @@ from .subscribestar_failed_media import SubscribeStarFailedMedia
 from .subscribestar_seen_media import SubscribeStarSeenMedia
 from .tag import Tag, TagKind, image_tag
 from .tag_alias import TagAlias
-from .tag_eval_run import TagEvalRun
 from .tag_head import TagHead
 from .tag_positive_confirmation import TagPositiveConfirmation
 from .tag_suggestion_rejection import TagSuggestionRejection
@@ -75,7 +74,6 @@ __all__ = [
     "HeadMetricsSnapshot",
     "HeadTrainingRun",
     "TagAlias",
-    "TagEvalRun",
     "TagHead",
     "TagPositiveConfirmation",
     "TagSuggestionRejection",

backend/app/models/head_training_run.py

@@ -1,7 +1,7 @@
 """HeadTrainingRun — persisted lifecycle of a head-training batch (#114).
 
-Mirrors TagEvalRun so the run SURVIVES navigation and the admin card can show
-live + historical status instead of holding it in transient frontend state.
+A persisted run row (not transient frontend state) so the run SURVIVES
+navigation and the admin card can show live + historical status.
 Training is idempotent (it upserts tag_head rows), so a SIGKILL'd run is harmless
 — a maintenance recovery sweep flips a stalled `running` row to `error`, and the
 next run re-trains. State machine: running → ready / error.
@@ -37,8 +37,8 @@ class HeadTrainingRun(Base):
     n_trained: Mapped[int | None] = mapped_column(Integer, nullable=True)
     n_skipped: Mapped[int | None] = mapped_column(Integer, nullable=True)
     error: Mapped[str | None] = mapped_column(Text, nullable=True)
-    # Last time the task made progress — the recovery sweep tells a live run from
-    # a SIGKILL'd one by this (mirrors TagEvalRun).
+    # Last time the task made progress — the recovery sweep tells a live run
+    # from a SIGKILL'd one by this.
     last_progress_at: Mapped[datetime | None] = mapped_column(
         DateTime(timezone=True), nullable=True
     )

backend/app/models/tag_eval_run.py

@@ -1,45 +0,0 @@
-"""TagEvalRun — persisted lifecycle of a head-vs-centroid tagging eval (#1130).
-
-Mirrors LibraryAuditRun so the result SURVIVES navigation: the run + its full
-report live in this row, and the admin card rehydrates from it on mount instead
-of holding the report in transient frontend state. State machine:
-running → ready / error. The async ml-queue task writes `report` (JSONB) when
-done; a maintenance recovery sweep flips a stalled `running` row to `error`.
-"""
-
-from datetime import datetime
-from typing import Any
-
-from sqlalchemy import DateTime, Integer, String, Text, func
-from sqlalchemy.dialects.postgresql import JSONB
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class TagEvalRun(Base):
-    __tablename__ = "tag_eval_run"
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    # The eval parameters: {concepts: [...], curve_points: [...], neg_ratio,
-    # cv_folds, ...} — echoed back so the report is self-describing.
-    params: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)
-    status: Mapped[str] = mapped_column(
-        String(16), nullable=False, default="running", index=True,
-    )
-    # running | ready | error
-    started_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now(),
-    )
-    finished_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True,
-    )
-    # The full result: per-concept metrics (head vs centroid), learning-curve
-    # points, and example image ids. Null until the task finishes.
-    report: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True)
-    error: Mapped[str | None] = mapped_column(Text, nullable=True)
-    # Last time the task made progress — the recovery sweep tells a live run
-    # from a SIGKILL'd one by this (mirrors LibraryAuditRun).
-    last_progress_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True,
-    )

backend/app/services/ml/heads.py

@@ -1,12 +1,13 @@
 """Production heads: train + score the per-concept classifiers (#114).
 
-The eval (#1130, tag_eval.py) proved the spine; this is its production form.
+The eval harness (#1130) proved the spine, then retired 2026-07-02 once the
+tagging system was accepted; this is the production form.
 - TRAIN (sync, ml worker — needs scikit-learn): for every general/character tag
   with enough labelled positives, fit a logistic-regression head on the FROZEN
   SigLIP embeddings (positives + negatives = rejections + sampled unlabeled),
   derive an honest suggest threshold + earned-auto-apply point from CROSS-
-  VALIDATED scores, and upsert a TagHead row. Reuses tag_eval's proven data
-  loaders + metric helpers so production heads match the eval's measured numbers.
+  VALIDATED scores, and upsert a TagHead row. Uses the eval-proven data loaders
+  + metric helpers (training_data.py) so heads match the measured numbers.
 - SCORE (async, API worker — numpy via pgvector, NO scikit-learn): score one
   image's embedding against all current heads → the suggestions the rail shows,
   REPLACING Camie predictions + per-tag centroids.
@@ -37,7 +38,7 @@ from ...models import (
     TagSuggestionRejection,
 )
 from ...models.tag import image_tag
-from .tag_eval import (
+from .training_data import (
     _auto_apply_point,
     _ids_with_tag,
     _l2norm,

backend/app/services/ml/tag_eval.py

@@ -1,430 +0,0 @@
-"""Head-vs-centroid tagging eval (#1130, milestone #114 slice 1).
-
-Proves the "frozen embedding + small trained head (with negatives)" spine on the
-operator's OWN data, reusing the SigLIP embeddings already stored on
-image_record. For each concept tag it compares:
-  - CENTROID baseline (the old approach): cosine to the mean of positive vectors.
-  - HEAD (the new approach): logistic regression trained on positives + negatives.
-and reports cross-validated precision/recall/AP for both, a LEARNING CURVE
-(accuracy as the number of tagged positives grows), and example image ids to
-eyeball.
-
-numpy + scikit-learn are imported LAZILY inside run_eval so the API worker (base
-image, no ML stack) can still import start_tag_eval_run to enqueue the ml-queue
-task — the heavy compute only runs on the ml worker.
-"""
-
-from __future__ import annotations
-
-import logging
-from datetime import UTC, datetime
-from typing import Any
-
-from sqlalchemy import func, select
-from sqlalchemy.orm import Session
-
-from ...models import (
-    ImageRecord,
-    Tag,
-    TagEvalRun,
-    TagKind,
-    TagPositiveConfirmation,
-    TagSuggestionRejection,
-)
-from ...models.tag import image_tag
-
-log = logging.getLogger(__name__)
-
-# The operator's real concept list (mix of whole-ish + small/local cues). The
-# admin trigger can override; this is the default eval set.
-DEFAULT_CONCEPTS = [
-    "glasses", "cat", "dog", "horse", "goblin",
-    "cum", "lactation", "fellatio", "xray", "stomach bulge",
-]
-DEFAULT_CURVE_POINTS = [10, 30, 100, 300]
-DEFAULT_NEG_RATIO = 3       # negatives per positive (rejections + sampled unlabeled)
-DEFAULT_CV_FOLDS = 5
-MIN_POSITIVES = 8           # below this, a concept can't be evaluated meaningfully
-_UNLABELED_POOL = 4000      # cap on sampled unlabeled rows pulled per concept
-_EXAMPLES_K = 12
-
-
-def start_tag_eval_run(session: Session, params: dict[str, Any]) -> int:
-    """Create a TagEvalRun (status='running') and dispatch the ml-queue task.
-    Returns the new run id. Light guard: one running eval at a time."""
-    existing = session.execute(
-        select(TagEvalRun.id).where(TagEvalRun.status == "running")
-    ).scalar_one_or_none()
-    if existing is not None:
-        raise EvalAlreadyRunning(existing)
-    norm = _normalize_params(params)
-    run = TagEvalRun(params=norm, status="running", last_progress_at=datetime.now(UTC))
-    session.add(run)
-    session.flush()
-    run_id = run.id
-    # Same enqueue-by-import pattern api/suggestions.py uses for ml tasks; the
-    # commit happens in the API handler so row + dispatch are visible together.
-    from ...tasks.ml import tag_eval_run as _task
-    _task.delay(run_id)
-    return run_id
-
-
-class EvalAlreadyRunning(Exception):
-    """Raised by start_tag_eval_run when an eval is already in flight."""
-
-
-def _normalize_params(params: dict[str, Any] | None) -> dict[str, Any]:
-    params = params or {}
-    concepts = [str(c).strip() for c in (params.get("concepts") or []) if str(c).strip()]
-    try:
-        neg_ratio = max(1, int(params.get("neg_ratio", DEFAULT_NEG_RATIO)))
-    except (TypeError, ValueError):
-        neg_ratio = DEFAULT_NEG_RATIO
-    try:
-        cv_folds = max(2, int(params.get("cv_folds", DEFAULT_CV_FOLDS)))
-    except (TypeError, ValueError):
-        cv_folds = DEFAULT_CV_FOLDS
-    try:
-        auto_top_n = min(max(int(params.get("auto_top_n", 0) or 0), 0), 200)
-    except (TypeError, ValueError):
-        auto_top_n = 0
-    try:
-        precision_target = min(max(float(params.get("precision_target", 0.97)), 0.5), 0.999)
-    except (TypeError, ValueError):
-        precision_target = 0.97
-    # No explicit concepts and auto-discovery off → fall back to the hand list.
-    if not concepts and not auto_top_n:
-        concepts = list(DEFAULT_CONCEPTS)
-    curve = params.get("curve_points") or DEFAULT_CURVE_POINTS
-    curve = sorted({int(n) for n in curve if int(n) > 0})
-    return {
-        "concepts": concepts,
-        "neg_ratio": neg_ratio,
-        "cv_folds": cv_folds,
-        "auto_top_n": auto_top_n,
-        "precision_target": round(precision_target, 4),
-        "curve_points": curve,
-    }
-
-
-def _top_general_concepts(session: Session, n: int, min_count: int) -> list[str]:
-    """The n most-tagged general (concept) tags with >= min_count images — a fast
-    server-side way to broaden the eval beyond the hand-picked list (counts all
-    sources; source-aware filtering is a separate concern)."""
-    rows = session.execute(
-        select(Tag.name)
-        .join(image_tag, image_tag.c.tag_id == Tag.id)
-        .where(Tag.kind == TagKind.general)
-        .group_by(Tag.id)
-        .having(func.count(image_tag.c.image_record_id) >= min_count)
-        .order_by(func.count(image_tag.c.image_record_id).desc())
-        .limit(n)
-    ).all()
-    return [r[0] for r in rows]
-
-
-def _resolve_tag_id(session: Session, name: str) -> int | None:
-    """Case-insensitive tag-name match; if several share a name, take the one
-    applied to the most images (the one the operator actually uses)."""
-    rows = session.execute(
-        select(Tag.id, func.count(image_tag.c.image_record_id))
-        .outerjoin(image_tag, image_tag.c.tag_id == Tag.id)
-        .where(func.lower(Tag.name) == name.lower())
-        .group_by(Tag.id)
-        .order_by(func.count(image_tag.c.image_record_id).desc())
-    ).all()
-    return rows[0][0] if rows else None
-
-
-def _ids_with_tag(session: Session, tag_id: int) -> list[int]:
-    return [
-        r[0] for r in session.execute(
-            select(image_tag.c.image_record_id).where(image_tag.c.tag_id == tag_id)
-        ).all()
-    ]
-
-
-def _rejected_ids(session: Session, tag_id: int) -> list[int]:
-    return [
-        r[0] for r in session.execute(
-            select(TagSuggestionRejection.image_record_id)
-            .where(TagSuggestionRejection.tag_id == tag_id)
-        ).all()
-    ]
-
-
-def _confirmed_ids(session: Session, tag_id: int) -> set[int]:
-    """Positives the operator explicitly affirmed ('keep') — excluded from the
-    doubts list so confirmed-correct images don't resurface every run."""
-    return {
-        r[0] for r in session.execute(
-            select(TagPositiveConfirmation.image_record_id)
-            .where(TagPositiveConfirmation.tag_id == tag_id)
-        ).all()
-    }
-
-
-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."""
-    stmt = (
-        select(ImageRecord.id)
-        .where(ImageRecord.siglip_embedding.is_not(None))
-        .order_by(func.random())
-        .limit(limit)
-    )
-    if exclude:
-        stmt = stmt.where(ImageRecord.id.not_in(exclude))
-    return [r[0] for r in session.execute(stmt).all()]
-
-
-def _load_embeddings(session: Session, ids: list[int]) -> dict[int, Any]:
-    import numpy as np
-
-    out: dict[int, Any] = {}
-    if not ids:
-        return out
-    # Chunk the IN list to stay well under psycopg's parameter ceiling.
-    for i in range(0, len(ids), 2000):
-        chunk = ids[i:i + 2000]
-        for rid, emb in session.execute(
-            select(ImageRecord.id, ImageRecord.siglip_embedding)
-            .where(ImageRecord.id.in_(chunk))
-            .where(ImageRecord.siglip_embedding.is_not(None))
-        ).all():
-            out[rid] = np.asarray(emb, dtype=np.float32)
-    return out
-
-
-def run_eval(session: Session, params: dict[str, Any]) -> dict[str, Any]:
-    """Compute the full report. Per-concept failures are captured, not fatal."""
-    import numpy as np
-
-    cfg = _normalize_params(params)
-    # Auto-discovery: union the explicit concepts with the top-N most-tagged
-    # general tags (server-side, fast) so the eval can broaden itself.
-    concepts = list(cfg["concepts"])
-    if cfg["auto_top_n"]:
-        seen = {c.lower() for c in concepts}
-        for name in _top_general_concepts(session, cfg["auto_top_n"], MIN_POSITIVES):
-            if name.lower() not in seen:
-                concepts.append(name)
-                seen.add(name.lower())
-    cfg["concepts"] = concepts
-    concepts_out = []
-    for name in cfg["concepts"]:
-        try:
-            concepts_out.append(_eval_concept(session, name, cfg, np))
-        except Exception as exc:  # one bad concept shouldn't kill the run
-            log.exception("tag-eval concept %r failed", name)
-            concepts_out.append({"name": name, "skipped": f"error: {exc}"})
-    return {
-        "generated_at": datetime.now(UTC).isoformat(),
-        "params": cfg,
-        "concepts": concepts_out,
-    }
-
-
-def _eval_concept(session: Session, name: str, cfg: dict, np) -> dict[str, Any]:
-    tag_id = _resolve_tag_id(session, name)
-    if tag_id is None:
-        return {"name": name, "skipped": "no such tag"}
-    pos_ids = _ids_with_tag(session, tag_id)
-    if len(pos_ids) < MIN_POSITIVES:
-        return {"name": name, "tag_id": tag_id, "n_pos": len(pos_ids),
-                "skipped": f"too few positives (<{MIN_POSITIVES})"}
-
-    neg_ratio = cfg["neg_ratio"]
-    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) * neg_ratio, _EXAMPLES_K * 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]
-
-    emb = _load_embeddings(session, pos_ids + neg_ids)
-    pos = [(i, emb[i]) for i in pos_ids if i in emb]
-    neg = [(i, emb[i]) for i in neg_ids if i in emb]
-    if len(pos) < MIN_POSITIVES or len(neg) < MIN_POSITIVES:
-        return {"name": name, "tag_id": tag_id, "n_pos": len(pos),
-                "n_neg": len(neg), "skipped": "too few embedded examples"}
-
-    ids = np.array([i for i, _ in pos] + [i for i, _ in neg])
-    X = np.vstack([v for _, v in pos] + [v for _, v in neg]).astype(np.float32)
-    y = np.array([1] * len(pos) + [0] * len(neg))
-    Xn = _l2norm(X, np)
-
-    head = _eval_head(Xn, y, cfg["cv_folds"], cfg["precision_target"], np)
-    centroid = _eval_centroid(Xn, y, cfg["cv_folds"], np)
-    curve = _learning_curve(Xn, y, cfg["curve_points"], neg_ratio, np)
-    confirmed = _confirmed_ids(session, tag_id)
-    examples = _examples(session, Xn, y, ids, np, set(rejected), confirmed)
-
-    return {
-        "name": name, "tag_id": tag_id,
-        "n_pos": len(pos), "n_neg": len(neg),
-        "n_rejected": len(rejected),
-        "head": head, "centroid": centroid,
-        "curve": curve, "examples": examples,
-    }
-
-
-def _l2norm(X, np):
-    n = np.linalg.norm(X, axis=1, keepdims=True)
-    n[n == 0] = 1.0
-    return X / n
-
-
-def _metrics_from_scores(y, scores, np) -> dict[str, float]:
-    from sklearn.metrics import average_precision_score, precision_recall_curve
-
-    ap = float(average_precision_score(y, scores))
-    prec, rec, thr = precision_recall_curve(y, scores)
-    f1 = (2 * prec * rec) / np.clip(prec + rec, 1e-9, None)
-    best = int(np.argmax(f1))
-    # thr has len = len(prec)-1; map best index safely.
-    t = float(thr[min(best, len(thr) - 1)]) if len(thr) else 0.5
-    return {
-        "ap": round(ap, 4),
-        "precision": round(float(prec[best]), 4),
-        "recall": round(float(rec[best]), 4),
-        "f1": round(float(f1[best]), 4),
-        "threshold": round(t, 4),
-    }
-
-
-def _safe_folds(y, folds, np) -> int:
-    minority = int(min(np.bincount(y)))
-    return max(2, min(folds, minority))
-
-
-def _eval_head(Xn, y, folds, target, np) -> dict[str, float]:
-    from sklearn.linear_model import LogisticRegression
-    from sklearn.model_selection import StratifiedKFold, cross_val_predict
-
-    clf = LogisticRegression(max_iter=1000, class_weight="balanced")
-    cv = StratifiedKFold(n_splits=_safe_folds(y, folds, np), shuffle=True,
-                         random_state=0)
-    probs = cross_val_predict(clf, Xn, y, cv=cv, method="predict_proba")[:, 1]
-    m = _metrics_from_scores(y, probs, np)
-    m["auto_apply"] = _auto_apply_point(y, probs, target, np)
-    return m
-
-
-def _auto_apply_point(y, scores, target, np) -> dict | None:
-    """The auto-apply operating point: the threshold that yields the MOST recall
-    while holding precision >= target. This answers 'could this concept fire
-    without a human, and how much would it catch?' Returns None if no threshold
-    reaches the precision target (concept not auto-apply-ready)."""
-    from sklearn.metrics import precision_recall_curve
-
-    prec, rec, thr = precision_recall_curve(y, scores)
-    best = None  # (threshold, precision, recall) maximizing recall s.t. prec>=target
-    for i in range(len(thr)):  # thr[i] corresponds to prec[i], rec[i]
-        if prec[i] >= target and (best is None or rec[i] > best[2]):
-            best = (float(thr[i]), float(prec[i]), float(rec[i]))
-    if best is None:
-        return None
-    return {
-        "target": round(float(target), 4),
-        "threshold": round(best[0], 4),
-        "precision": round(best[1], 4),
-        "recall": round(best[2], 4),
-    }
-
-
-def _eval_centroid(Xn, y, folds, np) -> dict[str, float]:
-    """Cross-validated cosine-to-positive-mean — the OLD method's quality."""
-    from sklearn.model_selection import StratifiedKFold
-
-    cv = StratifiedKFold(n_splits=_safe_folds(y, folds, np), shuffle=True,
-                         random_state=0)
-    scores = np.zeros(len(y), dtype=np.float32)
-    for train, test in cv.split(Xn, y):
-        c = Xn[train][y[train] == 1].mean(axis=0)
-        cn = c / (np.linalg.norm(c) or 1.0)
-        scores[test] = Xn[test] @ cn
-    return _metrics_from_scores(y, scores, np)
-
-
-def _learning_curve(Xn, y, points, neg_ratio, np) -> list[dict[str, float]]:
-    """Hold out a fixed test split; train the head on a growing number of
-    positives and watch AP/F1 climb — answers 'does tagging more sharpen it?'"""
-    from sklearn.linear_model import LogisticRegression
-    from sklearn.model_selection import train_test_split
-
-    rng = np.random.default_rng(0)
-    idx = np.arange(len(y))
-    try:
-        tr, te = train_test_split(idx, test_size=0.3, stratify=y, random_state=0)
-    except ValueError:
-        return []
-    tr_pos = tr[y[tr] == 1]
-    tr_neg = tr[y[tr] == 0]
-    out = []
-    for n in points:
-        if n > len(tr_pos):
-            break
-        sp = rng.choice(tr_pos, size=n, replace=False)
-        nn = min(len(tr_neg), n * neg_ratio)
-        sn = rng.choice(tr_neg, size=nn, replace=False)
-        sub = np.concatenate([sp, sn])
-        clf = LogisticRegression(max_iter=1000, class_weight="balanced")
-        clf.fit(Xn[sub], y[sub])
-        prob = clf.predict_proba(Xn[te])[:, 1]
-        m = _metrics_from_scores(y[te], prob, np)
-        out.append({"n_pos": int(n), "ap": m["ap"], "f1": m["f1"]})
-    return out
-
-
-def _examples(session, Xn, y, ids, np, rejected_set, confirmed_set) -> dict[str, list[dict]]:
-    """Train on all data, then surface: top-scoring negatives the operator has
-    NOT already rejected (= fresh suggestions) and lowest-scoring POSITIVES the
-    operator has NOT already confirmed (= unreviewed doubts). Excluding rejected
-    ids stops an adjudicated near-miss from resurfacing in 'would suggest';
-    excluding confirmed ids stops a 'kept' correct positive from resurfacing in
-    'head doubts' every run. Resolves thumbnail urls for a self-contained report."""
-    from sklearn.linear_model import LogisticRegression
-
-    clf = LogisticRegression(max_iter=1000, class_weight="balanced")
-    clf.fit(Xn, y)
-    s = clf.predict_proba(Xn)[:, 1]
-    neg_idx = np.where(y == 0)[0]
-    pos_idx = np.where(y == 1)[0]
-    top_neg = []
-    for i in neg_idx[np.argsort(s[neg_idx])[::-1]]:  # high score → low
-        rid = int(ids[i])
-        if rid in rejected_set:
-            continue  # already told the head 'no' — don't re-suggest it
-        top_neg.append(rid)
-        if len(top_neg) >= _EXAMPLES_K:
-            break
-    low_pos = []
-    for i in pos_idx[np.argsort(s[pos_idx])]:  # low score → high
-        rid = int(ids[i])
-        if rid in confirmed_set:
-            continue  # already kept/confirmed — don't re-doubt it
-        low_pos.append(rid)
-        if len(low_pos) >= _EXAMPLES_K:
-            break
-    thumbs = _resolve_thumbs(session, top_neg + low_pos)
-    return {
-        "head_would_suggest": [thumbs[i] for i in top_neg if i in thumbs],
-        "head_doubts_positive": [thumbs[i] for i in low_pos if i in thumbs],
-    }
-
-
-def _resolve_thumbs(session, ids: list[int]) -> dict[int, dict]:
-    from ..gallery_service import thumbnail_url
-
-    out: dict[int, dict] = {}
-    if not ids:
-        return out
-    for rid, tp, sha, mime in session.execute(
-        select(
-            ImageRecord.id, ImageRecord.thumbnail_path,
-            ImageRecord.sha256, ImageRecord.mime,
-        ).where(ImageRecord.id.in_(ids))
-    ).all():
-        out[rid] = {"id": rid, "thumbnail_url": thumbnail_url(tp, sha, mime)}
-    return out

backend/app/services/ml/training_data.py

@@ -0,0 +1,121 @@
+"""Shared data-selection + validated-metric helpers for the heads trainer.
+
+Born in the head-vs-centroid eval harness (#1130, tag_eval.py) that proved the
+"frozen embedding + small trained head (with negatives)" spine; the harness was
+retired 2026-07-02 (operator: the tagging system is proven, the eval isn't
+needed) and these survivors moved here — they ARE the heads' production data
+pipeline (heads.py trains and scores with them nightly).
+
+numpy/scikit-learn are imported lazily inside the functions that need them so
+the API worker (base image, no ML stack) can import this module.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from ...models import ImageRecord, TagSuggestionRejection
+from ...models.tag import image_tag
+
+
+def _ids_with_tag(session: Session, tag_id: int) -> list[int]:
+    return [
+        r[0] for r in session.execute(
+            select(image_tag.c.image_record_id).where(image_tag.c.tag_id == tag_id)
+        ).all()
+    ]
+
+
+def _rejected_ids(session: Session, tag_id: int) -> list[int]:
+    return [
+        r[0] for r in session.execute(
+            select(TagSuggestionRejection.image_record_id)
+            .where(TagSuggestionRejection.tag_id == tag_id)
+        ).all()
+    ]
+
+
+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."""
+    stmt = (
+        select(ImageRecord.id)
+        .where(ImageRecord.siglip_embedding.is_not(None))
+        .order_by(func.random())
+        .limit(limit)
+    )
+    if exclude:
+        stmt = stmt.where(ImageRecord.id.not_in(exclude))
+    return [r[0] for r in session.execute(stmt).all()]
+
+
+def _load_embeddings(session: Session, ids: list[int]) -> dict[int, Any]:
+    import numpy as np
+
+    out: dict[int, Any] = {}
+    if not ids:
+        return out
+    # Chunk the IN list to stay well under psycopg's parameter ceiling.
+    for i in range(0, len(ids), 2000):
+        chunk = ids[i:i + 2000]
+        for rid, emb in session.execute(
+            select(ImageRecord.id, ImageRecord.siglip_embedding)
+            .where(ImageRecord.id.in_(chunk))
+            .where(ImageRecord.siglip_embedding.is_not(None))
+        ).all():
+            out[rid] = np.asarray(emb, dtype=np.float32)
+    return out
+
+
+def _l2norm(X, np):
+    n = np.linalg.norm(X, axis=1, keepdims=True)
+    n[n == 0] = 1.0
+    return X / n
+
+
+def _metrics_from_scores(y, scores, np) -> dict[str, float]:
+    from sklearn.metrics import average_precision_score, precision_recall_curve
+
+    ap = float(average_precision_score(y, scores))
+    prec, rec, thr = precision_recall_curve(y, scores)
+    f1 = (2 * prec * rec) / np.clip(prec + rec, 1e-9, None)
+    best = int(np.argmax(f1))
+    # thr has len = len(prec)-1; map best index safely.
+    t = float(thr[min(best, len(thr) - 1)]) if len(thr) else 0.5
+    return {
+        "ap": round(ap, 4),
+        "precision": round(float(prec[best]), 4),
+        "recall": round(float(rec[best]), 4),
+        "f1": round(float(f1[best]), 4),
+        "threshold": round(t, 4),
+    }
+
+
+def _safe_folds(y, folds, np) -> int:
+    minority = int(min(np.bincount(y)))
+    return max(2, min(folds, minority))
+
+
+def _auto_apply_point(y, scores, target, np) -> dict | None:
+    """The auto-apply operating point: the threshold that yields the MOST recall
+    while holding precision >= target. This answers 'could this concept fire
+    without a human, and how much would it catch?' Returns None if no threshold
+    reaches the precision target (concept not auto-apply-ready)."""
+    from sklearn.metrics import precision_recall_curve
+
+    prec, rec, thr = precision_recall_curve(y, scores)
+    best = None  # (threshold, precision, recall) maximizing recall s.t. prec>=target
+    for i in range(len(thr)):  # thr[i] corresponds to prec[i], rec[i]
+        if prec[i] >= target and (best is None or rec[i] > best[2]):
+            best = (float(thr[i]), float(prec[i]), float(rec[i]))
+    if best is None:
+        return None
+    return {
+        "target": round(float(target), 4),
+        "threshold": round(best[0], 4),
+        "precision": round(best[1], 4),
+        "recall": round(best[2], 4),
+    }

backend/app/tasks/maintenance.py

@@ -21,7 +21,6 @@ from ..models import (
     ImportTask,
     LibraryAuditRun,
     Source,
-    TagEvalRun,
     TaskRun,
 )
 from ..utils.phash import compute_phash
@@ -96,9 +95,6 @@ BACKUP_DB_STALL_THRESHOLD_MINUTES = 40
 # Library audit: scan_library_for_rule has time_limit=7500s (2h5m).
 # 2h15m gives a 10-min buffer.
 LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES = 135
-# tag-eval (#1130) has a 30-min soft limit; flag a run with no progress past 40.
-TAG_EVAL_STALL_THRESHOLD_MINUTES = 40
-TAG_EVAL_KEEP_RUNS = 20
 # head training (#114) has a 60-min soft limit; flag no-progress past 75.
 HEAD_TRAINING_STALL_THRESHOLD_MINUTES = 75
 HEAD_TRAINING_KEEP_RUNS = 20
@@ -743,46 +739,6 @@ def recover_stalled_library_audit_runs() -> int:
         return recovered
 
 
-@celery.task(name="backend.app.tasks.maintenance.recover_stalled_tag_eval_runs")
-def recover_stalled_tag_eval_runs() -> int:
-    """Flip TagEvalRun rows stuck in 'running' past the stall threshold to
-    'error', and prune old runs to the last TAG_EVAL_KEEP_RUNS (retention,
-    rule 89). Runs every 5 min on the maintenance lane; no-op when idle."""
-    SessionLocal = _sync_session_factory()
-    now = datetime.now(UTC)
-    cutoff = now - timedelta(minutes=TAG_EVAL_STALL_THRESHOLD_MINUTES)
-    with SessionLocal() as session:
-        result = session.execute(
-            update(TagEvalRun)
-            .where(TagEvalRun.status == "running")
-            .where(
-                func.coalesce(TagEvalRun.last_progress_at, TagEvalRun.started_at)
-                < cutoff
-            )
-            .values(
-                status="error", finished_at=now,
-                error=(
-                    f"stranded by recovery sweep (no progress for "
-                    f"{TAG_EVAL_STALL_THRESHOLD_MINUTES} min)"
-                ),
-            )
-        )
-        # Retention: keep only the most recent N runs.
-        keep = session.execute(
-            select(TagEvalRun.id).order_by(TagEvalRun.id.desc())
-            .limit(TAG_EVAL_KEEP_RUNS)
-        ).scalars().all()
-        if keep:
-            session.execute(
-                delete(TagEvalRun).where(TagEvalRun.id.not_in(keep))
-            )
-        session.commit()
-        recovered = result.rowcount or 0
-        if recovered:
-            log.info("recover_stalled_tag_eval_runs: recovered %d rows", recovered)
-        return recovered
-
-
 @celery.task(name="backend.app.tasks.maintenance.recover_stalled_head_training_runs")
 def recover_stalled_head_training_runs() -> int:
     """Flip HeadTrainingRun rows stuck in 'running' past the stall threshold to

backend/app/tasks/ml.py

@@ -250,51 +250,6 @@ def backfill(self) -> int:
     return enqueued
 
 
-@celery.task(
-    name="backend.app.tasks.ml.tag_eval_run",
-    bind=True,
-    # The head-vs-centroid eval (#1130) loads embeddings + fits sklearn heads
-    # for several concepts — minutes, not seconds. Runs on the ml queue because
-    # only that worker has numpy/scikit-learn.
-    soft_time_limit=1800, time_limit=2100,
-)
-def tag_eval_run(self, run_id: int) -> str:
-    """Compute the eval report into the persisted TagEvalRun row so it survives
-    navigation (the admin card rehydrates from the row, not transient state)."""
-    from datetime import UTC, datetime
-
-    from ..models import TagEvalRun
-    from ..services.ml.tag_eval import run_eval
-
-    SessionLocal = _sync_session_factory()
-    with SessionLocal() as session:
-        run = session.get(TagEvalRun, run_id)
-        if run is None:
-            return "missing"
-        run.last_progress_at = datetime.now(UTC)
-        session.commit()
-        try:
-            report = run_eval(session, run.params)
-        except SoftTimeLimitExceeded:
-            run.status = "error"
-            run.error = "timed out"
-            run.finished_at = datetime.now(UTC)
-            session.commit()
-            raise
-        except Exception as exc:
-            log.exception("tag_eval_run %d failed", run_id)
-            run.status = "error"
-            run.error = str(exc)
-            run.finished_at = datetime.now(UTC)
-            session.commit()
-            return "error"
-        run.report = report
-        run.status = "ready"
-        run.finished_at = datetime.now(UTC)
-        session.commit()
-        return "ready"
-
-
 @celery.task(
     name="backend.app.tasks.ml.train_heads",
     bind=True,