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