feat(allowlist): coverage projection + applied-count + post-accept projection (#7a/#7b)

Cluster B, milestone #99. Backend for the allowlist tuning dashboard.

#7a: AllowlistService.coverage(tag_id, threshold) counts distinct images with
a prediction resolving to the tag (raw_name==tag.name OR (raw_name,category) in
the tag's aliases) scoring >= threshold — the gross candidate pool, mirroring
tasks.ml._confidence_for_tag resolution. list_all now carries applied_count
(grouped image_tag count) + coverage_count (at the row's threshold). New
GET /api/tags/<id>/allowlist/coverage?threshold= for the live what-if number.

#7b: /suggestions/accept + /alias return {allowlisted, tag_id, tag_name,
projected_count} (projection at the tag's threshold) instead of 204, so the UI
can show a non-blocking 'auto-applying to ~N images' toast. Apply still runs
async via apply_allowlist_tags — projected_count is an estimate.

Tests: coverage by threshold (direct + alias-with-category), list applied vs
coverage, coverage route (explicit/default/bad threshold), accept/alias payload
(newly-allowlisted vs already-on-list).

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

backend/app/services/ml/allowlist.py

@@ -5,11 +5,18 @@ image_tag AND to tag_allowlist; per-image removal/dismiss writes a rejection.
 from collections.abc import Sequence
 from dataclasses import dataclass
 
-from sqlalchemy import delete, select
+from sqlalchemy import and_, delete, distinct, func, or_, select
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from ...models import MLSettings, Tag, TagAllowlist, TagSuggestionRejection
+from ...models import (
+    ImagePrediction,
+    MLSettings,
+    Tag,
+    TagAlias,
+    TagAllowlist,
+    TagSuggestionRejection,
+)
 from ...models.tag import image_tag
 from .aliases import AliasService
 
@@ -20,6 +27,8 @@ class AllowlistRow:
     tag_name: str
     tag_kind: str
     min_confidence: float
+    applied_count: int      # image_tag rows currently carrying this tag
+    coverage_count: int     # images a sweep WOULD cover at min_confidence
 
 
 class AllowlistService:
@@ -116,6 +125,44 @@ class AllowlistService:
             delete(TagAllowlist).where(TagAllowlist.tag_id == tag_id)
         )
 
+    async def _coverage_match(self, tag: Tag):
+        """The predicate over image_prediction rows that resolve to `tag`,
+        mirroring tasks.ml._confidence_for_tag's resolution: a prediction whose
+        raw_name equals the tag name (any category), OR an alias maps
+        (raw_name, category) -> this tag. Returns a SQLAlchemy boolean clause.
+        """
+        alias_rows = (
+            await self.session.execute(
+                select(TagAlias.alias_string, TagAlias.alias_category).where(
+                    TagAlias.canonical_tag_id == tag.id
+                )
+            )
+        ).all()
+        name_clause = ImagePrediction.raw_name == tag.name
+        alias_clauses = [
+            and_(
+                ImagePrediction.raw_name == a,
+                ImagePrediction.category == c,
+            )
+            for a, c in alias_rows
+        ]
+        return or_(name_clause, *alias_clauses) if alias_clauses else name_clause
+
+    async def coverage(self, tag_id: int, threshold: float) -> int:
+        """How many distinct images a sweep WOULD cover for this tag at
+        `threshold`: images with a resolving prediction scoring >= threshold.
+        The gross candidate pool (NOT minus already-applied/rejected) — it's
+        the tuning signal for "lower the threshold and ~N more images qualify".
+        """
+        tag = await self.session.get(Tag, tag_id)
+        if tag is None:
+            return 0
+        match = await self._coverage_match(tag)
+        stmt = select(
+            func.count(distinct(ImagePrediction.image_record_id))
+        ).where(ImagePrediction.score >= threshold, match)
+        return (await self.session.execute(stmt)).scalar_one()
+
     async def list_all(self) -> Sequence[AllowlistRow]:
         stmt = (
             select(
@@ -128,12 +175,33 @@ class AllowlistService:
             .order_by(Tag.name.asc())
         )
         rows = (await self.session.execute(stmt)).all()
-        return [
-            AllowlistRow(
-                tag_id=r[0],
-                tag_name=r[1],
-                tag_kind=r[2].value if hasattr(r[2], "value") else str(r[2]),
-                min_confidence=r[3],
+        tag_ids = [r[0] for r in rows]
+
+        # Applied counts in ONE grouped query (vs N per-row counts).
+        applied: dict[int, int] = {}
+        if tag_ids:
+            applied = dict(
+                (
+                    await self.session.execute(
+                        select(image_tag.c.tag_id, func.count())
+                        .where(image_tag.c.tag_id.in_(tag_ids))
+                        .group_by(image_tag.c.tag_id)
+                    )
+                ).all()
             )
-            for r in rows
-        ]
+
+        result = []
+        for r in rows:
+            # Coverage is per-tag (alias set differs); allowlist is small.
+            cov = await self.coverage(r[0], r[3])
+            result.append(
+                AllowlistRow(
+                    tag_id=r[0],
+                    tag_name=r[1],
+                    tag_kind=r[2].value if hasattr(r[2], "value") else str(r[2]),
+                    min_confidence=r[3],
+                    applied_count=applied.get(r[0], 0),
+                    coverage_count=cov,
+                )
+            )
+        return result