feat(fc2a): schema migration 0002 — tag kinds + fandom hierarchy + import lifecycle

Adds Tag.kind enum (artist/character/fandom/general/series/archive/post/meta/rating),
Tag.fandom_id FK with CHECK constraint (only valid for kind='character'), and a
kind-aware uniqueness index so the same name can exist across kinds and the same
character name can exist in different fandoms.

Adds ImportBatch + ImportTask state-machine tables for scan tracking, plus a
single-row ImportSettings table (CHECK id=1) holding the importer's filter knobs.

Adds image_record.integrity_status column (defaults to 'unknown'); FC-2e
populates this via the integrity verifier.

Drops the unused tag.namespace column from FC-1 — superseded by kind.

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

alembic/versions/0002_fc2a_tag_kinds_and_import_tasks.py

@@ -0,0 +1,208 @@
+"""fc2a: tag kinds, import_task, import_batch, integrity_status
+
+Revision ID: 0002
+Revises: 0001
+Create Date: 2026-05-14
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0002"
+down_revision: Union[str, None] = "0001"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+TAG_KINDS = (
+    "artist",
+    "character",
+    "fandom",
+    "general",
+    "series",
+    "archive",
+    "post",
+    "meta",
+    "rating",
+)
+
+
+def upgrade() -> None:
+    # --- Tag kind enum + fandom_id ---
+    tag_kind = sa.Enum(*TAG_KINDS, name="tag_kind")
+    tag_kind.create(op.get_bind(), checkfirst=True)
+
+    op.add_column(
+        "tag",
+        sa.Column("kind", tag_kind, nullable=False, server_default="general"),
+    )
+    op.add_column(
+        "tag",
+        sa.Column("fandom_id", sa.Integer(), nullable=True),
+    )
+    op.create_foreign_key(
+        "fk_tag_fandom_id_tag",
+        "tag",
+        "tag",
+        ["fandom_id"],
+        ["id"],
+        ondelete="SET NULL",
+    )
+
+    # Drop the old global uniqueness on name; add kind+fandom-aware uniqueness.
+    op.drop_constraint("uq_tag_name", "tag", type_="unique")
+    op.drop_index("ix_tag_name", table_name="tag")
+    op.execute(
+        """
+        CREATE UNIQUE INDEX uq_tag_name_kind_fandom
+          ON tag (name, kind, COALESCE(fandom_id, 0))
+        """
+    )
+
+    # CHECK: fandom_id is only allowed for character kind.
+    op.create_check_constraint(
+        "ck_tag_fandom_requires_character",
+        "tag",
+        "(fandom_id IS NULL) OR (kind = 'character')",
+    )
+
+    # Drop the old namespace column — superseded by kind.
+    op.drop_index("ix_tag_namespace", table_name="tag")
+    op.drop_column("tag", "namespace")
+
+    # --- ImportBatch ---
+    op.create_table(
+        "import_batch",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("triggered_by", sa.String(length=32), nullable=False),
+        sa.Column("source_path", sa.Text(), nullable=False),
+        sa.Column("scan_mode", sa.String(length=16), nullable=False),
+        sa.Column(
+            "started_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("total_files", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("imported", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("skipped", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("failed", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("status", sa.String(length=16), nullable=False, server_default="running"),
+        sa.PrimaryKeyConstraint("id", name="pk_import_batch"),
+    )
+    op.create_index("ix_import_batch_status", "import_batch", ["status"])
+
+    # --- ImportTask ---
+    op.create_table(
+        "import_task",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("batch_id", sa.Integer(), nullable=False),
+        sa.Column("source_path", sa.Text(), nullable=False),
+        sa.Column("task_type", sa.String(length=16), nullable=False),
+        sa.Column("status", sa.String(length=16), nullable=False, server_default="pending"),
+        sa.Column("result_image_id", sa.Integer(), nullable=True),
+        sa.Column("error", sa.Text(), nullable=True),
+        sa.Column("size_bytes", sa.BigInteger(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["batch_id"],
+            ["import_batch.id"],
+            name="fk_import_task_batch_id_import_batch",
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["result_image_id"],
+            ["image_record.id"],
+            name="fk_import_task_result_image_id_image_record",
+            ondelete="SET NULL",
+        ),
+        sa.PrimaryKeyConstraint("id", name="pk_import_task"),
+    )
+    op.create_index("ix_import_task_batch_id", "import_task", ["batch_id"])
+    op.create_index("ix_import_task_status", "import_task", ["status"])
+    op.create_index(
+        "ix_import_task_created_at_desc",
+        "import_task",
+        [sa.text("created_at DESC")],
+    )
+
+    # --- ImportSettings (single-row table) ---
+    op.create_table(
+        "import_settings",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("import_scan_path", sa.Text(), nullable=False, server_default="/import"),
+        sa.Column("min_width", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("min_height", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column(
+            "skip_transparent", sa.Boolean(), nullable=False, server_default=sa.false()
+        ),
+        sa.Column(
+            "transparency_threshold",
+            sa.Float(),
+            nullable=False,
+            server_default="0.9",
+        ),
+        sa.Column(
+            "skip_single_color", sa.Boolean(), nullable=False, server_default=sa.false()
+        ),
+        sa.Column(
+            "single_color_threshold",
+            sa.Float(),
+            nullable=False,
+            server_default="0.95",
+        ),
+        sa.Column("single_color_tolerance", sa.Integer(), nullable=False, server_default="30"),
+        sa.PrimaryKeyConstraint("id", name="pk_import_settings"),
+        sa.CheckConstraint("id = 1", name="ck_import_settings_singleton"),
+    )
+    # Seed the single row immediately so callers can always SELECT id=1.
+    op.execute("INSERT INTO import_settings (id) VALUES (1)")
+
+    # --- ImageRecord additions ---
+    op.add_column(
+        "image_record",
+        sa.Column(
+            "integrity_status",
+            sa.String(length=24),
+            nullable=False,
+            server_default="unknown",
+        ),
+    )
+    op.create_index(
+        "ix_image_record_integrity_status",
+        "image_record",
+        ["integrity_status"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_image_record_integrity_status", table_name="image_record")
+    op.drop_column("image_record", "integrity_status")
+
+    op.drop_table("import_settings")
+    op.drop_index("ix_import_task_created_at_desc", table_name="import_task")
+    op.drop_index("ix_import_task_status", table_name="import_task")
+    op.drop_index("ix_import_task_batch_id", table_name="import_task")
+    op.drop_table("import_task")
+    op.drop_index("ix_import_batch_status", table_name="import_batch")
+    op.drop_table("import_batch")
+
+    op.drop_constraint("ck_tag_fandom_requires_character", "tag", type_="check")
+    op.execute("DROP INDEX uq_tag_name_kind_fandom")
+    op.add_column("tag", sa.Column("namespace", sa.String(length=64), nullable=True))
+    op.create_index("ix_tag_namespace", "tag", ["namespace"])
+    op.create_index("ix_tag_name", "tag", ["name"], unique=False)
+    op.create_unique_constraint("uq_tag_name", "tag", ["name"])
+    op.drop_constraint("fk_tag_fandom_id_tag", "tag", type_="foreignkey")
+    op.drop_column("tag", "fandom_id")
+    op.drop_column("tag", "kind")
+    sa.Enum(name="tag_kind").drop(op.get_bind(), checkfirst=True)

backend/app/models/__init__.py

@@ -6,9 +6,12 @@ from .credential import Credential
 from .download_event import DownloadEvent
 from .image_provenance import ImageProvenance
 from .image_record import ImageRecord
+from .import_batch import ImportBatch
+from .import_settings import ImportSettings
+from .import_task import ImportTask
 from .post import Post
 from .source import Source
-from .tag import Tag, image_tag
+from .tag import Tag, TagKind, image_tag
 
 __all__ = [
     "Base",
@@ -19,6 +22,10 @@ __all__ = [
     "ImageRecord",
     "ImageProvenance",
     "Tag",
+    "TagKind",
     "image_tag",
     "DownloadEvent",
+    "ImportBatch",
+    "ImportTask",
+    "ImportSettings",
 ]

backend/app/models/image_record.py

@@ -40,6 +40,12 @@ class ImageRecord(Base):
     width: Mapped[int | None] = mapped_column(Integer, nullable=True)
     height: Mapped[int | None] = mapped_column(Integer, nullable=True)
 
+    # Integrity verification status. FC-2e populates this; FC-2a leaves rows at 'unknown'.
+    # Values: 'unknown' (default), 'ok', 'corrupt', 'failed_verification'.
+    integrity_status: Mapped[str] = mapped_column(
+        String(24), nullable=False, default="unknown", index=True
+    )
+
     # Thumbnail (populated by FC-2)
     thumbnail_path: Mapped[str | None] = mapped_column(Text, nullable=True)
 

backend/app/models/import_batch.py

@@ -0,0 +1,32 @@
+"""ImportBatch — one scan run, aggregating many ImportTasks."""
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, Integer, String, Text, func
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from .base import Base
+
+
+class ImportBatch(Base):
+    __tablename__ = "import_batch"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    triggered_by: Mapped[str] = mapped_column(String(32), nullable=False)
+    source_path: Mapped[str] = mapped_column(Text, nullable=False)
+    scan_mode: Mapped[str] = mapped_column(String(16), nullable=False)  # quick|deep
+
+    started_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+
+    total_files: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    imported: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    skipped: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    failed: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+
+    status: Mapped[str] = mapped_column(String(16), nullable=False, default="running", index=True)
+    # running | complete | cancelled
+
+    tasks = relationship("ImportTask", back_populates="batch", cascade="all, delete-orphan")

backend/app/models/import_settings.py

@@ -0,0 +1,28 @@
+"""ImportSettings — single-row table holding the importer's tunable knobs.
+
+Enforced as a single row via a CHECK (id = 1) constraint. The application
+always SELECTs id=1 and never inserts/deletes after the initial migration.
+"""
+
+from sqlalchemy import Boolean, CheckConstraint, Float, Integer, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from .base import Base
+
+
+class ImportSettings(Base):
+    __tablename__ = "import_settings"
+    __table_args__ = (CheckConstraint("id = 1", name="ck_import_settings_singleton"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    import_scan_path: Mapped[str] = mapped_column(Text, nullable=False, default="/import")
+
+    min_width: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    min_height: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+
+    skip_transparent: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    transparency_threshold: Mapped[float] = mapped_column(Float, nullable=False, default=0.9)
+
+    skip_single_color: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    single_color_threshold: Mapped[float] = mapped_column(Float, nullable=False, default=0.95)
+    single_color_tolerance: Mapped[int] = mapped_column(Integer, nullable=False, default=30)

backend/app/models/import_task.py

@@ -0,0 +1,41 @@
+"""ImportTask — a single source file to import as part of a batch.
+
+State machine: pending -> queued -> processing -> complete | skipped | failed.
+Workers crashing mid-task leave rows in 'processing'; the recovery sweep
+(tasks/maintenance.py::recover_interrupted_tasks) re-queues rows that have
+been processing longer than the stuck-task threshold.
+"""
+
+from datetime import datetime
+
+from sqlalchemy import BigInteger, DateTime, ForeignKey, Integer, String, Text, func
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from .base import Base
+
+
+class ImportTask(Base):
+    __tablename__ = "import_task"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    batch_id: Mapped[int] = mapped_column(
+        ForeignKey("import_batch.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+
+    source_path: Mapped[str] = mapped_column(Text, nullable=False)
+    task_type: Mapped[str] = mapped_column(String(16), nullable=False)  # media|archive
+    status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending", index=True)
+
+    result_image_id: Mapped[int | None] = mapped_column(
+        ForeignKey("image_record.id", ondelete="SET NULL"), nullable=True
+    )
+    error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    size_bytes: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+
+    batch = relationship("ImportBatch", back_populates="tasks")

backend/app/models/tag.py

@@ -1,17 +1,42 @@
 """Tag + ImageTag association.
 
