feat(explore): cluster-consensus tag-gaps service + route (#94a)

Cluster C, milestone #94. BulkTagService.tag_gaps(image_ids, threshold) finds
tags applied to >= threshold fraction of a visual neighbour set but not all of
it (the '7 of 10 share Miku; these 3 don't' signal). Each gap carries the
laggard image ids minus any TagSuggestionRejection rows, so apply-to-cluster
never re-proposes a tag a neighbour dismissed. 100%-common tags and <2-image
sets are excluded. New POST /api/images/cluster/tag-gaps.

Tests: consensus found / common excluded / missing ids; rejected laggard
excluded from missing; tag dropped when all laggards rejected; <2 images empty;
route shape + bad input.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XCUHUGQLrBrkgyk1t49kpX

backend/app/api/tags.py

@@ -320,6 +320,27 @@ async def common_tags():
     return jsonify({"tags": tags})
 
 
+@tags_bp.route("/images/cluster/tag-gaps", methods=["POST"])
+async def cluster_tag_gaps():
+    """Cluster-consensus tag gaps for a visual neighbour set (#94 Explore):
+    tags on >= threshold of the images but not all. Each gap carries the
+    laggard image ids (minus rejections) for apply-to-cluster."""
+    body = await request.get_json()
+    ids, err = _parse_bulk_ids(body)
+    if err:
+        return err
+    try:
+        threshold = float(body.get("threshold", 0.6))
+    except (TypeError, ValueError):
+        threshold = 0.6
+    threshold = min(1.0, max(0.0, threshold))
+    async with get_session() as session:
+        gaps = await BulkTagService(session).tag_gaps(ids, threshold)
+    return jsonify(
+        {"gaps": gaps, "total": len(set(ids)), "threshold": threshold}
+    )
+
+
 @tags_bp.route("/images/bulk/tags", methods=["POST"])
 async def bulk_add_tag():
     body = await request.get_json()

backend/app/services/bulk_tag_service.py

@@ -1,5 +1,7 @@
 """Bulk tag operations over a set of images (Core, set-based, atomic)."""
 
+import math
+
 from sqlalchemy import and_, func, select
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -13,6 +15,85 @@ class BulkTagService:
     def __init__(self, session: AsyncSession):
         self.session = session
 
+    async def tag_gaps(
+        self, image_ids: list[int], threshold: float
+    ) -> list[dict]:
+        """Cluster-consensus tag gaps (#94 Explore): tags applied to at least
+        `threshold` fraction of the image set but NOT all of it — the
+        "7 of these 10 share Hatsune Miku; these 3 don't" signal.
+
+        For each gap, `missing_image_ids` are the set members that LACK the tag
+        and have NOT rejected it (TagSuggestionRejection) — so apply-to-cluster
+        never re-proposes a tag a neighbor explicitly dismissed. A tag on every
+        image is "common", not a gap; a tag on fewer than 2 isn't consensus.
+        """
+        ids = list({int(i) for i in image_ids})
+        if len(ids) < 2:
+            return []
+        n = len(ids)
+        min_present = max(2, math.ceil(threshold * n))
+        if min_present >= n:
+            return []  # threshold so high only a 100%-common tag could qualify
+
+        cnt = func.count(func.distinct(image_tag.c.image_record_id))
+        gap_rows = (
+            await self.session.execute(
+                select(Tag.id, Tag.name, Tag.kind)
+                .join(image_tag, image_tag.c.tag_id == Tag.id)
+                .where(image_tag.c.image_record_id.in_(ids))
+                .group_by(Tag.id, Tag.name, Tag.kind)
+                .having(and_(cnt >= min_present, cnt < n))
+                .order_by(cnt.desc(), Tag.name.asc())
+            )
+        ).all()
+        if not gap_rows:
+            return []
+        gap_ids = [r.id for r in gap_rows]
+
+        present: dict[int, set[int]] = {}
+        for tid, iid in (
+            await self.session.execute(
+                select(image_tag.c.tag_id, image_tag.c.image_record_id).where(
+                    image_tag.c.tag_id.in_(gap_ids),
+                    image_tag.c.image_record_id.in_(ids),
+                )
+            )
+        ).all():
+            present.setdefault(tid, set()).add(iid)
+
+        rejected: dict[int, set[int]] = {}
+        for tid, iid in (
+            await self.session.execute(
+                select(
+                    TagSuggestionRejection.tag_id,
+                    TagSuggestionRejection.image_record_id,
+                ).where(
+                    TagSuggestionRejection.tag_id.in_(gap_ids),
+                    TagSuggestionRejection.image_record_id.in_(ids),
+                )
+            )
+        ).all():
+            rejected.setdefault(tid, set()).add(iid)
+
+        id_set = set(ids)
+        gaps = []
+        for r in gap_rows:
+            have = present.get(r.id, set())
+            missing = id_set - have - rejected.get(r.id, set())
+            if not missing:
+                continue  # every laggard rejected it — nothing to apply
+            gaps.append(
+                {
+                    "tag_id": r.id,
+                    "name": r.name,
+                    "kind": r.kind.value if hasattr(r.kind, "value") else str(r.kind),
+                    "present_count": len(have),
+                    "total": n,
+                    "missing_image_ids": sorted(missing),
+                }
+            )
+        return gaps
+
     async def common_tags(self, image_ids: list[int]) -> list[dict]:
         """Tags present on EVERY image in image_ids."""
         if not image_ids: