fix(cleanup): live prune uses the same predicate as the preview (data loss)

The fandom/chapter exclusions added in fb05c5e only touched find_unused_tags
(the preview SAMPLE). prune_unused_tags re-implemented the predicate inline for
the dry-run COUNT and the live DELETE with only the image_tag + series_page
checks — so the preview showed a safe list of names while the delete removed
every fandom (and chaptered series). Operator-flagged 2026-06-08: real data loss
— assigned fandoms deleted, their characters SET-NULLed.

Extract _unused_tag_conditions() as the single source of truth and use it for
the preview, the count, AND the delete, so they can never diverge again. Added a
prune-commit test asserting the LIVE delete spares a character's fandom and a
chaptered series.

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

backend/app/services/cleanup_service.py

@@ -139,24 +139,25 @@ def count_tag_associations(session: Session, *, tag_id: int) -> int:
     ).scalar_one()
 
 
-def find_unused_tags(
-    session: Session, *, limit: int | None = None,
-) -> list[Tag]:
-    """Tags genuinely referenced by nothing — safe to sweep.
+def _unused_tag_conditions() -> list:
+    """The WHERE conditions that define an 'unused' tag — the SINGLE source of
+    truth shared by find_unused_tags (preview sample), the dry-run count, AND the
+    live delete, so the preview can NEVER diverge from what the delete removes.
 
-    Sorted by name. Used by both dry-run preview and the live prune. A tag is
-    "unused" iff it has zero references across ALL the ways a tag can be in use:
-      - image_tag           (applied to an image)
-      - series_page         (a series tag with ordered pages)
-      - series_chapter      (a series tag with chapters but no pages yet, e.g.
-                             all-placeholder)
-      - tag.fandom_id       (a fandom referenced by a character)
+    A tag is "unused" iff it has zero references across ALL the ways a tag can be
+    in use:
+      - image_tag       (applied to an image)
+      - series_page     (a series tag with ordered pages)
+      - series_chapter  (a series tag with chapters but no pages yet)
+      - tag.fandom_id   (a fandom referenced by a character)
 
     The fandom check is essential: fandom tags are NEVER applied to images — a
     character carries its fandom via fandom_id — so without it every assigned
-    fandom looked "unused", and the FK is ondelete=SET NULL, so deleting one
-    would silently strip the fandom off all its characters (operator-flagged
-    2026-06-08).
+    fandom looks "unused", and the FK is ondelete=SET NULL, so deleting one
+    silently strips the fandom off all its characters. The delete had only the
+    first two checks while the preview sample had all four, so the preview showed
+    a safe list but the delete removed every fandom anyway (operator-flagged
+    2026-06-08). Defining the predicate once makes that impossible.
     """
     used_via_image_tag = select(image_tag.c.tag_id).distinct()
     used_via_series = select(SeriesPage.series_tag_id).where(
@@ -166,14 +167,20 @@ def find_unused_tags(
     used_via_fandom = select(Tag.fandom_id).where(
         Tag.fandom_id.is_not(None)
     ).distinct()
-    stmt = (
-        select(Tag)
-        .where(Tag.id.not_in(used_via_image_tag))
-        .where(Tag.id.not_in(used_via_series))
-        .where(Tag.id.not_in(used_via_chapter))
-        .where(Tag.id.not_in(used_via_fandom))
-        .order_by(Tag.name)
-    )
+    return [
+        Tag.id.not_in(used_via_image_tag),
+        Tag.id.not_in(used_via_series),
+        Tag.id.not_in(used_via_chapter),
+        Tag.id.not_in(used_via_fandom),
+    ]
+
+
+def find_unused_tags(
+    session: Session, *, limit: int | None = None,
+) -> list[Tag]:
+    """Tags genuinely referenced by nothing — safe to sweep. Sorted by name.
+    Shares its predicate with the live prune via _unused_tag_conditions()."""
+    stmt = select(Tag).where(*_unused_tag_conditions()).order_by(Tag.name)
     if limit is not None:
         stmt = stmt.limit(limit)
     return list(session.execute(stmt).scalars().all())
@@ -381,30 +388,21 @@ def prune_unused_tags(session: Session, *, dry_run: bool = False) -> dict:
 
     Implementation note: the previous SELECT-ids → DELETE-WHERE-IN
     pattern was vulnerable to the psycopg 65535-parameter ceiling on
-    libraries with tag explosions. The live delete now runs a single
-    DELETE with the same NOT-IN predicate find_unused_tags uses, so
-    the row count scales without binding every id as a parameter.
-    Audit 2026-06-02.
+    libraries with tag explosions. The live delete runs a single DELETE
+    with the SAME predicate (_unused_tag_conditions) the preview uses, so
+    the row count scales without binding every id as a parameter AND the
+    delete can never remove a tag the preview deemed safe. Audit 2026-06-02;
+    predicate unified 2026-06-08.
     """
+    conditions = _unused_tag_conditions()
     sample_rows = find_unused_tags(session, limit=50)
     sample = [t.name for t in sample_rows]
-    used_via_image_tag = select(image_tag.c.tag_id).distinct()
-    used_via_series = select(SeriesPage.series_tag_id).where(
-        SeriesPage.series_tag_id.is_not(None)
-    ).distinct()
     if dry_run:
         count = session.execute(
-            select(func.count())
-            .select_from(Tag)
-            .where(Tag.id.not_in(used_via_image_tag))
-            .where(Tag.id.not_in(used_via_series))
+            select(func.count()).select_from(Tag).where(*conditions)
         ).scalar_one()
         return {"count": count, "sample_names": sample}
-    result = session.execute(
-        Tag.__table__.delete()
-        .where(Tag.id.not_in(used_via_image_tag))
-        .where(Tag.id.not_in(used_via_series))
-    )
+    result = session.execute(Tag.__table__.delete().where(*conditions))
     session.commit()
     return {"deleted": result.rowcount or 0, "sample_names": sample}
 

tests/test_cleanup_service.py

@@ -362,6 +362,30 @@ def test_prune_unused_tags_commit_deletes_them(db_sync, tmp_path):
     assert "bye" not in surviving_names
 
 
+def test_prune_unused_tags_commit_spares_fandom_and_chaptered_series(db_sync):
+    # The LIVE delete (not just the preview) must use the same predicate — it
+    # previously had only the image_tag/series_page checks and deleted every
+    # fandom the preview correctly excluded (operator-flagged 2026-06-08, real
+    # data loss). Both dry-run count and the commit must spare referenced tags.
+    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
+    series = _make_tag(db_sync, name="EmptySeries", kind=TagKind.series)
+    db_sync.add(SeriesChapter(series_tag_id=series.id, chapter_number=1))
+    _make_tag(db_sync, name="GenuinelyUnused")
+    db_sync.commit()
+
+    # dry-run count agrees with the delete: only the 1 truly-unused tag.
+    assert cleanup_service.prune_unused_tags(db_sync, dry_run=True)["count"] == 1
+    result = cleanup_service.prune_unused_tags(db_sync, dry_run=False)
+    assert result["deleted"] == 1
+
+    surviving = db_sync.execute(select(Tag.name)).scalars().all()
+    assert "Creux" in surviving          # fandom referenced by a character
+    assert "EmptySeries" in surviving    # series referenced by a chapter
+    assert "GenuinelyUnused" not in surviving
+
+
 # --- reset_content_tagging ------------------------------------------