diff --git a/alembic/versions/0045_image_prediction_table.py b/alembic/versions/0045_image_prediction_table.py
new file mode 100644
index 0000000..189fb1a
--- /dev/null
+++ b/alembic/versions/0045_image_prediction_table.py
@@ -0,0 +1,82 @@
+"""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.
+    # Filter to >= the store floor (ml_settings.tagger_store_floor, default
+    # 0.70) right here so this is self-sufficient — it does NOT depend on the
+    # #764 prune having run, and extracting only the >=floor tail keeps
+    # image_prediction small (~tens of rows/image) even from the full JSON.
+    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
+          AND (je.value ->> 'confidence')::double precision
+              >= COALESCE(
+                  (SELECT tagger_store_floor FROM ml_settings WHERE id = 1),
+                  0.70
+              )
+        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")
diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py
index 00047f9..19d444b 100644
--- a/backend/app/models/__init__.py
+++ b/backend/app/models/__init__.py
@@ -7,6 +7,7 @@ from .backup_run import BackupRun
 from .base import Base
 from .credential import Credential
 from .download_event import DownloadEvent
+from .image_prediction import ImagePrediction
 from .image_provenance import ImageProvenance
 from .image_record import ImageRecord
 from .import_batch import ImportBatch
@@ -45,6 +46,7 @@ __all__ = [
     "SeriesPage",
     "SeriesSuggestion",
     "ImageRecord",
+    "ImagePrediction",
     "ImageProvenance",
     "Tag",
     "TagKind",
diff --git a/backend/app/models/image_prediction.py b/backend/app/models/image_prediction.py
new file mode 100644
index 0000000..f532243
--- /dev/null
+++ b/backend/app/models/image_prediction.py
@@ -0,0 +1,37 @@
+"""ImagePrediction — one row per (image, tagger vocab prediction).
+
+Replaces the image_record.tagger_predictions JSON blob (#768). Storing the
+raw Camie/booru vocab name (not a tag_id) preserves the suggestion read
+path's semantics: raw_name → canonical Tag resolution happens at read time
+via the alias map, and accepting a prediction can CREATE the Tag. The store
+floor (ml_settings.tagger_store_floor) is applied at WRITE time, so only
+predictions >= the floor land here.
+"""
+
+from sqlalchemy import Float, ForeignKey, Index, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from .base import Base
+
+
+class ImagePrediction(Base):
+    __tablename__ = "image_prediction"
+    __table_args__ = (
+        UniqueConstraint(
+            "image_record_id", "raw_name", name="image_raw_name",
+        ),
+        # Per-image read (suggestion build) and the "images with tag X above
+        # Y" query the JSON blob never allowed.
+        Index("ix_image_prediction_image", "image_record_id"),
+        Index("ix_image_prediction_name_score", "raw_name", "score"),
+    )
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    image_record_id: Mapped[int] = mapped_column(
+        ForeignKey("image_record.id", ondelete="CASCADE"), nullable=False,
+    )
+    # The raw tagger vocab key (booru form) — NOT a tag_id. Resolved to a
+    # canonical Tag at read time, exactly as the old JSON keys were.
+    raw_name: Mapped[str] = mapped_column(String(255), nullable=False)
+    category: Mapped[str] = mapped_column(String(64), nullable=False)
+    score: Mapped[float] = mapped_column(Float, nullable=False)
diff --git a/backend/app/services/importer.py b/backend/app/services/importer.py
index a79da39..a1cd9b5 100644
--- a/backend/app/services/importer.py
+++ b/backend/app/services/importer.py
@@ -1090,6 +1090,16 @@ class Importer:
         existing.siglip_embedding = None
         existing.siglip_model_version = None
         existing.centroid_scores = None
+        # #768: predictions also live in the normalized image_prediction table
+        # now — clear them so a re-imported file re-derives a fresh set.
+        from sqlalchemy import delete as _delete
+
+        from ..models import ImagePrediction as _ImagePrediction
+        self.session.execute(
+            _delete(_ImagePrediction).where(
+                _ImagePrediction.image_record_id == existing.id
+            )
+        )
         # created_at intentionally preserved; updated_at auto-bumps.
         self.session.flush()
         self.session.commit()
diff --git a/backend/app/services/ml/suggestions.py b/backend/app/services/ml/suggestions.py
index b53e198..ca60c30 100644
--- a/backend/app/services/ml/suggestions.py
+++ b/backend/app/services/ml/suggestions.py
@@ -8,6 +8,7 @@ from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ...models import (
+    ImagePrediction,
     ImageRecord,
     MLSettings,
     Tag,
@@ -48,6 +49,25 @@ class SuggestionService:
             await self.session.execute(select(MLSettings).where(MLSettings.id == 1))
         ).scalar_one()
 
+    async def _load_predictions(self, image_id: int) -> dict:
+        """Predictions for one image from the normalized image_prediction
+        table (#768), in the {raw_name: {category, confidence}} shape the rest
+        of this service consumed from the old JSON column — so all downstream
+        threshold/alias/merge logic is unchanged."""
+        rows = (
+            await self.session.execute(
+                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 _threshold_for(
         self, s: MLSettings, category: str, override: float | None = None,
     ) -> float:
@@ -80,7 +100,7 @@ class SuggestionService:
             return SuggestionList()
 
         settings = await self._settings()
-        predictions: dict = img.tagger_predictions or {}
+        predictions: dict = await self._load_predictions(image_id)
 
         applied = set(
             (
diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py
index ff09258..3d62213 100644
--- a/backend/app/tasks/ml.py
+++ b/backend/app/tasks/ml.py
@@ -10,11 +10,11 @@ import logging
 from pathlib import Path
 
 from celery.exceptions import SoftTimeLimitExceeded
-from sqlalchemy import select
+from sqlalchemy import delete, select
 from sqlalchemy.exc import DBAPIError, OperationalError
 
 from ..celery_app import celery
-from ..models import ImageRecord, MLSettings
+from ..models import ImagePrediction, ImageRecord, MLSettings
 from ._sync_engine import sync_session_factory as _sync_session_factory
 
 log = logging.getLogger(__name__)
@@ -162,6 +162,23 @@ def tag_and_embed(self, image_id: int) -> dict:
             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-
+            # insert keeps a re-tag idempotent. tagger_store_floor was already
+            # applied in tagger.infer, so preds is the >=floor set. (Transitional
+            # dual-write alongside the JSON column until the read cutover lands.)
+            session.execute(
+                delete(ImagePrediction).where(
+                    ImagePrediction.image_record_id == image_id
+                )
+            )
+            session.add_all([
+                ImagePrediction(
+                    image_record_id=image_id, raw_name=name,
+                    category=p.get("category", "general"),
+                    score=float(p.get("confidence", 0.0)),
+                )
+                for name, p in preds.items()
+            ])
             session.commit()
     except SoftTimeLimitExceeded:
         log.error(
@@ -331,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(
@@ -377,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
diff --git a/tests/_prediction_helpers.py b/tests/_prediction_helpers.py
new file mode 100644
index 0000000..e689ccc
--- /dev/null
+++ b/tests/_prediction_helpers.py
@@ -0,0 +1,21 @@
+"""#768 test helper: seed image_prediction rows.
+
+Read-path tests used to seed ImageRecord(tagger_predictions={...}); predictions
+now live in the normalized image_prediction table, so seed there instead.
+"""
+from backend.app.models import ImagePrediction
+
+
+async def seed_predictions(session, image_id: int, predictions: dict) -> None:
+    """Insert image_prediction rows from a {raw_name: {category, confidence}}
+    dict (the old JSON shape). Caller commits/flushes as needed; this flushes."""
+    session.add_all([
+        ImagePrediction(
+            image_record_id=image_id,
+            raw_name=name,
+            category=p.get("category", "general"),
+            score=float(p.get("confidence", 0.0)),
+        )
+        for name, p in predictions.items()
+    ])
+    await session.flush()
diff --git a/tests/test_api_suggestions.py b/tests/test_api_suggestions.py
index f8bd49c..b9fd041 100644
--- a/tests/test_api_suggestions.py
+++ b/tests/test_api_suggestions.py
@@ -15,14 +15,17 @@ def eager():
 
 
 async def _img(db, preds):
+    from tests._prediction_helpers import seed_predictions
+
     img = ImageRecord(
         path="/images/s.jpg", sha256="s" * 64, size_bytes=1,
         mime="image/jpeg", width=1, height=1,
         origin="imported_filesystem", integrity_status="unknown",
-        tagger_predictions=preds,
     )
     db.add(img)
     await db.commit()
+    await seed_predictions(db, img.id, preds)
+    await db.commit()
     return img
 
 
diff --git a/tests/test_image_prediction.py b/tests/test_image_prediction.py
new file mode 100644
index 0000000..43e05e9
--- /dev/null
+++ b/tests/test_image_prediction.py
@@ -0,0 +1,57 @@
+"""#768: image_prediction table — model + constraints round-trip."""
+import pytest
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+
+from backend.app.models import ImagePrediction, ImageRecord
+
+pytestmark = pytest.mark.integration
+
+
+async def _make_image(db, path="/img/p0.jpg", sha="0"):
+    rec = ImageRecord(
+        path=path, sha256=sha.ljust(64, "0")[:64], size_bytes=10,
+        mime="image/jpeg", origin="imported_filesystem",
+    )
+    db.add(rec)
+    await db.flush()
+    return rec
+
+
+@pytest.mark.asyncio
+async def test_image_prediction_round_trip(db):
+    rec = await _make_image(db)
+    db.add_all([
+        ImagePrediction(
+            image_record_id=rec.id, raw_name="blue_eyes",
+            category="general", score=0.92,
+        ),
+        ImagePrediction(
+            image_record_id=rec.id, raw_name="hatsune_miku",
+            category="character", score=0.88,
+        ),
+    ])
+    await db.commit()
+
+    rows = (await db.execute(
+        select(ImagePrediction.raw_name, ImagePrediction.score)
+        .where(ImagePrediction.image_record_id == rec.id)
+        .order_by(ImagePrediction.score.desc())
+    )).all()
+    assert [r.raw_name for r in rows] == ["blue_eyes", "hatsune_miku"]
+
+
+@pytest.mark.asyncio
+async def test_image_prediction_unique_per_image_name(db):
+    rec = await _make_image(db, path="/img/p1.jpg", sha="1")
+    db.add(ImagePrediction(
+        image_record_id=rec.id, raw_name="dup",
+        category="general", score=0.9,
+    ))
+    await db.commit()
+    db.add(ImagePrediction(
+        image_record_id=rec.id, raw_name="dup",
+        category="general", score=0.7,
+    ))
+    with pytest.raises(IntegrityError):
+        await db.commit()
diff --git a/tests/test_ml_suggestions.py b/tests/test_ml_suggestions.py
index 9486ce8..891b54f 100644
--- a/tests/test_ml_suggestions.py
+++ b/tests/test_ml_suggestions.py
@@ -9,7 +9,7 @@ from backend.app.services.tag_service import TagService
 pytestmark = pytest.mark.integration
 
 
-def _img(sha: str, predictions: dict) -> ImageRecord:
+def _img(sha: str) -> ImageRecord:
     return ImageRecord(
         path=f"/images/{sha}.jpg",
         sha256=sha,
@@ -19,24 +19,34 @@ def _img(sha: str, predictions: dict) -> ImageRecord:
         height=1,
         origin="imported_filesystem",
         integrity_status="unknown",
-        tagger_predictions=predictions,
     )
 
 
+async def _seed_img(db, sha: str, predictions: dict) -> ImageRecord:
+    """#768: create an image + seed its predictions into image_prediction
+    (the read path's source), returning the flushed record."""
+    from tests._prediction_helpers import seed_predictions
+
+    img = _img(sha)
+    db.add(img)
+    await db.flush()
+    await seed_predictions(db, img.id, predictions)
+    return img
+
+
 @pytest.mark.asyncio
 async def test_threshold_filters_low_confidence_general(db):
     # Default general threshold is 0.50 (alembic 0029 lowered it from
     # 0.95). Use 0.30/0.60 to keep the test asserting threshold behavior
     # rather than the exact cutoff number.
-    img = _img(
+    img = await _seed_img(
+        db,
         "a" * 64,
         {
             "lowconf": {"category": "general", "confidence": 0.30},
             "sword": {"category": "general", "confidence": 0.97},
         },
     )
-    db.add(img)
-    await db.flush()
     sl = await SuggestionService(db).for_image(img.id)
     names = [s.display_name for s in sl.by_category.get("general", [])]
     # display_name is normalized (tag_name.normalize) before surfacing.
@@ -49,36 +59,34 @@ async def test_threshold_override_surfaces_low_confidence(db):
     # The typed-dropdown "show everything the model saw" mode: threshold_override
     # surfaces stored predictions below the configured threshold (in canonical
     # formatting) so they can be picked instead of hand-typed (2026-06-09).
-    img = _img(
+    img = await _seed_img(
+        db,
         "d" * 64,
         {
             "lowconf": {"category": "general", "confidence": 0.30},
             "sword": {"category": "general", "confidence": 0.97},
         },
     )
-    db.add(img)
-    await db.flush()
     sl = await SuggestionService(db).for_image(img.id, threshold_override=0.0)
     names = [s.display_name for s in sl.by_category.get("general", [])]
     assert "Sword" in names
     assert "Lowconf" in names  # below the configured threshold, surfaced anyway
 
     # Unsurfaced categories are still excluded even with the override.
-    img2 = _img("e" * 64, {"safe": {"category": "rating", "confidence": 0.99}})
-    db.add(img2)
-    await db.flush()
+    img2 = await _seed_img(
+        db, "e" * 64, {"safe": {"category": "rating", "confidence": 0.99}}
+    )
     sl2 = await SuggestionService(db).for_image(img2.id, threshold_override=0.0)
     assert "rating" not in sl2.by_category
 
 
 @pytest.mark.asyncio
 async def test_unsurfaced_category_dropped(db):
-    img = _img(
+    img = await _seed_img(
+        db,
         "b" * 64,
         {"safe": {"category": "rating", "confidence": 0.99}},
     )
-    db.add(img)
-    await db.flush()
     sl = await SuggestionService(db).for_image(img.id)
     assert "rating" not in sl.by_category
 
@@ -88,12 +96,11 @@ async def test_alias_resolution(db):
     tags = TagService(db)
     canonical = await tags.find_or_create("Sasuke Uchiha", TagKind.character)
     await AliasService(db).create("uchiha_sasuke", "character", canonical.id)
-    img = _img(
+    img = await _seed_img(
+        db,
         "c" * 64,
         {"uchiha_sasuke": {"category": "character", "confidence": 0.96}},
     )
-    db.add(img)
-    await db.flush()
     sl = await SuggestionService(db).for_image(img.id)
     chars = sl.by_category["character"]
     assert len(chars) == 1
@@ -104,12 +111,11 @@ async def test_alias_resolution(db):
 
 @pytest.mark.asyncio
 async def test_raw_tag_creates_new(db):
-    img = _img(
+    img = await _seed_img(
+        db,
         "d" * 64,
         {"brand_new_tag": {"category": "character", "confidence": 0.96}},
     )
-    db.add(img)
-    await db.flush()
     sl = await SuggestionService(db).for_image(img.id)
     chars = sl.by_category["character"]
     # display_name is the normalized Camie name (underscores -> spaces,
@@ -123,12 +129,11 @@ async def test_raw_tag_creates_new(db):
 async def test_applied_tag_not_suggested(db):
     tags = TagService(db)
     tag = await tags.find_or_create("alreadyhere", TagKind.character)
-    img = _img(
+    img = await _seed_img(
+        db,
         "e" * 64,
         {"alreadyhere": {"category": "character", "confidence": 0.96}},
     )
-    db.add(img)
-    await db.flush()
     await db.execute(
         image_tag.insert().values(
             image_record_id=img.id, tag_id=tag.id, source="manual"
diff --git a/tests/test_suggestions_bulk.py b/tests/test_suggestions_bulk.py
index 1bfe40d..d0d101a 100644
--- a/tests/test_suggestions_bulk.py
+++ b/tests/test_suggestions_bulk.py
@@ -5,16 +5,16 @@ from backend.app.models import ImageRecord, TagKind
 from backend.app.models.tag import image_tag
 from backend.app.services.ml.suggestions import SuggestionService
 from backend.app.services.tag_service import TagService
+from tests._prediction_helpers import seed_predictions
 
 pytestmark = pytest.mark.integration
 
 
-def _img(sha: str, predictions: dict) -> ImageRecord:
+def _img(sha: str) -> ImageRecord:
     return ImageRecord(
         path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1,
         mime="image/jpeg", width=1, height=1,
         origin="imported_filesystem", integrity_status="unknown",
-        tagger_predictions=predictions,
     )
 
 
@@ -22,10 +22,12 @@ def _img(sha: str, predictions: dict) -> ImageRecord:
 async def test_consensus_includes_tag_over_threshold(db):
     tags = TagService(db)
     t = await tags.find_or_create("sword", TagKind.general)
-    a = _img("a" * 64, {"sword": {"category": "general", "confidence": 0.97}})
-    b = _img("b" * 64, {"sword": {"category": "general", "confidence": 0.95}})
+    a = _img("a" * 64)
+    b = _img("b" * 64)
     db.add_all([a, b])
     await db.flush()
+    await seed_predictions(db, a.id, {"sword": {"category": "general", "confidence": 0.97}})
+    await seed_predictions(db, b.id, {"sword": {"category": "general", "confidence": 0.95}})
     res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8)
     gen = res["general"]
     assert any(s["canonical_tag_id"] == t.id for s in gen)
@@ -38,10 +40,11 @@ async def test_consensus_includes_tag_over_threshold(db):
 async def test_consensus_counts_already_applied_for_coverage(db):
     tags = TagService(db)
     t = await tags.find_or_create("sky", TagKind.general)
-    a = _img("c" * 64, {"sky": {"category": "general", "confidence": 0.96}})
-    b = _img("d" * 64, {})  # no prediction
+    a = _img("c" * 64)
+    b = _img("d" * 64)  # no prediction
     db.add_all([a, b])
     await db.flush()
+    await seed_predictions(db, a.id, {"sky": {"category": "general", "confidence": 0.96}})
     # b already has the tag applied -> counts toward coverage, not confidence
     await db.execute(
         image_tag.insert().values(
@@ -58,10 +61,11 @@ async def test_consensus_counts_already_applied_for_coverage(db):
 async def test_consensus_excludes_below_threshold(db):
     tags = TagService(db)
     await tags.find_or_create("rare", TagKind.general)
-    a = _img("e" * 64, {"rare": {"category": "general", "confidence": 0.96}})
-    b = _img("f" * 64, {})
+    a = _img("e" * 64)
+    b = _img("f" * 64)
     db.add_all([a, b])
     await db.flush()
+    await seed_predictions(db, a.id, {"rare": {"category": "general", "confidence": 0.96}})
     res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8)
     assert all(
         s["name"] != "rare" for s in res.get("general", [])
@@ -70,10 +74,12 @@ async def test_consensus_excludes_below_threshold(db):
 
 @pytest.mark.asyncio
 async def test_consensus_skips_creates_new_tag(db):
-    a = _img("g" * 64, {"neverseen": {"category": "general", "confidence": 0.99}})
-    b = _img("h" * 64, {"neverseen": {"category": "general", "confidence": 0.99}})
+    a = _img("g" * 64)
+    b = _img("h" * 64)
     db.add_all([a, b])
     await db.flush()
+    await seed_predictions(db, a.id, {"neverseen": {"category": "general", "confidence": 0.99}})
+    await seed_predictions(db, b.id, {"neverseen": {"category": "general", "confidence": 0.99}})
     res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8)
     # 'neverseen' has no Tag row -> creates_new_tag -> excluded from consensus
     assert all(s["name"] != "neverseen" for s in res.get("general", []))
@@ -90,9 +96,11 @@ async def test_bulk_suggestions_route(db):
 
     tags = TagService(db)
     await tags.find_or_create("sword", TagKind.general)
-    a = _img("i" * 64, {"sword": {"category": "general", "confidence": 0.97}})
+    a = _img("i" * 64)
     db.add(a)
     await db.commit()
+    await seed_predictions(db, a.id, {"sword": {"category": "general", "confidence": 0.97}})
+    await db.commit()
     app = create_app()
     async with app.test_client() as c:
         resp = await c.post(
diff --git a/tests/test_tasks_ml.py b/tests/test_tasks_ml.py
index e759a2e..eb59e47 100644
--- a/tests/test_tasks_ml.py
+++ b/tests/test_tasks_ml.py
@@ -63,6 +63,7 @@ async def test_apply_allowlist_applies_above_threshold(db):
     from backend.app.models.tag import image_tag
     from backend.app.services.tag_service import TagService
     from backend.app.tasks import ml as ml_tasks
+    from tests._prediction_helpers import seed_predictions
 
     tag = await TagService(db).find_or_create("autohero", TagKind.character)
     db.add(TagAllowlist(tag_id=tag.id, min_confidence=0.95))
@@ -70,12 +71,13 @@ async def test_apply_allowlist_applies_above_threshold(db):
         path="/images/al.jpg", sha256="al" + "0" * 62, size_bytes=1,
         mime="image/jpeg", width=1, height=1,
         origin="imported_filesystem", integrity_status="unknown",
-        tagger_predictions={
-            "autohero": {"category": "character", "confidence": 0.97}
-        },
     )
     db.add(img)
     await db.commit()
+    await seed_predictions(
+        db, img.id, {"autohero": {"category": "character", "confidence": 0.97}}
+    )
+    await db.commit()
 
     n = ml_tasks.apply_allowlist_tags(tag_id=tag.id)
     assert n >= 1
@@ -98,6 +100,7 @@ async def test_apply_allowlist_skips_below_threshold(db):
     from backend.app.models.tag import image_tag
     from backend.app.services.tag_service import TagService
     from backend.app.tasks import ml as ml_tasks
+    from tests._prediction_helpers import seed_predictions
 
     tag = await TagService(db).find_or_create("lowconf", TagKind.character)
     db.add(TagAllowlist(tag_id=tag.id, min_confidence=0.95))
@@ -105,12 +108,13 @@ async def test_apply_allowlist_skips_below_threshold(db):
         path="/images/lc.jpg", sha256="lc" + "0" * 62, size_bytes=1,
         mime="image/jpeg", width=1, height=1,
         origin="imported_filesystem", integrity_status="unknown",
-        tagger_predictions={
-            "lowconf": {"category": "character", "confidence": 0.40}
-        },
     )
     db.add(img)
     await db.commit()
+    await seed_predictions(
+        db, img.id, {"lowconf": {"category": "character", "confidence": 0.40}}
+    )
+    await db.commit()
     ml_tasks.apply_allowlist_tags(tag_id=tag.id)
     applied = (
         await db.execute(