From 27eb63abf6ba00f27e884270427981ac948dd9fe Mon Sep 17 00:00:00 2001
From: Claude Code <daniilerosov12@gmail.com>
Date: Thu, 6 Aug 2026 17:25:45 +0300
Subject: [PATCH 1/7] Proof: a dispatch refuses evidence its lanes cannot
 download (V5b2d-4c)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Наблюдаемый дефект: coordinator проверял, что --evidence-run-id положительное
целое, но не то, что такой прогон существует и несёт названный артефакт.
Положительность — не свидетельство существования. Я запустил диспатч против
run 99999999 и получил 133 обречённых прогона, каждый падал за 9-11 секунд на
скачивании несуществующего артефакта.

Одна ошибка координатора превращается в 256 обречённых задач, и узнаёшь ты об
этом 256 раз подряд вместо одного.

Закон: перед первым диспатчем координатор спрашивает, несёт ли названный
прогон названный артефакт, и отказывает закрыто. Наблюдение GitHub — единственная
нечистая граница — инъектируется, поэтому само решение проверяется без сети.
Любой сбой наблюдения (нет прогона, нет токена, битый ответ) — тоже отказ, а не
падение и не молчаливый проход. Dry run остаётся офлайновым по контракту, гейт
живёт только на боевом пути.

Доказательства: четыре RED-теста (прогон без артефакта, недостижимый прогон,
успешный допуск, anti-vacuity — наблюдателя спрашивают именно про названный
прогон) плюс проверка реальностью: admit_evidence_artifact_v1(99999999, ...)
отвергает тот самый вход, что стоил 133 прогонов, а 31089986150 допускается для
обоих движков.

Локально: 14 тестов координатора, OK.
---
 proof/region/v1/corpus_dispatch.py            | 65 +++++++++++++++++++
 proof/region/v1/tests/test_corpus_dispatch.py | 52 +++++++++++++++
 2 files changed, 117 insertions(+)

diff --git a/proof/region/v1/corpus_dispatch.py b/proof/region/v1/corpus_dispatch.py
index 2326569f..5c783ab9 100644
--- a/proof/region/v1/corpus_dispatch.py
+++ b/proof/region/v1/corpus_dispatch.py
@@ -188,6 +188,62 @@ def verification_dispatch_commands_v1(
     )
 
 
