feat(run_pipeline): add --force-reextract option for bulk re-extraction runs

MPCAutofill/cardpicker/management/commands/run_pipeline.py

@@ -348,6 +348,13 @@ def add_arguments(self, parser: CommandParser) -> None:
             default=None,
             help="Stop after N micro-batches (default: process all cards).",
         )
+        parser.add_argument(
+            "--force-reextract",
+            dest="force_reextract",
+            action="store_true",
+            default=False,
+            help="Force re-extraction of Stage C evidence for every card, ignoring prior runs.",
+        )
         parser.add_argument("--skip-stage-c", dest="skip_stage_c", action="store_true", default=False)
         parser.add_argument("--skip-stage-d", dest="skip_stage_d", action="store_true", default=False)
         parser.add_argument("--skip-clustering", dest="skip_clustering", action="store_true", default=False)
@@ -656,6 +663,14 @@ def _run_streaming_stages(
         limit: Optional[int] = options.get("limit")
         short_circuit = False if options.get("no_shortcircuit") else None
 
+        # Each micro-batch's PilotRunLedger row needs a UNIQUE id (`PilotRunLedger.run_id` is a
+        # unique constraint), while every data row this pass writes must stay under the operator's
+        # clean `run_id` (channel_report scopes by the run_id on the rows - comment above the
+        # pipeline's own ledger create). The attempt timestamp makes the ledger id unique across
+        # resumes too: re-running `--run-id <same>` re-dispatches batch 0, and its prior-attempt
+        # ledger row must not collide (stage_e_dispatch.dispatch_micro_batch's `ledger_run_id`).
+        attempt = timezone.now().strftime("%Y%m%dT%H%M%S%f")
+
         queryset = Card.objects.filter(content_phash__isnull=False).order_by("pk")
 
         after_pk = 0
@@ -710,9 +725,10 @@ def _run_streaming_stages(
                 trigger_reason="pipeline",
                 run_id=run_id,
                 batch_size=len(chunk),
-                force_stage_c_reextract=False,
+                force_stage_c_reextract=options["force_reextract"],
                 short_circuit=short_circuit,
                 dry_run=dry_run,
+                ledger_run_id=f"{run_id}-{attempt}Z-b{batch_count}",
             )
 
             batch_count += 1

MPCAutofill/cardpicker/stage_e_dispatch.py

@@ -1356,6 +1356,7 @@ def dispatch_micro_batch(
     force_stage_c_reextract: bool = False,
     short_circuit: Optional[bool] = None,
     dry_run: bool = False,
+    ledger_run_id: Optional[str] = None,
 ) -> DispatchOutcome:
     """
     The CONVEYOR itself - one micro-batch dispatch decision (docs/proposals/stage-e-streaming.md
@@ -1382,6 +1383,18 @@ def dispatch_micro_batch(
     `--dry-run` flag is the sole production caller; dispatch from the event system always passes
     `dry_run=False`.
 
+    `ledger_run_id` (2026-07-31): DECOUPLES the micro-batch's PilotRunLedger row identity from
+    the run_id its data is stamped with. By default the ledger row takes `dispatch_run_id`
+    (`run_id` or the auto-minted `stage-e-stream-*` id) - identical to every caller's data stamp,
+    which is what every drill/test below pins. `run_pipeline` is the one caller whose DATA must
+    keep the operator's clean run_id (`channel_report` scopes by the run_id ON THE ROWS, see its
+    own ledger-comment at run_pipeline.py) while each micro-batch's ledger row must be UNIQUE
+    (`PilotRunLedger.run_id` is a unique constraint) - a multi-batch pass passing the same
+    `run_id` for every dispatch would collide on the second batch. Passing `ledger_run_id`
+    (`<clean run_id>-<attempt timestamp>-b<batch num>` from the pipeline) gives that caller a
+    unique per-attempt, per-batch ledger row while `run_id` keeps stamping every data row with
+    the clean identity. When `ledger_run_id` is None this function behaves exactly as before.
+
     Ordering: no-self-resume gate -> fresh envelope sample -> batch selection ->
     concurrency-cap slot acquire (`cardpicker.stage_e_concurrency`) -> Stage C (sequential, per-card)
     -> Stage D (AS-IS entry points, scoped) -> ledger write -> slot release. Every gate below returns
@@ -1471,9 +1484,13 @@ def dispatch_micro_batch(
         # "Phase 2" section): one PilotRunLedger row per micro-batch dispatch, `command=
         # "stage_e_streaming_dispatch"`. `dry_run` is False for event-system dispatches and forwarded
         # from the CLI `--dry-run` flag for pipeline dispatches (2026-07-30, the streaming pipeline
-        # always runs all stages and reports, just without persisting when --dry-run is set).
+        # always runs all stages and reports, just without persisting when --dry-run is set). When a
+        # caller passes `ledger_run_id`, that becomes this row's identity (the pipeline needs a
+        # UNIQUE per-attempt, per-batch id here while `run_id` keeps stamping the data); otherwise the
+        # row takes `dispatch_run_id`, the same id its data is stamped with - see the param's own
+        # docstring paragraph.
         ledger = PilotRunLedger.objects.create(
-            run_id=dispatch_run_id,
+            run_id=ledger_run_id or dispatch_run_id,
             command="stage_e_streaming_dispatch",
             dry_run=dry_run,
             status=PilotRunLedger.Status.RUNNING,

MPCAutofill/cardpicker/tests/test_run_pipeline.py

@@ -358,6 +358,30 @@ def test_a_bare_invocation_runs_every_stage_and_produces_rows(
         assert "channel_report exit=" in out
         assert "VOTE CHANNELS" in out
 
+    def test_a_multi_batch_run_gives_every_micro_batch_a_unique_ledger_row(self, cohort: dict[str, Any]) -> None:
+        """
+        The 2026-07-31 run_id collision: a multi-batch pass under one `--run-id` used to hand the
+        SAME id to every `dispatch_micro_batch`, and `PilotRunLedger.run_id` is UNIQUE - so batch 1
+        (and every later batch) died with an IntegrityError and the run could never finish a whole
+        catalogue. The data must stay under the operator's clean run_id (channel_report scopes by
+        the run_id on the rows) while each micro-batch's ledger row is unique (`ledger_run_id`).
+        """
+        _run("--batch-size", "1", run_id="multi-batch")
+
+        summary = PilotRunLedger.objects.get(command="run_pipeline", run_id="multi-batch-pipeline")
+        assert summary.status == PilotRunLedger.Status.COMPLETED
+
+        batches = list(
+            PilotRunLedger.objects.filter(command="stage_e_streaming_dispatch", run_id__startswith="multi-batch-")
+        )
+        assert len(batches) >= 2, "the --batch-size 1 pass must span more than one micro-batch"
+        assert all(row.status == PilotRunLedger.Status.COMPLETED for row in batches)
+        assert len({row.run_id for row in batches}) == len(batches), "every dispatch ledger row must be unique"
+        assert all("-b" in row.run_id for row in batches), "ledger rows must be suffixed per batch, not bare run_id"
+        assert not PilotRunLedger.objects.filter(command="stage_e_streaming_dispatch", run_id="multi-batch").exists()
+
+        assert ImageEvidence.objects.filter(run_id="multi-batch").count() == 2
+
     def test_cluster_propagation_gives_an_unfetched_member_its_groups_verdict(self, cohort: dict[str, Any]) -> None:
         """
         THE HIGHEST-VALUE CARRIED FEATURE. `absorbed` never fetched, never extracted and never

