alembic/versions/0056_tag_eval_run.py

@@ -0,0 +1,43 @@
+"""tag_eval_run: persisted head-vs-centroid tagging eval runs (#1130)
+
+Milestone #114 slice 1. A long ml-queue eval whose full report must SURVIVE
+navigation, so the run + report live in a row the admin card rehydrates from
+(mirrors library_audit_run). running -> ready / error.
+
+Revision ID: 0056
+Revises: 0055
+Create Date: 2026-06-28
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects.postgresql import JSONB
+
+revision: str = "0056"
+down_revision: Union[str, None] = "0055"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "tag_eval_run",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("params", JSONB(), nullable=False),
+        sa.Column("status", sa.String(length=16), nullable=False, server_default="running"),
+        sa.Column(
+            "started_at", sa.DateTime(timezone=True), nullable=False,
+            server_default=sa.func.now(),
+        ),
+        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("report", JSONB(), nullable=True),
+        sa.Column("error", sa.Text(), nullable=True),
+        sa.Column("last_progress_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.create_index("ix_tag_eval_run_status", "tag_eval_run", ["status"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_tag_eval_run_status", table_name="tag_eval_run")
+    op.drop_table("tag_eval_run")

backend/app/api/__init__.py

@@ -36,6 +36,7 @@ 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 [
@@ -56,6 +57,7 @@ def all_blueprints() -> list[Blueprint]:
         suggestions_bp,
         allowlist_bp,
         aliases_bp,
+        tag_eval_bp,
         ml_admin_bp,
         thumbnails_bp,
         sources_bp,

backend/app/api/tag_eval.py

@@ -0,0 +1,70 @@
+"""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

@@ -156,6 +156,10 @@ 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-import-batches": {
                 "task": "backend.app.tasks.maintenance.recover_stalled_import_batches",
                 "schedule": 300.0,

backend/app/models/__init__.py

@@ -29,6 +29,7 @@ from .subscribestar_seen_media import SubscribeStarSeenMedia
 from .tag import Tag, TagKind, image_tag
 from .tag_alias import TagAlias
 from .tag_allowlist import TagAllowlist
+from .tag_eval_run import TagEvalRun
 from .tag_reference_embedding import TagReferenceEmbedding
 from .tag_suggestion_rejection import TagSuggestionRejection
 from .task_run import TaskRun
@@ -65,6 +66,7 @@ __all__ = [
     "MLSettings",
     "TagAlias",
     "TagAllowlist",
+    "TagEvalRun",
     "TagReferenceEmbedding",
     "TagSuggestionRejection",
     "TaskRun",

backend/app/models/tag_eval_run.py

@@ -0,0 +1,45 @@
+"""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/tag_eval.py

@@ -0,0 +1,335 @@
+"""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, 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 = params.get("concepts") or DEFAULT_CONCEPTS
+    concepts = [str(c).strip() for c in concepts 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
+    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,
+        "curve_points": curve,
+    }
+
+
+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 _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)
+    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"], np)
+    centroid = _eval_centroid(Xn, y, cfg["cv_folds"], np)
+    curve = _learning_curve(Xn, y, cfg["curve_points"], neg_ratio, np)
+    examples = _examples(session, Xn, y, ids, np)
+
+    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, 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]
+    return _metrics_from_scores(y, probs, np)
+
+
+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) -> dict[str, list[dict]]:
+    """Train on all data, then surface: top-scoring UNLABELED-ish (highest among
+    the negative pool = what the head would newly suggest) and lowest-scoring
+    POSITIVES (where the head disagrees with the operator's tag — likely the
+    most informative to review). Resolves thumbnail urls so the stored report
+    renders without per-id lookups (survives navigation as a self-contained
+    artifact)."""
+    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 = [int(ids[i]) for i in neg_idx[np.argsort(s[neg_idx])[::-1][:_EXAMPLES_K]]]
+    low_pos = [int(ids[i]) for i in pos_idx[np.argsort(s[pos_idx])[:_EXAMPLES_K]]]
+    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/tasks/maintenance.py

@@ -19,6 +19,7 @@ from ..models import (
     ImportTask,
     LibraryAuditRun,
     Source,
+    TagEvalRun,
     TaskRun,
 )
 from ..utils.phash import compute_phash
