2026-06-11 19:32:05 -04:00
17 changed files with 74 additions and 445 deletions
@@ -0,0 +1,43 @@
 """drop image_record.tagger_predictions (predictions normalized to image_prediction)
 Final step of #768. The per-tag predictions now live in the image_prediction
 table (backfilled from the JSON, read by suggestions + allowlist, written by
 tag_and_embed). The old JSON column is dead weight — and it's the ~100 GB of
 sub-0.70 score tail that bloated image_record's TOAST and broke DB backups
 (#739). Dropping it is a fast catalog change; it does NOT reclaim the disk on
 its own — run `VACUUM FULL image_record` (or pg_repack) afterward, off-hours,
 to return the space to the OS so backups go small.
 DROP COLUMN needs a brief ACCESS EXCLUSIVE lock on image_record; env.py's
 lock_timeout guards it, so quiesce the ml-worker if a tagging run is in flight
 (see the migration-lock reference). tagger_model_version is kept — it's the
 "has this been tagged / is it current?" signal the backfill sweep reads.
 Revision ID: 0046
 Revises: 0045
 Create Date: 2026-06-11
 """
 from typing import Sequence, Union
 import sqlalchemy as sa
 from alembic import op
 revision: str = "0046"
 down_revision: Union[str, None] = "0045"
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 def upgrade() -> None:
    op.drop_column("image_record", "tagger_predictions")
 def downgrade() -> None:
    # Re-add the column empty. The JSON data is not restored (it lived only in
    # this column); a downgrade would re-tag or backfill from image_prediction
    # separately if ever needed.
    op.add_column(
        "image_record",
        sa.Column("tagger_predictions", sa.JSON(), nullable=True),
    )
@@ -251,7 +251,7 @@ async def tags_reset_content():
    """Tier-A: delete ALL general + character tags (the Camie-suggestable
    content vocabulary) so the operator can re-tag from scratch via
    auto-suggest. fandom + series tags + series_page ordering are preserved,
-    and image tagger_predictions are untouched so suggestions repopulate.
+    and image_prediction rows are untouched so suggestions repopulate.
    dry-run preview returns per-kind counts + applications + a sample so the
    UI shows exactly what'll go before the operator confirms (dry_run=false).
    Irreversible except via DB backup restore."""
@@ -348,28 +348,3 @@ async def trigger_reextract_archives():
    async_result = reextract_archive_attachments_task.delay()
    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
@admin_bp.route("/maintenance/prune-predictions", methods=["POST"])
 async def trigger_prune_predictions():
    """Operator-triggered #764 backfill: drop stored tagger predictions below
    the current ml_settings.tagger_store_floor and clamp allowlist thresholds
    up to it. Shrinks image_record's TOAST (~100 GB of sub-0.70 scores).
    Idempotent + self-resuming; runs on the maintenance_long lane."""
    from ..tasks.admin import prune_low_confidence_predictions_task
    async_result = prune_low_confidence_predictions_task.delay()
    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
@admin_bp.route("/maintenance/backfill-predictions", methods=["POST"])
 async def trigger_backfill_predictions():
    """Operator-triggered #768 backfill: copy stored tagger predictions from the
    image_record.tagger_predictions JSON into the normalized image_prediction
    table. Batched + resumable + idempotent; runs on the maintenance_long lane.
    Run this once after deploying migration 0045 (which creates the empty table)
    to populate predictions for the existing library."""
    from ..tasks.admin import backfill_image_predictions_task
    async_result = backfill_image_predictions_task.delay()
    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
@@ -60,8 +60,10 @@ class ImageRecord(Base):
        ForeignKey("artist.id", ondelete="SET NULL"), nullable=True, index=True
    )
-    # ML fields (populated by FC-2's ml-worker)
+    # ML fields (populated by FC-2's ml-worker). Per-tag predictions live in the
-    tagger_predictions: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+    # normalized image_prediction table (#768) — the tagger_predictions JSON
    # column was dropped in migration 0046. tagger_model_version stays as the
    # "has this been tagged / is it current?" signal the backfill sweep reads.
    tagger_model_version: Mapped[str | None] = mapped_column(String(128), nullable=True)
    # 1152 = SigLIP-so400m embedding dim. Swapping models in FC-2 may require
    # a column-width migration.
