Merge pull request 'Release: dev → main (Discord poster picker, chunk import deadline, CI repairs)' (#260) from dev into main

Two more reasons CI could not go green, found while fixing #6864.

1. sign-extension's "never go backwards" guard refused every weekly base
   refresh since at least 2026-10-04 (runs 8023, 9015, 9400, 9818). The
   refresh builds main, whose extension version (2026.9.25.1201) trails
   the highest one dev has signed since (2026.9.28.1403), so the guard
   exited before the cache lookup that would have found main's version
   already signed and reused it. The web image (:latest) was never
   refreshed. A version that is already signed now passes the guard: it
   is reused from its release, not signed again, so it cannot be a
   downgrade.

2. tests/test_api_workers.py read each lane's ceiling from the CI
   container's CPU quota and memory limit. Since run 10242 (2026-11-05)
   the runner's container allows fewer slots than two tests set (6 worker
   slots, 2 ML), so both got 400 "above what this container can hold".
   An autouse fixture now pins a 16-core, 64 GiB container, as
   test_worker_lanes.py already does.

Task #6864.

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

.forgejo/workflows/baseline.yml

@@ -81,8 +81,18 @@ jobs:
           set -eux
           # Same service-IP dance as build.yml's integration job; see the long
           # comment there for why the job name must stay separator-free.
-          PG=$(docker ps --filter "name=compare" --filter "ancestor=pgvector/pgvector:pg16" -q | head -n1)
-          test -n "$PG"
+          # Only THIS job's services. The runner's docker daemon is shared, so another
+          # repo's job can be running beside this one, and a job-name filter matches its
+          # containers too (Steward run 8358 and Inkwell run 8653 each took another repo's
+          # Postgres). act names every container of a task GITEA-ACTIONS-TASK-<n>-...: read
+          # <n> from our own container (its id is in the /etc/hostname bind mount) and
+          # require exactly one match.
+          SELF=$(grep -o '/containers/[0-9a-f]\{64\}/' /proc/self/mountinfo | head -n1 | cut -d/ -f3 || true)
+          SELF=${SELF:-$(hostname)}
+          TASK=$(docker inspect -f '{{.Name}}' "$SELF" | grep -o 'GITEA-ACTIONS-TASK-[0-9]*-')
+          test -n "$TASK"
+          PG=$(docker ps --filter "name=$TASK" --filter "ancestor=pgvector/pgvector:pg16" -q)
+          test "$(printf '%s\n' "$PG" | grep -c .)" -eq 1
           PG_IP=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "$PG")
           test -n "$PG_IP"
           echo "PG_CONTAINER=$PG" >> "$GITHUB_ENV"

.forgejo/workflows/build.yml

@@ -532,9 +532,20 @@ jobs:
           echo "=== container landscape (diagnostic for filter scoping) ==="
           docker ps -a --format '{{.ID}} {{.Image}} -> {{.Names}}'
           echo "=== end landscape ==="
-          PG=$(docker ps --filter "name=integration" --filter "ancestor=pgvector/pgvector:pg16" -q | head -n1)
-          RD=$(docker ps --filter "name=integration" --filter "ancestor=redis:8-alpine" -q | head -n1)
-          test -n "$PG" && test -n "$RD"
+          # Only THIS job's services. The runner's docker daemon is shared, so another
+          # repo's job can be running beside this one, and a job-name filter matches its
+          # containers too (Steward run 8358 and Inkwell run 8653 each took another repo's
+          # Postgres). act names every container of a task GITEA-ACTIONS-TASK-<n>-...: read
+          # <n> from our own container (its id is in the /etc/hostname bind mount) and
+          # require exactly one match.
+          SELF=$(grep -o '/containers/[0-9a-f]\{64\}/' /proc/self/mountinfo | head -n1 | cut -d/ -f3 || true)
+          SELF=${SELF:-$(hostname)}
+          TASK=$(docker inspect -f '{{.Name}}' "$SELF" | grep -o 'GITEA-ACTIONS-TASK-[0-9]*-')
+          test -n "$TASK"
+          PG=$(docker ps --filter "name=$TASK" --filter "ancestor=pgvector/pgvector:pg16" -q)
+          test "$(printf '%s\n' "$PG" | grep -c .)" -eq 1
+          RD=$(docker ps --filter "name=$TASK" --filter "ancestor=redis:8-alpine" -q)
+          test "$(printf '%s\n' "$RD" | grep -c .)" -eq 1
           PG_IP=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "$PG")
           RD_IP=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "$RD")
           test -n "$PG_IP" && test -n "$RD_IP"
