feat(fc2b): add allowlist-apply + centroid recompute tasks + beat

apply_allowlist_tags: 4 modes (tag-only / image-only / both / full sweep),
matches a tag to a prediction either by direct name or via alias
(name, category) resolution, gates on per-tag min_confidence, skips
applied/rejected, applies source='ml_auto'. recompute_centroid /
recompute_centroids: async-bridged calls into CentroidService, delta-gated.
Beat: daily backfill, daily centroid recompute, daily allowlist sweep.

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

backend/app/tasks/ml.py

@@ -189,9 +189,152 @@ def backfill(self) -> int:
     return enqueued
 
 
-# --- Defined fully in Task 9/10. Stub so tag_and_embed's .delay() resolves
-#     and the module imports cleanly between commits. Replaced in Task 9. ---
 @celery.task(name="backend.app.tasks.ml.apply_allowlist_tags", bind=True)
 def apply_allowlist_tags(self, tag_id: int | None = None,
                          image_id: int | None = None) -> int:
-    return 0  # replaced in Task 9
+    """Retroactively apply allowlisted tags.
+
+    Modes:
+    - tag_id only      : scan all images for this tag.
+    - image_id only    : scan all allowlisted tags for this image.
+    - both             : just the (image, tag) pair.
+    - neither          : full sweep (daily beat).
+
+    Skips: already-applied, rejected (tag_suggestion_rejection), or
+    confidence below the tag's allowlist min_confidence. Applied with
+    source='ml_auto'.
+    """
+    from sqlalchemy import and_
+    from sqlalchemy import select as sa_select
+    from sqlalchemy.dialects.postgresql import insert as pg_insert
+
+    from ..models import TagAllowlist, TagSuggestionRejection
+    from ..models.tag import image_tag
+
+    SessionLocal = _sync_session_factory()
+    applied = 0
+    with SessionLocal() as session:
+        allow_rows = session.execute(
+            sa_select(TagAllowlist.tag_id, TagAllowlist.min_confidence)
+            if tag_id is None
+            else sa_select(
+                TagAllowlist.tag_id, TagAllowlist.min_confidence
+            ).where(TagAllowlist.tag_id == tag_id)
+        ).all()
+        allow = {r[0]: r[1] for r in allow_rows}
+        if not allow:
+            return 0
+
+        img_query = sa_select(
+            ImageRecord.id, ImageRecord.tagger_predictions
+        ).where(ImageRecord.tagger_predictions.is_not(None))
+        if image_id is not None:
+            img_query = img_query.where(ImageRecord.id == image_id)
+
+        for img_id, preds in session.execute(img_query).all():
+            preds = preds or {}
+            for a_tag_id, min_conf in allow.items():
+                exists = session.execute(
+                    sa_select(image_tag.c.tag_id).where(
+                        and_(
+                            image_tag.c.image_record_id == img_id,
+                            image_tag.c.tag_id == a_tag_id,
+                        )
+                    )
+                ).scalar_one_or_none()
+                if exists is not None:
+                    continue
+                rej = session.get(
+                    TagSuggestionRejection, (img_id, a_tag_id)
+                )
+                if rej is not None:
+                    continue
+                from ..models import Tag
+
+                tag = session.get(Tag, a_tag_id)
+                if tag is None:
+                    continue
+                conf = _confidence_for_tag(session, tag, preds)
+                if conf is None or conf < min_conf:
+                    continue
+                stmt = pg_insert(image_tag).values(
+                    image_record_id=img_id,
+                    tag_id=a_tag_id,
+                    source="ml_auto",
+                )
+                stmt = stmt.on_conflict_do_nothing(
+                    index_elements=["image_record_id", "tag_id"]
+                )
+                session.execute(stmt)
+                applied += 1
+        session.commit()
+    return applied
+
+
+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
+    (prediction name, category) -> tag.id.
+    """
+    from sqlalchemy import select as sa_select
+
+    from ..models import TagAlias
+
+    best: float | None = None
+    direct = preds.get(tag.name)
+    if direct is not None:
+        best = float(direct.get("confidence", 0.0))
+    alias_rows = session.execute(
+        sa_select(TagAlias.alias_string, TagAlias.alias_category).where(
+            TagAlias.canonical_tag_id == tag.id
+        )
+    ).all()
+    for alias_string, alias_category in alias_rows:
+        p = preds.get(alias_string)
+        if p is None:
+            continue
+        if p.get("category") != alias_category:
+            continue
+        c = float(p.get("confidence", 0.0))
+        if best is None or c > best:
+            best = c
+    return best
+
+
+@celery.task(name="backend.app.tasks.ml.recompute_centroid", bind=True)
+def recompute_centroid(self, tag_id: int) -> bool:
+    import asyncio
+
+    from ..extensions import make_engine, make_session_factory
+    from ..services.ml.centroids import CentroidService
+
+    async def _run() -> bool:
+        engine = make_engine()
+        Session = make_session_factory(engine)
+        async with Session() as session:
+            svc = CentroidService(session)
+            result = await svc.recompute_for_tag(tag_id)
+            await session.commit()
+            return result
+
+    return asyncio.run(_run())
+
+
+@celery.task(name="backend.app.tasks.ml.recompute_centroids", bind=True)
+def recompute_centroids(self) -> int:
+    """Daily: find drifted centroids, enqueue recompute_centroid for each."""
+    import asyncio
+
+    from ..extensions import make_engine, make_session_factory
+    from ..services.ml.centroids import CentroidService
+
+    async def _list() -> list[int]:
+        engine = make_engine()
+        Session = make_session_factory(engine)
+        async with Session() as session:
+            return await CentroidService(session).list_drifted()
+
+    drifted = asyncio.run(_list())
+    for tid in drifted:
+        recompute_centroid.delay(tid)
+    return len(drifted)