@@ -1,6 +1,6 @@
 """TagAlias — maps a model's (name, category) prediction to the operator's
-canonical tag. Resolved at suggestion-read time so raw predictions stay
+canonical tag. Resolved at suggestion-read time so the raw predictions stored
-unmolested in image_record.tagger_predictions.
+in image_prediction stay unmolested.
 """
 from datetime import datetime
@@ -574,7 +574,7 @@ def reset_content_tagging(session: Session, *, dry_run: bool = False) -> dict:
    can re-tag from scratch via the Camie auto-suggest.
    PRESERVED: fandom + series tags and their series_page ordering, plus every
-    image's image_record.tagger_predictions (untouched) so suggestions
+    image's image_prediction rows (untouched) so suggestions
    repopulate immediately. CASCADE on image_tag / tag_alias / tag_allowlist /
    tag_reference_embedding / tag_suggestion_rejection clears each deleted
    tag's applications + metadata. Tag.fandom_id is SET NULL, so deleting
@@ -1085,7 +1085,6 @@ class Importer:
        existing.height = height
        existing.thumbnail_path = None
        existing.integrity_status = "unknown"
        existing.tagger_predictions = None
        existing.tagger_model_version = None
        existing.siglip_embedding = None
        existing.siglip_model_version = None
@@ -1,7 +1,7 @@
 """Alias resolution + CRUD.
 A tag_alias maps (model_name, model_category) -> canonical Tag. Resolution
-happens at suggestion-read time so raw tagger_predictions stay unmolested.
+happens at suggestion-read time so the raw image_prediction rows stay unmolested.
 """
 from collections.abc import Sequence
@@ -207,212 +207,3 @@ def rescan_series_suggestions_task(self, after_post_id: int = 0) -> dict:
        )
        rescan_series_suggestions_task.delay(summary["resume_after_id"])
    return summary