@@ -776,6 +787,15 @@ jobs:
               print("guard: no ext-* release yet — nothing to go backwards from")
               raise SystemExit(0)
 
+          # A version that is already signed is reused from its release, not
+          # signed again, so it cannot be a downgrade. The weekly refresh builds
+          # main, whose version trails whatever dev has signed since; refusing
+          # it here would stop :latest from ever being refreshed while dev is
+          # ahead.
+          if any(v == derived for v, _ in signed):
+              print(f"guard: {derived_s} is already signed — reused, not re-signed")
+              raise SystemExit(0)
+
           hi, hi_tag = signed[-1]
           print(f"guard: derived={derived_s}  highest already signed={hi_tag}")
           if derived < hi:
@@ -1718,9 +1738,20 @@ jobs:
           # Service discovery mirrors the integration lane above: these jobs run
           # in a container against a mounted docker socket, so the services are
           # SIBLINGS reachable by IP, not by hostname.
-          PG=$(docker ps --filter "name=smoke" --filter "ancestor=pgvector/pgvector:pg16" -q | head -n1)
-          RD=$(docker ps --filter "name=smoke" --filter "ancestor=redis:8-alpine" -q | head -n1)
-          test -n "$PG" && test -n "$RD"
+          # Only THIS job's services. The runner's docker daemon is shared, so another
+          # repo's job can be running beside this one, and a job-name filter matches its
+          # containers too (Steward run 8358 and Inkwell run 8653 each took another repo's
+          # Postgres). act names every container of a task GITEA-ACTIONS-TASK-<n>-...: read
+          # <n> from our own container (its id is in the /etc/hostname bind mount) and
+          # require exactly one match.
+          SELF=$(grep -o '/containers/[0-9a-f]\{64\}/' /proc/self/mountinfo | head -n1 | cut -d/ -f3 || true)
+          SELF=${SELF:-$(hostname)}
+          TASK=$(docker inspect -f '{{.Name}}' "$SELF" | grep -o 'GITEA-ACTIONS-TASK-[0-9]*-')
+          test -n "$TASK"
+          PG=$(docker ps --filter "name=$TASK" --filter "ancestor=pgvector/pgvector:pg16" -q)
+          test "$(printf '%s\n' "$PG" | grep -c .)" -eq 1
+          RD=$(docker ps --filter "name=$TASK" --filter "ancestor=redis:8-alpine" -q)
+          test "$(printf '%s\n' "$RD" | grep -c .)" -eq 1
           PG_IP=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "$PG")
           RD_IP=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "$RD")
           test -n "$PG_IP" && test -n "$RD_IP"

backend/app/services/download_service.py

@@ -13,6 +13,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import time
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
@@ -39,7 +40,7 @@ from .gallery_dl import (
     walk_completed,
 )
 from .importer import Importer
-from .ingest_core import DEFAULT_REVISIT_DAYS
+from .ingest_core import DEFAULT_REVISIT_DAYS, PHASE3_DEADLINE_SECONDS
 from .platforms import auth_type_for
 from .scheduler_service import set_platform_cooldown
 
@@ -82,6 +83,7 @@ class DownloadService:
 
     async def download_source(self, source_id: int) -> int:
         """Returns DownloadEvent.id. Idempotent: in-flight events are returned as-is."""
+        started = time.monotonic()
         setup = await self._phase1_setup(source_id)
         if setup["status"] in ("skipped", "in_flight"):
             return setup["event_id"]
@@ -174,6 +176,7 @@ class DownloadService:
 
         return await self._phase3_persist(
             ctx["event_id"], ctx, dl_result, resolved_campaign_id,
+            started=started,
         )
 
     async def _run_download(
@@ -268,6 +271,8 @@ class DownloadService:
         ctx: dict,
         dl_result,
         resolved_campaign_id: str | None,
+        *,
+        started: float | None = None,
     ) -> int:
         # Cache Patreon campaign id on the source.
         if resolved_campaign_id:
@@ -303,7 +308,23 @@ class DownloadService:
 
         loop = asyncio.get_running_loop()
 
-        for path_str in dl_result.written_paths or []:
+        # Phase 3 shares the task's soft limit with the walk, and an import's
+        # cost is only estimated there. Native platforms stop starting imports
+        # at the deadline: what is left stays on disk with no record, which the
+        # next chunk's walk imports. gallery-dl platforms have no such pickup,
+        # so they import everything.
+        deadline = (
+            started + PHASE3_DEADLINE_SECONDS
+            if started is not None and uses_native_ingester(ctx["platform"])
+            else None
+        )
+        written_paths = list(dl_result.written_paths or [])
+        deferred_from: int | None = None
+
+        for idx, path_str in enumerate(written_paths):
+            if deadline is not None and time.monotonic() >= deadline:
+                deferred_from = idx
+                break
             if path_str in (dl_result.quarantined_paths or []):
                 continue
             path = Path(path_str)
@@ -427,8 +448,16 @@ class DownloadService:
         # has been through the importer. Had the run died before here they stay
         # unmarked, and the next walk imports them from disk (ingest_core).
         mark_seen = getattr(dl_result, "mark_seen_after_import", None)
+        deferred = written_paths[deferred_from:] if deferred_from is not None else []
         if mark_seen is not None:
-            await loop.run_in_executor(None, mark_seen)
+            if deferred:
+                await loop.run_in_executor(
+                    None, lambda: mark_seen(skip_paths=frozenset(deferred)),
+                )
+            else:
+                await loop.run_in_executor(None, mark_seen)
+        if deferred:
+            self._defer_unimported(dl_result, deferred_from, len(deferred))
 
         # #830 recapture: backfill source_filehash on EXISTING on-disk images so
         # their post-body inline <img src=CDN> remaps to the local copy. A
@@ -530,6 +559,33 @@ class DownloadService:
         await self.async_session.commit()
         return event_id
 
