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")

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")

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
+        ],
+    })

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
 
 

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,

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",

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
+    )

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()
+    )

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)

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"
     )

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}

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

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

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"

frontend/src/components/settings/HeadsCard.vue

@@ -92,6 +92,105 @@
         </table>
       </div>
     </div>
+
+    <!-- Earned auto-apply -->
+    <div class="fc-auto mt-6">
+      <div class="d-flex align-center mb-1" style="gap: 10px;">
+        <v-icon size="18" color="accent">mdi-lightning-bolt</v-icon>
+        <span class="fc-section-h">Auto-apply</span>
+        <v-switch
+          v-model="autoEnabled" :loading="settingBusy" hide-details density="compact"
+          color="success" class="ml-auto" @update:model-value="onToggleAuto"
+        />
+      </div>
+      <p class="fc-muted text-body-2 mb-3">
+        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.
+      </p>
+
+      <div class="d-flex mb-3" style="gap: 12px;">
+        <v-text-field
+          v-model.number="autoPrecisionInput" label="Precision target"
+          type="number" min="0.5" max="0.999" step="0.01" density="compact"
+          hide-details style="max-width: 200px;" :disabled="settingBusy"
+          @change="onSaveSettings"
+        />
+        <v-text-field
+          v-model.number="autoMinPosInput" label="Min examples to fire"
+          type="number" min="1" density="compact" hide-details
+          style="max-width: 200px;" :disabled="settingBusy"
+          @change="onSaveSettings"
+        />
+      </div>
+
+      <div class="d-flex" style="gap: 8px;">
+        <v-btn
+          size="small" variant="tonal" color="accent" rounded="pill"
+          prepend-icon="mdi-eye-outline" :loading="autoBusy" @click="onPreview"
+        >Preview</v-btn>
+        <v-btn
+          size="small" variant="flat" color="accent" rounded="pill"
+          prepend-icon="mdi-lightning-bolt"
+          :loading="autoBusy || autoRunning" :disabled="!autoEnabled"
+          @click="onApplyNow"
+        >Apply now</v-btn>
+      </div>
+
+      <div v-if="autoRunning" class="mt-3">
+        <v-progress-linear indeterminate color="accent" />
+        <div class="text-body-2 mt-2 fc-muted">Sweeping the library…</div>
+      </div>
+
+      <div v-if="lastSweep && !autoRunning" class="mt-3">
+        <div class="fc-muted text-caption mb-1">
+          {{ lastSweep.dry_run ? 'Preview' : 'Applied' }} ·
+          {{ formatTime(lastSweep.finished_at) }} ·
+          {{ lastSweep.dry_run ? 'would apply' : 'applied' }}
+          <strong>{{ sweepTotal(lastSweep) }}</strong>
+          tag{{ sweepTotal(lastSweep) === 1 ? '' : 's' }}
+        </div>
+        <div v-if="sweepConcepts(lastSweep).length" class="fc-chips">
+          <span v-for="c in sweepConcepts(lastSweep)" :key="c.tag_id" class="fc-chip">
+            {{ c.name }} <strong>{{ c.applied }}</strong>
+          </span>
+        </div>
+      </div>
+    </div>
+
+    <!-- Performance / tuning -->
+    <div v-if="metricsConcepts.length" class="mt-5">
+      <div class="fc-section-h mb-1">How auto-apply is landing</div>
+      <div class="fc-muted text-caption mb-2">
+        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.
+      </div>
+      <div class="fc-table-wrap">
+        <table class="fc-table">
+          <thead>
+            <tr>
+              <th class="fc-l">Concept</th>
+              <th class="fc-r" title="Tags currently auto-applied">applied</th>
+              <th class="fc-r" title="Auto-tags you removed">misfires</th>
+              <th class="fc-r" title="Removed / (applied + removed)">rate</th>
+              <th class="fc-r" title="Tags you added by hand that a head exists for">missed</th>
+            </tr>
+          </thead>
+          <tbody>
+            <tr v-for="c in metricsConcepts" :key="c.tag_id">
+              <td class="fc-l">{{ c.name }}</td>
+              <td class="fc-r fc-mono">{{ c.n_auto_applied }}</td>
+              <td class="fc-r fc-mono">{{ c.n_misfires }}</td>
+              <td class="fc-r fc-mono" :class="rateClass(c.misfire_rate)">
+                {{ ratePct(c.misfire_rate) }}
+              </td>
+              <td class="fc-r fc-mono">{{ c.n_underfires }}</td>
+            </tr>
+          </tbody>
+        </table>
+      </div>
+    </div>
   </MaintenanceTile>
 </template>
 
@@ -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) {
 <style scoped>
 .fc-muted { color: rgb(var(--v-theme-on-surface-variant)); }
 
+.fc-section-h {
+  font-size: 13px; font-weight: 700; letter-spacing: 0.03em;
+  text-transform: uppercase; color: rgb(var(--v-theme-on-surface));
+}
+.fc-auto {
+  border-top: 1px solid rgb(var(--v-theme-surface-light)); padding-top: 16px;
+}
+.fc-chips { display: flex; flex-wrap: wrap; gap: 6px; }
+.fc-chip {
+  font-size: 12px; padding: 2px 8px; border-radius: 999px;
+  background: rgb(var(--v-theme-surface-light));
+  color: rgb(var(--v-theme-on-surface-variant));
+}
+.fc-chip strong { color: rgb(var(--v-theme-on-surface)); }
+
 .fc-stats { display: flex; gap: 28px; }
 .fc-stat__n {
   font-size: 22px; font-weight: 700; line-height: 1.1;

frontend/src/composables/useHeadTraining.js

@@ -0,0 +1,67 @@
+import { computed, ref } from 'vue'
+
+import { useHeadsStore } from '../stores/heads.js'
+import { toast } from '../utils/toast.js'
+
+// Shared "(re)train the concept heads" behaviour (#114): trigger + poll status,
+// with toasts on start/finish. Backs both the Settings card and the Explore
+// view's inline button so the active retrain works the same everywhere.
+// Call start() on mount and stop() on unmount to manage the poll timer.
+export function useHeadTraining() {
+  const store = useHeadsStore()
+  const summary = ref(null)
+  const busy = ref(false)        // the trigger POST is in flight
+  let timer = null
+
+  const running = computed(() => summary.value?.running_id != null)
+  const headCount = computed(() => summary.value?.head_count ?? 0)
+
+  async function refresh() {
+    try {
+      summary.value = await store.status()
+    } catch { /* non-fatal — the button still offers a fresh train */ }
+  }
+
+  function startPoll() {
+    stopPoll()
+    timer = setInterval(async () => {
+      const wasRunning = running.value
+      await refresh()
+      // Announce completion once, on the running → done transition.
+      if (wasRunning && !running.value) {
+        toast({
+          text: `Heads retrained — ${headCount.value} concept${headCount.value === 1 ? '' : 's'}`,
+          type: 'success',
+        })
+      }
+      if (!running.value) stopPoll()
+    }, 5000)
+  }
+
+  function stopPoll() {
+    if (timer) { clearInterval(timer); timer = null }
+  }
+
+  // Reflect an already-running run (e.g. the nightly one) on mount.
+  async function start() {
+    await refresh()
+    if (running.value) startPoll()
+  }
+
+  async function train() {
+    busy.value = true
+    try {
+      await store.train()
+      toast({ text: 'Head training started…', type: 'info' })
+      await refresh()
+      startPoll()
+    } catch (e) {
+      const msg = e.body?.running_id ? 'Training is already running.' : e.message
+      toast({ text: `Could not start training: ${msg}`, type: 'error' })
+    } finally {
+      busy.value = false
+    }
+  }
+
+  return { summary, running, busy, headCount, refresh, train, start, stop: stopPoll }
+}

frontend/src/stores/explore.js

@@ -19,6 +19,10 @@ export const useExploreStore = defineStore('explore', () => {
   const anchor = ref(null)       // /api/gallery/image/<id> payload
   const neighbors = ref([])      // [{id, thumbnail_url, ...}]
   const breadcrumb = ref([])     // [{id, thumbnail_url}] walked path
+  // Index of the current anchor within breadcrumb — browser-style back/forward.
+  // The trail keeps its forward branch when you step back (so ← / → can move
+  // through visited items); a NEW walk off a back-step truncates that branch.
+  const cursor = ref(-1)
   const loading = ref(false)
   const error = ref(null)
 
@@ -54,13 +58,39 @@ export const useExploreStore = defineStore('explore', () => {
     }
   }
 
-  // Forward walk appends; navigating to an id already in the trail (a
-  // breadcrumb click, or a loop back) TRIMS to it — so the route stays the
-  // single source of truth and the crumb bar never grows stale branches.
+  // The route is the source of truth; this reconciles the trail + cursor to it.
+  // Revisiting an id already on the path (← / →, a crumb click, or a loop) just
+  // MOVES the cursor there — the forward branch is preserved so → can return to
+  // it. A genuinely new image off a back-step truncates the stale forward branch
+  // (browser semantics), then appends and points the cursor at the new tip.
   function _reconcileTrail (id, thumbnailUrl) {
     const idx = breadcrumb.value.findIndex((c) => c.id === id)
-    if (idx >= 0) breadcrumb.value = breadcrumb.value.slice(0, idx + 1)
-    else breadcrumb.value = [...breadcrumb.value, { id, thumbnail_url: thumbnailUrl }]
+    if (idx >= 0) {
+      cursor.value = idx
+      return
+    }
+    const base = cursor.value < breadcrumb.value.length - 1
+      ? breadcrumb.value.slice(0, cursor.value + 1)
+      : breadcrumb.value
+    breadcrumb.value = [...base, { id, thumbnail_url: thumbnailUrl }]
+    cursor.value = breadcrumb.value.length - 1
+  }
+
+  // ← target: the previous crumb (null at the start of the trail).
+  function backTarget () {
+    return cursor.value > 0 ? breadcrumb.value[cursor.value - 1].id : null
+  }
+
+  // → target: the next already-visited crumb if we'd stepped back, else a
+  // RANDOM neighbour to keep the rabbit-hole going. Null if neither exists.
+  function forwardTarget () {
+    if (cursor.value >= 0 && cursor.value < breadcrumb.value.length - 1) {
+      return breadcrumb.value[cursor.value + 1].id
+    }
+    if (neighbors.value.length) {
+      return neighbors.value[Math.floor(Math.random() * neighbors.value.length)].id
+    }
+    return null
   }
 
   function reset () {
@@ -68,6 +98,7 @@ export const useExploreStore = defineStore('explore', () => {
     anchor.value = null
     neighbors.value = []
     breadcrumb.value = []
+    cursor.value = -1
     error.value = null
     loading.value = false
   }
@@ -145,8 +176,8 @@ export const useExploreStore = defineStore('explore', () => {
   function close () {}
 
   return {
-    anchor, neighbors, breadcrumb, loading, error, NEIGHBOR_LIMIT,
-    anchorOn, reset,
+    anchor, neighbors, breadcrumb, cursor, loading, error, NEIGHBOR_LIMIT,
+    anchorOn, reset, backTarget, forwardTarget,
     // host surface
     current, currentImageId,
     reloadTags, addExistingTag, removeTag, createAndAdd, close,

frontend/src/stores/heads.js

@@ -20,5 +20,22 @@ export const useHeadsStore = defineStore('heads', () => {
     return await api.post('/api/heads/train', { body: { params } })
   }
 
-  return { status, train }
+  // Earned auto-apply: trigger a sweep. dry_run previews (writes nothing);
+  // a real sweep needs head_auto_apply_enabled on (else 400).
+  async function autoApply(dryRun = false) {
+    return await api.post('/api/heads/auto-apply', { body: { dry_run: dryRun } })
+  }
+
+  // Recent sweeps + per-concept report (volume / projected per head).
+  async function autoApplyStatus() {
+    return await api.get('/api/heads/auto-apply')
+  }
+
+  // Observability: per-concept counts (volume, misfires, under-fires, realized
+  // misfire rate, head quality) + the daily time-series, to tune from.
+  async function metrics() {
+    return await api.get('/api/heads/metrics')
+  }
+
+  return { status, train, autoApply, autoApplyStatus, metrics }
 })

frontend/src/views/ExploreView.vue

@@ -22,19 +22,30 @@
         <button
           v-for="(c, i) in store.breadcrumb" :key="c.id"
           class="fc-ex__crumb"
-          :class="{ 'fc-ex__crumb--current': i === store.breadcrumb.length - 1 }"
+          :class="{ 'fc-ex__crumb--current': i === store.cursor }"
           type="button"
-          :aria-current="i === store.breadcrumb.length - 1 ? 'page' : undefined"
+          :aria-current="i === store.cursor ? 'page' : undefined"
           :title="`Step ${i + 1}`"
           @click="goTo(c.id)"
         >
           <img :src="c.thumbnail_url" alt="" loading="lazy" />
         </button>
-        <v-btn
-          class="fc-ex__reseed" size="small" variant="text" color="accent"
-          prepend-icon="mdi-shuffle-variant" :loading="seeding"
-          @click="reseed"
-        >Random image</v-btn>
+        <div class="fc-ex__trail-actions">
+          <!-- Active retrain right where you tag: fold the +/- you just gave
+               into the heads without a trip to Settings (the nightly beat is the
+               passive cadence). -->
+          <v-btn
+            size="small" variant="text" color="accent"
+            prepend-icon="mdi-brain" :loading="headsBusy || headsRunning"
+            title="Retrain the concept heads on your latest tags"
+            @click="trainHeads"
+          >{{ headsRunning ? 'Training…' : 'Retrain heads' }}</v-btn>
+          <v-btn
+            size="small" variant="text" color="accent"
+            prepend-icon="mdi-shuffle-variant" :loading="seeding"
+            @click="reseed"
+          >Random image</v-btn>
+        </div>
       </nav>
 
       <v-alert v-if="store.error" type="error" variant="tonal" class="ma-3">
@@ -115,6 +126,7 @@ import { useRoute, useRouter } from 'vue-router'
 import { useApi } from '../composables/useApi.js'
 import { useExploreStore } from '../stores/explore.js'
 import { useModalStore } from '../stores/modal.js'
+import { useHeadTraining } from '../composables/useHeadTraining.js'
 import { isTextEntry } from '../utils/textEntry.js'
 import ImageCanvas from '../components/modal/ImageCanvas.vue'
 import ImageMetaBar from '../components/modal/ImageMetaBar.vue'
@@ -132,6 +144,13 @@ const seeding = ref(false)
 const seedError = ref(null)
 const tagPanelRef = ref(null)
 
+// Inline head-retrain (shared with the Settings card) so banking your latest
+// +/- feedback is one click while you walk content.
+const {
+  running: headsRunning, busy: headsBusy, train: trainHeads,
+  start: startHeads, stop: stopHeads,
+} = useHeadTraining()
+
 // Auto-focus the tag input after any action so tagging needs no extra click —
 // the whole point of the workspace (operator-asked 2026-06-26). nextTick waits
 // for the post-navigation re-render, then rAF lands the focus AFTER paint so a
@@ -191,16 +210,39 @@ function goTo (id) {
 
 function openInViewer (id) { modal.open(id) }
 
-// Modal-parity focus: T or "/" jumps to the tag input (the same shortcut the
-// image modal binds), unless the caret is already in a text field.
+// Keyboard: T or "/" jumps to the tag input (modal-parity); ←/→ walk the
+// breadcrumb trail — ← steps back, → goes forward to an already-visited item
+// or, with no forward history, jumps to a random neighbour (keep rabbit-holing).
 function onKeyDown (ev) {
-  if ((ev.key === 't' || ev.key === '/') && !isTextEntry(ev.target)) {
-    const input = document.querySelector('.fc-tag-autocomplete input')
-    if (input) { ev.preventDefault(); input.focus() }
+  if (ev.metaKey || ev.ctrlKey || ev.altKey) return
+  const inText = isTextEntry(ev.target)
+  if (ev.key === 't' || ev.key === '/') {
+    if (!inText) {
+      const input = document.querySelector('.fc-tag-autocomplete input')
+      if (input) { ev.preventDefault(); input.focus() }
+    }
+    return
   }
+  if (ev.key !== 'ArrowLeft' && ev.key !== 'ArrowRight') return
+  // The tag input auto-focuses after every walk, so also allow navigation while
+  // it's focused-but-EMPTY (the caret has nowhere to go) — otherwise arrow-nav
+  // would dead-end after one step. Once a tag is being typed, arrows move the
+  // caret instead.
+  const t = ev.target
+  const inEmptyTagInput =
+    inText && t.value === '' && t.closest?.('.fc-tag-autocomplete')
+  if (inText && !inEmptyTagInput) return
+  const id = ev.key === 'ArrowLeft' ? store.backTarget() : store.forwardTarget()
+  if (id != null) { ev.preventDefault(); goTo(id) }
 }
-onMounted(() => document.addEventListener('keydown', onKeyDown))
-onUnmounted(() => document.removeEventListener('keydown', onKeyDown))
+onMounted(() => {
+  document.addEventListener('keydown', onKeyDown)
+  startHeads()   // reflect a run already in flight (e.g. the nightly one)
+})
+onUnmounted(() => {
+  document.removeEventListener('keydown', onKeyDown)
+  stopHeads()
+})
 </script>
 
 <style scoped>
@@ -233,7 +275,9 @@ onUnmounted(() => document.removeEventListener('keydown', onKeyDown))
 .fc-ex__crumb img { width: 100%; height: 100%; object-fit: cover; }
 .fc-ex__crumb--current { border-color: rgb(var(--v-theme-accent)); }
 .fc-ex__crumb:focus-visible { outline: 2px solid rgb(var(--v-theme-accent)); outline-offset: 1px; }
-.fc-ex__reseed { margin-left: auto; }
+.fc-ex__trail-actions {
+  margin-left: auto; display: flex; align-items: center; gap: 4px; flex: 0 0 auto;
+}
 
 /* The three panes fill the remaining height; each scrolls on its own.
    grid-template-rows: minmax(0, 1fr) BOUNDS the single row to the container

tests/test_head_auto_apply.py

@@ -0,0 +1,150 @@
+"""Earned auto-apply (#114). The sweep is numpy-only (no scikit-learn), so the
+apply logic is tested directly via the sync session; the API guards (disabled /
+dry-run / conflict) via the async client."""
+import pytest
+from sqlalchemy import select
+
+from backend.app.models import (
+    HeadAutoApplyRun,
+    ImageRecord,
+    MLSettings,
+    Tag,
+    TagHead,
+    TagKind,
+)
+from backend.app.models.tag import image_tag
+from backend.app.services.ml.heads import auto_apply_sweep
+
+pytestmark = pytest.mark.integration
+
+
+def _emb(slot: int) -> list[float]:
+    v = [0.0] * 1152
+    v[slot] = 3.0
+    return v
+
+
+def _img(db, sha: str, emb) -> ImageRecord:
+    img = ImageRecord(
+        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
+        width=1, height=1, origin="imported_filesystem",
+        integrity_status="unknown", siglip_embedding=emb,
+    )
+    db.add(img)
+    db.flush()
+    return img
+
+
+def _head(db, tag_id: int, slot: int, *, threshold=0.5, n_pos=30):
+    s = db.execute(select(MLSettings).where(MLSettings.id == 1)).scalar_one()
+    w = [0.0] * 1152
+    w[slot] = 1.0
+    db.add(TagHead(
+        tag_id=tag_id, embedding_version=s.embedder_model_version,
+        weights=w, bias=0.0, suggest_threshold=0.5, auto_apply_threshold=threshold,
+        n_pos=n_pos, n_neg=90, ap=0.9, precision_cv=0.98, recall=0.7,
+    ))
+
+
+def _run(db, dry_run=False) -> HeadAutoApplyRun:
+    run = HeadAutoApplyRun(dry_run=dry_run, params={"dry_run": dry_run}, status="running")
+    db.add(run)
+    db.flush()
+    return run
+
+
+def _applied_source(db, image_id, tag_id):
+    return db.execute(
+        select(image_tag.c.source)
+        .where(image_tag.c.image_record_id == image_id)
+        .where(image_tag.c.tag_id == tag_id)
+    ).scalar_one_or_none()
+
+
+def test_sweep_applies_to_matching_image(db_sync):
+    img = _img(db_sync, "a" * 64, _emb(0))
+    tag = Tag(name="autotag", kind=TagKind.general)
+    db_sync.add(tag)
+    db_sync.flush()
+    _head(db_sync, tag.id, 0)
+    run = _run(db_sync)
+    db_sync.commit()
+    result = auto_apply_sweep(db_sync, run, dry_run=False)
+    assert result["n_applied"] == 1
+    assert _applied_source(db_sync, img.id, tag.id) == "head_auto"
+
+
+def test_sweep_dry_run_counts_but_writes_nothing(db_sync):
+    img = _img(db_sync, "b" * 64, _emb(0))
+    tag = Tag(name="previewtag", kind=TagKind.general)
+    db_sync.add(tag)
+    db_sync.flush()
+    _head(db_sync, tag.id, 0)
+    run = _run(db_sync, dry_run=True)
+    db_sync.commit()
+    result = auto_apply_sweep(db_sync, run, dry_run=True)
+    assert result["n_applied"] == 1                      # it WOULD apply
+    assert _applied_source(db_sync, img.id, tag.id) is None  # but wrote nothing
+
+
+def test_sweep_skips_under_supported_head(db_sync):
+    # n_pos below head_auto_apply_min_positives (default 30) → a precise-looking
+    # but under-supported head never fires.
+    img = _img(db_sync, "c" * 64, _emb(0))
+    tag = Tag(name="weaktag", kind=TagKind.general)
+    db_sync.add(tag)
+    db_sync.flush()
+    _head(db_sync, tag.id, 0, n_pos=5)
+    run = _run(db_sync)
+    db_sync.commit()
+    result = auto_apply_sweep(db_sync, run, dry_run=False)
+    assert result["n_applied"] == 0
+    assert _applied_source(db_sync, img.id, tag.id) is None
+
+
+def test_sweep_skips_ungraduated_head(db_sync):
+    # auto_apply_threshold is None (head never reached the precision bar).
+    _img(db_sync, "d" * 64, _emb(0))
+    tag = Tag(name="nograd", kind=TagKind.general)
+    db_sync.add(tag)
+    db_sync.flush()
+    _head(db_sync, tag.id, 0, threshold=None)
+    run = _run(db_sync)
+    db_sync.commit()
+    result = auto_apply_sweep(db_sync, run, dry_run=False)
+    assert result["n_applied"] == 0
+
+
+@pytest.mark.asyncio
+async def test_auto_apply_disabled_blocks_real_run(client, db):
+    # With the master switch OFF, a real sweep is refused (400). (It defaults ON
+    # now — opt-out — so the test disables it explicitly to exercise this path.)
+    s = (await db.execute(select(MLSettings).where(MLSettings.id == 1))).scalar_one()
+    s.head_auto_apply_enabled = False
+    await db.commit()
+    resp = await client.post("/api/heads/auto-apply", json={"dry_run": False})
+    assert resp.status_code == 400
+    assert (await resp.get_json())["error"] == "auto_apply_disabled"
+
+
+@pytest.mark.asyncio
+async def test_auto_apply_dry_run_allowed_when_disabled(client, db, monkeypatch):
+    monkeypatch.setattr(
+        "backend.app.tasks.ml.apply_head_tags.delay", lambda *a, **k: None
+    )
+    resp = await client.post("/api/heads/auto-apply", json={"dry_run": True})
+    assert resp.status_code == 202
+    assert (await resp.get_json())["status"] == "running"
+
+
+@pytest.mark.asyncio
+async def test_auto_apply_conflict_when_one_running(client, db, monkeypatch):
+    monkeypatch.setattr(
+        "backend.app.tasks.ml.apply_head_tags.delay", lambda *a, **k: None
+    )
+    db.add(HeadAutoApplyRun(dry_run=True, params={}, status="running"))
+    await db.flush()
+    await db.commit()
+    resp = await client.post("/api/heads/auto-apply", json={"dry_run": True})
+    assert resp.status_code == 409
+    assert (await resp.get_json())["error"] == "auto_apply_already_running"

tests/test_head_metrics.py

@@ -0,0 +1,107 @@
+"""Auto-apply observability (#114): misfire/under-fire counters captured on
+operator corrections, the daily snapshot time-series, and the metrics API."""
+import pytest
+from sqlalchemy import select
+
+from backend.app.models import HeadMetric, HeadMetricsSnapshot, ImageRecord, TagHead, TagKind
+from backend.app.models.tag import image_tag
+from backend.app.services.tag_service import TagService
+
+pytestmark = pytest.mark.integration
+
+
+async def _img(db, sha) -> ImageRecord:
+    img = ImageRecord(
+        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
+        width=1, height=1, origin="imported_filesystem", integrity_status="unknown",
+    )
+    db.add(img)
+    await db.flush()
+    return img
+
+
+def _head(tag_id):
+    return TagHead(
+        tag_id=tag_id, embedding_version="siglip-test", weights=[0.0] * 1152,
+        bias=0.0, suggest_threshold=0.5, auto_apply_threshold=0.6,
+        n_pos=30, n_neg=90, ap=0.9, precision_cv=0.95, recall=0.7,
+    )
+
+
+@pytest.mark.asyncio
+async def test_removing_head_auto_tag_counts_misfire(db):
+    img = await _img(db, "a" * 64)
+    tag = await TagService(db).find_or_create("misfire", TagKind.general)
+    await db.execute(image_tag.insert().values(
+        image_record_id=img.id, tag_id=tag.id, source="head_auto",
+    ))
+    await db.commit()
+    await TagService(db).remove_from_image(img.id, tag.id)
+    await db.commit()
+    m = await db.get(HeadMetric, tag.id)
+    assert m is not None and m.n_misfires == 1 and m.n_underfires == 0
+
+
+@pytest.mark.asyncio
+async def test_removing_manual_tag_is_not_a_misfire(db):
+    img = await _img(db, "b" * 64)
+    tag = await TagService(db).find_or_create("manualrm", TagKind.general)
+    await db.execute(image_tag.insert().values(
+        image_record_id=img.id, tag_id=tag.id, source="manual",
+    ))
+    await db.commit()
+    await TagService(db).remove_from_image(img.id, tag.id)
+    await db.commit()
+    assert await db.get(HeadMetric, tag.id) is None
+
+
+@pytest.mark.asyncio
+async def test_manual_add_with_head_counts_underfire(db):
+    img = await _img(db, "c" * 64)
+    tag = await TagService(db).find_or_create("underfire", TagKind.general)
+    db.add(_head(tag.id))
+    await db.commit()
+    await TagService(db).add_to_image(img.id, tag.id, source="manual")
+    await db.commit()
+    m = await db.get(HeadMetric, tag.id)
+    assert m is not None and m.n_underfires == 1
+
+
+@pytest.mark.asyncio
+async def test_manual_add_without_head_no_underfire(db):
+    img = await _img(db, "d" * 64)
+    tag = await TagService(db).find_or_create("nohead", TagKind.general)
+    await db.commit()
+    await TagService(db).add_to_image(img.id, tag.id, source="manual")
+    await db.commit()
+    assert await db.get(HeadMetric, tag.id) is None
+
+
+@pytest.mark.asyncio
+async def test_snapshot_records_timeseries_point(db):
+    tag = await TagService(db).find_or_create("snap", TagKind.general)
+    db.add(_head(tag.id))
+    await db.commit()
+    from backend.app.tasks.maintenance import snapshot_head_metrics
+
+    n = snapshot_head_metrics()   # sync task, own session
+    assert n >= 1
+    snaps = (await db.execute(
+        select(HeadMetricsSnapshot).where(HeadMetricsSnapshot.tag_id == tag.id)
+    )).scalars().all()
+    assert len(snaps) == 1
+    assert snaps[0].name == "snap"
+
+
+@pytest.mark.asyncio
+async def test_metrics_api_returns_concept(client, db):
+    tag = await TagService(db).find_or_create("apimetric", TagKind.general)
+    db.add(_head(tag.id))
+    await db.commit()
+    resp = await client.get("/api/heads/metrics")
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    c = next(x for x in body["concepts"] if x["name"] == "apimetric")
+    assert c["auto_apply"] is True
+    assert c["n_misfires"] == 0
+    assert "snapshots" in body