feat(ml): image_prediction table + backfill + dual-write (#768 step 1)

Normalize tagger predictions out of the image_record.tagger_predictions JSON
blob into a queryable per-prediction table. Step 1 of the cutover (expand):
additive + low-risk — reads still use the JSON, this just adds the table and
keeps it populated.

- ImagePrediction(image_record_id, raw_name, category, score) — stores the
  RAW tagger vocab name (not tag_id) so read-time alias→canonical resolution
  is unchanged. Indexed for per-image reads + by (raw_name, score).
- Migration 0045: create table + set-based backfill from the JSON via
  json_each (fast post-#764-prune). The old column stays (vestigial) and is
  dropped in a later follow-up — DROP needs an ACCESS EXCLUSIVE lock on the
  hot image_record table, so it waits for a quiesced-worker window.
- tag_and_embed dual-writes the rows (delete-then-insert, idempotent);
  tagger_store_floor already applied in infer().

Next: switch suggestion + allowlist reads to the table, then drop the JSON
write. Plan-task #768.

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

alembic/versions/0045_image_prediction_table.py

@@ -0,0 +1,73 @@
+"""image_prediction table + backfill from image_record.tagger_predictions
+
+Normalizes the per-image tagger predictions out of the JSON blob into a
+queryable table (#768). Backfills from the existing JSON in one set-based
+INSERT…SELECT over json_each — fast because the #764 prune already shrank
+each row to its >=0.70 entries. The old image_record.tagger_predictions
+column is left in place here (vestigial) and dropped in a follow-up once the
+code cutover is verified — dropping it needs an ACCESS EXCLUSIVE lock on the
+hot image_record table (the 0044 lock class), so it's deferred to a
+quiesced-worker window.
+
+Revision ID: 0045
+Revises: 0044
+Create Date: 2026-06-10
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0045"
+down_revision: Union[str, None] = "0044"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "image_prediction",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "image_record_id", sa.Integer(),
+            sa.ForeignKey("image_record.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("raw_name", sa.String(length=255), nullable=False),
+        sa.Column("category", sa.String(length=64), nullable=False),
+        sa.Column("score", sa.Float(), nullable=False),
+        sa.UniqueConstraint(
+            "image_record_id", "raw_name", name="image_raw_name",
+        ),
+    )
+    op.create_index(
+        "ix_image_prediction_image", "image_prediction", ["image_record_id"],
+    )
+    op.create_index(
+        "ix_image_prediction_name_score", "image_prediction",
+        ["raw_name", "score"],
+    )
+    # Backfill from the JSON blob. json_each expands {name: {category,
+    # confidence}} into one row per prediction. category defaults to 'general'
+    # to mirror the suggestion read path; rows with no confidence are skipped.
+    op.execute(
+        """
+        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(ir.tagger_predictions) je
+        WHERE ir.tagger_predictions IS NOT NULL
+          AND je.value ->> 'confidence' IS NOT NULL
+        ON CONFLICT (image_record_id, raw_name) DO NOTHING
+        """
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_image_prediction_name_score", "image_prediction")
+    op.drop_index("ix_image_prediction_image", "image_prediction")
+    op.drop_table("image_prediction")