@celery.task(
    name="backend.app.tasks.admin.prune_low_confidence_predictions_task",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError),
    retry_backoff=15, retry_backoff_max=180, max_retries=1,
    soft_time_limit=3600, time_limit=4200,  # 60 min / 70 min
 )
 def prune_low_confidence_predictions_task(self, after_id: int = 0) -> dict:
    """One-time #764 backfill: drop tagger_predictions entries below the DB
    store floor (ml_settings.tagger_store_floor) from existing image_record
    rows, and clamp any allowlist min_confidence below the floor up to it.
    The Camie tagger emits ~10k tags; the old 0.05 floor stored the entire
    near-zero tail, bloating image_record's TOAST to ~100 GB. This rewrites
    each row to the new floor. Keyset by id ASC (restart-safe via after_id);
    idempotent — already-pruned rows rewrite to themselves and are skipped.
    Rewriting rows generates bloat, so run VACUUM FULL / pg_repack on
    image_record afterward to return the disk to the OS.
    The keep predicate (confidence >= floor) mirrors Tagger.infer's store
    gate so backfilled rows match what new imports store. Self-resumes on the
    soft time limit (re-enqueues from the last committed id)."""
    from celery.exceptions import SoftTimeLimitExceeded
    from sqlalchemy import select, update
    from ..models import ImageRecord, MLSettings, TagAllowlist
    SessionLocal = _sync_session_factory()
    scanned = 0
    pruned = 0
    clamped = 0
    last_id = after_id
    try:
        with SessionLocal() as session:
            floor = session.execute(
                select(MLSettings.tagger_store_floor).where(MLSettings.id == 1)
            ).scalar_one()
            # Clamp allowlist thresholds below the new floor once, on the
            # first pass (#764 consumer #4) — a sub-floor min_confidence can't
            # apply more permissively now that nothing below it is stored.
            if after_id == 0:
                clamped = session.execute(
                    update(TagAllowlist)
                    .where(TagAllowlist.min_confidence < floor)
                    .values(min_confidence=floor)
                ).rowcount or 0
                session.commit()
            while True:
                rows = session.execute(
                    select(ImageRecord.id, ImageRecord.tagger_predictions)
                    .where(ImageRecord.id > last_id)
                    .where(ImageRecord.tagger_predictions.is_not(None))
                    .order_by(ImageRecord.id.asc())
                    .limit(500)
                ).all()
                if not rows:
                    break
                for image_id, preds in rows:
                    scanned += 1
                    if not preds:
                        continue
                    kept = {
                        name: p for name, p in preds.items()
                        if float(p.get("confidence", 0.0)) >= floor
                    }
                    if len(kept) != len(preds):
                        session.execute(
                            update(ImageRecord)
                            .where(ImageRecord.id == image_id)
                            .values(tagger_predictions=kept)
                        )
                        pruned += 1
                session.commit()
                last_id = rows[-1].id  # advance only after commit, for resume
    except SoftTimeLimitExceeded:
        log.warning(
            "prune_low_confidence_predictions soft-limited at id=%s "
            "(scanned=%d pruned=%d) — re-enqueuing", last_id, scanned, pruned,
        )
        prune_low_confidence_predictions_task.delay(last_id)
        return {
            "partial": True, "last_id": last_id,
            "scanned": scanned, "pruned": pruned,
        }
    log.info(
        "prune_low_confidence_predictions complete: floor=%s scanned=%d "
        "pruned=%d allowlist_clamped=%d", floor, scanned, pruned, clamped,
    )
    return {
        "floor": floor, "scanned": scanned, "pruned": pruned,
        "allowlist_clamped": clamped, "last_id": last_id,
    }
 # Backfill image_prediction from image_record.tagger_predictions (#768).
 # Deliberately NOT done in migration 0045: a single INSERT…SELECT over the
 # ~100 GB TOAST is one transaction — invisible until commit, unmonitorable, and
 # the MATERIALIZED-CTE form spilled the whole 100 GB to temp on NFS. Instead we
 # walk image_record in id WINDOWS, running a bounded INSERT…SELECT over each
 # window and committing per chunk: progress is visible (image_prediction grows
 # live), it's resumable (re-enqueues from the last committed id), and json_each
 # stays in the DB executor streaming each window (no Python-side 100 GB load, no
 # materialization). Idempotent via ON CONFLICT DO NOTHING.
 _BACKFILL_PRED_CHUNK_SECONDS = 600  # re-enqueue boundary, like normalize_tags
 _BACKFILL_PRED_ID_WINDOW = 2000     # image_record ids per committed batch
