From 0418b413539615accdf683aa2027f9c4f8553e41 Mon Sep 17 00:00:00 2001 From: Claude Code Date: Fri, 7 Aug 2026 00:13:04 +0300 Subject: [PATCH 1/2] Proof: a dispatch admits the evidence its lanes can read, not the name (V5b2d-4g) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The verification dispatch binds an evidence run and an exact artifact name, and a name is not a layout. An evidence run built by an older producer, with a different evidence-out/ shape, carries the right artifact name and the wrong contents: it passes admission, fans out to 256 lanes, and every one of them dies seconds in on `test -f`. That is the incident of the 133 wasted runs again, on its third axis. So the admission reads what it names. The coordinator downloads the bound artifact once — two files well under a megabyte — and dispatches only when it carries every path the lane workflow guards. The module's shape is kept: the download is the one impure boundary and decides nothing; which paths must be there is a pure rule a test can reach without a network; a missing input and any failure to observe are the same typed refusal, never a crash and never a silent proceed. The temporary directory is owned by the download and removed on both paths. A dry run stays offline by contract. The path list is a contract with verification-lanes.yml, so it is not restated by hand: tests read it back out of the workflow text in both directions — every guarded path must be admitted, and every evidence path the lane uses must be covered — so the two cannot drift apart silently. --- proof/region/v1/corpus_dispatch.py | 127 ++++++ .../v1/tests/test_verification_dispatch.py | 383 ++++++++++++++++++ 2 files changed, 510 insertions(+) diff --git a/proof/region/v1/corpus_dispatch.py b/proof/region/v1/corpus_dispatch.py index 2326569f..af0021c0 100644 --- a/proof/region/v1/corpus_dispatch.py +++ b/proof/region/v1/corpus_dispatch.py @@ -17,6 +17,8 @@ import json import subprocess import sys +import tempfile +from collections.abc import Iterable from pathlib import Path PROOF = Path(__file__).resolve().parent @@ -36,6 +38,14 @@ "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", +) DEFAULT_LANE_WIDTH = 1 << 16 DEFAULT_SHARD_WIDTH = corpus_lane.DEFAULT_SHARD_POINTS FULL_DOMAIN = protocol.OUTPUT_CARDINALITY_V1 @@ -188,6 +198,111 @@ def verification_dispatch_commands_v1( ) +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, + ) + 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. CalledProcessError.__repr__ drops stderr, and stderr + # is where the only distinguishing text lives. + 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( @@ -242,6 +357,18 @@ def main(argv: list[str] | None = None) -> int: args.evidence_run_id, args.evidence_artifact, ) + if type(commands) is tuple and not args.dry_run: + # Fail closed before the first dispatch: a dry run stays offline + # by contract, so the download belongs to the live path only. + 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 else: commands = dispatch_commands_v1(plan, args.shard_width) for command in commands: diff --git a/proof/region/v1/tests/test_verification_dispatch.py b/proof/region/v1/tests/test_verification_dispatch.py index 8cc7dfc6..296ffe29 100644 --- a/proof/region/v1/tests/test_verification_dispatch.py +++ b/proof/region/v1/tests/test_verification_dispatch.py @@ -15,16 +15,27 @@ rejections before any dispatch exists. Plans that cannot cover the domain exactly and evidence run ids that are not positive integers are likewise typed rejections. + +A name is not a layout. An evidence run built by an older producer carries +the right artifact name and the wrong `evidence-out/` contents, passes every +check above, and then dies in all 256 lanes on the first `test -f`. So the +admission also reads the artifact: one download of two small files decides +what 256 lanes would otherwise discover one at a time. The list of paths a +lane needs is a contract with `verification-lanes.yml`, so it is bound to the +workflow text here rather than restated by hand. """ from __future__ import annotations import io +import re +import subprocess import sys import tempfile import unittest from contextlib import redirect_stdout from pathlib import Path +from unittest import mock PROOF = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROOF)) @@ -35,6 +46,13 @@ REPO = PROOF.parents[2] ARB_ARTIFACT = "verification-evidence-arb" MPFI_ARTIFACT = "verification-evidence-mpfi" +VERIFICATION_WORKFLOW = ( + REPO / ".github" / "workflows" / "verification-lanes.yml" +) +# `test -f "${evidence}/"` — the lane's own statement of what it needs. +LANE_INPUT_GUARD_V1 = re.compile(r'test -f "\$\{evidence\}/([^"]+)"') +# Every use of the downloaded artifact root, guard or runner argument alike. +LANE_EVIDENCE_USE_V1 = re.compile(r'\$\{evidence\}/([^"\s]+)') class EvidenceArtifactAllowlistTests(unittest.TestCase): @@ -247,5 +265,370 @@ def test_producer_workflow_publishes_exactly_the_allowlisted_names(self) -> None self.assertIn(artifact, text) +class LaneInputContractTests(unittest.TestCase): + """The admitted paths are the lane's requirements, not a private guess. + + The coordinator refuses an evidence artifact that lacks what the lane + reads. If that list and the workflow drift apart, the admission either + passes runs the lane cannot use — the incident it exists to prevent — or + refuses runs the lane could. So the list is read back out of the + workflow text, in both directions. + """ + + def test_the_declared_lane_inputs_are_exactly_the_workflow_guards(self) -> None: + text = VERIFICATION_WORKFLOW.read_text(encoding="utf-8") + guarded = tuple(LANE_INPUT_GUARD_V1.findall(text)) + self.assertTrue( + guarded, "the lane guards no evidence file: the contract is unreadable" + ) + self.assertEqual( + len(set(guarded)), len(guarded), "the lane guards a path twice" + ) + self.assertEqual( + sorted(guarded), sorted(corpus_dispatch.EVIDENCE_LANE_INPUTS_V1) + ) + + def test_every_evidence_path_the_lane_uses_is_admitted(self) -> None: + # A guard can be forgotten; a runner argument cannot. Every path the + # lane reads out of the artifact must be a declared lane input or a + # directory on the way to one, or the admission has a blind spot. + text = VERIFICATION_WORKFLOW.read_text(encoding="utf-8") + used = tuple(LANE_EVIDENCE_USE_V1.findall(text)) + self.assertTrue(used, "the lane reads nothing out of the artifact") + for path in used: + covered = path in corpus_dispatch.EVIDENCE_LANE_INPUTS_V1 or any( + required.startswith(f"{path}/") + for required in corpus_dispatch.EVIDENCE_LANE_INPUTS_V1 + ) + self.assertTrue( + covered, + f"the lane reads {path!r} but the dispatch does not admit it", + ) + + +class LaneInputRuleTests(unittest.TestCase): + """The rule: which lane inputs a downloaded artifact does not carry.""" + + def _complete(self) -> tuple[str, ...]: + # The rest of the exported evidence: bundle contents addressed by + # hex, and the engine's sealed run coordinates. The lane does not + # read them, so they neither satisfy nor block an admission. + return corpus_dispatch.EVIDENCE_LANE_INPUTS_V1 + ( + "comparator-bundle/content/9f86d081884c7d65", + "transcript.bin", + "run-claim.bin", + ) + + def test_a_complete_artifact_is_missing_nothing(self) -> None: + self.assertEqual( + corpus_dispatch.missing_lane_inputs_v1(self._complete()), () + ) + + def test_each_lane_input_is_reported_when_it_is_the_one_absent(self) -> None: + for required in corpus_dispatch.EVIDENCE_LANE_INPUTS_V1: + observed = tuple( + path for path in self._complete() if path != required + ) + self.assertEqual( + corpus_dispatch.missing_lane_inputs_v1(observed), + (required,), + f"{required!r} was not reported as absent", + ) + + def test_an_empty_artifact_misses_every_lane_input_in_declared_order(self) -> None: + self.assertEqual( + corpus_dispatch.missing_lane_inputs_v1(()), + corpus_dispatch.EVIDENCE_LANE_INPUTS_V1, + ) + + def test_a_near_miss_does_not_satisfy_a_lane_input(self) -> None: + # The lane runs `test -f` on the exact path. Anything that is not + # that path — a suffixed sibling, the directory above it, a + # backslash-joined or trailing-slash spelling, a different case — is + # absent as far as the lane is concerned, and must read absent here. + for near in ( + "job.bin.txt", + "evidence-out/job.bin", + "comparator-bundle", + "comparator-bundle/", + "comparator-bundle\\comparator-manifest-v2.bin", + "comparator-bundle/comparator-manifest-v2.BIN", + "comparator-bundle/comparator-manifest-v1.bin", + ): + self.assertEqual( + corpus_dispatch.missing_lane_inputs_v1((near,)), + corpus_dispatch.EVIDENCE_LANE_INPUTS_V1, + f"{near!r} was accepted for a lane input it is not", + ) + + def test_foreign_observed_entries_are_ignored_not_matched(self) -> None: + observed = ( + b"job.bin", + None, + 17, + Path("comparator-bundle/comparator-manifest-v2.bin"), + ) + self.assertEqual( + corpus_dispatch.missing_lane_inputs_v1(observed), + corpus_dispatch.EVIDENCE_LANE_INPUTS_V1, + ) + + +class EvidenceDownloadObserverTests(unittest.TestCase): + """The impure boundary: download the artifact, report it, leave no trace.""" + + def test_the_download_reports_its_files_and_removes_its_directory(self) -> None: + seen: dict[str, object] = {} + + def fake_run(argv, **kwargs): + argv = tuple(argv) + seen["argv"] = argv + directory = Path(argv[argv.index("--dir") + 1]) + seen["dir"] = directory + (directory / "job.bin").write_bytes(b"\x00") + bundle = directory / "comparator-bundle" + (bundle / "content").mkdir(parents=True) + (bundle / "comparator-manifest-v2.bin").write_bytes(b"\x01") + (bundle / "content" / "9f86d081884c7d65").write_bytes(b"\x02") + return subprocess.CompletedProcess(argv, 0, "", "") + + with mock.patch.object(corpus_dispatch.subprocess, "run", fake_run): + observed = corpus_dispatch.gh_evidence_artifact_paths_v1( + 31000000001, ARB_ARTIFACT + ) + + # Nested files are reported by their path under the artifact root, + # which is the shape the rule matches against. + self.assertEqual( + observed, + ( + "comparator-bundle/comparator-manifest-v2.bin", + "comparator-bundle/content/9f86d081884c7d65", + "job.bin", + ), + ) + argv = seen["argv"] + self.assertEqual(argv[:4], ("gh", "run", "download", "31000000001")) + self.assertEqual(argv[argv.index("--name") + 1], ARB_ARTIFACT) + self.assertFalse( + Path(seen["dir"]).exists(), + "the download directory outlived the observation", + ) + + def test_a_failed_download_still_removes_its_directory(self) -> None: + seen: dict[str, object] = {} + + def fake_run(argv, **kwargs): + argv = tuple(argv) + directory = Path(argv[argv.index("--dir") + 1]) + seen["dir"] = directory + (directory / "partial.bin").write_bytes(b"\x00") + raise subprocess.CalledProcessError(1, argv, "", "no artifact matches") + + with mock.patch.object(corpus_dispatch.subprocess, "run", fake_run): + with self.assertRaises(subprocess.CalledProcessError): + corpus_dispatch.gh_evidence_artifact_paths_v1(1, MPFI_ARTIFACT) + + self.assertFalse( + Path(seen["dir"]).exists(), + "a failed download left its directory behind", + ) + + +class EvidenceContentAdmissionTests(unittest.TestCase): + """The admission: refuse an artifact the lanes could not use.""" + + def _complete(self) -> tuple[str, ...]: + return corpus_dispatch.EVIDENCE_LANE_INPUTS_V1 + ("run-claim.bin",) + + def test_a_complete_artifact_is_admitted_after_one_observation(self) -> None: + calls: list[tuple[object, ...]] = [] + + def observer(run_id, artifact): + calls.append((run_id, artifact)) + return self._complete() + + self.assertIsNone( + corpus_dispatch.admit_evidence_artifact_content_v1( + 31000000001, ARB_ARTIFACT, observer + ) + ) + self.assertEqual(calls, [(31000000001, ARB_ARTIFACT)]) + + def test_a_missing_lane_input_is_a_typed_refusal_that_names_it(self) -> None: + for required in corpus_dispatch.EVIDENCE_LANE_INPUTS_V1: + observed = tuple( + path for path in self._complete() if path != required + ) + refusal = corpus_dispatch.admit_evidence_artifact_content_v1( + 31000000001, MPFI_ARTIFACT, lambda *_: observed + ) + self.assertIs(type(refusal), corpus.ShardCorpusRejectedV1) + self.assertEqual( + refusal.reason, corpus.ShardCorpusReasonV1.FOREIGN_INPUT + ) + self.assertIn(required, refusal.detail) + self.assertIn("31000000001", refusal.detail) + self.assertIn(MPFI_ARTIFACT, refusal.detail) + + def test_an_unobservable_run_is_a_refusal_carrying_the_cause(self) -> None: + def observer(run_id, artifact): + raise subprocess.CalledProcessError( + 1, ("gh",), "", "HTTP 404: Not Found (run 31000000001)" + ) + + refusal = corpus_dispatch.admit_evidence_artifact_content_v1( + 31000000001, ARB_ARTIFACT, observer + ) + self.assertIs(type(refusal), corpus.ShardCorpusRejectedV1) + self.assertEqual( + refusal.reason, corpus.ShardCorpusReasonV1.FOREIGN_INPUT + ) + # An operator must be able to tell "no such run" from "no token"; + # a bare refusal makes those look identical. + self.assertIn("HTTP 404: Not Found", refusal.detail) + + def test_any_failure_to_observe_is_a_refusal_not_a_crash(self) -> None: + def raiser(error): + def observer(run_id, artifact): + raise error + + return observer + + for observer in ( + raiser(RuntimeError("gh: command not found")), + raiser(OSError(13, "Permission denied")), + raiser(TimeoutError("download timed out")), + lambda run_id, artifact: 17, + lambda run_id, artifact: None, + ): + refusal = corpus_dispatch.admit_evidence_artifact_content_v1( + 31000000001, ARB_ARTIFACT, observer + ) + self.assertIs( + type(refusal), + corpus.ShardCorpusRejectedV1, + f"{observer!r} did not land as a refusal", + ) + self.assertEqual( + refusal.reason, corpus.ShardCorpusReasonV1.FOREIGN_INPUT + ) + + def test_the_default_observer_is_resolved_at_call_time(self) -> None: + # A default argument would bind the module attribute at definition + # time: the seam would look injectable and not be. + calls: list[tuple[object, ...]] = [] + + def observer(run_id, artifact): + calls.append((run_id, artifact)) + return corpus_dispatch.EVIDENCE_LANE_INPUTS_V1 + + with mock.patch.object( + corpus_dispatch, "gh_evidence_artifact_paths_v1", observer + ): + self.assertIsNone( + corpus_dispatch.admit_evidence_artifact_content_v1( + 31000000001, ARB_ARTIFACT + ) + ) + self.assertEqual(calls, [(31000000001, ARB_ARTIFACT)]) + + +class VerificationDispatchContentAdmissionCliTests(unittest.TestCase): + """One download decides; 256 lanes never start on evidence they cannot read.""" + + def _dispatch(self, argv: list[str]) -> tuple[int, list[tuple[str, ...]]]: + dispatched: list[tuple[str, ...]] = [] + + def fake_run(command, **kwargs): + dispatched.append(tuple(command)) + return subprocess.CompletedProcess(tuple(command), 0) + + with mock.patch.object(corpus_dispatch.subprocess, "run", fake_run): + with redirect_stdout(io.StringIO()): + exit_code = corpus_dispatch.main(argv) + return exit_code, dispatched + + def test_the_live_path_refuses_before_the_first_dispatch(self) -> None: + observed = tuple( + path + for path in corpus_dispatch.EVIDENCE_LANE_INPUTS_V1 + if path != "job.bin" + ) + with tempfile.TemporaryDirectory() as tmp: + with mock.patch.object( + corpus_dispatch, + "gh_evidence_artifact_paths_v1", + lambda *_: observed, + ): + exit_code, dispatched = self._dispatch( + [ + "--mode", + "verification-dispatch", + "--evidence-run-id", + "31000000001", + "--evidence-artifact", + ARB_ARTIFACT, + "--out", + str(Path(tmp) / "out"), + ] + ) + self.assertEqual(exit_code, 64) + self.assertEqual(dispatched, []) + + def test_all_256_lanes_dispatch_after_one_admitted_download(self) -> None: + calls: list[tuple[object, ...]] = [] + + def observer(run_id, artifact): + calls.append((run_id, artifact)) + return corpus_dispatch.EVIDENCE_LANE_INPUTS_V1 + + with tempfile.TemporaryDirectory() as tmp: + with mock.patch.object( + corpus_dispatch, "gh_evidence_artifact_paths_v1", observer + ): + exit_code, dispatched = self._dispatch( + [ + "--mode", + "verification-dispatch", + "--evidence-run-id", + "31000000001", + "--evidence-artifact", + MPFI_ARTIFACT, + "--out", + str(Path(tmp) / "out"), + ] + ) + self.assertEqual(exit_code, 0) + self.assertEqual(len(dispatched), 256) + self.assertEqual(calls, [(31000000001, MPFI_ARTIFACT)]) + + def test_a_dry_run_stays_offline(self) -> None: + def observer(run_id, artifact): + raise AssertionError("a dry run downloaded the evidence") + + with tempfile.TemporaryDirectory() as tmp: + with mock.patch.object( + corpus_dispatch, "gh_evidence_artifact_paths_v1", observer + ): + exit_code, dispatched = self._dispatch( + [ + "--mode", + "verification-dispatch", + "--lane-width", + str(1 << 23), + "--evidence-run-id", + "31000000001", + "--evidence-artifact", + ARB_ARTIFACT, + "--dry-run", + "--out", + str(Path(tmp) / "out"), + ] + ) + self.assertEqual(exit_code, 0) + self.assertEqual(dispatched, []) + + if __name__ == "__main__": unittest.main() From 9f1d7e3ddbfb5c312ddaf1ed8aed905b5e6811fd Mon Sep 17 00:00:00 2001 From: Claude Code Date: Fri, 7 Aug 2026 08:57:15 +0300 Subject: [PATCH 2/2] Proof: keep the promise this slice wrote down MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Final-state review anchored on the commit object — after catching its own first mutation pass reading a stale file through WSL — ran 26 mutants and found two blockers plus one theatre test. THE CRASH. A refused command set is not iterable, and `main` fell straight into the loop with it: a mistyped artifact name — the likeliest operator slip — crashed with a TypeError while this very commit's docstring promises never a crash. The same fallthrough was fixed on the #548 branch, but this branch grows from main, where it still lived. Typed exit now, and the CLI test covers three foreign coordinates while also proving the expensive download never ran for any of them — a coordinate the pure builder already refused must not cost a network call. THE FALSE CLAIM. The comment said `CalledProcessError.__repr__` drops stderr. Measured: it does not — `BaseException` keeps every constructor argument in `args`. The preference for `stderr` over `repr` is legibility, not recovery, and the comment now says so instead of inventing a stronger reason. THE THEATRE. The foreign-entries test fed bytes, None and an int to the type guard — inputs that can never equal a string, so the guard was semantically free and the test held without the code it pointed at. An unhashable list is what makes the guard load-bearing: without it the observation raises instead of refusing. The test feeds one now, and stripping the guard fails it. Also: the download carries a deadline. A hung `gh` stalled the coordinator forever on the last check standing between an operator and 256 dispatches, indistinguishable from work in progress. Each repair proven by its mutant: the restored fallthrough reddens the new CLI test, the stripped guard reddens the strengthened one. Verified: 31 dispatch tests pass; 426 on Linux with the three pre-existing `test_build` failures that also fail on an untouched tree. --- proof/region/v1/corpus_dispatch.py | 19 ++++++++- .../v1/tests/test_verification_dispatch.py | 42 +++++++++++++++++++ 2 files changed, 59 insertions(+), 2 deletions(-) diff --git a/proof/region/v1/corpus_dispatch.py b/proof/region/v1/corpus_dispatch.py index af0021c0..61974276 100644 --- a/proof/region/v1/corpus_dispatch.py +++ b/proof/region/v1/corpus_dispatch.py @@ -46,6 +46,9 @@ "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 DEFAULT_LANE_WIDTH = 1 << 16 DEFAULT_SHARD_WIDTH = corpus_lane.DEFAULT_SHARD_POINTS FULL_DOMAIN = protocol.OUTPUT_CARDINALITY_V1 @@ -245,6 +248,10 @@ def gh_evidence_artifact_paths_v1(run_id: int, artifact: str) -> tuple[str, ...] 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( @@ -285,8 +292,9 @@ def admit_evidence_artifact_content_v1( # 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. CalledProcessError.__repr__ drops stderr, and stderr - # is where the only distinguishing text lives. + # 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, @@ -371,6 +379,13 @@ def main(argv: list[str] | None = None) -> int: return 64 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 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 for command in commands: rendered = " ".join(command) if args.dry_run: diff --git a/proof/region/v1/tests/test_verification_dispatch.py b/proof/region/v1/tests/test_verification_dispatch.py index 296ffe29..f413c794 100644 --- a/proof/region/v1/tests/test_verification_dispatch.py +++ b/proof/region/v1/tests/test_verification_dispatch.py @@ -362,10 +362,16 @@ def test_a_near_miss_does_not_satisfy_a_lane_input(self) -> None: ) def test_foreign_observed_entries_are_ignored_not_matched(self) -> None: + # The unhashable entry is the one that matters. A non-string simply + # never equals a string, so dropping the type guard would change + # nothing for `bytes` or `int` — the test would assert an invariant + # that holds without the code it points at. A list makes the guard + # load-bearing: without it the observation raises instead of refusing. observed = ( b"job.bin", None, 17, + ["job.bin"], Path("comparator-bundle/comparator-manifest-v2.bin"), ) self.assertEqual( @@ -576,6 +582,42 @@ def test_the_live_path_refuses_before_the_first_dispatch(self) -> None: self.assertEqual(exit_code, 64) self.assertEqual(dispatched, []) + def test_a_foreign_coordinate_exits_typed_instead_of_crashing(self) -> None: + # `verification_dispatch_commands_v1` returns a refusal, and a refusal + # is not iterable. A mistyped artifact name is the likeliest way an + # operator reaches this, and the docstring of the admission promises + # never a crash — so the promise has to be true here. + asked: list[object] = [] + for artifact, run_id in ( + ("verification-evidence-abr", "31000000001"), + (ARB_ARTIFACT, "0"), + (ARB_ARTIFACT, "-5"), + ): + with self.subTest(artifact=artifact, run_id=run_id): + with tempfile.TemporaryDirectory() as tmp: + with mock.patch.object( + corpus_dispatch, + "gh_evidence_artifact_paths_v1", + lambda *args: asked.append(args) or (), + ): + exit_code, dispatched = self._dispatch( + [ + "--mode", + "verification-dispatch", + "--evidence-run-id", + run_id, + "--evidence-artifact", + artifact, + "--out", + str(Path(tmp) / "out"), + ] + ) + self.assertEqual(exit_code, 64) + self.assertEqual(dispatched, []) + # And the expensive check never ran: a coordinate the pure builder + # already refused must not cost a download. + self.assertEqual(asked, []) + def test_all_256_lanes_dispatch_after_one_admitted_download(self) -> None: calls: list[tuple[object, ...]] = []