feat(native-ingest): durable run logging that survives a worker kill (#899 L1/L3, DRY 3/3)

DRY pass commit 3 — the observability half. ingest_core accumulated ALL
human-readable progress in log_lines → DownloadResult.stdout, persisted to the
DownloadEvent ONLY at phase 3; the real logger was used almost nowhere. So a
worker SIGKILL/OOM/hard-time-limit mid-walk left NO trace (the "task died,
no trace" mode from the recovery-sweep work).

Route run milestones through the container log too, each carrying source_id (L3
context):
- run START (platform/mode/source/campaign/resume_cursor)
- per-PAGE breadcrumb (posts/downloaded/skipped/errors/quarantined/gated/cursor)
  at each page boundary — pages are minutes apart on big backfills, so this shows
  how far a since-died walk got
- final SUMMARY (same string as the stdout summary)
- operator STOP

(L2 — quarantines log.warning'd — already landed in commit 1's base
_validate_path; failures log.warning via the base _failure_result; the #862 body
canary already log.error's.) log_lines/stdout content is unchanged (summary just
captured in a var), so existing assertions hold.

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

backend/app/services/ingest_core.py

@@ -240,6 +240,15 @@ class Ingester:
                 ),
             )
 
+        # #899 L1: emit run milestones through the real logger (not only the
+        # in-memory log_lines → DownloadResult.stdout, which is persisted to the
+        # DownloadEvent ONLY at phase 3). A worker SIGKILL/OOM/hard-time-limit
+        # mid-walk would otherwise leave NO trace; these land in the container log
+        # in real time regardless of whether the event gets finalized.
+        log.info(
+            "%s ingest START (%s): source=%s campaign=%s resume_cursor=%s",
+            self._platform, mode, source_id, campaign_id, resume_cursor,
+        )
         try:
             for post, included, page_cursor in self.client.iter_posts(
                 campaign_id, cursor=resume_cursor
@@ -249,6 +258,15 @@ class Ingester:
                 # after it. Carried as DownloadResult.cursor (plan #704).
                 if page_cursor and page_cursor != emitted_cursor:
                     emitted_cursor = page_cursor
+                    # #899 L1: a per-page breadcrumb in the container log (pages can
+                    # be minutes apart on image-dense backfills) — survives a worker
+                    # kill so the operator sees how far a since-died walk got.
+                    log.info(
+                        "%s ingest progress (%s, source=%s): posts=%d downloaded=%d "
+                        "skipped=%d errors=%d quarantined=%d gated=%d cursor=%s",
+                        self._platform, mode, source_id, posts_processed, downloaded,
+                        skipped_count, errors, quarantined, gated_skipped, emitted_cursor,
+                    )
                     # plan #705 #6: persist the cursor at each page boundary so a
                     # worker SIGKILL mid-chunk resumes near the crash, not the
                     # chunk start. (phase 3 still writes the final cursor — same
@@ -450,6 +468,10 @@ class Ingester:
         # posts), so don't re-write them — return PARTIAL (reads as "ok/progress",
         # the lifecycle no-ops since state is gone) instead of a false "complete".
         if stopped:
+            log.info(
+                "%s ingest STOPPED by operator (%s, source=%s): %d file(s) this chunk",
+                self._platform, mode, source_id, downloaded,
+            )
             return _result(
                 success=False, return_code=-1,
                 error_type=ErrorType.PARTIAL,
@@ -467,7 +489,7 @@ class Ingester:
             log_lines.append(f"{quarantined} media item(s) quarantined (invalid)")
         if dead_lettered:
             log_lines.append(f"{dead_lettered} media item(s) skipped (dead-lettered)")
-        log_lines.append(
+        summary = (
             f"{self._platform} ingest ({mode}): {downloaded} downloaded, "
             f"{skipped_count} skipped, {quarantined} quarantined, "
             f"{dead_lettered} dead-lettered, {errors} error(s), "
@@ -481,6 +503,9 @@ class Ingester:
             + (", reached end" if reached_bottom else "")
             + (", time-boxed" if budget_hit else "")
         )
+        log_lines.append(summary)
+        # #899 L1: also to the container log (survives event-finalization failure).
+        log.info("%s (source=%s)", summary, source_id)
 
         if budget_hit:
             # A chunk that hit its time-box but made forward progress is a