+def gh_run_artifact_names_v1(run_id: int) -> tuple[str, ...]:
+    """Artifact names the given workflow run carries, read through `gh`.
+
+    The one impure boundary of this admission: it observes GitHub.  It is
+    injected into `admit_evidence_artifact_v1` so the decision itself stays
+    testable without a network.
+    """
+
+    completed = subprocess.run(
+        (
+            "gh",
+            "api",
+            f"repos/{{owner}}/{{repo}}/actions/runs/{run_id}/artifacts",
+            "--jq",
+            ".artifacts[].name",
+        ),
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    return tuple(
+        line.strip() for line in completed.stdout.splitlines() if line.strip()
+    )
+
+
+def admit_evidence_artifact_v1(
+    evidence_run_id: int,
+    evidence_artifact: str,
+    observer: object = gh_run_artifact_names_v1,
+) -> corpus.ShardCorpusRejectedV1 | None:
+    """Refuse a dispatch whose evidence run cannot carry what it names.
+
+    Every lane downloads `evidence_artifact` from `evidence_run_id`; a run
+    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.
+    """
+
+    try:
+        names = tuple(observer(evidence_run_id))  # type: ignore[operator]
+    except Exception:
+        # The observation is a hostile boundary: an unreachable run, a
+        # missing token or a malformed reply must all land as one refusal.
+        return corpus._reject(
+            corpus.ShardCorpusReasonV1.FOREIGN_INPUT,
+            f"verification dispatch cannot observe run {evidence_run_id}",
+        )
+    if evidence_artifact not in names:
+        return corpus._reject(
+            corpus.ShardCorpusReasonV1.FOREIGN_INPUT,
+            f"run {evidence_run_id} carries no artifact {evidence_artifact!r}",
+        )
+    return None
+
+
 def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument(
@@ -242,6 +298,15 @@ 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 observation belongs to the live path only.
+            refusal = admit_evidence_artifact_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_corpus_dispatch.py b/proof/region/v1/tests/test_corpus_dispatch.py
index 2ed438fd..ae6ded64 100644
--- a/proof/region/v1/tests/test_corpus_dispatch.py
+++ b/proof/region/v1/tests/test_corpus_dispatch.py
@@ -198,5 +198,57 @@ def test_unknown_mode_is_rejected(self) -> None:
             corpus_dispatch.main(["--mode", "launch"])
 
 
+class EvidenceAdmissionTests(unittest.TestCase):
+    """A dispatch names a run that must actually carry the artifact.
+
+    The lane workflow downloads the named artifact from the named run, so a
+    run that does not exist, or exists without that artifact, produces one
+    doomed job per lane — 256 of them, each failing seconds in.  The
+    coordinator refuses before any dispatch exists instead of discovering it
+    256 times over.
+    """
+
+    def test_a_run_without_the_artifact_is_a_typed_rejection(self) -> None:
+        result = corpus_dispatch.admit_evidence_artifact_v1(
+            424242,
+            "verification-evidence-arb",
+            lambda run_id: ("verification-evidence-mpfi",),
+        )
+        self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
+        self.assertEqual(result.reason, corpus.ShardCorpusReasonV1.FOREIGN_INPUT)
+
+    def test_an_unreachable_run_is_a_typed_rejection_not_a_crash(self) -> None:
+        def unreachable(run_id: int):
+            raise OSError("no such run")
+
+        result = corpus_dispatch.admit_evidence_artifact_v1(
+            424242, "verification-evidence-arb", unreachable
+        )
+        self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
+        self.assertEqual(result.reason, corpus.ShardCorpusReasonV1.FOREIGN_INPUT)
+
+    def test_a_run_carrying_the_artifact_is_admitted(self) -> None:
+        result = corpus_dispatch.admit_evidence_artifact_v1(
+            424242,
+            "verification-evidence-arb",
+            lambda run_id: ("verification-evidence-arb", "verification-evidence-mpfi"),
+        )
+        self.assertIsNone(result)
+
+    def test_the_observer_is_asked_about_the_named_run(self) -> None:
+        # Anti-vacuity: an observer that ignores its argument would let a
+        # dispatch bind one run while its artifacts were read from another.
+        seen: list[int] = []
+
+        def observer(run_id: int):
+            seen.append(run_id)
+            return ("verification-evidence-arb",)
+
+        corpus_dispatch.admit_evidence_artifact_v1(
+            99, "verification-evidence-arb", observer
+        )
+        self.assertEqual(seen, [99])
+
+
 if __name__ == "__main__":
     unittest.main()

From b16f217bb642b76fcaac7ffb94ebf1da37be84a8 Mon Sep 17 00:00:00 2001
From: Claude Code <daniilerosov12@gmail.com>
Date: Thu, 6 Aug 2026 18:06:23 +0300
Subject: [PATCH 2/7] Proof: the gate must refuse what cannot be downloaded,
 and be wired in provably
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Две независимые линзы ревью дали FAIL. Находки закрыты; две из них означали,
что гейт лишь ВЫГЛЯДЕЛ рабочим.

1. Истёкшие артефакты. Листинг GitHub включает артефакты с expired:true, скачать
   которые нельзя. Гейт брал имя не глядя, поэтому старый evidence-run проходил
   бы допуск и снова давал 256 обречённых полос — ровно тот класс, который срез
   объявляет закрытым, только через соседний путь. Теперь jq отбирает
   `select(.expired | not)`.

2. Шов не был покрыт. Четыре теста доказывали РЕШЕНИЕ допуска, но ни один не
   доказывал, что main() спрашивает. Удаление прошивки оставляло все 14 тестов
   зелёными. Добавлены четыре теста шва: отказ не диспатчит ничего, допуск
   диспатчит все 256 полос, dry run никого не спрашивает и ничего не запускает,
   отвергнутый набор команд уходит через exit 64. Саботаж подтверждает
   чувствительность: удаление прошивки красит test_a_refused_evidence_run_
   dispatches_nothing.

3. Точка инъекции была декларативной. Значение по умолчанию связывало
   gh_run_artifact_names_v1 при определении функции, поэтому подмена наблюдателя
   не действовала: шов выглядел инъектируемым и не был им. Дефект вскрыл сам
   тест шва — резолвинг перенесён в тело.

4. Отвергнутый набор команд попадал в цикл и давал нетипизированный
   TypeError: 'ShardCorpusRejectedV1' object is not iterable. Дефект
   предсуществующий, но мой guard `type(commands) is tuple` был написан ВОКРУГ
   отказа, не обработав его, и нарушал тот самый инвариант, который срез
   заявляет. Теперь типизированный exit 64.

5. Пагинация: gh api без --paginate отдаёт одну страницу; имя на второй странице
   читалось бы как отсутствующее. Добавлен --paginate. Направление и без того
   было fail-closed, но ложный отказ на валидном прогоне — тоже дефект.

6. Причина отказа теперь едет с отказом: оператор обязан отличать «нет такого
   прогона» от «нет токена», иначе диагностика инцидента требует ручного
   повтора вызова.

Проверка реальностью: 99999999 отвергается, 31089986150 и 31022375756 (оба
движка, живые артефакты) допускаются.

Локально: 18 тестов координатора, OK.
---
 proof/region/v1/corpus_dispatch.py            | 29 +++++-
 proof/region/v1/tests/test_corpus_dispatch.py | 94 +++++++++++++++++++
 2 files changed, 119 insertions(+), 4 deletions(-)

diff --git a/proof/region/v1/corpus_dispatch.py b/proof/region/v1/corpus_dispatch.py
index 5c783ab9..39bef4f4 100644
--- a/proof/region/v1/corpus_dispatch.py
+++ b/proof/region/v1/corpus_dispatch.py
@@ -200,9 +200,15 @@ def gh_run_artifact_names_v1(run_id: int) -> tuple[str, ...]:
         (
             "gh",
             "api",
+            "--paginate",
             f"repos/{{owner}}/{{repo}}/actions/runs/{run_id}/artifacts",
             "--jq",
-            ".artifacts[].name",
+            # An expired artifact is still listed but can no longer be
+            # downloaded, so naming it would let a stale evidence run through
+            # the gate and reproduce the very failure this admission exists to
+            # prevent.  Pagination matters for the same reason: a name on the
+            # second page must not read as absent.
+            ".artifacts[] | select(.expired | not) | .name",
         ),
         capture_output=True,
         text=True,
@@ -216,7 +222,7 @@ def gh_run_artifact_names_v1(run_id: int) -> tuple[str, ...]:
 def admit_evidence_artifact_v1(
     evidence_run_id: int,
     evidence_artifact: str,
-    observer: object = gh_run_artifact_names_v1,
+    observer: object | None = None,
 ) -> corpus.ShardCorpusRejectedV1 | None:
     """Refuse a dispatch whose evidence run cannot carry what it names.
 
@@ -227,14 +233,23 @@ def admit_evidence_artifact_v1(
     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
+    # a caller but invisible to anything that replaces the observer — the
+    # seam would look injectable and not be.
+    if observer is None:
+        observer = gh_run_artifact_names_v1
     try:
         names = tuple(observer(evidence_run_id))  # type: ignore[operator]
-    except Exception:
+    except Exception as error:
         # The observation is a hostile boundary: an unreachable run, a
         # missing token or a malformed reply 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.
         return corpus._reject(
             corpus.ShardCorpusReasonV1.FOREIGN_INPUT,
-            f"verification dispatch cannot observe run {evidence_run_id}",
+            f"verification dispatch cannot observe run {evidence_run_id}:"
+            f" {error!r}",
         )
     if evidence_artifact not in names:
         return corpus._reject(
@@ -309,6 +324,12 @@ 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 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)
+        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 ae6ded64..e35089ac 100644
--- a/proof/region/v1/tests/test_corpus_dispatch.py
+++ b/proof/region/v1/tests/test_corpus_dispatch.py
@@ -17,6 +17,7 @@
 import sys
 import tempfile
 import unittest
+import unittest.mock
 from pathlib import Path
 
 PROOF = Path(__file__).resolve().parents[1]
@@ -250,5 +251,98 @@ def observer(run_id: int):
         self.assertEqual(seen, [99])
 
 
+class DispatchSeamTests(unittest.TestCase):
+    """The gate has to be wired in, not merely defined.
+
+    Four unit tests over the admission prove the decision; none of them
+    prove that `main` asks.  Deleting the call would leave those green while
+    the live path dispatched blind again, so the seam gets its own contract:
+    a refusal must stop the campaign before the first invocation, and an
+    admission must let it through.
+    """
+
+    def _argv(self, out: Path) -> list[str]:
+        return [
+            "--mode",
+            "verification-dispatch",
+            "--evidence-run-id",
+            "424242",
+            "--evidence-artifact",
+            "verification-evidence-arb",
+            "--out",
+            str(out),
+        ]
+
+    def test_a_refused_evidence_run_dispatches_nothing(self) -> None:
+        launched: list[tuple[str, ...]] = []
+        with tempfile.TemporaryDirectory() as tmp:
+            with unittest.mock.patch.object(
+                corpus_dispatch, "gh_run_artifact_names_v1", lambda run_id: ()
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: launched.append(tuple(command)),
+            ):
+                code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt"))
+        self.assertEqual(code, 64)
+        self.assertEqual(launched, [])
+
+    def test_an_admitted_evidence_run_dispatches_every_lane(self) -> None:
+        launched: list[tuple[str, ...]] = []
+
+        class _Completed:
+            returncode = 0
+
+        with tempfile.TemporaryDirectory() as tmp:
+            with unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifact_names_v1",
+                lambda run_id: ("verification-evidence-arb",),
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: (
+                    launched.append(tuple(command)) or _Completed()
+                ),
+            ):
+                code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt"))
+        self.assertEqual(code, 0)
+        self.assertEqual(len(launched), 256)
+        self.assertTrue(
+            all(command[:3] == ("gh", "workflow", "run") for command in launched)
+        )
+
+    def test_a_dry_run_stays_offline_and_asks_nobody(self) -> None:
+        # The dry run is documented as offline: it must not even observe.
+        observed: list[int] = []
+        launched: list[tuple[str, ...]] = []
+        with tempfile.TemporaryDirectory() as tmp:
+            with unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifact_names_v1",
+                lambda run_id: observed.append(run_id) or (),
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: launched.append(tuple(command)),
+            ):
+                code = corpus_dispatch.main(
+                    self._argv(Path(tmp) / "out.txt") + ["--dry-run"]
+                )
+        self.assertEqual(code, 0)
+        self.assertEqual(observed, [])
+        self.assertEqual(launched, [])
+
+    def test_a_rejected_command_set_is_exit_64_not_a_type_error(self) -> None:
+        # A foreign artifact name is refused by the pure builder; that
+        # refusal must leave through the typed exit, never as a TypeError
+        # raised by iterating a rejection.
+        with tempfile.TemporaryDirectory() as tmp:
+            argv = self._argv(Path(tmp) / "out.txt")
+            argv[argv.index("verification-evidence-arb")] = "verification-evidence-foreign"
+            code = corpus_dispatch.main(argv + ["--dry-run"])
+        self.assertEqual(code, 64)
+
+
 if __name__ == "__main__":
     unittest.main()

From 546ff045343b955cd416066d045256f89efd8b70 Mon Sep 17 00:00:00 2001
From: Claude Code <daniilerosov12@gmail.com>
Date: Thu, 6 Aug 2026 18:31:43 +0300
Subject: [PATCH 3/7] Proof: pin the coordinate the incident turned on, and put
 the expiry rule where a test can reach it
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Пере-ревью финального состояния. Два дефекта, оба доказаны мутацией, и оба
означали, что защита выглядит рабочей и не является ею.

1. Тесты шва не закрепляли КООРДИНАТУ RUN_ID. Оба подменённых наблюдателя
   игнорировали аргумент, поэтому мутант, зашивающий чужой id в вызов допуска
   (admit_evidence_artifact_v1(1, args.evidence_artifact)), выживал при всех 18
   зелёных тестах. Координата артефакта была закреплена, а run_id — тот самый,
   на котором и произошёл инцидент с 99999999, — нет. Живой сценарий: гейт
   наблюдает прогон A и допускает, а 256 полос диспатчатся связанными с
   прогоном B. Наблюдатели шва сделаны чувствительными к run_id; мутант гибнет.

2. Правило истёкших артефактов жило в jq-выражении, то есть за сетью, и не было
   покрыто ничем: удаление select(.expired | not) оставляло все 18 тестов
   зелёными. Наблюдение теперь только НАБЛЮДАЕТ и возвращает пары (имя,
   expired), а решение, что считается свидетельством, применяется в чистом
   ядре, где до него дотягивается тест. Два теста: истёкший артефакт — отказ;
   живой рядом с истёкшим — допуск (иначе правило выродилось бы в сплошной
   отказ). Мутант «убрать фильтр» гибнет.

3. Причина отказа несла repr(error), а CalledProcessError.__repr__ отбрасывает
   stderr — ровно тот текст, который отличает «нет такого прогона» от «нет
   токена». Комментарий обещал различимость, код её не давал. Теперь detail
   несёт stderr: живая проверка даёт «cannot observe run 99999999: gh: Not
   Found (HTTP 404)».

Проверка реальностью: 99999999 отвергается с причиной, 31089986150 допускается,
наблюдение возвращает обе пары с expired=False.

Локально: 20 тестов координатора, OK.
---
 proof/region/v1/corpus_dispatch.py            | 44 ++++++++++------
 proof/region/v1/tests/test_corpus_dispatch.py | 52 ++++++++++++++++---
 2 files changed, 72 insertions(+), 24 deletions(-)

diff --git a/proof/region/v1/corpus_dispatch.py b/proof/region/v1/corpus_dispatch.py
index 39bef4f4..1926bab4 100644
--- a/proof/region/v1/corpus_dispatch.py
+++ b/proof/region/v1/corpus_dispatch.py
@@ -188,12 +188,13 @@ def verification_dispatch_commands_v1(
     )
 
 
-def gh_run_artifact_names_v1(run_id: int) -> tuple[str, ...]:
-    """Artifact names the given workflow run carries, read through `gh`.
+def gh_run_artifacts_v1(run_id: int) -> tuple[tuple[str, bool], ...]:
+    """Every artifact the run lists, as `(name, expired)` pairs.
 
-    The one impure boundary of this admission: it observes GitHub.  It is
-    injected into `admit_evidence_artifact_v1` so the decision itself stays
-    testable without a network.
+    The one impure boundary of this admission: it observes GitHub and reports
+    what it saw, without deciding anything.  Which of those artifacts count is
+    a rule, so it lives in `admit_evidence_artifact_v1` where a test can reach
+    it — filtering here would put the rule behind the network.
     """
 
     completed = subprocess.run(
@@ -203,20 +204,19 @@ def gh_run_artifact_names_v1(run_id: int) -> tuple[str, ...]:
             "--paginate",
             f"repos/{{owner}}/{{repo}}/actions/runs/{run_id}/artifacts",
             "--jq",
-            # An expired artifact is still listed but can no longer be
-            # downloaded, so naming it would let a stale evidence run through
-            # the gate and reproduce the very failure this admission exists to
-            # prevent.  Pagination matters for the same reason: a name on the
-            # second page must not read as absent.
-            ".artifacts[] | select(.expired | not) | .name",
+            ".artifacts[] | [.name, (.expired // false)] | @tsv",
         ),
         capture_output=True,
         text=True,
         check=True,
     )
-    return tuple(
-        line.strip() for line in completed.stdout.splitlines() if line.strip()
-    )
+    observed: list[tuple[str, bool]] = []
+    for line in completed.stdout.splitlines():
+        if not line.strip():
+            continue
+        name, _, expired = line.partition("	")
+        observed.append((name.strip(), expired.strip().lower() == "true"))
+    return tuple(observed)
 
 
 def admit_evidence_artifact_v1(
@@ -238,18 +238,28 @@ def admit_evidence_artifact_v1(
     # a caller but invisible to anything that replaces the observer — the
     # seam would look injectable and not be.
     if observer is None:
-        observer = gh_run_artifact_names_v1
+        observer = gh_run_artifacts_v1
     try:
-        names = tuple(observer(evidence_run_id))  # type: ignore[operator]
+        observed = tuple(observer(evidence_run_id))  # type: ignore[operator]
+        # An expired artifact is still listed but can no longer be
+        # downloaded, so naming it would admit a stale evidence run and
+        # reproduce the very failure this admission exists to prevent.  The
+        # rule is applied here, not in the query, so a test can prove it.
+        names = tuple(name for name, expired in observed if not expired)
     except Exception as error:
         # The observation is a hostile boundary: an unreachable run, a
         # missing token or a malformed reply 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: "Not Found" against "no token"
+        # against a network failure. Without it the operator debugs the wrong
+        # axis, which is exactly what the refusal is supposed to prevent.
+        cause = getattr(error, "stderr", None) or repr(error)
         return corpus._reject(
             corpus.ShardCorpusReasonV1.FOREIGN_INPUT,
             f"verification dispatch cannot observe run {evidence_run_id}:"
-            f" {error!r}",
+            f" {str(cause).strip()}",
         )
     if evidence_artifact not in names:
         return corpus._reject(
diff --git a/proof/region/v1/tests/test_corpus_dispatch.py b/proof/region/v1/tests/test_corpus_dispatch.py
index e35089ac..ab3dd8c2 100644
--- a/proof/region/v1/tests/test_corpus_dispatch.py
+++ b/proof/region/v1/tests/test_corpus_dispatch.py
@@ -213,7 +213,7 @@ def test_a_run_without_the_artifact_is_a_typed_rejection(self) -> None:
         result = corpus_dispatch.admit_evidence_artifact_v1(
             424242,
             "verification-evidence-arb",
-            lambda run_id: ("verification-evidence-mpfi",),
+            lambda run_id: (("verification-evidence-mpfi", False),),
         )
         self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
         self.assertEqual(result.reason, corpus.ShardCorpusReasonV1.FOREIGN_INPUT)
@@ -232,7 +232,35 @@ def test_a_run_carrying_the_artifact_is_admitted(self) -> None:
         result = corpus_dispatch.admit_evidence_artifact_v1(
             424242,
             "verification-evidence-arb",
-            lambda run_id: ("verification-evidence-arb", "verification-evidence-mpfi"),
+            lambda run_id: (
+                ("verification-evidence-arb", False),
+                ("verification-evidence-mpfi", False),
+            ),
+        )
+        self.assertIsNone(result)
+
+    def test_an_expired_artifact_is_not_evidence(self) -> None:
+        # An expired artifact is still listed by GitHub but can no longer be
+        # downloaded, so admitting it would send every lane to a download
+        # that cannot succeed — the incident class, through the stale-run
+        # path instead of the missing-run one.
+        result = corpus_dispatch.admit_evidence_artifact_v1(
+            424242,
+            "verification-evidence-arb",
+            lambda run_id: (("verification-evidence-arb", True),),
+        )
+        self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
+        self.assertEqual(result.reason, corpus.ShardCorpusReasonV1.FOREIGN_INPUT)
+
+    def test_a_live_artifact_beside_an_expired_one_is_admitted(self) -> None:
+        # Anti-vacuity for the rule above: expiry must filter, not blanket-refuse.
+        result = corpus_dispatch.admit_evidence_artifact_v1(
+            424242,
+            "verification-evidence-arb",
+            lambda run_id: (
+                ("verification-evidence-mpfi", True),
+                ("verification-evidence-arb", False),
+            ),
         )
         self.assertIsNone(result)
 
@@ -243,7 +271,7 @@ def test_the_observer_is_asked_about_the_named_run(self) -> None:
 
         def observer(run_id: int):
             seen.append(run_id)
-            return ("verification-evidence-arb",)
+            return (("verification-evidence-arb", False),)
 
         corpus_dispatch.admit_evidence_artifact_v1(
             99, "verification-evidence-arb", observer
@@ -277,7 +305,11 @@ def test_a_refused_evidence_run_dispatches_nothing(self) -> None:
         launched: list[tuple[str, ...]] = []
         with tempfile.TemporaryDirectory() as tmp:
             with unittest.mock.patch.object(
-                corpus_dispatch, "gh_run_artifact_names_v1", lambda run_id: ()
+                corpus_dispatch,
+                "gh_run_artifacts_v1",
+                lambda run_id: (("verification-evidence-arb", False),)
+                if run_id == 999
+                else (),
             ), unittest.mock.patch.object(
                 corpus_dispatch.subprocess,
                 "run",
@@ -296,8 +328,14 @@ class _Completed:
         with tempfile.TemporaryDirectory() as tmp:
             with unittest.mock.patch.object(
                 corpus_dispatch,
-                "gh_run_artifact_names_v1",
-                lambda run_id: ("verification-evidence-arb",),
+                "gh_run_artifacts_v1",
+                # Sensitive to the run id on purpose: an observer that ignores
+                # it lets a mutant hardcode the id at the admission call site
+                # and still pass — and the run id is the exact coordinate the
+                # 99999999 incident turned on.
+                lambda run_id: (("verification-evidence-arb", False),)
+                if run_id == 424242
+                else (),
             ), unittest.mock.patch.object(
                 corpus_dispatch.subprocess,
                 "run",
@@ -319,7 +357,7 @@ def test_a_dry_run_stays_offline_and_asks_nobody(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             with unittest.mock.patch.object(
                 corpus_dispatch,
-                "gh_run_artifact_names_v1",
+                "gh_run_artifacts_v1",
                 lambda run_id: observed.append(run_id) or (),
             ), unittest.mock.patch.object(
                 corpus_dispatch.subprocess,

From e45b74682cdcfc6ee74a051db4e71bc774b37059 Mon Sep 17 00:00:00 2001
From: Claude Code <daniilerosov12@gmail.com>
Date: Thu, 6 Aug 2026 21:46:46 +0300
Subject: [PATCH 4/7] Proof: make the safe run the default, and test the rules
 that were only claimed
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two independent reviews of the final state — one of them running actual
mutants — showed this slice narrowed the incident class without closing it,
and that its own commit message overstated what was pinned.

THE POLARITY.  The incident was a forgotten `--dry-run`, and the fix left the
dangerous run as the default: safety still meant remembering a flag.  Printing
is now what happens, `--execute` is the opt-in, and it has to name the
campaign's size with `--expect-lanes`.  Swapping the two widths is a realistic
slip — `--lane-width 256 --shard-width 4` is 65536 runs — and the only cheap
moment to notice is before the first invocation.

THE ARTIFACT COORDINATE.  A mutant hardcoding `EVIDENCE_ARTIFACTS_V1[0]` at
the admission call site survived all twenty tests: both seam tests always
passed the arb name, so pinning one engine was indistinguishable.  The
producer runs its two engines as independent jobs, so a run really can carry
one artifact and not the other, and admitting the wrong one buys 256 doomed
lanes — the incident down a second axis.  The previous message claimed this
coordinate was pinned; it was pinned where commands are built, not where the
admission asks.

THE EXPIRY RULE.  Moving the filter out of the query put the decision where a
test could reach it and left the decoding that produces it behind the network.
Two mutants survived: one that never marks anything expired, one that drops
the field from the query.  `parse_artifact_listing_v1` is pure now and refuses
a listing that is not the two-column form it asked for — the old parser read
an unrecognised line as "live", which admits exactly the stale evidence run
this gate exists to refuse.  The repository currently holds 127 expired
artifacts, so the path is not hypothetical.

A campaign that dies mid-flight now reports how many lanes it launched and
returns the typed exit instead of a traceback, so the retry can resume rather
than duplicate.

Every repair is proven by the mutant it kills, each failing exactly one test:
the hardcoded artifact, the never-expired listing, the field dropped from the
query, and `repr(error)` in place of the stderr that tells "no such run" from
"no token".

Verified: 276 local tests, 7 errors — the exact Windows baseline.

Left out deliberately, and not hidden: the admission still checks that the
named artifact exists, not that it carries `job.bin` and the comparator
bundle, and not which workflow or commit produced it.  Those are a different
class — supply-chain provenance rather than operator error — and want their
own slice.
---
 proof/region/v1/corpus_dispatch.py            |  85 +++++++--
 proof/region/v1/tests/test_corpus_dispatch.py | 174 +++++++++++++++++-
 .../v1/tests/test_verification_dispatch.py    |   3 -
 3 files changed, 237 insertions(+), 25 deletions(-)

diff --git a/proof/region/v1/corpus_dispatch.py b/proof/region/v1/corpus_dispatch.py
index 1926bab4..17ce02d3 100644
--- a/proof/region/v1/corpus_dispatch.py
+++ b/proof/region/v1/corpus_dispatch.py
@@ -7,8 +7,10 @@
 cover [0, 2^24) exactly, with every seam landing on the packing alignment
 and on a shard boundary — and turns it into one `gh workflow run` invocation
 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.
+aligned cover are typed rejections before any dispatch exists.  Printing the
+commands is the default and dispatching is the opt-in `--execute`, which has
+to name the campaign's size: the incident this coordinator exists to prevent
+was a forgotten flag turning one mistake into 133 runs.
 """
 
 from __future__ import annotations
@@ -210,12 +212,36 @@ def gh_run_artifacts_v1(run_id: int) -> tuple[tuple[str, bool], ...]:
         text=True,
         check=True,
     )
+    return parse_artifact_listing_v1(completed.stdout)
+
+
+def parse_artifact_listing_v1(stdout: str) -> tuple[tuple[str, bool], ...]:
+    """Decode the artifact listing wire form into names and expiry.
+
+    Reading `expired` is a rule like any other, so it lives here rather than
+    behind the network where nothing can reach it.  A listing that does not
+    look like the two-column form it is asked for raises: an unrecognised
+    line silently read as "live" would admit exactly the stale evidence run
+    the admission exists to refuse.
+    """
+
     observed: list[tuple[str, bool]] = []
-    for line in completed.stdout.splitlines():
+    for number, line in enumerate(stdout.splitlines(), 1):
         if not line.strip():
+            # Blank separators carry no record; anything with content has to
+            # be exactly the requested shape.
             continue
-        name, _, expired = line.partition("	")
-        observed.append((name.strip(), expired.strip().lower() == "true"))
+        fields = line.split("\t")
+        if len(fields) != 2:
+            raise ValueError(f"artifact listing line {number} is not two columns")
+        name, expired = fields[0].strip(), fields[1].strip().lower()
+        if not name:
+            raise ValueError(f"artifact listing line {number} has no name")
+        if expired not in ("true", "false"):
+            raise ValueError(
+                f"artifact listing line {number} has no boolean expiry: {expired!r}"
+            )
+        observed.append((name, expired == "true"))
     return tuple(observed)
 
 
@@ -280,9 +306,18 @@ def main(argv: list[str] | None = None) -> int:
     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("--dry-run", action="store_true")
+    # Printing is the default and dispatching is the opt-in.  The incident
+    # that made this coordinator dangerous was a forgotten `--dry-run`: a
+    # polarity where the safe run is the one you have to remember turns one
+    # missing flag into hundreds of runs.  `--expect-lanes` makes the operator
+    # state the scale before it happens, so a mistyped width is refused
+    # instead of dispatched — swapping the two widths is a realistic slip and
+    # silently means tens of thousands of jobs.
+    parser.add_argument("--execute", action="store_true")
+    parser.add_argument("--expect-lanes", type=int, default=None)
     parser.add_argument("--out", type=Path, required=True)
     args = parser.parse_args(argv)
+    dispatching = args.execute and args.mode != "plan"
 
     plan = lane_plan_v1(args.lane_width, args.shard_width)
     if type(plan) is not tuple:
@@ -323,7 +358,7 @@ 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:
+        if type(commands) is tuple and dispatching:
             # Fail closed before the first dispatch: a dry run stays offline
             # by contract, so the observation belongs to the live path only.
             refusal = admit_evidence_artifact_v1(
@@ -340,13 +375,37 @@ def main(argv: list[str] | None = None) -> int:
         # TypeError at the boundary it is supposed to guard.
         print(f"lane dispatch rejected: {commands!r}", file=sys.stderr)
         return 64
+    if not dispatching:
+        for command in commands:
+            print(" ".join(command))
+        return 0
+    if args.expect_lanes != len(commands):
+        # The operator has to name the scale, and reality has to agree.  A
+        # width typed one token wrong produces a different campaign, and the
+        # only cheap moment to notice is before the first invocation.
+        print(
+            f"dispatch refused: --expect-lanes={args.expect_lanes} but the plan"
+            f" has {len(commands)} lanes",
+            file=sys.stderr,
+        )
+        return 64
+    print(f"dispatching {len(commands)} lanes", file=sys.stderr)
+    launched = 0
     for command in commands:
-        rendered = " ".join(command)
-        if args.dry_run:
-            print(rendered)
-            continue
-        subprocess.run(command, check=True)
-        print(rendered)
+        try:
+            subprocess.run(command, check=True)
+        except subprocess.CalledProcessError as error:
+            # A campaign that dies mid-flight leaves runs already created.
+            # Reporting where it stopped is what makes the retry resumable
+            # instead of a duplicate of everything already dispatched.
+            print(
+                f"dispatch stopped after {launched} of {len(commands)} lanes:"
+                f" {str(getattr(error, 'stderr', None) or error).strip()}",
+                file=sys.stderr,
+            )
+            return 64
+        launched += 1
+        print(" ".join(command))
     return 0
 
 
diff --git a/proof/region/v1/tests/test_corpus_dispatch.py b/proof/region/v1/tests/test_corpus_dispatch.py
index ab3dd8c2..98e7c072 100644
--- a/proof/region/v1/tests/test_corpus_dispatch.py
+++ b/proof/region/v1/tests/test_corpus_dispatch.py
@@ -6,12 +6,14 @@
 plan whose windows cover [0, 2^24) exactly — contiguous, packing-aligned,
 no gaps, no overlaps — and must turn that plan into one `gh workflow run`
 invocation per lane.  Any width that cannot produce an exact aligned cover
-is a typed rejection before any dispatch exists, and a dry run must emit
-the dispatch commands without touching the network.
+is a typed rejection before any dispatch exists.  Printing the commands is
+the default and dispatching is the opt-in: the incident this coordinator
+exists to prevent was a forgotten flag.
 """
 
 from __future__ import annotations
 
+import inspect
 import json
 import subprocess
 import sys
@@ -141,7 +143,6 @@ def boom(*args: object, **kwargs: object) -> None:
                         "1024",
                         "--shard-width",
                         "256",
-                        "--dry-run",
                         "--out",
                         out,
                     ]
@@ -188,7 +189,7 @@ def test_invalid_widths_exit_64_before_any_dispatch(self) -> None:
                 ["--mode", "plan", "--lane-width", "65536", "--shard-width",
                  "12288", "--out", out],
                 ["--mode", "dispatch", "--lane-width", "0", "--shard-width",
-                 "256", "--dry-run", "--out", out],
+                 "256", "--out", out],
             )
             for argv in invalid:
                 self.assertEqual(corpus_dispatch.main(argv), 64, argv)
@@ -279,6 +280,80 @@ def observer(run_id: int):
         self.assertEqual(seen, [99])
 
 
+class ArtifactListingWireTests(unittest.TestCase):
+    """Reading `expired` off the wire is a rule, and rules get tests.
+
+    Moving the expiry filter out of the query put the decision where a test
+    can reach it, but the decoding that produces `expired` stayed behind the
+    network — so a listing whose shape drifted read as "live" and admitted a
+    stale run.  The captured form below is what `gh` actually emits for the
+    query this module sends.
+    """
+
+    CAPTURED_V1 = (
+        "verification-evidence-arb\tfalse\n"
+        "verification-evidence-mpfi\tfalse\n"
+        "corpus-lane-0-65536\ttrue\n"
+    )
+
+    def test_the_captured_wire_form_decodes_to_names_and_expiry(self) -> None:
+        self.assertEqual(
+            corpus_dispatch.parse_artifact_listing_v1(self.CAPTURED_V1),
+            (
+                ("verification-evidence-arb", False),
+                ("verification-evidence-mpfi", False),
+                ("corpus-lane-0-65536", True),
+            ),
+        )
+
+    def test_an_unrecognised_line_refuses_instead_of_reading_as_live(self) -> None:
+        for hostile in (
+            "verification-evidence-arb\n",
+            "verification-evidence-arb\tfalse\textra\n",
+            "\tfalse\n",
+            "verification-evidence-arb\tno\n",
+            "verification-evidence-arb\t\n",
+            "verification-evidence-arb\t1\n",
+        ):
+            with self.subTest(hostile=hostile):
+                with self.assertRaises(ValueError):
+                    corpus_dispatch.parse_artifact_listing_v1(hostile)
+
+    def test_an_unreadable_listing_becomes_a_typed_refusal(self) -> None:
+        # The decode raises; the admission is what turns that into a refusal,
+        # so the two are proven together rather than separately assumed.
+        def observer(_run_id: int) -> tuple[tuple[str, bool], ...]:
+            return corpus_dispatch.parse_artifact_listing_v1("garbage\n")
+
+        result = corpus_dispatch.admit_evidence_artifact_v1(
+            1, "verification-evidence-arb", observer
+        )
+        self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
+
+    def test_the_refusal_carries_the_text_that_tells_causes_apart(self) -> None:
+        # `repr` of a CalledProcessError drops stderr, and stderr is the only
+        # place "no such run" reads differently from "no token". Without it
+        # the operator debugs the wrong axis — which is what the refusal is
+        # supposed to prevent.
+        def observer(_run_id: int) -> tuple[tuple[str, bool], ...]:
+            raise subprocess.CalledProcessError(
+                1, "gh", stderr="gh: Not Found (HTTP 404)"
+            )
+
+        result = corpus_dispatch.admit_evidence_artifact_v1(
+            99999999, "verification-evidence-arb", observer
+        )
+        self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
+        self.assertIn("Not Found", result.detail)
+
+    def test_the_query_stays_paginated(self) -> None:
+        # A run carrying more than one page of artifacts would otherwise lose
+        # the evidence name and refuse a healthy campaign.
+        source = inspect.getsource(corpus_dispatch.gh_run_artifacts_v1)
+        self.assertIn("--paginate", source)
+        self.assertIn("(.expired // false)", source)
+
+
 class DispatchSeamTests(unittest.TestCase):
     """The gate has to be wired in, not merely defined.
 
@@ -289,8 +364,8 @@ class DispatchSeamTests(unittest.TestCase):
     admission must let it through.
     """
 
-    def _argv(self, out: Path) -> list[str]:
-        return [
+    def _argv(self, out: Path, *, live: bool = True) -> list[str]:
+        argv = [
             "--mode",
             "verification-dispatch",
             "--evidence-run-id",
@@ -300,6 +375,9 @@ def _argv(self, out: Path) -> list[str]:
             "--out",
             str(out),
         ]
+        # Dispatching is the opt-in, and the scale is declared with it: the
+        # seam only exists on the live path.
+        return [*argv, "--execute", "--expect-lanes", "256"] if live else argv
 
     def test_a_refused_evidence_run_dispatches_nothing(self) -> None:
         launched: list[tuple[str, ...]] = []
@@ -350,6 +428,84 @@ class _Completed:
             all(command[:3] == ("gh", "workflow", "run") for command in launched)
         )
 
+    def test_the_admission_asks_about_the_artifact_the_operator_named(self) -> None:
+        # Mirror of the run-id pin: an observer blind to the artifact lets a
+        # mutant hardcode one engine's name at the call site and stay green.
+        # The producer runs the two engines as independent jobs, so a run
+        # really can carry one artifact and not the other — and admitting the
+        # wrong one buys 256 doomed lanes, which is the incident again.
+        launched: list[tuple[str, ...]] = []
+        argv = [
+            "--mode",
+            "verification-dispatch",
+            "--evidence-run-id",
+            "424242",
+            "--evidence-artifact",
+            "verification-evidence-mpfi",
+            "--out",
+        ]
+        with tempfile.TemporaryDirectory() as tmp:
+            with unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifacts_v1",
+                lambda run_id: (("verification-evidence-arb", False),),
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: launched.append(tuple(command)),
+            ):
+                code = corpus_dispatch.main(
+                    [
+                        *argv,
+                        str(Path(tmp) / "out.txt"),
+                        "--execute",
+                        "--expect-lanes",
+                        "256",
+                    ]
+                )
+        self.assertEqual(code, 64)
+        self.assertEqual(launched, [])
+
+    def test_a_campaign_of_an_unexpected_size_is_refused_before_it_starts(
+        self,
+    ) -> None:
+        # Swapping the two widths is a realistic slip and silently means a
+        # different campaign; the operator declares the scale so reality can
+        # disagree out loud.
+        launched: list[tuple[str, ...]] = []
+        with tempfile.TemporaryDirectory() as tmp:
+            with unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifacts_v1",
+                lambda run_id: (("verification-evidence-arb", False),),
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: launched.append(tuple(command)),
+            ):
+                argv = self._argv(Path(tmp) / "out.txt", live=False)
+                code = corpus_dispatch.main(
+                    [*argv, "--execute", "--expect-lanes", "16"]
+                )
+        self.assertEqual(code, 64)
+        self.assertEqual(launched, [])
+
+    def test_without_execute_the_campaign_only_prints(self) -> None:
+        # The polarity is the guard: the incident was a forgotten flag, so
+        # forgetting one now costs nothing.
+        launched: list[tuple[str, ...]] = []
+        with tempfile.TemporaryDirectory() as tmp:
+            with unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: launched.append(tuple(command)),
+            ):
+                code = corpus_dispatch.main(
+                    self._argv(Path(tmp) / "out.txt", live=False)
+                )
+        self.assertEqual(code, 0)
+        self.assertEqual(launched, [])
+
     def test_a_dry_run_stays_offline_and_asks_nobody(self) -> None:
         # The dry run is documented as offline: it must not even observe.
         observed: list[int] = []
@@ -365,7 +521,7 @@ def test_a_dry_run_stays_offline_and_asks_nobody(self) -> None:
                 lambda command, **kwargs: launched.append(tuple(command)),
             ):
                 code = corpus_dispatch.main(
-                    self._argv(Path(tmp) / "out.txt") + ["--dry-run"]
+                    self._argv(Path(tmp) / "out.txt", live=False)
                 )
         self.assertEqual(code, 0)
         self.assertEqual(observed, [])
@@ -376,9 +532,9 @@ def test_a_rejected_command_set_is_exit_64_not_a_type_error(self) -> None:
         # refusal must leave through the typed exit, never as a TypeError
         # raised by iterating a rejection.
         with tempfile.TemporaryDirectory() as tmp:
-            argv = self._argv(Path(tmp) / "out.txt")
+            argv = self._argv(Path(tmp) / "out.txt", live=False)
             argv[argv.index("verification-evidence-arb")] = "verification-evidence-foreign"
-            code = corpus_dispatch.main(argv + ["--dry-run"])
+            code = corpus_dispatch.main(argv)
         self.assertEqual(code, 64)
 
 
diff --git a/proof/region/v1/tests/test_verification_dispatch.py b/proof/region/v1/tests/test_verification_dispatch.py
index 8cc7dfc6..c6961d6a 100644
--- a/proof/region/v1/tests/test_verification_dispatch.py
+++ b/proof/region/v1/tests/test_verification_dispatch.py
@@ -161,7 +161,6 @@ def test_dry_run_prints_one_command_per_lane(self) -> None:
                         "31000000001",
                         "--evidence-artifact",
                         ARB_ARTIFACT,
-                        "--dry-run",
                         "--out",
                         str(Path(tmp) / "out"),
                     ]
@@ -182,7 +181,6 @@ def test_missing_evidence_run_id_exits_64(self) -> None:
                     "verification-dispatch",
                     "--evidence-artifact",
                     ARB_ARTIFACT,
-                    "--dry-run",
                     "--out",
                     str(Path(tmp) / "out"),
                 ]