@celery.task(
    name="backend.app.tasks.admin.backfill_image_predictions_task",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError),
    retry_backoff=15, retry_backoff_max=180, max_retries=1,
    soft_time_limit=1800, time_limit=2400,  # 30 min / 40 min
 )
 def backfill_image_predictions_task(self, after_id: int = 0) -> dict:
    """One-time #768 backfill: copy each image_record's stored tagger
    predictions (the >= store-floor entries) from the tagger_predictions JSON
    into the normalized image_prediction table.
    Batched by id window + committed per chunk so it's monitorable and
    resumable; idempotent (ON CONFLICT DO NOTHING) so re-running is safe.
    Filters to >= ml_settings.tagger_store_floor (default 0.70) so the table
    stays small even from the full pre-prune JSON tail. Guards json_each against
    non-object rows (scalar/null tagger_predictions → "cannot deconstruct a
    scalar") via an inline CASE. Self-resumes on the soft time limit."""
    import time
    from celery.exceptions import SoftTimeLimitExceeded
    from sqlalchemy import func, select, text
    from ..models import ImageRecord, MLSettings
    _INSERT_WINDOW = text(
        """
        INSERT INTO image_prediction (image_record_id, raw_name, category, score)
        SELECT ir.id,
               je.key,
               COALESCE(je.value ->> 'category', 'general'),
               (je.value ->> 'confidence')::double precision
        FROM image_record ir,
             json_each(
                 CASE WHEN json_typeof(ir.tagger_predictions) = 'object'
                      THEN ir.tagger_predictions
                      ELSE '{}'::json END
             ) je
        WHERE ir.id > :lo AND ir.id <= :hi
          AND je.value ->> 'confidence' IS NOT NULL
          AND (je.value ->> 'confidence')::double precision >= :floor
        ON CONFLICT (image_record_id, raw_name) DO NOTHING
        """
    )
    SessionLocal = _sync_session_factory()
    started = time.monotonic()
    last_id = after_id
    inserted = 0
    windows = 0
    with SessionLocal() as session:
        floor = session.execute(
            select(MLSettings.tagger_store_floor).where(MLSettings.id == 1)
        ).scalar_one()
        max_id = session.execute(
            select(func.max(ImageRecord.id))
        ).scalar() or 0
        try:
            while last_id < max_id:
                hi = last_id + _BACKFILL_PRED_ID_WINDOW
                res = session.execute(
                    _INSERT_WINDOW, {"lo": last_id, "hi": hi, "floor": floor}
                )
                session.commit()
                inserted += res.rowcount or 0
                windows += 1
                last_id = hi  # advance only after commit, for resume
                if time.monotonic() - started > _BACKFILL_PRED_CHUNK_SECONDS:
                    log.info(
                        "backfill_image_predictions chunk done (windows=%d "
                        "inserted=%d up to id=%d/%d) — re-enqueuing",
                        windows, inserted, min(last_id, max_id), max_id,
                    )
                    backfill_image_predictions_task.delay(last_id)
                    return {
                        "partial": True, "last_id": last_id, "max_id": max_id,
                        "inserted": inserted, "windows": windows,
                    }
        except SoftTimeLimitExceeded:
            log.warning(
                "backfill_image_predictions soft-limited at id=%d "
                "(inserted=%d) — re-enqueuing", last_id, inserted,
            )
            backfill_image_predictions_task.delay(last_id)
            return {
                "partial": True, "last_id": last_id, "max_id": max_id,
                "inserted": inserted, "windows": windows,
            }
    log.info(
        "backfill_image_predictions complete: floor=%s inserted=%d windows=%d "
        "max_id=%d", floor, inserted, windows, max_id,
    )
    return {
        "floor": floor, "inserted": inserted, "windows": windows,
        "max_id": max_id, "last_id": max_id,
    }
@@ -157,15 +157,15 @@ def tag_and_embed(self, image_id: int) -> dict:
                )
            phase = "persist"
            record.tagger_predictions = preds
            record.tagger_model_version = settings.tagger_model_version
            record.siglip_embedding = embedding.tolist()
            record.siglip_model_version = settings.embedder_model_version
            session.add(record)
