From c46ea5087efa23cc7e660a3850084aecb0210919 Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Fri, 24 Jul 2026 17:40:51 +0000 Subject: [PATCH] Add Stage E Phase 2 streaming dispatch loop (issue #153) Builds the conveyor Phase 1's envelope primitive had no caller for: an event-driven card-create/evidence-change trigger plus a cron backstop sweep, micro-batched dispatch that calls the existing Stage C extraction and Stage D calculator entry points as-is (scoped via a new optional card_ids parameter), and a per-batch PilotRunLedger convention. Ships default-off behind settings.STAGE_E_STREAMING_ENABLED; BULK-mode commands are unaffected. Also fixes a pre-existing mypy type-arg gap in operating_envelope.py that was blocking this commit's pre-commit hook. Co-Authored-By: Claude Fable 5 --- MPCAutofill/MPCAutofill/settings.py | 19 + MPCAutofill/cardpicker/apps.py | 9 + .../cardpicker/local_calculate_verdicts.py | 63 ++- .../commands/stream_backstop_sweep.py | 151 ++++++ MPCAutofill/cardpicker/operating_envelope.py | 4 +- MPCAutofill/cardpicker/stage_e_dispatch.py | 466 +++++++++++++++++ MPCAutofill/cardpicker/stage_e_signals.py | 63 +++ .../cardpicker/tests/test_stage_e_dispatch.py | 467 ++++++++++++++++++ docs/README.md | 8 +- docs/features/stage-e-operations.md | 187 ++++++- docs/proposals/stage-e-streaming.md | 19 + 11 files changed, 1413 insertions(+), 43 deletions(-) create mode 100644 MPCAutofill/cardpicker/management/commands/stream_backstop_sweep.py create mode 100644 MPCAutofill/cardpicker/stage_e_dispatch.py create mode 100644 MPCAutofill/cardpicker/stage_e_signals.py create mode 100644 MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py diff --git a/MPCAutofill/MPCAutofill/settings.py b/MPCAutofill/MPCAutofill/settings.py index d00da745e..c329941b4 100755 --- a/MPCAutofill/MPCAutofill/settings.py +++ b/MPCAutofill/MPCAutofill/settings.py @@ -424,3 +424,22 @@ "label": "Django Q2", "orm": "default", } + +# Stage E Phase 2 - streaming dispatch loop (docs/proposals/stage-e-streaming.md §3, docs/features/ +# stage-e-operations.md's "Phase 2" section). Default OFF, exactly matching this phase's own +# NOT-IN-SCOPE line ("actually enabling the trigger in prod settings ... turning it on is the +# phase-3 shakedown's polled owner action") - the event-driven card-create/evidence-change signal +# receivers (cardpicker/stage_e_signals.py) and the cron backstop sweep +# (management/commands/stream_backstop_sweep.py) are both wired unconditionally, but every one of +# them checks this flag first and is a no-op while it's False. Flipping it to True is the ONLY +# action phase 3 needs to take to go live - no redeploy of this module required. +STAGE_E_STREAMING_ENABLED = env.bool("STAGE_E_STREAMING_ENABLED", default=False) + +# Micro-batch size (docs/proposals/stage-e-streaming.md §3 decision (2), sharpened by §10(c)): NOT +# a value this brief or this change invents precision for - §10(c) ratifies that the real number +# ships as a MEASURED OUTPUT of the Bug-A tail shakedown's own instrumentation (phase 3, not yet +# run). This default is a placeholder sized to the brief's own "roughly 10-100 cards per batch" +# sanity range (§3 decision (2)) - a mid-range, conservative starting point pending that +# measurement, not a considered answer. Tunable without a code change (env var) precisely so the +# shakedown can adjust it without a redeploy. +STAGE_E_MICRO_BATCH_SIZE = env.int("STAGE_E_MICRO_BATCH_SIZE", default=25) diff --git a/MPCAutofill/cardpicker/apps.py b/MPCAutofill/cardpicker/apps.py index b69e38d4b..48dbc13e2 100644 --- a/MPCAutofill/cardpicker/apps.py +++ b/MPCAutofill/cardpicker/apps.py @@ -3,3 +3,12 @@ class CardpickerConfig(AppConfig): name = "cardpicker" + + def ready(self) -> None: + # Stage E Phase 2 (docs/proposals/stage-e-streaming.md §3 decision (1)) - registers the + # card-create/evidence-change post_save receivers (cardpicker/stage_e_signals.py). + # Connecting a signal receiver is cheap and side-effect-free by itself; each receiver is + # its own no-op while settings.STAGE_E_STREAMING_ENABLED is False (default), so importing + # this module here has no observable effect until that flag flips - see that module's own + # docstring. + from cardpicker import stage_e_signals # noqa: F401 diff --git a/MPCAutofill/cardpicker/local_calculate_verdicts.py b/MPCAutofill/cardpicker/local_calculate_verdicts.py index 0c1fd6ab5..841dd9d07 100644 --- a/MPCAutofill/cardpicker/local_calculate_verdicts.py +++ b/MPCAutofill/cardpicker/local_calculate_verdicts.py @@ -373,7 +373,7 @@ import logging from dataclasses import dataclass, field from pathlib import Path -from typing import Optional +from typing import Iterable, Optional import imagehash @@ -875,7 +875,9 @@ class JoinKeyCalculatorResult: def _eligible_cards_queryset( - anonymous_id: str, rescannable_skip_reasons: frozenset[str] = JOIN_KEY_RESCANNABLE_SKIP_REASONS + anonymous_id: str, + rescannable_skip_reasons: frozenset[str] = JOIN_KEY_RESCANNABLE_SKIP_REASONS, + card_ids: Optional[Iterable[int]] = None, ) -> "QuerySet[Card]": """ Mirrors `local_identify_printing_tags._eligible_base_queryset`'s shape (unresolved, no @@ -889,6 +891,17 @@ def _eligible_cards_queryset( instead, since the two calculators' own skip vocabularies mean different things by the same "transient, re-selectable" concept. + `card_ids` (2026-07-24, docs/proposals/stage-e-streaming.md §3 decision (4)/§10(c), Stage E + Phase 2): an OPTIONAL additional `.filter(pk__in=card_ids)` - purely a scope narrowing, exactly + the same shape `run_image_evidence_cohort.py`'s own pre-existing `--card-ids-file` flag already + applies to Stage C, extended here to Stage D. `None` (the default) is a no-op - every existing + BULK-mode caller (`run_join_key_calculator`/`run_fallback_calculator`/`run_slow_path_calculator` + with no `card_ids` argument, exactly as every one of today's management commands calls them) + sees byte-identical behaviour to before this parameter existed. Never changes WHICH cards among + those given are accepted or how - only narrows the pool a caller (Stage E's own streaming + micro-batch dispatcher, `cardpicker.stage_e_dispatch`) considers at all, so this is pure + dispatch-side scoping, not a change to any decode/accept-reject rule. + Idempotence for a repeated multi-pass Stage D fire comes entirely from the stable, per- calculator `anonymous_id` exclusion above (`.exclude(printing_tags__anonymous_id=anonymous_id)`) - deliberately the ONLY vote-population exclusion here. An earlier draft of this module also @@ -925,7 +938,7 @@ def _eligible_cards_queryset( .exclude(skip_reason__in=rescannable_skip_reasons) .values_list("card_id", flat=True) ) - return ( + queryset = ( Card.objects.filter( printing_tag_status=PrintingTagStatus.UNRESOLVED, canonical_card__isnull=True, @@ -939,6 +952,9 @@ def _eligible_cards_queryset( .distinct() .select_related("source") ) + if card_ids is not None: + queryset = queryset.filter(pk__in=card_ids) + return queryset def run_join_key_calculator( @@ -947,6 +963,7 @@ def run_join_key_calculator( chunk_size: int = 500, audit_sample_size: int = 20, default_cards_path: Optional[Path] = None, + card_ids: Optional[Iterable[int]] = None, ) -> JoinKeyCalculatorResult: """ Batch runner over every currently-eligible card with a CURRENT `ImageEvidence` row (its @@ -958,6 +975,10 @@ def run_join_key_calculator( straight through to `_resolve_candidates_for_card`'s own `is_back_face` call - `None` (the default, used in production) resolves to the real on-disk Scryfall cache; only ever overridden by a test. + + `card_ids` (2026-07-24, Stage E Phase 2 - see `_eligible_cards_queryset`'s own docstring for + the full rationale): forwarded straight through as a pure scope narrowing. `None` (the + default) is every existing caller's own behaviour, unchanged. """ run_id = run_id or generate_run_id() index = CandidateNameIndex() @@ -969,7 +990,7 @@ def run_join_key_calculator( scan_log_batch: list[CardScanLog] = [] touched_card_ids: list[int] = [] - for card in _eligible_cards_queryset(JOIN_KEY_ANONYMOUS_ID).iterator(chunk_size=chunk_size): + for card in _eligible_cards_queryset(JOIN_KEY_ANONYMOUS_ID, card_ids=card_ids).iterator(chunk_size=chunk_size): if card.content_phash is None: continue # no stable hash yet to key a CURRENT ImageEvidence lookup against @@ -1215,7 +1236,7 @@ class FallbackCalculatorResult: audit: list[dict[str, object]] = field(default_factory=list) -def _fallback_eligible_cards_queryset() -> "QuerySet[Card]": +def _fallback_eligible_cards_queryset(card_ids: Optional[Iterable[int]] = None) -> "QuerySet[Card]": """ Cards the join-key calculator already concluded have no confident hit - the SAME population `_slow_path_eligible_cards_queryset` below selects from (a real `is_no_match` vote, or a @@ -1223,6 +1244,9 @@ def _fallback_eligible_cards_queryset() -> "QuerySet[Card]": `STAGE_D_FALLBACK_ANONYMOUS_ID` hasn't already processed (scanned OR voted), via the shared `_eligible_cards_queryset` helper (idempotence mechanism only - see that function's own docstring for why a deduction-vote exclusion was considered and deliberately not added). + + `card_ids` (2026-07-24, Stage E Phase 2): forwarded straight through to + `_eligible_cards_queryset` - see that function's own docstring for the full rationale. """ join_key_no_match_card_ids = CardPrintingTag.objects.filter( anonymous_id=JOIN_KEY_ANONYMOUS_ID, is_no_match=True @@ -1231,7 +1255,7 @@ def _fallback_eligible_cards_queryset() -> "QuerySet[Card]": anonymous_id=JOIN_KEY_ANONYMOUS_ID, skip_reason__in=JOIN_KEY_NO_HIT_SKIP_REASONS ).values_list("card_id", flat=True) return _eligible_cards_queryset( - STAGE_D_FALLBACK_ANONYMOUS_ID, rescannable_skip_reasons=FALLBACK_RESCANNABLE_SKIP_REASONS + STAGE_D_FALLBACK_ANONYMOUS_ID, rescannable_skip_reasons=FALLBACK_RESCANNABLE_SKIP_REASONS, card_ids=card_ids ).filter(Q(pk__in=join_key_no_match_card_ids) | Q(pk__in=join_key_no_hit_scanned_card_ids)) @@ -1241,6 +1265,7 @@ def run_fallback_calculator( chunk_size: int = 500, audit_sample_size: int = 20, default_cards_path: Optional[Path] = None, + card_ids: Optional[Iterable[int]] = None, ) -> FallbackCalculatorResult: """ Batch runner for PIECE 1 (module docstring) - mirrors `run_join_key_calculator`'s own shape @@ -1251,7 +1276,8 @@ def run_fallback_calculator( Stage D's own "pass 2", the same relationship `local_fallback.py`'s own module docstring documents between the pilot's pass 1 (OCR/phash) and pass 2 (fallback). `default_cards_path` is threaded through to `_resolve_candidates_for_card` exactly as `run_join_key_calculator`'s own - parameter is. + parameter is. `card_ids` (2026-07-24, Stage E Phase 2) is forwarded straight through to + `_fallback_eligible_cards_queryset` - see `_eligible_cards_queryset`'s own docstring. """ run_id = run_id or generate_run_id() index = CandidateNameIndex() @@ -1261,7 +1287,7 @@ def run_fallback_calculator( scan_log_batch: list[CardScanLog] = [] touched_card_ids: list[int] = [] - for card in _fallback_eligible_cards_queryset().iterator(chunk_size=chunk_size): + for card in _fallback_eligible_cards_queryset(card_ids=card_ids).iterator(chunk_size=chunk_size): if card.content_phash is None: continue # no stable hash yet to key a CURRENT ImageEvidence lookup against @@ -1458,7 +1484,7 @@ class SlowPathCalculatorResult: audit: list[dict[str, object]] = field(default_factory=list) -def _slow_path_eligible_cards_queryset() -> "QuerySet[Card]": +def _slow_path_eligible_cards_queryset(card_ids: Optional[Iterable[int]] = None) -> "QuerySet[Card]": """ Cards the join-key calculator (JOIN_KEY_ANONYMOUS_ID) already concluded have no confident hit - either a real `is_no_match` vote, or a non-rescannable skip in @@ -1477,6 +1503,9 @@ def _slow_path_eligible_cards_queryset() -> "QuerySet[Card]": (`no-evidence-types-used`/`eliminated`/`ambiguous`) is deliberately NOT excluded here - it still has no confident automated hit from either calculator and belongs in the review queue exactly as before this PR. + + `card_ids` (2026-07-24, Stage E Phase 2): a pure scope narrowing, same convention as + `_eligible_cards_queryset`'s own `card_ids` parameter - see that function's own docstring. """ join_key_no_match_card_ids = CardPrintingTag.objects.filter( anonymous_id=JOIN_KEY_ANONYMOUS_ID, is_no_match=True @@ -1490,7 +1519,7 @@ def _slow_path_eligible_cards_queryset() -> "QuerySet[Card]": fallback_voted_card_ids = CardPrintingTag.objects.filter( anonymous_id=STAGE_D_FALLBACK_ANONYMOUS_ID, is_no_match=False ).values_list("card_id", flat=True) - return ( + queryset = ( Card.objects.filter( printing_tag_status=PrintingTagStatus.UNRESOLVED, canonical_card__isnull=True, @@ -1502,10 +1531,17 @@ def _slow_path_eligible_cards_queryset() -> "QuerySet[Card]": .distinct() .select_related("source") ) + if card_ids is not None: + queryset = queryset.filter(pk__in=card_ids) + return queryset def run_slow_path_calculator( - run_id: Optional[str] = None, dry_run: bool = True, chunk_size: int = 500, audit_sample_size: int = 20 + run_id: Optional[str] = None, + dry_run: bool = True, + chunk_size: int = 500, + audit_sample_size: int = 20, + card_ids: Optional[Iterable[int]] = None, ) -> SlowPathCalculatorResult: """ Batch runner over every card the join-key calculator already routed to no-hit (see @@ -1515,14 +1551,15 @@ def run_slow_path_calculator( a reviewer) and writes a `CardScanLog(anonymous_id=SLOW_PATH_ANONYMOUS_ID, skip_reason=SLOW_PATH_TO_REVIEW_REASON)` durable routing marker. `dry_run=True` (the default, matching `run_join_key_calculator`'s own convention) computes and counts everything without - writing. + writing. `card_ids` (2026-07-24, Stage E Phase 2) is forwarded straight through to + `_slow_path_eligible_cards_queryset` - see `_eligible_cards_queryset`'s own docstring. """ run_id = run_id or generate_run_id() result = SlowPathCalculatorResult(dry_run=dry_run, run_id=run_id) scan_log_batch: list[CardScanLog] = [] - for card in _slow_path_eligible_cards_queryset().iterator(chunk_size=chunk_size): + for card in _slow_path_eligible_cards_queryset(card_ids=card_ids).iterator(chunk_size=chunk_size): if card.content_phash is None: continue # no stable hash yet to key a CURRENT ImageEvidence lookup against diff --git a/MPCAutofill/cardpicker/management/commands/stream_backstop_sweep.py b/MPCAutofill/cardpicker/management/commands/stream_backstop_sweep.py new file mode 100644 index 000000000..020c8db2e --- /dev/null +++ b/MPCAutofill/cardpicker/management/commands/stream_backstop_sweep.py @@ -0,0 +1,151 @@ +""" +Stage E Phase 2 - the cron backstop sweep (docs/proposals/stage-e-streaming.md §3 decision (1)'s +"low-frequency cron sweep as a correctness backstop, not the primary path"). Re-runs the SAME +eligibility selectors the event-driven trigger (`cardpicker/stage_e_signals.py`) and the conveyor's +own backlog-fill (`cardpicker/stage_e_dispatch.py`'s `_select_micro_batch`) already use, catching +anything a lost/never-fired django-q dispatch missed - django-q2's own delivery guarantee is +at-least-once-ATTEMPTED, not exactly-once-DELIVERED (§3 decision (1)'s own reasoning), and this +project has no audited "no dispatch was ever silently dropped" property. + +DEFAULT-OFF, same gate as every other Phase 2 entry point (`settings.STAGE_E_STREAMING_ENABLED`) - +this command exits immediately, doing nothing, whenever that flag is False, matching the "ships the +mechanism, never turns it on" posture the whole of Phase 2 follows (see `stage_e_dispatch.py`'s own +module docstring). Not scheduled anywhere by this change either - a django-q `Schedule` row that +actually runs this on a cadence is a live-DB write this change deliberately does not make (NOT IN +SCOPE per the phase-2 task brief: "actually enabling the trigger in prod settings... turning it on +is the phase-3 shakedown's polled owner action"). + +Drives repeated `dispatch_micro_batch(card_ids=None, ...)` calls - `card_ids=None` lets +`_select_micro_batch` fill each batch entirely from the backlog, no seed card. Two backlogs, tried +in order per batch: (a) the Stage C backlog (cards with no CURRENT full-manifest `ImageEvidence` row +- catches a lost card-create dispatch), (b) once (a) is empty for a given batch, the Stage D +join-key-eligible backlog (cards with current evidence that have never received a join-key vote OR +scan-log row - catches a lost evidence-change dispatch; `_select_micro_batch` itself deliberately +does not fill from this backlog, see its own docstring, so the sweep covers it here instead). Stops +when both backlogs come back empty, when the envelope trips ("halted-new-trip"/"halted-open-trip" - +the sweep does not retry past a halt; the next scheduled sweep invocation picks up where this one +stopped, exactly like a re-invoked BULK command would), or when `--max-batches` is reached (a safety +bound for a single invocation, not a design limit). + +IDEMPOTENCE: every batch this command dispatches goes through the exact same `dispatch_micro_batch` +conveyor the event trigger uses - Stage C's own resume filter and Stage D's own +anonymous_id-exclusion eligibility queries are what make a RE-RUN of this same command produce zero +additional writes once the backlog is genuinely exhausted (the conveyor's own idempotence, not a new +mechanism this command adds). +""" + +from typing import Any + +from django.conf import settings +from django.core.management.base import BaseCommand, CommandParser +from django.utils import timezone + +from cardpicker.local_calculate_verdicts import JOIN_KEY_ANONYMOUS_ID +from cardpicker.local_calculate_verdicts import ( + _eligible_cards_queryset as _stage_d_eligible_cards_queryset, +) +from cardpicker.stage_e_dispatch import DEFAULT_MICRO_BATCH_SIZE, dispatch_micro_batch + +DEFAULT_MAX_BATCHES = 1000 + +_HALT_STATUSES = ("halted-open-trip", "halted-new-trip") + + +def _next_stage_d_backlog_ids(batch_size: int) -> list[int]: + """ + The Stage-D-only backlog `_select_micro_batch` (`stage_e_dispatch.py`) deliberately does NOT + fill from (see that function's own docstring) - cards whose Stage C evidence is already + complete but that have never had a join-key pass at all. Reuses + `local_calculate_verdicts._eligible_cards_queryset` UNSCOPED (`card_ids=None`) - the exact same + pool `run_join_key_calculator`'s own BULK-mode invocation would consider - sliced to + `batch_size`, never materializing the whole backlog. A module-private helper reused here rather + than duplicated, the same "reuse, never re-derive" convention `cardpicker/tests/ + test_local_calculate_verdicts.py` already establishes for testing it directly. + """ + return list( + _stage_d_eligible_cards_queryset(JOIN_KEY_ANONYMOUS_ID).order_by("pk").values_list("pk", flat=True)[:batch_size] + ) + + +class Command(BaseCommand): + help = ( + "Stage E Phase 2 cron backstop sweep (docs/proposals/stage-e-streaming.md §3 decision (1)) " + "- a correctness backstop for the event-driven trigger, not the primary dispatch path. " + "No-op unless settings.STAGE_E_STREAMING_ENABLED is True. See this command's own module " + "docstring and docs/features/stage-e-operations.md's 'Phase 2' section." + ) + + def add_arguments(self, parser: CommandParser) -> None: + parser.add_argument( + "--max-batches", + type=int, + default=DEFAULT_MAX_BATCHES, + help=f"Safety bound on how many micro-batches one invocation will dispatch before " + f"exiting, even if the backlog isn't exhausted yet (default {DEFAULT_MAX_BATCHES}).", + ) + parser.add_argument( + "--batch-size", + type=int, + default=None, + help="Override settings.STAGE_E_MICRO_BATCH_SIZE for this invocation only.", + ) + + def handle(self, *args: Any, **options: Any) -> None: + if not getattr(settings, "STAGE_E_STREAMING_ENABLED", False): + self.stdout.write("STAGE_E_STREAMING_ENABLED is False - backstop sweep is a no-op.") + return + + max_batches: int = options["max_batches"] + batch_size: int = options["batch_size"] or getattr( + settings, "STAGE_E_MICRO_BATCH_SIZE", DEFAULT_MICRO_BATCH_SIZE + ) + run_id_prefix = f"stage-e-backstop-{timezone.now().strftime('%Y%m%dT%H%M%SZ')}" + + batches_dispatched = 0 + total_stage_c = 0 + total_stage_d_votes = 0 + halted_status = None + + for batch_num in range(max_batches): + outcome = dispatch_micro_batch( + card_ids=None, + trigger_reason="backstop-sweep", + run_id=f"{run_id_prefix}-{batch_num}", + batch_size=batch_size, + ) + if outcome.status in _HALT_STATUSES: + halted_status = outcome.status + self.stdout.write(f"Envelope halt ({outcome.status}, trip_id={outcome.trip_id}) - stopping sweep.") + break + + if outcome.status == "empty": + # Backlog (a) exhausted for this pass - try backlog (b) before concluding the whole + # sweep is done (module docstring). + stage_d_backlog_ids = _next_stage_d_backlog_ids(batch_size) + if not stage_d_backlog_ids: + self.stdout.write("Backlog exhausted - nothing left to dispatch.") + break + outcome = dispatch_micro_batch( + card_ids=stage_d_backlog_ids, + trigger_reason="backstop-sweep-stage-d", + run_id=f"{run_id_prefix}-{batch_num}-d", + batch_size=batch_size, + ) + if outcome.status in _HALT_STATUSES: + halted_status = outcome.status + self.stdout.write(f"Envelope halt ({outcome.status}, trip_id={outcome.trip_id}) - stopping sweep.") + break + if outcome.status == "empty": + self.stdout.write("Backlog exhausted - nothing left to dispatch.") + break + + batches_dispatched += 1 + total_stage_c += outcome.stage_c_completed + total_stage_d_votes += ( + outcome.stage_d_join_key_votes + outcome.stage_d_fallback_votes + outcome.stage_d_slow_path_routed + ) + + self.stdout.write( + f"DONE batches_dispatched={batches_dispatched} stage_c_completed={total_stage_c} " + f"stage_d_votes_or_routes={total_stage_d_votes} halted={halted_status}" + ) diff --git a/MPCAutofill/cardpicker/operating_envelope.py b/MPCAutofill/cardpicker/operating_envelope.py index 414caaeff..a889802e7 100644 --- a/MPCAutofill/cardpicker/operating_envelope.py +++ b/MPCAutofill/cardpicker/operating_envelope.py @@ -42,7 +42,7 @@ """ from dataclasses import dataclass -from typing import Optional +from typing import Any, Optional from django.db.models import Q from django.utils import timezone @@ -79,7 +79,7 @@ class EnvelopeSignals: google_lockout: bool = False -def _bar_breach(signals: EnvelopeSignals) -> Optional[tuple[str, dict]]: +def _bar_breach(signals: EnvelopeSignals) -> Optional[tuple[str, dict[str, Any]]]: """ Returns `(bar, detail)` for the FIRST bar breached, checked in the priority order the brief itself implies (§10(a) lists the instant lockout pause distinctly from the other three's diff --git a/MPCAutofill/cardpicker/stage_e_dispatch.py b/MPCAutofill/cardpicker/stage_e_dispatch.py new file mode 100644 index 000000000..804e5155c --- /dev/null +++ b/MPCAutofill/cardpicker/stage_e_dispatch.py @@ -0,0 +1,466 @@ +""" +Stage E Phase 2 - the streaming dispatch loop (docs/proposals/stage-e-streaming.md, GitHub issue +#153; docs/features/stage-e-operations.md's "Phase 2" section is the operator-facing runbook this +module implements). Phase 1 (docs/proposals/stage-e-streaming.md's own header, PR #440, +`cardpicker/operating_envelope.py`) built the envelope PRIMITIVE with no caller - this module is +that caller: the CONVEYOR a card travels through once an event (card-create, evidence-change, +`cardpicker.stage_e_signals`) or the cron backstop sweep (`management/commands/ +stream_backstop_sweep.py`) names it eligible. + +SCOPE, per the owner-approved Phase 2 task brief: this module is the DISPATCH LOOP only - it NEVER +reimplements Stage C extraction, Stage D calculator decode logic, or consensus resolution. Every +actual decision (does this OCR read match a candidate, does this vote clear the human-backed gate) +still happens inside `cardpicker.image_evidence`/`cardpicker.local_calculate_verdicts`/ +`cardpicker.printing_consensus` exactly as it does for BULK mode - this module only decides WHEN +and on WHICH cards to call those existing entry points, and records what happened. BULK-mode +commands (`run_image_evidence_cohort`, `local_calculate_verdicts`, `reparse_collector_evidence`, +`consensus_recompute`, etc.) are untouched and keep working exactly as before: none of their own +call sites pass the new `card_ids` scoping parameter `local_calculate_verdicts.py` gained for this +module's benefit (see that module's own docstring on `_eligible_cards_queryset`'s `card_ids` +parameter, `None` by default = unchanged behaviour), so BULK mode's own behaviour is byte-identical +to before this change. + +DEFAULT-OFF (NOT IN SCOPE for this phase, per the task brief): `settings.STAGE_E_STREAMING_ENABLED` +gates every entry point below - `dispatch_micro_batch` is a no-op whenever it's False, and so is +every event trigger built on top of it (`cardpicker.stage_e_signals`) and the backstop sweep. Ships +default False (`MPCAutofill/settings.py`). Flipping it to True is the phase-3 shakedown's own +polled owner action - this change ships the mechanism, never turns it on. + +NO SELF-RESUME (binding Tron-gate note from Phase 1's review, restated here since this is the first +caller that actually enforces it): `dispatch_micro_batch` checks `operating_envelope.current_trip()` +BEFORE doing any work and refuses to dispatch, full stop, whenever it returns non-None - no code +path in this module ever calls `acknowledge_trip` or otherwise clears a trip. Resume is +`resolve_envelope_trip`'s own management command, always a fresh, explicit owner action (see +docs/features/stage-e-operations.md's runbook) - never automatic, never from inside this module. + +FETCH-FAILURE WINDOW SIZING (the second binding Tron-gate note): the rolling window this module +samples `fetch_failures_in_window`/`fetch_total_in_window` from is sized to +`operating_envelope.FETCH_FAILURE_WINDOW` (500) exactly - `check_envelope` computes its rate on +whatever it's handed, so getting this deque's `maxlen` right is entirely this module's own +responsibility, not that primitive's. See `_FetchOutcomeWindow` below. Process-local (module-level +singleton `_window`, one per worker process) - a multi-worker streaming deployment aggregating this +window across processes is a phase-3 operational concern, not a Phase 2 design gap: +`operating_envelope.EnvelopeSignals`'s own docstring already documents the caller as owning +windowing, with no cross-process aggregation promised anywhere in the ratified design (§10(a) sizes +the window, it doesn't mandate a shared store). + +PIPELINE STAGES, in order, per micro-batch (task brief scope item 5): Stage C extraction +(`cardpicker.image_evidence.compute_card_evidence`/`persist_evidence`, called per-card, +SEQUENTIALLY - fed by `cardpicker.image_cdn_fetch.fetch_card_image_bytes`) -> Stage D calculators +(`cardpicker.local_calculate_verdicts.run_join_key_calculator`/`run_fallback_calculator`/ +`run_slow_path_calculator`, called AS-IS with the new `card_ids` scope, in the same join-key -> +fallback -> slow-path escalation order every BULK-mode command already uses) -> ledger write. +Sequential, not pooled, on purpose: PASSIVE mode's own micro-batches (§3 decision (2), a handful to +a few dozen cards) are far too small for BULK mode's process-pool concurrency to buy anything - it +would only add a fork's worth of startup overhead per batch. This matches the brief's own "a +single-worker, single-core floor mode must be correct, just slow, never a degraded/unsound mode" +requirement (§5). + +CONSENSUS RECOMPUTE (decision (4)) NEEDS NO SEPARATE STEP HERE: all three Stage D calculators +already call `resolve_and_persist_printing(touched_card)` internally for every card they cast a +vote on (see e.g. `run_join_key_calculator`'s own final loop, unchanged by this module) - scoping +those calculators to the micro-batch via `card_ids` already scopes their consensus recompute calls +to exactly the same set, satisfying decision (4)'s "scoped incremental per-touch" requirement for +free. This module never imports `printing_consensus`/`vote_consensus`/`tag_consensus`/ +`artist_consensus` (PROTECTED CORE) directly at all. +""" + +import logging +import os +import time +from collections import deque +from dataclasses import dataclass, field +from typing import Deque, Iterable, Optional + +from django.conf import settings +from django.utils import timezone + +from cardpicker.harvest_fetch_limiter import GoogleFetchLockoutError +from cardpicker.local_calculate_verdicts import ( + known_set_codes, + run_fallback_calculator, + run_join_key_calculator, + run_slow_path_calculator, +) +from cardpicker.models import Card, EnvelopeTrip, ImageEvidence, PilotRunLedger +from cardpicker.operating_envelope import ( + FETCH_FAILURE_WINDOW, + EnvelopeSignals, + check_envelope, + current_trip, +) +from cardpicker.pilot_run_lifecycle import mark_ledger_failed, merge_counters +from cardpicker.process_metrics import get_process_rss_mb +from cardpicker.utils import get_baked_git_sha + +logger = logging.getLogger(__name__) + +# Placeholder pending §10(c)'s own measurement (see MPCAutofill/settings.py's own +# STAGE_E_MICRO_BATCH_SIZE comment for the full citation) - not invented precision, a +# conservative default inside the brief's own "roughly 10-100" sanity range. +DEFAULT_MICRO_BATCH_SIZE = 25 + + +def _stage_c_manifest_extractor_keys() -> "frozenset[str]": + """ + Lazy import (this module's own "avoid a hard import-time dependency between sibling engines" + posture, mirrored from `local_calculate_verdicts.py`'s own `JOIN_KEY_CONFIDENCE_BOTH` comment) - + a management-command module isn't normally imported from a library module at Django app-startup + time (this module is imported from `cardpicker.stage_e_signals`, wired in `apps.py`'s `ready()`), + so this stays call-time-only rather than a module-level import. `MANIFEST_EXTRACTOR_KEYS` itself + is untouched by this change - imported, never duplicated, so the two eligibility notions (BULK + mode's own resume filter, this module's own backlog fill) can never drift apart silently. + """ + from cardpicker.management.commands.run_image_evidence_cohort import ( + MANIFEST_EXTRACTOR_KEYS, + ) + + return MANIFEST_EXTRACTOR_KEYS + + +class _FetchOutcomeWindow: + """ + The rolling fetch-outcome window `dispatch_micro_batch` samples + `fetch_failures_in_window`/`fetch_total_in_window` from before every envelope check - sized to + `operating_envelope.FETCH_FAILURE_WINDOW` (500) exactly, per the binding Phase-1 Tron-gate note + (module docstring's "FETCH-FAILURE WINDOW SIZING" section). A `deque(maxlen=...)` is the + mechanism that actually enforces the size: once 500 outcomes have been recorded, the 501st push + silently evicts the oldest, so `len(self._window)` can never exceed `FETCH_FAILURE_WINDOW` + regardless of how many cards this worker process has ever touched. + """ + + def __init__(self, maxlen: int = FETCH_FAILURE_WINDOW) -> None: + self._window: Deque[bool] = deque(maxlen=maxlen) + + def record(self, success: bool) -> None: + self._window.append(success) + + def failures_and_total(self) -> tuple[int, int]: + total = len(self._window) + failures = sum(1 for success in self._window if not success) + return failures, total + + def __len__(self) -> int: + return len(self._window) + + +# Process-local singleton (module docstring's "FETCH-FAILURE WINDOW SIZING" section) - one per +# worker process, spanning that process's whole uptime, not reset per batch. +_window = _FetchOutcomeWindow() + + +@dataclass +class DispatchOutcome: + """ + What `dispatch_micro_batch` returns - never raises for an ordinary halt (streaming-disabled, + trip-open, freshly-tripped) since none of those are failures of the dispatch loop itself, only + reasons it correctly declined to do work this call. `status` is one of: + - "disabled" - `settings.STAGE_E_STREAMING_ENABLED` is False. + - "halted-open-trip" - `current_trip()` was already non-None; no self-resume (module docstring). + - "halted-new-trip" - this call's own fresh envelope sample breached a bar. + - "empty" - streaming is enabled and the envelope is clear, but nothing was eligible. + - "completed" - did real work; does not itself guarantee zero failures inside the batch + (a card can still fail its own fetch/extraction), only that the DISPATCH LOOP didn't halt. + - "completed-with-trip" - did real work, but a `GoogleFetchLockoutError` observed mid-batch + tripped the envelope (instant-pause bar) partway through - this batch's own already-fetched + work still drains (ops doc's "in-flight work drains, nothing NEW starts"), but the NEXT + `dispatch_micro_batch` call will see `current_trip()` non-None and refuse. + """ + + status: str + run_id: Optional[str] = None + card_ids: list[int] = field(default_factory=list) + stage_c_completed: int = 0 + stage_c_fetch_failures: int = 0 + stage_d_join_key_votes: int = 0 + stage_d_fallback_votes: int = 0 + stage_d_slow_path_routed: int = 0 + trip_id: Optional[str] = None + + +def _sample_envelope_signals(google_lockout: bool = False) -> EnvelopeSignals: + """ + Live signals sampled fresh before every dispatch decision (`operating_envelope.py`'s own module + docstring: "the caller owns sampling"). `load_avg`/`rss_mb_per_worker` are best-effort - `None` + on a platform without `/proc`/`os.getloadavg` (matches `get_process_rss_mb`'s own documented + convention: a caller must treat `None` as "skip this bar", never as an error). + """ + try: + load_avg: Optional[float] = os.getloadavg()[0] + except (OSError, AttributeError): + load_avg = None + failures, total = _window.failures_and_total() + return EnvelopeSignals( + load_avg=load_avg, + rss_mb_per_worker=get_process_rss_mb(), + fetch_failures_in_window=failures, + fetch_total_in_window=total, + google_lockout=google_lockout, + ) + + +def _select_micro_batch(seed_card_ids: Iterable[int], batch_size: int) -> list[int]: + """ + Builds one micro-batch's own card-id list (docs/proposals/stage-e-streaming.md §3 decision (2)): + starts with `seed_card_ids` (the event trigger's own touched card, or an empty seed for the + backstop sweep) and fills up to `batch_size` from the general Stage C backlog - cards with a + stable content hash but no CURRENT `ImageEvidence` row carrying every manifest extractor key + (the SAME shape `run_image_evidence_cohort.py`'s own resume filter uses, imported not + reimplemented - see `_stage_c_manifest_extractor_keys`). Order preserved (seed first), + de-duplicated. Bounded reads only (`[:batch_size]`/`[:remaining]` slices, never a full-table + materialization) - the whole point of a micro-batch is a bounded-cost dispatch (§3 decision (2)'s + own "one batch's wall-clock cost stays in the few-seconds-to-low-tens-of-seconds range"). + + Deliberately does NOT also backfill from the Stage-D-only backlog (cards whose Stage C evidence + is already complete but that have never had a Stage D pass) - the seed card itself always gets a + Stage D attempt regardless (`dispatch_micro_batch` scopes Stage D to the WHOLE returned batch, + seed included), and Stage C is the dominant wall-clock cost driver `batch_size` is sized against + (§3 decision (2)/§1's own worst-case floor), so backlog-filling from Stage C's own queue is the + lever that matters for keeping a batch's wall-clock cost bounded. + """ + seen: list[int] = [] + seen_set: set[int] = set() + for card_id in seed_card_ids: + if card_id not in seen_set: + seen.append(card_id) + seen_set.add(card_id) + if len(seen) >= batch_size: + return seen[:batch_size] + + remaining = batch_size - len(seen) + manifest_keys = list(_stage_c_manifest_extractor_keys()) + fully_processed_ids = ImageEvidence.objects.filter(extractor_versions__has_keys=manifest_keys).values_list( + "card_id", flat=True + ) + backlog_ids = ( + Card.objects.filter(content_phash__isnull=False) + .exclude(pk__in=seen_set) + .exclude(pk__in=fully_processed_ids) + .order_by("pk") + .values_list("pk", flat=True)[:remaining] + ) + for card_id in backlog_ids: + if card_id not in seen_set: + seen.append(card_id) + seen_set.add(card_id) + return seen[:batch_size] + + +def _run_stage_c(batch_ids: list[int], run_id: str, outcome: DispatchOutcome) -> Optional[EnvelopeTrip]: + """ + Sequential, per-card Stage C extraction over whichever of `batch_ids` still lack a full + manifest - the SAME per-card unit (`image_evidence.compute_card_evidence` + + `image_evidence.persist_evidence`, fed by `image_cdn_fetch.fetch_card_image_bytes`) + `run_image_evidence_cohort.py`'s own fetch/compute stages call, just driven one card at a time + (module docstring's own "PIPELINE STAGES" section explains why). Every fetch outcome is recorded + onto `_window` regardless of whether it ends up mattering to THIS batch's own envelope decision - + the window spans the whole worker process's uptime, not one batch. + + Returns the `EnvelopeTrip` this call itself recorded (only possible via the instant Google + lockout bar - see `GoogleFetchLockoutError` below), or `None`. A lockout stops Stage C + IMMEDIATELY for this batch - in-flight work already committed stays committed (each card's + `persist_evidence` call is already durable the instant it returns, matching the resume + contract's own "one-transaction batch commit or explicit evidence-first statement" - here, every + card's own persist is its own transaction, so there is no partial-card state to roll back) - and + records a fresh trip via `check_envelope(google_lockout=True)` so the NEXT dispatch call refuses + until an owner acknowledges it, matching the "instant pause" bar exactly. + """ + from io import BytesIO + + from PIL import Image + + from cardpicker.image_cdn_fetch import DEFAULT_FETCH_DPI, fetch_card_image_bytes + from cardpicker.image_evidence import compute_card_evidence, persist_evidence + + manifest_keys = list(_stage_c_manifest_extractor_keys()) + already_done_ids = set( + ImageEvidence.objects.filter(card_id__in=batch_ids, extractor_versions__has_keys=manifest_keys).values_list( + "card_id", flat=True + ) + ) + lexicon = known_set_codes() + + for card_id in batch_ids: + if card_id in already_done_ids: + continue + try: + card = Card.objects.select_related("source").get(pk=card_id) + except Card.DoesNotExist: + continue + if card.content_phash is None: + continue + + fetch_started_at = time.monotonic() + try: + image_bytes = fetch_card_image_bytes(card, dpi=DEFAULT_FETCH_DPI) + except GoogleFetchLockoutError: + _window.record(success=False) + logger.error("Stage E dispatch: GoogleFetchLockoutError observed - halting Stage C for this batch") + return check_envelope(_sample_envelope_signals(google_lockout=True), run_id=run_id) + fetch_latency_ms = (time.monotonic() - fetch_started_at) * 1000 + + if image_bytes is None: + _window.record(success=False) + outcome.stage_c_fetch_failures += 1 + continue + + _window.record(success=True) + image = Image.open(BytesIO(image_bytes)) + result = compute_card_evidence( + card_id, card.content_phash, image, fetch_latency_ms=fetch_latency_ms, known_set_codes=lexicon + ) + persist_evidence(result, run_id=run_id) + outcome.stage_c_completed += 1 + + return None + + +def _run_stage_d(batch_ids: list[int], run_id: str, outcome: DispatchOutcome) -> None: + """ + Stage D over the SAME micro-batch, scoped via the `card_ids` parameter + `local_calculate_verdicts.py` gained for this module (see that module's own docstring) - the + join-key -> fallback -> slow-path escalation order every BULK-mode command already uses, + unchanged (module docstring's "PIPELINE STAGES" section explains the consensus-recompute + piece). Runs unconditionally for every card in `batch_ids`, including ones Stage C never + reached this round (e.g. a lockout stopped Stage C partway, or the card already had current + evidence and never needed Stage C at all this dispatch) - each calculator's own eligibility + query simply finds nothing to do for a card with no current evidence (a "no-evidence" named + skip, not an error), so this is always safe to call. + """ + join_key_result = run_join_key_calculator(run_id=run_id, dry_run=False, card_ids=batch_ids) + outcome.stage_d_join_key_votes = join_key_result.votes_written + join_key_result.no_match_votes_written + + fallback_result = run_fallback_calculator(run_id=run_id, dry_run=False, card_ids=batch_ids) + outcome.stage_d_fallback_votes = fallback_result.votes_written + + slow_path_result = run_slow_path_calculator(run_id=run_id, dry_run=False, card_ids=batch_ids) + outcome.stage_d_slow_path_routed = slow_path_result.routed_written + + +def dispatch_micro_batch( + card_ids: Optional[Iterable[int]] = None, + trigger_reason: str = "event", + run_id: Optional[str] = None, + batch_size: Optional[int] = None, +) -> DispatchOutcome: + """ + The CONVEYOR itself - one micro-batch dispatch decision (docs/proposals/stage-e-streaming.md + §3, this module's own docstring). Called by `cardpicker.stage_e_signals`'s own event receivers + (via `dispatch_for_card`, `card_ids=[the triggering card's own pk]`) and by + `stream_backstop_sweep` (`card_ids=None`, letting `_select_micro_batch` fill the whole batch + from the backlog). + + Ordering: default-off gate -> no-self-resume gate -> fresh envelope sample -> Stage C + (sequential, per-card) -> Stage D (AS-IS entry points, scoped) -> ledger write. Every gate below + returns WITHOUT touching the DB (aside from the envelope check's own trip-persist side effect) + the instant it applies - a halted 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 + # current_trip() BEFORE ever calling [check_envelope]"). + existing_trip = current_trip(run_id=run_id) + if existing_trip is not None: + logger.info( + "Stage E dispatch refused - envelope trip %s (%s) is still open, no self-resume", + existing_trip.trip_id, + existing_trip.bar, + ) + return DispatchOutcome(status="halted-open-trip", run_id=run_id, trip_id=existing_trip.trip_id) + + signals = _sample_envelope_signals() + fresh_trip = check_envelope(signals, run_id=run_id) + if fresh_trip is not None: + logger.warning( + "Stage E dispatch halted - envelope bar %s breached (%s), trip %s persisted", + fresh_trip.bar, + fresh_trip.detail, + fresh_trip.trip_id, + ) + return DispatchOutcome(status="halted-new-trip", run_id=run_id, trip_id=fresh_trip.trip_id) + + effective_batch_size = ( + batch_size + if batch_size is not None + else getattr(settings, "STAGE_E_MICRO_BATCH_SIZE", DEFAULT_MICRO_BATCH_SIZE) + ) + batch_ids = _select_micro_batch(card_ids or (), effective_batch_size) + if not batch_ids: + return DispatchOutcome(status="empty", run_id=run_id) + + dispatch_run_id = run_id or f"stage-e-stream-{timezone.now().strftime('%Y%m%dT%H%M%S%f')}Z" + + # 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). + ledger = PilotRunLedger.objects.create( + run_id=dispatch_run_id, + command="stage_e_streaming_dispatch", + dry_run=False, + status=PilotRunLedger.Status.RUNNING, + git_sha=get_baked_git_sha(), + counters={"trigger_reason": trigger_reason, "batch_size": len(batch_ids)}, + ) + + outcome = DispatchOutcome(status="completed", run_id=dispatch_run_id, card_ids=batch_ids) + batch_start = time.monotonic() + + try: + lockout_trip = _run_stage_c(batch_ids, dispatch_run_id, outcome) + # 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) + + if lockout_trip is not None: + outcome.status = "completed-with-trip" + outcome.trip_id = lockout_trip.trip_id + + peak_rss_mb = get_process_rss_mb() + ledger.status = PilotRunLedger.Status.COMPLETED + ledger.finished_at = timezone.now() + ledger.counters = merge_counters( + ledger.counters, + { + "elapsed_s": round(time.monotonic() - batch_start, 3), + "stage_c_completed": outcome.stage_c_completed, + "stage_c_fetch_failures": outcome.stage_c_fetch_failures, + "stage_d_join_key_votes": outcome.stage_d_join_key_votes, + "stage_d_fallback_votes": outcome.stage_d_fallback_votes, + "stage_d_slow_path_routed": outcome.stage_d_slow_path_routed, + "peak_rss_mb": peak_rss_mb, + "lockout_trip_id": lockout_trip.trip_id if lockout_trip is not None else None, + }, + ) + ledger.save(update_fields=["status", "finished_at", "counters"]) + except Exception as exc: + # Shared FAILED-transition rail (cardpicker.pilot_run_lifecycle.mark_ledger_failed) - a + # no-op if this invocation already reached the COMPLETED save above, otherwise records a + # triage-able counters["failure_reason"] alongside FAILED (docs/proposals/ + # stage-e-streaming.md §3 decision (6)'s "empty-failed-row" gap fix, reused here rather than + # duplicated). A crash mid-Stage-C-loop leaves every already-`persist_evidence`-committed + # card durably written (each card's own persist is its own transaction) - the resume + # contract (docs/features/stage-e-operations.md) holds: a fresh dispatch over the same or an + # overlapping card set skips whatever's already current and picks up the rest, exactly the + # same "truthful ledger, idempotent re-entry" property the batch kill-test already proves. + mark_ledger_failed(ledger, exc) + raise + + return outcome + + +def dispatch_for_card(card_id: int, reason: str = "event") -> None: + """ + The django-q `async_task` entry point (`cardpicker.stage_e_signals`'s own event receivers, + docs/proposals/stage-e-streaming.md §3 decision (1)) - a thin wrapper around + `dispatch_micro_batch` scoping the seed to exactly the one card that triggered this task. A + bare module-level function (not a closure/lambda): `async_task` needs a string dotted path it + can re-import inside the worker process (`"cardpicker.stage_e_dispatch.dispatch_for_card"` - + see `cardpicker.stage_e_signals` for the exact call site). + """ + dispatch_micro_batch(card_ids=[card_id], trigger_reason=reason) diff --git a/MPCAutofill/cardpicker/stage_e_signals.py b/MPCAutofill/cardpicker/stage_e_signals.py new file mode 100644 index 000000000..558df1929 --- /dev/null +++ b/MPCAutofill/cardpicker/stage_e_signals.py @@ -0,0 +1,63 @@ +""" +Stage E Phase 2 - the event-driven trigger half of docs/proposals/stage-e-streaming.md §3 +decision (1) ("event-driven ... dispatched on card-create and on evidence-change, with a +low-frequency cron sweep as a correctness backstop, not the primary path"). Two `post_save` +receivers, wired unconditionally in `cardpicker.apps.CardpickerConfig.ready()` (connecting a +Django signal receiver is cheap and side-effect-free by itself) but each a no-op whenever +`settings.STAGE_E_STREAMING_ENABLED` is False (`MPCAutofill/settings.py`'s own docstring) - so this +module ships DEFAULT-OFF exactly like the rest of Phase 2, with no redeploy needed to turn it on. + +Both receivers dispatch via `django_q.tasks.async_task`, never inline - a `post_save` handler +running Stage C/D synchronously inside the same request/transaction that just created the +`Card`/`ImageEvidence` row would (a) block whatever view/command triggered the save on a +network-fetch-plus-OCR-cost pipeline stage, and (b) risk seeing the just-committed row before its +own transaction has actually committed if the save happened inside a wider atomic block (a real +risk `local_calculate_verdicts.py`'s own commands avoid by never running inline off a signal at +all). `async_task` queues the work onto django-q2's existing worker pool (`Q_CLUSTER`, already +provisioned in this project - see `settings.py`) instead. + +CARD-CREATE: fires once, only on `created=True` - never on an ordinary field-update save (matches +decision (1)'s own "card-create" framing exactly; a re-save of an existing card is not a new-card +event). + +EVIDENCE-CHANGE: fires on every `ImageEvidence` save, created or updated - `dispatch_for_card`'s own +downstream Stage C step is naturally idempotent (its own resume filter skips a card whose evidence +is already current, see `stage_e_dispatch._run_stage_c`), and Stage D's own eligibility queries +already exclude a card once it's carrying a vote from a given calculator's own `anonymous_id` - so +a burst of `ImageEvidence` saves for the same card (e.g. one extractor group's write, then +another's, both landing on the SAME row within one Stage C pass) triggers several dispatch calls +that mostly resolve to fast, cheap no-ops rather than repeated real work. This is the SAME +"evidence-change event re-opens a card to re-scan, never an elapsed-time trigger" contract issue +#278's own selector already specifies (docs/proposals/stage-e-streaming.md §4 item 4) - deliberately +generic here (every evidence-change fires an attempt, not just #278's own specific detector), since +this module only decides WHETHER to attempt a dispatch, never what any downstream engine does with +it. +""" + +from typing import Any + +from django.conf import settings +from django.db.models.signals import post_save +from django.dispatch import receiver + +from cardpicker.models import Card, ImageEvidence + + +@receiver(post_save, sender=Card) +def _dispatch_on_card_create(sender: Any, instance: Card, created: bool, **kwargs: Any) -> None: + if not created: + return + if not getattr(settings, "STAGE_E_STREAMING_ENABLED", False): + return + from django_q.tasks import async_task + + async_task("cardpicker.stage_e_dispatch.dispatch_for_card", instance.pk, "card-create") + + +@receiver(post_save, sender=ImageEvidence) +def _dispatch_on_evidence_change(sender: Any, instance: ImageEvidence, **kwargs: Any) -> None: + if not getattr(settings, "STAGE_E_STREAMING_ENABLED", False): + return + from django_q.tasks import async_task + + async_task("cardpicker.stage_e_dispatch.dispatch_for_card", instance.card_id, "evidence-change") diff --git a/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py b/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py new file mode 100644 index 000000000..68df9f610 --- /dev/null +++ b/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py @@ -0,0 +1,467 @@ +""" +Tests for cardpicker.stage_e_dispatch - Stage E Phase 2's streaming dispatch loop +(docs/proposals/stage-e-streaming.md, docs/features/stage-e-operations.md's "Phase 2" section). + +No network calls, no live image fetch - `fetch_card_image_bytes`/`compute_card_evidence` are +monkeypatched at their SOURCE module (`cardpicker.image_cdn_fetch`/`cardpicker.image_evidence`), +never at `cardpicker.stage_e_dispatch` itself, because `_run_stage_c` imports both lazily (inside +the function body, module docstring's own "avoid a hard import-time dependency" convention) - a +patch applied to the source module before the call is what a fresh `from ... import ...` inside the +function body actually observes. `persist_evidence` itself is left REAL (unmocked) in every test +below - it's a small, already-tested, non-network function, and exercising it for real is what +proves ImageEvidence rows actually land, matching `_evidence()`'s own convention in +`test_local_calculate_verdicts.py`. +""" + +import io +from typing import Any + +import pytest +from PIL import Image + +from django.core.management import call_command +from django.test import override_settings + +from cardpicker import stage_e_dispatch +from cardpicker.harvest_fetch_limiter import GoogleFetchLockoutError +from cardpicker.image_evidence import ExtractionResult +from cardpicker.local_calculate_verdicts import JOIN_KEY_ANONYMOUS_ID +from cardpicker.management.commands.run_image_evidence_cohort import ( + MANIFEST_EXTRACTOR_KEYS, +) +from cardpicker.models import ( + CardPrintingTag, + EnvelopeTrip, + ImageEvidence, + PilotRunLedger, + PrintingTagStatus, +) +from cardpicker.operating_envelope import ( + FETCH_FAILURE_WINDOW, + acknowledge_trip, + check_envelope, + current_trip, +) +from cardpicker.stage_e_dispatch import ( + _FetchOutcomeWindow, + _select_micro_batch, + dispatch_for_card, + dispatch_micro_batch, +) +from cardpicker.tests.factories import ( + CanonicalCardFactory, + CardFactory, + ImageEvidenceFactory, +) + +STREAMING_ON = override_settings(STAGE_E_STREAMING_ENABLED=True) + + +@pytest.fixture(autouse=True) +def _reset_fetch_failure_window(monkeypatch: pytest.MonkeyPatch) -> None: + """The rolling fetch-outcome window (`stage_e_dispatch._window`) is a process-local module + singleton spanning a worker process's whole uptime by design (module docstring) - reset it + before every test in this file so no test observes another's fetch outcomes.""" + monkeypatch.setattr(stage_e_dispatch, "_window", _FetchOutcomeWindow()) + + +def _png_bytes() -> bytes: + """A tiny, genuinely-decodable PNG - `_run_stage_c` calls the real `PIL.Image.open` on + whatever `fetch_card_image_bytes` returns before handing it to `compute_card_evidence` (which + is separately stubbed below), so this needs to be real image bytes, not an arbitrary literal.""" + buffer = io.BytesIO() + Image.new("RGB", (10, 10)).save(buffer, format="PNG") + return buffer.getvalue() + + +def _stub_compute_card_evidence_ok(**field_overrides: Any): + """Builds a stand-in for `cardpicker.image_evidence.compute_card_evidence` that returns an + `ExtractionResult` carrying every Stage C manifest key (so the resume filter treats the card as + fully processed after one pass) plus whatever join-key-relevant fields the caller wants to + steer Stage D's own verdict.""" + + def _stub( + card_id: int, content_hash, image, fetch_latency_ms=0.0, profile=None, short_circuit=None, known_set_codes=None + ): + fields = { + "fetch_ok": True, + "collector_line_raw_text": "", + "collector_line_set_code": "", + "collector_line_collector_number": "", + "legal_line_proxy_marker_detected": False, + "symbol_phash": None, + } + fields.update(field_overrides) + return ExtractionResult( + card_id=card_id, + content_hash=content_hash, + fields=fields, + extractor_versions={key: f"{key}-v1" for key in MANIFEST_EXTRACTOR_KEYS}, + ) + + return _stub + + +def _install_stage_c_stub(monkeypatch: pytest.MonkeyPatch, fetch_result: Any = b"", **field_overrides: Any) -> None: + import cardpicker.image_cdn_fetch as image_cdn_fetch_module + import cardpicker.image_evidence as image_evidence_module + + if callable(fetch_result): + monkeypatch.setattr(image_cdn_fetch_module, "fetch_card_image_bytes", fetch_result) + else: + monkeypatch.setattr(image_cdn_fetch_module, "fetch_card_image_bytes", lambda card, dpi=None: fetch_result) + monkeypatch.setattr( + image_evidence_module, "compute_card_evidence", _stub_compute_card_evidence_ok(**field_overrides) + ) + + +class TestDefaultOff: + def test_disabled_by_default_returns_disabled_status(self, db: Any) -> None: + outcome = dispatch_micro_batch(card_ids=[1]) + assert outcome.status == "disabled" + assert PilotRunLedger.objects.count() == 0 + assert EnvelopeTrip.objects.count() == 0 + + def test_dispatch_for_card_is_a_silent_no_op_when_disabled(self, db: Any) -> None: + card = CardFactory(content_phash=42) + dispatch_for_card(card.pk, "card-create") + assert PilotRunLedger.objects.count() == 0 + + def test_backstop_sweep_is_a_no_op_when_disabled(self, db: Any, capsys: pytest.CaptureFixture) -> None: + call_command("stream_backstop_sweep") + assert PilotRunLedger.objects.count() == 0 + assert "no-op" in capsys.readouterr().out + + +class TestEnvelopeTripHaltsAndNoSelfResume: + @STREAMING_ON + def test_a_breached_envelope_halts_before_any_work_and_records_a_trip( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + card = CardFactory(content_phash=42) + monkeypatch.setattr( + stage_e_dispatch, + "_sample_envelope_signals", + lambda google_lockout=False: stage_e_dispatch.EnvelopeSignals(load_avg=8.0), + ) + + outcome = dispatch_micro_batch(card_ids=[card.pk]) + + assert outcome.status == "halted-new-trip" + assert outcome.trip_id is not None + trip = EnvelopeTrip.objects.get(trip_id=outcome.trip_id) + assert trip.bar == EnvelopeTrip.Bar.HOST_LOAD + # halted BEFORE any ledger row/Stage C/D work - a halted dispatch never partially starts. + assert PilotRunLedger.objects.count() == 0 + assert ImageEvidence.objects.count() == 0 + + @STREAMING_ON + def test_an_open_trip_refuses_dispatch_even_with_healthy_signals_no_self_resume( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + card = CardFactory(content_phash=42) + open_trip = check_envelope(stage_e_dispatch.EnvelopeSignals(load_avg=9.0)) + assert open_trip is not None + # Every signal is now healthy - the trip alone must still gate dispatch (no self-resume). + monkeypatch.setattr( + stage_e_dispatch, + "_sample_envelope_signals", + lambda google_lockout=False: stage_e_dispatch.EnvelopeSignals(load_avg=0.1), + ) + + outcome = dispatch_micro_batch(card_ids=[card.pk]) + + assert outcome.status == "halted-open-trip" + assert outcome.trip_id == open_trip.trip_id + assert PilotRunLedger.objects.count() == 0 + open_trip.refresh_from_db() + assert open_trip.acknowledged_at is None # this module never clears a trip itself + + @STREAMING_ON + def test_dispatch_resumes_only_after_an_explicit_acknowledge( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + card = CardFactory(content_phash=42) + _full_evidence(card) # already Stage-C-complete, isolates this to "did dispatch proceed" + trip = check_envelope(stage_e_dispatch.EnvelopeSignals(load_avg=9.0)) + assert trip is not None + monkeypatch.setattr( + stage_e_dispatch, + "_sample_envelope_signals", + lambda google_lockout=False: stage_e_dispatch.EnvelopeSignals(load_avg=0.1), + ) + + still_halted = dispatch_micro_batch(card_ids=[card.pk]) + assert still_halted.status == "halted-open-trip" + + acknowledge_trip(trip.trip_id, "load confirmed back to normal") + resumed = dispatch_micro_batch(card_ids=[card.pk]) + assert resumed.status in ("completed", "empty") + assert current_trip() is None + + +class TestFetchFailureWindowSizing: + def test_window_maxlen_matches_the_ratified_500_constant(self) -> None: + window = _FetchOutcomeWindow() + assert window._window.maxlen == FETCH_FAILURE_WINDOW == 500 + + def test_window_caps_at_500_and_evicts_oldest(self) -> None: + window = _FetchOutcomeWindow() + for _ in range(500): + window.record(success=False) + failures, total = window.failures_and_total() + assert (failures, total) == (500, 500) + + window.record(success=True) # the 501st push evicts the oldest (a failure) + failures, total = window.failures_and_total() + assert total == 500 # capped, never grows past the ratified window size + assert failures == 499 # one failure evicted, replaced by a success + + def test_window_feeds_check_envelope_at_exactly_the_ratified_rate(self, db: Any) -> None: + """Ties the window's own sizing directly to operating_envelope's ratified >1%-over-500 + math (docs/proposals/stage-e-streaming.md §10(a)) - not just that this module's deque is + sized 500, but that a real 500-card window built via this module trips (or doesn't) exactly + where the primitive says it should.""" + window = _FetchOutcomeWindow() + for _ in range(494): + window.record(success=True) + for _ in range(6): + window.record(success=False) # 6/500 = 1.2% > 1% ceiling + failures, total = window.failures_and_total() + signals = stage_e_dispatch.EnvelopeSignals(fetch_failures_in_window=failures, fetch_total_in_window=total) + trip = check_envelope(signals) + assert trip is not None + assert trip.bar == EnvelopeTrip.Bar.FETCH_FAILURE_RATE + + def test_exactly_5_of_500_does_not_trip(self, db: Any) -> None: + window = _FetchOutcomeWindow() + for _ in range(495): + window.record(success=True) + for _ in range(5): + window.record(success=False) # exactly 1.0% - the ceiling itself, not a breach + failures, total = window.failures_and_total() + signals = stage_e_dispatch.EnvelopeSignals(fetch_failures_in_window=failures, fetch_total_in_window=total) + assert check_envelope(signals) is None + + +def _full_evidence(card, **overrides: Any) -> ImageEvidence: + """A CURRENT ImageEvidence row carrying every Stage C manifest key - makes Stage C's own resume + filter treat this card as already-done, isolating a test to the Stage D leg (or to pure + dispatch-gating behaviour) without needing to mock the fetch/compute chain at all.""" + defaults = dict( + content_hash=card.content_phash or 0, + extractor_versions={key: f"{key}-v1" for key in MANIFEST_EXTRACTOR_KEYS}, + collector_line_raw_text="", + collector_line_set_code="", + collector_line_collector_number="", + legal_line_proxy_marker_detected=False, + symbol_phash=None, + ) + defaults.update(overrides) + return ImageEvidenceFactory(card=card, **defaults) + + +class TestSelectMicroBatch: + def test_seed_cards_come_first_and_are_deduplicated(self, db: Any) -> None: + card = CardFactory(content_phash=42) + _full_evidence(card) + batch = _select_micro_batch([card.pk, card.pk], batch_size=5) + assert batch == [card.pk] + + def test_backlog_fills_up_to_batch_size_excluding_already_processed_cards(self, db: Any) -> None: + done = CardFactory(content_phash=1) + _full_evidence(done) + pending_a = CardFactory(content_phash=2) + pending_b = CardFactory(content_phash=3) + no_hash = CardFactory(content_phash=None) + + batch = _select_micro_batch([], batch_size=10) + + assert done.pk not in batch + assert no_hash.pk not in batch + assert set(batch) == {pending_a.pk, pending_b.pk} + + def test_backlog_fill_is_bounded_by_batch_size(self, db: Any) -> None: + for _ in range(5): + CardFactory(content_phash=100) + batch = _select_micro_batch([], batch_size=2) + assert len(batch) == 2 + + def test_seed_alone_already_at_batch_size_skips_the_backlog_query_entirely(self, db: Any) -> None: + seed = CardFactory(content_phash=1) + CardFactory(content_phash=2) # would be backlog-eligible, but batch_size=1 leaves no room + batch = _select_micro_batch([seed.pk], batch_size=1) + assert batch == [seed.pk] + + +class TestEndToEndMicroBatch: + """event -> batch -> Stage C extraction + Stage D calculators invoked -> counters written.""" + + @STREAMING_ON + def test_dispatch_for_card_runs_the_full_conveyor(self, db: Any, monkeypatch: pytest.MonkeyPatch) -> None: + card = CardFactory(name="Some Card", content_phash=42) + printing = CanonicalCardFactory(name="Some Card", expansion__code="mom", collector_number="158") + _install_stage_c_stub( + monkeypatch, + fetch_result=_png_bytes(), + collector_line_set_code="mom", + collector_line_collector_number="158", + ) + + dispatch_for_card(card.pk, "card-create") + + evidence = ImageEvidence.objects.get(card=card) + assert MANIFEST_EXTRACTOR_KEYS.issubset(evidence.extractor_versions.keys()) + + vote = CardPrintingTag.objects.get(card=card, anonymous_id=JOIN_KEY_ANONYMOUS_ID) + assert vote.printing_id == printing.pk + assert vote.is_no_match is False + # a single VoteSource.OCR vote can never resolve a card alone (the human-backed gate). + card.refresh_from_db() + assert card.printing_tag_status == PrintingTagStatus.UNRESOLVED + + ledger = PilotRunLedger.objects.get(command="stage_e_streaming_dispatch") + assert ledger.status == PilotRunLedger.Status.COMPLETED + assert ledger.counters["trigger_reason"] == "card-create" + assert ledger.counters["stage_c_completed"] == 1 + assert ledger.counters["stage_d_join_key_votes"] == 1 + assert "peak_rss_mb" in ledger.counters + assert "elapsed_s" in ledger.counters + + @STREAMING_ON + def test_a_card_with_current_evidence_skips_stage_c_but_still_runs_stage_d( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + card = CardFactory(name="Some Card", content_phash=42) + printing = CanonicalCardFactory(name="Some Card", expansion__code="mom", collector_number="158") + _full_evidence(card, collector_line_set_code="mom", collector_line_collector_number="158") + + def _fail_if_called(card, dpi=None): + raise AssertionError("Stage C should have been skipped - evidence is already current") + + _install_stage_c_stub(monkeypatch, fetch_result=_fail_if_called) + + outcome = dispatch_micro_batch(card_ids=[card.pk]) + + assert outcome.status == "completed" + assert outcome.stage_c_completed == 0 + vote = CardPrintingTag.objects.get(card=card, anonymous_id=JOIN_KEY_ANONYMOUS_ID) + assert vote.printing_id == printing.pk + + +class TestGoogleLockoutMidBatch: + @STREAMING_ON + def test_lockout_stops_stage_c_trips_the_envelope_and_refuses_the_next_dispatch( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + card_a = CardFactory(name="Card A", content_phash=1) + card_b = CardFactory(name="Card B", content_phash=2) + + def _lockout_fetch(card, dpi=None): + raise GoogleFetchLockoutError("locked out") + + _install_stage_c_stub(monkeypatch, fetch_result=_lockout_fetch) + + outcome = dispatch_micro_batch(card_ids=[card_a.pk, card_b.pk]) + + assert outcome.status == "completed-with-trip" + assert outcome.trip_id is not None + trip = EnvelopeTrip.objects.get(trip_id=outcome.trip_id) + assert trip.bar == EnvelopeTrip.Bar.GOOGLE_LOCKOUT + assert ImageEvidence.objects.count() == 0 # lockout hit before any card's fetch succeeded + + ledger = PilotRunLedger.objects.get(run_id=outcome.run_id) + assert ledger.counters["lockout_trip_id"] == trip.trip_id + + # no self-resume: the next dispatch call refuses outright. + refused = dispatch_micro_batch(card_ids=[card_a.pk]) + assert refused.status == "halted-open-trip" + + +class TestKillSafetyResumeContract: + """Extends the batch kill-test's own assertions (scripts/ops/crash_drill.sh, docs/proposals/ + stage-e-streaming.md §7) to a streamed micro-batch: a mid-batch crash leaves a truthful FAILED + ledger row and every already-committed card durably written, and a re-invocation over the same + (or an overlapping) card set completes idempotently with zero manual cleanup.""" + + @STREAMING_ON + def test_mid_batch_crash_leaves_truthful_ledger_and_durable_partial_work( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + card_a = CardFactory(name="Card A", content_phash=1) + card_b = CardFactory(name="Card B", content_phash=2) + card_c = CardFactory(name="Card C", content_phash=3) + calls = {"n": 0} + + def _fetch_crashes_on_second_card(card, dpi=None): + calls["n"] += 1 + if calls["n"] == 2: + raise RuntimeError("simulated kill mid-batch") + return _png_bytes() + + _install_stage_c_stub(monkeypatch, fetch_result=_fetch_crashes_on_second_card) + + with pytest.raises(RuntimeError, match="simulated kill mid-batch"): + dispatch_micro_batch(card_ids=[card_a.pk, card_b.pk, card_c.pk], run_id="kill-drill-1") + + ledger = PilotRunLedger.objects.get(run_id="kill-drill-1") + assert ledger.status == PilotRunLedger.Status.FAILED + assert "RuntimeError" in ledger.counters["failure_reason"] + + # durable partial work: the first card's evidence committed before the crash. + assert ImageEvidence.objects.filter(card=card_a).count() == 1 + # nothing committed for the card that crashed or anything after it in this pass. + assert ImageEvidence.objects.filter(card=card_b).count() == 0 + assert ImageEvidence.objects.filter(card=card_c).count() == 0 + + # RESUME: fix the fault and re-invoke over the SAME card set - zero manual cleanup. + _install_stage_c_stub(monkeypatch, fetch_result=_png_bytes()) + resumed = dispatch_micro_batch(card_ids=[card_a.pk, card_b.pk, card_c.pk], run_id="kill-drill-2") + + assert resumed.status == "completed" + # idempotent re-entry: card_a's evidence is not duplicated, despite being re-included. + assert ImageEvidence.objects.filter(card=card_a).count() == 1 + assert ImageEvidence.objects.filter(card=card_b).count() == 1 + assert ImageEvidence.objects.filter(card=card_c).count() == 1 + resumed_ledger = PilotRunLedger.objects.get(run_id="kill-drill-2") + assert resumed_ledger.status == PilotRunLedger.Status.COMPLETED + # only the two cards the crashed run never reached needed real Stage C work this time. + assert resumed_ledger.counters["stage_c_completed"] == 2 + + +class TestBackstopSweep: + @STREAMING_ON + def test_sweep_processes_the_stage_d_backlog_and_is_idempotent_on_rerun( + self, db: Any, capsys: pytest.CaptureFixture + ) -> None: + card = CardFactory(name="Some Card", content_phash=42) + printing = CanonicalCardFactory(name="Some Card", expansion__code="mom", collector_number="158") + _full_evidence(card, collector_line_set_code="mom", collector_line_collector_number="158") + + call_command("stream_backstop_sweep") + first_output = capsys.readouterr().out + assert "batches_dispatched=1" in first_output or "stage_d_votes_or_routes=1" in first_output + + vote = CardPrintingTag.objects.get(card=card, anonymous_id=JOIN_KEY_ANONYMOUS_ID) + assert vote.printing_id == printing.pk + votes_after_first_run = CardPrintingTag.objects.count() + + call_command("stream_backstop_sweep") + second_output = capsys.readouterr().out + assert "batches_dispatched=0" in second_output + + assert CardPrintingTag.objects.count() == votes_after_first_run # idempotent - no dup votes + + @STREAMING_ON + def test_sweep_stops_on_an_envelope_halt(self, db: Any, monkeypatch: pytest.MonkeyPatch) -> None: + CardFactory(name="Some Card", content_phash=42) + monkeypatch.setattr( + stage_e_dispatch, + "_sample_envelope_signals", + lambda google_lockout=False: stage_e_dispatch.EnvelopeSignals(load_avg=9.0), + ) + + call_command("stream_backstop_sweep") + + assert EnvelopeTrip.objects.filter(bar=EnvelopeTrip.Bar.HOST_LOAD).count() == 1 + assert PilotRunLedger.objects.count() == 0 # halted before any batch ledger row was written diff --git a/docs/README.md b/docs/README.md index 8defec90b..00a736d91 100644 --- a/docs/README.md +++ b/docs/README.md @@ -185,9 +185,11 @@ Deployment, incidents, and cross-session lessons. vs. local venv trust, worktree port collisions, ES mapping drift, and more). - [`features/stage-e-operations.md`](features/stage-e-operations.md) — the - Stage E Phase 1 envelope enforcement primitive: PASSIVE vs. BULK mode, - the four ratified pause bars, and the trip/resume runbook - (`resolve_envelope_trip --acknowledge-trip`). Companion to + Stage E Phase 1 envelope enforcement primitive (PASSIVE vs. BULK mode, the + four ratified pause bars, the trip/resume runbook via + `resolve_envelope_trip --acknowledge-trip`) and Phase 2 streaming dispatch + loop (event trigger, cron backstop, micro-batching, the streaming-run + ledger convention) — both **default-OFF**. Companion to [`proposals/stage-e-streaming.md`](proposals/stage-e-streaming.md), which remains the design authority. diff --git a/docs/features/stage-e-operations.md b/docs/features/stage-e-operations.md index 23dae76ca..0960a6f54 100644 --- a/docs/features/stage-e-operations.md +++ b/docs/features/stage-e-operations.md @@ -1,21 +1,24 @@ As of: 2026-07-24 What this is: the admin-facing operational truth for Stage E's envelope -enforcement primitive (Phase 1 of -[`docs/proposals/stage-e-streaming.md`](../proposals/stage-e-streaming.md), -issue [#153](https://github.com/ProxyPrints/ProxyPrints.github.io/issues/153)). +enforcement primitive (Phase 1) and streaming dispatch loop (Phase 2), both +implementing [`docs/proposals/stage-e-streaming.md`](../proposals/stage-e-streaming.md) +(issue [#153](https://github.com/ProxyPrints/ProxyPrints.github.io/issues/153)). That brief is the design authority (still **HOLD**, owner review pending on §3-§5 as a whole) and is not restated here — this doc covers what an operator actually does: the two operating modes, what the envelope bars -mean, and the trip/resume runbook. See +mean, the trip/resume runbook, and (new in Phase 2) the dispatch loop itself +— its trigger, batching, and observability. See [`docs/theory.md`](../theory.md)'s new "Streaming and continuous operation" section for why none of this changes the soundness model. -Phase 1 ships the envelope PRIMITIVE only (`cardpicker/operating_envelope.py`, -the `EnvelopeTrip` model, and the `resolve_envelope_trip` management command) -— nothing in the codebase calls `check_envelope` yet. There is no streaming -dispatch loop today; this doc describes the mechanism a Phase 2 dispatcher -will consume, written now so the runbook exists before the first real trip -does, not after. +Phase 1 shipped the envelope PRIMITIVE (`cardpicker/operating_envelope.py`, +the `EnvelopeTrip` model, and the `resolve_envelope_trip` management command). +Phase 2 (this update) is the first CALLER of that primitive: the streaming +dispatch loop itself (`cardpicker/stage_e_dispatch.py`) — see "Phase 2 — the +streaming dispatch loop" below. **Both phases ship default-OFF** +(`settings.STAGE_E_STREAMING_ENABLED = False`) — turning streaming on against +production is the phase-3 shakedown's own polled owner action, not something +either phase does by merging. --- @@ -131,27 +134,161 @@ not an oversight: the admin is a monitoring surface for finding a `trip_id` to hand to the command, not a second, less-visible resume path that could bypass the mandatory `--note` and the CLI's own audit trail. -## What Phase 2/3 will add +## Phase 2 — the streaming dispatch loop -This page describes Phase 1 (the primitive) only. Not yet built, and not -promised on any particular timeline — see `stage-e-streaming.md` itself for -the full design (still HOLD pending owner review of §3-§5): +Built 2026-07-24, per the owner-approved Phase 2 implementation task for +`stage-e-streaming.md` §3-§5 (still HOLD as a brief — this is the +owner-pre-approved implementation of what it already specced, the same +posture Phase 1 shipped under). Ships **default-OFF** +(`settings.STAGE_E_STREAMING_ENABLED = False`) — every mechanism below is +live code, wired unconditionally, but every entry point checks the flag +first and is a no-op while it's False. Flipping it to `True` is the ONLY +action the phase-3 shakedown needs to take to go live; no redeploy of this +code is required. -- **Phase 2**: the actual streaming dispatch loop — the event-driven - qcluster trigger (`stage-e-streaming.md` §3 decision (1)), micro-batch - sizing derived from the Bug-A shakedown cohort measurement (§10(c)), and - the wiring that calls `check_envelope`/`current_trip` before every - dispatch (this primitive has no caller yet). -- **Phase 3** (informal shorthand, not a brief-defined phase number): the - dispatcher-kill acceptance test (`stage-e-streaming.md` §7) and the - `CardScanLog` retention tripwire mechanism (§10(b)) — both specced in the - brief, neither built in this change. +### What it is + +`cardpicker/stage_e_dispatch.py`'s `dispatch_micro_batch` is the CONVEYOR — +the first real caller of Phase 1's `check_envelope`/`current_trip` primitive. +It is a DISPATCH LOOP only: it never reimplements Stage C extraction, Stage D +calculator decode logic, or consensus resolution — every actual accept/reject +decision still happens inside the same `cardpicker.image_evidence`/ +`cardpicker.local_calculate_verdicts`/`cardpicker.printing_consensus` code +BULK mode already uses, called via their existing entry points. BULK-mode +commands (`run_image_evidence_cohort`, `local_calculate_verdicts`, +`reparse_collector_evidence`, `consensus_recompute`, etc.) are byte-identical +to before this change — none of their own call sites pass the new optional +`card_ids` scoping parameter `local_calculate_verdicts.py`'s three calculator +entry points (`run_join_key_calculator`/`run_fallback_calculator`/ +`run_slow_path_calculator`) gained for this module's benefit. + +### Ordering, every dispatch call + +1. **Default-off gate** — `settings.STAGE_E_STREAMING_ENABLED` must be `True`, + or the call is a no-op (`status="disabled"`). +2. **No-self-resume gate** — `operating_envelope.current_trip()` must be + `None`, or the call refuses outright (`status="halted-open-trip"`) with + zero DB writes beyond the lookup itself. This is the binding rule from + Phase 1's own review: no code path in `stage_e_dispatch.py` ever calls + `acknowledge_trip` — resume is always `resolve_envelope_trip`'s own + command, a fresh, explicit owner action (see the runbook above). +3. **Fresh envelope sample** — live host load (`os.getloadavg()`), this + worker process's own RSS (`cardpicker.process_metrics.get_process_rss_mb`), + and a rolling fetch-outcome window feed `check_envelope`. If THIS sample + breaches a bar, a new trip is recorded and the call halts + (`status="halted-new-trip"`) before touching Stage C/D at all. +4. **Micro-batch selection** — `_select_micro_batch` builds the card-id list: + the triggering event's own card first (if any), filled up to + `settings.STAGE_E_MICRO_BATCH_SIZE` from the Stage C backlog (cards + lacking a full-manifest `ImageEvidence` row — the same shape + `run_image_evidence_cohort.py`'s own resume filter uses, imported, not + reimplemented). +5. **Stage C** (sequential, per-card, not pooled — a micro-batch is far too + small for BULK mode's process-pool concurrency to help) — the same + `compute_card_evidence`/`persist_evidence` unit `run_image_evidence_cohort.py` + drives, one card at a time. A `GoogleFetchLockoutError` stops Stage C for + this batch immediately and records a fresh trip (instant-pause bar) — + in-flight, already-committed work stays committed; Stage D below still + runs against whatever was reached ("in-flight work drains, nothing NEW + starts"). +6. **Stage D** — `run_join_key_calculator`/`run_fallback_calculator`/ + `run_slow_path_calculator`, called AS-IS with the new `card_ids` scope, in + the same escalation order every BULK-mode invocation already uses. Each of + these already calls `resolve_and_persist_printing` internally for every + card it touches — this is what satisfies §3 decision (4)'s "scoped + incremental per-touch consensus recompute" with no separate consensus step + in the dispatcher at all. +7. **Ledger write** — one `PilotRunLedger` row per micro-batch (see + "Observability" below). + +### Trigger: event-driven, plus a cron backstop (§3 decision (1)) + +- **Event-driven** (`cardpicker/stage_e_signals.py`, wired in + `cardpicker.apps.CardpickerConfig.ready()`): a `post_save` receiver on + `Card` (only `created=True` — "card-create") and on `ImageEvidence` (every + save — "evidence-change") queues `dispatch_for_card` via django-q2's + `async_task`, never inline. Both receivers check + `STAGE_E_STREAMING_ENABLED` before doing anything, including before + importing `django_q.tasks` — connecting the receivers themselves is always + cheap and side-effect-free, only the flag gates real work. +- **Cron backstop** (`manage.py stream_backstop_sweep`): re-runs the same + eligibility selectors against the Stage C backlog, then (once that's empty) + the Stage D join-key-eligible backlog, dispatching micro-batches until both + are exhausted, the envelope trips, or `--max-batches` is reached. Catches + anything a lost/never-fired django-q dispatch missed (django-q2's own + delivery guarantee is at-least-once-attempted, not exactly-once-delivered). + **Not scheduled anywhere by this change** — no django-q `Schedule` row is + created; wiring an actual cadence is a phase-3/live-deploy action, not a + code change. + +### Micro-batch sizing (§3 decision (2), sharpened by §10(c)) + +`settings.STAGE_E_MICRO_BATCH_SIZE` (default `25`, env-tunable, no code +change needed to adjust) is a **placeholder**, not a considered answer — +§10(c) ratifies that the real number ships as a MEASURED OUTPUT of the Bug-A +tail shakedown's own instrumentation (phase 3, not yet run). The default +sits inside the brief's own "roughly 10-100 cards per batch" sanity range as +a conservative starting point pending that measurement. + +### Observability: the streaming-run ledger convention + +Every micro-batch — from either trigger — writes one `PilotRunLedger` row: +`command="stage_e_streaming_dispatch"`, `dry_run=False` (PASSIVE mode has no +per-batch dry-run leg — see `stage-e-streaming.md` §3 decision (5)), and +`counters` carrying `trigger_reason` (`"card-create"`/`"evidence-change"`/ +`"backstop-sweep"`/`"backstop-sweep-stage-d"`), `batch_size`, +`stage_c_completed`, `stage_c_fetch_failures`, `stage_d_join_key_votes`, +`stage_d_fallback_votes`, `stage_d_slow_path_routed`, `elapsed_s`, +`peak_rss_mb` (via the same `process_metrics.get_process_rss_mb` Phase 1 +wired in), and `lockout_trip_id` (non-null only when a Google lockout tripped +mid-batch). A halted call (`disabled`/`halted-open-trip`/`halted-new-trip`) +writes NO ledger row at all — a halted dispatch never partially starts, so +there's nothing to record beyond the `EnvelopeTrip` row `check_envelope` +itself already persists. A crashed batch (any other exception) is marked +`FAILED` with `counters["failure_reason"]` via the same +`pilot_run_lifecycle.mark_ledger_failed` rail every BULK-mode command uses — +no new failure-handling mechanism. + +### Resume contract, extended to a streamed micro-batch + +Each card's own `persist_evidence` call is its own transaction — a crash +mid-batch leaves every already-persisted card durably written and nothing +partially written for the card the crash interrupted. A re-invocation over +the same (or an overlapping) card-id set is idempotent: Stage C's resume +filter skips cards already fully processed, and Stage D's own +anonymous_id-exclusion eligibility queries skip cards already voted on — the +same "truthful ledger, idempotent re-entry, zero manual cleanup" property the +batch kill-test (`scripts/ops/crash_drill.sh`) already proves for BULK mode, +now covered for the streamed path by `cardpicker/tests/test_stage_e_dispatch.py`'s +`TestKillSafetyResumeContract` (a mid-batch exception, a truthful `FAILED` +ledger row, and an idempotent re-invocation, at unit-test granularity). The +LIVE, host-level dispatcher-kill drill `stage-e-streaming.md` §7(b) specs +(killing the dispatcher PROCESS itself, not a simulated exception) is still +open — see that section for why it's sequenced into the phase-3 shakedown, +not this change. + +## Phase 3 (not yet built) + +Informal shorthand, not a brief-defined phase number — see +`stage-e-streaming.md` for the full design (still HOLD pending owner review +of §3-§5 as a whole): + +- **Turning `STAGE_E_STREAMING_ENABLED` on** against production — the + phase-3 shakedown's own polled owner action, explicitly not done by either + Phase 1 or Phase 2 landing. +- **The live, host-level dispatcher-kill acceptance test** (`stage-e-streaming.md` + §7(b)) — killing the dispatcher process itself mid-stream, not a simulated + exception. +- **The `CardScanLog` retention tripwire mechanism** (§10(b)) — specced in + the brief, not built in this change. +- **The Bug-A tail shakedown itself** (§6 item 1, issue #418) — the cohort + that measures the real `STAGE_E_MICRO_BATCH_SIZE` (§10(c)). ## See also - [`docs/proposals/stage-e-streaming.md`](../proposals/stage-e-streaming.md) - — the full design brief this page implements Phase 1 of; the design - authority for every number and decision cited above. + — the full design brief this page implements Phase 1 and Phase 2 of; the + design authority for every number and decision cited above. - [`docs/theory.md`](../theory.md) — "Streaming and continuous operation": why moving from batch to streaming (and this envelope's pause/resume mechanism) changes nothing about the pipeline's soundness model. diff --git a/docs/proposals/stage-e-streaming.md b/docs/proposals/stage-e-streaming.md index 776b77775..8eb11fbb6 100644 --- a/docs/proposals/stage-e-streaming.md +++ b/docs/proposals/stage-e-streaming.md @@ -31,6 +31,25 @@ no streaming dispatch loop exists yet, and §3-§5 as a whole still need owner review before Phase 2 (the loop that actually consumes this primitive) is built. +**Phase 2 built 2026-07-24** (the streaming dispatch loop itself, per the +owner-approved implementation task for this brief's §3-§5 as specced) - +`cardpicker/stage_e_dispatch.py` (the conveyor: default-off gate, the +no-self-resume check against Phase 1's `current_trip`, a fresh +`check_envelope` sample, micro-batch selection, sequential Stage C, scoped +Stage D via the new `card_ids` parameter `local_calculate_verdicts.py`'s +three calculator entry points gained, and the per-batch `PilotRunLedger` +row), `cardpicker/stage_e_signals.py` (the event-driven card-create/ +evidence-change trigger, §3 decision (1)), and +`manage.py stream_backstop_sweep` (the cron backstop, same decision). Ships +**default-OFF** (`settings.STAGE_E_STREAMING_ENABLED = False`) - see +[`docs/features/stage-e-operations.md`](../features/stage-e-operations.md)'s +new "Phase 2" section for the full operator-facing detail (trigger, batching, +observability, resume contract). **This does not lift the HOLD above either** - +turning streaming on in production, the live host-level dispatcher-kill drill +(§7(b)), the `CardScanLog` retention tripwire (§10(b)), and the Bug-A tail +shakedown that measures the real micro-batch size (§10(c)) are all still +open, tracked as "Phase 3" in the ops doc. + **§10 update (2026-07-24, owner):** four of §9's open items are now ratified — the rate-control/authorization-envelope question (§9 item 1) and the `CardScanLog` retention question (§9 item 2) are RESOLVED, and