feat(tags): protective alias creation per observed tagger category with kind fallback

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

backend/app/services/tag_service.py

@@ -412,4 +412,47 @@ class TagService:
     async def _create_protective_aliases(
         self, src_name: str, src_kind: TagKind, tgt: int
     ) -> bool:
-        return False
+        """One alias per category the tagger has actually emitted for
+        src_name (so future predictions resolve to target); fall back to
+        the tag's kind when the tagger never predicted this exact name.
+        Idempotent — never clobbers a pre-existing operator alias."""
+        from ..models.tag_alias import TagAlias
+
+        rows = (
+            await self.session.execute(
+                text(
+                    "SELECT DISTINCT "
+                    "  (tagger_predictions::jsonb -> :n ->> 'category') "
+                    "  AS cat "
+                    "FROM image_record "
+                    "WHERE tagger_predictions IS NOT NULL "
+                    "  AND (tagger_predictions::jsonb) ? :n"
+                ),
+                {"n": src_name},
+            )
+        ).all()
+        categories = {r.cat for r in rows if r.cat}
+        if not categories:
+            kind_val = (
+                src_kind.value
+                if hasattr(src_kind, "value")
+                else str(src_kind)
+            )
+            categories = {kind_val}
+
+        created = False
+        for cat in categories:
+            res = await self.session.execute(
+                pg_insert(TagAlias)
+                .values(
+                    alias_string=src_name,
+                    alias_category=cat,
+                    canonical_tag_id=tgt,
+                )
+                .on_conflict_do_nothing(
+                    index_elements=["alias_string", "alias_category"]
+                )
+            )
+            if res.rowcount:
+                created = True
+        return created

tests/test_tag_merge.py

@@ -302,3 +302,103 @@ async def test_merge_fandom_reparents_children(db):
     db.expire_all()  # merge() uses Core DML; drop stale identity-map state
     refreshed = await db.get(Tag, child.id)
     assert refreshed.fandom_id == f_tgt.id
+
+
+@pytest.mark.asyncio
+async def test_no_alias_when_purely_manual(db):
+    from backend.app.models.tag_alias import TagAlias
+
+    svc = TagService(db)
+    a = await svc.find_or_create("ManualOnly", TagKind.general)
+    b = await svc.find_or_create("CanonM", TagKind.general)
+    img = await _img(db)
+    await svc.add_to_image(img, a.id, source="manual")
+    result = await svc.merge(a.id, b.id)
+    assert result.alias_created is False
+    assert (await db.execute(select(TagAlias))).first() is None
+
+
+@pytest.mark.asyncio
+async def test_alias_per_observed_prediction_category(db):
+    from backend.app.models import ImageRecord
+    from backend.app.models.tag_alias import TagAlias
+
+    svc = TagService(db)
+    a = await svc.find_or_create("predname", TagKind.general)
+    b = await svc.find_or_create("CanonP", TagKind.general)
+    img = await _img(db)
+    # mark source machine-known so keep_as_alias is True
+    await svc.add_to_image(img, a.id, source="ml_auto")
+    r1 = await db.get(ImageRecord, img)
+    r1.tagger_predictions = {
+        "predname": {"category": "general", "confidence": 0.9}
+    }
+    i2 = await _img(db)
+    r2 = await db.get(ImageRecord, i2)
+    r2.tagger_predictions = {
+        "predname": {"category": "copyright", "confidence": 0.8}
+    }
+    await db.flush()
+    result = await svc.merge(a.id, b.id)
+    assert result.alias_created is True
+    db.expire_all()
+    rows = (await db.execute(select(TagAlias))).scalars().all()
+    assert {(r.alias_string, r.alias_category) for r in rows} == {
+        ("predname", "general"),
+        ("predname", "copyright"),
+    }
+    assert all(r.canonical_tag_id == b.id for r in rows)
+
+
+@pytest.mark.asyncio
+async def test_alias_fallback_to_kind_when_no_predictions(db):
+    from backend.app.models.tag_alias import TagAlias
+
+    svc = TagService(db)
+    a = await svc.find_or_create("AllowNoPred", TagKind.character)
+    b = await svc.find_or_create("CanonF", TagKind.character)
+    db.add(TagAllowlist(tag_id=a.id))
+    await db.flush()
+    result = await svc.merge(a.id, b.id)
+    assert result.alias_created is True
+    db.expire_all()
+    al = (
+        await db.execute(
+            select(TagAlias).where(TagAlias.alias_string == "AllowNoPred")
+        )
+    ).scalar_one()
+    assert al.alias_category == "character"
+    assert al.canonical_tag_id == b.id
+
+
+@pytest.mark.asyncio
+async def test_alias_create_does_not_clobber_existing(db):
+    from backend.app.models import ImageRecord
+    from backend.app.models.tag_alias import TagAlias
+
+    svc = TagService(db)
+    a = await svc.find_or_create("dupalias", TagKind.general)
+    b = await svc.find_or_create("CanonD", TagKind.general)
+    other = await svc.find_or_create("OtherCanon", TagKind.general)
+    db.add(
+        TagAlias(
+            alias_string="dupalias",
+            alias_category="general",
+            canonical_tag_id=other.id,
+        )
+    )
+    img = await _img(db)
+    await svc.add_to_image(img, a.id, source="ml_auto")
+    r = await db.get(ImageRecord, img)
+    r.tagger_predictions = {
+        "dupalias": {"category": "general", "confidence": 0.9}
+    }
+    await db.flush()
+    await svc.merge(a.id, b.id)
+    db.expire_all()
+    al = (
+        await db.execute(
+            select(TagAlias).where(TagAlias.alias_string == "dupalias")
+        )
+    ).scalar_one()
+    assert al.canonical_tag_id == other.id  # pre-existing alias untouched