Add Stage C evidence transfer, decoupled fetch-ahead, and echo suppression

MPCAutofill/MPCAutofill/settings.py

@@ -442,6 +442,17 @@
 # action phase 3 needs to take to go live - no redeploy of this module required.
 STAGE_E_STREAMING_ENABLED = env.bool("STAGE_E_STREAMING_ENABLED", default=False)
 
+# Stage C evidence transfer kill-switch (issue #473 PR-2, Tron §8 gate condition, 2026-07-25) -
+# cardpicker/evidence_transfer.py's own module docstring has the full mechanism writeup. Default
+# `True` (transfer is ON by default for this first pass) so the feature actually runs without an
+# operator having to flip it on first - `False` makes `find_transfer_source` return `None`
+# unconditionally (no query issued), so both call sites (`stage_e_dispatch._run_stage_c`,
+# `run_image_evidence_cohort._fetch_one_card`) fall straight through to their own pre-existing
+# real-fetch path, exactly as if this feature didn't exist. Exists for first-pass reversibility -
+# a single settings flip isolates whether a live-run anomaly originates in transfer, no code change
+# or redeploy needed.
+STAGE_C_EVIDENCE_TRANSFER_ENABLED = env.bool("STAGE_C_EVIDENCE_TRANSFER_ENABLED", default=True)
+
 # Micro-batch size (docs/proposals/stage-e-streaming.md §3 decision (2), sharpened by §10(c)): NOT
 # a value this brief or this change invents precision for - §10(c) ratifies that the real number
 # ships as a MEASURED OUTPUT of the Bug-A tail shakedown's own instrumentation (phase 3, not yet

MPCAutofill/cardpicker/image_evidence.py

@@ -188,6 +188,8 @@
 
 import imagehash
 
