Proof: a dispatch admits the evidence its lanes can read, not the name (V5b2d-4g)

proof/region/v1/corpus_dispatch.py

@@ -31,6 +31,8 @@
 import re
 import subprocess
 import sys
+import tempfile
+from collections.abc import Iterable
 from dataclasses import dataclass
 from pathlib import Path
 
@@ -51,6 +53,17 @@
     "verification-evidence-arb",
     "verification-evidence-mpfi",
 )
+# What every lane reads out of the artifact it downloads, relative to the
+# artifact root.  This is a contract with `verification-lanes.yml`, which
+# guards exactly these paths before it replays anything; a test binds the
+# two together in both directions so they cannot drift apart.
+EVIDENCE_LANE_INPUTS_V1 = (
+    "job.bin",
+    "comparator-bundle/comparator-manifest-v2.bin",
+)
+# One download of one artifact: generous for a slow network, far short of a
+# wait an operator would mistake for work in progress.
+OBSERVATION_TIMEOUT_SECONDS_V1 = 120
 # An artifact name proves nothing about where the artifact came from.  Only a
 # successful operator-triggered run of the producer workflow may be replayed:
 # the path pins which workflow built the bundle, and the trigger pins whose
@@ -88,8 +101,9 @@
 _SUMMARY_WINDOWS_V1 = 8
 FULL_DOMAIN = protocol.OUTPUT_CARDINALITY_V1
 ALIGNMENT = corpus.CORPUS_SHARD_ALIGNMENT_V1
-# One listing of one run: generous for a slow network, far short of a wait an
-# operator would mistake for work in progress.
+# One observation — a listing, a projection or a two-small-file download:
+# generous for a slow network, far short of a wait an operator would mistake
+# for work in progress.
 OBSERVATION_TIMEOUT_SECONDS_V1 = 60
 
 