+    @staticmethod
+    def _defer_unimported(dl_result, first: int, count: int) -> None:
+        """Turn this chunk into a chunk boundary at the first file phase 3 left.
+
+        The next chunk resumes on the page that file came from, so its walk
+        meets every left file again, on disk with no record, and imports it. A
+        walk that reached its end has not finished while files wait, so a clean
+        result becomes PARTIAL; a real failure keeps its own error.
+        """
+        cursors = getattr(dl_result, "written_cursors", None) or []
+        if first < len(cursors):
+            dl_result.cursor = cursors[first]
+        dl_result.phase3_imported_any = first > 0
+        msg = (
+            f"Imports stopped at the chunk deadline: {count} file(s) left "
+            "for the next chunk"
+        )
+        dl_result.stdout = f"{dl_result.stdout}\n{msg}" if dl_result.stdout else msg
+        log.info("%s", msg)
+        if dl_result.error_type in (None, ErrorType.PARTIAL, ErrorType.TIMEOUT) or (
+            is_informational(dl_result.error_type)
+        ):
+            dl_result.success = False
+            dl_result.return_code = -1
+            dl_result.error_type = ErrorType.PARTIAL
+            dl_result.error_message = msg
+
     async def _apply_backfill_lifecycle(self, ctx: dict, dl_result) -> None:
         """Backfill state machine (plan #693, building on the cursor of #689).
 
@@ -597,9 +653,12 @@ class DownloadService:
         new_cursor = (
             dl_result.cursor if uses_native_ingester(ctx["platform"]) else None
         )
+        # Imports phase 3 finished before its deadline are progress even when
+        # the cursor went back to the page holding the files it left.
         advanced = bool(
             (new_cursor and new_cursor != old_cursor)
             or dl_result.files_downloaded > 0
+            or getattr(dl_result, "phase3_imported_any", False)
         )
         if advanced:
             if new_cursor:

backend/app/services/gallery_dl.py

@@ -169,6 +169,10 @@ class DownloadResult:
     # backfilled (inline-image localization) WITHOUT re-download or unlink. Empty
     # on the gallery-dl path and outside recapture.
     relink_source_paths: list[tuple[str, str, str]] = field(default_factory=list)
+    # Native ingester only: one entry per `written_paths` entry, the backfill
+    # cursor of the page that file came from. Phase 3 resumes the next chunk
+    # there when it leaves files unimported at its deadline. Empty on gallery-dl.
+    written_cursors: list[str | None] = field(default_factory=list)
     stdout: str = ""
     stderr: str = ""
     return_code: int = 0
@@ -194,8 +198,10 @@ class DownloadResult:
     # but not yet imported is invisible to every later walk, so a run killed in
     # between orphaned it for good (TamadaHeijun's 12PCG post lost 7 of 13
     # images to a stranded run, 2026-09-24). Unmarked, the next walk finds the
-    # file on disk with no ImageRecord and imports it. None on gallery-dl.
-    mark_seen_after_import: Callable[[], None] | None = None
+    # file on disk with no ImageRecord and imports it. Takes `skip_paths`: the
+    # written paths phase 3 did not import, which stay unmarked for the same
+    # reason. None on gallery-dl.
+    mark_seen_after_import: Callable[..., None] | None = None
 
 
 def extract_errors_warnings(stderr: str) -> str:

backend/app/services/ingest_core.py

@@ -114,13 +114,29 @@ _CANARY_MIN_SAMPLE = 30
 # on-disk files, and died at the soft limit.
 #
 # So the walk also stops when its elapsed time PLUS phase 3's estimated cost
-# would pass CHUNK_TOTAL_SECONDS. Costs measured on the live instance: 431
-# imports took 976s (~2.3s each: hash, pHash, sidecar, provenance); a relink is
-# a sha256 over NFS, 0.15s for the 8.8 MB average file, plus a lookup.
-# test_download_source_task pins CHUNK_TOTAL_SECONDS under the soft limit.
+# would pass CHUNK_TOTAL_SECONDS. An import (hash, pHash, sidecar, provenance,
+# thumbnail enqueue) runs 2-5s depending on the files; the estimate takes the
+# slow end. A relink is a sha256 over NFS plus a lookup.
+#
+# The estimate is a plan, not a guard. PHASE3_DEADLINE_SECONDS is the guard:
+# phase 3 starts no import past it (download_service), and leaves the rest on
+# disk, unmarked, for the next chunk. It sits under the soft limit by enough to
+# finish the import in hand, upsert post records and finalize the event.
+# test_download_source_task pins both under the soft limit.
 CHUNK_TOTAL_SECONDS = 1200.0
-PHASE3_IMPORT_SECONDS = 2.5
+PHASE3_IMPORT_SECONDS = 5.0
 PHASE3_RELINK_SECONDS = 0.25
+PHASE3_DEADLINE_SECONDS = 1200.0
+
+
+def _unskipped(
+    fetched: list[tuple[str, str]],
+    written_keys: dict[str, tuple[str, str]],
+    skip_paths,
+) -> list[tuple[str, str]]:
+    """`fetched` without the ledger keys of files phase 3 left unimported."""
+    skip = {written_keys[p] for p in skip_paths if p in written_keys}
+    return [kv for kv in fetched if kv not in skip] if skip else list(fetched)
 
 
 def _parse_published(raw: object) -> datetime | None:
@@ -289,6 +305,12 @@ class Ingester:
         # Media handed to phase 3 for import. Marked seen by phase 3 once the
         # import has run (`mark_seen_after_import`), not here — see there.
         fetched: list[tuple[str, str]] = []
+        # Per entry of `written`: the cursor of the page it came from, and its
+        # ledger key. Phase 3 can stop importing at its deadline; the files it
+        # leaves must stay unmarked, and the next chunk has to resume on the
+        # page holding the first of them, or no walk sees them again.
+        written_cursors: list[str | None] = []
+        written_keys: dict[str, tuple[str, str]] = {}
         # Post-record keys written this walk. Marked with the media, after
         # phase 3 has upserted the records — marking them at write time left a
         # walk killed before phase 3 with posts the ledger calls recorded that
@@ -347,8 +369,10 @@ class Ingester:
                 written_paths=written,
                 post_record_paths=list(post_records),
                 relink_source_paths=list(relink),
-                mark_seen_after_import=lambda: self._mark_seen(
-                    source_id, fetched + recorded,
+                written_cursors=list(written_cursors),
+                mark_seen_after_import=lambda skip_paths=(): self._mark_seen(
+                    source_id,
+                    _unskipped(fetched, written_keys, skip_paths) + recorded,
                 ),
                 stdout="\n".join(log_lines),
                 stderr="",
@@ -577,6 +601,8 @@ class Ingester:
                         downloaded += 1
                         if outcome.path is not None:
                             written.append(str(outcome.path))
+                            written_cursors.append(emitted_cursor)
+                            written_keys[str(outcome.path)] = (key, media_item.post_id)
                         fetched.append((key, media_item.post_id))
                         to_clear.append(key)
                         consecutive_seen = 0
@@ -590,6 +616,8 @@ class Ingester:
                         # attach_in_place because no record owns this path —
                         # the unlink below is about a file that IS the record.
                         written.append(str(outcome.path))
+                        written_cursors.append(emitted_cursor)
+                        written_keys[str(outcome.path)] = (key, media_item.post_id)
                         fetched.append((key, media_item.post_id))
                         to_clear.append(key)
                         skipped_count += 1

tests/conftest.py

@@ -85,6 +85,29 @@ def db_sync(sync_engine):
             session.rollback()
 
 
+@pytest.fixture
+def celery_conf():
+    """Set Celery config keys for one test: `celery_conf(task_always_eager=True)`.
+    The values they had before the test come back after it.
+
+    Use this, not `monkeypatch.setattr(celery.conf, ...)`. Celery serves config
+    keys through `__getattr__`, so monkeypatch sees no attribute to restore and
+    deletes the key on undo instead; the delete raises, and every later undo of
+    that test is skipped, leaking its patches into the tests after it.
+    """
+    from backend.app.celery_app import celery
+
+    saved: dict[str, object] = {}
+
+    def _set(**values: object) -> None:
+        for key in values:
+            saved.setdefault(key, celery.conf[key])
+        celery.conf.update(values)
+
+    yield _set
+    celery.conf.update(saved)
+
+
 @pytest_asyncio.fixture
 async def app():
     return create_app()

tests/test_api_cleanup.py

@@ -11,15 +11,14 @@ import pytest
 from PIL import Image
 from sqlalchemy import func, select
 
-from backend.app.celery_app import celery
 from backend.app.models import ImageRecord, LibraryAuditRun
 
 pytestmark = pytest.mark.integration
 
 
 @pytest.fixture(autouse=True)
-def disable_celery_eager(monkeypatch):
-    monkeypatch.setattr(celery.conf, "task_always_eager", False)
+def disable_celery_eager(celery_conf):
+    celery_conf(task_always_eager=False)
 
 
 def _sha256_min_dim_token(min_w: int, min_h: int) -> str:

tests/test_api_workers.py

@@ -13,12 +13,25 @@ from sqlalchemy import select
 from backend.app.api import workers as workers_api
 from backend.app.models import WorkerLane
 from backend.app.services import worker_control as wc
+from backend.app.services import worker_lanes as wl
 from backend.app.services.worker_lanes import LANES, SWEEP_PERIOD_SECONDS
 from tests.doubles import RecordingSession
 
 pytestmark = pytest.mark.integration
 
 
+@pytest.fixture(autouse=True)
+def a_known_machine(monkeypatch):
+    """A 16-core, 64 GiB container, whatever runs the suite.
+
+    Each lane's ceiling is derived from the container's CPU quota and memory
+    limit, so without this the caps these tests may set depend on the CI
+    runner's size, and a smaller runner refuses them as above the ceiling.
+    """
+    monkeypatch.setattr(wl, "container_cpu_count", lambda: 16)
+    monkeypatch.setattr(wl, "container_memory_bytes", lambda: 64 * wl.GIB)
+
+
 @pytest_asyncio.fixture
 async def no_live_workers(monkeypatch):
     """Nothing is running — which is the CI lane's actual truth, asserted

