fix(posts): link duplicate items to every post + prune bare shells

The native Patreon backfill flooded the feed with bare 'Post <id>' shells
(1589 for Anduo). Root cause: PostAttachment.sha256 was GLOBALLY unique, so a
non-art file reused across posts only ever linked to the first one, and
_capture_attachment created the Post before that dedup check — leaving later
posts with no image and no attachment. Duplicate IMAGES had the mirror gap:
attach_in_place returned duplicate_hash/duplicate_phash before _apply_sidecar,
so the second post got no provenance row, and the feed only rendered via
primary_post_id (one post per image).

Operator requirement: a duplicate item must show on EVERY post it appears in.
Unify the fix as link-not-suppress:

- importer: on duplicate_hash / duplicate_phash(larger_exists), append an
  image_provenance row for the new post (keep primary on the first). Both the
  download path (attach_in_place) and the filesystem path (_import_media).
- post_feed_service: render thumbnails by image_provenance UNION primary_post_id,
  so a cross-posted image shows on every post (and legacy primary-only images
  still show).
- PostAttachment: per-post uniqueness — drop UNIQUE(sha256), add partial
  UNIQUE(post_id, sha256) + partial UNIQUE(sha256) WHERE post_id IS NULL
  (migration 0043); _capture_attachment dedups per-(post,sha) over the shared
  sha-addressed blob, so no post is left bare.
- cleanup: new prune-bare-posts maintenance action (cleanup_service
  _bare_post_conditions shared by preview/count/delete per preview/apply parity;
  admin endpoint; PostMaintenanceCard). Deletes posts with zero image links
  (primary or provenance) AND zero attachments. Run after the feed fix so a
  hidden provenance link spares the post instead of deleting it.

Tests: dup image shows on both posts; dup attachment shows on both posts; feed
renders provenance-linked duplicates; prune-bare delete-path == preview.

Operator redeploys (migration 0043) then runs the prune to clear the shells.

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

backend/app/api/admin.py

@@ -6,6 +6,7 @@ Five action surfaces:
   DELETE /api/admin/tags/<int:tag_id>              (Tier B)
   POST   /api/admin/tags/<int:dest_id>/merge       (Tier B)
   POST   /api/admin/tags/prune-unused              (Tier A)
+  POST   /api/admin/posts/prune-bare               (Tier A)
   POST   /api/admin/tags/purge-legacy              (Tier A)
   GET    /api/admin/tags/<int:tag_id>/usage-count  (helper)
 
@@ -204,6 +205,27 @@ async def tags_prune_unused():
     return jsonify(result)
 
 
+@admin_bp.route("/posts/prune-bare", methods=["POST"])
+async def posts_prune_bare():
+    """Tier-A: delete bare posts — Post rows with no linked images (primary OR
+    provenance) and no attachments. Dry-run preview list IS the prompt: UI calls
+    with dry_run=true first, shows the count + sample, operator confirms by
+    re-calling with dry_run=false. Same preview/apply-parity predicate as the
+    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)
+
+
 @admin_bp.route("/tags/purge-legacy", methods=["POST"])
 async def tags_purge_legacy():
     """Tier-A: delete legacy IR-migration tags — archive/post/artist

backend/app/models/post_attachment.py

@@ -14,10 +14,12 @@ from sqlalchemy import (
     BigInteger,
     DateTime,
     ForeignKey,
+    Index,
     Integer,
     String,
     Text,
     func,
+    text,
 )
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -26,6 +28,24 @@ from .base import Base
 
 class PostAttachment(Base):
     __tablename__ = "post_attachment"
+    # Dedup is PER-POST, not global (2026-06-08): the same non-art file attached
+    # to many posts gets one row per post over a single sha-addressed blob, so no
+    # post is left a bare shell. Partial uniques: (post_id, sha256) for real posts;
+    # (sha256) alone for the NULL-post filesystem case (one row per file there).
+    __table_args__ = (
+        Index(
+            "uq_post_attachment_post_sha",
+            "post_id", "sha256",
+            unique=True,
+            postgresql_where=text("post_id IS NOT NULL"),
+        ),
+        Index(
+            "uq_post_attachment_null_post_sha",
+            "sha256",
+            unique=True,
+            postgresql_where=text("post_id IS NULL"),
+        ),
+    )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
     post_id: Mapped[int | None] = mapped_column(
@@ -35,7 +55,7 @@ class PostAttachment(Base):
         ForeignKey("artist.id", ondelete="SET NULL"), nullable=True, index=True
     )
     sha256: Mapped[str] = mapped_column(
-        String(64), nullable=False, unique=True, index=True
+        String(64), nullable=False, index=True
     )
     path: Mapped[str] = mapped_column(Text, nullable=False)
     original_filename: Mapped[str] = mapped_column(Text, nullable=False)

