feat(patreon): phase-2 ingester integration — build step 3 (plan #697)

Branch download_service phase 2 by platform: Patreon now routes to the
native PatreonIngester (zero per-file HEADs, native cursor/resume, loud
drift detection) instead of gallery-dl; the other 5 platforms are
unchanged. The ingester returns a DownloadResult-shaped object so phase 1
(DB setup) and phase 3 (import → pHash → thumbs → ML) are untouched.

Three modes wired from config_overrides state:
  - tick: skip seen (tier-1 ledger + tier-2 disk), early-out after N
    contiguous already-have-it items.
  - backfill: full-history time-boxed chunk, cursor checkpoint via
    gallery-dl-style "Cursor: <token>" lines in stdout (reuses the #693
    lifecycle + parse_last_cursor verbatim).
  - recovery: backfill that BYPASSES the tier-1 seen-ledger so
    dropped-and-deleted near-dups get re-fetched and re-evaluated under
    the current pHash threshold. Rides the #693 state machine via a
    _backfill_bypass_seen flag, cleared on completion / stop.

The seen-ledger uses short-lived sync sessions (injected sessionmaker),
never held across the walk (avoids the connection-reaping trap). Campaign
id resolves from override, an id: URL, or a vanity lookup; unresolvable =
loud NOT_FOUND, never a silent empty success.

Tests: new test_patreon_ingester.py (modes, ledger skip/idempotency,
budget→PARTIAL, recovery bypass, tier-2 disk, drift). The patreon-oriented
download_service tests now drive the ingester branch via a stub; the
gallery-dl campaign-retry test is replaced by resolution/caching coverage.

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

tests/test_download_service.py

@@ -94,6 +94,23 @@ def _fake_gdl_with_result(result):
     return fake
 
 
+def _stub_patreon_ingester(svc, result, resolved_campaign_id=None):
+    """Route the Patreon phase-2 branch (plan #697 native ingester) to a canned
+    DownloadResult so these tests exercise download_service's
+    phase-2-result→phase-3 handling (status mapping, backfill lifecycle, health)
+    without the real ingester's network/DB. The ingester itself is covered by
+    test_patreon_ingester.py. Returns a list capturing (ctx, source_config, mode)
+    per call so a test can assert mode / resume_cursor threading."""
+    calls = []
+
+    async def fake(ctx, source_config, mode):
+        calls.append({"ctx": ctx, "source_config": source_config, "mode": mode})
+        return result, resolved_campaign_id
+
+    svc._run_patreon_ingester = fake
+    return calls
+
+
 @pytest.mark.asyncio
 async def test_download_source_attaches_written_files(
     db, db_sync, tmp_path, seed_artist_and_source,
@@ -112,12 +129,13 @@ async def test_download_source_attaches_written_files(
     _make_jpg(f1, split="h")
     _make_jpg(f2, split="v")
 
-    fake_gdl = _fake_gdl_with_result(_make_fake_dl_result(
+    result = _make_fake_dl_result(
         success=True,
         written_paths=[str(f1), str(f2)],
         files_downloaded=2,
         stdout=f"{f1}\n{f2}\n",
-    ))
+    )
+    fake_gdl = _fake_gdl_with_result(result)
 
     sync_settings = db_sync.execute(
         select(ImportSettings).where(ImportSettings.id == 1)
@@ -137,6 +155,7 @@ async def test_download_source_attaches_written_files(
         async_session=db, sync_session=db_sync,
         gdl=fake_gdl, importer=importer, cred_service=cred_service,
     )
+    _stub_patreon_ingester(svc, result)
     event_id = await svc.download_source(source.id)
 
     ev = (await db.execute(
@@ -214,40 +233,19 @@ async def test_in_flight_idempotency_returns_existing_event(
 
 
 @pytest.mark.asyncio
-async def test_patreon_retry_caches_campaign_id(
-    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
+async def test_patreon_resolved_campaign_id_is_cached(
+    db, db_sync, tmp_path, seed_artist_and_source,
 ):
+    """plan #697: the native ingester resolves a Patreon vanity → campaign id
+    on the fly (replacing gallery-dl's reactive campaign-id retry). When phase 2
+    reports a freshly-resolved id, phase 3 caches it on the source so later runs
+    skip the lookup. Stub the ingester to report a resolved id; assert it lands
+    in config_overrides."""
     from backend.app.services.download_service import DownloadService
     from backend.app.services.importer import Importer
 
     _artist, source = seed_artist_and_source
 
-    failed = _make_fake_dl_result(
-        success=False, written_paths=[], stdout="",
-        stderr="[patreon][error] Failed to extract campaign ID for alice\n",
-    )
-    succeeded = _make_fake_dl_result(success=True, written_paths=[])
-
-    download_calls = []
-
-    async def fake_download(url, **k):
-        download_calls.append(url)
-        return failed if len(download_calls) == 1 else succeeded
-
-    fake_gdl = MagicMock()
-    fake_gdl.download = fake_download
-    fake_gdl._compute_run_stats = lambda *a, **k: {
-        "exit_code": 0, "downloaded_count": 0, "skipped_count": 0,
-        "per_item_failures": 0, "warning_count": 0, "tier_gated_count": 0,
-    }
-    fake_gdl._extract_errors_warnings = lambda *a, **k: ""
-    fake_gdl._truncate_log = lambda x, **k: x
-
-    monkeypatch.setattr(
-        "backend.app.services.download_service.resolve_campaign_id",
-        AsyncMock(return_value="99"),
-    )
-
     sync_settings = db_sync.execute(
         select(ImportSettings).where(ImportSettings.id == 1)
     ).scalar_one()
@@ -261,19 +259,102 @@ async def test_patreon_retry_caches_campaign_id(
 
     svc = DownloadService(
         async_session=db, sync_session=db_sync,
-        gdl=fake_gdl, importer=importer, cred_service=cred_service,
+        gdl=_fake_gdl_with_result(_make_fake_dl_result(success=True)),
+        importer=importer, cred_service=cred_service,
+    )
+    _stub_patreon_ingester(
+        svc, _make_fake_dl_result(success=True, written_paths=[]),
+        resolved_campaign_id="99",
     )
     await svc.download_source(source.id)
 
-    assert len(download_calls) == 2
-    assert "id:99" in download_calls[1]
-
     overrides = db_sync.execute(
         select(Source.config_overrides).where(Source.id == source.id)
     ).scalar_one()
     assert overrides.get("patreon_campaign_id") == "99"
 
 
+@pytest.mark.asyncio
+async def test_run_patreon_ingester_resolves_vanity_and_runs(
+    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
+):
+    """_run_patreon_ingester: with no cached campaign id, resolve the vanity and
+    pass the resolved id straight to the ingester; report it back so phase 3 can
+    cache it."""
+    from backend.app.services import download_service as dl_mod
+    from backend.app.services.download_service import DownloadService
+    from backend.app.services.gallery_dl import SourceConfig
+
+    _artist, source = seed_artist_and_source
+
+    monkeypatch.setattr(
+        dl_mod, "resolve_campaign_id", AsyncMock(return_value="4242"),
+    )
+    run_kwargs = {}
+
+    class _FakeIngester:
+        def __init__(self, **kw):
+            pass
+
+        def run(self, **kw):
+            run_kwargs.update(kw)
+            return _make_fake_dl_result(success=True, written_paths=[])
+
+    monkeypatch.setattr(dl_mod, "PatreonIngester", _FakeIngester)
+
+    svc = DownloadService(
+        async_session=db, sync_session=db_sync,
+        gdl=_fake_gdl_with_result(_make_fake_dl_result(success=True)),
+        importer=MagicMock(), cred_service=MagicMock(),
+        sync_session_factory=MagicMock(),
+    )
+    ctx = {
+        "source_id": source.id, "url": "https://patreon.com/alice",
+        "artist_slug": "alice", "cookies_path": None,
+        "config_overrides": {},
+    }
+    result, resolved = await svc._run_patreon_ingester(
+        ctx, SourceConfig.from_dict({}), "tick",
+    )
+    assert result.success is True
+    assert resolved == "4242"
+    assert run_kwargs["campaign_id"] == "4242"
+    assert run_kwargs["mode"] == "tick"
+
+
+@pytest.mark.asyncio
+async def test_run_patreon_ingester_unresolvable_fails_loud(
+    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
+):
+    """A campaign id we can't resolve is a loud NOT_FOUND failure, never a
+    silent empty success."""
+    from backend.app.services import download_service as dl_mod
+    from backend.app.services.download_service import DownloadService
+    from backend.app.services.gallery_dl import ErrorType, SourceConfig
+
+    _artist, source = seed_artist_and_source
+
+    monkeypatch.setattr(
+        dl_mod, "resolve_campaign_id", AsyncMock(return_value=None),
+    )
+    svc = DownloadService(
+        async_session=db, sync_session=db_sync,
+        gdl=MagicMock(), importer=MagicMock(), cred_service=MagicMock(),
+        sync_session_factory=MagicMock(),
+    )
+    ctx = {
+        "source_id": source.id, "url": "https://patreon.com/alice",
+        "artist_slug": "alice", "cookies_path": None,
+        "config_overrides": {},
+    }
+    result, resolved = await svc._run_patreon_ingester(
+        ctx, SourceConfig.from_dict({}), "tick",
+    )
+    assert result.success is False
+    assert result.error_type == ErrorType.NOT_FOUND
+    assert resolved is None
+
+
 # --- FC-3d: finalize hook updates Source health columns -------------------
 
 
@@ -376,8 +457,12 @@ async def test_finalize_skipped_preserves_failures_clears_error(db):
 
 
 def _backfill_svc(db, db_sync, tmp_path, result):
-    """DownloadService wired with a fake gdl returning `result` + a real
-    importer (so an empty written_paths just attaches nothing)."""
+    """DownloadService wired with a real importer (so empty written_paths just
+    attaches nothing) whose Patreon phase-2 branch is stubbed to return `result`.
+
+    The seeded source is Patreon, so phase 2 routes to the native ingester
+    (plan #697); the stub returns the canned result + captures the per-call
+    (ctx, source_config, mode). Returns (svc, ingester_calls)."""
     from backend.app.services.download_service import DownloadService
     from backend.app.services.importer import Importer
 
@@ -395,7 +480,8 @@ def _backfill_svc(db, db_sync, tmp_path, result):
         async_session=db, sync_session=db_sync,
         gdl=fake_gdl, importer=importer, cred_service=cred_service,
     )
-    return svc, fake_gdl
+    calls = _stub_patreon_ingester(svc, result)
+    return svc, calls
 
 
 @pytest.mark.asyncio
@@ -430,28 +516,25 @@ async def test_backfill_chunk_progress_advances_cursor(
 async def test_backfill_state_running_selects_backfill_mode_and_resumes(
     db, db_sync, tmp_path, seed_artist_and_source,
 ):
-    """state=='running' drives backfill mode (skip:True + chunk budget) and
-    threads the stored cursor to gallery-dl as the resume point."""
-    from backend.app.services.gallery_dl import (
-        BACKFILL_CHUNK_SECONDS,
-        BACKFILL_SKIP_VALUE,
-    )
+    """state=='running' drives backfill mode (chunk budget) and threads the
+    stored cursor to the native ingester as the resume point."""
+    from backend.app.services.gallery_dl import BACKFILL_CHUNK_SECONDS
 
     _artist, source = seed_artist_and_source
     source.config_overrides = {"_backfill_state": "running", "_backfill_cursor": "03:RESUME:here"}
     source.backfill_runs_remaining = 5
     await db.commit()
 
-    svc, fake_gdl = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
+    svc, calls = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
         success=False, written_paths=[],
         stderr="[patreon][debug] Cursor: 03:RESUME2:next\n",
     ))
     await svc.download_source(source.id)
 
-    kwargs = fake_gdl.download.call_args.kwargs
-    assert kwargs["skip_value"] is BACKFILL_SKIP_VALUE
-    assert kwargs["source_config"].resume_cursor == "03:RESUME:here"
-    assert kwargs["source_config"].timeout == BACKFILL_CHUNK_SECONDS
+    assert len(calls) == 1
+    assert calls[0]["mode"] == "backfill"
+    assert calls[0]["source_config"].resume_cursor == "03:RESUME:here"
+    assert calls[0]["source_config"].timeout == BACKFILL_CHUNK_SECONDS
 
     co = (await db.execute(
         select(Source.config_overrides).where(Source.id == source.id)
@@ -605,6 +688,32 @@ async def test_tick_mode_when_not_running_leaves_state_untouched(
     assert (co or {}).get("_backfill_state") is None
 
 
+@pytest.mark.asyncio
+async def test_recovery_mode_selected_and_flag_cleared_on_complete(
+    db, db_sync, tmp_path, seed_artist_and_source,
+):
+    """plan #697: `_backfill_bypass_seen` alongside a running backfill selects
+    recovery mode (ingester bypasses the seen-ledger). A clean rc=0 walk
+    completes the shared lifecycle AND clears the bypass flag so the next tick
+    honors the ledger again."""
+    _artist, source = seed_artist_and_source
+    source.config_overrides = {"_backfill_state": "running", "_backfill_bypass_seen": True}
+    source.backfill_runs_remaining = 5
+    await db.commit()
+
+    svc, calls = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
+        success=True, written_paths=[], files_downloaded=0,
+    ))
+    await svc.download_source(source.id)
+
+    assert calls[0]["mode"] == "recovery"
+    co = (await db.execute(
+        select(Source.config_overrides).where(Source.id == source.id)
+    )).scalar_one()
+    assert co.get("_backfill_state") == "complete"
+    assert "_backfill_bypass_seen" not in co
+
+
 @pytest.mark.asyncio
 async def test_partial_error_type_maps_to_ok_status(
     db, db_sync, tmp_path, seed_artist_and_source,
@@ -642,6 +751,7 @@ async def test_partial_error_type_maps_to_ok_status(
         async_session=db, sync_session=db_sync,
         gdl=fake_gdl, importer=importer, cred_service=cred_service,
     )
+    _stub_patreon_ingester(svc, fake_result)
     event_id = await svc.download_source(source.id)
 
     ev = (await db.execute(
@@ -677,10 +787,11 @@ async def test_download_enqueues_thumbnail_and_ml_per_attached_image(
     _make_jpg(f1, split="h")
     _make_jpg(f2, split="v")
 
-    fake_gdl = _fake_gdl_with_result(_make_fake_dl_result(
+    result = _make_fake_dl_result(
         success=True, written_paths=[str(f1), str(f2)],
         files_downloaded=2, stdout=f"{f1}\n{f2}\n",
-    ))
+    )
+    fake_gdl = _fake_gdl_with_result(result)
 
     sync_settings = db_sync.execute(
         select(ImportSettings).where(ImportSettings.id == 1)
@@ -713,6 +824,7 @@ async def test_download_enqueues_thumbnail_and_ml_per_attached_image(
         async_session=db, sync_session=db_sync,
         gdl=fake_gdl, importer=importer, cred_service=cred_service,
     )
+    _stub_patreon_ingester(svc, result)
     await svc.download_source(source.id)
 
     # Two files attached → two thumbnail enqueues + two ML enqueues, IDs
@@ -727,11 +839,11 @@ async def test_download_enqueues_thumbnail_and_ml_per_attached_image(
 async def test_releases_db_connections_before_subprocess(
     db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
 ):
-    """Phase 1's DB connections must be released BEFORE the (up to ~19.5-min
-    in backfill) gallery-dl subprocess, so they don't idle-die and strand
-    phase 3 with ConnectionDoesNotExistError (Anduo #40014). Spy on the
-    async session close and assert it happened before gdl.download runs;
-    phase 3 must still finalize the event."""
+    """Phase 1's DB connections must be released BEFORE the (multi-minute)
+    phase-2 fetch, so they don't idle-die and strand phase 3 with
+    ConnectionDoesNotExistError (Anduo #40014). Spy on the async session close
+    and assert it happened before the phase-2 work (here the native Patreon
+    ingester) runs; phase 3 must still finalize the event."""
     from backend.app.services.download_service import DownloadService
     from backend.app.services.importer import Importer
 
@@ -747,19 +859,9 @@ async def test_releases_db_connections_before_subprocess(
     monkeypatch.setattr(db, "close", spy_close)
 
     seen = {}
-
-    async def fake_download(url, **k):
-        seen["closed_before_download"] = closed["async"]
-        return _make_fake_dl_result(success=True, written_paths=[], stdout="")
-
-    fake_gdl = MagicMock()
-    fake_gdl.download = fake_download
-    fake_gdl._compute_run_stats = lambda *a, **k: {
-        "exit_code": 0, "downloaded_count": 0, "skipped_count": 0,
-        "per_item_failures": 0, "warning_count": 0, "tier_gated_count": 0,
-    }
-    fake_gdl._extract_errors_warnings = lambda *a, **k: ""
-    fake_gdl._truncate_log = lambda x, **k: x
+    fake_gdl = _fake_gdl_with_result(
+        _make_fake_dl_result(success=True, written_paths=[], stdout="")
+    )
 
     sync_settings = db_sync.execute(
         select(ImportSettings).where(ImportSettings.id == 1)
@@ -776,9 +878,15 @@ async def test_releases_db_connections_before_subprocess(
         async_session=db, sync_session=db_sync,
         gdl=fake_gdl, importer=importer, cred_service=cred_service,
     )
+
+    async def fake_ingest(ctx, source_config, mode):
+        seen["closed_before_phase2"] = closed["async"]
+        return _make_fake_dl_result(success=True, written_paths=[], stdout=""), None
+
+    svc._run_patreon_ingester = fake_ingest
     event_id = await svc.download_source(source.id)
 
-    assert seen["closed_before_download"] is True
+    assert seen["closed_before_phase2"] is True
     ev = (await db.execute(
         select(DownloadEvent).where(DownloadEvent.id == event_id)
     )).scalar_one()

tests/test_patreon_ingester.py

@@ -0,0 +1,324 @@
+"""PatreonIngester tests (native ingester build step 3).
+
+The HTTP client and the media downloader are stubbed (injected seams), so these
+exercise the ingester's own logic — mode behavior, the two-tier skip, cursor
+emission, the budget cutoff, and the Postgres seen-ledger — without network or a
+real CDN. The ledger is real (a sync sessionmaker bound to the test engine), so
+the tier-1 skip and the idempotent mark-seen run against actual rows.
+"""
+
+import pytest
+from sqlalchemy import func, select
+from sqlalchemy.orm import sessionmaker
+
+from backend.app.models import Artist, PatreonSeenMedia, Source
+from backend.app.services.gallery_dl import ErrorType
+from backend.app.services.patreon_client import MediaItem, PatreonDriftError
+from backend.app.services.patreon_downloader import MediaOutcome
+from backend.app.services.patreon_ingester import PatreonIngester, _ledger_key
+
+pytestmark = pytest.mark.integration
+
+
+# --- fakes ----------------------------------------------------------------
+
+
+def _media(post_id, n, *, filehash=None, kind="images"):
+    # Deterministic 32-hex-ish key, unique per (post_id, n).
+    fh = filehash if filehash is not None else f"{post_id}{n}".encode().hex().ljust(32, "0")[:32]
+    return MediaItem(
+        url=f"https://cdn.patreon.com/{fh}/{n}.jpg",
+        filename=f"{n}.jpg",
+        kind=kind,
+        filehash=fh,
+        post_id=post_id,
+    )
+
+
+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):
+        self._pages = pages
+        self._raise_exc = raise_exc
+        self.consumed_posts = 0
+
+    def iter_posts(self, campaign_id, cursor=None):
+        if self._raise_exc is not None:
+            raise self._raise_exc
+        for page_cursor, posts in self._pages:
+            for post_id, media in posts:
+                self.consumed_posts += 1
+                yield {"id": post_id, "_media": media}, {}, page_cursor
+
+    def extract_media(self, post, included_index):
+        return post["_media"]
+
+
+class _FakeDownloader:
+    """Stub PatreonDownloader. Honors the injected is_seen (tier-1); media in
+    `on_disk` report skipped_disk (tier-2); everything else downloads."""
+
+    def __init__(self, tmp_path, on_disk=None):
+        self.tmp_path = tmp_path
+        self.on_disk = set(on_disk or ())
+        self.download_calls = 0
+
+    def download_post(self, post, media_items, artist_slug, *, is_seen):
+        outcomes = []
+        for m in media_items:
+            if is_seen(m):
+                outcomes.append(MediaOutcome(media=m, status="skipped_seen", path=None, error=None))
+            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))
+            else:
+                self.download_calls += 1
+                p = self.tmp_path / f"{m.post_id}_{m.filename}"
+                p.write_bytes(b"x")
+                outcomes.append(MediaOutcome(media=m, status="downloaded", path=p, error=None))
+        return outcomes
+
+
+@pytest.fixture
+async def source_id(db):
+    artist = Artist(name="Ingest", slug="ingest")
+    db.add(artist)
+    await db.flush()
+    source = Source(
+        artist_id=artist.id, platform="patreon",
+        url="https://patreon.com/ingest", enabled=True, config_overrides={},
+    )
+    db.add(source)
+    await db.commit()
+    return source.id
+
+
+def _ingester(sync_engine, tmp_path, client, downloader):
+    factory = sessionmaker(sync_engine, expire_on_commit=False)
+    return PatreonIngester(
+        images_root=tmp_path, cookies_path=None, session_factory=factory,
+        client=client, downloader=downloader,
+    )
+
+
+def _count_ledger(sync_engine, source_id):
+    factory = sessionmaker(sync_engine, expire_on_commit=False)
+    with factory() as s:
+        return s.execute(
+            select(func.count(PatreonSeenMedia.id)).where(
+                PatreonSeenMedia.source_id == source_id
+            )
+        ).scalar_one()
+
+
+# --- tick -----------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_tick_downloads_unseen_and_marks_seen(source_id, sync_engine, tmp_path):
+    m1, m2 = _media("p1", 1), _media("p1", 2)
+    client = _FakeClient([(None, [("p1", [m1, m2])])])
+    downloader = _FakeDownloader(tmp_path)
+    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.success is True
+    assert result.return_code == 0
+    assert result.files_downloaded == 2
+    assert len(result.written_paths) == 2
+    assert _count_ledger(sync_engine, source_id) == 2
+
+
+@pytest.mark.asyncio
+async def test_tick_skips_seen_via_ledger(source_id, sync_engine, tmp_path):
+    m1 = _media("p1", 1)
+    # Pre-seed the ledger with m1's key.
+    factory = sessionmaker(sync_engine, expire_on_commit=False)
+    with factory() as s:
+        s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m1), post_id="p1"))
+        s.commit()
+
+    client = _FakeClient([(None, [("p1", [m1])])])
+    downloader = _FakeDownloader(tmp_path)
+    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 == 0
+    assert downloader.download_calls == 0
+
+
+@pytest.mark.asyncio
+async def test_tick_early_out_after_threshold(source_id, sync_engine, tmp_path):
+    # Three posts, one already-seen media each; threshold 2 → stop before the 3rd.
+    seen_media = [_media(f"p{i}", 1) for i in range(1, 4)]
+    factory = sessionmaker(sync_engine, expire_on_commit=False)
+    with factory() as s:
+        for m in seen_media:
+            s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m), post_id=m.post_id))
+        s.commit()
+
+    pages = [(None, [(m.post_id, [m]) for m in seen_media])]
+    client = _FakeClient(pages)
+    downloader = _FakeDownloader(tmp_path)
+    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", seen_threshold=2,
+    )
+    assert result.success is True
+    # Stopped after the 2nd contiguous seen item — never consumed the 3rd post.
+    assert client.consumed_posts == 2
+
+
+# --- backfill -------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_backfill_reaches_bottom_signals_complete(source_id, sync_engine, tmp_path):
+    pages = [
+        (None, [("p1", [_media("p1", 1)])]),
+        ("CUR2", [("p2", [_media("p2", 1)])]),
+    ]
+    client = _FakeClient(pages)
+    downloader = _FakeDownloader(tmp_path)
+    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",
+    )
+    # rc 0 + error_type None is what _apply_backfill_lifecycle reads as "complete".
+    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
+
+
+@pytest.mark.asyncio
+async def test_backfill_budget_cut_returns_partial_with_progress(
+    source_id, sync_engine, tmp_path, monkeypatch,
+):
+    # Deterministic clock: start=0, post1 check=10 (ok), post2 check=200 (>budget).
+    import backend.app.services.patreon_ingester as mod
+    ticks = iter([0.0, 10.0, 200.0, 250.0])
+    last = [0.0]
+
+    def fake_monotonic():
+        try:
+            last[0] = next(ticks)
+        except StopIteration:
+            pass
+        return last[0]
+
+    monkeypatch.setattr(mod.time, "monotonic", fake_monotonic)
+
+    pages = [
+        ("CUR1", [("p1", [_media("p1", 1)])]),
+        ("CUR2", [("p2", [_media("p2", 1)])]),
+    ]
+    client = _FakeClient(pages)
+    downloader = _FakeDownloader(tmp_path)
+    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",
+        resume_cursor=None, time_budget_seconds=100.0,
+    )
+    assert result.error_type == ErrorType.PARTIAL
+    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
+
+
+# --- recovery -------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_recovery_bypasses_seen_ledger(source_id, sync_engine, tmp_path):
+    m1 = _media("p1", 1)
+    factory = sessionmaker(sync_engine, expire_on_commit=False)
+    with factory() as s:
+        s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m1), post_id="p1"))
+        s.commit()
+
+    client = _FakeClient([(None, [("p1", [m1])])])
+    downloader = _FakeDownloader(tmp_path)
+    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="recovery",
+    )
+    # Ledger says seen, but recovery bypasses tier-1 → re-downloaded for a fresh
+    # pHash evaluation under the current threshold.
+    assert result.files_downloaded == 1
+    assert downloader.download_calls == 1
+
+
+@pytest.mark.asyncio
+async def test_recovery_tier2_disk_still_skips(source_id, sync_engine, tmp_path):
+    m1 = _media("p1", 1)
+    client = _FakeClient([(None, [("p1", [m1])])])
+    # File still on disk (a kept image) → tier-2 spares it even under recovery.
+    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="recovery",
+    )
+    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
+
+
+# --- ledger + drift -------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_mark_seen_is_idempotent(source_id, sync_engine, tmp_path):
+    m1 = _media("p1", 1)
+    client = _FakeClient([(None, [("p1", [m1])])])
+    downloader = _FakeDownloader(tmp_path)
+    ing = _ingester(sync_engine, tmp_path, client, downloader)
+
+    # Same media downloaded twice across two runs — the second marks seen again.
+    ing._mark_seen(source_id, [(_ledger_key(m1), "p1")])
+    ing._mark_seen(source_id, [(_ledger_key(m1), "p1")])
+    assert _count_ledger(sync_engine, source_id) == 1
+
+
+@pytest.mark.asyncio
+async def test_drift_fails_loud(source_id, sync_engine, tmp_path):
+    client = _FakeClient([], raise_exc=PatreonDriftError("missing top-level 'data' key"))
+    downloader = _FakeDownloader(tmp_path)
+    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.success is False
+    assert "Patreon API changed" in (result.error_message or "")
+
+
+def test_ledger_key_video_has_no_filehash():
+    video = MediaItem(
+        url="https://stream.mux.com/abc.m3u8", filename="clip.mp4",
+        kind="postfile", filehash=None, post_id="p9",
+    )
+    assert _ledger_key(video) == "p9:clip.mp4"