From 1a3d2d201e9e3a68f83a7b47584ece183b31edf5 Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Thu, 30 Jul 2026 16:22:01 +0000 Subject: [PATCH] Three owner-ruled fixes: md5 group as one unit, name-frequency soundness + census leak, mid-pass envelope re-sampling FIX 1 - THE MD5 GROUP BEHAVES AS ONE UNIT THROUGH THE MONOLITH Owner: "the md5 dedupe should only fetch each identical image once across sources and then apply votes to the entire group as the fetched card passes through the monolith." The fetch half already existed (`evidence_transfer`, keyed on `Card.md5_checksum`). The vote half existed only on the phash distance-0 key, so the set that got a fetch saved and the set that got a vote propagated were DIFFERENT SETS - byte-identical files always share a phash, but files sharing a phash are not necessarily byte-identical. CHECKED BEFORE BUILDING, as instructed: propagation is NOT redundant. `evidence_transfer` gives every md5 sibling its own ImageEvidence row with byte-identical extractor values, so it is reasonable to ask whether each member already reaches the same conclusion independently. It does not, structurally: a Stage D printing deduction is not a function of the evidence row alone. `_resolve_candidates_for_card` keys the candidate list on `Card.name`, and md5-identical uploads from different sources routinely carry different names. Members also differ on per-card eligibility. `test_the_unfetched_twin_has_no_verdict_of_its_own_without_propagation` falsifies the "N independent deductions already agree" hypothesis on the fixture rather than arguing it. Stage C+ now runs TWO tiers through ONE propagation engine (`_propagate_over_groups`), which takes the grouping as a parameter and knows nothing about how it was keyed: md5 exact identity -> shares a PRINTING vote. New. Runs first. phash d0 unchanged from PR #660. Runs second, filling only what md5 did not. Two defects in the PR #660 propagation are fixed on the way: - SOURCE VOTES WERE READ FROM REPRESENTATIVES ONLY. Stage D has no reason to reach a group's lowest pk first, so whenever it reached any other member, nothing propagated. Source votes are now read across every member, and one source per (group, identity) is chosen deterministically so two vote-holders in one group cannot generate duplicate rows inside a single write batch. - REPRESENTATIVES WERE NEVER PROPAGATION TARGETS. Same root cause, opposite direction. PROPAGATION NEVER OVERRIDES A MEMBER'S OWN INELIGIBILITY (owner constraint). A member already resolved, already confirmed to a `canonical_card`, not a CARD, or carrying a resolved custom-art/non-english tag is skipped. custom-art is the catalogue DECLARING the image is not a faithful depiction of a printing; a checksum must not overturn that. THE PHASH TIER IS LEFT IN PLACE, flagged rather than accreted. Issue #661 holds what phash grouping is FOR - the owner's direction is that it should eventually share an ILLUSTRATION (same artwork, possibly a different printing), not a printing verdict. Removing it now would itself be a behaviour change, and it is currently the only propagation reaching cards with no md5 at all (md5 is NULL for every LOCAL_FILE source by design). The `groups` parameter is the seam that tier plugs into later. FIX 2a - `run_name_frequency_elimination` NEVER LOOKED AT THE IMAGE Owner: "just because a card was printed exactly once doesn't mean that the image in our catalogue is an accurate depiction of that card, it may have a different border or another issue." Owner leaned toward adding the conjunct rather than dropping the tier; the conjunct is what shipped, and the reasoning for keeping the tier is in its own docstring. Everything the 1:1 gate checked was a COUNT. Counting establishes that IF the card depicts one of the name's printings THEN it is the uncovered one; nothing established the antecedent, and the only filters that spoke to it were the DECLARED custom-art/non-english tags - so an untagged altered border sailed through. "It is only a vote" is weaker than it sounds: #593 established a machine vote is what the question feed renders as the suggestion to confirm, and the human's click returns as a full-weight USER vote. The missing conjunct now requires the card's ALREADY-STORED evidence to be consistent with the candidate printing. NOT a new implementation: `_apply_agreement_checks`' border/frame check was lifted to `local_identify_printing_tags.printing_attribute_disagreement` and both callers now share it. That direction is forced - `local_calculate_verdicts` imports `local_identify_printing_tags`, never the reverse. Sharing also inherits PR #656's `artist_ocr` gate for free, which is the half a second copy would most likely have got wrong. NO STORED EVIDENCE MEANS ABSTAIN. This module's "missing data is not evidence" rule protects a match from being VETOED by silence; here silence is being asked to ESTABLISH something, so it points the other way. Counted separately from mismatches so the cost is legible. FIX 2b - THE CENSUS LEAK (a fresh wrong positive, not a stale vote) `_eligible_base_queryset(NAME_FREQUENCY_ANONYMOUS_ID)` was called with no `run_id`, making its "exclude cards already carrying this calculator's vote" LIFETIME. The gate is a COUNT over exactly that population, so the calculator was taking a census over a pool it permanently shrinks itself: run 1 votes on a card, a second upload of that name arrives, and run 2 sees one unresolved card where there are really two - and votes. Nothing about the second card changed; only the size of the population the gate counts. `compute_covered_printing_pks()` stays catalogue-wide and unscoped, deliberately: "covered" is a fact about the world, not about this calculator's progress. `run_pilot`'s `select_candidates` and `count_below_resolution_floor` are LEFT UNSCOPED - neither gates on a count over the returned population, so neither has this defect. Stated in `_eligible_base_queryset`'s docstring so the asymmetry is visible from the function rather than only from its callers. FIX 3 - THE MONOLITH RE-SAMPLES THE ENVELOPE MID-PASS Owner: "host resampling is likely required (for steps that aren't fetch) as the same monolith will run for small datasets and large ones so needs to fit the available compute appropriately." PR #660 checked the envelope ONCE, before Stage C. `_EnvelopeSentry` now re-samples at every stage seam: after Stage C, between each of Stage D's calculators/casters (via a new OPTIONAL `envelope_check` callback on `stage_e_dispatch._run_stage_d`, defaulting to None so the conveyor is byte-identical), and before each Stage C+ tier. Sample counts land on the ledger. HALT SEMANTICS PRESERVED. A breach still persists an EnvelopeTrip, still exits 3, still needs `resolve_envelope_trip` - no self-resume, and NOT converted to a throttle (that is rate pressure's channel, beneath Stage C, PR #644). A mid-pass halt message differs from the preflight's: rows already written STAY written, and it names the `--run-id` to resume with. Interval-gated at 60s so the check cannot become its own load. The number is derived, not tuned: the host-load bar reads the ONE-MINUTE load average, so sampling faster re-reads a number that has not finished moving. RESIDUAL, reported not hidden: the seams are BETWEEN calculators, not inside them. Closing that gap means threading a callback into each of seven calculators' own batch loops - a real refactor of shared code, deliberately not done here. DELIBERATE DUPLICATION, WITH A TRIPWIRE. `_members_eligible_for_a_propagated_vote` expresses four catalogue-level facts `_eligible_base_queryset` also expresses. It does not call that function (which bundles workload rules wrong for a propagation target) and that function could not be refactored to expose them (its own docstring records that tests and `stream_backstop_sweep` assert against its COMPILED SQL). `TestPropagationEligibilityMatchesTheBaseQueryset` fails if the two ever disagree. VERIFICATION - mutation red, restore green (7 mutants, all red; 263 tests green restored): M1 md5 tier returns no groups 3 failed M2 source votes read from representatives only 2 failed M3 propagation ignores member ineligibility 1 failed M4 envelope re-sampling reverted to preflight 2 failed M5 visual conjunct never disagrees 1 failed M6 no-evidence no longer abstains 1 failed M7 propagation eligibility drops the tag excludes 1 failed (the tripwire) Suites: test_run_pipeline, test_local_identify_printing_tags, test_local_calculate_verdicts, test_stage_e_dispatch - 520+ tests, all green. No model changes, so no migration. Docs: living pages only, no dated report - docs/identification-pipeline.md (Stage C+ md5 section), docs/features/printing-tags.md (both name-frequency fixes), docs/features/stage-e-operations.md (mid-pass re-sampling). Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013NhYmT1PxCcyemA16dFDxN --- .../cardpicker/local_calculate_verdicts.py | 82 +-- .../local_identify_printing_tags.py | 248 ++++++++- .../management/commands/run_pipeline.py | 506 +++++++++++++++--- MPCAutofill/cardpicker/stage_e_dispatch.py | 43 +- .../test_local_identify_printing_tags.py | 211 +++++++- .../cardpicker/tests/test_run_pipeline.py | 283 +++++++++- docs/features/printing-tags.md | 69 +++ docs/features/stage-e-operations.md | 52 ++ docs/identification-pipeline.md | 44 +- 9 files changed, 1397 insertions(+), 141 deletions(-) diff --git a/MPCAutofill/cardpicker/local_calculate_verdicts.py b/MPCAutofill/cardpicker/local_calculate_verdicts.py index 13579f3a5..bb28884e1 100644 --- a/MPCAutofill/cardpicker/local_calculate_verdicts.py +++ b/MPCAutofill/cardpicker/local_calculate_verdicts.py @@ -416,16 +416,18 @@ FALLBACK_CONFIDENCE_SINGLE_EVIDENCE, SYMBOL_DISTANCE_THRESHOLD, SYMBOL_MARGIN, - classify_frame_style, filter_by_border_color, - frame_style_is_consistent, match_artist, render_set_symbol, ) from cardpicker.local_identify_printing_tags import ( + ATTRIBUTE_BORDER_MISMATCH, + ATTRIBUTE_FRAME_MISMATCH, + FRAME_CHECK_REQUIRED_EXTRACTOR_KEYS, CandidateNameIndex, CandidatePrinting, generate_run_id, + printing_attribute_disagreement, ) # IMPORTED, NOT DUPLICATED AS A LITERAL (2026-07-30). This module's convention elsewhere is to @@ -571,7 +573,12 @@ # 2026-07-29 composition audit found, is the one that had to be gated: it is the only one whose # missing-data degradation is STRICT, and the skip reason it produces is not rescannable, so its # wrong answer is permanent for that content hash. -FRAME_CHECK_REQUIRED_EXTRACTOR_KEYS = frozenset({"collector_line_ocr", "artist_ocr"}) +# +# RE-EXPORTED, NOT REDEFINED (2026-07-30). The value now lives beside the check it gates, in +# `local_identify_printing_tags.printing_attribute_disagreement`, because that check acquired a +# SECOND caller (`run_name_frequency_elimination`) and a requirement defined twice is a +# requirement that can be raised in one place and not the other. The name stays exported from this +# module because that is where every existing reader looks for it; it is an alias, never a copy. # THE SET-CODE LEXICON GATE (module docstring) - a parsed `set_code` that matches no # `CanonicalExpansion.code` at all, same permanent-conclusion category as "no-text"/"ambiguous" @@ -801,59 +808,24 @@ def _apply_agreement_checks( metadata = getattr(canonical, "printing_metadata", None) if canonical is not None else None if metadata is not None: - if evidence.layout_class and metadata.border_color and evidence.layout_class != metadata.border_color: - # THE BORDER AGREEMENT VETO (module docstring) - layout_class mirrors - # local_fallback.classify_border_color's own return convention ("black"/"white"/ - # "silver"/"borderless"), the SAME value space Scryfall's own border_color field uses - # (confirmed via BORDER_COLOR_TO_TAG's own key set), so a direct string comparison is - # correct - no value-to-class remapping needed, unlike frame below. - return JoinKeyVerdict(card_id=card_id, skip_reason=JOIN_KEY_BORDER_MISMATCH_SKIP_REASON, detail=detail) - - # frame_class is re-derived here (not read from a stored ImageEvidence field - no such - # field exists) via the SAME two OCR-derived inputs local_identify_printing_tags.py's own - # live-pilot pass already uses to compute it: whether a collector NUMBER was parsed - # (post-2003 templates print one; pre-M15 templates never do) and whether the "Illus." - # anchor fired (artist_ocr's own byproduct). PROTECTED CORE call, not a reimplementation. - # - # GATED ON `artist_ocr` HAVING ACTUALLY RUN (2026-07-30, closing the 2026-07-29 composition - # audit's §5 second row). This calculator's eligibility query filters on - # `extractor_versions__has_key="collector_line_ocr"` and then reads SIX extractors' fields - # ungated. Most of those degrade PERMISSIVELY when their extractor never ran - a blank - # legal line reads as "nothing to compare", a null `image_is_truncated` reads as "not - # truncated" - and a permissive degradation is recoverable, because the human-backed - # consensus gate still stands between it and any resolution. + # THE BORDER AND FRAME AGREEMENT VETOES (module docstring), both of them, in one shared + # call. The implementation moved to `local_identify_printing_tags. + # printing_attribute_disagreement` on 2026-07-30 - unchanged in behaviour, and moved for a + # specific reason rather than for tidiness: `run_name_frequency_elimination` deduces a + # printing purely by COUNTING, with no look at the image at all, and the owner-ruled fix + # for that unsoundness is to require this exact cross-check. Two implementations of a + # check whose frame half has a STRICT missing-data degradation (PR #656, and that + # function's own docstring) is precisely the shape that has drifted three times in this + # project, so there is now ONE implementation and two callers. # - # THIS ONE DEGRADES STRICT, WHICH IS WHY IT IS THE ONE FIXED FIRST. `illus_anchor_fired` is - # NULLABLE, and `bool(None)` is `False`, which is indistinguishable from "artist_ocr ran - # and found no anchor". With no collector number either, `classify_frame_style` then - # returns "modern" for a card it has no anchor evidence about at all - and a genuine - # OLD-frame printing is vetoed `frame-mismatch`. That reason is deliberately NOT in - # `JOIN_KEY_RESCANNABLE_SKIP_REASONS`, so the wrong conclusion is PERMANENT for that - # content hash: the card never becomes eligible again, and no later Stage C pass can undo - # it. A wrong answer nothing can revisit is strictly worse than a missing one. - # - # So an absent `artist_ocr` skips the frame check entirely rather than evaluating it on - # invented input. That is not a new rule - it is the "missing data is not evidence" rule - # this function's own docstring already states, and which the copyright-year check and the - # `metadata is None` case above already follow. The card keeps its match at its - # already-computed confidence, and once Stage C fills `artist_ocr` in, the check runs for - # real. `artist_ocr` is at 220,579/220,579 coverage in production today, so this changes - # nothing about the current catalogue: it removes a trap, it does not loosen a live gate. - # - # `REQUIRED_EXTRACTOR_KEYS` is the pattern `local_detect_ai_art`, `local_lands_identify` - # and `local_layout_class_cast` already use, applied here per-CHECK rather than - # per-calculator: this calculator's other checks have genuinely different key - # requirements, and one calculator-wide gate would drop cards that only ever needed the - # collector line. - if FRAME_CHECK_REQUIRED_EXTRACTOR_KEYS <= evidence.extractor_versions.keys(): - frame_class = classify_frame_style( - parsed_a_collector_number=bool(evidence.collector_line_collector_number), - illus_anchor_fired=bool(evidence.illus_anchor_fired), - ) - if not frame_style_is_consistent(frame_class, metadata.frame): - # THE FRAME AGREEMENT VETO (module docstring) - mirrors - # local_identify_printing_tags.py's own frame-mismatch-withholding exactly. - return JoinKeyVerdict(card_id=card_id, skip_reason=JOIN_KEY_FRAME_MISMATCH_SKIP_REASON, detail=detail) + # The mapping back to THIS calculator's own skip vocabulary stays here: the shared helper + # names the FINDING, each calculator names its own SKIP. The two constants below are + # unchanged and still what `question_feed` reads. + disagreement = printing_attribute_disagreement(evidence, metadata) + if disagreement == ATTRIBUTE_BORDER_MISMATCH: + return JoinKeyVerdict(card_id=card_id, skip_reason=JOIN_KEY_BORDER_MISMATCH_SKIP_REASON, detail=detail) + if disagreement == ATTRIBUTE_FRAME_MISMATCH: + return JoinKeyVerdict(card_id=card_id, skip_reason=JOIN_KEY_FRAME_MISMATCH_SKIP_REASON, detail=detail) # THE COPYRIGHT-YEAR ERA CHECK (module docstring) - reuses the SAME metadata row the # border/frame checks above already fetched, no second query. Skipped entirely (not a diff --git a/MPCAutofill/cardpicker/local_identify_printing_tags.py b/MPCAutofill/cardpicker/local_identify_printing_tags.py index c0f259553..343f88b7d 100644 --- a/MPCAutofill/cardpicker/local_identify_printing_tags.py +++ b/MPCAutofill/cardpicker/local_identify_printing_tags.py @@ -75,7 +75,7 @@ from concurrent.futures import ThreadPoolExecutor from contextlib import nullcontext from dataclasses import dataclass, field -from typing import Iterable, Literal, Optional, cast +from typing import Any, Iterable, Literal, Optional, cast from PIL import Image @@ -504,14 +504,30 @@ def _eligible_base_queryset( equivalent. `run_id=None` (the default) is EXACTLY the pre-2026-07-29 behaviour, and today it is still what - most callers get. Only `local_lands_identify._land_pool_selected_cards` passes a run_id, because - lands is the one caller of this function named in the Stage D printing-channel directive. The - remaining callers - `run_pilot`'s `select_candidates`, `count_below_resolution_floor`, - `run_name_frequency_elimination` - are the OCR/phash pilot, a different workload with its own - fetch budgets and its own resume semantics, and flipping them is a separate decision with a - separate blast radius rather than a free ride on this one. That is a deliberate scoping of the - change, not an oversight, and it is recorded here so the asymmetry is visible from this - function rather than only from its callers. + most callers get. Two callers pass a run_id: `local_lands_identify._land_pool_selected_cards` + (lands is the one caller of this function named in the Stage D printing-channel directive) and + `run_name_frequency_elimination` (2026-07-30 - see the SECOND reason below, which is a + different and stronger argument than the resume one). The remaining callers - `run_pilot`'s + `select_candidates` and `count_below_resolution_floor` - are the OCR/phash pilot, a different + workload with its own fetch budgets and its own resume semantics, and flipping them is a + separate decision with a separate blast radius rather than a free ride on this one. That is a + deliberate scoping of the change, not an oversight, and it is recorded here so the asymmetry is + visible from this function rather than only from its callers. + + THE SECOND REASON TO PASS `run_id`, WHICH IS NOT ABOUT RESUMING (2026-07-30). For most callers + the own-vote exclude is a per-card idempotence checkpoint, and leaving it lifetime only costs + work that a later run skips. For a caller whose PREDICATE IS A COUNT OVER THE RETURNED + POPULATION, a lifetime exclude is not a missed vote - it is a source of FRESH WRONG POSITIVES, + because the calculator is taking a census over a population it is itself permanently shrinking. + `run_name_frequency_elimination` is exactly that caller: it votes only when a name has exactly + ONE unresolved eligible card, so every card it votes on leaves the pool forever under a + lifetime exclude, and a name that correctly ABSTAINED on run 1 (two eligible cards, so + elimination says nothing about which is which) can pass the gate on run 2 purely because run 1 + removed one of them. Its own docstring calls that count "the difference between a sound + inference and a coin flip"; a self-depleting pool turns it back into the coin flip. Any FUTURE + caller whose gate is a count over this queryset must pass `run_id` for the same reason - the + distinction to apply is the same one the BATCH SCOPING note below draws, and it points the + opposite way for `run_id` than it does for `card_ids`. THE DEDUCTIVE-BACKFILL EXCLUDE BELOW IS NEVER RUN-SCOPED, whatever `run_id` says: it is a workload choice about ANOTHER identity's votes ("don't pile a weaker vote onto a card the @@ -585,6 +601,11 @@ def _eligible_base_queryset( Narrowing this queryset to a micro-batch would silently reinterpret that count as "exactly one WITHIN THE BATCH" and cast votes for names whose other unresolved cards merely sat outside it; `compute_covered_printing_pks()` alongside it is catalog-wide in the same way. + NOTE THAT `run_id` IS THE OPPOSITE CASE AND MUST NOT BE READ OFF THIS PARAGRAPH: `card_ids` + narrows the population by a slice that has nothing to do with the question being asked, which + corrupts the count; `run_id` REMOVES a narrowing the calculator inflicted on itself in an + earlier run, which RESTORES the count to the catalog-wide one the gate is specified against. + One must not be passed here and the other must, for the same underlying reason. `run_lands_identify` (the caller scoped under issue #533's first prerequisite) is safe because its own predicate is strictly per-card - `is_lands_target` on that card's own name and candidate count - never a count over the returned population. Any further caller must be @@ -1248,6 +1269,78 @@ class AttributeReport: cards_absorbed_into_clusters: int = 0 +# THE BORDER/FRAME AGREEMENT CHECK'S OWN VOCABULARY AND EXTRACTOR REQUIREMENT (2026-07-30). +# `printing_attribute_disagreement` below is the single implementation of "does this card's +# already-stored evidence CONTRADICT this candidate printing"; these are the two answers it gives. +# The strings match `local_calculate_verdicts`' own `JOIN_KEY_BORDER_MISMATCH_SKIP_REASON` / +# `JOIN_KEY_FRAME_MISMATCH_SKIP_REASON` values, deliberately rather than incidentally - they name +# the same finding, and the per-calculator constants stay separate only because each calculator +# owns its own skip vocabulary under a different `anonymous_id`. +ATTRIBUTE_BORDER_MISMATCH = "border-mismatch" +ATTRIBUTE_FRAME_MISMATCH = "frame-mismatch" + +# `classify_frame_style` takes exactly two inputs and each comes from a DIFFERENT extractor: +# `collector_line_collector_number` from `collector_line_ocr`, `illus_anchor_fired` from +# `artist_ocr`. `local_calculate_verdicts.FRAME_CHECK_REQUIRED_EXTRACTOR_KEYS` is now an alias for +# this, so the requirement has one definition and cannot drift between callers. +FRAME_CHECK_REQUIRED_EXTRACTOR_KEYS = frozenset({"collector_line_ocr", "artist_ocr"}) + + +def printing_attribute_disagreement(evidence: Any, metadata: Any) -> Optional[str]: + """ + DOES THIS CARD'S ALREADY-STORED EVIDENCE CONTRADICT THIS CANDIDATE PRINTING? Returns the + disagreement's name (`ATTRIBUTE_BORDER_MISMATCH` / `ATTRIBUTE_FRAME_MISMATCH`) or `None` for + "nothing contradicts it". NO IMAGE FETCH, no OCR, no new extraction - every input is a field + already sitting on the row. + + Lifted verbatim out of `local_calculate_verdicts._apply_agreement_checks` (2026-07-30) so it + has ONE implementation and two callers: that function, unchanged in behaviour, and + `run_name_frequency_elimination`, which needed exactly this check and had no visual cross-check + of any kind. It lives HERE rather than beside its original caller purely because of import + direction - `local_calculate_verdicts` already imports this module and never the reverse, so + this is the only side of the pair both callers can reach without a cycle. + + "MISSING DATA IS NOT EVIDENCE" throughout, which is this check's oldest rule and the one most + easily broken by accident: + * no `metadata` sidecar at all -> nothing to compare, no disagreement. + * a blank `layout_class` or a blank `metadata.border_color` -> that half is skipped. + * `frame_style_is_consistent` returns True whenever either side is unresolved (see its own + docstring), so an unclassifiable frame never manufactures a mismatch. + + THE FRAME HALF IS GATED ON `artist_ocr` HAVING ACTUALLY RUN (PR #656), and that gate is the + single best reason this is shared rather than re-derived. `illus_anchor_fired` is NULLABLE and + `bool(None)` is `False`, indistinguishable from "artist_ocr ran and found no anchor". With no + collector number either, `classify_frame_style` then confidently answers "modern" for a card it + has no anchor evidence about at all, and a genuine OLD-frame printing gets vetoed. That is the + one degradation here that is STRICT rather than permissive, so an absent `artist_ocr` skips the + frame half entirely rather than evaluating it on invented input. A second implementation of + this check would very likely have re-introduced that trap. + + BORDER IS A DIRECT STRING COMPARISON, FRAME IS NOT. `layout_class` mirrors + `local_fallback.classify_border_color`'s return convention ("black"/"white"/"silver"/ + "borderless"), the SAME value space Scryfall's own `border_color` uses - so no remapping is + correct there. `frame` is a Scryfall frame YEAR ("1993"/"2015"/...) and must go through + `FRAME_VALUE_TO_CLASS`, which is what `frame_style_is_consistent` owns. + """ + if metadata is None or evidence is None: + return None + + layout_class = getattr(evidence, "layout_class", None) + border_color = getattr(metadata, "border_color", None) + if layout_class and border_color and layout_class != border_color: + return ATTRIBUTE_BORDER_MISMATCH + + if FRAME_CHECK_REQUIRED_EXTRACTOR_KEYS <= (evidence.extractor_versions or {}).keys(): + frame_class = local_fallback.classify_frame_style( + parsed_a_collector_number=bool(evidence.collector_line_collector_number), + illus_anchor_fired=bool(evidence.illus_anchor_fired), + ) + if not local_fallback.frame_style_is_consistent(frame_class, getattr(metadata, "frame", None)): + return ATTRIBUTE_FRAME_MISMATCH + + return None + + def build_propagated_cluster_votes( *, representative_card_id: int, @@ -1972,6 +2065,42 @@ class NameFrequencyResult: run_id: str = "" votes_written: int = 0 gate_violations: list[int] = field(default_factory=list) + # THE VISUAL CONJUNCT'S OWN ABSTENTION COUNTERS (2026-07-30). Reported separately from + # `votes_written` so the cost of the new gate is legible on the first run rather than + # inferred from a smaller total: `abstained_no_evidence` is "we have never looked at this + # image", `abstained_attribute_mismatch` is "we looked and it contradicts the printing". + # Those are very different findings and collapsing them would hide which one is doing the work. + abstained_no_evidence: int = 0 + abstained_attribute_mismatch: int = 0 + + +def _current_evidence_for(card_id: int) -> Optional[Any]: + """This card's CURRENT `ImageEvidence` row, or `None` when it has never been extracted (or its + only row is stale against the card's live hash/checksum). `current_evidence_queryset` is THE + shared definition of "current" - content hash matches the card's live `content_phash`, and a + stamped md5 agrees wherever both sides carry one - reused rather than re-expressed, since this + module would otherwise become the Nth inline copy of a rule that already got centralised once. + + Imported at CALL time, not module scope: `image_evidence` imports `local_fallback`, which + TYPE_CHECKING-imports this module, so a module-scope import here would turn a type-only cycle + into a real one. Same posture `evidence_transfer`'s own call-time import documents. + + Only reached for a name that has ALREADY passed both counting gates (measured live 2026-07-16: + 1,678 names catalogue-wide qualify), so this per-name fetch is bounded by that number rather + than by the eligible card population.""" + from cardpicker.image_evidence import current_evidence_queryset + + card = Card.objects.filter(pk=card_id).first() + if card is None: + return None + return current_evidence_queryset(card).first() + + +def _printing_metadata_for(printing_pk: int) -> Optional[Any]: + """The candidate printing's `CanonicalPrintingMetadata` sidecar, or `None` when it has none - + which `printing_attribute_disagreement` already treats as "nothing to compare".""" + canonical = CanonicalCard.objects.filter(pk=printing_pk).select_related("printing_metadata").first() + return getattr(canonical, "printing_metadata", None) if canonical is not None else None def run_name_frequency_elimination( @@ -2000,6 +2129,90 @@ def run_name_frequency_elimination( match, card_type=CARD, not deductive-backfill-covered, no custom-art/non-english tag) plus this function's own anonymous_id for idempotence, and the SAME batch-flush + gate-check pattern as run_pilot (a kill loses at most one batch; a plain re-invocation resumes cleanly). + + THE DEDUCTION MUST LOOK AT THE IMAGE (owner ruling, 2026-07-30): "just because a card was + printed exactly once doesn't mean that the image in our catalogue is an accurate depiction of + that card, it may have a different border or another issue." + + Everything the 1:1 gate above checks is a COUNT. Counting establishes that IF this card is a + depiction of one of this name's printings, THEN it must be the uncovered one. It establishes + NOTHING about the antecedent, and the only filters that spoke to it at all were the DECLARED + `custom-art` / non-English tags - so an altered border, a custom frame or a misnamed upload + that nobody had tagged yet sailed straight through and got a full-confidence structural vote. + + "IT IS ONLY A VOTE" IS NOT THE DEFENCE IT SOUNDS LIKE, which is why this was worth fixing + rather than tolerating. Issue #593 established that a machine vote is what the question feed + renders as THE SUGGESTION TO CONFIRM, and a human's click returns as a full-weight USER vote. + A visually-unverified deduction therefore becomes a one-click rubber stamp, and the human- + backed consensus gate - the thing that normally makes a wrong machine vote recoverable - is + exactly the mechanism that launders it. + + SO THE MISSING CONJUNCT IS ADDED: the card's ALREADY-STORED evidence must be consistent with + the candidate printing (`printing_attribute_disagreement` - border class and frame style, the + same check `local_calculate_verdicts._apply_agreement_checks` has always applied to the + join-key channel, now shared rather than re-derived). This is the same shape as the D1 fix: a + tier claimed a cross-check it never performed, and adding the real one made it sound. + + NO NEW FETCH, NO NEW EXTRACTION. Every input is a field already on the card's current + `ImageEvidence` row. That matters for what this function IS - a pure structural deduction that + costs no network - and it is why the conjunct is affordable at catalogue scale. + + NO STORED EVIDENCE MEANS ABSTAIN, NOT PROCEED. If the card has no current `ImageEvidence` row + at all, we have never looked at this image, so the antecedent is exactly as unestablished as it + was before - and "we have no evidence" must not read as "no evidence against". This is the one + place where this module's usual "missing data is not evidence" rule points the other way, and + deliberately: that rule protects a match from being VETOED by silence, whereas here silence is + being asked to ESTABLISH something. Counted as `abstained_no_evidence` so the size of that + population is visible on the first run rather than inferred. + + WHY NOT SIMPLY DROP THIS CALCULATOR. It has never run in production (zero `PilotRunLedger` + rows for `local_name_frequency_elimination`, ever), so nothing is contaminated and there is no + retraction to plan - dropping it would have been clean. It is kept because the deduction is + genuinely sound ONCE the antecedent is established, and elimination reaches a population the + image-based channels structurally cannot: a name whose single uncovered printing has no + distinguishing collector line to read. Deleting a sound tier because it was missing a guard is + a worse trade than adding the guard. + + THE CENSUS MUST BE RUN-SCOPED (2026-07-30). `_eligible_base_queryset` is called with THIS + run's `run_id`, and that argument is a soundness fix, not a resume convenience. The gate above + is a COUNT OVER THE POPULATION THIS QUERYSET RETURNS - "exactly one unresolved eligible card + for this name, catalog-wide" - while one of that queryset's exclusions is "cards already + carrying THIS calculator's vote". Left unscoped, that exclusion is LIFETIME, so the calculator + is taking a census over a population it is itself permanently shrinking: + + run 1 name N has TWO unresolved eligible cards, A and B. len(card_ids) != 1, so the + gate correctly ABSTAINS - elimination cannot say which of A and B is the + uncovered printing, and voting for either would be a coin flip. + (later) some other channel votes on A, or A is resolved/confirmed by a human, or this + calculator votes on A for some OTHER name it also qualifies under. + run 2 A is gone from the pool. Name N now returns exactly ONE card, B. The gate PASSES + and votes B for the printing - not because anything new was learned about B, but + because the pool was depleted underneath the question. + + That is a FRESH WRONG POSITIVE, which is a strictly worse failure than the stale-vote and + missed-vote cases the 2026-07-29 run-scoping directive was written for: nothing about B's + image, evidence or name changed between the two runs, only the size of the population the + gate counts. Passing `run_id` narrows the self-suppressing excludes to rows THIS run wrote, so + a fresh run_id restores the catalog-wide count the gate is specified against and name N + abstains on run 2 exactly as it did on run 1. Within-run resume is unaffected and still works: + re-invoking with the SAME run_id still sees this run's own already-voted cards excluded, so a + killed run picks up where it stopped rather than redoing completed batches. + + WHAT THIS DOES *NOT* CHANGE, deliberately. `compute_covered_printing_pks()` above stays + catalog-wide and un-scoped, and must: "covered" is a fact about the CATALOGUE (a confirmed + `canonical_card`, or a RESOLVED `inferred_canonical_card`), not about this calculator's + progress, and run-scoping it would make every run re-derive coverage from an empty set and + treat every printing as uncovered. The two halves of the gate are therefore scoped + DIFFERENTLY, on purpose: the coverage half asks a question about the world, the census half + asks a question about a population this calculator mutates. It also means a name whose single + uncovered printing genuinely got covered between runs drops out of the gate naturally on the + coverage half, which is why restoring the census half does not simply re-vote everything. + + `run_pilot`'s own `select_candidates` and `count_below_resolution_floor` are LEFT UNSCOPED - + neither gates on a count over the returned population (the pilot's predicate is per-card and + its selection is a fetch-budget ordering; the floor count is a report metric), so neither has + the defect this fixes, and flipping them is the separate decision with the separate blast + radius `_eligible_base_queryset`'s own docstring already records. """ # a separate invocation entrypoint from run_pilot (own management command, own gate-check # loop) - generates its OWN run_id, never one shared with a run_pilot() call. @@ -2010,7 +2223,10 @@ def run_name_frequency_elimination( cards_by_name: dict[str, list[int]] = collections.defaultdict(list) for card_id, name in ( - _eligible_base_queryset(NAME_FREQUENCY_ANONYMOUS_ID) + # `run_id` PASSED (2026-07-30) - THE CENSUS LEAK. See this function's own docstring's + # "THE CENSUS MUST BE RUN-SCOPED" section for why this one argument is a correctness fix + # and not a resume convenience. + _eligible_base_queryset(NAME_FREQUENCY_ANONYMOUS_ID, run_id=run_id) .values_list("pk", "name") .order_by("pk") .iterator(chunk_size=5000) @@ -2052,6 +2268,18 @@ def flush() -> None: if len(uncovered) != 1: continue + # THE MISSING CONJUNCT (owner ruling, 2026-07-30) - see this function's own docstring's + # "THE DEDUCTION MUST LOOK AT THE IMAGE" section. Everything above is a COUNT; nothing + # above establishes that this card is a depiction of that printing at all. + evidence = _current_evidence_for(card_ids[0]) + if evidence is None: + result.abstained_no_evidence += 1 + continue + disagreement = printing_attribute_disagreement(evidence, _printing_metadata_for(uncovered[0].pk)) + if disagreement is not None: + result.abstained_attribute_mismatch += 1 + continue + votes_batch.append( CardPrintingTag( card_id=card_ids[0], diff --git a/MPCAutofill/cardpicker/management/commands/run_pipeline.py b/MPCAutofill/cardpicker/management/commands/run_pipeline.py index a27b24d6f..0c761ff83 100644 --- a/MPCAutofill/cardpicker/management/commands/run_pipeline.py +++ b/MPCAutofill/cardpicker/management/commands/run_pipeline.py @@ -176,6 +176,7 @@ from cardpicker.local_clustering import compute_two_threshold_clusters from cardpicker.local_identify_printing_tags import ( + EXCLUDED_RESOLVED_TAGS, build_propagated_cluster_votes, verify_zero_resolutions, ) @@ -183,7 +184,14 @@ EXIT_ENVELOPE_HALT, run_stage_zero_freshness, ) -from cardpicker.models import Card, CardPrintingTag, PilotRunLedger, VoteSource +from cardpicker.models import ( + Card, + CardPrintingTag, + CardTypes, + PilotRunLedger, + PrintingTagStatus, + VoteSource, +) from cardpicker.operating_envelope import check_envelope, current_trip from cardpicker.pilot_run_lifecycle import ( mark_ledger_failed, @@ -211,6 +219,17 @@ # not the summary row, keep the unsuffixed name. LEDGER_RUN_ID_SUFFIX = "-pipeline" +# MINIMUM WALL-CLOCK GAP BETWEEN ENVELOPE SAMPLES during a pass (`_EnvelopeSentry`). Owner brief: +# "do not re-sample so often it becomes its own load." A sample is one `/proc` load read, one RSS +# read and one DB query (`current_trip`; `check_envelope` only writes on an actual breach), so the +# DB round trip is the real cost, not the host reads. 60s is chosen against what the bar actually +# measures rather than tuned by feel: `HOST_LOAD_CEILING` is compared against the ONE-MINUTE load +# average (`os.getloadavg()[0]`, see `stage_e_dispatch._sample_envelope_signals`), so sampling +# faster than 60s re-reads a number that has not finished moving - it cannot detect a breach any +# earlier, it only multiplies queries. Sampling much slower would let a breach persist for longer +# than the window the signal is derived from. +ENVELOPE_RESAMPLE_INTERVAL_SECONDS = 60.0 + class Command(BaseCommand): help = ( @@ -340,6 +359,19 @@ def handle(self, *args: Any, **options: Any) -> None: self.stdout.write(f"To resume this run after a stop: --run-id {run_id}") self.stdout.write("=" * 78) + # THE ENVELOPE IS SAMPLED THROUGHOUT THE PASS, NOT ONLY BEFORE IT - see `_EnvelopeSentry`. + # Constructed here rather than inside the preflight because every stage below shares this + # one instance: it is what carries the interval gate, so the whole pass samples at one + # cadence instead of each stage re-deciding. + # `interval_seconds` passed EXPLICITLY from the module constant rather than inherited from + # the parameter default, so the cadence is resolvable (and overridable) at CALL time - a + # default bound at `def` time cannot be reached by a test that needs to prove re-sampling + # happens at all without making the test sleep for a real minute. + self._envelope = _EnvelopeSentry( + run_id=run_id, write=self.stdout.write, interval_seconds=ENVELOPE_RESAMPLE_INTERVAL_SECONDS + ) + envelope_check: Optional[Any] = None if options["skip_envelope"] else self._envelope.check + 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 @@ -385,6 +417,14 @@ def handle(self, *args: Any, **options: Any) -> None: else: counters["stage_c"] = self._run_stage_c(run_id=run_id, options=options, dry_run=dry_run) + # Stage C is the one stage whose inside this command cannot reach: it is delegated + # whole to `run_image_evidence_cohort` via `call_command`, and that command owns its + # own RSS guard and its own limiter. So the envelope is re-sampled at the seam AFTER + # it - the first point where a Stage C that spent hours saturating the box can be + # observed by this command at all. + if envelope_check is not None: + envelope_check("stage-d") + 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. @@ -400,7 +440,9 @@ def handle(self, *args: Any, **options: Any) -> None: 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) + counters["stage_d"] = self._run_stage_d_bulk( + run_id=run_id, cohort_ids=cohort_ids, dry_run=dry_run, envelope_check=envelope_check + ) # -- STAGE C+ : CLUSTER VOTE PROPAGATION ------------------------------------------- if options["skip_clustering"]: @@ -408,7 +450,7 @@ def handle(self, *args: Any, **options: Any) -> None: counters["clustering"] = {"skipped": True} else: counters["clustering"] = self._propagate_cluster_votes( - run_id=run_id, cohort_ids=cohort_ids, dry_run=dry_run + run_id=run_id, cohort_ids=cohort_ids, dry_run=dry_run, envelope_check=envelope_check ) # -- STAGE E : FIDELITY GATE ------------------------------------------------------- @@ -432,6 +474,10 @@ def handle(self, *args: Any, **options: Any) -> None: counters["channel_report"] = self._run_channel_report(run_id=run_id) counters["elapsed_s"] = round(time.monotonic() - started, 1) + # How often the envelope actually got sampled, on the run's own ledger row. A pass that + # reports one sample is a pass that never re-sampled, which is the PR #660 behaviour + # this fix exists to end - so it is recorded rather than left to be inferred. + counters["envelope"] = self._envelope.stats() ledger.status = PilotRunLedger.Status.COMPLETED ledger.finished_at = timezone.now() ledger.counters = merge_counters(ledger.counters, counters) @@ -466,36 +512,15 @@ def handle(self, *args: Any, **options: Any) -> None: # ------------------------------------------------------------------------------------------ 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. + The operating envelope's PREFLIGHT - the first sample the sentry takes, forced rather than + interval-gated, before anything is written. Delegates to `_EnvelopeSentry` so the preflight + and every mid-pass re-sample are literally the same check; see that class's own docstring + for the bars, the halt semantics, and why re-sampling exists at all. """ 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._envelope.check("preflight", force=True) self.stdout.write("STAGE E: operating envelope clear.") # ------------------------------------------------------------------------------------------ @@ -542,7 +567,12 @@ def _run_stage_c(self, *, run_id: str, options: dict[str, Any], dry_run: bool = # ------------------------------------------------------------------------------------------ def _run_stage_d_bulk( - self, *, run_id: str, cohort_ids: Optional[list[int]], dry_run: bool = False + self, + *, + run_id: str, + cohort_ids: Optional[list[int]], + dry_run: bool = False, + envelope_check: Optional[Any] = None, ) -> dict[str, Any]: """ Stage D, called EXPLICITLY - `stage_e_dispatch._run_stage_d`, the single place the @@ -563,7 +593,10 @@ def _run_stage_d_bulk( # `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) + # `envelope_check` THREADED IN (2026-07-30). `_run_stage_d` calls it at each seam between + # its calculators, which is the finest granularity reachable without refactoring all of + # them - see that function's own docstring for the residual gap that leaves. + _run_stage_d(cohort_ids, run_id, outcome, dry_run=dry_run, envelope_check=envelope_check) result = { "join_key_votes": outcome.stage_d_join_key_votes, @@ -582,15 +615,146 @@ def _run_stage_d_bulk( # ------------------------------------------------------------------------------------------ def _propagate_cluster_votes( - self, *, run_id: str, cohort_ids: Optional[list[int]], dry_run: bool = False + self, + *, + run_id: str, + cohort_ids: Optional[list[int]], + dry_run: bool = False, + envelope_check: Optional[Any] = None, ) -> dict[str, Any]: """ - STAGE C+ - the pilot capability that was reachable from no engine. + STAGE C+ - GROUP VOTE PROPAGATION. Two grouping keys, run in a fixed order, sharing ONE + propagation engine (`_propagate_over_groups`). + + MD5 FIRST, AND MD5 IS THE ONE THAT SHARES A PRINTING VOTE (owner ruling, 2026-07-30: + "the md5 dedupe should only fetch each identical image once across sources and then apply + votes to the entire group as the fetched card passes through the monolith"). Two cards with + the same `Card.md5_checksum` are the SAME BYTES, so they are depictions of the same + printing - a claim of exact identity whose correctness argument is trivial. That is the + claim a printing verdict needs. + + WHY THE FETCH HALF AND THE VOTE HALF HAD TO BE RE-KEYED ONTO THE SAME GROUP. Before this, + the two halves were keyed DIFFERENTLY: `evidence_transfer` saves a FETCH on the md5 group, + while this stage saved a DEDUCTION on the phash distance-0 group. Byte-identical files + always share a phash, but files sharing a phash are not necessarily byte-identical, so the + set that got a fetch saved and the set that got a vote propagated were not the same set - + which is exactly the thing that is hard to reason about and easy to get wrong. The md5 + group now behaves as one unit end to end: fetched once (`evidence_transfer`), deduced once + (Stage D on whichever member Stage D reached), and the conclusion applied across the group + here, under the casting calculator's own identity. + + PROPAGATION IS NOT REDUNDANT, WHICH WAS CHECKED BEFORE IT WAS BUILT. `evidence_transfer` + already gives every md5 sibling its own `ImageEvidence` row with byte-identical extractor + field values, so it is reasonable to ask whether each member already reaches the same + conclusion independently, making this stage unnecessary. It does not, for a reason that is + structural rather than incidental: a Stage D printing deduction is NOT a function of the + evidence row alone. `local_calculate_verdicts._resolve_candidates_for_card` keys the + candidate list on `Card.name`, and two md5-identical uploads from different sources + routinely carry different names (that is the ordinary state of a cross-source catalogue). + Members also differ on per-card eligibility. So members genuinely reach DIFFERENT + conclusions, or none at all, from identical evidence - and the group's one sound conclusion + has to be carried to them deliberately. + + PROPAGATION NEVER OVERRIDES A MEMBER'S OWN INELIGIBILITY. See + `_members_eligible_for_a_propagated_vote`: a member that is already RESOLVED, already + confirmed to a `canonical_card`, not a `CARD`, or carrying a resolved `custom-art` / + `non-english` tag is skipped. Being byte-identical to a card we identified does not entitle + this stage to overwrite a fact the catalogue already holds about the member, and the + `custom-art` case is the sharp one: that tag is the catalogue DECLARING the image is not a + faithful depiction of a printing, so voting a printing onto it would contradict a + human-visible declaration on the strength of a checksum. + + THE PHASH TIER IS LEFT EXACTLY AS PR #660 SHIPPED IT, and that is a flagged decision rather + than an accreted default. Issue #661 holds the question of what phash grouping is FOR; the + owner's stated direction is that phash should eventually share an ILLUSTRATION (same + artwork, possibly a DIFFERENT printing - a near-identity claim at a grain where a weaker + claim is appropriate), not a printing verdict. Until that is built, removing the existing + phash printing propagation would itself be a behaviour change, and it is currently the ONLY + propagation reaching cards that have no md5 at all (md5 is NULL for every `LOCAL_FILE` + source by design and is never invented - see `Card.md5_checksum`'s own docstring). So it + stays, it runs SECOND, and the ordering is the point: md5's exact-identity votes land + first, and the phash tier can only fill what md5 did not, because + `_propagate_over_groups` re-reads the already-voted set per tier. + + THE SEAM FOR PHASH-AS-ILLUSTRATION IS THE `groups` PARAMETER. `_propagate_over_groups` + takes `members_by_representative` and knows nothing about how it was keyed, so adding the + illustration-grain tier later means computing a different grouping and calling the same + engine - not restructuring this stage. That is the "single grouping abstraction with the + key as a parameter" #661 asks for, arrived at here rather than deferred to it. + """ + stats: dict[str, Any] = {} + + md5_groups = self._md5_groups(cohort_ids) + stats["md5"] = self._propagate_over_groups( + groups=md5_groups, + tier="md5", + run_id=run_id, + dry_run=dry_run, + envelope_check=envelope_check, + ) + + phash_groups = self._phash_distance_zero_groups(cohort_ids) + stats["phash_d0"] = self._propagate_over_groups( + groups=phash_groups, + tier="phash_d0", + run_id=run_id, + dry_run=dry_run, + envelope_check=envelope_check, + ) + + key = "would_propagate" if dry_run else "votes_propagated" + stats[key] = stats["md5"].get(key, 0) + stats["phash_d0"].get(key, 0) + self.stdout.write(f"STAGE C+: {stats}") + return stats + + # ------------------------------------------------------------------------------------------ + def _md5_groups(self, cohort_ids: Optional[list[int]]) -> dict[int, list[int]]: + """ + BYTE-IDENTICAL GROUPS, keyed on `Card.md5_checksum` - the same key `evidence_transfer` + already groups on, so the fetch saving and the vote saving now describe the same set. + + A NULL OR UNIQUE md5 IS A GROUP OF ONE (issue #473's ruling 3, inherited verbatim rather + than re-decided here) and simply does not appear in the result: a checksum is copied from + the source listing and is NEVER invented, so a card without one groups with nothing. The + empty string is excluded alongside NULL - `md5_checksum` is a `CharField`, and an empty + value is an absent checksum, not a value that thousands of cards genuinely share. + + Representative = `min(pk)`, matching `local_clustering._compute_exact_match_clusters`' + own convention exactly, so the two tiers cannot disagree about what a representative IS. + Note the representative is only a STABLE NAME for the group here - it is NOT required to be + the card that holds the vote, because `_propagate_over_groups` looks for source votes + across every member. That matters: Stage D reaches whichever member it reaches, and there + is no reason that is the lowest pk. + + THE POOL IS THE CATALOGUE, NOT THIS RUN'S SELECTION - the same property the phash tier's + own note describes. `--scope-stage-d-to-cohort` narrows it, and narrows that independence + with it. + """ + cards = Card.objects.filter(md5_checksum__isnull=False).exclude(md5_checksum="") + if cohort_ids is not None: + cards = cards.filter(pk__in=cohort_ids) + by_checksum: dict[str, list[int]] = {} + for pk, checksum in cards.values_list("pk", "md5_checksum").iterator(): + if not checksum: + # Unreachable given the filter above; kept so that "a NULL or empty checksum is a + # group of ONE" is true by construction rather than by the queryset alone. Fusing + # every checksum-less card into a single group is the worst failure available + # here, and it is one narrowed filter away at all times. + continue + by_checksum.setdefault(checksum, []).append(pk) + groups: dict[int, list[int]] = {} + for card_ids in by_checksum.values(): + if len(card_ids) < 2: + continue + representative = min(card_ids) + groups[representative] = sorted(pk for pk in card_ids if pk != representative) + return groups - `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. + # ------------------------------------------------------------------------------------------ + def _phash_distance_zero_groups(self, cohort_ids: Optional[list[int]]) -> dict[int, list[int]]: + """ + The phash distance-0 tier PR #660 shipped, unchanged in behaviour and merely lifted into + its own method so both tiers hand the SAME shape to the SAME propagation engine. 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 @@ -599,10 +763,6 @@ def _propagate_cluster_votes( 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: @@ -613,47 +773,173 @@ def _propagate_cluster_votes( # 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} + return compute_two_threshold_clusters(cast(Any, selected)).members_by_representative + + # ------------------------------------------------------------------------------------------ + def _members_eligible_for_a_propagated_vote(self, member_ids: set[int]) -> set[int]: + """ + WHICH GROUP MEMBERS MAY RECEIVE A PROPAGATED PRINTING VOTE AT ALL (owner constraint, + 2026-07-30: "propagation must not override a member's own ineligibility - a card excluded + for a real reason stays excluded"). + + These are CATALOGUE-LEVEL facts about the member, not workload preferences: + * `printing_tag_status` is still UNRESOLVED - a resolved card's printing is settled. + * no confirmed `canonical_card` - a human-confirmed indexing match outranks any machine + vote, and contradicting it from a checksum would be the worst available failure. + * `card_type=CARD` - tokens and cardbacks are excluded from every printing channel in + this codebase for structural reasons (`_eligible_base_queryset`'s own docstring). + * no resolved `custom-art` / `non-english` tag - the sharp one. `custom-art` is the + catalogue DECLARING this image is not a faithful depiction of a printing. Byte + identity with a card we identified does not overturn that declaration. + + DELIBERATELY NOT `local_identify_printing_tags._eligible_base_queryset`, and this is the + one place in this change where a predicate is expressed rather than reused. That function + computes the same four facts, but bundles them with WORKLOAD rules that are wrong here: a + scan-log exclusion keyed to the PILOT's own rescannable vocabulary (a Stage D identity's + vocabulary differs), and a deductive-backfill exclusion that is a "don't spend a scan" + choice rather than an ineligibility. It also cannot simply be refactored to expose these + four: its own docstring records that several tests and `stream_backstop_sweep` assert + against its COMPILED SQL, so re-ordering its `.exclude()` chain to share a helper would + change that SQL for every legacy caller. `TestPropagationEligibilityMatchesTheBaseQueryset` + is the drift tripwire that keeps the two honest instead - see its own docstring. + + DELIBERATELY NOT `local_calculate_verdicts._eligible_cards_queryset` either: that one + additionally requires a CURRENT `ImageEvidence` row, and propagating to a member that was + never fetched or computed is the entire point of this stage. + """ + return set( + Card.objects.filter( + pk__in=member_ids, + printing_tag_status=PrintingTagStatus.UNRESOLVED, + canonical_card__isnull=True, + card_type=CardTypes.CARD, + ) + .exclude(tags__contains=[EXCLUDED_RESOLVED_TAGS[0]]) + .exclude(tags__contains=[EXCLUDED_RESOLVED_TAGS[1]]) + .values_list("pk", flat=True) + ) + + # ------------------------------------------------------------------------------------------ + def _propagate_over_groups( + self, + *, + groups: dict[int, list[int]], + tier: str, + run_id: str, + dry_run: bool = False, + envelope_check: Optional[Any] = None, + ) -> dict[str, Any]: + """ + THE ONE PROPAGATION ENGINE, shared by every grouping key. It is handed + `members_by_representative` and knows nothing about how the grouping was computed - which + is what makes adding a tier (issue #661's illustration-grain phash) a matter of computing a + different grouping, not restructuring this stage. + + THE SOURCE VOTE MAY BE HELD BY ANY MEMBER, NOT ONLY THE REPRESENTATIVE. PR #660 looked for + votes on representatives only, which silently propagated nothing whenever Stage D happened + to reach a non-representative member - and Stage D has no reason to prefer the lowest pk. + This reads this run's votes across EVERY member of every group, then propagates each one to + the rest of ITS OWN group. + + ONE SOURCE VOTE PER (GROUP, IDENTITY), CHOSEN DETERMINISTICALLY (lowest card id). Without + this, two members of one group both holding a vote would each generate rows for the other + members, producing duplicate `(card, printing, anonymous_id)` rows inside a SINGLE write + batch - which the already-voted guard cannot catch, because it is computed from the DB + before any of this batch is written. + + A GROUP WHOSE MEMBERS DISAGREE IS COUNTED, NOT SILENTLY RESOLVED. Two members of one md5 + group can hold DIFFERENT printing verdicts under the same identity, because their + candidate lists came from different `Card.name`s. Byte-identical images cannot depict two + different printings, so a disagreement is a real signal about the upstream deduction, not + noise to average away. The deterministic pick keeps the write well-defined; the counter + (`groups_with_conflicting_verdicts`) is what makes the condition visible on the ledger + rather than lost. + + `members_already_voted` is one query per identity, up front, and is RE-READ for each tier + so an earlier tier's writes are visible to a later one. That is what gives md5 precedence + over phash without either tier knowing about the other. Propagating to a member that + already holds a vote under the same `anonymous_id` would violate `CardPrintingTag`'s own + (card, printing, anonymous_id) uniqueness constraint anyway. + """ + absorbed_ids = {m for members in groups.values() for m in members} stats: dict[str, Any] = { - "cluster_count": len(members_by_representative), - "cards_absorbed_into_clusters": len(member_ids), + "group_count": len(groups), + "cards_absorbed_into_groups": len(absorbed_ids), "votes_propagated": 0, + "members_skipped_ineligible": 0, + "groups_with_conflicting_verdicts": 0, } - if not member_ids: - self.stdout.write(f"STAGE C+: {stats} - nothing to propagate.") + if not absorbed_ids: return stats - # The representatives' own verdicts, as cast by THIS run's Stage D. + if envelope_check is not None: + envelope_check(f"stage-c+:{tier}") + + group_of_member: dict[int, int] = {} + for representative, members in groups.items(): + group_of_member[representative] = representative + for member in members: + group_of_member[member] = representative + + # EVERY CARD IN EVERY GROUP, REPRESENTATIVES INCLUDED - not just the absorbed members. Two + # separate things depend on this and both are wrong if the representative is left out: + # a representative can HOLD the source vote (Stage D has no reason to reach the lowest pk + # first), and a representative can equally be a propagation TARGET when some other member + # holds it. `absorbed_ids` above stays members-only because it reports "how many cards were + # absorbed into a group", which is a different question from "who participates here". + all_group_card_ids = set(group_of_member.keys()) 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) + CardPrintingTag.objects.filter(card_id__in=all_group_card_ids, run_id=run_id, is_no_match=False).exclude( + printing_id=None + ) ) + if not source_votes: + return stats + + # One source vote per (group, identity), lowest card id wins; disagreements counted. + chosen: dict[tuple[int, str], CardPrintingTag] = {} + conflicted: set[tuple[int, str]] = set() + for vote in sorted(source_votes, key=lambda v: v.card_id): + 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 + key = (group_of_member[vote.card_id], vote.anonymous_id) + incumbent = chosen.get(key) + if incumbent is None: + chosen[key] = vote + elif incumbent.printing_id != vote.printing_id: + conflicted.add(key) + stats["groups_with_conflicting_verdicts"] = len({group for group, _identity in conflicted}) + + eligible_member_ids = self._members_eligible_for_a_propagated_vote(all_group_card_ids) + stats["members_skipped_ineligible"] = len(all_group_card_ids) - len(eligible_member_ids) + already_voted_by_identity: dict[str, set[int]] = {} - for anonymous_id in {vote.anonymous_id for vote in source_votes}: + for anonymous_id in {vote.anonymous_id for vote in chosen.values()}: already_voted_by_identity[anonymous_id] = set( - CardPrintingTag.objects.filter(card_id__in=member_ids, anonymous_id=anonymous_id).values_list( + CardPrintingTag.objects.filter(card_id__in=all_group_card_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 + for (representative, anonymous_id), vote in chosen.items(): + assert vote.printing_id is not None and vote.confidence is not None + # Every card in the group EXCEPT the one holding the source vote. Built here rather + # than reusing `groups` directly because the source vote is not necessarily the + # representative, so "the others" is relative to the VOTE, not to the group's name. + others = [pk for pk in ([representative] + groups[representative]) if pk != vote.card_id] + skip = already_voted_by_identity.get(anonymous_id, set()) | (set(others) - eligible_member_ids) rows.extend( build_propagated_cluster_votes( representative_card_id=vote.card_id, printing_pk=vote.printing_id, - anonymous_id=vote.anonymous_id, + anonymous_id=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()), + members_by_representative={vote.card_id: others}, + members_already_voted=skip, source=VoteSource(vote.source), ) ) @@ -663,7 +949,6 @@ def _propagate_cluster_votes( # 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 # ------------------------------------------------------------------------------------------ @@ -714,6 +999,101 @@ def _run_channel_report(self, *, run_id: str) -> dict[str, Any]: return {"exit_code": exit_code} +class _EnvelopeSentry: + """ + THE OPERATING ENVELOPE, RE-SAMPLED DURING THE PASS (2026-07-30) - not just before it. + + PR #660 shipped Stage E as a PREFLIGHT ONLY: `current_trip`/`check_envelope` once, before + Stage C, never again. Owner brief: "host resampling is likely required (for steps that aren't + fetch) as the same monolith will run for small datasets and large ones so needs to fit the + available compute appropriately." A single check at launch is right for a 200-card run and + wrong for a 230k one - the box's free compute at hour six is not what it was at hour zero, and + this command's whole point is that ONE invocation serves both sizes. + + THREE SEPARATE PROTECTIONS, AND THIS CLASS IS ONLY THE MIDDLE ONE. They are easy to conflate + and must not be: + * the 7/s ceiling protects GOOGLE, applies per request, and lives beneath Stage C in + `harvest_rate_coordinator` (PR #649). Nothing here. + * THE ENVELOPE protects THIS HOST. That is this class. + * compute gating paces everything else. Also not here. + + HALT SEMANTICS ARE PRESERVED EXACTLY, and this is the sharp edge. A genuine breach HALTS and + never self-resumes: `check_envelope` persists an `EnvelopeTrip`, this raises, and the only way + back is an owner action through `resolve_envelope_trip`. A re-sample is NOT a throttle and must + never become one - PR #644 converted RATE PRESSURE (429/503) to a throttle precisely so that it + would stop masquerading as an envelope breach, and converting a host-load breach the other way + would undo that distinction from the opposite direction. Load average is this box's own + saturation; going slower on Google does not reduce it. + + WHAT A MID-PASS HALT LEAVES BEHIND, stated plainly because it differs from the preflight's + "nothing was written". By the time a re-sample fires, real rows exist. They stay - every one of + them carries this run's `run_id` and is queryable by it, and the run resumes with + `--run-id ` once the trip is acknowledged. That is the same posture the fidelity gate + already takes (exit 7 is "read this run before trusting it", not a rollback), and it is why + this raises rather than attempting any unwind. + + INTERVAL-GATED so it cannot become its own load (see `ENVELOPE_RESAMPLE_INTERVAL_SECONDS` for + why 60s is derived from the one-minute load average rather than picked). `check(..., force=True)` + bypasses the gate and is used for the preflight, which must always sample. + + `current_trip` BEFORE `check_envelope`, never the reverse - the order `operating_envelope`'s own + docstring requires. An OPEN trip refuses outright, including a trip some OTHER process opened + while this pass was running, which is the case a preflight-only design could not see at all. + """ + + __slots__ = ("_run_id", "_write", "_interval", "_last_sampled_at", "samples", "skipped") + + def __init__(self, *, run_id: str, write: Any, interval_seconds: float = ENVELOPE_RESAMPLE_INTERVAL_SECONDS): + self._run_id = run_id + self._write = write + self._interval = interval_seconds + self._last_sampled_at: Optional[float] = None + self.samples = 0 + self.skipped = 0 + + def check(self, step: str, *, force: bool = False) -> None: + """ + Sample the envelope unless the interval gate says it is too soon. Raises `CommandError` + with `EXIT_ENVELOPE_HALT` on an open trip or a fresh breach; returns silently otherwise. + `step` names the stage about to start and appears in the halt message, so an operator + reading a halted run knows where in the pass it stopped without correlating timestamps. + """ + now = time.monotonic() + if not force and self._last_sampled_at is not None and (now - self._last_sampled_at) < self._interval: + self.skipped += 1 + return + self._last_sampled_at = now + self.samples += 1 + + existing = current_trip(run_id=self._run_id) + if existing is not None: + raise CommandError( + f"ENVELOPE HALT before {step}: trip {existing.trip_id} ({existing.bar}) is still " + "open. No self-resume - clear it with `resolve_envelope_trip` after investigating. " + + self._written_so_far(step), + returncode=EXIT_ENVELOPE_HALT, + ) + fresh = check_envelope(_sample_envelope_signals(), run_id=self._run_id) + if fresh is not None: + raise CommandError( + f"ENVELOPE HALT before {step}: bar {fresh.bar} breached ({fresh.detail}); trip " + f"{fresh.trip_id} persisted. " + self._written_so_far(step), + returncode=EXIT_ENVELOPE_HALT, + ) + + def _written_so_far(self, step: str) -> str: + if step == "preflight": + return "Nothing was written." + return ( + f"Everything this run wrote before {step} STAYS WRITTEN and is queryable by " + f"run_id={self._run_id}; resume with --run-id {self._run_id} once the trip is " + "acknowledged." + ) + + def stats(self) -> dict[str, Any]: + return {"samples": self.samples, "skipped_by_interval": self.skipped, "interval_s": self._interval} + + class _ClusterInput: """ The two-attribute shape `local_clustering.compute_two_threshold_clusters` reads (`.card.pk`, diff --git a/MPCAutofill/cardpicker/stage_e_dispatch.py b/MPCAutofill/cardpicker/stage_e_dispatch.py index 08f98c2d4..a948f521f 100644 --- a/MPCAutofill/cardpicker/stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/stage_e_dispatch.py @@ -1134,7 +1134,13 @@ def _run_attribute_chip_casters( ) -def _run_stage_d(batch_ids: Optional[list[int]], run_id: str, outcome: DispatchOutcome, dry_run: bool = False) -> None: +def _run_stage_d( + batch_ids: Optional[list[int]], + run_id: str, + outcome: DispatchOutcome, + dry_run: bool = False, + envelope_check: Optional[Callable[[str], None]] = None, +) -> 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 @@ -1189,24 +1195,59 @@ def _run_stage_d(batch_ids: Optional[list[int]], run_id: str, outcome: DispatchO stop eight concurrent dispatches from re-tripping the load bar the moment streaming resumes - 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. + + `envelope_check` (2026-07-30, OPTIONAL, DEFAULT `None` = today's behaviour byte for byte). + A callable invoked at each seam BETWEEN the calculators below, given the name of the step + about to start. It exists for ONE caller, `run_pipeline`, whose pass is whole-catalogue and + long: PR #660 checked the operating envelope once as a PREFLIGHT and never again, which is + right for a 200-card run and wrong for a 230k one, because the same command has to fit + whatever compute is free DURING the pass, not what was free at launch. + + THE CONVEYOR DELIBERATELY PASSES NOTHING. `dispatch_micro_batch` already samples the envelope + per micro-batch, above this function, so a second check inside would be redundant sampling on + the hot path. `None` therefore means "my caller already owns this", not "nobody is checking". + + THE CALLBACK MAY RAISE, AND RAISING IS THE POINT. A genuine envelope breach HALTS and does not + self-resume (`operating_envelope`'s own resume semantics; cleared only by + `resolve_envelope_trip`). So this function does not catch, translate or count anything the + callback raises - it propagates, and whatever Stage D had already written stays written and + stays queryable by `run_id`. This is deliberately NOT the throttle path: rate pressure is + handled beneath Stage C by `harvest_fetch_limiter` (PR #644/#649) and never reaches here. + + THE SEAMS ARE BETWEEN CALCULATORS, NOT INSIDE THEM. Each of the steps below is one call into a + calculator that owns its own internal batching, so the finest granularity reachable WITHOUT + changing all of those calculators is one check per step. At catalogue scale a single + calculator can run for a long time between checks; closing that gap means threading a progress + callback into each calculator's own batch loop, which is a real refactor of shared code and is + deliberately not done here. """ + + def seam(step: str) -> None: + if envelope_check is not None: + envelope_check(step) + + seam("stage-d:join-key") 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 + seam("stage-d:fallback") 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 + seam("stage-d:illustration") 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 + seam("stage-d:slow-path") 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. + seam("stage-d:attribute-chips") _run_attribute_chip_casters(run_id=run_id, card_ids=batch_ids, outcome=outcome, dry_run=dry_run) diff --git a/MPCAutofill/cardpicker/tests/test_local_identify_printing_tags.py b/MPCAutofill/cardpicker/tests/test_local_identify_printing_tags.py index 10ac3cc28..3ec8c52a1 100644 --- a/MPCAutofill/cardpicker/tests/test_local_identify_printing_tags.py +++ b/MPCAutofill/cardpicker/tests/test_local_identify_printing_tags.py @@ -70,6 +70,7 @@ CanonicalPrintingMetadataFactory, CardFactory, CardPrintingTagFactory, + ImageEvidenceFactory, SourceFactory, TagFactory, ) @@ -1518,6 +1519,26 @@ def fake_ocr(selected, image, crop_box, bleed_class=None, known_set_codes=None): assert attributes.uncovered_printings_closed == 0 +def _extracted(card, layout_class="black", illus_anchor_fired=True, collector_number="158"): + """A card that has actually been LOOKED AT: a current `ImageEvidence` row carrying a full frame + manifest. Required by `run_name_frequency_elimination` as of 2026-07-30 - its visual conjunct + abstains outright on a card with no stored evidence, because "we have no evidence" must not + read as "no evidence against" for a deduction whose entire weakness is that it never looked at + the image. `content_phash` is set on the card so the row satisfies `current_evidence_queryset`'s + own currency rule (`content_hash == card.content_phash`); `CardFactory` leaves it NULL by + default, which would make the row permanently non-current and silently defeat the fixture.""" + card.content_phash = card.content_phash or 0x0A0A0A0A0A0A0A0A + card.save(update_fields=["content_phash"]) + return ImageEvidenceFactory( + card=card, + content_hash=card.content_phash, + extractor_versions={"collector_line_ocr": 1, "artist_ocr": 1}, + layout_class=layout_class, + illus_anchor_fired=illus_anchor_fired, + collector_line_collector_number=collector_number, + ) + + class TestNameFrequencyElimination: """Fast-follow (2026-07-16): run_name_frequency_elimination's SAFE 1:1 gate - exactly one uncovered printing AND exactly one unresolved-eligible card for that name - not just "one @@ -1529,6 +1550,7 @@ def test_votes_for_the_single_uncovered_printing_when_exactly_one_card_and_one_g uncovered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="bbb")) CardFactory(canonical_card=covered_printing) # confirms "aaa" as covered card = CardFactory(name="Forest") # the single unresolved card for this name + _extracted(card) result = run_name_frequency_elimination(dry_run=False) @@ -1575,26 +1597,201 @@ def test_dry_run_writes_nothing(self, db): covered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="aaa")) CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="bbb")) CardFactory(canonical_card=covered_printing) - CardFactory(name="Forest") + _extracted(CardFactory(name="Forest")) result = run_name_frequency_elimination(dry_run=True) assert result.votes_written == 1 # counted, even though nothing is persisted assert not CardPrintingTag.objects.exists() - def test_idempotent_on_a_second_invocation(self, db): + def test_a_fresh_run_recasts_the_same_verdict_without_duplicating_the_row(self, db): + """2026-07-30, the census-leak fix's deliberate consequence. This test used to assert + `second.votes_written == 0` under the name `test_idempotent_on_a_second_invocation`. That + property was a SIDE EFFECT of the lifetime own-vote exclusion this calculator's census + cannot correctly use (see `test_a_prior_runs_own_vote_does_not_deplete_the_census...` + below), not a property anyone specified, and it directly contradicted the 2026-07-29 + run-scoping directive: "prior runs must not suppress work in a new run". + + The property that MATTERS - and the one an idempotence test should have been asserting all + along - is that re-running never DUPLICATES or CHANGES a verdict: one row, same printing, + superseded in place by `purge_and_write_votes` rather than accumulated. That is asserted + here and is unchanged by the fix.""" covered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="aaa")) - CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="bbb")) + uncovered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="bbb")) CardFactory(canonical_card=covered_printing) - CardFactory(name="Forest") + card = CardFactory(name="Forest") + _extracted(card) first = run_name_frequency_elimination(dry_run=False) second = run_name_frequency_elimination(dry_run=False) + assert first.votes_written == 1 + # A FRESH run_id reconsiders every card, so the same sound verdict is cast again. + assert second.votes_written == 1 + # ...but the row is superseded, never duplicated, and the verdict never drifts. + votes = CardPrintingTag.objects.filter(anonymous_id=NAME_FREQUENCY_ANONYMOUS_ID) + assert votes.count() == 1 + assert votes.get().card_id == card.pk + assert votes.get().printing_id == uncovered_printing.pk + + def test_within_run_resume_still_skips_cards_this_run_already_voted_on(self, db): + """The half of the own-vote exclusion that run-scoping KEEPS. A killed run re-invoked with + the SAME run_id must pick up where it stopped rather than redoing completed batches - so + the exclusion is narrowed to this run's rows, not removed.""" + covered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="aaa")) + CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="bbb")) + CardFactory(canonical_card=covered_printing) + _extracted(CardFactory(name="Forest")) + + first = run_name_frequency_elimination(dry_run=False, run_id="resume-me") + second = run_name_frequency_elimination(dry_run=False, run_id="resume-me") + assert first.votes_written == 1 assert second.votes_written == 0 assert CardPrintingTag.objects.filter(anonymous_id=NAME_FREQUENCY_ANONYMOUS_ID).count() == 1 + def test_a_prior_runs_own_vote_does_not_deplete_the_census_for_a_later_run(self, db): + """THE CENSUS LEAK (2026-07-30). `_eligible_base_queryset` was called with no `run_id`, so + its "exclude cards already carrying this calculator's vote" was LIFETIME. The gate is a + COUNT over exactly that population ("exactly one unresolved eligible card for this name"), + so the calculator was taking a census over a pool it permanently shrinks itself. + + The sequence below is the concrete defect, and note that the wrong answer is a FRESH WRONG + POSITIVE, not a stale or missed vote - nothing about card B changed between the runs, only + the size of the population the gate counts: + + run 1 "Forest" has ONE unresolved eligible card (A) and ONE uncovered printing. + Sound: A is that printing by elimination. A is voted. + later a second "Forest" upload, B, arrives from another source - the ordinary way this + catalogue grows. + run 2 WITH THE LEAK: A is excluded forever by its own run-1 vote, so the name presents + as having exactly one unresolved card (B) and votes B for the same printing. + That is a coin flip wearing the gate's clothes - either A or B is a redundant + depiction of the already-covered printing and elimination cannot say which. + CORRECTLY: the pool is {A, B}, the count is 2, and the gate ABSTAINS. + + A machine vote does not resolve a card (`compute_covered_printing_pks` counts only + confirmed `canonical_card` / RESOLVED `inferred_canonical_card`), so run 1's vote leaves + the printing uncovered and the coverage half of the gate still passes on run 2 - which is + precisely why the census half has to be the thing that stops it.""" + covered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="aaa")) + CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="bbb")) + CardFactory(canonical_card=covered_printing) + card_a = CardFactory(name="Forest") + _extracted(card_a) + + first = run_name_frequency_elimination(dry_run=False, run_id="run-1") + assert first.votes_written == 1 + assert CardPrintingTag.objects.filter( + card=card_a, anonymous_id=NAME_FREQUENCY_ANONYMOUS_ID + ).exists(), "precondition: run 1 must have cast the sound vote this test then depletes the pool with" + + card_b = CardFactory(name="Forest") + _extracted(card_b) + + second = run_name_frequency_elimination(dry_run=False, run_id="run-2") + + assert second.votes_written == 0, ( + "run 2 voted despite TWO unresolved eligible cards sharing the name - the census was " + "depleted by run 1's own vote, which is the leak this asserts against" + ) + assert not CardPrintingTag.objects.filter(card=card_b, anonymous_id=NAME_FREQUENCY_ANONYMOUS_ID).exists() + + def test_abstains_when_the_stored_evidence_contradicts_the_candidate_printings_border(self, db): + """ + THE MISSING CONJUNCT (owner ruling, 2026-07-30): "just because a card was printed exactly + once doesn't mean that the image in our catalogue is an accurate depiction of that card, it + may have a different border or another issue." + + The 1:1 count is entirely satisfied here - one uncovered printing, one unresolved eligible + card - and the OLD calculator would have voted. The card's own stored evidence says its + border is WHITE while the printing Scryfall describes is BLACK-bordered, so the image is + not a faithful depiction of the only printing elimination could assign it. Counting cannot + see that; looking can. + + This matters more than "it is only a vote" suggests: issue #593 established that a machine + vote is what the question feed renders as the suggestion to confirm, and a human's click + returns as a full-weight USER vote - so an unverified deduction becomes a one-click rubber + stamp rather than a harmless guess. + """ + covered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="aaa")) + uncovered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="bbb")) + CanonicalPrintingMetadataFactory(canonical_card=uncovered_printing, border_color="black") + CardFactory(canonical_card=covered_printing) + card = CardFactory(name="Forest") + _extracted(card, layout_class="white") + + result = run_name_frequency_elimination(dry_run=False) + + assert result.votes_written == 0 + assert result.abstained_attribute_mismatch == 1 + assert not CardPrintingTag.objects.filter(card=card, anonymous_id=NAME_FREQUENCY_ANONYMOUS_ID).exists() + + def test_votes_when_the_stored_evidence_agrees_with_the_candidate_printing(self, db): + """The other side of the conjunct - it must NARROW the tier, not disable it. Same fixture + as the border-mismatch test with one field flipped, so the two together prove the check is + discriminating rather than uniformly refusing.""" + covered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="aaa")) + uncovered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="bbb")) + CanonicalPrintingMetadataFactory(canonical_card=uncovered_printing, border_color="black") + CardFactory(canonical_card=covered_printing) + card = CardFactory(name="Forest") + _extracted(card, layout_class="black") + + result = run_name_frequency_elimination(dry_run=False) + + assert result.votes_written == 1 + assert result.abstained_attribute_mismatch == 0 + assert ( + CardPrintingTag.objects.get(card=card, anonymous_id=NAME_FREQUENCY_ANONYMOUS_ID).printing_id + == uncovered_printing.pk + ) + + def test_abstains_when_the_card_has_never_been_extracted_at_all(self, db): + """ + NO STORED EVIDENCE MEANS ABSTAIN, NOT PROCEED. This is the one place where this module's + usual "missing data is not evidence" rule points the OTHER way, and deliberately: that rule + exists to stop a match being VETOED by silence, whereas here silence is being asked to + ESTABLISH that the image depicts one of the name's printings. A card nobody has ever looked + at leaves that premise exactly as unestablished as it was before the fix. + """ + covered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="aaa")) + CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="bbb")) + CardFactory(canonical_card=covered_printing) + card = CardFactory(name="Forest") # deliberately NO _extracted(...) + + result = run_name_frequency_elimination(dry_run=False) + + assert result.votes_written == 0 + assert result.abstained_no_evidence == 1 + assert not CardPrintingTag.objects.filter(card=card, anonymous_id=NAME_FREQUENCY_ANONYMOUS_ID).exists() + + def test_the_frame_half_of_the_conjunct_is_gated_on_artist_ocr_having_run(self, db): + """ + PR #656's trap, inherited by this caller for free BECAUSE the check is shared rather than + re-derived. `illus_anchor_fired` is nullable and `bool(None)` is `False`, which is + indistinguishable from "artist_ocr ran and found no anchor" - with no collector number + either, `classify_frame_style` then confidently answers "modern" for a card it has no + anchor evidence about, and a genuine OLD-frame printing gets vetoed. + + Here `artist_ocr` never ran, so the frame half must be SKIPPED rather than evaluated on + invented input: the card still votes. A private second copy of this check would very + likely have re-introduced the trap, which is the argument for sharing it. + """ + covered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="aaa")) + uncovered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="bbb")) + CanonicalPrintingMetadataFactory(canonical_card=uncovered_printing, border_color="", frame="1993") + CardFactory(canonical_card=covered_printing) + card = CardFactory(name="Forest") + evidence = _extracted(card, layout_class="", illus_anchor_fired=None, collector_number="") + evidence.extractor_versions = {"collector_line_ocr": 1} # artist_ocr never ran + evidence.save(update_fields=["extractor_versions"]) + + result = run_name_frequency_elimination(dry_run=False) + + assert result.votes_written == 1, "an absent artist_ocr must skip the frame half, not veto on it" + assert result.abstained_attribute_mismatch == 0 + def test_excludes_tokens_and_cardbacks(self, db): covered_printing = CanonicalCardFactory(name="Beast", expansion=CanonicalExpansionFactory(code="aaa")) CanonicalCardFactory(name="Beast", expansion=CanonicalExpansionFactory(code="bbb")) @@ -1613,7 +1810,7 @@ def test_dry_run_writes_nothing(self, db, capsys): covered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="aaa")) CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="bbb")) CardFactory(canonical_card=covered_printing) - CardFactory(name="Forest") + _extracted(CardFactory(name="Forest")) call_command("local_name_frequency_elimination", "--dry-run") @@ -1629,6 +1826,7 @@ def test_real_run_writes_and_passes_gate_check(self, db, capsys): CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="bbb")) CardFactory(canonical_card=covered_printing) card = CardFactory(name="Forest") + _extracted(card) call_command("local_name_frequency_elimination") @@ -1902,7 +2100,7 @@ def test_an_explicit_run_id_is_respected_not_regenerated(self, db, monkeypatch): def test_name_frequency_elimination_gets_its_own_run_id(self, db): printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="aaa")) - CardFactory(name="Forest") + _extracted(CardFactory(name="Forest")) result = run_name_frequency_elimination() @@ -3628,6 +3826,7 @@ def test_name_frequency_insert_failure_rolls_its_purge_back(self, db, monkeypatc uncovered_printing = CanonicalCardFactory(name="Forest", expansion=CanonicalExpansionFactory(code="bbb")) CardFactory(canonical_card=covered_printing) card = CardFactory(name="Forest") + _extracted(card) stale = CardPrintingTag.objects.create( card=card, printing=uncovered_printing, diff --git a/MPCAutofill/cardpicker/tests/test_run_pipeline.py b/MPCAutofill/cardpicker/tests/test_run_pipeline.py index 3c9f192fd..4944a23c1 100644 --- a/MPCAutofill/cardpicker/tests/test_run_pipeline.py +++ b/MPCAutofill/cardpicker/tests/test_run_pipeline.py @@ -207,6 +207,36 @@ def cohort(db: Any) -> dict[str, Any]: return {"printing": printing, "representative": representative, "absorbed": absorbed, "lone": lone} +SHARED_MD5 = "d41d8cd98f00b204e9800998ecf8427e" + + +@pytest.fixture +def md5_group(db: Any) -> dict[str, Any]: + """ + An MD5 GROUP WHOSE MEMBERS DO NOT SHARE A PHASH - the whole point of the 2026-07-30 md5 tier, + and constructed so the phash tier provably cannot account for the result. + + `unfetchable` is created FIRST, so it holds the LOWER pk and is therefore the group's + representative under the `min(pk)` convention - while the card that actually reaches Stage D + and casts a vote is `fetched`, a NON-representative. That is deliberate: PR #660 looked for + source votes on representatives only, so this fixture is also the regression cover for the + "Stage D reached a member that is not the lowest pk and nothing propagated" defect. + + The two carry DIFFERENT `content_phash` values, so no distance-0 phash cluster exists between + them and the phash tier contributes nothing here. + """ + 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") + unfetchable = CardFactory( + name=f"{FETCH_FAILS_PREFIX} Some Card", content_phash=0x1111111111111111, md5_checksum=SHARED_MD5 + ) + fetched = CardFactory(name="Some Card", content_phash=0x2222222222222222, md5_checksum=SHARED_MD5) + return {"printing": printing, "unfetchable": unfetchable, "fetched": fetched} + + def _run(*argv: str, run_id: str = "test-monolith") -> None: call_command("run_pipeline", "--run-id", run_id, *argv) @@ -260,9 +290,12 @@ def test_a_bare_invocation_runs_every_stage_and_produces_rows( 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 + # Stage C+ ran BOTH grouping tiers and found the distance-0 pair on the phash one. + # (The counters are per-tier as of the 2026-07-30 md5 change - `md5` is the exact-identity + # tier that shares a printing vote, `phash_d0` is the pre-existing one #661 holds.) + assert counters["clustering"]["phash_d0"]["group_count"] == 1 + assert counters["clustering"]["phash_d0"]["cards_absorbed_into_groups"] == 1 + assert "md5" in counters["clustering"], "the md5 tier must run even when it finds no groups" # 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 @@ -583,12 +616,18 @@ def test_stage_d_is_handed_write_mode_explicitly_never_by_inheritance( """ seen: dict[str, Any] = {} - def _capture(batch_ids: Any, run_id: str, outcome: Any, dry_run: bool = True) -> None: + def _capture( + batch_ids: Any, run_id: str, outcome: Any, dry_run: bool = True, envelope_check: Any = None + ) -> None: seen["dry_run"] = dry_run + seen["envelope_check"] = envelope_check monkeypatch.setattr(pipeline_command, "_run_stage_d", _capture) _run() assert seen["dry_run"] is False + # The mid-pass envelope sentry is handed to Stage D too, not only used between stages - + # without it, Stage D's own multi-calculator sequence would be a single unmonitored span. + assert seen["envelope_check"] is not None seen.clear() _run("--dry-run", run_id="test-monolith-dry") @@ -603,3 +642,239 @@ def test_a_bare_run_never_trips_the_forced_dry_run_precondition(self, cohort: di """ _run() # would raise CommandError("FORCED DRY-RUN GUARD: ...") if the guard applied assert ImageEvidence.objects.filter(run_id="test-monolith").exists() + + +# ================================================================================================== +# FIX 1 - THE MD5 GROUP AS ONE UNIT (owner ruling, 2026-07-30) +# ================================================================================================== +@pytest.mark.django_db(transaction=True) +class TestMd5GroupPropagation: + """ + "The md5 dedupe should only fetch each identical image once across sources and then apply votes + to the entire group as the fetched card passes through the monolith." + + The fetch half already existed (`evidence_transfer`, keyed on md5). The VOTE half existed only + on the phash distance-0 key, so the set that got a fetch saved and the set that got a vote + propagated were different sets. These tests pin the md5 key doing the vote half, on a fixture + where the phash tier provably cannot be responsible. + """ + + def test_an_md5_twin_that_shares_no_phash_still_gets_the_groups_verdict(self, md5_group: dict[str, Any]) -> None: + """ + THE FIX, in one assertion. `unfetchable` never fetched, never extracted, and reached no + Stage D calculator with evidence - and it shares NO phash with anything, so the pre-existing + distance-0 tier cannot reach it. It is byte-identical to `fetched` (same + `Card.md5_checksum`), so it must carry `fetched`'s printing verdict, under the same + identity, at the same confidence. + + It is also the regression cover for source-vote discovery: `unfetchable` has the LOWER pk + and is therefore the group's representative, while the vote is held by `fetched`, a + non-representative. Looking for source votes on representatives only - what PR #660 did - + finds nothing here. + """ + _run() + + source = CardPrintingTag.objects.get( + card_id=md5_group["fetched"].pk, anonymous_id=JOIN_KEY_ANONYMOUS_ID, run_id="test-monolith" + ) + propagated = CardPrintingTag.objects.get( + card_id=md5_group["unfetchable"].pk, anonymous_id=JOIN_KEY_ANONYMOUS_ID, run_id="test-monolith" + ) + assert propagated.printing_id == source.printing_id + assert propagated.confidence == source.confidence + assert propagated.is_no_match is False + + # ...and it came from the md5 tier, not the phash one. Asserting only the row above would + # pass if some future change made the phash tier reach this card by another route. + ledger = PilotRunLedger.objects.get(command="run_pipeline", run_id="test-monolith-pipeline") + assert ledger.counters["clustering"]["md5"]["votes_propagated"] == 1 + assert ledger.counters["clustering"]["phash_d0"]["votes_propagated"] == 0 + + def test_the_unfetched_twin_has_no_verdict_of_its_own_without_propagation(self, md5_group: dict[str, Any]) -> None: + """ + The precondition that makes the test above mean something. With Stage C+ unwired, the md5 + twin has NO printing vote at all - so the row asserted above is genuinely produced by + propagation and is not something Stage D would have reached on its own from transferred + evidence. This is the "N independent deductions already agree" hypothesis being falsified + on the fixture rather than argued about. + """ + _run("--skip-clustering") + assert not CardPrintingTag.objects.filter(card_id=md5_group["unfetchable"].pk).exists() + + def test_propagation_never_overrides_a_members_own_ineligibility(self, md5_group: dict[str, Any]) -> None: + """ + Owner constraint: "a card excluded for a real reason stays excluded." `custom-art` is the + catalogue DECLARING that an image is not a faithful depiction of a printing. Byte identity + with a card we did identify must not overturn that - otherwise a checksum silently + outranks a human-visible declaration. + """ + unfetchable = md5_group["unfetchable"] + unfetchable.tags = ["custom-art"] + unfetchable.save(update_fields=["tags"]) + + _run() + + assert CardPrintingTag.objects.filter( + card_id=md5_group["fetched"].pk, anonymous_id=JOIN_KEY_ANONYMOUS_ID + ).exists(), "precondition: the source vote must still have been cast" + assert not CardPrintingTag.objects.filter(card_id=unfetchable.pk).exists() + + ledger = PilotRunLedger.objects.get(command="run_pipeline", run_id="test-monolith-pipeline") + assert ledger.counters["clustering"]["md5"]["members_skipped_ineligible"] == 1 + + def test_a_null_md5_is_a_group_of_one(self, md5_group: dict[str, Any]) -> None: + """Issue #473's ruling 3, inherited not re-decided: a checksum is copied from the source + listing and never invented, so cards without one group with nothing. Two cards with a NULL + md5 must NOT be treated as sharing the "same" (absent) checksum - the failure mode that + would silently fuse every `LOCAL_FILE` card in the catalogue into one group.""" + CardFactory(name="Some Card", content_phash=0x3333333333333333, md5_checksum=None) + CardFactory(name=f"{FETCH_FAILS_PREFIX} Other", content_phash=0x4444444444444444, md5_checksum=None) + + _run() + + ledger = PilotRunLedger.objects.get(command="run_pipeline", run_id="test-monolith-pipeline") + # Only the one real md5 group from the fixture, never a group of NULL-checksum cards. + assert ledger.counters["clustering"]["md5"]["group_count"] == 1 + + +# ================================================================================================== +# FIX 3 - THE ENVELOPE IS RE-SAMPLED DURING THE PASS +# ================================================================================================== +@pytest.mark.django_db(transaction=True) +class TestEnvelopeResampling: + """ + Owner: "host resampling is likely required (for steps that aren't fetch) as the same monolith + will run for small datasets and large ones so needs to fit the available compute + appropriately." PR #660 sampled once, as a preflight, and never again. + """ + + def test_the_envelope_is_re_sampled_during_the_pass_not_only_at_preflight( + self, cohort: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + """ + The interval gate is dropped to 0 so the test does not have to spend a real minute proving + the cadence exists. What is asserted is the thing that was missing: MORE THAN ONE sample. + A preflight-only command reports exactly one, forever, at any interval. + """ + monkeypatch.setattr(pipeline_command, "ENVELOPE_RESAMPLE_INTERVAL_SECONDS", 0.0) + _run() + + ledger = PilotRunLedger.objects.get(command="run_pipeline", run_id="test-monolith-pipeline") + assert ledger.counters["envelope"]["samples"] > 1 + + def test_the_interval_gate_stops_it_becoming_its_own_load( + self, cohort: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + """The other half of the owner's instruction - "do not re-sample so often it becomes its + own load". At the real 60s interval a short pass samples ONCE (the forced preflight) and + every later seam is gated out, so the seams are counted rather than queried.""" + monkeypatch.setattr(pipeline_command, "ENVELOPE_RESAMPLE_INTERVAL_SECONDS", 3600.0) + _run() + + ledger = PilotRunLedger.objects.get(command="run_pipeline", run_id="test-monolith-pipeline") + assert ledger.counters["envelope"]["samples"] == 1 + assert ledger.counters["envelope"]["skipped_by_interval"] > 1 + + def test_a_breach_appearing_mid_pass_halts_the_run_and_says_rows_were_kept( + self, cohort: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + """ + A HOST-LOAD BREACH THAT APPEARS AFTER THE PREFLIGHT MUST HALT. It must NOT be converted + into a throttle (that is rate pressure's channel, beneath Stage C, PR #644) and it must not + self-resume. The halt message must also tell the truth about what is on disk: unlike the + preflight's "nothing was written", a mid-pass halt leaves real rows behind. + """ + from cardpicker.operating_envelope import EnvelopeSignals + + monkeypatch.setattr(pipeline_command, "ENVELOPE_RESAMPLE_INTERVAL_SECONDS", 0.0) + calls = {"n": 0} + + def _clean_then_breach(*args: Any, **kwargs: Any) -> EnvelopeSignals: + calls["n"] += 1 + if calls["n"] == 1: # the preflight sees a clear box + return EnvelopeSignals(load_avg=0.5, rss_mb_per_worker=128.0) + return EnvelopeSignals(load_avg=99.0, rss_mb_per_worker=128.0) + + monkeypatch.setattr(pipeline_command, "_sample_envelope_signals", _clean_then_breach) + + with pytest.raises(CommandError) as excinfo: + _run() + + message = str(excinfo.value) + assert "ENVELOPE HALT" in message + assert "host_load" in message + assert "STAYS WRITTEN" in message, "a mid-pass halt must not claim nothing was written" + assert "--run-id test-monolith" in message, "the halt must name the resume handle" + + # The trip is DURABLE - the run cannot decide for itself that it is fine now. + from cardpicker.models import EnvelopeTrip + + assert EnvelopeTrip.objects.filter(acknowledged_at__isnull=True).exists() + + def test_skip_envelope_disables_the_mid_pass_checks_too( + self, cohort: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + """`--skip-envelope` must turn off the WHOLE bar, not just the preflight - otherwise the + flag would silently stop meaning what it says the moment re-sampling landed.""" + from cardpicker.operating_envelope import EnvelopeSignals + + monkeypatch.setattr(pipeline_command, "ENVELOPE_RESAMPLE_INTERVAL_SECONDS", 0.0) + monkeypatch.setattr( + pipeline_command, + "_sample_envelope_signals", + lambda *a, **k: EnvelopeSignals(load_avg=99.0, rss_mb_per_worker=128.0), + ) + + _run("--skip-envelope") # must not raise + + from cardpicker.models import EnvelopeTrip + + assert not EnvelopeTrip.objects.exists() + + +@pytest.mark.django_db +class TestPropagationEligibilityMatchesTheBaseQueryset: + """ + THE DRIFT TRIPWIRE for `_members_eligible_for_a_propagated_vote`. + + That method expresses four CATALOGUE-LEVEL facts (unresolved, no confirmed `canonical_card`, + `card_type=CARD`, no resolved `custom-art`/`non-english` tag) which + `local_identify_printing_tags._eligible_base_queryset` also expresses. It does not simply CALL + that function, for reasons its own docstring gives: that queryset bundles the four with + WORKLOAD rules that are wrong for a propagation target (a scan-log exclusion keyed to the + pilot's rescannable vocabulary, and a deductive-backfill exclusion that is a "don't spend a + scan" choice rather than an ineligibility). Nor can `_eligible_base_queryset` be refactored to + expose the four - its own docstring records that several tests and `stream_backstop_sweep` + assert against its COMPILED SQL, so re-ordering its `.exclude()` chain would change that SQL + for every legacy caller. + + So there are two expressions of the same four facts, and this test is what stops them drifting: + over a fixture that triggers each ineligibility reason exactly once - and that deliberately + contains NO votes, NO scan logs and NO deductive-backfill rows, so the workload excludes are + inert and the two are being compared on the four facts alone - both must return the same set. + + If this fails, the two have diverged. Fix the divergence; do not relax the assertion. + """ + + def test_the_two_expressions_of_catalogue_level_eligibility_agree(self, db: Any) -> None: + from cardpicker.local_identify_printing_tags import _eligible_base_queryset + from cardpicker.models import CardTypes, PrintingTagStatus + + eligible = CardFactory(name="Eligible Card") + resolved = CardFactory(name="Resolved Card", printing_tag_status=PrintingTagStatus.RESOLVED) + confirmed = CardFactory(name="Confirmed Card", canonical_card=CanonicalCardFactory(name="Confirmed Card")) + token = CardFactory(name="Token Card", card_type=CardTypes.TOKEN) + custom = CardFactory(name="Custom Card", tags=["custom-art"]) + foreign = CardFactory(name="Foreign Card", tags=["non-english"]) + + every_id = {c.pk for c in (eligible, resolved, confirmed, token, custom, foreign)} + + command = pipeline_command.Command() + from_propagation = command._members_eligible_for_a_propagated_vote(every_id) + from_base = set( + _eligible_base_queryset("some-identity-with-no-rows").filter(pk__in=every_id).values_list("pk", flat=True) + ) + + assert from_propagation == from_base + # ...and both actually discriminate, rather than agreeing by returning everything. + assert from_propagation == {eligible.pk} diff --git a/docs/features/printing-tags.md b/docs/features/printing-tags.md index 4b606ac17..e47803a4e 100644 --- a/docs/features/printing-tags.md +++ b/docs/features/printing-tags.md @@ -3611,6 +3611,75 @@ deduction is weaker evidence than an engine that actually looked at the image, e resolve - same consensus/gate-check discipline as every other engine in this module, same batch-flush checkpointing pattern as `run_pilot`. +#### The counting gate was not sufficient on its own (2026-07-30) + +Owner ruling: _"just because a card was printed exactly once doesn't mean that the image in our +catalogue is an accurate depiction of that card, it may have a different border or another +issue."_ + +Everything the 1:1 gate checks is a **count**. Counting establishes that _if_ this card is a +depiction of one of the name's printings, _then_ it must be the uncovered one. It establishes +nothing about the antecedent, and the only filters that spoke to it at all were the **declared** +`custom-art` and non-English tags — so an altered border, a custom frame, or a misnamed upload +that nobody had tagged yet passed the gate and received a full-confidence structural vote. + +"It is only a vote" is a weaker defence than it sounds. Issue #593 established that a machine +vote is what the question feed renders as _the suggestion to confirm_, and a human's click +returns as a full-weight USER vote — so a visually-unverified deduction becomes a one-click +rubber stamp, and the human-backed consensus gate is the mechanism that launders it. + +**The missing conjunct is now required**: the card's already-stored evidence must be consistent +with the candidate printing — border class and frame style, via +`local_identify_printing_tags.printing_attribute_disagreement`, which is the _same_ check the +Stage D join-key channel has always applied, now shared between the two callers rather than +re-derived. No image fetch and no new extraction: every input is a field already sitting on the +card's current `ImageEvidence` row. Sharing it also inherits PR #656's gate for free — the frame +half is skipped entirely unless `artist_ocr` actually ran, because `illus_anchor_fired` is +nullable and `bool(None)` is indistinguishable from "ran and found no anchor". + +**No stored evidence means abstain, not proceed.** If the card has never been extracted, we have +never looked at the image, so the antecedent is exactly as unestablished as before. This is the +one place where this module's usual "missing data is not evidence" rule points the other way, and +deliberately so: that rule protects a match from being _vetoed_ by silence, whereas here silence +is being asked to _establish_ something. The two abstention populations are reported separately +(`abstained_no_evidence`, `abstained_attribute_mismatch`) so the cost of the gate is legible on +the first run rather than inferred from a smaller total. + +The tier was **not** dropped, although it could have been cleanly: it has never run in production +(zero `PilotRunLedger` rows for `local_name_frequency_elimination`, ever), so nothing is +contaminated and no retraction was needed. It is kept because the deduction is genuinely sound +once the antecedent is established, and elimination reaches a population the image-based channels +structurally cannot — a name whose single uncovered printing has no distinguishing collector line +to read. Deleting a sound tier for want of a guard is a worse trade than adding the guard. + +#### The census was leaking (2026-07-30) + +Separately from soundness, the 1:1 gate was counting over a population it was itself permanently +shrinking. `_eligible_base_queryset` was called with **no `run_id`**, so its exclusion of cards +already carrying this calculator's vote was **lifetime** rather than per-run: + +- **run 1** — a name has one unresolved eligible card and one uncovered printing. Sound; it votes. +- **later** — a second upload of that name arrives from another source, the ordinary way this + catalogue grows. +- **run 2** — the run-1 card is excluded _forever_ by its own vote, so the name presents as having + exactly one unresolved card again and votes for the new one. Correctly, the pool is two cards + and the gate should abstain. + +That is a **fresh wrong positive**, not a stale or missed vote: nothing about the second card +changed, only the size of the population the gate counts. The fix is to pass this run's `run_id`, +which narrows the self-suppressing excludes to rows _this_ run wrote. Within-run resume is +unaffected — re-invoking with the same `run_id` still skips cards this run already voted on. + +`compute_covered_printing_pks()` stays catalogue-wide and un-scoped, deliberately: "covered" is a +fact about the world (a confirmed `canonical_card`, or a RESOLVED `inferred_canonical_card`), not +about this calculator's progress, and run-scoping it would make every run treat every printing as +uncovered. The two halves of the gate are scoped differently on purpose. + +`run_pilot`'s own `select_candidates` and `count_below_resolution_floor` are **left unscoped** — +neither gates on a count over the returned population (the pilot's predicate is per-card and its +selection is a fetch-budget ordering; the floor count is a report metric), so neither has this +defect. + ## Incident: per-chunk thread pool leaked Postgres connections, crashed the live run (2026-07-16) The second full-catalog relaunch (post cluster-dedup removal) died ~3 minutes in with diff --git a/docs/features/stage-e-operations.md b/docs/features/stage-e-operations.md index 48c8ac198..12ecd8e23 100644 --- a/docs/features/stage-e-operations.md +++ b/docs/features/stage-e-operations.md @@ -120,6 +120,58 @@ and never touches that window. **This is a narrowing, not a weakening.** Every bar still halts for everything it ever halted for, _except_ a destination asking us to slow down. +### The monolith re-samples the envelope during a pass (2026-07-30) + +Owner ruling: _"host resampling is likely required (for steps that aren't +fetch) as the same monolith will run for small datasets and large ones so needs +to fit the available compute appropriately."_ + +`manage.py run_pipeline` shipped (PR #660) with the envelope checked **once**, +as a preflight, before Stage C — and never again. That is correct for a +200-card run and wrong for a 230,000-card one: the same command has to fit +whatever compute is free **during** the pass, not what happened to be free at +launch. + +The envelope is now re-sampled at every stage seam: after Stage C, between each +of Stage D's calculators and chip casters, and before each Stage C+ propagation +tier. `PilotRunLedger.counters["envelope"]` records `samples` and +`skipped_by_interval`, so a run reporting one sample is visibly a run that never +re-sampled. + +**Three separate protections, and this is only the middle one.** They are easy +to conflate: + +| protection | protects | where it lives | +| ---------------- | --------------- | ------------------------------------------ | +| 7/sec ceiling | Google | `harvest_rate_coordinator`, per request | +| **the envelope** | **this host** | **`operating_envelope`, sampled per seam** | +| compute gating | everything else | `stage_e_batch_sizing` / concurrency slots | + +**Halt semantics are unchanged, and a re-sample is not a throttle.** A genuine +breach found mid-pass still persists an `EnvelopeTrip`, still exits 3, and still +requires `resolve_envelope_trip` — no self-resume. Converting a host-load breach +into a throttle would undo the 429/503 distinction above from the opposite +direction: going slower on Google does not reduce this box's own load. + +**What a mid-pass halt leaves behind differs from the preflight's, and the +message says so.** The preflight can honestly say "nothing was written"; by the +time a re-sample fires, real rows exist. They stay — every one stamped with the +run's `run_id` — and the run resumes with `--run-id ` once the trip is +acknowledged. + +**Cadence: 60 seconds minimum between samples**, interval-gated so the check +cannot become its own load. The number is derived rather than tuned by feel: the +host-load bar compares against the **one-minute** load average +(`os.getloadavg()[0]`), so sampling faster re-reads a number that has not +finished moving — it cannot detect a breach any earlier, it only multiplies DB +round trips. + +**Known residual.** The seams are _between_ calculators, not inside them. Each +Stage D calculator owns its own internal batching, so at catalogue scale a +single calculator can run for a long time between checks. Closing that gap means +threading a progress callback into each calculator's own batch loop — a real +refactor of shared code, deliberately not done here. + **The ceiling itself is now 7 req/sec** (`harvest_fetch_limiter.GOOGLE_IMAGE .rate_per_sec`, down from 8.0). The "**or hardware, whichever comes first**" half of the ruling needs no second number and deliberately does not get one: the limiter is a strict _minimum-interval_ pacer, so it can only ever **delay** diff --git a/docs/identification-pipeline.md b/docs/identification-pipeline.md index d91689160..a0873fcf1 100644 --- a/docs/identification-pipeline.md +++ b/docs/identification-pipeline.md @@ -104,14 +104,54 @@ executed: ``` Stage 0 Scryfall reference refresh, once, at the front -Stage E operating-envelope preflight +Stage E operating-envelope preflight, then RE-SAMPLED at every stage seam 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 C+ group vote propagation — md5 first, then phash distance-0 Stage E fidelity gate — machine-only resolutions must be zero end channel_report ``` +### Stage C+ — the md5 group behaves as one unit + +An **md5 group** is a set of cards whose stored `Card.md5_checksum` is +identical: the same bytes, uploaded by different sources. Two things now key on +that same group, which is the whole point: + +- **one fetch per group** — `evidence_transfer` copies an md5-sibling's + evidence instead of re-fetching (this half already existed); +- **one deduction per group** — whichever member Stage D reached casts the + verdict, and Stage C+ applies it across the rest of the group, under the + casting calculator's own identity and with its already-voted guard intact. + +Before this, the fetch half keyed on md5 and the vote half keyed on phash +distance-0, so _the set that got a fetch saved and the set that got a vote +propagated were different sets_. Byte-identical files always share a phash; +files sharing a phash are not necessarily byte-identical. + +Propagation is **not** redundant with each member deducing for itself off +transferred evidence, which is worth stating because it looks like it should +be. A Stage D printing deduction is not a function of the evidence row alone: +candidates are resolved from `Card.name`, and md5-identical uploads from +different sources routinely carry different names. Members therefore reach +genuinely different conclusions — or none at all — from byte-identical +evidence. + +Propagation **never overrides a member's own ineligibility**. A member that is +already resolved, already confirmed to a `canonical_card`, not a `CARD`, or +carrying a resolved `custom-art` / `non-english` tag is skipped. `custom-art` is +the catalogue declaring the image is _not_ a faithful depiction of a printing, +and a checksum must not overturn that. + +**The phash distance-0 tier still runs, second, unchanged.** Its future is +issue #661: the intended direction is that phash shares an _illustration_ (same +artwork, possibly a different printing — a near-identity claim at a grain where +a weaker claim is appropriate), not a printing verdict. Until that is built it +stays, because it is the only propagation reaching cards with no md5 at all — +md5 is NULL for every `LOCAL_FILE` source by design and is never invented. Both +tiers call one propagation engine that takes the grouping as a parameter, so +adding the illustration-grain tier later is a new grouping, not a restructure. + 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