backend/app/services/cleanup_service.py

@@ -20,7 +20,15 @@ from typing import Any
 from sqlalchemy import func, or_, select, update
 from sqlalchemy.orm import Session
 
-from ..models import Artist, ImageRecord, LibraryAuditRun, Tag
+from ..models import (
+    Artist,
+    ImageProvenance,
+    ImageRecord,
+    LibraryAuditRun,
+    Post,
+    PostAttachment,
+    Tag,
+)
 from ..models.series_chapter import SeriesChapter
 from ..models.series_page import SeriesPage
 from ..models.tag import image_tag
@@ -407,6 +415,88 @@ def prune_unused_tags(session: Session, *, dry_run: bool = False) -> dict:
     return {"deleted": result.rowcount or 0, "sample_names": sample}
 
 
+def _bare_post_conditions() -> list:
+    """The WHERE conditions that define a 'bare' post — the SINGLE source of truth
+    shared by find_bare_posts (preview sample), the dry-run count, AND the live
+    delete, so the preview can NEVER diverge from what the delete removes
+    ([[feedback_preview_apply_parity]]).
+
+    A post is "bare" iff NOTHING is attached to it across every way content links
+    to a post:
+      - image_record.primary_post_id  (a post's own canonical images)
+      - image_provenance.post_id      (cross-posted/duplicate images linked here)
+      - post_attachment.post_id        (preserved non-art files)
+
+    These are exactly the shells the empty-post flood produced: the native Patreon
+    ingester synthesized a Post per walked post, but when its only content was a
+    duplicate image/attachment that linked to an EARLIER post, the new post was
+    left with none of the three (operator-flagged 2026-06-08). Every FK to post.id
+    is SET NULL or CASCADE, so deleting a bare post is non-destructive by
+    construction — there is nothing pointing at it to orphan. Must run AFTER the
+    provenance-render fix so a post that DOES have a hidden provenance link is
+    spared, not deleted.
+    """
+    has_primary = select(ImageRecord.id).where(
+        ImageRecord.primary_post_id == Post.id
+    )
+    has_provenance = select(ImageProvenance.id).where(
+        ImageProvenance.post_id == Post.id
+    )
+    has_attachment = select(PostAttachment.id).where(
+        PostAttachment.post_id == Post.id
+    )
+    return [
+        ~has_primary.exists(),
+        ~has_provenance.exists(),
+        ~has_attachment.exists(),
+    ]
+
+
+def find_bare_posts(
+    session: Session, *, limit: int | None = None,
+) -> list[Post]:
+    """Posts with zero linked images (primary OR provenance) AND zero
+    attachments — safe to sweep. Sorted by id. Shares its predicate with the
+    live prune via _bare_post_conditions()."""
+    stmt = select(Post).where(*_bare_post_conditions()).order_by(Post.id)
+    if limit is not None:
+        stmt = stmt.limit(limit)
+    return list(session.execute(stmt).scalars().all())
+
+
+def _bare_post_label(post: Post) -> str:
+    """Human label for the preview sample — mirrors the feed's fallback title."""
+    if post.post_title:
+        return post.post_title
+    return f"Post {post.external_post_id or post.id}"
+
+
+def prune_bare_posts(session: Session, *, dry_run: bool = False) -> dict:
+    """Find posts with no images and no attachments and (unless dry_run) delete
+    them.
+
+    Returns:
+      dry_run=True:  {"count": N, "sample_names": [first 50]}
+      dry_run=False: {"deleted": N, "sample_names": [first 50]}
+
+    The live delete runs a single DELETE with the SAME predicate
+    (_bare_post_conditions) the preview uses, so the row count scales without
+    binding every id as a parameter ([[reference_psycopg_65535_param_ceiling]])
+    AND the delete can never remove a post the preview deemed kept.
+    """
+    conditions = _bare_post_conditions()
+    sample_rows = find_bare_posts(session, limit=50)
+    sample = [_bare_post_label(p) for p in sample_rows]
+    if dry_run:
+        count = session.execute(
+            select(func.count()).select_from(Post).where(*conditions)
+        ).scalar_one()
+        return {"count": count, "sample_names": sample}
+    result = session.execute(Post.__table__.delete().where(*conditions))
+    session.commit()
+    return {"deleted": result.rowcount or 0, "sample_names": sample}
+
+
 # Legacy tags FC no longer uses, in two shapes:
 #   (1) kinds the tag input never produces — archive/post/artist.
 #       provenance (post grouping) + archive membership are their own

