From 6fd8c7fbe08b47c7dad2814ad69dee171365d924 Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:37:02 +0000 Subject: [PATCH 1/5] Thin-extract the two seams the monolith needs: Stage 0 freshness, cluster-vote propagation Neither piece is changed in behaviour; both are lifted from a caller-private scope (a BaseCommand method / a closure over six pieces of run_pilot state) to a module-level callable so a second engine can reach them without a copy. - stream_full_catalog._run_stage_zero_freshness -> module-level run_stage_zero_freshness, taking write/warn callables instead of closing over self.stdout/self.style. It now also RETURNS the bulk-file vintage (remote updated_at, cache path, cache mtime age, whether it refreshed, the import stats) so a caller can date its own conclusions; the existing method is a wrapper that discards it and prints exactly what it always printed. - local_identify_printing_tags.run_pilot's propagate_cluster_vote closure -> module-level build_propagated_cluster_votes, which returns the rows rather than writing them. The closure now calls it and keeps its own batching and written-id ledgers. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013NhYmT1PxCcyemA16dFDxN --- .../local_identify_printing_tags.py | 95 ++++++-- .../commands/stream_full_catalog.py | 215 +++++++++++------- 2 files changed, 204 insertions(+), 106 deletions(-) diff --git a/MPCAutofill/cardpicker/local_identify_printing_tags.py b/MPCAutofill/cardpicker/local_identify_printing_tags.py index e32a1e71b..c0f259553 100644 --- a/MPCAutofill/cardpicker/local_identify_printing_tags.py +++ b/MPCAutofill/cardpicker/local_identify_printing_tags.py @@ -1248,6 +1248,56 @@ class AttributeReport: cards_absorbed_into_clusters: int = 0 +def build_propagated_cluster_votes( + *, + representative_card_id: int, + printing_pk: int, + anonymous_id: str, + confidence: float, + run_id: Optional[str], + members_by_representative: dict[int, list[int]], + members_already_voted: set[int], + source: VoteSource = VoteSource.OCR, +) -> list[CardPrintingTag]: + """ + THE CLUSTER-VOTE PROPAGATION RULE (addendum item 2a), lifted out of `run_pilot`'s own + `propagate_cluster_vote` closure 2026-07-30 so it has exactly one implementation and two + callers - `run_pilot` (unchanged behaviour: the closure now calls this and does its own + batching) and `run_pipeline`, the one-command monolith, which had no way to reach it at all + while it lived inside a closure over six pieces of `run_pilot`-local state. + + An accepted vote on a distance-0 cluster REPRESENTATIVE propagates as an identical vote (same + `anonymous_id`, `printing`, `confidence`) to every OTHER member of its cluster. Absorbed + members are, by construction, cards whose stored `content_phash` is bit-identical to the + representative's - the same image - so this is an identity property first and a throughput + lever second: it is what makes an identity group AGREE, and it does it without the member + ever being fetched or computed. + + `members_already_voted` is the caller's pre-computed set of member card ids that ALREADY carry + a vote under this same `anonymous_id` (one query, up front, never re-queried per call - see + `run_pilot`'s own call site for why a member can legitimately be in that state). Propagating + to one anyway would violate `CardPrintingTag`'s own (card, printing, anonymous_id) uniqueness + constraint, and would silently double-vote or overwrite regardless. + + Returns the rows to write; it never writes them itself, so the caller keeps ownership of its + own batching, purge-and-write discipline and gate accounting. + """ + member_ids = members_by_representative.get(representative_card_id, []) + return [ + CardPrintingTag( + card_id=member_id, + printing_id=printing_pk, + is_no_match=False, + anonymous_id=anonymous_id, + source=source, + confidence=confidence, + run_id=run_id, + ) + for member_id in member_ids + if member_id not in members_already_voted + ] + + def run_pilot( engine: Literal["ocr", "phash", "both"] = "both", limit: int = 300, @@ -1483,29 +1533,28 @@ def propagate_cluster_vote( engine this run) - propagating anyway would violate CardPrintingTag's own (card, printing, anonymous_id) uniqueness constraint, and would silently double-vote or attempt to overwrite an existing vote regardless. Returns how many propagated votes were - actually queued, for the engine's votes_written tally.""" - member_ids = cluster_result.members_by_representative.get(representative_card_id, []) - already_voted = members_already_voted_by_anonymous_id.get(anonymous_id, set()) - propagated = 0 - for member_id in member_ids: - if member_id in already_voted: - continue - votes_batch.append( - CardPrintingTag( - card_id=member_id, - printing_id=printing_pk, - is_no_match=False, - anonymous_id=anonymous_id, - source=VoteSource.OCR, - confidence=confidence, - run_id=run_id, - ) - ) - if member_id not in written_card_ids: - written_card_ids.append(member_id) - batch_written_card_ids.append(member_id) - propagated += 1 - return propagated + actually queued, for the engine's votes_written tally. + + The row-building half was lifted verbatim into the module-level + `build_propagated_cluster_votes` (2026-07-30) so a second engine - `run_pipeline`, the + one-command monolith - can propagate cluster votes without a second copy of it. This + closure keeps everything that is genuinely `run_pilot`-local: which batch the rows join, + and the two written-id ledgers the gate check reads.""" + rows = build_propagated_cluster_votes( + representative_card_id=representative_card_id, + printing_pk=printing_pk, + anonymous_id=anonymous_id, + confidence=confidence, + run_id=run_id, + members_by_representative=cluster_result.members_by_representative, + members_already_voted=members_already_voted_by_anonymous_id.get(anonymous_id, set()), + ) + for row in rows: + votes_batch.append(row) + if row.card_id not in written_card_ids: + written_card_ids.append(row.card_id) + batch_written_card_ids.append(row.card_id) + return len(rows) # Fetch budget (pre-scale program item 3b): every image fetch is one request against the # image CDN Worker, which shares its daily request quota with live site traffic diff --git a/MPCAutofill/cardpicker/management/commands/stream_full_catalog.py b/MPCAutofill/cardpicker/management/commands/stream_full_catalog.py index a1797b5e9..7535c32aa 100644 --- a/MPCAutofill/cardpicker/management/commands/stream_full_catalog.py +++ b/MPCAutofill/cardpicker/management/commands/stream_full_catalog.py @@ -313,7 +313,7 @@ import logging import random import time -from typing import Any, List, Optional +from typing import Any, Callable, List, Optional from django.conf import settings from django.core.management.base import BaseCommand, CommandError, CommandParser @@ -517,6 +517,125 @@ def deterministic_sample_pks(sample_size: int, source_keys: Optional[List[str]] return sorted(random.Random(sample_size).sample(all_pks, sample_size)) +def run_stage_zero_freshness( + *, + require_fresh: bool, + is_resume: bool, + write: Callable[[str], None], + warn: Callable[[str], str], +) -> dict[str, Any]: + """ + STAGE 0 (GitHub issue #513 item 2) - verify Scryfall printing-metadata freshness and + refresh it if stale, ONCE, before any batch is dispatched. See this module's own docstring + for the three binding properties (once-only and why, fail-before-any-dispatch, report what + it decided). Every failure path below raises `CommandError` with + `returncode=EXIT_STAGE_ZERO_FAILED` (2, the module docstring's EXIT CODES table), i.e. a + non-zero exit with no batch dispatched and nothing written. + + Calls `printing_metadata_import`'s own entry points, never a reimplementation of them. + `_get_default_cards_entry` is what makes the remote comparison possible at all (it returns + the live `/bulk-data` entry carrying `updated_at`), `_is_fresh` is that module's own + verdict, and `import_scryfall_printing_metadata` is the refresh - which repeats the same + check internally and so re-downloads only if it agrees the cache is stale. + """ + try: + entry = _get_default_cards_entry() + except Exception as exc: # noqa: BLE001 - any failure here must abort the run, loudly + raise CommandError( + f"STAGE 0 FAILED: could not read Scryfall's /bulk-data entry ({exc!r}). Refusing " + "to start a full-catalog pass against reference data of unknown age. Nothing was " + "dispatched and nothing was written.", + returncode=EXIT_STAGE_ZERO_FAILED, + ) + + cache_path = _cache_path() + age_text = "cache absent" + age_days: Optional[float] = None + if cache_path.exists(): + age_days = (time.time() - cache_path.stat().st_mtime) / 86400.0 + age_text = f"local cache mtime age {age_days:.2f}d" + fresh = _is_fresh(cache_path, entry) + + if fresh: + write( + f"STAGE 0: Scryfall printing metadata is FRESH - skipping refresh " + f"(remote updated_at={entry.updated_at}, {age_text}, path={cache_path})." + ) + return _stage_zero_vintage(entry, cache_path, age_days, refreshed=False) + + if require_fresh: + raise CommandError( + f"STAGE 0 FAILED: Scryfall printing metadata is STALE and --require-fresh was " + f"passed (remote updated_at={entry.updated_at}, {age_text}, path={cache_path}). " + "Refresh it first, or drop --require-fresh to let stage 0 refresh it. Nothing was " + "dispatched and nothing was written.", + returncode=EXIT_STAGE_ZERO_FAILED, + ) + + write( + f"STAGE 0: Scryfall printing metadata is STALE - refreshing now, once, before any " + f"batch (remote updated_at={entry.updated_at}, {age_text}, path={cache_path})." + ) + if is_resume: + # Same early/late inconsistency a mid-run refresh would cause, only spread across two + # invocations: whatever the prior invocation already dispatched was deduced against the + # OLD reference set, and everything from here on is deduced against the new one. + write( + warn( + "STAGE 0 WARNING: this is a RESUMED run and the reference data CHANGED since " + "the previous invocation. Batches already completed under the stored " + "high-water mark were deduced against the OLD CanonicalPrintingMetadata; " + "everything from here on uses the new one. The completed pass will not be " + "internally consistent. Consider restarting with --start-pk 0." + ) + ) + logger.warning( + "stream_full_catalog stage 0: reference data changed on a RESUMED run - " + "pre-resume batches used the older CanonicalPrintingMetadata" + ) + + try: + stats = import_scryfall_printing_metadata() + except Exception as exc: # noqa: BLE001 - see this method's own docstring + raise CommandError( + f"STAGE 0 FAILED: Scryfall printing-metadata refresh raised {exc!r}. Refusing to " + "start a full-catalog pass against half-imported reference data. Nothing was " + "dispatched.", + returncode=EXIT_STAGE_ZERO_FAILED, + ) + write( + f"STAGE 0: refresh complete - created={stats.get('created')} " + f"updated={stats.get('updated')} deleted={stats.get('deleted')} " + f"skipped={stats.get('skipped')} no_matching_card={stats.get('no_matching_card')}. " + "This will NOT run again for the lifetime of this invocation." + ) + return _stage_zero_vintage(entry, cache_path, age_days, refreshed=True, import_stats=stats) + + +def _stage_zero_vintage( + entry: Any, + cache_path: Any, + age_days: Optional[float], + *, + refreshed: bool, + import_stats: Optional[dict[str, Any]] = None, +) -> dict[str, Any]: + """ + THE BULK-FILE VINTAGE, as a recordable dict rather than only a printed line (2026-07-30). A + run's conclusions can only be dated if the run says which Scryfall bulk file it reasoned + against - `remote_updated_at` IS that date. Returned by `run_stage_zero_freshness` so a caller + can fold it into its own `PilotRunLedger.counters`; `stream_full_catalog`'s own method wrapper + discards it and keeps printing exactly what it always printed. + """ + return { + "remote_updated_at": str(getattr(entry, "updated_at", "")), + "cache_path": str(cache_path), + "cache_age_days": round(age_days, 4) if age_days is not None else None, + "refreshed": refreshed, + "import_stats": import_stats, + } + + class Command(BaseCommand): help = ( "Push the FULL catalog through the Stage E streaming conveyor (cardpicker.stage_e_dispatch) " @@ -1183,90 +1302,20 @@ def _write_verdict( f"(relaunch with the SAME --source flags to continue, or --start-pk {resume_pk})" ) - def _run_stage_zero_freshness(self, *, require_fresh: bool, is_resume: bool) -> None: + def _run_stage_zero_freshness(self, *, require_fresh: bool, is_resume: bool) -> dict[str, Any]: """ - STAGE 0 (GitHub issue #513 item 2) - verify Scryfall printing-metadata freshness and - refresh it if stale, ONCE, before any batch is dispatched. See this module's own docstring - for the three binding properties (once-only and why, fail-before-any-dispatch, report what - it decided). Every failure path below raises `CommandError` with - `returncode=EXIT_STAGE_ZERO_FAILED` (2, the module docstring's EXIT CODES table), i.e. a - non-zero exit with no batch dispatched and nothing written. - - Calls `printing_metadata_import`'s own entry points, never a reimplementation of them. - `_get_default_cards_entry` is what makes the remote comparison possible at all (it returns - the live `/bulk-data` entry carrying `updated_at`), `_is_fresh` is that module's own - verdict, and `import_scryfall_printing_metadata` is the refresh - which repeats the same - check internally and so re-downloads only if it agrees the cache is stale. + Stage 0 for THIS command - a thin wrapper (2026-07-30) over the module-level + `run_stage_zero_freshness` the body was lifted into verbatim so a second driver + (`run_pipeline`, the one-command monolith) can run the identical stage without a second + copy of it. Behaviour, output strings and the `EXIT_STAGE_ZERO_FAILED` contract are + unchanged; only the `self.stdout`/`self.style` bindings are passed in rather than closed + over. The returned vintage dict is unused here and recorded by the monolith instead. """ - try: - entry = _get_default_cards_entry() - except Exception as exc: # noqa: BLE001 - any failure here must abort the run, loudly - raise CommandError( - f"STAGE 0 FAILED: could not read Scryfall's /bulk-data entry ({exc!r}). Refusing " - "to start a full-catalog pass against reference data of unknown age. Nothing was " - "dispatched and nothing was written.", - returncode=EXIT_STAGE_ZERO_FAILED, - ) - - cache_path = _cache_path() - age_text = "cache absent" - if cache_path.exists(): - age_days = (time.time() - cache_path.stat().st_mtime) / 86400.0 - age_text = f"local cache mtime age {age_days:.2f}d" - fresh = _is_fresh(cache_path, entry) - - if fresh: - self.stdout.write( - f"STAGE 0: Scryfall printing metadata is FRESH - skipping refresh " - f"(remote updated_at={entry.updated_at}, {age_text}, path={cache_path})." - ) - return - - if require_fresh: - raise CommandError( - f"STAGE 0 FAILED: Scryfall printing metadata is STALE and --require-fresh was " - f"passed (remote updated_at={entry.updated_at}, {age_text}, path={cache_path}). " - "Refresh it first, or drop --require-fresh to let stage 0 refresh it. Nothing was " - "dispatched and nothing was written.", - returncode=EXIT_STAGE_ZERO_FAILED, - ) - - self.stdout.write( - f"STAGE 0: Scryfall printing metadata is STALE - refreshing now, once, before any " - f"batch (remote updated_at={entry.updated_at}, {age_text}, path={cache_path})." - ) - if is_resume: - # Same early/late inconsistency a mid-run refresh would cause, only spread across two - # invocations: whatever the prior invocation already dispatched was deduced against the - # OLD reference set, and everything from here on is deduced against the new one. - self.stdout.write( - self.style.WARNING( - "STAGE 0 WARNING: this is a RESUMED run and the reference data CHANGED since " - "the previous invocation. Batches already completed under the stored " - "high-water mark were deduced against the OLD CanonicalPrintingMetadata; " - "everything from here on uses the new one. The completed pass will not be " - "internally consistent. Consider restarting with --start-pk 0." - ) - ) - logger.warning( - "stream_full_catalog stage 0: reference data changed on a RESUMED run - " - "pre-resume batches used the older CanonicalPrintingMetadata" - ) - - try: - stats = import_scryfall_printing_metadata() - except Exception as exc: # noqa: BLE001 - see this method's own docstring - raise CommandError( - f"STAGE 0 FAILED: Scryfall printing-metadata refresh raised {exc!r}. Refusing to " - "start a full-catalog pass against half-imported reference data. Nothing was " - "dispatched.", - returncode=EXIT_STAGE_ZERO_FAILED, - ) - self.stdout.write( - f"STAGE 0: refresh complete - created={stats.get('created')} " - f"updated={stats.get('updated')} deleted={stats.get('deleted')} " - f"skipped={stats.get('skipped')} no_matching_card={stats.get('no_matching_card')}. " - "This will NOT run again for the lifetime of this invocation." + return run_stage_zero_freshness( + require_fresh=require_fresh, + is_resume=is_resume, + write=self.stdout.write, + warn=self.style.WARNING, ) def _emit_progress_summary( From e6847b66a80bf3413df44b36422514fa88eeb83f Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:01:06 +0000 Subject: [PATCH 2/5] THE MONOLITH: one command that runs the whole pipeline end to end manage.py run_pipeline. A bare invocation is a complete, from-scratch, whole-catalogue run that WRITES. Every flag either narrows the cohort, disables a stage, or withholds the write; none is a precondition. Stage 0 stream_full_catalog.run_stage_zero_freshness, and the bulk-file vintage it returns is recorded on the run's ledger row so the run can be dated Stage E operating_envelope.current_trip / check_envelope preflight Stage C run_image_evidence_cohort via call_command (pooled engine, run-scoped resume, md5 evidence transfer, RSS guard - none of it re-derived here) Stage D stage_e_dispatch._run_stage_d, called explicitly rather than reached by the post_save echo: join-key -> fallback -> illustration -> slow-path, then the border / frame-style / bleed-edge chips Stage C+ local_clustering.compute_two_threshold_clusters + local_identify_printing_tags.build_propagated_cluster_votes - the pilot capability no engine had Stage E verify_zero_resolutions fidelity gate end channel_report, reported and never folded into the exit status _run_stage_d/_run_attribute_chip_casters/_run_illustration_calculator gain two parameters so one Stage D serves both engines: batch_ids=None is bulk mode, and dry_run is threaded through. Both defaults preserve existing behaviour exactly. Write polarity is inverted relative to every organ called: all six Stage D calculators/casters default to dry_run=True, so the monolith passes dry_run explicitly at every seam. Inheriting those defaults would compute a whole pass and persist nothing while every log line reported success. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013NhYmT1PxCcyemA16dFDxN --- .../management/commands/run_pipeline.py | 728 ++++++++++++++++++ MPCAutofill/cardpicker/stage_e_dispatch.py | 42 +- .../cardpicker/tests/test_run_pipeline.py | 583 ++++++++++++++ 3 files changed, 1342 insertions(+), 11 deletions(-) create mode 100644 MPCAutofill/cardpicker/management/commands/run_pipeline.py create mode 100644 MPCAutofill/cardpicker/tests/test_run_pipeline.py diff --git a/MPCAutofill/cardpicker/management/commands/run_pipeline.py b/MPCAutofill/cardpicker/management/commands/run_pipeline.py new file mode 100644 index 000000000..a27b24d6f --- /dev/null +++ b/MPCAutofill/cardpicker/management/commands/run_pipeline.py @@ -0,0 +1,728 @@ +""" +THE MONOLITH - one command that runs the whole identification pipeline end to end. + +Owner brief, 2026-07-30: "1 click". Every stage below already existed and every stage below was +run separately, by hand, in an order carried in an operator's head. Nothing here is a new +inference, a new calculator, or a new heuristic: this module is imports, sequencing, `run_id` +threading and error handling. If a future change adds pipeline LOGIC to this file, that logic is +in the wrong place - it belongs in the stage that owns it. + + manage.py run_pipeline + +is a complete, working, from-scratch, whole-catalogue run THAT WRITES. That is a hard +requirement, not a nicety (owner rulings: "a bulk run redoes everything from scratch; flags tell +it to narrow", "default the default things, disable them with flags", and - the polarity, stated +explicitly - "the eventual intention for the monolith is that default is to write and flags are +what prevents it. (opposite)"). Every flag on this command either NARROWS the cohort, DISABLES a +stage, or WITHHOLDS the write. None is a precondition. + +THE WRITE POLARITY IS INVERTED RELATIVE TO EVERY STAGE THIS COMMAND CALLS, deliberately, and this +is the single easiest thing here to get wrong. `local_calculate_verdicts`' four calculators, +`local_layout_class_cast.run_layout_class_cast` and +`local_attribute_chip_cast.run_attribute_chip_cast` all default to `dry_run=True` - the correct +default for a command an operator invokes deliberately, and a catastrophic one to INHERIT here, +because a monolith that computes a whole 230k-card pass and persists nothing fails looking exactly +like success: full logs, every stage reporting, zero rows. This command therefore passes +`dry_run` EXPLICITLY at every seam and defaults it to False. The pieces whose default had to be +overridden, in full: + + run_join_key_calculator dry_run=True -> passed False + run_fallback_calculator dry_run=True -> passed False + run_illustration_calculator dry_run=True -> passed False + run_slow_path_calculator dry_run=True -> passed False + run_layout_class_cast dry_run=True -> passed False + run_attribute_chip_cast dry_run=True -> passed False + +(all six via `stage_e_dispatch._run_stage_d`, which already passed False and now takes the flag as +a parameter). `run_image_evidence_cohort` is the one stage already write-by-default; its +`--dry-run` is forwarded rather than overridden. `import_scryfall_printing_metadata` has no +dry-run mode at all, which is why `--dry-run` SKIPS stage 0 rather than pretending to run it - +the same rule `stream_full_catalog` already documents, and for the same reason: a refresh is a +real download and a real DB write. + +`--dry-run` IS A REAL PASS THAT WITHHOLDS THE WRITE, not a plan. Every stage runs, every +calculator reports what it WOULD cast, clustering computes and reports what it WOULD propagate, +and `channel_report` still runs so an operator gets the preview in the shape they will read it in +afterwards. It exits 0: it did what was asked. + +NOTE ON `add_dry_run_guard_arguments`. The repo's forced-dry-run PRECONDITION (a `--write` refuses +unless a matching COMPLETED dry-run row exists) is a guard on commands whose default is dry-run; +it is not added here, because a command that writes by default has no `--write` to gate and +requiring an operator to have run a dry-run first would put a flag back in front of the working +run. The one place it still applies is inside Stage C's own `--card-ids-file` path, and this +command forwards `--skip-dryrun-check` for exactly that. + +THE SEQUENCE, and why it is this sequence +========================================= + + Stage 0 Scryfall reference refresh once, at the front + Stage E envelope preflight before anything is written + Stage C evidence extraction, pooled the run's only network stage + Stage D the four calculators + three chips explicit, in dependency order + Stage C+ md5/phash cluster vote propagation the pilot capability no engine had + Stage E fidelity gate machine-only resolutions must be zero + end channel_report what did every channel actually produce? + +STAGE 0 - SCRYFALL AT THE FRONT. Owner ruling: "the scryfall importer is meant to be wired to the +front of the entire monolith, not run separately." This calls +`stream_full_catalog.run_stage_zero_freshness`, the SAME stage-0 `stream_full_catalog` runs (its +body was lifted to module level for this; see its own docstring), which in turn calls +`printing_metadata_import`'s own entry points. `import_scryfall_printing_metadata` is a full-set +VALUE-DIFFING upsert, so re-running it IS the backfill - it repopulates `face_illustrations` and +picks up any drift, and a re-import against an unchanged bulk file issues no row writes at all. + +Once, at the front, never during the run. This is the non-obvious constraint and it is inherited +verbatim: Stage D's illustration deduction builds its matching index from +`CanonicalPrintingMetadata`, exactly the table a refresh rewrites. A mid-run refresh would have +early cards deduced against one reference set and late cards against another under a single +`run_id`, producing results neither comparable across the run nor reproducible from it. + +THE RUN RECORDS ITS BULK-FILE VINTAGE. `run_stage_zero_freshness` returns the remote `updated_at` +it compared against, the cache path, the cache's mtime age and whether it refreshed; all of it +lands in this run's `PilotRunLedger.counters["stage_0"]`. A run's conclusions can only be dated if +the run says which Scryfall bulk file it reasoned against. + +STAGE C - POOLED, RUN-SCOPED. Delegates to `run_image_evidence_cohort` via `call_command`, which +is the whole point: that command owns the pooled engine (ThreadPoolExecutor fetch -> bounded queue +-> ProcessPoolExecutor compute), the priority ordering, the resume filter, the RSS guard, the +md5 evidence-transfer path and its own ledger row. Re-deriving any of that here would be a second +copy to keep in sync. Two properties it already has and this command must not break: + + - A FRESH `--run-id` REDOES EVERYTHING (PR #645). `already_extracted_card_ids(run_id)` scopes + the resume filter to rows THIS run wrote, so a new run reconsiders every card. Within-run + resume still survives a crash: re-invoke with the same `--run-id`. + - `bleed_diff_mm` IS FILLED BY THIS PASS, with no extra wiring and no backfill command. It was + 97.9% NULL only because the field was added without an extractor version bump, so no existing + row was ever re-extracted. `image_evidence.compute_card_evidence` already calls + `local_fallback.compute_bleed_diff_mm` unconditionally (image_evidence.py:976) inside the + `geometry_bleed` group. A from-scratch run re-extracts every card, so every card gets it. + +STAGE D - EXPLICIT, NOT BY ECHO. Owner ruling: "everything in the conveyor should be running in +the monolith." This calls `stage_e_dispatch._run_stage_d` DIRECTLY, with `batch_ids=None` (bulk +mode - see that function's own docstring). It deliberately does NOT rely on the pooled runner's +`post_save` echo into the conveyor: that dependency is implicit, it runs one micro-batch at a time +under the streaming envelope, and it is exactly what left the Stage D route unestablished (#618). + +Calling `_run_stage_d` rather than re-listing its calls is the point. It is the one place the +order lives, and the order is load-bearing (PR #604): join-key -> fallback -> illustration -> +slow-path, then the three attribute chips. The asymmetry is the correctness argument - a +calculator's `run_id` narrows its OWN progress, never an UPSTREAM verdict. Run-scoping the +upstream selectors would hand downstream calculators an empty pool while reporting success. +`_run_stage_d` also carries the three attribute-chip casters PR #654 wired in (border via +`local_layout_class_cast`, frame-style and bleed-edge via `local_attribute_chip_cast`), which is +how the monolith reaches all three without naming them itself. + +STAGE C+ - CLUSTER VOTE PROPAGATION, the capability no engine had. `local_clustering. +compute_two_threshold_clusters` computes distance-0 clusters over stored `content_phash` values - +pure, no fetch, no writes. A d=0 member is an image bit-identical to its representative's. +`local_identify_printing_tags.build_propagated_cluster_votes` (lifted out of `run_pilot`'s closure +for this) then gives every absorbed member its representative's printing verdict, under the same +identity, WITHOUT the member ever being fetched or computed. This is a correctness property first +- identity groups must agree - and a throughput lever second. + +WHAT IT PROPAGATES, AND WHY THAT DIFFERS FROM THE PILOT. `run_pilot` propagates its own OCR/phash +votes. The monolith does not carry pilot OCR/phash voting at all (standing owner deferral; Stage +D's join-key already superseded the OCR half). So there is no OCR/phash vote here to propagate, +and the monolith propagates STAGE D's printing verdicts instead - the votes this run actually +cast. Same rule, same guard, same identity discipline, different upstream. See DEVIATIONS in this +change's report. + +PILOT OCR/PHASH VOTING IS DELIBERATELY NOT CARRIED. Standing owner deferral: "we have been +deferring the pilot phash until the end, with the expectation that our new pipeline will either +shake out what it needs to identify or render it obsolete." `local_calculate_verdicts.py:221` +records that Stage D's join-key superseded the OCR half. + +STAGE E - REUSED, NOT REIMPLEMENTED. `operating_envelope.current_trip` / `check_envelope` and +`stage_e_dispatch._sample_envelope_signals` are called as-is. The throttle behaviour from PR #644 +(rate pressure throttles, genuine breaches halt) and the global 7/s ceiling from PR #649 live +BELOW this command, inside `harvest_fetch_limiter` / `harvest_rate_coordinator`, and apply to +Stage C's fetches whether this command knows about them or not - which is why this file does not +mention them again. The fidelity gate is `local_identify_printing_tags.verify_zero_resolutions`, +the same gate `local_calculate_verdicts` runs between its calculators. + +END - `channel_report`. Run at the end of the pass, ALWAYS non-gating for this command's own exit +status. Expect exit 1 on the first run: `ZERO_DECLARATIONS` ships empty and there are known-silent +channels. That is the instrument working, and silencing it here would be building the instrument +and suppressing its first reading in the same change. + +THE RUN'S IDENTITY IS ITS `run_id`, AND NOTHING ELSE. There is no `--test-mode`, no provisional +marker, no separate table and no weight discount, deliberately. The first run of this command is a +shakedown, but provisionality is a property of our CONFIDENCE, not of the data - it lives in the +ledger and in whatever ruling follows from reading `channel_report`. What makes a later run able +to disregard this one is the from-scratch default: a fresh `run_id` reconsiders every card, and +`local_calculate_verdicts._split_new_printing_tag_votes` supersedes a changed verdict while +`models.purge_stale_machine_votes` archives the old row into `ArchivedCardPrintingTag` first, so +the two generations stay diffable via `local_calculate_verdicts --generation-diff`. The default +`run_id` is therefore self-describing (`monolith--`) rather than opaque, and +is printed prominently at the start AND the end so an operator can name this run later. + +EXIT CODES - the supervisor contract, matching `stream_full_catalog`'s own table where they +overlap: + + 0 the pass ran to completion (or, under `--dry-run`, previewed one). `channel_report`'s + own verdict is REPORTED, never folded in. + 2 Stage 0 failed. Nothing was dispatched and nothing was written. + 3 the operating envelope halted the run (an open trip, or a fresh breach). + 7 the pass completed but the FIDELITY GATE found machine-only resolutions. Everything is + written; this is a loud "read this run before trusting it", not a rollback. +""" + +import time +from typing import Any, Optional, cast + +from django.core.management import call_command +from django.core.management.base import BaseCommand, CommandError, CommandParser +from django.utils import timezone + +from cardpicker.local_clustering import compute_two_threshold_clusters +from cardpicker.local_identify_printing_tags import ( + build_propagated_cluster_votes, + verify_zero_resolutions, +) +from cardpicker.management.commands.stream_full_catalog import ( + EXIT_ENVELOPE_HALT, + run_stage_zero_freshness, +) +from cardpicker.models import Card, CardPrintingTag, PilotRunLedger, VoteSource +from cardpicker.operating_envelope import check_envelope, current_trip +from cardpicker.pilot_run_lifecycle import ( + mark_ledger_failed, + merge_counters, + resilient_terminal_output, +) +from cardpicker.stage_e_dispatch import ( + DispatchOutcome, + _run_stage_d, + _sample_envelope_signals, +) +from cardpicker.utils import get_baked_git_sha +from cardpicker.vote_write import purge_and_write_votes + +EXIT_FIDELITY_GATE_VIOLATION = 7 + +# The cohort bound handed to `run_image_evidence_cohort` when the operator has not narrowed the +# run. That command's own `--limit` defaults to 3000 (a pilot-sized default for a pilot-sized +# invocation); a monolith run is whole-catalogue by default, so this passes a bound larger than +# the catalogue rather than inventing an "unbounded" mode that command does not have. +WHOLE_CATALOGUE_LIMIT = 100_000_000 + +# Suffix distinguishing THIS command's own `PilotRunLedger` row from the row its delegated Stage C +# invocation writes under the same run identity. See `handle`'s own comment for why the data rows, +# not the summary row, keep the unsuffixed name. +LEDGER_RUN_ID_SUFFIX = "-pipeline" + + +class Command(BaseCommand): + help = ( + "THE MONOLITH: run the whole identification pipeline end to end in one command - Scryfall " + "refresh, pooled Stage C evidence extraction, all four Stage D calculators plus the three " + "attribute-chip casters, md5/phash cluster vote propagation, the fidelity gate, and " + "channel_report. A bare invocation is a complete from-scratch whole-catalogue run; every " + "flag either narrows the cohort or disables a stage." + ) + + def add_arguments(self, parser: CommandParser) -> None: + parser.add_argument( + "--run-id", + dest="run_id", + default=None, + help=( + "Identity of this run, stamped onto every row it writes and onto its " + "PilotRunLedger row. Defaults to a self-describing monolith-. A " + "FRESH run-id redoes everything from scratch; re-passing an EARLIER run-id " + "resumes that run where it stopped." + ), + ) + # ---- cohort narrowing ------------------------------------------------------------- + parser.add_argument( + "--limit", + dest="limit", + type=int, + default=None, + help=( + "Narrow Stage C to the highest-priority N cards. Default: the whole catalogue. " + "Stage D still runs in bulk mode unless --scope-stage-d-to-cohort is passed." + ), + ) + parser.add_argument( + "--card-ids-file", + dest="card_ids_file", + default=None, + help="Narrow Stage C to an explicit newline-delimited card-id file.", + ) + parser.add_argument( + "--scope-stage-d-to-cohort", + dest="scope_stage_d", + action="store_true", + default=False, + help=( + "Scope Stage D and cluster propagation to the Stage C cohort instead of the whole " + "eligible catalogue. Only meaningful alongside --limit/--card-ids-file, and only " + "advisable for small cohorts: card ids are pushed into every dependency subquery, " + "which is a large win at batch size 25 and a pathology at catalogue scale." + ), + ) + # ---- Stage C engine tunables (forwarded verbatim) --------------------------------- + parser.add_argument("--workers", dest="workers", type=int, default=None) + parser.add_argument("--fetch-threads", dest="fetch_threads", type=int, default=None) + parser.add_argument("--queue-depth", dest="queue_depth", type=int, default=None) + parser.add_argument("--max-rss-mb", dest="max_rss_mb", type=float, default=None) + parser.add_argument( + "--no-shortcircuit", + dest="no_shortcircuit", + action="store_true", + default=False, + help="Forwarded to Stage C: extract every card fully rather than short-circuiting.", + ) + parser.add_argument( + "--skip-dryrun-check", + dest="skip_dryrun_check", + action="store_true", + default=False, + help=( + "Forwarded to Stage C's own forced-dry-run guard, which arms only for a " + "--card-ids-file write. Prominently logged wherever it applies." + ), + ) + # ---- stage disable flags (every stage is ON by default) --------------------------- + parser.add_argument( + "--skip-freshness", + dest="skip_freshness", + action="store_true", + default=False, + help="Skip Stage 0 entirely (tests, bounded trials, a resumed run whose data has not moved).", + ) + parser.add_argument( + "--require-fresh", + dest="require_fresh", + action="store_true", + default=False, + help="Make Stage 0 verify-only: fail rather than refresh.", + ) + parser.add_argument( + "--dry-run", + dest="dry_run", + action="store_true", + default=False, + help=( + "Run every stage and report what WOULD be written, without writing anything. THE " + "WRITE IS THE DEFAULT - this flag is the only thing that prevents it. Stage 0 is " + "skipped under --dry-run (a Scryfall refresh is a real download and a real DB " + "write, with no preview mode of its own). Exits 0: it did what was asked." + ), + ) + 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) + parser.add_argument( + "--skip-envelope", + dest="skip_envelope", + action="store_true", + default=False, + help="Skip the operating-envelope preflight. Never appropriate for a real bulk run.", + ) + parser.add_argument("--skip-gate", dest="skip_gate", action="store_true", default=False) + parser.add_argument("--skip-channel-report", dest="skip_channel_report", action="store_true", default=False) + + # ------------------------------------------------------------------------------------------ + def handle(self, *args: Any, **options: Any) -> None: + started = time.monotonic() + run_id: str = options["run_id"] or f"monolith-{time.strftime('%Y%m%dT%H%M%SZ', time.gmtime())}" + + self.stdout.write("=" * 78) + dry_run: bool = options["dry_run"] + self.stdout.write(f"MONOLITH RUN run_id={run_id} mode={'DRY-RUN (writes nothing)' if dry_run else 'WRITE'}") + self.stdout.write(f"git_sha={get_baked_git_sha()}") + self.stdout.write( + "Every row this run writes is stamped with that run_id. It is the ONLY thing that " + "identifies this run's output - there is no test marker and no separate table." + ) + self.stdout.write(f"To resume this run after a stop: --run-id {run_id}") + self.stdout.write("=" * 78) + + counters: dict[str, Any] = {} + # THE LEDGER ROW'S OWN id IS SUFFIXED; EVERY DATA ROW'S IS NOT. `PilotRunLedger.run_id` is + # UNIQUE (models.py), one row per run identity - and Stage C is delegated to + # `run_image_evidence_cohort`, which writes its own row under the run identity it is given. + # Two rows cannot share one. The choice is therefore which of the two things gets the clean + # `run_id`: this command's summary row, or every `ImageEvidence`/`CardPrintingTag`/ + # `CardTagVote`/`CardScanLog` row the run produces. It must be the DATA - `channel_report` + # scopes a channel's run-column counts by the run_id ON THE ROWS, and a Stage C whose + # evidence carried a different run_id from the votes would read as a silent channel, which + # is exactly the reading "nothing is culled" exists to make impossible. So the pipeline's + # own summary row takes the suffix and the pipeline's OUTPUT keeps the name the operator + # typed. This mirrors `stream_full_catalog`'s own prefix/per-batch-suffix convention. + ledger = PilotRunLedger.objects.create( + command="run_pipeline", + run_id=f"{run_id}{LEDGER_RUN_ID_SUFFIX}", + dry_run=dry_run, + status=PilotRunLedger.Status.RUNNING, + git_sha=get_baked_git_sha(), + ) + + try: + # -- STAGE 0 --------------------------------------------------------------------- + if options["skip_freshness"] or dry_run: + why = "--skip-freshness" if options["skip_freshness"] else "--dry-run" + self.stdout.write(f"STAGE 0 skipped ({why}).") + counters["stage_0"] = {"skipped": True, "reason": why} + else: + counters["stage_0"] = run_stage_zero_freshness( + require_fresh=options["require_fresh"], + is_resume=bool(options["run_id"]), + write=self.stdout.write, + warn=self.style.WARNING, + ) + + # -- STAGE E PREFLIGHT ------------------------------------------------------------ + self._envelope_preflight(run_id=run_id, skip=options["skip_envelope"]) + + # -- STAGE C ---------------------------------------------------------------------- + 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} + else: + counters["stage_c"] = self._run_stage_c(run_id=run_id, options=options, dry_run=dry_run) + + 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) + + # -- STAGE C+ : CLUSTER VOTE PROPAGATION ------------------------------------------- + if options["skip_clustering"]: + self.stdout.write("STAGE C+ clustering skipped (--skip-clustering).") + counters["clustering"] = {"skipped": True} + else: + counters["clustering"] = self._propagate_cluster_votes( + run_id=run_id, cohort_ids=cohort_ids, dry_run=dry_run + ) + + # -- STAGE E : FIDELITY GATE ------------------------------------------------------- + gate_violations: list[int] = [] + if dry_run: + # Nothing was written, so there is no resolution state for the gate to inspect. + self.stdout.write("FIDELITY GATE skipped (--dry-run wrote no votes to check).") + counters["fidelity_gate"] = {"skipped": True, "reason": "--dry-run"} + elif options["skip_gate"]: + self.stdout.write("FIDELITY GATE skipped (--skip-gate).") + counters["fidelity_gate"] = {"skipped": True} + else: + gate_violations = self._run_fidelity_gate(run_id=run_id) + counters["fidelity_gate"] = {"violations": len(gate_violations)} + + # -- END : channel_report ---------------------------------------------------------- + if options["skip_channel_report"]: + self.stdout.write("channel_report skipped (--skip-channel-report).") + counters["channel_report"] = {"skipped": True} + else: + counters["channel_report"] = self._run_channel_report(run_id=run_id) + + counters["elapsed_s"] = round(time.monotonic() - started, 1) + ledger.status = PilotRunLedger.Status.COMPLETED + ledger.finished_at = timezone.now() + ledger.counters = merge_counters(ledger.counters, counters) + ledger.save(update_fields=["status", "finished_at", "counters"]) + + with resilient_terminal_output(): + self.stdout.write("=" * 78) + self.stdout.write( + f"MONOLITH DONE run_id={run_id} " + f"mode={'DRY-RUN (nothing written)' if dry_run else 'WRITE'} " + f"elapsed={counters['elapsed_s']:.0f}s" + ) + self.stdout.write(f"Name this run by its run_id: {run_id}") + self.stdout.write("=" * 78) + + if gate_violations: + raise CommandError( + f"FIDELITY GATE: {len(gate_violations)} card(s) reached a RESOLVED printing " + "state on machine votes alone in this run. Everything this run computed is " + f"written and is queryable by run_id={run_id}; this exit is a loud 'read this " + "run before trusting it', not a rollback.", + returncode=EXIT_FIDELITY_GATE_VIOLATION, + ) + + except Exception as exc: # noqa: BLE001 - see pilot_run_lifecycle.mark_ledger_failed + # A no-op when this invocation already marked the row COMPLETED above (the gate- + # violation CommandError path), matching every other long-running command's + # counters-before-output convention. + mark_ledger_failed(ledger, exc) + raise + + # ------------------------------------------------------------------------------------------ + def _envelope_preflight(self, *, run_id: str, skip: bool) -> None: + """ + The operating envelope, checked ONCE before anything is written - `operating_envelope`'s + own two entry points, called in the order that module's docstring requires (`current_trip` + BEFORE `check_envelope`, never the reverse) and with the same no-self-resume rule the + conveyor has: an open trip refuses outright and is cleared by an owner action + (`resolve_envelope_trip`), never by a run deciding for itself that it is fine now. + + The rate-pressure half of the envelope (PR #644's throttle-instead-of-halt, PR #649's + global 7/s ceiling) is NOT checked here and must not be: it lives underneath Stage C in + `harvest_fetch_limiter`/`harvest_rate_coordinator`, applies per request, and its whole + point is that rate pressure slows the pass rather than stopping it. + """ + if skip: + self.stdout.write("STAGE E envelope preflight skipped (--skip-envelope).") + return + + existing = current_trip(run_id=run_id) + if existing is not None: + raise CommandError( + f"ENVELOPE HALT: trip {existing.trip_id} ({existing.bar}) is still open. No " + "self-resume - clear it with `resolve_envelope_trip` after investigating. Nothing " + "was written.", + returncode=EXIT_ENVELOPE_HALT, + ) + fresh = check_envelope(_sample_envelope_signals(), run_id=run_id) + if fresh is not None: + raise CommandError( + f"ENVELOPE HALT: bar {fresh.bar} breached ({fresh.detail}); trip {fresh.trip_id} " + "persisted. Nothing was written.", + returncode=EXIT_ENVELOPE_HALT, + ) + self.stdout.write("STAGE E: operating envelope clear.") + + # ------------------------------------------------------------------------------------------ + def _run_stage_c(self, *, run_id: str, options: dict[str, Any], dry_run: bool = False) -> dict[str, Any]: + """ + Stage C, delegated whole to `run_image_evidence_cohort` - the pooled engine, its priority + ordering, its run-scoped resume filter (PR #645), its RSS guard, its md5 evidence-transfer + path and its own ledger row, none of which is re-derived here. Counters are read back off + THAT command's ledger row rather than returned, because a `call_command` gives no return + value and inventing a channel for one would mean editing the command to suit this caller. + """ + argv: list[str] = ["--run-id", run_id] + if options["card_ids_file"]: + argv += ["--card-ids-file", options["card_ids_file"]] + else: + argv += ["--limit", str(options["limit"] if options["limit"] is not None else WHOLE_CATALOGUE_LIMIT)] + for flag, key in ( + ("--workers", "workers"), + ("--fetch-threads", "fetch_threads"), + ("--queue-depth", "queue_depth"), + ): + if options[key] is not None: + argv += [flag, str(options[key])] + if options["max_rss_mb"] is not None: + argv += ["--max-rss-mb", str(options["max_rss_mb"])] + if options["no_shortcircuit"]: + argv.append("--no-shortcircuit") + if options["skip_dryrun_check"]: + argv.append("--skip-dryrun-check") + if dry_run: + # Stage C is the one delegated stage that is ALREADY write-by-default, so this + # forwards its own flag rather than overriding a default. + argv.append("--dry-run") + + self.stdout.write(f"STAGE C: run_image_evidence_cohort {' '.join(argv)}") + call_command("run_image_evidence_cohort", *argv) + + row = ( + PilotRunLedger.objects.filter(command="run_image_evidence_cohort", run_id=run_id) + .order_by("-started_at") + .first() + ) + return dict(row.counters or {}) if row is not None else {} + + # ------------------------------------------------------------------------------------------ + def _run_stage_d_bulk( + self, *, run_id: str, cohort_ids: Optional[list[int]], dry_run: bool = False + ) -> dict[str, Any]: + """ + Stage D, called EXPLICITLY - `stage_e_dispatch._run_stage_d`, the single place the + dependency order lives, invoked in bulk mode (`batch_ids=None`). This is deliberately not + the pooled runner's `post_save` echo into the conveyor: that route is implicit, runs one + micro-batch at a time under the streaming envelope, and is what left Stage D unestablished. + + Four calculators in dependency order plus the three attribute-chip casters, all of it + owned by `_run_stage_d`. `DispatchOutcome` is that function's output channel; it is + constructed here purely to receive the counters. + """ + self.stdout.write( + "STAGE D: join-key -> fallback -> illustration -> slow-path, then the border / " + "frame-style / bleed-edge attribute chips" + + ("" if cohort_ids is None else f" (scoped to {len(cohort_ids)} cards)") + ) + outcome = DispatchOutcome(status="monolith", run_id=run_id) + # `dry_run` PASSED EXPLICITLY. All six calculators/casters underneath default to + # dry_run=True; inheriting that here would compute a whole pass and persist nothing while + # every log line and counter still reported success. See this module's docstring. + _run_stage_d(cohort_ids, run_id, outcome, dry_run=dry_run) + + result = { + "join_key_votes": outcome.stage_d_join_key_votes, + "join_key_already_voted": outcome.stage_d_join_key_already_voted, + "fallback_votes": outcome.stage_d_fallback_votes, + "fallback_already_voted": outcome.stage_d_fallback_already_voted, + "illustration_votes": outcome.stage_d_illustration_votes, + "illustration_already_voted": outcome.stage_d_illustration_already_voted, + "slow_path_routed": outcome.stage_d_slow_path_routed, + "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, + } + self.stdout.write(f"STAGE D: {result}") + return result + + # ------------------------------------------------------------------------------------------ + def _propagate_cluster_votes( + self, *, run_id: str, cohort_ids: Optional[list[int]], dry_run: bool = False + ) -> dict[str, Any]: + """ + STAGE C+ - the pilot capability that was reachable from no engine. + + `compute_two_threshold_clusters` groups cards by their stored `content_phash`; a distance-0 + cluster is a set of BIT-IDENTICAL images. `build_propagated_cluster_votes` then gives every + absorbed member its representative's printing verdict under the same identity, with no + fetch and no compute for the member. Both functions are called, never reimplemented. + + THE CLUSTER POOL IS THE CATALOGUE, NOT THIS RUN'S SELECTION. `run_pilot` clusters over its + own eligibility-narrowed selection pool, which makes membership a function of what earlier + runs already voted on: a genuine 3-card cluster can present as a 2-card one, and dropping + the lowest-pk member changes which card becomes representative. Clustering over `Card` by + stored hash - the whole catalogue by default - is what makes the answer independent of run + history. When `--scope-stage-d-to-cohort` narrows this, that independence is narrowed too; + that is the cost of the flag and the reason it is not the default. + + `members_already_voted` is one query, up front, per identity - a member that already holds + a vote under the same `anonymous_id` is skipped, because propagating anyway would violate + `CardPrintingTag`'s own (card, printing, anonymous_id) uniqueness constraint. + """ + cards = Card.objects.filter(content_phash__isnull=False) + if cohort_ids is not None: + cards = cards.filter(pk__in=cohort_ids) + selected = [_ClusterInput(card=card) for card in cards.only("pk", "content_phash").iterator()] + + # `compute_two_threshold_clusters` is ANNOTATED against the pilot's `SelectedCard` but its + # runtime contract is only `.card.pk` and `.card.content_phash` (`SelectedCard` is a + # TYPE_CHECKING-only import there, and it carries a pilot candidate list this command has + # no use for). Cast rather than widen that pure module's signature for this caller. + cluster_result = compute_two_threshold_clusters(cast(Any, selected)) + members_by_representative = cluster_result.members_by_representative + member_ids = {m for members in members_by_representative.values() for m in members} + stats: dict[str, Any] = { + "cluster_count": len(members_by_representative), + "cards_absorbed_into_clusters": len(member_ids), + "votes_propagated": 0, + } + if not member_ids: + self.stdout.write(f"STAGE C+: {stats} - nothing to propagate.") + return stats + + # The representatives' own verdicts, as cast by THIS run's Stage D. + source_votes = list( + CardPrintingTag.objects.filter( + card_id__in=list(members_by_representative.keys()), run_id=run_id, is_no_match=False + ).exclude(printing_id=None) + ) + already_voted_by_identity: dict[str, set[int]] = {} + for anonymous_id in {vote.anonymous_id for vote in source_votes}: + already_voted_by_identity[anonymous_id] = set( + CardPrintingTag.objects.filter(card_id__in=member_ids, anonymous_id=anonymous_id).values_list( + "card_id", flat=True + ) + ) + + rows: list[CardPrintingTag] = [] + for vote in source_votes: + if vote.printing_id is None or vote.confidence is None: + # Cannot happen given the queryset above; asserted here so a later change to that + # filter cannot silently start propagating a vote with no printing or no weight. + continue + rows.extend( + build_propagated_cluster_votes( + representative_card_id=vote.card_id, + printing_pk=vote.printing_id, + anonymous_id=vote.anonymous_id, + confidence=vote.confidence, + run_id=run_id, + members_by_representative=members_by_representative, + members_already_voted=already_voted_by_identity.get(vote.anonymous_id, set()), + source=VoteSource(vote.source), + ) + ) + if rows and not dry_run: + # The same write rail every other printing-vote writer uses: the purge is atomic with + # the insert and scoped to exactly the rows being inserted, and a superseded row is + # archived into `ArchivedCardPrintingTag` before deletion. + purge_and_write_votes(CardPrintingTag, rows, target_field="card_id") + stats["would_propagate" if dry_run else "votes_propagated"] = len(rows) + self.stdout.write(f"STAGE C+: {stats}") + return stats + + # ------------------------------------------------------------------------------------------ + def _run_fidelity_gate(self, *, run_id: str) -> list[int]: + """ + The Stage D fidelity gate - `verify_zero_resolutions`, the same check + `local_calculate_verdicts` runs between its calculators, applied here once over every card + this run cast a printing vote for. It answers one question: did any card reach a RESOLVED + printing state on machine votes alone? The answer must be zero. + """ + card_ids = list(CardPrintingTag.objects.filter(run_id=run_id).values_list("card_id", flat=True).distinct()) + if not card_ids: + self.stdout.write("FIDELITY GATE: this run cast no printing votes - nothing to check.") + return [] + violations = verify_zero_resolutions(card_ids) + if violations: + self.stdout.write( + self.style.ERROR(f"FIDELITY GATE VIOLATION: {len(violations)} card(s): {violations[:20]}") + ) + else: + self.stdout.write(f"FIDELITY GATE: clear over {len(card_ids)} cards.") + return violations + + # ------------------------------------------------------------------------------------------ + def _run_channel_report(self, *, run_id: str) -> dict[str, Any]: + """ + `channel_report`, run at the end of the pass and NEVER folded into this command's own exit + status. Its exit 1 is expected on a first run - `ZERO_DECLARATIONS` ships empty and there + are known-silent channels - and that is the instrument working. Gating the monolith on it + would mean a correct reading of a real gap failing a run that did exactly what was asked. + """ + self.stdout.write("=" * 78) + self.stdout.write("CHANNEL REPORT (non-gating for this command's exit status)") + exit_code = 0 + try: + call_command("channel_report", "--run-id", run_id) + except SystemExit as exc: + exit_code = int(exc.code or 0) + self.stdout.write( + f"channel_report exit={exit_code}" + + ( + " <- EXPECTED on a first run: ZERO_DECLARATIONS ships empty and there are " + "known-silent channels. Read the findings; do not silence them here." + if exit_code + else "" + ) + ) + return {"exit_code": exit_code} + + +class _ClusterInput: + """ + The two-attribute shape `local_clustering.compute_two_threshold_clusters` reads (`.card.pk`, + `.card.content_phash`). `SelectedCard`, the type that function is annotated against, is a + `TYPE_CHECKING`-only import there and carries a pilot candidate list this command has no use + for; the runtime contract is these two attributes and nothing else. + """ + + __slots__ = ("card",) + + def __init__(self, card: Card) -> None: + self.card = card diff --git a/MPCAutofill/cardpicker/stage_e_dispatch.py b/MPCAutofill/cardpicker/stage_e_dispatch.py index 25a2fa3ed..08f98c2d4 100644 --- a/MPCAutofill/cardpicker/stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/stage_e_dispatch.py @@ -1058,7 +1058,7 @@ def _run_stage_c( return trip -def _run_illustration_calculator(run_id: str, card_ids: list[int]) -> Any: +def _run_illustration_calculator(run_id: str, card_ids: Optional[list[int]], dry_run: bool = False) -> Any: """ Lazy-import wrapper for `cardpicker.local_illustration.run_illustration_calculator` - mirrors `_stage_c_manifest_extractor_keys`'s own posture of avoiding a hard import-time dependency @@ -1069,10 +1069,12 @@ def _run_illustration_calculator(run_id: str, card_ids: list[int]) -> Any: """ from cardpicker.local_illustration import run_illustration_calculator - return run_illustration_calculator(run_id=run_id, dry_run=False, card_ids=card_ids) + return run_illustration_calculator(run_id=run_id, dry_run=dry_run, card_ids=card_ids) -def _run_attribute_chip_casters(run_id: str, card_ids: list[int], outcome: DispatchOutcome) -> None: +def _run_attribute_chip_casters( + run_id: str, card_ids: Optional[list[int]], outcome: DispatchOutcome, dry_run: bool = False +) -> None: """ THE ATTRIBUTE-CHIP CASTERS, wired into the conveyor (2026-07-30, closing the 2026-07-29 composition audit's §1 Q1 items 1-3). Same lazy-import posture as @@ -1115,10 +1117,10 @@ def _run_attribute_chip_casters(run_id: str, card_ids: list[int], outcome: Dispa from cardpicker.local_layout_class_cast import run_layout_class_cast try: - border_result = run_layout_class_cast(run_id=run_id, dry_run=False, card_ids=card_ids) + border_result = run_layout_class_cast(run_id=run_id, dry_run=dry_run, card_ids=card_ids) outcome.stage_d_border_chip_votes = border_result.votes_written - chip_result = run_attribute_chip_cast(run_id=run_id, dry_run=False, card_ids=card_ids) + chip_result = run_attribute_chip_cast(run_id=run_id, dry_run=dry_run, card_ids=card_ids) outcome.stage_d_frame_chip_votes = chip_result.frame_votes_written outcome.stage_d_bleed_chip_votes = chip_result.bleed_votes_written except RuntimeError as exc: @@ -1132,7 +1134,7 @@ def _run_attribute_chip_casters(run_id: str, card_ids: list[int], outcome: Dispa ) -def _run_stage_d(batch_ids: list[int], run_id: str, outcome: DispatchOutcome) -> None: +def _run_stage_d(batch_ids: Optional[list[int]], run_id: str, outcome: DispatchOutcome, dry_run: bool = False) -> 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 @@ -1144,6 +1146,24 @@ def _run_stage_d(batch_ids: list[int], run_id: str, outcome: DispatchOutcome) -> 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. + `batch_ids=None` IS BULK MODE (2026-07-30, for `run_pipeline`, the one-command monolith). + Every calculator and caster below already accepts `card_ids=None` and has always treated it as + "the whole eligible catalogue" - that is the mode `local_calculate_verdicts`' own bulk command + invokes them in. Passing it through here means the monolith runs THIS sequence, in THIS order, + rather than keeping a second copy of it: the conveyor's per-micro-batch scoping and a + full-catalogue pass become the same Stage D, differing only in that one argument. A + full-catalogue pass must NOT instead pass its whole cohort as an explicit id list - `card_ids` + is pushed down into every dependency subquery (PR #579), which is a large win at batch 25 and + a `pk__in` list of ~230,000 ids at catalogue scale. + + `dry_run` DEFAULTS TO FALSE HERE AND MUST STAY THAT WAY. Every caller of this function is an + ENGINE, and an engine that computes a whole pass and persists nothing fails in the worst + available shape: full logs, every counter reporting, zero rows. The `dry_run=True` path exists + for `run_pipeline --dry-run`, whose whole job is to let an operator preview a 230k-card pass + before committing to it - each calculator below already had the parameter and already reports + `would_cast` alongside `votes_written`, so a dry run is a real, fully-executed pass that + withholds only the write. + CONCURRENT-DISPATCH VOTE COLLISION (2026-07-24, shakedown run tripping envelope trip envtrip-20260724T214616-be6e5db9): this is the FIRST caller ever to invoke `run_join_key_calculator`/`run_fallback_calculator` concurrently (django-q2 runs @@ -1170,24 +1190,24 @@ def _run_stage_d(batch_ids: list[int], run_id: str, outcome: DispatchOutcome) -> see `settings.STAGE_E_MAX_CONCURRENT_DISPATCHES` (companion change) for the actual fix to that, and `docs/features/stage-e-operations.md`'s runbook for acknowledging the open trip itself. """ - join_key_result = run_join_key_calculator(run_id=run_id, dry_run=False, card_ids=batch_ids) + join_key_result = run_join_key_calculator(run_id=run_id, dry_run=dry_run, card_ids=batch_ids) outcome.stage_d_join_key_votes = join_key_result.votes_written + join_key_result.no_match_votes_written outcome.stage_d_join_key_already_voted = join_key_result.already_voted - fallback_result = run_fallback_calculator(run_id=run_id, dry_run=False, card_ids=batch_ids) + fallback_result = run_fallback_calculator(run_id=run_id, dry_run=dry_run, card_ids=batch_ids) outcome.stage_d_fallback_votes = fallback_result.votes_written outcome.stage_d_fallback_already_voted = fallback_result.already_voted - illustration_result = _run_illustration_calculator(run_id=run_id, card_ids=batch_ids) + illustration_result = _run_illustration_calculator(run_id=run_id, card_ids=batch_ids, dry_run=dry_run) outcome.stage_d_illustration_votes = illustration_result.votes_written outcome.stage_d_illustration_already_voted = illustration_result.already_voted - slow_path_result = run_slow_path_calculator(run_id=run_id, dry_run=False, card_ids=batch_ids) + slow_path_result = run_slow_path_calculator(run_id=run_id, dry_run=dry_run, card_ids=batch_ids) outcome.stage_d_slow_path_routed = slow_path_result.routed_written # See `_run_attribute_chip_casters`' own docstring: three chip families that were reachable # from neither engine, two of them at zero rows with no substitute. Zero image fetches. - _run_attribute_chip_casters(run_id=run_id, card_ids=batch_ids, outcome=outcome) + _run_attribute_chip_casters(run_id=run_id, card_ids=batch_ids, outcome=outcome, dry_run=dry_run) def dispatch_micro_batch( diff --git a/MPCAutofill/cardpicker/tests/test_run_pipeline.py b/MPCAutofill/cardpicker/tests/test_run_pipeline.py new file mode 100644 index 000000000..1ad27ca83 --- /dev/null +++ b/MPCAutofill/cardpicker/tests/test_run_pipeline.py @@ -0,0 +1,583 @@ +""" +END-TO-END COVER FOR THE MONOLITH (`manage.py run_pipeline`). + +This suite drives the WHOLE command over a small fixture cohort and asserts that each stage ran +and produced rows - because the thing that ships in `run_pipeline.py` is the WIRING, and the only +defect class that matters for wiring is "a stage silently did not run". A unit test per stage +cannot catch that; the stages all had unit tests already, and were still not connected. + +WHAT IS STUBBED, AND WHY IT IS EXACTLY THIS MUCH. Two seams, both of them the network: + + - `run_image_evidence_cohort._fetch_one_card` - the only place the pipeline touches the image + CDN. The stub decides per card whether the "fetch" succeeded, which is how the propagation + test below gets a cluster member with no evidence of its own. + - `run_image_evidence_cohort._compute_one_card` - the CPU-bound extractor stage. The stub writes + the `ImageEvidence` row a real extraction would have written. Running the real extractors + would mean shipping real card images and a tesseract binary into this test for no wiring + coverage at all: every extractor already has its own unit test, and none of them is what this + file is about. + - `run_pipeline.run_stage_zero_freshness` - a real Scryfall bulk download. Stage 0's own + behaviour is covered by `test_stream_full_catalog.TestStageZeroFreshness`; what THIS file + asserts is that the monolith calls it, once, at the front, and records the vintage it returned. + +Everything downstream of those three - every Stage D calculator, all three attribute-chip casters, +the clustering, the propagation, the fidelity gate, `channel_report`, and every `run_id` handoff +between them - runs FOR REAL against the test database. + +`transaction=True` throughout: `run_image_evidence_cohort.handle` closes its parent DB +connections before forking its pools (`_parent_connections.close_all()`), which under the plain +`django_db` marker's outer `atomic()` takes the `closed_in_transaction=True` path and makes the +next query raise. Same rule, same reason, as `test_run_image_evidence_cohort.py`'s own docstring. +""" + +from typing import Any, Optional + +import pytest + +from django.core.management import call_command +from django.core.management.base import CommandError + +from cardpicker.local_attribute_chip_cast import ( + BLEED_EDGE_CAST_ANONYMOUS_ID, + FRAME_STYLE_CAST_ANONYMOUS_ID, +) +from cardpicker.local_calculate_verdicts import JOIN_KEY_ANONYMOUS_ID +from cardpicker.local_layout_class_cast import LAYOUT_CLASS_CAST_ANONYMOUS_ID +from cardpicker.management.commands import run_image_evidence_cohort as cohort_command +from cardpicker.management.commands import run_pipeline as pipeline_command +from cardpicker.management.commands.run_image_evidence_cohort import ( + MANIFEST_EXTRACTOR_CURRENT_VERSIONS, +) +from cardpicker.models import ( + CardPrintingTag, + CardTagVote, + ImageEvidence, + PilotRunLedger, +) +from cardpicker.tests.factories import CanonicalCardFactory, CardFactory + +# A card whose name starts with this is treated by the fetch stub as a card the CDN could not +# serve. It therefore reaches Stage D with NO evidence row, abstains everywhere, and is the only +# way a distance-0 cluster member can still be missing a vote by the time propagation runs. +FETCH_FAILS_PREFIX = "FETCHFAIL" + +CLUSTER_HASH = 0x0F0F0F0F0F0F0F0F +LONE_HASH = 0x1234567812345678 + +STAGE_ZERO_VINTAGE = { + "remote_updated_at": "2026-07-30T00:00:00.000Z", + "cache_path": "/tmp/default_cards.jsonl", + "cache_age_days": 0.5, + "refreshed": False, + "import_stats": None, +} + + +class _SyncPoolStub: + """Runs submitted work inline, so the pooled Stage C engine is exercised in-process.""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + pass + + def __enter__(self) -> "_SyncPoolStub": + return self + + def __exit__(self, *args: Any) -> None: + return None + + def submit(self, fn: Any, *args: Any, **kwargs: Any) -> Any: + from concurrent.futures import Future + + future: "Future[Any]" = Future() + try: + future.set_result(fn(*args, **kwargs)) + except BaseException as exc: # noqa: BLE001 - mirror a real pool's failure surfacing + future.set_exception(exc) + return future + + +def _stub_fetch(card_id: int, stop_event: Any, run_id: str = "", dry_run: bool = False) -> Any: + from cardpicker.models import Card + + card = Card.objects.get(pk=card_id) + if card.name.startswith(FETCH_FAILS_PREFIX): + return cohort_command._FetchOutcome(card_id=card_id, outcome="fetch_failed", card_name=card.name) + return cohort_command._FetchOutcome( + card_id=card_id, + content_hash=card.content_phash, + md5_checksum=f"md5-{card_id}", + sha256_checksum=f"sha-{card_id}", + image_bytes=b"not-really-an-image", + card_name=card.name, + ) + + +def _stub_compute( + card_id: int, + content_hash: Optional[int], + image_bytes: Optional[bytes], + fetch_latency_ms: float, + dry_run: bool, + run_id: str, + profile: bool = False, + short_circuit: Optional[bool] = None, + known_set_codes: Optional[frozenset] = None, + md5_checksum: Optional[str] = None, + sha256_checksum: Optional[str] = None, + card_artist_names: tuple = (), +) -> tuple: + """ + Stands in for the extractor stage by writing the `ImageEvidence` row a real extraction of a + well-behaved card would have written: a complete extractor manifest, a collector line that + resolves against the fixture's `CanonicalCard`, and the three fields the attribute chips read + (`layout_class` -> border, `collector_line_collector_number` + `illus_anchor_fired` -> frame + style, `bleed_class` -> bleed edge). + """ + if not dry_run: + ImageEvidence.objects.update_or_create( + card_id=card_id, + defaults=dict( + content_hash=content_hash 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, + # Chip inputs. `trimmed` is the only bleed reading that casts a vote (the caster is + # negative-only by design), and `black` is one of BORDER_COLOR_TO_TAG's keys. + layout_class="black", + bleed_class="trimmed", + bleed_diff_mm=0.5, + illus_anchor_fired=True, + ), + ) + return card_id, "ok", None, False + + +@pytest.fixture(autouse=True) +def _no_network(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(cohort_command, "ThreadPoolExecutor", _SyncPoolStub) + monkeypatch.setattr(cohort_command, "ProcessPoolExecutor", _SyncPoolStub) + monkeypatch.setattr(cohort_command, "_fetch_one_card", _stub_fetch) + monkeypatch.setattr(cohort_command, "_compute_one_card", _stub_compute) + monkeypatch.setattr( + pipeline_command, + "run_stage_zero_freshness", + lambda **kwargs: dict(STAGE_ZERO_VINTAGE), + ) + + +@pytest.fixture +def cohort(db: Any) -> dict[str, Any]: + """ + Three cards and the reference row they resolve against. + + `representative` and `absorbed` share one `content_phash`, so they form a distance-0 cluster; + `compute_exact_match_clusters` makes the LOWEST pk the representative, which is why the + representative is created first. `absorbed`'s name makes the fetch stub fail it, so it reaches + Stage D with no evidence and abstains - leaving it as the one card in the fixture whose only + possible vote is a propagated one. + """ + call_command("seed_default_tags") + call_command("seed_attribute_tags") + call_command("seed_sensitive_tags") + + printing = CanonicalCardFactory(name="Some Card", expansion__code="mom", collector_number="158") + representative = CardFactory(name="Some Card", content_phash=CLUSTER_HASH) + absorbed = CardFactory(name=f"{FETCH_FAILS_PREFIX} Some Card", content_phash=CLUSTER_HASH) + lone = CardFactory(name="Some Card", content_phash=LONE_HASH) + return {"printing": printing, "representative": representative, "absorbed": absorbed, "lone": lone} + + +def _run(*argv: str, run_id: str = "test-monolith") -> None: + call_command("run_pipeline", "--run-id", run_id, *argv) + + +# ================================================================================================== +# THE END-TO-END PASS +# ================================================================================================== +@pytest.mark.django_db(transaction=True) +class TestEndToEndPass: + def test_a_bare_invocation_runs_every_stage_and_produces_rows( + self, cohort: dict[str, Any], capsys: pytest.CaptureFixture + ) -> None: + """ + The whole point of the command, in one assertion block: no flag is required, and each of + the seven stages both RAN and left evidence that it ran. + """ + _run() + out = capsys.readouterr().out + + ledger = PilotRunLedger.objects.get(command="run_pipeline", run_id="test-monolith-pipeline") + assert ledger.status == PilotRunLedger.Status.COMPLETED + counters = ledger.counters + + # Stage 0 ran and its bulk-file VINTAGE is on the run's own ledger row, so this run's + # conclusions can be dated. + assert counters["stage_0"]["remote_updated_at"] == STAGE_ZERO_VINTAGE["remote_updated_at"] + + # Stage E preflight ran. + assert "operating envelope clear" in out + + # Stage C ran, stamped its rows with THIS run_id, and filled bleed_diff_mm on the way - + # the field that was 97.9% NULL and that the brief says needs no separate backfill. + evidence = ImageEvidence.objects.filter(run_id="test-monolith") + assert evidence.count() == 2 # the two fetchable cards + assert evidence.filter(bleed_diff_mm__isnull=True).count() == 0 + + # Stage D ran: the join-key calculator resolved the fixture's collector line. + 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 + + # 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. + for identity in ( + LAYOUT_CLASS_CAST_ANONYMOUS_ID, + FRAME_STYLE_CAST_ANONYMOUS_ID, + BLEED_EDGE_CAST_ANONYMOUS_ID, + ): + assert CardTagVote.objects.filter( + run_id="test-monolith", anonymous_id=identity + ).exists(), f"no chip votes for {identity}" + + # Clustering ran and found the distance-0 pair. + assert counters["clustering"]["cluster_count"] == 1 + assert counters["clustering"]["cards_absorbed_into_clusters"] == 1 + + # The fidelity gate ran and is clear - a machine vote alone must never resolve a card. + assert counters["fidelity_gate"]["violations"] == 0 + + # channel_report ran at the end and its verdict is REPORTED, not folded into the exit. + assert "CHANNEL REPORT" in out + assert "channel_report exit=" in out + + def test_cluster_propagation_gives_an_unfetched_member_its_groups_verdict(self, cohort: dict[str, Any]) -> None: + """ + THE HIGHEST-VALUE CARRIED FEATURE. `absorbed` never fetched, never extracted and never + reached any Stage D calculator with evidence - so on its own it has no printing vote at + all. Because its stored `content_phash` is bit-identical to `representative`'s, it + inherits `representative`'s verdict, under the same identity, with no fetch of its own. + """ + _run() + + representative_vote = CardPrintingTag.objects.get( + card_id=cohort["representative"].pk, + anonymous_id=JOIN_KEY_ANONYMOUS_ID, + run_id="test-monolith", + ) + absorbed_vote = CardPrintingTag.objects.get(card_id=cohort["absorbed"].pk, anonymous_id=JOIN_KEY_ANONYMOUS_ID) + + assert absorbed_vote.printing_id == representative_vote.printing_id + assert absorbed_vote.confidence == representative_vote.confidence + assert absorbed_vote.run_id == "test-monolith" + # It genuinely never had evidence - the vote cannot have come from a calculator. + assert not ImageEvidence.objects.filter(card_id=cohort["absorbed"].pk).exists() + + ledger = PilotRunLedger.objects.get(command="run_pipeline", run_id="test-monolith-pipeline") + assert ledger.counters["clustering"]["votes_propagated"] == 1 + + def test_the_run_id_is_printed_at_the_start_and_at_the_end( + self, cohort: dict[str, Any], capsys: pytest.CaptureFixture + ) -> None: + """The run's identity is the only thing that marks its output, so it must be impossible to + miss in a terminal an operator scrolls back through later.""" + _run(run_id="shakedown-01") + out = capsys.readouterr().out + assert "MONOLITH RUN run_id=shakedown-01" in out + assert "MONOLITH DONE run_id=shakedown-01" in out + assert "To resume this run after a stop: --run-id shakedown-01" in out + + def test_a_default_run_id_is_self_describing_not_an_opaque_timestamp(self, cohort: dict[str, Any]) -> None: + call_command("run_pipeline") + row = PilotRunLedger.objects.get(command="run_pipeline") + assert row.run_id.startswith("monolith-") + assert row.run_id.endswith(pipeline_command.LEDGER_RUN_ID_SUFFIX) + + def test_a_fresh_run_id_redoes_stage_c_from_scratch(self, cohort: dict[str, Any]) -> None: + """ + The from-scratch default, at the seam that used to break it: Stage C's resume filter is + run-scoped (PR #645), so a second run under a NEW run_id re-extracts every card rather + than skipping cards a previous run finished. + """ + _run(run_id="run-a") + assert ImageEvidence.objects.filter(run_id="run-a").count() == 2 + + _run(run_id="run-b") + assert ImageEvidence.objects.filter(run_id="run-b").count() == 2 + assert ImageEvidence.objects.filter(run_id="run-a").count() == 0 # re-stamped, not skipped + + def test_a_failure_marks_the_ledger_row_failed_with_a_reason( + self, cohort: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + def _boom(*args: Any, **kwargs: Any) -> None: + raise RuntimeError("stage d exploded") + + monkeypatch.setattr(pipeline_command, "_run_stage_d", _boom) + with pytest.raises(RuntimeError): + _run() + row = PilotRunLedger.objects.get(command="run_pipeline", run_id="test-monolith-pipeline") + assert row.status == PilotRunLedger.Status.FAILED + assert "stage d exploded" in row.counters["failure_reason"] + + +# ================================================================================================== +# THE MUTATION TABLE - unwire a stage, go red +# ================================================================================================== +@pytest.mark.django_db(transaction=True) +class TestUnwiringAStageIsCaught: + """ + Each case below simulates the exact regression this suite exists to catch: a stage that stops + being called while the command still exits 0. The disable flags are the honest stand-in for + "someone deleted the call" - they take the same code path (the stage does not run) without + needing to monkeypatch the module under test into a shape it cannot really have. + + | stage unwired | flag | the assertion that goes red | + |--------------------|-------------------------|---------------------------------------------| + | Stage 0 | --skip-freshness | no bulk-file vintage on the ledger row | + | Stage C | --skip-stage-c | no ImageEvidence rows for this run | + | Stage D | --skip-stage-d | no printing votes, no chip votes | + | attribute chips | (Stage D carries them) | covered by the Stage D row above | + | clustering | --skip-clustering | the unfetched member never gets a verdict | + | fidelity gate | --skip-gate | no gate result recorded | + | channel_report | --skip-channel-report | the report never runs | + """ + + def test_unwiring_stage_zero_loses_the_bulk_file_vintage(self, cohort: dict[str, Any]) -> None: + _run("--skip-freshness") + counters = PilotRunLedger.objects.get(command="run_pipeline", run_id="test-monolith-pipeline").counters + assert counters["stage_0"] == {"skipped": True, "reason": "--skip-freshness"} + assert "remote_updated_at" not in counters["stage_0"] + + def test_unwiring_stage_c_produces_no_evidence(self, cohort: dict[str, Any]) -> None: + _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") + assert not CardPrintingTag.objects.filter(run_id="test-monolith", anonymous_id=JOIN_KEY_ANONYMOUS_ID).exists() + for identity in ( + LAYOUT_CLASS_CAST_ANONYMOUS_ID, + FRAME_STYLE_CAST_ANONYMOUS_ID, + BLEED_EDGE_CAST_ANONYMOUS_ID, + ): + assert not CardTagVote.objects.filter(run_id="test-monolith", anonymous_id=identity).exists() + + def test_unwiring_clustering_leaves_the_unfetched_member_with_no_verdict(self, cohort: dict[str, Any]) -> None: + """The mutation that matters most: with propagation unwired the command still exits 0 and + Stage D still reports votes, but the distance-0 member silently disagrees with its own + identity group by having no verdict at all.""" + _run("--skip-clustering") + assert not CardPrintingTag.objects.filter(card_id=cohort["absorbed"].pk).exists() + + def test_unwiring_the_gate_records_no_gate_result(self, cohort: dict[str, Any]) -> None: + _run("--skip-gate") + counters = PilotRunLedger.objects.get(command="run_pipeline", run_id="test-monolith-pipeline").counters + assert counters["fidelity_gate"] == {"skipped": True} + + def test_unwiring_channel_report_never_runs_it(self, cohort: dict[str, Any], capsys: pytest.CaptureFixture) -> None: + _run("--skip-channel-report") + assert "CHANNEL REPORT" not in capsys.readouterr().out + + +# ================================================================================================== +# STAGE E - the envelope actually gates +# ================================================================================================== +@pytest.mark.django_db(transaction=True) +class TestEnvelopeGating: + def test_an_open_trip_halts_before_anything_is_written( + self, cohort: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + class _Trip: + trip_id = "envtrip-test" + bar = "host_load" + + monkeypatch.setattr(pipeline_command, "current_trip", lambda run_id=None: _Trip()) + with pytest.raises(CommandError) as excinfo: + _run() + assert excinfo.value.returncode == pipeline_command.EXIT_ENVELOPE_HALT + assert not ImageEvidence.objects.filter(run_id="test-monolith").exists() + + def test_a_fresh_breach_halts_before_anything_is_written( + self, cohort: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + class _Trip: + trip_id = "envtrip-fresh" + bar = "rss" + detail = "rss over ceiling" + + monkeypatch.setattr(pipeline_command, "current_trip", lambda run_id=None: None) + monkeypatch.setattr(pipeline_command, "check_envelope", lambda signals, run_id=None: _Trip()) + with pytest.raises(CommandError) as excinfo: + _run() + assert excinfo.value.returncode == pipeline_command.EXIT_ENVELOPE_HALT + assert not ImageEvidence.objects.filter(run_id="test-monolith").exists() + + def test_skip_envelope_lets_the_run_proceed(self, cohort: dict[str, Any], monkeypatch: pytest.MonkeyPatch) -> None: + class _Trip: + trip_id = "envtrip-test" + bar = "host_load" + + monkeypatch.setattr(pipeline_command, "current_trip", lambda run_id=None: _Trip()) + _run("--skip-envelope") + assert ImageEvidence.objects.filter(run_id="test-monolith").exists() + + +# ================================================================================================== +# STAGE 0 - once, at the front +# ================================================================================================== +@pytest.mark.django_db(transaction=True) +class TestStageZeroIsCalledOnceAtTheFront: + def test_stage_zero_runs_exactly_once_and_before_stage_c( + self, cohort: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + """ + The binding constraint inherited from `stream_full_catalog`: a refresh rewrites + `CanonicalPrintingMetadata`, which is the table Stage D's illustration deduction builds its + index from. Refreshing more than once - or after Stage C has started - would have early + and late cards deduced against different reference sets under one run_id. + """ + calls: list[str] = [] + + def _record_stage_zero(**kwargs: Any) -> dict[str, Any]: + calls.append("stage_0") + return dict(STAGE_ZERO_VINTAGE) + + def _record_fetch(*args: Any, **kwargs: Any) -> Any: + calls.append("stage_c") + return _stub_fetch(*args, **kwargs) + + monkeypatch.setattr(pipeline_command, "run_stage_zero_freshness", _record_stage_zero) + monkeypatch.setattr(cohort_command, "_fetch_one_card", _record_fetch) + + _run() + + assert calls.count("stage_0") == 1 + assert calls[0] == "stage_0" + + def test_require_fresh_is_forwarded(self, cohort: dict[str, Any], monkeypatch: pytest.MonkeyPatch) -> None: + seen: dict[str, Any] = {} + + def _capture(**kwargs: Any) -> dict[str, Any]: + seen.update(kwargs) + return dict(STAGE_ZERO_VINTAGE) + + monkeypatch.setattr(pipeline_command, "run_stage_zero_freshness", _capture) + _run("--require-fresh") + assert seen["require_fresh"] is True + + +# ================================================================================================== +# WRITE POLARITY - the monolith writes by default; only --dry-run prevents it +# ================================================================================================== +@pytest.mark.django_db(transaction=True) +class TestWritesByDefault: + """ + Owner ruling: "the eventual intention for the monolith is that default is to write and flags + are what prevents it. (opposite)". This is the inverse of every organ the monolith calls, and + the failure it guards against is the dangerous one - a pass that computes everything, logs + everything, reports success and persists nothing is indistinguishable from a working run + except by row counts, and on a 230k pass that is hours before anyone notices. + + THE ORGAN TABLE. Every write-capable organ this command reaches, its own gate, and what the + monolith passes. A mutation restoring any organ's own default must fail + `test_every_organ_persists_rows_with_no_flags` below. + + | organ | its own gate | monolith passes | + |------------------------------|-------------------------|---------------------------| + | import_scryfall_printing_... | none (always writes) | called; skipped on dry-run| + | run_image_evidence_cohort | --dry-run (write-first) | forwards --dry-run only | + | run_join_key_calculator | dry_run=True default | dry_run=False | + | run_fallback_calculator | dry_run=True default | dry_run=False | + | run_illustration_calculator | dry_run=True default | dry_run=False | + | run_slow_path_calculator | dry_run=True default | dry_run=False | + | run_layout_class_cast | dry_run=True default | dry_run=False | + | run_attribute_chip_cast | dry_run=True default | dry_run=False | + | cluster vote propagation | none (this command's) | gated on `not dry_run` | + """ + + def test_every_organ_persists_rows_with_no_flags(self, cohort: dict[str, Any]) -> None: + """A bare invocation - no flags at all - must leave rows behind from EVERY organ.""" + _run() + + assert ImageEvidence.objects.filter(run_id="test-monolith").exists(), "Stage C wrote nothing" + assert CardPrintingTag.objects.filter( + run_id="test-monolith", anonymous_id=JOIN_KEY_ANONYMOUS_ID + ).exists(), "the join-key calculator wrote nothing" + for identity in ( + LAYOUT_CLASS_CAST_ANONYMOUS_ID, + FRAME_STYLE_CAST_ANONYMOUS_ID, + BLEED_EDGE_CAST_ANONYMOUS_ID, + ): + assert CardTagVote.objects.filter( + run_id="test-monolith", anonymous_id=identity + ).exists(), f"the {identity} caster wrote nothing" + assert CardPrintingTag.objects.filter( + card_id=cohort["absorbed"].pk + ).exists(), "cluster propagation wrote nothing" + + def test_dry_run_persists_nothing_through_any_organ(self, cohort: dict[str, Any]) -> None: + """A dry run that writes through ONE organ is worse than no dry run, because it is + trusted. Nothing may reach the database through any of them.""" + _run("--dry-run") + + assert not ImageEvidence.objects.filter(run_id="test-monolith").exists() + assert not CardPrintingTag.objects.filter(run_id="test-monolith").exists() + assert not CardTagVote.objects.filter(run_id="test-monolith").exists() + assert not CardPrintingTag.objects.filter(card_id=cohort["absorbed"].pk).exists() + + def test_dry_run_still_runs_every_stage_and_reports_what_it_would_write( + self, cohort: dict[str, Any], capsys: pytest.CaptureFixture + ) -> None: + """A dry run must be a genuinely useful preview of a 230k pass, not a plan: every stage + executes and reports, and it exits 0 because it did what was asked.""" + _run("--dry-run") + out = capsys.readouterr().out + + assert "mode=DRY-RUN (writes nothing)" in out + assert "STAGE 0 skipped (--dry-run)" in out + assert "operating envelope clear" in out + assert "STAGE C:" in out + assert "STAGE D:" in out + assert "STAGE C+:" in out + assert "CHANNEL REPORT" in out + assert "MONOLITH DONE" in out + + row = PilotRunLedger.objects.get(command="run_pipeline", run_id="test-monolith-pipeline") + assert row.dry_run is True + assert row.status == PilotRunLedger.Status.COMPLETED + + def test_stage_d_is_handed_write_mode_explicitly_never_by_inheritance( + self, cohort: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + """ + The pin. `_run_stage_d`'s own `dry_run` parameter defaults to False, but the monolith must + pass it EXPLICITLY - a future edit that drops the argument would silently re-inherit + whatever that default becomes. This asserts the argument is actually sent. + """ + seen: dict[str, Any] = {} + + def _capture(batch_ids: Any, run_id: str, outcome: Any, dry_run: bool = True) -> None: + seen["dry_run"] = dry_run + + monkeypatch.setattr(pipeline_command, "_run_stage_d", _capture) + _run() + assert seen["dry_run"] is False + + seen.clear() + _run("--dry-run", run_id="test-monolith-dry") + assert seen["dry_run"] is True + + def test_a_bare_run_never_trips_the_forced_dry_run_precondition(self, cohort: dict[str, Any]) -> None: + """ + `enforce_dry_run_precondition` (issue #362) is a FORCED dry-run, not just a default, and it + would put a flag in front of the working run if it applied. Stage C arms it only for a + `--card-ids-file` write (`write_mode=(not dry_run) and bool(card_ids_file_for_scope)`), and + a bare monolith run goes down the `--limit` path - so it must be a no-op here. + """ + _run() # would raise CommandError("FORCED DRY-RUN GUARD: ...") if the guard applied + assert ImageEvidence.objects.filter(run_id="test-monolith").exists() From 6dd59f1369d1e4eb47f770580b47e924c28a03cb Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:05:24 +0000 Subject: [PATCH 3/5] Docs: the monolith, in the living pipeline and operations pages docs/identification-pipeline.md gains a 'How to run all of it' section at the front of the stage walkthrough - the sequence, that the command holds no pipeline logic of its own, and the two load-bearing defaults (writes by default, redoes everything from scratch). Stage 0's placement is noted where the reference set is introduced. docs/features/stage-e-operations.md gains the operational section: the write-polarity organ table (which organ, its own gate, what run_pipeline passes), how enforce_dry_run_precondition interacts with a write-by-default command, what a dry run actually does, Stage 0's returned bulk-file vintage, why Stage D is called explicitly rather than reached by the post_save echo, the two deliberate differences between the monolith's cluster propagation and the pilot's, exit codes, the ledger run_id suffix and why the DATA keeps the clean name, and the recorded list of ways a fresh run still inherits from an earlier one. No dated report. docs_lint.py --strict clean. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013NhYmT1PxCcyemA16dFDxN --- docs/features/stage-e-operations.md | 191 ++++++++++++++++++++++++++++ docs/identification-pipeline.md | 45 +++++++ 2 files changed, 236 insertions(+) diff --git a/docs/features/stage-e-operations.md b/docs/features/stage-e-operations.md index 337c808e2..48c8ac198 100644 --- a/docs/features/stage-e-operations.md +++ b/docs/features/stage-e-operations.md @@ -1583,6 +1583,197 @@ Nothing new — every batch gets its own `PilotRunLedger` row via is `stage-e-fullcat-b-`, the same collision-safe shape the shakedown driver adopted. +## The monolith — `run_pipeline` (2026-07-30) + +`manage.py run_pipeline` — **one command that runs the whole identification +pipeline end to end.** Owner brief: "1 click". Every stage it runs already +existed and every stage was already run separately; what did not exist was +anything that ran them together, in order, under one `run_id`. + +``` +Stage 0 Scryfall reference refresh, once, at the front +Stage E operating-envelope preflight +Stage C evidence extraction (the pooled engine) +Stage D join-key → fallback → illustration → slow-path, then the three chips +Stage C+ distance-0 cluster vote propagation +Stage E fidelity gate — machine-only resolutions must be zero +end channel_report +``` + +**It contains no pipeline logic.** Every stage is reached by importing and +calling the module that already owned it. If a future change adds an +inference, a threshold or a calculator to `run_pipeline.py`, that logic is in +the wrong file. + +### Two defaults, both load-bearing + +**It writes, and only `--dry-run` prevents that.** Owner ruling: _"the eventual +intention for the monolith is that default is to write and flags are what +prevents it. (opposite)"_. This is the inverse of every organ it calls, and +inheriting their defaults is the dangerous failure — a pass that computes +everything, logs every card, reports success and **persists nothing** is +indistinguishable from a working run except by row counts, and on a 230k pass +that is hours before anyone notices. The command therefore passes `dry_run` +explicitly at every seam: + +| organ | its own gate | what `run_pipeline` passes | +| ----------------------------------------------- | --------------------------------- | --------------------------------------------------------------------------------------------------------- | +| `import_scryfall_printing_metadata` | none — always writes | called; **skipped** under `--dry-run` (no preview mode; a refresh is a real download and a real DB write) | +| `run_image_evidence_cohort` | `--dry-run` (already write-first) | forwards `--dry-run`; nothing overridden | +| `run_join_key_calculator` | `dry_run=True` default | `dry_run=False` | +| `run_fallback_calculator` | `dry_run=True` default | `dry_run=False` | +| `run_illustration_calculator` | `dry_run=True` default | `dry_run=False` | +| `run_slow_path_calculator` | `dry_run=True` default | `dry_run=False` | +| `run_layout_class_cast` (border chip) | `dry_run=True` default | `dry_run=False` | +| `run_attribute_chip_cast` (frame + bleed chips) | `dry_run=True` default | `dry_run=False` | +| cluster vote propagation | none (this command's own write) | gated on `not dry_run` | + +All six calculators/casters are reached through +`stage_e_dispatch._run_stage_d`, which now takes `dry_run` as a parameter +rather than hard-coding `False`, so there is exactly one Stage D and both +engines use it. + +`enforce_dry_run_precondition` (the **forced** dry-run of issue #362, not just +a default) is deliberately not added to this command: a command that writes by +default has no `--write` to gate, and requiring a prior dry-run would put a +flag back in front of the working run. The one place it still applies is +inside Stage C's own `--card-ids-file` path +(`write_mode=(not dry_run) and bool(card_ids_file_for_scope)`), and +`run_pipeline` forwards `--skip-dryrun-check` for exactly that. A bare +invocation goes down the `--limit` path and never arms it. + +**A dry run is a real pass that withholds the write**, not a plan: every stage +executes, every calculator reports what it _would_ cast, clustering reports +what it _would_ propagate, and `channel_report` still runs so the preview +arrives in the shape it will be read in afterwards. Exits 0. + +**It redoes everything from scratch.** A fresh `--run-id` means no earlier run +suppresses work — Stage C's resume filter is run-scoped (PR #645) and each +calculator's own eligibility is run-scoped (PR #604). Re-passing an earlier +`--run-id` resumes that run instead. No flag is required for a working run. + +### Stage 0 — the same stage, one implementation + +Stage 0 is `stream_full_catalog`'s own freshness stage, whose body was lifted +to a module-level `run_stage_zero_freshness` so both drivers call one copy. +Everything the "Stage 0 — Scryfall freshness" section above states — once at +the start and never during the run, fail before any dispatch with exit 2, +report what it decided, `--skip-freshness` / `--require-fresh` — applies here +unchanged. + +What is new is that it **returns the bulk-file vintage** rather than only +printing it: the remote `updated_at` it compared against, the cache path, the +cache's mtime age, whether it refreshed, and the import's own +created/updated/deleted counts. `run_pipeline` records all of it under +`PilotRunLedger.counters["stage_0"]`, so a run's conclusions can be dated +against the Scryfall bulk file they were reasoned from. + +### Stage D — explicit, never by echo + +`run_pipeline` calls `stage_e_dispatch._run_stage_d` directly with +`batch_ids=None` (bulk mode). It deliberately does **not** rely on the pooled +Stage C runner's `post_save` echo into the conveyor: that route is implicit, +runs one micro-batch at a time under the streaming envelope, and is what left +the Stage D route unestablished. + +The order is load-bearing and unchanged — join-key → fallback → illustration → +slow-path, then the three attribute chips. Its asymmetry is the correctness +argument: a calculator's `run_id` narrows its OWN progress, never an UPSTREAM +verdict. Run-scoping the upstream selectors would hand downstream calculators +an empty pool while reporting success. + +### Stage C+ — distance-0 cluster vote propagation + +`local_clustering.compute_two_threshold_clusters` groups cards by stored +`content_phash`; a distance-0 cluster is a set of bit-identical images. +`local_identify_printing_tags.build_propagated_cluster_votes` (lifted out of +`run_pilot`'s closure so it has one implementation and two callers) then gives +every absorbed member its representative's printing verdict, under the same +identity, **without the member ever being fetched or computed**. Correctness +property first — identity groups must agree — throughput lever second. + +Two differences from the pilot's use of the same rule, both deliberate: + +- **What is propagated.** `run_pilot` propagates its own OCR/phash votes. The + monolith does not carry pilot OCR/phash voting (standing owner deferral; + Stage D's join-key superseded the OCR half), so it propagates **Stage D's** + printing verdicts — the votes the run actually cast. +- **What is clustered.** `run_pilot` clusters over its own + eligibility-narrowed selection pool, which makes cluster membership a + function of what earlier runs already voted on: a genuine 3-card cluster can + present as a 2-card one, and dropping the lowest-pk member changes which + card becomes representative. `run_pipeline` clusters over `Card` by stored + hash — the whole catalogue by default — which is what makes the answer + independent of run history. `--scope-stage-d-to-cohort` narrows that, and + narrows the independence with it; that is why it is not the default. + +### Exit codes + +| code | meaning | +| ---- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `0` | the pass ran to completion, or previewed one under `--dry-run`. `channel_report`'s verdict is REPORTED, never folded in. | +| `2` | **Stage 0 failed** — nothing dispatched, nothing written. | +| `3` | **operating envelope halted the run** — an open trip (no self-resume) or a fresh breach. Nothing written. | +| `7` | the pass completed but the **fidelity gate** found machine-only resolutions. Everything is written; this is a loud "read this run before trusting it", not a rollback. | + +`channel_report` runs at the end and is **never** folded into the exit status. +Expect its own exit 1 on the first run: `ZERO_DECLARATIONS` ships empty and +there are known-silent channels. That is the instrument working. + +### Ledger convention + +One `PilotRunLedger` row, `command="run_pipeline"`, `run_id` = +`-pipeline`. The suffix is not cosmetic: `PilotRunLedger.run_id` is +UNIQUE, and Stage C is delegated to `run_image_evidence_cohort`, which writes +its own row under the run identity it is given. Two rows cannot share one, so +the choice is which of the two gets the clean name — this command's summary +row, or every data row the run produces. **The data wins:** `channel_report` +scopes a channel's run-column counts by the `run_id` on the rows, and a Stage +C whose evidence carried a different `run_id` from the votes would read as a +silent channel, which is exactly the reading "nothing is culled" exists to +make impossible. + +The row's `counters` carry one key per stage — `stage_0` (including the bulk +file vintage), `stage_c`, `stage_d`, `clustering`, `fidelity_gate`, +`channel_report`, `elapsed_s`. + +### What a fresh run still inherits from an earlier one + +The from-scratch default covers **selection** — which cards get looked at. +It does not cover every route by which a prior run's rows reach a fresh run's +**answer**, and the following were found while wiring this command. They are +recorded, not fixed: this command's first run is how the pipeline's real state +gets measured, and changing what it measures while building it would make that +first reading untrustworthy. + +- **`run_name_frequency_elimination` is a census predicate and it leaks.** Its + "exactly one unresolved eligible card for this name" test counts over + `_eligible_base_queryset` called with no `run_id`, so the pool is depleted by + its own prior votes. A name that gains a second card after run 1 can have + that second card voted on in run 2 while an empty-vote catalogue would + abstain — a fresh wrong assertion, not a carried-forward one. It has never + run in production and `run_pipeline` does **not** wire it. +- **The attribute-chip and layout casters have no `run_id` scoping at all** + and no value-comparison split, so a card whose evidence is later re-extracted + can never have its chip re-read under a new run. Monotonic suppression, not a + false fire. +- **`_split_new_printing_tag_votes` compares VALUES**, so a recomputed + identical verdict is skipped and the original row survives **with its + original `run_id`**. Correct and desirable — but it means a "fresh" run's + live output legitimately contains rows stamped with an older run, and + `filter(run_id=R)` returns only what R _changed_, not what R concluded. +- **Fallback / slow-path eligibility read upstream verdicts unscoped**, by + design and with the argument recorded in-source: scoping them would hand a + downstream calculator an empty pool while reporting success. +- **Only `CardPrintingTag` is archived on supersession.** `CardTagVote` (all + three chip families) and `CardIllustrationVote` are deleted outright — + `models.vote_archive_model` maps only that one model. So the printing grain + stays diffable across generations via `--generation-diff`; the tag and + illustration grains do not. `CardIllustrationVote` is additionally the only + production caller of `purge_stale_machine_votes` that bypasses + `vote_write.purge_and_write_votes`, so it never records + `superseded_by_run_id` either. + ## Phase 3 (not yet built) Informal shorthand, not a brief-defined phase number — see diff --git a/docs/identification-pipeline.md b/docs/identification-pipeline.md index ffd461652..0d4b883a9 100644 --- a/docs/identification-pipeline.md +++ b/docs/identification-pipeline.md @@ -96,11 +96,56 @@ source of truth for gate status, for how many cards are currently sitting at each stage. Absolute counts return to this diagram once real user confirmations start accumulating in volume. +## How to run all of it: `run_pipeline` (2026-07-30) + +Everything below this heading used to be run **one command at a time**, in an +order carried in an operator's head. `manage.py run_pipeline` is that order, +executed: + +``` +Stage 0 Scryfall reference refresh, once, at the front +Stage E operating-envelope preflight +Stage C evidence extraction (the pooled engine) +Stage D join-key → fallback → illustration → slow-path, then the three chips +Stage C+ distance-0 cluster vote propagation +Stage E fidelity gate — machine-only resolutions must be zero +end channel_report +``` + +It contains **no pipeline logic of its own**. Each stage below is reached by +importing and calling the thing that already owned it; the command is +sequencing, `run_id` threading and error handling. Everything it writes is +stamped with one `run_id`, and that `run_id` is the only thing identifying a +run's output — there is no test mode, no provisional marker and no separate +table. + +Two defaults are load-bearing and are the opposite of every command it calls: + +- **It writes.** A bare `manage.py run_pipeline` is a complete, working run + that persists rows. `--dry-run` is the only thing that prevents the write, + and a dry run still executes every stage and reports what it _would_ write. + Every Stage D calculator and attribute-chip caster defaults to + `dry_run=True`, so `run_pipeline` passes the flag explicitly at every seam — + inheriting those defaults would compute a whole pass and persist nothing + while every log line reported success. +- **It redoes everything from scratch.** A fresh `--run-id` means no prior + run suppresses work: Stage C's resume filter is run-scoped and so is each + calculator's own eligibility. Flags narrow; nothing is required to get a + working run. Re-passing an earlier `--run-id` resumes that run instead. + +Operational detail — flags, exit codes, what a dry run reports, and which +prior-run influences survive a fresh `run_id` — is in +[`features/stage-e-operations.md`](features/stage-e-operations.md). + ## What exists before anything runs - A **Card row**: name, source drive, and a content phash of the image. - The **reference set**: every real printing of every card name (CanonicalCard / CanonicalExpansion, from Scryfall) — set code, collector number, denominator. + Refreshed by **Stage 0**, at the front of a `run_pipeline` pass and never + during one: Stage D's illustration deduction builds its matching index from + exactly the table a refresh rewrites, so refreshing mid-pass would have early + and late cards deduced against different reference sets under one `run_id`. - **No pixels stored, ever.** Images are fetched transiently, read, discarded. ## Stage C — evidence extraction (`run_image_evidence_cohort`) From ae6cfc5614c2f3f6011ea5b81274be810b7a09d2 Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:19:53 +0000 Subject: [PATCH 4/5] Strengthen two end-to-end assertions that survived their own mutation The fidelity-gate and channel_report checks both passed with the stage unwired: 'violations == 0' is what you get whether the gate found nothing or was never called, and the CHANNEL REPORT banner is printed by run_pipeline itself either way. Now assert a string only the stage itself can emit - 'FIDELITY GATE: clear over N cards' and channel_report's own roster family title. Both mutations now go red. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013NhYmT1PxCcyemA16dFDxN --- MPCAutofill/cardpicker/tests/test_run_pipeline.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/MPCAutofill/cardpicker/tests/test_run_pipeline.py b/MPCAutofill/cardpicker/tests/test_run_pipeline.py index 1ad27ca83..88692268e 100644 --- a/MPCAutofill/cardpicker/tests/test_run_pipeline.py +++ b/MPCAutofill/cardpicker/tests/test_run_pipeline.py @@ -249,12 +249,19 @@ def test_a_bare_invocation_runs_every_stage_and_produces_rows( assert counters["clustering"]["cluster_count"] == 1 assert counters["clustering"]["cards_absorbed_into_clusters"] == 1 - # The fidelity gate ran and is clear - a machine vote alone must never resolve a card. + # The fidelity gate ran, INSPECTED CARDS, and is clear - a machine vote alone must never + # resolve a card. Asserting the count alone would survive the gate never being called at + # all, since "no violations" and "never looked" produce the same zero. assert counters["fidelity_gate"]["violations"] == 0 + assert "FIDELITY GATE: clear over " in out # channel_report ran at the end and its verdict is REPORTED, not folded into the exit. + # The banner and the exit line are printed by THIS command either way, so they cannot + # distinguish "the report ran" from "the call was deleted"; the roster family title is + # emitted only by channel_report itself. assert "CHANNEL REPORT" in out assert "channel_report exit=" in out + assert "VOTE CHANNELS" in out def test_cluster_propagation_gives_an_unfetched_member_its_groups_verdict(self, cohort: dict[str, Any]) -> None: """ From 30fec7deebcc52517ccbce44e28302cc8fd0b497 Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:33:56 +0000 Subject: [PATCH 5/5] Reset the process-global fetch-outcome window in the monolith's own tests test_run_pipeline.py passed alone and halted every test at the envelope preflight inside the full suite: stage_e_dispatch._window is a process-global rolling window that _sample_envelope_signals reads, and four fetch failures recorded by test_stage_e_dispatch.py earlier in the same pytest process tripped the fetch_failure_rate bar at 4/4. Same autouse reset fixture, same reasoning, as that file already carries. Full cardpicker/tests/: 3541 passed, 8 skipped. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013NhYmT1PxCcyemA16dFDxN --- MPCAutofill/cardpicker/tests/test_run_pipeline.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/MPCAutofill/cardpicker/tests/test_run_pipeline.py b/MPCAutofill/cardpicker/tests/test_run_pipeline.py index 88692268e..3c9f192fd 100644 --- a/MPCAutofill/cardpicker/tests/test_run_pipeline.py +++ b/MPCAutofill/cardpicker/tests/test_run_pipeline.py @@ -37,6 +37,7 @@ from django.core.management import call_command from django.core.management.base import CommandError +from cardpicker import stage_e_dispatch from cardpicker.local_attribute_chip_cast import ( BLEED_EDGE_CAST_ANONYMOUS_ID, FRAME_STYLE_CAST_ANONYMOUS_ID, @@ -157,6 +158,20 @@ def _stub_compute( return card_id, "ok", None, False +@pytest.fixture(autouse=True) +def _reset_fetch_failure_window(monkeypatch: pytest.MonkeyPatch) -> None: + """ + `stage_e_dispatch._window` is a PROCESS-GLOBAL rolling fetch-outcome window, and + `_sample_envelope_signals` - which this command's envelope preflight calls - reads it. Without + this reset, fetch failures recorded by any earlier test in the same pytest process leak in and + trip the `fetch_failure_rate` bar here: running this file alone passed while running it inside + the full suite halted every test at the preflight with 4/4 failures inherited from + `test_stage_e_dispatch.py`. Same fixture, same reasoning, as that file's own + `_reset_fetch_failure_window`. + """ + monkeypatch.setattr(stage_e_dispatch, "_window", stage_e_dispatch._FetchOutcomeWindow()) + + @pytest.fixture(autouse=True) def _no_network(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(cohort_command, "ThreadPoolExecutor", _SyncPoolStub)