feat(download): tick/backfill modes + partial-success classifier (plan #544)

Routine subscription polls walked the entire post history every tick
even when nothing had changed, because gallery-dl's default `skip: True`
continues iterating archived posts. A creator with ~550 archived posts
(Knuxy patreon) saturates the 870s wall-clock cap before completing,
even with zero downloads needed. Plus, a tier-limited run that
downloaded hundreds of files but ran out the clock should be a
warning, not an error.

Two coupled changes, both operator-flagged 2026-06-01:

* **Tick mode (default, cron polls).** New `TICK_SKIP_VALUE = "exit:20"`
  asks gallery-dl to exit after 20 contiguous archived items. Fresh
  subscriptions + new-content cases still walk normally; established
  subscription with zero new content exits in ~30s of HEAD requests
  instead of pegging the timeout. 20 (not 5) gives headroom against
  paywall warnings interleaving with archived items.
* **Backfill mode (explicit, operator-triggered).** Sticky for N runs
  via new `Source.backfill_runs_remaining` (alembic 0031). While > 0,
  downloads use `skip: True` + 1800s timeout. Auto-decrements per run
  with early-reset to 0 when a clean run finds zero files (queue
  drained). N defaults to 3 — multiple runs give the system enough
  budget to finish a deep walk across timeout boundaries. New
  `POST /api/sources/{id}/backfill` arms the source; "Deep scan"
  button on each SourceRow (chip shows remaining count) wires it.

Plus partial-success classifier: non-zero gallery-dl exit + ≥1 file
downloaded + no source-level error fires `ErrorType.PARTIAL`, which
download_service maps to `status=\"ok\"`. The run did real work; the
next tick continues via gallery-dl's archive. No more red events for
"timed out mid-walk after downloading 300 files."

Retires `SourceConfig.skip_existing` — skip value is now derived from
the source state and passed as a separate `skip_value` parameter
through download() / _build_config_for_source(). `GD_DEFAULTS` drops
the now-dead key (was inert data after this refactor).

Tests cover:
* tick + backfill skip-value emission in _build_config_for_source
* PARTIAL classifier branch + TIER_LIMITED-wins-over-PARTIAL ordering
* SourceService.set_backfill_runs validation + persistence
* /api/sources/{id}/backfill 200/400/404 paths
* download_service auto-decrement / auto-reset / tick-mode-no-touch
* PARTIAL → status=ok in the orchestrator (no consecutive_failures bump)

tests/test_download_service.py

@@ -370,3 +370,179 @@ async def test_finalize_skipped_preserves_failures_clears_error(db):
     assert row.consecutive_failures == 4
     assert row.last_error is None
     assert row.last_checked_at is not None
+
+
+# --- Plan #544: backfill lifecycle + PARTIAL → status=ok -------------------
+
+
+@pytest.mark.asyncio
+async def test_backfill_decrements_after_run(
+    db, db_sync, tmp_path, seed_artist_and_source,
+):
+    """When backfill_runs_remaining > 0 going in, a non-clean / non-empty
+    run decrements by 1 — operator gets N runs to complete the deep scan
+    before tick mode resumes."""
+    from backend.app.services.download_service import DownloadService
+    from backend.app.services.importer import Importer
+
+    _artist, source = seed_artist_and_source
+    source.backfill_runs_remaining = 3
+    await db.commit()
+
+    images_root = tmp_path / "images"
+    f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
+    _make_jpg(f1, split="h")
+
+    fake_gdl = _fake_gdl_with_result(_make_fake_dl_result(
+        success=True, written_paths=[str(f1)], files_downloaded=1,
+        stdout=f"{f1}\n",
+    ))
+
+    sync_settings = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    importer = Importer(
+        session=db_sync, images_root=images_root, import_root=images_root,
+        thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
+    )
+    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64"))
+    svc = DownloadService(
+        async_session=db, sync_session=db_sync,
+        gdl=fake_gdl, importer=importer, cred_service=cred_service,
+    )
+    await svc.download_source(source.id)
+
+    remaining = (await db.execute(
+        select(Source.backfill_runs_remaining).where(Source.id == source.id)
+    )).scalar_one()
+    assert remaining == 2
+
+
+@pytest.mark.asyncio
+async def test_backfill_auto_resets_on_clean_zero_files(
+    db, db_sync, tmp_path, seed_artist_and_source,
+):
+    """A clean run (rc=0) that downloaded zero files means the backfill
+    queue drained — reset to 0 immediately instead of wasting the rest of
+    the N-run budget on no-op walks."""
+    from backend.app.services.download_service import DownloadService
+    from backend.app.services.importer import Importer
+
+    _artist, source = seed_artist_and_source
+    source.backfill_runs_remaining = 3
+    await db.commit()
+
+    fake_result = _make_fake_dl_result(
+        success=True, written_paths=[], files_downloaded=0,
+    )
+    fake_gdl = _fake_gdl_with_result(fake_result)
+
+    sync_settings = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    importer = Importer(
+        session=db_sync, images_root=tmp_path / "images",
+        import_root=tmp_path / "images",
+        thumbnailer=Thumbnailer(images_root=tmp_path / "images"),
+        settings=sync_settings,
+    )
+    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64"))
+    svc = DownloadService(
+        async_session=db, sync_session=db_sync,
+        gdl=fake_gdl, importer=importer, cred_service=cred_service,
+    )
+    await svc.download_source(source.id)
+
+    remaining = (await db.execute(
+        select(Source.backfill_runs_remaining).where(Source.id == source.id)
+    )).scalar_one()
+    assert remaining == 0
+
+
+@pytest.mark.asyncio
+async def test_tick_mode_does_not_touch_backfill_counter(
+    db, db_sync, tmp_path, seed_artist_and_source,
+):
+    """When backfill_runs_remaining is already 0, downloads don't go
+    negative or otherwise mutate the counter."""
+    from backend.app.services.download_service import DownloadService
+    from backend.app.services.importer import Importer
+
+    _artist, source = seed_artist_and_source
+    assert source.backfill_runs_remaining == 0
+
+    fake_gdl = _fake_gdl_with_result(_make_fake_dl_result(
+        success=True, written_paths=[], files_downloaded=0,
+    ))
+
+    sync_settings = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    importer = Importer(
+        session=db_sync, images_root=tmp_path / "images",
+        import_root=tmp_path / "images",
+        thumbnailer=Thumbnailer(images_root=tmp_path / "images"),
+        settings=sync_settings,
+    )
+    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64"))
+    svc = DownloadService(
+        async_session=db, sync_session=db_sync,
+        gdl=fake_gdl, importer=importer, cred_service=cred_service,
+    )
+    await svc.download_source(source.id)
+
+    remaining = (await db.execute(
+        select(Source.backfill_runs_remaining).where(Source.id == source.id)
+    )).scalar_one()
+    assert remaining == 0
+
+
+@pytest.mark.asyncio
+async def test_partial_error_type_maps_to_ok_status(
+    db, db_sync, tmp_path, seed_artist_and_source,
+):
+    """A non-zero gallery-dl exit accompanied by PARTIAL error_type
+    (real files were downloaded before the run was cut short) is treated
+    as status=ok by the orchestrator — next tick continues."""
+    from backend.app.services.download_service import DownloadService
+    from backend.app.services.gallery_dl import ErrorType
+    from backend.app.services.importer import Importer
+
+    _artist, source = seed_artist_and_source
+
+    images_root = tmp_path / "images"
+    f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
+    _make_jpg(f1, split="h")
+
+    fake_result = _make_fake_dl_result(
+        success=False, written_paths=[str(f1)], files_downloaded=1,
+        error_type=ErrorType.PARTIAL,
+        error_message="Downloaded 1 file; run did not complete in budget",
+        stdout=f"{f1}\n",
+    )
+    fake_gdl = _fake_gdl_with_result(fake_result)
+
+    sync_settings = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    importer = Importer(
+        session=db_sync, images_root=images_root, import_root=images_root,
+        thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
+    )
+    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64"))
+    svc = DownloadService(
+        async_session=db, sync_session=db_sync,
+        gdl=fake_gdl, importer=importer, cred_service=cred_service,
+    )
+    event_id = await svc.download_source(source.id)
+
+    ev = (await db.execute(
+        select(DownloadEvent).where(DownloadEvent.id == event_id)
+    )).scalar_one()
+    assert ev.status == "ok"
+    assert ev.error is None
+    # And the source's failure counter isn't bumped — PARTIAL isn't a failure.
+    src_after = (await db.execute(
+        select(Source.consecutive_failures).where(Source.id == source.id)
+    )).scalar_one()
+    assert src_after == 0