feat(fc2b): add AliasService — (name, category) -> canonical tag

resolve() / resolve_many() (batch, used by the suggestion read path),
idempotent create, remove, list_all. Category-scoped so 'naruto' as
character vs copyright map to different canonicals. Tests marked
integration (real DB).

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

backend/app/services/ml/aliases.py

@@ -0,0 +1,104 @@
+"""Alias resolution + CRUD.
+
+A tag_alias maps (model_name, model_category) -> canonical Tag. Resolution
+happens at suggestion-read time so raw tagger_predictions stay unmolested.
+"""
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+from sqlalchemy import delete, select
+from sqlalchemy.dialects.postgresql import insert
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ...models import Tag, TagAlias
+
+
+@dataclass(frozen=True)
+class AliasRow:
+    alias_string: str
+    alias_category: str
+    canonical_tag_id: int
+    canonical_tag_name: str
+
+
+class AliasService:
+    def __init__(self, session: AsyncSession):
+        self.session = session
+
+    async def resolve(self, name: str, category: str) -> Tag | None:
+        """Return the canonical Tag for (name, category), or None if no alias."""
+        stmt = (
+            select(Tag)
+            .join(TagAlias, TagAlias.canonical_tag_id == Tag.id)
+            .where(TagAlias.alias_string == name)
+            .where(TagAlias.alias_category == category)
+        )
+        return (await self.session.execute(stmt)).scalar_one_or_none()
+
+    async def resolve_many(
+        self, pairs: list[tuple[str, str]]
+    ) -> dict[tuple[str, str], Tag]:
+        """Batch-resolve. Returns only the pairs that have an alias.
+
+        Used by SuggestionService so it does one query instead of N.
+        """
+        if not pairs:
+            return {}
+        strings = {p[0] for p in pairs}
+        stmt = (
+            select(TagAlias, Tag)
+            .join(Tag, Tag.id == TagAlias.canonical_tag_id)
+            .where(TagAlias.alias_string.in_(strings))
+        )
+        rows = (await self.session.execute(stmt)).all()
+        wanted = set(pairs)
+        out: dict[tuple[str, str], Tag] = {}
+        for alias, tag in rows:
+            key = (alias.alias_string, alias.alias_category)
+            if key in wanted:
+                out[key] = tag
+        return out
+
+    async def create(
+        self, alias_string: str, alias_category: str, canonical_tag_id: int
+    ) -> None:
+        """Idempotent create (ON CONFLICT DO NOTHING)."""
+        stmt = insert(TagAlias).values(
+            alias_string=alias_string,
+            alias_category=alias_category,
+            canonical_tag_id=canonical_tag_id,
+        )
+        stmt = stmt.on_conflict_do_nothing(
+            index_elements=["alias_string", "alias_category"]
+        )
+        await self.session.execute(stmt)
+
+    async def remove(self, alias_string: str, alias_category: str) -> None:
+        await self.session.execute(
+            delete(TagAlias)
+            .where(TagAlias.alias_string == alias_string)
+            .where(TagAlias.alias_category == alias_category)
+        )
+
+    async def list_all(self) -> Sequence[AliasRow]:
+        stmt = (
+            select(
+                TagAlias.alias_string,
+                TagAlias.alias_category,
+                TagAlias.canonical_tag_id,
+                Tag.name,
+            )
+            .join(Tag, Tag.id == TagAlias.canonical_tag_id)
+            .order_by(TagAlias.alias_string.asc())
+        )
+        rows = (await self.session.execute(stmt)).all()
+        return [
+            AliasRow(
+                alias_string=r[0],
+                alias_category=r[1],
+                canonical_tag_id=r[2],
+                canonical_tag_name=r[3],
+            )
+            for r in rows
+        ]

tests/test_ml_aliases.py

@@ -0,0 +1,85 @@
+import pytest
+
+from backend.app.models import TagKind
+from backend.app.services.ml.aliases import AliasService
+from backend.app.services.tag_service import TagService
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.mark.asyncio
+async def test_create_and_resolve(db):
+    tags = TagService(db)
+    canonical = await tags.find_or_create("Sasuke Uchiha", TagKind.character)
+    aliases = AliasService(db)
+    await aliases.create("uchiha_sasuke", "character", canonical.id)
+
+    resolved = await aliases.resolve("uchiha_sasuke", "character")
+    assert resolved is not None
+    assert resolved.id == canonical.id
+
+
+@pytest.mark.asyncio
+async def test_resolve_returns_none_when_no_alias(db):
+    aliases = AliasService(db)
+    assert await aliases.resolve("nonexistent", "general") is None
+
+
+@pytest.mark.asyncio
+async def test_category_scoped(db):
+    tags = TagService(db)
+    char = await tags.find_or_create("Naruto", TagKind.character)
+    copy = await tags.find_or_create("Naruto", TagKind.fandom)
+    aliases = AliasService(db)
+    await aliases.create("naruto", "character", char.id)
+    await aliases.create("naruto", "copyright", copy.id)
+
+    assert (await aliases.resolve("naruto", "character")).id == char.id
+    assert (await aliases.resolve("naruto", "copyright")).id == copy.id
+
+
+@pytest.mark.asyncio
+async def test_create_idempotent(db):
+    tags = TagService(db)
+    t = await tags.find_or_create("X", TagKind.general)
+    aliases = AliasService(db)
+    await aliases.create("x_alias", "general", t.id)
+    await aliases.create("x_alias", "general", t.id)  # no error
+    assert (await aliases.resolve("x_alias", "general")).id == t.id
+
+
+@pytest.mark.asyncio
+async def test_resolve_many(db):
+    tags = TagService(db)
+    a = await tags.find_or_create("Alice", TagKind.character)
+    b = await tags.find_or_create("Bob", TagKind.artist)
+    aliases = AliasService(db)
+    await aliases.create("alice_x", "character", a.id)
+    await aliases.create("bob_x", "artist", b.id)
+
+    out = await aliases.resolve_many(
+        [("alice_x", "character"), ("bob_x", "artist"), ("unknown", "general")]
+    )
+    assert out[("alice_x", "character")].id == a.id
+    assert out[("bob_x", "artist")].id == b.id
+    assert ("unknown", "general") not in out
+
+
+@pytest.mark.asyncio
+async def test_remove(db):
+    tags = TagService(db)
+    t = await tags.find_or_create("Y", TagKind.general)
+    aliases = AliasService(db)
+    await aliases.create("y_alias", "general", t.id)
+    await aliases.remove("y_alias", "general")
+    assert await aliases.resolve("y_alias", "general") is None
+
+
+@pytest.mark.asyncio
+async def test_list_all(db):
+    tags = TagService(db)
+    t = await tags.find_or_create("Z", TagKind.general)
+    aliases = AliasService(db)
+    await aliases.create("z1", "general", t.id)
+    rows = await aliases.list_all()
+    assert any(r.alias_string == "z1" and r.canonical_tag_name == "Z" for r in rows)