-Tags use namespaced strings (``artist:Name``, ``character:Name``,
+Tags carry a ``kind`` enum and (for kind='character') an optional
-``rating:safe``, etc.). Artist tags are kept in sync with the Artist table —
+``fandom_id`` FK to another tag of kind='fandom'. The uniqueness key is
-the sync trigger lands in FC-2.
+(name, kind, COALESCE(fandom_id, 0)) so the same name can exist across
+different kinds and the same character name can exist in different fandoms.
 """
 
 from datetime import datetime
+from enum import Enum
 
-from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Table, func
+from sqlalchemy import (
+    CheckConstraint,
+    Column,
+    DateTime,
+    Enum as SQLEnum,
+    ForeignKey,
+    Integer,
+    String,
+    Table,
+    func,
+)
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from .base import Base
 
+
+class TagKind(str, Enum):
+    artist = "artist"
+    character = "character"
+    fandom = "fandom"
+    general = "general"
+    series = "series"
+    archive = "archive"
+    post = "post"
+    meta = "meta"
+    rating = "rating"
+
+
 image_tag = Table(
     "image_tag",
     Base.metadata,
@@ -21,20 +46,34 @@ image_tag = Table(
         primary_key=True,
     ),
     Column("tag_id", ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True),
-    Column("source", String(32), nullable=False, default="manual"),  # manual|auto|ml
+    Column("source", String(32), nullable=False, default="manual"),
     Column("created_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
 )
 
 
 class Tag(Base):
     __tablename__ = "tag"
+    __table_args__ = (
+        CheckConstraint(
+            "(fandom_id IS NULL) OR (kind = 'character')",
+            name="ck_tag_fandom_requires_character",
+        ),
+    )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    name: Mapped[str] = mapped_column(String(255), nullable=False, unique=True, index=True)
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
-    namespace: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    kind: Mapped[TagKind] = mapped_column(
+        SQLEnum(TagKind, name="tag_kind", values_callable=lambda e: [m.value for m in e]),
+        nullable=False,
+        default=TagKind.general,
+    )
+    fandom_id: Mapped[int | None] = mapped_column(
+        ForeignKey("tag.id", ondelete="SET NULL"), nullable=True, index=True
+    )
 
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()
     )
 
     images = relationship("ImageRecord", secondary=image_tag, backref="tags")
+    fandom = relationship("Tag", remote_side="Tag.id", foreign_keys=[fandom_id])

tests/test_migration_0002.py

@@ -0,0 +1,56 @@
+"""Smoke test for migration 0002: confirms model classes import and the
+tag-kind uniqueness rule shape is correct.
+"""
+
+from backend.app.models import (
+    Base,
+    ImportBatch,
+    ImportSettings,
+    ImportTask,
+    Tag,
+    TagKind,
+)
+
+
+def test_new_tables_registered():
+    expected = {"import_batch", "import_task", "import_settings"}
+    assert expected.issubset(Base.metadata.tables.keys())
+
+
+def test_tag_has_kind_and_fandom_id():
+    cols = {c.name for c in Tag.__table__.columns}
+    assert "kind" in cols
+    assert "fandom_id" in cols
+    assert "namespace" not in cols
+
+
+def test_tag_kind_enum_values():
+    expected = {
+        "artist",
+        "character",
+        "fandom",
+        "general",
+        "series",
+        "archive",
+        "post",
+        "meta",
+        "rating",
+    }
+    assert {k.value for k in TagKind} == expected
+
+
+def test_image_record_has_integrity_status():
+    from backend.app.models import ImageRecord
+    cols = {c.name for c in ImageRecord.__table__.columns}
+    assert "integrity_status" in cols
+
+
+def test_import_task_has_state_columns():
+    cols = {c.name for c in ImportTask.__table__.columns}
+    for required in ("batch_id", "source_path", "task_type", "status", "result_image_id"):
+        assert required in cols
+
+
+def test_import_settings_singleton_constraint():
+    constraints = {c.name for c in ImportSettings.__table__.constraints}
+    assert "ck_import_settings_singleton" in constraints