@@ -93,6 +94,9 @@ 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
 # Import batches finalize only after every child ImportTask hits a
 # terminal state. The recovery sweep targets the case where every
 # task is done but the batch never got its closing UPDATE
@@ -709,6 +713,46 @@ 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_import_batches")
 def recover_stalled_import_batches() -> int:
     """Finalize ImportBatch rows stuck in running past the hard limit

backend/app/tasks/ml.py

@@ -538,3 +538,48 @@ def recompute_centroids(self) -> int:
     for tid in drifted:
         recompute_centroid.delay(tid)
     return len(drifted)
+
+
+@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"

frontend/src/components/settings/AliasTable.vue

@@ -6,7 +6,7 @@
   >
     <v-data-table-virtual
       :headers="headers" :items="store.rows" :loading="store.loading"
-      height="360" density="compact"
+      height="360" density="compact" fixed-header
       no-data-text="No aliases yet — create one from a suggestion's ⋮ menu."
     >
       <template #item.mapping="{ item }">

frontend/src/components/settings/AllowlistTable.vue

@@ -7,7 +7,7 @@
   >
     <v-data-table-virtual
       :headers="headers" :items="store.rows" :loading="store.loading"
-      height="360" density="compact"
+      height="360" density="compact" fixed-header
       no-data-text="No tags on the allowlist yet — accept a suggestion to add one."
     >
       <template #item.applied_count="{ item }">

frontend/src/components/settings/ImportTaskList.vue

@@ -31,7 +31,7 @@
     </CardHeading>
     <v-data-table-virtual
       :headers="headers" :items="store.tasks" :loading="store.tasksLoading"
-      height="480" density="compact" no-data-text="No tasks yet — trigger a scan above."
+      height="480" density="compact" fixed-header no-data-text="No tasks yet — trigger a scan above."
     >
       <template #item.status="{ item }">
         <v-chip :color="statusColor(item.status)" size="small" variant="tonal">

frontend/src/components/settings/MaintenancePanel.vue

@@ -28,6 +28,7 @@
         <MLThresholdSliders />
         <AllowlistTable />
         <AliasTable />
+        <TagEvalCard />
       </div>
     </section>
 
@@ -53,6 +54,7 @@ import DbMaintenanceCard from './DbMaintenanceCard.vue'
 import MLThresholdSliders from './MLThresholdSliders.vue'
 import AllowlistTable from './AllowlistTable.vue'
 import AliasTable from './AliasTable.vue'
+import TagEvalCard from './TagEvalCard.vue'
 import BackupCard from './BackupCard.vue'
 import { useSystemActivityStore } from '../../stores/systemActivity.js'
 

frontend/src/components/settings/TagEvalCard.vue

@@ -0,0 +1,211 @@
+<template>
+  <MaintenanceTile
+    icon="mdi-flask-outline"
+    title="Tagging eval (heads vs centroid)"
+    blurb="Measure whether a trained head beats the old centroid on your own tags — and whether tagging more sharpens it."
+    :open="!!run"
+  >
+    <p class="fc-muted text-body-2 mb-3">
+      Reuses the SigLIP embeddings already stored on your images (no re-embed, no
+      GPU). For each concept it trains a logistic-regression <strong>head</strong>
+      on your positives + negatives and compares it to the old single
+      <strong>centroid</strong>, with cross-validated AP/F1 and a learning curve.
+      Runs as a background task; the result is saved and reloads here.
+    </p>
+
+    <v-textarea
+      v-model="conceptsText" label="Concepts (comma-separated)"
+      rows="2" auto-grow density="compact" hide-details class="mb-3"
+      :disabled="running"
+    />
+
+    <v-btn
+      v-if="!running"
+      color="accent" variant="flat" rounded="pill"
+      prepend-icon="mdi-play" :loading="busy" @click="onStart"
+    >Run eval</v-btn>
+
+    <div v-if="running" class="mt-3">
+      <v-progress-linear indeterminate color="accent" />
+      <div class="text-body-2 mt-2 fc-muted">Running… (started {{ startedAgo }})</div>
+    </div>
+
+    <v-alert
+      v-if="run && run.status === 'error'"
+      type="error" variant="tonal" density="compact" class="mt-3"
+    >Eval failed: {{ run.error }}</v-alert>
+
+    <div v-if="report" class="mt-4">
+      <div class="fc-muted text-caption mb-2">
+        Ran {{ formatTime(report.generated_at) }} ·
+        {{ report.concepts.length }} concept(s) ·
+        neg ratio {{ report.params.neg_ratio }}, {{ report.params.cv_folds }}-fold CV
+      </div>
+
+      <div v-for="c in report.concepts" :key="c.name" class="fc-cc">
+        <div class="fc-cc__head">
+          <span class="fc-cc__name">{{ c.name }}</span>
+          <span v-if="c.skipped" class="fc-muted text-caption">— skipped: {{ c.skipped }}</span>
+          <span v-else class="fc-muted text-caption">
+            {{ c.n_pos }} pos · {{ c.n_neg }} neg<span v-if="c.n_rejected"> ({{ c.n_rejected }} rejected)</span>
+          </span>
+        </div>
+
+        <template v-if="!c.skipped">
+          <table class="fc-metrics">
+            <thead>
+              <tr><th></th><th>AP</th><th>F1</th><th>Prec</th><th>Rec</th></tr>
+            </thead>
+            <tbody>
+              <tr>
+                <td class="fc-metrics__lbl">Head</td>
+                <td class="fc-num fc-win">{{ c.head.ap }}</td>
+                <td class="fc-num">{{ c.head.f1 }}</td>
+                <td class="fc-num">{{ c.head.precision }}</td>
+                <td class="fc-num">{{ c.head.recall }}</td>
+              </tr>
+              <tr>
+                <td class="fc-metrics__lbl fc-muted">Centroid</td>
+                <td class="fc-num fc-muted">{{ c.centroid.ap }}</td>
+                <td class="fc-num fc-muted">{{ c.centroid.f1 }}</td>
+                <td class="fc-num fc-muted">{{ c.centroid.precision }}</td>
+                <td class="fc-num fc-muted">{{ c.centroid.recall }}</td>
+              </tr>
+            </tbody>
+          </table>
+          <div class="text-caption mb-2" :class="apDelta(c) >= 0 ? 'fc-up' : 'fc-down'">
+            Δ AP {{ apDelta(c) >= 0 ? '+' : '' }}{{ apDelta(c).toFixed(3) }}
+            (head − centroid)
+          </div>
+
+          <div v-if="c.curve && c.curve.length" class="fc-curve">
+            <span class="fc-muted text-caption">Learning curve (AP @ N positives):</span>
+            <span v-for="p in c.curve" :key="p.n_pos" class="fc-curve__pt">
+              {{ p.n_pos }}→<strong>{{ p.ap }}</strong>
+            </span>
+          </div>
+
+          <div v-if="c.examples" class="fc-ex">
+            <div class="fc-ex__row">
+              <div class="fc-muted text-caption mb-1">Head would suggest (untagged, high score)</div>
+              <div class="fc-ex__thumbs">
+                <img
+                  v-for="it in c.examples.head_would_suggest" :key="`s${it.id}`"
+                  :src="it.thumbnail_url" class="fc-ex__thumb" loading="lazy"
+                />
+              </div>
+            </div>
+            <div class="fc-ex__row">
+              <div class="fc-muted text-caption mb-1">Head doubts these (your positives, low score)</div>
+              <div class="fc-ex__thumbs">
+                <img
+                  v-for="it in c.examples.head_doubts_positive" :key="`d${it.id}`"
+                  :src="it.thumbnail_url" class="fc-ex__thumb" loading="lazy"
+                />
+              </div>
+            </div>
+          </div>
+        </template>
+      </div>
+    </div>
+  </MaintenanceTile>
+</template>
+
+<script setup>
+import { toast } from '../../utils/toast.js'
+import { computed, onMounted, onUnmounted, ref } from 'vue'
+
+import MaintenanceTile from '../common/MaintenanceTile.vue'
+import { useTagEvalStore } from '../../stores/tagEval.js'
+
+const DEFAULT_CONCEPTS =
+  'glasses, cat, dog, horse, goblin, cum, lactation, fellatio, xray, stomach bulge'
+
+const store = useTagEvalStore()
+const run = ref(null)
+const conceptsText = ref(DEFAULT_CONCEPTS)
+const busy = ref(false)
+let pollTimer = null
+
+const running = computed(() => run.value?.status === 'running')
+const report = computed(() => (run.value?.status === 'ready' ? run.value.report : null))
+const startedAgo = computed(() =>
+  run.value?.started_at ? formatTime(run.value.started_at) : '')
+
+// Rehydrate the persisted run on mount so the report survives navigation — the
+// task runs backend-side regardless; we just reconnect to its row.
+onMounted(async () => {
+  try {
+    const latest = await store.latest()
+    if (latest) {
+      run.value = await store.getRun(latest.id)
+      if (run.value.status === 'running') startPoll(latest.id)
+    }
+  } catch { /* non-fatal — card still works for a fresh run */ }
+})
+onUnmounted(stopPoll)
+
+function startPoll(id) {
+  stopPoll()
+  pollTimer = setInterval(async () => {
+    try {
+      run.value = await store.getRun(id)
+      if (run.value.status !== 'running') stopPoll()
+    } catch (e) {
+      stopPoll()
+      toast({ text: `Eval poll failed: ${e.message}`, type: 'error' })
+    }
+  }, 5000)
+}
+function stopPoll() {
+  if (pollTimer) { clearInterval(pollTimer); pollTimer = null }
+}
+
+async function onStart() {
+  busy.value = true
+  try {
+    const concepts = conceptsText.value.split(',').map(s => s.trim()).filter(Boolean)
+    const res = await store.start({ concepts })
+    run.value = await store.getRun(res.run_id)
+    startPoll(res.run_id)
+  } catch (e) {
+    const msg = e.body?.running_id
+      ? 'An eval is already running.'
+      : e.message
+    toast({ text: `Could not start eval: ${msg}`, type: 'error' })
+  } finally {
+    busy.value = false
+  }
+}
+
+function apDelta(c) { return (c.head?.ap ?? 0) - (c.centroid?.ap ?? 0) }
+function formatTime(iso) {
+  if (!iso) return ''
+  try { return new Date(iso).toLocaleString() } catch { return iso }
+}
+</script>
+
+<style scoped>
+.fc-muted { color: rgb(var(--v-theme-on-surface-variant)); }
+.fc-cc {
+  padding: 12px 0;
+  border-top: 1px solid rgb(var(--v-theme-surface-light));
+}
+.fc-cc__head { display: flex; align-items: baseline; gap: 8px; margin-bottom: 6px; }
+.fc-cc__name { font-weight: 600; }
+.fc-metrics { width: 100%; max-width: 360px; border-collapse: collapse; font-size: 13px; }
+.fc-metrics th { text-align: right; font-weight: 600; color: rgb(var(--v-theme-on-surface-variant)); padding: 0 8px; }
+.fc-metrics__lbl { text-align: left; }
+.fc-num { text-align: right; font-variant-numeric: tabular-nums; padding: 1px 8px; }
+.fc-win { color: rgb(var(--v-theme-accent)); font-weight: 600; }
+.fc-up { color: rgb(var(--v-theme-success)); }
+.fc-down { color: rgb(var(--v-theme-error)); }
+.fc-curve { margin-bottom: 8px; }
+.fc-curve__pt { margin-left: 10px; font-size: 13px; font-variant-numeric: tabular-nums; }
+.fc-ex__row { margin-top: 6px; }
+.fc-ex__thumbs { display: flex; flex-wrap: wrap; gap: 4px; }
+.fc-ex__thumb {
+  width: 56px; height: 56px; object-fit: cover; border-radius: 4px;
+  background: rgb(var(--v-theme-surface-light));
+}
+</style>

frontend/src/stores/tagEval.js

@@ -0,0 +1,27 @@
+import { defineStore } from 'pinia'
+
+import { useApi } from '../composables/useApi.js'
+
+// Tag-eval (#1130): trigger + revisit the head-vs-centroid learning-curve eval.
+// The run + full report live server-side (tag_eval_run), so the card rehydrates
+// from getRun() on mount — the report survives navigation.
+export const useTagEvalStore = defineStore('tagEval', () => {
+  const api = useApi()
+
+  async function start(params) {
+    return await api.post('/api/tag-eval', { body: { params } })
+  }
+
+  async function getRun(id) {
+    return await api.get(`/api/tag-eval/${id}`)   // includes the full report
+  }
+
+  // The most recent run (light row, no report) — the card calls getRun() with
+  // its id to pull the persisted report on mount.
+  async function latest() {
+    const body = await api.get('/api/tag-eval', { params: { limit: 1 } })
+    return (body.runs && body.runs[0]) || null
+  }
+
+  return { start, getRun, latest }
+})

requirements-ml.txt

@@ -19,3 +19,10 @@ transformers>=5.8,<6.0
 onnxruntime>=1.26,<2.0
 huggingface-hub>=1.14,<2.0
 opencv-python-headless>=4.13,<5.0
+
+# scikit-learn powers the tag-eval (#1130) head-vs-centroid comparison: logistic
+# regression + cross-validated precision/recall/AP. Battle-tested metrics matter
+# because that eval's whole purpose is producing trustworthy numbers. numpy is
+# left to resolve transitively (torch/transformers/sklearn all pull it) to avoid
+# pinning against their constraints.
+scikit-learn>=1.7,<2.0

tests/test_api_tag_eval.py

@@ -0,0 +1,77 @@
+import pytest
+
+from backend.app.models import TagEvalRun
+from backend.app.services.ml.tag_eval import (
+    DEFAULT_CONCEPTS,
+    _normalize_params,
+)
+
+pytestmark = pytest.mark.integration
+
+
+def test_normalize_params_defaults_and_overrides():
+    d = _normalize_params(None)
+    assert d["concepts"] == DEFAULT_CONCEPTS
+    assert d["neg_ratio"] >= 1 and d["cv_folds"] >= 2
+    over = _normalize_params(
+        {"concepts": ["glasses", " ", "cat"], "neg_ratio": "4",
+         "cv_folds": "1", "curve_points": [30, 10, 10]}
+    )
+    assert over["concepts"] == ["glasses", "cat"]   # blanks dropped
+    assert over["neg_ratio"] == 4
+    assert over["cv_folds"] == 2                      # clamped to >=2
+    assert over["curve_points"] == [10, 30]           # deduped + sorted
+
+
+@pytest.mark.asyncio
+async def test_history_and_detail_rehydrate(client, db):
+    # A finished run with a report — the persisted row IS the survives-navigation
+    # source: history is light (no report), detail carries it.
+    run = TagEvalRun(
+        params={"concepts": ["glasses"]},
+        status="ready",
+        report={"concepts": [{"name": "glasses", "head": {"ap": 0.9}}]},
+    )
+    db.add(run)
+    await db.flush()
+    await db.commit()
+    rid = run.id
+
+    h = await client.get("/api/tag-eval?limit=10")
+    assert h.status_code == 200
+    hbody = await h.get_json()
+    row = next(r for r in hbody["runs"] if r["id"] == rid)
+    assert row["status"] == "ready"
+    assert "report" not in row                        # list stays light
+
+    d = await client.get(f"/api/tag-eval/{rid}")
+    assert d.status_code == 200
+    dbody = await d.get_json()
+    assert dbody["report"]["concepts"][0]["name"] == "glasses"
+
+
+@pytest.mark.asyncio
+async def test_create_enqueues_running(client, db, monkeypatch):
+    monkeypatch.setattr(
+        "backend.app.tasks.ml.tag_eval_run.delay", lambda *a, **k: None
+    )
+    resp = await client.post("/api/tag-eval", json={"params": {"concepts": ["cat"]}})
+    assert resp.status_code == 202
+    body = await resp.get_json()
+    assert body["status"] == "running"
+    got = await db.get(TagEvalRun, body["run_id"])
+    assert got is not None and got.status == "running"
+
+
+@pytest.mark.asyncio
+async def test_create_conflicts_when_one_running(client, db, monkeypatch):
+    monkeypatch.setattr(
+        "backend.app.tasks.ml.tag_eval_run.delay", lambda *a, **k: None
+    )
+    db.add(TagEvalRun(params={}, status="running"))
+    await db.flush()
+    await db.commit()
+    resp = await client.post("/api/tag-eval", json={"params": {}})
+    assert resp.status_code == 409
+    body = await resp.get_json()
+    assert body["error"] == "eval_already_running"