docs/features/stage-e-operations.md

@@ -1704,6 +1704,16 @@ suppresses work — Stage C's resume filter is run-scoped (PR #645) and each
 calculator's own eligibility is run-scoped (PR #604). Re-passing an earlier
 `--run-id` resumes that run instead. No flag is required for a working run.
 
+**`--force-reextract` overrides the one not-run-scoped skip.** Stage C's
+manifest check (`already_done_ids`, the extractor-manifest `ImageEvidence`
+filter) is deliberately global — its whole point is that evidence already
+extracted with the current extractor versions needs no redo, on any run.
+`--force-reextract` clears that filter for the pass, so every eligible card
+is extracted fresh and its evidence row overwritten in place
+(`ImageEvidence` is keyed on `(card_id, content_hash)`). Use it for a full
+bulk re-extraction pass when you want every card's evidence rebuilt
+unconditionally, not merely cards an earlier run has not touched.
+
 ### Stage 0 — the same stage, one implementation
 
 Stage 0 is `stream_full_catalog`'s own freshness stage, whose body was lifted
@@ -1789,6 +1799,17 @@ The row's `counters` carry one key per stage — `stage_0` (including the bulk
 file vintage), `stage_c`, `stage_d`, `clustering`, `fidelity_gate`,
 `channel_report`, `elapsed_s`.
 
+**Every micro-batch's dispatch also gets its own UNIQUE ledger row.**
+`dispatch_micro_batch`'s `ledger_run_id` parameter (2026-07-31) decouples a
+micro-batch's ledger identity from the `run_id` its data is stamped with.
+The pipeline passes `ledger_run_id=<run_id>-<attempt timestamp>-b<batch num>`
+so a multi-batch pass under one `--run-id` survives: `PilotRunLedger.run_id`
+is UNIQUE, and handing every dispatch the same bare `run_id` used to raise an
+`IntegrityError` on the second micro-batch. Data rows keep the operator's
+clean `run_id` (channel_report scopes by the run_id on the rows); only each
+dispatch's own ledger row carries the suffixed id. Event-system dispatches
+pass `ledger_run_id=None` and are byte-identical to before.
+
 ### What a fresh run still inherits from an earlier one
 
 The from-scratch default covers **selection** — which cards get looked at.