+from django.db.models import Q, QuerySet
+
 from cardpicker.harvest_fetch_limiter import GoogleFetchLockoutError
 from cardpicker.image_cdn_fetch import DEFAULT_FETCH_DPI, fetch_card_image
 from cardpicker.local_fallback import (
@@ -461,6 +463,8 @@ def extract_card_evidence(
         profile=profile,
         short_circuit=short_circuit,
         known_set_codes=known_set_codes,
+        md5_checksum=card.md5_checksum,
+        sha256_checksum=card.sha256_checksum,
     )
 
 
@@ -472,6 +476,8 @@ def compute_card_evidence(
     profile: Optional[dict[str, float]] = None,
     short_circuit: Optional[bool] = None,
     known_set_codes: Optional[frozenset[str]] = None,
+    md5_checksum: Optional[str] = None,
+    sha256_checksum: Optional[str] = None,
 ) -> ExtractionResult:
     """
     Compute-only continuation of `extract_card_evidence` above - everything that function does
@@ -554,11 +560,27 @@ def compute_card_evidence(
     the gate entirely - every parse is accepted exactly as before, the pre-2026-07-23 behavior -
     so this is purely additive: a card whose first parse is already lexicon-valid (the overwhelming
     majority) sees zero behavior or compute change either way.
+
+    `md5_checksum`/`sha256_checksum` (2026-07-25, issue #473 PR-2, folded with issue #472): the
+    calling card's own live `Card.md5_checksum`/`Card.sha256_checksum` at the moment of THIS real
+    extraction pass, stamped verbatim onto the result's `fields` (so
+    `persist_evidence` writes them the same way every other field is written - no special-casing).
+    `None` (the default) is every pre-#473 caller's own behavior, unchanged - both fields are
+    nullable and null-tolerant everywhere they're read (`evidence_transfer.md5_currency_q`,
+    `current_evidence_queryset` below). Real extraction ALWAYS re-stamps to the card's own live
+    values regardless of what a prior row (including a prior TRANSFERRED row - see
+    `evidence_transfer.transfer_evidence`'s own docstring) happened to carry - this is the
+    "computed-once-forever, but re-extraction always re-stamps the truth" half of the staleness fix
+    the transfer half's own stamping mirrors.
     """
     if short_circuit is None:
         short_circuit = _short_circuit_enabled_by_env()
 
-    fields: dict[str, Any] = {"fetch_latency_ms": fetch_latency_ms}
+    fields: dict[str, Any] = {
+        "fetch_latency_ms": fetch_latency_ms,
+        "md5_checksum": md5_checksum,
+        "sha256_checksum": sha256_checksum,
+    }
     extractor_versions: dict[str, str] = {}
     skip_reasons: dict[str, str] = {}
     if profile is not None:
@@ -916,6 +938,52 @@ def compute_card_evidence(
     )
 
 
+def current_evidence_queryset(card: Card) -> "QuerySet[ImageEvidence]":
+    """
+    THE single "which `ImageEvidence` row(s) are CURRENT for this card" queryset (2026-07-25,
+    issue #473 PR-2's staleness fix) - replaces what used to be N independent inline copies of
+    `ImageEvidence.objects.filter(card_id=card.pk, content_hash=card.content_phash)` scattered
+    across `local_calculate_verdicts.py` (all three Stage D calculators),
+    `local_layout_class_cast.py`, `local_detect_ai_art.py`, `local_lands_identify.py`'s own
+    `_current_evidence_for_card`, and `reparse_collector_evidence.py`'s own same-named function -
+    one shared definition, so the staleness rule below can never drift between call sites the way
+    that many independent inline copies eventually would have.
+
+    Currency = TWO conditions, BOTH required:
+
+    1. `content_hash` matches the card's own LIVE `content_phash` (the pre-existing rule, unchanged
+       - an evidence row from a prior image upload is never reused once the upload changes).
+    2. (2026-07-25, #473 PR-2) The row's own STAMPED `md5_checksum` agrees with `Card.md5_checksum`
+       WHENEVER BOTH are non-null - closes the silent in-place-file-replacement hole a
+       content_phash-only currency check can miss (a source file replaced at the same Drive
+       location changes the Drive `md5Checksum` on the next listing walk without necessarily
+       producing a different perceptual hash on every re-fetch, e.g. a lightly re-encoded but
+       visually-identical re-upload).
+
+    NULL-TOLERANT BY DESIGN, stated explicitly per this PR's own scope: a legacy evidence row
+    written before the `md5_checksum` stamp existed (`row.md5_checksum is None`) OR a card whose
+    source never carries an md5 at all (`card.md5_checksum is None`, e.g. `LOCAL_FILE`) stays
+    CURRENT under condition 1 alone - only a row that stamped a REAL md5 which now actively
+    DISAGREES with the card's own live md5 is excluded. Forcing every legacy row to fail currency
+    the day this ships would be its own multi-day mass-recompute for zero new information about
+    those specific cards; this fix is scoped to catching a REAL disagreement, not to retroactively
+    distrusting everything that predates the stamp.
+
+    `card.content_phash is None` (no stable hash yet) always returns an empty queryset - every
+    existing call site already guarded this case itself before this function existed (the "no
+    stable hash yet to key a CURRENT ImageEvidence lookup against" comment repeated at each one),
+    but this function enforces it directly too rather than trusting every future caller to keep
+    doing so - `ImageEvidence.content_hash` is a non-nullable column, so a `None` here could never
+    have matched a real row anyway.
+    """
+    if card.content_phash is None:
+        return ImageEvidence.objects.none()
+    qs = ImageEvidence.objects.filter(card_id=card.pk, content_hash=card.content_phash)
+    if card.md5_checksum is not None:
+        qs = qs.filter(Q(md5_checksum__isnull=True) | Q(md5_checksum=card.md5_checksum))
+    return qs
+
+
 def persist_evidence(result: ExtractionResult, run_id: Optional[str] = None) -> Optional[ImageEvidence]:
     """
     The thin, separate DB-write step (see module docstring for why this is split from
@@ -933,6 +1001,16 @@ def persist_evidence(result: ExtractionResult, run_id: Optional[str] = None) ->
     local_identify_printing_tags.py, local_residual_classify.py, local_layout_class_cast.py,
     local_detect_ai_art.py, local_lands_identify.py) - this was the one outlier (2026-07-24 IO
     audit, finding 3).
+
+    `evidence.transferred`/`transferred_from_card_id` are unconditionally reset to
+    `False`/`None` here (2026-07-25, issue #473 PR-2) - `persist_evidence` is called ONLY for a
+    REAL extraction pass (`evidence_transfer.transfer_evidence` is the separate, only other writer
+    of an `ImageEvidence` row, and it never calls this function), so every call here represents
+    genuine fresh extraction. A row that was previously TRANSFERRED (`transferred=True`) and later
+    receives a real extraction pass (e.g. `stage_e_shakedown`'s own `force_stage_c_reextract`) is
+    no longer a transferred row once this returns - leaving the flag stale would wrongly keep it
+    excluded from Stage D machine voting (the interim guard, `local_calculate_verdicts.
+    _eligible_cards_queryset`) even though it now carries a genuine independent extraction.
     """
 
     if result.content_hash is None:
@@ -944,6 +1022,8 @@ def persist_evidence(result: ExtractionResult, run_id: Optional[str] = None) ->
         setattr(evidence, field_name, value)
     evidence.extractor_versions = {**evidence.extractor_versions, **result.extractor_versions}
     evidence.run_id = run_id
+    evidence.transferred = False
+    evidence.transferred_from_card_id = None
     evidence.save()
 
     scan_log_batch = [
@@ -1015,6 +1095,7 @@ def build_reconciliation_report(
     "ExtractionResult",
     "extract_card_evidence",
     "compute_card_evidence",
+    "current_evidence_queryset",
     "persist_evidence",
     "ReconciliationReport",
     "build_reconciliation_report",

MPCAutofill/cardpicker/local_calculate_verdicts.py

@@ -379,6 +379,7 @@
 
 from django.db.models import Count, Max, Q, QuerySet
 
+from cardpicker.image_evidence import current_evidence_queryset
 from cardpicker.local_fallback import (
     FALLBACK_CONFIDENCE_MULTI_EVIDENCE,
     FALLBACK_CONFIDENCE_SINGLE_EVIDENCE,
@@ -456,11 +457,32 @@
 # two already-established tiers immediately above and below it.
 JOIN_KEY_CONFIDENCE_ARTIST_DISAGREEMENT = 0.65
 
+# INTERIM STAGE D GUARD (issue #473 PR-2, TEMPORARY BY DESIGN - see `ImageEvidence.transferred`'s
+# own model-field docstring and `evidence_transfer.transfer_evidence`'s own docstring for the full
+# rationale): a card whose CURRENT evidence row was created by `evidence_transfer.transfer_evidence`
+# rather than a real per-card extraction pass is excluded from the two MACHINE-VOTING Stage D
+# calculators below (join-key/fallback - both cast a `CardPrintingTag` vote) - its own "machine
+# observation" is the SAME bytes an md5-sibling card already voted from, not an independent one,
+# so casting a vote from it here would fabricate the independence the vote-weight matrix assumes
+# is real. The THIRD Stage D calculator, slow-path, is deliberately NOT guarded - it casts no
+# machine vote at all, only a `CardScanLog` routing marker handing the card to a HUMAN reviewer
+# (see `run_slow_path_calculator`'s own loop comment), which is exactly the safety net this guard
+# exists to preserve, not a case it needs to protect against. RESCANNABLE (a future real extraction
+# pass, or PR-3's own group-level vote pooling landing and removing this guard entirely, both
+# un-stick a card stuck here) - included in each of the two guarded calculators' own
+# RESCANNABLE_SKIP_REASONS set below. ISSUE #473's OWN COORDINATION NOTE (PR-3 build-plan section):
+# "Removes PR-2's interim Stage D guard" - do not remove this guard, or the `transferred` flag it
+# reads, before PR-3
+# (group-level vote pooling) actually merges and the group-aware calculators no longer need it.
+TRANSFERRED_INTERIM_GUARD_SKIP_REASON = "transferred-interim-guard"
+
 # A degenerate/skip outcome that stays eligible for re-selection on a future invocation, same
 # convention as local_identify_printing_tags.RESCANNABLE_SKIP_REASONS - "no-evidence" here
 # because ImageEvidence simply hadn't been extracted yet for this card at selection time is a
 # transient state (a future extraction run may still land it), not a permanent conclusion.
-JOIN_KEY_RESCANNABLE_SKIP_REASONS = frozenset({"no-evidence"})
+# TRANSFERRED_INTERIM_GUARD_SKIP_REASON (above) is rescannable for the same reason - both a real
+# extraction landing later and PR-3's own guard removal un-stick a card stuck here.
+JOIN_KEY_RESCANNABLE_SKIP_REASONS = frozenset({"no-evidence", TRANSFERRED_INTERIM_GUARD_SKIP_REASON})
 
 # 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"
@@ -1197,7 +1219,7 @@ def run_join_key_calculator(
             continue  # no stable hash yet to key a CURRENT ImageEvidence lookup against
 
         evidence = (
-            ImageEvidence.objects.filter(card_id=card.pk, content_hash=card.content_phash)
+            current_evidence_queryset(card)
             .filter(extractor_versions__has_key="collector_line_ocr")
             .order_by("-updated_at")
             .first()
@@ -1212,6 +1234,23 @@ def run_join_key_calculator(
                 )
             continue
 
+        # INTERIM STAGE D GUARD (issue #473 PR-2, temporary by design - see
+        # TRANSFERRED_INTERIM_GUARD_SKIP_REASON's own module-level comment above).
+        if evidence.transferred:
+            result.skip_counts[TRANSFERRED_INTERIM_GUARD_SKIP_REASON] = (
+                result.skip_counts.get(TRANSFERRED_INTERIM_GUARD_SKIP_REASON, 0) + 1
+            )
+            if not dry_run:
+                scan_log_batch.append(
+                    CardScanLog(
+                        card_id=card.pk,
+                        anonymous_id=JOIN_KEY_ANONYMOUS_ID,
+                        run_id=run_id,
+                        skip_reason=TRANSFERRED_INTERIM_GUARD_SKIP_REASON,
+                    )
+                )
+            continue
+
         result.cards_considered += 1
         if index is None:
             index = _get_cached_candidate_name_index()
@@ -1293,7 +1332,9 @@ def run_join_key_calculator(
 # two carry the same meaning here as there, no rename needed.
 FALLBACK_NO_EVIDENCE_SKIP_REASON = "no-evidence"  # this calculator's own ImageEvidence-row-missing case, same meaning as JOIN_KEY's own identical string, different anonymous_id scope
 FALLBACK_NO_SUB_CHECK_EVIDENCE_SKIP_REASON = "no-sub-check-evidence"  # local_fallback.FallbackOutcome's own "no-evidence" concept, renamed to avoid colliding with the line above
-FALLBACK_RESCANNABLE_SKIP_REASONS = frozenset({FALLBACK_NO_EVIDENCE_SKIP_REASON})
+# TRANSFERRED_INTERIM_GUARD_SKIP_REASON (module-level comment above, issue #473 PR-2) is
+# rescannable here too - same reasoning as JOIN_KEY_RESCANNABLE_SKIP_REASONS' own inclusion of it.
+FALLBACK_RESCANNABLE_SKIP_REASONS = frozenset({FALLBACK_NO_EVIDENCE_SKIP_REASON, TRANSFERRED_INTERIM_GUARD_SKIP_REASON})
 
 
 @dataclass(frozen=True)
@@ -1511,7 +1552,7 @@ def run_fallback_calculator(
             continue  # no stable hash yet to key a CURRENT ImageEvidence lookup against
 
         evidence = (
-            ImageEvidence.objects.filter(card_id=card.pk, content_hash=card.content_phash)
+            current_evidence_queryset(card)
             .filter(extractor_versions__has_key="collector_line_ocr")
             .order_by("-updated_at")
             .first()
@@ -1531,6 +1572,23 @@ def run_fallback_calculator(
                 )
             continue
 
+        # INTERIM STAGE D GUARD (issue #473 PR-2, temporary by design - see
+        # TRANSFERRED_INTERIM_GUARD_SKIP_REASON's own module-level comment above).
+        if evidence.transferred:
+            result.skip_counts[TRANSFERRED_INTERIM_GUARD_SKIP_REASON] = (
+                result.skip_counts.get(TRANSFERRED_INTERIM_GUARD_SKIP_REASON, 0) + 1
+            )
+            if not dry_run:
+                scan_log_batch.append(
+                    CardScanLog(
+                        card_id=card.pk,
+                        anonymous_id=STAGE_D_FALLBACK_ANONYMOUS_ID,
+                        run_id=run_id,
+                        skip_reason=TRANSFERRED_INTERIM_GUARD_SKIP_REASON,
+                    )
+                )
+            continue
+
         result.cards_considered += 1
         if index is None:
             index = _get_cached_candidate_name_index()
@@ -1802,8 +1860,14 @@ def run_slow_path_calculator(
         if card.content_phash is None:
             continue  # no stable hash yet to key a CURRENT ImageEvidence lookup against
 
+        # NOTE (issue #473 PR-2): the interim Stage D guard (TRANSFERRED_INTERIM_GUARD_SKIP_REASON,
+        # see its own module-level comment) deliberately does NOT apply here - this calculator
+        # casts no machine vote at all, only a CardScanLog routing marker that hands the card to a
+        # HUMAN reviewer. A human looking at transferred-evidence-derived signals is exactly the
+        # safety net the guard exists to preserve, not a case it needs to protect against - the
+        # fabricated-independence risk is specific to an automated vote, never a human decision.
         evidence = (
-            ImageEvidence.objects.filter(card_id=card.pk, content_hash=card.content_phash)
+            current_evidence_queryset(card)
             .filter(extractor_versions__has_key="collector_line_ocr")
             .order_by("-updated_at")
             .first()
@@ -1862,6 +1926,7 @@ def run_slow_path_calculator(
     "JOIN_KEY_NO_MATCH_CONFIDENCE",
     "JOIN_KEY_CONFIDENCE_ARTIST_DISAGREEMENT",
     "JOIN_KEY_RESCANNABLE_SKIP_REASONS",
+    "TRANSFERRED_INTERIM_GUARD_SKIP_REASON",
     "JOIN_KEY_NO_HIT_SKIP_REASONS",
     "JOIN_KEY_UNKNOWN_SET_CODE_SKIP_REASON",
     "known_set_codes",

MPCAutofill/cardpicker/local_detect_ai_art.py

@@ -66,6 +66,7 @@
 
 from django.db.models import QuerySet
 
+from cardpicker.image_evidence import current_evidence_queryset
 from cardpicker.local_identify_printing_tags import generate_run_id
 from cardpicker.models import (
     Card,
@@ -393,11 +394,7 @@ def run_ai_art_detector(
         if card.content_phash is None:
             continue  # no stable hash yet to key a CURRENT ImageEvidence lookup against
 
-        evidence = (
-            ImageEvidence.objects.filter(card_id=card.pk, content_hash=card.content_phash)
-            .order_by("-updated_at")
-            .first()
-        )
+        evidence = current_evidence_queryset(card).order_by("-updated_at").first()
         if evidence is None:
             result.skip_counts["no-evidence"] = result.skip_counts.get("no-evidence", 0) + 1
             if not dry_run: