diff --git a/MPCAutofill/cardpicker/management/commands/run_pipeline.py b/MPCAutofill/cardpicker/management/commands/run_pipeline.py index 61c6f8aa6..8c1250868 100644 --- a/MPCAutofill/cardpicker/management/commands/run_pipeline.py +++ b/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 ` 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 diff --git a/MPCAutofill/cardpicker/stage_e_dispatch.py b/MPCAutofill/cardpicker/stage_e_dispatch.py index 69672858f..89b068fb3 100644 --- a/MPCAutofill/cardpicker/stage_e_dispatch.py +++ b/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` + (`--b` 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, diff --git a/MPCAutofill/cardpicker/tests/test_run_pipeline.py b/MPCAutofill/cardpicker/tests/test_run_pipeline.py index 868a38a7a..66bdb635a 100644 --- a/MPCAutofill/cardpicker/tests/test_run_pipeline.py +++ b/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 diff --git a/docs/features/stage-e-operations.md b/docs/features/stage-e-operations.md index 12ecd8e23..68ccb66fe 100644 --- a/docs/features/stage-e-operations.md +++ b/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=--b` +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.