@@ -197,7 +195,6 @@ def test_missing_evidence_artifact_exits_64(self) -> None:
                     "verification-dispatch",
                     "--evidence-run-id",
                     "31000000001",
-                    "--dry-run",
                     "--out",
                     str(Path(tmp) / "out"),
                 ]

From ebb08dd611aba1df2f3e7a9b8b8b9705e1c202f5 Mon Sep 17 00:00:00 2001
From: Claude Code <daniilerosov12@gmail.com>
Date: Thu, 6 Aug 2026 22:12:16 +0300
Subject: [PATCH 5/7] Proof: pin the two new handles, and the coordinate whose
 test claimed it was pinned
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A third review ran fifteen mutants against the previous commit; nine survived,
and all but one lived in the code that commit added.  The polarity change and
the partial-failure report were shipped without a single test that could hold
them.

THE RUN ID.  All three seam tests take the run id from one helper that
hardcodes 424242, so a call site hardcoding that same number was
indistinguishable from one that reads the argument — while the test's own
comment claimed the opposite in as many words.  The mirror axis was fixed last
round and this one left standing.  A seam test now names a different run and
records what the observer was actually asked.

THE DECLARED SCALE.  Making `--expect-lanes` optional survived every test,
because falling through to the mismatch comparison refuses too — with a
message that tells the operator nothing.  It has its own refusal now, and the
test reads the message rather than only the exit code: the two differ exactly
where the operator does.

