Merge pull request 'Nested-archive extraction (#718) + post-first ingest (#67) + post-body canary (#862)' (#109) from dev into main

The dedup branches of _import_media linked the existing image to the new post via
_apply_sidecar(artist=None), relying on the SIDECAR to carry the artist. But an
archive member's artist comes from its path, and under post-first the per-media
sidecar is minimal (no artist) — so a re-packed / cross-posted archive image
deduped and was left UNLINKED from the new post, i.e. the post showed "no images".

Resolve the path-anchored artist (derive_top_level_artist) up-front in
_import_media and pass it to both enrich-on-duplicate branches (sha256 + phash
larger_exists) and the new-record path. Drop the now-dead _attach_artist helper
(its logic is inlined at the single new-record call site).

Surfaced by the new test_archive_all_deduped_is_benign_not_flagged (was asserting
2==4: the second post got no provenance links).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/app/services/download_service.py

@@ -274,6 +274,14 @@ class DownloadService:
             artist = self.sync_session.get(Artist, ctx["artist_id"])
         source_row = self.sync_session.get(Source, ctx["source_id"])
 
+        # Post-first (#856): on the native ingester the post-record is the sole
+        # writer of the post body/links, so the per-media import must NOT apply post
+        # fields. gallery-dl platforms keep writing them via the sidecar. The
+        # Importer is per-task, so this per-instance flag is safe. uses_native_ingester
+        # is the future-proof seam — a platform migrating onto the native core
+        # flips to post-first automatically (milestone #67, step 3).
+        self.importer.post_first = uses_native_ingester(ctx["platform"])
+
         import_summary = {"attached": 0, "skipped": 0, "errors": 0}
         # Archives detected but captured WITHOUT extracting any image (probe
         # rejected / corrupt / missing extractor backend). Surfaced on the event

backend/app/services/importer.py

@@ -79,6 +79,12 @@ IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tif", ".tiff"}
 VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".m4v", ".wmv", ".flv"}
 ALL_EXTS = IMAGE_EXTS | VIDEO_EXTS
 
+# A "high-resolution files" pack often wraps per-chapter .rar/.zip inside one
+# outer archive (incase #718) — recurse into nested archives so their images
+# land instead of being silently dropped. Cap the depth so a maliciously
+# deep-nested archive can't recurse forever; the bomb-probe re-runs per level.
+_ARCHIVE_MAX_DEPTH = 3
+
 
 def is_supported(path: Path) -> bool:
     return path.suffix.lower() in ALL_EXTS
@@ -149,6 +155,14 @@ class Importer:
         self.thumbnailer = thumbnailer
         self.settings = settings
         self.deep = deep
+        # Post-first ingest (#856, milestone #67): on the NATIVE core ingester the
+        # post-record (`upsert_post_record`) is the SOLE writer of a post's
+        # body/links/metadata; the per-media import only links image provenance +
+        # localization, NOT the post body. download_service sets this True for
+        # native platforms before the phase-3 media loop. Default False keeps the
+        # gallery-dl path writing post fields via `_apply_sidecar` (unchanged).
+        # Safe as a per-instance flag: Importer is constructed per Celery task.
+        self.post_first = False
         self.attachments = AttachmentStore(images_root)
         # phash near-dup candidate cache. Archive imports call _import_media
         # per-member; without this cache the per-member SELECT *FROM
