Merge pull request 'Provenance archive linkage, post reconciliation close-out, and the cleanup/admin DRY pass' (#127) from dev into main

Frontend pattern-consistency sweep (note #1026, the last DRY-thread item).
TagMaintenanceCard (4 flows) + PostMaintenanceCard (2 flows) each hand-rolled the
same sync preview→commit state machine: a previewData/previewing/committing
triple + onPreview/onCommit that dry-run-previews, then applies and collapses the
projection (the apply shares the backend predicate, so afterward it's empty).

Extract usePreviewCommit({preview, commit, emptyPreview}) owning that lifecycle.
The 6 flows become declarative: supply the two thunks + the collapse shape. The
normalize flow (commit dispatches a self-resuming background task, not a sync
apply) omits emptyPreview so the projection stays and a truthy result = queued.

Composable returns are aliased to the cards' existing local names, so the
templates only change where they read the apply result (the success badges).
Long-Celery-task cards (GatedPurge/VideoDedup) keep useMaintenanceTask — a
different pattern (navigable-away task lifecycle), deliberately not merged.

Exhaustiveness: no card hand-rolls the refs anymore; the only dryRun:false
callers are these two cards, both via the composable. Added a vitest spec for
the primitive (collapse static + fn, dispatch-variant, re-preview clears result).

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

alembic/versions/0055_image_provenance_from_attachment.py

@@ -0,0 +1,55 @@
+"""image_provenance: from_attachment_id (which archive an image was extracted from)
+
+Milestone #87. When an image is pulled out of a .zip/.rar, record WHICH archive
+PostAttachment it came from, so the provenance UI can show the single archive a
+file lives inside instead of every attachment on the post. Nullable FK with
+ON DELETE SET NULL — a loose (non-archive) download leaves it NULL, and deleting
+the archive attachment forgets the linkage without destroying the (image, post)
+provenance edge. Existing rows are NULL until the reextract backfill stamps them.
+
+Revision ID: 0055
+Revises: 0054
+Create Date: 2026-06-22
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0055"
+down_revision: Union[str, None] = "0054"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "image_provenance",
+        sa.Column("from_attachment_id", sa.Integer(), nullable=True),
+    )
+    op.create_index(
+        "ix_image_provenance_from_attachment_id",
+        "image_provenance",
+        ["from_attachment_id"],
+    )
+    op.create_foreign_key(
+        "fk_image_provenance_from_attachment",
+        "image_provenance",
+        "post_attachment",
+        ["from_attachment_id"],
+        ["id"],
+        ondelete="SET NULL",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint(
+        "fk_image_provenance_from_attachment",
+        "image_provenance",
+        type_="foreignkey",
+    )
+    op.drop_index(
+        "ix_image_provenance_from_attachment_id",
+        table_name="image_provenance",
+    )
+    op.drop_column("image_provenance", "from_attachment_id")

backend/app/api/admin.py

@@ -39,6 +39,31 @@ def _bulk_image_confirm_token(image_ids: list[int]) -> str:
     return digest[:8]
 
 
+async def _run_dry_run_op(service_fn, **service_kwargs):
+    """Shared body for the Tier-A dry-run/apply endpoints: read the `dry_run`
+    flag, run the cleanup_service predicate under `run_sync`, and return its
+    result dict. The SAME `service_fn` drives both preview and apply (the flag
+    just toggles), so a handler physically can't let its preview diverge from
+    its delete (rule 93). Default False preserves the existing contract — the UI
+    always passes `dry_run` explicitly (true to preview, false to apply). Extra
+    service kwargs (e.g. `source_id`) pass straight through."""
+    body = await request.get_json(silent=True) or {}
+    dry_run = bool(body.get("dry_run", False))
+    async with get_session() as session:
+        result = await session.run_sync(
+            lambda sync_sess: service_fn(sync_sess, dry_run=dry_run, **service_kwargs)
+        )
+    return jsonify(result)
+
+
+def _queued(async_result):
+    """Standard 202 for an operator-triggered maintenance task: hand the UI the
+    Celery task id so it can tail /maintenance/task-result (or the activity
+    dashboard) for the summary. (trigger_vacuum stays bespoke — the UI doesn't
+    poll it, so it returns no task id.)"""
+    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+
+
 @admin_bp.route("/artists/<slug>/cascade-delete", methods=["POST"])
 async def artist_cascade_delete(slug: str):
     body = await request.get_json(silent=True) or {}
@@ -193,16 +218,7 @@ async def tags_prune_unused():
     re-call with dry_run=false."""
     from ..services.cleanup_service import prune_unused_tags
 
-    body = await request.get_json(silent=True) or {}
-    dry_run = bool(body.get("dry_run", False))
-
-    async with get_session() as session:
-        result = await session.run_sync(
-            lambda sync_sess: prune_unused_tags(
-                sync_sess, dry_run=dry_run,
-            )
-        )
-    return jsonify(result)
+    return await _run_dry_run_op(prune_unused_tags)
 
 
 @admin_bp.route("/posts/prune-bare", methods=["POST"])
@@ -214,16 +230,7 @@ async def posts_prune_bare():
     prune itself, so the preview can't diverge from the delete."""
     from ..services.cleanup_service import prune_bare_posts
 
-    body = await request.get_json(silent=True) or {}
-    dry_run = bool(body.get("dry_run", False))
-
-    async with get_session() as session:
-        result = await session.run_sync(
-            lambda sync_sess: prune_bare_posts(
-                sync_sess, dry_run=dry_run,
-            )
-        )
-    return jsonify(result)
+    return await _run_dry_run_op(prune_bare_posts)
 
 
 @admin_bp.route("/posts/reconcile-duplicates", methods=["POST"])
@@ -237,20 +244,13 @@ async def posts_reconcile_duplicates():
     from ..services.cleanup_service import reconcile_duplicate_posts
 
     body = await request.get_json(silent=True) or {}
-    dry_run = bool(body.get("dry_run", False))
     raw_source = body.get("source_id")
     try:
         source_id = int(raw_source) if raw_source is not None else None
     except (TypeError, ValueError):
         return _bad("invalid_source_id", detail="source_id must be an integer")
 
-    async with get_session() as session:
-        result = await session.run_sync(
-            lambda sync_sess: reconcile_duplicate_posts(
-                sync_sess, source_id=source_id, dry_run=dry_run,
-            )
-        )
-    return jsonify(result)
+    return await _run_dry_run_op(reconcile_duplicate_posts, source_id=source_id)
 
 
 @admin_bp.route("/tags/purge-legacy", methods=["POST"])
@@ -263,14 +263,7 @@ async def tags_purge_legacy():
     operator confirms with dry_run=false."""
     from ..services.cleanup_service import purge_legacy_tags
 
-    body = await request.get_json(silent=True) or {}
-    dry_run = bool(body.get("dry_run", False))
-
-    async with get_session() as session:
-        result = await session.run_sync(
-            lambda sync_sess: purge_legacy_tags(sync_sess, dry_run=dry_run)
-        )
-    return jsonify(result)
+    return await _run_dry_run_op(purge_legacy_tags)
 
 
 @admin_bp.route("/tags/reset-content", methods=["POST"])
@@ -284,14 +277,7 @@ async def tags_reset_content():
     Irreversible except via DB backup restore."""
     from ..services.cleanup_service import reset_content_tagging
 
-    body = await request.get_json(silent=True) or {}
-    dry_run = bool(body.get("dry_run", False))
-
-    async with get_session() as session:
-        result = await session.run_sync(
-            lambda sync_sess: reset_content_tagging(sync_sess, dry_run=dry_run)
-        )
-    return jsonify(result)
+    return await _run_dry_run_op(reset_content_tagging)
 
 
 @admin_bp.route("/tags/normalize", methods=["POST"])
@@ -317,7 +303,7 @@ async def tags_normalize():
     from ..tasks.admin import normalize_tags_task
 
     async_result = normalize_tags_task.delay()
-    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+    return _queued(async_result)
 
 
 @admin_bp.route("/maintenance/db-stats", methods=["GET"])
@@ -374,7 +360,7 @@ async def trigger_reextract_archives():
     from ..tasks.admin import reextract_archive_attachments_task
 
     async_result = reextract_archive_attachments_task.delay()
-    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+    return _queued(async_result)
 
 
 @admin_bp.route("/maintenance/prune-missing-files", methods=["POST"])
@@ -387,7 +373,7 @@ async def trigger_prune_missing_files():
     from ..tasks.admin import prune_missing_file_records_task
 
     async_result = prune_missing_file_records_task.delay()
-    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+    return _queued(async_result)
 
 
 @admin_bp.route("/maintenance/dedup-videos", methods=["POST"])
@@ -403,7 +389,7 @@ async def trigger_dedup_videos():
     body = await request.get_json(silent=True) or {}
     dry_run = bool(body.get("dry_run", True))  # default to the SAFE preview
     async_result = dedup_videos_task.delay(dry_run=dry_run)
-    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+    return _queued(async_result)
 
 
 @admin_bp.route("/maintenance/purge-gated-previews", methods=["POST"])
@@ -419,7 +405,7 @@ async def trigger_purge_gated_previews():
     body = await request.get_json(silent=True) or {}
     dry_run = bool(body.get("dry_run", True))  # default to the SAFE preview
     async_result = purge_gated_previews_task.delay(dry_run=dry_run)
-    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+    return _queued(async_result)
 
 
 @admin_bp.route("/maintenance/task-result/<task_id>", methods=["GET"])

backend/app/models/image_provenance.py

@@ -41,6 +41,16 @@ class ImageProvenance(Base):
     source_id: Mapped[int | None] = mapped_column(
         ForeignKey("source.id", ondelete="SET NULL"), nullable=True, index=True
     )
+    # The archive PostAttachment this image was extracted FROM, when it came
+    # out of a .zip/.rar rather than as a loose file (milestone #87). Lets the
+    # provenance UI show the exact archive a file lives inside instead of every
+    # attachment on the post. NULL for loose downloads and pre-backfill rows.
+    # SET NULL so deleting the archive attachment never destroys the (image,
+    # post) edge — it just forgets which archive it came from.
+    from_attachment_id: Mapped[int | None] = mapped_column(
+        ForeignKey("post_attachment.id", ondelete="SET NULL"),
+        nullable=True, index=True,
+    )
     captured_metadata: Mapped[dict | None] = mapped_column(JSON, nullable=True)
     captured_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()

backend/app/services/cleanup_service.py

@@ -573,6 +573,29 @@ def _repoint_post_links(session: Session, loser_id: int, keeper_id: int) -> None
     dup_imgs = select(ImageProvenance.image_record_id).where(
         ImageProvenance.post_id == keeper_id
     )
+    # Before dropping the colliding loser rows, carry their from_attachment_id
+    # (which archive the file came out of, milestone #87) onto the keeper's
+    # surviving row when the keeper didn't record one. For the gallery-dl→native
+    # case this very milestone targets, the keeper is the native stub (no
+    # archive) and the loser is the gallery-dl row that extracted the member, so
+    # a blind delete would silently lose the containing-archive linkage.
+    for img_id, att_id in session.execute(
+        select(ImageProvenance.image_record_id, ImageProvenance.from_attachment_id)
+        .where(
+            ImageProvenance.post_id == loser_id,
+            ImageProvenance.image_record_id.in_(dup_imgs),
+            ImageProvenance.from_attachment_id.is_not(None),
+        )
+    ).all():
+        session.execute(
+            update(ImageProvenance)
+            .where(
+                ImageProvenance.post_id == keeper_id,
+                ImageProvenance.image_record_id == img_id,
+                ImageProvenance.from_attachment_id.is_(None),
+            )
+            .values(from_attachment_id=att_id)
+        )
     session.execute(
         delete(ImageProvenance).where(
             ImageProvenance.post_id == loser_id,

backend/app/services/importer.py

@@ -17,7 +17,7 @@ from enum import StrEnum
 from pathlib import Path
 
 from PIL import Image
-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -506,6 +506,12 @@ class Importer:
         artist_use = artist if artist is not None else self._resolve_artist(source)
         post = self._post_for_sidecar(source, artist_use)
         member_ids: list[int] = []
+        # Every member image touched (new + superseded + deduped), so the
+        # from_attachment_id stamp below covers files that already existed in the
+        # library and were merely re-linked to this post — those matter most
+        # (the HR copy a bundle re-ships). Separate from member_ids, which is
+        # the NEWLY-imported subset feeding the ImportResult contract.
+        member_record_ids: set[int] = set()
         # Per-outcome tally so the "no images" reason names the ACTUAL cause
         # (#718): nested-archive packs, all-deduped (benign), unsupported formats,
         # or failed/corrupt members — instead of one catch-all string.
@@ -516,11 +522,21 @@ class Importer:
         self._collect_archive_members(
             source, attribution=source, source_row=source_row,
             depth=0, member_ids=member_ids, counts=counts,
+            member_record_ids=member_record_ids,
         )
         # Preserve the archive itself (links to the same Post/Artist).
         self._capture_attachment(
             source, post=post, artist=artist_use, resolved=True
         )
+        # Stamp each member's provenance row for THIS post with the archive it
+        # came out of (milestone #87). Done as a post-pass rather than threaded
+        # through _import_media/_apply_sidecar so the many dedup/supersede
+        # branches stay untouched. NULL-only so a re-extract never re-stamps and
+        # the backfill (reextract task → this same path) is idempotent. Nested
+        # members link to this OUTER archive — the only one stored as a blob.
+        self._stamp_member_archive(
+            post.id if post is not None else None, source, member_record_ids,
+        )
         if member_ids:
             return ImportResult(
                 status="imported", image_id=member_ids[0],
@@ -555,6 +571,7 @@ class Importer:
         self, archive_path: Path, *, attribution: Path,
         source_row: Source | None, depth: int,
         member_ids: list[int], counts: dict,
+        member_record_ids: set[int],
     ) -> None:
         """Extract `archive_path` and import its image/video members, RECURSING
         into nested archives (#718). Members attribute to `attribution` — the
@@ -590,6 +607,7 @@ class Importer:
                             member_path, attribution=attribution,
                             source_row=source_row, depth=depth + 1,
                             member_ids=member_ids, counts=counts,
+                            member_record_ids=member_record_ids,
                         )
                         continue
                     counts["media"] += 1
@@ -601,10 +619,16 @@ class Importer:
                     )
                     if res.status in ("imported", "superseded") and res.image_id:
                         member_ids.append(res.image_id)
+                        member_record_ids.add(res.image_id)
                     elif res.status == "skipped" and res.skip_reason in (
                         SkipReason.duplicate_hash, SkipReason.duplicate_phash
                     ):
                         counts["deduped"] += 1
+                        # A deduped member still links provenance to this post
+                        # (enrich-on-duplicate); record it so its archive origin
+                        # gets stamped too.
+                        if res.image_id:
+                            member_record_ids.add(res.image_id)
                     else:
                         counts["failed"] += 1
         except Exception as exc:  # noqa: BLE001 — defensive per level; keep going
@@ -613,6 +637,39 @@ class Importer:
                 archive_path.name, depth, exc,
             )
 
+    def _stamp_member_archive(
+        self, post_id: int | None, archive_source: Path, member_record_ids: set[int],
+    ) -> None:
+        """Record which archive each extracted member came from (milestone #87).
+
+        Resolves the archive's own PostAttachment (by post + sha — it was just
+        captured) and stamps from_attachment_id on every member's provenance row
+        FOR THIS POST. NULL-only, so re-extracting the same archive (the backfill
+        path) never overwrites and stays idempotent. No-op when the archive isn't
+        post-attached (filesystem import with no post) or yielded no members.
+        """
+        if post_id is None or not member_record_ids:
+            return
+        sha = _sha256_of(archive_source)
+        att_id = self.session.execute(
+            select(PostAttachment.id).where(
+                PostAttachment.post_id == post_id,
+                PostAttachment.sha256 == sha,
+            )
+        ).scalar_one_or_none()
+        if att_id is None:
+            return
+        self.session.execute(
+            update(ImageProvenance)
+            .where(
+                ImageProvenance.image_record_id.in_(member_record_ids),
+                ImageProvenance.post_id == post_id,
+                ImageProvenance.from_attachment_id.is_(None),
+            )
+            .values(from_attachment_id=att_id)
+        )
+        self.session.commit()
+
     @staticmethod
     def _video_aspect_matches(w, h, cw, ch) -> bool:
         """True when two (w,h) pairs share an aspect ratio within tolerance.

backend/app/services/provenance_service.py

@@ -66,6 +66,10 @@ class ProvenanceService:
         ).scalars().all()
         return [_attachment_dict(a) for a in rows]
 
+    async def _attachment_by_id(self, attachment_id: int) -> list[dict]:
+        att = await self.session.get(PostAttachment, attachment_id)
+        return [_attachment_dict(att)] if att is not None else []
+
     async def for_image(self, image_id: int) -> dict | None:
         rec = await self.session.get(ImageRecord, image_id)
         if rec is None:
@@ -85,7 +89,33 @@ class ProvenanceService:
         )
         rows = (await self.session.execute(stmt)).all()
         post_ids = [ip.post_id for ip, _p, _s, _a in rows]
-        attachments = await self._attachments_for_posts(post_ids)
+        # Prefer the EXACT archive this file came out of (milestone #87): if the
+        # originating post's provenance row records from_attachment_id, the image
+        # was extracted from that one .zip/.rar, so show only it — not the dozens
+        # of unrelated archives a "High Resolution Files" bundle post carries.
+        from_att_id = next(
+            (
+                ip.from_attachment_id
+                for ip, _p, _s, _a in rows
+                if ip.post_id == rec.primary_post_id
+                and ip.from_attachment_id is not None
+            ),
+            None,
+        )
+        if from_att_id is not None:
+            attachments = await self._attachment_by_id(from_att_id)
+        else:
+            # No recorded containing archive (loose download, or pre-backfill):
+            # scope to the originating post only, not every pHash-linked post.
+            # primary_post_id is the post this file was actually captured from;
+            # fall back to all linked posts when it's unset (older rows /
+            # filesystem imports).
+            attach_post_ids = (
+                [rec.primary_post_id]
+                if rec.primary_post_id is not None
+                else post_ids
+            )
+            attachments = await self._attachments_for_posts(attach_post_ids)
         return {
             "image_id": image_id,
             "provenance": [

frontend/src/components/modal/ProvenancePanel.vue

@@ -70,7 +70,13 @@
     </div>
 
     <div v-if="attachments.length" class="fc-prov__attach">
-      <h4 class="fc-prov__attach-title">Attachments</h4>
+      <h4 class="fc-prov__attach-title">
+        {{ attachments.length === 1 ? 'Attachment' : `Attachments (${attachments.length})` }}
+      </h4>
+      <!-- Scroll-capped: a single post can carry dozens of archives (HR
+           bundle posts), which previously ballooned the panel past the
+           viewport. Mirror the cards' independent-scroll treatment. -->
+      <div class="fc-prov__attach-list">
         <a
           v-for="at in attachments" :key="at.id"
           class="fc-prov__attach-row"
@@ -79,6 +85,7 @@
           <span class="fc-prov__attach-size">({{ at.size_bytes }} B)</span>
         </a>
       </div>
+    </div>
   </section>
 </template>
 
@@ -229,6 +236,15 @@ function openPost(postId, artistId) {
   font-size: 13px; color: rgb(var(--v-theme-on-surface-variant));
   margin: 8px 0 4px;
 }
+.fc-prov__attach-list {
+  /* ~7 rows visible before scrolling; the clipped row hints at more.
+     Hairline scrollbar matching .fc-prov__cards. */
+  max-height: 180px;
+  overflow-y: auto;
+  scrollbar-width: thin;
+  scrollbar-color: rgb(var(--v-theme-surface-light)) transparent;
+  padding-right: 4px;
+}
 .fc-prov__attach-row {
   display: block; font-size: 13px; text-decoration: none;
   color: rgb(var(--v-theme-accent)); padding: 2px 0;

frontend/src/components/settings/PostMaintenanceCard.vue

@@ -42,8 +42,8 @@
         :loading="committing"
         @click="onCommit"
       >Delete {{ preview.count }} bare post(s)</v-btn>
-      <span v-if="deleted != null" class="ml-3 text-caption text-success">
-        Deleted {{ deleted }} ✓
+      <span v-if="bareResult" class="ml-3 text-caption text-success">
+        Deleted {{ bareResult.deleted ?? 0 }} ✓
       </span>
     </div>
   </MaintenanceTile>
@@ -88,78 +88,47 @@
         :loading="merging"
         @click="onMergeDupes"
       >Merge {{ dupPreview.posts_to_merge }} duplicate(s)</v-btn>
-      <span v-if="merged != null" class="ml-3 text-caption text-success">
-        Merged {{ merged }} ✓
+      <span v-if="dupResult" class="ml-3 text-caption text-success">
+        Merged {{ dupResult.merged ?? 0 }} ✓
       </span>
     </div>
   </MaintenanceTile>
 </template>
 
 <script setup>
-import { computed, ref } from 'vue'
+import { computed } from 'vue'
 
 import { useAdminStore } from '../../stores/admin.js'
+import { usePreviewCommit } from '../../composables/usePreviewCommit.js'
 import SampleNameGrid from '../common/SampleNameGrid.vue'
 import MaintenanceTile from '../common/MaintenanceTile.vue'
 
 const store = useAdminStore()
-const preview = ref(null)
-const loadingPreview = ref(false)
-const committing = ref(false)
-const deleted = ref(null)
 
-const dupPreview = ref(null)
-const loadingDupPreview = ref(false)
-const merging = ref(false)
-const merged = ref(null)
+// Bare-post prune: preview the count, then delete. After the apply the same
+// predicate is empty, so collapse to count 0.
+const {
+  previewData: preview, previewing: loadingPreview, committing,
+  result: bareResult, runPreview: onPreview, runCommit: onCommit,
+} = usePreviewCommit({
+  preview: () => store.pruneBarePosts({ dryRun: true }),
+  commit: () => store.pruneBarePosts({ dryRun: false }),
+  emptyPreview: { count: 0, sample_names: [] },
+})
+
+// Duplicate-post reconcile: same shape, collapse to zero groups after merge.
+const {
+  previewData: dupPreview, previewing: loadingDupPreview, committing: merging,
+  result: dupResult, runPreview: onPreviewDupes, runCommit: onMergeDupes,
+} = usePreviewCommit({
+  preview: () => store.reconcileDuplicatePosts({ dryRun: true }),
+  commit: () => store.reconcileDuplicatePosts({ dryRun: false }),
+  emptyPreview: { groups: 0, posts_to_merge: 0, sample: [] },
+})
+
 const dupSampleNames = computed(() =>
   (dupPreview.value?.sample ?? []).map(
     (g) => `${g.title} — ${g.rows} rows`,
   ),
 )
-
-async function onPreview() {
-  loadingPreview.value = true
-  deleted.value = null
-  try {
-    preview.value = await store.pruneBarePosts({ dryRun: true })
-  } finally {
-    loadingPreview.value = false
-  }
-}
-
-async function onCommit() {
-  committing.value = true
-  try {
-    const result = await store.pruneBarePosts({ dryRun: false })
-    deleted.value = result.deleted ?? 0
-    // Reflect the completed sweep — the predicate is identical to the preview,
-    // so after a commit there is nothing left to delete.
-    preview.value = { count: 0, sample_names: [] }
-  } finally {
-    committing.value = false
-  }
-}
-
-async function onPreviewDupes() {
-  loadingDupPreview.value = true
-  merged.value = null
-  try {
-    dupPreview.value = await store.reconcileDuplicatePosts({ dryRun: true })
-  } finally {
-    loadingDupPreview.value = false
-  }
-}
-
-async function onMergeDupes() {
-  merging.value = true
-  try {
-    const result = await store.reconcileDuplicatePosts({ dryRun: false })
-    merged.value = result.merged ?? 0
-    // Same predicate as the preview — after the merge there are no dup groups left.
-    dupPreview.value = { groups: 0, posts_to_merge: 0, sample: [] }
-  } finally {
-    merging.value = false
-  }
-}
 </script>

frontend/src/components/settings/TagMaintenanceCard.vue

@@ -186,11 +186,11 @@
       <v-btn
         color="error" variant="flat" rounded="pill"
         prepend-icon="mdi-format-letter-case"
-        :disabled="!normPreview.total_changes || normResult === 'queued'"
+        :disabled="!normPreview.total_changes || !!normResult"
         :loading="normCommitting"
         @click="onNormCommit"
       >Standardize {{ normPreview.total_changes }} tag group(s)</v-btn>
-      <span v-if="normResult === 'queued'" class="ml-3 text-caption text-success">
+      <span v-if="normResult" class="ml-3 text-caption text-success">
         Queued ✓ — runs in the background (you can leave this page). It
         processes in chunks, so re-run “Preview” later to confirm it's all done.
       </span>
@@ -199,106 +199,53 @@
 </template>
 
 <script setup>
-import { ref } from 'vue'
-
 import { useAdminStore } from '../../stores/admin.js'
+import { usePreviewCommit } from '../../composables/usePreviewCommit.js'
 import SampleNameGrid from '../common/SampleNameGrid.vue'
 import MaintenanceTile from '../common/MaintenanceTile.vue'
 
 const store = useAdminStore()
-const preview = ref(null)
-const loadingPreview = ref(false)
-const committing = ref(false)
-const kindPreview = ref(null)
-const loadingKindPreview = ref(false)
-const kindCommitting = ref(false)
-const resetPreview = ref(null)
-const loadingResetPreview = ref(false)
-const resetCommitting = ref(false)
-const normPreview = ref(null)
-const loadingNormPreview = ref(false)
-const normCommitting = ref(false)
-const normResult = ref(null)
 
-async function onPreview() {
-  loadingPreview.value = true
-  try {
-    preview.value = await store.pruneUnusedTags({ dryRun: true })
-  } finally {
-    loadingPreview.value = false
-  }
-}
+// Unused-tag prune. Collapse to count 0 but carry the apply's sample_names.
+const {
+  previewData: preview, previewing: loadingPreview, committing,
+  runPreview: onPreview, runCommit: onCommit,
+} = usePreviewCommit({
+  preview: () => store.pruneUnusedTags({ dryRun: true }),
+  commit: () => store.pruneUnusedTags({ dryRun: false }),
+  emptyPreview: (r) => ({ count: 0, sample_names: r.sample_names || [] }),
+})
 
-async function onCommit() {
-  committing.value = true
-  try {
-    const result = await store.pruneUnusedTags({ dryRun: false })
-    preview.value = { count: 0, sample_names: result.sample_names || [] }
-  } finally {
-    committing.value = false
-  }
-}
+// Legacy migration-tag purge.
+const {
+  previewData: kindPreview, previewing: loadingKindPreview,
+  committing: kindCommitting, runPreview: onKindPreview, runCommit: onKindCommit,
+} = usePreviewCommit({
+  preview: () => store.purgeLegacyTags({ dryRun: true }),
+  commit: () => store.purgeLegacyTags({ dryRun: false }),
+  emptyPreview: { count: 0, by_kind: {}, by_prefix: {}, sample_names: [] },
+})
 
-async function onKindPreview() {
-  loadingKindPreview.value = true
-  try {
-    kindPreview.value = await store.purgeLegacyTags({ dryRun: true })
-  } finally {
-    loadingKindPreview.value = false
-  }
-}
+// Reset content tagging (general + character).
+const {
+  previewData: resetPreview, previewing: loadingResetPreview,
+  committing: resetCommitting, runPreview: onResetPreview, runCommit: onResetCommit,
+} = usePreviewCommit({
+  preview: () => store.resetContentTagging({ dryRun: true }),
+  commit: () => store.resetContentTagging({ dryRun: false }),
+  emptyPreview: { count: 0, by_kind: {}, applications: 0, sample_names: [] },
+})
 
-async function onKindCommit() {
-  kindCommitting.value = true
-  try {
-    await store.purgeLegacyTags({ dryRun: false })
-    kindPreview.value = { count: 0, by_kind: {}, by_prefix: {}, sample_names: [] }
-  } finally {
-    kindCommitting.value = false
-  }
-}
-
-async function onResetPreview() {
-  loadingResetPreview.value = true
-  try {
-    resetPreview.value = await store.resetContentTagging({ dryRun: true })
-  } finally {
-    loadingResetPreview.value = false
-  }
-}
-
-async function onResetCommit() {
-  resetCommitting.value = true
-  try {
-    await store.resetContentTagging({ dryRun: false })
-    resetPreview.value = { count: 0, by_kind: {}, applications: 0, sample_names: [] }
-  } finally {
-    resetCommitting.value = false
-  }
-}
-
-async function onNormPreview() {
-  loadingNormPreview.value = true
-  normResult.value = null
-  try {
-    normPreview.value = await store.normalizeTags({ dryRun: true })
-  } finally {
-    loadingNormPreview.value = false
-  }
-}
-
-async function onNormCommit() {
-  normCommitting.value = true
-  normResult.value = null
-  try {
-    // Fire-and-forget: the task is time-boxed and self-resuming across chunks
-    // (a large back-catalog can't finish in one run), so we DON'T poll-until-
-    // done — that would falsely report "complete" after the first chunk. Just
-    // confirm it's queued; the operator can re-run Preview later to verify.
-    await store.normalizeTags({ dryRun: false })
-    normResult.value = 'queued'
-  } finally {
-    normCommitting.value = false
-  }
-}
+// Standardize casing. The apply DISPATCHES a self-resuming background task (no
+// poll-until-done — that would falsely report complete after the first chunk),
+// so there's no emptyPreview: leave the projection up; a truthy normResult means
+// "queued". The operator re-runs Preview later to confirm it's all done.
+const {
+  previewData: normPreview, previewing: loadingNormPreview,
+  committing: normCommitting, result: normResult,
+  runPreview: onNormPreview, runCommit: onNormCommit,
+} = usePreviewCommit({
+  preview: () => store.normalizeTags({ dryRun: true }),
+  commit: () => store.normalizeTags({ dryRun: false }),
+})
 </script>

frontend/src/composables/usePreviewCommit.js

@@ -0,0 +1,52 @@
+import { ref } from 'vue'
+
+// Canonical sync preview→commit flow for a maintenance tile: the operator
+// previews (dry-run) → sees the projection → commits (apply), where the apply
+// returns its result inline (no long Celery task). Owns the shared lifecycle —
+// the two loading flags, the preview payload, and the apply result — so each
+// tile stops hand-rolling its own `preview`/`loading`/`committing` refs +
+// onPreview/onCommit handlers (this was duplicated across 6 maintenance flows;
+// DRY pass #753).
+//
+// The tile supplies the `preview`/`commit` thunks (which call the right store
+// action with dryRun true/false) and, optionally, `emptyPreview`: the shape to
+// collapse the preview to after a successful apply — the apply uses the SAME
+// backend predicate as the preview, so afterward there's nothing left. Pass a
+// function to derive it from the apply result; omit it entirely for a commit
+// that dispatches a background task and should leave the preview in place.
+//
+// Errors surface via the store's own lastError (shown in the tile's alert), so
+// they're intentionally NOT captured here. For an apply that runs as a LONG
+// Celery task the operator can navigate away from, use useMaintenanceTask.
+export function usePreviewCommit ({ preview, commit, emptyPreview = null }) {
+  const previewData = ref(null)
+  const previewing = ref(false)
+  const committing = ref(false)
+  const result = ref(null)
+
+  async function runPreview () {
+    previewing.value = true
+    result.value = null  // clear any prior apply badge when re-previewing
+    try {
+      previewData.value = await preview()
+    } finally {
+      previewing.value = false
+    }
+  }
+
+  async function runCommit () {
+    committing.value = true
+    try {
+      result.value = await commit()
+      if (emptyPreview !== null) {
+        previewData.value = typeof emptyPreview === 'function'
+          ? emptyPreview(result.value)
+          : emptyPreview
+      }
+    } finally {
+      committing.value = false
+    }
+  }
+
+  return { previewData, previewing, committing, result, runPreview, runCommit }
+}

frontend/src/stores/admin.js

@@ -7,186 +7,114 @@ export const useAdminStore = defineStore('admin', () => {
   const api = useApi()
   const lastError = ref(null)
 
+  // Every admin action runs through _guard: reset lastError, run the call,
+  // surface its message to lastError on failure, and re-throw so callers still
+  // see the rejection. One copy of the capture/rethrow instead of 13.
+  async function _guard(fn) {
+    lastError.value = null
+    try {
+      return await fn()
+    } catch (e) {
+      lastError.value = e.message
+      throw e
+    }
+  }
+
+  // The Tier-A maintenance endpoints share one shape: POST a dry_run flag
+  // (default the SAFE preview), optionally with extra body fields. The backend's
+  // same predicate drives preview + apply, so the UI just toggles dryRun.
+  function _dryRunPost(url, { dryRun = true, ...extra } = {}) {
+    return _guard(() => api.post(url, { body: { dry_run: dryRun, ...extra } }))
+  }
+
   // --- Tier-C: artist cascade ---------------------------------------
 
-  async function projectArtistCascade(slug) {
-    lastError.value = null
-    try {
-      return await api.post(
+  function projectArtistCascade(slug) {
+    return _guard(() => api.post(
       `/api/admin/artists/${encodeURIComponent(slug)}/cascade-delete`,
       { body: { dry_run: true } },
-      )
-    } catch (e) {
-      lastError.value = e.message
-      throw e
-    }
+    ))
   }
 
-  async function dispatchArtistCascade(slug, confirm) {
-    lastError.value = null
-    try {
-      return await api.post(
+  function dispatchArtistCascade(slug, confirm) {
+    return _guard(() => api.post(
       `/api/admin/artists/${encodeURIComponent(slug)}/cascade-delete`,
       { body: { dry_run: false, confirm } },
-      )
-    } catch (e) {
-      lastError.value = e.message
-      throw e
-    }
+    ))
   }
 
   // --- Tier-C: bulk image delete ------------------------------------
 
-  async function projectBulkImageDelete(imageIds) {
-    lastError.value = null
-    try {
-      return await api.post(
+  function projectBulkImageDelete(imageIds) {
+    return _guard(() => api.post(
       '/api/admin/images/bulk-delete',
       { body: { image_ids: imageIds, dry_run: true } },
-      )
-    } catch (e) {
-      lastError.value = e.message
-      throw e
-    }
+    ))
   }
 
-  async function dispatchBulkImageDelete(imageIds, confirm) {
-    lastError.value = null
-    try {
-      return await api.post(
+  function dispatchBulkImageDelete(imageIds, confirm) {
+    return _guard(() => api.post(
       '/api/admin/images/bulk-delete',
       { body: { image_ids: imageIds, dry_run: false, confirm } },
-      )
-    } catch (e) {
-      lastError.value = e.message
-      throw e
-    }
+    ))
   }
 
   // --- Tier-B: tag delete + merge -----------------------------------
 
-  async function deleteTag(tagId) {
-    lastError.value = null
-    try {
-      return await api.delete(`/api/admin/tags/${tagId}`)
-    } catch (e) {
-      lastError.value = e.message
-      throw e
-    }
+  function deleteTag(tagId) {
+    return _guard(() => api.delete(`/api/admin/tags/${tagId}`))
   }
 
-  async function mergeTags(destId, sourceId) {
-    lastError.value = null
-    try {
-      return await api.post(
+  function mergeTags(destId, sourceId) {
+    return _guard(() => api.post(
       `/api/admin/tags/${destId}/merge`,
       { body: { source_id: sourceId } },
-      )
-    } catch (e) {
-      lastError.value = e.message
-      throw e
-    }
+    ))
   }
 
   async function tagUsageCount(tagId) {
-    lastError.value = null
-    try {
+    return _guard(async () => {
       const body = await api.get(`/api/admin/tags/${tagId}/usage-count`)
       return body.count
-    } catch (e) {
-      lastError.value = e.message
-      throw e
-    }
+    })
   }
 
-  // --- Tier-A: prune unused -----------------------------------------
+  // --- Tier-A: dry-run/apply maintenance ----------------------------
 
-  async function pruneUnusedTags({ dryRun = true } = {}) {
-    lastError.value = null
-    try {
-      return await api.post(
-        '/api/admin/tags/prune-unused',
-        { body: { dry_run: dryRun } },
-      )
-    } catch (e) {
-      lastError.value = e.message
-      throw e
-    }
+  function pruneUnusedTags(opts = {}) {
+    return _dryRunPost('/api/admin/tags/prune-unused', opts)
   }
 
-  // Tier-A: delete bare posts (no images + no attachments) — the empty-post
-  // flood shells. Preview/apply parity on the backend; UI previews then confirms.
-  async function pruneBarePosts({ dryRun = true } = {}) {
-    lastError.value = null
-    try {
-      return await api.post(
-        '/api/admin/posts/prune-bare',
-        { body: { dry_run: dryRun } },
-      )
-    } catch (e) {
-      lastError.value = e.message
-      throw e
-    }
+  // Delete bare posts (no images + no attachments) — the empty-post flood
+  // shells. Preview/apply parity on the backend; UI previews then confirms.
+  function pruneBarePosts(opts = {}) {
+    return _dryRunPost('/api/admin/posts/prune-bare', opts)
   }
 
-  // Tier-A: unify duplicate post rows (gallery-dl attachment-id + native post-id
-  // for the same real post) onto one post-id-keyed keeper. Images untouched.
-  // Preview/apply parity on the backend; UI previews then confirms.
-  async function reconcileDuplicatePosts({ dryRun = true, sourceId = null } = {}) {
-    lastError.value = null
-    try {
-      return await api.post(
-        '/api/admin/posts/reconcile-duplicates',
-        { body: { dry_run: dryRun, source_id: sourceId } },
-      )
-    } catch (e) {
-      lastError.value = e.message
-      throw e
-    }
+  // Unify duplicate post rows (gallery-dl attachment-id + native post-id for the
+  // same real post) onto one post-id-keyed keeper. Images untouched. sourceId
+  // (optional) scopes to one source; sent as source_id in the body.
+  function reconcileDuplicatePosts({ sourceId = null, ...opts } = {}) {
+    return _dryRunPost('/api/admin/posts/reconcile-duplicates', {
+      ...opts, source_id: sourceId,
+    })
   }
 
-  async function purgeLegacyTags({ dryRun = true } = {}) {
-    lastError.value = null
-    try {
-      return await api.post(
-        '/api/admin/tags/purge-legacy',
-        { body: { dry_run: dryRun } },
-      )
-    } catch (e) {
-      lastError.value = e.message
-      throw e
-    }
+  function purgeLegacyTags(opts = {}) {
+    return _dryRunPost('/api/admin/tags/purge-legacy', opts)
   }
 
   // Destructive: deletes ALL general + character tags so the operator can
   // re-tag from scratch via auto-suggest. fandom + series preserved.
-  async function resetContentTagging({ dryRun = true } = {}) {
-    lastError.value = null
-    try {
-      return await api.post(
-        '/api/admin/tags/reset-content',
-        { body: { dry_run: dryRun } },
-      )
-    } catch (e) {
-      lastError.value = e.message
-      throw e
-    }
+  function resetContentTagging(opts = {}) {
+    return _dryRunPost('/api/admin/tags/reset-content', opts)
   }
 
   // #714: Title-Case the back-catalog + merge case/whitespace-variant tags.
   // dry-run returns a projection inline; live returns {task_id} (long op —
   // FK repoints) the caller polls via pollTaskUntilDone.
-  async function normalizeTags({ dryRun = true } = {}) {
-    lastError.value = null
-    try {
-      return await api.post(
-        '/api/admin/tags/normalize',
-        { body: { dry_run: dryRun } },
-      )
-    } catch (e) {
-      lastError.value = e.message
-      throw e
-    }
+  function normalizeTags(opts = {}) {
+    return _dryRunPost('/api/admin/tags/normalize', opts)
   }
 
   // --- Task progress polling (taps FC-3i activity dashboard) --------

frontend/test/adminStore.spec.js

@@ -0,0 +1,85 @@
+import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'
+import { setActivePinia, createPinia } from 'pinia'
+import { useAdminStore } from '../src/stores/admin.js'
+
+// Covers the two helpers the admin store actions route through (DRY Finding C,
+// #753): _dryRunPost (URL + dry_run body, sourceId→source_id) and _guard
+// (lastError capture + rethrow). The store had no frontend test before.
+
+function stubFetch(handler) {
+  globalThis.fetch = vi.fn(async (url, init) => {
+    const { status, body } = handler(url, init)
+    return {
+      ok: status >= 200 && status < 300,
+      status,
+      statusText: String(status),
+      text: async () => (body == null ? '' : JSON.stringify(body)),
+    }
+  })
+}
+
+function lastCallBody(calls) {
+  return JSON.parse(calls.at(-1).init.body)
+}
+
+describe('admin store', () => {
+  beforeEach(() => setActivePinia(createPinia()))
+  afterEach(() => vi.restoreAllMocks())
+
+  it('_dryRunPost sends the apply flag to the right endpoint', async () => {
+    const s = useAdminStore()
+    const calls = []
+    stubFetch((url, init) => {
+      calls.push({ url, init })
+      return { status: 200, body: { deleted: 0 } }
+    })
+    await s.pruneUnusedTags({ dryRun: false })
+    const c = calls.at(-1)
+    expect(c.url).toContain('/api/admin/tags/prune-unused')
+    expect(c.init.method).toBe('POST')
+    expect(lastCallBody(calls)).toEqual({ dry_run: false })
+  })
+
+  it('_dryRunPost defaults to the safe preview', async () => {
+    const s = useAdminStore()
+    const calls = []
+    stubFetch((url, init) => {
+      calls.push({ url, init })
+      return { status: 200, body: { count: 0 } }
+    })
+    await s.pruneBarePosts()
+    expect(lastCallBody(calls)).toEqual({ dry_run: true })
+  })
+
+  it('reconcileDuplicatePosts maps sourceId → source_id', async () => {
+    const s = useAdminStore()
+    const calls = []
+    stubFetch((url, init) => {
+      calls.push({ url, init })
+      return { status: 200, body: { groups: 0 } }
+    })
+    await s.reconcileDuplicatePosts({ dryRun: false, sourceId: 42 })
+    const c = calls.at(-1)
+    expect(c.url).toContain('/api/admin/posts/reconcile-duplicates')
+    expect(lastCallBody(calls)).toEqual({ dry_run: false, source_id: 42 })
+  })
+
+  it('_guard captures the error message on lastError and rethrows', async () => {
+    const s = useAdminStore()
+    stubFetch(() => ({ status: 500, body: { error: 'boom' } }))
+    await expect(s.deleteTag(7)).rejects.toThrow('boom')
+    expect(s.lastError).toBe('boom')
+  })
+
+  it('_guard clears lastError on a subsequent success', async () => {
+    const s = useAdminStore()
+    stubFetch(() => ({ status: 500, body: { error: 'boom' } }))
+    await expect(s.deleteTag(7)).rejects.toThrow()
+    expect(s.lastError).toBe('boom')
+
+    stubFetch(() => ({ status: 200, body: { count: 3 } }))
+    const n = await s.tagUsageCount(7)
+    expect(n).toBe(3)  // tagUsageCount returns body.count, not the raw body
+    expect(s.lastError).toBe(null)
+  })
+})

frontend/test/usePreviewCommit.spec.js

@@ -0,0 +1,66 @@
+import { describe, it, expect } from 'vitest'
+import { usePreviewCommit } from '../src/composables/usePreviewCommit.js'
+
+// Canonical sync preview→commit flow for maintenance tiles (DRY pattern sweep,
+// #753) — the primitive TagMaintenanceCard + PostMaintenanceCard's 6 flows route
+// through.
+
+describe('usePreviewCommit', () => {
+  it('runPreview loads the projection and toggles previewing', async () => {
+    const pc = usePreviewCommit({
+      preview: async () => ({ count: 3 }),
+      commit: async () => ({ deleted: 3 }),
+    })
+    expect(pc.previewData.value).toBe(null)
+    const p = pc.runPreview()
+    expect(pc.previewing.value).toBe(true)
+    await p
+    expect(pc.previewing.value).toBe(false)
+    expect(pc.previewData.value).toEqual({ count: 3 })
+  })
+
+  it('runCommit stores the result and collapses to a static emptyPreview', async () => {
+    const pc = usePreviewCommit({
+      preview: async () => ({ count: 3 }),
+      commit: async () => ({ deleted: 3 }),
+      emptyPreview: { count: 0, sample_names: [] },
+    })
+    await pc.runPreview()
+    await pc.runCommit()
+    expect(pc.result.value).toEqual({ deleted: 3 })
+    expect(pc.previewData.value).toEqual({ count: 0, sample_names: [] })
+  })
+
+  it('emptyPreview as a function derives the collapsed shape from the result', async () => {
+    const pc = usePreviewCommit({
+      preview: async () => ({ count: 2 }),
+      commit: async () => ({ sample_names: ['a', 'b'] }),
+      emptyPreview: (r) => ({ count: 0, sample_names: r.sample_names }),
+    })
+    await pc.runCommit()
+    expect(pc.previewData.value).toEqual({ count: 0, sample_names: ['a', 'b'] })
+  })
+
+  it('without emptyPreview the projection stays (dispatch-on-commit variant)', async () => {
+    const pc = usePreviewCommit({
+      preview: async () => ({ total_changes: 5 }),
+      commit: async () => ({ status: 'queued' }),
+    })
+    await pc.runPreview()
+    await pc.runCommit()
+    expect(pc.result.value).toEqual({ status: 'queued' })
+    expect(pc.previewData.value).toEqual({ total_changes: 5 })  // unchanged
+  })
+
+  it('runPreview clears a prior apply result (re-preview hides the badge)', async () => {
+    const pc = usePreviewCommit({
+      preview: async () => ({ count: 1 }),
+      commit: async () => ({ deleted: 1 }),
+      emptyPreview: { count: 0 },
+    })
+    await pc.runCommit()
+    expect(pc.result.value).toEqual({ deleted: 1 })
+    await pc.runPreview()
+    expect(pc.result.value).toBe(null)
+  })
+})

tests/test_api_admin.py

@@ -5,10 +5,11 @@ without actually queuing. Tier-B/A endpoints run synchronously
 through the real service.
 """
 import hashlib
+from types import SimpleNamespace
 
 import pytest
 
-from backend.app.models import Artist, ImageRecord, Tag, TagKind
+from backend.app.models import Artist, ImageRecord, Post, Tag, TagKind
 from backend.app.models.tag import image_tag
 
 pytestmark = pytest.mark.integration
@@ -411,6 +412,67 @@ async def test_purge_legacy_commit_deletes_only_legacy(client, db):
     assert gone == 0
 
 
+# --- Tier-A: POST /posts/prune-bare + /posts/reconcile-duplicates ---
+# These two routes share _run_dry_run_op with the tag prunes (DRY pass,
+# task #753); cover their apply path + reconcile's source_id passthrough so
+# every helper consumer is exercised at the route level, not just the service.
+
+
+@pytest.mark.asyncio
+async def test_prune_bare_commit_deletes_bare_posts(client, db):
+    from sqlalchemy import func, select
+
+    a = Artist(name="BP", slug="bp-api")
+    db.add(a)
+    await db.flush()
+    db.add(Post(artist_id=a.id, external_post_id="bare1"))  # no images/attachments
+    await db.commit()
+
+    resp = await client.post(
+        "/api/admin/posts/prune-bare", json={"dry_run": False},
+    )
+    body = await resp.get_json()
+    assert body["deleted"] >= 1
+    remaining = (await db.execute(
+        select(func.count()).select_from(Post).where(Post.external_post_id == "bare1")
+    )).scalar_one()
+    assert remaining == 0
+
+
+@pytest.mark.asyncio
+async def test_reconcile_duplicates_commit_merges_via_endpoint(client, db):
+    from sqlalchemy import select
+
+    a = Artist(name="RC", slug="rc-api")
+    db.add(a)
+    await db.flush()
+    db.add_all([
+        Post(artist_id=a.id, external_post_id="711509",
+             raw_metadata={"post_id": 1923726, "category": "subscribestar"},
+             description="body"),
+        Post(artist_id=a.id, external_post_id="1923726",
+             raw_metadata={"post_id": 1923726, "category": "subscribestar"}),
+    ])
+    await db.commit()
+
+    resp = await client.post(
+        "/api/admin/posts/reconcile-duplicates", json={"dry_run": False},
+    )
+    body = await resp.get_json()
+    assert body["merged"] == 1
+    rows = (await db.execute(select(Post.external_post_id))).scalars().all()
+    assert rows == ["1923726"]  # one post-id-keyed keeper survives
+
+
+@pytest.mark.asyncio
+async def test_reconcile_duplicates_rejects_bad_source_id(client):
+    resp = await client.post(
+        "/api/admin/posts/reconcile-duplicates",
+        json={"dry_run": True, "source_id": "abc"},
+    )
+    assert resp.status_code == 400
+
+
 # --- DB maintenance: bloat readout + manual VACUUM trigger ----------
 
 
@@ -437,6 +499,87 @@ async def test_trigger_vacuum_queues_the_task(client, monkeypatch):
     assert calls == [1]
 
 
+# --- maintenance triggers route through _queued() (DRY Finding B, #753) ---
+# Each operator-triggered task returns 202 + the Celery task id via the shared
+# _queued() helper. These trigger endpoints had no route-level tests; cover every
+# helper consumer, and assert the dry_run flag threads through where applicable.
+
+
+def _fake_delay(captured):
+    def _delay(*args, **kwargs):
+        captured.append((args, kwargs))
+        return SimpleNamespace(id="task-xyz")
+    return _delay
+
+
+@pytest.mark.asyncio
+async def test_trigger_reextract_archives_queues(client, monkeypatch):
+    from backend.app.tasks import admin as admin_tasks
+
+    calls = []
+    monkeypatch.setattr(
+        admin_tasks.reextract_archive_attachments_task, "delay", _fake_delay(calls)
+    )
+    resp = await client.post("/api/admin/maintenance/reextract-archives")
+    assert resp.status_code == 202
+    assert (await resp.get_json())["task_id"] == "task-xyz"
+    assert len(calls) == 1
+
+
+@pytest.mark.asyncio
+async def test_trigger_prune_missing_files_queues(client, monkeypatch):
+    from backend.app.tasks import admin as admin_tasks
+
+    calls = []
+    monkeypatch.setattr(
+        admin_tasks.prune_missing_file_records_task, "delay", _fake_delay(calls)
+    )
+    resp = await client.post("/api/admin/maintenance/prune-missing-files")
+    assert resp.status_code == 202
+    assert (await resp.get_json())["task_id"] == "task-xyz"
+
+
+@pytest.mark.asyncio
+async def test_trigger_dedup_videos_queues_and_threads_dry_run(client, monkeypatch):
+    from backend.app.tasks import admin as admin_tasks
+
+    calls = []
+    monkeypatch.setattr(admin_tasks.dedup_videos_task, "delay", _fake_delay(calls))
+    resp = await client.post(
+        "/api/admin/maintenance/dedup-videos", json={"dry_run": False},
+    )
+    assert resp.status_code == 202
+    assert (await resp.get_json())["task_id"] == "task-xyz"
+    assert calls[0][1] == {"dry_run": False}  # flag threaded to the task
+
+
+@pytest.mark.asyncio
+async def test_trigger_purge_gated_previews_queues_and_threads_dry_run(client, monkeypatch):
+    from backend.app.tasks import admin as admin_tasks
+
+    calls = []
+    monkeypatch.setattr(
+        admin_tasks.purge_gated_previews_task, "delay", _fake_delay(calls)
+    )
+    resp = await client.post(
+        "/api/admin/maintenance/purge-gated-previews", json={"dry_run": True},
+    )
+    assert resp.status_code == 202
+    assert (await resp.get_json())["task_id"] == "task-xyz"
+    assert calls[0][1] == {"dry_run": True}
+
+
+@pytest.mark.asyncio
+async def test_trigger_normalize_live_queues(client, monkeypatch):
+    from backend.app.tasks import admin as admin_tasks
+
+    calls = []
+    monkeypatch.setattr(admin_tasks.normalize_tags_task, "delay", _fake_delay(calls))
+    resp = await client.post("/api/admin/tags/normalize", json={"dry_run": False})
+    assert resp.status_code == 202
+    assert (await resp.get_json())["task_id"] == "task-xyz"
+
+
 # --- Tier-A: POST /tags/reset-content -------------------------------
 
 

tests/test_cleanup_service.py

@@ -853,3 +853,36 @@ def test_reconcile_dedups_provenance_collision(db_sync, tmp_path):
         select(ImageProvenance.post_id).where(ImageProvenance.image_record_id == img_id)
     ).scalars().all()
     assert prov == [native_id]
+
+
+def test_reconcile_preserves_from_attachment_on_provenance_collision(db_sync, tmp_path):
+    # The gallery-dl loser row recorded which archive the image came out of
+    # (#87); the native keeper row didn't. Merging drops the loser provenance row
+    # on the (image, post) collision — but must first carry its from_attachment_id
+    # onto the keeper so the containing-archive linkage survives.
+    a = _make_artist(db_sync, slug="rc4")
+    img = _make_image(db_sync, artist=a, path=str(tmp_path / "z.jpg"), sha256="e" * 64)
+    legacy = Post(artist_id=a.id, external_post_id="9", raw_metadata={"post_id": 77})
+    native = Post(artist_id=a.id, external_post_id="77", raw_metadata={"post_id": 77})
+    db_sync.add_all([legacy, native])
+    db_sync.flush()
+    att = PostAttachment(
+        post_id=legacy.id, artist_id=a.id, sha256="f" * 64,
+        path=str(tmp_path / "bundle.cbz"), original_filename="bundle.cbz",
+        ext=".cbz", size_bytes=9,
+    )
+    db_sync.add(att)
+    db_sync.flush()
+    db_sync.add(ImageProvenance(
+        image_record_id=img.id, post_id=legacy.id, from_attachment_id=att.id,
+    ))
+    db_sync.add(ImageProvenance(image_record_id=img.id, post_id=native.id))
+    img_id, native_id, att_id = img.id, native.id, att.id
+    db_sync.commit()
+
+    cleanup_service.reconcile_duplicate_posts(db_sync, dry_run=False)
+    rows = db_sync.execute(
+        select(ImageProvenance.post_id, ImageProvenance.from_attachment_id)
+        .where(ImageProvenance.image_record_id == img_id)
+    ).all()
+    assert rows == [(native_id, att_id)]

tests/test_importer_archive.py

@@ -164,6 +164,86 @@ def test_archive_all_deduped_is_benign_not_flagged(importer, import_layout):
     assert provs == 4  # 2 images × 2 posts (enrich-on-duplicate)
 
 
+def test_archive_members_record_containing_archive(importer, import_layout):
+    """Milestone #87: each extracted member's provenance row records the archive
+    PostAttachment it came out of, so provenance can show the one archive a file
+    lives inside."""
+    import_root, _ = import_layout
+    importer.settings.phash_threshold = 0
+    arc = import_root / "Bob" / "set.cbz"
+    arc.parent.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(arc, "w") as zf:
+        zf.writestr("a.jpg", _split_bytes("v"))
+        zf.writestr("b.jpg", _split_bytes("h"))
+    arc.with_suffix(".cbz.json").write_text(json.dumps(
+        {"category": "patreon", "id": "777", "title": "Set"}))
+
+    importer.import_one(arc)
+    att = importer.session.execute(select(PostAttachment)).scalar_one()
+    provs = importer.session.execute(select(ImageProvenance)).scalars().all()
+    assert len(provs) == 2
+    assert all(p.from_attachment_id == att.id for p in provs)
+
+
+def test_nested_archive_member_records_outer_archive(importer, import_layout):
+    """Nested members link to the OUTER stored archive — the only one persisted
+    as a PostAttachment (the inner archive lives in a tempdir)."""
+    import_root, _ = import_layout
+    importer.settings.phash_threshold = 0
+    inner = io.BytesIO()
+    with zipfile.ZipFile(inner, "w") as zf:
+        zf.writestr("p1.jpg", _split_bytes("v"))
+        zf.writestr("p2.jpg", _split_bytes("h"))
+    arc = import_root / "Nessie" / "hr.cbz"
+    arc.parent.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(arc, "w") as zf:
+        zf.writestr("chapter.zip", inner.getvalue())
+    arc.with_suffix(".cbz.json").write_text(json.dumps(
+        {"category": "patreon", "id": "900", "title": "HR Pack"}))
+
+    importer.import_one(arc)
+    att = importer.session.execute(select(PostAttachment)).scalar_one()
+    provs = importer.session.execute(select(ImageProvenance)).scalars().all()
+    assert len(provs) == 2
+    assert all(p.from_attachment_id == att.id for p in provs)
+
+
+def test_deduped_archive_member_records_each_posts_archive(importer, import_layout):
+    """A member re-shipped in a second post's archive gets its NEW provenance row
+    (for the second post) stamped with the SECOND archive — the UPDATE-of-existing
+    path the reextract backfill relies on. Each post's rows point at its own
+    archive."""
+    import_root, _ = import_layout
+    importer.settings.phash_threshold = 0
+    pv, ph = _split_bytes("v"), _split_bytes("h")
+    first = import_root / "Dup" / "a.cbz"
+    first.parent.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(first, "w") as zf:
+        zf.writestr("a.jpg", pv)
+        zf.writestr("b.jpg", ph)
+    first.with_suffix(".cbz.json").write_text(json.dumps(
+        {"category": "patreon", "id": "111", "title": "First"}))
+    importer.import_one(first)
+
+    second = import_root / "Dup" / "b.cbz"  # different post, same two images
+    with zipfile.ZipFile(second, "w") as zf:
+        zf.writestr("a.jpg", pv)
+        zf.writestr("b.jpg", ph)
+    second.with_suffix(".cbz.json").write_text(json.dumps(
+        {"category": "patreon", "id": "222", "title": "Second"}))
+    importer.import_one(second)
+
+    posts = {p.external_post_id: p.id for p in
+             importer.session.execute(select(Post)).scalars().all()}
+    atts = {a.post_id: a.id for a in
+            importer.session.execute(select(PostAttachment)).scalars().all()}
+    provs = importer.session.execute(select(ImageProvenance)).scalars().all()
+    assert len(provs) == 4  # 2 images × 2 posts
+    for p in provs:
+        assert p.from_attachment_id == atts[p.post_id]
+    assert atts[posts["111"]] != atts[posts["222"]]
+
+
 def test_corrupt_archive_still_stored(importer, import_layout):
     import_root, _ = import_layout
     arc = import_root / "Bob" / "broken.zip"

tests/test_provenance_service.py

@@ -134,6 +134,98 @@ async def test_for_image_null_post_fields_serialize_null(db):
     assert e["post"]["attachment_count"] is None
 
 
+@pytest.mark.asyncio
+async def test_for_image_attachments_scoped_to_primary_post(db):
+    # Image linked to TWO posts; only the primary (originating) post's
+    # attachments should surface — not the other post's mega-bundle.
+    rec = await _seed_image(db)
+    a1, s1, primary = await _seed_post(db, artist_name="P1", slug="p1",
+                                       platform="patreon", ext_id="100")
+    a2, s2, bundle = await _seed_post(db, artist_name="P2", slug="p2",
+                                      platform="patreon", ext_id="200")
+    rec.primary_post_id = primary.id
+    db.add(ImageProvenance(image_record_id=rec.id, post_id=primary.id,
+                           source_id=s1.id))
+    db.add(ImageProvenance(image_record_id=rec.id, post_id=bundle.id,
+                           source_id=s2.id))
+    db.add(PostAttachment(
+        post_id=primary.id, artist_id=a1.id, sha256="p" + "0" * 63,
+        path="/images/attachments/p00/keep.zip", original_filename="keep.zip",
+        ext=".zip", mime="application/zip", size_bytes=9,
+    ))
+    db.add(PostAttachment(
+        post_id=bundle.id, artist_id=a2.id, sha256="b" + "0" * 63,
+        path="/images/attachments/b00/drop.rar", original_filename="drop.rar",
+        ext=".rar", mime="application/x-rar", size_bytes=9,
+    ))
+    await db.flush()
+
+    payload = await ProvenanceService(db).for_image(rec.id)
+    names = [a["original_filename"] for a in payload["attachments"]]
+    assert names == ["keep.zip"]
+
+
+@pytest.mark.asyncio
+async def test_for_image_attachments_filtered_to_containing_archive(db):
+    # Milestone #87: when the originating post's provenance row records which
+    # archive the file came from, show ONLY that archive — not the post's other
+    # attachments.
+    rec = await _seed_image(db)
+    a1, s1, post = await _seed_post(db, artist_name="Arc", slug="arc",
+                                    platform="patreon", ext_id="500")
+    rec.primary_post_id = post.id
+    att_in = PostAttachment(
+        post_id=post.id, artist_id=a1.id, sha256="e" + "0" * 63,
+        path="/images/attachments/e00/in.cbz", original_filename="in.cbz",
+        ext=".cbz", mime="application/zip", size_bytes=9,
+    )
+    att_other = PostAttachment(
+        post_id=post.id, artist_id=a1.id, sha256="f" + "0" * 63,
+        path="/images/attachments/f00/other.rar", original_filename="other.rar",
+        ext=".rar", mime="application/x-rar", size_bytes=9,
+    )
+    db.add(att_in)
+    db.add(att_other)
+    await db.flush()
+    db.add(ImageProvenance(image_record_id=rec.id, post_id=post.id,
+                           source_id=s1.id, from_attachment_id=att_in.id))
+    await db.flush()
+
+    payload = await ProvenanceService(db).for_image(rec.id)
+    names = [a["original_filename"] for a in payload["attachments"]]
+    assert names == ["in.cbz"]
+
+
+@pytest.mark.asyncio
+async def test_for_image_attachments_fallback_to_all_posts_when_no_primary(db):
+    # No primary_post_id (older rows / filesystem imports) → preserve the
+    # aggregate-across-linked-posts behavior so attachments aren't lost.
+    rec = await _seed_image(db)
+    a1, s1, p1 = await _seed_post(db, artist_name="F1", slug="f1",
+                                  platform="patreon", ext_id="300")
+    a2, s2, p2 = await _seed_post(db, artist_name="F2", slug="f2",
+                                  platform="patreon", ext_id="400")
+    db.add(ImageProvenance(image_record_id=rec.id, post_id=p1.id,
+                           source_id=s1.id))
+    db.add(ImageProvenance(image_record_id=rec.id, post_id=p2.id,
+                           source_id=s2.id))
+    db.add(PostAttachment(
+        post_id=p1.id, artist_id=a1.id, sha256="c" + "0" * 63,
+        path="/images/attachments/c00/one.zip", original_filename="one.zip",
+        ext=".zip", mime="application/zip", size_bytes=9,
+    ))
+    db.add(PostAttachment(
+        post_id=p2.id, artist_id=a2.id, sha256="d" + "0" * 63,
+        path="/images/attachments/d00/two.zip", original_filename="two.zip",
+        ext=".zip", mime="application/zip", size_bytes=9,
+    ))
+    await db.flush()
+
+    payload = await ProvenanceService(db).for_image(rec.id)
+    names = sorted(a["original_filename"] for a in payload["attachments"])
+    assert names == ["one.zip", "two.zip"]
+
+
 @pytest.mark.asyncio
 async def test_for_post_missing_returns_none(db):
     svc = ProvenanceService(db)

tests/test_reextract_archives.py

@@ -74,10 +74,13 @@ def test_reextract_links_archive_members_to_post(db_sync, tmp_path, monkeypatch)
     with zipfile.ZipFile(arc, "w") as zf:
         zf.writestr("inside.jpg", _jpeg("red"))
     sha = hashlib.sha256(arc.read_bytes()).hexdigest()
-    db_sync.add(PostAttachment(
+    att = PostAttachment(
         post_id=post.id, artist_id=artist.id, sha256=sha, path=str(arc),
         original_filename=arc.name, ext="", size_bytes=arc.stat().st_size,
-    ))
+    )
+    db_sync.add(att)
+    db_sync.flush()
+    att_id = att.id
     db_sync.commit()
 
     summary = cleanup_service.reextract_archive_attachments(
@@ -94,6 +97,9 @@ def test_reextract_links_archive_members_to_post(db_sync, tmp_path, monkeypatch)
     ).scalars().all()
     assert len(prov) == 1
     assert prov[0].image_record_id == images[0].id
+    # Milestone #87: re-extraction stamps which archive the member came from —
+    # this is the backfill path for pre-existing opaque archives.
+    assert prov[0].from_attachment_id == att_id
 
     # Idempotent — a second run imports nothing new (member dedups by sha256).
     again = cleanup_service.reextract_archive_attachments(