tests/test_download_service.py

@@ -570,6 +570,64 @@ async def test_backfill_chunk_progress_advances_cursor(
     assert remaining == 4
 
 
+@pytest.mark.asyncio
+async def test_phase_3_leaves_imports_past_its_deadline_for_the_next_chunk(
+    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
+):
+    """A walk that reached its end handed phase 3 more imports than fit before
+    the deadline. Phase 3 imports what fits, leaves the rest on disk and
+    unmarked, keeps the backfill running, and resumes it on the page holding
+    the first file it left."""
+    import backend.app.services.download_service as ds
+
+    _artist, source = seed_artist_and_source
+    source.config_overrides = {"_backfill_state": "running", "_backfill_cursor": "03:PAGE1"}
+    source.backfill_runs_remaining = 5
+    await db.commit()
+
+    images_root = tmp_path / "images"
+    f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
+    f2 = images_root / "alice" / "patreon" / "post" / "b.jpg"
+    _make_jpg(f1, split="h")
+    _make_jpg(f2, split="v")
+    result = _make_fake_dl_result(
+        success=True, written_paths=[str(f1), str(f2)], cursor="03:PAGE3",
+    )
+    result.written_cursors = ["03:PAGE1", "03:PAGE2"]
+    left_unmarked = []
+    result.mark_seen_after_import = lambda skip_paths=(): left_unmarked.append(
+        set(skip_paths)
+    )
+
+    # The clock is past the deadline once the first import is done.
+    past = ds.PHASE3_DEADLINE_SECONDS + 1
+    ticks = iter([0.0, 0.0])
+    monkeypatch.setattr(ds, "time", SimpleNamespace(monotonic=lambda: next(ticks, past)))
+
+    svc, _ = _backfill_svc(db, db_sync, tmp_path, result)
+    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.files_count == 1
+    assert "1 file(s) left for the next chunk" in ev.metadata_["stdout"]
+    assert left_unmarked == [{str(f2)}]
+    assert f2.exists()
+    assert db_sync.execute(
+        select(func.count(ImageRecord.id)).where(ImageRecord.path == str(f2))
+    ).scalar_one() == 0
+
+    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"
+    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,

tests/test_download_source_task.py

@@ -55,6 +55,15 @@ def test_the_native_chunk_leaves_room_to_tear_down_before_the_soft_limit():
     assert CHUNK_TOTAL_SECONDS <= DOWNLOAD_SOFT_TIME_LIMIT - 120
 
 
+def test_phase_3_stops_importing_with_room_to_finalize():
+    """Past its deadline phase 3 starts no import; what remains after it (the
+    import in hand, post records, the event) has to fit before the soft limit."""
+    from backend.app.services.ingest_core import PHASE3_DEADLINE_SECONDS
+    from backend.app.tasks.download import DOWNLOAD_SOFT_TIME_LIMIT
+
+    assert PHASE3_DEADLINE_SECONDS <= DOWNLOAD_SOFT_TIME_LIMIT - 120
+
+
 def test_decorated_limits_match_module_constants():
     """The @celery.task decorator must use the audited constants, not
     drifted literals."""
@@ -158,7 +167,7 @@ async def test_finalize_soft_limited_is_noop_without_running_event(db_sync):
 @pytest.mark.integration
 @pytest.mark.asyncio
 async def test_download_source_catches_soft_limit_and_salvages_event(
-    db_sync, monkeypatch,
+    db_sync, monkeypatch, celery_conf,
 ):
     """End-to-end wiring: when the inner run raises SoftTimeLimitExceeded,
     download_source's handler must flip the in-flight event to error
@@ -169,8 +178,7 @@ async def test_download_source_catches_soft_limit_and_salvages_event(
     import backend.app.tasks.download as dl
     from backend.app.models import DownloadEvent
 
-    monkeypatch.setattr(celery.conf, "task_always_eager", True)
-    monkeypatch.setattr(celery.conf, "task_eager_propagates", False)
+    celery_conf(task_always_eager=True, task_eager_propagates=False)
     # This test exercises the soft-limit salvage path, not the per-platform
     # serialization. Neutralize the Redis lock so it can't defer/recurse under
     # eager mode (and so it doesn't contend with test_platform_lock).

tests/test_patreon_ingester.py

@@ -984,6 +984,31 @@ async def test_recapture_imports_a_seen_file_nothing_imported(
     assert downloader.download_calls == 0
 
 
+@pytest.mark.asyncio
+async def test_files_phase_3_leaves_unimported_stay_unmarked(
+    source_id, sync_engine, tmp_path,
+):
+    """Phase 3 can stop importing at its deadline. Each written file names the
+    page it came from, so the next chunk can resume there; the files left keep
+    no ledger mark, so that walk meets them on disk and imports them."""
+    m1, m2 = _media("p1", 1), _media("p2", 1)
+    pages = [("CUR1", [("p1", [m1])]), ("CUR2", [("p2", [m2])])]
+    ing = _ingester(sync_engine, tmp_path, _FakeClient(pages), _FakeDownloader(tmp_path))
+    result = ing.run(source_id=source_id, campaign_id="c1", artist_slug="ingest",
+                     url="https://patreon.com/ingest", mode="backfill")
+    assert result.written_cursors == ["CUR1", "CUR2"]
+
+    result.mark_seen_after_import(skip_paths={str(tmp_path / "p2_1.jpg")})
+    assert _count_ledger(sync_engine, source_id) == 3   # m1 and both post records
+
+    ing2 = _ingester(sync_engine, tmp_path, _FakeClient(pages),
+                     _FakeDownloader(tmp_path, on_disk={_ledger_key(m1), _ledger_key(m2)}))
+    again = ing2.run(source_id=source_id, campaign_id="c1", artist_slug="ingest",
+                     url="https://patreon.com/ingest", mode="backfill")
+    assert again.written_paths == [str(tmp_path / "p2_1.jpg")]
+    assert "on disk but never imported: p2_1.jpg" in again.stdout
+
+
 # --- dead-letter ledger (plan #705 #7) ------------------------------------
 
 

tests/test_task_run_signals.py

@@ -20,10 +20,9 @@ pytestmark = pytest.mark.integration
 
 
 @pytest.fixture(autouse=True)
-def _eager_celery(monkeypatch):
+def _eager_celery(celery_conf):
     """task_always_eager so signals fire in-process during the test."""
-    monkeypatch.setattr(celery.conf, "task_always_eager", True)
-    monkeypatch.setattr(celery.conf, "task_eager_propagates", False)
+    celery_conf(task_always_eager=True, task_eager_propagates=False)
 
 
 @pytest.mark.asyncio

tests/test_tasks_admin.py

@@ -15,9 +15,8 @@ pytestmark = pytest.mark.integration
 
 
 @pytest.fixture(autouse=True)
-def _eager_celery(monkeypatch):
-    monkeypatch.setattr(celery.conf, "task_always_eager", True)
-    monkeypatch.setattr(celery.conf, "task_eager_propagates", False)
+def _eager_celery(celery_conf):
+    celery_conf(task_always_eager=True, task_eager_propagates=False)
 
 
 @pytest.fixture(autouse=True)

tests/test_tasks_backup.py

@@ -10,16 +10,14 @@ from pathlib import Path
 import pytest
 from sqlalchemy import select
 
-from backend.app.celery_app import celery
 from backend.app.models import BackupRun, ImportSettings
 
 pytestmark = pytest.mark.integration
 
 
 @pytest.fixture(autouse=True)
-def _eager_celery(monkeypatch):
-    monkeypatch.setattr(celery.conf, "task_always_eager", True)
-    monkeypatch.setattr(celery.conf, "task_eager_propagates", False)
+def _eager_celery(celery_conf):
+    celery_conf(task_always_eager=True, task_eager_propagates=False)
 
 
 @pytest.fixture(autouse=True)

tests/test_tasks_library_audit.py

@@ -9,7 +9,6 @@ from PIL import Image
 from sqlalchemy import select
 
 import backend.app.tasks.library_audit  # noqa: F401 — celery registration
-from backend.app import celery_app
 from backend.app.models import ImageRecord, LibraryAuditRun
 
 pytestmark = pytest.mark.integration
@@ -37,7 +36,7 @@ def _mk_image(db_sync, tmp_path, *, mode, color, name):
 
 
 def test_scan_library_for_rule_populates_matched_ids_for_transparency(
-    db_sync, tmp_path, monkeypatch,
+    db_sync, tmp_path, celery_conf,
 ):
     transparent_rec, _ = _mk_image(
         db_sync, tmp_path, mode="RGBA", color=(0, 0, 0, 0), name="trans.png",
@@ -53,10 +52,10 @@ def test_scan_library_for_rule_populates_matched_ids_for_transparency(
     db_sync.commit()
     audit_id = audit.id
 
-    monkeypatch.setattr(celery_app.celery.conf, "task_always_eager", True)
+    celery_conf(task_always_eager=True)
     from backend.app.tasks.library_audit import scan_library_for_rule
     scan_library_for_rule.run(audit_id)
-    monkeypatch.setattr(celery_app.celery.conf, "task_always_eager", False)
+    celery_conf(task_always_eager=False)
 
     matched = db_sync.execute(
         select(LibraryAuditRun.matched_ids).where(LibraryAuditRun.id == audit_id)
@@ -124,7 +123,7 @@ def test_scan_resumes_and_accumulates_matched(db_sync, tmp_path, monkeypatch):
 
 
 def test_scan_library_for_rule_skips_missing_files_gracefully(
-    db_sync, tmp_path, monkeypatch,
+    db_sync, tmp_path, celery_conf,
 ):
     rec, path = _mk_image(
         db_sync, tmp_path, mode="RGBA", color=(0, 0, 0, 0), name="ghost.png",
@@ -138,10 +137,10 @@ def test_scan_library_for_rule_skips_missing_files_gracefully(
     db_sync.commit()
     audit_id = audit.id
 
-    monkeypatch.setattr(celery_app.celery.conf, "task_always_eager", True)
+    celery_conf(task_always_eager=True)
     from backend.app.tasks.library_audit import scan_library_for_rule
     scan_library_for_rule.run(audit_id)
-    monkeypatch.setattr(celery_app.celery.conf, "task_always_eager", False)
+    celery_conf(task_always_eager=False)
 
     status = db_sync.execute(
         select(LibraryAuditRun.status).where(LibraryAuditRun.id == audit_id)