@@ -483,17 +497,17 @@ 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] = []
-        member_total = 0
+        # Per-outcome tally so the "no images" reason names the ACTUAL cause
-        with extract_archive(source) as members:
+        # (#718): nested-archive packs, all-deduped (benign), unsupported formats,
-            for _name, member_path in members:
+        # or failed/corrupt members — instead of one catch-all string.
-                member_total += 1
+        counts = {
-                if not is_supported(member_path):
+            "media": 0, "deduped": 0, "unsupported": 0,
-                    continue  # non-media preserved via the stored archive
+            "failed": 0, "nested": 0, "nested_rejected": 0,
-                res = self._import_media(
+        }
-                    member_path, source, explicit_source=source_row,
+        self._collect_archive_members(
+            source, attribution=source, source_row=source_row,
+            depth=0, member_ids=member_ids, counts=counts,
         )
-                if res.status in ("imported", "superseded") and res.image_id:
-                    member_ids.append(res.image_id)
         # Preserve the archive itself (links to the same Post/Artist).
         self._capture_attachment(
             source, post=post, artist=artist_use, resolved=True
@@ -503,20 +517,93 @@ class Importer:
                 status="imported", image_id=member_ids[0],
                 member_image_ids=member_ids,
             )
-        # No images landed — surface WHY so a post showing "no images" beside an
+        # No NEW images landed. Name WHY precisely so a post showing "no images"
-        # archive is diagnosable instead of silent. Zero members usually means
+        # beside an archive is diagnosable — and DON'T flag the benign all-deduped
-        # the extractor backend is missing/failed (unar for rar, py7zr for 7z)
+        # case as a problem: those images already exist in the library and were
-        # or the file is corrupt; non-zero-but-no-images means it held only
+        # re-linked to this post (enrich-on-duplicate), so the post DOES show them.
-        # non-media files.
+        if counts["media"] == 0 and counts["nested"] == 0:
+            reason = ("archive yielded no members (unsupported/corrupt, or the "
+                      "extractor backend failed)")
+        elif counts["deduped"] and not (
+            counts["unsupported"] or counts["failed"] or counts["nested_rejected"]
+        ):
+            log.info(
+                "%s: all %d image member(s) already in library (deduped + linked "
+                "to this post) — archive preserved, no new import",
+                source.name, counts["deduped"],
+            )
+            return ImportResult(status="attached")
+        else:
             reason = (
-            "archive extracted but held no supported image/video members"
+                f"no new images — media={counts['media']} deduped={counts['deduped']} "
-            if member_total
+                f"unsupported/non-media={counts['unsupported']} failed={counts['failed']} "
-            else "archive yielded no members (unsupported/corrupt, or the "
+                f"nested={counts['nested']} nested-rejected={counts['nested_rejected']}"
-            "extractor backend failed)"
             )
         log.warning("%s: %s", source.name, reason)
         return ImportResult(status="attached", error=reason)
 
+    def _collect_archive_members(
+        self, archive_path: Path, *, attribution: Path,
+        source_row: Source | None, depth: int,
+        member_ids: list[int], counts: dict,
+    ) -> None:
+        """Extract `archive_path` and import its image/video members, RECURSING
+        into nested archives (#718). Members attribute to `attribution` — the
+        OUTER archive's path, so its sidecar resolves them to the right Post even
+        when they came from a nested archive. Depth-capped + bomb-probed per
+        nested level. Mutates `member_ids` and `counts` in place.
+
+        extract_archive is fail-soft, but each level is wrapped so one bad nested
+        archive can't abort the whole import."""
+        try:
+            with extract_archive(archive_path) as members:
+                for _name, member_path in members:
+                    if not member_path.is_file():
+                        continue  # directory entries etc. — not media
+                    if is_archive(member_path):
+                        counts["nested"] += 1
+                        if depth + 1 > _ARCHIVE_MAX_DEPTH:
+                            counts["nested_rejected"] += 1
+                            log.warning(
+                                "nested archive past depth cap %d, skipped: %s",
+                                _ARCHIVE_MAX_DEPTH, member_path.name,
+                            )
+                            continue
+                        probe = safe_probe.probe_archive(member_path)
+                        if not probe.ok:
+                            counts["nested_rejected"] += 1
+                            log.warning(
+                                "nested archive rejected (%s): %s",
+                                probe.reason, member_path.name,
+                            )
+                            continue
+                        self._collect_archive_members(
+                            member_path, attribution=attribution,
+                            source_row=source_row, depth=depth + 1,
+                            member_ids=member_ids, counts=counts,
+                        )
+                        continue
+                    counts["media"] += 1
+                    if not is_supported(member_path):
+                        counts["unsupported"] += 1
+                        continue  # non-media preserved via the stored archive
+                    res = self._import_media(
+                        member_path, attribution, explicit_source=source_row,
+                    )
+                    if res.status in ("imported", "superseded") and res.image_id:
+                        member_ids.append(res.image_id)
+                    elif res.status == "skipped" and res.skip_reason in (
+                        SkipReason.duplicate_hash, SkipReason.duplicate_phash
+                    ):
+                        counts["deduped"] += 1
+                    else:
+                        counts["failed"] += 1
+        except Exception as exc:  # noqa: BLE001 — defensive per level; keep going
+            log.warning(
+                "archive extraction failed for %s (depth %d): %s",
+                archive_path.name, depth, exc,
+            )
+
     def _import_media(
         self, source: Path, attribution_path: Path,
         *, explicit_source: Source | None = None,
@@ -595,6 +682,15 @@ class Importer:
                         error=f"{pct:.2%} transparent",
                     )
 
+        # Artist anchored to the attribution path (folder→artist), resolved
+        # UP-FRONT so the enrich-on-duplicate branches link provenance with the
+        # right artist even when the sidecar carries none — which is now the norm
+        # for archive members under post-first (the per-media sidecar is minimal).
+        # Without this a cross-posted / re-packed archive image deduped and was
+        # left UNLINKED from the new post = a post showing "no images" (#718).
+        _artist_name = derive_top_level_artist(attribution_path, self.import_root)
+        path_artist = self._upsert_artist(_artist_name) if _artist_name else None
+
         # Hash dedup (exact).
         sha = _sha256_of(source)
         existing_stmt = select(ImageRecord).where(ImageRecord.sha256 == sha)
@@ -604,9 +700,9 @@ class Importer:
                 # 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).
+                # post.
                 self._apply_sidecar(
-                    existing, attribution_path, None,
+                    existing, attribution_path, path_artist,
                     explicit_source=explicit_source,
                 )
                 self.session.commit()
@@ -640,7 +736,7 @@ class Importer:
                     larger = self.session.get(ImageRecord, match_id)
                     if larger is not None:
                         self._apply_sidecar(
-                            larger, attribution_path, None,
+                            larger, attribution_path, path_artist,
                             explicit_source=explicit_source,
                         )
                         self.session.commit()
@@ -674,13 +770,11 @@ class Importer:
         self.session.flush()
         self._phash_cache_append(phash, width, height, record.id)
 
-        # Folder→artist (anchored to attribution_path).
+        # Folder→artist (path_artist resolved up-front above); bind it to the
-        artist = None
+        # new record so it carries the canonical Artist.
-        artist_name = derive_top_level_artist(
+        if path_artist is not None and record.artist_id is None:
-            attribution_path, self.import_root
+            record.artist_id = path_artist.id
-        )
+            self.session.flush()
-        if artist_name:
-            artist = self._attach_artist(record, artist_name)
 
         # Sidecar provenance (best-effort; never fails the import).
         # explicit_source lets the FC-3c download path bind the new
@@ -690,7 +784,7 @@ class Importer:
         # subscription-downloaded zip previously lost subscription
         # linkage if the on-disk layout didn't match assumptions.
         self._apply_sidecar(
-            record, attribution_path, artist, explicit_source=explicit_source,
+            record, attribution_path, path_artist, explicit_source=explicit_source,
         )
 
         # Thumbnail is queued separately by the calling task; the importer
@@ -989,17 +1083,6 @@ class Importer:
             self.session.flush()
         return artist
 
-    def _attach_artist(self, record: ImageRecord, artist_name: str) -> Artist:
-        """Upsert the Artist and set it as the image's canonical artist.
-        Artist-kind tags were retired in FC-2d-vii-c — the Artist row is
-        the single source of truth. Returns the Artist (sidecar
-        provenance attaches its Source to it)."""
-        artist = self._upsert_artist(artist_name)
-        if record.artist_id is None:
-            record.artist_id = artist.id
-            self.session.flush()
-        return artist
-
     @staticmethod
     def _sidecar_artist_name(data: dict) -> str | None:
         for k in ("artist", "author_name"):
@@ -1110,6 +1193,13 @@ class Importer:
             external_post_id=epid,
             artist_id=artist.id,
         )
