feat(ingest): recapture body + links for every walked post (Phase 5)

Operator reframed backfill as inherent to the existing walk: you can't fill
links the system never had by re-downloading media that's already on disk, so
the body/link recapture has to ride the walk itself.

Hoist the post-record capture out of the media-less branch so it runs for EVERY
post — gated once per post by the synthetic post key in the seen-ledger
(detail-fetch for an empty feed body happens at most once; recovery re-captures
unconditionally). A normal BACKFILL now walks history and recaptures each post's
body + external links (which phase 3 imports via upsert_post_record →
_sync_external_links → the download sweep, all already wired). A tick captures
new posts going forward. No separate button — the backfill is the backfill.

Tests: media posts now also carry a synthetic post-key ledger row (count
assertions +1); new test proves an already-on-disk media post still recaptures
its body/links on a re-walk.

Completes the core of #830 (Phase 5). Phase 2 (inline-image localization)
remains.

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

backend/app/services/ingest_core.py

@@ -227,25 +227,31 @@ class Ingester:
                 # resume_cursor None, so everything counts.
                 if not (resume_cursor and page_cursor == resume_cursor):
                     chunk_new_posts += 1
+                # Capture the post body + external links ONCE per post (gated by
+                # the synthetic post key in the seen-ledger), for EVERY post —
+                # whether or not it has downloadable media. This is what makes a
+                # backfill/recovery re-walk RECAPTURE bodies + links for posts
+                # whose media is already on disk: re-downloading existing media
+                # never fills links the system never had, so the body recapture
+                # has to ride the walk itself. Detail-fetch (for an empty feed
+                # body) happens at most once per post — the gate then spares it on
+                # later walks. bypass_seen (recovery) re-captures unconditionally.
+                if post_record_key and write_post_record:
+                    rk = post_record_key(post)
+                    if rk is not None:
+                        pkey, ppid = rk
+                        already = (
+                            set() if bypass_seen
+                            else self._seen_keys(source_id, [pkey])
+                        )
+                        if pkey not in already:
+                            rec_path = write_post_record(post, artist_slug)
+                            if rec_path is not None:
+                                post_records.append(str(rec_path))
+                                self._mark_seen(source_id, [(pkey, ppid)])
+
                 media = self.client.extract_media(post, included)
                 if not media:
-                    # No downloadable media — still capture the post itself
-                    # (title/body/external links) so text-only posts aren't lost.
-                    # Gate via the seen-ledger (synthetic post key) so a text post
-                    # is detail-fetched + recorded ONCE, not re-walked every tick.
-                    if post_record_key and write_post_record:
-                        rk = post_record_key(post)
-                        if rk is not None:
-                            pkey, ppid = rk
-                            already = (
-                                set() if bypass_seen
-                                else self._seen_keys(source_id, [pkey])
-                            )
-                            if pkey not in already:
-                                rec_path = write_post_record(post, artist_slug)
-                                if rec_path is not None:
-                                    post_records.append(str(rec_path))
-                                    self._mark_seen(source_id, [(pkey, ppid)])
                     continue
 
                 keys = [ledger_key(m) for m in media]

tests/test_patreon_ingester.py

@@ -173,7 +173,10 @@ async def test_tick_downloads_unseen_and_marks_seen(source_id, sync_engine, tmp_
     # plan #704: structured run_stats carry the real counts.
     assert result.run_stats["downloaded_count"] == 2
     assert result.posts_processed == 1
-    assert _count_ledger(sync_engine, source_id) == 2
+    # 2 media keys + 1 synthetic post key (body/links recaptured per post).
+    assert _count_ledger(sync_engine, source_id) == 3
+    # The post body + links are captured for media posts too (rides the walk).
+    assert len(result.post_record_paths) == 1
 
 
 @pytest.mark.asyncio
@@ -196,8 +199,9 @@ async def test_quarantined_media_surfaced_in_result(source_id, sync_engine, tmp_
     assert result.run_stats["quarantined_count"] == 1
     assert result.run_stats["downloaded_count"] == 1
     assert len(result.written_paths) == 1        # quarantined NOT written
-    # Quarantined media is NOT marked seen (a fixed file may be re-fetched).
-    assert _count_ledger(sync_engine, source_id) == 1
+    # Quarantined media is NOT marked seen (a fixed file may be re-fetched);
+    # m1 + the synthetic post key (body/links captured per post) = 2.
+    assert _count_ledger(sync_engine, source_id) == 2
 
 
 @pytest.mark.asyncio
@@ -550,8 +554,31 @@ async def test_recovery_tier2_disk_still_skips(source_id, sync_engine, tmp_path)
     )
     assert result.files_downloaded == 0
     assert downloader.download_calls == 0
-    # Disk-skip still reconciles the ledger so a later tick skips at tier-1.
-    assert _count_ledger(sync_engine, source_id) == 1
+    # Disk-skip reconciles the media key + the synthetic post key (recovery
+    # recaptures the body/links per post) = 2.
+    assert _count_ledger(sync_engine, source_id) == 2
+
+
+@pytest.mark.asyncio
+async def test_backfill_recaptures_body_for_already_downloaded_post(
+    source_id, sync_engine, tmp_path,
+):
+    """Phase 5 guarantee: a post whose media is ALREADY on disk still gets its
+    body + external links recaptured on a re-walk. Re-downloading existing media
+    can never fill links the system never had — so the recapture rides the walk
+    itself, and a normal backfill is the backfill."""
+    m1 = _media("p1", 1)
+    client = _FakeClient([(None, [("p1", [m1])])])
+    downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)})
+    ing = _ingester(sync_engine, tmp_path, client, downloader)
+
+    result = ing.run(
+        source_id=source_id, campaign_id="c1", artist_slug="ingest",
+        url="https://patreon.com/ingest", mode="backfill",
+    )
+    assert result.files_downloaded == 0          # media already on disk
+    assert len(result.post_record_paths) == 1    # body/links recaptured anyway
+    assert downloader.post_records == 1
 
 
 # --- dead-letter ledger (plan #705 #7) ------------------------------------