THE PARTIAL FAILURE.  Deleting the whole handler, returning 0 mid-campaign, or
moving the counter before the call all stayed green.  The count is the point —
a retry that cannot tell three from zero duplicates everything already
dispatched — so the test now reads "after 3 of 256" out of the report.  The
handler also catches OSError: a missing `gh` abandoned a campaign with a
traceback and no count.

ORDER.  The authenticated observation ran before the free local check, so a
mistyped width spent an API call and answered with a diagnosis about the
evidence run.  The cheap refusals come first now, which is what the comment
claimed all along.

`--mode plan --execute` was a silent no-op: a flag with live semantics that is
quietly ignored is worse than one that refuses, because the operator believes
a campaign started.

`--paginate` was checked by reading the function's source, which a mutant that
drops the flag from argv and leaves the word in a comment passes untouched.
The check is behavioural now — it asserts the argv actually built, plus
`check` and `capture_output`, which the source read never saw.

Names that referred to the removed flag are gone; only the comment recounting
the incident still says `--dry-run`, which is history and stays.

Every repair is proven by the mutant it kills, each failing exactly one test:
the hardcoded run id, the optional scale, the vanished counter, `return 0`
mid-campaign, the ignored `--execute` in plan mode, and `--paginate` demoted
to a comment.

Verified: 282 local tests, 7 errors — the exact Windows baseline.
---
 proof/region/v1/corpus_dispatch.py            |  57 ++++--
 proof/region/v1/tests/test_corpus_dispatch.py | 176 +++++++++++++++++-
 .../v1/tests/test_verification_dispatch.py    |   2 +-
 3 files changed, 209 insertions(+), 26 deletions(-)

diff --git a/proof/region/v1/corpus_dispatch.py b/proof/region/v1/corpus_dispatch.py
index 17ce02d3..be26e514 100644
--- a/proof/region/v1/corpus_dispatch.py
+++ b/proof/region/v1/corpus_dispatch.py
@@ -317,7 +317,13 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--expect-lanes", type=int, default=None)
     parser.add_argument("--out", type=Path, required=True)
     args = parser.parse_args(argv)
-    dispatching = args.execute and args.mode != "plan"
+    if args.execute and args.mode == "plan":
+        # A flag with live semantics must never be silently ignored: an
+        # operator who believes a campaign started is worse off than one who
+        # is told it did not.
+        print("plan mode cannot dispatch: drop --execute", file=sys.stderr)
+        return 64
+    dispatching = args.execute
 
     plan = lane_plan_v1(args.lane_width, args.shard_width)
     if type(plan) is not tuple:
@@ -358,15 +364,6 @@ def main(argv: list[str] | None = None) -> int:
             args.evidence_run_id,
             args.evidence_artifact,
         )
-        if type(commands) is tuple and dispatching:
-            # Fail closed before the first dispatch: a dry run stays offline
-            # by contract, so the observation belongs to the live path only.
-            refusal = admit_evidence_artifact_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)
     if type(commands) is not tuple:
@@ -379,28 +376,58 @@ def main(argv: list[str] | None = None) -> int:
         for command in commands:
             print(" ".join(command))
         return 0
