feat(ingest): post-body schema-drift canary — fail a native walk red when zero bodies extracted (#862)

If Patreon renames/restructures the post body field again (as content →
content_json_string already did), every body silently comes back empty and we'd
archive empty posts without noticing. Surface that as a loud failure.

Research-grounded design (Patreon `content` is officially null|string, body has
no post_type gate, gallery-dl independently added the same content_json_string
fallback): empty bodies are LEGITIMATE for gallery/art posts, so a fraction
threshold would false-positive constantly. The robust, creator-independent break
signature is "a meaningful sample of posts, a body extracted from NONE of them."

- ingest_core counts posts_recorded / posts_with_body on the native post-record
  path (gallery-dl never enters it, so the canary is native-only by construction).
- When posts_recorded >= _CANARY_MIN_SAMPLE (30) and posts_with_body == 0, return
  ErrorType.API_DRIFT (maps to task_run status "error" — red; its semantics are
  literally "fix the field-set/parser, not creds"). Placed after the timeout/stop
  returns so it never masks a more specific failure.
- Run summary always appends "bodies X/Y" for sub-threshold observability (a
  partial regression that still extracts some bodies shows in the Raw stdout).

Tests: zero bodies over the sample -> API_DRIFT; bodies present -> success;
below the sample floor -> success (tick safety).

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

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