-            # Write the normalized image_prediction rows (#768). Delete-then-
+            # Write the normalized image_prediction rows (#768) — the sole home
-            # insert keeps a re-tag idempotent. tagger_store_floor was already
+            # for predictions now (image_record.tagger_predictions was dropped in
-            # applied in tagger.infer, so preds is the >=floor set. (Transitional
+            # migration 0046). Delete-then-insert keeps a re-tag idempotent;
-            # dual-write alongside the JSON column until the read cutover lands.)
+            # tagger_store_floor was already applied in tagger.infer, so preds is
            # the >=floor set.
            session.execute(
                delete(ImagePrediction).where(
                    ImagePrediction.image_record_id == image_id
@@ -282,7 +282,7 @@ def backfill(self) -> int:
                select(ImageRecord.id)
                .where(ImageRecord.id > last_id)
                .where(
-                    (ImageRecord.tagger_predictions.is_(None))
+                    (ImageRecord.tagger_model_version.is_(None))
                    | (
                        ImageRecord.tagger_model_version
                        != settings.tagger_model_version
@@ -1,50 +0,0 @@
 <template>
  <!-- #768: one-time copy of stored tagger predictions from the
       image_record.tagger_predictions JSON into the normalized
       image_prediction table. Migration 0045 creates the empty table; this
       populates it for the existing library. -->
  <v-card>
    <v-card-title>Backfill normalized predictions</v-card-title>
    <v-card-text>
      <p class="text-body-2 mb-3">
        Copies each image's stored tagger predictions into the new
        <code>image_prediction</code> table (the source the suggestions and
        allowlist now read from). Run this <strong>once</strong> after the
        upgrade so existing images get their suggestions back — newly tagged
        images populate it automatically. Batched, resumable and idempotent;
        safe to run more than once and to leave running in the background.
      </p>
      <v-btn color="primary" rounded="pill" :loading="busy" @click="run">
        <v-icon start>mdi-database-import-outline</v-icon> Backfill predictions now
      </v-btn>
      <span v-if="queued" class="ml-3 text-caption text-success">Queued ✓</span>
      <QueueStatusBar queue="maintenance_long" queue-label="Maintenance (long)" />
    </v-card-text>
  </v-card>
 </template>
 <script setup>
 import { ref } from 'vue'
 import { useApi } from '../../composables/useApi.js'
 import { toast } from '../../utils/toast.js'
 import QueueStatusBar from './QueueStatusBar.vue'
 const api = useApi()
 const busy = ref(false)
 const queued = ref(false)
 async function run () {
  busy.value = true
  queued.value = false
  try {
    await api.post('/api/admin/maintenance/backfill-predictions')
    queued.value = true
    toast({ text: 'Prediction backfill queued', type: 'success' })
  } catch (e) {
    toast({ text: e?.body?.detail || e?.message || 'Failed to queue', type: 'error' })
  } finally {
    busy.value = false
  }
 }
 </script>
@@ -12,8 +12,6 @@
      <ThumbnailBackfillCard />
    </div>
    <MLThresholdSliders class="mt-4" />
    <BackfillPredictionsCard class="mt-4" />
    <PrunePredictionsCard class="mt-4" />
    <AllowlistTable class="mt-4" />
    <AliasTable class="mt-4" />
    <DbMaintenanceCard class="mt-6" />
@@ -33,8 +31,6 @@ import MLBackfillCard from './MLBackfillCard.vue'
 import CentroidRecomputeCard from './CentroidRecomputeCard.vue'
 import ThumbnailBackfillCard from './ThumbnailBackfillCard.vue'
 import MLThresholdSliders from './MLThresholdSliders.vue'
 import BackfillPredictionsCard from './BackfillPredictionsCard.vue'
 import PrunePredictionsCard from './PrunePredictionsCard.vue'
 import AllowlistTable from './AllowlistTable.vue'
 import AliasTable from './AliasTable.vue'
 import DbMaintenanceCard from './DbMaintenanceCard.vue'
@@ -1,58 +0,0 @@
 <template>
  <!-- #764: drop stored tagger predictions below the store floor to shrink
       image_record's TOAST (the sub-0.70 score tail had grown it to ~100 GB). -->
  <v-card>
    <v-card-title>Prune low-confidence predictions</v-card-title>
    <v-card-text>
      <p class="text-body-2 mb-3">
        Removes stored tagger predictions below the current store floor
        (<strong>{{ floorPct }}</strong>) from every image, and clamps any
        allowlist threshold below the floor up to it. This is what shrinks the
        database — the low-confidence tail was the bulk of its size. Idempotent
        and resumable; safe to run more than once. Afterward, reclaim the freed
        space with <code>VACUUM FULL</code> / <code>pg_repack</code> on
        <code>image_record</code>.
      </p>
      <v-btn color="primary" rounded="pill" :loading="busy" @click="run">
        <v-icon start>mdi-database-minus-outline</v-icon> Prune predictions now
      </v-btn>
      <span v-if="queued" class="ml-3 text-caption text-success">Queued ✓</span>
      <QueueStatusBar queue="maintenance_long" queue-label="Maintenance (long)" />
    </v-card-text>
  </v-card>
 </template>
 <script setup>
 import { computed, onMounted, ref } from 'vue'
 import { useApi } from '../../composables/useApi.js'
 import { toast } from '../../utils/toast.js'
 import { useMLStore } from '../../stores/ml.js'
 import QueueStatusBar from './QueueStatusBar.vue'
 const api = useApi()
 const ml = useMLStore()
 const busy = ref(false)
 const queued = ref(false)
 const floorPct = computed(() => {
  const f = ml.settings?.tagger_store_floor
  return f == null ? '—' : `${Math.round(f * 100)}%`
 })
 onMounted(() => { if (!ml.settings) ml.loadSettings() })
 async function run () {
  busy.value = true
  queued.value = false
  try {
    await api.post('/api/admin/maintenance/prune-predictions')
    queued.value = true
    toast({ text: 'Prediction prune queued', type: 'success' })
  } catch (e) {
    toast({ text: e?.body?.detail || e?.message || 'Failed to queue', type: 'error' })
  } finally {
    busy.value = false
  }
 }
 </script>
@@ -24,8 +24,10 @@ def test_new_tables_registered():
 def test_image_record_columns_renamed():
    cols = {c.name for c in ImageRecord.__table__.columns}
-    assert "tagger_predictions" in cols
+    # tagger_predictions (the renamed wd14_predictions) was later dropped in
    # migration 0046 — predictions live in image_prediction now (#768).
    assert "tagger_model_version" in cols
    assert "tagger_predictions" not in cols
    assert "wd14_predictions" not in cols
    assert "wd14_model_version" not in cols
@@ -11,6 +11,7 @@ from PIL import Image
 from sqlalchemy import func, select
 from backend.app.models import (
    ImagePrediction,
    ImageProvenance,
    ImageRecord,
    ImportSettings,
@@ -118,7 +119,11 @@ def test_smaller_existing_is_superseded(importer, import_layout):
            image_record_id=eid, tag_id=tag.id, source="manual"
        )
    )
-    old.tagger_predictions = {"x": 1}
+    importer.session.add(
        ImagePrediction(
            image_record_id=eid, raw_name="x", category="general", score=0.9
        )
    )
    old.siglip_embedding = [0.0] * 1152
    old.integrity_status = "ok"
    importer.session.commit()
@@ -136,7 +141,11 @@ def test_smaller_existing_is_superseded(importer, import_layout):
    assert row.path != old_path
    assert row.phash is not None
    assert row.integrity_status == "unknown"
-    assert row.tagger_predictions is None
+    # #768: re-import clears the normalized predictions too
    assert importer.session.execute(
        select(func.count()).select_from(ImagePrediction)
        .where(ImagePrediction.image_record_id == eid)
    ).scalar_one() == 0
    assert row.siglip_embedding is None
    linked = importer.session.execute(
        select(image_tag.c.tag_id).where(
@@ -324,9 +324,7 @@ async def test_protective_alias_uses_tag_kind(db):
    # The protective alias category is the tag's KIND — the tagger maps each name
    # to exactly one category and a tag's kind is set from it, so kind already IS
    # the tagger's category. The merge no longer scans image_record's predictions
-    # to rediscover it. Even with a (contrived) differing prediction category
+    # to rediscover it — it writes a single (name, kind) alias from the tag kind.
    # present, the merge writes a single (name, kind) alias.
    from backend.app.models import ImageRecord
    from backend.app.models.tag_alias import TagAlias
    svc = TagService(db)
@@ -335,10 +333,6 @@ async def test_protective_alias_uses_tag_kind(db):
    img = await _img(db)
    # mark source machine-known so keep_as_alias is True
    await svc.add_to_image(img, a.id, source="ml_auto")
    r1 = await db.get(ImageRecord, img)
    r1.tagger_predictions = {
        "predname": {"category": "copyright", "confidence": 0.8}
    }
    await db.flush()
    result = await svc.merge(a.id, b.id)
    assert result.alias_created is True
@@ -381,7 +375,6 @@ async def test_alias_fallback_to_kind_when_no_predictions(db):
@pytest.mark.asyncio
 async def test_alias_create_does_not_clobber_existing(db):
    from backend.app.models import ImageRecord
    from backend.app.models.tag_alias import TagAlias
    svc = TagService(db)
@@ -397,10 +390,6 @@ async def test_alias_create_does_not_clobber_existing(db):
    )
    img = await _img(db)
    await svc.add_to_image(img, a.id, source="ml_auto")
    r = await db.get(ImageRecord, img)
    r.tagger_predictions = {
        "dupalias": {"category": "general", "confidence": 0.9}
    }
    await db.flush()
    await svc.merge(a.id, b.id)
    cid = await db.scalar(
@@ -42,75 +42,6 @@ def test_bulk_delete_images_task_registered():
    )
 def test_prune_low_confidence_predictions_task_registered():
    assert (
        "backend.app.tasks.admin.prune_low_confidence_predictions_task"
        in celery.tasks
    )
 def test_backfill_image_predictions_task_registered():
    assert (
        "backend.app.tasks.admin.backfill_image_predictions_task"
        in celery.tasks
    )
@pytest.mark.asyncio
 async def test_prune_low_confidence_predictions(db_sync, tmp_path):
    # #764: drop stored tagger predictions below the store floor (default
    # 0.70) and clamp allowlist thresholds up to it.
    from backend.app.models import Tag, TagAllowlist, TagKind
    from backend.app.tasks.admin import prune_low_confidence_predictions_task
    f0 = tmp_path / "p0.jpg"
    f0.write_bytes(b"x")
    img0 = ImageRecord(
        path=str(f0), sha256=f"{0:064x}", size_bytes=10, mime="image/jpeg",
        origin="imported_filesystem",
        tagger_predictions={
            "keep_high": {"category": "general", "confidence": 0.92},
            "keep_edge": {"category": "general", "confidence": 0.70},
            "drop_mid": {"category": "general", "confidence": 0.40},
            "drop_tiny": {"category": "general", "confidence": 0.06},
        },
    )
    db_sync.add(img0)
    f1 = tmp_path / "p1.jpg"
    f1.write_bytes(b"x")
    img1 = ImageRecord(
        path=str(f1), sha256=f"{1:064x}", size_bytes=10, mime="image/jpeg",
        origin="imported_filesystem",
        tagger_predictions={"only": {"category": "general", "confidence": 0.99}},
    )
    db_sync.add(img1)
    tag = Tag(name="lowthr-tag", kind=TagKind.general)
    db_sync.add(tag)
    db_sync.flush()
    db_sync.add(TagAllowlist(tag_id=tag.id, min_confidence=0.30))
    db_sync.commit()
    img0_id, img1_id, tag_id = img0.id, img1.id, tag.id
    result = prune_low_confidence_predictions_task.delay().get()
    assert result["floor"] == pytest.approx(0.70)
    assert result["pruned"] == 1            # only img0 had sub-floor entries
    assert result["allowlist_clamped"] == 1
    db_sync.expire_all()
    p0 = db_sync.execute(
        select(ImageRecord.tagger_predictions).where(ImageRecord.id == img0_id)
    ).scalar_one()
    assert set(p0) == {"keep_high", "keep_edge"}   # >=0.70 kept, <0.70 dropped
    p1 = db_sync.execute(
        select(ImageRecord.tagger_predictions).where(ImageRecord.id == img1_id)
    ).scalar_one()
    assert set(p1) == {"only"}                      # already clean — untouched
    clamped = db_sync.execute(
        select(TagAllowlist.min_confidence).where(TagAllowlist.tag_id == tag_id)
    ).scalar_one()
    assert clamped == pytest.approx(0.70)
 # --- delete_artist_cascade_task -------------------------------------
@@ -44,7 +44,7 @@ async def test_backfill_enqueues_missing(db, monkeypatch):
        path="/images/n.jpg", sha256="n" * 64, size_bytes=1,
        mime="image/jpeg", width=1, height=1,
        origin="imported_filesystem", integrity_status="unknown",
-        tagger_predictions=None, siglip_embedding=None,
+        siglip_embedding=None,
    )
    db.add(img)
    await db.commit()