feat(heads): earned auto-apply — sweep mechanism, off by default (#114 auto-apply A)

Graduated heads can now apply their tag without a human — gated so it's safe:
- FIRING GATE: a head fires only when the master switch (head_auto_apply_enabled,
  default OFF) is on AND it has >= head_auto_apply_min_positives (default 30)
  clean labels. A precise-looking but under-supported low-N head can't spray tags.
- auto_apply_sweep (heads.py): streams every embedded image in chunks, scores
  against the eligible heads (numpy, no sklearn), applies each head's tag where
  score >= its auto_apply_threshold and the tag isn't already applied/rejected,
  with source='head_auto' (distinguishable + reversible). dry_run counts only.
- HeadAutoApplyRun (migration 0059) tracks each sweep / preview; apply_head_tags
  task (ml queue) + scheduled_apply_head_tags daily beat (no-op unless enabled)
  + recovery sweep + retention(20).
- API: POST /api/heads/auto-apply {dry_run} (202 / 409 running / 400 disabled),
  GET /api/heads/auto-apply (recent runs + per-concept report). Settings
  head_auto_apply_enabled + min_positives via /api/ml/settings.

Tests: sweep applies above threshold, dry-run writes nothing, skips under-
supported + ungraduated heads; API disabled/dry-run/conflict guards.

NEXT (slice 2): the observability the operator asked for — per-concept misfire
(auto-applied-then-removed) + under-fire tracking, time-series snapshots, and a
reporting API to tune. Slice 3: the UI (enable, preview, trends).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa

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.false(),
+        ),
+    )
+    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")

backend/app/api/heads.py

@@ -12,8 +12,14 @@ from quart import Blueprint, jsonify, request
 from sqlalchemy import desc, func, select
 
 from ..extensions import get_session
-from ..models import HeadTrainingRun, Tag, TagHead
+from ..models import HeadAutoApplyRun, HeadTrainingRun, Tag, TagHead
-from ..services.ml.heads import HeadTrainingAlreadyRunning, start_head_training_run
+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 +122,62 @@ 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],
+    })

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

@@ -113,6 +113,10 @@ def make_celery() -> Celery:
                 "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
+            },
             "integrity-verify-weekly": {
                 "task": "backend.app.tasks.maintenance.verify_integrity",
                 "schedule": 604800.0,  # weekly
@@ -168,6 +172,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,7 @@ 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_training_run import HeadTrainingRun
 from .image_prediction import ImagePrediction
 from .image_provenance import ImageProvenance
@@ -67,6 +68,7 @@ __all__ = [
     "ImportSettings",
     "LibraryAuditRun",
     "MLSettings",
+    "HeadAutoApplyRun",
     "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/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,17 @@ 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) ONLY 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. Off by
+    # default; the operator enables after previewing. Operator-tunable.
+    head_auto_apply_enabled: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False
+    )
+    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/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,48 @@ 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
+
+
 @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

@@ -659,3 +659,82 @@ def scheduled_train_heads() -> str:
         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"

tests/test_head_auto_apply.py

@@ -0,0 +1,146 @@
+"""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 = _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):
+    # head_auto_apply_enabled defaults False → a real sweep is refused (400).
+    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"