From 4da2ae7ede67c7f5dc67765c982ec69d64072f0e Mon Sep 17 00:00:00 2001 From: Claude Code Date: Thu, 6 Aug 2026 23:22:13 +0300 Subject: [PATCH 1/4] Proof: a dispatched campaign hands the dual proof its lane run ids (V5b2d-4e) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Наблюдаемый дефект: полное покрытие домена — 512 отдельных прогонов полос (256 на движок), и дуальному доказательству нужен список их идентификаторов. Собрать его нечем: `corpus_dispatch.py` печатает команды `gh workflow run`, а они id не возвращают; сбор по времени создания ломается ровно там, где он и нужен — два движка реплеят одни и те же окна одной evidence-сборки одновременно. Закон: прогон полосы назван своими координатами (`verification-lanes.yml` рендерит `lane + of `), поэтому список кампании — запрос по именам, а не догадка по часам. Кампанию отделяют артефакт и evidence run id, не время. Режим `--mode collect` устроен по границе, уже принятой в модуле: - `gh_lane_runs_v1` — единственное нечистое наблюдение. Проекция сделана самим `--jq` в запросе, поэтому в процесс входят ровно три поля (databaseId, displayTitle, conclusion) и ни одно непросмотренное поле ответа API не может попасть ни в решение, ни в лог. Запись, которая не проецируется точно, отвергается, а не угадывается. - `match_lane_runs_v1` — чистое сопоставление имён с планом. Все правила — чья кампания, чей движок, какое заключение считается, что такое полное покрытие — достижимы тестом без сети. - `collect_lane_runs_v1` — шов между ними: недоступный API, отсутствующий токен или нечитаемый ответ становятся типизированным отказом с причиной, а не пустой кампанией; пустая кампания и сломанный запрос ведут оператора к противоположным действиям. Неполный сбор — типизированный отказ `LaneRunCollectionRejectedV1`, а не тихий частичный список: окна без прогона и окна с двумя едут в нём данными (`missing`/`duplicated`), оператор передиспатчит ровно их, а CLI не пишет ничего и выходит 64. Частичный список неотличим от полного для того, кто прочитает его следующим. Парсер имени допускает ровно канонический рендеринг: `int()` принял бы `+7`, `007`, `1_0` и не-ASCII цифры, и каждое из них позволило бы чужому заголовку занять окно плана. Доказательства (WSL, python3 -m unittest): - RED до реализации: 17 ошибок «module 'corpus_dispatch' has no attribute parse_lane_run_name_v1 / match_lane_runs_v1 / collect_lane_runs_v1». - 19 новых тестов; каждый заявленный класс убивает мутацию: порядок плана → порядок наблюдения (M1); дыра терпится (M2); дубль терпится (M3); фильтр артефакта снят (M4); фильтр evidence run снят (M5); заключение игнорируется (M6); ordinals по `isdigit()` (M7); `except Exception` сужен до `OSError` на шве наблюдения (M8). Все восемь падают на финальном коде. - Контакт с реальностью: `gh run list --workflow verification-lanes.yml --json databaseId,displayTitle,conclusion --jq ...` на живом репозитории возвращает реальные записи; `--mode collect` против прогона 31104030757 отказывает с 256 missing и не пишет ничего (в `main` run-name ещё нет — он приходит с V5b2d-4d, до него ни один заголовок не является полосой кампании). - Набор proof: 427 тестов, 3 падения — ровно предсуществующие в test_build на Linux (два golden-digest и post_popen_handler_gap). Связность: режим полагается на `run-name` из `verification-lanes.yml` (V5b2d-4d). Без него сбор не выдумывает совпадений — он отказывает громко. --- proof/region/v1/corpus_dispatch.py | 464 ++++++++++++++++- proof/region/v1/tests/test_corpus_dispatch.py | 474 ++++++++++++++++++ 2 files changed, 937 insertions(+), 1 deletion(-) diff --git a/proof/region/v1/corpus_dispatch.py b/proof/region/v1/corpus_dispatch.py index 2326569f..ca38c5df 100644 --- a/proof/region/v1/corpus_dispatch.py +++ b/proof/region/v1/corpus_dispatch.py @@ -9,14 +9,22 @@ per lane of `full-domain-corpus.yml`. Widths that cannot produce an exact aligned cover are typed rejections before any dispatch exists; `--dry-run` prints every dispatch command without touching the network. + +`gh workflow run` returns no run id, so a dispatched campaign is only a set +of names. The collect mode closes that gap: `verification-lanes.yml` titles +every lane run after its own coordinates, so the campaign's run ids are a +query against those titles instead of a guess about creation times — which is +exactly what breaks when two campaigns overlap in time. """ from __future__ import annotations import argparse import json +import re import subprocess import sys +from dataclasses import dataclass from pathlib import Path PROOF = Path(__file__).resolve().parent @@ -38,6 +46,24 @@ ) DEFAULT_LANE_WIDTH = 1 << 16 DEFAULT_SHARD_WIDTH = corpus_lane.DEFAULT_SHARD_POINTS +LANE_RUN_COLLECTION_SCHEMA_V1 = "corpus-lane-runs-v1" +LANE_RUN_COLLECTION_NAME_V1 = "lane-runs.json" +# Only a successful lane produced the fragment the dual proof will replay, so +# a run in any other terminal state — and an unfinished one, whose conclusion +# is still empty — leaves its window uncovered. +LANE_RUN_SUCCESS_CONCLUSION_V1 = "success" +# A full-domain cover is 512 runs (256 per engine) before a single rerun, and +# a truncated listing can only hide runs, which surfaces as missing windows — +# a loud refusal, never a short list. +LANE_RUN_QUERY_LIMIT_V1 = 2000 +# The run-name renders ordinals through GitHub's expression syntax, which +# emits plain decimals. `int()` also accepts `+7`, `007`, `1_0` and non-ASCII +# digits, and each of those would let a foreign title claim a plan window, so +# the parser admits exactly the canonical rendering. +_CANONICAL_ORDINAL_V1 = re.compile(r"(?:0|[1-9][0-9]*)") +# A refusal's prose stays readable at 256 windows; the machine-readable set +# travels in the refusal's own fields, so nothing is lost by truncating here. +_SUMMARY_WINDOWS_V1 = 8 FULL_DOMAIN = protocol.OUTPUT_CARDINALITY_V1 ALIGNMENT = corpus.CORPUS_SHARD_ALIGNMENT_V1 @@ -188,17 +214,414 @@ def verification_dispatch_commands_v1( ) +@dataclass(frozen=True) +class LaneRunObservationV1: + """Whitelisted fields of one workflow run, exactly as GitHub reported them. + + Three fields are the whole observation surface: the id the dual proof + needs, the title that carries the lane coordinates, and the conclusion + that says whether the lane produced anything. Nothing else from the API + reply is admitted, so no unreviewed field can reach a decision or a log. + """ + + run_id: int + display_title: str + conclusion: str + + def __post_init__(self) -> None: + if ( + type(self.run_id) is not int + or self.run_id <= 0 + or type(self.display_title) is not str + or type(self.conclusion) is not str + ): + raise TypeError("invalid lane run observation") + + +@dataclass(frozen=True) +class LaneRunNameV1: + """The coordinates one lane run title carries.""" + + evidence_artifact: str + window_start: int + window_points: int + evidence_run_id: int + + def __post_init__(self) -> None: + if ( + type(self.evidence_artifact) is not str + or not self.evidence_artifact + or type(self.window_start) is not int + or self.window_start < 0 + or type(self.window_points) is not int + or self.window_points < 1 + or type(self.evidence_run_id) is not int + or self.evidence_run_id < 1 + ): + raise TypeError("invalid lane run name") + + +@dataclass(frozen=True) +class LaneRunCollectionV1: + """One successful lane run per plan window, in plan order.""" + + evidence_run_id: int + evidence_artifact: str + lanes: tuple[tuple[int, int, int], ...] + + def __post_init__(self) -> None: + if ( + type(self.evidence_run_id) is not int + or self.evidence_run_id < 1 + or type(self.evidence_artifact) is not str + or self.evidence_artifact not in EVIDENCE_ARTIFACTS_V1 + or type(self.lanes) is not tuple + or not self.lanes + or any( + type(lane) is not tuple + or len(lane) != 3 + or any(type(value) is not int for value in lane) + for lane in self.lanes + ) + ): + raise TypeError("invalid lane run collection") + + +@dataclass(frozen=True) +class LaneRunCollectionRejectedV1: + """Why no list of lane run ids exists. + + A partial cover is the failure this mode is built to catch, so the windows + that have no run and the windows that have more than one travel with the + refusal as data: the operator re-dispatches exactly those, and the caller + never has to parse prose to learn what is missing. + """ + + reason: corpus.ShardCorpusReasonV1 + detail: str + missing: tuple[tuple[int, int], ...] = () + duplicated: tuple[tuple[int, int], ...] = () + + def __post_init__(self) -> None: + if type(self.reason) is not corpus.ShardCorpusReasonV1: + raise TypeError("invalid lane run collection rejection reason") + if type(self.detail) is not str or not self.detail: + raise TypeError("invalid lane run collection rejection detail") + for windows in (self.missing, self.duplicated): + if type(windows) is not tuple or any( + type(window) is not tuple + or len(window) != 2 + or any(type(value) is not int for value in window) + for window in windows + ): + raise TypeError("invalid lane run collection window set") + + +def _collection_reject( + reason: corpus.ShardCorpusReasonV1, + detail: str, + missing: tuple[tuple[int, int], ...] = (), + duplicated: tuple[tuple[int, int], ...] = (), +) -> LaneRunCollectionRejectedV1: + return LaneRunCollectionRejectedV1(reason, detail, missing, duplicated) + + +def render_windows_v1(windows: tuple[tuple[int, int], ...]) -> str: + """The wire rendering of a window set, in the run-name's own notation.""" + + return ",".join(f"{start}+{points}" for start, points in windows) or "none" + + +def _window_summary_v1(label: str, windows: tuple[tuple[int, int], ...]) -> str: + """One bounded human line; the refusal itself carries the full set.""" + + head = render_windows_v1(windows[:_SUMMARY_WINDOWS_V1]) + if len(windows) <= _SUMMARY_WINDOWS_V1: + return f"{label}={head}" + return f"{label}={head} (+{len(windows) - _SUMMARY_WINDOWS_V1} more)" + + +def _ordinal_v1(token: str, minimum: int) -> int | None: + if _CANONICAL_ORDINAL_V1.fullmatch(token) is None: + return None + value = int(token) + return value if value >= minimum else None + + +def parse_lane_run_name_v1(display_title: object) -> LaneRunNameV1 | None: + """The lane coordinates a run title carries, or None if it carries none. + + `verification-lanes.yml` titles every lane run + + lane + of + + from a folded scalar, which collapses to exactly single-space separation. + Anything else — another workflow's run, another naming scheme, a title + that merely resembles the form — is not a lane run of this campaign, and + returning None keeps that judgement out of the network boundary. + """ + + if type(display_title) is not str: + return None + tokens = display_title.split(" ") + if len(tokens) != 5 or tokens[0] != "lane" or tokens[3] != "of": + return None + artifact, window, run_id_token = tokens[1], tokens[2], tokens[4] + if not artifact: + return None + start_token, separator, points_token = window.partition("+") + if not separator: + return None + window_start = _ordinal_v1(start_token, 0) + window_points = _ordinal_v1(points_token, 1) + evidence_run_id = _ordinal_v1(run_id_token, 1) + if window_start is None or window_points is None or evidence_run_id is None: + return None + return LaneRunNameV1(artifact, window_start, window_points, evidence_run_id) + + +def _canonical_plan_v1(plan: object) -> bool: + """Windows a run can be matched against one at a time. + + Overlap is the one thing that must not pass: two windows sharing an + ordinal would make "exactly one run per window" undefined before any run + is even observed. Whether the windows also cover the domain exactly is + `lane_plan_v1`'s judgement, not this one's. + """ + + if type(plan) is not tuple or not plan: + return False + cursor = 0 + for window in plan: + if ( + type(window) is not tuple + or len(window) != 2 + or any(type(value) is not int for value in window) + or window[0] < cursor + or window[1] < 1 + ): + return False + cursor = window[0] + window[1] + return True + + +def match_lane_runs_v1( + plan: object, + evidence_run_id: object, + evidence_artifact: object, + observations: object, +) -> LaneRunCollectionV1 | LaneRunCollectionRejectedV1: + """Bind observed run titles to the plan windows, or refuse with the gaps. + + Pure: it decides only from the plan and the observation it is handed, so + every rule below — which campaign a title belongs to, which engine, which + conclusion counts, what a complete cover is — is reachable by a test + without a network. A window with no successful run and a window with two + are the same class of failure: the campaign's run ids are not yet a fact, + and a list that hid either would let the dual proof rest on a cover + nobody checked. + """ + + if not _canonical_plan_v1(plan): + return _collection_reject( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, + "lane run collection requires a canonical lane plan", + ) + if type(evidence_run_id) is not int or evidence_run_id <= 0: + return _collection_reject( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, + "lane run collection requires a positive evidence run id", + ) + if ( + type(evidence_artifact) is not str + or evidence_artifact not in EVIDENCE_ARTIFACTS_V1 + ): + return _collection_reject( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, + "lane run collection requires an allowlisted engine artifact", + ) + try: + observed = tuple(observations) # type: ignore[arg-type] + except Exception: + # The observation is a hostile boundary: any iterator failure — not + # just a non-iterable — must land as the typed refusal. + return _collection_reject( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, + "lane run collection requires an iterable observation", + ) + if any(type(seen) is not LaneRunObservationV1 for seen in observed): + return _collection_reject( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, + "lane run collection requires canonical lane run observations", + ) + + covering: dict[tuple[int, int], list[int]] = {} + for seen in observed: + if seen.conclusion != LANE_RUN_SUCCESS_CONCLUSION_V1: + continue + name = parse_lane_run_name_v1(seen.display_title) + if name is None: + continue + # Two campaigns overlap in time by design — the two engines replay the + # same windows of the same evidence run — so the artifact and the + # evidence run id, not the clock, are what separate them. + if ( + name.evidence_run_id != evidence_run_id + or name.evidence_artifact != evidence_artifact + ): + continue + run_ids = covering.setdefault((name.window_start, name.window_points), []) + # One run listed twice is still one run: deduplicating by id keeps a + # repetitive listing from manufacturing a duplicate-cover refusal. + if seen.run_id not in run_ids: + run_ids.append(seen.run_id) + + missing = tuple(window for window in plan if not covering.get(window)) + duplicated = tuple( + window for window in plan if len(covering.get(window, ())) > 1 + ) + if missing or duplicated: + return _collection_reject( + corpus.ShardCorpusReasonV1.INCOMPLETE_COVER, + "the lane run cover is incomplete: " + f"{_window_summary_v1('missing', missing)} " + f"{_window_summary_v1('duplicated', duplicated)}", + missing, + duplicated, + ) + return LaneRunCollectionV1( + evidence_run_id, + evidence_artifact, + tuple( + (start, points, covering[(start, points)][0]) for start, points in plan + ), + ) + + +def lane_runs_json_v1( + collection: object, +) -> str | LaneRunCollectionRejectedV1: + """The deterministic wire form of one collected campaign.""" + + if type(collection) is not LaneRunCollectionV1: + return _collection_reject( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, + "the lane run wire form requires a collected campaign", + ) + return json.dumps( + { + "schema": LANE_RUN_COLLECTION_SCHEMA_V1, + "evidence_run_id": collection.evidence_run_id, + "evidence_artifact": collection.evidence_artifact, + "lane_count": len(collection.lanes), + "lanes": [ + { + "window_start": start, + "window_points": points, + "run_id": run_id, + } + for start, points, run_id in collection.lanes + ], + }, + sort_keys=True, + separators=(",", ":"), + ) + + +def gh_lane_runs_v1(limit: int = LANE_RUN_QUERY_LIMIT_V1) -> tuple[ + LaneRunObservationV1, ... +]: + """Every verification lane run GitHub lists, as whitelisted observations. + + The one impure boundary of the collection: it observes and reports what + it saw, without deciding anything. The projection is done by `--jq` in + the query itself, so no unreviewed field of the API reply ever enters the + process; which of these runs belongs to a campaign is a rule, and it lives + in `match_lane_runs_v1` where a test can reach it. + """ + + completed = subprocess.run( + ( + "gh", + "run", + "list", + "--workflow", + VERIFICATION_WORKFLOW_V1, + "--limit", + str(limit), + "--json", + "databaseId,displayTitle,conclusion", + "--jq", + ".[] | [.databaseId, .displayTitle, .conclusion] | @tsv", + ), + capture_output=True, + text=True, + check=True, + ) + observed: list[LaneRunObservationV1] = [] + for line in completed.stdout.splitlines(): + if not line: + continue + fields = line.split("\t") + if len(fields) != 3: + # A title carrying a tab would shift every field after it, so a + # record that does not project exactly is refused, not guessed. + raise ValueError("gh listed a lane run in an unreadable shape") + run_id, display_title, conclusion = fields + observed.append( + LaneRunObservationV1(int(run_id), display_title, conclusion) + ) + return tuple(observed) + + +def collect_lane_runs_v1( + plan: object, + evidence_run_id: object, + evidence_artifact: object, + observer: object | None = None, + limit: int = LANE_RUN_QUERY_LIMIT_V1, +) -> LaneRunCollectionV1 | LaneRunCollectionRejectedV1: + """Observe the lane runs once, then match them against the plan. + + The seam between the two is where a failure to observe becomes a typed + refusal instead of a crash — an unreachable API, a missing token or an + unreadable reply must not look like an empty campaign, because an empty + campaign and a broken query lead an operator to opposite actions. + """ + + # Resolved here, not captured as a default: a default binds the module + # attribute at definition time, which would make the injection point look + # real to a caller that replaces the observer and quietly not be. + if observer is None: + observer = gh_lane_runs_v1 + try: + observed = tuple(observer(limit)) # type: ignore[operator] + except Exception as error: + # The cause travels with the refusal: an operator has to tell "no such + # workflow" from "no token" from a network failure, and + # CalledProcessError.__repr__ drops stderr — the only place that text + # lives. + cause = getattr(error, "stderr", None) or repr(error) + return _collection_reject( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, + f"lane run collection cannot observe {VERIFICATION_WORKFLOW_V1}:" + f" {str(cause).strip()}", + ) + return match_lane_runs_v1(plan, evidence_run_id, evidence_artifact, observed) + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--mode", - choices=("plan", "dispatch", "verification-dispatch"), + choices=("plan", "dispatch", "verification-dispatch", "collect"), required=True, ) parser.add_argument("--lane-width", type=int, default=DEFAULT_LANE_WIDTH) parser.add_argument("--shard-width", type=int, default=DEFAULT_SHARD_WIDTH) parser.add_argument("--evidence-run-id", type=int, default=None) parser.add_argument("--evidence-artifact", type=str, default=None) + parser.add_argument("--run-limit", type=int, default=LANE_RUN_QUERY_LIMIT_V1) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--out", type=Path, required=True) args = parser.parse_args(argv) @@ -223,6 +646,45 @@ def main(argv: list[str] | None = None) -> int: ) return 0 + if args.mode == "collect": + if args.evidence_run_id is None: + print("lane run collection requires --evidence-run-id", file=sys.stderr) + return 64 + if args.evidence_artifact is None: + print("lane run collection requires --evidence-artifact", file=sys.stderr) + return 64 + collection = collect_lane_runs_v1( + plan, + args.evidence_run_id, + args.evidence_artifact, + limit=args.run_limit, + ) + if type(collection) is not LaneRunCollectionV1: + # Nothing is written: a partial list of run ids reads exactly like + # a complete one to whatever consumes it next. + print(f"lane run collection refused: {collection.detail}", file=sys.stderr) + for label, windows in ( + ("missing", collection.missing), + ("duplicated", collection.duplicated), + ): + if windows: + print( + f"{label} windows ({len(windows)}):" + f" {render_windows_v1(windows)}", + file=sys.stderr, + ) + return 64 + args.out.mkdir(parents=True, exist_ok=True) + (args.out / LANE_RUN_COLLECTION_NAME_V1).write_bytes( + lane_runs_json_v1(collection).encode("ascii") + ) + print( + f"collected lanes={len(collection.lanes)} " + f"artifact={collection.evidence_artifact} " + f"evidence_run={collection.evidence_run_id}" + ) + return 0 + if args.mode == "verification-dispatch": if args.evidence_run_id is None: print( diff --git a/proof/region/v1/tests/test_corpus_dispatch.py b/proof/region/v1/tests/test_corpus_dispatch.py index 2ed438fd..d8bd6351 100644 --- a/proof/region/v1/tests/test_corpus_dispatch.py +++ b/proof/region/v1/tests/test_corpus_dispatch.py @@ -29,6 +29,74 @@ FULL_DOMAIN = protocol.OUTPUT_CARDINALITY_V1 ALIGNMENT = corpus.CORPUS_SHARD_ALIGNMENT_V1 +ARB_ARTIFACT = "verification-evidence-arb" +MPFI_ARTIFACT = "verification-evidence-mpfi" +EVIDENCE_RUN = 4242424242 +FOREIGN_EVIDENCE_RUN = 5353535353 + + +def lane_run_name( + artifact: str, + window_start: int, + window_points: int, + evidence_run_id: int, +) -> str: + """The exact title `verification-lanes.yml` renders for one lane run. + + The workflow's folded `run-name` scalar + + lane ${{ inputs.evidence_artifact }} + ${{ inputs.window_start }}+${{ inputs.window_points }} + of ${{ inputs.evidence_run_id }} + + collapses to one line of single-space-separated coordinates, so the + collector's parser is tested against that literal shape and not against a + convenient invention. + """ + + return f"lane {artifact} {window_start}+{window_points} of {evidence_run_id}" + + +def observation( + run_id: int, + artifact: str = ARB_ARTIFACT, + window_start: int = 0, + window_points: int = 65536, + evidence_run_id: int = EVIDENCE_RUN, + conclusion: str = "success", +) -> object: + return corpus_dispatch.LaneRunObservationV1( + run_id, + lane_run_name(artifact, window_start, window_points, evidence_run_id), + conclusion, + ) + + +def cover( + plan: tuple[tuple[int, int], ...], + artifact: str = ARB_ARTIFACT, + evidence_run_id: int = EVIDENCE_RUN, + first_run_id: int = 900000, +) -> list[object]: + """One successful lane run per plan window, in reverse plan order. + + GitHub lists runs newest first, so a collector that returned observation + order instead of plan order would look right only by accident; the fixture + makes that accident impossible. + """ + + return [ + observation( + first_run_id + index, + artifact, + start, + points, + evidence_run_id, + ) + for index, (start, points) in reversed(list(enumerate(plan))) + ] + + class LanePlanCoverTests(unittest.TestCase): def test_default_plan_covers_the_full_domain_exactly(self) -> None: @@ -198,5 +266,411 @@ def test_unknown_mode_is_rejected(self) -> None: corpus_dispatch.main(["--mode", "launch"]) +class LaneRunNameTests(unittest.TestCase): + """The parser is the only thing that turns a run title into coordinates.""" + + def test_parses_the_exact_workflow_run_name_form(self) -> None: + parsed = corpus_dispatch.parse_lane_run_name_v1( + "lane verification-evidence-arb 0+65536 of 4242424242" + ) + self.assertIs(type(parsed), corpus_dispatch.LaneRunNameV1) + self.assertEqual(parsed.evidence_artifact, ARB_ARTIFACT) + self.assertEqual(parsed.window_start, 0) + self.assertEqual(parsed.window_points, 65536) + self.assertEqual(parsed.evidence_run_id, 4242424242) + + last = corpus_dispatch.parse_lane_run_name_v1( + lane_run_name(MPFI_ARTIFACT, FULL_DOMAIN - 65536, 65536, 7) + ) + self.assertIs(type(last), corpus_dispatch.LaneRunNameV1) + self.assertEqual(last.evidence_artifact, MPFI_ARTIFACT) + self.assertEqual(last.window_start, FULL_DOMAIN - 65536) + self.assertEqual(last.window_points, 65536) + self.assertEqual(last.evidence_run_id, 7) + + def test_every_noncanonical_title_is_not_a_lane_run(self) -> None: + titles = ( + # not a lane run at all + "", + "Verification lane replay", + "lane", + "full-domain lane 0+65536 of 7", + # arity drift + "lane verification-evidence-arb 0+65536 of", + "lane verification-evidence-arb 0+65536 of 7 rerun", + "lane verification-evidence-arb 0+65536 7", + # whitespace drift: a folded scalar emits exactly one space + "lane verification-evidence-arb 0+65536 of 7", + "lane verification-evidence-arb 0+65536 of 7 ", + " lane verification-evidence-arb 0+65536 of 7", + "lane verification-evidence-arb 0+65536\tof 7", + # window drift + "lane verification-evidence-arb 0-65536 of 7", + "lane verification-evidence-arb 65536 of 7", + "lane verification-evidence-arb 0+65536+7 of 7", + "lane verification-evidence-arb 0+0 of 7", + # noncanonical ordinals that int() would happily swallow + "lane verification-evidence-arb 00+65536 of 7", + "lane verification-evidence-arb +0+65536 of 7", + "lane verification-evidence-arb 0+6_5536 of 7", + "lane verification-evidence-arb 0+65536 of 007", + "lane verification-evidence-arb ٠+65536 of 7", + "lane verification-evidence-arb 0+65536 of -7", + "lane verification-evidence-arb 0+65536 of 0", + # foreign types + None, + 7, + b"lane verification-evidence-arb 0+65536 of 7", + ) + for title in titles: + self.assertIsNone(corpus_dispatch.parse_lane_run_name_v1(title), title) + + +class LaneRunMatchTests(unittest.TestCase): + """Matching names against the plan is pure: no network reaches it.""" + + def test_a_complete_cover_yields_every_run_id_in_plan_order(self) -> None: + plan = corpus_dispatch.lane_plan_v1() + self.assertIs(type(plan), tuple) + collected = corpus_dispatch.match_lane_runs_v1( + plan, EVIDENCE_RUN, ARB_ARTIFACT, cover(plan) + ) + self.assertIs(type(collected), corpus_dispatch.LaneRunCollectionV1) + self.assertEqual(collected.evidence_run_id, EVIDENCE_RUN) + self.assertEqual(collected.evidence_artifact, ARB_ARTIFACT) + self.assertEqual(len(collected.lanes), len(plan)) + self.assertEqual( + collected.lanes, + tuple( + (start, points, 900000 + index) + for index, (start, points) in enumerate(plan) + ), + ) + + def test_a_missing_window_is_a_typed_refusal_naming_the_hole(self) -> None: + plan = corpus_dispatch.lane_plan_v1() + self.assertIs(type(plan), tuple) + observations = [ + seen + for seen in cover(plan) + if f" {65536 * 3}+65536 " not in seen.display_title + ] + self.assertEqual(len(observations), len(plan) - 1) + refusal = corpus_dispatch.match_lane_runs_v1( + plan, EVIDENCE_RUN, ARB_ARTIFACT, observations + ) + self.assertIs(type(refusal), corpus_dispatch.LaneRunCollectionRejectedV1) + self.assertEqual(refusal.reason, corpus.ShardCorpusReasonV1.INCOMPLETE_COVER) + self.assertEqual(refusal.missing, ((65536 * 3, 65536),)) + self.assertEqual(refusal.duplicated, ()) + self.assertIn("196608+65536", refusal.detail) + + def test_a_window_in_two_runs_is_a_typed_refusal_naming_the_duplicate( + self, + ) -> None: + plan = ((0, 65536), (65536, 65536), (131072, 65536)) + observations = cover(plan) + [observation(777, ARB_ARTIFACT, 65536, 65536)] + refusal = corpus_dispatch.match_lane_runs_v1( + plan, EVIDENCE_RUN, ARB_ARTIFACT, observations + ) + self.assertIs(type(refusal), corpus_dispatch.LaneRunCollectionRejectedV1) + self.assertEqual(refusal.reason, corpus.ShardCorpusReasonV1.INCOMPLETE_COVER) + self.assertEqual(refusal.missing, ()) + self.assertEqual(refusal.duplicated, ((65536, 65536),)) + self.assertIn("65536+65536", refusal.detail) + + def test_a_foreign_engine_run_never_fills_this_engines_window(self) -> None: + plan = ((0, 65536), (65536, 65536), (131072, 65536)) + # The other engine's lane covers the same window of the same evidence + # run: only the artifact tells the two campaigns apart. + observations = [ + seen for seen in cover(plan) if " 65536+65536 " not in seen.display_title + ] + [observation(555, MPFI_ARTIFACT, 65536, 65536)] + refusal = corpus_dispatch.match_lane_runs_v1( + plan, EVIDENCE_RUN, ARB_ARTIFACT, observations + ) + self.assertIs(type(refusal), corpus_dispatch.LaneRunCollectionRejectedV1) + self.assertEqual(refusal.missing, ((65536, 65536),)) + + # And a complete Arb cover is not disturbed by the MPFI campaign + # running beside it. + collected = corpus_dispatch.match_lane_runs_v1( + plan, + EVIDENCE_RUN, + ARB_ARTIFACT, + cover(plan) + cover(plan, MPFI_ARTIFACT, first_run_id=800000), + ) + self.assertIs(type(collected), corpus_dispatch.LaneRunCollectionV1) + self.assertEqual( + collected.lanes, ((0, 65536, 900000), (65536, 65536, 900001), (131072, 65536, 900002)) + ) + + def test_a_run_of_another_campaign_never_fills_this_ones_window(self) -> None: + plan = ((0, 65536), (65536, 65536)) + observations = [ + seen for seen in cover(plan) if " 0+65536 " not in seen.display_title + ] + [ + observation(444, ARB_ARTIFACT, 0, 65536, FOREIGN_EVIDENCE_RUN), + ] + refusal = corpus_dispatch.match_lane_runs_v1( + plan, EVIDENCE_RUN, ARB_ARTIFACT, observations + ) + self.assertIs(type(refusal), corpus_dispatch.LaneRunCollectionRejectedV1) + self.assertEqual(refusal.missing, ((0, 65536),)) + + def test_an_unsuccessful_run_never_covers_its_window(self) -> None: + plan = ((0, 65536), (65536, 65536)) + for conclusion in ("failure", "cancelled", "timed_out", "skipped", ""): + observations = [ + seen for seen in cover(plan) if " 0+65536 " not in seen.display_title + ] + [observation(333, ARB_ARTIFACT, 0, 65536, conclusion=conclusion)] + refusal = corpus_dispatch.match_lane_runs_v1( + plan, EVIDENCE_RUN, ARB_ARTIFACT, observations + ) + self.assertIs( + type(refusal), corpus_dispatch.LaneRunCollectionRejectedV1, conclusion + ) + self.assertEqual(refusal.missing, ((0, 65536),), conclusion) + + def test_a_failed_run_beside_the_successful_rerun_is_not_a_duplicate(self) -> None: + plan = ((0, 65536),) + collected = corpus_dispatch.match_lane_runs_v1( + plan, + EVIDENCE_RUN, + ARB_ARTIFACT, + cover(plan) + [observation(222, ARB_ARTIFACT, 0, 65536, conclusion="failure")], + ) + self.assertIs(type(collected), corpus_dispatch.LaneRunCollectionV1) + self.assertEqual(collected.lanes, ((0, 65536, 900000),)) + + def test_gaps_and_duplicates_are_reported_together(self) -> None: + plan = ((0, 65536), (65536, 65536), (131072, 65536)) + observations = [ + seen for seen in cover(plan) if " 0+65536 " not in seen.display_title + ] + [observation(111, ARB_ARTIFACT, 131072, 65536)] + refusal = corpus_dispatch.match_lane_runs_v1( + plan, EVIDENCE_RUN, ARB_ARTIFACT, observations + ) + self.assertIs(type(refusal), corpus_dispatch.LaneRunCollectionRejectedV1) + self.assertEqual(refusal.missing, ((0, 65536),)) + self.assertEqual(refusal.duplicated, ((131072, 65536),)) + + def test_foreign_inputs_are_typed_refusals(self) -> None: + plan = ((0, 65536),) + cases = ( + (corpus.ShardCorpusRejectedV1( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, "foreign" + ), EVIDENCE_RUN, ARB_ARTIFACT, cover(plan)), + (((0, 65536), (65536,)), EVIDENCE_RUN, ARB_ARTIFACT, cover(plan)), + ((), EVIDENCE_RUN, ARB_ARTIFACT, ()), + (plan, 0, ARB_ARTIFACT, cover(plan)), + (plan, -1, ARB_ARTIFACT, cover(plan)), + (plan, "4242424242", ARB_ARTIFACT, cover(plan)), + (plan, True, ARB_ARTIFACT, cover(plan)), + (plan, EVIDENCE_RUN, "verification-evidence", cover(plan)), + (plan, EVIDENCE_RUN, None, cover(plan)), + (plan, EVIDENCE_RUN, ARB_ARTIFACT, None), + (plan, EVIDENCE_RUN, ARB_ARTIFACT, 7), + (plan, EVIDENCE_RUN, ARB_ARTIFACT, ["lane verification-evidence-arb 0+65536 of 4242424242"]), + ) + for case in cases: + refusal = corpus_dispatch.match_lane_runs_v1(*case) + self.assertIs( + type(refusal), corpus_dispatch.LaneRunCollectionRejectedV1, case + ) + self.assertEqual( + refusal.reason, corpus.ShardCorpusReasonV1.FOREIGN_INPUT, case + ) + + def test_the_collection_is_immutable_and_deterministic(self) -> None: + plan = ((0, 65536), (65536, 65536)) + first = corpus_dispatch.match_lane_runs_v1( + plan, EVIDENCE_RUN, ARB_ARTIFACT, cover(plan) + ) + second = corpus_dispatch.match_lane_runs_v1( + plan, EVIDENCE_RUN, ARB_ARTIFACT, cover(plan) + ) + self.assertEqual( + corpus_dispatch.lane_runs_json_v1(first), + corpus_dispatch.lane_runs_json_v1(second), + ) + decoded = json.loads(corpus_dispatch.lane_runs_json_v1(first)) + self.assertEqual(decoded["schema"], "corpus-lane-runs-v1") + self.assertEqual(decoded["evidence_run_id"], EVIDENCE_RUN) + self.assertEqual(decoded["evidence_artifact"], ARB_ARTIFACT) + self.assertEqual(decoded["lane_count"], 2) + self.assertEqual( + decoded["lanes"][0], + {"window_start": 0, "window_points": 65536, "run_id": 900000}, + ) + with self.assertRaises(Exception): + first.lanes = () + + +class GhObservationWhitelistTests(unittest.TestCase): + """The query itself is the whitelist: nothing else may enter the process.""" + + def _observe(self, stdout: str) -> tuple[object, list[str]]: + argv: list[str] = [] + + class Completed: + def __init__(self, out: str) -> None: + self.stdout = out + + def fake_run(command: tuple[str, ...], **kwargs: object) -> object: + argv.extend(command) + self.assertEqual(kwargs.get("check"), True) + self.assertEqual(kwargs.get("capture_output"), True) + return Completed(stdout) + + original = subprocess.run + subprocess.run = fake_run # type: ignore[assignment] + try: + return corpus_dispatch.gh_lane_runs_v1(7), argv + finally: + subprocess.run = original # type: ignore[assignment] + + def test_the_query_projects_exactly_three_fields(self) -> None: + observed, argv = self._observe( + "31\tlane verification-evidence-arb 0+65536 of 4242424242\tsuccess\n" + "30\tlane verification-evidence-mpfi 0+65536 of 4242424242\tfailure\n" + "29\tsome other run\t\n" + ) + joined = " ".join(argv) + self.assertIn("gh run list --workflow verification-lanes.yml", joined) + self.assertIn("--limit 7", joined) + self.assertIn("--json databaseId,displayTitle,conclusion", joined) + self.assertIn( + "--jq .[] | [.databaseId, .displayTitle, .conclusion] | @tsv", joined + ) + self.assertEqual( + observed, + ( + corpus_dispatch.LaneRunObservationV1( + 31, "lane verification-evidence-arb 0+65536 of 4242424242", "success" + ), + corpus_dispatch.LaneRunObservationV1( + 30, + "lane verification-evidence-mpfi 0+65536 of 4242424242", + "failure", + ), + corpus_dispatch.LaneRunObservationV1(29, "some other run", ""), + ), + ) + + def test_an_unreadable_record_is_never_guessed(self) -> None: + for stdout in ( + "31\tlane verification-evidence-arb 0+65536 of 42\n", + "31\tlane\twith\ttab\tsuccess\n", + "not-an-id\tlane verification-evidence-arb 0+65536 of 42\tsuccess\n", + ): + with self.assertRaises(ValueError, msg=stdout): + self._observe(stdout) + + +class LaneRunObservationBoundaryTests(unittest.TestCase): + def test_an_observer_failure_is_a_refusal_not_a_crash(self) -> None: + def boom(limit: int) -> tuple[object, ...]: + raise subprocess.CalledProcessError(1, ("gh",), stderr="gh: Not Found") + + refusal = corpus_dispatch.collect_lane_runs_v1( + ((0, 65536),), EVIDENCE_RUN, ARB_ARTIFACT, observer=boom + ) + self.assertIs(type(refusal), corpus_dispatch.LaneRunCollectionRejectedV1) + self.assertEqual(refusal.reason, corpus.ShardCorpusReasonV1.FOREIGN_INPUT) + self.assertIn("Not Found", refusal.detail) + + def test_collection_matches_exactly_what_the_observer_reported(self) -> None: + plan = ((0, 65536), (65536, 65536)) + seen: list[int] = [] + + def observer(limit: int) -> tuple[object, ...]: + seen.append(limit) + return tuple(cover(plan)) + + collected = corpus_dispatch.collect_lane_runs_v1( + plan, EVIDENCE_RUN, ARB_ARTIFACT, observer=observer, limit=13 + ) + self.assertIs(type(collected), corpus_dispatch.LaneRunCollectionV1) + self.assertEqual(collected.lanes, ((0, 65536, 900000), (65536, 65536, 900001))) + self.assertEqual(seen, [13]) + + +class CollectCliTests(unittest.TestCase): + def _with_observer(self, observer: object, argv: list[str]) -> int: + original = corpus_dispatch.gh_lane_runs_v1 + corpus_dispatch.gh_lane_runs_v1 = observer # type: ignore[assignment] + try: + return corpus_dispatch.main(argv) + finally: + corpus_dispatch.gh_lane_runs_v1 = original # type: ignore[assignment] + + def test_collect_mode_writes_the_machine_readable_run_ids(self) -> None: + plan = corpus_dispatch.lane_plan_v1() + with tempfile.TemporaryDirectory() as out: + status = self._with_observer( + lambda limit: tuple(cover(plan)), + [ + "--mode", + "collect", + "--evidence-run-id", + str(EVIDENCE_RUN), + "--evidence-artifact", + ARB_ARTIFACT, + "--out", + out, + ], + ) + self.assertEqual(status, 0) + decoded = json.loads((Path(out) / "lane-runs.json").read_text()) + self.assertEqual(decoded["schema"], "corpus-lane-runs-v1") + self.assertEqual(decoded["lane_count"], 256) + self.assertEqual(len(decoded["lanes"]), 256) + self.assertEqual( + [lane["run_id"] for lane in decoded["lanes"]], + [900000 + index for index in range(256)], + ) + + def test_an_incomplete_collection_writes_nothing_and_exits_64(self) -> None: + plan = corpus_dispatch.lane_plan_v1() + holed = [ + seen for seen in cover(plan) if " 65536+65536 " not in seen.display_title + ] + with tempfile.TemporaryDirectory() as out: + status = self._with_observer( + lambda limit: tuple(holed), + [ + "--mode", + "collect", + "--evidence-run-id", + str(EVIDENCE_RUN), + "--evidence-artifact", + ARB_ARTIFACT, + "--out", + out, + ], + ) + self.assertEqual(status, 64) + self.assertEqual(list(Path(out).iterdir()), []) + + def test_collect_requires_both_evidence_coordinates(self) -> None: + with tempfile.TemporaryDirectory() as out: + incomplete = ( + ["--mode", "collect", "--evidence-artifact", ARB_ARTIFACT, + "--out", out], + ["--mode", "collect", "--evidence-run-id", str(EVIDENCE_RUN), + "--out", out], + ) + for argv in incomplete: + self.assertEqual( + self._with_observer( + lambda limit: self.fail("collect must not observe"), argv + ), + 64, + argv, + ) + self.assertEqual(list(Path(out).iterdir()), []) + + if __name__ == "__main__": unittest.main() From 8f4c924602365fd14259424876bfa66248ae5a40 Mon Sep 17 00:00:00 2001 From: Claude Code Date: Fri, 7 Aug 2026 00:33:39 +0300 Subject: [PATCH 2/4] Proof: defend the three invariants the collection only asserted MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Independent verification ran mutants and found the same class three times: an invariant stated in code and in the comment beside it, with nothing that would notice its removal. The expensive one is the second engine. Both engines replay the same evidence build, so a listing carries both campaigns — and a collector that hardcoded the first artifact answered an MPFI request with Arb's run ids and exited 0. Half a dual proof, silently wrong, reported as success. Every CLI case used the Arb artifact, so nothing could tell the two apart. The overlap guard was theatre in the literal sense: its docstring calls overlap "the one thing that must not pass", and the refusal suite contained no overlapping plan at all — the case it did contain died on the tuple-shape branch beside it. A plan of two windows sharing an ordinal is in the suite now. Deduplicating by run id had the same shape: a listing that repeats a run must not manufacture a duplicate-cover refusal, and removing the check left everything green. Each is proven by the mutant it kills, one test apiece: the hardcoded artifact at the collect call site, `window[0] < cursor` weakened to a tautology, and the unconditional append. Verified: 31 dispatch tests pass; 429 on Linux with the three pre-existing `test_build` failures that also fail on an untouched tree. --- proof/region/v1/tests/test_corpus_dispatch.py | 65 +++++++++++++++++++ 1 file changed, 65 insertions(+) diff --git a/proof/region/v1/tests/test_corpus_dispatch.py b/proof/region/v1/tests/test_corpus_dispatch.py index d8bd6351..f2edb4c0 100644 --- a/proof/region/v1/tests/test_corpus_dispatch.py +++ b/proof/region/v1/tests/test_corpus_dispatch.py @@ -462,6 +462,9 @@ def test_foreign_inputs_are_typed_refusals(self) -> None: corpus.ShardCorpusReasonV1.FOREIGN_INPUT, "foreign" ), EVIDENCE_RUN, ARB_ARTIFACT, cover(plan)), (((0, 65536), (65536,)), EVIDENCE_RUN, ARB_ARTIFACT, cover(plan)), + # Overlap is what the guard exists for, and a plan short of one + # window kills only the tuple-shape branch beside it. + (((0, 65536), (32768, 65536)), EVIDENCE_RUN, ARB_ARTIFACT, cover(plan)), ((), EVIDENCE_RUN, ARB_ARTIFACT, ()), (plan, 0, ARB_ARTIFACT, cover(plan)), (plan, -1, ARB_ARTIFACT, cover(plan)), @@ -631,6 +634,68 @@ def test_collect_mode_writes_the_machine_readable_run_ids(self) -> None: [900000 + index for index in range(256)], ) + def test_one_run_listed_twice_is_still_one_run(self) -> None: + # A repetitive listing must not manufacture a duplicate-cover refusal: + # the same run id twice is the same run, and the collection has to say + # so rather than report the campaign broken. + plan = corpus_dispatch.lane_plan_v1() + listing = cover(plan) + doubled = listing + listing + with tempfile.TemporaryDirectory() as out: + status = self._with_observer( + lambda limit: tuple(doubled), + [ + "--mode", + "collect", + "--evidence-run-id", + str(EVIDENCE_RUN), + "--evidence-artifact", + ARB_ARTIFACT, + "--out", + out, + ], + ) + self.assertEqual(status, 0) + decoded = json.loads((Path(out) / "lane-runs.json").read_text()) + self.assertEqual(len(decoded["lanes"]), 256) + self.assertEqual( + [lane["run_id"] for lane in decoded["lanes"]], + [900000 + index for index in range(256)], + ) + + def test_the_second_engine_is_collected_when_both_campaigns_are_listed( + self, + ) -> None: + # Both engines replay the same evidence build, so a listing carries + # both campaigns. A collector that hardcoded the first artifact would + # answer the MPFI request with Arb's run ids and exit 0 — half the + # dual proof, silently wrong. + plan = corpus_dispatch.lane_plan_v1() + listing = cover(plan, artifact=ARB_ARTIFACT, first_run_id=900000) + cover( + plan, artifact=MPFI_ARTIFACT, first_run_id=700000 + ) + with tempfile.TemporaryDirectory() as out: + status = self._with_observer( + lambda limit: tuple(listing), + [ + "--mode", + "collect", + "--evidence-run-id", + str(EVIDENCE_RUN), + "--evidence-artifact", + MPFI_ARTIFACT, + "--out", + out, + ], + ) + self.assertEqual(status, 0) + decoded = json.loads((Path(out) / "lane-runs.json").read_text()) + self.assertEqual(decoded["evidence_artifact"], MPFI_ARTIFACT) + self.assertEqual( + [lane["run_id"] for lane in decoded["lanes"]], + [700000 + index for index in range(256)], + ) + def test_an_incomplete_collection_writes_nothing_and_exits_64(self) -> None: plan = corpus_dispatch.lane_plan_v1() holed = [ From aa3251474b6709d47f1d24797f1827792097fe09 Mon Sep 17 00:00:00 2001 From: Claude Code Date: Fri, 7 Aug 2026 08:52:02 +0300 Subject: [PATCH 3/4] Proof: the collect CLI judges its arguments before it spends a query MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Замечание ревью верно по существу: `--mode collect` передавал свои аргументы в `collect_lane_runs_v1` без единой проверки, а тот первым делом делает сетевой вызов. Наблюдённое поведение до правки (WSL, наблюдатель-заглушка вместо `gh`): - `--evidence-run-id 0` и `-1`: запрос выполнен (observer_called=[2000]), отказ гласит «lane run collection cannot observe verification-lanes.yml: …» — вина переложена на workflow, а не на аргумент; - `--evidence-artifact verification-evidence-flint`: то же самое; - `--run-limit 0` и `-5`: значение уходит в `gh run list --limit` дословно, а пустой ответ превращается в отказ «cover is incomplete: missing=…» с 256 окнами — оператор идёт передиспатчивать кампанию, которая цела. Закон: координата, которая будет потрачена на сетевой запрос, проверяется до запроса, и отказ называет виновный аргумент. Код возврата назвать его не может — он 64 у всех соседних отказов, поэтому именно на текст и на отсутствие запроса опираются тесты. Allowlist не продублирован: он рендерится из `EVIDENCE_ARTIFACTS_V1`, так что оператор узнаёт допустимый набор из самого отказа. Тот же инвариант рядом (предсуществующий дефект, закрыт этим же срезом): `--mode verification-dispatch` с непозитивным run id или чужим артефактом обходил построитель команд, тот возвращал типизированный отказ, а `main` обходил его циклом как список команд — оператор получал `TypeError: 'ShardCorpusRejectedV1' object is not iterable` вместо причины. Теперь отказ построителя печатается с его собственной причиной и выходит 64. Доказательства (WSL, python3 -m unittest): - RED до реализации: три падения по заявленной причине («--evidence-run-id» не найден в тексте отказа; «--run-limit» не найден в отказе про неполное покрытие) и одна ошибка TypeError в verification-dispatch. - 8 мутантов, каждый убит: M1/M2/M3 — снятие каждой из трёх проверок; M4 — общий текст «requires valid arguments» вместо имени аргумента (убивает все три теста); M5 — проверки перенесены ПОСЛЕ запроса (observed=[2000] != []); M6 — проверка лимита сделана слишком строгой (`>= 0`), убита тестом на допустимые аргументы; M7 — отказ без причины; M8 — снятие защиты от обхода отказа (тот самый TypeError). - Прежний `test_collect_requires_both_evidence_coordinates` заменён: его наблюдатель звал `self.fail`, а `collect_lane_runs_v1` глотает любое исключение наблюдателя в типизированный отказ с тем же кодом 64 — тест не мог упасть по своей заявленной причине. - Набор proof: 433 теста, 3 падения — ровно предсуществующие в test_build (два golden-digest и post_popen_handler_gap), они же падают на нетронутом дереве (429 тестов, те же три). - Контакт с реальностью: после правки все пять случаев дают observer_called=[], ничего не записано, текст называет аргумент. Rust не затронут: изменены три файла в proof/region/v1. --- proof/region/v1/corpus_dispatch.py | 33 +++- proof/region/v1/tests/test_corpus_dispatch.py | 141 ++++++++++++++++-- .../v1/tests/test_verification_dispatch.py | 41 ++++- 3 files changed, 194 insertions(+), 21 deletions(-) diff --git a/proof/region/v1/corpus_dispatch.py b/proof/region/v1/corpus_dispatch.py index ca38c5df..66ee7072 100644 --- a/proof/region/v1/corpus_dispatch.py +++ b/proof/region/v1/corpus_dispatch.py @@ -647,11 +647,31 @@ def main(argv: list[str] | None = None) -> int: return 0 if args.mode == "collect": - if args.evidence_run_id is None: - print("lane run collection requires --evidence-run-id", file=sys.stderr) + # Every one of these coordinates is spent on a network query, so all + # three are judged before it is made. A query fired with a foreign + # coordinate comes back as a refusal about `verification-lanes.yml` or + # as an empty listing — the first blames the workflow for the + # operator's typo, the second reports a 256-window hole in a campaign + # that may be intact. Each refusal names the argument at fault, + # because the exit code cannot: it is 64 for all of them. + if args.evidence_run_id is None or args.evidence_run_id <= 0: + print( + "lane run collection requires a positive --evidence-run-id", + file=sys.stderr, + ) return 64 - if args.evidence_artifact is None: - print("lane run collection requires --evidence-artifact", file=sys.stderr) + if args.evidence_artifact not in EVIDENCE_ARTIFACTS_V1: + print( + "lane run collection requires --evidence-artifact from " + + ", ".join(EVIDENCE_ARTIFACTS_V1), + file=sys.stderr, + ) + return 64 + if args.run_limit <= 0: + print( + "lane run collection requires a positive --run-limit", + file=sys.stderr, + ) return 64 collection = collect_lane_runs_v1( plan, @@ -706,6 +726,11 @@ def main(argv: list[str] | None = None) -> int: ) else: commands = dispatch_commands_v1(plan, args.shard_width) + if type(commands) is not tuple: + # The builder already decided; walking its rejection as if it were the + # command list turned that decision into a TypeError traceback. + print(f"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_corpus_dispatch.py b/proof/region/v1/tests/test_corpus_dispatch.py index f2edb4c0..24c5607f 100644 --- a/proof/region/v1/tests/test_corpus_dispatch.py +++ b/proof/region/v1/tests/test_corpus_dispatch.py @@ -12,11 +12,13 @@ from __future__ import annotations +import io import json import subprocess import sys import tempfile import unittest +from contextlib import redirect_stderr from pathlib import Path PROOF = Path(__file__).resolve().parents[1] @@ -718,23 +720,130 @@ def test_an_incomplete_collection_writes_nothing_and_exits_64(self) -> None: self.assertEqual(status, 64) self.assertEqual(list(Path(out).iterdir()), []) - def test_collect_requires_both_evidence_coordinates(self) -> None: - with tempfile.TemporaryDirectory() as out: - incomplete = ( - ["--mode", "collect", "--evidence-artifact", ARB_ARTIFACT, - "--out", out], - ["--mode", "collect", "--evidence-run-id", str(EVIDENCE_RUN), - "--out", out], - ) - for argv in incomplete: - self.assertEqual( - self._with_observer( - lambda limit: self.fail("collect must not observe"), argv - ), - 64, - argv, + def _refuse(self, argv: list[str]) -> tuple[int, str, list[int]]: + """Run the CLI, recording every query the arguments would have fired. + + The observer answers with an empty listing rather than raising: a + raising observer is swallowed into a typed refusal that also exits 64, + so it would make "the query never happened" untestable. Here an + argument that reaches the query produces the *cover* refusal — same + exit code, different text — which is exactly why these tests read the + text and the recorded queries, never the exit code alone. + """ + + observed: list[int] = [] + + def observer(limit: int) -> tuple[object, ...]: + observed.append(limit) + return () + + stderr = io.StringIO() + with redirect_stderr(stderr): + status = self._with_observer(observer, argv) + return status, stderr.getvalue(), observed + + def _assert_names_only(self, stderr: str, argument: str, context: object) -> None: + """The refusal names the argument at fault and no other.""" + + self.assertIn(argument, stderr, context) + for other in ("--evidence-run-id", "--evidence-artifact", "--run-limit"): + if other != argument: + self.assertNotIn(other, stderr, context) + + def test_a_foreign_evidence_run_id_is_named_and_never_queried(self) -> None: + # Every neighbouring refusal also exits 64, so the exit code cannot + # tell an operator which argument to fix; the text has to. And the + # query must not happen at all: fired with a foreign coordinate it can + # only come back as a refusal that blames `verification-lanes.yml`. + for value in (None, "0", "-1"): + argv = ["--mode", "collect", "--evidence-artifact", ARB_ARTIFACT] + if value is not None: + argv += ["--evidence-run-id", value] + with tempfile.TemporaryDirectory() as out: + status, stderr, observed = self._refuse(argv + ["--out", out]) + self.assertEqual(status, 64, value) + self._assert_names_only(stderr, "--evidence-run-id", value) + self.assertEqual(observed, [], value) + self.assertEqual(list(Path(out).iterdir()), [], value) + + def test_a_foreign_evidence_artifact_is_named_and_never_queried(self) -> None: + for value in ( + None, + "", + "verification-evidence", + "verification-evidence-flint", + "verification-evidence-arb/", + ): + argv = ["--mode", "collect", "--evidence-run-id", str(EVIDENCE_RUN)] + if value is not None: + argv += ["--evidence-artifact", value] + with tempfile.TemporaryDirectory() as out: + status, stderr, observed = self._refuse(argv + ["--out", out]) + self.assertEqual(status, 64, value) + self._assert_names_only(stderr, "--evidence-artifact", value) + # The operator learns the admissible set from the refusal, and + # it is the module's allowlist rather than a second copy. + for artifact in corpus_dispatch.EVIDENCE_ARTIFACTS_V1: + self.assertIn(artifact, stderr, value) + self.assertEqual(observed, [], value) + self.assertEqual(list(Path(out).iterdir()), [], value) + + def test_a_nonpositive_run_limit_is_named_and_never_queried(self) -> None: + # `gh run list --limit 0` is not a query anyone can act on, and a + # negative limit reaches the network verbatim. + for value in ("0", "-5"): + with tempfile.TemporaryDirectory() as out: + status, stderr, observed = self._refuse( + [ + "--mode", + "collect", + "--evidence-run-id", + str(EVIDENCE_RUN), + "--evidence-artifact", + ARB_ARTIFACT, + "--run-limit", + value, + "--out", + out, + ] ) - self.assertEqual(list(Path(out).iterdir()), []) + self.assertEqual(status, 64, value) + self._assert_names_only(stderr, "--run-limit", value) + self.assertEqual(observed, [], value) + self.assertEqual(list(Path(out).iterdir()), [], value) + + def test_the_admissible_arguments_reach_the_query_unchanged(self) -> None: + # The guard above must refuse foreign arguments, not narrow the + # admissible ones: every allowlisted artifact and a positive limit + # still reach the observer exactly as given. + plan = corpus_dispatch.lane_plan_v1() + for artifact in corpus_dispatch.EVIDENCE_ARTIFACTS_V1: + observed: list[int] = [] + + def observer(limit: int) -> tuple[object, ...]: + observed.append(limit) + return tuple(cover(plan, artifact=artifact)) + + with tempfile.TemporaryDirectory() as out: + status = self._with_observer( + observer, + [ + "--mode", + "collect", + "--evidence-run-id", + str(EVIDENCE_RUN), + "--evidence-artifact", + artifact, + "--run-limit", + "1", + "--out", + out, + ], + ) + self.assertEqual(status, 0, artifact) + self.assertEqual(observed, [1], artifact) + decoded = json.loads((Path(out) / "lane-runs.json").read_text()) + self.assertEqual(decoded["evidence_artifact"], artifact) if __name__ == "__main__": diff --git a/proof/region/v1/tests/test_verification_dispatch.py b/proof/region/v1/tests/test_verification_dispatch.py index 8cc7dfc6..7fce51ef 100644 --- a/proof/region/v1/tests/test_verification_dispatch.py +++ b/proof/region/v1/tests/test_verification_dispatch.py @@ -23,7 +23,7 @@ import sys import tempfile import unittest -from contextlib import redirect_stdout +from contextlib import redirect_stderr, redirect_stdout from pathlib import Path PROOF = Path(__file__).resolve().parents[1] @@ -189,6 +189,45 @@ def test_missing_evidence_run_id_exits_64(self) -> None: ) self.assertEqual(exit_code, 64) + def test_a_rejected_dispatch_is_reported_by_cause_not_raised(self) -> None: + # The command builder answers a foreign coordinate with a typed + # rejection, and `main` used to walk that rejection as if it were the + # command list — an operator got a TypeError traceback instead of the + # reason. The cause the builder named must reach stderr, and the + # process must leave by the same door as every other refusal. + for coordinates, cause in ( + ( + ["--evidence-run-id", "0", "--evidence-artifact", ARB_ARTIFACT], + "positive evidence run id", + ), + ( + [ + "--evidence-run-id", + "31000000001", + "--evidence-artifact", + "verification-evidence-flint", + ], + "allowlisted engine artifact", + ), + ): + with tempfile.TemporaryDirectory() as tmp: + stderr = io.StringIO() + with redirect_stderr(stderr): + exit_code = corpus_dispatch.main( + [ + "--mode", + "verification-dispatch", + "--lane-width", + str(1 << 23), + *coordinates, + "--dry-run", + "--out", + str(Path(tmp) / "out"), + ] + ) + self.assertEqual(exit_code, 64, coordinates) + self.assertIn(cause, stderr.getvalue(), coordinates) + def test_missing_evidence_artifact_exits_64(self) -> None: with tempfile.TemporaryDirectory() as tmp: exit_code = corpus_dispatch.main( From b4450f66fced7c404d7dc6bb708816198c427720 Mon Sep 17 00:00:00 2001 From: Claude Code Date: Fri, 7 Aug 2026 10:12:45 +0300 Subject: [PATCH 4/4] Proof: a saturated listing blames the limit, not the campaign MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Independent verification found the class this branch claimed closed was closed only at the zero boundary: a positive but truncating --run-limit spent the query and then reported the campaign incomplete — sending the operator to re-dispatch 256 lanes when the fix is one flag. `gh run list --limit N` drops the oldest runs, so a listing that came back exactly at the limit cannot distinguish a real hole from its own truncation. The refusal now names the flag when the listing is saturated, and keeps blaming the campaign when there is room to spare — the anti-vacuity case, because hiding real damage behind the flag would be the opposite failure. Also corrected here: the comment claiming CalledProcessError.__repr__ drops stderr — measured false on the #555 branch, repr carries every constructor argument; stderr is preferred for legibility, not recovery. Verified: 36 dispatch tests pass, both new ones read the refusal text. --- proof/region/v1/corpus_dispatch.py | 26 +++++++-- proof/region/v1/tests/test_corpus_dispatch.py | 58 +++++++++++++++++++ 2 files changed, 80 insertions(+), 4 deletions(-) diff --git a/proof/region/v1/corpus_dispatch.py b/proof/region/v1/corpus_dispatch.py index 66ee7072..1806d779 100644 --- a/proof/region/v1/corpus_dispatch.py +++ b/proof/region/v1/corpus_dispatch.py @@ -598,16 +598,34 @@ def collect_lane_runs_v1( observed = tuple(observer(limit)) # type: ignore[operator] except Exception as error: # The cause travels with the refusal: an operator has to tell "no such - # workflow" from "no token" from a network failure, and - # CalledProcessError.__repr__ drops stderr — the only place that text - # lives. + # workflow" from "no token" from a network failure. `stderr` is + # preferred over `repr` for legibility — the text itself rather than + # the text wrapped in a constructor call. cause = getattr(error, "stderr", None) or repr(error) return _collection_reject( corpus.ShardCorpusReasonV1.FOREIGN_INPUT, f"lane run collection cannot observe {VERIFICATION_WORKFLOW_V1}:" f" {str(cause).strip()}", ) - return match_lane_runs_v1(plan, evidence_run_id, evidence_artifact, observed) + matched = match_lane_runs_v1(plan, evidence_run_id, evidence_artifact, observed) + if ( + type(matched) is LaneRunCollectionRejectedV1 + and matched.reason is corpus.ShardCorpusReasonV1.INCOMPLETE_COVER + and len(observed) == limit + ): + # A listing saturated at the limit truncates the oldest runs, so the + # gap may be the query's, not the campaign's. Blaming the campaign + # sends the operator to re-dispatch 256 lanes when the fix is one + # flag — the exact misdirection this admission exists to prevent. + return _collection_reject( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, + f"the listing is saturated at --run-limit {limit}, so older lane" + f" runs may lie beyond it; raise the limit before blaming the" + f" campaign ({matched.detail})", + matched.missing, + matched.duplicated, + ) + return matched def main(argv: list[str] | None = None) -> int: diff --git a/proof/region/v1/tests/test_corpus_dispatch.py b/proof/region/v1/tests/test_corpus_dispatch.py index 24c5607f..f11731e5 100644 --- a/proof/region/v1/tests/test_corpus_dispatch.py +++ b/proof/region/v1/tests/test_corpus_dispatch.py @@ -12,6 +12,7 @@ from __future__ import annotations +import contextlib import io import json import subprocess @@ -636,6 +637,63 @@ def test_collect_mode_writes_the_machine_readable_run_ids(self) -> None: [900000 + index for index in range(256)], ) + def test_a_saturated_listing_blames_the_limit_not_the_campaign(self) -> None: + # `gh run list --limit N` truncates the oldest runs. A positive but + # short limit used to spend the query and then report the campaign + # incomplete — sending the operator to re-dispatch 256 lanes when the + # fix is one flag. + plan = corpus_dispatch.lane_plan_v1() + listing = cover(plan) + errors = io.StringIO() + with tempfile.TemporaryDirectory() as out: + with contextlib.redirect_stderr(errors): + status = self._with_observer( + lambda limit: tuple(listing[:limit]), + [ + "--mode", + "collect", + "--evidence-run-id", + str(EVIDENCE_RUN), + "--evidence-artifact", + ARB_ARTIFACT, + "--run-limit", + "255", + "--out", + out, + ], + ) + self.assertEqual(status, 64) + self.assertIn("--run-limit 255", errors.getvalue()) + self.assertIn("saturated", errors.getvalue()) + + def test_an_unsaturated_incomplete_cover_still_blames_the_campaign(self) -> None: + # Anti-vacuity for the saturation rule: a genuine hole with room to + # spare in the listing is the campaign's fault, and saying otherwise + # would hide real damage behind the flag. + plan = corpus_dispatch.lane_plan_v1() + holed = [ + seen for seen in cover(plan) if " 65536+65536 " not in seen.display_title + ] + errors = io.StringIO() + with tempfile.TemporaryDirectory() as out: + with contextlib.redirect_stderr(errors): + status = self._with_observer( + lambda limit: tuple(holed), + [ + "--mode", + "collect", + "--evidence-run-id", + str(EVIDENCE_RUN), + "--evidence-artifact", + ARB_ARTIFACT, + "--out", + out, + ], + ) + self.assertEqual(status, 64) + self.assertNotIn("saturated", errors.getvalue()) + self.assertIn("incomplete", errors.getvalue()) + def test_one_run_listed_twice_is_still_one_run(self) -> None: # A repetitive listing must not manufacture a duplicate-cover refusal: # the same run id twice is the same run, and the collection has to say