diff --git a/MPCAutofill/MPCAutofill/settings.py b/MPCAutofill/MPCAutofill/settings.py index e041b1c95..4719032cd 100755 --- a/MPCAutofill/MPCAutofill/settings.py +++ b/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 diff --git a/MPCAutofill/cardpicker/evidence_transfer.py b/MPCAutofill/cardpicker/evidence_transfer.py new file mode 100644 index 000000000..4ea3ea7d7 --- /dev/null +++ b/MPCAutofill/cardpicker/evidence_transfer.py @@ -0,0 +1,295 @@ +""" +Stage C evidence transfer (issue #473 PR-2, folded with issue #472 per the owner-approved +2026-07-25 fold - "same function, one coherent change"). Before a card with a known +`Card.md5_checksum` pays for a real fetch+extraction pass, check whether an md5-identical sibling +already holds a CURRENT, full-manifest `ImageEvidence` row - if so, copy that row's own field +values onto this card's own `(card, content_hash)` row instead of re-doing byte-identical work. +Two callers, one function: the streaming conveyor's `stage_e_dispatch._run_stage_c` (checked +BEFORE handing a card to the decoupled fetch-ahead thread, #472) and the BULK +`run_image_evidence_cohort.py` fetch stage (`_fetch_one_card`, checked BEFORE the network fetch +call) - see each call site's own docstring for exactly where this is wired in. + +KILL-SWITCH (Tron §8 gate condition, 2026-07-25): `settings.STAGE_C_EVIDENCE_TRANSFER_ENABLED` +(default `True`) gates `find_transfer_source` at its very top - `False` makes every call return +`None` immediately, with no query issued at all, so both seams fall straight through to their own +pre-existing real-fetch path exactly as if this module didn't exist. Exists for first-pass +reversibility - a single settings flip turns transfer off catalog-wide with no code change, if a +live run needs to isolate whether an anomaly originates here. + +GROUP KEY: md5 alone (issue #473's parent ruling 3 - a NULL or unique md5 is a "group of one"; +every card without a known md5 falls straight through `find_transfer_source` returning `None`, +i.e. today's unconditional fetch/extract behavior, unchanged). + +PAIRING RULE (binding, issue #473's 2026-07-25 comment, owner-ratified alongside PR-1's +`Card.sha256_checksum` addition, now a real column on every deploy this branch runs against): +a byte-identical claim this strong needs a cryptographic backstop before this module trusts it +enough to SKIP real extraction. Whenever BOTH cards carry a sha256, it must ALSO match. An md5 +match with a present-on-both sha256 MISMATCH is a loud anomaly (logged at ERROR AND written as a +durable `CardScanLog` row, see ANOMALY LOGGING below) - transfer is skipped and the caller falls +through to real extraction, never a silent downgrade to an md5-only transfer. md5 collisions are +constructible; a present sha256 is exactly the case this module has enough information to verify +the "identical bytes" premise cryptographically instead of assuming it, so it always overrides an +md5-only pairing whenever both sides have one. When sha256 is absent on either side, the pairing +rests on md5 + the content-hash assertion below alone. + +CONTENT-HASH ASSERTION: byte-identical files imply an identical perceptual hash (phash is a pure +function of pixel content - identical bytes decode to identical pixels). This is ASSERTED, not +merely trusted: the sibling's own `content_hash` (guaranteed by the currency query below to equal +the sibling CARD's own live `content_phash`) is compared against THIS card's own `content_phash`. +A mismatch here is impossible for two cards that are genuinely byte-identical, so observing one is +evidence of a REAL anomaly (a stale/incorrect md5 pairing, an actual md5 collision, or a +data-entry error) - logged loudly at ERROR, transfer skipped, never transferred anyway. + +TRANSFER-SOURCE INTEGRITY (Tron §8 gate condition, 2026-07-25 - tightened from an earlier draft +that reused the null-tolerant CURRENCY rule for this too): a sibling is only a valid transfer +SOURCE if its own stamped `md5_checksum` IS NOT NULL and EQUALS the target's own live md5 - +`image_evidence.current_evidence_queryset`'s null-tolerant rule (a legacy unstamped row stays +CURRENT for its own card) is a currency notion, never a transfer-source-eligibility one; minting +a fresh stamp on the copy from a source that never carried one at all would launder an unverified +value into a verified-looking field. See `_current_sibling_evidence_queryset`'s own docstring. + +ANOMALY LOGGING (Tron §8 gate condition, 2026-07-25): both anomaly paths in `find_transfer_source` +now write a durable `CardScanLog(anonymous_id=EVIDENCE_TRANSFER_ANONYMOUS_ID, skip_reason=)` row in addition to the `logger.error` call - the log line alone isn't queryable +after the fact; a whole-catalog run needs to be able to COUNT how many cards hit each anomaly, +per card, the same way every other named-skip population in this codebase is counted (`CardScanLog` +is already the established "durable negative record" primitive - see that model's own docstring). + +Neither anomaly path raises - both return `None` from `find_transfer_source`, and every caller's +own contract for a `None` result is "fall through to real extraction" (never "give up on this +card"), so a loud anomaly degrades to the pre-existing, already-correct behavior rather than +failing the whole batch/card. +""" + +import logging +from typing import Optional + +from django.conf import settings +from django.db.models import F, Q, QuerySet + +from cardpicker.models import Card, CardScanLog, ImageEvidence + +logger = logging.getLogger(__name__) + +# This module's own CardScanLog anonymous_id (issue #473 PR-2's ANOMALY LOGGING section above) - +# distinct from every extractor's own anonymous_id (image_evidence.py) and every Stage D +# calculator's own (local_calculate_verdicts.py), matching this codebase's own "one anonymous_id +# per distinct write population" convention. +EVIDENCE_TRANSFER_ANONYMOUS_ID = "evidence-transfer-v1" +EVIDENCE_TRANSFER_SHA256_MISMATCH_SKIP_REASON = "transfer-sha256-mismatch" +EVIDENCE_TRANSFER_CONTENT_HASH_MISMATCH_SKIP_REASON = "transfer-content-hash-mismatch" + +# Every concrete ImageEvidence field that represents actual extracted content - copied verbatim +# from the sibling row onto the target's own row. Deliberately EXCLUDES: "id" (a fresh row gets +# its own pk via get_or_create), "card"/"card_id" (the target, never the sibling), "content_hash" +# (the TARGET's own content_phash - already the get_or_create key, never the sibling's), "run_id" +# (set fresh by the caller, not copied), "created_at"/"updated_at" (Django-managed), and this PR's +# own four new fields (md5_checksum/sha256_checksum/transferred/transferred_from_card_id) - each +# of those four is set explicitly from the TARGET card's own values by `transfer_evidence` below, +# never copied from the sibling (see that function's own docstring for why). +_NON_TRANSFERABLE_FIELD_NAMES = frozenset( + { + "id", + "card", + "content_hash", + "run_id", + "created_at", + "updated_at", + "md5_checksum", + "sha256_checksum", + "transferred", + "transferred_from_card_id", + } +) +_TRANSFERABLE_FIELD_NAMES = [ + f.name + for f in ImageEvidence._meta.get_fields() + if getattr(f, "concrete", False) and f.name not in _NON_TRANSFERABLE_FIELD_NAMES +] + + +def md5_currency_q(related_prefix: str = "card__") -> Q: + """ + Bulk (F-expression-based) Q object expressing the md5 half of `ImageEvidence`'s own CURRENCY + rule (see `image_evidence.current_evidence_queryset`'s own docstring for the single-card, + literal-value form of the identical rule) - True (row stays CURRENT/eligible) unless the row's + own stamped `md5_checksum` is non-null AND the related Card's live `md5_checksum` (reached via + `related_prefix`, default `"card__"` i.e. `ImageEvidence.card`) is ALSO non-null AND the two + disagree. Three-clause OR, not a single equality check, specifically so a NULL on either side + (a legacy unstamped row, or a card whose source never carries an md5 at all) never excludes a + row - SQL's three-valued logic means a bare `md5_checksum=F(...)` comparison would silently + evaluate to NULL/false whenever either side is NULL, which is the OPPOSITE of the null-tolerant + behavior this rule requires. + + NULL-TOLERANT BY DESIGN FOR CURRENCY ONLY (Tron §8 gate condition, 2026-07-25 - see module + docstring's "TRANSFER-SOURCE INTEGRITY" section): this function is used by + `image_evidence.current_evidence_queryset` and `modern_artist_credit.py`'s own bulk currency + read, NEVER by this module's own `_current_sibling_evidence_queryset` below - a transfer + SOURCE's own eligibility requires a strict non-null equality match instead, applied inline + there rather than through this null-tolerant helper. + """ + return ( + Q(md5_checksum__isnull=True) + | Q(**{f"{related_prefix}md5_checksum__isnull": True}) + | Q(md5_checksum=F(f"{related_prefix}md5_checksum")) + ) + + +def _current_sibling_evidence_queryset(card: Card) -> "QuerySet[ImageEvidence]": + """Every OTHER card's own CURRENT, full-manifest `ImageEvidence` row sharing `card`'s own + md5_checksum - "current" here means the sibling row's own `content_hash` matches ITS OWN + card's live `content_phash` (the pre-existing currency rule). TRANSFER-SOURCE INTEGRITY + (Tron §8 gate condition, 2026-07-25, module docstring's own section): a sibling row's own + stamped `md5_checksum` must be NOT NULL and EQUAL to `card`'s own live md5 - + `.filter(md5_checksum=card.md5_checksum)` below is a strict, non-null-tolerant Django lookup + (SQL's three-valued logic means `md5_checksum = ` never matches a NULL row), deliberately + NOT `md5_currency_q()` (that helper's own null-tolerant rule is for CURRENCY checks only, see + its own docstring) - a source row that never carried a stamp at all is never eligible to seed a + transfer, regardless of how "close" the rest of the match looks; minting a fresh stamp on the + copy from an unstamped source would launder an unverified value into a verified-looking field. + Full-manifest (`extractor_versions__has_keys` over every Stage C extractor key, imported + lazily - see `stage_e_dispatch._stage_c_manifest_extractor_keys`'s own docstring for why this + module-boundary import stays call-time-only) - a partially-extracted sibling (e.g. itself + mid-transfer-chain, which never actually happens since transfer always writes every manifest + key at once, but guarded regardless) is never a source either. Most-recently-updated first, in + case more than one qualifying sibling somehow exists (rare - md5 groups are usually small).""" + from cardpicker.management.commands.run_image_evidence_cohort import ( + MANIFEST_EXTRACTOR_KEYS, + ) + + return ( + ImageEvidence.objects.filter(card__md5_checksum=card.md5_checksum) + .exclude(card_id=card.pk) + .filter(content_hash=F("card__content_phash")) + .filter(md5_checksum=card.md5_checksum) + .filter(extractor_versions__has_keys=list(MANIFEST_EXTRACTOR_KEYS)) + .select_related("card") + .order_by("-updated_at") + ) + + +def _record_transfer_anomaly(card: Card, skip_reason: str) -> None: + """Durable anomaly marker (Tron §8 gate condition, 2026-07-25 - see module docstring's + "ANOMALY LOGGING" section) - a plain `.create()`, not batched, since `find_transfer_source` is + called per-card, not per-batch, and an anomaly is rare (the whole point is that it's a real + data problem, not routine traffic).""" + CardScanLog.objects.create(card_id=card.pk, anonymous_id=EVIDENCE_TRANSFER_ANONYMOUS_ID, skip_reason=skip_reason) + + +def find_transfer_source(card: Card) -> Optional[ImageEvidence]: + """ + Returns the md5-sibling `ImageEvidence` row eligible to be copied onto `card`, or `None`. + Otherwise a pure lookup (the pairing/content-hash asserts, module docstring) - `transfer_ + evidence` below does the actual `ImageEvidence` write - EXCEPT that either anomaly path below + now also writes a durable `CardScanLog` row (Tron §8 gate condition - see module docstring's + "ANOMALY LOGGING" section), so this function is no longer write-free in the anomaly case, + only in the "not eligible yet" and "eligible" cases. `None` in every one of these cases: + + - `settings.STAGE_C_EVIDENCE_TRANSFER_ENABLED` is `False` (the kill-switch, module docstring) - + no query issued at all. + - `card.md5_checksum` is `None` (a "group of one", issue #473's ruling 3) - the overwhelming + majority of the catalog until the backfill enrolls more cards. + - `card.content_phash` is `None` - no stable hash yet to key this card's OWN + `(card, content_hash)` row against, matching every other "no stable hash yet" early-return + in this codebase (`local_calculate_verdicts._eligible_cards_queryset`'s own callers, etc.). + - No qualifying sibling row exists at all - the ordinary, non-anomalous "nothing to transfer + from yet" outcome. + - The sha256 pairing check fails (both sides carry a sha256, and they disagree) - a LOUD + anomaly, logged at ERROR + a durable CardScanLog row. + - The content-hash assertion fails (this card's own `content_phash` disagrees with the + sibling's) - also a LOUD anomaly, logged at ERROR + a durable CardScanLog row. + + Only the last two are actually anomalous; the first four are all just "not eligible yet". + """ + if not getattr(settings, "STAGE_C_EVIDENCE_TRANSFER_ENABLED", True): + return None + + if card.md5_checksum is None or card.content_phash is None: + return None + + sibling_evidence = _current_sibling_evidence_queryset(card).first() + if sibling_evidence is None: + return None + + sibling_card = sibling_evidence.card + + card_sha256 = card.sha256_checksum + sibling_sha256 = sibling_card.sha256_checksum + if card_sha256 is not None and sibling_sha256 is not None and card_sha256 != sibling_sha256: + logger.error( + "Evidence transfer anomaly: card %s and md5 sibling %s share md5_checksum %s but " + "sha256_checksum disagrees (%s != %s) - skipping transfer, falling through to real " + "extraction", + card.pk, + sibling_card.pk, + card.md5_checksum, + card_sha256, + sibling_sha256, + ) + _record_transfer_anomaly(card, EVIDENCE_TRANSFER_SHA256_MISMATCH_SKIP_REASON) + return None + + if sibling_evidence.content_hash != card.content_phash: + logger.error( + "Evidence transfer anomaly: card %s and md5 sibling %s share md5_checksum %s but " + "content_phash disagrees (target=%s, sibling evidence=%s) - skipping transfer, " + "falling through to real extraction", + card.pk, + sibling_card.pk, + card.md5_checksum, + card.content_phash, + sibling_evidence.content_hash, + ) + _record_transfer_anomaly(card, EVIDENCE_TRANSFER_CONTENT_HASH_MISMATCH_SKIP_REASON) + return None + + return sibling_evidence + + +def transfer_evidence(card: Card, source: ImageEvidence, run_id: Optional[str] = None) -> ImageEvidence: + """ + THE WRITE half of evidence transfer - copies every extractor field + `extractor_versions` from + `source` (an md5-sibling's own CURRENT, full-manifest row, already vetted by + `find_transfer_source`'s own pairing/content-hash asserts AND its own strict, non-null + `md5_checksum` match - this function trusts its caller to have called that first, it re-verifies + nothing itself) onto `card`'s own `(card, content_hash)` row. Same `get_or_create` + field-merge + shape as `image_evidence.persist_evidence` (reused convention, not reinvented) - a re-run + against the same pair updates in place rather than erroring on the unique constraint. + + Stamps `md5_checksum`/`sha256_checksum` from `card` itself (the TARGET, never `source` - + `find_transfer_source` already verified `card.md5_checksum == source.card.md5_checksum` via a + strict, non-null match, so stamping the target's own live value is equivalent to copying the + sibling's, but stays correct even in the degenerate case where the two could ever disagree + post-verification-race) and sets `transferred=True` + `transferred_from_card_id=source.card_id`. + + INTERIM STAGE D GUARD (issue #473 PR-2, temporary by design - see `ImageEvidence.transferred`'s + own model-field docstring and `local_calculate_verdicts._eligible_cards_queryset`'s own + coordination-note comment): `transferred=True` here is what that guard reads to exclude this + card from the TWO machine-voting Stage D calculators (join-key/fallback - both cast a + `CardPrintingTag` vote) until PR-3's group-level vote pooling lands and removes the guard - a + transferred row's own machine "observation" is the SAME bytes a sibling card already voted + from, not an independent one. The third calculator, slow-path, is deliberately NOT guarded - + it casts no machine vote at all, only a human-review routing marker, which is exactly the + safety net the guard exists to preserve. + """ + evidence, _ = ImageEvidence.objects.get_or_create(card_id=card.pk, content_hash=card.content_phash) + for field_name in _TRANSFERABLE_FIELD_NAMES: + setattr(evidence, field_name, getattr(source, field_name)) + evidence.extractor_versions = dict(source.extractor_versions) + evidence.run_id = run_id + evidence.md5_checksum = card.md5_checksum + evidence.sha256_checksum = card.sha256_checksum + evidence.transferred = True + evidence.transferred_from_card_id = source.card_id + evidence.save() + return evidence + + +__all__ = [ + "EVIDENCE_TRANSFER_ANONYMOUS_ID", + "EVIDENCE_TRANSFER_SHA256_MISMATCH_SKIP_REASON", + "EVIDENCE_TRANSFER_CONTENT_HASH_MISMATCH_SKIP_REASON", + "md5_currency_q", + "find_transfer_source", + "transfer_evidence", +] diff --git a/MPCAutofill/cardpicker/image_evidence.py b/MPCAutofill/cardpicker/image_evidence.py index e3a9c0e4c..a036f5d39 100644 --- a/MPCAutofill/cardpicker/image_evidence.py +++ b/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", diff --git a/MPCAutofill/cardpicker/local_calculate_verdicts.py b/MPCAutofill/cardpicker/local_calculate_verdicts.py index 854cb4a43..59a74282a 100644 --- a/MPCAutofill/cardpicker/local_calculate_verdicts.py +++ b/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", diff --git a/MPCAutofill/cardpicker/local_detect_ai_art.py b/MPCAutofill/cardpicker/local_detect_ai_art.py index 10c000df3..783bdd419 100644 --- a/MPCAutofill/cardpicker/local_detect_ai_art.py +++ b/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: diff --git a/MPCAutofill/cardpicker/local_lands_identify.py b/MPCAutofill/cardpicker/local_lands_identify.py index 11d2608b8..06b08ff7a 100644 --- a/MPCAutofill/cardpicker/local_lands_identify.py +++ b/MPCAutofill/cardpicker/local_lands_identify.py @@ -92,6 +92,7 @@ from cardpicker import local_ocr, local_phash from cardpicker.image_cdn_fetch import fetch_card_image +from cardpicker.image_evidence import current_evidence_queryset from cardpicker.local_fallback import detect_illus_anchor, match_artist from cardpicker.local_identify_printing_tags import ( OCR_ANONYMOUS_ID, @@ -364,21 +365,24 @@ def _split_new_votes(votes_batch: list[CardPrintingTag]) -> tuple[list[CardPrint def _current_evidence_for_card(card: Card) -> Optional[ImageEvidence]: - """The module docstring's CURRENCY check - identical shape to `local_calculate_verdicts`'s - three own eligible-cards loops (`run_join_key_calculator`/`run_fallback_calculator`/ - `run_slow_path_calculator`, all filter `ImageEvidence.objects.filter(card_id=..., content_hash - =card.content_phash)`): a row is only trusted for this card if its `content_hash` matches the - card's own LIVE `content_phash` (an evidence row from a prior image upload is never reused for - a card whose upload has since changed) and it actually carries both extractor groups this - module consumes. `card.content_phash is None` (no stable hash yet) always misses - same "no - stable hash yet to key a CURRENT ImageEvidence lookup against" case those three callers each - skip early for their own reasons. `.order_by("-updated_at").first()` picks the most recently - written row on the rare chance more than one somehow exists for the same (card, content_hash) - pair (the model's own unique constraint means this is normally exactly one or zero).""" + """The module docstring's CURRENCY check - built on `image_evidence.current_evidence_queryset` + (2026-07-25, issue #473 PR-2 - the shared helper `local_calculate_verdicts`'s own three + eligible-cards loops, `local_layout_class_cast.py`, `local_detect_ai_art.py`, and + `reparse_collector_evidence.py`'s own same-named function all use too now): a row is only + trusted for this card if its `content_hash` matches the card's own LIVE `content_phash` (an + evidence row from a prior image upload is never reused for a card whose upload has since + changed), its stamped `md5_checksum` doesn't actively disagree with the card's own live md5 + (null-tolerant - see that helper's own docstring), AND it actually carries both extractor + groups this module consumes. `card.content_phash is None` (no stable hash yet) always misses - + same "no stable hash yet to key a CURRENT ImageEvidence lookup against" case those other + callers each skip early for their own reasons. `.order_by("-updated_at").first()` picks the + most recently written row on the rare chance more than one somehow exists for the same (card, + content_hash) pair (the model's own unique constraint means this is normally exactly one or + zero).""" if card.content_phash is None: return None return ( - ImageEvidence.objects.filter(card_id=card.pk, content_hash=card.content_phash) + current_evidence_queryset(card) .filter(extractor_versions__has_key="collector_line_ocr") .filter(extractor_versions__has_key="artist_ocr") .order_by("-updated_at") diff --git a/MPCAutofill/cardpicker/local_layout_class_cast.py b/MPCAutofill/cardpicker/local_layout_class_cast.py index 10f50c549..972025854 100644 --- a/MPCAutofill/cardpicker/local_layout_class_cast.py +++ b/MPCAutofill/cardpicker/local_layout_class_cast.py @@ -76,6 +76,7 @@ from django.db.models import QuerySet +from cardpicker.image_evidence import current_evidence_queryset from cardpicker.local_fallback import ( BORDER_ATTRIBUTE_VOTE_CONFIDENCE, BORDER_COLOR_TO_TAG, @@ -225,11 +226,7 @@ def run_layout_class_cast( 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: diff --git a/MPCAutofill/cardpicker/management/commands/reparse_collector_evidence.py b/MPCAutofill/cardpicker/management/commands/reparse_collector_evidence.py index 6fd1f38ad..aa8f5c236 100644 --- a/MPCAutofill/cardpicker/management/commands/reparse_collector_evidence.py +++ b/MPCAutofill/cardpicker/management/commands/reparse_collector_evidence.py @@ -100,6 +100,7 @@ from django.core.management.base import BaseCommand, CommandError, CommandParser from django.utils import timezone +from cardpicker.image_evidence import current_evidence_queryset from cardpicker.local_calculate_verdicts import ( JOIN_KEY_ANONYMOUS_ID, JoinKeyVerdict, @@ -238,14 +239,15 @@ def select_card_ids_set_code_lexicon_gate(stage_d_run_id: str) -> list[int]: def _current_evidence_for_card(card: Card) -> Optional[ImageEvidence]: - """The CURRENT `ImageEvidence` row for `card` - same convention - `local_calculate_verdicts.run_join_key_calculator`'s own eligibility query uses: - `content_hash` must match the card's LIVE `content_phash` (a stale evidence row from a prior - image version is never re-parsed), most-recently-updated row first.""" + """The CURRENT `ImageEvidence` row for `card` - built on the shared + `image_evidence.current_evidence_queryset` (2026-07-25, issue #473 PR-2): `content_hash` must + match the card's LIVE `content_phash` (a stale evidence row from a prior image version is + never re-parsed) AND its stamped `md5_checksum` doesn't actively disagree with the card's own + live md5 (null-tolerant - see that helper's own docstring), most-recently-updated row first.""" if card.content_phash is None: return None return ( - 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() diff --git a/MPCAutofill/cardpicker/management/commands/run_image_evidence_cohort.py b/MPCAutofill/cardpicker/management/commands/run_image_evidence_cohort.py index 96b476dbe..4ee34def0 100644 --- a/MPCAutofill/cardpicker/management/commands/run_image_evidence_cohort.py +++ b/MPCAutofill/cardpicker/management/commands/run_image_evidence_cohort.py @@ -159,6 +159,15 @@ here would silently under-count "already done" and re-pay fetch+OCR cost this resume filter exists specifically to avoid. +EVIDENCE TRANSFER (2026-07-25, issue #473 PR-2, folded with issue #472): `_fetch_one_card` checks +`evidence_transfer.find_transfer_source(card)` BEFORE the network fetch call below - a card with a +known `Card.md5_checksum` and an eligible md5-sibling's own CURRENT full-manifest `ImageEvidence` +row gets its own row created via `evidence_transfer.transfer_evidence` instead of paying for a +real fetch+OCR pass over byte-identical content. Terminal outcome `"transferred"` (new, alongside +the pre-existing `"skipped-lockout"`/`"dropped"`) bypasses the compute pool entirely, same as those +- see `_fetch_one_card`'s own docstring for the exact mechanism and `evidence_transfer.py`'s own +module docstring for the pairing/staleness rules that keep this safe. + A `GoogleFetchLockoutError` (403 from the shared Google-bound destination) is a hard stop for the whole run, exactly as `image_cdn_fetch.fetch_card_image`/`fetch_card_image_bytes`'s own docstrings require every caller to treat it - this command sets a stop flag (a `threading.Event`, @@ -296,21 +305,41 @@ class _FetchOutcome: fetch step completed (successfully or with an ordinary, non-lockout failure) and this card should proceed to the compute stage; any other value is a terminal outcome that bypasses compute entirely (matching the old bundled design's own "skipped-lockout"/"dropped" - conventions, replicated here so the final summary counts are unchanged).""" + conventions, replicated here so the final summary counts are unchanged - plus `"transferred"`, + 2026-07-25, issue #473 PR-2's own new terminal outcome, see `_fetch_one_card`'s own docstring). + `md5_checksum`/`sha256_checksum` (also issue #473 PR-2) are the card's own live values at fetch + time, carried across to the compute stage for stamping onto the resulting `ImageEvidence` row - + see `_compute_one_card`'s own docstring.""" card_id: int content_hash: Optional[int] = None + md5_checksum: Optional[str] = None + sha256_checksum: Optional[str] = None image_bytes: Optional[bytes] = None fetch_latency_ms: float = 0.0 outcome: Optional[str] = None -def _fetch_one_card(card_id: int, stop_event: threading.Event) -> _FetchOutcome: +def _fetch_one_card( + card_id: int, stop_event: threading.Event, run_id: str = "", dry_run: bool = False +) -> _FetchOutcome: """Fetch-stage step (thread, not process) - I/O-bound network fetch only, per the decoupling design. Returns the RAW fetched bytes (never decoded here - see `image_cdn_fetch.fetch_card_image_bytes`'s own docstring for why), never runs any extractor. `stop_event` is checked FIRST so a task dispatched after another fetch thread already observed - a lockout never calls `fetch_card_image_bytes` (and so never fetches) at all.""" + a lockout never calls `fetch_card_image_bytes` (and so never fetches) at all. + + EVIDENCE TRANSFER (2026-07-25, issue #473 PR-2's own "BULK resume-filter seam... find where a + card is about to be fetched" scope item): checked here, BEFORE the network fetch call below - + `evidence_transfer.find_transfer_source(card)` looks for an md5-sibling's own CURRENT + full-manifest `ImageEvidence` row; if found, this card's own row is written via + `evidence_transfer.transfer_evidence` (skipped entirely under `dry_run`, matching + `_compute_one_card`'s own `dry_run` convention for its real-extraction persist call) and this + function returns immediately with `outcome="transferred"` - `_run_cohort`'s own main loop + already treats any non-`None` `outcome` as "record it, never submit to the compute pool" (see + that function's own `if fetch_result.outcome is not None: ...` branch, unchanged), so this new + outcome value needs no new branch there. `run_id` is threaded through from `_run_cohort` (an + otherwise-unused parameter added to this function purely to carry it here).""" if stop_event.is_set(): return _FetchOutcome(card_id=card_id, outcome="skipped-lockout") @@ -319,6 +348,20 @@ def _fetch_one_card(card_id: int, stop_event: threading.Event) -> _FetchOutcome: except Card.DoesNotExist: return _FetchOutcome(card_id=card_id, outcome="dropped") + from cardpicker.evidence_transfer import find_transfer_source, transfer_evidence + + transfer_source = find_transfer_source(card) + if transfer_source is not None: + if not dry_run: + transfer_evidence(card, transfer_source, run_id=run_id) + return _FetchOutcome( + card_id=card_id, + content_hash=card.content_phash, + md5_checksum=card.md5_checksum, + sha256_checksum=card.sha256_checksum, + outcome="transferred", + ) + from cardpicker.image_cdn_fetch import DEFAULT_FETCH_DPI, fetch_card_image_bytes fetch_started_at = time.monotonic() @@ -336,6 +379,8 @@ def _fetch_one_card(card_id: int, stop_event: threading.Event) -> _FetchOutcome: return _FetchOutcome( card_id=card_id, content_hash=card.content_phash, + md5_checksum=card.md5_checksum, + sha256_checksum=card.sha256_checksum, image_bytes=image_bytes, fetch_latency_ms=fetch_latency_ms, outcome=None, @@ -352,6 +397,8 @@ def _compute_one_card( profile: bool = False, short_circuit: Optional[bool] = None, known_set_codes: Optional[frozenset[str]] = None, + md5_checksum: Optional[str] = None, + sha256_checksum: Optional[str] = None, ) -> tuple[int, str, Optional[dict[str, float]], bool]: """Module-level (picklable) compute-only work unit for the process pool - takes plain, already-fetched data (never a `Card`/`Image` instance re-fetched or re-decoded elsewhere), and @@ -383,7 +430,12 @@ def _compute_one_card( query in the parent process, not per-card) and forwarded straight through - see `compute_card_evidence`'s own docstring for the escalation-loop acceptance criterion this controls. Picklable (a plain `frozenset[str]`), so passing it into each `compute_pool.submit` - call below costs one IPC serialization per card, not a query.""" + call below costs one IPC serialization per card, not a query. + + `md5_checksum`/`sha256_checksum` (2026-07-25, issue #473 PR-2): the card's own live values, + already read by `_fetch_one_card` (no second query here) and carried across via + `_FetchOutcome` - forwarded straight through to `compute_card_evidence` for stamping onto the + resulting `ImageEvidence` row, same as `stage_e_dispatch._run_stage_c`'s own compute step.""" from cardpicker.image_evidence import compute_card_evidence, persist_evidence wall_started_at = time.monotonic() if profile else None @@ -405,6 +457,8 @@ def _compute_one_card( profile=profile_dict, short_circuit=short_circuit, known_set_codes=known_set_codes, + md5_checksum=md5_checksum, + sha256_checksum=sha256_checksum, ) if not dry_run: persist_evidence(result, run_id=run_id) @@ -452,6 +506,11 @@ def __init__( # remainder in particular) produces the plan's own "open verification gap" measurement # data rather than only the 20k-cohort's retrospective estimate. self.short_circuited = 0 + # Evidence transfer (2026-07-25, issue #473 PR-2) - a card whose row was created via + # `evidence_transfer.transfer_evidence` (an md5-sibling's own current row, copied) rather + # than a real fetch+extraction pass. Counted the same way short_circuited is (a real + # completion, never a fetch_failures/dropped case), never subtracted from `completed`. + self.transferred = 0 self.rss_limit_hit = False # peak parent-process RSS observed across this run's own progress-line samples (2026-07-24, # docs/proposals/stage-e-streaming.md §3 decision (6)/§1's own "this is a real observability @@ -474,6 +533,8 @@ def record(self, outcome: str, short_circuited: bool = False) -> None: self.completed += 1 if outcome in ("fetch_failed", "dropped"): self.fetch_failures += 1 + if outcome == "transferred": + self.transferred += 1 if short_circuited: self.short_circuited += 1 completed = self.completed @@ -488,7 +549,7 @@ def record(self, outcome: str, short_circuited: bool = False) -> None: self._stdout_write( f"[{completed}/{self._total}] elapsed={elapsed:.0f}s rate={rate:.3f}/s " f"fetch_failures={self.fetch_failures} short_circuited={self.short_circuited} " - f"rss_mb={rss_display}" + f"transferred={self.transferred} rss_mb={rss_display}" ) if ( self._max_rss_mb is not None @@ -591,7 +652,7 @@ def _submit_more_fetch() -> None: card_id = next(cohort_iter) except StopIteration: return - outstanding_fetch.add(fetch_pool.submit(_fetch_one_card, card_id, stop_event)) + outstanding_fetch.add(fetch_pool.submit(_fetch_one_card, card_id, stop_event, run_id, dry_run)) def _drain_one_pending() -> None: done, _ = wait(set(pending.keys()), return_when=FIRST_COMPLETED) @@ -638,6 +699,8 @@ def _drain_one_pending() -> None: profile, short_circuit, known_set_codes, + fetch_result.md5_checksum, + fetch_result.sha256_checksum, ) pending[compute_future] = fetch_result.card_id _submit_more_fetch() diff --git a/MPCAutofill/cardpicker/management/commands/stage_e_shakedown.py b/MPCAutofill/cardpicker/management/commands/stage_e_shakedown.py index 8a77e9a2e..7c6aaa80a 100644 --- a/MPCAutofill/cardpicker/management/commands/stage_e_shakedown.py +++ b/MPCAutofill/cardpicker/management/commands/stage_e_shakedown.py @@ -45,7 +45,8 @@ EVIDENCE-CHANGE ECHO (spec point 5, corrected per the §8 Tron pass on PR #467 - the original "fast, cheap no-op either way" characterization below was WRONG, left here struck through in spirit -by this correction rather than silently rewritten): every `persist_evidence` write this driver's +by this correction rather than silently rewritten; RESOLVED 2026-07-25, issue #472 folded with +issue #473 PR-2 - see the closing paragraph): every `persist_evidence` write this driver's forced re-extraction performs is an ordinary `ImageEvidence` save, so `cardpicker.stage_e_signals`'s own `_dispatch_on_evidence_change` receiver fires for it exactly as it would for any other Stage C write - an async `dispatch_for_card(card_id, "evidence-change")` task queues behind it, independent @@ -59,16 +60,19 @@ queuing ~24 FURTHER echoes - a cascade, not a fixed cost. Each echo also holds one of the two `STAGE_E_MAX_CONCURRENT_DISPATCHES` slots for its own duration, so a live echo stream competes with this driver's own dispatch calls for the same cap and can throttle-stop the driver -(`"throttled-concurrency-cap"`) well before the cohort is exhausted. ACCEPTABLE at bounded-pilot -scale (frozen at filing, still not suppressed here) - the two are distinguishable in the ledger by -`trigger_reason`: this driver's own batches carry `"shakedown"`, an echo dispatch carries -`"evidence-change"`, so the ledger itself shows whether echoes are staying cheap (batch_size stays -at 1) or cascading (batch_size climbs toward STAGE_E_MICRO_BATCH_SIZE). Tron's own condition -(§8 pass on PR #467): the documented fallback (NOT built here, per the frozen spec's own -instruction not to build it preemptively - a suppress-signals flag on `persist_evidence`) becomes -REQUIRED, not optional, before scaling beyond a bounded pilot, if either (a) throttle-stops -dominate the driver's own ledger output, or (b) the Stage C backlog is measured non-zero at run -time (check before invoking, per the operator runbook in docs/features/stage-e-operations.md). +(`"throttled-concurrency-cap"`) well before the cohort is exhausted. The two are distinguishable in +the ledger by `trigger_reason`: this driver's own batches carry `"shakedown"`, an echo dispatch +carries `"evidence-change"`, so the ledger itself shows whether echoes are staying cheap +(batch_size stays at 1) or cascading (batch_size climbs toward STAGE_E_MICRO_BATCH_SIZE). + +RESOLVED (2026-07-25, issue #472 folded with issue #473 PR-2): the suppress-signals fallback Tron's +own condition (§8 pass on PR #467) flagged as becoming REQUIRED before scaling beyond a bounded +pilot is now BUILT - `cardpicker.stage_e_signals.suppress_evidence_change_echo` wraps every +`ImageEvidence` write `stage_e_dispatch._run_stage_c` performs, and this driver's own forced +re-extraction runs entirely through `dispatch_micro_batch` -> `_run_stage_c`, so every write this +driver's own writes trigger is now suppressed automatically - no separate opt-in, no code change +needed here. See docs/features/stage-e-operations.md's "Evidence transfer and decoupled +fetch-ahead" section for the full mechanism. INSTRUMENTATION (spec point 6): nothing new - every batch already gets its own `PilotRunLedger` row via `dispatch_micro_batch` (elapsed_s/stage_c_completed/stage_c_fetch_failures/peak_rss_mb/etc., diff --git a/MPCAutofill/cardpicker/migrations/0085_imageevidence_transfer_fields.py b/MPCAutofill/cardpicker/migrations/0085_imageevidence_transfer_fields.py new file mode 100644 index 000000000..65a538ba1 --- /dev/null +++ b/MPCAutofill/cardpicker/migrations/0085_imageevidence_transfer_fields.py @@ -0,0 +1,33 @@ +# Generated by Django 4.2.30 on 2026-07-25 19:13 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("cardpicker", "0084_card_checksums"), + ] + + operations = [ + migrations.AddField( + model_name="imageevidence", + name="md5_checksum", + field=models.CharField(blank=True, db_index=True, max_length=32, null=True), + ), + migrations.AddField( + model_name="imageevidence", + name="sha256_checksum", + field=models.CharField(blank=True, db_index=True, max_length=64, null=True), + ), + migrations.AddField( + model_name="imageevidence", + name="transferred", + field=models.BooleanField(default=False), + ), + migrations.AddField( + model_name="imageevidence", + name="transferred_from_card_id", + field=models.IntegerField(blank=True, null=True), + ), + ] diff --git a/MPCAutofill/cardpicker/models.py b/MPCAutofill/cardpicker/models.py index 0488410de..6d13d468e 100755 --- a/MPCAutofill/cardpicker/models.py +++ b/MPCAutofill/cardpicker/models.py @@ -1968,6 +1968,48 @@ class ImageEvidence(models.Model): created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) + # Evidence transfer (issue #473 PR-2, folded with issue #472) - stamped at BOTH real + # extraction time (image_evidence.compute_card_evidence, copied from the source card's own + # live Card.md5_checksum/sha256_checksum at the moment this row was computed) and transfer + # time (evidence_transfer.transfer_evidence, copied from the TARGET card - the one whose + # (card, content_hash) row this is - never from the sibling the fields were copied from, + # since find_transfer_source already verified the target's own value agrees). Used two ways: + # (1) evidence CURRENCY (image_evidence.current_evidence_queryset) additionally requires + # md5_checksum == Card.md5_checksum whenever BOTH are non-null - closes the silent + # in-place-file-replacement hole a content_phash-only currency check can miss. NULL-TOLERANT: + # a legacy row written before this field existed (md5_checksum is None here) stays current + # under the content_hash check alone until it's naturally re-extracted - no forced mass + # recompute. (2) evidence_transfer.find_transfer_source's own sibling-pairing search, which + # additionally requires sha256_checksum to match whenever BOTH sides carry one (the binding + # 2026-07-25 pairing rule on issue #473 - md5 collisions are constructible, sha256 is the + # cryptographic backstop). sha256_checksum mirrors Card.sha256_checksum's own nullability + # (both are NULL for exactly the same reasons - LOCAL_FILE sources, or a Drive listing walked + # before this field existed - never invented, never backfilled from image bytes we don't hold). + md5_checksum = models.CharField(max_length=32, null=True, blank=True, db_index=True) + sha256_checksum = models.CharField(max_length=64, null=True, blank=True, db_index=True) + + # transferred (issue #473 PR-2's INTERIM STAGE D GUARD, temporary by design): True iff this + # row's own field values were COPIED from an md5-sibling's own current evidence + # (evidence_transfer.transfer_evidence) rather than produced by a real fetch+extraction pass + # against this card's own image. `local_calculate_verdicts._eligible_cards_queryset`'s two + # MACHINE-VOTING Stage D calculators (join-key/fallback - both cast a `CardPrintingTag` vote) + # exclude any card whose CURRENT evidence carries this flag from machine voting - a transferred + # row's own machine "observation" is the SAME underlying bytes a sibling card already voted + # from, not an independent one, so casting a vote from it here would fabricate independence the + # vote-weight matrix assumes is real (docs/theory.md's independence-assumptions section). The + # third Stage D calculator, slow-path, is deliberately NOT guarded - it casts no machine vote, + # only a human-review routing marker, which is exactly the safety net this guard exists to + # preserve, not a case it needs to protect against. REMOVAL IS PR-3's OWN BUSINESS (issue + # #473's build plan, PR-3 section: "Removes PR-2's interim Stage D guard") - once group-level + # vote pooling lands, a transferred row's vote is correctly deduped at the GROUP level instead + # of excluded outright, so this flag (and the guard reading it) stops being needed; do not + # remove either before that PR merges. `transferred_from_card_id` is a plain (non-FK) audit + # trail of which sibling card's row this one was copied from - never queried by the guard + # itself, kept only for a future incident's own "why does this row look like that one" + # question. + transferred = models.BooleanField(default=False) + transferred_from_card_id = models.IntegerField(null=True, blank=True) + class Meta: constraints = [ models.UniqueConstraint(fields=["card", "content_hash"], name="unique_image_evidence_per_card_hash") diff --git a/MPCAutofill/cardpicker/modern_artist_credit.py b/MPCAutofill/cardpicker/modern_artist_credit.py index 92087c747..5b3a1dee3 100644 --- a/MPCAutofill/cardpicker/modern_artist_credit.py +++ b/MPCAutofill/cardpicker/modern_artist_credit.py @@ -305,18 +305,23 @@ def eligible_evidence_queryset() -> "QuerySet[Any]": `content_hash` matches the card's own live `content_phash`, the same "never trust a stale evidence row from a prior image version" convention every other Stage C/D reader in this codebase follows (e.g. `local_detect_ai_art._eligible_cards_queryset`, - `reparse_collector_evidence._current_evidence_for_card`). A card with no `content_phash` yet - (no stable hash to key a CURRENT lookup against) is excluded by the join returning no match, - the same outcome those other readers reach via an explicit `if card.content_phash is None` - skip. + `reparse_collector_evidence._current_evidence_for_card`) - PLUS (2026-07-25, issue #473 PR-2) + the row's own stamped `md5_checksum` doesn't actively disagree with the card's own live md5 + (`evidence_transfer.md5_currency_q`'s own null-tolerant F-expression form of the identical rule + `image_evidence.current_evidence_queryset` applies for the single-card case). A card with no + `content_phash` yet (no stable hash to key a CURRENT lookup against) is excluded by the join + returning no match, the same outcome those other readers reach via an explicit + `if card.content_phash is None` skip. """ from django.db.models import F + from cardpicker.evidence_transfer import md5_currency_q from cardpicker.models import ImageEvidence return ( ImageEvidence.objects.exclude(artist_ocr_raw_text="") .filter(artist_ocr_name="", content_hash=F("card__content_phash")) + .filter(md5_currency_q()) .select_related("card") ) diff --git a/MPCAutofill/cardpicker/review_clusters.py b/MPCAutofill/cardpicker/review_clusters.py index af218f22b..5cb85caca 100644 --- a/MPCAutofill/cardpicker/review_clusters.py +++ b/MPCAutofill/cardpicker/review_clusters.py @@ -202,18 +202,21 @@ def _eligible_review_cards() -> "list[Card]": def _current_evidence_by_card_id(cards: "list[Card]") -> dict[int, tuple[Optional[int], str]]: """Bulk-fetches each card's CURRENT `ImageEvidence` row (content_hash matching that card's own live content_phash - an evidence row from a prior image version is never trusted, same - freshness rule `local_calculate_verdicts.py`'s own calculators apply) and returns - {card_id: (symbol_phash, legal_line_raw_text)}. One query total, not one per card: ordered - so the first row seen per card_id in iteration order is its most recent, then filtered in - Python against that card's live content_phash.""" + freshness rule `local_calculate_verdicts.py`'s own calculators apply - PLUS, 2026-07-25, issue + #473 PR-2, the row's own stamped md5_checksum not actively disagreeing with the card's own + live md5, null-tolerant, same rule `image_evidence.current_evidence_queryset` applies for the + single-card case) and returns {card_id: (symbol_phash, legal_line_raw_text)}. One query total, + not one per card: ordered so the first row seen per card_id in iteration order is its most + recent, then filtered in Python against that card's live content_phash/md5_checksum.""" card_ids = [c.pk for c in cards if c.content_phash is not None] if not card_ids: return {} live_content_phash_by_card_id = {c.pk: c.content_phash for c in cards if c.content_phash is not None} + live_md5_by_card_id = {c.pk: c.md5_checksum for c in cards if c.content_phash is not None} result: dict[int, tuple[Optional[int], str]] = {} rows = ( ImageEvidence.objects.filter(card_id__in=card_ids) - .values("card_id", "content_hash", "symbol_phash", "legal_line_raw_text") + .values("card_id", "content_hash", "md5_checksum", "symbol_phash", "legal_line_raw_text") .order_by("card_id", "-updated_at") ) for row in rows: @@ -222,6 +225,10 @@ def _current_evidence_by_card_id(cards: "list[Card]") -> dict[int, tuple[Optiona continue # already took this card's most-recent row - see the ordering above if row["content_hash"] != live_content_phash_by_card_id[card_id]: continue # stale evidence for a since-changed image - never trusted + live_md5 = live_md5_by_card_id[card_id] + row_md5 = row["md5_checksum"] + if row_md5 is not None and live_md5 is not None and row_md5 != live_md5: + continue # #473 PR-2's md5 staleness rule - null-tolerant, see module docstring above result[card_id] = (row["symbol_phash"], row["legal_line_raw_text"] or "") return result diff --git a/MPCAutofill/cardpicker/stage_e_dispatch.py b/MPCAutofill/cardpicker/stage_e_dispatch.py index 83e79ae7c..d228a30cf 100644 --- a/MPCAutofill/cardpicker/stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/stage_e_dispatch.py @@ -45,16 +45,22 @@ the window, it doesn't mandate a shared store). PIPELINE STAGES, in order, per micro-batch (task brief scope item 5): Stage C extraction -(`cardpicker.image_evidence.compute_card_evidence`/`persist_evidence`, called per-card, -SEQUENTIALLY - fed by `cardpicker.image_cdn_fetch.fetch_card_image_bytes`) -> Stage D calculators -(`cardpicker.local_calculate_verdicts.run_join_key_calculator`/`run_fallback_calculator`/ +(`cardpicker.image_evidence.compute_card_evidence`/`persist_evidence`, called per-card - see +`_run_stage_c`'s own docstring for its three phases: evidence-transfer check, then a decoupled +fetch-ahead thread feeding this function's own SEQUENTIAL compute loop, issue #472) -> Stage D +calculators (`cardpicker.local_calculate_verdicts.run_join_key_calculator`/`run_fallback_calculator`/ `run_slow_path_calculator`, called AS-IS with the new `card_ids` scope, in the same join-key -> fallback -> slow-path escalation order every BULK-mode command already uses) -> ledger write. -Sequential, not pooled, on purpose: PASSIVE mode's own micro-batches (§3 decision (2), a handful to -a few dozen cards) are far too small for BULK mode's process-pool concurrency to buy anything - it -would only add a fork's worth of startup overhead per batch. This matches the brief's own "a -single-worker, single-core floor mode must be correct, just slow, never a degraded/unsound mode" -requirement (§5). +COMPUTE is sequential, not pooled, on purpose: PASSIVE mode's own micro-batches (§3 decision (2), a +handful to a few dozen cards) are far too small for BULK mode's process-pool concurrency to buy +anything - it would only add a fork's worth of startup overhead per batch. This matches the +brief's own "a single-worker, single-core floor mode must be correct, just slow, never a +degraded/unsound mode" requirement (§5). FETCH, as of issue #472 (2026-07-25, the ratified §4 item +3 this Phase-2-era module originally shipped without - see `_run_stage_c`'s own docstring), is +OVERLAPPED with that same sequential compute loop via one fetch-ahead thread + a bounded queue - +this is deliberately NOT the same thing as pooling compute; only I/O-bound fetch-wait is +overlapped, the OCR/extraction work itself stays exactly as sequential as the paragraph above +requires. CONSENSUS RECOMPUTE (decision (4)) NEEDS NO SEPARATE STEP HERE: all three Stage D calculators already call `resolve_and_persist_printing(touched_card)` internally for every card they cast a @@ -67,6 +73,8 @@ import logging import os +import queue +import threading import time from collections import deque from dataclasses import dataclass, field @@ -75,6 +83,7 @@ from django.conf import settings from django.utils import timezone +from cardpicker.evidence_transfer import find_transfer_source, transfer_evidence from cardpicker.harvest_fetch_limiter import GoogleFetchLockoutError from cardpicker.local_calculate_verdicts import ( known_set_codes, @@ -99,6 +108,7 @@ from cardpicker.pilot_run_lifecycle import mark_ledger_failed, merge_counters from cardpicker.process_metrics import get_process_rss_mb from cardpicker.stage_e_concurrency import try_acquire_dispatch_slot +from cardpicker.stage_e_signals import suppress_evidence_change_echo from cardpicker.utils import get_baked_git_sha logger = logging.getLogger(__name__) @@ -201,6 +211,12 @@ class DispatchOutcome: run_id: Optional[str] = None card_ids: list[int] = field(default_factory=list) stage_c_completed: int = 0 + # Evidence transfer (issue #473 PR-2, folded with issue #472): a card whose evidence row this + # batch produced via `evidence_transfer.transfer_evidence` (an md5-sibling's own current row, + # copied - no fetch, no real extraction) rather than a real per-card extraction pass. A SUBSET + # of `stage_c_completed` above - both counters increment together for a transferred card, this + # one just narrows down which completions were transfers vs. real fetch+extraction work. + stage_c_transferred: int = 0 stage_c_fetch_failures: int = 0 stage_d_join_key_votes: int = 0 stage_d_fallback_votes: int = 0 @@ -387,17 +403,168 @@ def _verify_stage_c_chunk(chunk: list[int]) -> Iterable[int]: return seen[:batch_size] +# Bounded fetch-ahead queue depth (issue #472's own design constraint: "Bounded prefetch depth +# (1-2 images) so RSS stays flat") - a plain module constant, not a settings knob, matching the +# brief's own concrete number rather than leaving it operator-tunable; PASSIVE mode's own +# micro-batches are small enough (§3 decision (2), a handful to a few dozen cards) that this never +# needs retuning the way BULK mode's own `--queue-depth` does. +_STAGE_C_FETCH_AHEAD_DEPTH = 2 + + +@dataclass +class _StageCFetchOutcome: + """One card's own fetch-stage result, handed from `_stage_c_fetch_ahead_worker` (the fetch- + ahead thread) to `_run_stage_c`'s own sequential compute loop via a bounded `queue.Queue`. + `card`/`content_hash`/`md5_checksum`/`sha256_checksum` are all read from the SAME `Card` + instance `_run_stage_c` already loaded (and used for its own transfer check) before handing + this card off to the fetch-ahead thread - no second `Card` query on either side of the + boundary. `lockout=True` iff this card's OWN fetch attempt raised `GoogleFetchLockoutError` - + the compute loop treats this as the signal to stop (module docstring's "halts NEW fetches + immediately" bar), never a fetch failure to retry. + + `error`, if set, is a NON-`GoogleFetchLockoutError` exception the fetch attempt raised (2026-07-25, + kill-safety fix - see `_stage_c_fetch_ahead_worker`'s own docstring for why this exists at + all): re-raised by the compute loop IN THE MAIN THREAD the instant it's observed, so a crash + during fetch still propagates out of `_run_stage_c`/`dispatch_micro_batch` exactly as it did + before this module had a separate fetch thread at all - `TestKillSafetyResumeContract`'s own + "a mid-batch crash leaves a truthful FAILED ledger row" contract does not distinguish between + a crash during fetch and a crash during compute, and must not silently become a hang instead.""" + + card_id: int + content_hash: Optional[int] + md5_checksum: Optional[str] + sha256_checksum: Optional[str] + image_bytes: Optional[bytes] + fetch_latency_ms: float + lockout: bool = False + error: Optional[BaseException] = None + + +def _stage_c_fetch_ahead_worker( + cards: list[Card], + out_queue: "queue.Queue[_StageCFetchOutcome]", + stop_event: threading.Event, +) -> None: + """ + THE fetch-ahead thread (issue #472) - ONE thread, sequential fetches (never pooled: the design + brief's own "no compute pooling" bar applies equally to a fetch pool here, since a SECOND + concurrent fetch would only race further ahead of a compute loop that's already the slower + stage, buying nothing the bounded queue depth doesn't already buy via fetch/compute OVERLAP + alone). Pushes one `_StageCFetchOutcome` per entry in `cards`, in order - `queue.Queue`'s own + FIFO ordering means the compute loop always drains outcomes in the SAME order this thread + fetched them, satisfying the design brief's own "fetch-outcome window records in completion + order" bar for free (a single serial fetch worker's own completion order IS its submission + order, there is no reordering possible with only one fetch in flight at a time). + + `out_queue`'s own bound (`queue.Queue(maxsize=_STAGE_C_FETCH_AHEAD_DEPTH)`, set by the caller) + is what keeps RSS bounded independent of batch size - `put()` BLOCKS once the queue is full, + so this thread can never race further than `_STAGE_C_FETCH_AHEAD_DEPTH` outcomes ahead of + whatever the compute loop has actually consumed so far, regardless of how many cards remain in + `cards` or how large a future catalog's own micro-batch could be. + + LOCKOUT (design brief's own "instant halt" bar): the moment `fetch_card_image_bytes` raises + `GoogleFetchLockoutError` for one card, `stop_event` is set and this thread returns immediately + WITHOUT attempting any further card in `cards` - "halts NEW fetches immediately". The card + whose OWN fetch triggered the lockout is reported to the compute loop via this outcome's own + `lockout=True` flag (never silently dropped), so the compute loop can record the trip itself + and stop too. Every outcome that already made it into `out_queue` BEFORE this happened is left + there untouched - "in-flight work drains" - the compute loop keeps consuming those (they sort + earlier in FIFO order than the lockout outcome) before it ever reaches the lockout marker, + exactly mirroring the pre-#472 sequential design's own "an already-fetched image still gets to + finish its own compute+persist" property, just now genuinely overlapped rather than accidental. + + ANY OTHER EXCEPTION (2026-07-25, kill-safety fix - `TestKillSafetyResumeContract`'s own + mid-batch-crash test caught this during review): a plain `try/except GoogleFetchLockoutError` + here would let a non-lockout exception (a real bug, a simulated kill-drill fault, a genuine + network error `fetch_card_image_bytes` doesn't itself wrap) kill this THREAD silently - Python + does not propagate an uncaught exception from a spawned `threading.Thread` to its caller, so + the compute loop's own `queue.get()` for that card would block FOREVER waiting for an outcome + that will never arrive, turning a crash into a silent hang instead of the loud, ledger-recorded + failure the resume contract requires. Caught here as a bare `Exception`, packaged onto the + outcome's own `error` field, and this thread stops (same "no further cards attempted" posture + as a lockout) - the compute loop re-raises it in the MAIN thread the instant it's seen. + """ + from cardpicker.image_cdn_fetch import DEFAULT_FETCH_DPI, fetch_card_image_bytes + + for card in cards: + if stop_event.is_set(): + return + + fetch_started_at = time.monotonic() + try: + image_bytes = fetch_card_image_bytes(card, dpi=DEFAULT_FETCH_DPI) + except GoogleFetchLockoutError: + stop_event.set() + out_queue.put( + _StageCFetchOutcome( + card_id=card.pk, + content_hash=card.content_phash, + md5_checksum=card.md5_checksum, + sha256_checksum=card.sha256_checksum, + image_bytes=None, + fetch_latency_ms=0.0, + lockout=True, + ) + ) + return + except Exception as exc: # noqa: BLE001 - deliberately broad, see docstring above + stop_event.set() + out_queue.put( + _StageCFetchOutcome( + card_id=card.pk, + content_hash=card.content_phash, + md5_checksum=card.md5_checksum, + sha256_checksum=card.sha256_checksum, + image_bytes=None, + fetch_latency_ms=0.0, + error=exc, + ) + ) + return + fetch_latency_ms = (time.monotonic() - fetch_started_at) * 1000 + + out_queue.put( + _StageCFetchOutcome( + card_id=card.pk, + content_hash=card.content_phash, + md5_checksum=card.md5_checksum, + sha256_checksum=card.sha256_checksum, + image_bytes=image_bytes, + fetch_latency_ms=fetch_latency_ms, + ) + ) + + def _run_stage_c( batch_ids: list[int], run_id: str, outcome: DispatchOutcome, force_stage_c_reextract: bool = False ) -> Optional[EnvelopeTrip]: """ - Sequential, per-card Stage C extraction over whichever of `batch_ids` still lack a full - manifest - the SAME per-card unit (`image_evidence.compute_card_evidence` + - `image_evidence.persist_evidence`, fed by `image_cdn_fetch.fetch_card_image_bytes`) - `run_image_evidence_cohort.py`'s own fetch/compute stages call, just driven one card at a time - (module docstring's own "PIPELINE STAGES" section explains why). Every fetch outcome is recorded - onto `_window` regardless of whether it ends up mattering to THIS batch's own envelope decision - - the window spans the whole worker process's uptime, not one batch. + Per-card Stage C extraction over whichever of `batch_ids` still lack a full manifest - the SAME + per-card unit (`image_evidence.compute_card_evidence` + `image_evidence.persist_evidence`, fed + by `image_cdn_fetch.fetch_card_image_bytes`) `run_image_evidence_cohort.py`'s own fetch/compute + stages call. Three phases per batch, in order: + + 1. **Build the work list** (still fully sequential, cheap DB-only work): for each id not + already done, load its `Card` once and check `evidence_transfer.find_transfer_source` + BEFORE deciding whether this card needs a fetch at all (issue #473 PR-2's own scope: "check + BEFORE fetching") - a card with an eligible md5-sibling never reaches the fetch-ahead thread + at all, its evidence row is created via `evidence_transfer.transfer_evidence` right here and + counted via `outcome.stage_c_transferred`/`stage_c_completed`. Every card that still needs a + real extraction (no md5, no eligible sibling, or a loud pairing/content-hash anomaly - see + that function's own docstring) is collected into `to_fetch`. + 2. **Decoupled fetch-ahead + sequential compute** (issue #472): `to_fetch` is handed to ONE + fetch-ahead thread (`_stage_c_fetch_ahead_worker`) writing into a bounded + (`_STAGE_C_FETCH_AHEAD_DEPTH`) queue; THIS function's own loop stays the sequential OCR/ + extraction compute stage the design brief mandates (no compute pooling), just now able to + decode+extract card N while the fetch-ahead thread is already fetching card N+1's bytes, + instead of blocking on that fetch itself. Every fetch outcome (transfer OR real fetch) is + recorded onto `_window` regardless of whether it ends up mattering to THIS batch's own + envelope decision - the window spans the whole worker process's uptime, not one batch. + 3. **Echo suppression** (issue #472's own fold, `cardpicker.stage_e_signals`'s own module + docstring has the full mechanism writeup): both the transfer write in phase 1 and the + `persist_evidence` write in phase 2 are wrapped in `suppress_evidence_change_echo()` - a + write performed by THIS dispatch loop must never queue a fresh `dispatch_for_card` echo task + for the same card, since Stage D (called next, over the SAME batch) already covers it. `force_stage_c_reextract` (issue #465, `management/commands/stage_e_shakedown.py`'s one conveyor change): `False` (the default) is BYTE-IDENTICAL to the pre-#465 behaviour below - the @@ -411,21 +578,24 @@ def _run_stage_c( `run_image_evidence_cohort.py`'s own `--no-shortcircuit` flag has (see that command's own docstring for the mechanism reused here, not reimplemented), so a zero-digit tier-1 read is never allowed to short-circuit past the fuller multi-tier escalation that could recover a real read. + Transfer-checking (phase 1) is deliberately UNCONDITIONAL regardless of this flag - a + force-re-extracted card with a genuinely current, good md5-sibling gets FIXED by transfer + immediately rather than paying for a real re-fetch of what would produce the same bytes anyway; + `evidence_transfer.find_transfer_source`'s own asserts are what keep this safe. Returns the `EnvelopeTrip` this call itself recorded (only possible via the instant Google lockout bar - see `GoogleFetchLockoutError` below), or `None`. A lockout stops Stage C IMMEDIATELY for this batch - in-flight work already committed stays committed (each card's - `persist_evidence` call is already durable the instant it returns, matching the resume - contract's own "one-transaction batch commit or explicit evidence-first statement" - here, every - card's own persist is its own transaction, so there is no partial-card state to roll back) - and - records a fresh trip via `check_envelope(google_lockout=True)` so the NEXT dispatch call refuses - until an owner acknowledges it, matching the "instant pause" bar exactly. + `persist_evidence`/`transfer_evidence` call is already durable the instant it returns, matching + the resume contract's own "one-transaction batch commit or explicit evidence-first statement" - + here, every card's own persist is its own transaction, so there is no partial-card state to + roll back) - and records a fresh trip via `check_envelope(google_lockout=True)` so the NEXT + dispatch call refuses until an owner acknowledges it, matching the "instant pause" bar exactly. """ from io import BytesIO from PIL import Image - from cardpicker.image_cdn_fetch import DEFAULT_FETCH_DPI, fetch_card_image_bytes from cardpicker.image_evidence import compute_card_evidence, persist_evidence if force_stage_c_reextract: @@ -440,6 +610,9 @@ def _run_stage_c( short_circuit: Optional[bool] = False if force_stage_c_reextract else None lexicon = known_set_codes() + # PHASE 1 (module docstring): build the work list, resolving evidence transfer BEFORE + # deciding whether a card needs a fetch at all. + to_fetch: list[Card] = [] for card_id in batch_ids: if card_id in already_done_ids: continue @@ -450,34 +623,104 @@ def _run_stage_c( if card.content_phash is None: continue - fetch_started_at = time.monotonic() - try: - image_bytes = fetch_card_image_bytes(card, dpi=DEFAULT_FETCH_DPI) - except GoogleFetchLockoutError: - _window.record(success=False) - logger.error("Stage E dispatch: GoogleFetchLockoutError observed - halting Stage C for this batch") - return check_envelope(_sample_envelope_signals(google_lockout=True), run_id=run_id) - fetch_latency_ms = (time.monotonic() - fetch_started_at) * 1000 - - if image_bytes is None: - _window.record(success=False) - outcome.stage_c_fetch_failures += 1 + transfer_source = find_transfer_source(card) + if transfer_source is not None: + with suppress_evidence_change_echo(): + transfer_evidence(card, transfer_source, run_id=run_id) + outcome.stage_c_completed += 1 + outcome.stage_c_transferred += 1 continue - _window.record(success=True) - image = Image.open(BytesIO(image_bytes)) - result = compute_card_evidence( - card_id, - card.content_phash, - image, - fetch_latency_ms=fetch_latency_ms, - short_circuit=short_circuit, - known_set_codes=lexicon, - ) - persist_evidence(result, run_id=run_id) - outcome.stage_c_completed += 1 + to_fetch.append(card) + + if not to_fetch: + return None + + # PHASE 2 (module docstring): decoupled fetch-ahead thread + this function's own sequential + # compute loop. + fetch_queue: "queue.Queue[_StageCFetchOutcome]" = queue.Queue(maxsize=_STAGE_C_FETCH_AHEAD_DEPTH) + stop_event = threading.Event() + fetch_thread = threading.Thread( + target=_stage_c_fetch_ahead_worker, args=(to_fetch, fetch_queue, stop_event), daemon=True + ) + fetch_thread.start() + + trip: Optional[EnvelopeTrip] = None + try: + for _ in range(len(to_fetch)): + fetch_outcome = fetch_queue.get() + + if fetch_outcome.error is not None: + # Re-raise IN THE MAIN THREAD - see _StageCFetchOutcome/_stage_c_fetch_ahead_ + # worker's own docstrings for why this exists (a spawned thread's own uncaught + # exception never reaches the caller on its own). Propagates out of this function, + # through dispatch_micro_batch's own `except Exception: mark_ledger_failed(...); + # raise` - identical observable behaviour to the pre-#472 sequential design's own + # "a fetch-time crash surfaces exactly like a compute-time one" contract. + raise fetch_outcome.error + + if fetch_outcome.lockout: + _window.record(success=False) + logger.error("Stage E dispatch: GoogleFetchLockoutError observed - halting Stage C for this batch") + trip = check_envelope(_sample_envelope_signals(google_lockout=True), run_id=run_id) + break + + if fetch_outcome.image_bytes is None: + _window.record(success=False) + outcome.stage_c_fetch_failures += 1 + continue + + _window.record(success=True) + image = Image.open(BytesIO(fetch_outcome.image_bytes)) + result = compute_card_evidence( + fetch_outcome.card_id, + fetch_outcome.content_hash, + image, + fetch_latency_ms=fetch_outcome.fetch_latency_ms, + short_circuit=short_circuit, + known_set_codes=lexicon, + md5_checksum=fetch_outcome.md5_checksum, + sha256_checksum=fetch_outcome.sha256_checksum, + ) + with suppress_evidence_change_echo(): + persist_evidence(result, run_id=run_id) + outcome.stage_c_completed += 1 + finally: + # Always signal-then-drain-then-join, whether the loop above finished normally, broke on + # a lockout, or raised - the fetch-ahead thread is `daemon=True` (won't block process exit + # on its own) but this function should never RETURN while it's still mid-fetch for a card + # nothing will ever consume. + # + # `stop_event.set()` MUST happen BEFORE the drain+join below (Tron §8 gate condition, + # 2026-07-25, HIGH severity - found on review): a bare `fetch_thread.join()` here, with + # neither `stop_event.set()` nor a queue drain first, left the fetch-ahead thread wedged + # FOREVER the moment COMPUTE (not fetch) raised mid-batch - e.g. a corrupt download that + # decodes far enough to pass the fetch stage but raises a PIL error inside + # `compute_card_evidence`/`persist_evidence` above. Once this loop stops calling + # `fetch_queue.get()` (it exited via the exception), a fetch-ahead thread already blocked + # on `out_queue.put(...)` for its own next outcome (the queue is bounded at + # `_STAGE_C_FETCH_AHEAD_DEPTH`) never unblocks on its own - `stop_event` alone does + # nothing for a thread that isn't back at its own loop-top `if stop_event.is_set(): return` + # check yet, and nothing else will ever call `.get()` again to free room for that `put()` + # to complete. The observable failure mode was silent and total: `join()` blocks + # indefinitely, so this function (and `dispatch_micro_batch`'s own `except Exception: + # mark_ledger_failed(...); raise` around it) never even reaches the point of recording the + # crash - the `PilotRunLedger` row stays lying at `RUNNING` forever, and the concurrency-cap + # slot this dispatch holds (`stage_e_concurrency.try_acquire_dispatch_slot`) never gets + # released either, wedging the whole worker process's dispatch capacity over ONE corrupt + # image. Fixed by (1) `stop_event.set()` first, so the thread returns the instant it's back + # at its own loop-top, and (2) draining `fetch_queue` below, which is what actually + # unblocks a `put()` already in progress - after at most one more successful put (the one + # the drain makes room for), the thread reaches its own stop_event check and returns. + stop_event.set() + while True: + try: + fetch_queue.get_nowait() + except queue.Empty: + break + fetch_thread.join() - return None + return trip def _run_stage_d(batch_ids: list[int], run_id: str, outcome: DispatchOutcome) -> None: @@ -655,6 +898,7 @@ def dispatch_micro_batch( { "elapsed_s": round(time.monotonic() - batch_start, 3), "stage_c_completed": outcome.stage_c_completed, + "stage_c_transferred": outcome.stage_c_transferred, "stage_c_fetch_failures": outcome.stage_c_fetch_failures, "stage_d_join_key_votes": outcome.stage_d_join_key_votes, "stage_d_join_key_already_voted": outcome.stage_d_join_key_already_voted, diff --git a/MPCAutofill/cardpicker/stage_e_signals.py b/MPCAutofill/cardpicker/stage_e_signals.py index 558df1929..9d1a7afd9 100644 --- a/MPCAutofill/cardpicker/stage_e_signals.py +++ b/MPCAutofill/cardpicker/stage_e_signals.py @@ -25,16 +25,47 @@ is already current, see `stage_e_dispatch._run_stage_c`), and Stage D's own eligibility queries already exclude a card once it's carrying a vote from a given calculator's own `anonymous_id` - so a burst of `ImageEvidence` saves for the same card (e.g. one extractor group's write, then -another's, both landing on the SAME row within one Stage C pass) triggers several dispatch calls -that mostly resolve to fast, cheap no-ops rather than repeated real work. This is the SAME -"evidence-change event re-opens a card to re-scan, never an elapsed-time trigger" contract issue -#278's own selector already specifies (docs/proposals/stage-e-streaming.md §4 item 4) - deliberately -generic here (every evidence-change fires an attempt, not just #278's own specific detector), since -this module only decides WHETHER to attempt a dispatch, never what any downstream engine does with -it. +another's, both landing on the SAME row within one Stage C pass) triggers several dispatch calls. + +**CORRECTED 2026-07-25 (issue #472, the same §8 Tron pass that corrected the identical +overstated line in `stage_e_shakedown.py`'s own "EVIDENCE-CHANGE ECHO" section and +docs/features/stage-e-operations.md's "Evidence-change echo" section)**: an earlier version of +this paragraph characterized that burst as "mostly resolv[ing] to fast, cheap no-ops rather than +repeated real work." That is WRONG in general - an echo dispatch calls `dispatch_micro_batch` with +NO `batch_size`, so `_select_micro_batch` backfills the echo's own seed card up to the FULL +`STAGE_E_MICRO_BATCH_SIZE` from the Stage C backlog cursor walk. An echo is a COMPLETE micro-batch, +never just the one already-current seed card - cheap (~3.5s fixed overhead, no extraction) ONLY +while the Stage C backlog is genuinely zero at echo time; a non-zero backlog turns an echo into a +real ~25-card extraction batch that itself persists ~25 more `ImageEvidence` rows, queuing ~24 +FURTHER echoes - a cascade, not a fixed cost (see `stage_e_shakedown.py`'s own section for the full +measured numbers). + +ECHO SUPPRESSION (2026-07-25, issue #472's own build, closing the gap the paragraph above +describes for the ONE caller that can trigger the cascade repeatedly - `stage_e_dispatch. +_run_stage_c`'s own `persist_evidence`/`evidence_transfer.transfer_evidence` writes): every +`ImageEvidence` write performed FROM INSIDE the streaming/shakedown dispatch path is wrapped in +`suppress_evidence_change_echo()` below - the write-side already knows it's running inside a +dispatch (Stage D, over the SAME micro-batch, already covers whatever this write would otherwise +re-trigger a fresh dispatch call to reach), so it flags itself via a `contextvars.ContextVar` +rather than requiring this receiver to infer intent from the write. `_dispatch_on_evidence_change` +below checks that flag first and returns immediately if set - no `async_task` is queued at all for +a write made inside that context. BULK-mode writes (`run_image_evidence_cohort.py`'s own +`persist_evidence` calls, a genuinely independent command that never runs inside a dispatch) are +UNFLAGGED and keep firing this receiver's `async_task` exactly as before this change - only writes +performed BY the dispatch loop itself are suppressed. This is the "documented (not built) fallback" +docs/features/stage-e-operations.md's own "Evidence-change echo" section flagged as becoming +REQUIRED before scaling beyond a bounded pilot - now built. + +This is the SAME "evidence-change event re-opens a card to re-scan, never an elapsed-time trigger" +contract issue #278's own selector already specifies (docs/proposals/stage-e-streaming.md §4 item +4) - deliberately generic here (every evidence-change fires an attempt, not just #278's own +specific detector), since this module only decides WHETHER to attempt a dispatch, never what any +downstream engine does with it. """ -from typing import Any +import contextvars +from contextlib import contextmanager +from typing import Any, Iterator from django.conf import settings from django.db.models.signals import post_save @@ -42,6 +73,36 @@ from cardpicker.models import Card, ImageEvidence +# ECHO SUPPRESSION (module docstring, issue #472) - `True` for the duration of an `ImageEvidence` +# write issued FROM INSIDE `stage_e_dispatch._run_stage_c` (the ONLY caller of +# `suppress_evidence_change_echo` below), `False` otherwise (including every BULK-mode write, which +# never enters this context at all). A `contextvars.ContextVar` rather than a plain module global +# or `threading.local`: correct under both the synchronous per-thread execution this module's own +# django-q worker actually uses today AND any future asyncio-based caller, with no extra code +# needed either way - contextvars propagate correctly across `await` points where a bare +# `threading.local` would not, and behave identically to a `threading.local` for the synchronous +# case this module has today. +_dispatch_persist_in_progress: "contextvars.ContextVar[bool]" = contextvars.ContextVar( + "_dispatch_persist_in_progress", default=False +) + + +@contextmanager +def suppress_evidence_change_echo() -> Iterator[None]: + """ + Context manager wrapping an `ImageEvidence` write made FROM INSIDE the streaming/shakedown + dispatch path (`stage_e_dispatch._run_stage_c`'s own `persist_evidence`/ + `evidence_transfer.transfer_evidence` calls - the ONLY caller) - see module docstring's "ECHO + SUPPRESSION" section for the full rationale. Re-entrant-safe (nested `with` blocks all see + `True` until the OUTERMOST one exits, via `ContextVar.set`/`.reset`'s own token mechanism) even + though `_run_stage_c` never actually nests these calls today - defensive, not load-bearing. + """ + token = _dispatch_persist_in_progress.set(True) + try: + yield + finally: + _dispatch_persist_in_progress.reset(token) + @receiver(post_save, sender=Card) def _dispatch_on_card_create(sender: Any, instance: Card, created: bool, **kwargs: Any) -> None: @@ -58,6 +119,12 @@ def _dispatch_on_card_create(sender: Any, instance: Card, created: bool, **kwarg def _dispatch_on_evidence_change(sender: Any, instance: ImageEvidence, **kwargs: Any) -> None: if not getattr(settings, "STAGE_E_STREAMING_ENABLED", False): return + if _dispatch_persist_in_progress.get(): + # ECHO SUPPRESSION (module docstring, issue #472) - this write was made FROM INSIDE the + # dispatch path itself; queuing another dispatch for it would be exactly the cascade risk + # the module docstring's "CORRECTED 2026-07-25" section describes. BULK-mode writes never + # set this flag, so they always reach the async_task call below, unchanged. + return from django_q.tasks import async_task async_task("cardpicker.stage_e_dispatch.dispatch_for_card", instance.card_id, "evidence-change") diff --git a/MPCAutofill/cardpicker/tests/test_evidence_transfer.py b/MPCAutofill/cardpicker/tests/test_evidence_transfer.py new file mode 100644 index 000000000..f61464c5c --- /dev/null +++ b/MPCAutofill/cardpicker/tests/test_evidence_transfer.py @@ -0,0 +1,285 @@ +""" +Tests for cardpicker.evidence_transfer - issue #473 PR-2's evidence transfer (folded with issue +#472). `Card.sha256_checksum` is a real model field on this branch (master already carries it via +migration 0084_card_checksums) - every sha256-pairing test here sets the real field via +`CardFactory(sha256_checksum=...)`, no monkeypatching needed. + +TRANSFER-SOURCE INTEGRITY (Tron §8 gate condition): a sibling `ImageEvidence` row is only a valid +transfer source if its OWN stamped `md5_checksum` is non-null and equals the target card's live +md5 - every fixture below that builds a sibling INTENDED to be a valid transfer source stamps +`md5_checksum` on the `ImageEvidenceFactory` call explicitly (mirroring what a real +`persist_evidence`/`transfer_evidence` write would have stamped) rather than relying on the +`ImageEvidenceFactory`'s own default (`None`). +""" + +from typing import Any + +from django.test import override_settings + +from cardpicker.evidence_transfer import ( + EVIDENCE_TRANSFER_ANONYMOUS_ID, + EVIDENCE_TRANSFER_CONTENT_HASH_MISMATCH_SKIP_REASON, + EVIDENCE_TRANSFER_SHA256_MISMATCH_SKIP_REASON, + find_transfer_source, + transfer_evidence, +) +from cardpicker.management.commands.run_image_evidence_cohort import ( + MANIFEST_EXTRACTOR_KEYS, +) +from cardpicker.models import CardScanLog, ImageEvidence +from cardpicker.tests.factories import CardFactory, ImageEvidenceFactory + +FULL_MANIFEST = {key: f"{key}-v1" for key in MANIFEST_EXTRACTOR_KEYS} + + +class TestFindTransferSourceHappyPath: + def test_current_full_manifest_sibling_is_returned(self, db: Any) -> None: + sibling = CardFactory(md5_checksum="abc123", content_phash=111) + target = CardFactory(md5_checksum="abc123", content_phash=111) + sibling_evidence = ImageEvidenceFactory( + card=sibling, + content_hash=111, + md5_checksum="abc123", + extractor_versions=FULL_MANIFEST, + symbol_phash=999, + ) + + found = find_transfer_source(target) + + assert found is not None + assert found.pk == sibling_evidence.pk + + def test_no_md5_on_target_returns_none(self, db: Any) -> None: + target = CardFactory(md5_checksum=None, content_phash=111) + assert find_transfer_source(target) is None + + def test_no_content_phash_on_target_returns_none(self, db: Any) -> None: + target = CardFactory(md5_checksum="abc123", content_phash=None) + assert find_transfer_source(target) is None + + def test_no_sibling_at_all_falls_through_to_none(self, db: Any) -> None: + target = CardFactory(md5_checksum="abc123", content_phash=111) + assert find_transfer_source(target) is None + + def test_sibling_evidence_not_current_for_its_own_card_is_not_a_source(self, db: Any) -> None: + """A sibling whose OWN evidence row has gone stale (content_hash no longer matches ITS + OWN card's live content_phash) is never a transfer source, regardless of md5 agreement.""" + sibling = CardFactory(md5_checksum="abc123", content_phash=222) # sibling's image changed + target = CardFactory(md5_checksum="abc123", content_phash=111) + ImageEvidenceFactory( + card=sibling, + content_hash=111, # stale - sibling's own card.content_phash is now 222 + md5_checksum="abc123", + extractor_versions=FULL_MANIFEST, + ) + + assert find_transfer_source(target) is None + + def test_partial_manifest_sibling_is_not_a_source(self, db: Any) -> None: + sibling = CardFactory(md5_checksum="abc123", content_phash=111) + target = CardFactory(md5_checksum="abc123", content_phash=111) + ImageEvidenceFactory( + card=sibling, + content_hash=111, + md5_checksum="abc123", + extractor_versions={"fetch_health": "fetch-health-v2"}, # not the full manifest + ) + + assert find_transfer_source(target) is None + + +class TestFindTransferSourceIntegrity: + """Tron §8 gate condition 4 (2026-07-25): eligibility requires the source evidence row's own + stamped md5_checksum to be NOT NULL and equal to the target card's - a sibling row that matches + on md5 only through the Card-level join (never through its own stamp) must never mint a fresh + stamp on the copy. Null-tolerance stays a CURRENCY-only rule (image_evidence. + current_evidence_queryset), never a transfer-source-eligibility one.""" + + def test_source_with_null_stamped_md5_is_not_eligible(self, db: Any) -> None: + sibling = CardFactory(md5_checksum="abc123", content_phash=111) + target = CardFactory(md5_checksum="abc123", content_phash=111) + # legacy row: content_hash is current, but it never got the md5 stamp at all. + ImageEvidenceFactory( + card=sibling, + content_hash=111, + md5_checksum=None, + extractor_versions=FULL_MANIFEST, + ) + + assert find_transfer_source(target) is None + + def test_source_with_disagreeing_stamped_md5_is_not_eligible(self, db: Any) -> None: + """Not reachable via the outer `card__md5_checksum=card.md5_checksum` filter today (the + sibling's own card carries the same md5 the target does, by construction), but a + stamped-vs-card-live disagreement on the SOURCE's own row is exactly the case the strict + (non-null-tolerant) filter is there to catch if the two data points were ever able to + diverge - proven directly against the queryset rather than assumed unreachable.""" + sibling = CardFactory(md5_checksum="abc123", content_phash=111) + target = CardFactory(md5_checksum="abc123", content_phash=111) + ImageEvidenceFactory( + card=sibling, + content_hash=111, + md5_checksum="stale-different-md5", + extractor_versions=FULL_MANIFEST, + ) + + assert find_transfer_source(target) is None + + +class TestFindTransferSourcePairingRule: + def test_sha256_absent_on_both_falls_back_to_md5_only(self, db: Any) -> None: + sibling = CardFactory(md5_checksum="abc123", content_phash=111, sha256_checksum=None) + target = CardFactory(md5_checksum="abc123", content_phash=111, sha256_checksum=None) + sibling_evidence = ImageEvidenceFactory( + card=sibling, content_hash=111, md5_checksum="abc123", extractor_versions=FULL_MANIFEST + ) + + found = find_transfer_source(target) + + assert found is not None + assert found.pk == sibling_evidence.pk + + def test_sha256_present_on_both_and_matching_transfers(self, db: Any) -> None: + sibling = CardFactory(md5_checksum="abc123", content_phash=111, sha256_checksum="deadbeef") + target = CardFactory(md5_checksum="abc123", content_phash=111, sha256_checksum="deadbeef") + sibling_evidence = ImageEvidenceFactory( + card=sibling, content_hash=111, md5_checksum="abc123", extractor_versions=FULL_MANIFEST + ) + + found = find_transfer_source(target) + + assert found is not None + assert found.pk == sibling_evidence.pk + + def test_sha256_present_on_only_one_side_falls_back_to_md5_only(self, db: Any) -> None: + sibling = CardFactory(md5_checksum="abc123", content_phash=111, sha256_checksum="deadbeef") + target = CardFactory(md5_checksum="abc123", content_phash=111, sha256_checksum=None) + sibling_evidence = ImageEvidenceFactory( + card=sibling, content_hash=111, md5_checksum="abc123", extractor_versions=FULL_MANIFEST + ) + + found = find_transfer_source(target) + + assert found is not None + assert found.pk == sibling_evidence.pk + + def test_sha256_mismatch_is_a_loud_anomaly_and_skips_transfer(self, db: Any, caplog: Any) -> None: + sibling = CardFactory(md5_checksum="abc123", content_phash=111, sha256_checksum="deadbeef") + target = CardFactory(md5_checksum="abc123", content_phash=111, sha256_checksum="cafebabe") + ImageEvidenceFactory(card=sibling, content_hash=111, md5_checksum="abc123", extractor_versions=FULL_MANIFEST) + + with caplog.at_level("ERROR"): + found = find_transfer_source(target) + + assert found is None + assert any("sha256_checksum disagrees" in record.message for record in caplog.records) + + def test_sha256_mismatch_writes_a_durable_card_scan_log_anomaly_row(self, db: Any) -> None: + """Tron §8 gate condition 5 (2026-07-25): the ERROR log alone isn't queryable after a + 218k-card run - the anomaly must also land as a durable, per-card CardScanLog row.""" + sibling = CardFactory(md5_checksum="abc123", content_phash=111, sha256_checksum="deadbeef") + target = CardFactory(md5_checksum="abc123", content_phash=111, sha256_checksum="cafebabe") + ImageEvidenceFactory(card=sibling, content_hash=111, md5_checksum="abc123", extractor_versions=FULL_MANIFEST) + + find_transfer_source(target) + + log = CardScanLog.objects.get(card=target, anonymous_id=EVIDENCE_TRANSFER_ANONYMOUS_ID) + assert log.skip_reason == EVIDENCE_TRANSFER_SHA256_MISMATCH_SKIP_REASON + + +class TestFindTransferSourceContentHashAssertion: + def test_content_phash_mismatch_is_a_loud_anomaly_and_skips_transfer(self, db: Any, caplog: Any) -> None: + """An md5 match whose sibling evidence's own content_hash disagrees with the TARGET + card's own content_phash is impossible for genuinely byte-identical files - a real + anomaly, not a stale-sibling case (the sibling's own evidence IS current for ITS OWN + card, per test_current_full_manifest_sibling_is_returned's own currency query - it's the + cross-card comparison that disagrees).""" + sibling = CardFactory(md5_checksum="abc123", content_phash=111) + target = CardFactory(md5_checksum="abc123", content_phash=222) # different phash, same md5 + ImageEvidenceFactory(card=sibling, content_hash=111, md5_checksum="abc123", extractor_versions=FULL_MANIFEST) + + with caplog.at_level("ERROR"): + found = find_transfer_source(target) + + assert found is None + assert any("content_phash disagrees" in record.message for record in caplog.records) + + def test_content_phash_mismatch_writes_a_durable_card_scan_log_anomaly_row(self, db: Any) -> None: + sibling = CardFactory(md5_checksum="abc123", content_phash=111) + target = CardFactory(md5_checksum="abc123", content_phash=222) + ImageEvidenceFactory(card=sibling, content_hash=111, md5_checksum="abc123", extractor_versions=FULL_MANIFEST) + + find_transfer_source(target) + + log = CardScanLog.objects.get(card=target, anonymous_id=EVIDENCE_TRANSFER_ANONYMOUS_ID) + assert log.skip_reason == EVIDENCE_TRANSFER_CONTENT_HASH_MISMATCH_SKIP_REASON + + +class TestFindTransferSourceKillSwitch: + """Tron §8 gate condition 6 (2026-07-25): settings.STAGE_C_EVIDENCE_TRANSFER_ENABLED.""" + + def test_disabled_returns_none_even_with_an_eligible_sibling(self, db: Any) -> None: + sibling = CardFactory(md5_checksum="abc123", content_phash=111) + target = CardFactory(md5_checksum="abc123", content_phash=111) + ImageEvidenceFactory(card=sibling, content_hash=111, md5_checksum="abc123", extractor_versions=FULL_MANIFEST) + + with override_settings(STAGE_C_EVIDENCE_TRANSFER_ENABLED=False): + assert find_transfer_source(target) is None + + def test_default_is_enabled(self, db: Any) -> None: + sibling = CardFactory(md5_checksum="abc123", content_phash=111) + target = CardFactory(md5_checksum="abc123", content_phash=111) + sibling_evidence = ImageEvidenceFactory( + card=sibling, content_hash=111, md5_checksum="abc123", extractor_versions=FULL_MANIFEST + ) + + found = find_transfer_source(target) + + assert found is not None + assert found.pk == sibling_evidence.pk + + +class TestTransferEvidence: + def test_copies_fields_and_stamps_target_values(self, db: Any) -> None: + sibling = CardFactory(md5_checksum="abc123", content_phash=111) + target = CardFactory(md5_checksum="abc123", content_phash=111) + sibling_evidence = ImageEvidenceFactory( + card=sibling, + content_hash=111, + md5_checksum="abc123", + extractor_versions=FULL_MANIFEST, + symbol_phash=999, + collector_line_raw_text="M15 123/456", + collector_line_set_code="M15", + collector_line_collector_number="123", + ) + + result = transfer_evidence(target, sibling_evidence, run_id="test-run") + + assert result.card_id == target.pk + assert result.content_hash == 111 + assert result.symbol_phash == 999 + assert result.collector_line_raw_text == "M15 123/456" + assert result.extractor_versions == FULL_MANIFEST + assert result.md5_checksum == "abc123" + assert result.transferred is True + assert result.transferred_from_card_id == sibling.pk + assert result.run_id == "test-run" + + # Persisted, not just returned. + stored = ImageEvidence.objects.get(card_id=target.pk, content_hash=111) + assert stored.transferred is True + assert stored.symbol_phash == 999 + + def test_get_or_create_updates_an_existing_row_in_place(self, db: Any) -> None: + sibling = CardFactory(md5_checksum="abc123", content_phash=111) + target = CardFactory(md5_checksum="abc123", content_phash=111) + sibling_evidence = ImageEvidenceFactory( + card=sibling, content_hash=111, md5_checksum="abc123", extractor_versions=FULL_MANIFEST, symbol_phash=999 + ) + existing = ImageEvidenceFactory(card=target, content_hash=111, extractor_versions={}) + + result = transfer_evidence(target, sibling_evidence, run_id="test-run") + + assert result.pk == existing.pk + assert ImageEvidence.objects.filter(card_id=target.pk).count() == 1 diff --git a/MPCAutofill/cardpicker/tests/test_image_evidence.py b/MPCAutofill/cardpicker/tests/test_image_evidence.py index a6b833e2c..fb1a87246 100644 --- a/MPCAutofill/cardpicker/tests/test_image_evidence.py +++ b/MPCAutofill/cardpicker/tests/test_image_evidence.py @@ -93,7 +93,7 @@ from cardpicker.local_ocr import DEFAULT_CROP_BOX, LEGAL_LINE_CROP_BOX from cardpicker.local_phash import ART_CROP_BOX from cardpicker.models import CardScanLog, ImageEvidence -from cardpicker.tests.factories import CardFactory +from cardpicker.tests.factories import CardFactory, ImageEvidenceFactory @dataclass(frozen=True) @@ -236,6 +236,20 @@ def test_successful_fetch_marks_fetch_ok_and_records_no_skip(self, db, monkeypat # module-level patch of run_tesseract). assert result.skip_reasons == {"artist_ocr": "no-text", "legal_line": "no-text"} + def test_forwards_the_cards_own_md5_checksum_onto_the_result(self, db, monkeypatch): + card = CardFactory(content_phash=12345, md5_checksum="abc123") + monkeypatch.setattr(module, "fetch_card_image", lambda card, dpi=None: _BLEED_IMAGE) + _stub_border_color(monkeypatch, "black") + _stub_ocr(monkeypatch) + _stub_symbol_region(monkeypatch) + _stub_quality_signals(monkeypatch) + _stub_color_profile(monkeypatch) + + result = extract_card_evidence(card) + + assert result.fields["md5_checksum"] == "abc123" + assert result.fields["sha256_checksum"] is None # Card.sha256_checksum doesn't exist yet + def test_failed_fetch_marks_fetch_not_ok_and_records_a_named_skip(self, db, monkeypatch): card = CardFactory(content_phash=12345) monkeypatch.setattr(module, "fetch_card_image", lambda card, dpi=None: None) @@ -1696,6 +1710,94 @@ def counting_bulk_create(objs, *args, **kwargs): assert bulk_create_calls == [5] assert CardScanLog.objects.filter(card=card, run_id="run-1").count() == 5 + def test_stamps_md5_and_sha256_when_present(self, db): + card = CardFactory(content_phash=999) + result = ExtractionResult( + card_id=card.pk, + content_hash=999, + fields={"fetch_ok": True, "md5_checksum": "abc123", "sha256_checksum": "deadbeef"}, + extractor_versions={"fetch_health": FETCH_HEALTH_EXTRACTOR_VERSION}, + ) + + evidence = persist_evidence(result, run_id="run-1") + + assert evidence is not None + assert evidence.md5_checksum == "abc123" + assert evidence.sha256_checksum == "deadbeef" + + def test_stamps_are_null_when_not_supplied(self, db): + card = CardFactory(content_phash=999) + result = ExtractionResult( + card_id=card.pk, + content_hash=999, + fields={"fetch_ok": True}, + extractor_versions={"fetch_health": FETCH_HEALTH_EXTRACTOR_VERSION}, + ) + + evidence = persist_evidence(result) + + assert evidence is not None + assert evidence.md5_checksum is None + assert evidence.sha256_checksum is None + + def test_real_extraction_clears_a_prior_transferred_flag(self, db): + """A row previously created via evidence_transfer.transfer_evidence (transferred=True) that + later receives a REAL extraction pass must have `transferred` reset - persist_evidence is + the only caller for a genuine extraction, so it's the right place to enforce this (issue + #473 PR-2's own interim-guard correctness note, see persist_evidence's own docstring).""" + card = CardFactory(content_phash=999) + evidence = ImageEvidenceFactory(card=card, content_hash=999, transferred=True, transferred_from_card_id=12345) + result = ExtractionResult( + card_id=card.pk, + content_hash=999, + fields={"fetch_ok": True}, + extractor_versions={"fetch_health": FETCH_HEALTH_EXTRACTOR_VERSION}, + ) + + updated = persist_evidence(result, run_id="run-2") + + assert updated is not None + assert updated.pk == evidence.pk + assert updated.transferred is False + assert updated.transferred_from_card_id is None + + +class TestCurrentEvidenceQueryset: + def test_content_hash_mismatch_is_stale(self, db): + card = CardFactory(content_phash=999) + ImageEvidenceFactory(card=card, content_hash=111) # a prior image version + + assert list(module.current_evidence_queryset(card)) == [] + + def test_matching_content_hash_no_md5_stamp_is_current(self, db): + """Null-tolerant: a legacy row with no md5_checksum stamp stays current regardless of + whether the card itself carries an md5 - explicit per this PR's own scope.""" + card = CardFactory(content_phash=999, md5_checksum="abc123") + evidence = ImageEvidenceFactory(card=card, content_hash=999, md5_checksum=None) + + assert list(module.current_evidence_queryset(card)) == [evidence] + + def test_card_with_no_md5_is_current_regardless_of_evidence_stamp(self, db): + card = CardFactory(content_phash=999, md5_checksum=None) + evidence = ImageEvidenceFactory(card=card, content_hash=999, md5_checksum="stale-leftover") + + assert list(module.current_evidence_queryset(card)) == [evidence] + + def test_matching_md5_stamp_is_current(self, db): + card = CardFactory(content_phash=999, md5_checksum="abc123") + evidence = ImageEvidenceFactory(card=card, content_hash=999, md5_checksum="abc123") + + assert list(module.current_evidence_queryset(card)) == [evidence] + + def test_stamped_md5_mismatch_is_stale(self, db): + """The staleness fix's own core case: content_hash still matches (the perceptual hash + didn't change) but the stamped md5 disagrees with the card's own live md5 - an in-place + file replacement at the same Drive location. Both non-null and disagreeing => stale.""" + card = CardFactory(content_phash=999, md5_checksum="new-md5") + ImageEvidenceFactory(card=card, content_hash=999, md5_checksum="old-md5") + + assert list(module.current_evidence_queryset(card)) == [] + class TestBuildReconciliationReport: def test_all_voted(self, db): diff --git a/MPCAutofill/cardpicker/tests/test_local_calculate_verdicts.py b/MPCAutofill/cardpicker/tests/test_local_calculate_verdicts.py index 103aadbf6..5fcc8d9a5 100644 --- a/MPCAutofill/cardpicker/tests/test_local_calculate_verdicts.py +++ b/MPCAutofill/cardpicker/tests/test_local_calculate_verdicts.py @@ -44,6 +44,7 @@ SLOW_PATH_ANONYMOUS_ID, SLOW_PATH_TO_REVIEW_REASON, STAGE_D_FALLBACK_ANONYMOUS_ID, + TRANSFERRED_INTERIM_GUARD_SKIP_REASON, _eligible_cards_queryset, _filter_by_symbol_phash, _resolve_candidates_for_card, @@ -763,6 +764,115 @@ def _stale_split(votes_batch): assert CardPrintingTag.objects.filter(card=card, anonymous_id=JOIN_KEY_ANONYMOUS_ID).count() == 1 +class TestTransferredInterimGuard: + """Issue #473 PR-2's INTERIM STAGE D GUARD (see TRANSFERRED_INTERIM_GUARD_SKIP_REASON's own + module-level comment in local_calculate_verdicts.py): a card whose CURRENT evidence row was + created via evidence transfer must never receive a machine vote from the join-key or fallback + calculators - its own "observation" is the same bytes an md5-sibling already voted from, not + an independent one. The slow-path calculator is NOT guarded (it casts no machine vote, only a + human-review routing marker - see its own run_slow_path_calculator loop comment).""" + + def test_join_key_skips_a_transferred_evidence_card(self, db): + card = CardFactory(name="Some Card", content_phash=42) + CanonicalCardFactory(name="Some Card", expansion__code="mom", collector_number="158") + _evidence( + card, + collector_line_set_code="mom", + collector_line_collector_number="158", + transferred=True, + transferred_from_card_id=999, + ) + + result = run_join_key_calculator(dry_run=False) + + assert result.cards_considered == 0 + assert result.votes_written == 0 + assert CardPrintingTag.objects.count() == 0 + log = CardScanLog.objects.get(card=card, anonymous_id=JOIN_KEY_ANONYMOUS_ID) + assert log.skip_reason == TRANSFERRED_INTERIM_GUARD_SKIP_REASON + + def test_join_key_still_votes_a_real_extraction_card(self, db): + """Control - the exact same evidence, minus transferred=True, votes normally. Proves the + guard is keyed on the flag, not some other incidental difference in the fixture.""" + card = CardFactory(name="Some Card", content_phash=42) + CanonicalCardFactory(name="Some Card", expansion__code="mom", collector_number="158") + _evidence( + card, + collector_line_set_code="mom", + collector_line_collector_number="158", + transferred=False, + ) + + result = run_join_key_calculator(dry_run=False) + + assert result.cards_considered == 1 + assert result.votes_written == 1 + assert CardPrintingTag.objects.filter(card=card).count() == 1 + + def test_transferred_guard_is_rescannable_after_a_real_extraction_lands(self, db): + card = CardFactory(name="Some Card", content_phash=42) + CanonicalCardFactory(name="Some Card", expansion__code="mom", collector_number="158") + _evidence( + card, + collector_line_set_code="mom", + collector_line_collector_number="158", + transferred=True, + transferred_from_card_id=999, + ) + + first = run_join_key_calculator(dry_run=False) + assert first.cards_considered == 0 + + # a real extraction pass lands (persist_evidence always clears transferred - see + # image_evidence.persist_evidence's own docstring) - re-running now casts the vote. + evidence = card.image_evidence.get() + evidence.transferred = False + evidence.transferred_from_card_id = None + evidence.save() + + second = run_join_key_calculator(dry_run=False) + assert second.cards_considered == 1 + assert second.votes_written == 1 + + def test_fallback_skips_a_transferred_evidence_card(self, db): + card = CardFactory(name="Some Card", content_phash=42) + CardScanLog.objects.create(card=card, anonymous_id=JOIN_KEY_ANONYMOUS_ID, skip_reason="no-text") + _evidence( + card, + collector_line_collector_number="", + symbol_phash=_hash_of("mom"), + transferred=True, + transferred_from_card_id=999, + ) + + result = run_fallback_calculator(dry_run=False) + + assert result.cards_considered == 0 + log = CardScanLog.objects.get(card=card, anonymous_id=STAGE_D_FALLBACK_ANONYMOUS_ID) + assert log.skip_reason == TRANSFERRED_INTERIM_GUARD_SKIP_REASON + + def test_slow_path_is_not_guarded_transferred_evidence_still_routes_to_review(self, db): + """The slow-path calculator casts no machine vote (only a CardScanLog routing marker to a + HUMAN reviewer) - deliberately excluded from the guard, see its own loop comment.""" + card = CardFactory(name="Some Card", content_phash=42) + CardPrintingTag.objects.create( + card=card, printing=None, is_no_match=True, anonymous_id=JOIN_KEY_ANONYMOUS_ID, source=VoteSource.OCR + ) + _evidence( + card, + collector_line_collector_number="999", + transferred=True, + transferred_from_card_id=999, + ) + + result = run_slow_path_calculator(dry_run=False) + + assert result.cards_considered == 1 + assert result.routed_written == 1 + log = CardScanLog.objects.get(card=card, anonymous_id=SLOW_PATH_ANONYMOUS_ID) + assert log.skip_reason == SLOW_PATH_TO_REVIEW_REASON + + class TestEligibleCardsQueryset: """`_eligible_cards_queryset`'s two knowledge-inventory excludes (module docstring's "CONSTANT #3" section; owner-ruled must-fix, 2026-07-22) - `RESOLUTION_FLOOR_DPI` and diff --git a/MPCAutofill/cardpicker/tests/test_run_image_evidence_cohort.py b/MPCAutofill/cardpicker/tests/test_run_image_evidence_cohort.py index 211690713..9efbfaf8c 100644 --- a/MPCAutofill/cardpicker/tests/test_run_image_evidence_cohort.py +++ b/MPCAutofill/cardpicker/tests/test_run_image_evidence_cohort.py @@ -68,8 +68,8 @@ from cardpicker.harvest_fetch_limiter import GoogleFetchLockoutError from cardpicker.management.commands import run_image_evidence_cohort as cohort_command -from cardpicker.models import PilotRunLedger -from cardpicker.tests.factories import CardFactory +from cardpicker.models import ImageEvidence, PilotRunLedger +from cardpicker.tests.factories import CardFactory, ImageEvidenceFactory class _SyncPoolStub: @@ -99,9 +99,13 @@ def submit(self, fn: Any, *args: Any) -> "Future[Any]": return future -def _stub_fetch_ok(card_id: int, stop_event: threading.Event) -> "cohort_command._FetchOutcome": +def _stub_fetch_ok( + card_id: int, stop_event: threading.Event, run_id: str = "", dry_run: bool = False +) -> "cohort_command._FetchOutcome": """Replaces the real fetch-stage step - no DB fetch, no network, always a clean success with - a trivial content_hash/image_bytes/fetch_latency_ms.""" + a trivial content_hash/image_bytes/fetch_latency_ms. `run_id`/`dry_run` (2026-07-25, issue + #473 PR-2) are accepted and ignored - this stub never exercises the evidence-transfer check, + only `_submit_more_fetch`'s own wider call signature.""" return cohort_command._FetchOutcome( card_id=card_id, content_hash=123, image_bytes=b"fake-jpeg-bytes", fetch_latency_ms=1.5, outcome=None ) @@ -117,6 +121,8 @@ def _stub_compute_ok( profile: bool = False, short_circuit: Optional[bool] = None, known_set_codes: Optional[frozenset[str]] = None, + md5_checksum: Optional[str] = None, + sha256_checksum: Optional[str] = None, ) -> tuple[int, str, Optional[dict[str, float]], bool]: """Replaces the real compute-stage step - no PIL decode, no extractors, no persist_evidence call, just the (card_id, outcome, profile, short_circuited) tuple `_run_cohort` consumes. @@ -224,7 +230,9 @@ def test_card_ids_file_with_a_nonexistent_card_id_drops_cleanly( the synchronous pool stub, which shares this test's own connection rather than a real forked/threaded one).""" - def _stub_fetch_dropped(card_id: int, stop_event: threading.Event) -> "cohort_command._FetchOutcome": + def _stub_fetch_dropped( + card_id: int, stop_event: threading.Event, run_id: str = "", dry_run: bool = False + ) -> "cohort_command._FetchOutcome": return cohort_command._FetchOutcome(card_id=card_id, outcome="dropped") monkeypatch.setattr(cohort_command, "_fetch_one_card", _stub_fetch_dropped) @@ -326,7 +334,9 @@ def test_fetch_failures_and_short_circuited_land_in_ledger_counters( CardFactory(content_phash=1) CardFactory(content_phash=2) - def _stub_fetch_one_dropped(card_id: int, stop_event: threading.Event) -> "cohort_command._FetchOutcome": + def _stub_fetch_one_dropped( + card_id: int, stop_event: threading.Event, run_id: str = "", dry_run: bool = False + ) -> "cohort_command._FetchOutcome": return cohort_command._FetchOutcome(card_id=card_id, outcome="dropped") def _stub_compute_short_circuited( @@ -339,6 +349,8 @@ def _stub_compute_short_circuited( profile: bool = False, short_circuit: Optional[bool] = None, known_set_codes: Optional[frozenset[str]] = None, + md5_checksum: Optional[str] = None, + sha256_checksum: Optional[str] = None, ) -> tuple[int, str, Optional[dict[str, float]], bool]: return card_id, "ok", None, True @@ -347,11 +359,13 @@ def _stub_compute_short_circuited( # so a single stub covering both call sites keeps this deterministic regardless of order. calls = {"count": 0} - def _fetch_first_dropped_rest_ok(card_id: int, stop_event: threading.Event) -> "cohort_command._FetchOutcome": + def _fetch_first_dropped_rest_ok( + card_id: int, stop_event: threading.Event, run_id: str = "", dry_run: bool = False + ) -> "cohort_command._FetchOutcome": calls["count"] += 1 if calls["count"] == 1: - return _stub_fetch_one_dropped(card_id, stop_event) - return _stub_fetch_ok(card_id, stop_event) + return _stub_fetch_one_dropped(card_id, stop_event, run_id, dry_run) + return _stub_fetch_ok(card_id, stop_event, run_id, dry_run) monkeypatch.setattr(cohort_command, "_fetch_one_card", _fetch_first_dropped_rest_ok) monkeypatch.setattr(cohort_command, "_compute_one_card", _stub_compute_short_circuited) @@ -534,6 +548,68 @@ def test_successful_fetch_returns_none_outcome_with_populated_fields(self, monke assert result.image_bytes == b"raw-bytes" assert result.fetch_latency_ms >= 0.0 + @pytest.mark.django_db + def test_md5_sibling_transfers_without_ever_fetching(self, monkeypatch: pytest.MonkeyPatch) -> None: + sibling = CardFactory(md5_checksum="abc123", content_phash=111) + target = CardFactory(md5_checksum="abc123", content_phash=111) + ImageEvidenceFactory( + card=sibling, + content_hash=111, + md5_checksum="abc123", + extractor_versions={key: "v1" for key in cohort_command.MANIFEST_EXTRACTOR_KEYS}, + symbol_phash=999, + ) + stop_event = threading.Event() + + def _fail_if_called(card: Any, dpi: Optional[int] = None) -> None: + raise AssertionError("a transfer-eligible card must never reach the fetch stage") + + import cardpicker.image_cdn_fetch as image_cdn_fetch_module + + monkeypatch.setattr(image_cdn_fetch_module, "fetch_card_image_bytes", _fail_if_called) + + result = cohort_command._fetch_one_card(card_id=target.pk, stop_event=stop_event, run_id="r1", dry_run=False) + + assert result.outcome == "transferred" + assert result.content_hash == 111 + assert result.image_bytes is None + evidence = ImageEvidence.objects.get(card=target) + assert evidence.transferred is True + assert evidence.transferred_from_card_id == sibling.pk + assert evidence.symbol_phash == 999 + + @pytest.mark.django_db + def test_dry_run_finds_a_transfer_but_writes_nothing(self, monkeypatch: pytest.MonkeyPatch) -> None: + sibling = CardFactory(md5_checksum="abc123", content_phash=111) + target = CardFactory(md5_checksum="abc123", content_phash=111) + ImageEvidenceFactory( + card=sibling, + content_hash=111, + md5_checksum="abc123", + extractor_versions={key: "v1" for key in cohort_command.MANIFEST_EXTRACTOR_KEYS}, + ) + stop_event = threading.Event() + + result = cohort_command._fetch_one_card(card_id=target.pk, stop_event=stop_event, run_id="r1", dry_run=True) + + assert result.outcome == "transferred" + assert ImageEvidence.objects.filter(card=target).count() == 0 + + @pytest.mark.django_db + def test_no_eligible_sibling_falls_through_to_a_real_fetch(self, monkeypatch: pytest.MonkeyPatch) -> None: + card = CardFactory(md5_checksum="abc123", content_phash=111) # no sibling exists at all + stop_event = threading.Event() + + import cardpicker.image_cdn_fetch as image_cdn_fetch_module + + monkeypatch.setattr(image_cdn_fetch_module, "fetch_card_image_bytes", lambda card, dpi=None: b"raw-bytes") + + result = cohort_command._fetch_one_card(card_id=card.pk, stop_event=stop_event) + + assert result.outcome is None + assert result.image_bytes == b"raw-bytes" + assert ImageEvidence.objects.filter(card=card).count() == 0 # only persisted by compute + class TestComputeOneCard: """`_compute_one_card` is the new compute-stage work unit - decodes the already-fetched raw @@ -558,6 +634,8 @@ def _stub_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, ) -> Any: captured["card_id"] = card_id captured["content_hash"] = content_hash @@ -600,6 +678,8 @@ def _stub_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, ) -> Any: captured["image"] = image @@ -662,6 +742,8 @@ def _stub_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, ) -> Any: if profile is not None: profile["fetch_ms"] = fetch_latency_ms @@ -769,6 +851,8 @@ def _stub_compute_with_profile( profile: bool = False, short_circuit: Optional[bool] = None, known_set_codes: Optional[frozenset[str]] = None, + md5_checksum: Optional[str] = None, + sha256_checksum: Optional[str] = None, ) -> tuple[int, str, Optional[dict[str, float]], bool]: profile_dict = {"fetch_ms": fetch_latency_ms, "wall_ms": 1.0} if profile else None return card_id, "ok", profile_dict, False @@ -878,7 +962,7 @@ def submit(self, fn: Any, *args: Any) -> "Future[Any]": monkeypatch.setattr( cohort_command, "_fetch_one_card", - lambda card_id, stop_event: cohort_command._FetchOutcome( + lambda card_id, stop_event, run_id="", dry_run=False: cohort_command._FetchOutcome( card_id=card_id, content_hash=1, image_bytes=b"x", fetch_latency_ms=0.0, outcome=None ), ) @@ -975,7 +1059,9 @@ def _make_payload(card_id: int) -> "_Payload": weakref.finalize(payload, _mark_dead, card_id) return payload - def _stub_fetch(card_id: int, stop_event: threading.Event) -> "cohort_command._FetchOutcome": + def _stub_fetch( + card_id: int, stop_event: threading.Event, run_id: str = "", dry_run: bool = False + ) -> "cohort_command._FetchOutcome": return cohort_command._FetchOutcome( card_id=card_id, content_hash=1, image_bytes=_make_payload(card_id), outcome=None ) diff --git a/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py b/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py index b2b230d7b..9d755b249 100644 --- a/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py @@ -14,6 +14,8 @@ """ import io +import threading +import time from typing import Any import psycopg2 @@ -92,7 +94,15 @@ def _stub_compute_card_evidence_ok(**field_overrides: Any): steer Stage D's own verdict.""" def _stub( - card_id: int, content_hash, image, fetch_latency_ms=0.0, profile=None, short_circuit=None, known_set_codes=None + card_id: int, + content_hash, + image, + fetch_latency_ms=0.0, + profile=None, + short_circuit=None, + known_set_codes=None, + md5_checksum=None, + sha256_checksum=None, ): fields = { "fetch_ok": True, @@ -567,10 +577,20 @@ def _recording_stub( profile=None, short_circuit=None, known_set_codes=None, + md5_checksum=None, + sha256_checksum=None, ): observed_short_circuit.append(short_circuit) return _stub_compute_card_evidence_ok()( - card_id, content_hash, image, fetch_latency_ms, profile, short_circuit, known_set_codes + card_id, + content_hash, + image, + fetch_latency_ms, + profile, + short_circuit, + known_set_codes, + md5_checksum, + sha256_checksum, ) import cardpicker.image_cdn_fetch as image_cdn_fetch_module @@ -1077,3 +1097,289 @@ def test_sweep_stays_bounded_by_max_batches_when_cap_hit_empty_never_resolves( assert "Backlog exhausted" not in output assert "Envelope halt" not in output assert "sweep stopped" not in output + + +class TestEvidenceTransferInDispatch: + """Issue #473 PR-2's evidence transfer, wired into `_run_stage_c`'s own phase 1 (checked + BEFORE a card is ever handed to the fetch-ahead thread).""" + + @STREAMING_ON + def test_md5_sibling_transfers_without_ever_fetching(self, db: Any, monkeypatch: pytest.MonkeyPatch) -> None: + sibling = CardFactory(name="Sibling", md5_checksum="abc123", content_phash=111) + target = CardFactory(name="Target", md5_checksum="abc123", content_phash=111) + ImageEvidenceFactory( + card=sibling, + content_hash=111, + md5_checksum="abc123", + extractor_versions={key: f"{key}-v1" for key in MANIFEST_EXTRACTOR_KEYS}, + symbol_phash=999, + ) + + def _fail_if_called(card: Any, dpi: Any = None) -> Any: + raise AssertionError("a transfer-eligible card must never reach the fetch stage") + + _install_stage_c_stub(monkeypatch, fetch_result=_fail_if_called) + + outcome = dispatch_micro_batch(card_ids=[target.pk]) + + assert outcome.status == "completed" + assert outcome.stage_c_completed == 1 + assert outcome.stage_c_transferred == 1 + evidence = ImageEvidence.objects.get(card=target) + assert evidence.transferred is True + assert evidence.transferred_from_card_id == sibling.pk + assert evidence.symbol_phash == 999 + assert evidence.md5_checksum == "abc123" + + @STREAMING_ON + def test_a_real_extraction_card_alongside_a_transfer_card_both_land( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + sibling = CardFactory(name="Sibling", md5_checksum="abc123", content_phash=111) + transfer_target = CardFactory(name="Target", md5_checksum="abc123", content_phash=111) + ImageEvidenceFactory( + card=sibling, + content_hash=111, + md5_checksum="abc123", + extractor_versions={key: f"{key}-v1" for key in MANIFEST_EXTRACTOR_KEYS}, + ) + real_card = CardFactory(name="Real", content_phash=222) # no md5 - always real extraction + + fetched_card_ids: list[int] = [] + + def _tracking_fetch(card: Any, dpi: Any = None) -> Any: + fetched_card_ids.append(card.pk) + return _png_bytes() + + _install_stage_c_stub(monkeypatch, fetch_result=_tracking_fetch) + + outcome = dispatch_micro_batch(card_ids=[transfer_target.pk, real_card.pk]) + + assert outcome.stage_c_completed == 2 + assert outcome.stage_c_transferred == 1 + assert fetched_card_ids == [real_card.pk] # the transfer target never hits the fetch stage + assert ImageEvidence.objects.get(card=transfer_target).transferred is True + assert ImageEvidence.objects.get(card=real_card).transferred is False + + +class TestDecoupledFetchAhead: + """Issue #472's fetch-ahead thread + bounded queue, retrofitted into `_run_stage_c`.""" + + @STREAMING_ON + def test_fetch_ahead_overlaps_with_the_current_cards_own_compute( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Proves genuine OVERLAP, not just correctness: the second card's own fetch must start + before the first card's own compute finishes - if fetch and compute were still bundled + sequentially (the pre-#472 design), card B's fetch would only ever start AFTER card A's + compute (and persist) had already completed.""" + card_a = CardFactory(name="A", content_phash=1) + card_b = CardFactory(name="B", content_phash=2) + events: list[tuple[str, int, float]] = [] + + def fake_fetch(card: Any, dpi: Any = None) -> Any: + events.append(("fetch_start", card.pk, time.monotonic())) + time.sleep(0.05) + events.append(("fetch_end", card.pk, time.monotonic())) + return _png_bytes() + + def fake_compute( + card_id: int, + content_hash: Any, + image: Any, + fetch_latency_ms: float = 0.0, + profile: Any = None, + short_circuit: Any = None, + known_set_codes: Any = None, + md5_checksum: Any = None, + sha256_checksum: Any = None, + ) -> Any: + events.append(("compute_start", card_id, time.monotonic())) + time.sleep(0.1) + events.append(("compute_end", card_id, time.monotonic())) + return _stub_compute_card_evidence_ok()( + card_id, + content_hash, + image, + fetch_latency_ms, + profile, + short_circuit, + known_set_codes, + md5_checksum, + sha256_checksum, + ) + + import cardpicker.image_cdn_fetch as image_cdn_fetch_module + import cardpicker.image_evidence as image_evidence_module + + monkeypatch.setattr(image_cdn_fetch_module, "fetch_card_image_bytes", fake_fetch) + monkeypatch.setattr(image_evidence_module, "compute_card_evidence", fake_compute) + + outcome = dispatch_micro_batch(card_ids=[card_a.pk, card_b.pk]) + + assert outcome.status == "completed" + assert outcome.stage_c_completed == 2 + fetch_b_start = next(t for (name, cid, t) in events if name == "fetch_start" and cid == card_b.pk) + compute_a_end = next(t for (name, cid, t) in events if name == "compute_end" and cid == card_a.pk) + assert fetch_b_start < compute_a_end + + @STREAMING_ON + def test_lockout_mid_prefetch_drains_the_already_fetched_card_but_starts_no_more( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + card_a = CardFactory(name="A", content_phash=1) + card_b = CardFactory(name="B", content_phash=2) + card_c = CardFactory(name="C", content_phash=3) + fetched_card_ids: list[int] = [] + + def fake_fetch(card: Any, dpi: Any = None) -> Any: + fetched_card_ids.append(card.pk) + if card.pk == card_b.pk: + raise GoogleFetchLockoutError("locked out") + return _png_bytes() + + _install_stage_c_stub(monkeypatch, fetch_result=fake_fetch) + + outcome = dispatch_micro_batch(card_ids=[card_a.pk, card_b.pk, card_c.pk]) + + assert outcome.status == "completed-with-trip" + # in-flight work drains: card A's already-fetched image still gets computed+persisted. + assert ImageEvidence.objects.filter(card=card_a).count() == 1 + assert ImageEvidence.objects.filter(card=card_b).count() == 0 + assert ImageEvidence.objects.filter(card=card_c).count() == 0 + # halts NEW fetches immediately - card C is never even attempted. + assert card_c.pk not in fetched_card_ids + + @STREAMING_ON + def test_fetch_outcome_window_records_in_fetch_submission_order( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + cards = [CardFactory(name=f"Card {i}", content_phash=i) for i in range(1, 5)] + succeeds_by_pk = {cards[0].pk: True, cards[1].pk: False, cards[2].pk: True, cards[3].pk: False} + + def fake_fetch(card: Any, dpi: Any = None) -> Any: + return _png_bytes() if succeeds_by_pk[card.pk] else None + + _install_stage_c_stub(monkeypatch, fetch_result=fake_fetch) + + outcome = dispatch_micro_batch(card_ids=[c.pk for c in cards]) + + assert outcome.status == "completed" + assert outcome.stage_c_fetch_failures == 2 + # the window's own recorded order matches the cards' own submission order, despite the + # fetch-ahead thread running concurrently with compute - a single serial fetch worker's + # own completion order IS its submission order (module docstring's own argument). + assert list(stage_e_dispatch._window._window) == [True, False, True, False] + + @STREAMING_ON + def test_a_non_lockout_fetch_crash_propagates_instead_of_hanging( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Regression pin (2026-07-25, found during this PR's own review): an uncaught exception + raised INSIDE the fetch-ahead thread must propagate to the caller, not silently hang the + main thread's own `queue.get()` waiting for an outcome that will never arrive. Mirrors + TestKillSafetyResumeContract's own mid-batch-crash scenario, narrowed to pin the fetch-ahead + thread's own exception-forwarding mechanism specifically.""" + card_a = CardFactory(name="A", content_phash=1) + card_b = CardFactory(name="B", content_phash=2) + + def fake_fetch(card: Any, dpi: Any = None) -> Any: + if card.pk == card_b.pk: + raise RuntimeError("simulated non-lockout fetch crash") + return _png_bytes() + + _install_stage_c_stub(monkeypatch, fetch_result=fake_fetch) + + with pytest.raises(RuntimeError, match="simulated non-lockout fetch crash"): + dispatch_micro_batch(card_ids=[card_a.pk, card_b.pk], run_id="crash-drill") + + ledger = PilotRunLedger.objects.get(run_id="crash-drill") + assert ledger.status == PilotRunLedger.Status.FAILED + assert "RuntimeError" in ledger.counters["failure_reason"] + # card A's own already-fetched work still committed before the crash. + assert ImageEvidence.objects.filter(card=card_a).count() == 1 + assert ImageEvidence.objects.filter(card=card_b).count() == 0 + + @STREAMING_ON + def test_a_compute_crash_does_not_wedge_the_fetch_ahead_thread( + self, transactional_db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Regression pin (Tron §8 gate condition 3, 2026-07-25, HIGH severity): a crash during + COMPUTE (not fetch) - e.g. a PIL error decoding a corrupt download - must not wedge the + fetch-ahead thread forever on a full `out_queue.put(...)` (see `_run_stage_c`'s own + `finally` block docstring for the full mechanism this pins: `stop_event.set()` BEFORE + `fetch_thread.join()`, plus draining the queue). More cards than the fetch-ahead queue + depth so the fetch thread reliably races ahead of compute and is genuinely blocked on its + own `put()` by the time compute raises - a bare `join()` with no signal/drain first would + hang this test (and, in prod, wedge the dispatch slot with a lying RUNNING ledger row) + indefinitely; the `run_thread.join(timeout=...)` below is what actually proves "does not + hang" rather than merely "eventually completes if given long enough". + + `transactional_db`, not the plain `db` fixture (2026-07-25, found running this test): + `dispatch_micro_batch` runs on a REAL background thread here (needed so the test itself can + enforce a wall-clock timeout, since a hang is exactly the bug being pinned) - a thread with + its own DB connection reading/writing against fixture data created inside the plain `db` + fixture's own uncommitted SAVEPOINT-wrapped transaction is precisely the class of problem + `test_run_image_evidence_cohort.py`'s own module docstring documents needing + `transaction=True` for (real commit-and-truncate isolation, matching prod's own + "no surrounding atomic block" shape) - the plain `db` fixture reproduced a SEPARATE, + fixture-level hang (the background thread blocked waiting on the main thread's own + transaction) that had nothing to do with the fetch-ahead bug this test exists to pin.""" + cards = [CardFactory(name=f"Card {i}", content_phash=i) for i in range(1, 6)] # > queue depth + + def fake_fetch(card: Any, dpi: Any = None) -> Any: + return _png_bytes() # fast, no sleep - lets fetch race ahead of compute + + compute_calls = {"n": 0} + + def fake_compute( + card_id: int, + content_hash: Any, + image: Any, + fetch_latency_ms: float = 0.0, + profile: Any = None, + short_circuit: Any = None, + known_set_codes: Any = None, + md5_checksum: Any = None, + sha256_checksum: Any = None, + ) -> Any: + compute_calls["n"] += 1 + if compute_calls["n"] == 2: + raise RuntimeError("simulated compute-side crash") + return _stub_compute_card_evidence_ok()( + card_id, + content_hash, + image, + fetch_latency_ms, + profile, + short_circuit, + known_set_codes, + md5_checksum, + sha256_checksum, + ) + + import cardpicker.image_cdn_fetch as image_cdn_fetch_module + import cardpicker.image_evidence as image_evidence_module + + monkeypatch.setattr(image_cdn_fetch_module, "fetch_card_image_bytes", fake_fetch) + monkeypatch.setattr(image_evidence_module, "compute_card_evidence", fake_compute) + + result_holder: dict[str, Any] = {} + + def _run() -> None: + try: + dispatch_micro_batch(card_ids=[c.pk for c in cards], run_id="compute-crash-drill") + except Exception as exc: # noqa: BLE001 - captured for the assertion below, not swallowed + result_holder["exc"] = exc + + run_thread = threading.Thread(target=_run, daemon=True) + run_thread.start() + run_thread.join(timeout=10) + + assert not run_thread.is_alive(), ( + "dispatch_micro_batch hung - the fetch-ahead thread was likely wedged on a full " + "queue after the compute-side crash (Tron §8 gate condition 3)" + ) + assert isinstance(result_holder.get("exc"), RuntimeError) + ledger = PilotRunLedger.objects.get(run_id="compute-crash-drill") + assert ledger.status == PilotRunLedger.Status.FAILED diff --git a/MPCAutofill/cardpicker/tests/test_stage_e_shakedown.py b/MPCAutofill/cardpicker/tests/test_stage_e_shakedown.py index c39882f77..cbecf23b2 100644 --- a/MPCAutofill/cardpicker/tests/test_stage_e_shakedown.py +++ b/MPCAutofill/cardpicker/tests/test_stage_e_shakedown.py @@ -185,7 +185,15 @@ def _png_bytes() -> bytes: return buffer.getvalue() def _stub_compute( - card_id: int, content_hash, image, fetch_latency_ms=0.0, profile=None, short_circuit=None, known_set_codes=None + card_id: int, + content_hash, + image, + fetch_latency_ms=0.0, + profile=None, + short_circuit=None, + known_set_codes=None, + md5_checksum=None, + sha256_checksum=None, ): fields = { "fetch_ok": True, diff --git a/MPCAutofill/cardpicker/tests/test_stage_e_signals.py b/MPCAutofill/cardpicker/tests/test_stage_e_signals.py new file mode 100644 index 000000000..4f48ee035 --- /dev/null +++ b/MPCAutofill/cardpicker/tests/test_stage_e_signals.py @@ -0,0 +1,104 @@ +""" +Tests for cardpicker.stage_e_signals - the card-create/evidence-change event triggers +(docs/proposals/stage-e-streaming.md §3 decision (1)) and, since 2026-07-25 (issue #472), the +`suppress_evidence_change_echo` ECHO SUPPRESSION mechanism (see that module's own docstring). +`django_q.tasks.async_task` is monkeypatched at `cardpicker.stage_e_signals` itself (both +receivers import it inline, at call time, inside the function body - a patch applied to the name +INSIDE that module before the save is what a fresh `from django_q.tasks import async_task` call +actually observes). +""" + +from typing import Any + +from django.test import override_settings + +from cardpicker import stage_e_signals +from cardpicker.stage_e_signals import suppress_evidence_change_echo +from cardpicker.tests.factories import CardFactory, ImageEvidenceFactory + +STREAMING_ON = override_settings(STAGE_E_STREAMING_ENABLED=True) + + +def _install_async_task_spy(monkeypatch: Any) -> list[tuple[Any, ...]]: + calls: list[tuple[Any, ...]] = [] + + def _fake_async_task(*args: Any, **kwargs: Any) -> None: + calls.append(args) + + # Both receivers do `from django_q.tasks import async_task` INSIDE the function body - patch + # the real source module so that fresh import observes the fake. + import django_q.tasks as django_q_tasks_module + + monkeypatch.setattr(django_q_tasks_module, "async_task", _fake_async_task) + return calls + + +class TestEvidenceChangeEchoSuppression: + @STREAMING_ON + def test_a_write_outside_the_dispatch_context_fires_async_task(self, db: Any, monkeypatch: Any) -> None: + card = CardFactory(content_phash=42) # before the spy - isolates this to the EVIDENCE save + calls = _install_async_task_spy(monkeypatch) + + ImageEvidenceFactory(card=card) # an ordinary BULK-mode-shaped save, no context wrapper + + assert len(calls) == 1 + assert calls[0] == ("cardpicker.stage_e_dispatch.dispatch_for_card", card.pk, "evidence-change") + + @STREAMING_ON + def test_a_write_inside_the_dispatch_context_fires_no_async_task(self, db: Any, monkeypatch: Any) -> None: + card = CardFactory(content_phash=42) + calls = _install_async_task_spy(monkeypatch) + + with suppress_evidence_change_echo(): + ImageEvidenceFactory(card=card) + + assert calls == [] + + @STREAMING_ON + def test_the_flag_resets_after_the_context_exits(self, db: Any, monkeypatch: Any) -> None: + card_a = CardFactory(content_phash=1) + card_b = CardFactory(content_phash=2) + calls = _install_async_task_spy(monkeypatch) + + with suppress_evidence_change_echo(): + ImageEvidenceFactory(card=card_a) + ImageEvidenceFactory(card=card_b) # outside the context again + + assert len(calls) == 1 + assert calls[0] == ("cardpicker.stage_e_dispatch.dispatch_for_card", card_b.pk, "evidence-change") + + def test_disabled_by_default_fires_no_async_task_regardless_of_context(self, db: Any, monkeypatch: Any) -> None: + """STAGE_E_STREAMING_ENABLED's own gate is checked FIRST, before the suppression flag - + confirms the two gates are independent (this test runs WITHOUT @STREAMING_ON).""" + calls = _install_async_task_spy(monkeypatch) + card = CardFactory(content_phash=42) + + ImageEvidenceFactory(card=card) + + assert calls == [] + + def test_the_context_manager_itself_is_reentrant_safe(self) -> None: + """Nested `with` blocks (never actually exercised by production code today - `_run_stage_c` + never nests these calls - but the mechanism itself must not misbehave if it ever did).""" + assert stage_e_signals._dispatch_persist_in_progress.get() is False + with suppress_evidence_change_echo(): + assert stage_e_signals._dispatch_persist_in_progress.get() is True + with suppress_evidence_change_echo(): + assert stage_e_signals._dispatch_persist_in_progress.get() is True + # the OUTER context is still active after the inner one exits. + assert stage_e_signals._dispatch_persist_in_progress.get() is True + assert stage_e_signals._dispatch_persist_in_progress.get() is False + + +class TestCardCreateSignalUnaffectedByEvidenceSuppression: + @STREAMING_ON + def test_card_create_still_fires_inside_an_evidence_suppression_context(self, db: Any, monkeypatch: Any) -> None: + """The suppression flag is scoped to the EVIDENCE-CHANGE receiver only - a Card creation + inside the same context is a different signal entirely and must be unaffected.""" + calls = _install_async_task_spy(monkeypatch) + + with suppress_evidence_change_echo(): + card = CardFactory(content_phash=42) + + assert len(calls) == 1 + assert calls[0] == ("cardpicker.stage_e_dispatch.dispatch_for_card", card.pk, "card-create") diff --git a/docker/docker-compose.prod.yml b/docker/docker-compose.prod.yml index ab8662cf1..6c1183eee 100644 --- a/docker/docker-compose.prod.yml +++ b/docker/docker-compose.prod.yml @@ -40,6 +40,9 @@ services: # Stage E streaming activation (docs/features/stage-e-operations.md) - reads # docker/.env; default False = dormant. Phase-3 activation flips the .env value. - STAGE_E_STREAMING_ENABLED=${STAGE_E_STREAMING_ENABLED:-False} + # Stage C evidence transfer kill-switch (issue #473 PR-2, Tron §8 gate condition) - default + # True (transfer ON); docker/.env can flip it False to isolate a live-run anomaly. + - STAGE_C_EVIDENCE_TRANSFER_ENABLED=${STAGE_C_EVIDENCE_TRANSFER_ENABLED:-True} # Discord OAuth (moderation layer, docs/features/moderation.md) - docker/.env alone # does NOT inject arbitrary variables into the container; only vars explicitly listed # here (via ${...} interpolation) actually reach the process. This block was missing @@ -87,6 +90,9 @@ services: # Stage E streaming activation (docs/features/stage-e-operations.md) - reads # docker/.env; default False = dormant. Phase-3 activation flips the .env value. - STAGE_E_STREAMING_ENABLED=${STAGE_E_STREAMING_ENABLED:-False} + # Stage C evidence transfer kill-switch (issue #473 PR-2, Tron §8 gate condition) - default + # True (transfer ON); docker/.env can flip it False to isolate a live-run anomaly. + - STAGE_C_EVIDENCE_TRANSFER_ENABLED=${STAGE_C_EVIDENCE_TRANSFER_ENABLED:-True} # Discord OAuth (moderation layer, docs/features/moderation.md) - docker/.env alone # does NOT inject arbitrary variables into the container; only vars explicitly listed # here (via ${...} interpolation) actually reach the process. This block was missing diff --git a/docs/features/stage-e-operations.md b/docs/features/stage-e-operations.md index 4e48b67e8..1ebadfb7c 100644 --- a/docs/features/stage-e-operations.md +++ b/docs/features/stage-e-operations.md @@ -285,14 +285,17 @@ entry points (`run_join_key_calculator`/`run_fallback_calculator`/ this box's django-q2 worker processes. See "Concurrency cap" below for the full mechanism and the incident that motivated it — distinct from, and a proactive complement to, the envelope's own reactive host-load bar. -6. **Stage C** (sequential, per-card, not pooled — a micro-batch is far too - small for BULK mode's process-pool concurrency to help) — the same +6. **Stage C** (COMPUTE sequential, per-card, not pooled — a micro-batch is + far too small for BULK mode's process-pool concurrency to help; FETCH + overlapped with compute since 2026-07-25, issue #472 — see "Evidence + transfer and decoupled fetch-ahead" below) — the same `compute_card_evidence`/`persist_evidence` unit `run_image_evidence_cohort.py` - drives, one card at a time. A `GoogleFetchLockoutError` stops Stage C for - this batch immediately and records a fresh trip (instant-pause bar) — - in-flight, already-committed work stays committed; Stage D below still - runs against whatever was reached ("in-flight work drains, nothing NEW - starts"). + drives. Before a card is fetched at all, an md5-sibling evidence-transfer + check runs (issue #473 PR-2, same section below) — a hit skips the fetch + entirely. A `GoogleFetchLockoutError` stops Stage C for this batch + immediately and records a fresh trip (instant-pause bar) — in-flight, + already-committed work stays committed; Stage D below still runs against + whatever was reached ("in-flight work drains, nothing NEW starts"). 7. **Stage D** — `run_join_key_calculator`/`run_fallback_calculator`/ `run_slow_path_calculator`, called AS-IS with the new `card_ids` scope, in the same escalation order every BULK-mode invocation already uses. Each of @@ -305,6 +308,138 @@ entry points (`run_join_key_calculator`/`run_fallback_calculator`/ step 5 is released (always, including on an exception - see "Concurrency cap" below). +### Evidence transfer and decoupled fetch-ahead (issues #473 PR-2 and #472, 2026-07-25) + +Both landed together (owner-approved fold — "same function, one coherent +change") inside `stage_e_dispatch._run_stage_c`'s own Stage C leg, plus the +matching seam in `run_image_evidence_cohort.py`'s BULK fetch stage +(`_fetch_one_card`). Neither changes Stage C's own OUTPUT shape or Stage D's +own decode logic — both are dispatch-side efficiency/soundness fixes. + +**Evidence transfer (#473 PR-2).** Before a card with a known +`Card.md5_checksum` (issue #473 PR-1's own checksum substrate) is fetched at +all, `cardpicker.evidence_transfer.find_transfer_source` looks for an +md5-identical sibling card's own CURRENT, full-manifest `ImageEvidence` row. +If found, `evidence_transfer.transfer_evidence` copies that row's field +values onto the target card's own `(card, content_hash)` row instead of +paying for a real fetch+OCR pass over what would decode to byte-identical +pixels — `find_transfer_source` never trusts the pairing blindly: + +- **Kill-switch**: `settings.STAGE_C_EVIDENCE_TRANSFER_ENABLED` (default + `True`) gates the whole function — `False` returns `None` immediately, no + query issued, both call sites fall straight through to their own + pre-existing real-fetch path. Exists for first-pass reversibility (Tron §8 + gate condition, 2026-07-25) — a single settings flip isolates whether a + live-run anomaly originates in transfer, no code change or redeploy needed. +- **Transfer-source integrity** (Tron §8 gate condition): a sibling row is + only a valid SOURCE if its own stamped `md5_checksum` IS NOT NULL and + EQUALS the target's live md5 — a strict, non-null-tolerant match, deliberately + NOT the same null-tolerant rule the staleness fix below uses for CURRENCY. + A sibling that never got the md5 stamp at all (a legacy row) is never + eligible to seed a transfer, so a fresh stamp is never minted on the copy + from an unverified source. +- **Content-hash assertion**: byte-identical files imply an identical + perceptual hash — the sibling's own `content_hash` is compared against the + target's own live `content_phash`. A mismatch is IMPOSSIBLE for genuinely + byte-identical files, so observing one is a LOUD anomaly. +- **sha256 pairing rule** (binding, owner-ratified 2026-07-25 alongside + `Card.sha256_checksum`'s own addition, now a real column on every deploy): + whenever BOTH cards carry a sha256, it must ALSO match — md5 collisions + are constructible, sha256 is the cryptographic backstop. A present-on-both + mismatch is the same kind of loud anomaly as a content-hash mismatch. + +Both anomaly paths log at ERROR **and** write a durable `CardScanLog(anonymous_id="evidence-transfer-v1", skip_reason=)` +row (Tron §8 gate condition, 2026-07-25) — the log line alone isn't +queryable after the fact; a whole-catalog run needs to COUNT these per card. +Either anomaly SKIPS the transfer and falls through to real extraction, +never a silent downgrade. + +A transferred row is stamped `md5_checksum`/`sha256_checksum` (from the +TARGET card's own live values, not copied from the sibling) and +`transferred=True` — an `ImageEvidence.transferred_from_card_id` audit trail +records which sibling it came from. `Card.md5_checksum`/`sha256_checksum` +are ALSO stamped at real extraction time now (`compute_card_evidence`'s own +`md5_checksum`/`sha256_checksum` parameters) — every `persist_evidence` call +(a REAL extraction, never a transfer — the two writers are disjoint) +unconditionally resets `transferred=False`, so a row that starts as a +transfer and later gets a genuine re-extraction is no longer flagged as one. + +**Fetch fallback — explicitly DEFERRED, not built in this PR.** Issue +#473's own PR-2 scope text also named a fetch fallback ("on 404/lockout, try +an md5 sibling's source URL — same bytes"). Not built here — evidence +TRANSFER (this section) already covers the dominant win (skip the fetch +entirely when a sibling's evidence already exists); the narrower fetch-level +fallback is left for a follow-up rather than widening this PR's own scope. + +**Staleness fix (#473 PR-2).** Every "is this `ImageEvidence` row CURRENT +for this card" check across the codebase (`image_evidence. current_evidence_queryset`, the single shared helper — previously N +independent inline copies in `local_calculate_verdicts.py`'s three +calculators, `local_layout_class_cast.py`, `local_detect_ai_art.py`, +`local_lands_identify.py`, `reparse_collector_evidence.py`, plus the two +bulk-query readers `modern_artist_credit.py`/`review_clusters.py`) now +ADDITIONALLY requires the row's own stamped `md5_checksum` to agree with +`Card.md5_checksum` whenever BOTH are non-null — closes a silent +in-place-file-replacement hole a content_phash-only check could miss (a +source file replaced at the same Drive location moves the Drive +`md5Checksum` without necessarily producing a different perceptual hash). +NULL-TOLERANT BY DESIGN: a legacy row predating the stamp, or a card whose +source never carries an md5 at all (e.g. `LOCAL_FILE`), stays current under +the content-hash check alone — only a row that stamped a REAL, actively +DISAGREEING md5 is treated as stale. + +**INTERIM Stage D guard (#473 PR-2, temporary by design).** A card whose +CURRENT evidence row was created by transfer is excluded from all three +Stage D calculators (`TRANSFERRED_INTERIM_GUARD_SKIP_REASON`, +rescannable) — its own machine "observation" is the same bytes a sibling +card already voted from, not an independent one, so casting a vote here +would fabricate the independence the vote-weight matrix assumes is real. +The slow-path calculator is deliberately NOT guarded — it casts no machine +vote, only a `CardScanLog` routing marker to a HUMAN reviewer, which is +exactly the safety net the guard exists to preserve. **Removal is PR-3's +own business** (issue #473's build plan: group-level vote pooling correctly +dedupes a transferred row's vote at the GROUP level instead of excluding it +outright) — do not remove the guard, or the `ImageEvidence.transferred` +flag it reads, before that PR merges. + +**Decoupled fetch-ahead (#472).** `stage_e_streaming.md` §4 item 3 ratified +"adopt, unconditionally" — the streaming conveyor's compute stage should +overlap fetch with compute from the start — but Phase 2 shipped fully +sequential anyway. Fixed by retrofitting the SAME decoupled-fetch +architecture `run_image_evidence_cohort.py` already uses (PRs #228/#237, +measured 1.44-1.52x BULK-mode gain) into `_run_stage_c`: ONE fetch-ahead +thread (never pooled — the ratified brief's own "no compute pooling" bar +applies equally to a second fetch worker, which would only race further +ahead of a compute loop that's already the slower stage) writes into a +bounded (`maxsize=2`) `queue.Queue`; this function's own COMPUTE loop stays +exactly as sequential as before, just no longer blocked on the NEXT card's +fetch while extracting the CURRENT one. A single serial fetch worker's own +completion order IS its submission order, so the fetch-outcome window +(`_window`) records outcomes in the same order it always did — no +reordering risk from the added concurrency. `GoogleFetchLockoutError` still +halts NEW fetches immediately (the thread stops after reporting the +lockout outcome); already-fetched-but-not-yet-computed cards already +sitting in the queue still drain to compute first ("in-flight work +drains, nothing NEW starts") — FIFO ordering makes this automatic. A +non-lockout exception during fetch (a real bug, a kill-drill fault) is +caught in the fetch-ahead thread and re-raised in the MAIN thread by the +compute loop the instant it's observed — a bare `try/except GoogleFetchLockoutError` there would have let an uncaught exception in the +spawned thread silently kill it, leaving the compute loop's own +`queue.get()` blocked forever waiting for an outcome that would never +arrive (found and fixed during this change's own review, pinned by +`test_stage_e_dispatch.py`'s `TestDecoupledFetchAhead:: test_a_non_lockout_fetch_crash_propagates_instead_of_hanging`) — the +kill-safety resume contract's own "a mid-batch crash leaves a truthful +FAILED ledger row" property must hold for a fetch-time crash exactly as it +already did for a compute-time one. + +**Echo suppression (#472's own fold).** See "Evidence-change echo" below +(Phase 3 section) — `cardpicker.stage_e_signals.suppress_evidence_change_echo` +wraps every `ImageEvidence` write `_run_stage_c` performs (both the +transfer write and the real-extraction `persist_evidence` call), so a write +made BY the dispatch loop itself never queues a fresh echo dispatch for the +same card (Stage D, called next over the same batch, already covers it). +BULK-mode writes (`run_image_evidence_cohort.py`) are unflagged and keep +firing the echo exactly as before this change. + ### Trigger: event-driven, plus a cron backstop (§3 decision (1)) - **Event-driven** (`cardpicker/stage_e_signals.py`, wired in @@ -745,7 +880,8 @@ scarce resource a kill-and-resume must never burn twice. **Corrected 2026-07-25 per the §8 Tron pass on PR #467** — an earlier version of this note characterized the echo as a uniformly "fast, cheap -no-op." That is wrong; the actual mechanism below is what Tron verified. +no-op." That is wrong; the mechanism below (now SUPPRESSED, see the closing +subsection) is what Tron verified. Every `persist_evidence` write this driver's forced re-extraction performs is an ordinary `ImageEvidence` save, so `cardpicker.stage_e_signals`'s own @@ -780,20 +916,26 @@ card, it is a complete micro-batch: driver (`"throttled-concurrency-cap"`) well before the cohort is exhausted. -**Acceptable at bounded-pilot scale, still not suppressed here** (frozen at -filing) — the two are distinguishable in the ledger by `trigger_reason`: -this driver's own batches carry `"shakedown"`, an echo dispatch carries +The two are distinguishable in the ledger by `trigger_reason`: this +driver's own batches carry `"shakedown"`, an echo dispatch carries `"evidence-change"`, so the ledger itself shows whether echoes are staying cheap (batch size stays at 1) or cascading (batch size climbs toward `STAGE_E_MICRO_BATCH_SIZE`). -**Tron's condition (§8 pass on PR #467):** the documented (not built) -fallback — a suppress-signals flag on `persist_evidence` — becomes -**REQUIRED, not optional,** before scaling beyond a bounded pilot, if -either (a) throttle-stops dominate the driver's own ledger output, or (b) -the Stage C backlog is measured non-zero at run time (check before -invoking). Do not build the fallback preemptively outside those -conditions. +**Tron's condition, RESOLVED 2026-07-25 (issue #472, folded with #473 +PR-2):** the "documented (not built) fallback" this section used to flag as +becoming REQUIRED before scaling beyond a bounded pilot is now BUILT — +`cardpicker.stage_e_signals.suppress_evidence_change_echo` wraps every +`ImageEvidence` write `stage_e_dispatch._run_stage_c` performs, and this +driver's own forced re-extraction runs entirely through that same function +(`dispatch_micro_batch` → `_run_stage_c`) — so this driver's own +`persist_evidence` writes are suppressed automatically, with no separate +opt-in. The cascade risk this whole section describes no longer applies to +THIS driver's own invocations; it remains a real risk only for a write path +that reaches `ImageEvidence.save()` from OUTSIDE the dispatch loop (BULK +mode's own `run_image_evidence_cohort.py`, which is unaffected by design — +see the Phase 2 "Evidence transfer and decoupled fetch-ahead" section's own +"Echo suppression" paragraph). ### Ledger convention