diff --git a/backend/app/api/suggestions.py b/backend/app/api/suggestions.py
index 09d70e2..25bd2fe 100644
--- a/backend/app/api/suggestions.py
+++ b/backend/app/api/suggestions.py
@@ -85,3 +85,35 @@ async def dismiss_suggestion(image_id: int):
         await AllowlistService(session).dismiss(image_id, body["tag_id"])
         await session.commit()
     return "", 204
+
+
+@suggestions_bp.route("/suggestions/bulk", methods=["POST"])
+async def bulk_suggestions():
+    body = await request.get_json()
+    if not body or "image_ids" not in body:
+        return jsonify({"error": "image_ids required"}), 400
+    raw = body["image_ids"]
+    if not isinstance(raw, list) or not raw:
+        return jsonify({"error": "image_ids must be a non-empty list"}), 400
+    try:
+        ids = [int(x) for x in raw]
+    except (TypeError, ValueError):
+        return jsonify({"error": "image_ids must be integers"}), 400
+    if len(ids) > 200:
+        return jsonify({"error": "selection too large (max 200)"}), 400
+    try:
+        threshold = float(body.get("threshold", 0.8))
+    except (TypeError, ValueError):
+        threshold = 0.8
+    threshold = min(1.0, max(0.0, threshold))
+    async with get_session() as session:
+        suggestions = await SuggestionService(session).for_selection(
+            ids, threshold=threshold
+        )
+    return jsonify(
+        {
+            "suggestions": suggestions,
+            "evaluated": len(ids),
+            "threshold": threshold,
+        }
+    )
diff --git a/backend/app/services/ml/suggestions.py b/backend/app/services/ml/suggestions.py
index b725d9c..2c9ccae 100644
--- a/backend/app/services/ml/suggestions.py
+++ b/backend/app/services/ml/suggestions.py
@@ -197,3 +197,77 @@ class SuggestionService:
         for cat in result.by_category:
             result.by_category[cat].sort(key=lambda s: s.score, reverse=True)
         return result
+
+    async def for_selection(
+        self,
+        image_ids: list[int],
+        threshold: float = 0.8,
+        top_k: int = 10,
+    ) -> dict[str, list[dict]]:
+        """Consensus suggestions across image_ids. A tag is included iff it
+        was suggested for (or already applied to) >= threshold fraction of
+        the selection AND was acceptable on >= 1 image. Confidence is the
+        mean over images where it was suggested. Aggregated by
+        canonical_tag_id; creates-new (no canonical id) suggestions are
+        skipped (bulk Accept applies by tag id)."""
+        if not image_ids:
+            return {}
+        threshold = min(1.0, max(0.0, threshold))
+        total = len(image_ids)
+
+        stats: dict[int, dict] = {}
+        for image_id in image_ids:
+            sl = await self.for_image(image_id)
+            for category, items in sl.by_category.items():
+                for s in items:
+                    if s.canonical_tag_id is None or s.creates_new_tag:
+                        continue
+                    st = stats.get(s.canonical_tag_id)
+                    if st is None:
+                        st = {
+                            "tag_id": s.canonical_tag_id,
+                            "name": s.display_name,
+                            "category": category,
+                            "source": s.source,
+                            "suggested_count": 0,
+                            "sum_score": 0.0,
+                        }
+                        stats[s.canonical_tag_id] = st
+                    st["suggested_count"] += 1
+                    st["sum_score"] += s.score
+
+        rows = (
+            await self.session.execute(
+                select(
+                    image_tag.c.image_record_id, image_tag.c.tag_id
+                ).where(image_tag.c.image_record_id.in_(image_ids))
+            )
+        ).all()
+        applied_by_tag: dict[int, set[int]] = {}
+        for iid, tid in rows:
+            applied_by_tag.setdefault(tid, set()).add(iid)
+
+        result: dict[str, list[dict]] = {}
+        for st in stats.values():
+            existing_count = len(applied_by_tag.get(st["tag_id"], set()))
+            covered = st["suggested_count"] + existing_count
+            coverage = covered / total
+            if coverage < threshold or st["suggested_count"] < 1:
+                continue
+            result.setdefault(st["category"], []).append(
+                {
+                    "canonical_tag_id": st["tag_id"],
+                    "name": st["name"],
+                    "category": st["category"],
+                    "confidence": round(
+                        st["sum_score"] / st["suggested_count"], 4
+                    ),
+                    "coverage": round(coverage, 4),
+                    "covered_count": covered,
+                    "source": st["source"],
+                }
+            )
+        for cat in result:
+            result[cat].sort(key=lambda x: x["confidence"], reverse=True)
+            result[cat] = result[cat][:top_k]
+        return result
diff --git a/tests/test_suggestions_bulk.py b/tests/test_suggestions_bulk.py
new file mode 100644
index 0000000..2b00649
--- /dev/null
+++ b/tests/test_suggestions_bulk.py
@@ -0,0 +1,115 @@
+import pytest
+
+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
+
+pytestmark = pytest.mark.integration
+
+
+def _img(sha: str, predictions: dict) -> 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,
+    )
+
+
+@pytest.mark.asyncio
+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}})
+    db.add_all([a, b])
+    await db.flush()
+    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)
+    s = next(s for s in gen if s["canonical_tag_id"] == t.id)
+    assert s["coverage"] == 1.0
+    assert 0.95 <= s["confidence"] <= 0.97
+
+
+@pytest.mark.asyncio
+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
+    db.add_all([a, b])
+    await db.flush()
+    # b already has the tag applied -> counts toward coverage, not confidence
+    await db.execute(
+        image_tag.insert().values(
+            image_record_id=b.id, tag_id=t.id, source="manual"
+        )
+    )
+    res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8)
+    s = next(s for s in res["general"] if s["canonical_tag_id"] == t.id)
+    assert s["coverage"] == 1.0          # 1 suggested + 1 applied / 2
+    assert s["confidence"] == pytest.approx(0.96, abs=1e-4)
+
+
+@pytest.mark.asyncio
+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, {})
+    db.add_all([a, b])
+    await db.flush()
+    res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8)
+    assert all(
+        s["name"] != "rare" for s in res.get("general", [])
+    )  # coverage 0.5 < 0.8
+
+
+@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}})
+    db.add_all([a, b])
+    await db.flush()
+    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", []))
+
+
+@pytest.mark.asyncio
+async def test_consensus_threshold_clamped_and_empty_for_no_ids(db):
+    res = await SuggestionService(db).for_selection([], threshold=5.0)
+    assert res == {}
+
+
+@pytest.mark.asyncio
+async def test_bulk_suggestions_route(db):
+    from backend.app import create_app
+
+    tags = TagService(db)
+    await tags.find_or_create("sword", TagKind.general)
+    a = _img("i" * 64, {"sword": {"category": "general", "confidence": 0.97}})
+    db.add(a)
+    await db.commit()
+    app = create_app()
+    async with app.test_client() as c:
+        resp = await c.post(
+            "/api/suggestions/bulk", json={"image_ids": [a.id]}
+        )
+        assert resp.status_code == 200
+        body = await resp.get_json()
+        assert body["evaluated"] == 1
+        assert body["threshold"] == 0.8
+        assert "suggestions" in body
+
+
+@pytest.mark.asyncio
+async def test_bulk_suggestions_requires_ids(db):
+    from backend.app import create_app
+
+    app = create_app()
+    async with app.test_client() as c:
+        resp = await c.post("/api/suggestions/bulk", json={})
+        assert resp.status_code == 400