+    if args.expect_lanes is None:
+        # Its own refusal, not a comparison against `None`: the operator who
+        # forgot the flag has to read what is missing, not a mismatch.
+        print(
+            "dispatch refused: --execute requires --expect-lanes",
+            file=sys.stderr,
+        )
+        return 64
     if args.expect_lanes != len(commands):
-        # The operator has to name the scale, and reality has to agree.  A
-        # width typed one token wrong produces a different campaign, and the
-        # only cheap moment to notice is before the first invocation.
+        # The operator names the scale and reality has to agree.  A width
+        # typed one token wrong produces a different campaign, and this is
+        # the cheap moment to notice — before anything is observed or run.
         print(
             f"dispatch refused: --expect-lanes={args.expect_lanes} but the plan"
             f" has {len(commands)} lanes",
             file=sys.stderr,
         )
         return 64
+    if args.mode == "verification-dispatch":
+        # Fail closed before the first dispatch, and only here: the printing
+        # path stays offline by contract, and this observation costs an
+        # authenticated call that a mistyped width should never spend.
+        refusal = admit_evidence_artifact_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:
         try:
             subprocess.run(command, check=True)
+        except OSError as error:
+            # Not only CalledProcessError: a missing `gh`, an exhausted file
+            # descriptor or a killed child all abandon a campaign mid-flight,
+            # and all of them must leave the same resumable report.
+            print(
+                f"dispatch stopped after {launched} of {len(commands)} lanes:"
+                f" {str(error).strip()}",
+                file=sys.stderr,
+            )
+            return 64
         except subprocess.CalledProcessError as error:
             # A campaign that dies mid-flight leaves runs already created.
             # Reporting where it stopped is what makes the retry resumable
-            # instead of a duplicate of everything already dispatched.
+            # instead of a duplicate of everything already dispatched.  The
+            # child's stderr is not captured here on purpose — it belongs on
+            # the operator's terminal — so only the exit status is quotable.
             print(
                 f"dispatch stopped after {launched} of {len(commands)} lanes:"
-                f" {str(getattr(error, 'stderr', None) or error).strip()}",
+                f" {str(error).strip()}",
                 file=sys.stderr,
             )
             return 64
diff --git a/proof/region/v1/tests/test_corpus_dispatch.py b/proof/region/v1/tests/test_corpus_dispatch.py
index 98e7c072..cfae9aa7 100644
--- a/proof/region/v1/tests/test_corpus_dispatch.py
+++ b/proof/region/v1/tests/test_corpus_dispatch.py
@@ -13,7 +13,8 @@
 
 from __future__ import annotations
 
-import inspect
+import contextlib
+import io
 import json
 import subprocess
 import sys
@@ -123,13 +124,13 @@ def test_dispatch_commands_reject_a_rejected_plan(self) -> None:
         self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
 
 
-class DryRunTests(unittest.TestCase):
-    def test_dry_run_never_touches_the_network(self) -> None:
+class DefaultIsPrintingTests(unittest.TestCase):
+    def test_the_default_run_never_touches_the_network(self) -> None:
         calls: list[list[str]] = []
 
         def boom(*args: object, **kwargs: object) -> None:
             calls.append([args, kwargs])
-            raise AssertionError("dry run must not invoke subprocess")
+            raise AssertionError("printing must not invoke subprocess")
 
         original = subprocess.run
         subprocess.run = boom  # type: ignore[assignment]
@@ -346,12 +347,36 @@ def observer(_run_id: int) -> tuple[tuple[str, bool], ...]:
         self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
         self.assertIn("Not Found", result.detail)
 
-    def test_the_query_stays_paginated(self) -> None:
+    def test_the_query_asks_for_every_page_and_for_expiry(self) -> None:
+        # Behavioural, not a look at the source: a mutant that drops the flag
+        # from argv while leaving the words in a comment reads identically.
         # A run carrying more than one page of artifacts would otherwise lose
         # the evidence name and refuse a healthy campaign.
-        source = inspect.getsource(corpus_dispatch.gh_run_artifacts_v1)
-        self.assertIn("--paginate", source)
-        self.assertIn("(.expired // false)", source)
+        seen: list[tuple[str, ...]] = []
+
+        class _Completed:
+            stdout = "verification-evidence-arb\tfalse\n"
+            returncode = 0
+
+        def record(command: tuple[str, ...], **kwargs: object) -> object:
+            seen.append(tuple(command))
+            assert kwargs.get("check") is True
+            assert kwargs.get("capture_output") is True
+            return _Completed()
+
+        with unittest.mock.patch.object(corpus_dispatch.subprocess, "run", record):
+            observed = corpus_dispatch.gh_run_artifacts_v1(31116022208)
+
+        self.assertEqual(observed, (("verification-evidence-arb", False),))
+        self.assertEqual(len(seen), 1)
+        argv = seen[0]
+        self.assertEqual(argv[:3], ("gh", "api", "--paginate"))
+        self.assertIn("repos/{owner}/{repo}/actions/runs/31116022208/artifacts", argv)
+        self.assertIn("--jq", argv)
+        self.assertIn(
+            ".artifacts[] | [.name, (.expired // false)] | @tsv",
+            argv[argv.index("--jq") + 1],
+        )
 
 
 class DispatchSeamTests(unittest.TestCase):
@@ -506,8 +531,139 @@ def test_without_execute_the_campaign_only_prints(self) -> None:
         self.assertEqual(code, 0)
         self.assertEqual(launched, [])
 
-    def test_a_dry_run_stays_offline_and_asks_nobody(self) -> None:
-        # The dry run is documented as offline: it must not even observe.
+    def test_the_admission_asks_about_the_run_the_operator_named(self) -> None:
+        # The other seam tests all pass 424242, so a call site that hardcoded
+        # that very number would be indistinguishable from one that reads the
+        # argument.  This one names a different run and records what the
+        # observer was actually asked.
+        observed: list[int] = []
+        launched: list[tuple[str, ...]] = []
+
+        class _Completed:
+            returncode = 0
+
+        with tempfile.TemporaryDirectory() as tmp:
+            argv = self._argv(Path(tmp) / "out.txt")
+            argv[argv.index("424242")] = "31116022208"
+            with unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifacts_v1",
+                lambda run_id: (
+                    observed.append(run_id)
+                    or (("verification-evidence-arb", False),)
+                ),
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: (
+                    launched.append(tuple(command)) or _Completed()
+                ),
+            ):
+                code = corpus_dispatch.main(argv)
+        self.assertEqual(code, 0)
+        self.assertEqual(observed, [31116022208])
+        self.assertEqual(len(launched), 256)
+
+    def test_execute_without_a_declared_scale_dispatches_nothing(self) -> None:
+        # The scale declaration is the guard, so it has to be required — an
+        # optional one restores the forgotten-flag class it replaced.
+        launched: list[tuple[str, ...]] = []
+        with tempfile.TemporaryDirectory() as tmp:
+            with unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifacts_v1",
+                lambda run_id: (("verification-evidence-arb", False),),
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: launched.append(tuple(command)),
+            ):
+                argv = self._argv(Path(tmp) / "out.txt", live=False)
+                errors = io.StringIO()
+                with contextlib.redirect_stderr(errors):
+                    code = corpus_dispatch.main([*argv, "--execute"])
+        self.assertEqual(code, 64)
+        self.assertEqual(launched, [])
+        # The refusal has to name what is missing: falling through to the
+        # mismatch comparison refuses too, and tells the operator nothing.
+        self.assertIn("--expect-lanes", errors.getvalue())
+        self.assertNotIn("but the plan has", errors.getvalue())
+
+    def test_a_wrong_scale_is_refused_before_the_network_is_asked(self) -> None:
+        # The cheap local check must precede the authenticated call: a
+        # mistyped width should not spend one, and the operator should read
+        # "wrong scale" rather than a diagnosis about the evidence run.
+        observed: list[int] = []
+        with tempfile.TemporaryDirectory() as tmp:
+            with unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifacts_v1",
+                lambda run_id: observed.append(run_id) or (),
+            ):
+                argv = self._argv(Path(tmp) / "out.txt", live=False)
+                code = corpus_dispatch.main(
+                    [*argv, "--execute", "--expect-lanes", "16"]
+                )
+        self.assertEqual(code, 64)
+        self.assertEqual(observed, [])
+
+    def test_plan_mode_refuses_execute_instead_of_ignoring_it(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            code = corpus_dispatch.main(
+                ["--mode", "plan", "--out", tmp, "--execute", "--expect-lanes", "256"]
+            )
+        self.assertEqual(code, 64)
+
+    def test_a_campaign_that_dies_reports_how_far_it_got(self) -> None:
+        # Without this the failure is a traceback: the operator cannot tell a
+        # retry from a duplicate of everything already dispatched.
+        launched: list[tuple[str, ...]] = []
+
+        class _Completed:
+            returncode = 0
+
+        def flaky(command: tuple[str, ...], **_kwargs: object) -> object:
+            if len(launched) == 3:
+                raise subprocess.CalledProcessError(1, list(command))
+            launched.append(tuple(command))
+            return _Completed()
+
+        with tempfile.TemporaryDirectory() as tmp:
+            with unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifacts_v1",
+                lambda run_id: (("verification-evidence-arb", False),),
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess, "run", flaky
+            ):
+                errors = io.StringIO()
+                with contextlib.redirect_stderr(errors):
+                    code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt"))
+        self.assertEqual(code, 64)
+        # Exactly the ones that succeeded, and no further attempt after.
+        self.assertEqual(len(launched), 3)
+        # The count is the whole point: a retry that cannot tell 3 from 0
+        # duplicates everything already dispatched.
+        self.assertIn("after 3 of 256", errors.getvalue())
+
+    def test_a_missing_gh_is_a_typed_stop_not_a_traceback(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            with unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifacts_v1",
+                lambda run_id: (("verification-evidence-arb", False),),
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: (_ for _ in ()).throw(
+                    FileNotFoundError("gh")
+                ),
+            ):
+                code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt"))
+        self.assertEqual(code, 64)
+
+    def test_printing_stays_offline_and_asks_nobody(self) -> None:
+        # Printing is documented as offline: it must not even observe.
         observed: list[int] = []
         launched: list[tuple[str, ...]] = []
         with tempfile.TemporaryDirectory() as tmp:
diff --git a/proof/region/v1/tests/test_verification_dispatch.py b/proof/region/v1/tests/test_verification_dispatch.py
index c6961d6a..40869e50 100644
--- a/proof/region/v1/tests/test_verification_dispatch.py
+++ b/proof/region/v1/tests/test_verification_dispatch.py
@@ -147,7 +147,7 @@ def test_foreign_evidence_artifact_is_a_typed_rejection(self) -> None:
 
 
 class VerificationDispatchCliTests(unittest.TestCase):
