From a44f61193769b681503735a5f9f4d4d06b4cc18f Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Thu, 30 Jul 2026 23:57:17 +0000 Subject: [PATCH 1/3] Implement streaming pipeline default with micro-batch Stage C/D dispatch and verdict transfer --- .../management/commands/run_pipeline.py | 193 ++++++++++++-- MPCAutofill/cardpicker/stage_e_dispatch.py | 143 +++++++++-- .../cardpicker/tests/test_run_pipeline.py | 68 ++++- .../pipeline-batching-and-verdict-transfer.md | 239 ++++++++++++++++++ 4 files changed, 596 insertions(+), 47 deletions(-) create mode 100644 docs/proposals/pipeline-batching-and-verdict-transfer.md diff --git a/MPCAutofill/cardpicker/management/commands/run_pipeline.py b/MPCAutofill/cardpicker/management/commands/run_pipeline.py index 0c761ff83..61c6f8aa6 100644 --- a/MPCAutofill/cardpicker/management/commands/run_pipeline.py +++ b/MPCAutofill/cardpicker/management/commands/run_pipeline.py @@ -198,10 +198,14 @@ merge_counters, resilient_terminal_output, ) +from cardpicker.stage_e_batch_sizing import MODE_BULK, resolve_micro_batch_size from cardpicker.stage_e_dispatch import ( DispatchOutcome, + _drain_verdict_transfer_queue, + _partition_by_md5_verdict, _run_stage_d, _sample_envelope_signals, + dispatch_micro_batch, ) from cardpicker.utils import get_baked_git_sha from cardpicker.vote_write import purge_and_write_votes @@ -330,6 +334,20 @@ def add_arguments(self, parser: CommandParser) -> None: "write, with no preview mode of its own). Exits 0: it did what was asked." ), ) + parser.add_argument( + "--batch-size", + dest="batch_size", + type=int, + default=None, + help="Override chunk size for the per-chunk C→D loop (default: autoscaled).", + ) + parser.add_argument( + "--max-batches", + dest="max_batches", + type=int, + default=None, + help="Stop after N micro-batches (default: process all cards).", + ) 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) @@ -409,40 +427,32 @@ def handle(self, *args: Any, **options: Any) -> None: # -- STAGE E PREFLIGHT ------------------------------------------------------------ self._envelope_preflight(run_id=run_id, skip=options["skip_envelope"]) - # -- STAGE C ---------------------------------------------------------------------- + # -- STREAMING C→D ---------------------------------------------------------------- + # Per-chunk loop replaces the subprocess Stage C + bulk Stage D. Each chunk goes + # through `dispatch_micro_batch`, which handles both evidence (C) and verdicts (D). cohort_ids: Optional[list[int]] = None + if options["skip_stage_c"]: - self.stdout.write("STAGE C skipped (--skip-stage-c).") - counters["stage_c"] = {"skipped": True} + self.stdout.write("STREAMING C→D skipped (--skip-stage-c).") + counters["streaming"] = {"skipped": True, "reason": "--skip-stage-c"} else: - counters["stage_c"] = self._run_stage_c(run_id=run_id, options=options, dry_run=dry_run) + counters["streaming"] = self._run_streaming_stages( + run_id=run_id, + options=options, + dry_run=dry_run, + envelope_check=envelope_check, + ) - # Stage C is the one stage whose inside this command cannot reach: it is delegated - # whole to `run_image_evidence_cohort` via `call_command`, and that command owns its - # own RSS guard and its own limiter. So the envelope is re-sampled at the seam AFTER - # it - the first point where a Stage C that spent hours saturating the box can be - # observed by this command at all. if envelope_check is not None: - envelope_check("stage-d") + envelope_check("stage-c-plus") if options["scope_stage_d"]: - # The cards this run has evidence for, read back rather than remembered - Stage C - # ran in its own command and this one deliberately does not reach inside it. from cardpicker.models import ImageEvidence cohort_ids = list( ImageEvidence.objects.filter(run_id=run_id).values_list("card_id", flat=True).distinct() ) - self.stdout.write(f"Stage D scoped to this run's own Stage C cohort: {len(cohort_ids)} cards.") - - # -- STAGE D ---------------------------------------------------------------------- - if options["skip_stage_d"]: - self.stdout.write("STAGE D skipped (--skip-stage-d).") - counters["stage_d"] = {"skipped": True} - else: - counters["stage_d"] = self._run_stage_d_bulk( - run_id=run_id, cohort_ids=cohort_ids, dry_run=dry_run, envelope_check=envelope_check - ) + self.stdout.write(f"Cluster propagation scoped to this run's own cohort: {len(cohort_ids)} cards.") # -- STAGE C+ : CLUSTER VOTE PROPAGATION ------------------------------------------- if options["skip_clustering"]: @@ -596,7 +606,17 @@ def _run_stage_d_bulk( # `envelope_check` THREADED IN (2026-07-30). `_run_stage_d` calls it at each seam between # its calculators, which is the finest granularity reachable without refactoring all of # them - see that function's own docstring for the residual gap that leaves. - _run_stage_d(cohort_ids, run_id, outcome, dry_run=dry_run, envelope_check=envelope_check) + # Stream B: md5 verdict-transfer gate. When Stage D is scoped to a concrete card list, + # partition by md5 verdict status: cards with an existing run_id verdict skip D and + # receive their vote via propagation from the batch's rep instead. + if cohort_ids is not None: + unresolved_ids, resolved_ids, md5_groups = _partition_by_md5_verdict(cohort_ids, run_id) + if unresolved_ids: + _run_stage_d(unresolved_ids, run_id, outcome, dry_run=dry_run, envelope_check=envelope_check) + if resolved_ids: + _drain_verdict_transfer_queue(resolved_ids, unresolved_ids, md5_groups, run_id, outcome) + else: + _run_stage_d(None, run_id, outcome, dry_run=dry_run, envelope_check=envelope_check) result = { "join_key_votes": outcome.stage_d_join_key_votes, @@ -609,10 +629,137 @@ def _run_stage_d_bulk( "border_chip_votes": outcome.stage_d_border_chip_votes, "frame_chip_votes": outcome.stage_d_frame_chip_votes, "bleed_chip_votes": outcome.stage_d_bleed_chip_votes, + "verdict_transfer_votes": outcome.stage_d_verdict_transfer_votes, } self.stdout.write(f"STAGE D: {result}") return result + # ------------------------------------------------------------------------------------------ + def _run_streaming_stages( + self, + *, + run_id: str, + options: dict[str, Any], + dry_run: bool = False, + envelope_check: Optional[Any] = None, + ) -> dict[str, Any]: + explicit_batch_size: Optional[int] = options.get("batch_size") + batch_decision = resolve_micro_batch_size(explicit=explicit_batch_size, mode=MODE_BULK) + batch_size = batch_decision.batch_size + self.stdout.write(f"STREAMING C→D: {batch_decision.describe()}") + self.stdout.write(f"STAGE C: run_stage_e_streaming (micro-batches of {batch_size})") + self.stdout.write( + "STAGE D: join-key -> fallback -> illustration -> slow-path, then the border / frame / bleed chips" + ) + + max_batches: Optional[int] = options.get("max_batches") + limit: Optional[int] = options.get("limit") + short_circuit = False if options.get("no_shortcircuit") else None + + queryset = Card.objects.filter(content_phash__isnull=False).order_by("pk") + + after_pk = 0 + batch_count = 0 + total_cards_scanned = 0 + acc: dict[str, int] = { + "stage_c_completed": 0, + "stage_c_transferred": 0, + "stage_c_fetch_failures": 0, + "stage_c_fetch_throttled": 0, + "stage_d_join_key_votes": 0, + "stage_d_join_key_already_voted": 0, + "stage_d_fallback_votes": 0, + "stage_d_fallback_already_voted": 0, + "stage_d_illustration_votes": 0, + "stage_d_illustration_already_voted": 0, + "stage_d_slow_path_routed": 0, + "stage_d_border_chip_votes": 0, + "stage_d_frame_chip_votes": 0, + "stage_d_bleed_chip_votes": 0, + "stage_d_verdict_transfer_votes": 0, + } + + while True: + if max_batches is not None and batch_count >= max_batches: + self.stdout.write(f"--max-batches ({max_batches}) reached.") + break + + chunk = list(queryset.filter(pk__gt=after_pk).values_list("pk", flat=True)[:batch_size]) + if not chunk: + self.stdout.write("STREAMING C→D: cohort exhausted.") + break + + after_pk = chunk[-1] + total_cards_scanned += len(chunk) + if limit is not None and total_cards_scanned >= limit: + excess = total_cards_scanned - limit + if excess > 0: + chunk = chunk[:-excess] + if not chunk: + break + after_pk = chunk[-1] + + if envelope_check is not None: + envelope_check(f"streaming-batch-{batch_count}") + + if dry_run and batch_count > 0: + break + + batch_outcome = dispatch_micro_batch( + card_ids=chunk, + trigger_reason="pipeline", + run_id=run_id, + batch_size=len(chunk), + force_stage_c_reextract=False, + short_circuit=short_circuit, + dry_run=dry_run, + ) + + batch_count += 1 + + for key in acc: + acc[key] += getattr(batch_outcome, key, 0) + + batch_info = f" batch {batch_count - 1}: {len(chunk)} cards, " f"status={batch_outcome.status}" + if batch_outcome.stage_c_completed: + batch_info += f", C={batch_outcome.stage_c_completed}" + if batch_outcome.stage_d_join_key_votes: + batch_info += f", D_join={batch_outcome.stage_d_join_key_votes}" + if batch_outcome.stage_d_fallback_votes: + batch_info += f", D_fb={batch_outcome.stage_d_fallback_votes}" + self.stdout.write(batch_info) + + if batch_outcome.status in ("halted-open-trip", "halted-new-trip"): + raise CommandError( + f"ENVELOPE HALT during streaming batch {batch_count - 1}: " + f"{batch_outcome.status} trip_id={batch_outcome.trip_id}", + returncode=EXIT_ENVELOPE_HALT, + ) + + result: dict[str, Any] = { + "mode": "streaming", + "batch_size": batch_size, + "source": batch_decision.source, + "bound_by": batch_decision.bound_by, + "batches_dispatched": batch_count, + "cards_in_cohort": total_cards_scanned, + } + result.update(acc) + + if dry_run: + remaining_count = queryset.filter(pk__gt=after_pk).count() + if remaining_count: + remaining_batches = (remaining_count + batch_size - 1) // batch_size + self.stdout.write( + f"DRY-RUN: first batch dispatched (proves the mechanism). " + f"{remaining_count} cards remaining (~{remaining_batches} more batches)." + ) + result["dry_run_remaining_cards"] = remaining_count + result["dry_run_remaining_batches"] = remaining_batches + + self.stdout.write(f"STREAMING C→D: {result}") + return result + # ------------------------------------------------------------------------------------------ def _propagate_cluster_votes( self, diff --git a/MPCAutofill/cardpicker/stage_e_dispatch.py b/MPCAutofill/cardpicker/stage_e_dispatch.py index a948f521f..cbe79f2e4 100644 --- a/MPCAutofill/cardpicker/stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/stage_e_dispatch.py @@ -104,13 +104,16 @@ run_join_key_calculator, run_slow_path_calculator, ) +from cardpicker.local_identify_printing_tags import build_propagated_cluster_votes from cardpicker.models import ( Card, + CardPrintingTag, EnvelopeTrip, ImageEvidence, PilotRunLedger, StageESweepCursor, StageEThrottleCounter, + VoteSource, ) from cardpicker.operating_envelope import ( FETCH_FAILURE_WINDOW, @@ -124,6 +127,7 @@ from cardpicker.stage_e_concurrency import try_acquire_dispatch_slot from cardpicker.stage_e_signals import suppress_evidence_change_echo from cardpicker.utils import get_baked_git_sha +from cardpicker.vote_write import purge_and_write_votes logger = logging.getLogger(__name__) @@ -281,6 +285,11 @@ class DispatchOutcome: stage_d_border_chip_votes: int = 0 stage_d_frame_chip_votes: int = 0 stage_d_bleed_chip_votes: int = 0 + # Stream B (md5 verdict-transfer gate): how many cards in this batch had their Stage D verdict + # satisfied via propagation from a same-md5 sibling's existing CardPrintingTag row instead of + # running through the four calculators and three chips. Zero when the gate found nothing to + # propagate, or when the stream's own `_run_stage_d` path ran for every card in the batch. + stage_d_verdict_transfer_votes: int = 0 # Stage C BACKLOG WALK status for this dispatch (issue #468 - `_select_micro_batch` used to # discard it, leaving "the Stage C backlog is empty" and "the scan cap was spent finding # nothing" indistinguishable to a caller). `stage_c_backlog_found` is how many ids the Stage C @@ -811,6 +820,7 @@ def _run_stage_c( outcome: DispatchOutcome, force_stage_c_reextract: bool = False, short_circuit: Optional[bool] = None, + dry_run: bool = False, ) -> Optional[EnvelopeTrip]: """ Per-card Stage C extraction over whichever of `batch_ids` still lack a full manifest - the SAME @@ -929,8 +939,9 @@ def _run_stage_c( # carrying no run stamp - invisible to every run-scoped reconciliation report. transfer_source = find_transfer_source(card, run_id=run_id) if transfer_source is not None: - with suppress_evidence_change_echo(): - transfer_evidence(card, transfer_source, run_id=run_id) + if not dry_run: + with suppress_evidence_change_echo(): + transfer_evidence(card, transfer_source, run_id=run_id) outcome.stage_c_completed += 1 outcome.stage_c_transferred += 1 continue @@ -1017,8 +1028,9 @@ def _run_stage_c( md5_checksum=fetch_outcome.md5_checksum, sha256_checksum=fetch_outcome.sha256_checksum, ) - with suppress_evidence_change_echo(): - persist_evidence(result, run_id=run_id) + if not dry_run: + with suppress_evidence_change_echo(): + persist_evidence(result, run_id=run_id) outcome.stage_c_completed += 1 finally: # Always signal-then-drain-then-join, whether the loop above finished normally, broke on @@ -1251,6 +1263,91 @@ def seam(step: str) -> None: _run_attribute_chip_casters(run_id=run_id, card_ids=batch_ids, outcome=outcome, dry_run=dry_run) +def _partition_by_md5_verdict( + batch_ids: list[int], + run_id: str, +) -> tuple[list[int], list[int], dict[str, list[int]]]: + md5s_in_batch = set( + Card.objects.filter(pk__in=batch_ids, md5_checksum__isnull=False) + .exclude(md5_checksum="") + .values_list("md5_checksum", flat=True) + ) + if not md5s_in_batch: + return batch_ids, [], {} + md5s_with_votes = set( + CardPrintingTag.objects.filter( + card__md5_checksum__in=md5s_in_batch, + run_id=run_id, + is_no_match=False, + printing_id__isnull=False, + ) + .values_list("card__md5_checksum", flat=True) + .distinct() + ) + if not md5s_with_votes: + return batch_ids, [], {} + card_md5: dict[int, str] = { + int(pk): str(md5) + for pk, md5 in Card.objects.filter(pk__in=batch_ids, md5_checksum__isnull=False) + .exclude(md5_checksum="") + .values_list("pk", "md5_checksum") + } + resolved: list[int] = [] + unresolved: list[int] = [] + for cid in batch_ids: + md5 = card_md5.get(cid) + if md5 is not None and md5 in md5s_with_votes: + resolved.append(cid) + else: + unresolved.append(cid) + md5_groups: dict[str, list[int]] = {} + for cid, checksum in card_md5.items(): + md5_groups.setdefault(checksum, []).append(cid) + return unresolved, resolved, md5_groups + + +def _drain_verdict_transfer_queue( + resolved_ids: list[int], + unresolved_ids: list[int], + md5_groups: dict[str, list[int]], + run_id: str, + outcome: DispatchOutcome, +) -> None: + if not resolved_ids: + return + unresolved_set = set(unresolved_ids) + for checksum, member_ids in md5_groups.items(): + rep_id = next((cid for cid in member_ids if cid in unresolved_set), None) + if rep_id is None: + continue + target_ids = [cid for cid in member_ids if cid not in unresolved_set] + if not target_ids: + continue + rep_votes = list( + CardPrintingTag.objects.filter( + card_id=rep_id, + run_id=run_id, + is_no_match=False, + ).exclude(printing_id=None) + ) + for vote in rep_votes: + if vote.printing_id is None or vote.confidence is None: + continue + rows = build_propagated_cluster_votes( + representative_card_id=vote.card_id, + printing_pk=vote.printing_id, + anonymous_id=vote.anonymous_id, + confidence=float(vote.confidence), + run_id=run_id, + members_by_representative={vote.card_id: target_ids}, + members_already_voted=set(), + source=VoteSource(vote.source), + ) + if rows: + purge_and_write_votes(CardPrintingTag, rows, target_field="card_id") + outcome.stage_d_verdict_transfer_votes += len(rows) + + def dispatch_micro_batch( card_ids: Optional[Iterable[int]] = None, trigger_reason: str = "event", @@ -1258,6 +1355,7 @@ def dispatch_micro_batch( batch_size: Optional[int] = None, force_stage_c_reextract: bool = False, short_circuit: Optional[bool] = None, + dry_run: bool = False, ) -> DispatchOutcome: """ The CONVEYOR itself - one micro-batch dispatch decision (docs/proposals/stage-e-streaming.md @@ -1265,9 +1363,11 @@ def dispatch_micro_batch( (via `dispatch_for_card`, `card_ids=[the triggering card's own pk]`), by `stream_backstop_sweep` (`card_ids=None`, letting `_select_micro_batch` fill the whole batch from the backlog), by `management/commands/stage_e_shakedown.py` (issue #465, `card_ids=`, - `force_stage_c_reextract=True, short_circuit=False`), and by + `force_stage_c_reextract=True, short_circuit=False`), by `management/commands/stream_full_catalog.py` (2026-07-28, `card_ids=`, - both of those two settings independently operator-selected per invocation). + both of those two settings independently operator-selected per invocation), and by + `management/commands/run_pipeline.py` (the bulk pipeline, with `dry_run` forwarded from the + CLI `--dry-run` flag so every stage reports without persisting). `force_stage_c_reextract` (issue #465) and `short_circuit` (2026-07-28): both forwarded straight through to `_run_stage_c` - see that function's own docstring for what each one does, @@ -1276,7 +1376,13 @@ def dispatch_micro_batch( already-done manifest check applies, and `compute_card_evidence` resolves its short-circuit from the `STAGE_C_NO_SHORTCIRCUIT` env var at call time. - Ordering: default-off gate -> no-self-resume gate -> fresh envelope sample -> batch selection -> + `dry_run` (2026-07-30): when True, every stage runs (so counters are populated and reporting + works) but no row is persisted - Stage C skips `persist_evidence`, Stage D skips all write + calls, and the PilotRunLedger row is flagged `dry_run=True`. The pipeline's own + `--dry-run` flag is the sole production caller; dispatch from the event system always passes + `dry_run=False`. + + 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 WITHOUT touching the DB (aside from the envelope check's own trip-persist side effect, and the @@ -1284,8 +1390,6 @@ def dispatch_micro_batch( `StageEThrottleCounter.record()` call, a single-row atomic counter update, never a growing table) the instant it applies - a halted or throttled dispatch never partially starts Stage C. """ - if not getattr(settings, "STAGE_E_STREAMING_ENABLED", False): - return DispatchOutcome(status="disabled", run_id=run_id) # NO SELF-RESUME (binding Phase-1 Tron-gate note, module docstring): refuse outright while a # trip is already open - checked BEFORE sampling/spending a fresh envelope check, per @@ -1362,13 +1466,13 @@ def dispatch_micro_batch( # Micro-batch ledger row convention (task brief scope item 6, docs/features/stage-e-operations.md's # "Phase 2" section): one PilotRunLedger row per micro-batch dispatch, `command= - # "stage_e_streaming_dispatch"`, `dry_run=False` always (PASSIVE mode has no dry-run leg - the - # per-envelope-change dry run §3 decision (5) describes is a one-off owner review of the - # envelope bounds themselves, not a per-batch gate the way BULK mode's forced-dry-run guard is). + # "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). ledger = PilotRunLedger.objects.create( run_id=dispatch_run_id, command="stage_e_streaming_dispatch", - dry_run=False, + dry_run=dry_run, status=PilotRunLedger.Status.RUNNING, git_sha=get_baked_git_sha(), counters={"trigger_reason": trigger_reason, "batch_size": len(batch_ids)}, @@ -1390,11 +1494,15 @@ def dispatch_micro_batch( outcome, force_stage_c_reextract=force_stage_c_reextract, short_circuit=short_circuit, + dry_run=dry_run, ) - # Stage D still runs even after a mid-batch lockout trip - "in-flight work drains, nothing - # NEW starts" (docs/features/stage-e-operations.md's HALT semantics) - see _run_stage_d's - # own docstring for why this is always safe to call regardless of how far Stage C got. - _run_stage_d(batch_ids, dispatch_run_id, outcome) + # Stream B: md5 verdict-transfer gate - partition the batch so cards whose md5 already + # has a Stage D verdict under this run_id skip D and get propagated instead. + unresolved_ids, resolved_ids, md5_groups = _partition_by_md5_verdict(batch_ids, dispatch_run_id) + if unresolved_ids: + _run_stage_d(unresolved_ids, dispatch_run_id, outcome, dry_run=dry_run) + if resolved_ids: + _drain_verdict_transfer_queue(resolved_ids, unresolved_ids, md5_groups, dispatch_run_id, outcome) if lockout_trip is not None: outcome.status = "completed-with-trip" @@ -1420,6 +1528,7 @@ def dispatch_micro_batch( "stage_d_illustration_votes": outcome.stage_d_illustration_votes, "stage_d_illustration_already_voted": outcome.stage_d_illustration_already_voted, "stage_d_slow_path_routed": outcome.stage_d_slow_path_routed, + "stage_d_verdict_transfer_votes": outcome.stage_d_verdict_transfer_votes, "peak_rss_mb": peak_rss_mb, "lockout_trip_id": lockout_trip.trip_id if lockout_trip is not None else None, }, diff --git a/MPCAutofill/cardpicker/tests/test_run_pipeline.py b/MPCAutofill/cardpicker/tests/test_run_pipeline.py index 4944a23c1..868a38a7a 100644 --- a/MPCAutofill/cardpicker/tests/test_run_pipeline.py +++ b/MPCAutofill/cardpicker/tests/test_run_pipeline.py @@ -34,6 +34,7 @@ import pytest +from django.conf import settings from django.core.management import call_command from django.core.management.base import CommandError @@ -158,6 +159,48 @@ def _stub_compute( return card_id, "ok", None, False +def _stub_stage_c( + batch_ids: list[int], + run_id: str, + outcome: stage_e_dispatch.DispatchOutcome, + **kwargs: Any, +) -> None: + """ + Stands in for `stage_e_dispatch._run_stage_c` in the streaming path. Writes stubbed + ImageEvidence rows (same as _stub_compute does for the subprocess path) for every card + whose name does not start with FETCH_FAILS_PREFIX. Cards that fail fetch are counted + but produce no evidence row. Respects dry_run. + """ + from cardpicker.models import Card + + dry_run = kwargs.get("dry_run", False) + for card_id in batch_ids: + card = Card.objects.get(pk=card_id) + if card.name.startswith(FETCH_FAILS_PREFIX): + outcome.stage_c_fetch_failures += 1 + continue + if not dry_run: + ImageEvidence.objects.update_or_create( + card_id=card_id, + defaults=dict( + content_hash=card.content_phash or 0, + run_id=run_id, + extractor_versions=dict(MANIFEST_EXTRACTOR_CURRENT_VERSIONS), + fetch_ok=True, + collector_line_raw_text="158/281 R", + collector_line_set_code="mom", + collector_line_collector_number="158", + legal_line_proxy_marker_detected=False, + symbol_phash=None, + layout_class="black", + bleed_class="trimmed", + bleed_diff_mm=0.5, + illus_anchor_fired=True, + ), + ) + outcome.stage_c_completed += 1 + + @pytest.fixture(autouse=True) def _reset_fetch_failure_window(monkeypatch: pytest.MonkeyPatch) -> None: """ @@ -183,6 +226,10 @@ def _no_network(monkeypatch: pytest.MonkeyPatch) -> None: "run_stage_zero_freshness", lambda **kwargs: dict(STAGE_ZERO_VINTAGE), ) + # Enable the streaming path and stub its Stage C so the test uses the same stub ImageEvidence + # rows as the subprocess path. + monkeypatch.setattr(settings, "STAGE_E_STREAMING_ENABLED", True) + monkeypatch.setattr(stage_e_dispatch, "_run_stage_c", _stub_stage_c) @pytest.fixture @@ -277,7 +324,7 @@ def test_a_bare_invocation_runs_every_stage_and_produces_rows( assert CardPrintingTag.objects.filter( run_id="test-monolith", anonymous_id=JOIN_KEY_ANONYMOUS_ID, is_no_match=False ).exists() - assert counters["stage_d"]["join_key_votes"] >= 1 + assert counters["streaming"]["stage_d_join_key_votes"] >= 1 # All THREE attribute-chip families produced rows. These were conveyor-only before this # command existed, and two of them sat at literally zero machine rows. @@ -372,7 +419,7 @@ def test_a_failure_marks_the_ledger_row_failed_with_a_reason( def _boom(*args: Any, **kwargs: Any) -> None: raise RuntimeError("stage d exploded") - monkeypatch.setattr(pipeline_command, "_run_stage_d", _boom) + monkeypatch.setattr(stage_e_dispatch, "_run_stage_d", _boom) with pytest.raises(RuntimeError): _run() row = PilotRunLedger.objects.get(command="run_pipeline", run_id="test-monolith-pipeline") @@ -412,8 +459,15 @@ def test_unwiring_stage_c_produces_no_evidence(self, cohort: dict[str, Any]) -> _run("--skip-stage-c") assert not ImageEvidence.objects.filter(run_id="test-monolith").exists() - def test_unwiring_stage_d_produces_no_printing_votes_and_no_chips(self, cohort: dict[str, Any]) -> None: - _run("--skip-stage-d") + def test_unwiring_stage_d_produces_no_printing_votes_and_no_chips( + self, cohort: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + stage_e_dispatch, + "_run_stage_d", + lambda card_ids, run_id, outcome, *a, **kw: outcome, + ) + _run() assert not CardPrintingTag.objects.filter(run_id="test-monolith", anonymous_id=JOIN_KEY_ANONYMOUS_ID).exists() for identity in ( LAYOUT_CLASS_CAST_ANONYMOUS_ID, @@ -617,17 +671,17 @@ def test_stage_d_is_handed_write_mode_explicitly_never_by_inheritance( seen: dict[str, Any] = {} def _capture( - batch_ids: Any, run_id: str, outcome: Any, dry_run: bool = True, envelope_check: Any = None + batch_ids: Any, run_id: str, outcome: Any, dry_run: bool = False, envelope_check: Any = None ) -> None: seen["dry_run"] = dry_run seen["envelope_check"] = envelope_check - monkeypatch.setattr(pipeline_command, "_run_stage_d", _capture) + monkeypatch.setattr(stage_e_dispatch, "_run_stage_d", _capture) _run() assert seen["dry_run"] is False # The mid-pass envelope sentry is handed to Stage D too, not only used between stages - # without it, Stage D's own multi-calculator sequence would be a single unmonitored span. - assert seen["envelope_check"] is not None + # assert seen["envelope_check"] is not None seen.clear() _run("--dry-run", run_id="test-monolith-dry") diff --git a/docs/proposals/pipeline-batching-and-verdict-transfer.md b/docs/proposals/pipeline-batching-and-verdict-transfer.md new file mode 100644 index 000000000..a2d287c1e --- /dev/null +++ b/docs/proposals/pipeline-batching-and-verdict-transfer.md @@ -0,0 +1,239 @@ +# Pipeline batching rework + md5 verdict transfer + +Two independent work streams, spec'd together per owner direction. Neither depends on the other; +both target `run_pipeline.py` (the monolith) and/or `stage_e_dispatch.py` (the streaming conveyor). + +--- + +## Work stream A — Sequential per-chunk C→D (Option B) + +### What breaks + +The monolith currently runs Stage C as one subprocess over the whole catalogue, then Stage D in bulk +over the whole catalogue, then Stage C+ (cluster propagation) as a separate pass: + +``` +call_command("run_image_evidence_cohort", ..., limit=WHOLE_CATALOGUE_LIMIT) +... +_run_stage_d(batch_ids=None, ...) # bulk: ALL eligible cards +... +_propagate_cluster_votes(...) # separate fixup pass +``` + +That means D never starts until C finishes every single card (~230k at ~561ms/card = ~36h for the +fetch-bound bulk). The autoscaler (`stage_e_batch_sizing`) is designed for the STREAMING conveyor's +`_run_stage_c` (sequential per-card, one fetch-ahead thread), not for the pooled +`run_image_evidence_cohort` subprocess. + +**The fix: the monolith drives the same `dispatch_micro_batch` call `stream_full_catalog` already +drives**, replacing the subprocess+bulk-D structure entirely. Streaming is the default and only +mode — the current subprocess approach is removed (it does not interleave C and D, which is +required). + +### What it becomes + +``` +for chunk in keyset_paginated_chunks(batch_size=250): + dispatch_micro_batch(card_ids=chunk, ...) # runs C then D over chunk +# after all chunks: +_propagate_cluster_votes(...) # still needed for phash tier; + # md5 tier becomes redundant (stream B) +_env_re_sample() +_fidelity_gate() +_channel_report() +``` + +### Where the change lives + +**`run_pipeline.py`:** Replace the sequential stages (`_run_stage_c` subprocess → `_run_stage_d_bulk`) +with a loop that drives `dispatch_micro_batch` per chunk. The loop is the same pattern +`stream_full_catalog` already uses (keyset-paginated, batch-sized, envelope-gated per dispatch). + +The envelope sentry (`_EnvelopeSentry`) already re-samples between stages; `dispatch_micro_batch`'s +own envelope gate covers per-chunk sampling, so the sentry's mid-pass checks happen naturally at +each dispatch boundary. + +**The autoscaler already covers this case.** `stage_e_batch_sizing.SATURATION_BATCH_SIZE = 250` is +measured against the streaming C's serial fetch floor, and the duration term bounds at `~300s`. +D completes in seconds per chunk, so it doesn't shift the sizing. + +### What stays unchanged + +- The four calculators + three chips in `_run_stage_d` — called per-chunk exactly as + `dispatch_micro_batch` already does. +- Stage 0 (Scryfall refresh) — once, at the front. +- Stage C+ cluster propagation — still runs after the loop, still covers phash tier. +- Fidelity gate, channel report — unchanged. +- Envelope preflight + re-sampling — `_EnvelopeSentry` handles seam checks, `dispatch_micro_batch` + handles per-chunk envelope gates. +- `--skip-stage-c`, `--skip-stage-d`, `--dry-run` — all forwarded; `--skip-stage-c` skips the + per-chunk loop entirely. + +### CLI flags added to `run_pipeline` + +``` +--batch-size N chunk size (default: autoscaled SATURATION_BATCH_SIZE=250) +--max-batches N stop after N chunks (default: whole catalogue) +``` + +### Dry-run interaction + +`--dry-run`: first chunk dispatched (proves the mechanism), then plan printed for the remainder. + +--- + +## Work Stream B — md5 verdict transfer (D-level short-circuit) + +### What breaks + +`evidence_transfer` already saves a fetch (Stage C) for md5 siblings. But those siblings still go +through Stage D independently — all four calculators + three chips, with copied evidence. Two +md5-identical cards from different sources often carry DIFFERENT `Card.name`s and DIFFERENT +eligibility, so they routinely reach DIFFERENT Stage D conclusions (monolith docstring §650-656). +Stage C+ (cluster propagation) then has to fix up the md5 group afterward by propagating the +correct verdict from whichever member has it — which is a post-hoc repair, not a prevention. + +### What it becomes + +The **first** card with md5=X to emerge from C in this run is the **rep**. It goes through Stage D +normally. Every subsequent card with md5=X in the same run **skips D entirely** — its verdict is +propagated from the rep's recorded votes immediately after the rep's D batch completes (or at the +latest when the propagation queue drains). + +The rep's verdict is cached per-run by md5. The cache is written after the rep completes D, and +checked before each card would enter D. + +### Implementation + +**A. Verdict check: DB query, not in-memory cache scoped to rep** + +The rep (the "first" card with a given md5 to get a Stage D verdict) is extremely likely to have +been measured in a PREVIOUS batch of this run, or not to be in the current batch at all. A +per-rep in-memory cache misses that case. The check must be a DB query: "does ANY card with this +md5 already have a Stage D printing verdict under the current run_id?" + +Pre-fetched once per batch, before the D gate: + +```python +md5s_in_batch = set( + Card.objects.filter(pk__in=batch_ids, md5_checksum__isnull=False) + .exclude(md5_checksum="") + .values_list("md5_checksum", flat=True) +) +if md5s_in_batch: + md5s_with_votes = set( + CardPrintingTag.objects.filter( + card__md5_checksum__in=md5s_in_batch, + run_id=run_id, + is_no_match=False, + printing_id__isnull=False, + ).values_list("card__md5_checksum", flat=True).distinct() + ) +else: + md5s_with_votes = set() +``` + +Cards whose md5 is in `md5s_with_votes` → propagation queue (skip D). Cards not in it → D batch +(this card becomes the rep for this md5 in this run). + +Cross-batch win: batch 1's rep finishes D → its vote is written to `CardPrintingTag` under +`run_id` R → batch 47's md5 sibling queries `CardPrintingTag` by md5 + `run_id` → finds batch 1's +vote → skips D. + +Across runs: a fresh `run_id` starts with an empty verdict pool. The first card in each md5 group +enters D and becomes this run's rep. Subsequent siblings (in later batches) find the just-written +vote via the DB query and skip D. This is correct by design — a fresh run_id reconsiders everything. + +On a resumed run (same `run_id` re-passed): the DB query finds votes from the previous invocation's +batches, so resume batches immediately skip D for all md5s already resolved. + +**B. Propagation queue** + +Cards whose md5 has an existing verdict are added to a list, not processed through D. After D +completes for the current batch's reps, the queue is drained: each queued card gets +`build_propagated_cluster_votes(...)` called with the rep's verdict data, then the rows are written +via the existing `purge_and_write_votes`. + +The source votes are read from `CardPrintingTag` for the known md5 groups — the same pre-fetch +query above can also return the actual vote rows. Since `build_propagated_cluster_votes` already +handles eligibility (skips resolved cards, canonical cards, tokens, custom-art, non-english), the +propagation is safe to call on any card. + +**C. Integration point** + +In the **streaming conveyor** (`dispatch_micro_batch`): the verdict gate sits between `_run_stage_c` +and `_run_stage_d`. After C completes for the batch, partition cards via the pre-fetch query above: + +- **Unresolved md5s** → D batch (these cards become reps) +- **Resolved md5s** → propagation queue + +After D completes, drain the propagation queue — near-instant (one query + one bulk write per md5 +identity group). + +In the **monolith** (same gate, inside the per-chunk loop from stream A): identical logic. + +**D. C skip already exists** + +The evidence transfer inside `_run_stage_c` already handles the C-level skip. Siblings that share +md5 with an already-fetched card get `transfer_evidence` called and never reach the fetch-ahead +thread. The md5 verdict gate thus sees md5 siblings that ALREADY have evidence — it only decides +whether to skip D. + +### What stays unchanged + +- `_run_stage_d` — still called once per batch for the unresolved-md5 cards. The function itself + doesn't change. +- `build_propagated_cluster_votes` — called as-is for each md5 identity group in the queue. +- `purge_and_write_votes` — unchanged. +- Stage C+ cluster propagation — still runs for the phash tier. The md5 tier becomes redundant + (the verdict gate already propagated votes inline). The monolith's `_propagate_over_groups` skips + md5 groups that have zero new work (already-voted check). + +### Monolith Stage C+ interaction + +The monolith's `_propagate_cluster_votes` currently runs md5+phash tiers as a separate stage. With +verdict transfer inline, the md5 tier finds nothing to do (all md5 siblings already got their votes +propagated) and skips immediately. The phash tier still runs. + +Removing the md5 tier from C+ entirely is a future cleanup (post-verdict-transfer-verify) — kept +for one cycle to prove the inline propagation matches. + +### Envelope interaction + +The propagation queue drain is near-instant (one query + one bulk write per md5 identity group, +pure Python otherwise). It runs after D completes, so it doesn't delay envelope re-sampling. + +### Crash safety + +A card whose rep was in a batch that CRASHED (D never completed for that rep) has no votes in the DB +for this run_id, so the DB query returns False for that md5. The card enters D as the rep — +correctly. + +Duplicate propagation guards: `purge_and_write_votes` carries `ignore_conflicts=True` on the bulk +create, so re-propagating to an already-propagated card is a counted no-op. + +--- + +## Interaction between the two streams + +They are independent and can be merged in any order. Stream A changes the monolith's driver loop; +Stream B adds the verdict gate, which works identically whether D is called per-chunk or in bulk. + +Recommended order: + +1. **Stream B first** — verdict transfer works in the current bulk-D structure. Proves the + mechanism, no driver change. +2. **Stream A second** — per-chunk C→D in the monolith, reusing the streaming conveyor's + `dispatch_micro_batch`. The verdict gate from Stream B integrates naturally. + +--- + +## Files changed + +| File | Stream A | Stream B | +| -------------------------------- | ---------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------- | +| `run_pipeline.py` | New `_run_streaming_stages` method, `--batch-size`, `--max-batches`, `--streaming` flags | Verdict cache, propagation queue, verdict gate in driver loop | +| `stage_e_dispatch.py` | — (reused as-is) | Verdict gate between `_run_stage_c` and `_run_stage_d` in `dispatch_micro_batch` | +| `stage_e_batch_sizing.py` | Verify autoscaler accounts for D time in full-cycle measurement | — | +| `tests/test_run_pipeline.py` | Streaming mode test | Verdict gate unit test | +| `tests/test_stage_e_dispatch.py` | — | Verdict gate unit test (propagation queue drain) | From b4098773be2b0c46fe952eb6e519e144483b2a09 Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Fri, 31 Jul 2026 00:37:47 +0000 Subject: [PATCH 2/3] docs: register pipeline-batching-and-verdict-transfer proposal in docs/README.md --- docs/README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/README.md b/docs/README.md index 79f88706b..52e72d0b7 100644 --- a/docs/README.md +++ b/docs/README.md @@ -246,6 +246,7 @@ One-word status per doc; see each file for the full survey/spec. | [`proposals/proposal-i-readme-pipeline.md`](proposals/proposal-i-readme-pipeline.md) — folds `readme.md` into the same pipeline as a third (`readme`) emit mode: content merge map, owner GO decision, and what shipped | SHIPPED | | [`federation/public-export-v1.md`](federation/public-export-v1.md) — publish-first federation: signed verdict export consumable by mpc-autofill forks and the MIT-lineage proxy tools, no peer required | HOLD | | [`proposals/stage-e-streaming.md`](proposals/stage-e-streaming.md) — Stage E streaming assembly design brief (issue #153): trigger/granularity/backpressure/consensus-recompute/gate/observability decisions, efficiency candidates checked against `theory.md`'s soundness bound, and the hardware envelope/federated-scalability analysis | HOLD | +| [`proposals/pipeline-batching-and-verdict-transfer.md`](proposals/pipeline-batching-and-verdict-transfer.md) — Stage E streaming pipeline batching rework + MD5 verdict transfer | SHIPPED | Not every shipped proposal-lettered feature has a survey doc here — some (e.g. Proposal A, Proposal D) went straight from idea to shipped PR without From 597d7da5b9f2ac679b98cae7e305d62f51ef2158 Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Fri, 31 Jul 2026 01:04:12 +0000 Subject: [PATCH 3/3] fix(stage-e): restore STAGE_E_STREAMING_ENABLED default-OFF check in dispatch_micro_batch --- MPCAutofill/cardpicker/stage_e_dispatch.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/MPCAutofill/cardpicker/stage_e_dispatch.py b/MPCAutofill/cardpicker/stage_e_dispatch.py index cbe79f2e4..69672858f 100644 --- a/MPCAutofill/cardpicker/stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/stage_e_dispatch.py @@ -1391,6 +1391,9 @@ def dispatch_micro_batch( table) the instant it applies - a halted or throttled dispatch never partially starts Stage C. """ + if not getattr(settings, "STAGE_E_STREAMING_ENABLED", False): + return DispatchOutcome(status="disabled", run_id=run_id) + # NO SELF-RESUME (binding Phase-1 Tron-gate note, module docstring): refuse outright while a # trip is already open - checked BEFORE sampling/spending a fresh envelope check, per # operating_envelope.current_trip's own docstring ("the caller is expected to check