Merge pull request 'Smarter backfill: time-boxed chunks, run-until-done (plan #693)' (#71) from dev into main

Plan #693 (frontend). Backend landed in 96fffaf.

- sources store: setBackfill(runs) → startBackfill/stopBackfill ({action}).
- SubscriptionsTab: the deep-scan window.prompt for N runs becomes a
  Start/Stop toggle keyed on source.backfill_state.
- SourceRow + SourceCard: the 'backfill (N×)' chip becomes a state badge —
  Backfilling (chunk N) / Backfilled / Stalled — and the scan button
  toggles between Backfill (mdi-magnify-scan) and Stop (mdi-stop).

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

backend/app/api/sources.py

@@ -122,26 +122,25 @@ async def delete_source(source_id: int):
 
 @sources_bp.route("/<int:source_id>/backfill", methods=["POST"])
 async def set_backfill(source_id: int):
-    """Plan #544: arm a source for backfill mode for the next N download
-    runs. Body: `{"runs": int}` (1..10, default 3). Returns the updated
-    source dict. While backfill_runs_remaining > 0, downloads use
-    gallery-dl's full-walk config (skip: True + 30-min timeout) instead
-    of the catch-up default (skip: "exit:20" + 14.5-min timeout)."""
+    """Plan #693: start or stop a run-until-done backfill. Body:
+    `{"action": "start" | "stop"}` (default "start"). 'start' walks the full
+    post history in time-boxed chunks until it reaches the bottom (then the
+    source shows 'complete'); 'stop' cancels back to tick mode. Returns the
+    updated source dict (incl. backfill_state / backfill_chunks)."""
     payload = await request.get_json(silent=True) or {}
-    runs = payload.get("runs", 3)
-    try:
-        runs = int(runs)
-    except (TypeError, ValueError):
-        return _bad("invalid_runs", detail="runs must be an integer")
+    action = payload.get("action", "start")
+    if action not in ("start", "stop"):
+        return _bad("invalid_action", detail="action must be 'start' or 'stop'")
     async with get_session() as session:
         try:
-            record = await SourceService(session).set_backfill_runs(
-                source_id, runs,
+            svc = SourceService(session)
+            record = (
+                await svc.start_backfill(source_id)
+                if action == "start"
+                else await svc.stop_backfill(source_id)
             )
         except LookupError:
             return _bad("not_found", status=404)
-        except ValueError as exc:
-            return _bad("invalid_runs", detail=str(exc))
     return jsonify(record.to_dict())
 
 

backend/app/services/download_service.py

@@ -26,12 +26,13 @@ from sqlalchemy.orm import joinedload
 from ..models import Artist, DownloadEvent, Source
 from .credential_service import CredentialService
 from .gallery_dl import (
+    BACKFILL_CHUNK_SECONDS,
     BACKFILL_SKIP_VALUE,
-    BACKFILL_TIMEOUT_SECONDS,
     TICK_SKIP_VALUE,
     ErrorType,
     GalleryDLService,
     SourceConfig,
+    parse_last_cursor,
 )
 from .importer import Importer
 from .patreon_resolver import resolve_campaign_id
@@ -107,15 +108,22 @@ class DownloadService:
         self.sync_session.close()
 
         source_config = SourceConfig.from_dict(ctx["config_overrides"] or {})
-        # alembic 0031 / plan #544: derive skip_value + timeout from the
-        # source's backfill_runs_remaining counter. When > 0, walk the full
-        # post history (skip: True + 1170s); when 0, exit gallery-dl after
-        # 20 contiguous archived items (skip: "exit:20" + the default
-        # 870s). Operator sets backfill via POST /api/sources/{id}/backfill.
-        backfill_remaining = ctx.get("backfill_runs_remaining", 0) or 0
-        if backfill_remaining > 0:
+        # Backfill mode (plan #693): the source's `_backfill_state == "running"`
+        # selects a time-boxed deep-walk chunk — skip: True (walk full history)
+        # + the BACKFILL_CHUNK_SECONDS budget, resuming from the cursor
+        # checkpoint (plan #689). State is the single source of truth; it stays
+        # "running" across ticks until the walk reaches the bottom (phase 3
+        # flips it to "complete"). Otherwise tick mode: exit gallery-dl after
+        # 20 contiguous archived items (skip: "exit:20" + the default 870s).
+        # Operator drives this via POST /api/sources/{id}/backfill {action}.
+        overrides = ctx["config_overrides"] or {}
+        in_backfill = overrides.get("_backfill_state") == "running"
+        if in_backfill:
             skip_value: bool | str = BACKFILL_SKIP_VALUE
-            source_config.timeout = BACKFILL_TIMEOUT_SECONDS
+            source_config.timeout = BACKFILL_CHUNK_SECONDS
+            pending_cursor = overrides.get("_backfill_cursor")
+            if ctx["platform"] == "patreon" and pending_cursor:
+                source_config.resume_cursor = pending_cursor
         else:
             skip_value = TICK_SKIP_VALUE
 
@@ -160,6 +168,24 @@ class DownloadService:
                         skip_value=skip_value,
                     )
 
+        # A backfill chunk that hit its time-box but made forward progress is
+        # NORMAL, not a failure — reclassify TIMEOUT → PARTIAL so it reads as
+        # "ok/progress" (PARTIAL maps to status "ok"), not a red error, and the
+        # next chunk just resumes from the new cursor (plan #693). A chunk that
+        # timed out with NO progress stays TIMEOUT and feeds phase 3's
+        # stall-guard. Leave RATE_LIMITED alone so the platform-cooldown fires.
+        if in_backfill and dl_result.error_type == ErrorType.TIMEOUT:
+            new_cursor = parse_last_cursor(dl_result.stdout, dl_result.stderr)
+            advanced = bool(
+                (new_cursor and new_cursor != overrides.get("_backfill_cursor"))
+                or dl_result.files_downloaded > 0
+            )
+            if advanced:
+                dl_result.error_type = ErrorType.PARTIAL
+                dl_result.error_message = (
+                    f"Backfill chunk: {dl_result.files_downloaded} file(s) — continuing"
+                )
+
         return await self._phase3_persist(
             ctx["event_id"], ctx, dl_result, resolved_campaign_id,
         )
@@ -395,36 +421,88 @@ class DownloadService:
             source_id=ctx["source_id"], status=status, error_message=ev.error,
             error_type=dl_result.error_type.value if dl_result.error_type else None,
         )
-        # Plan #544: backfill lifecycle — auto-complete when a clean
-        # backfill run drained the queue (gallery-dl exited 0 + zero files
-        # downloaded means there was nothing to fetch); otherwise decrement
-        # the counter. Next tick falls back to tick mode once it hits 0.
-        #
-        # Audit 2026-06-02 gating: VALIDATION_FAILED also exits the
-        # subprocess with return_code=0 and files_downloaded=0 (every
-        # file was quarantined), which used to match the auto-complete
-        # predicate exactly — zeroing the operator's armed budget on
-        # the FIRST quarantine run instead of decrementing. Require
-        # dl_result.success + no error_type so only genuinely-empty
-        # successful runs drain the counter.
-        backfill_remaining = ctx.get("backfill_runs_remaining", 0) or 0
-        if backfill_remaining > 0:
-            src = (await self.async_session.execute(
-                select(Source).where(Source.id == ctx["source_id"])
-            )).scalar_one()
-            queue_drained = (
-                dl_result.success
-                and dl_result.error_type is None
-                and dl_result.return_code == 0
-                and dl_result.files_downloaded == 0
-            )
-            if queue_drained:
-                src.backfill_runs_remaining = 0
-            else:
-                src.backfill_runs_remaining = max(0, backfill_remaining - 1)
+        await self._apply_backfill_lifecycle(ctx, dl_result)
         await self.async_session.commit()
         return event_id
 