backend/app/services/importer.py

@@ -361,22 +361,39 @@ class Importer:
             artist = self._resolve_artist(source)
             post = self._post_for_sidecar(source, artist)
         sha = _sha256_of(source)
-        select_existing = select(PostAttachment).where(PostAttachment.sha256 == sha)
+        post_id = post.id if post else None
+        # Dedup is PER-POST, not global: a non-art file the creator attaches to
+        # many posts (a standard pdf/zip/link-card) must get a row on EVERY post
+        # so none is left a bare shell. The on-disk blob stays sha-deduped
+        # (attachments.store is sha-addressed + idempotent); only the rows are
+        # per-post. PostAttachment's unique is partial (post_id, sha256) for
+        # real posts and (sha256) for the NULL-post filesystem case. Before
+        # 2026-06-08 the global UNIQUE(sha256) left every post after the first
+        # with no attachment row → 1589 empty Anduo shells.
+        if post_id is not None:
+            select_existing = select(PostAttachment).where(
+                PostAttachment.post_id == post_id,
+                PostAttachment.sha256 == sha,
+            )
+        else:
+            select_existing = select(PostAttachment).where(
+                PostAttachment.post_id.is_(None),
+                PostAttachment.sha256 == sha,
+            )
         existing = self.session.execute(select_existing).scalar_one_or_none()
         if existing is not None:
             self.session.commit()
             return ImportResult(status="attached")
