diff --git a/alembic/versions/0059_head_auto_apply.py b/alembic/versions/0059_head_auto_apply.py new file mode 100644 index 0000000..d0bb9b8 --- /dev/null +++ b/alembic/versions/0059_head_auto_apply.py @@ -0,0 +1,70 @@ +"""head_auto_apply_run + earned-auto-apply settings (#114) + +A graduated head can apply its tag without a human, gated by a master switch + +a support floor. head_auto_apply_run tracks each sweep / dry-run preview. + +Revision ID: 0059 +Revises: 0058 +Create Date: 2026-06-29 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects.postgresql import JSONB + +revision: str = "0059" +down_revision: Union[str, None] = "0058" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "head_auto_apply_run", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column( + "dry_run", sa.Boolean(), nullable=False, server_default=sa.false() + ), + 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("n_applied", sa.Integer(), 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_head_auto_apply_run_status", "head_auto_apply_run", ["status"], + ) + + op.add_column( + "ml_settings", + sa.Column( + "head_auto_apply_enabled", sa.Boolean(), nullable=False, + server_default=sa.true(), # opt-out: on by default (operator-asked) + ), + ) + op.add_column( + "ml_settings", + sa.Column( + "head_auto_apply_min_positives", sa.Integer(), nullable=False, + server_default="30", + ), + ) + + +def downgrade() -> None: + op.drop_column("ml_settings", "head_auto_apply_min_positives") + op.drop_column("ml_settings", "head_auto_apply_enabled") + op.drop_index( + "ix_head_auto_apply_run_status", table_name="head_auto_apply_run" + ) + op.drop_table("head_auto_apply_run") diff --git a/alembic/versions/0060_head_metrics.py b/alembic/versions/0060_head_metrics.py new file mode 100644 index 0000000..e94edb8 --- /dev/null +++ b/alembic/versions/0060_head_metrics.py @@ -0,0 +1,74 @@ +"""head_metric + head_metrics_snapshot: auto-apply observability (#114) + +Running misfire/under-fire counters per concept (captured at correction time, +since image_tag.source is lost on delete) + a daily per-concept time-series so +the operator can tune the precision target + support floor from real data. + +Revision ID: 0060 +Revises: 0059 +Create Date: 2026-06-29 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0060" +down_revision: Union[str, None] = "0059" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "head_metric", + sa.Column( + "tag_id", sa.Integer(), + sa.ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True, + ), + sa.Column("n_misfires", sa.Integer(), nullable=False, server_default="0"), + sa.Column("n_underfires", sa.Integer(), nullable=False, server_default="0"), + sa.Column( + "updated_at", sa.DateTime(timezone=True), nullable=False, + server_default=sa.func.now(), + ), + ) + + op.create_table( + "head_metrics_snapshot", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column( + "tag_id", sa.Integer(), + sa.ForeignKey("tag.id", ondelete="CASCADE"), + ), + sa.Column("name", sa.String(length=255), nullable=False), + sa.Column( + "snapshot_at", sa.DateTime(timezone=True), nullable=False, + server_default=sa.func.now(), + ), + sa.Column("n_auto_applied", sa.Integer(), nullable=False, server_default="0"), + sa.Column("n_misfires", sa.Integer(), nullable=False, server_default="0"), + sa.Column("n_underfires", sa.Integer(), nullable=False, server_default="0"), + sa.Column("ap", sa.Float(), nullable=True), + sa.Column("precision_cv", sa.Float(), nullable=True), + sa.Column("recall", sa.Float(), nullable=True), + sa.Column("n_pos", sa.Integer(), nullable=True), + ) + op.create_index( + "ix_head_metrics_snapshot_tag_id", "head_metrics_snapshot", ["tag_id"], + ) + op.create_index( + "ix_head_metrics_snapshot_snapshot_at", "head_metrics_snapshot", + ["snapshot_at"], + ) + + +def downgrade() -> None: + op.drop_index( + "ix_head_metrics_snapshot_snapshot_at", table_name="head_metrics_snapshot" + ) + op.drop_index( + "ix_head_metrics_snapshot_tag_id", table_name="head_metrics_snapshot" + ) + op.drop_table("head_metrics_snapshot") + op.drop_table("head_metric") diff --git a/backend/app/api/heads.py b/backend/app/api/heads.py index 018ee2c..e6f10eb 100644 --- a/backend/app/api/heads.py +++ b/backend/app/api/heads.py @@ -12,8 +12,22 @@ from quart import Blueprint, jsonify, request from sqlalchemy import desc, func, select from ..extensions import get_session -from ..models import HeadTrainingRun, Tag, TagHead -from ..services.ml.heads import HeadTrainingAlreadyRunning, start_head_training_run +from ..models import ( + HeadAutoApplyRun, + HeadMetric, + HeadMetricsSnapshot, + HeadTrainingRun, + Tag, + TagHead, +) +from ..models.tag import image_tag +from ..services.ml.heads import ( + HeadAutoApplyAlreadyRunning, + HeadAutoApplyDisabled, + HeadTrainingAlreadyRunning, + start_head_auto_apply_run, + start_head_training_run, +) heads_bp = Blueprint("heads", __name__, url_prefix="/api/heads") @@ -116,3 +130,156 @@ async def status(): "runs": [_serialize_run(r) for r in runs], "heads": heads, }) + + +def _serialize_apply_run(run: HeadAutoApplyRun) -> dict: + return { + "id": run.id, + "dry_run": run.dry_run, + "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, + "n_applied": run.n_applied, + "report": run.report, + "error": run.error, + } + + +@heads_bp.route("/auto-apply", methods=["POST"]) +async def auto_apply(): + """Trigger an earned-auto-apply sweep. {dry_run:true} previews (writes + nothing); a real sweep needs head_auto_apply_enabled on.""" + body = await request.get_json(silent=True) or {} + params = {"dry_run": bool(body.get("dry_run", False))} + async with get_session() as session: + try: + run_id = await session.run_sync( + lambda s: start_head_auto_apply_run(s, params) + ) + except HeadAutoApplyAlreadyRunning as running: + return jsonify({ + "error": "auto_apply_already_running", + "running_id": int(running.args[0]), + }), 409 + except HeadAutoApplyDisabled: + return jsonify({"error": "auto_apply_disabled"}), 400 + await session.commit() + return jsonify({"run_id": run_id, "status": "running"}), 202 + + +@heads_bp.route("/auto-apply", methods=["GET"]) +async def auto_apply_status(): + async with get_session() as session: + running = ( + await session.execute( + select(HeadAutoApplyRun.id) + .where(HeadAutoApplyRun.status == "running") + .order_by(HeadAutoApplyRun.id.desc()) + .limit(1) + ) + ).scalar_one_or_none() + runs = ( + await session.execute( + select(HeadAutoApplyRun) + .order_by(HeadAutoApplyRun.id.desc()) + .limit(10) + ) + ).scalars().all() + return jsonify({ + "running_id": running, + "runs": [_serialize_apply_run(r) for r in runs], + }) + + +@heads_bp.route("/metrics", methods=["GET"]) +async def metrics(): + """Auto-apply observability: per-concept current counts (volume, misfires, + under-fires, realized misfire rate, head quality) + the daily time-series so + the operator can tune the precision target + support floor from real data.""" + async with get_session() as session: + head_rows = ( + await session.execute( + select( + TagHead.tag_id, Tag.name, TagHead.ap, TagHead.precision_cv, + TagHead.recall, TagHead.auto_apply_threshold, TagHead.n_pos, + ).join(Tag, Tag.id == TagHead.tag_id) + ) + ).all() + heads = {r.tag_id: r for r in head_rows} + metric_rows = ( + await session.execute( + select( + HeadMetric.tag_id, HeadMetric.n_misfires, HeadMetric.n_underfires + ) + ) + ).all() + mets = {r.tag_id: r for r in metric_rows} + applied = dict( + ( + await session.execute( + select(image_tag.c.tag_id, func.count()) + .where(image_tag.c.source == "head_auto") + .group_by(image_tag.c.tag_id) + ) + ).all() + ) + names = {r.tag_id: r.name for r in head_rows} + # Names for metric-only tags (head pruned but corrections recorded). + missing = [t for t in mets if t not in names] + if missing: + for tid, nm in ( + await session.execute( + select(Tag.id, Tag.name).where(Tag.id.in_(missing)) + ) + ).all(): + names[tid] = nm + + concepts = [] + for tid in set(heads) | set(mets): + h = heads.get(tid) + m = mets.get(tid) + n_applied = applied.get(tid, 0) + n_mis = m.n_misfires if m else 0 + denom = n_applied + n_mis + concepts.append({ + "tag_id": tid, + "name": names.get(tid, str(tid)), + "n_auto_applied": n_applied, + "n_misfires": n_mis, + "n_underfires": m.n_underfires if m else 0, + # Of everything this head ever auto-applied, the fraction you + # removed — the misfire rate (null until something fired). + "misfire_rate": round(n_mis / denom, 4) if denom else None, + "ap": h.ap if h else None, + "precision_cv": h.precision_cv if h else None, + "recall": h.recall if h else None, + "auto_apply": bool(h and h.auto_apply_threshold is not None), + "n_pos": h.n_pos if h else None, + }) + concepts.sort(key=lambda c: (c["n_misfires"], c["n_auto_applied"]), reverse=True) + + snaps = ( + await session.execute( + select(HeadMetricsSnapshot) + .order_by(HeadMetricsSnapshot.snapshot_at.desc()) + .limit(1000) + ) + ).scalars().all() + return jsonify({ + "concepts": concepts, + "snapshots": [ + { + "tag_id": s.tag_id, + "name": s.name, + "snapshot_at": s.snapshot_at.isoformat() if s.snapshot_at else None, + "n_auto_applied": s.n_auto_applied, + "n_misfires": s.n_misfires, + "n_underfires": s.n_underfires, + "ap": s.ap, + "precision_cv": s.precision_cv, + "recall": s.recall, + "n_pos": s.n_pos, + } + for s in snaps + ], + }) diff --git a/backend/app/api/ml_admin.py b/backend/app/api/ml_admin.py index a3d3b74..1b0e1d1 100644 --- a/backend/app/api/ml_admin.py +++ b/backend/app/api/ml_admin.py @@ -19,6 +19,8 @@ _EDITABLE = ( "video_min_tag_frames", "head_min_positives", "head_auto_apply_precision", + "head_auto_apply_enabled", + "head_auto_apply_min_positives", ) @@ -44,6 +46,8 @@ async def get_settings(): "embedder_model_version": s.embedder_model_version, "head_min_positives": s.head_min_positives, "head_auto_apply_precision": s.head_auto_apply_precision, + "head_auto_apply_enabled": s.head_auto_apply_enabled, + "head_auto_apply_min_positives": s.head_auto_apply_min_positives, } ) @@ -109,6 +113,8 @@ def _validate(p: dict) -> str | None: return "head_min_positives must be >= 1" if not (0.5 <= float(p["head_auto_apply_precision"]) <= 0.999): return "head_auto_apply_precision must be between 0.5 and 0.999" + if int(p["head_auto_apply_min_positives"]) < 1: + return "head_auto_apply_min_positives must be >= 1" return None diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index be1b368..9617d6c 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -109,6 +109,18 @@ def make_celery() -> Celery: "task": "backend.app.tasks.ml.apply_allowlist_tags", "schedule": 86400.0, }, + "train-heads-nightly": { + "task": "backend.app.tasks.ml.scheduled_train_heads", + "schedule": 86400.0, # passive cadence; manual retrain stays available + }, + "apply-head-tags-daily": { + "task": "backend.app.tasks.ml.scheduled_apply_head_tags", + "schedule": 86400.0, # no-op unless head_auto_apply_enabled + }, + "snapshot-head-metrics-daily": { + "task": "backend.app.tasks.maintenance.snapshot_head_metrics", + "schedule": 86400.0, + }, "integrity-verify-weekly": { "task": "backend.app.tasks.maintenance.verify_integrity", "schedule": 604800.0, # weekly @@ -164,6 +176,10 @@ def make_celery() -> Celery: "task": "backend.app.tasks.maintenance.recover_stalled_head_training_runs", "schedule": 300.0, }, + "recover-stalled-head-auto-apply-runs": { + "task": "backend.app.tasks.maintenance.recover_stalled_head_auto_apply_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 04025ed..681adb7 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -8,6 +8,9 @@ from .base import Base from .credential import Credential from .download_event import DownloadEvent from .external_link import ExternalLink +from .head_auto_apply_run import HeadAutoApplyRun +from .head_metric import HeadMetric +from .head_metrics_snapshot import HeadMetricsSnapshot from .head_training_run import HeadTrainingRun from .image_prediction import ImagePrediction from .image_provenance import ImageProvenance @@ -67,6 +70,9 @@ __all__ = [ "ImportSettings", "LibraryAuditRun", "MLSettings", + "HeadAutoApplyRun", + "HeadMetric", + "HeadMetricsSnapshot", "HeadTrainingRun", "TagAlias", "TagAllowlist", diff --git a/backend/app/models/head_auto_apply_run.py b/backend/app/models/head_auto_apply_run.py new file mode 100644 index 0000000..08c359a --- /dev/null +++ b/backend/app/models/head_auto_apply_run.py @@ -0,0 +1,46 @@ +"""HeadAutoApplyRun — persisted lifecycle of an earned-auto-apply sweep (#114). + +A graduated head can apply its tag to images it scores above the head's +auto-apply threshold, without a human. This row tracks one such sweep (or a +dry-run PREVIEW of it) so the result survives navigation and the admin card can +show what fired / what would fire. Mirrors HeadTrainingRun. State machine: +running → ready / error. The `report` JSONB holds per-concept counts +(applied / projected / scanned). +""" + +from datetime import datetime +from typing import Any + +from sqlalchemy import Boolean, DateTime, Integer, String, Text, func +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from .base import Base + + +class HeadAutoApplyRun(Base): + __tablename__ = "head_auto_apply_run" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + # dry_run=True is a PREVIEW: scores + counts what WOULD apply, writes nothing + # (preview/apply parity, rule 93). + dry_run: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + 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 + ) + # Total tags applied across all heads this sweep (0 for a clean dry-run). + n_applied: Mapped[int | None] = mapped_column(Integer, nullable=True) + # Per-concept breakdown: [{tag_id, name, applied, scanned, threshold}, ...]. + report: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) + error: Mapped[str | None] = mapped_column(Text, nullable=True) + last_progress_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) diff --git a/backend/app/models/head_metric.py b/backend/app/models/head_metric.py new file mode 100644 index 0000000..a034e51 --- /dev/null +++ b/backend/app/models/head_metric.py @@ -0,0 +1,32 @@ +"""HeadMetric — running correction counters per concept (#114 observability). + +Earned auto-apply fires graduated heads; to TUNE it we need to know how often a +head's auto-applied tag was wrong (the operator removed it = a MISFIRE) and how +often the operator had to add a tag a head exists for by hand (an UNDER-FIRE, +the head missed it). image_tag.source is lost when a row is deleted, so these +are captured as durable cumulative counters at correction time — they survive +head retrain/prune (keyed by tag, not by the head row). The daily snapshot reads +them into the time-series. +""" + +from datetime import datetime + +from sqlalchemy import DateTime, ForeignKey, Integer, func +from sqlalchemy.orm import Mapped, mapped_column + +from .base import Base + + +class HeadMetric(Base): + __tablename__ = "head_metric" + + tag_id: Mapped[int] = mapped_column( + ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True + ) + # An auto-applied (source='head_auto') tag the operator later REMOVED. + n_misfires: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + # A tag with a head that the operator added by HAND (the head missed it). + n_underfires: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) diff --git a/backend/app/models/head_metrics_snapshot.py b/backend/app/models/head_metrics_snapshot.py new file mode 100644 index 0000000..a9ec7ac --- /dev/null +++ b/backend/app/models/head_metrics_snapshot.py @@ -0,0 +1,38 @@ +"""HeadMetricsSnapshot — a daily per-concept time-series point (#114). + +The "amount of change over time" reporting the operator asked for: once a day, +record each concept's auto-applied VOLUME (current head_auto tags), cumulative +misfires/under-fires, and the head's measured quality. Plotting these rows over +time shows whether auto-apply is landing better/worse and whether tagging more is +sharpening a concept — the signal for tuning the precision target + support floor. +""" + +from datetime import datetime + +from sqlalchemy import DateTime, Float, ForeignKey, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from .base import Base + + +class HeadMetricsSnapshot(Base): + __tablename__ = "head_metrics_snapshot" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + tag_id: Mapped[int] = mapped_column( + ForeignKey("tag.id", ondelete="CASCADE"), index=True + ) + # Denormalized so a snapshot stays readable even if the tag is later renamed. + name: Mapped[str] = mapped_column(String(255), nullable=False) + snapshot_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), index=True + ) + # Current count of source='head_auto' applications still standing. + n_auto_applied: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + n_misfires: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + n_underfires: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + # The head's measured quality at snapshot time (null if no head exists). + ap: Mapped[float | None] = mapped_column(Float, nullable=True) + precision_cv: Mapped[float | None] = mapped_column(Float, nullable=True) + recall: Mapped[float | None] = mapped_column(Float, nullable=True) + n_pos: Mapped[int | None] = mapped_column(Integer, nullable=True) diff --git a/backend/app/models/ml_settings.py b/backend/app/models/ml_settings.py index 243387a..696787b 100644 --- a/backend/app/models/ml_settings.py +++ b/backend/app/models/ml_settings.py @@ -2,7 +2,15 @@ from datetime import datetime -from sqlalchemy import CheckConstraint, DateTime, Float, Integer, String, func +from sqlalchemy import ( + Boolean, + CheckConstraint, + DateTime, + Float, + Integer, + String, + func, +) from sqlalchemy.orm import Mapped, mapped_column from .base import Base @@ -66,6 +74,18 @@ class MLSettings(Base): head_auto_apply_precision: Mapped[float] = mapped_column( Float, nullable=False, default=0.97 ) + # Earned auto-apply (#114). A graduated head fires (tags images without a + # human) when this master switch is on AND the head has at least + # head_auto_apply_min_positives clean labels — so a precise-looking but + # under-supported low-N head can't spray tags across the library. ON by + # default (operator-asked 2026-06-29: opt-OUT, not opt-in); the support + + # measured-precision gates keep it safe, and every auto-tag is reversible. + head_auto_apply_enabled: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=True + ) + head_auto_apply_min_positives: Mapped[int] = mapped_column( + Integer, nullable=False, default=30 + ) tagger_model_version: Mapped[str] = mapped_column( String(128), nullable=False, default="camie-tagger-v2" ) diff --git a/backend/app/services/ml/heads.py b/backend/app/services/ml/heads.py index aed4e69..4b07e68 100644 --- a/backend/app/services/ml/heads.py +++ b/backend/app/services/ml/heads.py @@ -26,12 +26,14 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session from ...models import ( + HeadAutoApplyRun, HeadTrainingRun, ImageRecord, MLSettings, Tag, TagHead, TagKind, + TagSuggestionRejection, ) from ...models.tag import image_tag from .tag_eval import ( @@ -328,3 +330,138 @@ async def _settings_async(session: AsyncSession) -> MLSettings: return ( await session.execute(select(MLSettings).where(MLSettings.id == 1)) ).scalar_one() + + +# --- Earned auto-apply (sync, ml worker) --------------------------------- +# A graduated head can apply its tag to images it scores above the head's +# auto_apply_threshold, without a human. Gated by a master switch + a support +# floor so a precise-looking but under-supported head can't spray tags. + +_AUTO_APPLY_CHUNK = 5000 + + +class HeadAutoApplyAlreadyRunning(Exception): + """Raised when an auto-apply sweep is already in flight.""" + + +class HeadAutoApplyDisabled(Exception): + """Raised when a real (non-dry-run) sweep is requested but the master + switch (head_auto_apply_enabled) is off.""" + + +def start_head_auto_apply_run(session: Session, params: dict[str, Any]) -> int: + """Create a HeadAutoApplyRun + dispatch the ml-queue sweep. dry_run previews + (writes nothing); a real sweep needs the master switch on. One run at a time.""" + dry_run = bool((params or {}).get("dry_run", False)) + existing = session.execute( + select(HeadAutoApplyRun.id).where(HeadAutoApplyRun.status == "running") + ).scalar_one_or_none() + if existing is not None: + raise HeadAutoApplyAlreadyRunning(existing) + if not dry_run and not _settings(session).head_auto_apply_enabled: + raise HeadAutoApplyDisabled() + run = HeadAutoApplyRun( + dry_run=dry_run, params={"dry_run": dry_run}, status="running", + last_progress_at=datetime.now(UTC), + ) + session.add(run) + session.flush() + run_id = run.id + from ...tasks.ml import apply_head_tags as _task + _task.delay(run_id) + return run_id + + +def _auto_apply_heads(session: Session, embedding_version: str, min_pos: int): + """Eligible heads to fire: graduated (auto_apply_threshold set), enough + support, current embedding. Returns the row list (tag_id/name/weights/...).""" + return session.execute( + select( + TagHead.tag_id, Tag.name, TagHead.weights, TagHead.bias, + TagHead.auto_apply_threshold, + ) + .join(Tag, Tag.id == TagHead.tag_id) + .where(TagHead.embedding_version == embedding_version) + .where(TagHead.auto_apply_threshold.is_not(None)) + .where(TagHead.n_pos >= min_pos) + ).all() + + +def auto_apply_sweep( + session: Session, run: HeadAutoApplyRun, dry_run: bool +) -> dict[str, Any]: + """Score every embedded image against the eligible heads and apply (or, for + dry_run, just count) each head's tag where score >= its auto_apply_threshold + and the tag isn't already applied or rejected on that image. Streams + embeddings in chunks; commits per chunk on a real run. Returns + {n_applied, concepts:[{tag_id,name,applied,scanned,threshold}]}.""" + import numpy as np + from sqlalchemy.dialects.postgresql import insert as pg_insert + + settings = _settings(session) + rows = _auto_apply_heads( + session, settings.embedder_model_version, + settings.head_auto_apply_min_positives, + ) + if not rows: + return {"n_applied": 0, "concepts": []} + + W = np.vstack([np.asarray(r.weights, dtype=np.float32) for r in rows]) + b = np.asarray([r.bias for r in rows], dtype=np.float32) + thr = np.asarray([r.auto_apply_threshold for r in rows], dtype=np.float32) + tag_ids = [r.tag_id for r in rows] + names = [r.name for r in rows] + + # Skip images that already carry, or have rejected, each tag. + skip = {tid: set() for tid in tag_ids} + for tid in tag_ids: + for (iid,) in session.execute( + select(image_tag.c.image_record_id).where(image_tag.c.tag_id == tid) + ): + skip[tid].add(iid) + for (iid,) in session.execute( + select(TagSuggestionRejection.image_record_id).where( + TagSuggestionRejection.tag_id == tid + ) + ): + skip[tid].add(iid) + + applied = [0] * len(rows) + scanned = 0 + all_ids = list(session.execute( + select(ImageRecord.id).where(ImageRecord.siglip_embedding.is_not(None)) + ).scalars()) + for start in range(0, len(all_ids), _AUTO_APPLY_CHUNK): + chunk = all_ids[start:start + _AUTO_APPLY_CHUNK] + emb = _load_embeddings(session, chunk) + cids = [i for i in chunk if i in emb] + if not cids: + continue + Xn = _l2norm(np.vstack([emb[i] for i in cids]).astype(np.float32), np) + probs = 1.0 / (1.0 + np.exp(-(Xn @ W.T + b))) # (N, H) + scanned += len(cids) + for h in range(len(rows)): + tid = tag_ids[h] + for idx in np.where(probs[:, h] >= thr[h])[0]: + iid = cids[int(idx)] + if iid in skip[tid]: + continue + skip[tid].add(iid) + applied[h] += 1 + if not dry_run: + session.execute( + pg_insert(image_tag) + .values(image_record_id=iid, tag_id=tid, source="head_auto") + .on_conflict_do_nothing() + ) + if not dry_run: + session.commit() + run.last_progress_at = datetime.now(UTC) + session.commit() + + concepts = [ + {"tag_id": tag_ids[h], "name": names[h], "applied": applied[h], + "scanned": scanned, "threshold": float(thr[h])} + for h in range(len(rows)) + ] + return {"n_applied": sum(applied), "concepts": concepts} diff --git a/backend/app/services/tag_service.py b/backend/app/services/tag_service.py index 495875a..839071c 100644 --- a/backend/app/services/tag_service.py +++ b/backend/app/services/tag_service.py @@ -9,7 +9,7 @@ from sqlalchemy import and_, case, exists, func, select, text, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession -from ..models import Tag, TagKind, image_tag +from ..models import HeadMetric, Tag, TagHead, TagKind, image_tag from ..models.tag_allowlist import TagAllowlist from ..models.tag_reference_embedding import TagReferenceEmbedding from .db_helpers import get_or_create @@ -215,6 +215,18 @@ class TagService: async def add_to_image(self, image_id: int, tag_id: int, source: str = "manual") -> None: """Idempotent: re-adding an existing tag does nothing.""" + # A genuinely-new MANUAL add of a tag that already has a head is an + # UNDER-FIRE signal — the auto-system should have caught it (#114 obs). + is_new = source == "manual" and ( + await self.session.execute( + select(image_tag.c.tag_id).where( + and_( + image_tag.c.image_record_id == image_id, + image_tag.c.tag_id == tag_id, + ) + ) + ) + ).first() is None stmt = pg_insert(image_tag).values( image_record_id=image_id, tag_id=tag_id, source=source ) @@ -222,8 +234,22 @@ class TagService: index_elements=["image_record_id", "tag_id"] ) await self.session.execute(stmt) + if is_new: + await self._note_under_fire(tag_id) async def remove_from_image(self, image_id: int, tag_id: int) -> None: + # Removing an auto-applied (source='head_auto') tag is a MISFIRE — read + # the source BEFORE deleting, since it's lost with the row (#114 obs). + src = ( + await self.session.execute( + select(image_tag.c.source).where( + and_( + image_tag.c.image_record_id == image_id, + image_tag.c.tag_id == tag_id, + ) + ) + ) + ).scalar_one_or_none() await self.session.execute( image_tag.delete().where( and_( @@ -232,6 +258,31 @@ class TagService: ) ) ) + if src == "head_auto": + await self._bump_metric(tag_id, "n_misfires") + + async def _note_under_fire(self, tag_id: int) -> None: + """Count an under-fire only when the tag actually has a head.""" + has_head = ( + await self.session.execute( + select(TagHead.tag_id).where(TagHead.tag_id == tag_id) + ) + ).first() is not None + if has_head: + await self._bump_metric(tag_id, "n_underfires") + + async def _bump_metric(self, tag_id: int, column: str) -> None: + """Increment a HeadMetric counter (upsert), keyed by tag so it survives + head retrain/prune.""" + col = HeadMetric.__table__.c[column] + await self.session.execute( + pg_insert(HeadMetric) + .values(tag_id=tag_id, **{column: 1}) + .on_conflict_do_update( + index_elements=["tag_id"], + set_={column: col + 1, "updated_at": func.now()}, + ) + ) async def list_for_image(self, image_id: int) -> Sequence: """Tags on an image, ordered (kind, name). Each row carries the fandom's diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index fa0397b..e9e334e 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -13,6 +13,7 @@ from ..celery_app import celery from ..models import ( BackupRun, DownloadEvent, + HeadAutoApplyRun, HeadTrainingRun, ImageRecord, ImportBatch, @@ -101,6 +102,9 @@ 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 +# head auto-apply (#114) shares the 60-min soft limit; flag past 75. +HEAD_AUTO_APPLY_STALL_THRESHOLD_MINUTES = 75 +HEAD_AUTO_APPLY_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 @@ -800,6 +804,125 @@ def recover_stalled_head_training_runs() -> int: return recovered +@celery.task(name="backend.app.tasks.maintenance.recover_stalled_head_auto_apply_runs") +def recover_stalled_head_auto_apply_runs() -> int: + """Flip stalled HeadAutoApplyRun 'running' rows to 'error' + prune to the + last HEAD_AUTO_APPLY_KEEP_RUNS (retention, rule 89). 5-min maintenance lane.""" + SessionLocal = _sync_session_factory() + now = datetime.now(UTC) + cutoff = now - timedelta(minutes=HEAD_AUTO_APPLY_STALL_THRESHOLD_MINUTES) + with SessionLocal() as session: + result = session.execute( + update(HeadAutoApplyRun) + .where(HeadAutoApplyRun.status == "running") + .where( + func.coalesce( + HeadAutoApplyRun.last_progress_at, HeadAutoApplyRun.started_at + ) + < cutoff + ) + .values( + status="error", finished_at=now, + error=( + f"stranded by recovery sweep (no progress for " + f"{HEAD_AUTO_APPLY_STALL_THRESHOLD_MINUTES} min)" + ), + ) + ) + keep = session.execute( + select(HeadAutoApplyRun.id).order_by(HeadAutoApplyRun.id.desc()) + .limit(HEAD_AUTO_APPLY_KEEP_RUNS) + ).scalars().all() + if keep: + session.execute( + delete(HeadAutoApplyRun).where(HeadAutoApplyRun.id.not_in(keep)) + ) + session.commit() + recovered = result.rowcount or 0 + if recovered: + log.info( + "recover_stalled_head_auto_apply_runs: recovered %d rows", recovered + ) + return recovered + + +# Keep ~6 months of daily head-metric snapshots (enough to see tuning trends). +HEAD_METRICS_SNAPSHOT_RETENTION_DAYS = 180 + + +@celery.task(name="backend.app.tasks.maintenance.snapshot_head_metrics") +def snapshot_head_metrics() -> int: + """Daily per-concept observability point (#114): record each head-bearing + concept's auto-applied volume, cumulative misfires/under-fires, and the + head's measured quality — the time-series the operator tunes from. Prunes + points older than the retention window.""" + from ..models import ( + HeadMetric, + HeadMetricsSnapshot, + Tag, + TagHead, + ) + from ..models.tag import image_tag + + SessionLocal = _sync_session_factory() + now = datetime.now(UTC) + with SessionLocal() as session: + heads = { + r.tag_id: r for r in session.execute( + select( + TagHead.tag_id, TagHead.ap, TagHead.precision_cv, + TagHead.recall, TagHead.n_pos, + ) + ) + } + metrics = { + r.tag_id: r for r in session.execute( + select( + HeadMetric.tag_id, HeadMetric.n_misfires, HeadMetric.n_underfires + ) + ) + } + # .all() first: dict() of a bare Result tries the mapping protocol (a + # Result exposes .keys()) and subscripts it, which fails. + applied = dict( + session.execute( + select(image_tag.c.tag_id, func.count()) + .where(image_tag.c.source == "head_auto") + .group_by(image_tag.c.tag_id) + ).all() + ) + tag_ids = set(heads) | set(metrics) + if not tag_ids: + return 0 + names = dict( + session.execute( + select(Tag.id, Tag.name).where(Tag.id.in_(tag_ids)) + ).all() + ) + for tid in tag_ids: + h = heads.get(tid) + m = metrics.get(tid) + session.add(HeadMetricsSnapshot( + tag_id=tid, name=names.get(tid, str(tid)), + snapshot_at=now, + n_auto_applied=applied.get(tid, 0), + n_misfires=m.n_misfires if m else 0, + n_underfires=m.n_underfires if m else 0, + ap=h.ap if h else None, + precision_cv=h.precision_cv if h else None, + recall=h.recall if h else None, + n_pos=h.n_pos if h else None, + )) + session.execute( + delete(HeadMetricsSnapshot).where( + HeadMetricsSnapshot.snapshot_at + < now - timedelta(days=HEAD_METRICS_SNAPSHOT_RETENTION_DAYS) + ) + ) + session.commit() + return len(tag_ids) + + @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 65d8866..bc22c5b 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -629,3 +629,112 @@ def train_heads(self, run_id: int) -> str: run.finished_at = datetime.now(UTC) session.commit() return "ready" + + +@celery.task(name="backend.app.tasks.ml.scheduled_train_heads") +def scheduled_train_heads() -> str: + """Nightly passive retrain (#114): fold the day's accepts/rejects + any + newly-eligible concepts into the heads without the operator clicking. Skips + if a run is already in flight (one at a time). Creates + COMMITS the run row + before dispatching so the ml-queue worker can always find it.""" + from datetime import UTC, datetime + + from sqlalchemy import select as sa_select + + from ..models import HeadTrainingRun + + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + running = session.execute( + sa_select(HeadTrainingRun.id).where(HeadTrainingRun.status == "running") + ).scalar_one_or_none() + if running is not None: + return "already running" + run = HeadTrainingRun( + params={"source": "scheduled"}, status="running", + last_progress_at=datetime.now(UTC), + ) + session.add(run) + session.commit() + run_id = run.id + train_heads.delay(run_id) + return "dispatched" + + +@celery.task( + name="backend.app.tasks.ml.apply_head_tags", + bind=True, + # Scores the whole library against the graduated heads and applies their + # tags (or, dry_run, just counts). Streams embeddings in chunks; numpy only, + # but ml queue keeps it off the API workers. Commits per chunk. + soft_time_limit=3600, time_limit=3900, +) +def apply_head_tags(self, run_id: int) -> str: + """Run an earned-auto-apply sweep into the persisted HeadAutoApplyRun row.""" + from datetime import UTC, datetime + + from ..models import HeadAutoApplyRun + from ..services.ml.heads import auto_apply_sweep + + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + run = session.get(HeadAutoApplyRun, run_id) + if run is None: + return "missing" + run.last_progress_at = datetime.now(UTC) + session.commit() + try: + result = auto_apply_sweep(session, run, run.dry_run) + except SoftTimeLimitExceeded: + run.status = "error" + run.error = "timed out" + run.finished_at = datetime.now(UTC) + session.commit() + raise + except Exception as exc: + log.exception("apply_head_tags %d failed", run_id) + run.status = "error" + run.error = str(exc) + run.finished_at = datetime.now(UTC) + session.commit() + return "error" + run.n_applied = result["n_applied"] + run.report = {"concepts": result["concepts"]} + run.status = "ready" + run.finished_at = datetime.now(UTC) + session.commit() + return "ready" + + +@celery.task(name="backend.app.tasks.ml.scheduled_apply_head_tags") +def scheduled_apply_head_tags() -> str: + """Daily passive auto-apply sweep (#114) — only when the master switch is on. + Skips if a sweep is already in flight. Creates + COMMITS the run before + dispatching so the worker always finds it.""" + from datetime import UTC, datetime + + from sqlalchemy import select as sa_select + + from ..models import HeadAutoApplyRun, MLSettings + + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + enabled = session.execute( + sa_select(MLSettings.head_auto_apply_enabled).where(MLSettings.id == 1) + ).scalar_one_or_none() + if not enabled: + return "disabled" + running = session.execute( + sa_select(HeadAutoApplyRun.id).where(HeadAutoApplyRun.status == "running") + ).scalar_one_or_none() + if running is not None: + return "already running" + run = HeadAutoApplyRun( + dry_run=False, params={"dry_run": False, "source": "scheduled"}, + status="running", last_progress_at=datetime.now(UTC), + ) + session.add(run) + session.commit() + run_id = run.id + apply_head_tags.delay(run_id) + return "dispatched" diff --git a/frontend/src/components/settings/HeadsCard.vue b/frontend/src/components/settings/HeadsCard.vue index 673ff1f..a4c5627 100644 --- a/frontend/src/components/settings/HeadsCard.vue +++ b/frontend/src/components/settings/HeadsCard.vue @@ -92,6 +92,105 @@ + + +
+
+ mdi-lightning-bolt + Auto-apply + +
+

+ Graduated heads (⚡, with ≥ {{ autoMinPosInput }} examples) apply their tag + on their own where they clear {{ Math.round((autoPrecisionInput || 0) * 100) }}% + precision. Every auto-tag is reversible; removing one teaches the head it + misfired. +

+ +
+ + +
+ +
+ Preview + Apply now +
+ +
+ +
Sweeping the library…
+
+ +
+
+ {{ lastSweep.dry_run ? 'Preview' : 'Applied' }} · + {{ formatTime(lastSweep.finished_at) }} · + {{ lastSweep.dry_run ? 'would apply' : 'applied' }} + {{ sweepTotal(lastSweep) }} + tag{{ sweepTotal(lastSweep) === 1 ? '' : 's' }} +
+
+ + {{ c.name }} {{ c.applied }} + +
+
+
+ + +
+
How auto-apply is landing
+
+ Misfire = an auto-tag you removed; missed = a tag you added by hand that a + head should have caught. Tune the precision target from the misfire rate. +
+
+ + + + + + + + + + + + + + + + + + + +
Conceptappliedmisfiresratemissed
{{ c.name }}{{ c.n_auto_applied }}{{ c.n_misfires }} + {{ ratePct(c.misfire_rate) }} + {{ c.n_underfires }}
+
+
@@ -109,6 +208,21 @@ const summary = ref(null) const busy = ref(false) let pollTimer = null +// --- Auto-apply state --- +const autoEnabled = ref(false) +const autoPrecisionInput = ref(0.97) +const autoMinPosInput = ref(30) +const settingBusy = ref(false) +const autoBusy = ref(false) +const autoStatus = ref(null) +const metricsData = ref(null) +let autoTimer = null + +const autoRunning = computed(() => autoStatus.value?.running_id != null) +const lastSweep = computed(() => + (autoStatus.value?.runs || []).find(r => r.status !== 'running') || null) +const metricsConcepts = computed(() => metricsData.value?.concepts ?? []) + const headCount = computed(() => summary.value?.head_count ?? 0) const graduatedCount = computed(() => summary.value?.graduated_count ?? 0) const heads = computed(() => summary.value?.heads ?? []) @@ -127,12 +241,21 @@ const startedAgo = computed(() => { }) onMounted(async () => { - // Settings power the "min N tags" copy; non-fatal if it fails. - mlSettings.loadSettings().catch(() => {}) + // Settings power the copy + the auto-apply tuning inputs. + try { + await mlSettings.loadSettings() + const s = mlSettings.settings || {} + autoEnabled.value = !!s.head_auto_apply_enabled + autoPrecisionInput.value = s.head_auto_apply_precision ?? 0.97 + autoMinPosInput.value = s.head_auto_apply_min_positives ?? 30 + } catch { /* non-fatal */ } await refresh() if (running.value) startPoll() + await refreshAuto() + if (autoRunning.value) startAutoPoll() + refreshMetrics() }) -onUnmounted(stopPoll) +onUnmounted(() => { stopPoll(); stopAutoPoll() }) async function refresh() { try { @@ -165,6 +288,82 @@ async function onTrain() { } } +// --- Auto-apply --- +async function refreshAuto() { + try { autoStatus.value = await store.autoApplyStatus() } catch { /* non-fatal */ } +} +async function refreshMetrics() { + try { metricsData.value = await store.metrics() } catch { /* non-fatal */ } +} +function startAutoPoll() { + stopAutoPoll() + autoTimer = setInterval(async () => { + const was = autoRunning.value + await refreshAuto() + // Sweep just finished → refresh the counts + landing metrics. + if (was && !autoRunning.value) { refreshMetrics(); refresh() } + if (!autoRunning.value) stopAutoPoll() + }, 4000) +} +function stopAutoPoll() { if (autoTimer) { clearInterval(autoTimer); autoTimer = null } } + +async function onToggleAuto(val) { + settingBusy.value = true + try { + await mlSettings.patchSettings({ head_auto_apply_enabled: !!val }) + toast({ text: val ? 'Auto-apply on' : 'Auto-apply off', type: 'success' }) + } catch (e) { + autoEnabled.value = !val // revert the switch + toast({ text: `Could not update: ${e.message}`, type: 'error' }) + } finally { + settingBusy.value = false + } +} +async function onSaveSettings() { + settingBusy.value = true + try { + await mlSettings.patchSettings({ + head_auto_apply_precision: Number(autoPrecisionInput.value), + head_auto_apply_min_positives: Number(autoMinPosInput.value), + }) + } catch (e) { + toast({ text: `Could not save: ${e.message}`, type: 'error' }) + } finally { + settingBusy.value = false + } +} +function onPreview() { startSweep(true) } +function onApplyNow() { startSweep(false) } +async function startSweep(dryRun) { + autoBusy.value = true + try { + await store.autoApply(dryRun) + await refreshAuto() + startAutoPoll() + } catch (e) { + const code = e.body?.error + const msg = code === 'auto_apply_already_running' ? 'A sweep is already running.' + : code === 'auto_apply_disabled' ? 'Enable auto-apply first.' + : e.message + toast({ text: `Could not start sweep: ${msg}`, type: 'error' }) + } finally { + autoBusy.value = false + } +} +function sweepConcepts(run) { + return (run?.report?.concepts || []) + .filter(c => c.applied > 0) + .sort((a, b) => b.applied - a.applied) +} +function sweepTotal(run) { return run?.n_applied ?? 0 } +function ratePct(x) { return x == null ? '—' : `${Math.round(x * 100)}%` } +function rateClass(x) { + if (x == null) return '' + if (x <= 0.03) return 'fc-good' + if (x <= 0.1) return 'fc-ok' + return 'fc-weak' +} + function pct(x) { return x == null ? '—' : `${Math.round(x * 100)}%` } function apClass(ap) { if (ap == null) return '' @@ -190,6 +389,21 @@ function relTime(iso) {