feat(tags): non-mutating merge preview + admin dry_run (#8a)

Cluster B, milestone #99. TagService.merge_preview(source, target) computes the
same counts the apply produces (rule 93 parity) without mutating: images_moving
(source links the apply UPDATEs), images_already_on_target (links it drops),
source_total, series_pages, will_alias (_keep_as_alias), a kind/fandom
compatible flag (surfaced, not raised, so the UI can warn), and up to 6
thumbnails of the moving images. The admin /tags/<dest>/merge route gains a
dry_run flag returning the preview JSON.

Tests: preview moving-count == apply merged_count (parity), incompatible flagged
without raising, self/missing raise, admin dry_run returns preview + no mutation.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XCUHUGQLrBrkgyk1t49kpX

backend/app/api/admin.py

@@ -171,6 +171,30 @@ async def tag_merge(dest_id: int):
     if not isinstance(source_id, int) or source_id == dest_id:
         return _bad("invalid_source_id", detail="source_id must be int and differ from dest")
 
+    # dry_run: non-mutating preview (counts + sample) so the operator can
+    # confirm the target before the irreversible merge (#8, rule 93 parity).
+    if body.get("dry_run"):
+        async with get_session() as session:
+            try:
+                p = await TagService(session).merge_preview(
+                    source_id=source_id, target_id=dest_id,
+                )
+            except TagValidationError as exc:
+                return _bad("tag_not_found", status=404, detail=str(exc))
+        return jsonify({
+            "preview": {
+                "source_id": p.source_id, "source_name": p.source_name,
+                "target_id": p.target_id, "target_name": p.target_name,
+                "compatible": p.compatible,
+                "images_moving": p.images_moving,
+                "images_already_on_target": p.images_already_on_target,
+                "source_total": p.source_total,
+                "series_pages": p.series_pages,
+                "will_alias": p.will_alias,
+                "sample_thumbnails": p.sample_thumbnails,
+            },
+        })
+
     async with get_session() as session:
         try:
             result = await TagService(session).merge(

backend/app/services/tag_service.py

@@ -79,6 +79,23 @@ class MergeResult:
     source_deleted: bool
 
 
+@dataclass(frozen=True)
+class MergePreview:
+    """Non-mutating projection of what merge(source→target) would do (#8).
+    Shares the apply's predicates (rule 93) so the preview can't drift."""
+    source_id: int
+    source_name: str
+    target_id: int
+    target_name: str
+    compatible: bool                # same kind + fandom — would apply succeed?
+    images_moving: int              # source links that move (image lacks target)
+    images_already_on_target: int   # source links dropped (image has target)
+    source_total: int               # images carrying source
+    series_pages: int               # series pages repointed
+    will_alias: bool                # source name kept as a protective alias
+    sample_thumbnails: list[str]    # a few of the moving images
+
+
 class TagService:
     def __init__(self, session: AsyncSession):
         self.session = session
@@ -376,6 +393,96 @@ class TagService:
         await self.session.flush()
         return tag
 
+    async def merge_preview(
+        self, source_id: int, target_id: int
+    ) -> MergePreview:
+        """Read-only dry-run of merge(source→target): the same counts the apply
+        would produce, plus a thumbnail sample of the images that move. Raises
+        TagValidationError only for missing/self-merge (so the UI can render a
+        kind/fandom-mismatch warning rather than a hard error)."""
+        if source_id == target_id:
+            raise TagValidationError("Cannot merge a tag into itself")
+        source = await self.session.get(Tag, source_id)
+        target = await self.session.get(Tag, target_id)
+        if source is None or target is None:
+            raise TagValidationError("Tag not found")
+
+        compatible = source.kind == target.kind and (
+            (source.fandom_id or 0) == (target.fandom_id or 0)
+        )
+
+        # Mirrors _repoint_image_tags: links whose image already has target are
+        # dropped (already_on_target); the rest move.
+        target_images = select(image_tag.c.image_record_id).where(
+            image_tag.c.tag_id == target_id
+        )
+        moving = await self.session.scalar(
+            select(func.count())
+            .select_from(image_tag)
+            .where(
+                image_tag.c.tag_id == source_id,
+                image_tag.c.image_record_id.notin_(target_images),
+            )
+        )
+        already = await self.session.scalar(
+            select(func.count())
+            .select_from(image_tag)
+            .where(
+                image_tag.c.tag_id == source_id,
+                image_tag.c.image_record_id.in_(target_images),
+            )
+        )
+
+        from ..models.series_page import SeriesPage
+
+        series_pages = await self.session.scalar(
+            select(func.count())
+            .select_from(SeriesPage)
+            .where(SeriesPage.series_tag_id == source_id)
+        )
+
+        will_alias = await self._keep_as_alias(source_id)
+        sample = await self._merge_sample_thumbnails(source_id, target_images)
+
+        return MergePreview(
+            source_id=source_id,
+            source_name=source.name,
+            target_id=target_id,
+            target_name=target.name,
+            compatible=compatible,
+            images_moving=moving or 0,
+            images_already_on_target=already or 0,
+            source_total=(moving or 0) + (already or 0),
+            series_pages=series_pages or 0,
+            will_alias=will_alias,
+            sample_thumbnails=sample,
+        )
+
+    async def _merge_sample_thumbnails(
+        self, source_id: int, target_images
+    ) -> list[str]:
+        """Up to 6 thumbnails of the images that WOULD move (carry source, not
+        target) — a sanity check that the merge targets the right content."""
+        from ..models import ImageRecord
+        from .gallery_service import thumbnail_url
+
+        rows = (
+            await self.session.execute(
+                select(
+                    ImageRecord.thumbnail_path,
+                    ImageRecord.sha256,
+                    ImageRecord.mime,
+                )
+                .join(image_tag, image_tag.c.image_record_id == ImageRecord.id)
+                .where(
+                    image_tag.c.tag_id == source_id,
+                    image_tag.c.image_record_id.notin_(target_images),
+                )
+                .limit(6)
+            )
+        ).all()
+        return [thumbnail_url(tp, sha, mime) for tp, sha, mime in rows]
+
     async def merge(self, source_id: int, target_id: int) -> MergeResult:
         """Transactionally repoint every FK from source→target, optionally
         keep source's name as a tagger alias, delete source. Atomic: any