-        # Savepoint + IntegrityError recovery — PostAttachment.sha256 is
-        # UNIQUE, so two workers can both pass the SELECT and only the
-        # second INSERT fails. Without savepoint, the outer transaction
-        # poisons and the calling task crashes. attachments.store is
-        # sha-addressed so both workers race to write the same target
-        # path; shutil.copy2 + rename is idempotent. Audit 2026-06-02.
+        # Savepoint + IntegrityError recovery — the partial UNIQUE means two
+        # workers can both pass the SELECT and only the second INSERT fails.
+        # Without savepoint, the outer transaction poisons and the calling task
+        # crashes. attachments.store is sha-addressed so both workers race to
+        # write the same target path; shutil.copy2 + rename is idempotent.
         sp = self.session.begin_nested()
         try:
             stored = self.attachments.store(source, sha)
             self.session.add(PostAttachment(
-                post_id=post.id if post else None,
+                post_id=post_id,
                 artist_id=artist.id if artist else None,
                 sha256=sha,
                 path=stored,
@@ -389,7 +406,7 @@ class Importer:
             sp.commit()
         except IntegrityError:
             sp.rollback()
-            # Lost the race — the other worker's row is canonical.
+            # Lost the race — the other worker's row for this (post, sha) wins.
             self.session.execute(select_existing).scalar_one()
         self.session.commit()
         return ImportResult(status="attached")
@@ -555,6 +572,15 @@ class Importer:
         existing = self.session.execute(existing_stmt).scalar_one_or_none()
         if existing:
             if not self.deep:
+                # Enrich-on-duplicate (parity with attach_in_place): a re-scanned
+                # file that is a byte-dup of an existing image still links its
+                # post via _apply_sidecar, so a cross-posted image shows on every
+                # post. Artist is derived from the sidecar here (not yet resolved).
+                self._apply_sidecar(
+                    existing, attribution_path, None,
+                    explicit_source=explicit_source,
+                )
+                self.session.commit()
                 return ImportResult(
                     status="skipped", skip_reason=SkipReason.duplicate_hash,
                     image_id=existing.id, error="sha256 already present",
@@ -581,6 +607,14 @@ class Importer:
                     candidates, self.settings.phash_threshold,
                 )
                 if rel == "larger_exists":
+                    # Enrich-on-duplicate (parity with attach_in_place).
+                    larger = self.session.get(ImageRecord, match_id)
+                    if larger is not None:
+                        self._apply_sidecar(
+                            larger, attribution_path, None,
+                            explicit_source=explicit_source,
+                        )
+                        self.session.commit()
                     return ImportResult(
                         status="skipped",
                         skip_reason=SkipReason.duplicate_phash,
@@ -764,6 +798,15 @@ class Importer:
             select(ImageRecord).where(ImageRecord.sha256 == sha)
         ).scalar_one_or_none()
         if existing:
+            # Enrich-on-duplicate (image_provenance docstring / spec §3): the
+            # same bytes appearing in another post must show on THAT post too,
+            # so append a provenance row for the new post rather than dropping
+            # the linkage. primary_post_id stays on the first post — _apply_sidecar
+            # only sets it when NULL. Without this the new post is left with no
+            # image AND (if its other content also deduped) no attachment, i.e. a
+            # bare shell — operator-flagged 2026-06-08 (1589 empty Anduo posts).
+            self._apply_sidecar(existing, path, artist, explicit_source=source)
+            self.session.commit()
             return ImportResult(
                 status="skipped", skip_reason=SkipReason.duplicate_hash,
                 image_id=existing.id, error="sha256 already present",
@@ -784,6 +827,15 @@ class Importer:
                     candidates, self.settings.phash_threshold,
                 )
                 if rel == "larger_exists":
+                    # Enrich-on-duplicate: link the near-dup's post to the
+                    # existing larger image so it shows on both (see duplicate_hash
+                    # above). smaller_exists already links via _supersede.
+                    larger = self.session.get(ImageRecord, match_id)
+                    if larger is not None:
+                        self._apply_sidecar(
+                            larger, path, artist, explicit_source=source,
+                        )
+                        self.session.commit()
                     return ImportResult(
                         status="skipped",
                         skip_reason=SkipReason.duplicate_phash,

backend/app/services/post_feed_service.py

@@ -6,7 +6,8 @@ COALESCE(Post.post_date, Post.downloaded_at) so posts without a
 publish date sort by when we captured them.
 
 Pure read-surface; no writes. The service composes the post dict
-(including thumbnails from ImageRecord.primary_post_id and non-media
+(thumbnails from every image linked to the post — its own primary images
+plus cross-posted duplicates via image_provenance — and non-media
 attachments from PostAttachment) so the API layer can jsonify directly.
 """
 from __future__ import annotations
@@ -17,7 +18,14 @@ from datetime import datetime
 from sqlalchemy import and_, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from ..models import Artist, ImageRecord, Post, PostAttachment, Source
+from ..models import (
+    Artist,
+    ImageProvenance,
+    ImageRecord,
+    Post,
+    PostAttachment,
+    Source,
+)
 from ..utils.text import html_to_plain, truncate_at_word
 from .gallery_service import thumbnail_url
 
@@ -205,28 +213,50 @@ class PostFeedService:
         """
         if not post_ids:
             return {}
+        # A post shows EVERY image linked to it — both its own primary images and
+        # cross-posted duplicates linked via image_provenance (a near-dup of an
+        # existing image gets a provenance row for the new post, not dropped; see
+        # image_provenance docstring + the importer enrich-on-duplicate path). The
+        # UNION dedups (image, post) pairs and also keeps any legacy image that has
+        # a primary_post_id but no provenance row. Partition the window on the
+        # link's post_id, not ImageRecord.primary_post_id, so a duplicate counts
+        # under each post it belongs to.
+        links = (
+            select(
+                ImageProvenance.image_record_id.label("image_id"),
+                ImageProvenance.post_id.label("post_id"),
+            )
+            .where(ImageProvenance.post_id.in_(post_ids))
+            .union(
+                select(
+                    ImageRecord.id.label("image_id"),
+                    ImageRecord.primary_post_id.label("post_id"),
+                ).where(ImageRecord.primary_post_id.in_(post_ids))
+            )
+            .subquery()
+        )
         # Rank images within each post; cap at `limit` rows per post when
         # limit is set, return all when limit is None.
         ranked = (
             select(
                 ImageRecord.id,
-                ImageRecord.primary_post_id,
+                links.c.post_id,
                 ImageRecord.sha256,
                 ImageRecord.mime,
                 ImageRecord.thumbnail_path,
                 func.row_number().over(
-                    partition_by=ImageRecord.primary_post_id,
+                    partition_by=links.c.post_id,
                     order_by=ImageRecord.id.asc(),
                 ).label("rn"),
                 func.count(ImageRecord.id).over(
-                    partition_by=ImageRecord.primary_post_id,
+                    partition_by=links.c.post_id,
                 ).label("total"),
             )
-            .where(ImageRecord.primary_post_id.in_(post_ids))
+            .join(ImageRecord, ImageRecord.id == links.c.image_id)
             .subquery()
         )
         stmt = select(
-            ranked.c.id, ranked.c.primary_post_id,
+            ranked.c.id, ranked.c.post_id,
             ranked.c.sha256, ranked.c.mime, ranked.c.thumbnail_path, ranked.c.total,
         )
         if limit is not None: