From 6e3c5f697f9978b9d23259d71efa20ee426962b3 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 27 Jun 2026 22:49:10 -0400 Subject: [PATCH] =?UTF-8?q?feat(ml):=20tag-eval=20backend=20=E2=80=94=20he?= =?UTF-8?q?ad-vs-centroid=20learning-curve=20eval=20(persisted)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Slice 1 of milestone #114 (tagging v2). Proves the frozen-embedding + trained- head spine on the operator's own data, reusing the SigLIP embeddings already stored on image_record — no re-embedding, no GPU. Per concept: train a logistic-regression HEAD (positives + negatives = explicit rejections + sampled unlabeled) vs the old single-CENTROID baseline; report cross-validated precision/recall/AP for both, a LEARNING CURVE (AP/F1 as tagged positives grow 10→30→100→300), and example image ids (head-would-suggest / head-doubts-positive) to eyeball. Persisted so the report SURVIVES navigation (operator-flagged): the run + full report live in a new tag_eval_run row (mirrors library_audit_run); the admin card will rehydrate from GET on mount, not transient state. - models.TagEvalRun + migration 0056; runs on the ml queue (only worker with numpy/sklearn) — numpy/sklearn lazy-imported so the API can still enqueue. - services/ml/tag_eval (compute + start helper, one-running guard), tasks.ml .tag_eval_run, api/tag-eval (POST create, GET history light / detail w/ report). - recover_stalled_tag_eval_runs sweep + retention (keep last 20) + 5-min beat (rule 89). scikit-learn added to requirements-ml. - tests: param normalization + the rehydrate read-path + create/conflict. Frontend admin card (trigger + render persisted report) follows next. Co-Authored-By: Claude Opus 4.8 (1M context) --- alembic/versions/0056_tag_eval_run.py | 43 ++++ backend/app/api/__init__.py | 2 + backend/app/api/tag_eval.py | 70 ++++++ backend/app/celery_app.py | 4 + backend/app/models/__init__.py | 2 + backend/app/models/tag_eval_run.py | 45 ++++ backend/app/services/ml/tag_eval.py | 316 ++++++++++++++++++++++++++ backend/app/tasks/maintenance.py | 44 ++++ backend/app/tasks/ml.py | 45 ++++ requirements-ml.txt | 7 + tests/test_api_tag_eval.py | 77 +++++++ 11 files changed, 655 insertions(+) create mode 100644 alembic/versions/0056_tag_eval_run.py create mode 100644 backend/app/api/tag_eval.py create mode 100644 backend/app/models/tag_eval_run.py create mode 100644 backend/app/services/ml/tag_eval.py create mode 100644 tests/test_api_tag_eval.py diff --git a/alembic/versions/0056_tag_eval_run.py b/alembic/versions/0056_tag_eval_run.py new file mode 100644 index 0000000..7d8e91f --- /dev/null +++ b/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") diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py index f97b72a..e0794e9 100644 --- a/backend/app/api/__init__.py +++ b/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, diff --git a/backend/app/api/tag_eval.py b/backend/app/api/tag_eval.py new file mode 100644 index 0000000..31fe10b --- /dev/null +++ b/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("/", 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)) diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index eae20e6..aba2a18 100644 --- a/backend/app/celery_app.py +++ b/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, diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 6061f31..1d8a751 100644 --- a/backend/app/models/__init__.py +++ b/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", diff --git a/backend/app/models/tag_eval_run.py b/backend/app/models/tag_eval_run.py new file mode 100644 index 0000000..d0775ed --- /dev/null +++ b/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, + ) diff --git a/backend/app/services/ml/tag_eval.py b/backend/app/services/ml/tag_eval.py new file mode 100644 index 0000000..132d3f1 --- /dev/null +++ b/backend/app/services/ml/tag_eval.py @@ -0,0 +1,316 @@ +"""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(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(Xn, y, ids, np) -> dict[str, list[int]]: + """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).""" + 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 = neg_idx[np.argsort(s[neg_idx])[::-1][:_EXAMPLES_K]] + low_pos = pos_idx[np.argsort(s[pos_idx])[:_EXAMPLES_K]] + return { + "head_would_suggest": [int(ids[i]) for i in top_neg], + "head_doubts_positive": [int(ids[i]) for i in low_pos], + } diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index b24fe3a..1c8a0b8 100644 --- a/backend/app/tasks/maintenance.py +++ b/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 diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index d45649f..9c8c477 100644 --- a/backend/app/tasks/ml.py +++ b/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" diff --git a/requirements-ml.txt b/requirements-ml.txt index af14624..c216487 100644 --- a/requirements-ml.txt +++ b/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 diff --git a/tests/test_api_tag_eval.py b/tests/test_api_tag_eval.py new file mode 100644 index 0000000..2addd1d --- /dev/null +++ b/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"