+        # Post-first (#856): on the native path the post-record owns the body/
+        # links/metadata, so the per-media import must NOT write post fields — its
+        # minimal sidecar carries no body, and applying it would clobber
+        # raw_metadata + re-run link-sync off empty data. The image still links
+        # provenance + localization below. gallery-dl (post_first False) is
+        # unchanged: its sidecar IS the only body source, so it still applies.
+        if not self.post_first:
             self._apply_post_fields(post, sd)
 
         # Race-safe (image_record_id, post_id) upsert — mirrors the

backend/app/services/ingest_core.py

@@ -57,6 +57,19 @@ _ERROR_MAX = 1000
 # page-tied update would lurch). Trivial churn (~one single-row UPDATE / 5s).
 _LIVE_PROGRESS_INTERVAL = 5.0
 
+# Post-body schema-drift canary (#862). Patreon's body lives in
+# content/content_json_string with NO post_type gate, so a field rename (as
+# content→content_json_string already was) zeroes EVERY body at once — across
+# every artist, every walk. If a native walk records at least this many posts
+# and extracts a body from NONE of them, treat it as that break (fail the run
+# API_DRIFT) rather than silently archiving empties. A *fraction* threshold would
+# false-positive on gallery/art creators who legitimately post images with no
+# caption, so the gate is "zero across a minimum sample": a real creator nearly
+# always has SOME text across this many posts, a broken parser has none. Set high
+# enough that a small tick (a few new posts) can't trip it — only a backfill /
+# recapture (the operator's schema-test flow) reaches the sample.
+_CANARY_MIN_SAMPLE = 30
+
 
 class Ingester:
     """Generic native-ingest orchestration. Subclass with a platform adapter
@@ -128,9 +141,18 @@ class Ingester:
         # tick has no resumable backfill state.
         checkpoint = mode in ("backfill", "recovery", "recapture")
         ledger_key = self._ledger_key
-        # Optional seams (Patreon native ingester): capture pure-text posts that
+        # POST-FIRST CONTRACT (milestone #67): these two optional seams make a
-        # have NO downloadable media so the artist archive is complete. Absent on
+        # platform "post-first" on the native core ingester — the post-record is
-        # stub clients/downloaders (unit tests) → media-less posts skipped as before.
+        # the single authoritative writer of the post body/links/metadata, and the
+        # per-media sidecar carries image identity only (download_service flips
+        # importer.post_first via uses_native_ingester, so the import side follows
+        # automatically). A platform migrating off gallery-dl onto the native core
+        # adopts post-first by implementing BOTH:
+        #   client.post_record_key(post)   -> (ledger_key, post_id) | None  (gate)
+        #   downloader.write_post_record(post, artist_slug) -> PostRecordOutcome
+        # Absent on stub clients/downloaders (unit tests) and on not-yet-migrated
+        # platforms → media-less posts are skipped as before and the body still
+        # comes from the per-media sidecar (gallery-dl path). See [[post-first-ingest-contract]].
         post_record_key = getattr(self.client, "post_record_key", None)
         write_post_record = getattr(self.downloader, "write_post_record", None)
         start = time.monotonic()
@@ -149,6 +171,12 @@ class Ingester:
         dead_lettered = 0
         skipped_count = 0
         posts_processed = 0
+        # Post-body schema-drift canary counters (#862, native ingester only —
+        # gallery-dl walks never enter the post-record block below so these stay 0
+        # and the canary can't fire there). posts_recorded = post-records attempted
+        # this walk; posts_with_body = how many yielded a non-empty body.
+        posts_recorded = 0
+        posts_with_body = 0
         # Net-new posts THIS chunk for the live progress badge (plan #704 #5);
         # excludes the re-walked resume page so _backfill_posts stays a monotonic
         # absolute across chunks instead of an inflating sum. posts_processed
@@ -261,6 +289,9 @@ class Ingester:
                         )
                         if pkey not in already:
                             rec = write_post_record(post, artist_slug)
+                            posts_recorded += 1
+                            if rec.body_chars:
+                                posts_with_body += 1
                             if rec.path is not None:
                                 post_records.append(str(rec.path))
                                 self._mark_seen(source_id, [(pkey, ppid)])
@@ -414,6 +445,10 @@ class Ingester:
             f"{dead_lettered} dead-lettered, {errors} error(s), "
             f"{posts_processed} post(s), {len(post_records)} post-record(s), "
             f"{len(relink)} relinked"
+            # Body-capture health (#862): even below the canary's red-alarm
+            # threshold, surfacing the ratio makes a partial extraction regression
+            # visible in the Raw stdout (e.g. "bodies 3/180" reads as off).
+            + (f", bodies {posts_with_body}/{posts_recorded}" if posts_recorded else "")
             + (", reached end" if reached_bottom else "")
             + (", time-boxed" if budget_hit else "")
         )
@@ -439,6 +474,27 @@ class Ingester:
                 error_message="Chunk timed out with no progress",
             )
 
+        # Post-body schema-drift canary (#862): a native walk recorded a
+        # meaningful sample of posts but extracted a body from NONE of them. Since
+        # the body field has no post_type gate, that's the signature of Patreon
+        # renaming/restructuring the body field (as content→content_json_string
+        # already was) — fail RED (API_DRIFT: "fix is the field-set/parser, not
+        # creds") so the breakage screams instead of silently archiving empties.
+        # Only reached on an otherwise-clean walk (timeout/stop/error returned
+        # above), so it never masks a more specific failure.
+        if posts_recorded >= _CANARY_MIN_SAMPLE and posts_with_body == 0:
+            msg = (
+                f"Post-body canary: extracted a body from 0 of {posts_recorded} "
+                "posts — Patreon's body field shape likely changed; the ingester "
+                "needs a field-set/parser update."
+            )
+            log_lines.append(msg)
+            log.error("%s (artist=%s)", msg, artist_slug)
+            return _result(
+                success=False, return_code=-1,
+                error_type=ErrorType.API_DRIFT, error_message=msg,
+            )
+
         # Normal success: reached the bottom, or a tick that early-outed. rc 0 +
         # error_type None is REQUIRED for a backfill/recovery walk that reached
         # the bottom to be marked COMPLETE by

backend/app/services/patreon_downloader.py

@@ -529,29 +529,38 @@ class PatreonDownloader:
     def _write_sidecar(
         self, post: dict, media_path: Path, *, source_url: str | None = None
     ) -> Path:
-        """Write the importer-consumed sidecar next to `media_path`.
+        """Write the per-media sidecar next to `media_path` — post-first (#856).
 
-        Patreon uses base-default sidecar keys (parse_sidecar maps
+        On the native ingester the POST-RECORD (`write_post_record` → `_post.json`)
-        category->platform, id->external_post_id, title->post_title,
+        is the sole writer of the post body/links/metadata, captured once per post
-        content->description, published_at->post_date, url->post_url). Patreon
+        BEFORE its media in the walk. So the per-media sidecar carries ONLY
-        registers no derive_post_url, so `url` is trusted as the permalink — we
+        image-specific identity: `category` (platform) + `id` (external_post_id, to
-        pass the post's attributes.url.
+        link provenance to the right Post) + this file's `source_url` (its CDN URL,
-
+        #830 Phase 2 — the importer persists its filehash so the body's inline
-        `source_url` is THIS file's CDN URL (#830 Phase 2). The importer persists
+        `<img src>` remaps to the local copy at render time). No body: writing it
-        its filehash on the ImageRecord so the post body's inline `<img src>` can
+        next to every image duplicated the post body N+1× and risked divergence
-        be remapped to the local copy at render time.
+        (milestone #67). The importer skips post fields for these (post_first).
         """
         return self._write_sidecar_data(
-            post, media_path.with_suffix(".json"), source_url=source_url
+            post, media_path.with_suffix(".json"), source_url=source_url,
+            minimal=True,
         )
 
     def _write_sidecar_data(
-        self, post: dict, sidecar_path: Path, *, source_url: str | None = None
+        self, post: dict, sidecar_path: Path, *, source_url: str | None = None,
+        minimal: bool = False,
     ) -> Path:
