fix: phase 3 stops importing at a deadline and leaves the rest for the next chunk

A Discord recapture chunk on #banana-land was killed at the 1350s soft limit
twice (events 92354 on 2026-09-28 and 92623 on 2026-09-29). The walk's
budget (CHUNK_TOTAL_SECONDS, #4407) assumed 2.5s per import; measured on
those chunks, a real import of the source's doodles took about 5s, so ~290
imports the walk had planned to fit took about twice the time reserved.

- PHASE3_DEADLINE_SECONDS (1200s from task start): on native platforms
  phase 3 starts no import past it. The files left stay on disk with no
  record and are not marked seen; the next chunk resumes on the page that
  holds the first of them and imports them through the existing "on disk
  but never imported" path. A walk that reached its end becomes PARTIAL so
  the backfill is not marked complete while files wait.
- DownloadResult.written_cursors records each written file's page cursor;
  mark_seen_after_import takes skip_paths.
- PHASE3_IMPORT_SECONDS 2.5 -> 5.0, the slow end of what was measured, so
  the walk rarely leans on the deadline.

Task #4493.

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

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