-    def test_dry_run_prints_one_command_per_lane(self) -> None:
+    def test_the_default_prints_one_command_per_lane(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             stdout = io.StringIO()
             with redirect_stdout(stdout):

From e18fb78553fa2d0085de8083b5b59524e091d999 Mon Sep 17 00:00:00 2001
From: Claude Code <daniilerosov12@gmail.com>
Date: Thu, 6 Aug 2026 23:06:25 +0300
Subject: [PATCH 6/7] Proof: a dispatch admits the run's origin, not only the
 artifact's name
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The evidence admission checked that a live artifact of the named name
existed inside the named run.  Three runs pass that and must not: one
produced by a different workflow, one a fork's pull request produced —
carrying a comparator bundle nobody reviewed — and one that never finished
successfully.  Each buys 256 lanes replaying something the operator did not
intend, and the reproduction is exit 0 with 256 dispatches while nothing was
ever asked about the run.

The run is now observed as a projection of five named fields (never the run
object: an unknown reply is how this project has leaked), decoded by a rule
that refuses shape drift instead of defaulting, and admitted by a pure rule
covering the workflow path, the workflow_dispatch trigger, completion with
conclusion success, and a readable head_sha.  Any failure to observe is a
typed refusal, as with the artifact.

Staleness cannot be decided here — last week's green producer run satisfies
every rule above — so the one coordinate that settles it is put in front of
the operator at the moment of admission, before the lanes start.

Twelve mutants of the new rules were run against the suite and all twelve
were killed, including a dropped seam call in `main`, a fail-open
observation, a decode that invents missing fields, and a query that fetches
the whole reply.

The artifact's *contents* are still unchecked: proving job.bin and the
comparator bundle exist requires downloading it, and that is its own slice.
---
 proof/region/v1/corpus_dispatch.py            | 209 ++++++++-
 proof/region/v1/tests/test_corpus_dispatch.py | 442 +++++++++++++++++-
 2 files changed, 634 insertions(+), 17 deletions(-)

diff --git a/proof/region/v1/corpus_dispatch.py b/proof/region/v1/corpus_dispatch.py
index be26e514..ad8459c7 100644
--- a/proof/region/v1/corpus_dispatch.py
+++ b/proof/region/v1/corpus_dispatch.py
@@ -11,6 +11,11 @@
 commands is the default and dispatching is the opt-in `--execute`, which has
 to name the campaign's size: the incident this coordinator exists to prevent
 was a forgotten flag turning one mistake into 133 runs.
+
+The verification lanes replay evidence produced by another run, so that run
+is admitted before the first dispatch: which workflow produced it, what
+triggered it, whether it succeeded, and which commit it stands on — the last
+of these cannot be decided here and is reported to the operator instead.
 """
 
 from __future__ import annotations
@@ -19,6 +24,7 @@
 import json
 import subprocess
 import sys
+from dataclasses import dataclass
 from pathlib import Path
 
 PROOF = Path(__file__).resolve().parent
@@ -38,6 +44,21 @@
     "verification-evidence-arb",
     "verification-evidence-mpfi",
 )
+# 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
+# code it was — a fork's pull request can run the same workflow and publish an
+# artifact of exactly the allowlisted name.
+EVIDENCE_WORKFLOW_PATH_V1 = ".github/workflows/full-domain-run.yml"
+EVIDENCE_RUN_EVENT_V1 = "workflow_dispatch"
+EVIDENCE_RUN_STATUS_V1 = "completed"
+EVIDENCE_RUN_CONCLUSION_V1 = "success"
+# A projection, never the whole run object: the reply carries fields this
+# module has no reason to read, and printing a reply of unknown shape is how
+# this project has leaked before.
+RUN_PROVENANCE_JQ_V1 = "[.path, .event, .status, .conclusion, .head_sha] | @tsv"
+COMMIT_SHA_LENGTH_V1 = 40
+COMMIT_SHA_ALPHABET_V1 = frozenset("0123456789abcdef")
 DEFAULT_LANE_WIDTH = 1 << 16
 DEFAULT_SHARD_WIDTH = corpus_lane.DEFAULT_SHARD_POINTS
 FULL_DOMAIN = protocol.OUTPUT_CARDINALITY_V1
@@ -245,6 +266,182 @@ def parse_artifact_listing_v1(stdout: str) -> tuple[tuple[str, bool], ...]:
     return tuple(observed)
 
 
+@dataclass(frozen=True)
+class RunProvenanceV1:
+    """The run fields the admission reads, and nothing else.
+
+    Where the evidence came from is four coordinates — which workflow built
+    it, what triggered that workflow, whether it finished successfully, and
+    which commit it stands on.  They travel as one record so no caller can
+    check two of them and forget the rest.
+    """
+
+    path: str
+    event: str
+    status: str
+    conclusion: str
+    head_sha: str
+
+    def __post_init__(self) -> None:
+        for field in (
+            self.path,
+            self.event,
+            self.status,
+            self.conclusion,
+            self.head_sha,
+        ):
+            if type(field) is not str:
+                raise TypeError("invalid run provenance field")
+
+
+def gh_run_provenance_v1(run_id: int) -> RunProvenanceV1:
+    """The run's origin as GitHub reports it, projected to the named fields.
+
+    The second impure boundary of this admission, and the same contract as
+    the first: it observes and reports, deciding nothing.  The query names
+    the fields instead of fetching the run object, because everything not
+    named here is a field this module would carry without ever reading it.
+    """
+
+    completed = subprocess.run(
+        (
+            "gh",
+            "api",
+            f"repos/{{owner}}/{{repo}}/actions/runs/{run_id}",
+            "--jq",
+            RUN_PROVENANCE_JQ_V1,
+        ),
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    return parse_run_provenance_v1(completed.stdout)
+
+
+def parse_run_provenance_v1(stdout: str) -> RunProvenanceV1:
+    """Decode one run's projected wire form into the observed record.
+
+    Shape only: which values are acceptable is `admit_run_provenance_v1`,
+    where a test can reach it.  What is refused here is a reply that is not
+    exactly one five-column record — an empty or drifted reply read as a
+    default would admit precisely the runs this observation exists to catch.
+    """
+
+    records = [line for line in stdout.splitlines() if line.strip()]
+    if len(records) != 1:
+        raise ValueError(
+            f"run provenance is not one record: {len(records)} lines"
+        )
+    fields = records[0].split("\t")
+    if len(fields) != 5:
+        raise ValueError(
+            f"run provenance record is not five columns: {len(fields)}"
+        )
+    path, event, status, conclusion, head_sha = (
+        field.strip() for field in fields
+    )
+    return RunProvenanceV1(path, event, status, conclusion, head_sha)
+
+
+def admit_run_provenance_v1(
+    provenance: object,
+) -> corpus.ShardCorpusRejectedV1 | None:
+    """Which observed run may be replayed — the whole rule, and nothing impure.
+
+    A run id and an artifact name say only that some run holds a file of the
+    right name.  Three runs pass that and must not pass this: one produced by
+    a different workflow, one a fork's pull request produced, and one that
+    never finished successfully.  Each of them sends 256 lanes to replay
+    something the operator did not intend.
+    """
+
+    if type(provenance) is not RunProvenanceV1:
+        return corpus._reject(
+            corpus.ShardCorpusReasonV1.FOREIGN_INPUT,
+            "evidence run provenance is not an observed record",
+        )
+    if provenance.path != EVIDENCE_WORKFLOW_PATH_V1:
+        # The path, not the file name: a workflow of the same basename in a
+        # foreign directory is a different producer.
+        return corpus._reject(
+            corpus.ShardCorpusReasonV1.FOREIGN_INPUT,
+            f"evidence run was produced by {provenance.path!r},"
+            f" not {EVIDENCE_WORKFLOW_PATH_V1}",
+        )
+    if provenance.event != EVIDENCE_RUN_EVENT_V1:
+        return corpus._reject(
+            corpus.ShardCorpusReasonV1.FOREIGN_INPUT,
+            f"evidence run was triggered by {provenance.event!r},"
+            f" not {EVIDENCE_RUN_EVENT_V1}",
+        )
+    if (
+        provenance.status != EVIDENCE_RUN_STATUS_V1
+        or provenance.conclusion != EVIDENCE_RUN_CONCLUSION_V1
+    ):
+        # Both halves: a run still in flight can already have uploaded one
+        # engine's artifact, and a conclusion is only final once the status
+        # says the run is.
+        return corpus._reject(
+            corpus.ShardCorpusReasonV1.FOREIGN_INPUT,
+            f"evidence run is {provenance.status!r}/{provenance.conclusion!r},"
+            f" not {EVIDENCE_RUN_STATUS_V1}/{EVIDENCE_RUN_CONCLUSION_V1}",
+        )
+    if (
+        len(provenance.head_sha) != COMMIT_SHA_LENGTH_V1
+        or not COMMIT_SHA_ALPHABET_V1.issuperset(provenance.head_sha)
+    ):
+        # The commit is what the operator checks the campaign against, so a
+        # run that carries no readable one cannot be admitted for being green:
+        # an abbreviated or absent sha is not something to paste into `git`.
+        return corpus._reject(
+            corpus.ShardCorpusReasonV1.FOREIGN_INPUT,
+            f"evidence run carries no commit sha: {provenance.head_sha!r}",
+        )
+    return None
+
+
+def admit_evidence_run_v1(
+    evidence_run_id: int,
+    observer: object | None = None,
+) -> corpus.ShardCorpusRejectedV1 | None:
+    """Refuse a dispatch whose evidence run is not the one the campaign means.
+
+    Observation and reporting only; the rule is `admit_run_provenance_v1`.
+    Any failure to observe is a refusal, never a crash and never a silent
+    proceed, on the same reasoning as the artifact admission.
+
+    An admitted run's commit goes to the operator here.  Nothing in this
+    process can tell last week's green producer run from this week's — the
+    rules above admit both — so the one coordinate that decides it is put in
+    front of the operator at the moment of admission, while 256 lanes have
+    still not started.
+    """
+
+    if observer is None:
+        observer = gh_run_provenance_v1
+    try:
+        provenance = observer(evidence_run_id)  # type: ignore[operator]
+    except Exception as error:
+        # Same hostile boundary, same reason to carry the cause: "no such
+        # run" and "no token" are indistinguishable without stderr, and
+        # `CalledProcessError.__repr__` drops it.
+        cause = getattr(error, "stderr", None) or repr(error)
+        return corpus._reject(
+            corpus.ShardCorpusReasonV1.FOREIGN_INPUT,
+            f"verification dispatch cannot observe run {evidence_run_id}:"
+            f" {str(cause).strip()}",
+        )
+    refusal = admit_run_provenance_v1(provenance)
+    if refusal is not None:
+        return refusal
+    print(
+        f"evidence run {evidence_run_id} admitted:"
+        f" head_sha={provenance.head_sha}",
+        file=sys.stderr,
+    )
+    return None
+
+
 def admit_evidence_artifact_v1(
     evidence_run_id: int,
     evidence_artifact: str,
@@ -396,8 +593,16 @@ def main(argv: list[str] | None = None) -> int:
         return 64
     if args.mode == "verification-dispatch":
         # Fail closed before the first dispatch, and only here: the printing
-        # path stays offline by contract, and this observation costs an
-        # authenticated call that a mistyped width should never spend.
+        # path stays offline by contract, and these observations cost
+        # authenticated calls that a mistyped width should never spend.
+        # Origin before contents: a run of the wrong workflow, the wrong
+        # trigger or the wrong outcome lists an artifact of exactly the right
+        # name, so asking about the artifact first would clear a run that
+        # should never have been considered.
+        refusal = admit_evidence_run_v1(args.evidence_run_id)
+        if refusal is not None:
+            print(f"verification dispatch refused: {refusal.detail}", file=sys.stderr)
+            return 64
         refusal = admit_evidence_artifact_v1(
             args.evidence_run_id, args.evidence_artifact
         )
diff --git a/proof/region/v1/tests/test_corpus_dispatch.py b/proof/region/v1/tests/test_corpus_dispatch.py
index cfae9aa7..a47386d4 100644
--- a/proof/region/v1/tests/test_corpus_dispatch.py
+++ b/proof/region/v1/tests/test_corpus_dispatch.py
@@ -379,16 +379,284 @@ def record(command: tuple[str, ...], **kwargs: object) -> object:
         )
 
 
-class DispatchSeamTests(unittest.TestCase):
-    """The gate has to be wired in, not merely defined.
+ADMITTED_SHA_V1 = "3f0f0a1b2c3d4e5f60718293a4b5c6d7e8f90a1b"
+
+
+def _admitted_provenance(**overrides: str) -> object:
+    """The run an operator is allowed to replay, with one field disturbed."""
+
+    fields = dict(
+        path=".github/workflows/full-domain-run.yml",
+        event="workflow_dispatch",
+        status="completed",
+        conclusion="success",
+        head_sha=ADMITTED_SHA_V1,
+    )
+    fields.update(overrides)
+    return corpus_dispatch.RunProvenanceV1(**fields)
+
+
+class RunProvenanceWireTests(unittest.TestCase):
+    """Observing a run means reading named fields, never a raw reply.
+
+    The artifact listing proves a name exists inside some run; it says
+    nothing about which workflow produced that run, what triggered it,
+    whether it finished, or which commit it stands on.  Asking GitHub for the
+    whole run object would drag a large reply of unknown shape through this
+    process — the project's standing hazard — so the query projects exactly
+    the fields the rule reads, and the decode refuses anything that is not
+    that shape rather than inventing a field.
+    """
+
+    CAPTURED_V1 = (
+        ".github/workflows/full-domain-run.yml\tworkflow_dispatch\tcompleted\t"
+        f"success\t{ADMITTED_SHA_V1}\n"
+    )
+
+    def test_the_captured_wire_form_decodes_to_the_named_fields(self) -> None:
+        self.assertEqual(
+            corpus_dispatch.parse_run_provenance_v1(self.CAPTURED_V1),
+            _admitted_provenance(),
+        )
+
+    def test_a_reply_that_is_not_one_five_column_record_refuses(self) -> None:
+        # An empty or drifted reply read as a default record would admit the
+        # exact runs this observation exists to refuse, so shape drift is a
+        # decode failure — which the admission turns into a typed refusal.
+        for hostile in (
+            "",
+            "\n",
+            ".github/workflows/full-domain-run.yml\tworkflow_dispatch\tcompleted"
+            f"\t{ADMITTED_SHA_V1}\n",
+            self.CAPTURED_V1.replace("\n", "\textra\n"),
+            self.CAPTURED_V1 + self.CAPTURED_V1,
+            f"{{\"path\": \".github/workflows/full-domain-run.yml\"}}\n",
+        ):
+            with self.subTest(hostile=hostile):
+                with self.assertRaises(ValueError):
+                    corpus_dispatch.parse_run_provenance_v1(hostile)
+
+    def test_the_query_projects_the_fields_and_not_the_whole_reply(self) -> None:
+        # Behavioural, not a look at the source, and load-bearing twice: the
+        # run object carries far more than this module reads, and printing an
+        # unknown reply is the known way this project leaks.
+        seen: list[tuple[str, ...]] = []
+
+        class _Completed:
+            stdout = RunProvenanceWireTests.CAPTURED_V1
+            returncode = 0
 
-    Four unit tests over the admission prove the decision; none of them
-    prove that `main` asks.  Deleting the call would leave those green while
-    the live path dispatched blind again, so the seam gets its own contract:
-    a refusal must stop the campaign before the first invocation, and an
-    admission must let it through.
+        def record(command: tuple[str, ...], **kwargs: object) -> object:
+            seen.append(tuple(command))
+            assert kwargs.get("check") is True
+            assert kwargs.get("capture_output") is True
+            return _Completed()
+
+        with unittest.mock.patch.object(corpus_dispatch.subprocess, "run", record):
+            observed = corpus_dispatch.gh_run_provenance_v1(31116022208)
+
+        self.assertEqual(observed, _admitted_provenance())
+        self.assertEqual(len(seen), 1)
+        argv = seen[0]
+        self.assertEqual(argv[:2], ("gh", "api"))
+        self.assertIn("repos/{owner}/{repo}/actions/runs/31116022208", argv)
+        self.assertIn("--jq", argv)
+        query = argv[argv.index("--jq") + 1]
+        for field in (".path", ".event", ".status", ".conclusion", ".head_sha"):
+            self.assertIn(field, query)
+        self.assertIn("@tsv", query)
+        # Exactly five selectors: a bare `.` or a dropped projection would
+        # pull the whole run object through this process.
+        self.assertEqual(query.count("."), 5)
+
+
+class RunProvenanceRuleTests(unittest.TestCase):
+    """Which observed run may be replayed is a rule, so it gets tests.
+
+    A run id plus an artifact name is not provenance.  A stale run of an old
+    commit, a run of a different workflow, and a run a fork's pull request
+    produced all list an artifact of exactly the right name, and all three
+    send 256 lanes to replay under something the operator did not intend.
     """
 
+    def test_the_producer_run_an_operator_dispatched_is_admitted(self) -> None:
+        self.assertIsNone(
+            corpus_dispatch.admit_run_provenance_v1(_admitted_provenance())
+        )
+
+    def test_a_run_of_another_workflow_is_refused(self) -> None:
+        # The path, not the file name: a workflow of the same basename in a
+        # foreign directory is a different producer.
+        for path in (
+            ".github/workflows/verification-lanes.yml",
+            ".github/workflows/ci.yml",
+            "full-domain-run.yml",
+            "vendor/.github/workflows/full-domain-run.yml",
+            "",
+        ):
+            with self.subTest(path=path):
+                result = corpus_dispatch.admit_run_provenance_v1(
+                    _admitted_provenance(path=path)
+                )
+                self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
+                self.assertEqual(
+                    result.reason, corpus.ShardCorpusReasonV1.FOREIGN_INPUT
+                )
+
+    def test_a_run_a_pull_request_produced_is_refused(self) -> None:
+        # A fork's pull request can run the producer workflow and upload an
+        # artifact of exactly the allowlisted name carrying a comparator
+        # bundle nobody reviewed.  Only an operator's own dispatch counts.
+        for event in ("pull_request", "pull_request_target", "push", "schedule", ""):
+            with self.subTest(event=event):
+                result = corpus_dispatch.admit_run_provenance_v1(
+                    _admitted_provenance(event=event)
+                )
+                self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
+
+    def test_a_run_that_did_not_finish_successfully_is_refused(self) -> None:
+        for status, conclusion in (
+            ("completed", "failure"),
+            ("completed", "cancelled"),
+            ("completed", "timed_out"),
+            # A run still in flight reports a null conclusion, which the wire
+            # form renders as an empty column.
+            ("in_progress", ""),
+            ("queued", ""),
+            # Hostile rather than observed: a finished-looking conclusion
+            # under an unfinished status must not pass on the conclusion
+            # alone.
+            ("in_progress", "success"),
+        ):
+            with self.subTest(status=status, conclusion=conclusion):
+                result = corpus_dispatch.admit_run_provenance_v1(
+                    _admitted_provenance(status=status, conclusion=conclusion)
+                )
+                self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
+
+    def test_a_run_without_a_commit_sha_is_refused(self) -> None:
+        # The sha is what the operator checks the campaign against, so a run
+        # that does not carry one cannot be admitted merely for being green.
+        for head_sha in (
+            "",
+            "3f0f0a1b",
+            ADMITTED_SHA_V1[:-1],
+            ADMITTED_SHA_V1 + "0",
+            ADMITTED_SHA_V1[:-1] + "g",
+            ADMITTED_SHA_V1.upper(),
+        ):
+            with self.subTest(head_sha=head_sha):
+                result = corpus_dispatch.admit_run_provenance_v1(
+                    _admitted_provenance(head_sha=head_sha)
+                )
+                self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
+
+    def test_a_record_that_was_never_observed_is_refused_not_crashed(self) -> None:
+        for foreign in (
+            None,
+            (
+                ".github/workflows/full-domain-run.yml",
+                "workflow_dispatch",
+                "completed",
+                "success",
+                ADMITTED_SHA_V1,
+            ),
+            {"path": ".github/workflows/full-domain-run.yml"},
+            "completed",
+            object(),
+        ):
+            with self.subTest(foreign=foreign):
+                result = corpus_dispatch.admit_run_provenance_v1(foreign)
+                self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
+
+
+class EvidenceRunAdmissionTests(unittest.TestCase):
+    """The impure half: observe, refuse on any failure, report the sha.
+
+    Nothing here decides which run is acceptable — that is the pure rule
+    above.  What is proven here is that observation failures land as typed
+    refusals carrying their cause, that the named run is the one observed,
+    and that an admitted run's commit reaches the operator.
+    """
+
+    def test_an_unreachable_run_is_a_typed_refusal_carrying_the_cause(self) -> None:
+        def observer(_run_id: int) -> object:
+            raise subprocess.CalledProcessError(
+                1, "gh", stderr="gh: Not Found (HTTP 404)"
+            )
+
+        result = corpus_dispatch.admit_evidence_run_v1(99999999, observer)
+        self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
+        self.assertIn("Not Found", result.detail)
+
+    def test_an_undecodable_reply_becomes_a_refusal_not_a_traceback(self) -> None:
+        def observer(_run_id: int) -> object:
+            return corpus_dispatch.parse_run_provenance_v1("garbage\n")
+
+        result = corpus_dispatch.admit_evidence_run_v1(1, observer)
+        self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
+
+    def test_the_observer_is_asked_about_the_named_run(self) -> None:
+        # Anti-vacuity: an observer blind to its argument would let one run be
+        # bound while another was vouched for.
+        seen: list[int] = []
+
+        def observer(run_id: int) -> object:
+            seen.append(run_id)
+            return _admitted_provenance()
+
+        errors = io.StringIO()
+        with contextlib.redirect_stderr(errors):
+            self.assertIsNone(corpus_dispatch.admit_evidence_run_v1(77, observer))
+        self.assertEqual(seen, [77])
+
+    def test_the_admitted_commit_reaches_the_operator(self) -> None:
+        # A green producer run of last week's comparator is admissible by
+        # every rule here and still wrong.  Nothing in this process can tell
+        # which commit the operator meant, so the one coordinate that decides
+        # it is put in front of them at the moment of admission.
+        errors = io.StringIO()
+        with contextlib.redirect_stderr(errors):
+            result = corpus_dispatch.admit_evidence_run_v1(
+                424242, lambda run_id: _admitted_provenance()
+            )
+        self.assertIsNone(result)
+        self.assertIn(ADMITTED_SHA_V1, errors.getvalue())
+
+    def test_a_refused_run_reports_no_commit(self) -> None:
+        # The report belongs to admission: a sha printed beside a refusal
+        # reads as a cleared campaign.
+        errors = io.StringIO()
+        with contextlib.redirect_stderr(errors):
+            result = corpus_dispatch.admit_evidence_run_v1(
+                424242, lambda run_id: _admitted_provenance(event="pull_request")
+            )
+        self.assertIs(type(result), corpus.ShardCorpusRejectedV1)
+        self.assertNotIn(ADMITTED_SHA_V1, errors.getvalue())
+
+
+class DispatchSeamTests(unittest.TestCase):
+    """The gates have to be wired in, not merely defined.
+
+    The unit tests above prove what each admission decides; none of them
+    prove that `main` asks.  Deleting either call would leave them green
+    while the live path dispatched blind again, so both seams get their own
+    contract: a refusal must stop the campaign before the first invocation,
+    an admission must let it through, and the admitted commit must reach the
+    operator before the lanes do.
+    """
+
+    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()),
+        )
+
     def _argv(self, out: Path, *, live: bool = True) -> list[str]:
         argv = [
             "--mode",
@@ -407,7 +675,7 @@ def _argv(self, out: Path, *, live: bool = True) -> list[str]:
     def test_a_refused_evidence_run_dispatches_nothing(self) -> None:
         launched: list[tuple[str, ...]] = []
         with tempfile.TemporaryDirectory() as tmp:
-            with unittest.mock.patch.object(
+            with self._provenance_patch(), unittest.mock.patch.object(
                 corpus_dispatch,
                 "gh_run_artifacts_v1",
                 lambda run_id: (("verification-evidence-arb", False),)
@@ -429,7 +697,7 @@ class _Completed:
             returncode = 0
 
         with tempfile.TemporaryDirectory() as tmp:
-            with unittest.mock.patch.object(
+            with self._provenance_patch(), unittest.mock.patch.object(
                 corpus_dispatch,
                 "gh_run_artifacts_v1",
                 # Sensitive to the run id on purpose: an observer that ignores
@@ -470,7 +738,7 @@ def test_the_admission_asks_about_the_artifact_the_operator_named(self) -> None:
             "--out",
         ]
         with tempfile.TemporaryDirectory() as tmp:
-            with unittest.mock.patch.object(
+            with self._provenance_patch(), unittest.mock.patch.object(
                 corpus_dispatch,
                 "gh_run_artifacts_v1",
                 lambda run_id: (("verification-evidence-arb", False),),
@@ -545,7 +813,7 @@ class _Completed:
         with tempfile.TemporaryDirectory() as tmp:
             argv = self._argv(Path(tmp) / "out.txt")
             argv[argv.index("424242")] = "31116022208"
-            with unittest.mock.patch.object(
+            with self._provenance_patch(), unittest.mock.patch.object(
                 corpus_dispatch,
                 "gh_run_artifacts_v1",
                 lambda run_id: (
@@ -595,7 +863,9 @@ def test_a_wrong_scale_is_refused_before_the_network_is_asked(self) -> None:
         # "wrong scale" rather than a diagnosis about the evidence run.
         observed: list[int] = []
         with tempfile.TemporaryDirectory() as tmp:
-            with unittest.mock.patch.object(
+            with self._provenance_patch(
+                lambda run_id: observed.append(run_id) or _admitted_provenance()
+            ), unittest.mock.patch.object(
                 corpus_dispatch,
                 "gh_run_artifacts_v1",
                 lambda run_id: observed.append(run_id) or (),
@@ -629,7 +899,7 @@ def flaky(command: tuple[str, ...], **_kwargs: object) -> object:
             return _Completed()
 
         with tempfile.TemporaryDirectory() as tmp:
-            with unittest.mock.patch.object(
+            with self._provenance_patch(), unittest.mock.patch.object(
                 corpus_dispatch,
                 "gh_run_artifacts_v1",
                 lambda run_id: (("verification-evidence-arb", False),),
@@ -648,7 +918,7 @@ def flaky(command: tuple[str, ...], **_kwargs: object) -> object:
 
     def test_a_missing_gh_is_a_typed_stop_not_a_traceback(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
-            with unittest.mock.patch.object(
+            with self._provenance_patch(), unittest.mock.patch.object(
                 corpus_dispatch,
                 "gh_run_artifacts_v1",
                 lambda run_id: (("verification-evidence-arb", False),),
@@ -667,7 +937,9 @@ def test_printing_stays_offline_and_asks_nobody(self) -> None:
         observed: list[int] = []
         launched: list[tuple[str, ...]] = []
         with tempfile.TemporaryDirectory() as tmp:
-            with unittest.mock.patch.object(
+            with self._provenance_patch(
+                lambda run_id: observed.append(run_id) or _admitted_provenance()
+            ), unittest.mock.patch.object(
                 corpus_dispatch,
                 "gh_run_artifacts_v1",
                 lambda run_id: observed.append(run_id) or (),
@@ -683,6 +955,146 @@ def test_printing_stays_offline_and_asks_nobody(self) -> None:
         self.assertEqual(observed, [])
         self.assertEqual(launched, [])
 
+    def test_a_run_of_a_foreign_workflow_dispatches_nothing(self) -> None:
+        # The artifact name is right, the artifact is live, and the run is
+        # still not the producer's.  Without the provenance seam this is 256
+        # lanes replaying whatever a foreign workflow happened to upload.
+        launched: list[tuple[str, ...]] = []
+        with tempfile.TemporaryDirectory() as tmp:
+            with self._provenance_patch(
+                lambda run_id: _admitted_provenance(
+                    path=".github/workflows/ci.yml"
+                )
+            ), unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifacts_v1",
+                lambda run_id: (("verification-evidence-arb", False),),
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: launched.append(tuple(command)),
+            ):
+                errors = io.StringIO()
+                with contextlib.redirect_stderr(errors):
+                    code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt"))
+        self.assertEqual(code, 64)
+        self.assertEqual(launched, [])
+
+    def test_a_run_a_pull_request_produced_dispatches_nothing(self) -> None:
+        # A fork's pull request can run the producer workflow and publish an
+        # artifact of exactly the allowlisted name; every earlier rule admits
+        # it, and the lanes then replay a comparator bundle nobody reviewed.
+        launched: list[tuple[str, ...]] = []
+        with tempfile.TemporaryDirectory() as tmp:
+            with self._provenance_patch(
+                lambda run_id: _admitted_provenance(event="pull_request")
+            ), unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifacts_v1",
+                lambda run_id: (("verification-evidence-arb", False),),
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: launched.append(tuple(command)),
+            ):
+                errors = io.StringIO()
+                with contextlib.redirect_stderr(errors):
+                    code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt"))
+        self.assertEqual(code, 64)
+        self.assertEqual(launched, [])
+
+    def test_a_run_that_failed_dispatches_nothing(self) -> None:
+        # A failed producer run can still have uploaded one engine's artifact
+        # before the other job died; the artifact listing cannot tell.
+        launched: list[tuple[str, ...]] = []
+        with tempfile.TemporaryDirectory() as tmp:
+            with self._provenance_patch(
+                lambda run_id: _admitted_provenance(conclusion="failure")
+            ), unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifacts_v1",
+                lambda run_id: (("verification-evidence-arb", False),),
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: launched.append(tuple(command)),
+            ):
+                errors = io.StringIO()
+                with contextlib.redirect_stderr(errors):
+                    code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt"))
+        self.assertEqual(code, 64)
+        self.assertEqual(launched, [])
+
+    def test_the_admitted_commit_reaches_the_operator_before_the_campaign(
+        self,
+    ) -> None:
+        # Which commit the evidence stands on is the one thing this process
+        # cannot decide and the operator can.  It has to be on their terminal
+        # before 256 lanes start, not discoverable afterwards in the run log.
+        launched: list[tuple[str, ...]] = []
+
+        class _Completed:
+            returncode = 0
+
+        with tempfile.TemporaryDirectory() as tmp:
+            with self._provenance_patch(), unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifacts_v1",
+                lambda run_id: (("verification-evidence-arb", False),),
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: (
+                    launched.append(tuple(command)) or _Completed()
+                ),
+            ):
+                errors = io.StringIO()
+                with contextlib.redirect_stderr(errors):
+                    code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt"))
+        self.assertEqual(code, 0)
+        self.assertEqual(len(launched), 256)
+        reported = errors.getvalue()
+        self.assertIn(ADMITTED_SHA_V1, reported)
+        # Before, not after: an operator reading it below "dispatching 256
+        # lanes" learns the commit once the campaign is already gone.
+        self.assertLess(
+            reported.index(ADMITTED_SHA_V1), reported.index("dispatching 256")
+        )
+
+    def test_the_provenance_admission_asks_about_the_run_the_operator_named(
+        self,
+    ) -> None:
+        # Mirror of the artifact pin: a call site that hardcoded a run id
+        # would vouch for one run while the lanes replayed another.
+        observed: list[int] = []
+        launched: list[tuple[str, ...]] = []
+
+        class _Completed:
+            returncode = 0
+
+        with tempfile.TemporaryDirectory() as tmp:
+            argv = self._argv(Path(tmp) / "out.txt")
+            argv[argv.index("424242")] = "31116022208"
+            with self._provenance_patch(
+                lambda run_id: observed.append(run_id) or _admitted_provenance()
+            ), unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifacts_v1",
+                lambda run_id: (("verification-evidence-arb", False),),
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: (
+                    launched.append(tuple(command)) or _Completed()
+                ),
+            ):
+                errors = io.StringIO()
+                with contextlib.redirect_stderr(errors):
+                    code = corpus_dispatch.main(argv)
+        self.assertEqual(code, 0)
+        self.assertEqual(observed, [31116022208])
+        self.assertEqual(len(launched), 256)
+
     def test_a_rejected_command_set_is_exit_64_not_a_type_error(self) -> None:
         # A foreign artifact name is refused by the pure builder; that
         # refusal must leave through the typed exit, never as a TypeError

From bcdc5eac001b94a7cc358f88115827b778f6abba Mon Sep 17 00:00:00 2001
From: Claude Code <daniilerosov12@gmail.com>
Date: Fri, 7 Aug 2026 00:03:11 +0300
Subject: [PATCH 7/7] Proof: pin what the origin admission only claimed
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Independent verification ran mutants against the previous commit and found
three rules that lived in the code and in comments beside it, with nothing to
hold them.

`text=True` was the dangerous one.  Removing it makes `gh` return bytes, the
decode raises, and every run is refused — a gate that looks alive and admits
nobody, fail-closed and silent.  No test saw it, and the same gap existed for
the artifact observation, so both are pinned now.

The comment claimed origin is asked before contents because a run of the wrong
workflow lists an artifact of exactly the right name.  True, and unproven:
swapping the two blocks left everything green.  The order is the claim, so the
order is what the test reads — the artifact observer must never be asked when
the origin is refused.

`RunProvenanceV1.__post_init__` was unreachable — the parser only ever feeds it
string slices — and raised a bare `TypeError` in a module whose whole thesis is
typed refusals.  Deleting it left the suite green, which is the definition of
speculative defence.  Gone.

Proven by the mutant each kills, one test apiece: the missing `text=True`, and
the swapped gate order.

Verified: 54 dispatch tests pass; 452 on Linux with the three pre-existing
`test_build` failures that also fail on an untouched tree.

This commit was authored on the #548 branch by mistake and moved to its own
before anything was pushed — #548 has been through three review rounds and
does not need a new feature landing inside it.
---
 proof/region/v1/corpus_dispatch.py            | 11 ------
 proof/region/v1/tests/test_corpus_dispatch.py | 39 +++++++++++++++++++
 2 files changed, 39 insertions(+), 11 deletions(-)

diff --git a/proof/region/v1/corpus_dispatch.py b/proof/region/v1/corpus_dispatch.py
index ad8459c7..696c25a2 100644
--- a/proof/region/v1/corpus_dispatch.py
+++ b/proof/region/v1/corpus_dispatch.py
@@ -282,17 +282,6 @@ class RunProvenanceV1:
     conclusion: str
     head_sha: str
 
-    def __post_init__(self) -> None:
-        for field in (
-            self.path,
-            self.event,
-            self.status,
-            self.conclusion,
-            self.head_sha,
-        ):
-            if type(field) is not str:
-                raise TypeError("invalid run provenance field")
-
 
 def gh_run_provenance_v1(run_id: int) -> RunProvenanceV1:
     """The run's origin as GitHub reports it, projected to the named fields.
diff --git a/proof/region/v1/tests/test_corpus_dispatch.py b/proof/region/v1/tests/test_corpus_dispatch.py
index a47386d4..443acb4e 100644
--- a/proof/region/v1/tests/test_corpus_dispatch.py
+++ b/proof/region/v1/tests/test_corpus_dispatch.py
@@ -362,6 +362,9 @@ def record(command: tuple[str, ...], **kwargs: object) -> object:
             seen.append(tuple(command))
             assert kwargs.get("check") is True
             assert kwargs.get("capture_output") is True
+            # Without this the reply arrives as bytes, the decode raises, and
+            # every run is refused — a gate that looks alive and admits nobody.
+            assert kwargs.get("text") is True
             return _Completed()
 
         with unittest.mock.patch.object(corpus_dispatch.subprocess, "run", record):
@@ -450,6 +453,9 @@ def record(command: tuple[str, ...], **kwargs: object) -> object:
             seen.append(tuple(command))
             assert kwargs.get("check") is True
             assert kwargs.get("capture_output") is True
+            # Without this the reply arrives as bytes, the decode raises, and
+            # every run is refused — a gate that looks alive and admits nobody.
+            assert kwargs.get("text") is True
             return _Completed()
 
         with unittest.mock.patch.object(corpus_dispatch.subprocess, "run", record):
@@ -932,6 +938,39 @@ def test_a_missing_gh_is_a_typed_stop_not_a_traceback(self) -> None:
                 code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt"))
         self.assertEqual(code, 64)
 
+    def test_the_origin_is_asked_before_the_artifact(self) -> None:
+        # A run of the wrong workflow lists an artifact of exactly the right
+        # name, so asking about the artifact first clears a run that should
+        # never have been considered — and spends a call to do it.  The order
+        # is the claim, so the order is what this pins.
+        asked: list[str] = []
+        with tempfile.TemporaryDirectory() as tmp:
+            with unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_provenance_v1",
+                lambda run_id: asked.append("origin")
+                or corpus_dispatch.RunProvenanceV1(
+                    corpus_dispatch.EVIDENCE_WORKFLOW_PATH_V1,
+                    "pull_request",
+                    "completed",
+                    "success",
+                    "0" * 40,
+                ),
+            ), unittest.mock.patch.object(
+                corpus_dispatch,
+                "gh_run_artifacts_v1",
+                lambda run_id: asked.append("artifact")
+                or (("verification-evidence-arb", False),),
+            ), unittest.mock.patch.object(
+                corpus_dispatch.subprocess,
+                "run",
+                lambda command, **kwargs: None,
+            ):
+                code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt"))
+        # The wrong trigger is refused, and the artifact was never asked about.
+        self.assertEqual(code, 64)
+        self.assertEqual(asked, ["origin"])
+
     def test_printing_stays_offline_and_asks_nobody(self) -> None:
         # Printing is documented as offline: it must not even observe.
         observed: list[int] = []