-        """Serialize the post's metadata to `sidecar_path`. Shared by the
+        """Serialize the post's metadata to `sidecar_path`. The post-only record
-        per-media sidecar (next to each file) and the post-only sidecar
+        (`write_post_record`) writes the FULL post (body/title/date/url); the
-        (`write_post_record`, for media-less posts). `source_url` is set only for
+        per-media sidecar (`_write_sidecar`, minimal=True) writes only image
-        the per-media sidecar — a media-less post has no source file."""
+        identity (category/id/source_url) — post-first (#856). `source_url` is set
+        only for the per-media sidecar — a media-less post has no source file."""
+        if minimal:
+            data = {"category": "patreon", "id": str(post.get("id") or "")}
+            if source_url:
+                data["source_url"] = source_url
+            sidecar_path.write_text(json.dumps(data, indent=2))
+            return sidecar_path
         attrs = post.get("attributes") or {}
         title = attrs.get("title")
         # Resolve the body HTML from the feed attrs: legacy flat `content`, else

tests/test_importer_archive.py

@@ -93,6 +93,77 @@ def test_archive_imports_members_and_stores_archive(importer, import_layout):
     assert att.post_id == post.id
 
 
+def test_nested_archive_members_imported(importer, import_layout):
+    """#718: a 'high-res' pack that wraps a nested archive must still import the
+    nested images and link them to the post — not silently drop them (incase's
+    per-chapter .rar inside an outer 'High Resolution files' pack)."""
+    import_root, _ = import_layout
+    importer.settings.phash_threshold = 0  # v/h splits are distinct only at 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())  # nested archive, no direct media
+    arc.with_suffix(".cbz.json").write_text(json.dumps(
+        {"category": "patreon", "id": "900", "title": "HR Pack"}))
+
+    r = importer.import_one(arc)
+    assert r.status == "imported"
+    assert len(r.member_image_ids) == 2  # both nested images surfaced
+
+    assert importer.session.execute(
+        select(func.count()).select_from(ImageRecord)
+    ).scalar_one() == 2
+    post = importer.session.execute(select(Post)).scalar_one()
+    provs = importer.session.execute(
+        select(func.count()).select_from(ImageProvenance)
+    ).scalar_one()
+    assert provs == 2  # both nested members linked to the one outer post
+    # The outer pack is still preserved as an attachment.
+    att = importer.session.execute(select(PostAttachment)).scalar_one()
+    assert att.original_filename == "hr.cbz"
+    assert att.post_id == post.id
+
+
+def test_archive_all_deduped_is_benign_not_flagged(importer, import_layout):
+    """#718 reason-string fix: when every image in an archive already exists
+    (cross-posted), the images re-link to the new post and the archive is NOT
+    flagged as an unextracted-archive problem (no error set)."""
+    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"}))
+    assert importer.import_one(first).status == "imported"
+
+    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"}))
+
+    r = importer.import_one(second)
+    assert r.status == "attached"
+    assert r.error is None  # benign all-deduped — NOT a flagged problem
+
+    assert importer.session.execute(
+        select(func.count()).select_from(ImageRecord)
+    ).scalar_one() == 2  # no new records
+    provs = importer.session.execute(
+        select(func.count()).select_from(ImageProvenance)
+    ).scalar_one()
+    assert provs == 4  # 2 images × 2 posts (enrich-on-duplicate)
+
+
 def test_corrupt_archive_still_stored(importer, import_layout):
     import_root, _ = import_layout
     arc = import_root / "Bob" / "broken.zip"

