feat(ml): read suggestions + allowlist from image_prediction (#768 step 2)

Switch every prediction READER off the JSON column onto the normalized
image_prediction table. Parity by construction: each reader loads the same
{raw_name: {category, confidence}} dict it consumed before (via small
_load_predictions helpers), so all downstream threshold/alias/merge/consensus
logic is byte-identical — only the data source changed.

- suggestions.SuggestionService.for_image (and for_selection via it)
- ml.apply_allowlist_tags (iterates images that have prediction rows)
- importer re-import reset deletes the image's prediction rows
The tagger_predictions JSON column is still dual-written (step 1) so it stays
valid during transition; the backfill task's NULL check still works. Removing
the JSON write + DROP column + retiring the #764 prune is the cleanup
follow-up (needs a quiesced-worker window for the DROP lock).

Tests: shared tests/_prediction_helpers.seed_predictions seeds the table;
read-path tests (suggestions, bulk consensus, allowlist apply, API) seed there
instead of ImageRecord.tagger_predictions.

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

backend/app/tasks/ml.py

@@ -348,14 +348,16 @@ def apply_allowlist_tags(self, tag_id: int | None = None,
         if not allow:
             return 0
 
-        img_query = sa_select(
-            ImageRecord.id, ImageRecord.tagger_predictions
-        ).where(ImageRecord.tagger_predictions.is_not(None))
+        # Images that have any predictions (#768: from image_prediction, not
+        # the old JSON column), optionally narrowed to one image.
+        img_ids_query = sa_select(ImagePrediction.image_record_id).distinct()
         if image_id is not None:
-            img_query = img_query.where(ImageRecord.id == image_id)
+            img_ids_query = img_ids_query.where(
+                ImagePrediction.image_record_id == image_id
+            )
 
-        for img_id, preds in session.execute(img_query).all():
-            preds = preds or {}
+        for (img_id,) in session.execute(img_ids_query).all():
+            preds = _load_predictions_sync(session, img_id)
             for a_tag_id, min_conf in allow.items():
                 exists = session.execute(
                     sa_select(image_tag.c.tag_id).where(
@@ -394,6 +396,25 @@ def apply_allowlist_tags(self, tag_id: int | None = None,
     return applied
 
 
+def _load_predictions_sync(session, image_id: int) -> dict:
+    """Predictions for one image from image_prediction (#768), in the
+    {raw_name: {category, confidence}} shape _confidence_for_tag consumes —
+    keeps the allowlist resolution logic unchanged."""
+    from sqlalchemy import select as sa_select
+
+    rows = session.execute(
+        sa_select(
+            ImagePrediction.raw_name,
+            ImagePrediction.category,
+            ImagePrediction.score,
+        ).where(ImagePrediction.image_record_id == image_id)
+    ).all()
+    return {
+        r.raw_name: {"category": r.category, "confidence": r.score}
+        for r in rows
+    }
+
+
 def _confidence_for_tag(session, tag, preds: dict) -> float | None:
     """Highest confidence among predictions that resolve to `tag` —
     either the prediction name equals the tag name, or an alias maps