fix(migration): make 0045 DDL-only; backfill image_prediction via batched task (#768)

The inline INSERT…SELECT backfill in migration 0045 wrapped the table
creation and a ~100 GB pass over image_record.tagger_predictions in one
transaction: nothing committed until the end, it was unmonitorable, and an
earlier MATERIALIZED-CTE form spilled the full 100 GB to temp on NFS. A
deploy got stuck on it for ~2h with image_prediction never appearing.

Split the concerns:
- 0045 now creates ONLY the table + indexes (instant DDL → web boots).
- New backend.app.tasks.admin.backfill_image_predictions_task copies the
  >= store-floor predictions from the JSON into image_prediction, batched by
  id window and committed per chunk: live progress, resumable (re-enqueues
  from the last committed id), idempotent (ON CONFLICT DO NOTHING). json_each
  stays in the DB executor streaming each window — no Python-side 100 GB load,
  no materialization.
- POST /api/admin/maintenance/backfill-predictions + a Maintenance-tab card
  to trigger the one-time run after upgrading.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/app/tasks/admin.py

@@ -303,3 +303,116 @@ def prune_low_confidence_predictions_task(self, after_id: int = 0) -> dict:
         "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,
+    }