@@ -468,8 +482,7 @@ def admit_evidence_run_v1(
     return None
 
 
-def admit_evidence_artifact_v1(
-    evidence_run_id: int,
+def admit_evidence_artifact_v1(    evidence_run_id: int,
     evidence_artifact: str,
     observer: object | None = None,
 ) -> corpus.ShardCorpusRejectedV1 | None:
@@ -479,8 +492,7 @@ def admit_evidence_artifact_v1(
     that does not exist, or carries no such artifact, turns one mistake into
     256 doomed jobs.  A positive integer is not evidence that a run exists,
     so the coordinator asks before it dispatches, and any failure to observe
-    is itself a refusal — never a crash and never a silent proceed.
-    """
+    is itself a refusal — never a crash and never a silent proceed.    """
 
     # Resolved here, not captured as a default: a default binds the module
     # attribute at definition time, which makes the injection point real for
@@ -929,6 +941,113 @@ def collect_lane_runs_v1(
         )
     return matched
 
+
+def missing_lane_inputs_v1(observed_paths: Iterable[str]) -> tuple[str, ...]:
+    """Which lane inputs a downloaded evidence artifact does not carry.
+
+    The rule of this admission, kept pure and off the network so a test can
+    reach it.  Paths match exactly, in the declared order: the lane runs
+    `test -f` on the exact path, so a suffixed sibling, the directory above
+    it, a backslash-joined spelling or a different case is absent to the
+    lane and must read absent here.  Entries that are not strings cannot be
+    a path the lane found, so they never satisfy a requirement.
+    """
+
+    present = frozenset(path for path in observed_paths if type(path) is str)
+    return tuple(
+        required
+        for required in EVIDENCE_LANE_INPUTS_V1
+        if required not in present
+    )
+
+
+def gh_evidence_artifact_paths_v1(run_id: int, artifact: str) -> tuple[str, ...]:
+    """Every file the named artifact of that run carries, relative to its root.
+
+    The one impure boundary of this admission: it downloads exactly what the
+    lanes will download, into a temporary directory it owns and always
+    removes, and reports what arrived without deciding anything.  Which of
+    those paths count is a rule, so it lives in `missing_lane_inputs_v1` —
+    filtering here would put the rule behind the network.  The evidence is
+    two files well under a megabyte, so this one download costs less than a
+    single doomed lane, let alone 256 of them.
+    """
+
+    with tempfile.TemporaryDirectory(prefix="verification-evidence-") as tmp:
+        root = Path(tmp)
+        subprocess.run(
+            (
+                "gh",
+                "run",
+                "download",
+                str(run_id),
+                "--name",
+                artifact,
+                "--dir",
+                str(root),
+            ),
+            capture_output=True,
+            text=True,
+            check=True,
+            # A hung download is worse than a refused one: without a deadline
+            # the coordinator waits forever on the last check standing between
+            # an operator and 256 dispatches.
+            timeout=OBSERVATION_TIMEOUT_SECONDS_V1,
+        )
+        return tuple(
+            sorted(
+                path.relative_to(root).as_posix()
+                for path in root.rglob("*")
+                if path.is_file()
+            )
+        )
+
+
+def admit_evidence_artifact_content_v1(    evidence_run_id: int,
+    evidence_artifact: str,
+    observer: object | None = None,
+) -> corpus.ShardCorpusRejectedV1 | None:
+    """Refuse a dispatch whose evidence artifact lacks what the lanes read.
+
+    A name is not a layout.  An evidence run produced before the exporter
+    settled on today's `evidence-out/` carries the right artifact name and
+    the wrong contents: it passes every name check and then dies in all 256
+    lanes, seconds apart, on the first `test -f`.  So the coordinator reads
+    the artifact once before it dispatches, and admits it only when every
+    lane input is there.  Any failure to observe is itself a refusal —
+    never a crash, never a silent proceed.  The run id and the artifact
+    name are admitted upstream; this admission speaks about content only.    """
+
+    # Resolved here, not captured as a default: a default binds the module
+    # attribute at definition time, which makes the injection point real for
+    # a caller but invisible to anything that replaces the observer — the
+    # seam would look injectable and not be.
+    if observer is None:
+        observer = gh_evidence_artifact_paths_v1
+    try:
+        observed = tuple(observer(evidence_run_id, evidence_artifact))  # type: ignore[operator]
+    except Exception as error:
+        # The download is a hostile boundary: a deleted run, an expired
+        # artifact, a missing token and a broken network must all land as
+        # one refusal.  The cause travels with it — an operator has to tell
+        # "no such run" from "no token", and a bare refusal makes those look
+        # identical.  `stderr` is preferred over `repr` because it is the text
+        # itself rather than the text wrapped in a constructor call; `repr`
+        # does carry it, so this is legibility, not recovery.
+        cause = getattr(error, "stderr", None) or repr(error)
+        return corpus._reject(
+            corpus.ShardCorpusReasonV1.FOREIGN_INPUT,
+            f"verification dispatch cannot read artifact {evidence_artifact!r}"
+            f" of run {evidence_run_id}: {str(cause).strip()}",
+        )
+    missing = missing_lane_inputs_v1(observed)
+    if missing:
+        return corpus._reject(
+            corpus.ShardCorpusReasonV1.FOREIGN_INPUT,
+            f"artifact {evidence_artifact!r} of run {evidence_run_id} carries"
+            f" no {', '.join(missing)}",        )
+    return None
+
 def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument(
@@ -1061,10 +1180,11 @@ def main(argv: list[str] | None = None) -> int:
     else:
         commands = dispatch_commands_v1(plan, args.shard_width)
     if type(commands) is not tuple:
-        # A rejected command set is a typed refusal, not an iterable: letting
-        # it reach the loop below turns the module's own contract into a
-        # TypeError at the boundary it is supposed to guard.
-        print(f"lane dispatch rejected: {commands!r}", file=sys.stderr)
+        # A rejected command set is a typed refusal, not an iterable.  Letting
+        # it reach the loop turns this module's own contract into a TypeError
+        # at the boundary it exists to guard — and a mistyped artifact name is
+        # the likeliest way an operator gets here.
+        print(f"lane dispatch rejected: {commands.detail}", file=sys.stderr)
         return 64
     if not dispatching:
         for command in commands:
@@ -1106,6 +1226,15 @@ def main(argv: list[str] | None = None) -> int:
         if refusal is not None:
             print(f"verification dispatch refused: {refusal.detail}", file=sys.stderr)
             return 64
+        # Contents last: this one downloads, and the free refusals above must
+        # never cost a fetch.  A name is not a layout — an older producer's
+        # run passes every check before this and dies in all 256 lanes.
+        refusal = admit_evidence_artifact_content_v1(
+            args.evidence_run_id, args.evidence_artifact
+        )
+        if refusal is not None:
+            print(f"verification dispatch refused: {refusal.detail}", file=sys.stderr)
+            return 64
     print(f"dispatching {len(commands)} lanes", file=sys.stderr)
     launched = 0
     for command in commands:

proof/region/v1/tests/test_corpus_dispatch.py

@@ -734,15 +734,34 @@ class DispatchSeamTests(unittest.TestCase):
     """
 
     def _provenance_patch(self, observer: object = None) -> object:
-        """Patch the run observation the live path performs before anything else."""
-
-        return unittest.mock.patch.object(
-            corpus_dispatch,
-            "gh_run_provenance_v1",
-            observer
-            if observer is not None
-            else (lambda run_id: _admitted_provenance()),
-        )
+        """Patch the two observations the live path performs before anything else.
+
+        The seam under test is the coordinator's, not the network's: an
+        admitted run also answers the content admission with exactly the
+        inputs a lane reads, so a seam test fails on its own claim rather
+        than on an unmocked download.
+        """
+
+        import contextlib as _ctx
+
+        @_ctx.contextmanager
+        def both():
+            with unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_provenance_v1",
+                observer
+                if observer is not None
+                else (lambda run_id: _admitted_provenance()),
+            ), unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_evidence_artifact_paths_v1",
+                lambda run_id, artifact: tuple(
+                    corpus_dispatch.EVIDENCE_LANE_INPUTS_V1
+                ),
+            ):
+                yield
+
+        return both()
 
 
     def _argv(self, out: Path, *, live: bool = True) -> list[str]: