diff --git a/alembic/versions/0059_head_auto_apply.py b/alembic/versions/0059_head_auto_apply.py new file mode 100644 index 0000000..fc9afaf --- /dev/null +++ b/alembic/versions/0059_head_auto_apply.py @@ -0,0 +1,70 @@ +"""head_auto_apply_run + earned-auto-apply settings (#114) + +A graduated head can apply its tag without a human, gated by a master switch + +a support floor. head_auto_apply_run tracks each sweep / dry-run preview. + +Revision ID: 0059 +Revises: 0058 +Create Date: 2026-06-29 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects.postgresql import JSONB + +revision: str = "0059" +down_revision: Union[str, None] = "0058" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "head_auto_apply_run", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column( + "dry_run", sa.Boolean(), nullable=False, server_default=sa.false() + ), + sa.Column("params", JSONB(), nullable=False), + sa.Column( + "status", sa.String(length=16), nullable=False, + server_default="running", + ), + sa.Column( + "started_at", sa.DateTime(timezone=True), nullable=False, + server_default=sa.func.now(), + ), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("n_applied", sa.Integer(), nullable=True), + sa.Column("report", JSONB(), nullable=True), + sa.Column("error", sa.Text(), nullable=True), + sa.Column("last_progress_at", sa.DateTime(timezone=True), nullable=True), + ) + op.create_index( + "ix_head_auto_apply_run_status", "head_auto_apply_run", ["status"], + ) + + op.add_column( + "ml_settings", + sa.Column( + "head_auto_apply_enabled", sa.Boolean(), nullable=False, + server_default=sa.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") diff --git a/backend/app/api/heads.py b/backend/app/api/heads.py index 018ee2c..5b6243f 100644 --- a/backend/app/api/heads.py +++ b/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 ..services.ml.heads import HeadTrainingAlreadyRunning, start_head_training_run +from ..models import HeadAutoApplyRun, HeadTrainingRun, Tag, TagHead +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], + }) diff --git a/backend/app/api/ml_admin.py b/backend/app/api/ml_admin.py index a3d3b74..1b0e1d1 100644 --- a/backend/app/api/ml_admin.py +++ b/backend/app/api/ml_admin.py @@ -19,6 +19,8 @@ _EDITABLE = ( "video_min_tag_frames", "head_min_positives", "head_auto_apply_precision", + "head_auto_apply_enabled", + "head_auto_apply_min_positives", ) @@ -44,6 +46,8 @@ async def get_settings(): "embedder_model_version": s.embedder_model_version, "head_min_positives": s.head_min_positives, "head_auto_apply_precision": s.head_auto_apply_precision, + "head_auto_apply_enabled": s.head_auto_apply_enabled, + "head_auto_apply_min_positives": s.head_auto_apply_min_positives, } ) @@ -109,6 +113,8 @@ def _validate(p: dict) -> str | None: return "head_min_positives must be >= 1" if not (0.5 <= float(p["head_auto_apply_precision"]) <= 0.999): return "head_auto_apply_precision must be between 0.5 and 0.999" + if int(p["head_auto_apply_min_positives"]) < 1: + return "head_auto_apply_min_positives must be >= 1" return None diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index f4d3751..80e77a3 100644 --- a/backend/app/celery_app.py +++ b/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, diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 04025ed..e688bf0 100644 --- a/backend/app/models/__init__.py +++ b/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", diff --git a/backend/app/models/head_auto_apply_run.py b/backend/app/models/head_auto_apply_run.py new file mode 100644 index 0000000..08c359a --- /dev/null +++ b/backend/app/models/head_auto_apply_run.py @@ -0,0 +1,46 @@ +"""HeadAutoApplyRun — persisted lifecycle of an earned-auto-apply sweep (#114). + +A graduated head can apply its tag to images it scores above the head's +auto-apply threshold, without a human. This row tracks one such sweep (or a +dry-run PREVIEW of it) so the result survives navigation and the admin card can +show what fired / what would fire. Mirrors HeadTrainingRun. State machine: +running → ready / error. The `report` JSONB holds per-concept counts +(applied / projected / scanned). +""" + +from datetime import datetime +from typing import Any + +from sqlalchemy import Boolean, DateTime, Integer, String, Text, func +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from .base import Base + + +class HeadAutoApplyRun(Base): + __tablename__ = "head_auto_apply_run" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + # dry_run=True is a PREVIEW: scores + counts what WOULD apply, writes nothing + # (preview/apply parity, rule 93). + dry_run: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + params: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False) + status: Mapped[str] = mapped_column( + String(16), nullable=False, default="running", index=True + ) + # running | ready | error + started_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + finished_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + # Total tags applied across all heads this sweep (0 for a clean dry-run). + n_applied: Mapped[int | None] = mapped_column(Integer, nullable=True) + # Per-concept breakdown: [{tag_id, name, applied, scanned, threshold}, ...]. + report: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) + error: Mapped[str | None] = mapped_column(Text, nullable=True) + last_progress_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) diff --git a/backend/app/models/ml_settings.py b/backend/app/models/ml_settings.py index 243387a..ea815f2 100644 --- a/backend/app/models/ml_settings.py +++ b/backend/app/models/ml_settings.py @@ -2,7 +2,15 @@ from datetime import datetime -from sqlalchemy import CheckConstraint, DateTime, Float, Integer, String, func +from sqlalchemy import ( + Boolean, + CheckConstraint, + DateTime, + Float, + Integer, + String, + func, +) from sqlalchemy.orm import Mapped, mapped_column from .base import Base @@ -66,6 +74,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" ) diff --git a/backend/app/services/ml/heads.py b/backend/app/services/ml/heads.py index aed4e69..4b07e68 100644 --- a/backend/app/services/ml/heads.py +++ b/backend/app/services/ml/heads.py @@ -26,12 +26,14 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session from ...models import ( + HeadAutoApplyRun, HeadTrainingRun, ImageRecord, MLSettings, Tag, TagHead, TagKind, + TagSuggestionRejection, ) from ...models.tag import image_tag from .tag_eval import ( @@ -328,3 +330,138 @@ async def _settings_async(session: AsyncSession) -> MLSettings: return ( await session.execute(select(MLSettings).where(MLSettings.id == 1)) ).scalar_one() + + +# --- Earned auto-apply (sync, ml worker) --------------------------------- +# A graduated head can apply its tag to images it scores above the head's +# auto_apply_threshold, without a human. Gated by a master switch + a support +# floor so a precise-looking but under-supported head can't spray tags. + +_AUTO_APPLY_CHUNK = 5000 + + +class HeadAutoApplyAlreadyRunning(Exception): + """Raised when an auto-apply sweep is already in flight.""" + + +class HeadAutoApplyDisabled(Exception): + """Raised when a real (non-dry-run) sweep is requested but the master + switch (head_auto_apply_enabled) is off.""" + + +def start_head_auto_apply_run(session: Session, params: dict[str, Any]) -> int: + """Create a HeadAutoApplyRun + dispatch the ml-queue sweep. dry_run previews + (writes nothing); a real sweep needs the master switch on. One run at a time.""" + dry_run = bool((params or {}).get("dry_run", False)) + existing = session.execute( + select(HeadAutoApplyRun.id).where(HeadAutoApplyRun.status == "running") + ).scalar_one_or_none() + if existing is not None: + raise HeadAutoApplyAlreadyRunning(existing) + if not dry_run and not _settings(session).head_auto_apply_enabled: + raise HeadAutoApplyDisabled() + run = HeadAutoApplyRun( + dry_run=dry_run, params={"dry_run": dry_run}, status="running", + last_progress_at=datetime.now(UTC), + ) + session.add(run) + session.flush() + run_id = run.id + from ...tasks.ml import apply_head_tags as _task + _task.delay(run_id) + return run_id + + +def _auto_apply_heads(session: Session, embedding_version: str, min_pos: int): + """Eligible heads to fire: graduated (auto_apply_threshold set), enough + support, current embedding. Returns the row list (tag_id/name/weights/...).""" + return session.execute( + select( + TagHead.tag_id, Tag.name, TagHead.weights, TagHead.bias, + TagHead.auto_apply_threshold, + ) + .join(Tag, Tag.id == TagHead.tag_id) + .where(TagHead.embedding_version == embedding_version) + .where(TagHead.auto_apply_threshold.is_not(None)) + .where(TagHead.n_pos >= min_pos) + ).all() + + +def auto_apply_sweep( + session: Session, run: HeadAutoApplyRun, dry_run: bool +) -> dict[str, Any]: + """Score every embedded image against the eligible heads and apply (or, for + dry_run, just count) each head's tag where score >= its auto_apply_threshold + and the tag isn't already applied or rejected on that image. Streams + embeddings in chunks; commits per chunk on a real run. Returns + {n_applied, concepts:[{tag_id,name,applied,scanned,threshold}]}.""" + import numpy as np + from sqlalchemy.dialects.postgresql import insert as pg_insert + + settings = _settings(session) + rows = _auto_apply_heads( + session, settings.embedder_model_version, + settings.head_auto_apply_min_positives, + ) + if not rows: + return {"n_applied": 0, "concepts": []} + + W = np.vstack([np.asarray(r.weights, dtype=np.float32) for r in rows]) + b = np.asarray([r.bias for r in rows], dtype=np.float32) + thr = np.asarray([r.auto_apply_threshold for r in rows], dtype=np.float32) + tag_ids = [r.tag_id for r in rows] + names = [r.name for r in rows] + + # Skip images that already carry, or have rejected, each tag. + skip = {tid: set() for tid in tag_ids} + for tid in tag_ids: + for (iid,) in session.execute( + select(image_tag.c.image_record_id).where(image_tag.c.tag_id == tid) + ): + skip[tid].add(iid) + for (iid,) in session.execute( + select(TagSuggestionRejection.image_record_id).where( + TagSuggestionRejection.tag_id == tid + ) + ): + skip[tid].add(iid) + + applied = [0] * len(rows) + scanned = 0 + all_ids = list(session.execute( + select(ImageRecord.id).where(ImageRecord.siglip_embedding.is_not(None)) + ).scalars()) + for start in range(0, len(all_ids), _AUTO_APPLY_CHUNK): + chunk = all_ids[start:start + _AUTO_APPLY_CHUNK] + emb = _load_embeddings(session, chunk) + cids = [i for i in chunk if i in emb] + if not cids: + continue + Xn = _l2norm(np.vstack([emb[i] for i in cids]).astype(np.float32), np) + probs = 1.0 / (1.0 + np.exp(-(Xn @ W.T + b))) # (N, H) + scanned += len(cids) + for h in range(len(rows)): + tid = tag_ids[h] + for idx in np.where(probs[:, h] >= thr[h])[0]: + iid = cids[int(idx)] + if iid in skip[tid]: + continue + skip[tid].add(iid) + applied[h] += 1 + if not dry_run: + session.execute( + pg_insert(image_tag) + .values(image_record_id=iid, tag_id=tid, source="head_auto") + .on_conflict_do_nothing() + ) + if not dry_run: + session.commit() + run.last_progress_at = datetime.now(UTC) + session.commit() + + concepts = [ + {"tag_id": tag_ids[h], "name": names[h], "applied": applied[h], + "scanned": scanned, "threshold": float(thr[h])} + for h in range(len(rows)) + ] + return {"n_applied": sum(applied), "concepts": concepts} diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index fa0397b..217b755 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -13,6 +13,7 @@ from ..celery_app import celery from ..models import ( BackupRun, DownloadEvent, + HeadAutoApplyRun, HeadTrainingRun, ImageRecord, ImportBatch, @@ -101,6 +102,9 @@ TAG_EVAL_KEEP_RUNS = 20 # head training (#114) has a 60-min soft limit; flag no-progress past 75. HEAD_TRAINING_STALL_THRESHOLD_MINUTES = 75 HEAD_TRAINING_KEEP_RUNS = 20 +# head auto-apply (#114) shares the 60-min soft limit; flag past 75. +HEAD_AUTO_APPLY_STALL_THRESHOLD_MINUTES = 75 +HEAD_AUTO_APPLY_KEEP_RUNS = 20 # Import batches finalize only after every child ImportTask hits a # terminal state. The recovery sweep targets the case where every # task is done but the batch never got its closing UPDATE @@ -800,6 +804,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 diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index b6366fe..bc22c5b 100644 --- a/backend/app/tasks/ml.py +++ b/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" diff --git a/tests/test_head_auto_apply.py b/tests/test_head_auto_apply.py new file mode 100644 index 0000000..e56b98b --- /dev/null +++ b/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"