tests/test_importer_attach_in_place.py

@@ -231,6 +231,78 @@ def test_attach_in_place_duplicate_attachment_links_both_posts(importer, db_sync
     assert set(linked) == {"301", "302"}  # one row per post, both present
 
 
+def test_attach_in_place_post_first_skips_post_fields(importer, db_sync):
+    """Post-first (#856, milestone #67): with post_first set (native path), the
+    per-media import links provenance + localization (source_filehash) but does
+    NOT write the post body/title — the post-record owns those. Guards the import
+    side even if a body sneaks into the sidecar."""
+    from backend.app.models import ImageProvenance, Post
+
+    importer.post_first = True
+    images_root = importer.images_root
+    artist = Artist(name="Pat", slug="pat")
+    db_sync.add(artist)
+    db_sync.flush()
+
+    target = images_root / "pat" / "patreon" / "post1" / "img.jpg"
+    _make_jpg(target)
+    cdn = "https://cdn.patreon.com/" + "a" * 32 + "/x.jpg"
+    target.with_suffix(target.suffix + ".json").write_text(
+        '{"category": "patreon", "id": "900", "title": "Should NOT apply", '
+        '"content": "<p>body should NOT apply</p>", '
+        f'"source_url": "{cdn}"}}'
+    )
+
+    result = importer.attach_in_place(target, artist=artist)
+    assert result.status == "imported"
+
+    # Image-specific work STILL happens: localization filehash + provenance link.
+    rec = db_sync.execute(
+        select(ImageRecord).where(ImageRecord.id == result.image_id)
+    ).scalar_one()
+    assert rec.source_filehash == "a" * 32
+    assert rec.primary_post_id is not None
+
+    post = db_sync.get(Post, rec.primary_post_id)
+    assert post.external_post_id == "900"
+    # Post-first: body/title NOT written by the media path (post-record owns them).
+    assert post.description is None
+    assert post.post_title is None
+    # Provenance link exists regardless of post_first.
+    prov = db_sync.execute(
+        select(ImageProvenance).where(
+            ImageProvenance.image_record_id == result.image_id
+        )
+    ).scalars().all()
+    assert len(prov) == 1
+
+
+def test_attach_in_place_post_first_false_applies_post_fields(importer, db_sync):
+    """Default (gallery-dl path): the sidecar IS the body source, so post fields
+    ARE applied. Guards that the post_first flag actually gates this — not a no-op."""
+    from backend.app.models import Post
+
+    assert importer.post_first is False  # default
+    images_root = importer.images_root
+    artist = Artist(name="Gal", slug="gal")
+    db_sync.add(artist)
+    db_sync.flush()
+
+    target = images_root / "gal" / "patreon" / "post1" / "img.jpg"
+    _make_jpg(target)
+    target.with_suffix(target.suffix + ".json").write_text(
+        '{"category": "patreon", "id": "901", "title": "Applies", '
+        '"content": "<p>body applies</p>"}'
+    )
+
+    result = importer.attach_in_place(target, artist=artist)
+    assert result.status == "imported"
+    rec = db_sync.get(ImageRecord, result.image_id)
+    post = db_sync.get(Post, rec.primary_post_id)
+    assert post.description == "<p>body applies</p>"
+    assert post.post_title == "Applies"
+
+
 def test_attach_in_place_invalid_image_returns_skipped(importer):
     images_root = importer.images_root
     bad = images_root / "fred" / "bad.jpg"

tests/test_patreon_downloader.py

@@ -174,8 +174,11 @@ def test_sidecar_written_and_findable(tmp_path):
     data = json.loads(sidecar.read_text())
     assert data["category"] == "patreon"
     assert data["id"] == "1001"
-    assert data["title"] == "My Post Title"
+    # Post-first (#856): the per-media sidecar carries image identity ONLY — no
-    assert data["url"] == "https://www.patreon.com/posts/1001"
+    # post body/title/url. The post-record (`_post.json`) owns those now.
+    assert "title" not in data
+    assert "url" not in data
+    assert "content" not in data
     # #830 Phase 2: the per-media sidecar records THIS file's CDN URL so the
     # importer can persist its filehash for inline-image localization.
     assert data["source_url"] == "https://cdn.patreon.com/media1.png"
@@ -594,40 +597,34 @@ def test_range_ignored_by_server_refetches_clean(tmp_path, monkeypatch):
 # -- content enrichment (adaptive detail-fetch of an empty feed body) -------
 
 
-def test_enriches_empty_content_via_fetcher(tmp_path):
+def test_media_sidecar_is_minimal_post_first(tmp_path):
-    """An empty feed body is backfilled from the detail-fetch seam — once per
+    """Post-first (#856): the per-media sidecar carries ONLY image identity
-    post (memoized on the shared dict) — and lands in the importer sidecar."""
+    (category/id/source_url), NOT the post body — the post-record (`_post.json`)
+    owns the body and is written first in the walk. So download_post never hits
+    the detail fetcher (that's write_post_record's job), and the body is no longer
+    duplicated next to every image."""
     calls: list[str] = []
-
-    def _fetcher(post_id: str) -> str:
-        calls.append(post_id)
-        return "<p>full body</p>"
-
     dl = PatreonDownloader(
         images_root=tmp_path, cookies_path=None, validate=False,
-        session=_FakeSession(), content_fetcher=_fetcher,
+        session=_FakeSession(),
+        content_fetcher=lambda pid: calls.append(pid) or "<p>full body</p>",
     )
     post = _post()
-    post["attributes"]["content"] = ""  # feed returned an empty body
+    post["attributes"]["content"] = ""  # even an empty feed body isn't fetched here
     dl.download_post(post, [_img("a.png"), _img("b.png")], "artist-x")
 
     post_dir = tmp_path / "artist-x" / "patreon" / pd_mod._post_dir_name(post)
     sc = find_sidecar(post_dir / "01_a.png")
     assert sc is not None
-    assert json.loads(sc.read_text())["content"] == "<p>full body</p>"
+    data = json.loads(sc.read_text())
-    # Fetched ONCE for the post, not once per media item.
+    assert data == {
-    assert calls == ["1001"]
+        "category": "patreon",
-
+        "id": "1001",
-
+        "source_url": "https://cdn.patreon.com/a.png",
-def test_no_enrichment_when_feed_body_present(tmp_path):
+    }
-    """A non-empty feed body is trusted as-is; the detail endpoint isn't hit."""
+    # No post body/title next to the image; the detail endpoint is never hit here.
-    calls: list[str] = []
+    assert "content" not in data
-    dl = PatreonDownloader(
+    assert "title" not in data
-        images_root=tmp_path, cookies_path=None, validate=False,
-        session=_FakeSession(),
-        content_fetcher=lambda pid: calls.append(pid) or "x",
-    )
-    dl.download_post(_post(), [_img("a.png")], "artist-x")  # _post body is non-empty
     assert calls == []
 
 

tests/test_patreon_ingester.py

@@ -18,6 +18,7 @@ from backend.app.models import (
     Source,
 )
 from backend.app.services.gallery_dl import ErrorType
+from backend.app.services.ingest_core import _CANARY_MIN_SAMPLE
 from backend.app.services.patreon_client import (
     MediaItem,
     PatreonAPIError,
@@ -49,9 +50,12 @@ class _FakeClient:
     """Stub PatreonClient. `pages` is a list of (page_cursor, [posts]); each post
     is (post_id, [MediaItem]). `raise_on_first` lets a test trip drift."""
 
-    def __init__(self, pages, raise_exc=None):
+    def __init__(self, pages, raise_exc=None, empty_body=False):
         self._pages = pages
         self._raise_exc = raise_exc
+        # empty_body simulates a body-field schema break: every post comes back
+        # with no content (the #862 canary's trip condition).
+        self._empty_body = empty_body
         self.consumed_posts = 0
 
     def iter_posts(self, campaign_id, cursor=None):
@@ -62,6 +66,7 @@ class _FakeClient:
                 self.consumed_posts += 1
                 # Carry feed attributes (title/post_type/content) like the real
                 # client — ingest_core reads these for the per-post diagnostic.
+                content = "" if self._empty_body else f"<p>body {post_id}</p>"
                 yield (
                     {
                         "id": post_id,
@@ -69,7 +74,7 @@ class _FakeClient:
                         "attributes": {
                             "title": f"Post {post_id}",
                             "post_type": "image_file",
-                            "content": f"<p>body {post_id}</p>",
+                            "content": content,
                         },
                     },
                     {},
@@ -941,3 +946,60 @@ async def test_recovery_bypasses_gate_and_re_records(source_id, sync_engine, tmp
                       url="https://patreon.com/ingest", mode="recovery")
     assert len(result.post_record_paths) == 1
     assert dl2.post_records == 1
+
+
+# --- post-body schema-drift canary (#862) --------------------------------
+
+
+def _media_less_posts(n):
+    # A single page of n media-less posts; each still gets a post-record, so the
+    # canary counts all n. Unique ids keep the seen-ledger gate from collapsing them.
+    return [(None, [(f"c{i}", []) for i in range(n)])]
+
+
+@pytest.mark.asyncio
+async def test_body_canary_fails_run_when_all_bodies_empty(source_id, sync_engine, tmp_path):
+    """#862: a native walk that records >= _CANARY_MIN_SAMPLE posts but extracts a
+    body from NONE of them is the signature of a Patreon body-field rename (as
+    content→content_json_string was) — fail the run RED (API_DRIFT) instead of
+    silently archiving empties."""
+    client = _FakeClient(_media_less_posts(_CANARY_MIN_SAMPLE), empty_body=True)
+    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
+
+    result = ing.run(
+        source_id=source_id, campaign_id="c1", artist_slug="ingest",
+        url="https://patreon.com/ingest", mode="backfill",
+    )
+    assert result.success is False
+    assert result.error_type == ErrorType.API_DRIFT
+    assert "canary" in (result.error_message or "").lower()
+
+
+@pytest.mark.asyncio
+async def test_body_canary_silent_when_some_bodies_present(source_id, sync_engine, tmp_path):
+    """No false alarm: a large walk that DOES capture bodies completes normally
+    (normal operation — e.g. a gallery creator who writes captions)."""
+    client = _FakeClient(_media_less_posts(_CANARY_MIN_SAMPLE))  # bodies present
+    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
+
+    result = ing.run(
+        source_id=source_id, campaign_id="c1", artist_slug="ingest",
+        url="https://patreon.com/ingest", mode="backfill",
+    )
+    assert result.success is True
+    assert result.error_type is None
+
+
+@pytest.mark.asyncio
+async def test_body_canary_silent_below_min_sample(source_id, sync_engine, tmp_path):
+    """Tick safety: a small walk (a few new captionless posts) is below the sample
+    floor and must NOT trip the canary, even with every body empty."""
+    client = _FakeClient(_media_less_posts(_CANARY_MIN_SAMPLE - 1), empty_body=True)
+    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
+
+    result = ing.run(
+        source_id=source_id, campaign_id="c1", artist_slug="ingest",
+        url="https://patreon.com/ingest", mode="backfill",
+    )
+    assert result.success is True
+    assert result.error_type is None