feat(tags): system tags — is_system column, seeded hygiene tags, protection guards

Training hygiene step 1 (milestone #128). Migration 0075 adds
tag.is_system and seeds wip / banner / editor screenshot (kind=general),
ADOPTING an existing same-(name,kind) tag case-insensitively instead of
duplicating. These rows drive the upcoming training exclusions, so they
are protected: rename and merge-away refuse system tags (merge-INTO
stays allowed — folding an operator's old hygiene tag into the system
row is the intended move; merge is the only tag-delete path, so that
guard covers deletion). is_system rides every tag serialization.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM

alembic/versions/0075_tag_is_system.py

@@ -0,0 +1,60 @@
+"""tag.is_system + seed the three hygiene system tags
+
+Training hygiene (operator 2026-07-03, milestone #128): rough WIPs tagged as a
+character poison that character's head and CCIP references; banners/editor
+screenshots pollute whole-image similarity. The fix keys on SYSTEM tags the
+product ships — not operator configuration — so the seed lives here.
+
+Seeding ADOPTS an existing same-(name, kind=general) tag (case-insensitive,
+matching TagService.rename's collision stance) instead of inserting a
+duplicate, so an operator who already tagged `wip` keeps their applications.
+
+Revision ID: 0075
+Revises: 0074
+Create Date: 2026-07-03
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0075"
+down_revision: Union[str, None] = "0074"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+SYSTEM_TAG_NAMES = ("wip", "banner", "editor screenshot")
+
+
+def upgrade() -> None:
+    op.add_column(
+        "tag",
+        sa.Column(
+            "is_system", sa.Boolean(), nullable=False,
+            server_default=sa.false(),
+        ),
+    )
+    conn = op.get_bind()
+    for name in SYSTEM_TAG_NAMES:
+        adopted = conn.execute(
+            sa.text(
+                "UPDATE tag SET is_system = true "
+                "WHERE lower(name) = lower(:name) AND kind = 'general'"
+            ),
+            {"name": name},
+        )
+        if adopted.rowcount == 0:
+            conn.execute(
+                sa.text(
+                    "INSERT INTO tag (name, kind, is_system) "
+                    "VALUES (:name, 'general', true)"
+                ),
+                {"name": name},
+            )
+
+
+def downgrade() -> None:
+    # The seeded rows survive as ordinary general tags — dropping the flag is
+    # enough to disarm the mechanism, and deleting rows would orphan any
+    # operator applications made while the flag existed.
+    op.drop_column("tag", "is_system")

backend/app/api/tags.py

@@ -324,6 +324,7 @@ async def get_tag(tag_id: int):
                 "name": tag.name,
                 "kind": tag.kind.value,
                 "fandom_id": tag.fandom_id,
+                "is_system": tag.is_system,
             }
         )
 
@@ -390,6 +391,7 @@ async def update_tag(tag_id: int):
                 "name": tag.name,
                 "kind": tag.kind.value,
                 "fandom_id": tag.fandom_id,
+                "is_system": tag.is_system,
             }
         )
 

backend/app/models/tag.py

@@ -10,6 +10,7 @@ from datetime import datetime
 from enum import StrEnum
 
 from sqlalchemy import (
+    Boolean,
     CheckConstraint,
     Column,
     DateTime,
@@ -17,6 +18,7 @@ from sqlalchemy import (
     Integer,
     String,
     Table,
+    false,
     func,
 )
 from sqlalchemy import (
@@ -74,6 +76,14 @@ class Tag(Base):
     fandom_id: Mapped[int | None] = mapped_column(
         ForeignKey("tag.id", ondelete="SET NULL"), nullable=True, index=True
     )
+    # System tags ship with FC (wip / banner / editor screenshot, seeded in
+    # migration 0075) and drive the training-hygiene exclusions: images
+    # carrying one are excluded from OTHER concepts' head training and from
+    # CCIP identity references. The mechanism keys on these exact rows, so
+    # they're protected from rename/merge-away/re-fandom in TagService.
+    is_system: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False, server_default=false()
+    )
 
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()

backend/app/services/tag_query.py

@@ -67,24 +67,28 @@ def fandom_join_alias():
 
 
 def tag_columns(fandom_alias):
-    """The canonical (id, name, kind, fandom_id, fandom_name) column set for a
-    tag select that outerjoins `fandom_alias` (from `fandom_join_alias()`)."""
+    """The canonical (id, name, kind, fandom_id, fandom_name, is_system)
+    column set for a tag select that outerjoins `fandom_alias` (from
+    `fandom_join_alias()`)."""
     return [
         Tag.id,
         Tag.name,
         Tag.kind,
         Tag.fandom_id,
         fandom_alias.c.name.label("fandom_name"),
+        Tag.is_system,
     ]
 
 
 def serialize_tag(row) -> dict:
     """Serialize a row carrying .id/.name/.kind/.fandom_id/.fandom_name to the
-    canonical tag dict. `kind` may be a TagKind enum or a plain string."""
+    canonical tag dict. `kind` may be a TagKind enum or a plain string.
+    `is_system` defaults False for callers whose selects predate the column."""
     return {
         "id": row.id,
         "name": row.name,
         "kind": row.kind.value if hasattr(row.kind, "value") else row.kind,
         "fandom_id": row.fandom_id,
         "fandom_name": row.fandom_name,
+        "is_system": bool(getattr(row, "is_system", False)),
     }

backend/app/services/tag_service.py

@@ -330,6 +330,12 @@ class TagService:
         tag = await self.session.get(Tag, tag_id)
         if tag is None:
             raise TagValidationError(f"Tag {tag_id} not found")
+        # The training-hygiene machinery keys on system tags by row; a rename
+        # would silently disarm it (milestone #128).
+        if tag.is_system:
+            raise TagValidationError(
+                f"'{tag.name}' is a system tag and cannot be renamed"
+            )
 
         # Case-insensitive clash (#701) — renaming onto a differently-cased tag
         # is still a merge, not a silent fork.
@@ -528,6 +534,14 @@ class TagService:
         source = await self.session.get(Tag, source_id)
         if source is None:
             raise TagValidationError(f"Tag {source_id} not found")
+        # Merge deletes the source row — the only tag-delete path — and the
+        # training-hygiene machinery keys on system rows (milestone #128).
+        # Merging INTO a system tag stays allowed (adopting an operator's old
+        # 'wip sketch' tag into the system 'wip' is the intended move).
+        if source.is_system:
+            raise TagValidationError(
+                f"'{source.name}' is a system tag and cannot be merged away"
+            )
         target = await self.session.get(Tag, target_id)
         if target is None:
             raise TagValidationError(f"Tag {target_id} not found")

tests/test_system_tags.py

@@ -0,0 +1,81 @@
+"""System tags (milestone #128): seeded rows, protection guards, serialization.
+
+The three hygiene tags (wip / banner / editor screenshot) ship via migration
+0075 and drive the training-hygiene exclusions. They must exist after
+migration, serialize with is_system, and refuse rename/merge-away — the
+mechanism keys on these exact rows. Merge-INTO stays allowed: adopting an
+operator's old hygiene tag into the system row is the intended move.
+"""
+import pytest
+from sqlalchemy import select
+
+from backend.app.models import Tag, TagKind
+
+pytestmark = pytest.mark.integration
+
+SYSTEM_NAMES = {"wip", "banner", "editor screenshot"}
+
+
+async def _system_tag(db, name):
+    return (await db.execute(
+        select(Tag).where(Tag.is_system.is_(True), Tag.name == name)
+    )).scalar_one()
+
+
+@pytest.mark.asyncio
+async def test_seeded_system_tags_exist(db):
+    rows = (await db.execute(
+        select(Tag.name, Tag.kind).where(Tag.is_system.is_(True))
+    )).all()
+    assert {r.name for r in rows} == SYSTEM_NAMES
+    assert all(r.kind == TagKind.general for r in rows)
+
+
+@pytest.mark.asyncio
+async def test_get_tag_serializes_is_system(client, db):
+    wip = await _system_tag(db, "wip")
+    resp = await client.get(f"/api/tags/{wip.id}")
+    assert resp.status_code == 200
+    assert (await resp.get_json())["is_system"] is True
+
+
+@pytest.mark.asyncio
+async def test_rename_system_tag_refused(client, db):
+    wip = await _system_tag(db, "wip")
+    resp = await client.patch(
+        f"/api/tags/{wip.id}", json={"name": "work in progress"}
+    )
+    assert resp.status_code == 400
+    body = await resp.get_json()
+    assert "system tag" in body["error"]
+
+
+@pytest.mark.asyncio
+async def test_merge_away_system_tag_refused(client, db):
+    banner = await _system_tag(db, "banner")
+    created = await client.post(
+        "/api/tags", json={"name": "ad banner", "kind": "general"}
+    )
+    target_id = (await created.get_json())["id"]
+    resp = await client.post(
+        f"/api/tags/{banner.id}/merge", json={"target_id": target_id}
+    )
+    assert resp.status_code == 400
+    body = await resp.get_json()
+    assert "system tag" in body["error"]
+
+
+@pytest.mark.asyncio
+async def test_merge_into_system_tag_allowed(client, db):
+    wip = await _system_tag(db, "wip")
+    created = await client.post(
+        "/api/tags", json={"name": "old wip", "kind": "general"}
+    )
+    source_id = (await created.get_json())["id"]
+    resp = await client.post(
+        f"/api/tags/{source_id}/merge", json={"target_id": wip.id}
+    )
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    assert body["target"]["id"] == wip.id
+    assert body["source_deleted"] is True

tests/test_tag_query.py

@@ -10,20 +10,29 @@ def test_serialize_tag_with_enum_kind():
     # ORM/column rows carry kind as a TagKind enum.
     row = SimpleNamespace(
         id=1, name="Sasuke Uchiha", kind=TagKind.character,
-        fandom_id=5, fandom_name="Naruto",
+        fandom_id=5, fandom_name="Naruto", is_system=False,
     )
     assert serialize_tag(row) == {
         "id": 1, "name": "Sasuke Uchiha", "kind": "character",
-        "fandom_id": 5, "fandom_name": "Naruto",
+        "fandom_id": 5, "fandom_name": "Naruto", "is_system": False,
     }
 
 
 def test_serialize_tag_with_string_kind_and_no_fandom():
-    # autocomplete hits already store kind as a plain string.
+    # autocomplete hits already store kind as a plain string. A row from a
+    # select that predates the is_system column serializes as non-system.
     row = SimpleNamespace(
         id=2, name="solo", kind="general", fandom_id=None, fandom_name=None,
     )
     assert serialize_tag(row) == {
         "id": 2, "name": "solo", "kind": "general",
-        "fandom_id": None, "fandom_name": None,
+        "fandom_id": None, "fandom_name": None, "is_system": False,
     }
+
+
+def test_serialize_tag_system_row():
+    row = SimpleNamespace(
+        id=3, name="wip", kind=TagKind.general,
+        fandom_id=None, fandom_name=None, is_system=True,
+    )
+    assert serialize_tag(row)["is_system"] is True