refactor(ml): remove the dead per-tag centroid subsystem (#1189)

The v2 pivot replaced per-tag SigLIP centroids with learned heads + CCIP.
Centroids were still recomputed (on every tag merge + a daily beat) but NOTHING
read them — suggestions come from heads+CCIP and apply_allowlist_tags applies
via Camie predictions, not centroids. Pure dead wiring; remove it.

Removed: CentroidService, recompute_centroid/recompute_centroids tasks, the
daily beat, POST /api/ml/recompute-centroids, the recompute-on-merge trigger,
the tag_reference_embedding table + model, the centroid_similarity_threshold +
min_reference_images settings (migration 0066), the CentroidRecomputeCard +
its store action + MaintenancePanel tile, and the centroid slider in
MLThresholdSliders. _keep_as_alias drops its vestigial has-centroid branch (the
allowlist branch already covers "could re-emit"); tag merge no longer clears a
table that no longer exists.

NOT touched (still live, parallel to heads): the Camie tagger, ImagePrediction,
and the allowlist bulk-apply — accepting a suggestion still allowlists + applies
it across the library. The tag-eval "centroid" baseline metric is unrelated
(in-memory) and stays. (image_record.centroid_scores JSON column also remains —
separate legacy field, its own micro-cleanup.)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa

backend/app/api/ml_admin.py

@@ -1,4 +1,4 @@
-"""ML admin API: settings, backfill trigger, centroid recompute trigger."""
+"""ML admin API: settings + backfill trigger."""
 
 from quart import Blueprint, jsonify, request
 
@@ -11,8 +11,6 @@ ml_admin_bp = Blueprint("ml_admin", __name__, url_prefix="/api/ml")
 _EDITABLE = (
     "suggestion_threshold_character",
     "suggestion_threshold_general",
-    "centroid_similarity_threshold",
-    "min_reference_images",
     "tagger_store_floor",
     "video_frame_interval_seconds",
     "video_max_frames",
@@ -41,8 +39,6 @@ async def get_settings():
         {
             "suggestion_threshold_character": s.suggestion_threshold_character,
             "suggestion_threshold_general": s.suggestion_threshold_general,
-            "centroid_similarity_threshold": s.centroid_similarity_threshold,
-            "min_reference_images": s.min_reference_images,
             "tagger_store_floor": s.tagger_store_floor,
             "video_frame_interval_seconds": s.video_frame_interval_seconds,
             "video_max_frames": s.video_max_frames,
@@ -142,11 +138,3 @@ async def trigger_backfill():
 
     r = backfill.delay()
     return jsonify({"celery_task_id": r.id}), 202
-
-
-@ml_admin_bp.route("/recompute-centroids", methods=["POST"])
-async def trigger_recompute():
-    from ..tasks.ml import recompute_centroids
-
-    r = recompute_centroids.delay()
-    return jsonify({"celery_task_id": r.id}), 202

backend/app/api/tags.py

@@ -304,12 +304,6 @@ async def merge_tag(source_id: int):
         from ..tasks.ml import apply_allowlist_tags
 
         apply_allowlist_tags.delay(tag_id=result.target_id)
-    # Tag merge invalidates the target's centroid (the merged-in source
-    # tag's images now contribute to it). Daily list_drifted catches it
-    # within 24h, but eager recompute closes the suggestion-quality dip
-    # in the meantime. Audit 2026-06-02.
-    from ..tasks.ml import recompute_centroid
-    recompute_centroid.delay(result.target_id)
     return jsonify(
         {
             "target": {

backend/app/celery_app.py

@@ -101,10 +101,6 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.ml.backfill",
                 "schedule": 86400.0,
             },
-            "recompute-centroids-daily": {
-                "task": "backend.app.tasks.ml.recompute_centroids",
-                "schedule": 86400.0,
-            },
             "apply-allowlist-sweep-daily": {
                 "task": "backend.app.tasks.ml.apply_allowlist_tags",
                 "schedule": 86400.0,

backend/app/models/__init__.py

@@ -38,7 +38,6 @@ from .tag_allowlist import TagAllowlist
 from .tag_eval_run import TagEvalRun
 from .tag_head import TagHead
 from .tag_positive_confirmation import TagPositiveConfirmation
-from .tag_reference_embedding import TagReferenceEmbedding
 from .tag_suggestion_rejection import TagSuggestionRejection
 from .task_run import TaskRun
 
@@ -83,7 +82,6 @@ __all__ = [
     "TagEvalRun",
     "TagHead",
     "TagPositiveConfirmation",
-    "TagReferenceEmbedding",
     "TagSuggestionRejection",
     "TaskRun",
 ]

backend/app/models/ml_settings.py

@@ -33,21 +33,14 @@ class MLSettings(Base):
     suggestion_threshold_general: Mapped[float] = mapped_column(
         Float, nullable=False, default=0.70
     )
-    centroid_similarity_threshold: Mapped[float] = mapped_column(
-        Float, nullable=False, default=0.55
-    )
     # Ingest floor: tagger predictions below this confidence are not stored
-    # (tagger.Tagger.infer). Default 0.70 — the suggestion path already
-    # filters at 0.70 and the centroid/learned path covers low-confidence
-    # preferred tags, so the sub-0.70 tail is redundant weight (it had
-    # bloated image_record's TOAST to ~100 GB; plan-task #764). Operator-
-    # tunable via Settings → ML; must stay ≤ the suggestion thresholds.
+    # (tagger.Tagger.infer). Default 0.70 — the suggestion path already filters
+    # there, so the sub-0.70 tail is redundant weight (it had bloated
+    # image_record's TOAST to ~100 GB; plan-task #764). Operator-tunable via
+    # Settings → ML; must stay ≤ the suggestion thresholds.
     tagger_store_floor: Mapped[float] = mapped_column(
         Float, nullable=False, default=0.70
     )
-    min_reference_images: Mapped[int] = mapped_column(
-        Integer, nullable=False, default=5
-    )
     # Video tagging (#747). Sample one frame every N seconds (fixed CADENCE, not a
     # fixed count) so a tag's frame-presence reflects real screen time regardless
     # of video length; cap the total so a long video can't explode into hundreds

backend/app/models/tag_reference_embedding.py

@@ -1,23 +0,0 @@
-"""TagReferenceEmbedding — per-tag centroid (mean SigLIP embedding of members)."""
-
-from datetime import datetime
-
-from pgvector.sqlalchemy import Vector
-from sqlalchemy import DateTime, ForeignKey, Integer, String, func
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import Base
-
-
-class TagReferenceEmbedding(Base):
-    __tablename__ = "tag_reference_embedding"
-
-    tag_id: Mapped[int] = mapped_column(
-        ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True
-    )
-    embedding: Mapped[list[float]] = mapped_column(Vector(1152), nullable=False)
-    reference_count: Mapped[int] = mapped_column(Integer, nullable=False)
-    model_version: Mapped[str] = mapped_column(String(128), nullable=False)
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, server_default=func.now()
-    )

backend/app/services/ml/centroids.py

@@ -1,163 +0,0 @@
-"""Tag centroids: the mean SigLIP embedding of a tag's member images.
-
-Powers centroid-augmented suggestions (a tag whose centroid is close to an
-image's embedding becomes a suggestion even if Camie didn't predict it).
-"""
-
-from dataclasses import dataclass
-
-import numpy as np
-from sqlalchemy import func, select
-from sqlalchemy.dialects.postgresql import insert
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from ...models import (
-    ImageRecord,
-    MLSettings,
-    Tag,
-    TagKind,
-    TagReferenceEmbedding,
-)
-from ...models.tag import image_tag
-
-ELIGIBLE_KINDS = {
-    TagKind.character,
-    TagKind.fandom,
-    TagKind.general,
-    TagKind.series,
-}
-
-
-@dataclass(frozen=True)
-class CentroidHit:
-    tag_id: int
-    similarity: float
-
-
-class CentroidService:
-    def __init__(self, session: AsyncSession):
-        self.session = session
-
-    async def _min_reference_images(self) -> int:
-        return (
-            await self.session.execute(
-                select(MLSettings.min_reference_images).where(MLSettings.id == 1)
-            )
-        ).scalar_one()
-
-    async def _model_version(self) -> str:
-        """Audit 2026-06-02: SigLIP model-version stamp comes from the
-        DB row, not the env constant. tag_and_embed (tasks/ml.py:110)
-        already reads from MLSettings.embedder_model_version, so by
-        sourcing centroid stamps + drift checks from the same row, we
-        eliminate the silent-drift case the audit flagged. env
-        SIGLIP_MODEL_VERSION still drives which model embedder.py
-        loads at runtime; the version stamp is purely the operator-
-        controlled identifier."""
-        return (
-            await self.session.execute(
-                select(MLSettings.embedder_model_version).where(MLSettings.id == 1)
-            )
-        ).scalar_one()
-
-    async def recompute_for_tag(self, tag_id: int) -> bool:
-        """Recompute one tag's centroid. Returns True if a centroid was
-        written, False if skipped (ineligible kind or too few members)."""
-        tag = await self.session.get(Tag, tag_id)
-        if tag is None or tag.kind not in ELIGIBLE_KINDS:
-            return False
-
-        min_refs = await self._min_reference_images()
-
-        stmt = (
-            select(ImageRecord.siglip_embedding)
-            .join(image_tag, image_tag.c.image_record_id == ImageRecord.id)
-            .where(image_tag.c.tag_id == tag_id)
-            .where(ImageRecord.siglip_embedding.is_not(None))
-        )
-        embeddings = [
-            np.array(e, dtype=np.float32)
-            for e in (await self.session.execute(stmt)).scalars().all()
-        ]
-        if len(embeddings) < min_refs:
-            return False
-
-        centroid = np.mean(np.stack(embeddings), axis=0).astype(np.float32)
-        model_version = await self._model_version()
-
-        stmt = insert(TagReferenceEmbedding).values(
-            tag_id=tag_id,
-            embedding=centroid.tolist(),
-            reference_count=len(embeddings),
-            model_version=model_version,
-        )
-        stmt = stmt.on_conflict_do_update(
-            index_elements=["tag_id"],
-            set_={
-                "embedding": centroid.tolist(),
-                "reference_count": len(embeddings),
-                "model_version": model_version,
-                "updated_at": func.now(),
-            },
-        )
-        await self.session.execute(stmt)
-        return True
-
-    async def list_drifted(self) -> list[int]:
-        """Tag ids whose centroid is stale: member count != reference_count,
-        OR no centroid row, OR centroid built on a different SigLIP version.
-        Only considers eligible-kind tags with embeddings present."""
-        current_model_version = await self._model_version()
-        member_counts = (
-            select(
-                image_tag.c.tag_id.label("tag_id"),
-                func.count(image_tag.c.image_record_id).label("members"),
-            )
-            .join(ImageRecord, ImageRecord.id == image_tag.c.image_record_id)
-            .where(ImageRecord.siglip_embedding.is_not(None))
-            .group_by(image_tag.c.tag_id)
-            .subquery()
-        )
-        stmt = (
-            select(Tag.id)
-            .join(member_counts, member_counts.c.tag_id == Tag.id)
-            .outerjoin(
-                TagReferenceEmbedding,
-                TagReferenceEmbedding.tag_id == Tag.id,
-            )
-            .where(Tag.kind.in_(ELIGIBLE_KINDS))
-            .where(
-                (TagReferenceEmbedding.tag_id.is_(None))
-                | (
-                    TagReferenceEmbedding.reference_count
-                    != member_counts.c.members
-                )
-                | (TagReferenceEmbedding.model_version != current_model_version)
-            )
-        )
-        return list((await self.session.execute(stmt)).scalars().all())
-
-    async def find_similar_tags(
-        self, image_id: int, limit: int = 20
-    ) -> list[CentroidHit]:
-        """Cosine similarity between an image's embedding and stored
-        centroids. Returns top-`limit` by similarity DESC. pgvector's
-        cosine_distance gives 1 - cosine_similarity."""
-        img = await self.session.get(ImageRecord, image_id)
-        if img is None or img.siglip_embedding is None:
-            return []
-        emb = img.siglip_embedding
-        distance = TagReferenceEmbedding.embedding.cosine_distance(emb)
-        stmt = (
-            select(
-                TagReferenceEmbedding.tag_id,
-                (1 - distance).label("similarity"),
-            )
-            .order_by(distance.asc())
-            .limit(limit)
-        )
-        rows = (await self.session.execute(stmt)).all()
-        return [
-            CentroidHit(tag_id=r.tag_id, similarity=float(r.similarity))
-            for r in rows
-        ]

backend/app/services/tag_service.py

@@ -11,7 +11,6 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import HeadMetric, Tag, TagHead, TagKind, image_tag
 from ..models.tag_allowlist import TagAllowlist
-from ..models.tag_reference_embedding import TagReferenceEmbedding
 from .db_helpers import get_or_create
 from .tag_query import fandom_join_alias, tag_columns
 
@@ -304,10 +303,10 @@ class TagService:
 
     async def _keep_as_alias(self, tag_id: int) -> bool:
         """A merged-away tag's old name must survive as an alias iff the ML
-        pipeline has ever applied it OR could re-emit it (allowlisted / has
-        a centroid) — otherwise the proactive apply_allowlist_tags worker
-        would silently regenerate it. Purely-manual, ML-unknown tags are
-        deleted outright (no DB bloat)."""
+        pipeline has ever applied it OR could re-emit it (allowlisted) —
+        otherwise the proactive apply_allowlist_tags worker would silently
+        regenerate it. Purely-manual, ML-unknown tags are deleted outright (no
+        DB bloat)."""
         is_machine = await self.session.scalar(
             select(
                 exists().where(
@@ -325,14 +324,7 @@ class TagService:
         allowlisted = await self.session.scalar(
             select(exists().where(TagAllowlist.tag_id == tag_id))
         )
-        if allowlisted:
-            return True
-        has_centroid = await self.session.scalar(
-            select(
-                exists().where(TagReferenceEmbedding.tag_id == tag_id)
-            )
-        )
-        return bool(has_centroid)
+        return bool(allowlisted)
 
     async def rename(self, tag_id: int, new_name: str) -> Tag:
         """Rename a tag. Raises TagMergeConflict if the new name collides
@@ -573,7 +565,6 @@ class TagService:
         merged_count = await self._repoint_image_tags(source_id, target_id)
         await self._repoint_rejections(source_id, target_id)
         await self._repoint_allowlist(source_id, target_id)
-        await self._repoint_embedding(source_id)
         await self._repoint_aliases(source_id, target_id)
         await self._repoint_fandom_children(
             source_id, target_id, source_kind
@@ -655,13 +646,6 @@ class TagService:
                 .values(tag_id=tgt)
             )
 
-    async def _repoint_embedding(self, src: int) -> None:
-        await self.session.execute(
-            text(
-                "DELETE FROM tag_reference_embedding WHERE tag_id = :src"
-            ),
-            {"src": src},
-        )
 
     async def _repoint_aliases(self, src: int, tgt: int) -> None:
         from ..models.tag_alias import TagAlias

backend/app/tasks/ml.py

@@ -1,9 +1,9 @@
-"""ML Celery tasks: per-image inference, backfill discovery, centroid
-recompute, allowlist auto-apply, model self-heal.
+"""ML Celery tasks: per-image inference, backfill discovery, head training,
+allowlist auto-apply, model self-heal.
 
-All run on the ml-worker (queue 'ml') except recompute_centroids and
-apply_allowlist_tags sweeps which are 'maintenance' lane. Sync sessions
-(Celery workers are sync processes), same pattern as FC-2a tasks.
+All run on the ml-worker (queue 'ml') except apply_allowlist_tags sweeps which
+are 'maintenance' lane. Sync sessions (Celery workers are sync processes), same
+pattern as FC-2a tasks.
 """
 
 import logging
@@ -487,59 +487,6 @@ def _confidence_for_tag(session, tag, preds: dict) -> float | None:
     return best
 
 
-@celery.task(name="backend.app.tasks.ml.recompute_centroid", bind=True)
-def recompute_centroid(self, tag_id: int) -> bool:
-    import asyncio
-
-    from ..services.ml.centroids import CentroidService
-    from ._async_session import async_session_factory
-
-    async def _run() -> bool:
-        # Per-task NullPool engine bound to THIS asyncio.run loop — the shared
-        # process-wide engine reuses connections across loops and raises
-        # "Future attached to a different loop" on every call after the first.
-        async_factory, async_engine = async_session_factory()
-        try:
-            async with async_factory() as session:
-                svc = CentroidService(session)
-                result = await svc.recompute_for_tag(tag_id)
-                await session.commit()
-                return result
-        finally:
-            await async_engine.dispose()
-
-    return asyncio.run(_run())
-
-
-@celery.task(
-    name="backend.app.tasks.ml.recompute_centroids",
-    bind=True,
-    # Audit 2026-06-02 — drifted-centroid rebuild over potentially
-    # hundreds of tags.
-    soft_time_limit=1800, time_limit=2100,
-)
-def recompute_centroids(self) -> int:
-    """Daily: find drifted centroids, enqueue recompute_centroid for each."""
-    import asyncio
-
-    from ..services.ml.centroids import CentroidService
-    from ._async_session import async_session_factory
-
-    async def _list() -> list[int]:
-        # Per-task NullPool engine bound to this loop (see recompute_centroid).
-        async_factory, async_engine = async_session_factory()
-        try:
-            async with async_factory() as session:
-                return await CentroidService(session).list_drifted()
-        finally:
-            await async_engine.dispose()
-
-    drifted = asyncio.run(_list())
-    for tid in drifted:
-        recompute_centroid.delay(tid)
-    return len(drifted)
-
-
 @celery.task(
     name="backend.app.tasks.ml.tag_eval_run",
     bind=True,