fix(tags): fandom views aggregate images via their characters

A fandom owns characters via Tag.fandom_id, but every image<->tag query
went purely through direct image_tag rows, so a fandom only surfaced
images literally tagged with it — images carrying one of its characters
were invisible to its browse count, previews, and gallery filter.

Derive membership at query time instead of materializing fandom rows
(which would drift on every reassign/merge/remove). Add one shared
predicate in tag_query.py — image_in_tag_scope / image_in_any_tag_scope:
an image belongs to a tag if tagged with it directly OR (when the tag is
a fandom) carrying a character whose fandom_id is that tag. The character
leg is empty for non-fandom tags, so it applies uniformly with no kind
branching. Route all read sites through it:

- gallery _apply_scope: include, OR-groups, and symmetric exclude
- directory image_count: correlated COUNT(DISTINCT) scalar subquery
- directory previews: UNION direct + via-character, then ROW_NUMBER<=3
- cleanup count_tag_associations: Tier-B delete prompt now reports a
  fandom's true blast radius (was 0 for fandoms with no direct rows)

find_unused_tags already protected fandoms via used_via_fandom; left as is.

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

backend/app/services/cleanup_service.py

@@ -18,7 +18,7 @@ from pathlib import Path
 from typing import Any
 
 from sqlalchemy import delete, func, or_, select, update
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, aliased
 
 from ..models import (
     Artist,
@@ -151,11 +151,22 @@ def project_bulk_image_delete(
 
 
 def count_tag_associations(session: Session, *, tag_id: int) -> int:
-    """COUNT(*) FROM image_tag WHERE tag_id=?. For Tier-B prompt."""
+    """Images affected by deleting this tag — the Tier-B blast-radius prompt.
+    Mirrors the gallery/directory membership predicate: images carrying the tag
+    DIRECTLY, plus — when it's a fandom — images carrying one of its characters
+    (member.fandom_id == tag_id). DISTINCT so each image counts once. Without
+    the character leg a fandom would report 0 here yet its delete still strips
+    the fandom off every character, badly understating the prompt."""
+    member = aliased(Tag)
     return session.execute(
-        select(func.count())
-        .select_from(image_tag)
-        .where(image_tag.c.tag_id == tag_id)
+        select(func.count(image_tag.c.image_record_id.distinct())).where(
+            or_(
+                image_tag.c.tag_id == tag_id,
+                image_tag.c.tag_id.in_(
+                    select(member.id).where(member.fandom_id == tag_id)
+                ),
+            )
+        )
     ).scalar_one()
 
 

backend/app/services/gallery_service.py

@@ -25,7 +25,13 @@ from sqlalchemy.orm import aliased
 from ..models import Artist, ImageProvenance, ImageRecord, Post, Source, Tag
 from ..models.tag import image_tag
 from .pagination import decode_cursor, encode_cursor
-from .tag_query import fandom_join_alias, serialize_tag, tag_columns
+from .tag_query import (
+    fandom_join_alias,
+    image_in_any_tag_scope,
+    image_in_tag_scope,
+    serialize_tag,
+    tag_columns,
+)
 
 # Reserved `platform` filter value selecting images with NO platformed
 # provenance (filesystem imports). Returned by facets() as a null-valued
@@ -180,29 +186,17 @@ def _apply_scope(
     - no_artist: ImageRecord.artist_id IS NULL.
     - date_from / date_to: half-open [from, to) bounds on effective_date.
     """
+    # Every tag clause goes through image_in_tag_scope/_any: a fandom tag also
+    # matches images carrying any of its characters (Tag.fandom_id). Include,
+    # OR-group, and exclude are all symmetric on that membership.
     for tid in tag_ids or []:
-        stmt = stmt.where(
-            exists().where(
-                image_tag.c.image_record_id == ImageRecord.id,
-                image_tag.c.tag_id == tid,
-            )
-        )
+        stmt = stmt.where(image_in_tag_scope(tid))
     for group in tag_or_groups or []:
         if not group:
             continue  # an empty OR-group would match nothing; treat as absent
-        stmt = stmt.where(
-            exists().where(
-                image_tag.c.image_record_id == ImageRecord.id,
-                image_tag.c.tag_id.in_(group),
-            )
-        )
+        stmt = stmt.where(image_in_any_tag_scope(group))
     if tag_exclude:
-        stmt = stmt.where(
-            ~exists().where(
-                image_tag.c.image_record_id == ImageRecord.id,
-                image_tag.c.tag_id.in_(tag_exclude),
-            )
-        )
+        stmt = stmt.where(~image_in_any_tag_scope(tag_exclude))
     prov = _provenance_clause(post_id, artist_id)
     if prov is not None:
         stmt = stmt.where(prov)

backend/app/services/tag_directory_service.py

@@ -53,12 +53,29 @@ class TagDirectoryService:
             raise ValueError("limit must be between 1 and 200")
 
         fandom = aliased(Tag)
-        count_col = func.count(image_tag.c.image_record_id).label("image_count")
-        stmt = (
-            select(Tag, fandom.name.label("fandom_name"), count_col)
-            .outerjoin(fandom, Tag.fandom_id == fandom.id)
-            .outerjoin(image_tag, image_tag.c.tag_id == Tag.id)
-            .group_by(Tag.id, fandom.name)
+        member = aliased(Tag)
+        # image_count aggregates the tag's images INCLUDING, for a fandom, every
+        # image carrying one of its characters (member.fandom_id == Tag.id) — the
+        # member subquery is empty for non-fandom tags, so they stay direct-only.
+        # DISTINCT so an image with both the fandom and ≥1 of its characters
+        # counts once; a correlated scalar subquery keeps it 1 row per tag with
+        # no group_by (the fandom self-join below is 1:1).
+        count_col = (
+            select(func.count(image_tag.c.image_record_id.distinct()))
+            .where(
+                or_(
+                    image_tag.c.tag_id == Tag.id,
+                    image_tag.c.tag_id.in_(
+                        select(member.id).where(member.fandom_id == Tag.id)
+                    ),
+                )
+            )
+            .correlate(Tag)
+            .scalar_subquery()
+            .label("image_count")
+        )
+        stmt = select(Tag, fandom.name.label("fandom_name"), count_col).outerjoin(
+            fandom, Tag.fandom_id == fandom.id
         )
         if kind is not None:
             stmt = stmt.where(Tag.kind == kind)
@@ -101,19 +118,34 @@ class TagDirectoryService:
     async def _previews(self, tag_ids: list[int]) -> dict[int, list[str]]:
         if not tag_ids:
             return {}
-        rn = func.row_number().over(
-            partition_by=image_tag.c.tag_id,
-            order_by=image_tag.c.image_record_id.desc(),
-        ).label("rn")
-        sub = (
+        # Preview pool per card = images tagged with the card's tag DIRECTLY,
+        # UNION images carrying a character whose fandom_id is the card's tag (so
+        # a fandom card previews its characters' images). UNION dedups the
+        # overlap. For non-fandom cards the via-fandom leg is empty.
+        member = aliased(Tag)
+        direct = select(
+            image_tag.c.tag_id.label("card_id"),
+            image_tag.c.image_record_id.label("image_record_id"),
+        ).where(image_tag.c.tag_id.in_(tag_ids))
+        via_fandom = (
             select(
-                image_tag.c.tag_id.label("tag_id"),
+                member.fandom_id.label("card_id"),
                 image_tag.c.image_record_id.label("image_record_id"),
-                rn,
             )
-            .where(image_tag.c.tag_id.in_(tag_ids))
-            .subquery()
+            .select_from(image_tag)
+            .join(member, member.id == image_tag.c.tag_id)
+            .where(member.fandom_id.in_(tag_ids))
         )
+        scoped = direct.union(via_fandom).subquery()
+        rn = func.row_number().over(
+            partition_by=scoped.c.card_id,
+            order_by=scoped.c.image_record_id.desc(),
+        ).label("rn")
+        sub = select(
+            scoped.c.card_id.label("tag_id"),
+            scoped.c.image_record_id.label("image_record_id"),
+            rn,
+        ).subquery()
         stmt = (
             select(
                 sub.c.tag_id,

backend/app/services/tag_query.py

@@ -1,4 +1,10 @@
-"""Shared tag-with-fandom query columns + serialization.
+"""Shared tag-with-fandom query columns, serialization, and membership.
+
+`image_in_tag_scope`/`image_in_any_tag_scope` are the SINGLE source of truth for
+"which images belong to a tag" — a fandom tag also owns the images of its
+characters (Tag.fandom_id), derived at query time rather than materialized. The
+gallery scope, directory count/previews, and the cleanup impact count all route
+through them so a fandom can never under-count its characters' images.
 
 Resolving a character tag's fandom NAME via a Tag self-join (Tag.fandom_id -> the
 fandom Tag) and serializing the canonical
@@ -11,7 +17,47 @@ TagDirectoryService selects the FULL Tag ORM plus an image-count aggregate (a
 different select shape), so it keeps its own variant — not folded in here.
 """
 
-from ..models import Tag
+from sqlalchemy import exists, or_, select
+from sqlalchemy.orm import aliased
+
+from ..models import ImageRecord, Tag
+from ..models.tag import image_tag
+
+
+def _fandom_member_char_ids(tids):
+    """Subquery of character tag ids owned by any fandom in `tids` (via
+    Tag.fandom_id). Empty for non-fandom tids, so callers can OR it in
+    unconditionally — a fandom aggregates its characters' images, every other
+    kind degrades to direct-only."""
+    char = aliased(Tag)
+    return select(char.id).where(char.fandom_id.in_(tids))
+
+
+def image_in_tag_scope(tid):
+    """Correlated EXISTS: the current ImageRecord belongs to tag `tid` — either
+    tagged with it DIRECTLY, or (when `tid` is a fandom) carrying a character
+    whose fandom_id == `tid`. This is the SINGLE membership predicate shared by
+    the gallery scope, the directory count/previews, and the cleanup impact
+    count, so a fandom never under-counts the images of its characters."""
+    return exists().where(
+        image_tag.c.image_record_id == ImageRecord.id,
+        or_(
+            image_tag.c.tag_id == tid,
+            image_tag.c.tag_id.in_(_fandom_member_char_ids([tid])),
+        ),
+    )
+
+
+def image_in_any_tag_scope(tids):
+    """Like `image_in_tag_scope` but for an OR-group: the image carries AT LEAST
+    ONE of `tids` directly, or a character of any fandom in `tids`."""
+    return exists().where(
+        image_tag.c.image_record_id == ImageRecord.id,
+        or_(
+            image_tag.c.tag_id.in_(tids),
+            image_tag.c.tag_id.in_(_fandom_member_char_ids(tids)),
+        ),
+    )
 
 
 def fandom_join_alias():

tests/test_cleanup_service.py

@@ -146,6 +146,27 @@ def test_count_tag_associations_counts_image_tag_rows(db_sync, tmp_path):
     assert cleanup_service.count_tag_associations(db_sync, tag_id=tag.id) == 1
 
 
+def test_count_tag_associations_fandom_includes_character_images(db_sync, tmp_path):
+    # Deleting a fandom affects images carrying its characters too, so the
+    # Tier-B blast-radius prompt must count them — not just direct image_tag
+    # rows on the fandom (a fandom usually has zero of those). DISTINCT: an
+    # image with both the fandom and its character counts once.
+    a = _make_artist(db_sync, slug="fc")
+    img0 = _make_image(db_sync, artist=a, path=str(tmp_path / "0.jpg"), sha256="e" * 64)
+    img1 = _make_image(db_sync, artist=a, path=str(tmp_path / "1.jpg"), sha256="f" * 64)
+    fandom = _make_tag(db_sync, name="Creux", kind=TagKind.fandom)
+    char = _make_tag(db_sync, name="OcChar", kind=TagKind.character)
+    char.fandom_id = fandom.id
+    db_sync.flush()
+    db_sync.execute(image_tag.insert().values([
+        {"image_record_id": img0.id, "tag_id": char.id},
+        {"image_record_id": img1.id, "tag_id": char.id},
+        {"image_record_id": img1.id, "tag_id": fandom.id},
+    ]))
+    db_sync.commit()
+    assert cleanup_service.count_tag_associations(db_sync, tag_id=fandom.id) == 2
+
+
 def test_find_unused_tags_returns_only_unreferenced(db_sync, tmp_path):
     a = _make_artist(db_sync, slug="fu")
     img = _make_image(

tests/test_gallery_service.py

@@ -103,6 +103,32 @@ async def test_scroll_with_tag_filter(db):
     assert page.images[0].id == images[0].id
 
 
+@pytest.mark.asyncio
+async def test_scroll_fandom_includes_images_via_its_characters(db):
+    # A fandom owns a character (Tag.fandom_id). Filtering by the fandom must
+    # surface images that carry the CHARACTER but were never tagged with the
+    # fandom directly — plus images tagged with the fandom directly. Excluding
+    # the fandom is symmetric: it hides both.
+    images = await _seed_images(db, 4)
+    fandom = Tag(name="Naruto", kind=TagKind.fandom)
+    db.add(fandom)
+    await db.flush()
+    char = Tag(name="Sasuke", kind=TagKind.character, fandom_id=fandom.id)
+    other = Tag(name="unrelated", kind=TagKind.general)
+    db.add_all([char, other])
+    await db.flush()
+    await db.execute(image_tag.insert().values([
+        {"image_record_id": images[0].id, "tag_id": char.id, "source": "manual"},
+        {"image_record_id": images[1].id, "tag_id": fandom.id, "source": "manual"},
+        {"image_record_id": images[2].id, "tag_id": other.id, "source": "manual"},
+    ]))
+    svc = GalleryService(db)
+    via = await svc.scroll(cursor=None, limit=10, tag_ids=[fandom.id])
+    assert {i.id for i in via.images} == {images[0].id, images[1].id}
+    excluded = await svc.scroll(cursor=None, limit=10, tag_exclude=[fandom.id])
+    assert {i.id for i in excluded.images} == {images[2].id, images[3].id}
+
+
 @pytest.mark.asyncio
 async def test_timeline_groups_by_month(db):
     await _seed_images(db, 3)

tests/test_tag_directory_service.py

@@ -36,6 +36,33 @@ async def test_directory_lists_tags_with_counts_and_previews(db):
     assert card["preview_thumbnails"][0].startswith("/images/thumbs/")
 
 
+@pytest.mark.asyncio
+async def test_directory_fandom_counts_and_previews_via_characters(db):
+    # A fandom card aggregates images of its characters (Tag.fandom_id), not
+    # just images tagged with the fandom directly. DISTINCT: an image carrying
+    # both the fandom and one of its characters counts once.
+    fandom = Tag(name="Bleach", kind=TagKind.fandom)
+    db.add(fandom)
+    await db.flush()
+    char = Tag(name="Ichigo", kind=TagKind.character, fandom_id=fandom.id)
+    db.add(char)
+    await db.flush()
+    # img0: character only; img1: fandom direct; img2: BOTH (dedup → counts once)
+    img0, img1, img2 = await _img(db, 0), await _img(db, 1), await _img(db, 2)
+    await db.execute(image_tag.insert().values([
+        {"image_record_id": img0.id, "tag_id": char.id, "source": "manual"},
+        {"image_record_id": img1.id, "tag_id": fandom.id, "source": "manual"},
+        {"image_record_id": img2.id, "tag_id": char.id, "source": "manual"},
+        {"image_record_id": img2.id, "tag_id": fandom.id, "source": "manual"},
+    ]))
+    await db.flush()
+    svc = TagDirectoryService(db)
+    page = await svc.list_tags(kind="fandom", q=None, cursor=None, limit=10)
+    card = next(c for c in page.cards if c["name"] == "Bleach")
+    assert card["image_count"] == 3
+    assert len(card["preview_thumbnails"]) == 3
+
+
 @pytest.mark.asyncio
 async def test_directory_kind_filter_and_search(db):
     db.add_all([