From 5c9f2eecd172d4a3caf8ac1e4f2b1d3575015192 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 07:13:33 +0500 Subject: [PATCH 1/9] feat(step7): the merge order becomes a predicate on the target tree MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The frozen T0 names an accepted harness identity that, on its own branch, does not exist yet: `104c384d01bf` arrives with the memory-semantics repair, which is a different pull request. That is fine while the merge order holds — and an order held by agreement is one stray click from becoming an archaeological artifact, with a document saying FROZEN in a tree where its referent is absent. So the order is checked instead of promised, and checked against the TARGET TREE rather than against pull-request numbers: a number proves someone pressed a button and says nothing about what the merged tree contains. The gate refuses unless, at the commit a merge would produce: - T0 declares FROZEN *and* collection_authorized: true; - the harness digest recomputed from that tree's own instrument sources, by the frozen formula, equals the one T0 names; - the policy freeze, the design constants and the training preregistration all bind that same digest — present is not the same as re-bound; - the capture, qualification and binding tools exist and actually enforce the campaign link and the authority state; - the step-7 note still revokes the automatic collection authority, so hosts plus a binding cannot again be enough to start a clock. The expected digest is read out of the frozen T0 and recomputed from the tree; a control proves no digest literal lives in the gate's own source, because a gate trusting its own constant would be checking itself. Exercised against real merges, not only fixtures. The full chain (#355 -> #356 -> #353 -> #354) is allowed. #354 alone is refused on three predicates at once — the instrument hashes to 562a7f7232da where T0 names 104c384d01bf, the bindings are stale, the machinery is absent. #355 plus #354 without the machinery is refused on one. T0-0 turns out to need no predicate of its own: #353 is an ancestor of the freeze commit, so it cannot be skipped. merge gate controls: 7 passed, 0 failed. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018xhcg5opoFbSTdYHpkSXCh --- scripts/step7/mergegate.py | 157 +++++++++++++++++++++ tests/test_step7_mergegate.py | 254 ++++++++++++++++++++++++++++++++++ 2 files changed, 411 insertions(+) create mode 100644 scripts/step7/mergegate.py create mode 100644 tests/test_step7_mergegate.py diff --git a/scripts/step7/mergegate.py b/scripts/step7/mergegate.py new file mode 100644 index 00000000..49620fa4 --- /dev/null +++ b/scripts/step7/mergegate.py @@ -0,0 +1,157 @@ +#!/usr/bin/env python3 +"""P-022 — the merge gate: does the world a freeze enters satisfy its references? + +The frozen T0 names things that live outside itself: an accepted harness +identity, the step-7 machinery that enforces the campaign join, the T0-0 +amendment that revoked automatic collection authority. On the branch where T0 +was frozen, some of those referents arrive through *other* pull requests. That +is fine while the merge order holds — and merge order held by agreement is one +stray click away from becoming an archaeological artifact, with a document +saying FROZEN in a tree where its referent does not exist yet. + +So the order is checked rather than promised, and it is checked against the +**target tree**, never against pull-request numbers. A PR number proves that +someone pressed a button; it proves nothing about what the merged tree contains. + +Run it on the commit a merge would produce: + + python scripts/step7/mergegate.py --repo . --commit + +Exit 0 only when every predicate holds. Any failure means the frozen contract +would become reachable from a tree that cannot satisfy it, and the merge is +refused however mergeable the forge believes it to be. + +The digest is not hard-coded here: it is read out of the frozen T0 and then +RECOMPUTED from the instrument sources in the target tree, by the frozen +formula, without importing the instrument. A gate that trusted a constant in +its own source would be checking itself. +""" + +from __future__ import annotations + +import argparse + +import json +import re +import subprocess +import sys +from pathlib import Path + +import execbinding as eb # the single implementation of the harness-digest formula + +T0_PATH = "docs/notes/p022-263-t0-protocol-freeze.md" +STEP7_NOTE = "docs/notes/p022-263a-step7-environment-capture.md" +BINDING_ARTIFACTS = ( + "docs/evidence/calibration/p022-263a-policy-freeze.json", + "docs/evidence/calibration/p022-263a-design-constants.json", + "docs/evidence/calibration/p022-263a-training-preregistration.json", +) +STEP7_TOOLS = ("scripts/step7/envcapture.py", "scripts/step7/hostqual.py", + "scripts/step7/execbinding.py") +DIGEST_RE = re.compile(r"measurement_harness_digest\s*\n?\s*([0-9a-f]{64})") + + +def blob(repo: Path, commit: str, path: str) -> bytes | None: + proc = subprocess.run(["git", "-C", str(repo), "cat-file", "blob", f"{commit}:{path}"], + capture_output=True, check=False) + return proc.stdout if proc.returncode == 0 else None + + +def check(name: str, ok: bool, detail: str) -> dict[str, object]: + return {"check": name, "result": "pass" if ok else "fail", "detail": detail} + + +def gate(repo: Path, commit: str) -> list[dict[str, object]]: + results: list[dict[str, object]] = [] + + raw_t0 = blob(repo, commit, T0_PATH) + if raw_t0 is None: + return [check("t0_present", False, f"{T0_PATH} does not exist at {commit}")] + text = raw_t0.decode("utf-8", "replace") + frozen = re.search(r"^\s*(NOT_FROZEN|FROZEN)\.?\s*$", text, re.MULTILINE) + flag = re.search(r"^\s*collection_authorized:\s*(true|false)\s*$", text, re.MULTILINE) + declared = frozen.group(1) if frozen else "" + authorized = (flag.group(1) == "true") if flag else None + results.append(check( + "t0_frozen_and_authorized", declared == "FROZEN" and authorized is True, + f"T0 declares {declared}, collection_authorized={json.dumps(authorized)}")) + + named = DIGEST_RE.search(text) + results.append(check("t0_names_a_digest", named is not None, + f"T0 names {named.group(1)[:12] if named else ''} as the accepted " + "harness identity")) + if named is None: + return results + expected = named.group(1) + + # Recomputed from the target tree's own sources, by the frozen formula. + live = eb.harness_digest_at(repo, commit) + results.append(check( + "instrument_matches_t0", live == expected, + f"the instrument at {commit[:12]} hashes to {live[:12] if live else ''}; " + f"T0 names {expected[:12]}. The frozen contract would otherwise point at an identity " + "this tree does not contain")) + + rebound = [] + for path in BINDING_ARTIFACTS: + raw = blob(repo, commit, path) + if raw is None: + rebound.append(f"{Path(path).name}: absent") + elif expected not in raw.decode("utf-8", "replace"): + rebound.append(f"{Path(path).name}: still bound to an older digest") + results.append(check( + "steps_4_5_6_rebound", not rebound, + "; ".join(rebound) if rebound else + f"the policy freeze, the design constants and the training preregistration all bind " + f"{expected[:12]}")) + + missing_tools = [t for t in STEP7_TOOLS if blob(repo, commit, t) is None] + hostqual = (blob(repo, commit, "scripts/step7/hostqual.py") or b"").decode("utf-8", "replace") + enforces_link = "CAMPAIGN_LINK_SCHEMA" in hostqual and "check_campaign_link" in hostqual + enforces_authority = "collection_authorized is" in hostqual or ( + "authorized is not True" in hostqual) + results.append(check( + "step7_machinery_present", not missing_tools and enforces_link and enforces_authority, + "; ".join(filter(None, [ + f"missing: {missing_tools}" if missing_tools else "", + "" if enforces_link else "hostqual does not enforce the campaign link", + "" if enforces_authority else "hostqual does not enforce the authority state", + ])) or "capture, qualification and binding tools are present, and the campaign link and " + "authority state are enforced")) + + note = (blob(repo, commit, STEP7_NOTE) or b"").decode("utf-8", "replace") + revoked = "AUTOMATIC AUTHORISATION OF THE FIRST STEP-7 COLLECTION IS REVOKED" in note + lingering = "SINGLE STEP-7 COLLECTION AUTHORISED automatically" in note + results.append(check( + "t0_zero_prerequisite", revoked and not lingering, + "the step-7 status block revokes the automatic collection authority" + if revoked and not lingering else + "the step-7 note still authorises the first collection automatically, so hosts plus a " + "binding would again be enough to start a clock")) + return results + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo", type=Path, default=Path(__file__).resolve().parents[2]) + parser.add_argument("--commit", default="HEAD") + parser.add_argument("--json", action="store_true") + args = parser.parse_args(argv) + + results = gate(args.repo, args.commit) + failed = [r for r in results if r["result"] != "pass"] + if args.json: + print(json.dumps({"commit": args.commit, "checks": results, + "merge_allowed": not failed}, indent=2)) + else: + for r in results: + print(f"{'ok ' if r['result'] == 'pass' else 'FAIL'} [{r['check']}] {r['detail']}") + print() + print("merge gate: allowed" if not failed else + f"merge gate: REFUSED, {len(failed)} predicate(s) unsatisfied — the frozen " + "contract must not become reachable from this tree") + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_step7_mergegate.py b/tests/test_step7_mergegate.py new file mode 100644 index 00000000..9340d2e4 --- /dev/null +++ b/tests/test_step7_mergegate.py @@ -0,0 +1,254 @@ +#!/usr/bin/env python3 +"""P-022 — controls on the merge gate. + + mergegate-complete-tree a world that satisfies the freeze is allowed + mergegate-missing-instrument the digest T0 names must exist in the tree + mergegate-stale-bindings steps 4/5/6 must be re-bound to that digest + mergegate-missing-machinery the campaign link and authority must be enforceable + mergegate-lingering-auto the revoked automatic authority must stay revoked + mergegate-unfrozen-t0 an unfrozen or unauthorised T0 is not merged as frozen + mergegate-reads-t0-digest the expected digest comes from T0, not from a constant + +Every control builds a throwaway repository, so the gate is exercised against +real git objects rather than against a mock of the thing it exists to read. + +Failures print `FAIL[]: `; nothing stops at the first one. + +Run: python tests/test_step7_mergegate.py +""" + +from __future__ import annotations + +import contextlib +import io +import re +import subprocess +import sys +import tempfile +from collections.abc import Callable +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT / "scripts")) +sys.path.insert(0, str(ROOT / "scripts" / "step7")) + +import execbinding as eb # noqa: E402 +import mergegate as mg # noqa: E402 + +_FAILURES: list[tuple[str, str]] = [] +_PASSES: list[str] = [] + + +def fail(check: str, detail: str) -> None: + _FAILURES.append((check, detail)) + print(f"FAIL[{check}]: {detail}") + + +def ok(check: str, detail: str = "") -> None: + _PASSES.append(check) + print(f"ok[{check}]: {detail}" if detail else f"ok[{check}]") + + +def guarded(check: str, control: Callable[[], None]) -> None: + try: + control() + except Exception as exc: + fail(check, f"the control raised {type(exc).__name__}: {exc}") + + +HOSTQUAL_STUB = ('CAMPAIGN_LINK_SCHEMA = "own.net/p022/campaign-link"\n' + 'def check_campaign_link():\n pass\n' + 'def gate():\n if authorized is not True:\n return "refuse"\n') +STEP7_NOTE_REVOKED = ("Status:\n AUTOMATIC AUTHORISATION OF THE FIRST STEP-7 COLLECTION " + "IS REVOKED (T0-0).\n") +STEP7_NOTE_AUTO = "Status:\n SINGLE STEP-7 COLLECTION AUTHORISED automatically after that gate.\n" + + +def commit_tree(repo: Path, files: dict[str, str]) -> str: + repo.mkdir(parents=True, exist_ok=True) + + def run(*args: str) -> None: + subprocess.run(["git", "-C", str(repo), *args], capture_output=True, check=True) + + if not (repo / ".git").exists(): + run("init", "-q") + run("config", "user.email", "control@example.invalid") + run("config", "user.name", "control") + for name, text in files.items(): + path = repo / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + run("add", "-A") + run("commit", "-q", "-m", "fixture") + return subprocess.run(["git", "-C", str(repo), "rev-parse", "HEAD"], + capture_output=True, text=True, check=True).stdout.strip() + + +def world(tmp: Path, *, frozen: bool = True, authorized: bool = True, + instrument: str = "print('instrument')\n", rebound: bool = True, + machinery: bool = True, auto_authority: bool = False, + name: str = "w") -> tuple[Path, str]: + """A synthetic target tree, and the commit a merge into it would produce.""" + repo = tmp / name + # The digest T0 will name is whatever this tree's own sources hash to, so a + # fixture cannot pass by agreeing with a constant this file also wrote. + files = {mg.STEP7_TOOLS[0]: "# capture\n", + eb.INSTRUMENT_SOURCES[0]: instrument, + eb.INSTRUMENT_SOURCES[1]: '{"decisive": []}\n', + mg.STEP7_NOTE: STEP7_NOTE_AUTO if auto_authority else STEP7_NOTE_REVOKED} + if machinery: + files[mg.STEP7_TOOLS[1]] = HOSTQUAL_STUB + files[mg.STEP7_TOOLS[2]] = "# binding\n" + probe = commit_tree(repo, files) + digest = eb.harness_digest_at(repo, probe) or "" + status = "FROZEN." if frozen else "NOT_FROZEN." + flag = "true" if authorized else "false" + files[mg.T0_PATH] = (f"```text\nStatus:\n {status}\n collection_authorized: {flag}\n```\n\n" + f" measurement_harness_digest\n {digest}\n") + bound = digest if rebound else "0" * 64 + for artifact in mg.BINDING_ARTIFACTS: + files[artifact] = '{"measurement_harness_digest": "' + bound + '"}\n' + return repo, commit_tree(repo, files) + + +def verdicts(repo: Path, commit: str) -> dict[str, str]: + return {str(r["check"]): str(r["result"]) for r in mg.gate(repo, commit)} + + +def control_complete_tree() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw)) + results = verdicts(repo, commit) + bad = sorted(k for k, v in results.items() if v != "pass") + if bad: + fail("mergegate-complete-tree", f"a satisfying world was refused on {bad}") + return + # Captured: a green run that prints a refusal teaches readers to skim past + # refusals, which is how the word stops meaning anything. + with contextlib.redirect_stdout(io.StringIO()): + rc = mg.main(["--repo", str(repo), "--commit", commit]) + if rc != 0: + fail("mergegate-complete-tree", "the CLI refused a satisfying world") + return + ok("mergegate-complete-tree", + "a tree carrying the named instrument, the re-bound artifacts, the machinery and the " + "revoked automatic authority is allowed") + + +def control_missing_instrument() -> None: + with tempfile.TemporaryDirectory() as raw: + tmp = Path(raw) + repo, commit = world(tmp) + # the instrument moves after T0 named it — the exact intermediate state a + # stray merge produces when the repair PR is not in yet + drifted = commit_tree(repo, {eb.INSTRUMENT_SOURCES[0]: "print('older instrument')\n"}) + results = verdicts(repo, drifted) + if results.get("instrument_matches_t0") != "fail": + fail("mergegate-missing-instrument", + "a tree whose instrument does not hash to the digest T0 names was allowed") + return + noise = io.StringIO() + with contextlib.redirect_stdout(noise): + rc = mg.main(["--repo", str(repo), "--commit", drifted]) + if rc == 0 or "REFUSED" not in noise.getvalue(): + fail("mergegate-missing-instrument", "the CLI allowed it") + return + ok("mergegate-missing-instrument", + "the frozen contract cannot become reachable from a tree whose instrument is not the " + "one it names") + + +def control_stale_bindings() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), rebound=False) + if verdicts(repo, commit).get("steps_4_5_6_rebound") != "fail": + fail("mergegate-stale-bindings", "artifacts bound to an older digest were allowed") + return + ok("mergegate-stale-bindings", + "steps 4, 5 and 6 must be re-bound to the digest T0 names, not merely present") + + +def control_missing_machinery() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), machinery=False) + if verdicts(repo, commit).get("step7_machinery_present") != "fail": + fail("mergegate-missing-machinery", + "a tree without the qualification and binding tools was allowed") + return + ok("mergegate-missing-machinery", + "without the tools that enforce the campaign link and the authority state, the freeze " + "would enter a world that cannot keep its promises") + + +def control_lingering_auto() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), auto_authority=True) + if verdicts(repo, commit).get("t0_zero_prerequisite") != "fail": + fail("mergegate-lingering-auto", + "a tree still authorising the first collection automatically was allowed") + return + ok("mergegate-lingering-auto", + "the revoked automatic authority must still be revoked in the target tree, or hosts " + "plus a binding are again enough to start a clock") + + +def control_unfrozen_t0() -> None: + with tempfile.TemporaryDirectory() as raw: + tmp = Path(raw) + for label, kwargs in (("not frozen", {"frozen": False}), + ("frozen but unauthorised", {"authorized": False}), + ("unfrozen but authorised", {"frozen": False, "authorized": True})): + repo, commit = world(tmp, name=f"w-{len(list(tmp.glob('w-*')))}", **kwargs) + if verdicts(repo, commit).get("t0_frozen_and_authorized") != "fail": + fail("mergegate-unfrozen-t0", f"{label} was treated as a freeze") + return + ok("mergegate-unfrozen-t0", + "only FROZEN together with collection_authorized: true is merged as a freeze; the other " + "three states are refused here as they are at the session gates") + + +def control_reads_t0_digest() -> None: + """The gate must not carry the expected digest in its own source.""" + source = (ROOT / "scripts" / "step7" / "mergegate.py").read_text(encoding="utf-8") + hardcoded = re.findall(r"\b[0-9a-f]{64}\b", source) + if hardcoded: + fail("mergegate-reads-t0-digest", + f"the gate carries {len(hardcoded)} literal digest(s) in its own source; it would " + "then be checking a constant it wrote rather than the contract") + return + with tempfile.TemporaryDirectory() as raw: + # two different instruments, two different digests, both accepted because + # each tree's T0 names its own + for text in ("print('one')\n", "print('a completely different instrument')\n"): + repo, commit = world(Path(raw), instrument=text, + name=f"d{abs(hash(text)) % 1000}") + if verdicts(repo, commit).get("instrument_matches_t0") != "pass": + fail("mergegate-reads-t0-digest", + "a tree whose T0 names its own instrument was refused") + return + ok("mergegate-reads-t0-digest", + "the expected digest is read out of the frozen T0 and recomputed from the target tree; " + "no digest is written into this gate's own source") + + +CONTROLS: list[tuple[str, Callable[[], None]]] = [ + ("mergegate-complete-tree", control_complete_tree), + ("mergegate-missing-instrument", control_missing_instrument), + ("mergegate-stale-bindings", control_stale_bindings), + ("mergegate-missing-machinery", control_missing_machinery), + ("mergegate-lingering-auto", control_lingering_auto), + ("mergegate-unfrozen-t0", control_unfrozen_t0), + ("mergegate-reads-t0-digest", control_reads_t0_digest), +] + + +def run() -> int: + for name, control in CONTROLS: + guarded(name, control) + print() + print(f"merge gate controls: {len(_PASSES)} passed, {len(_FAILURES)} failed") + return 1 if _FAILURES else 0 + + +if __name__ == "__main__": + sys.exit(run()) From 81b3a130018bb1ba0d504a56862f7ff93b84920c Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 07:46:14 +0500 Subject: [PATCH 2/9] fix(step7): the merge gate must run the machinery, not read its name MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first revision proved words. `step7_machinery_present` asked whether the two tool files existed and whether certain identifiers appeared in them, which a file containing `def check_campaign_link(): pass` satisfies — and the fixture that was supposed to exclude that world was exactly that file. A check for the name of a mechanism is not a check of the mechanism. `step7_machinery_enforces` now extracts the target tree's own hostqual and execbinding, runs them in a subprocess against a synthetic campaign, and requires all four attacks to be refused: a link naming another execution binding, a freeze edited after the link was made, FROZEN with collection_authorized false (on both readers), and a campaign swapped between preflight and postflight. A tool that accepts any of them fails the gate, and the refusal says which attack got through. `steps_4_5_6_rebound` searched each artifact's text for the digest. The right digest sitting in any field — a comment, a history entry, a field nobody binds — was read as a binding. It now walks the exact path each artifact actually binds at: measurement_harness_digest in the policy freeze, bound_measurement_harness_digest in the design constants, bindings.measurement_harness_digest in the training preregistration. Anywhere else is not a binding, and the refusal names the path and both digests. Fixtures ship the real tools; the two attack controls mutate one enforcement point each and require the gate to notice. Twelve controls, four of them new: a permissive link check is refused, an authority check that always passes is refused, a stale binding with the right digest in a decoy field is refused, and correct exact fields are allowed. Against the real merges: the full chain is allowed (rc 0); #354 without #355 and #356 is refused on three predicates (rc 1); #354 with the instrument but without the step-7 tools is refused on one (rc 1). 30d2f32 is untouched. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018xhcg5opoFbSTdYHpkSXCh --- scripts/step7/mergegate.py | 234 +++++++++++++++++++++++++++--- tests/test_step7_mergegate.py | 260 +++++++++++++++++++++++++++------- 2 files changed, 419 insertions(+), 75 deletions(-) diff --git a/scripts/step7/mergegate.py b/scripts/step7/mergegate.py index 49620fa4..fce995f6 100644 --- a/scripts/step7/mergegate.py +++ b/scripts/step7/mergegate.py @@ -35,17 +35,23 @@ import re import subprocess import sys +import tempfile from pathlib import Path import execbinding as eb # the single implementation of the harness-digest formula T0_PATH = "docs/notes/p022-263-t0-protocol-freeze.md" STEP7_NOTE = "docs/notes/p022-263a-step7-environment-capture.md" -BINDING_ARTIFACTS = ( - "docs/evidence/calibration/p022-263a-policy-freeze.json", - "docs/evidence/calibration/p022-263a-design-constants.json", - "docs/evidence/calibration/p022-263a-training-preregistration.json", -) +# Each artifact binds the digest at its own exact path. Named here so the check +# reads the binding rather than searching the file for a hopeful substring. +BINDING_ARTIFACTS = { + "docs/evidence/calibration/p022-263a-policy-freeze.json": + ("measurement_harness_digest",), + "docs/evidence/calibration/p022-263a-design-constants.json": + ("bound_measurement_harness_digest",), + "docs/evidence/calibration/p022-263a-training-preregistration.json": + ("bindings", "measurement_harness_digest"), +} STEP7_TOOLS = ("scripts/step7/envcapture.py", "scripts/step7/hostqual.py", "scripts/step7/execbinding.py") DIGEST_RE = re.compile(r"measurement_harness_digest\s*\n?\s*([0-9a-f]{64})") @@ -57,6 +63,183 @@ def blob(repo: Path, commit: str, path: str) -> bytes | None: return proc.stdout if proc.returncode == 0 else None +# The witness runs INSIDE the candidate tree's own modules, in a subprocess, so +# that importing them cannot bind to this gate's copies and a passing result +# cannot come from anything but the code being merged. +WITNESS_SOURCE = r""" +import json, sys, subprocess +from pathlib import Path +sys.path.insert(0, "scripts/step7") +import hostqual as hq +import execbinding as eb + +tmp = Path(sys.argv[1]) +POWER = {"platform": "windows", "plan_guid": hq.WIN_ACCEPTED_PLANS[0], + "processor_min_ac": 100, "processor_max_ac": 100, + "processor_min_dc": 100, "processor_max_dc": 100} +hq.power_snapshot = lambda: dict(POWER) +failures = [] + +def repo_with(name, files): + root = tmp / name + root.mkdir(parents=True, exist_ok=True) + run = lambda *a: subprocess.run(["git", "-C", str(root), *a], capture_output=True, check=True) + if not (root / ".git").exists(): + run("init", "-q"); run("config", "user.email", "w@x"); run("config", "user.name", "w") + for rel, text in files.items(): + f = root / rel; f.parent.mkdir(parents=True, exist_ok=True) + f.write_text(text, encoding="utf-8") + run("add", "-A"); run("commit", "-q", "-m", "w") + return root, subprocess.run(["git", "-C", str(root), "rev-parse", "HEAD"], + capture_output=True, text=True, check=True).stdout.strip() + +def w(name, doc): + f = tmp / name; f.write_text(json.dumps(doc), encoding="utf-8"); return f + +# --- W3/W4: the authority state must bite on both readers ------------------- +for label, status, flag, must_pass in (("FROZEN+false", "FROZEN.", "false", False), + ("NOT_FROZEN+true", "NOT_FROZEN.", "true", False), + ("FROZEN+true", "FROZEN.", "true", True)): + body = "```text\nStatus:\n %s\n collection_authorized: %s\n```\n" % (status, flag) + root, commit = repo_with("t0-" + flag + status[:4], {"t0.md": body}) + got = hq.bind_t0(root, "t0.md", commit)[1]["result"] == "pass" + if got != must_pass: + failures.append("hostqual accepted " + label if got else + "hostqual refused " + label) + try: + eb.t0_at(root, "t0.md", commit); accepted = True + except eb.BindingRefused: + accepted = False + if accepted != must_pass: + failures.append("execbinding accepted " + label if accepted else + "execbinding refused " + label) + +# --- W1/W2: the campaign link must be verified, not merely mentioned -------- +froot, fcommit = repo_with("freeze", {"d7/payload.json": '{"kind": "d7"}\n', + "d7/attestation.json": '{"kind": "att"}\n'}) +payload, att = froot / "d7/payload.json", froot / "d7/attestation.json" +blob = subprocess.run(["git", "-C", str(froot), "rev-parse", fcommit + ":d7/payload.json"], + capture_output=True, text=True, check=True).stdout.strip() +binding = w("binding.json", {"kind": "own.net/p022/execution-binding"}) +link = {"kind": hq.CAMPAIGN_LINK_SCHEMA, "schema": 1, + "execution_binding_sha256": hq.sha256_file(binding), + "d7_payload": {"path": "d7/payload.json", "sha256": hq.sha256_file(payload), + "blob_sha": blob, "commit": fcommit}, + "d7_attestation": {"path": "d7/attestation.json", "sha256": hq.sha256_file(att)}, + "recorded_at": "w"} +if hq.check_campaign_link(link, binding, froot)["result"] != "pass": + failures.append("a correct campaign link was refused") +wrong = dict(link, execution_binding_sha256="f" * 64) +if hq.check_campaign_link(wrong, binding, froot)["result"] != "fail": + failures.append("a link naming another execution binding was accepted") +payload.write_text('{"kind": "d7", "edited": true}\n', encoding="utf-8") +if hq.check_campaign_link(link, binding, froot)["result"] != "fail": + failures.append("a freeze edited after linking was accepted") + +# --- W5: a campaign cannot be swapped between preflight and postflight ------ +QUAL = {"kind": hq.QUALIFICATION_SCHEMA, "schema": 1, "stratum": "linux", + "t0": {"commit": "c" * 40, "path": "t0.md", "blob_sha": "b" * 40, + "sha256": "f" * 64, "status": "FROZEN"}, + "environment_id": "env-1", "host_fingerprint": "sha256:abc", + "provisioning": {"sha256": "0" * 64}, "environment_manifest": {"sha256": "1" * 64}, + "qualification_tool": {"sha256": "2" * 64}, "power_snapshot": dict(POWER), + "predicate": {k: "pass" for k in hq.PREDICATE_KEYS}, + "memory_metric": hq.STRATUM_METRIC["linux"], "qualified": True, "qualified_at": "w"} +MANIFEST = {"schema": hq.ENVCAPTURE_SCHEMA, + "identity": {"environment_id": {"status": "observed", "value": "env-1"}, + "host_fingerprint": {"status": "observed", "value": "sha256:abc"}}, + "provenance": {"ci": False}} +QUAL["environment_identity_sha256"] = hq.canonical_sha256(MANIFEST["identity"]) +CAND = b"candidate bytes" +cand = tmp / "cand.bin"; cand.write_bytes(CAND) +qpath = w("qual.json", QUAL) +stratum_block = lambda m, extra: {"qualification_sha256": hq.sha256_file(qpath), + "environment_id": "env-1", "host_fingerprint": "sha256:abc", + "environment_identity_sha256": QUAL["environment_identity_sha256"], + "candidate_sha256": hq.sha256_bytes(CAND), + "candidate_bytes": len(CAND), "memory_metric": m, **extra} +BINDING = {"kind": eb.BINDING_SCHEMA, "schema": 1, + "t0": {"commit": "c" * 40, "path": "t0.md", "blob_sha": "b" * 40, "sha256": "f" * 64}, + "instrument": {"accepted_commit": "a" * 40, "harness_digest": "d" * 64}, + "workloads": {"path": eb.WORKLOAD_MANIFEST, "manifest_sha256": "9" * 64}, + "linux": stratum_block(hq.STRATUM_METRIC["linux"], {}), + "windows": stratum_block(hq.STRATUM_METRIC["windows"], {})} +bpath = w("exec-binding.json", BINDING) +mpath = w("manifest.json", MANIFEST) +dpath = w("declaration.json", {"kind": hq.DECLARATION_SCHEMA, "schema": 1, + "no_campaign_workload": True, + "no_interactive_user_workload": True, + "no_prohibited_background_job_active": True, + "operator": "w", "recorded_at": "w"}) +probe = tmp / "probe.json"; probe.write_text("{}", encoding="utf-8") +quiet = {"eligible": True, "reason": "", "samples": [0.01], "mean": 0.01, "max": 0.01} + +def link_for(binding_file, root, commit_sha, payload_file, att_file, blob_sha): + return {"kind": hq.CAMPAIGN_LINK_SCHEMA, "schema": 1, + "execution_binding_sha256": hq.sha256_file(binding_file), + "d7_payload": {"path": "d7/payload.json", "sha256": hq.sha256_file(payload_file), + "blob_sha": blob_sha, "commit": commit_sha}, + "d7_attestation": {"path": "d7/attestation.json", "sha256": hq.sha256_file(att_file)}, + "recorded_at": "w"} + +r2, c2 = repo_with("freeze2", {"d7/payload.json": '{"kind": "d7", "campaign": "two"}\n', + "d7/attestation.json": '{"kind": "att"}\n'}) +p2, a2 = r2 / "d7/payload.json", r2 / "d7/attestation.json" +b2 = subprocess.run(["git", "-C", str(r2), "rev-parse", c2 + ":d7/payload.json"], + capture_output=True, text=True, check=True).stdout.strip() +r3, c3 = repo_with("freeze3", {"d7/payload.json": '{"kind": "d7", "campaign": "three"}\n', + "d7/attestation.json": '{"kind": "att"}\n'}) +p3, a3 = r3 / "d7/payload.json", r3 / "d7/attestation.json" +b3 = subprocess.run(["git", "-C", str(r3), "rev-parse", c3 + ":d7/payload.json"], + capture_output=True, text=True, check=True).stdout.strip() +l2 = w("link2.json", link_for(bpath, r2, c2, p2, a2, b2)) +l3 = w("link3.json", link_for(bpath, r3, c3, p3, a3, b3)) + +pre = hq.session_eligibility(bpath, qpath, mpath, dpath, cand, l2, r2, quiesce_result=quiet) +if not pre["eligible"]: + failures.append("a correctly linked session was refused: " + str(pre["reasons"])[:120]) +ppath = w("preflight.json", pre) +after = hq.session_admissibility(bpath, qpath, ppath, mpath, cand, probe, l3, r3) +if after["admissible"]: + failures.append("a campaign swapped between preflight and postflight was admissible") + +print(json.dumps(failures)) +""" + + +def witness(repo: Path, commit: str) -> list[str]: + """Run the target tree's own tools and require them to refuse. + + A name in a source file proves nothing: a comment, a dead function or + `def check_campaign_link(): pass` all satisfy a lexical check, and an earlier + revision of this gate was satisfied by exactly that — its own fixture shipped + the no-op and called the world compliant. So the tools are extracted from the + target tree, imported in a subprocess, and driven against attacks they are + claimed to stop. + """ + with tempfile.TemporaryDirectory() as raw: + work = Path(raw) + for tool in STEP7_TOOLS: + data = blob(repo, commit, tool) + if data is None: + return [f"{tool} is absent"] + target = work / tool + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(data) + script = work / "witness.py" + script.write_text(WITNESS_SOURCE, encoding="utf-8") + fixtures = work / "fixtures" + fixtures.mkdir() + proc = subprocess.run([sys.executable, str(script), str(fixtures)], + capture_output=True, text=True, cwd=str(work), check=False) + if proc.returncode != 0: + tail = (proc.stderr or proc.stdout).strip().splitlines()[-1:] or [""] + return [f"the machinery could not be exercised: {tail[0][:160]}"] + try: + return list(json.loads(proc.stdout.strip().splitlines()[-1])) + except (ValueError, IndexError): + return [f"the witness produced no verdict: {proc.stdout[:120]!r}"] + def check(name: str, ok: bool, detail: str) -> dict[str, object]: return {"check": name, "result": "pass" if ok else "fail", "detail": detail} @@ -93,12 +276,25 @@ def gate(repo: Path, commit: str) -> list[dict[str, object]]: "this tree does not contain")) rebound = [] - for path in BINDING_ARTIFACTS: + for path, field in BINDING_ARTIFACTS.items(): raw = blob(repo, commit, path) if raw is None: rebound.append(f"{Path(path).name}: absent") - elif expected not in raw.decode("utf-8", "replace"): - rebound.append(f"{Path(path).name}: still bound to an older digest") + continue + try: + doc = json.loads(raw.decode("utf-8")) + except ValueError as exc: + rebound.append(f"{Path(path).name}: not readable JSON ({exc})") + continue + # The exact field, by its own path in that artifact. A substring search + # would accept the right digest sitting in any passing field while the + # real binding stayed old — which is the same defect one level down. + node: object = doc + for key in field: + node = node.get(key) if isinstance(node, dict) else None + if node != expected: + got = str(node)[:12] if node is not None else "" + rebound.append(f"{Path(path).name}: {'.'.join(field)} is {got}, not {expected[:12]}") results.append(check( "steps_4_5_6_rebound", not rebound, "; ".join(rebound) if rebound else @@ -106,18 +302,16 @@ def gate(repo: Path, commit: str) -> list[dict[str, object]]: f"{expected[:12]}")) missing_tools = [t for t in STEP7_TOOLS if blob(repo, commit, t) is None] - hostqual = (blob(repo, commit, "scripts/step7/hostqual.py") or b"").decode("utf-8", "replace") - enforces_link = "CAMPAIGN_LINK_SCHEMA" in hostqual and "check_campaign_link" in hostqual - enforces_authority = "collection_authorized is" in hostqual or ( - "authorized is not True" in hostqual) - results.append(check( - "step7_machinery_present", not missing_tools and enforces_link and enforces_authority, - "; ".join(filter(None, [ - f"missing: {missing_tools}" if missing_tools else "", - "" if enforces_link else "hostqual does not enforce the campaign link", - "" if enforces_authority else "hostqual does not enforce the authority state", - ])) or "capture, qualification and binding tools are present, and the campaign link and " - "authority state are enforced")) + if missing_tools: + results.append(check("step7_machinery_enforces", False, f"missing: {missing_tools}")) + else: + refusals = witness(repo, commit) + results.append(check( + "step7_machinery_enforces", not refusals, + "; ".join(refusals) if refusals else + "the target tree's own tools were run and refused all four attacks: a link naming " + "another binding, a freeze edited after linking, FROZEN with collection_authorized " + "false, and a campaign swapped between preflight and postflight")) note = (blob(repo, commit, STEP7_NOTE) or b"").decode("utf-8", "replace") revoked = "AUTOMATIC AUTHORISATION OF THE FIRST STEP-7 COLLECTION IS REVOKED" in note diff --git a/tests/test_step7_mergegate.py b/tests/test_step7_mergegate.py index 9340d2e4..29e3a332 100644 --- a/tests/test_step7_mergegate.py +++ b/tests/test_step7_mergegate.py @@ -1,18 +1,25 @@ #!/usr/bin/env python3 """P-022 — controls on the merge gate. - mergegate-complete-tree a world that satisfies the freeze is allowed - mergegate-missing-instrument the digest T0 names must exist in the tree - mergegate-stale-bindings steps 4/5/6 must be re-bound to that digest - mergegate-missing-machinery the campaign link and authority must be enforceable - mergegate-lingering-auto the revoked automatic authority must stay revoked - mergegate-unfrozen-t0 an unfrozen or unauthorised T0 is not merged as frozen - mergegate-reads-t0-digest the expected digest comes from T0, not from a constant - -Every control builds a throwaway repository, so the gate is exercised against -real git objects rather than against a mock of the thing it exists to read. - -Failures print `FAIL[]: `; nothing stops at the first one. + mergegate-complete-tree a world that satisfies the freeze is allowed + mergegate-missing-instrument the digest T0 names must exist in the tree + mergegate-stale-bindings steps 4/5/6 must be re-bound to that digest + mergegate-exact-binding-field the digest is read at each artifact's own path + mergegate-decoy-digest the right digest in a field nobody binds is not a binding + mergegate-noop-campaign-link a permissive link check must not satisfy the gate + mergegate-dead-authority an authority check that always passes must not satisfy it + mergegate-missing-machinery the tools must be there at all + mergegate-lingering-auto the revoked automatic authority must stay revoked + mergegate-unfrozen-t0 an unfrozen or unauthorised T0 is not merged as frozen + mergegate-reads-t0-digest the expected digest comes from T0, not from a constant + control-inventory-complete this list and the executed set are the same set + +Fixtures ship the real tools, and two controls mutate them. An earlier revision +shipped a stub whose check_campaign_link was a bare pass and called that world +compliant: the fixture demonstrated the false positive it was meant to exclude. +Checking for the name of a mechanism is not checking the mechanism. + +Failures print FAIL[]: ; nothing stops at the first one. Run: python tests/test_step7_mergegate.py """ @@ -21,6 +28,7 @@ import contextlib import io +import json import re import subprocess import sys @@ -35,6 +43,8 @@ import execbinding as eb # noqa: E402 import mergegate as mg # noqa: E402 +NL = chr(10) + _FAILURES: list[tuple[str, str]] = [] _PASSES: list[str] = [] @@ -56,12 +66,55 @@ def guarded(check: str, control: Callable[[], None]) -> None: fail(check, f"the control raised {type(exc).__name__}: {exc}") -HOSTQUAL_STUB = ('CAMPAIGN_LINK_SCHEMA = "own.net/p022/campaign-link"\n' - 'def check_campaign_link():\n pass\n' - 'def gate():\n if authorized is not True:\n return "refuse"\n') -STEP7_NOTE_REVOKED = ("Status:\n AUTOMATIC AUTHORISATION OF THE FIRST STEP-7 COLLECTION " - "IS REVOKED (T0-0).\n") -STEP7_NOTE_AUTO = "Status:\n SINGLE STEP-7 COLLECTION AUTHORISED automatically after that gate.\n" +NOTE_REVOKED = """Status: + AUTOMATIC AUTHORISATION OF THE FIRST STEP-7 COLLECTION IS REVOKED (T0-0). +""" +NOTE_AUTO = """Status: + SINGLE STEP-7 COLLECTION AUTHORISED automatically after that gate. +""" + +T0_TEMPLATE = """```text +Status: + {status} + collection_authorized: {flag} +``` + + measurement_harness_digest + {digest} +""" + +# The digest sits at a different exact path in each artifact, so a fixture that +# wrote one flat shape everywhere would be exercising a gate that does not exist. +BINDING_SHAPES = { + "docs/evidence/calibration/p022-263a-policy-freeze.json": + lambda d: {"measurement_harness_digest": d}, + "docs/evidence/calibration/p022-263a-design-constants.json": + lambda d: {"bound_measurement_harness_digest": d}, + "docs/evidence/calibration/p022-263a-training-preregistration.json": + lambda d: {"bindings": {"measurement_harness_digest": d}}, +} + +# Mutations appended to the real hostqual: a later definition wins, so each stub +# disables exactly one enforcement point and leaves the rest genuine. +NOOP_LINK = ''' + +def check_campaign_link(link, binding_path, repo): + return check("campaign_link", True, "stubbed: what a lexical gate accepted") +''' + +DEAD_AUTHORITY = ''' + +def bind_t0(repo, path, commit): + block = {"commit": commit, "path": path, "blob_sha": "b" * 40, + "sha256": "f" * 64, "status": "FROZEN", "collection_authorized": True} + return block, check("t0", True, "stubbed: always authorised") +''' + + +def real_tools() -> dict[str, str]: + """The tools as they actually are. A fixture that shipped a stub would prove + only that the gate accepts stubs.""" + return {t: (ROOT / t).read_text(encoding="utf-8") for t in mg.STEP7_TOOLS} def commit_tree(repo: Path, files: dict[str, str]) -> str: @@ -85,29 +138,37 @@ def run(*args: str) -> None: def world(tmp: Path, *, frozen: bool = True, authorized: bool = True, - instrument: str = "print('instrument')\n", rebound: bool = True, - machinery: bool = True, auto_authority: bool = False, - name: str = "w") -> tuple[Path, str]: + instrument: str = "print('instrument')" + NL, rebound: bool = True, + machinery: bool = True, auto_authority: bool = False, decoy: bool = False, + wrong_field: bool = False, mutate: str = "", name: str = "w") -> tuple[Path, str]: """A synthetic target tree, and the commit a merge into it would produce.""" repo = tmp / name - # The digest T0 will name is whatever this tree's own sources hash to, so a - # fixture cannot pass by agreeing with a constant this file also wrote. - files = {mg.STEP7_TOOLS[0]: "# capture\n", - eb.INSTRUMENT_SOURCES[0]: instrument, - eb.INSTRUMENT_SOURCES[1]: '{"decisive": []}\n', - mg.STEP7_NOTE: STEP7_NOTE_AUTO if auto_authority else STEP7_NOTE_REVOKED} + files = {eb.INSTRUMENT_SOURCES[0]: instrument, + eb.INSTRUMENT_SOURCES[1]: '{"decisive": []}' + NL, + mg.STEP7_NOTE: NOTE_AUTO if auto_authority else NOTE_REVOKED} if machinery: - files[mg.STEP7_TOOLS[1]] = HOSTQUAL_STUB - files[mg.STEP7_TOOLS[2]] = "# binding\n" + tools = real_tools() + if mutate: + tools[mg.STEP7_TOOLS[1]] = tools[mg.STEP7_TOOLS[1]] + mutate + files.update(tools) + else: + files[mg.STEP7_TOOLS[0]] = "# capture only" + NL probe = commit_tree(repo, files) + # T0 names whatever THIS tree's own sources hash to, so a fixture cannot pass + # by agreeing with a constant this file also wrote. digest = eb.harness_digest_at(repo, probe) or "" - status = "FROZEN." if frozen else "NOT_FROZEN." - flag = "true" if authorized else "false" - files[mg.T0_PATH] = (f"```text\nStatus:\n {status}\n collection_authorized: {flag}\n```\n\n" - f" measurement_harness_digest\n {digest}\n") + files[mg.T0_PATH] = T0_TEMPLATE.format( + status="FROZEN." if frozen else "NOT_FROZEN.", + flag="true" if authorized else "false", + digest=digest) bound = digest if rebound else "0" * 64 - for artifact in mg.BINDING_ARTIFACTS: - files[artifact] = '{"measurement_harness_digest": "' + bound + '"}\n' + for artifact, shape in BINDING_SHAPES.items(): + doc = shape(bound) + if decoy: # the right digest, in a field nobody binds + doc["decoy"] = digest + if wrong_field and "training" in artifact: # right digest, wrong exact path + doc = {"measurement_harness_digest": digest} + files[artifact] = json.dumps(doc, indent=2) + NL return repo, commit_tree(repo, files) @@ -115,13 +176,18 @@ def verdicts(repo: Path, commit: str) -> dict[str, str]: return {str(r["check"]): str(r["result"]) for r in mg.gate(repo, commit)} +def details(repo: Path, commit: str) -> dict[str, str]: + return {str(r["check"]): str(r["detail"]) for r in mg.gate(repo, commit)} + + def control_complete_tree() -> None: with tempfile.TemporaryDirectory() as raw: repo, commit = world(Path(raw)) - results = verdicts(repo, commit) - bad = sorted(k for k, v in results.items() if v != "pass") + bad = sorted(k for k, v in verdicts(repo, commit).items() if v != "pass") if bad: - fail("mergegate-complete-tree", f"a satisfying world was refused on {bad}") + seen = details(repo, commit) + fail("mergegate-complete-tree", + f"a satisfying world was refused on {bad}: {[seen[k][:90] for k in bad]}") return # Captured: a green run that prints a refusal teaches readers to skim past # refusals, which is how the word stops meaning anything. @@ -131,19 +197,17 @@ def control_complete_tree() -> None: fail("mergegate-complete-tree", "the CLI refused a satisfying world") return ok("mergegate-complete-tree", - "a tree carrying the named instrument, the re-bound artifacts, the machinery and the " - "revoked automatic authority is allowed") + "a tree carrying the named instrument, the exactly re-bound artifacts, the real tools " + "and the revoked automatic authority is allowed") def control_missing_instrument() -> None: with tempfile.TemporaryDirectory() as raw: - tmp = Path(raw) - repo, commit = world(tmp) + repo, commit = world(Path(raw)) # the instrument moves after T0 named it — the exact intermediate state a # stray merge produces when the repair PR is not in yet - drifted = commit_tree(repo, {eb.INSTRUMENT_SOURCES[0]: "print('older instrument')\n"}) - results = verdicts(repo, drifted) - if results.get("instrument_matches_t0") != "fail": + drifted = commit_tree(repo, {eb.INSTRUMENT_SOURCES[0]: "print('older')" + NL}) + if verdicts(repo, drifted).get("instrument_matches_t0") != "fail": fail("mergegate-missing-instrument", "a tree whose instrument does not hash to the digest T0 names was allowed") return @@ -168,10 +232,78 @@ def control_stale_bindings() -> None: "steps 4, 5 and 6 must be re-bound to the digest T0 names, not merely present") +def control_exact_binding_field() -> None: + """Each artifact binds at its own path, and only that path counts.""" + with tempfile.TemporaryDirectory() as raw: + tmp = Path(raw) + repo, commit = world(tmp, name="exact") + if verdicts(repo, commit).get("steps_4_5_6_rebound") != "pass": + fail("mergegate-exact-binding-field", + "correct exact fields were refused: " + + details(repo, commit)["steps_4_5_6_rebound"]) + return + moved_repo, moved = world(tmp, wrong_field=True, name="wrongfield") + if verdicts(moved_repo, moved).get("steps_4_5_6_rebound") != "fail": + fail("mergegate-exact-binding-field", + "the training preregistration bound at the wrong path was accepted; the gate " + "is searching the file rather than reading the binding") + return + ok("mergegate-exact-binding-field", + "policy freeze at measurement_harness_digest, design constants at " + "bound_measurement_harness_digest, training preregistration at " + "bindings.measurement_harness_digest — and nowhere else") + + +def control_decoy_digest() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), rebound=False, decoy=True) + if verdicts(repo, commit).get("steps_4_5_6_rebound") != "fail": + fail("mergegate-decoy-digest", + "a stale binding passed because the right digest sat in a decoy field") + return + ok("mergegate-decoy-digest", + "the right digest in a field nobody binds is not a binding, and a substring search " + "would have called it one") + + +def control_noop_campaign_link() -> None: + """The attack the previous revision of this gate could not see.""" + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), mutate=NOOP_LINK, name="noop") + if verdicts(repo, commit).get("step7_machinery_enforces") != "fail": + fail("mergegate-noop-campaign-link", + "a permissive check_campaign_link satisfied the gate; the name of a mechanism " + "is not the mechanism") + return + detail = details(repo, commit)["step7_machinery_enforces"] + if "accepted" not in detail: + fail("mergegate-noop-campaign-link", f"refused for an unrelated reason: {detail}") + return + ok("mergegate-noop-campaign-link", + "a tree whose link check always passes is refused, and the refusal names the attack " + "that got through") + + +def control_dead_authority() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), mutate=DEAD_AUTHORITY, name="deadauth") + if verdicts(repo, commit).get("step7_machinery_enforces") != "fail": + fail("mergegate-dead-authority", + "a bind_t0 that always returns pass satisfied the gate") + return + detail = details(repo, commit)["step7_machinery_enforces"] + if "accepted" not in detail: + fail("mergegate-dead-authority", f"refused for an unrelated reason: {detail}") + return + ok("mergegate-dead-authority", + "a tree whose authority check always passes is refused, and the refusal says which " + "state it wrongly accepted") + + def control_missing_machinery() -> None: with tempfile.TemporaryDirectory() as raw: repo, commit = world(Path(raw), machinery=False) - if verdicts(repo, commit).get("step7_machinery_present") != "fail": + if verdicts(repo, commit).get("step7_machinery_enforces") != "fail": fail("mergegate-missing-machinery", "a tree without the qualification and binding tools was allowed") return @@ -193,12 +325,13 @@ def control_lingering_auto() -> None: def control_unfrozen_t0() -> None: + states = (("not frozen", {"frozen": False}), + ("frozen but unauthorised", {"authorized": False}), + ("unfrozen but authorised", {"frozen": False, "authorized": True})) with tempfile.TemporaryDirectory() as raw: tmp = Path(raw) - for label, kwargs in (("not frozen", {"frozen": False}), - ("frozen but unauthorised", {"authorized": False}), - ("unfrozen but authorised", {"frozen": False, "authorized": True})): - repo, commit = world(tmp, name=f"w-{len(list(tmp.glob('w-*')))}", **kwargs) + for i, (label, kwargs) in enumerate(states): + repo, commit = world(tmp, name=f"state{i}", **kwargs) if verdicts(repo, commit).get("t0_frozen_and_authorized") != "fail": fail("mergegate-unfrozen-t0", f"{label} was treated as a freeze") return @@ -210,18 +343,18 @@ def control_unfrozen_t0() -> None: def control_reads_t0_digest() -> None: """The gate must not carry the expected digest in its own source.""" source = (ROOT / "scripts" / "step7" / "mergegate.py").read_text(encoding="utf-8") - hardcoded = re.findall(r"\b[0-9a-f]{64}\b", source) + hardcoded = re.findall("(? None: "no digest is written into this gate's own source") +def control_inventory_complete() -> None: + listed = set(re.findall("^ ([a-z0-9-]+) +[^ ]", __doc__ or "", re.MULTILINE)) + executed = {name for name, _ in CONTROLS} + if listed != executed: + fail("control-inventory-complete", + f"listed but not executed: {sorted(listed - executed)}; executed but not listed: " + f"{sorted(executed - listed)}") + return + ok("control-inventory-complete", + f"{len(executed)} controls listed, {len(executed)} executed, same names in both") + + CONTROLS: list[tuple[str, Callable[[], None]]] = [ ("mergegate-complete-tree", control_complete_tree), ("mergegate-missing-instrument", control_missing_instrument), ("mergegate-stale-bindings", control_stale_bindings), + ("mergegate-exact-binding-field", control_exact_binding_field), + ("mergegate-decoy-digest", control_decoy_digest), + ("mergegate-noop-campaign-link", control_noop_campaign_link), + ("mergegate-dead-authority", control_dead_authority), ("mergegate-missing-machinery", control_missing_machinery), ("mergegate-lingering-auto", control_lingering_auto), ("mergegate-unfrozen-t0", control_unfrozen_t0), ("mergegate-reads-t0-digest", control_reads_t0_digest), + ("control-inventory-complete", control_inventory_complete), ] From 4ce410123eb46ad5be13ccc831bf017761bad49b Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 08:12:31 +0500 Subject: [PATCH 3/9] fix(step7): the witness needs positive controls, or refusal proves nothing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two holes in the proof, not in the mechanism. The authority table exercised three of the four states. R15 defines all four, and `NOT_FROZEN+false` was the one nobody checked — so a future tree in which a reader quietly authorised it would have passed the gate. It is in the table now, and `DEAD_AUTHORITY` has a twin that breaks `execbinding.t0_at` instead of `hostqual.bind_t0`: the gate's message claims enforcement on both readers, and a control that only breaks the first left half that claim resting on nothing. Both mutations must now be caught naming all three forbidden states. The campaign-swap witness asserted only that a swapped campaign is inadmissible. A `session_admissibility` returning `admissible: False` unconditionally would have satisfied it — the witness would have read "nothing is admissible" as "the swap was caught". It now requires the unchanged campaign to survive preflight to postflight first, and requires the swapped one to be refused by a reason naming the campaign link, so a broken closing probe cannot stand in for the continuity check. Two mutations prove each half bites: removing only the preflight/current campaign-link comparison while leaving `check_campaign_link` intact, and a postflight that refuses everything. A mutation that changes no bytes now raises rather than passing as a second positive control nobody reads as one. 15 controls, 0 failed. The probes are unchanged: full chain rc 0, #354 alone rc 1 on three predicates, #354 without the tools rc 1 on one. 30d2f32 is untouched. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018xhcg5opoFbSTdYHpkSXCh --- scripts/step7/mergegate.py | 20 ++++- tests/test_step7_mergegate.py | 133 +++++++++++++++++++++++++++++----- 2 files changed, 131 insertions(+), 22 deletions(-) diff --git a/scripts/step7/mergegate.py b/scripts/step7/mergegate.py index fce995f6..db4c71ba 100644 --- a/scripts/step7/mergegate.py +++ b/scripts/step7/mergegate.py @@ -99,6 +99,7 @@ def w(name, doc): # --- W3/W4: the authority state must bite on both readers ------------------- for label, status, flag, must_pass in (("FROZEN+false", "FROZEN.", "false", False), ("NOT_FROZEN+true", "NOT_FROZEN.", "true", False), + ("NOT_FROZEN+false", "NOT_FROZEN.", "false", False), ("FROZEN+true", "FROZEN.", "true", True)): body = "```text\nStatus:\n %s\n collection_authorized: %s\n```\n" % (status, flag) root, commit = repo_with("t0-" + flag + status[:4], {"t0.md": body}) @@ -199,9 +200,20 @@ def link_for(binding_file, root, commit_sha, payload_file, att_file, blob_sha): if not pre["eligible"]: failures.append("a correctly linked session was refused: " + str(pre["reasons"])[:120]) ppath = w("preflight.json", pre) +# The positive half first. Without it a postflight hard-wired to inadmissible +# would satisfy the negative half, and this witness would read "everything is +# refused" as "the swap is refused" — a broken closing probe standing in for the +# check it was meant to prove. +kept = hq.session_admissibility(bpath, qpath, ppath, mpath, cand, probe, l2, r2) +if not kept["admissible"]: + failures.append("an unchanged campaign was inadmissible at postflight: " + + str(kept.get("reasons"))[:140]) after = hq.session_admissibility(bpath, qpath, ppath, mpath, cand, probe, l3, r3) if after["admissible"]: failures.append("a campaign swapped between preflight and postflight was admissible") +elif not any("campaign link" in str(r) for r in after.get("reasons", ())): + failures.append("the swapped campaign was refused, but by no reason naming the campaign " + "link: " + str(after.get("reasons"))[:140]) print(json.dumps(failures)) """ @@ -309,9 +321,11 @@ def gate(repo: Path, commit: str) -> list[dict[str, object]]: results.append(check( "step7_machinery_enforces", not refusals, "; ".join(refusals) if refusals else - "the target tree's own tools were run and refused all four attacks: a link naming " - "another binding, a freeze edited after linking, FROZEN with collection_authorized " - "false, and a campaign swapped between preflight and postflight")) + "the target tree's own tools were run: both readers refused all three forbidden " + "authority states and accepted FROZEN+true; a link naming another binding and a " + "freeze edited after linking were refused; an unchanged campaign survived " + "preflight to postflight and a swapped one was refused by campaign-link " + "continuity")) note = (blob(repo, commit, STEP7_NOTE) or b"").decode("utf-8", "replace") revoked = "AUTOMATIC AUTHORISATION OF THE FIRST STEP-7 COLLECTION IS REVOKED" in note diff --git a/tests/test_step7_mergegate.py b/tests/test_step7_mergegate.py index 29e3a332..2771bee5 100644 --- a/tests/test_step7_mergegate.py +++ b/tests/test_step7_mergegate.py @@ -7,17 +7,22 @@ mergegate-exact-binding-field the digest is read at each artifact's own path mergegate-decoy-digest the right digest in a field nobody binds is not a binding mergegate-noop-campaign-link a permissive link check must not satisfy the gate - mergegate-dead-authority an authority check that always passes must not satisfy it + mergegate-dead-authority hostqual authorising anything must not satisfy it + mergegate-dead-authority-eb nor must execbinding — the claim covers both readers + mergegate-dead-continuity nor must a postflight that stops comparing campaign links + mergegate-always-inadmissible nor must one that refuses every campaign mergegate-missing-machinery the tools must be there at all mergegate-lingering-auto the revoked automatic authority must stay revoked mergegate-unfrozen-t0 an unfrozen or unauthorised T0 is not merged as frozen mergegate-reads-t0-digest the expected digest comes from T0, not from a constant control-inventory-complete this list and the executed set are the same set -Fixtures ship the real tools, and two controls mutate them. An earlier revision -shipped a stub whose check_campaign_link was a bare pass and called that world -compliant: the fixture demonstrated the false positive it was meant to exclude. -Checking for the name of a mechanism is not checking the mechanism. +Fixtures ship the real tools, and four controls mutate one enforcement point +each. An earlier revision shipped a stub whose check_campaign_link was a bare +pass and called that world compliant: the fixture demonstrated the false +positive it was meant to exclude. Checking for the name of a mechanism is not +checking the mechanism, and a witness that only ever sees refusals is not +checking one either — hence a positive control beside each negative one. Failures print FAIL[]: ; nothing stops at the first one. @@ -94,8 +99,11 @@ def guarded(check: str, control: Callable[[], None]) -> None: lambda d: {"bindings": {"measurement_harness_digest": d}}, } -# Mutations appended to the real hostqual: a later definition wins, so each stub -# disables exactly one enforcement point and leaves the rest genuine. +HOSTQUAL, EXECBINDING = mg.STEP7_TOOLS[1], mg.STEP7_TOOLS[2] + +# Mutations of the real tools, one enforcement point each. Appending a definition +# is enough where a later definition wins; the continuity check lives inside a +# larger function, so that one is a targeted edit of its condition. NOOP_LINK = ''' def check_campaign_link(link, binding_path, repo): @@ -110,6 +118,33 @@ def bind_t0(repo, path, commit): return block, check("t0", True, "stubbed: always authorised") ''' +# The second reader. The gate's message claims enforcement on both, so a control +# that only breaks the first leaves half of that claim unproved. +DEAD_AUTHORITY_EB = ''' + +def t0_at(repo, path, commit): + return {"commit": commit, "path": path, "blob_sha": "b" * 40, "sha256": "f" * 64, + "status": "FROZEN", "collection_authorized": True} +''' + +# The mutation the positive postflight control exists to catch: a tool that +# refuses every campaign satisfies a negative-only witness, which would then be +# reading "nothing is admissible" as "the swap was caught". +ALWAYS_INADMISSIBLE = ''' + +def session_admissibility(*args, **kwargs): + return {"admissible": False, "reasons": ["stubbed: nothing is ever admissible"]} +''' + +CONTINUITY_CONDITION = ('preflight.get("campaign_link_sha256") != ' + "sha256_file(campaign_link_path)") + + +def dead_continuity(src: str) -> str: + """Leave check_campaign_link working and remove only the comparison of the + preflight's campaign link with this pass's.""" + return src.replace(CONTINUITY_CONDITION, "False") + def real_tools() -> dict[str, str]: """The tools as they actually are. A fixture that shipped a stub would prove @@ -140,7 +175,8 @@ def run(*args: str) -> None: def world(tmp: Path, *, frozen: bool = True, authorized: bool = True, instrument: str = "print('instrument')" + NL, rebound: bool = True, machinery: bool = True, auto_authority: bool = False, decoy: bool = False, - wrong_field: bool = False, mutate: str = "", name: str = "w") -> tuple[Path, str]: + wrong_field: bool = False, mutate: dict[str, object] | None = None, + name: str = "w") -> tuple[Path, str]: """A synthetic target tree, and the commit a merge into it would produce.""" repo = tmp / name files = {eb.INSTRUMENT_SOURCES[0]: instrument, @@ -148,8 +184,14 @@ def world(tmp: Path, *, frozen: bool = True, authorized: bool = True, mg.STEP7_NOTE: NOTE_AUTO if auto_authority else NOTE_REVOKED} if machinery: tools = real_tools() - if mutate: - tools[mg.STEP7_TOOLS[1]] = tools[mg.STEP7_TOOLS[1]] + mutate + for tool, change in (mutate or {}).items(): + before = tools[tool] + tools[tool] = change(before) if callable(change) else before + str(change) + if tools[tool] == before: + # A mutation that lands nowhere turns an attack control into a + # second positive control that nobody reads as one. + raise AssertionError(f"the mutation for {tool} changed nothing; the source it " + "targets has moved") files.update(tools) else: files[mg.STEP7_TOOLS[0]] = "# capture only" + NL @@ -269,7 +311,7 @@ def control_decoy_digest() -> None: def control_noop_campaign_link() -> None: """The attack the previous revision of this gate could not see.""" with tempfile.TemporaryDirectory() as raw: - repo, commit = world(Path(raw), mutate=NOOP_LINK, name="noop") + repo, commit = world(Path(raw), mutate={HOSTQUAL: NOOP_LINK}, name="noop") if verdicts(repo, commit).get("step7_machinery_enforces") != "fail": fail("mergegate-noop-campaign-link", "a permissive check_campaign_link satisfied the gate; the name of a mechanism " @@ -284,20 +326,70 @@ def control_noop_campaign_link() -> None: "that got through") +def _dead_authority(check_name: str, tool: str, stub: str, reader: str) -> None: + """Both readers carry the authority state machine, and the gate says so. + A control that breaks one of them leaves the other half of that claim + standing on nothing.""" + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), mutate={tool: stub}, name="deadauth") + if verdicts(repo, commit).get("step7_machinery_enforces") != "fail": + fail(check_name, f"a {reader} authority check that always passes satisfied the gate") + return + detail = details(repo, commit)["step7_machinery_enforces"] + wrongly = [s for s in ("FROZEN+false", "NOT_FROZEN+true", "NOT_FROZEN+false") + if f"{reader} accepted {s}" in detail] + if len(wrongly) != 3: + fail(check_name, f"refused, but named {wrongly} rather than all three forbidden " + f"states: {detail}") + return + ok(check_name, + f"a tree whose {reader} authorises anything is refused, and the refusal names every " + "forbidden state it accepted: FROZEN+false, NOT_FROZEN+true, NOT_FROZEN+false") + + def control_dead_authority() -> None: + _dead_authority("mergegate-dead-authority", HOSTQUAL, DEAD_AUTHORITY, "hostqual") + + +def control_dead_authority_eb() -> None: + _dead_authority("mergegate-dead-authority-eb", EXECBINDING, DEAD_AUTHORITY_EB, "execbinding") + + +def control_dead_continuity() -> None: + """The campaign-swap witness needs its own sensitivity proof: a postflight + that refuses everything would satisfy a negative-only control.""" with tempfile.TemporaryDirectory() as raw: - repo, commit = world(Path(raw), mutate=DEAD_AUTHORITY, name="deadauth") + repo, commit = world(Path(raw), mutate={HOSTQUAL: dead_continuity}, name="deadcont") if verdicts(repo, commit).get("step7_machinery_enforces") != "fail": - fail("mergegate-dead-authority", - "a bind_t0 that always returns pass satisfied the gate") + fail("mergegate-dead-continuity", + "a postflight that no longer compares the preflight's campaign link with " + "this pass's satisfied the gate; the swap witness proves nothing") return detail = details(repo, commit)["step7_machinery_enforces"] - if "accepted" not in detail: - fail("mergegate-dead-authority", f"refused for an unrelated reason: {detail}") + if "swapped" not in detail: + fail("mergegate-dead-continuity", f"refused for an unrelated reason: {detail}") + return + ok("mergegate-dead-continuity", + "removing only the preflight/postflight campaign-link comparison, and leaving " + "check_campaign_link intact, is caught and named as the swap becoming admissible") + + +def control_always_inadmissible() -> None: + """A refusal is only evidence if acceptance was possible.""" + with tempfile.TemporaryDirectory() as raw: + repo, commit = world(Path(raw), mutate={HOSTQUAL: ALWAYS_INADMISSIBLE}, name="noadmit") + if verdicts(repo, commit).get("step7_machinery_enforces") != "fail": + fail("mergegate-always-inadmissible", + "a postflight that refuses every campaign satisfied the gate; the swap " + "witness was reading a blanket refusal as enforcement") + return + detail = details(repo, commit)["step7_machinery_enforces"] + if "unchanged campaign was inadmissible" not in detail: + fail("mergegate-always-inadmissible", f"refused for another reason: {detail}") return - ok("mergegate-dead-authority", - "a tree whose authority check always passes is refused, and the refusal says which " - "state it wrongly accepted") + ok("mergegate-always-inadmissible", + "the unchanged campaign must survive preflight to postflight, so a tool that refuses " + "everything cannot pose as one that caught the swap") def control_missing_machinery() -> None: @@ -384,6 +476,9 @@ def control_inventory_complete() -> None: ("mergegate-decoy-digest", control_decoy_digest), ("mergegate-noop-campaign-link", control_noop_campaign_link), ("mergegate-dead-authority", control_dead_authority), + ("mergegate-dead-authority-eb", control_dead_authority_eb), + ("mergegate-dead-continuity", control_dead_continuity), + ("mergegate-always-inadmissible", control_always_inadmissible), ("mergegate-missing-machinery", control_missing_machinery), ("mergegate-lingering-auto", control_lingering_auto), ("mergegate-unfrozen-t0", control_unfrozen_t0), From 96f6377f1d2484cb0f51b18f98abdc5f5a7dfe9b Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 08:21:01 +0500 Subject: [PATCH 4/9] ci(step7): the merge gate becomes a check, not a script in a drawer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #357 built a gate that refuses the wrong merge. Nothing ran it. A weapon in a safe is not a guard at the door, so this is the wiring: a workflow that runs it against the commit a merge would produce, under two names a ruleset can require. Wiring it exposed two things the gate could not answer on its own. **Applicability.** A required check runs on every pull request in the repository. `mergegate.py` refuses a tree with no T0 in it — correctly, as a question about a contract that must be there — so as a required check today it would refuse every pull request, including this one. The wrapper settles it first: no frozen T0 at the merge commit means no merge can make the contract reachable, the check passes and says so. The moment a tree carries T0 every predicate applies in full. **Co-change.** The workflow that runs on a pull request is the one on that pull request. A branch could otherwise carry the freeze and a weakened gate together and be judged by the gate it brought with it. So the gate's own files — `mergegate.py`, this wrapper, the workflow — may not change in the same merge that introduces or changes the frozen contract. Repairing the gate on its own stays ordinary work; that distinction is a control, not a promise. The applicability decision lives in the tool that owns `T0_PATH`, never a second copy in YAML, and a control fails if the workflow ever grows one. Another fails if the contexts the workflow declares stop matching the ones a ruleset is told to require — a required check nothing reports waits forever and reads as protection. Exit codes are the step-7 three: 0 proceed or not applicable, 1 refused, 2 the question could not be asked. 11 wiring controls, 0 failed; the 15 gate controls unchanged. Both suites also run green on Linux, which is where the job will run. Against real commits: this branch's head is not applicable, the full chain is allowed, #354 alone is refused on three predicates. This is half the wiring. Until a ruleset requires these two contexts, the job reports and nothing is prevented. #354 stays a draft until it does. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018xhcg5opoFbSTdYHpkSXCh --- .github/workflows/p022-merge-gate.yml | 70 +++++++ scripts/step7/mergegate_ci.py | 125 ++++++++++++ tests/test_step7_mergegate_ci.py | 277 ++++++++++++++++++++++++++ 3 files changed, 472 insertions(+) create mode 100644 .github/workflows/p022-merge-gate.yml create mode 100644 scripts/step7/mergegate_ci.py create mode 100644 tests/test_step7_mergegate_ci.py diff --git a/.github/workflows/p022-merge-gate.yml b/.github/workflows/p022-merge-gate.yml new file mode 100644 index 00000000..db536d73 --- /dev/null +++ b/.github/workflows/p022-merge-gate.yml @@ -0,0 +1,70 @@ +name: P-022 merge gate + +# The merge order for the #263 preregistration, as a check rather than an +# agreement. `mergegate.py` is run against the commit a merge would PRODUCE — +# for a pull_request event that is `github.sha`, the merge ref, not the branch +# head — because a PR number proves someone pressed a button and says nothing +# about what the merged tree contains. +# +# Two jobs, and the gate needs the controls: a gate whose own controls are +# failing must not be the thing that says a merge is safe. A failed controls job +# leaves the gate job unreported, which a required check reads as unsatisfied. +# +# This workflow is one half of the wiring. The other half is a ruleset requiring +# these two checks; without it the job reports and nothing is prevented. The +# names below are the contexts that ruleset must name: +# +# P-022 merge gate controls +# P-022 merge gate +# +# Least privilege as elsewhere in this repository: read-only, and every +# third-party `uses:` pinned to a commit SHA with its version in a comment. + +permissions: + contents: read + +on: + pull_request: + workflow_dispatch: + +jobs: + controls: + name: P-022 merge gate controls + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: "3.11" + - name: the gate's own controls (fixtures ship the real tools and mutate them) + run: python tests/test_step7_mergegate.py + - name: the wiring's controls (applicability and co-change) + run: python tests/test_step7_mergegate_ci.py + + gate: + name: P-022 merge gate + needs: controls + runs-on: ubuntu-latest + steps: + # Full history: the co-change rule compares the gate's own files against + # the base commit, which a shallow clone would not carry. + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 + with: + fetch-depth: 0 + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: "3.11" + - name: the gate, against the commit this merge would produce + env: + # Read through the environment rather than interpolated into the + # script body, so nothing from the event can be read as shell. + MERGE_COMMIT: ${{ github.sha }} + BASE_COMMIT: ${{ github.event.pull_request.base.sha }} + run: | + set -euo pipefail + if [ -n "${BASE_COMMIT}" ]; then + python scripts/step7/mergegate_ci.py --repo . \ + --commit "${MERGE_COMMIT}" --base "${BASE_COMMIT}" + else + python scripts/step7/mergegate_ci.py --repo . --commit "${MERGE_COMMIT}" + fi diff --git a/scripts/step7/mergegate_ci.py b/scripts/step7/mergegate_ci.py new file mode 100644 index 00000000..4709ab13 --- /dev/null +++ b/scripts/step7/mergegate_ci.py @@ -0,0 +1,125 @@ +#!/usr/bin/env python3 +"""P-022 — the merge gate as a status check. + +`mergegate.py` answers one question: may the frozen contract become reachable +from this tree? It answers it about a tree that *has* a frozen contract. A +required status check runs on every pull request in the repository, including +the ones that have nothing to do with P-022, so two rules have to be settled +before that gate can be wired to a branch protection rule. + +**Applicability.** If the frozen T0 does not exist at the merge commit, no merge +can make it reachable and there is nothing to protect. The check passes and says +so. This is not a way around the gate: the moment a tree carries T0, every +predicate applies in full. + +**Co-change.** A required check that a pull request can rewrite is decoration. +The workflow that runs on a pull request is the one *on that pull request*, so a +branch may carry both the freeze and a weakened gate and be judged by the gate +it brought with it. So: **the gate's own files may not change in the same merge +that introduces or changes the frozen contract.** Repairing the gate is still +ordinary work; doing it in the same breath as the freeze is not. + +Neither rule is enforcement isolation. A repository administrator can turn the +protection off, and this script cannot see that. What it removes is the quiet +path — a weakened gate arriving as part of the change it was meant to judge. + +Exit codes: 0 the merge may proceed (or the gate does not apply), 1 it is +refused, 2 the question could not be asked. + + python scripts/step7/mergegate_ci.py --repo . --commit \ + --base + +`--base` is what the co-change rule compares against. Without it the rule cannot +run, and the output says that rather than implying it passed. +""" + +from __future__ import annotations + +import argparse +import subprocess +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +import mergegate as mg # noqa: E402 + +# The gate's own surface: the predicates, this wrapper, and the workflow that +# runs them. A change to any of these changes what "the gate passed" means. +GATE_FILES = ("scripts/step7/mergegate.py", + "scripts/step7/mergegate_ci.py", + ".github/workflows/p022-merge-gate.yml") + + +def blob_sha(repo: Path, commit: str, path: str) -> str | None: + """The blob a path resolves to, or None when the path is not there.""" + proc = subprocess.run(["git", "-C", str(repo), "rev-parse", f"{commit}:{path}"], + capture_output=True, text=True, check=False) + return proc.stdout.strip() if proc.returncode == 0 else None + + +def applies(repo: Path, commit: str) -> bool: + return blob_sha(repo, commit, mg.T0_PATH) is not None + + +def co_change(repo: Path, commit: str, base: str) -> list[str]: + """Which gate files this merge changes, if it also touches the contract. + + Empty when the contract is untouched: repairing the gate is ordinary work. + """ + t0_now, t0_base = blob_sha(repo, commit, mg.T0_PATH), blob_sha(repo, base, mg.T0_PATH) + if t0_now == t0_base: + return [] + return [path for path in GATE_FILES + if blob_sha(repo, commit, path) != blob_sha(repo, base, path)] + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo", type=Path, default=Path(".")) + parser.add_argument("--commit", required=True, + help="the commit a merge would produce, not the branch head") + parser.add_argument("--base", help="the base commit the co-change rule compares against") + args = parser.parse_args(argv) + + proc = subprocess.run(["git", "-C", str(args.repo), "rev-parse", f"{args.commit}^{{commit}}"], + capture_output=True, text=True, check=False) + if proc.returncode != 0: + print(f"merge gate: MISUSE — {args.commit} is not a commit in {args.repo}") + return 2 + + if not applies(args.repo, args.commit): + print(f"merge gate: not applicable — {mg.T0_PATH} does not exist at {args.commit[:12]}, " + "so no merge here can make the frozen contract reachable. Every predicate " + "applies the moment a tree carries it.") + return 0 + + if args.base is None: + print("merge gate: the co-change rule did not run — no --base was given, so this pass " + "cannot say whether the gate itself changed alongside the contract") + else: + changed = co_change(args.repo, args.commit, args.base) + if changed: + print("merge gate: REFUSED — this merge changes the frozen contract and the gate " + f"that judges it in one act: {', '.join(changed)}. Repair the gate in its own " + "merge; a guard that arrives with what it guards is not a guard.") + return 1 + print("ok [gate_unchanged] the contract moves and the gate does not; this merge is " + "judged by the gate already on the base") + + results = mg.gate(args.repo, args.commit) + for row in results: + mark = "ok " if row["result"] == "pass" else "FAIL" + print(f"{mark} [{row['check']}] {row['detail']}") + bad = [r for r in results if r["result"] != "pass"] + print() + if bad: + print(f"merge gate: REFUSED, {len(bad)} predicate(s) unsatisfied — the frozen contract " + "must not become reachable from this tree") + return 1 + print("merge gate: allowed") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_step7_mergegate_ci.py b/tests/test_step7_mergegate_ci.py new file mode 100644 index 00000000..b85720b2 --- /dev/null +++ b/tests/test_step7_mergegate_ci.py @@ -0,0 +1,277 @@ +#!/usr/bin/env python3 +"""P-022 — controls on the wiring that makes the merge gate a status check. + + ci-not-applicable a tree without the frozen T0 passes, and says why + ci-applies-in-full a tree with it is judged by every predicate + ci-refusal-survives a refusal from the gate is still a refusal here + ci-co-change-refused the gate may not change in the merge that brings the contract + ci-gate-repair-allowed repairing the gate on its own is ordinary work + ci-no-base-is-stated without a base the co-change rule says it did not run + ci-misuse-is-two a commit that does not exist is misuse, not a verdict + ci-gate-files-exist every file the co-change rule watches is really there + ci-workflow-names the workflow names the contexts a ruleset must require + ci-workflow-derives-t0 the workflow hard-codes no path the tools already own + control-inventory this list and the executed set are the same set + +The applicability rule is the one that makes a required check possible at all: +without it the gate refuses every pull request in the repository, including the +one that adds it. The co-change rule is the one that makes it worth requiring: +a pull request runs its own copy of the workflow, so a branch could otherwise +carry the freeze and a weakened gate together and be judged by the gate it +brought with it. + +Run: python tests/test_step7_mergegate_ci.py +""" + +from __future__ import annotations + +import contextlib +import io +import sys +import tempfile +from collections.abc import Callable +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT / "scripts" / "step7")) +sys.path.insert(0, str(ROOT / "tests")) + +import mergegate_ci as ci # noqa: E402 +import test_step7_mergegate as mgt # noqa: E402 + +WORKFLOW = ROOT / ".github/workflows/p022-merge-gate.yml" +CONTEXTS = ("P-022 merge gate controls", "P-022 merge gate") + +_FAILURES: list[tuple[str, str]] = [] +_PASSES: list[str] = [] + + +def fail(check: str, detail: str) -> None: + _FAILURES.append((check, detail)) + print(f"FAIL[{check}]: {detail}") + + +def ok(check: str, detail: str = "") -> None: + _PASSES.append(check) + print(f"ok[{check}]: {detail}" if detail else f"ok[{check}]") + + +def guarded(check: str, control: Callable[[], None]) -> None: + try: + control() + except Exception as exc: + fail(check, f"the control raised {type(exc).__name__}: {exc}") + + +def run(*args: str) -> tuple[int, str]: + """The wrapper's exit code and everything it said, captured. A green run that + prints REFUSED teaches readers to skim past refusals.""" + out = io.StringIO() + with contextlib.redirect_stdout(out): + rc = ci.main(list(args)) + return rc, out.getvalue() + + +def wired(tmp: Path, name: str, **kwargs: object) -> tuple[Path, str, str]: + """A repository, the commit that carries T0, and the commit before it. + + `world()` builds its tree in two commits: the first carries the instrument + and the tools, the second adds T0 and the bindings. That first commit is + exactly the shape of a base branch that does not yet carry the contract. + """ + repo, head = mgt.world(tmp, name=name, **kwargs) + return repo, head, head + "^" + + +def control_not_applicable() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, _, before = wired(Path(raw), "na") + rc, said = run("--repo", str(repo), "--commit", before) + if rc != 0: + fail("ci-not-applicable", f"a tree with no frozen contract was refused: {said[:160]}") + return + if "not applicable" not in said: + fail("ci-not-applicable", f"it passed without saying why: {said[:160]}") + return + ok("ci-not-applicable", + "with no T0 in the tree no merge can make the frozen contract reachable, so the check " + "passes and names the reason — otherwise it would refuse every pull request in the repo") + + +def control_applies_in_full() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, head, before = wired(Path(raw), "full") + rc, said = run("--repo", str(repo), "--commit", head, "--base", before) + if rc != 0: + fail("ci-applies-in-full", f"a satisfying tree was refused: {said[:200]}") + return + missing = [c for c in ("t0_frozen_and_authorized", "instrument_matches_t0", + "steps_4_5_6_rebound", "step7_machinery_enforces") + if c not in said] + if missing: + fail("ci-applies-in-full", f"the wrapper did not report {missing}") + return + ok("ci-applies-in-full", + "the moment a tree carries T0 every predicate runs and is reported by name") + + +def control_refusal_survives() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, head, before = wired(Path(raw), "stale", rebound=False) + rc, said = run("--repo", str(repo), "--commit", head, "--base", before) + if rc != 1 or "REFUSED" not in said: + fail("ci-refusal-survives", + f"a stale binding came back as rc={rc}: {said[:200]}") + return + ok("ci-refusal-survives", + "the wrapper adds rules, it does not soften the ones underneath: a stale binding is " + "still exit 1") + + +def control_co_change_refused() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, head, before = wired(Path(raw), "cochange") + # the branch brings the contract and its own copy of the gate at once + attacked = mgt.commit_tree(repo, { + ci.GATE_FILES[0]: "# a gate that says yes\n", + ci.GATE_FILES[2]: "name: P-022 merge gate\n"}) + rc, said = run("--repo", str(repo), "--commit", attacked, "--base", before) + if rc != 1: + fail("ci-co-change-refused", + "a merge carrying both the freeze and a rewritten gate was allowed") + return + if ci.GATE_FILES[0] not in said: + fail("ci-co-change-refused", f"refused without naming the changed file: {said[:200]}") + return + _ = head + ok("ci-co-change-refused", + "a guard that arrives together with what it guards is not a guard; the refusal names " + "every gate file the merge rewrites") + + +def control_gate_repair_allowed() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, head, _ = wired(Path(raw), "repair") + # the contract is already in the base and does not move; only the gate does + repaired = mgt.commit_tree(repo, {ci.GATE_FILES[0]: "# a repaired gate\n"}) + rc, said = run("--repo", str(repo), "--commit", repaired, "--base", head) + if rc != 0: + fail("ci-gate-repair-allowed", + f"an ordinary repair of the gate was refused: {said[:200]}") + return + if "gate_unchanged" not in said: + fail("ci-gate-repair-allowed", f"it passed without saying so: {said[:200]}") + return + ok("ci-gate-repair-allowed", + "when the contract does not move, changing the gate is ordinary work and stays " + "possible — the rule is about co-change, not about freezing the gate forever") + + +def control_no_base_is_stated() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, head, _ = wired(Path(raw), "nobase") + rc, said = run("--repo", str(repo), "--commit", head) + if rc != 0: + fail("ci-no-base-is-stated", f"a satisfying tree was refused: {said[:200]}") + return + if "did not run" not in said: + fail("ci-no-base-is-stated", + f"a pass with no base did not say the co-change rule was skipped: {said[:200]}") + return + ok("ci-no-base-is-stated", + "a manual run cannot pose as a full one: without a base the wrapper says the co-change " + "rule did not run rather than implying it held") + + +def control_misuse_is_two() -> None: + with tempfile.TemporaryDirectory() as raw: + repo, _, _ = wired(Path(raw), "misuse") + rc, said = run("--repo", str(repo), "--commit", "f" * 40) + if rc != 2: + fail("ci-misuse-is-two", f"a commit that does not exist returned {rc}: {said[:160]}") + return + ok("ci-misuse-is-two", + "a question that could not be asked is exit 2, never a verdict — the same three codes " + "the step-7 tools already use") + + +def control_gate_files_exist() -> None: + missing = [p for p in ci.GATE_FILES if not (ROOT / p).exists()] + if missing: + fail("ci-gate-files-exist", + f"the co-change rule watches {missing}, which are not in this repository; a " + "renamed file would be watched by nobody") + return + ok("ci-gate-files-exist", + f"all {len(ci.GATE_FILES)} watched files exist: {', '.join(ci.GATE_FILES)}") + + +def control_workflow_names() -> None: + if not WORKFLOW.exists(): + fail("ci-workflow-names", f"{WORKFLOW.name} does not exist") + return + text = WORKFLOW.read_text(encoding="utf-8") + absent = [c for c in CONTEXTS if f"name: {c}" not in text] + if absent: + fail("ci-workflow-names", + f"the workflow declares no job named {absent}; a ruleset requiring that context " + "would wait forever on a check nothing reports") + return + if "mergegate_ci.py" not in text: + fail("ci-workflow-names", "the workflow does not run the wrapper") + return + ok("ci-workflow-names", + f"the workflow declares exactly the contexts a ruleset must require: {', '.join(CONTEXTS)}") + + +def control_workflow_derives_t0() -> None: + """The workflow must not carry a second copy of a path the tools own.""" + import mergegate as mg + text = WORKFLOW.read_text(encoding="utf-8") + if mg.T0_PATH in text: + fail("ci-workflow-derives-t0", + f"the workflow hard-codes {mg.T0_PATH}; applicability would then be decided in two " + "places that can drift apart") + return + ok("ci-workflow-derives-t0", + "applicability is decided once, in the tool that owns T0_PATH; the workflow only runs it") + + +def control_inventory() -> None: + import re + listed = set(re.findall("^ (ci-[a-z0-9-]+|control-inventory) +[^ ]", __doc__ or "", + re.MULTILINE)) + executed = {name for name, _ in CONTROLS} + if listed != executed: + fail("control-inventory", + f"listed but not executed: {sorted(listed - executed)}; executed but not listed: " + f"{sorted(executed - listed)}") + return + ok("control-inventory", f"{len(executed)} controls listed, {len(executed)} executed") + + +CONTROLS: list[tuple[str, Callable[[], None]]] = [ + ("ci-not-applicable", control_not_applicable), + ("ci-applies-in-full", control_applies_in_full), + ("ci-refusal-survives", control_refusal_survives), + ("ci-co-change-refused", control_co_change_refused), + ("ci-gate-repair-allowed", control_gate_repair_allowed), + ("ci-no-base-is-stated", control_no_base_is_stated), + ("ci-misuse-is-two", control_misuse_is_two), + ("ci-gate-files-exist", control_gate_files_exist), + ("ci-workflow-names", control_workflow_names), + ("ci-workflow-derives-t0", control_workflow_derives_t0), + ("control-inventory", control_inventory), +] + + +def main() -> int: + for name, control in CONTROLS: + guarded(name, control) + print() + print(f"merge gate wiring controls: {len(_PASSES)} passed, {len(_FAILURES)} failed") + return 1 if _FAILURES else 0 + + +if __name__ == "__main__": + sys.exit(main()) From 61cfea975ec7a0406526838f5e610ece9addaa89 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 08:32:03 +0500 Subject: [PATCH 5/9] style(step7): drop a noqa the linter does not need ruff allows a sys.path insertion before an import, so the E402 suppression was covering nothing and RUF100 said so. Caught by running the CI-pinned ruff (0.15.8) rather than the one that happened to be installed. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018xhcg5opoFbSTdYHpkSXCh --- scripts/step7/mergegate_ci.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/step7/mergegate_ci.py b/scripts/step7/mergegate_ci.py index 4709ab13..022dc9c4 100644 --- a/scripts/step7/mergegate_ci.py +++ b/scripts/step7/mergegate_ci.py @@ -42,7 +42,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parent)) -import mergegate as mg # noqa: E402 +import mergegate as mg # The gate's own surface: the predicates, this wrapper, and the workflow that # runs them. A change to any of these changes what "the gate passed" means. From 9bed20f70f4912e709f8a76713452b47fd31c937 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 08:49:26 +0500 Subject: [PATCH 6/9] style(step7): the gate's own source passes its own repository's linter Three ruff findings, all mine: an unsorted import block, one over-long line inside the witness source, and an unpacked name a control never reads. The long line is inside WITNESS_SOURCE, so wrapping it edits the script the gate runs against the target tree's tools. Hoisting the identity into a local changes nothing it does, and the controls and probes were re-run to say so rather than assumed: 15 merge gate controls pass, host qualification 26/26, and the three real merge probes still come back 0 / 1 / 1 with the same predicates. This is a lint pass only. No predicate, no attack and no message moved, and 30d2f32 is untouched. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018xhcg5opoFbSTdYHpkSXCh --- scripts/step7/mergegate.py | 7 ++++--- tests/test_step7_mergegate.py | 2 +- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/scripts/step7/mergegate.py b/scripts/step7/mergegate.py index db4c71ba..2ef635f3 100644 --- a/scripts/step7/mergegate.py +++ b/scripts/step7/mergegate.py @@ -30,7 +30,6 @@ from __future__ import annotations import argparse - import json import re import subprocess @@ -154,9 +153,11 @@ def w(name, doc): CAND = b"candidate bytes" cand = tmp / "cand.bin"; cand.write_bytes(CAND) qpath = w("qual.json", QUAL) +IDENT = QUAL["environment_identity_sha256"] stratum_block = lambda m, extra: {"qualification_sha256": hq.sha256_file(qpath), - "environment_id": "env-1", "host_fingerprint": "sha256:abc", - "environment_identity_sha256": QUAL["environment_identity_sha256"], + "environment_id": "env-1", + "host_fingerprint": "sha256:abc", + "environment_identity_sha256": IDENT, "candidate_sha256": hq.sha256_bytes(CAND), "candidate_bytes": len(CAND), "memory_metric": m, **extra} BINDING = {"kind": eb.BINDING_SCHEMA, "schema": 1, diff --git a/tests/test_step7_mergegate.py b/tests/test_step7_mergegate.py index 2771bee5..1593235f 100644 --- a/tests/test_step7_mergegate.py +++ b/tests/test_step7_mergegate.py @@ -245,7 +245,7 @@ def control_complete_tree() -> None: def control_missing_instrument() -> None: with tempfile.TemporaryDirectory() as raw: - repo, commit = world(Path(raw)) + repo, _commit = world(Path(raw)) # the instrument moves after T0 named it — the exact intermediate state a # stray merge produces when the repair PR is not in yet drifted = commit_tree(repo, {eb.INSTRUMENT_SOURCES[0]: "print('older')" + NL}) From 1e488fdf87902f2bf169b0c7507bc1773fdb41cc Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 08:54:41 +0500 Subject: [PATCH 7/9] fix(step7): an unreachable base is a question unasked, not a refusal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit If `--base` named a commit the clone does not carry, `blob_sha` returned None for every gate file on that side, the co-change rule read all of them as rewritten, and the merge was REFUSED — for a fact about the checkout's depth rather than about the merge. The workflow sets fetch-depth 0, so this would not have fired today; it would have fired the first time someone ran the wrapper by hand, or the first time a shallow clone was used to save a minute, and the refusal would have been about the gate being rewritten. A false refusal teaches people that the gate is noise, which is how a required check dies. The base is now validated like the merge commit, and an unreachable one is exit 2 with a message that names the reason. A control builds exactly that case and requires 2, refusing to accept 1. 12 wiring controls, 0 failed. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018xhcg5opoFbSTdYHpkSXCh --- scripts/step7/mergegate_ci.py | 17 ++++++++++++++--- tests/test_step7_mergegate_ci.py | 20 ++++++++++++++++++++ 2 files changed, 34 insertions(+), 3 deletions(-) diff --git a/scripts/step7/mergegate_ci.py b/scripts/step7/mergegate_ci.py index 022dc9c4..1d839e9c 100644 --- a/scripts/step7/mergegate_ci.py +++ b/scripts/step7/mergegate_ci.py @@ -62,6 +62,11 @@ def applies(repo: Path, commit: str) -> bool: return blob_sha(repo, commit, mg.T0_PATH) is not None +def is_commit(repo: Path, rev: str) -> bool: + return subprocess.run(["git", "-C", str(repo), "rev-parse", f"{rev}^{{commit}}"], + capture_output=True, check=False).returncode == 0 + + def co_change(repo: Path, commit: str, base: str) -> list[str]: """Which gate files this merge changes, if it also touches the contract. @@ -82,11 +87,17 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--base", help="the base commit the co-change rule compares against") args = parser.parse_args(argv) - proc = subprocess.run(["git", "-C", str(args.repo), "rev-parse", f"{args.commit}^{{commit}}"], - capture_output=True, text=True, check=False) - if proc.returncode != 0: + if not is_commit(args.repo, args.commit): print(f"merge gate: MISUSE — {args.commit} is not a commit in {args.repo}") return 2 + # A base that is not in the clone would make every gate file look changed, + # and the co-change rule would refuse for a reason that is about the checkout + # rather than about the merge. A question that cannot be asked is not a no. + if args.base is not None and not is_commit(args.repo, args.base): + print(f"merge gate: MISUSE — the base {args.base} is not a commit in {args.repo}; " + "the co-change rule cannot be evaluated, and a shallow checkout must not be " + "reported as a rewritten gate") + return 2 if not applies(args.repo, args.commit): print(f"merge gate: not applicable — {mg.T0_PATH} does not exist at {args.commit[:12]}, " diff --git a/tests/test_step7_mergegate_ci.py b/tests/test_step7_mergegate_ci.py index b85720b2..e3b42f03 100644 --- a/tests/test_step7_mergegate_ci.py +++ b/tests/test_step7_mergegate_ci.py @@ -8,6 +8,7 @@ ci-gate-repair-allowed repairing the gate on its own is ordinary work ci-no-base-is-stated without a base the co-change rule says it did not run ci-misuse-is-two a commit that does not exist is misuse, not a verdict + ci-absent-base-is-two nor may an unavailable base be reported as a rewritten gate ci-gate-files-exist every file the co-change rule watches is really there ci-workflow-names the workflow names the contexts a ruleset must require ci-workflow-derives-t0 the workflow hard-codes no path the tools already own @@ -195,6 +196,24 @@ def control_misuse_is_two() -> None: "the step-7 tools already use") +def control_absent_base_is_two() -> None: + """A shallow checkout must not be able to manufacture a refusal.""" + with tempfile.TemporaryDirectory() as raw: + repo, head, _ = wired(Path(raw), "absentbase") + rc, said = run("--repo", str(repo), "--commit", head, "--base", "e" * 40) + if rc == 1: + fail("ci-absent-base-is-two", + "a base that is not in the clone was reported as a rewritten gate; every gate " + "file looks changed against a base nobody has") + return + if rc != 2: + fail("ci-absent-base-is-two", f"an unavailable base returned {rc}: {said[:160]}") + return + ok("ci-absent-base-is-two", + "an unreachable base is exit 2 and says so; the co-change rule refuses to answer rather " + "than answering with the checkout's depth") + + def control_gate_files_exist() -> None: missing = [p for p in ci.GATE_FILES if not (ROOT / p).exists()] if missing: @@ -258,6 +277,7 @@ def control_inventory() -> None: ("ci-gate-repair-allowed", control_gate_repair_allowed), ("ci-no-base-is-stated", control_no_base_is_stated), ("ci-misuse-is-two", control_misuse_is_two), + ("ci-absent-base-is-two", control_absent_base_is_two), ("ci-gate-files-exist", control_gate_files_exist), ("ci-workflow-names", control_workflow_names), ("ci-workflow-derives-t0", control_workflow_derives_t0), From a49ef6cdd89cc4e77fe98e59a2e30d5ae77ee855 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 09:18:07 +0500 Subject: [PATCH 8/9] fix(tests): the wiring suite exposes run(), so the runner can run it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tests/run_tests.py imports every test_*.py and calls `run()` on it. The new suite called its entry point `main()`, so the runner reported it as a module with no run() and ended the whole test job — after my own controls had printed 15 green lines, which is exactly the shape of a failure that looks like someone else's. Found by reading CI rather than by running the file directly: `python tests/test_step7_mergegate_ci.py` passes either way, and that is precisely why the local run said nothing. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018xhcg5opoFbSTdYHpkSXCh --- tests/test_step7_mergegate_ci.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/test_step7_mergegate_ci.py b/tests/test_step7_mergegate_ci.py index e3b42f03..5359366b 100644 --- a/tests/test_step7_mergegate_ci.py +++ b/tests/test_step7_mergegate_ci.py @@ -285,7 +285,9 @@ def control_inventory() -> None: ] -def main() -> int: +def run() -> int: + # `run`, not `main`: tests/run_tests.py imports every test_*.py and calls + # run() on it, and a module without one is a suite that reports nothing. for name, control in CONTROLS: guarded(name, control) print() @@ -294,4 +296,4 @@ def main() -> int: if __name__ == "__main__": - sys.exit(main()) + sys.exit(run()) From 715bbe7facd28cffbf8d559c565bc3664d6d5bcb Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 09:24:35 +0500 Subject: [PATCH 9/9] fix(tests): rename the helper, not the entry point MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit gave this module a run() for tests/run_tests.py to call — onto a name a local helper already had. The helper won, the entry point was shadowed, and eight controls started invoking themselves recursively instead of the wrapper. Caught immediately by ruff (F811) and by the suite dropping to 4 passed, 8 failed, which is what a repository's own checks are for. The helper is `invoke` now and says in its docstring why it is not `run`. 12 wiring controls, 0 failed; ruff clean. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018xhcg5opoFbSTdYHpkSXCh --- tests/test_step7_mergegate_ci.py | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/tests/test_step7_mergegate_ci.py b/tests/test_step7_mergegate_ci.py index 5359366b..60de8d8b 100644 --- a/tests/test_step7_mergegate_ci.py +++ b/tests/test_step7_mergegate_ci.py @@ -64,9 +64,12 @@ def guarded(check: str, control: Callable[[], None]) -> None: fail(check, f"the control raised {type(exc).__name__}: {exc}") -def run(*args: str) -> tuple[int, str]: +def invoke(*args: str) -> tuple[int, str]: """The wrapper's exit code and everything it said, captured. A green run that - prints REFUSED teaches readers to skim past refusals.""" + prints REFUSED teaches readers to skim past refusals. + + Not named `run`: that name belongs to this module's own entry point, which + tests/run_tests.py calls.""" out = io.StringIO() with contextlib.redirect_stdout(out): rc = ci.main(list(args)) @@ -87,7 +90,7 @@ def wired(tmp: Path, name: str, **kwargs: object) -> tuple[Path, str, str]: def control_not_applicable() -> None: with tempfile.TemporaryDirectory() as raw: repo, _, before = wired(Path(raw), "na") - rc, said = run("--repo", str(repo), "--commit", before) + rc, said = invoke("--repo", str(repo), "--commit", before) if rc != 0: fail("ci-not-applicable", f"a tree with no frozen contract was refused: {said[:160]}") return @@ -102,7 +105,7 @@ def control_not_applicable() -> None: def control_applies_in_full() -> None: with tempfile.TemporaryDirectory() as raw: repo, head, before = wired(Path(raw), "full") - rc, said = run("--repo", str(repo), "--commit", head, "--base", before) + rc, said = invoke("--repo", str(repo), "--commit", head, "--base", before) if rc != 0: fail("ci-applies-in-full", f"a satisfying tree was refused: {said[:200]}") return @@ -119,7 +122,7 @@ def control_applies_in_full() -> None: def control_refusal_survives() -> None: with tempfile.TemporaryDirectory() as raw: repo, head, before = wired(Path(raw), "stale", rebound=False) - rc, said = run("--repo", str(repo), "--commit", head, "--base", before) + rc, said = invoke("--repo", str(repo), "--commit", head, "--base", before) if rc != 1 or "REFUSED" not in said: fail("ci-refusal-survives", f"a stale binding came back as rc={rc}: {said[:200]}") @@ -136,7 +139,7 @@ def control_co_change_refused() -> None: attacked = mgt.commit_tree(repo, { ci.GATE_FILES[0]: "# a gate that says yes\n", ci.GATE_FILES[2]: "name: P-022 merge gate\n"}) - rc, said = run("--repo", str(repo), "--commit", attacked, "--base", before) + rc, said = invoke("--repo", str(repo), "--commit", attacked, "--base", before) if rc != 1: fail("ci-co-change-refused", "a merge carrying both the freeze and a rewritten gate was allowed") @@ -155,7 +158,7 @@ def control_gate_repair_allowed() -> None: repo, head, _ = wired(Path(raw), "repair") # the contract is already in the base and does not move; only the gate does repaired = mgt.commit_tree(repo, {ci.GATE_FILES[0]: "# a repaired gate\n"}) - rc, said = run("--repo", str(repo), "--commit", repaired, "--base", head) + rc, said = invoke("--repo", str(repo), "--commit", repaired, "--base", head) if rc != 0: fail("ci-gate-repair-allowed", f"an ordinary repair of the gate was refused: {said[:200]}") @@ -171,7 +174,7 @@ def control_gate_repair_allowed() -> None: def control_no_base_is_stated() -> None: with tempfile.TemporaryDirectory() as raw: repo, head, _ = wired(Path(raw), "nobase") - rc, said = run("--repo", str(repo), "--commit", head) + rc, said = invoke("--repo", str(repo), "--commit", head) if rc != 0: fail("ci-no-base-is-stated", f"a satisfying tree was refused: {said[:200]}") return @@ -187,7 +190,7 @@ def control_no_base_is_stated() -> None: def control_misuse_is_two() -> None: with tempfile.TemporaryDirectory() as raw: repo, _, _ = wired(Path(raw), "misuse") - rc, said = run("--repo", str(repo), "--commit", "f" * 40) + rc, said = invoke("--repo", str(repo), "--commit", "f" * 40) if rc != 2: fail("ci-misuse-is-two", f"a commit that does not exist returned {rc}: {said[:160]}") return @@ -200,7 +203,7 @@ def control_absent_base_is_two() -> None: """A shallow checkout must not be able to manufacture a refusal.""" with tempfile.TemporaryDirectory() as raw: repo, head, _ = wired(Path(raw), "absentbase") - rc, said = run("--repo", str(repo), "--commit", head, "--base", "e" * 40) + rc, said = invoke("--repo", str(repo), "--commit", head, "--base", "e" * 40) if rc == 1: fail("ci-absent-base-is-two", "a base that is not in the clone was reported as a rewritten gate; every gate "