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([