+    async def _apply_backfill_lifecycle(self, ctx: dict, dl_result) -> None:
+        """Backfill state machine (plan #693, building on the cursor of #689).
+
+        A backfill runs in time-boxed chunks while
+        `config_overrides["_backfill_state"] == "running"`. Each chunk:
+          - COMPLETES the walk → clean rc=0 (gallery-dl with skip:True exits 0
+            only after exhausting the newest→oldest walk; a chunk cut short by
+            its time-box returns success=False / rc<0 via TimeoutExpired). On
+            completion: state="complete", clear the cursor, return to tick mode.
+          - made PROGRESS (cursor advanced and/or files written) → stay
+            "running", checkpoint the new cursor, bump the chunk counter, and
+            spend one of the safety-cap chunks (backfill_runs_remaining). If the
+            cap is exhausted without finishing → state="stalled".
+          - made NO progress → increment the stall counter; two strikes →
+            state="stalled", clear the cursor (a wedged walk can't loop).
+
+        State is the single source of truth; the cursor lives only inside a
+        running backfill. config_overrides is reassigned (not mutated in place)
+        for SQLAlchemy JSON change detection.
+        """
+        overrides = ctx["config_overrides"] or {}
+        if overrides.get("_backfill_state") != "running":
+            return  # not backfilling — tick mode, nothing to do
+
+        old_cursor = overrides.get("_backfill_cursor")
+        cap_remaining = ctx.get("backfill_runs_remaining", 0) or 0
+
+        src = (await self.async_session.execute(
+            select(Source).where(Source.id == ctx["source_id"])
+        )).scalar_one()
+        new_overrides = dict(src.config_overrides or {})
+        chunks = int(new_overrides.get("_backfill_chunks", 0)) + 1
+        new_overrides["_backfill_chunks"] = chunks
+
+        completed = (
+            dl_result.success
+            and dl_result.error_type is None
+            and dl_result.return_code == 0
+        )
+        if completed:
+            new_overrides["_backfill_state"] = "complete"
+            new_overrides.pop("_backfill_cursor", None)
+            new_overrides.pop("_backfill_cursor_stalls", None)
+            src.config_overrides = new_overrides
+            src.backfill_runs_remaining = 0
+            return
+
+        # Did not finish. Patreon checkpoints + resumes via cursor; other
+        # platforms have no resumable cursor (every chunk re-walks from the
+        # top), so they advance only by the download archive growing.
+        new_cursor = (
+            parse_last_cursor(dl_result.stdout, dl_result.stderr)
+            if ctx["platform"] == "patreon" else None
+        )
+        advanced = bool(
+            (new_cursor and new_cursor != old_cursor)
+            or dl_result.files_downloaded > 0
+        )
+        if advanced:
+            if new_cursor:
+                new_overrides["_backfill_cursor"] = new_cursor
+            new_overrides.pop("_backfill_cursor_stalls", None)
+            cap_remaining = max(0, cap_remaining - 1)
+            src.backfill_runs_remaining = cap_remaining
+            if cap_remaining == 0:
+                # Safety cap hit before reaching the bottom — pause, don't loop.
+                new_overrides["_backfill_state"] = "stalled"
+        else:
+            stalls = int(new_overrides.get("_backfill_cursor_stalls", 0)) + 1
+            if stalls >= 2:
+                new_overrides["_backfill_state"] = "stalled"
+                new_overrides.pop("_backfill_cursor", None)
+                new_overrides.pop("_backfill_cursor_stalls", None)
+            else:
+                new_overrides["_backfill_cursor_stalls"] = stalls
+
+        src.config_overrides = new_overrides
+
     async def _update_source_health(
         self, *, source_id: int, status: str, error_message: str | None,
         error_type: str | None = None,

backend/app/services/extension_service.py

@@ -16,7 +16,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import Artist, Source
 from ..utils.slug import slugify
-from .source_service import NEW_SOURCE_BACKFILL_RUNS
+from .source_service import BACKFILL_MAX_CHUNKS
 
 
 class UnknownPlatformError(Exception):
@@ -205,16 +205,17 @@ class ExtensionService:
             return existing, False
         sp = await self.session.begin_nested()
         try:
-            # New subscription sources arm a few backfill runs so the
-            # first ticks walk the full history (otherwise gallery-dl's
-            # exit:20 short-circuits before the archive is built).
-            # Mirrors SourceService.create — without it, Firefox quick-
-            # add on a creator with >20 unsynced posts would surface
-            # as "check failed" with no diagnosis. Audit 2026-06-02.
+            # New subscription sources arm run-until-done backfill (plan #693)
+            # so the first ticks walk the full history (otherwise gallery-dl's
+            # exit:20 short-circuits before the archive is built). Mirrors
+            # SourceService.create — without it, Firefox quick-add on a creator
+            # with >20 unsynced posts would surface as "check failed" with no
+            # diagnosis. Audit 2026-06-02.
             src = Source(
                 artist_id=artist_id, platform=platform,
                 url=url, enabled=True,
-                backfill_runs_remaining=NEW_SOURCE_BACKFILL_RUNS,
+                config_overrides={"_backfill_state": "running"},
+                backfill_runs_remaining=BACKFILL_MAX_CHUNKS,
             )
             self.session.add(src)
             await self.session.flush()

backend/app/services/gallery_dl.py

@@ -13,6 +13,7 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
+import re
 import subprocess
 import sys
 import tempfile
@@ -55,25 +56,24 @@ class ErrorType(StrEnum):
 # 20 contiguous HEADs is still negligible.
 TICK_SKIP_VALUE = "exit:20"
 
-# Backfill mode (operator-triggered deep scan): walk the full history.
-# Source.backfill_runs_remaining > 0 selects this mode; the longer
-# timeout below absorbs creators with thousands of posts.
+# Backfill mode (operator-triggered deep scan): walk the full history,
+# one TIME-BOXED CHUNK per run (plan #693). config_overrides["_backfill_state"]
+# == "running" selects this mode; the cursor checkpoint (plan #689) lets each
+# chunk resume where the last stopped, so the walk advances across chunks
+# until gallery-dl exits cleanly (= reached the bottom).
 #
-# Sits below download_source's Celery soft_time_limit
-# (DOWNLOAD_SOFT_TIME_LIMIT=1350, tasks/download.py) with ~180s of
-# headroom for phase-3 persist. subprocess.run MUST raise TimeoutExpired
-# before Celery raises SoftTimeLimitExceeded — that exception path
-# captures partial stdout/stderr and finalizes the event; the soft-limit
-# path (until the 2026-06-03 fix) did not. Audit history: 1800 guaranteed
-# SIGKILL against the old hard limit (Knuxy #38275); 1170 was then sized
-# "30s shy of the hard limit (1200)" but still EXCEEDED the soft limit
-# (900), so SoftTimeLimitExceeded preempted TimeoutExpired and every
-# backfill stranded empty (Anduo #39912). Raising the Celery soft/hard
-# limits to 1350/1500 (tasks/download.py) is what made 1170 safe.
-# backfill_runs_remaining=3 still gives ~58 minutes of cumulative walk
-# across three runs for prolific creators.
+# The chunk budget is deliberately FAR below download_source's Celery
+# soft_time_limit (DOWNLOAD_SOFT_TIME_LIMIT=1350, tasks/download.py), not
+# "just under" it. Hitting this budget is the NORMAL chunk boundary, not a
+# failure: subprocess.run raises TimeoutExpired (which captures partial
+# stdout/stderr + the last emitted cursor), and download_service reclassifies
+# a chunk that made progress as PARTIAL (status "ok"), not an error. The huge
+# headroom means a stuck file or a slow chunk can never let Celery's
+# SoftTimeLimitExceeded preempt TimeoutExpired (the failure mode behind Knuxy
+# #38275 / Anduo #39912/#40411). Earlier we ran one ~1170s run-to-the-wall
+# per arming; that died as a timeout error every time on large catalogs.
 BACKFILL_SKIP_VALUE = True
-BACKFILL_TIMEOUT_SECONDS = 1170
+BACKFILL_CHUNK_SECONDS = 600
 
 
 # Sits well below download_source's Celery soft_time_limit
@@ -98,6 +98,15 @@ class SourceConfig:
     Source.backfill_runs_remaining column at the download_service layer
     and is passed to _build_config_for_source as `skip_value`. Same for
     the per-run subprocess timeout.
+
+    `resume_cursor` is RUNTIME state, not persisted operator config: the
+    cursor-paged backfill checkpoint (see parse_last_cursor + the
+    download_service backfill lifecycle). When set on a patreon backfill
+    run, gallery-dl resumes its newest→oldest walk from that pagination
+    cursor instead of restarting from the top. It is threaded by
+    download_service from config_overrides["_backfill_cursor"]; SourceConfig
+    deliberately does NOT read it in from_dict (config_overrides also holds
+    operator config, and resume_cursor must only apply in backfill mode).
     """
     content_types: list[str] = field(default_factory=lambda: ["all"])
     sleep: float | None = None
@@ -106,6 +115,7 @@ class SourceConfig:
     filename_pattern: str | None = None
     save_metadata: bool = True
     timeout: int = _DEFAULT_GDL_TIMEOUT_SECONDS
+    resume_cursor: str | None = None
 
     @classmethod
     def from_dict(cls, data: dict) -> SourceConfig:
@@ -154,6 +164,21 @@ def _summarize_validation_failures(failures: list[dict]) -> str:
     return f"{n} files quarantined ({top_count}× {top_reason}, mixed)"
 
 
+# gallery-dl logs its pagination cursor (when extractor.*.cursor is truthy)
+# as `Cursor: <token>` — once per fetched page, at debug verbosity. The LAST
+# such line is the furthest-progressed page, i.e. the resume point. We scan
+# both streams (the debug log lands on stderr, but be defensive) and take the
+# final match; passing it back as extractor.patreon.cursor resumes the walk.
+# Survives a timed-out run too: the TimeoutExpired path returns the partial
+# stdout/stderr, so an interrupted backfill still yields its last cursor.
+_CURSOR_RE = re.compile(r"Cursor:\s*(\S+)")
+
+
+def parse_last_cursor(stdout: str, stderr: str) -> str | None:
+    matches = _CURSOR_RE.findall(f"{stdout or ''}\n{stderr or ''}")
+    return matches[-1] if matches else None
+
+
 class GalleryDLService:
     """Service for executing gallery-dl downloads."""
 
@@ -255,7 +280,12 @@ class GalleryDLService:
                 "skip": True,
                 "sleep": self._rate_limit,
                 "sleep-request": max(0.5, self._rate_limit / 4),
-                "retries": 3,
+                # 2 (not 3) retries — a stuck CDN host shouldn't burn a whole
+                # backfill chunk on one file. Anduo #40838 spent ~600s on a
+                # single image (4 extractor HEAD retries × 30s + 4 downloader
+                # GET retries × 120s); halving retries + the downloader
+                # timeout below caps a wedged file at ~1-2 min instead.
+                "retries": 2,
                 "timeout": 30.0,
                 "verify": True,
                 "postprocessors": [
@@ -270,8 +300,12 @@ class GalleryDLService:
             "downloader": {
                 "part": True,
                 "part-directory": str(self._config_dir / "temp"),
-                "retries": 3,
-                "timeout": 120.0,
+                # See the extractor retries note above (Anduo #40838): 2
+                # retries + a 60s read-timeout (was 120) so a stalled
+                # connection fails fast. 60s is a per-read timeout, not a
+                # transfer cap — large GIFs that keep streaming are unaffected.
+                "retries": 2,
+                "timeout": 60.0,
                 # Forward Patreon as Referer/Origin to yt-dlp when it
                 # fetches video manifests. Operator-flagged 2026-06-01
                 # (DaferQ patreon): video posts hosted on Mux carry a JWT
@@ -349,6 +383,15 @@ class GalleryDLService:
                 ]
             else:
                 platform_section["files"] = source_config.content_types
+            # Cursor-paged backfill: resume the newest→oldest walk from the
+            # checkpoint saved by the previous run instead of restarting from
+            # the top. Without this, a large catalog (Anduo's weekly
+            # image-dense Reports) never reaches its frontier inside the
+            # subprocess budget — the HEAD walk alone times out, 0 files
+            # written, no progress (event #40411). The PLATFORM_DEFAULTS leave
+            # `cursor: True` (log-only) for the first/fresh run.
+            if source_config.resume_cursor:
+                platform_section["cursor"] = source_config.resume_cursor
         elif platform == "hentaifoundry":
             if "pictures" in source_config.content_types or "all" in source_config.content_types:
                 platform_section["include"] = "all"

backend/app/services/source_service.py

@@ -64,6 +64,9 @@ class SourceRecord:
     consecutive_failures: int
     next_check_at: str | None
     backfill_runs_remaining: int
+    # plan #693: derived from config_overrides for the UI badge.
+    backfill_state: str | None  # "running" | "complete" | "stalled" | None (idle)
+    backfill_chunks: int
 
     def to_dict(self) -> dict:
         return {
@@ -82,6 +85,8 @@ class SourceRecord:
             "consecutive_failures": self.consecutive_failures,
             "next_check_at": self.next_check_at,
             "backfill_runs_remaining": self.backfill_runs_remaining,
+            "backfill_state": self.backfill_state,
+            "backfill_chunks": self.backfill_chunks,
         }
 
 
@@ -89,10 +94,13 @@ class SourceRecord:
 
 _EDITABLE = {"enabled", "url", "config_overrides", "check_interval_override", "platform"}
 
-# Plan #544 follow-up: newly created enabled sources pre-arm backfill so
-# their first N polls walk gallery-dl's full post history with the longer
-# timeout (matches the manual "Deep scan" button's default).
-NEW_SOURCE_BACKFILL_RUNS = 3
+# Plan #693: backfill safety cap. "Start backfill" (and a newly created
+# enabled source) arms a run-until-done walk; this caps how many time-boxed
+# chunks it may spend before pausing as "stalled", so a pathological walk that
+# never reaches the bottom can't run forever. Generous on purpose — at
+# BACKFILL_CHUNK_SECONDS (600s) per chunk this is ~33h of cumulative walk, far
+# beyond any real catalog; the cursor stall-guard is the real terminator.
+BACKFILL_MAX_CHUNKS = 200
 
 
 class SourceService:
@@ -135,6 +143,7 @@ class SourceService:
         self, source: Source, artist: Artist, settings: ImportSettings,
     ) -> SourceRecord:
         nxt = compute_next_check_at(source, artist, settings)
+        co = source.config_overrides or {}
         return SourceRecord(
             id=source.id,
             artist_id=source.artist_id,
@@ -151,6 +160,8 @@ class SourceService:
             consecutive_failures=source.consecutive_failures or 0,
             next_check_at=nxt.isoformat() if nxt else None,
             backfill_runs_remaining=source.backfill_runs_remaining or 0,
+            backfill_state=co.get("_backfill_state"),
+            backfill_chunks=int(co.get("_backfill_chunks", 0)),
         )
 
     async def _row_to_record(self, source: Source) -> SourceRecord:
@@ -212,16 +223,17 @@ class SourceService:
             select(func.count(Source.id)).where(Source.artist_id == artist_id)
         )).scalar_one()
 
-        # Plan #544 follow-up: a freshly added subscription has no archive
-        # yet, so the first few polls would walk the full post history in
-        # tick mode and trip exit:20 after ~20 contiguous archive hits —
-        # except there are none yet, so tick mode would walk forever and
-        # blow the wall-clock cap. Pre-arm backfill so the initial syncs
-        # use the longer timeout + skip:True walk. Tick mode resumes once
-        # the budget is spent or the queue drains.
-        # Disabled sources (incl. sidecar synthetics, url='sidecar:...')
-        # are never polled, so leave their counter at 0.
-        backfill_runs = NEW_SOURCE_BACKFILL_RUNS if enabled else 0
+        # Plan #693: a freshly added subscription has no archive yet, so it
+        # should walk its full post history once. Arm run-until-done backfill
+        # (state="running" + the chunk cap); the time-boxed chunks march to the
+        # bottom across ticks, then flip to "complete" and tick mode takes over.
+        # Disabled sources (incl. sidecar synthetics, url='sidecar:...') are
+        # never polled, so leave them idle.
+        if enabled:
+            config_overrides = {**(config_overrides or {}), "_backfill_state": "running"}
+            backfill_runs = BACKFILL_MAX_CHUNKS
+        else:
+            backfill_runs = 0
         source = Source(
             artist_id=artist_id, platform=platform, url=url,
             enabled=enabled, config_overrides=config_overrides,
@@ -286,22 +298,41 @@ class SourceService:
         await self.session.commit()
         return await self._row_to_record(source)
 
-    async def set_backfill_runs(
-        self, source_id: int, runs: int,
-    ) -> SourceRecord:
-        """Plan #544: arm a source for backfill mode. The next `runs`
-        download runs will use gallery-dl's full-walk config (skip: True
-        + 30-min timeout) instead of the catch-up default. Runs must be
-        1..10 — bigger is rejected to keep the operator from accidentally
-        setting a runaway budget."""
-        if not isinstance(runs, int) or runs < 1 or runs > 10:
-            raise ValueError("runs must be an integer in [1, 10]")
+    async def start_backfill(self, source_id: int) -> SourceRecord:
+        """Plan #693: arm a run-until-done backfill. Sets state="running" and
+        the chunk cap; download runs then walk the full post history in
+        time-boxed chunks (skip:True + BACKFILL_CHUNK_SECONDS), resuming from
+        the cursor each chunk, until gallery-dl reaches the bottom (→ state
+        "complete") or the cap/stall-guard pauses it (→ "stalled"). Clears any
+        prior cursor/chunk/stall state so a re-start walks fresh from the top."""
         source = (await self.session.execute(
             select(Source).where(Source.id == source_id)
         )).scalar_one_or_none()
         if source is None:
             raise LookupError(f"source id={source_id} not found")
-        source.backfill_runs_remaining = runs
+        co = dict(source.config_overrides or {})
+        co["_backfill_state"] = "running"
+        for k in ("_backfill_cursor", "_backfill_cursor_stalls", "_backfill_chunks"):
+            co.pop(k, None)
+        source.config_overrides = co
+        source.backfill_runs_remaining = BACKFILL_MAX_CHUNKS
+        await self.session.commit()
+        return await self._row_to_record(source)
+
+    async def stop_backfill(self, source_id: int) -> SourceRecord:
+        """Plan #693: cancel an in-progress backfill — back to idle/tick mode.
+        Clears the running state + cursor/chunk/stall bookkeeping."""
+        source = (await self.session.execute(
+            select(Source).where(Source.id == source_id)
+        )).scalar_one_or_none()
+        if source is None:
+            raise LookupError(f"source id={source_id} not found")
+        co = dict(source.config_overrides or {})
+        for k in ("_backfill_state", "_backfill_cursor", "_backfill_cursor_stalls",
+                  "_backfill_chunks"):
+            co.pop(k, None)
+        source.config_overrides = co
+        source.backfill_runs_remaining = 0
         await self.session.commit()
         return await self._row_to_record(source)
 

backend/app/tasks/download.py

@@ -31,7 +31,7 @@ _KEY_PATH = IMAGES_ROOT / "secrets" / "credential_key.b64"
 # SoftTimeLimitExceeded in-process, whereas the HARD limit SIGKILLs the
 # worker (no chance to finalize). Both gallery-dl subprocess budgets
 # (gallery_dl.py: _DEFAULT_GDL_TIMEOUT_SECONDS=870 tick,
-# BACKFILL_TIMEOUT_SECONDS=1170 backfill) MUST sit below the soft limit
+# BACKFILL_CHUNK_SECONDS=600 per backfill chunk, plan #693) MUST sit below the soft limit
 # so subprocess.run raises its own TimeoutExpired first — that path
 # captures partial stdout/stderr and finalizes the DownloadEvent. soft is
 # max-subprocess (1170) + ~180s phase-3 persist headroom; hard is soft +

frontend/src/components/modal/ImageViewer.vue

@@ -203,14 +203,37 @@ function nextFrame() {
 }
 
 @media (max-width: 900px) {
-  .fc-viewer__body { flex-direction: column; }
-  /* Side panel drops below the image — the next arrow uses the full width. */
-  .fc-viewer__nav--next { right: 16px; }
+  /* Stacked layout (operator-flagged 2026-06-05): pin the image pane and
+     its controls at the top while the metadata panel scrolls beneath it.
+     Previously the image + a 40vh panel shared one fixed viewport and the
+     controls could be pushed out of view; now the body scrolls and the
+     media is sticky, so the image + prev/next/close (and the integrity
+     badge) stay visible no matter how far the panel is scrolled. */
+  .fc-viewer__body {
+    flex-direction: column;
+    overflow-y: auto;
+  }
+  .fc-viewer__media {
+    flex: none;
+    height: 55vh;
+    position: sticky;
+    top: 0;
+    z-index: 1;
+    /* Opaque obsidian so the scrolling panel never bleeds through the
+       haze behind the pinned image. */
+    background: rgb(20, 23, 26);
+  }
   .fc-viewer__side {
     width: 100%;
-    max-height: 40vh;
+    max-height: none;
     border-left: none;
     border-top: 1px solid rgb(var(--v-theme-surface-light));
   }
+  /* Re-center the prev/next arrows over the 55vh image band (their base
+     top:50% would land on the scrolling panel); next uses the full width
+     now that the panel is below, not beside. Close + integrity badge keep
+     their top:16px/72px — already within the pinned band. */
+  .fc-viewer__nav { top: 27.5vh; }
+  .fc-viewer__nav--next { right: 16px; }
 }
 </style>

frontend/src/components/modal/TagAutocomplete.vue

@@ -65,8 +65,17 @@ const store = useTagStore()
 // Vuetify's v-text-field exposes .focus() on the component instance;
 // nextTick waits for the modal's mount to finish so the inner <input>
 // element exists.
+//
+// EXCEPTION — not on mobile (operator-flagged 2026-06-05): focusing the
+// field pops the soft keyboard, which shrinks the visual viewport and
+// shoves the pinned image + its nav/close controls out of view. 900px is
+// the modal's stacked-layout breakpoint (ImageViewer.vue). matchMedia is
+// safe on plain HTTP (not a secure-context API).
 const inputRef = ref(null)
-onMounted(() => { nextTick(() => inputRef.value?.focus?.()) })
+onMounted(() => {
+  if (window.matchMedia?.('(max-width: 900px)')?.matches) return
+  nextTick(() => inputRef.value?.focus?.())
+})
 
 // Single text input; no kind dropdown. Client-side mirror of the
 // backend's parse_kind_prefix lives below — kept in sync with

frontend/src/components/subscriptions/SourceCard.vue

@@ -25,9 +25,17 @@
         size="x-small" color="error" variant="tonal" label
       >{{ source.consecutive_failures }} err</v-chip>
       <v-chip
-        v-else-if="(source.backfill_runs_remaining || 0) > 0"
+        v-else-if="source.backfill_state === 'running'"
         size="x-small" color="info" variant="tonal" label
-      >backfill ({{ source.backfill_runs_remaining }}×)</v-chip>
+      >Backfilling{{ source.backfill_chunks ? ` (${source.backfill_chunks})` : '' }}</v-chip>
+      <v-chip
+        v-else-if="source.backfill_state === 'complete'"
+        size="x-small" color="success" variant="tonal" label
+      >Backfilled</v-chip>
+      <v-chip
+        v-else-if="source.backfill_state === 'stalled'"
+        size="x-small" color="warning" variant="tonal" label
+      >Stalled</v-chip>
     </div>
 
     <div class="fc-source-card__actions">
@@ -40,11 +48,13 @@
       </v-btn>
       <v-btn
         size="x-small" variant="text"
-        :disabled="(source.backfill_runs_remaining || 0) > 0"
+        :color="source.backfill_state === 'running' ? 'warning' : undefined"
         @click.stop="$emit('backfill', source)"
       >
-        <v-icon>mdi-magnify-scan</v-icon>
-        <v-tooltip activator="parent" location="top">Deep scan</v-tooltip>
+        <v-icon>{{ source.backfill_state === 'running' ? 'mdi-stop' : 'mdi-magnify-scan' }}</v-icon>
+        <v-tooltip activator="parent" location="top">
+          {{ source.backfill_state === 'running' ? 'Stop backfill' : 'Backfill full history' }}
+        </v-tooltip>
       </v-btn>
       <v-btn size="x-small" variant="text" @click.stop="$emit('edit', source)">
         <v-icon>mdi-pencil</v-icon>

frontend/src/components/subscriptions/SourceRow.vue

@@ -32,11 +32,17 @@
         size="x-small" color="error" variant="tonal" label
       >{{ source.consecutive_failures }}</v-chip>
       <v-chip
-        v-else-if="(source.backfill_runs_remaining || 0) > 0"
+        v-else-if="source.backfill_state === 'running'"
         size="x-small" color="info" variant="tonal" label
-      >
-        backfill ({{ source.backfill_runs_remaining }}×)
-      </v-chip>
+      >Backfilling{{ source.backfill_chunks ? ` (${source.backfill_chunks})` : '' }}</v-chip>
+      <v-chip
+        v-else-if="source.backfill_state === 'complete'"
+        size="x-small" color="success" variant="tonal" label
+      >Backfilled</v-chip>
+      <v-chip
+        v-else-if="source.backfill_state === 'stalled'"
+        size="x-small" color="warning" variant="tonal" label
+      >Stalled</v-chip>
       <span v-else class="fc-source-row__zero">0</span>
     </td>
     <td class="fc-source-row__actions">
@@ -49,13 +55,15 @@
         <v-tooltip activator="parent" location="top">Check now</v-tooltip>
       </v-btn>
       <v-btn
-        icon="mdi-magnify-scan" size="x-small" variant="text"
-        :disabled="(source.backfill_runs_remaining || 0) > 0"
+        size="x-small" variant="text"
+        :color="source.backfill_state === 'running' ? 'warning' : undefined"
         @click.stop="$emit('backfill', source)"
       >
-        <v-icon>mdi-magnify-scan</v-icon>
+        <v-icon>{{ source.backfill_state === 'running' ? 'mdi-stop' : 'mdi-magnify-scan' }}</v-icon>
         <v-tooltip activator="parent" location="top">
-          Deep scan — walk full history for next few runs
+          {{ source.backfill_state === 'running'
+            ? 'Stop backfill'
+            : 'Backfill — walk the full post history until complete' }}
         </v-tooltip>
       </v-btn>
       <v-btn

frontend/src/components/subscriptions/SubscriptionsTab.vue

@@ -532,31 +532,23 @@ async function onCheck(source) {
   }
 }
 
-// Plan #544: arm a source for backfill mode (gallery-dl walks the full
-// post history) for the next N download runs. Default 3 — enough budget
-// to finish a deep creator without re-prompting the operator across
-// timeout boundaries. The chip on the row reflects the remaining count.
+// Plan #693: toggle a run-until-done backfill. Starting walks the full post
+// history in time-boxed chunks across ticks until it reaches the bottom (the
+// row badge tracks progress / completion); stopping cancels back to tick mode.
 async function onBackfill(source) {
-  const raw = globalThis.window?.prompt(
-    `Deep scan "${source.artist_name} (${source.platform})" — walk full history for the next how many download runs? (1–10, default 3)`,
-    '3',
-  )
-  if (raw == null) return
-  const runs = parseInt(raw, 10)
-  if (!Number.isFinite(runs) || runs < 1 || runs > 10) {
-    toast({ text: 'Deep scan: runs must be 1–10', type: 'error' })
-    return
-  }
+  const running = source.backfill_state === 'running'
   try {
-    await store.setBackfill(source.id, runs, source.artist_id)
-    toast({
-      text: `Deep scan armed for ${runs} run${runs === 1 ? '' : 's'}`,
-      type: 'success',
-    })
+    if (running) {
+      await store.stopBackfill(source.id, source.artist_id)
+      toast({ text: `Backfill stopped for ${source.artist_name}`, type: 'success' })
+    } else {
+      await store.startBackfill(source.id, source.artist_id)
+      toast({ text: `Backfill started for ${source.artist_name}`, type: 'success' })
+    }
     await store.loadAll()
   } catch (e) {
     toast({
-      text: `Deep scan failed: ${e?.detail || e?.message || e}`,
+      text: `Backfill ${running ? 'stop' : 'start'} failed: ${e?.detail || e?.message || e}`,
       type: 'error',
     })
   }

frontend/src/stores/sources.js

@@ -85,11 +85,16 @@ export const useSourcesStore = defineStore('sources', () => {
     }
   }
 
-  // Plan #544: arm a source for backfill mode. The next `runs` download
-  // runs (default 3) walk gallery-dl's full post history instead of
-  // exiting early at the first contiguous archived block.
-  async function setBackfill(id, runs = 3, artistIdHint = null) {
-    const body = await api.post(`/api/sources/${id}/backfill`, { body: { runs } })
+  // Plan #693: start/stop a run-until-done backfill. 'start' walks the full
+  // post history in time-boxed chunks until it reaches the bottom (then the
+  // source shows backfill_state 'complete'); 'stop' cancels back to tick mode.
+  async function startBackfill(id, artistIdHint = null) {
+    const body = await api.post(`/api/sources/${id}/backfill`, { body: { action: 'start' } })
+    _invalidate(artistIdHint ?? body.artist_id)
+    return body
+  }
+  async function stopBackfill(id, artistIdHint = null) {
+    const body = await api.post(`/api/sources/${id}/backfill`, { body: { action: 'stop' } })
     _invalidate(artistIdHint ?? body.artist_id)
     return body
   }
@@ -120,7 +125,8 @@ export const useSourcesStore = defineStore('sources', () => {
     loadAll, loadForArtist,
     create, update, remove,
     checkNow,
-    setBackfill,
+    startBackfill,
+    stopBackfill,
     findOrCreateArtist, autocompleteArtist,
     loadScheduleStatus,
     sourcesByArtistGrouped,

tests/test_api_sources.py

@@ -199,7 +199,7 @@ async def test_list_derives_next_check_at_when_last_checked_set(
 
 
 @pytest.mark.asyncio
-async def test_backfill_endpoint_arms_source(client, artist, db):
+async def test_backfill_endpoint_start_and_stop(client, artist, db):
     src = Source(
         artist_id=artist.id, platform="patreon",
         url="https://patreon.com/alice-backfill", enabled=True,
@@ -208,18 +208,21 @@ async def test_backfill_endpoint_arms_source(client, artist, db):
     await db.commit()
     sid = src.id
 
-    resp = await client.post(f"/api/sources/{sid}/backfill", json={"runs": 5})
+    resp = await client.post(f"/api/sources/{sid}/backfill", json={"action": "start"})
     assert resp.status_code == 200
     body = await resp.get_json()
-    assert body["backfill_runs_remaining"] == 5
+    assert body["backfill_state"] == "running"
 
-    # GET reflects the new state.
+    # GET reflects the running state.
     one = await client.get(f"/api/sources/{sid}")
-    assert (await one.get_json())["backfill_runs_remaining"] == 5
+    assert (await one.get_json())["backfill_state"] == "running"
+
+    stopped = await client.post(f"/api/sources/{sid}/backfill", json={"action": "stop"})
+    assert (await stopped.get_json())["backfill_state"] is None
 
 
 @pytest.mark.asyncio
-async def test_backfill_endpoint_defaults_to_three(client, artist, db):
+async def test_backfill_endpoint_defaults_to_start(client, artist, db):
     src = Source(
         artist_id=artist.id, platform="patreon",
         url="https://patreon.com/alice-backfill-default", enabled=True,
@@ -228,26 +231,22 @@ async def test_backfill_endpoint_defaults_to_three(client, artist, db):
     await db.commit()
     resp = await client.post(f"/api/sources/{src.id}/backfill", json={})
     body = await resp.get_json()
-    assert body["backfill_runs_remaining"] == 3
+    assert body["backfill_state"] == "running"
 
 
 @pytest.mark.asyncio
-async def test_backfill_endpoint_rejects_out_of_range(client, artist, db):
+async def test_backfill_endpoint_rejects_bad_action(client, artist, db):
     src = Source(
         artist_id=artist.id, platform="patreon",
         url="https://patreon.com/alice-backfill-bad", enabled=True,
     )
     db.add(src)
     await db.commit()
-    bad = await client.post(f"/api/sources/{src.id}/backfill", json={"runs": 0})
+    bad = await client.post(f"/api/sources/{src.id}/backfill", json={"action": "nope"})
     assert bad.status_code == 400
-    too_big = await client.post(
-        f"/api/sources/{src.id}/backfill", json={"runs": 99}
-    )
-    assert too_big.status_code == 400
 
 
 @pytest.mark.asyncio
 async def test_backfill_endpoint_404_when_source_missing(client):
-    resp = await client.post("/api/sources/999999/backfill", json={"runs": 3})
+    resp = await client.post("/api/sources/999999/backfill", json={"action": "start"})
     assert resp.status_code == 404

tests/test_download_service.py

@@ -372,32 +372,16 @@ async def test_finalize_skipped_preserves_failures_clears_error(db):
     assert row.last_checked_at is not None
 
 
-# --- Plan #544: backfill lifecycle + PARTIAL → status=ok -------------------
+# --- Plan #693: backfill state machine (time-boxed chunks, run-until-done) ---
 
 
-@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."""
+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)."""
     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()
@@ -406,97 +390,221 @@ async def test_backfill_decrements_after_run(
         thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
     )
     cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
+    fake_gdl = _fake_gdl_with_result(result)
     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
+    return svc, fake_gdl
 
 
 @pytest.mark.asyncio
-async def test_backfill_auto_resets_on_clean_zero_files(
+async def test_backfill_chunk_progress_advances_cursor(
     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
-
+    """A running backfill chunk that didn't finish but advanced (new cursor)
+    stays 'running', checkpoints the cursor, bumps the chunk counter, and
+    spends one safety-cap chunk."""
     _artist, source = seed_artist_and_source
-    source.backfill_runs_remaining = 3
+    source.config_overrides = {"_backfill_state": "running"}
+    source.backfill_runs_remaining = 5
     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", bootstrap_ok=True))
-    svc = DownloadService(
-        async_session=db, sync_session=db_sync,
-        gdl=fake_gdl, importer=importer, cred_service=cred_service,
-    )
+    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
+        success=False, written_paths=[], files_downloaded=0,
+        stderr="[patreon][debug] Cursor: 03:PAGE2:xyz\n",
+    ))
     await svc.download_source(source.id)
 
-    remaining = (await db.execute(
-        select(Source.backfill_runs_remaining).where(Source.id == source.id)
+    remaining, co = (await db.execute(
+        select(Source.backfill_runs_remaining, Source.config_overrides)
+        .where(Source.id == source.id)
+    )).one()
+    assert co.get("_backfill_state") == "running"
+    assert co.get("_backfill_cursor") == "03:PAGE2:xyz"
+    assert co.get("_backfill_chunks") == 1
+    assert remaining == 4
+
+
+@pytest.mark.asyncio
+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,
+    )
+
+    _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(
+        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
+
+    co = (await db.execute(
+        select(Source.config_overrides).where(Source.id == source.id)
     )).scalar_one()
+    assert co.get("_backfill_cursor") == "03:RESUME2:next"
+
+
+@pytest.mark.asyncio
+async def test_backfill_clean_exit_marks_complete(
+    db, db_sync, tmp_path, seed_artist_and_source,
+):
+    """A clean rc=0 chunk = gallery-dl reached the bottom → state 'complete',
+    cursor cleared, returns to tick mode."""
+    _artist, source = seed_artist_and_source
+    source.config_overrides = {"_backfill_state": "running", "_backfill_cursor": "03:NEAR:bottom"}
+    source.backfill_runs_remaining = 5
+    await db.commit()
+
+    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
+        success=True, written_paths=[], files_downloaded=0,
+    ))
+    await svc.download_source(source.id)
+
+    remaining, co = (await db.execute(
+        select(Source.backfill_runs_remaining, Source.config_overrides)
+        .where(Source.id == source.id)
+    )).one()
+    assert co.get("_backfill_state") == "complete"
+    assert "_backfill_cursor" not in co
     assert remaining == 0
 
 
 @pytest.mark.asyncio
-async def test_tick_mode_does_not_touch_backfill_counter(
+async def test_backfill_cap_exhaustion_stalls(
     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
-
+    """A progressing chunk that spends the last safety-cap chunk without
+    reaching the bottom pauses as 'stalled' (doesn't loop forever)."""
     _artist, source = seed_artist_and_source
-    assert source.backfill_runs_remaining == 0
+    source.config_overrides = {"_backfill_state": "running"}
+    source.backfill_runs_remaining = 1
+    await db.commit()
 
-    fake_gdl = _fake_gdl_with_result(_make_fake_dl_result(
-        success=True, written_paths=[], files_downloaded=0,
+    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
+        success=False, written_paths=[],
+        stderr="[patreon][debug] Cursor: 03:MORE:left\n",
     ))
-
-    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", bootstrap_ok=True))
-    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()
+    remaining, co = (await db.execute(
+        select(Source.backfill_runs_remaining, Source.config_overrides)
+        .where(Source.id == source.id)
+    )).one()
+    assert co.get("_backfill_state") == "stalled"
     assert remaining == 0
 
 
+@pytest.mark.asyncio
+async def test_backfill_stuck_guard_stalls_after_two_no_advance(
+    db, db_sync, tmp_path, seed_artist_and_source,
+):
+    """Two consecutive chunks that fail to advance the cursor → 'stalled',
+    cursor cleared, so a wedged walk can't re-strand forever."""
+    from unittest.mock import MagicMock
+
+    from backend.app.services.download_service import DownloadService
+
+    _artist, source = seed_artist_and_source
+    source.config_overrides = {"_backfill_state": "running", "_backfill_cursor": "stuck"}
+    source.backfill_runs_remaining = 5
+    await db.commit()
+
+    svc = DownloadService(
+        async_session=db, sync_session=db_sync,
+        gdl=MagicMock(), importer=MagicMock(), cred_service=MagicMock(),
+    )
+    ctx = {
+        "source_id": source.id, "platform": "patreon",
+        "config_overrides": {"_backfill_state": "running", "_backfill_cursor": "stuck"},
+        "backfill_runs_remaining": 5,
+    }
+    stuck = _make_fake_dl_result(success=False, stderr="Cursor: stuck\n")
+
+    await svc._apply_backfill_lifecycle(ctx, stuck)
+    await db.commit()
+    co = (await db.execute(
+        select(Source.config_overrides).where(Source.id == source.id)
+    )).scalar_one()
+    assert co.get("_backfill_state") == "running"
+    assert co.get("_backfill_cursor_stalls") == 1
+
+    ctx["config_overrides"] = dict(co)
+    await svc._apply_backfill_lifecycle(ctx, stuck)
+    await db.commit()
+    co2 = (await db.execute(
+        select(Source.config_overrides).where(Source.id == source.id)
+    )).scalar_one()
+    assert co2.get("_backfill_state") == "stalled"
+    assert "_backfill_cursor" not in co2
+
+
+@pytest.mark.asyncio
+async def test_backfill_timeout_chunk_reclassified_to_ok(
+    db, db_sync, tmp_path, seed_artist_and_source,
+):
+    """A backfill chunk that hit its time-box (TIMEOUT) but advanced is
+    reclassified to PARTIAL → event status 'ok' (progress, not error), and
+    consecutive_failures is not bumped."""
+    from backend.app.services.gallery_dl import ErrorType
+
+    _artist, source = seed_artist_and_source
+    source.config_overrides = {"_backfill_state": "running"}
+    source.backfill_runs_remaining = 5
+    await db.commit()
+
+    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
+        success=False, files_downloaded=3,
+        error_type=ErrorType.TIMEOUT,
+        error_message="Download timed out",
+        stderr="[patreon][debug] Cursor: 03:NEXT:page\n",
+    ))
+    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"
+    failures = (await db.execute(
+        select(Source.consecutive_failures).where(Source.id == source.id)
+    )).scalar_one()
+    assert failures == 0
+
+
+@pytest.mark.asyncio
+async def test_tick_mode_when_not_running_leaves_state_untouched(
+    db, db_sync, tmp_path, seed_artist_and_source,
+):
+    """No _backfill_state → not backfilling; the lifecycle is a no-op and the
+    counter/state aren't mutated."""
+    _artist, source = seed_artist_and_source
+    assert (source.config_overrides or {}).get("_backfill_state") is None
+
+    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
+        success=True, written_paths=[], files_downloaded=0,
+    ))
+    await svc.download_source(source.id)
+
+    co = (await db.execute(
+        select(Source.config_overrides).where(Source.id == source.id)
+    )).scalar_one()
+    assert (co or {}).get("_backfill_state") is None
+
+
 @pytest.mark.asyncio
 async def test_partial_error_type_maps_to_ok_status(
     db, db_sync, tmp_path, seed_artist_and_source,

tests/test_download_source_task.py

@@ -34,7 +34,7 @@ def test_timeout_ladder_keeps_subprocess_budgets_under_soft_limit():
     preempts it. soft must in turn sit below the hard SIGKILL cap."""
     from backend.app.services.gallery_dl import (
         _DEFAULT_GDL_TIMEOUT_SECONDS,
-        BACKFILL_TIMEOUT_SECONDS,
+        BACKFILL_CHUNK_SECONDS,
     )
     from backend.app.tasks.download import (
         DOWNLOAD_HARD_TIME_LIMIT,
@@ -42,7 +42,7 @@ def test_timeout_ladder_keeps_subprocess_budgets_under_soft_limit():
     )
 
     assert _DEFAULT_GDL_TIMEOUT_SECONDS < DOWNLOAD_SOFT_TIME_LIMIT
-    assert BACKFILL_TIMEOUT_SECONDS < DOWNLOAD_SOFT_TIME_LIMIT
+    assert BACKFILL_CHUNK_SECONDS < DOWNLOAD_SOFT_TIME_LIMIT
     assert DOWNLOAD_SOFT_TIME_LIMIT < DOWNLOAD_HARD_TIME_LIMIT
 
 

tests/test_gallery_dl_service.py

@@ -8,6 +8,7 @@ from backend.app.services.gallery_dl import (
     ErrorType,
     GalleryDLService,
     SourceConfig,
+    parse_last_cursor,
 )
 
 
@@ -326,3 +327,38 @@ def test_default_config_forwards_patreon_referer_to_ytdl(gdl):
     headers = cfg["downloader"]["ytdl"]["raw-options"]["http_headers"]
     assert headers["Referer"] == "https://www.patreon.com/"
     assert headers["Origin"] == "https://www.patreon.com"
+
+
+# --- Cursor-paged backfill (plan #689) -------------------------------------
+
+
+def test_parse_last_cursor_returns_last_match_across_streams():
+    # gallery-dl logs `Cursor: <token>` per page (debug → stderr); the last
+    # is the furthest-progressed resume point.
+    stderr = (
+        "[patreon][debug] Cursor: 03:AAAA:bbb\n"
+        "[urllib3] GET ...\n"
+        "[patreon][debug] Cursor: 03:CCCC:ddd\n"
+    )
+    assert parse_last_cursor("", stderr) == "03:CCCC:ddd"
+
+
+def test_parse_last_cursor_scans_stdout_too_and_none_when_absent():
+    assert parse_last_cursor("Cursor: 99:ZZZ", "") == "99:ZZZ"
+    assert parse_last_cursor("no marker here", "still nothing") is None
+    assert parse_last_cursor("", "") is None
+
+
+def test_build_config_patreon_resume_cursor_overrides_default(gdl):
+    # No resume cursor → PLATFORM_DEFAULTS leaves the log-only `cursor: True`.
+    fresh = gdl._build_config_for_source(
+        "patreon", SourceConfig(), "alice", skip_value=True,
+    )
+    assert fresh["extractor"]["patreon"]["cursor"] is True
+
+    # Resume cursor set → gallery-dl restarts the walk from that page.
+    resumed = gdl._build_config_for_source(
+        "patreon", SourceConfig(resume_cursor="03:CCCC:ddd"), "alice",
+        skip_value=True,
+    )
+    assert resumed["extractor"]["patreon"]["cursor"] == "03:CCCC:ddd"

tests/test_source_service.py

@@ -175,58 +175,75 @@ async def test_list_hides_sidecar_synthetic_anchors(db):
 
 
 @pytest.mark.asyncio
-async def test_set_backfill_runs_arms_source(db):
-    """The service method overrides backfill_runs_remaining (regardless of
-    the auto-arm-on-create starting value) and returns the updated record
-    so the API can echo it back."""
+async def test_start_backfill_arms_run_until_done(db):
+    """start_backfill sets state=running + the chunk cap and clears any prior
+    cursor/chunk state, returning the updated record for the API to echo."""
+    from backend.app.services.source_service import BACKFILL_MAX_CHUNKS
+
     artist = await _artist(db, "Alice")
     svc = SourceService(db)
     rec = await svc.create(
         artist_id=artist.id, platform="patreon",
         url="https://patreon.com/alice",
     )
-    updated = await svc.set_backfill_runs(rec.id, 5)
-    assert updated.backfill_runs_remaining == 5
+    # Simulate a prior, finished walk leaving stale checkpoint state.
+    await db.execute(
+        Source.__table__.update().where(Source.id == rec.id).values(
+            config_overrides={"_backfill_state": "complete", "_backfill_cursor": "old",
+                              "_backfill_chunks": 7},
+        )
+    )
+    await db.commit()
 
-    db_value = (await db.execute(
-        select(Source.backfill_runs_remaining).where(Source.id == rec.id)
+    updated = await svc.start_backfill(rec.id)
+    assert updated.backfill_state == "running"
+    assert updated.backfill_chunks == 0
+    assert updated.backfill_runs_remaining == BACKFILL_MAX_CHUNKS
+
+    co = (await db.execute(
+        select(Source.config_overrides).where(Source.id == rec.id)
     )).scalar_one()
-    assert db_value == 5
+    assert co.get("_backfill_state") == "running"
+    assert "_backfill_cursor" not in co
 
 
 @pytest.mark.asyncio
-async def test_set_backfill_runs_rejects_out_of_range(db):
+async def test_stop_backfill_returns_to_idle(db):
     artist = await _artist(db, "Alice")
     svc = SourceService(db)
     rec = await svc.create(
         artist_id=artist.id, platform="patreon",
         url="https://patreon.com/alice",
     )
-    with pytest.raises(ValueError):
-        await svc.set_backfill_runs(rec.id, 0)
-    with pytest.raises(ValueError):
-        await svc.set_backfill_runs(rec.id, 11)
+    await svc.start_backfill(rec.id)
+    updated = await svc.stop_backfill(rec.id)
+    assert updated.backfill_state is None
+    assert updated.backfill_runs_remaining == 0
+    co = (await db.execute(
+        select(Source.config_overrides).where(Source.id == rec.id)
+    )).scalar_one()
+    assert "_backfill_state" not in (co or {})
 
 
 @pytest.mark.asyncio
-async def test_set_backfill_runs_raises_when_source_missing(db):
+async def test_start_backfill_raises_when_source_missing(db):
     svc = SourceService(db)
     with pytest.raises(LookupError):
-        await svc.set_backfill_runs(99999, 3)
+        await svc.start_backfill(99999)
 
 
 @pytest.mark.asyncio
 async def test_new_enabled_source_starts_in_backfill_mode(db):
-    """Plan #544 follow-up: freshly added enabled sources have no archive
-    yet, so the first few polls would blow the wall-clock cap in tick
-    mode. Pre-arm backfill so the initial walks use the longer timeout."""
+    """Plan #693: freshly added enabled sources have no archive yet, so they
+    arm run-until-done backfill — state 'running' — to walk the full history
+    on the first ticks instead of blowing the wall-clock cap in tick mode."""
     artist = await _artist(db, "Alice")
     svc = SourceService(db)
     rec = await svc.create(
         artist_id=artist.id, platform="patreon",
         url="https://patreon.com/alice-new",
     )
-    assert rec.backfill_runs_remaining == 3
+    assert rec.backfill_state == "running"
 
 
 @pytest.mark.asyncio