feat(patreon): structured ingester results + quarantine surfacing — #704 step 1

The native ingester faked gallery-dl stdout (`Cursor:` lines, summary) and
phase 3 regex-scraped it back — so Patreon run-stats were mostly zero and
quarantine stats blank. We own the ingester, so it now RETURNS structured
data and phase 3 reads it directly.

- DownloadResult gains run_stats/cursor/posts_processed (None/0 on the
  gallery-dl path, which keeps the text route).
- Ingester builds real run_stats from per-media outcome counts, sets the
  checkpoint cursor structurally (no fake `Cursor:` stdout), and counts
  posts processed. download_service phase 3 uses dl_result.run_stats when
  present; the backfill lifecycle + TIMEOUT→PARTIAL block checkpoint
  dl_result.cursor instead of parse_last_cursor(stdout).
- #4 quarantine: PatreonDownloader reports a distinct "quarantined"
  MediaOutcome (with the _quarantine dest); the ingester surfaces a real
  files_quarantined + quarantined_paths + run_stats.quarantined_count
  (was hardcoded 0). Quarantined media isn't written or marked seen.
- Cleanup: parse_last_cursor + _CURSOR_RE (and the now-unused `import re`)
  removed from gallery_dl — the structured cursor replaced the scrape.

Tests: ingester result carries real run_stats/cursor/posts_processed +
quarantine counts; downloader quarantines an invalid file as "quarantined";
backfill cursor tests pass cursor= structurally; dropped the
parse_last_cursor tests.

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

tests/test_patreon_ingester.py

@@ -63,11 +63,13 @@ class _FakeClient:
 
 class _FakeDownloader:
     """Stub PatreonDownloader. Honors the injected is_seen (tier-1); media in
-    `on_disk` report skipped_disk (tier-2); everything else downloads."""
+    `on_disk` report skipped_disk (tier-2); media in `quarantine` report
+    quarantined; everything else downloads."""
 
-    def __init__(self, tmp_path, on_disk=None):
+    def __init__(self, tmp_path, on_disk=None, quarantine=None):
         self.tmp_path = tmp_path
         self.on_disk = set(on_disk or ())
+        self.quarantine = set(quarantine or ())
         self.download_calls = 0
 
     def download_post(self, post, media_items, artist_slug, *, is_seen):
@@ -78,6 +80,10 @@ class _FakeDownloader:
             elif _ledger_key(m) in self.on_disk:
                 p = self.tmp_path / f"{m.post_id}_{m.filename}"
                 outcomes.append(MediaOutcome(media=m, status="skipped_disk", path=p, error=None))
+            elif _ledger_key(m) in self.quarantine:
+                self.download_calls += 1
+                q = self.tmp_path / "_quarantine" / f"{m.post_id}_{m.filename}"
+                outcomes.append(MediaOutcome(media=m, status="quarantined", path=q, error="bad image"))
             else:
                 self.download_calls += 1
                 p = self.tmp_path / f"{m.post_id}_{m.filename}"
@@ -137,9 +143,36 @@ async def test_tick_downloads_unseen_and_marks_seen(source_id, sync_engine, tmp_
     assert result.return_code == 0
     assert result.files_downloaded == 2
     assert len(result.written_paths) == 2
+    # 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
 
 
+@pytest.mark.asyncio
+async def test_quarantined_media_surfaced_in_result(source_id, sync_engine, tmp_path):
+    """plan #704 (#4): a media that fails validation is reported as quarantined —
+    a real files_quarantined count + paths + run_stats — not a blank 0, and is
+    NOT written or marked seen."""
+    m1, m2 = _media("p1", 1), _media("p1", 2)
+    client = _FakeClient([(None, [("p1", [m1, m2])])])
+    downloader = _FakeDownloader(tmp_path, quarantine={_ledger_key(m2)})
+    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="tick",
+    )
+    assert result.files_downloaded == 1          # only m1
+    assert result.files_quarantined == 1         # m2
+    assert len(result.quarantined_paths) == 1
+    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
+
+
 @pytest.mark.asyncio
 async def test_tick_skips_seen_via_ledger(source_id, sync_engine, tmp_path):
     m1 = _media("p1", 1)
@@ -206,8 +239,10 @@ async def test_backfill_reaches_bottom_signals_complete(source_id, sync_engine,
     assert result.return_code == 0
     assert result.error_type is None
     assert result.files_downloaded == 2
-    # The page-2 cursor was emitted for checkpointing.
-    assert "Cursor: CUR2" in result.stdout
+    # The page-2 cursor is carried structurally for checkpointing (plan #704).
+    assert result.cursor == "CUR2"
+    assert result.posts_processed == 2
+    assert result.run_stats["downloaded_count"] == 2
 
 
 @pytest.mark.asyncio
@@ -245,7 +280,7 @@ async def test_backfill_budget_cut_returns_partial_with_progress(
     assert result.return_code == -1
     assert result.files_downloaded == 1  # post1 downloaded before the cut
     assert downloader.download_calls == 1  # post2's body never ran
-    assert "Cursor: CUR1" in result.stdout
+    assert result.cursor == "CUR1"  # structured checkpoint (plan #704)
 
 
 # --- recovery -------------------------------------------------------------