diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5253bb6..2dcce28 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -57,6 +57,8 @@ jobs: run: cargo clippy -p psyche-store --all-targets --features test-fault-injection -- -D warnings - name: G2 evidence checker unit tests run: python3 scripts/check-g2-evidence-test.py + - name: G2 attestation scope tests + run: python3 scripts/check-g2-attestation-scope-test.py - name: G2 evidence relationships env: GH_TOKEN: ${{ github.token }} diff --git a/docs/G2-ATTESTATION-SCOPE.md b/docs/G2-ATTESTATION-SCOPE.md new file mode 100644 index 0000000..8bc0c1e --- /dev/null +++ b/docs/G2-ATTESTATION-SCOPE.md @@ -0,0 +1,87 @@ +# G2 retained-attestation scope + +**Status:** active trust-root policy +**Schema:** `psyche.g2-attestation-scope/v1` +**Machine-readable policy:** `scripts/g2-attestation-scope.json` +**Scope evidence:** `docs/G2-SCOPE-EVIDENCE.md` +**Historic implementation evidence:** `docs/G2-EVIDENCE.md` + +## Purpose + +The historic G2 evidence proves a specific implementation, test matrix, reviewed workflow, immutable Coven inputs, and CI run. It must remain verifiable without freezing every unrelated repository file forever. + +The retained-attestation scope answers one narrower question: + +> Has any file that can alter the G2 contract, implementation, test, migration, fixture, dependency, toolchain, workflow, verifier, or evidence policy changed since the scope was tested? + +The answer is fail-closed. A path is protected unless the versioned scope manifest explicitly classifies that exact path as outside the historic G2 claim. + +## What remains protected + +The default classification is `protected`. This includes, without relying on a fragile exhaustive prose list: + +- all Rust crates, tests, fixtures, migrations, and schemas; +- the npm wrapper covered by the reviewed workflow; +- Cargo manifests, lockfile, Rust toolchain, clippy and dependency policy; +- CI workflows and every setup/action/command they contain; +- the historic evidence checker core and its mutation tests; +- the scope wrapper, scope tests, manifest, and scope evidence; +- the G2 test manifest; +- architecture, schema, testing, plan, and evidence documents; +- every new or unknown path. + +A changed protected path invalidates the retained scope attestation until a new candidate head passes the complete reviewed CI and an evidence-only commit records that immutable run. + +## Explicitly unattested paths + +The manifest names a small exact set of repository-readiness and governance files that the historic G2 implementation evidence does not claim to prove. Each rule constrains the allowed operation and Git mode. + +The initial set exists only to permit the roadmap/agent-readiness work tracked in #10 and draft PR #15: + +- root onboarding, contribution, security, license, and agent guidance; +- the active roadmap and proposed protocol-ownership document; +- the machine-readable agent repository manifest; +- the bootstrap and fast/full verification wrappers. + +Unknown paths do not inherit safety from a directory or filename pattern. Adding another unattested path changes the protected scope policy and therefore requires a new scope attestation. + +## Two-layer evidence model + +1. `docs/G2-EVIDENCE.md` remains the immutable historic implementation attestation. Its tested source, CI run, matrix, source hashes, and approved plan provenance do not change merely because repository-readiness files evolve. +2. `docs/G2-SCOPE-EVIDENCE.md` attests the current scope checker, scope manifest, reviewed workflow, and complete current G2 test run. + +The scope evidence has two states: + +- `candidate`: the current pull-request head runs the full matrix but makes no retained scope claim; +- `passed`: an evidence-only commit records the successful candidate head and immutable pull-request run. + +The candidate-to-passed transition is intentionally a separate commit. The relationship checker permits that evidence-file modification and no protected source change. + +## Relationship verification + +For the repository's actual historic G2 evidence: + +- a candidate scope validates structure, full tests, the reviewed workflow, historic run provenance, and immutable Coven inputs, but does not claim a retained source relationship; +- a passed scope verifies its own immutable successful CI run and compares the tested source tree with the current terminal tree; +- GitHub Actions uses commit/tree APIs and rejects truncated recursive trees; +- local verification uses exact Git tree entries; +- directory tree hashes are ignored, but every blob and submodule path, mode, type, and object ID is compared; +- the scope evidence may be modified in place; +- explicitly unattested files may only use their declared operations and modes; +- deletions, mode changes, symlinks, submodules, renames, copies, unknown paths, and protected changes fail closed. + +The original evidence-only relationship behavior remains available for the legacy checker mutation suite and for synthetic evidence tests. It is not used to bless later protected-source changes. + +## Updating the protected surface + +When a protected source, test, workflow, dependency, policy, verifier, or evidence document must change: + +1. open a focused R4 issue and identify the affected invariant; +2. change the scope evidence to `candidate` or introduce the candidate trust-root update in the same reviewed slice; +3. run the full CI matrix on the exact candidate head; +4. inspect all positive, denial, crash, restart, migration, supply-chain, secret, npm, and scope tests; +5. add an evidence-only commit naming the candidate source SHA and immutable successful CI run; +6. rerun CI on the evidence commit; +7. merge only after protected-owner review and retained evidence are complete. + +Never add a path to the unattested set merely to make a red check green. diff --git a/docs/G2-SCOPE-EVIDENCE.md b/docs/G2-SCOPE-EVIDENCE.md new file mode 100644 index 0000000..a00df87 --- /dev/null +++ b/docs/G2-SCOPE-EVIDENCE.md @@ -0,0 +1,11 @@ +# G2 Attestation Scope Evidence + +**Status:** candidate +**Tested source commit:** not recorded before remote review +**CI attestation:** not recorded before remote review +**Scope schema:** `psyche.g2-attestation-scope/v1` +**Scope manifest:** `scripts/g2-attestation-scope.json` +**Scope manifest SHA-256:** `sha256:4e535973d4ab49de80e78cd35177bde658d5c31f6d0fd3a5d91edb579a118fa4` +**Historic G2 evidence:** `docs/G2-EVIDENCE.md` + +This document attests the fail-closed path scope used to determine whether the historic G2 implementation evidence still applies to a later repository tree. A candidate proves the checker, full G2 suite, workflow, and scope on its own pull-request head; it does not become the retained scope attestation until an evidence-only commit records that successful immutable run. diff --git a/scripts/check-g2-attestation-scope-test.py b/scripts/check-g2-attestation-scope-test.py new file mode 100644 index 0000000..04bb3a5 --- /dev/null +++ b/scripts/check-g2-attestation-scope-test.py @@ -0,0 +1,241 @@ +#!/usr/bin/env python3 +"""Focused regression tests for the retained G2 attestation scope.""" + +from __future__ import annotations + +import importlib.util +import pathlib +import unittest + + +ROOT = pathlib.Path(__file__).resolve().parents[1] +CHECKER_PATH = ROOT / "scripts/check-g2-evidence.py" + + +def load_checker(): + spec = importlib.util.spec_from_file_location("check_g2_evidence_scope", CHECKER_PATH) + if spec is None or spec.loader is None: + raise RuntimeError(f"cannot load {CHECKER_PATH}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class G2AttestationScopeTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.checker = load_checker() + cls.scope = cls.checker._read_scope_manifest(ROOT) + cls.blob_a = ("100644", "blob", "a" * 40) + cls.blob_b = ("100644", "blob", "b" * 40) + cls.exec_a = ("100755", "blob", "c" * 40) + cls.exec_b = ("100755", "blob", "d" * 40) + + def assert_rejected(self, before, after, pattern="retained G2 attestation"): + with self.assertRaisesRegex(self.checker.EvidenceError, pattern): + self.checker._validate_attested_changes(before, after, self.scope) + + def compare_response(self, *, status="diverged", merge_base=None, files=None): + tested = "1" * 40 + terminal = "2" * 40 + return { + "status": status, + "ahead_by": 1, + "total_commits": 1, + "base_commit": {"sha": tested}, + "merge_base_commit": {"sha": merge_base or ("3" * 40)}, + "commits": [{"sha": terminal}], + "files": files or [ + {"filename": "docs/G2-SCOPE-EVIDENCE.md", "status": "modified"}, + {"filename": "README.md", "status": "added"}, + ], + }, tested, terminal + + def test_repository_scope_manifest_is_exact_and_fail_closed(self): + self.assertEqual(set(self.scope), set(self.checker.EXPECTED_UNATTESTED_PATHS)) + self.assertNotIn("crates/psyche-core/src/contracts/mod.rs", self.scope) + self.assertNotIn(".github/workflows/ci.yml", self.scope) + self.assertNotIn("scripts/check-g2-evidence.py", self.scope) + self.assertNotIn("docs/G2-EVIDENCE.md", self.scope) + + def test_scope_manifest_rejects_duplicate_json_keys(self): + with self.assertRaisesRegex(self.checker.EvidenceError, "repeats a key"): + self.checker._json_object_without_duplicates([("default", "protected"), ("default", "open")]) + + def test_candidate_scope_evidence_is_explicit(self): + markdown = (ROOT / self.checker.SCOPE_EVIDENCE_PATH).read_text() + status, tested, run_url = self.checker._validate_scope_evidence(markdown) + self.assertEqual((status, tested, run_url), ("candidate", None, None)) + + def test_passed_scope_evidence_requires_immutable_source_and_run(self): + markdown = (ROOT / self.checker.SCOPE_EVIDENCE_PATH).read_text() + passed = markdown.replace("**Status:** candidate", "**Status:** passed") + passed = passed.replace( + "**Tested source commit:** not recorded before remote review", + f"**Tested source commit:** {'1' * 40}", + ) + passed = passed.replace( + "**CI attestation:** not recorded before remote review", + "**CI attestation:** https://github.com/OpenCoven/psyche/actions/runs/123456", + ) + status, tested, run_url = self.checker._validate_scope_evidence(passed) + self.assertEqual(status, "passed") + self.assertEqual(tested, "1" * 40) + self.assertEqual(run_url, "https://github.com/OpenCoven/psyche/actions/runs/123456") + + def test_scope_evidence_only_change_is_allowed(self): + path = self.checker.SCOPE_EVIDENCE_PATH + changes = self.checker._validate_attested_changes( + {path: self.blob_a}, + {path: self.blob_b}, + self.scope, + ) + self.assertEqual(changes, [(path, "modified")]) + + def test_explicit_readiness_additions_and_modifications_are_allowed(self): + before = { + "README.md": self.blob_a, + "scripts/agent-check": self.exec_a, + } + after = { + "README.md": self.blob_b, + "AGENTS.md": self.blob_a, + "scripts/agent-check": self.exec_b, + "scripts/agent-bootstrap": self.exec_a, + } + changes = self.checker._validate_attested_changes(before, after, self.scope) + self.assertEqual( + changes, + [ + ("AGENTS.md", "added"), + ("README.md", "modified"), + ("scripts/agent-bootstrap", "added"), + ("scripts/agent-check", "modified"), + ], + ) + + def test_unknown_and_g2_protected_paths_are_rejected(self): + paths = ( + "new-top-level.md", + "crates/psyche-core/src/contracts/mod.rs", + "crates/psyche-store/tests/migrations.rs", + ".github/workflows/ci.yml", + "Cargo.lock", + "rust-toolchain.toml", + "scripts/check-g2-evidence.py", + "scripts/check-g2-evidence-test.py", + "scripts/g2-test-manifest.json", + "scripts/g2-attestation-scope.json", + "docs/G2-EVIDENCE.md", + "docs/ARCHITECTURE.md", + "docs/SCHEMAS.md", + "docs/TESTING.md", + ) + for path in paths: + with self.subTest(path=path): + self.assert_rejected({path: self.blob_a}, {path: self.blob_b}) + + def test_safe_path_deletion_mode_change_symlink_and_submodule_are_rejected(self): + self.assert_rejected({"README.md": self.blob_a}, {}) + self.assert_rejected( + {"README.md": self.blob_a}, + {"README.md": ("100755", "blob", "b" * 40)}, + ) + self.assert_rejected( + {}, + {"README.md": ("120000", "blob", "b" * 40)}, + ) + self.assert_rejected( + {}, + {"README.md": ("160000", "commit", "b" * 40)}, + ) + + def test_rename_is_rejected_as_a_safe_path_deletion(self): + self.assert_rejected( + {"README.md": self.blob_a}, + {"AGENTS.md": self.blob_a}, + ) + + def test_remote_tree_rejects_truncation_submodules_and_duplicate_paths(self): + base = { + "sha": "a" * 40, + "truncated": True, + "tree": [], + } + with self.assertRaisesRegex(self.checker.EvidenceError, "truncated"): + self.checker._entry_map_from_remote_tree(base, "a" * 40) + + submodule = { + "sha": "a" * 40, + "truncated": False, + "tree": [ + {"path": "README.md", "mode": "160000", "type": "commit", "sha": "b" * 40}, + ], + } + entries = self.checker._entry_map_from_remote_tree(submodule, "a" * 40) + self.assert_rejected({}, entries) + + duplicate = { + "sha": "a" * 40, + "truncated": False, + "tree": [ + {"path": "README.md", "mode": "100644", "type": "blob", "sha": "b" * 40}, + {"path": "README.md", "mode": "100644", "type": "blob", "sha": "c" * 40}, + ], + } + with self.assertRaisesRegex(self.checker.EvidenceError, "duplicate"): + self.checker._entry_map_from_remote_tree(duplicate, "a" * 40) + + def test_compare_requires_tree_scope_for_nonancestor_safe_changes(self): + compare, tested, terminal = self.compare_response() + self.assertTrue( + self.checker._compare_requires_tree_scope(compare, tested, terminal, self.scope) + ) + + def test_compare_keeps_original_evidence_only_ancestor_contract(self): + tested = "1" * 40 + terminal = "2" * 40 + compare = { + "status": "ahead", + "ahead_by": 1, + "total_commits": 1, + "base_commit": {"sha": tested}, + "merge_base_commit": {"sha": tested}, + "commits": [{"sha": terminal}], + "files": [{"filename": "docs/G2-EVIDENCE.md", "status": "modified"}], + } + self.assertFalse( + self.checker._compare_requires_tree_scope(compare, tested, terminal, self.scope) + ) + + def test_nonancestor_legacy_evidence_only_compare_fails_closed(self): + compare, tested, terminal = self.compare_response( + files=[{"filename": "docs/G2-EVIDENCE.md", "status": "modified"}] + ) + with self.assertRaisesRegex(self.checker.EvidenceError, "ancestor"): + self.checker._compare_requires_tree_scope(compare, tested, terminal, self.scope) + + def test_compare_rejects_copy_rename_truncation_and_protected_files(self): + mutations = ( + [{"filename": "README.md", "status": "copied"}], + [{"filename": "README.md", "status": "renamed"}], + [{"filename": "src/main.rs", "status": "modified"}], + [{"filename": f"safe-{index}.md", "status": "added"} for index in range(300)], + ) + for files in mutations: + compare, tested, terminal = self.compare_response(files=files) + with self.subTest(status=files[0]["status"], count=len(files)): + with self.assertRaises(self.checker.EvidenceError): + self.checker._compare_requires_tree_scope( + compare, tested, terminal, self.scope + ) + + def test_path_parser_rejects_absolute_parent_backslash_and_directory_paths(self): + for path in ("/README.md", "../README.md", "docs\\README.md", "docs/"): + with self.subTest(path=path): + with self.assertRaises(self.checker.EvidenceError): + self.checker._safe_repository_path(path) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/scripts/check-g2-evidence-v1.py b/scripts/check-g2-evidence-v1.py new file mode 100644 index 0000000..f5d8ce9 --- /dev/null +++ b/scripts/check-g2-evidence-v1.py @@ -0,0 +1,1049 @@ +#!/usr/bin/env python3 +"""Verify that G2 source, tests, CI, and review evidence remain connected.""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import os +import pathlib +import re +import subprocess +import sys +import urllib.parse +from collections.abc import Mapping + + +class EvidenceError(RuntimeError): + """A G2 evidence relationship is absent or inconsistent.""" + + +FIXED_SPEC_COMMIT = "42dcbc4334cb48ecaf63efb550345e3eea2fb7ad" +APPROVED_PLAN_COMMIT = "5f22ebef1e23d045a10f2ec0a3c87be029446cf6" +APPROVED_PLAN_SHA256 = "4fba002ad9f969cd01866ea08f270654f82b53c7d90b73d28643a9abb12cba68" +PLAN_PATH = "docs/superpowers/plans/2026-08-05-psyche-w2-g2-foundation.md" +SOURCE_ROWS = { + "PLAN": ( + "https://github.com/OpenCoven/coven/blob/42dcbc43%334cb48ec%61f63efb5%350345e3e%65a2fb7ad/specs/psyche/PLAN.md", + "01382f8a0d2bca95ddd535634dd6a9f09ac4a80d588ccbebd72f163eaf56bc1e", + ), + "RUNTIME_DESIGN": ( + "https://github.com/OpenCoven/coven/blob/42dcbc43%334cb48ec%61f63efb5%350345e3e%65a2fb7ad/specs/psyche/RUNTIME_DESIGN.md", + "ab8c922214b8f1179ebf71fb8dfb55bd6d0ff2d6dfced4551bf90503767bb6b8", + ), + "TECH": ( + "https://github.com/OpenCoven/coven/blob/42dcbc43%334cb48ec%61f63efb5%350345e3e%65a2fb7ad/specs/psyche/TECH.md", + "1d00fb2b725f384ca027db60d0afbd0a62a7ec6c7dcbb5637bf14d30d40e2e1c", + ), + "COVEN_PREREQUISITES": ( + "https://github.com/OpenCoven/coven/blob/42dcbc43%334cb48ec%61f63efb5%350345e3e%65a2fb7ad/specs/psyche/COVEN_PREREQUISITES.md", + "33994a28921e70f824b0260ce08231b2117c50430c54e996ed47582d060e72f9", + ), + "COVEN_W1_AUDIT": ( + "https://github.com/OpenCoven/coven/blob/42dcbc43%334cb48ec%61f63efb5%350345e3e%65a2fb7ad/specs/psyche/COVEN_W1_AUDIT.md", + "eab9028bf7ef9c8a96d4c6bed69e4ef0b3497b470ca26589cb3ffcd80677322d", + ), +} + +# The commands are plan-owned literals. Hashing keeps this checker readable +# while still rejecting any byte of drift in a command cell. +MATRIX_COMMAND_SHA256 = { + "Canonical ID prefixes and execution-binding identity": "99555c557ce768a2691ea7a7e9e238d73596f0d6d302152fd128e7e2c1f55356", + "Complete canonical error enum": "8f65d986781fb10422a0261e17138201e7b3379ef8c0a09defab7be815dd4bee", + "Canonical delivery v1 shape": "832a79d4126a7cb77b8e0b0341674ca7348ef356091dab02257ccd390a69a384", + "Surface and quarantine owned types": "c5915caab82a15c37e3af5ddc41ec948043daf158fe78c4decc8ed677a17f1f9", + "Package-local nullable-binding fixtures": "2ac01fe93e9fb2580b65464cefaf2a1e942c2240048130d24597688e7d7b8d7a", + "Exhaustive registered decode": "7f744575d04e26cdc6a798d0e315371fcf9bf1b598c4aca9e8052da9d34cd6aa", + "Unknown kind/version/enum denial and quarantine": "ec7296a6b7fb26e7df19fb2c7147bb6225f45c279937b02046110773e468427e", + "Quarantine resolution": "0529a41bff3a7533173055d8d015d1d8a46ae922b0772c8f19fd0bde752bb5d3", + "Direct typed insert validation": "77f4e53c38d571775bf74280f50ea56ce8f802296d1e075425417e269210f5b4", + "Append-only execution-binding revisions": "a3823e0c95245f67996706bc37171569e9369e72166a4b42643da4c90deef71b", + "Transition contract and append-only rules": "65a4b8ac3d7e29f39d2d896ad222477576830b7a2829b41b5c147f0118dde3be", + "Checkpoint-failure shutdown": "979c2261ae0b36b6a1ab5bd34c70a5ec253c5f3120f87876fe33835d18207a69", + "Migrations": "f6d8857ebfc786480c20c2d693ed6cd4398bb5ad5979bb6454d15564d3532580", + "State-machine/property": "ec8c6e799d722b7a382d019853a8405949b464f44f68fb9c6fdb097a789b73bd", + "Crash/restart": "88e020450df496ffb6d915efacb4fc640a2bc325f023ac6332870ec847b71d16", + "Fake boundaries and durable termination ordering": "71355a2738db952c35c0fc18734eb80287602dbf0c10e2aabc96293c05ed9b15", + "Execution request RFC3339 golden bytes": "5315ed856e8dd813132a338a8024779707d8251efd0bdf71b92c2eae57c3a6e1", + "Validated termination dispatch": "68d7ac817187f977be07ffa7668a340222eba600a116380e46ba2b2cd43a67c1", + "G2 cancellation-state vocabulary": "b1822ccbdbd9cb0488227c16b94fd9ad75809c479bbfb01c3b9b8dba08b87caa", + "Full execution-request digest binding": "1ae9e298ddfd97efbd0b673655fbc90fbd78ef650f75fb20a4aaa803d8e166ff", + "C-S1 scripted contract negotiation": "0a0749461006ab5af3c1efd18cbff9b03b05e55045894153ce081160de337a83", + "C-S2 scripted session lifecycle": "19ee8df21953e783e4fa7ac82cee53728deb944ee333a19646b0b4b52cda80b6", + "C-S3 scripted snapshot/attempt binding": "02b6ecde06ab9bb935b81f7d1c003058ed7519ae272b139f1e82e10664510c3c", + "C-S4 scripted stable adoption": "7fa24fdf7b9465a457d33aa1c61c00b6372de8509b4142dbe109b9e6bdc9eb5c", + "C-S5 scripted non-adoption proof": "d25d4488d7d4d8832dca0ac72eec5fbaf09fbb870ff9ff035aeec21b6c039b55", + "C-S6 scripted ambiguity reconciliation/fence": "228e9790d023509f64a3871dfd58c2fd9b9afb4c1c18efa8e827d4a07f93dd84", + "C-S7 scripted ordered cursor": "220c990305d5ab9f080aafbab8b5224da1fc67e29f3a5f5a3d5cf141365675a8", + "C-S8 scripted terminal authority": "5e3223be4f8d8280792509a78661ae892691c6a0513e71fe35ea6752bbb86dc9", + "C-S9 scripted O5 cancellation acknowledgement": "fe4d01e7088f7376df97614522b4eebe3ee3cb7493d216c748d30bc2bf99b26a", + "C-S10 scripted result/artifact binding": "6bcfe3d642acc1a0c258cbaa3ddefd2257bba8a8d8b6e2251d93ce8d07366af3", + "C-S11 scripted restart persistence": "24866d81ddab9d0131cef455721cd2aee7f06fe61dd0320420bdf2617c128860", + "C-S12 scripted structured denial": "3e38ff6947e6e91504727847c5adae7cd3471f998d90c3980b67f520ce955a4a", +} + +CI_COMMANDS = ( + "cargo fmt --all -- --check", + "cargo clippy --workspace --all-targets -- -D warnings", + "cargo test --workspace --locked", + "cargo test -p psyche-test-support --test state_machine", + "cargo test -p psyche-test-support --test conformance", + "cargo test -p psyche-store --test migrations", + "cargo test -p psyche-store --features test-fault-injection --test crash", + "cargo clippy -p psyche-store --all-targets --features test-fault-injection -- -D warnings", + "cargo deny check licenses advisories bans sources", + 'gitleaks detect --no-banner --redact --log-opts="--all"', + "python3 scripts/check-g2-evidence-test.py", + "python3 scripts/check-g2-evidence.py", +) +NON_RUST_CI_COMMAND_JOBS = { + "cargo deny check licenses advisories bans sources": "supply-chain", + 'gitleaks detect --no-banner --redact --log-opts="--all"': "secrets", +} +# Pins the complete reviewed workflow, including setup/actions and every writer +# that could poison GITHUB_ENV or GITHUB_PATH. Newlines are normalized first so +# the same reviewed content verifies on Windows checkouts. +REVIEWED_WORKFLOW_SHA256 = "1f908303c1a8940ce5ec8c81182ddaf5d82e5c6bddd5ece7fb33e1baf9a087f1" +CI_WORKFLOW_ID = 326408880 + + +def fail(message: str) -> None: + raise EvidenceError(message) + + +def read_text(root: pathlib.Path, path: str, overrides: Mapping[str, str]) -> str: + if path in overrides: + return overrides[path] + candidate = root / path + if not candidate.is_file(): + fail(f"required file is absent: {path}") + return candidate.read_text(encoding="utf-8") + + +def parse_tables(markdown: str) -> tuple[list[list[str]], list[list[str]]]: + source: list[list[str]] = [] + matrix: list[list[str]] = [] + active: list[list[str]] | None = None + for line in markdown.splitlines(): + if line == "| Coven source | Immutable URL | SHA-256 |": + active = source + continue + if line == "| Criterion | Command | Result | Artifact |": + active = matrix + continue + if active is not None and line.startswith("|---"): + continue + if active is not None and line.startswith("|"): + cells = [cell.strip() for cell in line.strip().strip("|").split("|")] + if not all(cells): + fail("evidence table contains an empty cell") + active.append(cells) + elif active is not None: + active = None + if any(len(row) != 3 for row in source) or any(len(row) != 4 for row in matrix): + fail("evidence table has an invalid column count") + return source, matrix + + +def field(markdown: str, label: str) -> str: + matches = re.findall(rf"^\*\*{re.escape(label)}:\*\* (.+)$", markdown, re.MULTILINE) + if not matches: + fail(f"missing evidence field: {label}") + if len(matches) != 1: + fail(f"evidence field must occur exactly once: {label}") + return matches[0].strip().strip("`") + + +def yaml_scalar(value: str) -> str: + value = value.strip() + if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: + return value[1:-1] + return value + + +def parse_workflow_steps(workflow: str) -> list[dict[str, object]]: + """Extract active named/uses steps without treating YAML comments as data.""" + lines = workflow.splitlines() + try: + jobs_at = lines.index("jobs:") + except ValueError: + fail("workflow jobs mapping is absent") + job_starts = [ + (index, match.group("job")) + for index, line in enumerate(lines[jobs_at + 1:], start=jobs_at + 1) + if (match := re.fullmatch(r" (?P[A-Za-z0-9_-]+):\s*", line)) + ] + starts = [ + (index, len(match.group("indent")), match.group("key")) + for index, line in enumerate(lines) + if (match := re.match(r"^(?P\s*)-\s+(?Pname|uses):", line)) + ] + steps: list[dict[str, object]] = [] + for position, (start, indent, header_key) in enumerate(starts): + owners = [(index, job) for index, job in job_starts if index < start] + if not owners: + fail("workflow step is not contained by a job") + owner_at, owner = owners[-1] + end = next((index for index, _ in job_starts if index > owner_at), len(lines)) + for candidate, candidate_indent, _ in starts[position + 1:]: + if candidate < end and candidate_indent == indent: + end = candidate + break + child = " " * (indent + 2) + grandchild = " " * (indent + 4) + run_values: list[str] = [] + env: dict[str, str] = {} + env_counts: dict[str, int] = {} + direct_counts = {header_key: 1} + invalid = False + in_env = False + for line in lines[start + 1:end]: + if not line.strip() or line.lstrip().startswith("#"): + continue + line_indent = len(line) - len(line.lstrip()) + if in_env and line_indent == indent + 4: + match = re.fullmatch(r"(?P[A-Za-z_][A-Za-z0-9_-]*):(?P.*)", line[len(grandchild):]) + if not match: + invalid = True + continue + key = match.group("key") + env_counts[key] = env_counts.get(key, 0) + 1 + env[key] = yaml_scalar(match.group("value")) + continue + if line_indent == indent + 2: + in_env = False + match = re.fullmatch(r"(?P[A-Za-z_][A-Za-z0-9_-]*):(?P.*)", line[len(child):]) + if not match: + invalid = True + continue + key = match.group("key") + value = match.group("value") + direct_counts[key] = direct_counts.get(key, 0) + 1 + if key == "env": + if value.strip(): + invalid = True + else: + in_env = True + elif key == "run": + run_values.append(yaml_scalar(value)) + continue + if line_indent <= indent + 2: + in_env = False + if len(run_values) > 1: + fail("workflow step contains more than one active run value") + steps.append({ + "run": run_values[0] if run_values else None, + "env": env, + "env_counts": env_counts, + "direct_counts": direct_counts, + "invalid": invalid, + "job": owner, + }) + return steps + + +def workflow_job_lines(workflow: str, job: str) -> list[str]: + lines = workflow.splitlines() + job_pattern = re.compile(rf"^ {re.escape(job)}:\s*$") + starts = [index for index, line in enumerate(lines) if job_pattern.fullmatch(line)] + if len(starts) != 1: + fail(f"workflow must contain exactly one {job} job") + start = starts[0] + end = len(lines) + for index in range(start + 1, len(lines)): + if re.match(r"^ [A-Za-z0-9_-]+:\s*$", lines[index]): + end = index + break + return lines[start + 1:end] + + +def mapping_values(lines: list[str], indent: int) -> dict[str, list[str]]: + values: dict[str, list[str]] = {} + for line in lines: + if not line.strip() or line.lstrip().startswith("#"): + continue + if len(line) - len(line.lstrip()) == indent: + match = re.fullmatch(r"(?P[A-Za-z_][A-Za-z0-9_-]*):(?P.*)", line[indent:]) + if not match: + fail(f"workflow contains a noncanonical key at indentation {indent}: {line.strip()}") + values.setdefault(match.group("key"), []).append(yaml_scalar(match.group("value"))) + return values + + +def nested_mapping_lines(lines: list[str], indent: int, key: str) -> list[str]: + header = " " * indent + key + ":" + starts = [index for index, line in enumerate(lines) if line == header] + if len(starts) != 1: + fail(f"workflow mapping must contain exactly one active {key}") + start = starts[0] + end = len(lines) + for index in range(start + 1, len(lines)): + line = lines[index] + if line.strip() and not line.lstrip().startswith("#") and len(line) - len(line.lstrip()) <= indent: + end = index + break + return lines[start + 1:end] + + +def validate_workflow_scope(workflow: str) -> None: + lines = workflow.splitlines() + for line in lines: + if not line.strip() or line.lstrip().startswith("#"): + continue + if re.search(r"(?:^|[\s:\[\{,])(?:&|\*)[A-Za-z_][A-Za-z0-9_-]*", line) or re.match(r"^\s*<<\s*:", line): + fail("workflow anchors, aliases, and merge keys are not allowed") + root = mapping_values(lines, 0) + expected_root = {"name", "on", "concurrency", "env", "jobs"} + if set(root) != expected_root or any(len(values) != 1 for values in root.values()): + fail("workflow root must use the exact canonical CI structure") + if root["on"] != [""]: + fail("workflow triggers must use the canonical block mapping") + triggers = nested_mapping_lines(lines, 0, "on") + trigger_values = mapping_values(triggers, 2) + if trigger_values != {"push": [""], "pull_request": [""]}: + fail("workflow must run only for main pushes and pull requests") + push = mapping_values(nested_mapping_lines(triggers, 2, "push"), 4) + if push != {"branches": ["[main]"]}: + fail("workflow push trigger must target exactly main") + if mapping_values(nested_mapping_lines(triggers, 2, "pull_request"), 4): + fail("workflow pull_request trigger must be unqualified") + global_env = mapping_values(nested_mapping_lines(lines, 0, "env"), 2) + if global_env != {"CARGO_TERM_COLOR": ["always"], "RUSTFLAGS": ["-D warnings"]}: + fail("workflow global env must contain only fixed non-overriding values") + + +def validate_required_job_shapes(workflow: str) -> None: + expected = { + "rust": {"name", "runs-on", "strategy", "steps"}, + "supply-chain": {"name", "runs-on", "steps"}, + "secrets": {"name", "runs-on", "steps"}, + } + for job, keys in expected.items(): + values = mapping_values(workflow_job_lines(workflow, job), 4) + if set(values) != keys or any(len(entries) != 1 for entries in values.values()): + fail(f"CI required-command job {job} must use its exact canonical direct keys") + + +def validate_rust_matrix(workflow: str) -> None: + rust = workflow_job_lines(workflow, "rust") + direct = mapping_values(rust, 4) + if direct.get("runs-on") != ["${{ matrix.os }}"]: + fail("CI rust job must run on the active matrix.os value") + strategy = nested_mapping_lines(rust, 4, "strategy") + strategy_values = mapping_values(strategy, 6) + if strategy_values.get("fail-fast") != ["false"]: + fail("CI rust strategy must actively set fail-fast to false") + matrix = nested_mapping_lines(strategy, 6, "matrix") + matrix_values = mapping_values(matrix, 8) + if set(matrix_values) != {"os"} or len(matrix_values["os"]) != 1: + fail("CI rust matrix must contain only the supported os axis") + os_value = matrix_values["os"][0] + if not (os_value.startswith("[") and os_value.endswith("]")): + fail("CI rust matrix os axis must be an inline list") + systems = [yaml_scalar(item) for item in os_value[1:-1].split(",") if item.strip()] + expected = {"ubuntu-latest", "macos-latest", "windows-latest"} + if len(systems) != 3 or set(systems) != expected: + fail("CI rust matrix must cover exactly ubuntu, macOS, and Windows") + + +def validate_ci_structure(workflow: str) -> None: + validate_workflow_scope(workflow) + validate_required_job_shapes(workflow) + steps = parse_workflow_steps(workflow) + env_requirements = { + "cargo test -p psyche-test-support --test state_machine": { + "PROPTEST_CASES": "2048", + "PROPTEST_RNG_SEED": "0" * 32, + }, + "python3 scripts/check-g2-evidence.py": {"GH_TOKEN": "${{ github.token }}"}, + } + for command in CI_COMMANDS: + matching = [step for step in steps if step["run"] == command] + if len(matching) != 1: + fail(f"CI workflow must run exact G2 command once: {command}") + expected_job = NON_RUST_CI_COMMAND_JOBS.get(command, "rust") + if matching[0]["job"] != expected_job: + fail(f"CI workflow runs required command outside {expected_job}: {command}") + step = matching[0] + required_env = env_requirements.get(command, {}) + required_direct = {"name": 1, "run": 1} + if required_env: + required_direct["env"] = 1 + if step["invalid"] or step["direct_counts"] != required_direct: + fail(f"CI required command step has noncanonical direct keys: {command}") + if step["env"] != required_env or step["env_counts"] != {key: 1 for key in required_env}: + fail(f"CI required command step has noncanonical env: {command}") + validate_rust_matrix(workflow) + + +def validate_ci_workflow(workflow: str) -> None: + normalized = workflow.replace("\r\n", "\n").replace("\r", "\n") + if hashlib.sha256(normalized.encode("utf-8")).hexdigest() != REVIEWED_WORKFLOW_SHA256: + fail("CI workflow differs from the complete reviewed workflow") + validate_ci_structure(normalized) + + +def normalize_grouped(value: str, prefix: str = "") -> str: + value = value.removeprefix(prefix).replace("-", "") + if not re.fullmatch(r"[0-9a-f]+", value): + fail(f"invalid hexadecimal value: {value}") + return value + + +def parse_blob_url(url: str) -> tuple[str, str]: + parsed = urllib.parse.urlparse(url) + if parsed.scheme != "https" or parsed.netloc != "github.com": + fail(f"Coven source URL is not immutable HTTPS: {url}") + match = re.fullmatch(r"/OpenCoven/coven/blob/([^/]+)/(.+)", parsed.path) + if not match: + fail(f"Coven source URL is not an OpenCoven/coven blob URL: {url}") + commit = urllib.parse.unquote(match.group(1)).replace("-", "") + path = urllib.parse.unquote(match.group(2)) + if not re.fullmatch(r"[0-9a-f]{40}", commit) or path.startswith("/") or ".." in pathlib.PurePosixPath(path).parts: + fail(f"Coven source URL does not name a 40-hex commit and safe path: {url}") + return commit, path + + +def validate_evidence(markdown: str) -> tuple[str, list[list[str]], list[list[str]]]: + status = field(markdown, "Status") + if status not in {"candidate", "passed"}: + fail("evidence status must be candidate or passed") + source, matrix = parse_tables(markdown) + source_names = [row[0] for row in source] + if len(source_names) != len(set(source_names)) or set(source_names) != set(SOURCE_ROWS): + fail("Coven source table must contain each fixed source exactly once") + for name, url_cell, digest_cell in source: + url = url_cell.strip("`") + digest = normalize_grouped(digest_cell.strip("`"), "sha256:") + if (url, digest) != SOURCE_ROWS[name]: + fail(f"immutable Coven source row drifted: {name}") + commit, _ = parse_blob_url(url) + if commit != FIXED_SPEC_COMMIT: + fail(f"Coven specification source commit drifted: {name}") + if normalize_grouped(field(markdown, "Coven specification source commit")) != FIXED_SPEC_COMMIT: + fail("Coven specification source field drifted") + + criteria = [row[0] for row in matrix] + if len(criteria) != len(set(criteria)) or set(criteria) != set(MATRIX_COMMAND_SHA256): + fail("evidence matrix must contain every required criterion exactly once") + for criterion, command_cell, result, artifact in matrix: + if not (command_cell.startswith("`") and command_cell.endswith("`")): + fail(f"matrix command is not a code literal: {criterion}") + command = command_cell[1:-1] + if hashlib.sha256(command.encode()).hexdigest() != MATRIX_COMMAND_SHA256[criterion]: + fail(f"matrix command differs from the plan allowlist: {criterion}") + for atomic in command.split(" && "): + if not re.search(r" -- --exact [A-Za-z0-9_:]+$", atomic): + fail(f"matrix command is not an exact filtered test: {atomic}") + + if status == "candidate": + expected = { + "Tested source commit": "not recorded before remote review", + "CI attestation": "not recorded before remote review", + "Coven plan source commit": "not recorded before plan approval", + "Coven plan URL": "not recorded before plan approval", + "Coven plan SHA-256": "not recorded before plan approval", + } + for label, value in expected.items(): + if field(markdown, label) != value: + fail(f"candidate field must use its exact placeholder: {label}") + if any(row[2:] != ["not run remotely", "none"] for row in matrix): + fail("candidate matrix must use the exact result and artifact placeholders") + else: + tested = field(markdown, "Tested source commit") + run_url = field(markdown, "CI attestation") + plan_commit = normalize_grouped(field(markdown, "Coven plan source commit")) + plan_url = field(markdown, "Coven plan URL") + plan_digest = normalize_grouped(field(markdown, "Coven plan SHA-256"), "sha256:") + if not re.fullmatch(r"[0-9a-f]{40}", tested): + fail("passed evidence requires a 40-hex tested source") + if plan_commit != APPROVED_PLAN_COMMIT or plan_digest != APPROVED_PLAN_SHA256: + fail("passed evidence does not name the approved Coven plan provenance") + url_commit, url_path = parse_blob_url(plan_url) + if url_commit != plan_commit or url_path != PLAN_PATH: + fail("passed evidence plan URL does not match the approved plan") + if not re.fullmatch(r"https://github\.com/OpenCoven/psyche/actions/runs/[0-9]+", run_url): + fail("passed evidence requires an immutable Actions run URL") + if any(row[2] != "passed" or row[3] != run_url for row in matrix): + fail("every passed matrix result and artifact must attest the same run") + placeholders = re.compile(r"not run remotely|\bnone\b|not recorded|pending|placeholder|TBD|TODO", re.IGNORECASE) + if placeholders.search(markdown): + fail("passed evidence contains a candidate placeholder") + return status, source, matrix + + +def manifest_atomic_commands(manifest: Mapping[str, object]) -> tuple[dict[str, tuple[str, str]], dict[str, str]]: + targets = manifest.get("targets") + if not isinstance(targets, dict) or not targets: + fail("manifest.targets must be a non-empty object") + atomic: dict[str, tuple[str, str]] = {} + lists: dict[str, str] = {} + for target, raw in targets.items(): + if not isinstance(target, str) or not isinstance(raw, dict): + fail("manifest target entries must be objects") + list_command = raw.get("list_command") + tests = raw.get("tests") + if not isinstance(list_command, str) or not list_command.endswith(" -- --list --format terse"): + fail(f"invalid list command for {target}") + if not isinstance(tests, list) or not tests or any(not isinstance(name, str) or not name for name in tests): + fail(f"manifest target has no exact tests: {target}") + if len(tests) != len(set(tests)): + fail(f"manifest target repeats a test: {target}") + prefix = list_command.removesuffix(" -- --list --format terse") + lists[target] = list_command + for name in tests: + command = f"{prefix} -- --exact {name}" + if command in atomic: + fail(f"manifest maps an atomic command more than once: {command}") + atomic[command] = (target, name) + return atomic, lists + + +def list_tests(root: pathlib.Path, commands: Mapping[str, str]) -> dict[str, str]: + outputs: dict[str, str] = {} + for target, command in commands.items(): + completed = subprocess.run(command.split(), cwd=root, text=True, capture_output=True, check=False) + if completed.returncode: + fail(f"test listing failed for {target}:\n{completed.stdout}{completed.stderr}") + outputs[target] = completed.stdout + return outputs + + +def validate_manifest( + root: pathlib.Path, + manifest: Mapping[str, object], + matrix: list[list[str]], + listed_tests: Mapping[str, str] | None, +) -> None: + atomic_manifest, list_commands = manifest_atomic_commands(manifest) + matrix_commands: list[str] = [] + for row in matrix: + matrix_commands.extend(row[1][1:-1].split(" && ")) + if len(matrix_commands) != len(set(matrix_commands)): + fail("an atomic matrix command is duplicated") + if set(matrix_commands) != set(atomic_manifest): + missing = sorted(set(matrix_commands) - set(atomic_manifest)) + unused = sorted(set(atomic_manifest) - set(matrix_commands)) + fail(f"matrix/manifest mismatch; missing={missing}, unused={unused}") + outputs = dict(listed_tests) if listed_tests is not None else list_tests(root, list_commands) + if set(outputs) != set(list_commands): + fail("test listing output does not cover exactly the manifest targets") + for target, output in outputs.items(): + names = { + line.rsplit(": test", 1)[0] + for line in output.splitlines() + if line.endswith(": test") + } + if not names: + fail(f"test target lists zero tests: {target}") + for name in manifest["targets"][target]["tests"]: # type: ignore[index] + if name not in names: + fail(f"exact manifest test is absent from {target}: {name}") + + +def require_terms(path: str, text: str, terms: tuple[str, ...]) -> None: + missing = [term for term in terms if term not in text] + if missing: + fail(f"{path} is missing required relationships: {missing}") + + +def require_digest_mutation_matrix(path: str, text: str) -> None: + common = ( + "schema_version", "request_id", "graph_id", "node_id", "attempt_id", + "principal_id", "familiar_snapshot_id", "project_id", + "context_manifest_digest", "required_artifact_bindings", "payload_digest", + "created_at", "valid_until", + ) + launch = common + ("project_root", "cwd", "harness", "delegation_digest", "budget_digest") + input_request = common + ("session_id", "input_digest") + artifact = ("artifact_id", "digest", "media_type", "size") + for name in launch + input_request: + if f'"/input/{name}"' not in text: + fail(f"{path} does not stale-digest mutate request field: {name}") + for name in artifact: + if f'"/input/required_artifact_bindings/0/{name}"' not in text: + fail(f"{path} does not stale-digest mutate artifact field: {name}") + if 'mutations.push(("/input", other_input))' not in text: + fail(f"{path} does not stale-digest mutate the request variant") + + +def validate_record_kinds(source: str) -> None: + enum = re.search(r"pub enum RecordKind \{(.*?)\n\}", source, re.DOTALL) + if not enum: + fail("RecordKind declaration is absent") + variants = re.findall(r"^\s{4}([A-Z][A-Za-z0-9]+),$", enum.group(1), re.MULTILINE) + if len(variants) != 15 or "Attempt" not in variants or "ExecutionBinding" in variants: + fail(f"RecordKind must have exactly 15 variants with Attempt only: {variants}") + all_block = re.search(r"pub const ALL: \[RecordKind; 15\] = \[(.*?)\];", source, re.DOTALL) + prefixes = re.search(r"pub const fn prefix\(self\).*?match self \{(.*?)\n\s*\}", source, re.DOTALL) + if not all_block or re.findall(r"RecordKind::([A-Za-z0-9]+)", all_block.group(1)) != variants: + fail("RecordKind::ALL is not exhaustive and declaration-ordered") + if not prefixes: + fail("RecordKind prefix match is absent") + pairs = re.findall(r'RecordKind::([A-Za-z0-9]+) => "([a-z]{3}_)"', prefixes.group(1)) + if [name for name, _ in pairs] != variants: + fail("RecordKind prefix match does not use the exact variant set") + if dict(pairs).get("Attempt") != "att_" or sum(prefix == "att_" for _, prefix in pairs) != 1: + fail("Attempt must be the only att_ record kind") + if source.count("SchemaKind::ExecutionBinding => Some(RecordKind::Attempt)") != 1: + fail("ExecutionBinding must map exactly once to RecordKind::Attempt") + + +def validate_result_fixture(text: str) -> None: + try: + bundle = json.loads(text) + except json.JSONDecodeError as error: + fail(f"result-bundle fixture is invalid JSON: {error}") + if set(bundle) != {"artifacts", "correlation", "result", "session_id"}: + fail("result-bundle fixture is not strict and complete") + if not isinstance(bundle["artifacts"], list) or not bundle["artifacts"]: + fail("result-bundle fixture has no artifact") + references = [bundle["result"]] + [artifact.get("content", {}) for artifact in bundle["artifacts"]] + for reference in references: + if set(reference) != {"digest", "expires_at", "media_type", "size_bytes"}: + fail("every result/artifact content reference needs digest, media_type, size_bytes, expires_at") + if not re.fullmatch(r"sha256:[0-9a-f]{64}", str(reference["digest"])): + fail("content reference digest is not canonical") + if not isinstance(reference["size_bytes"], int) or reference["size_bytes"] <= 0: + fail("content reference size is invalid") + if not re.fullmatch(r"[^/\s]+/[^/\s]+", str(reference["media_type"])): + fail("content reference media type is invalid") + if not re.fullmatch(r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ", str(reference["expires_at"])): + fail("content reference expiry is not canonical RFC3339 UTC") + correlation = bundle["correlation"] + for artifact in bundle["artifacts"]: + if artifact.get("correlation") != correlation or artifact.get("session_id") != bundle["session_id"]: + fail("artifact lifetime/correlation does not match its result bundle") + + +def validate_golden(path: str, raw: bytes, expected_sha: str) -> None: + if raw.endswith(b"\n"): + fail(f"golden fixture has a trailing newline: {path}") + try: + value = json.loads(raw) + except json.JSONDecodeError as error: + fail(f"golden fixture is invalid JSON: {path}: {error}") + for key in ("created_at", "valid_until"): + if not isinstance(value.get(key), str) or not re.fullmatch(r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ", value[key]): + fail(f"golden fixture lacks an RFC3339 string {key}: {path}") + canonical = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + if canonical != raw: + fail(f"golden fixture is not canonical JSON: {path}") + if hashlib.sha256(raw).hexdigest() != expected_sha: + fail(f"golden fixture SHA-256 drifted: {path}") + + +def validate_sources(root: pathlib.Path, overrides: Mapping[str, str]) -> None: + port_path = "crates/psyche-coven/src/port.rs" + port = read_text(root, port_path, overrides) + require_terms(port_path, port, ( + "pub fn new(input: ExecutionRequestInput)", "let request_digest = digest(&input)?;", + "pub fn recompute_digest", "pub trait CovenPort", "async fn reconcile(", + "pub struct ResultBundle", "CancellationAcknowledgementEvidence", + )) + if re.search(r"(?:struct|enum)\s+\w*Acknowledgement\w*", port): + fail("psyche-coven owns a duplicate acknowledgement wire type") + + suite_path = "crates/psyche-test-support/src/suites/coven.rs" + suite = read_text(root, suite_path, overrides) + for number in range(1, 13): + if suite.count(f"pub async fn assert_c_s{number}_") != 1: + fail(f"reusable C-S{number} function is absent or duplicated") + require_terms(suite_path, suite, ( + "stale_digest_mutations", "request.recompute_digest()", "RequestDigestMismatch", + "RAW_LEDGER_STATES", '"killed"', '"orphaned"', "CancellationAcknowledgementEvidence", + "assert_c_s6_ambiguity_fence", "ReconciliationDisposition::Returned", + "ReconciliationDisposition::Fenced", "DurableDispositionKind::Returned", + "DurableDispositionKind::Fenced", "fixture.restart().await", "require_fault(", + "redispatch_eligibility", "EligibleAfterFence", "assert_c_s10_result_artifact_binding", + "mutate_result_digest", "mutate_result_media_type", "mutate_result_size", + "mutate_result_expiry", "mutate_artifact_digest", "mutate_artifact_media_type", + "mutate_artifact_size", "mutate_artifact_expiry", + )) + require_digest_mutation_matrix(suite_path, suite) + fake_path = "crates/psyche-test-support/src/coven.rs" + fake = read_text(root, fake_path, overrides) + require_terms(fake_path, fake, ("async fn adopt", "request.validate_digest()?;")) + if suite.count("request.validate_digest()?;") != 1: + fail("scripted Coven boundary must recompute the request digest exactly once") + conformance_path = "crates/psyche-test-support/tests/conformance.rs" + conformance = read_text(root, conformance_path, overrides) + for number in range(1, 13): + if len(re.findall(rf"async fn c_s{number}_[a-z0-9_]+\(\)", conformance)) != 1: + fail(f"exact C-S{number} wrapper is absent or duplicated") + + state_path = "crates/psyche-test-support/tests/state_machine.rs" + state = read_text(root, state_path, overrides) + require_terms(state_path, state, ( + "c_s6_model_never_redispatches_without_fence", "request_digest_binds_every_typed_field", + "stale_digest_requests", "MutateRequestFieldRetainDigest", "RequestDigestMismatch", + "ReconciliationDisposition::Returned", "ReconciliationDisposition::Fenced", + "fixture.restart().await", "select_fault(", "RedispatchEligibility::EligibleAfterFence", + )) + require_digest_mutation_matrix(state_path, state) + + core_path = "crates/psyche-core/src/contracts/mod.rs" + core = read_text(root, core_path, overrides) + validate_record_kinds(core) + require_terms(core_path, core, ( + "const ALL: [SchemaKind; 16]", "CanonicalDocument::ExecutionBinding", + "SchemaKind::Error => None", "UnknownSchema", "UnsupportedMajor", "UnknownEnumValue", + )) + error_path = "crates/psyche-core/src/contracts/error.rs" + require_terms(error_path, read_text(root, error_path, overrides), ("pub const ALL: [Self; 36]",)) + contracts_path = "crates/psyche-core/tests/contracts.rs" + require_terms(contracts_path, read_text(root, contracts_path, overrides), ( + "all_canonical_error_codes_decode", "delivery_v1_fixture_round_trips_canonically", + "surface_event_and_effect_fixtures_round_trip", "cancellation_state_vocabulary_requires_matching_o5_evidence", + "graph_and_node_accept_only_the_two_frozen_nullable_bindings", + )) + records_path = "crates/psyche-store/tests/records.rs" + require_terms(records_path, read_text(root, records_path, overrides), ( + "direct_insert_rejects_acknowledged_cancellation_without_evidence", + "CancellationAcknowledgementEvidence", "direct_insert_rejects_mismatched_cancellation_evidence", + )) + result_path = "crates/psyche-coven/tests/fixtures/result-bundle.json" + validate_result_fixture(read_text(root, result_path, overrides)) + bindings_path = "crates/psyche-coven/tests/bindings.rs" + require_terms(bindings_path, read_text(root, bindings_path, overrides), ( + "result_bundle_fixture_round_trips_complete_content_references", + "result_bundle_fixture_uses_launch_request_correlation", + "content_reference_rejects_digest_size_media_type_and_lifetime_mismatch", + '"/artifacts/0/correlation/request_digest"', '"/artifacts/0/correlation/valid_until"', + )) + request_test_path = "crates/psyche-coven/tests/request_digest.rs" + request_tests = read_text(root, request_test_path, overrides) + require_terms(request_test_path, request_tests, ( + "execution_request_launch_matches_golden_bytes_and_digest", + "execution_request_input_matches_golden_bytes_and_digest", + "75d651c5eb7f6e3ccd65631fce08afdcb8ac2a800bc0d8db55eaf9cf43519d04", + "c8c3d0cad99f65d0fdac7b2bb577cf1278412a7ea6255d443e45394109311c61", + )) + for path, digest in ( + ("crates/psyche-coven/tests/fixtures/execution-request-launch.json", "75d651c5eb7f6e3ccd65631fce08afdcb8ac2a800bc0d8db55eaf9cf43519d04"), + ("crates/psyche-coven/tests/fixtures/execution-request-input.json", "c8c3d0cad99f65d0fdac7b2bb577cf1278412a7ea6255d443e45394109311c61"), + ): + raw = overrides[path].encode() if path in overrides else (root / path).read_bytes() + validate_golden(path, raw, digest) + + +def validate_docs(root: pathlib.Path, overrides: Mapping[str, str]) -> None: + architecture = read_text(root, "docs/ARCHITECTURE.md", overrides) + for line in ( + "psyche-core <- psyche-config", "psyche-core <- psyche-store", "psyche-core <- psyche-coven", + "psyche-core <- psyche-surfaces", + "psyche-core + psyche-coven + psyche-surfaces + psyche-store <- psyche-test-support", + "psyche-config + psyche-store <- psyche-runtime <- psyche-cli", + "psyche-config <- psyche-cli", "psyche-store <- psyche-cli", + ): + if architecture.count(line) != 1: + fail(f"architecture dependency direction is absent or duplicated: {line}") + schemas = read_text(root, "docs/SCHEMAS.md", overrides) + require_terms("docs/SCHEMAS.md", schemas, ( + "psyche.identity_snapshot.v1", "psyche.intent.v1", "psyche.surface_event.v1", "psyche.graph.v1", + "psyche.graph_node.v1", "psyche.delegation.v1", "psyche.budget.v1", "psyche.approval.v1", + "psyche.execution_binding.v1", "psyche.evidence.v1", "psyche.verdict.v1", "psyche.recovery.v1", + "psyche.addon.v1", "psyche.surface_effect.v1", "psyche.delivery.v1", "psyche.error.v1", + "unknown kind", "unknown major", "unknown enum", "Transition", "del_", "dlg_", "QuarantineId", + "qua_", "CancellationState", "CancellationAcknowledgementEvidence", "killed", "orphaned", + "ResultBundle", "digest", "media_type", "size_bytes", "expires_at", "Attempt", "att_", + "ExecutionBinding", "retention", "Deferred", + )) + delivery_fields = ( + "schema_version", "delivery_id", "intent_id", "action_class", "account_id", "chat_id", + "topic", "relationship", "effect", "effect_digest", "surface_decision", + "logical_response_id", "logical_part", "state", "attempt_count", "telegram_message_id", + ) + ordered_delivery_shape = "The canonical delivery v1 fields are " + ", ".join( + f"`{field}`" for field in delivery_fields[:-1] + ) + f", and `{delivery_fields[-1]}`." + if ordered_delivery_shape not in " ".join(schemas.split()): + fail("docs/SCHEMAS.md does not freeze the exact ordered delivery v1 fields") + testing = read_text(root, "docs/TESTING.md", overrides) + require_terms("docs/TESTING.md", testing, ( + "scripts", "PROPTEST_CASES", "PROPTEST_RNG_SEED", "crash", "fault", "observation", + "full-request digest", "RFC3339", "g2-test-manifest.json", "killed", "orphaned", + "Immutable Coven", "return", "fence", "restart", "no-redispatch", "ExpectedUnsupported", + ) + tuple(f"C-S{number}" for number in range(1, 13))) + + +def run_json(command: list[str], root: pathlib.Path) -> object: + completed = subprocess.run(command, cwd=root, text=True, capture_output=True, check=False) + if completed.returncode: + fail(f"command failed: {' '.join(command)}\n{completed.stdout}{completed.stderr}") + try: + return json.loads(completed.stdout) + except json.JSONDecodeError as error: + fail(f"command did not return JSON: {' '.join(command)}: {error}") + + +def verify_coven_blob(root: pathlib.Path, url: str, expected_digest: str) -> None: + commit, path = parse_blob_url(url) + response = run_json(["gh", "api", f"repos/OpenCoven/coven/contents/{path}?ref={commit}"], root) + if not isinstance(response, dict) or response.get("type") != "file" or not response.get("sha"): + fail(f"Coven content API did not return a commit-owned blob: {path}") + try: + content = base64.b64decode(str(response["content"]), validate=False) + except (ValueError, TypeError) as error: + fail(f"Coven content API returned invalid base64: {path}: {error}") + if hashlib.sha256(content).hexdigest() != expected_digest: + fail(f"Coven content SHA-256 disagrees with evidence: {path}") + + +def verify_local_source_relationship(root: pathlib.Path, tested: str) -> None: + subprocess.run(["git", "merge-base", "--is-ancestor", tested, "HEAD"], cwd=root, check=True) + changed = subprocess.run( + ["git", "diff", "--name-only", f"{tested}..HEAD"], cwd=root, text=True, capture_output=True, check=True + ).stdout.splitlines() + if changed != ["docs/G2-EVIDENCE.md"]: + fail(f"passed source-to-HEAD diff is not evidence-only: {changed}") + + +def validate_evidence_only_compare(compare: object, tested: str, terminal: str) -> None: + if not isinstance(compare, dict): + fail("GitHub compare response is invalid") + base = compare.get("base_commit") + merge_base = compare.get("merge_base_commit") + commits = compare.get("commits") + if not isinstance(base, dict) or base.get("sha") != tested: + fail("GitHub compare response does not match the tested source") + if ( + not isinstance(commits, list) + or not commits + or not isinstance(commits[-1], dict) + or commits[-1].get("sha") != terminal + ): + fail("GitHub compare response does not end at the pull-request terminal commit") + if compare.get("ahead_by") != len(commits) or compare.get("total_commits") != len(commits): + fail("GitHub compare response commit counts are inconsistent") + if ( + compare.get("status") != "ahead" + or not isinstance(merge_base, dict) + or merge_base.get("sha") != tested + ): + fail("tested source is not the pull-request head's merge-base ancestor") + files = compare.get("files") + if ( + not isinstance(files, list) + or len(files) != 1 + or not isinstance(files[0], dict) + or files[0].get("filename") != "docs/G2-EVIDENCE.md" + or files[0].get("status") != "modified" + ): + fail(f"passed source-to-pull-request diff is not one modified evidence file: {files}") + + +def squash_merge_terminal(root: pathlib.Path, event: dict[str, object]) -> str: + before = event.get("before") + after = event.get("after") + if ( + event.get("ref") != "refs/heads/main" + or not isinstance(before, str) + or not re.fullmatch(r"[0-9a-f]{40}", before) + or not isinstance(after, str) + or not re.fullmatch(r"[0-9a-f]{40}", after) + ): + fail("GitHub Actions main push provenance is invalid") + + pulls = run_json( + [ + "gh", "api", "-H", "Accept: application/vnd.github+json", + f"repos/OpenCoven/psyche/commits/{after}/pulls", + ], + root, + ) + if not isinstance(pulls, list) or len(pulls) != 1 or not isinstance(pulls[0], dict): + fail("GitHub squash commit must belong to exactly one pull request") + pull = pulls[0] + base = pull.get("base") + head = pull.get("head") + base_repository = base.get("repo") if isinstance(base, dict) else None + head_repository = head.get("repo") if isinstance(head, dict) else None + terminal = head.get("sha") if isinstance(head, dict) else None + if ( + pull.get("state") != "closed" + or not isinstance(pull.get("merged_at"), str) + or not pull["merged_at"] + or pull.get("merge_commit_sha") != after + or not isinstance(base, dict) + or base.get("ref") != "main" + or not isinstance(base_repository, dict) + or base_repository.get("full_name") != "OpenCoven/psyche" + or not isinstance(head_repository, dict) + or head_repository.get("full_name") != "OpenCoven/psyche" + or not isinstance(terminal, str) + or not re.fullmatch(r"[0-9a-f]{40}", terminal) + ): + fail("GitHub squash pull-request provenance is invalid") + + merge_commit = run_json(["gh", "api", f"repos/OpenCoven/psyche/git/commits/{after}"], root) + terminal_commit = run_json( + ["gh", "api", f"repos/OpenCoven/psyche/git/commits/{terminal}"], + root, + ) + merge_tree = merge_commit.get("tree") if isinstance(merge_commit, dict) else None + terminal_tree = terminal_commit.get("tree") if isinstance(terminal_commit, dict) else None + parents = merge_commit.get("parents") if isinstance(merge_commit, dict) else None + if ( + not isinstance(merge_commit, dict) + or merge_commit.get("sha") != after + or not isinstance(parents, list) + or len(parents) != 1 + or not isinstance(parents[0], dict) + or parents[0].get("sha") != before + or not isinstance(merge_tree, dict) + or not isinstance(terminal_commit, dict) + or terminal_commit.get("sha") != terminal + or not isinstance(terminal_tree, dict) + or not isinstance(merge_tree.get("sha"), str) + or not re.fullmatch(r"[0-9a-f]{40}", merge_tree["sha"]) + or merge_tree.get("sha") != terminal_tree.get("sha") + ): + fail("GitHub squash commit tree does not match the reviewed pull-request tree") + return terminal + + +def verify_actions_source_relationship(root: pathlib.Path, tested: str) -> None: + event_path = os.environ.get("GITHUB_EVENT_PATH") + if not event_path: + fail("GitHub Actions event payload path is absent") + try: + event = json.loads(pathlib.Path(event_path).read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as error: + fail(f"GitHub Actions event payload is invalid: {error}") + if not isinstance(event, dict): + fail("GitHub Actions event payload is not an object") + + repository = event.get("repository") + if not isinstance(repository, dict) or repository.get("full_name") != "OpenCoven/psyche": + fail("GitHub Actions repository provenance is invalid") + + pull_request = event.get("pull_request") + if isinstance(pull_request, dict): + head = pull_request.get("head") + head_repository = head.get("repo") if isinstance(head, dict) else None + terminal = head.get("sha") if isinstance(head, dict) else None + if ( + not isinstance(head_repository, dict) + or head_repository.get("full_name") != "OpenCoven/psyche" + ): + fail("GitHub Actions pull-request repository provenance is invalid") + if not isinstance(terminal, str) or not re.fullmatch(r"[0-9a-f]{40}", terminal): + fail("GitHub Actions pull-request head provenance is invalid") + else: + terminal = squash_merge_terminal(root, event) + + compare = run_json( + ["gh", "api", f"repos/OpenCoven/psyche/compare/{tested}...{terminal}"], + root, + ) + validate_evidence_only_compare(compare, tested, terminal) + + +def verify_passed(root: pathlib.Path, markdown: str, source_rows: list[list[str]]) -> None: + tested = field(markdown, "Tested source commit") + run_url = field(markdown, "CI attestation") + if os.environ.get("GITHUB_ACTIONS") == "true": + verify_actions_source_relationship(root, tested) + else: + verify_local_source_relationship(root, tested) + match = re.fullmatch(r"https://github\.com/(OpenCoven)/(psyche)/actions/runs/([0-9]+)", run_url) + if not match: + fail("CI attestation URL is malformed") + owner, repo, run_id = match.groups() + run = run_json( + ["gh", "run", "view", run_id, "--repo", f"{owner}/{repo}", "--json", "conclusion,event,headSha,url,workflowName"], + root, + ) + expected_run = { + "conclusion": "success", + "event": "pull_request", + "headSha": tested, + "url": run_url, + "workflowName": "CI", + } + if not isinstance(run, dict) or run != expected_run: + fail(f"CI attestation does not match the tested source: {run}") + rest = run_json(["gh", "api", f"repos/OpenCoven/psyche/actions/runs/{run_id}"], root) + expected_rest = { + "conclusion": "success", + "event": "pull_request", + "head_sha": tested, + "html_url": run_url, + "id": int(run_id), + "path": ".github/workflows/ci.yml", + "status": "completed", + "workflow_id": CI_WORKFLOW_ID, + } + if ( + not isinstance(rest, dict) + or any(rest.get(key) != value for key, value in expected_rest.items()) + or not isinstance(rest.get("repository"), dict) + or rest["repository"].get("full_name") != "OpenCoven/psyche" + or not isinstance(rest.get("head_repository"), dict) + or rest["head_repository"].get("full_name") != "OpenCoven/psyche" + ): + fail(f"CI REST attestation does not match the tested workflow run: {rest}") + workflow = run_json( + ["gh", "api", f"repos/OpenCoven/psyche/actions/workflows/{CI_WORKFLOW_ID}"], + root, + ) + expected_workflow = { + "id": CI_WORKFLOW_ID, + "name": "CI", + "path": ".github/workflows/ci.yml", + "state": "active", + } + if not isinstance(workflow, dict) or any(workflow.get(key) != value for key, value in expected_workflow.items()): + fail(f"CI workflow metadata is not the active reviewed workflow: {workflow}") + for _, url, digest in source_rows: + verify_coven_blob(root, url.strip("`"), normalize_grouped(digest.strip("`"), "sha256:")) + verify_coven_blob(root, field(markdown, "Coven plan URL"), APPROVED_PLAN_SHA256) + + +def validate_repository( + root: pathlib.Path, + *, + manifest: Mapping[str, object] | None = None, + evidence: str | None = None, + listed_tests: Mapping[str, str] | None = None, + source_overrides: Mapping[str, str] | None = None, + verify_remote: bool = True, +) -> None: + root = root.resolve() + overrides = source_overrides or {} + workflow = read_text(root, ".github/workflows/ci.yml", overrides) + validate_ci_workflow(workflow) + for path in ("crates/psyche-store/tests/migrations.rs", "crates/psyche-store/tests/crash.rs"): + if not (root / path).is_file(): + fail(f"store evidence target is absent: {path}") + + manifest_data = manifest + if manifest_data is None: + try: + manifest_data = json.loads(read_text(root, "scripts/g2-test-manifest.json", overrides)) + except json.JSONDecodeError as error: + fail(f"G2 manifest is invalid JSON: {error}") + evidence_text = evidence if evidence is not None else read_text(root, "docs/G2-EVIDENCE.md", overrides) + status, source_rows, matrix = validate_evidence(evidence_text) + validate_manifest(root, manifest_data, matrix, listed_tests) + validate_sources(root, overrides) + validate_docs(root, overrides) + if status == "passed" and verify_remote: + verify_passed(root, evidence_text, source_rows) + + +def main() -> int: + root = pathlib.Path(__file__).resolve().parents[1] + try: + validate_repository(root) + except (EvidenceError, subprocess.CalledProcessError, OSError) as error: + print(f"G2 evidence check failed: {error}", file=sys.stderr) + return 1 + print("G2 evidence relationships verified") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/check-g2-evidence.py b/scripts/check-g2-evidence.py index f5d8ce9..3d752db 100644 --- a/scripts/check-g2-evidence.py +++ b/scripts/check-g2-evidence.py @@ -1,806 +1,314 @@ #!/usr/bin/env python3 -"""Verify that G2 source, tests, CI, and review evidence remain connected.""" +"""Verify G2 evidence and its fail-closed retained-attestation scope.""" from __future__ import annotations -import base64 -import hashlib -import json -import os -import pathlib -import re -import subprocess -import sys -import urllib.parse -from collections.abc import Mapping - - -class EvidenceError(RuntimeError): - """A G2 evidence relationship is absent or inconsistent.""" - - -FIXED_SPEC_COMMIT = "42dcbc4334cb48ecaf63efb550345e3eea2fb7ad" -APPROVED_PLAN_COMMIT = "5f22ebef1e23d045a10f2ec0a3c87be029446cf6" -APPROVED_PLAN_SHA256 = "4fba002ad9f969cd01866ea08f270654f82b53c7d90b73d28643a9abb12cba68" -PLAN_PATH = "docs/superpowers/plans/2026-08-05-psyche-w2-g2-foundation.md" -SOURCE_ROWS = { - "PLAN": ( - "https://github.com/OpenCoven/coven/blob/42dcbc43%334cb48ec%61f63efb5%350345e3e%65a2fb7ad/specs/psyche/PLAN.md", - "01382f8a0d2bca95ddd535634dd6a9f09ac4a80d588ccbebd72f163eaf56bc1e", - ), - "RUNTIME_DESIGN": ( - "https://github.com/OpenCoven/coven/blob/42dcbc43%334cb48ec%61f63efb5%350345e3e%65a2fb7ad/specs/psyche/RUNTIME_DESIGN.md", - "ab8c922214b8f1179ebf71fb8dfb55bd6d0ff2d6dfced4551bf90503767bb6b8", - ), - "TECH": ( - "https://github.com/OpenCoven/coven/blob/42dcbc43%334cb48ec%61f63efb5%350345e3e%65a2fb7ad/specs/psyche/TECH.md", - "1d00fb2b725f384ca027db60d0afbd0a62a7ec6c7dcbb5637bf14d30d40e2e1c", - ), - "COVEN_PREREQUISITES": ( - "https://github.com/OpenCoven/coven/blob/42dcbc43%334cb48ec%61f63efb5%350345e3e%65a2fb7ad/specs/psyche/COVEN_PREREQUISITES.md", - "33994a28921e70f824b0260ce08231b2117c50430c54e996ed47582d060e72f9", - ), - "COVEN_W1_AUDIT": ( - "https://github.com/OpenCoven/coven/blob/42dcbc43%334cb48ec%61f63efb5%350345e3e%65a2fb7ad/specs/psyche/COVEN_W1_AUDIT.md", - "eab9028bf7ef9c8a96d4c6bed69e4ef0b3497b470ca26589cb3ffcd80677322d", - ), -} - -# The commands are plan-owned literals. Hashing keeps this checker readable -# while still rejecting any byte of drift in a command cell. -MATRIX_COMMAND_SHA256 = { - "Canonical ID prefixes and execution-binding identity": "99555c557ce768a2691ea7a7e9e238d73596f0d6d302152fd128e7e2c1f55356", - "Complete canonical error enum": "8f65d986781fb10422a0261e17138201e7b3379ef8c0a09defab7be815dd4bee", - "Canonical delivery v1 shape": "832a79d4126a7cb77b8e0b0341674ca7348ef356091dab02257ccd390a69a384", - "Surface and quarantine owned types": "c5915caab82a15c37e3af5ddc41ec948043daf158fe78c4decc8ed677a17f1f9", - "Package-local nullable-binding fixtures": "2ac01fe93e9fb2580b65464cefaf2a1e942c2240048130d24597688e7d7b8d7a", - "Exhaustive registered decode": "7f744575d04e26cdc6a798d0e315371fcf9bf1b598c4aca9e8052da9d34cd6aa", - "Unknown kind/version/enum denial and quarantine": "ec7296a6b7fb26e7df19fb2c7147bb6225f45c279937b02046110773e468427e", - "Quarantine resolution": "0529a41bff3a7533173055d8d015d1d8a46ae922b0772c8f19fd0bde752bb5d3", - "Direct typed insert validation": "77f4e53c38d571775bf74280f50ea56ce8f802296d1e075425417e269210f5b4", - "Append-only execution-binding revisions": "a3823e0c95245f67996706bc37171569e9369e72166a4b42643da4c90deef71b", - "Transition contract and append-only rules": "65a4b8ac3d7e29f39d2d896ad222477576830b7a2829b41b5c147f0118dde3be", - "Checkpoint-failure shutdown": "979c2261ae0b36b6a1ab5bd34c70a5ec253c5f3120f87876fe33835d18207a69", - "Migrations": "f6d8857ebfc786480c20c2d693ed6cd4398bb5ad5979bb6454d15564d3532580", - "State-machine/property": "ec8c6e799d722b7a382d019853a8405949b464f44f68fb9c6fdb097a789b73bd", - "Crash/restart": "88e020450df496ffb6d915efacb4fc640a2bc325f023ac6332870ec847b71d16", - "Fake boundaries and durable termination ordering": "71355a2738db952c35c0fc18734eb80287602dbf0c10e2aabc96293c05ed9b15", - "Execution request RFC3339 golden bytes": "5315ed856e8dd813132a338a8024779707d8251efd0bdf71b92c2eae57c3a6e1", - "Validated termination dispatch": "68d7ac817187f977be07ffa7668a340222eba600a116380e46ba2b2cd43a67c1", - "G2 cancellation-state vocabulary": "b1822ccbdbd9cb0488227c16b94fd9ad75809c479bbfb01c3b9b8dba08b87caa", - "Full execution-request digest binding": "1ae9e298ddfd97efbd0b673655fbc90fbd78ef650f75fb20a4aaa803d8e166ff", - "C-S1 scripted contract negotiation": "0a0749461006ab5af3c1efd18cbff9b03b05e55045894153ce081160de337a83", - "C-S2 scripted session lifecycle": "19ee8df21953e783e4fa7ac82cee53728deb944ee333a19646b0b4b52cda80b6", - "C-S3 scripted snapshot/attempt binding": "02b6ecde06ab9bb935b81f7d1c003058ed7519ae272b139f1e82e10664510c3c", - "C-S4 scripted stable adoption": "7fa24fdf7b9465a457d33aa1c61c00b6372de8509b4142dbe109b9e6bdc9eb5c", - "C-S5 scripted non-adoption proof": "d25d4488d7d4d8832dca0ac72eec5fbaf09fbb870ff9ff035aeec21b6c039b55", - "C-S6 scripted ambiguity reconciliation/fence": "228e9790d023509f64a3871dfd58c2fd9b9afb4c1c18efa8e827d4a07f93dd84", - "C-S7 scripted ordered cursor": "220c990305d5ab9f080aafbab8b5224da1fc67e29f3a5f5a3d5cf141365675a8", - "C-S8 scripted terminal authority": "5e3223be4f8d8280792509a78661ae892691c6a0513e71fe35ea6752bbb86dc9", - "C-S9 scripted O5 cancellation acknowledgement": "fe4d01e7088f7376df97614522b4eebe3ee3cb7493d216c748d30bc2bf99b26a", - "C-S10 scripted result/artifact binding": "6bcfe3d642acc1a0c258cbaa3ddefd2257bba8a8d8b6e2251d93ce8d07366af3", - "C-S11 scripted restart persistence": "24866d81ddab9d0131cef455721cd2aee7f06fe61dd0320420bdf2617c128860", - "C-S12 scripted structured denial": "3e38ff6947e6e91504727847c5adae7cd3471f998d90c3980b67f520ce955a4a", -} - -CI_COMMANDS = ( - "cargo fmt --all -- --check", - "cargo clippy --workspace --all-targets -- -D warnings", - "cargo test --workspace --locked", - "cargo test -p psyche-test-support --test state_machine", - "cargo test -p psyche-test-support --test conformance", - "cargo test -p psyche-store --test migrations", - "cargo test -p psyche-store --features test-fault-injection --test crash", - "cargo clippy -p psyche-store --all-targets --features test-fault-injection -- -D warnings", - "cargo deny check licenses advisories bans sources", - 'gitleaks detect --no-banner --redact --log-opts="--all"', - "python3 scripts/check-g2-evidence-test.py", - "python3 scripts/check-g2-evidence.py", +# Keep the original reviewed checker byte-for-byte and layer the scope policy +# around it. This makes the trust-root change small, reviewable, and reversible +# while retaining every existing source, workflow, matrix, and remote-run check. +import pathlib as _bootstrap_pathlib + +_LEGACY_PATH = _bootstrap_pathlib.Path(__file__).with_name("check-g2-evidence-v1.py") +try: + _legacy_source = _LEGACY_PATH.read_text(encoding="utf-8") +except OSError as error: + raise RuntimeError(f"cannot load immutable G2 checker core: {error}") from error +_main_marker = '\nif __name__ == "__main__":\n' +if _legacy_source.count(_main_marker) != 1: + raise RuntimeError("immutable G2 checker core has an unexpected main boundary") +exec( + compile(_legacy_source.split(_main_marker, 1)[0], str(_LEGACY_PATH), "exec"), + globals(), ) -NON_RUST_CI_COMMAND_JOBS = { - "cargo deny check licenses advisories bans sources": "supply-chain", - 'gitleaks detect --no-banner --redact --log-opts="--all"': "secrets", +del _legacy_source, _main_marker + +_LEGACY_VERIFY_LOCAL_SOURCE_RELATIONSHIP = verify_local_source_relationship +_LEGACY_VALIDATE_EVIDENCE_ONLY_COMPARE = validate_evidence_only_compare +_LEGACY_VERIFY_ACTIONS_SOURCE_RELATIONSHIP = verify_actions_source_relationship +_LEGACY_VERIFY_PASSED = verify_passed +_LEGACY_VALIDATE_REPOSITORY = validate_repository + +SCOPE_SCHEMA = "psyche.g2-attestation-scope/v1" +SCOPE_MANIFEST_PATH = "scripts/g2-attestation-scope.json" +SCOPE_EVIDENCE_PATH = "docs/G2-SCOPE-EVIDENCE.md" +HISTORIC_EVIDENCE_PATH = "docs/G2-EVIDENCE.md" +SCOPE_MANIFEST_SHA256 = "4e535973d4ab49de80e78cd35177bde658d5c31f6d0fd3a5d91edb579a118fa4" +_SCOPE_TEST_COMMAND = "python3 scripts/check-g2-attestation-scope-test.py" + +if CI_COMMANDS[-1] != "python3 scripts/check-g2-evidence.py" or _SCOPE_TEST_COMMAND in CI_COMMANDS: + fail("immutable G2 checker core has an unexpected CI command boundary") +CI_COMMANDS = CI_COMMANDS[:-1] + (_SCOPE_TEST_COMMAND, CI_COMMANDS[-1]) +REVIEWED_WORKFLOW_SHA256 = "bce224826ebea5f2edfe9a63a45ccb332d1a20189462a5895f403c08dca3088d" + +EXPECTED_UNATTESTED_PATHS = { + "AGENTS.md": {"operations": ("added", "modified"), "modes": ("100644",)}, + "CONTRIBUTING.md": {"operations": ("added", "modified"), "modes": ("100644",)}, + "LICENSE": {"operations": ("added", "modified"), "modes": ("100644",)}, + "README.md": {"operations": ("added", "modified"), "modes": ("100644",)}, + "SECURITY.md": {"operations": ("added", "modified"), "modes": ("100644",)}, + "agent/manifest.yaml": {"operations": ("added", "modified"), "modes": ("100644",)}, + "docs/PROTOCOL-OWNERSHIP.md": {"operations": ("added", "modified"), "modes": ("100644",)}, + "docs/ROADMAP.md": {"operations": ("added", "modified"), "modes": ("100644",)}, + "scripts/agent-bootstrap": {"operations": ("added", "modified"), "modes": ("100755",)}, + "scripts/agent-check": {"operations": ("added", "modified"), "modes": ("100755",)}, } -# Pins the complete reviewed workflow, including setup/actions and every writer -# that could poison GITHUB_ENV or GITHUB_PATH. Newlines are normalized first so -# the same reviewed content verifies on Windows checkouts. -REVIEWED_WORKFLOW_SHA256 = "1f908303c1a8940ce5ec8c81182ddaf5d82e5c6bddd5ece7fb33e1baf9a087f1" -CI_WORKFLOW_ID = 326408880 - - -def fail(message: str) -> None: - raise EvidenceError(message) - - -def read_text(root: pathlib.Path, path: str, overrides: Mapping[str, str]) -> str: - if path in overrides: - return overrides[path] - candidate = root / path - if not candidate.is_file(): - fail(f"required file is absent: {path}") - return candidate.read_text(encoding="utf-8") - - -def parse_tables(markdown: str) -> tuple[list[list[str]], list[list[str]]]: - source: list[list[str]] = [] - matrix: list[list[str]] = [] - active: list[list[str]] | None = None - for line in markdown.splitlines(): - if line == "| Coven source | Immutable URL | SHA-256 |": - active = source - continue - if line == "| Criterion | Command | Result | Artifact |": - active = matrix - continue - if active is not None and line.startswith("|---"): - continue - if active is not None and line.startswith("|"): - cells = [cell.strip() for cell in line.strip().strip("|").split("|")] - if not all(cells): - fail("evidence table contains an empty cell") - active.append(cells) - elif active is not None: - active = None - if any(len(row) != 3 for row in source) or any(len(row) != 4 for row in matrix): - fail("evidence table has an invalid column count") - return source, matrix - - -def field(markdown: str, label: str) -> str: - matches = re.findall(rf"^\*\*{re.escape(label)}:\*\* (.+)$", markdown, re.MULTILINE) - if not matches: - fail(f"missing evidence field: {label}") - if len(matches) != 1: - fail(f"evidence field must occur exactly once: {label}") - return matches[0].strip().strip("`") -def yaml_scalar(value: str) -> str: - value = value.strip() - if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: - return value[1:-1] +def _json_object_without_duplicates(pairs: list[tuple[str, object]]) -> dict[str, object]: + value: dict[str, object] = {} + for key, item in pairs: + if key in value: + fail(f"attestation scope JSON repeats a key: {key}") + value[key] = item return value -def parse_workflow_steps(workflow: str) -> list[dict[str, object]]: - """Extract active named/uses steps without treating YAML comments as data.""" - lines = workflow.splitlines() - try: - jobs_at = lines.index("jobs:") - except ValueError: - fail("workflow jobs mapping is absent") - job_starts = [ - (index, match.group("job")) - for index, line in enumerate(lines[jobs_at + 1:], start=jobs_at + 1) - if (match := re.fullmatch(r" (?P[A-Za-z0-9_-]+):\s*", line)) - ] - starts = [ - (index, len(match.group("indent")), match.group("key")) - for index, line in enumerate(lines) - if (match := re.match(r"^(?P\s*)-\s+(?Pname|uses):", line)) - ] - steps: list[dict[str, object]] = [] - for position, (start, indent, header_key) in enumerate(starts): - owners = [(index, job) for index, job in job_starts if index < start] - if not owners: - fail("workflow step is not contained by a job") - owner_at, owner = owners[-1] - end = next((index for index, _ in job_starts if index > owner_at), len(lines)) - for candidate, candidate_indent, _ in starts[position + 1:]: - if candidate < end and candidate_indent == indent: - end = candidate - break - child = " " * (indent + 2) - grandchild = " " * (indent + 4) - run_values: list[str] = [] - env: dict[str, str] = {} - env_counts: dict[str, int] = {} - direct_counts = {header_key: 1} - invalid = False - in_env = False - for line in lines[start + 1:end]: - if not line.strip() or line.lstrip().startswith("#"): - continue - line_indent = len(line) - len(line.lstrip()) - if in_env and line_indent == indent + 4: - match = re.fullmatch(r"(?P[A-Za-z_][A-Za-z0-9_-]*):(?P.*)", line[len(grandchild):]) - if not match: - invalid = True - continue - key = match.group("key") - env_counts[key] = env_counts.get(key, 0) + 1 - env[key] = yaml_scalar(match.group("value")) - continue - if line_indent == indent + 2: - in_env = False - match = re.fullmatch(r"(?P[A-Za-z_][A-Za-z0-9_-]*):(?P.*)", line[len(child):]) - if not match: - invalid = True - continue - key = match.group("key") - value = match.group("value") - direct_counts[key] = direct_counts.get(key, 0) + 1 - if key == "env": - if value.strip(): - invalid = True - else: - in_env = True - elif key == "run": - run_values.append(yaml_scalar(value)) - continue - if line_indent <= indent + 2: - in_env = False - if len(run_values) > 1: - fail("workflow step contains more than one active run value") - steps.append({ - "run": run_values[0] if run_values else None, - "env": env, - "env_counts": env_counts, - "direct_counts": direct_counts, - "invalid": invalid, - "job": owner, - }) - return steps - - -def workflow_job_lines(workflow: str, job: str) -> list[str]: - lines = workflow.splitlines() - job_pattern = re.compile(rf"^ {re.escape(job)}:\s*$") - starts = [index for index, line in enumerate(lines) if job_pattern.fullmatch(line)] - if len(starts) != 1: - fail(f"workflow must contain exactly one {job} job") - start = starts[0] - end = len(lines) - for index in range(start + 1, len(lines)): - if re.match(r"^ [A-Za-z0-9_-]+:\s*$", lines[index]): - end = index - break - return lines[start + 1:end] - - -def mapping_values(lines: list[str], indent: int) -> dict[str, list[str]]: - values: dict[str, list[str]] = {} - for line in lines: - if not line.strip() or line.lstrip().startswith("#"): - continue - if len(line) - len(line.lstrip()) == indent: - match = re.fullmatch(r"(?P[A-Za-z_][A-Za-z0-9_-]*):(?P.*)", line[indent:]) - if not match: - fail(f"workflow contains a noncanonical key at indentation {indent}: {line.strip()}") - values.setdefault(match.group("key"), []).append(yaml_scalar(match.group("value"))) - return values - - -def nested_mapping_lines(lines: list[str], indent: int, key: str) -> list[str]: - header = " " * indent + key + ":" - starts = [index for index, line in enumerate(lines) if line == header] - if len(starts) != 1: - fail(f"workflow mapping must contain exactly one active {key}") - start = starts[0] - end = len(lines) - for index in range(start + 1, len(lines)): - line = lines[index] - if line.strip() and not line.lstrip().startswith("#") and len(line) - len(line.lstrip()) <= indent: - end = index - break - return lines[start + 1:end] - - -def validate_workflow_scope(workflow: str) -> None: - lines = workflow.splitlines() - for line in lines: - if not line.strip() or line.lstrip().startswith("#"): - continue - if re.search(r"(?:^|[\s:\[\{,])(?:&|\*)[A-Za-z_][A-Za-z0-9_-]*", line) or re.match(r"^\s*<<\s*:", line): - fail("workflow anchors, aliases, and merge keys are not allowed") - root = mapping_values(lines, 0) - expected_root = {"name", "on", "concurrency", "env", "jobs"} - if set(root) != expected_root or any(len(values) != 1 for values in root.values()): - fail("workflow root must use the exact canonical CI structure") - if root["on"] != [""]: - fail("workflow triggers must use the canonical block mapping") - triggers = nested_mapping_lines(lines, 0, "on") - trigger_values = mapping_values(triggers, 2) - if trigger_values != {"push": [""], "pull_request": [""]}: - fail("workflow must run only for main pushes and pull requests") - push = mapping_values(nested_mapping_lines(triggers, 2, "push"), 4) - if push != {"branches": ["[main]"]}: - fail("workflow push trigger must target exactly main") - if mapping_values(nested_mapping_lines(triggers, 2, "pull_request"), 4): - fail("workflow pull_request trigger must be unqualified") - global_env = mapping_values(nested_mapping_lines(lines, 0, "env"), 2) - if global_env != {"CARGO_TERM_COLOR": ["always"], "RUSTFLAGS": ["-D warnings"]}: - fail("workflow global env must contain only fixed non-overriding values") - - -def validate_required_job_shapes(workflow: str) -> None: - expected = { - "rust": {"name", "runs-on", "strategy", "steps"}, - "supply-chain": {"name", "runs-on", "steps"}, - "secrets": {"name", "runs-on", "steps"}, - } - for job, keys in expected.items(): - values = mapping_values(workflow_job_lines(workflow, job), 4) - if set(values) != keys or any(len(entries) != 1 for entries in values.values()): - fail(f"CI required-command job {job} must use its exact canonical direct keys") +def _safe_repository_path(path: object) -> str: + if not isinstance(path, str) or not path or "\\" in path or path.startswith("/") or path.endswith("/"): + fail(f"attestation scope contains an unsafe path: {path}") + pure = pathlib.PurePosixPath(path) + if any(part in {"", ".", ".."} for part in pure.parts): + fail(f"attestation scope contains an unsafe path: {path}") + return path -def validate_rust_matrix(workflow: str) -> None: - rust = workflow_job_lines(workflow, "rust") - direct = mapping_values(rust, 4) - if direct.get("runs-on") != ["${{ matrix.os }}"]: - fail("CI rust job must run on the active matrix.os value") - strategy = nested_mapping_lines(rust, 4, "strategy") - strategy_values = mapping_values(strategy, 6) - if strategy_values.get("fail-fast") != ["false"]: - fail("CI rust strategy must actively set fail-fast to false") - matrix = nested_mapping_lines(strategy, 6, "matrix") - matrix_values = mapping_values(matrix, 8) - if set(matrix_values) != {"os"} or len(matrix_values["os"]) != 1: - fail("CI rust matrix must contain only the supported os axis") - os_value = matrix_values["os"][0] - if not (os_value.startswith("[") and os_value.endswith("]")): - fail("CI rust matrix os axis must be an inline list") - systems = [yaml_scalar(item) for item in os_value[1:-1].split(",") if item.strip()] - expected = {"ubuntu-latest", "macos-latest", "windows-latest"} - if len(systems) != 3 or set(systems) != expected: - fail("CI rust matrix must cover exactly ubuntu, macOS, and Windows") - - -def validate_ci_structure(workflow: str) -> None: - validate_workflow_scope(workflow) - validate_required_job_shapes(workflow) - steps = parse_workflow_steps(workflow) - env_requirements = { - "cargo test -p psyche-test-support --test state_machine": { - "PROPTEST_CASES": "2048", - "PROPTEST_RNG_SEED": "0" * 32, - }, - "python3 scripts/check-g2-evidence.py": {"GH_TOKEN": "${{ github.token }}"}, +def _read_scope_manifest( + root: pathlib.Path, + overrides: Mapping[str, str] | None = None, +) -> dict[str, dict[str, tuple[str, ...]]]: + text = read_text(root, SCOPE_MANIFEST_PATH, overrides or {}) + if hashlib.sha256(text.encode("utf-8")).hexdigest() != SCOPE_MANIFEST_SHA256: + fail("G2 attestation scope manifest SHA-256 drifted") + try: + raw = json.loads(text, object_pairs_hook=_json_object_without_duplicates) + except json.JSONDecodeError as error: + fail(f"G2 attestation scope manifest is invalid JSON: {error}") + if not isinstance(raw, dict) or set(raw) != { + "default", "schema_version", "scope_evidence_path", "unattested_paths" + }: + fail("G2 attestation scope manifest has an unexpected top-level shape") + if raw.get("schema_version") != SCOPE_SCHEMA: + fail("G2 attestation scope schema drifted") + if raw.get("default") != "protected": + fail("G2 attestation scope must default to protected") + if raw.get("scope_evidence_path") != SCOPE_EVIDENCE_PATH: + fail("G2 attestation scope evidence path drifted") + paths = raw.get("unattested_paths") + if not isinstance(paths, dict) or set(paths) != set(EXPECTED_UNATTESTED_PATHS): + fail("G2 unattested path allowlist drifted") + normalized: dict[str, dict[str, tuple[str, ...]]] = {} + for raw_path, rule in paths.items(): + path = _safe_repository_path(raw_path) + if not isinstance(rule, dict) or set(rule) != {"operations", "modes"}: + fail(f"G2 unattested path rule has an unexpected shape: {path}") + operations = rule.get("operations") + modes = rule.get("modes") + if ( + not isinstance(operations, list) + or any(not isinstance(item, str) for item in operations) + or not isinstance(modes, list) + or any(not isinstance(item, str) for item in modes) + ): + fail(f"G2 unattested path rule is not a string-list rule: {path}") + normalized_rule = { + "operations": tuple(operations), + "modes": tuple(modes), + } + if normalized_rule != EXPECTED_UNATTESTED_PATHS[path]: + fail(f"G2 unattested path rule drifted: {path}") + normalized[path] = normalized_rule + return normalized + + +def _validate_scope_evidence(markdown: str) -> tuple[str, str | None, str | None]: + labels = re.findall(r"^\*\*([^:*]+):\*\*", markdown, re.MULTILINE) + expected_labels = { + "Status", + "Tested source commit", + "CI attestation", + "Scope schema", + "Scope manifest", + "Scope manifest SHA-256", + "Historic G2 evidence", } - for command in CI_COMMANDS: - matching = [step for step in steps if step["run"] == command] - if len(matching) != 1: - fail(f"CI workflow must run exact G2 command once: {command}") - expected_job = NON_RUST_CI_COMMAND_JOBS.get(command, "rust") - if matching[0]["job"] != expected_job: - fail(f"CI workflow runs required command outside {expected_job}: {command}") - step = matching[0] - required_env = env_requirements.get(command, {}) - required_direct = {"name": 1, "run": 1} - if required_env: - required_direct["env"] = 1 - if step["invalid"] or step["direct_counts"] != required_direct: - fail(f"CI required command step has noncanonical direct keys: {command}") - if step["env"] != required_env or step["env_counts"] != {key: 1 for key in required_env}: - fail(f"CI required command step has noncanonical env: {command}") - validate_rust_matrix(workflow) - - -def validate_ci_workflow(workflow: str) -> None: - normalized = workflow.replace("\r\n", "\n").replace("\r", "\n") - if hashlib.sha256(normalized.encode("utf-8")).hexdigest() != REVIEWED_WORKFLOW_SHA256: - fail("CI workflow differs from the complete reviewed workflow") - validate_ci_structure(normalized) - - -def normalize_grouped(value: str, prefix: str = "") -> str: - value = value.removeprefix(prefix).replace("-", "") - if not re.fullmatch(r"[0-9a-f]+", value): - fail(f"invalid hexadecimal value: {value}") - return value - - -def parse_blob_url(url: str) -> tuple[str, str]: - parsed = urllib.parse.urlparse(url) - if parsed.scheme != "https" or parsed.netloc != "github.com": - fail(f"Coven source URL is not immutable HTTPS: {url}") - match = re.fullmatch(r"/OpenCoven/coven/blob/([^/]+)/(.+)", parsed.path) - if not match: - fail(f"Coven source URL is not an OpenCoven/coven blob URL: {url}") - commit = urllib.parse.unquote(match.group(1)).replace("-", "") - path = urllib.parse.unquote(match.group(2)) - if not re.fullmatch(r"[0-9a-f]{40}", commit) or path.startswith("/") or ".." in pathlib.PurePosixPath(path).parts: - fail(f"Coven source URL does not name a 40-hex commit and safe path: {url}") - return commit, path - - -def validate_evidence(markdown: str) -> tuple[str, list[list[str]], list[list[str]]]: + if len(labels) != len(expected_labels) or set(labels) != expected_labels: + fail("G2 scope evidence fields are missing, duplicated, or unexpected") status = field(markdown, "Status") if status not in {"candidate", "passed"}: - fail("evidence status must be candidate or passed") - source, matrix = parse_tables(markdown) - source_names = [row[0] for row in source] - if len(source_names) != len(set(source_names)) or set(source_names) != set(SOURCE_ROWS): - fail("Coven source table must contain each fixed source exactly once") - for name, url_cell, digest_cell in source: - url = url_cell.strip("`") - digest = normalize_grouped(digest_cell.strip("`"), "sha256:") - if (url, digest) != SOURCE_ROWS[name]: - fail(f"immutable Coven source row drifted: {name}") - commit, _ = parse_blob_url(url) - if commit != FIXED_SPEC_COMMIT: - fail(f"Coven specification source commit drifted: {name}") - if normalize_grouped(field(markdown, "Coven specification source commit")) != FIXED_SPEC_COMMIT: - fail("Coven specification source field drifted") - - criteria = [row[0] for row in matrix] - if len(criteria) != len(set(criteria)) or set(criteria) != set(MATRIX_COMMAND_SHA256): - fail("evidence matrix must contain every required criterion exactly once") - for criterion, command_cell, result, artifact in matrix: - if not (command_cell.startswith("`") and command_cell.endswith("`")): - fail(f"matrix command is not a code literal: {criterion}") - command = command_cell[1:-1] - if hashlib.sha256(command.encode()).hexdigest() != MATRIX_COMMAND_SHA256[criterion]: - fail(f"matrix command differs from the plan allowlist: {criterion}") - for atomic in command.split(" && "): - if not re.search(r" -- --exact [A-Za-z0-9_:]+$", atomic): - fail(f"matrix command is not an exact filtered test: {atomic}") - + fail("G2 scope evidence status must be candidate or passed") + if field(markdown, "Scope schema") != SCOPE_SCHEMA: + fail("G2 scope evidence schema drifted") + if field(markdown, "Scope manifest") != SCOPE_MANIFEST_PATH: + fail("G2 scope evidence manifest path drifted") + if normalize_grouped(field(markdown, "Scope manifest SHA-256"), "sha256:") != SCOPE_MANIFEST_SHA256: + fail("G2 scope evidence manifest digest drifted") + if field(markdown, "Historic G2 evidence") != HISTORIC_EVIDENCE_PATH: + fail("G2 scope evidence historic evidence path drifted") + tested = field(markdown, "Tested source commit") + run_url = field(markdown, "CI attestation") if status == "candidate": - expected = { - "Tested source commit": "not recorded before remote review", - "CI attestation": "not recorded before remote review", - "Coven plan source commit": "not recorded before plan approval", - "Coven plan URL": "not recorded before plan approval", - "Coven plan SHA-256": "not recorded before plan approval", - } - for label, value in expected.items(): - if field(markdown, label) != value: - fail(f"candidate field must use its exact placeholder: {label}") - if any(row[2:] != ["not run remotely", "none"] for row in matrix): - fail("candidate matrix must use the exact result and artifact placeholders") - else: - tested = field(markdown, "Tested source commit") - run_url = field(markdown, "CI attestation") - plan_commit = normalize_grouped(field(markdown, "Coven plan source commit")) - plan_url = field(markdown, "Coven plan URL") - plan_digest = normalize_grouped(field(markdown, "Coven plan SHA-256"), "sha256:") - if not re.fullmatch(r"[0-9a-f]{40}", tested): - fail("passed evidence requires a 40-hex tested source") - if plan_commit != APPROVED_PLAN_COMMIT or plan_digest != APPROVED_PLAN_SHA256: - fail("passed evidence does not name the approved Coven plan provenance") - url_commit, url_path = parse_blob_url(plan_url) - if url_commit != plan_commit or url_path != PLAN_PATH: - fail("passed evidence plan URL does not match the approved plan") - if not re.fullmatch(r"https://github\.com/OpenCoven/psyche/actions/runs/[0-9]+", run_url): - fail("passed evidence requires an immutable Actions run URL") - if any(row[2] != "passed" or row[3] != run_url for row in matrix): - fail("every passed matrix result and artifact must attest the same run") - placeholders = re.compile(r"not run remotely|\bnone\b|not recorded|pending|placeholder|TBD|TODO", re.IGNORECASE) - if placeholders.search(markdown): - fail("passed evidence contains a candidate placeholder") - return status, source, matrix - - -def manifest_atomic_commands(manifest: Mapping[str, object]) -> tuple[dict[str, tuple[str, str]], dict[str, str]]: - targets = manifest.get("targets") - if not isinstance(targets, dict) or not targets: - fail("manifest.targets must be a non-empty object") - atomic: dict[str, tuple[str, str]] = {} - lists: dict[str, str] = {} - for target, raw in targets.items(): - if not isinstance(target, str) or not isinstance(raw, dict): - fail("manifest target entries must be objects") - list_command = raw.get("list_command") - tests = raw.get("tests") - if not isinstance(list_command, str) or not list_command.endswith(" -- --list --format terse"): - fail(f"invalid list command for {target}") - if not isinstance(tests, list) or not tests or any(not isinstance(name, str) or not name for name in tests): - fail(f"manifest target has no exact tests: {target}") - if len(tests) != len(set(tests)): - fail(f"manifest target repeats a test: {target}") - prefix = list_command.removesuffix(" -- --list --format terse") - lists[target] = list_command - for name in tests: - command = f"{prefix} -- --exact {name}" - if command in atomic: - fail(f"manifest maps an atomic command more than once: {command}") - atomic[command] = (target, name) - return atomic, lists - - -def list_tests(root: pathlib.Path, commands: Mapping[str, str]) -> dict[str, str]: - outputs: dict[str, str] = {} - for target, command in commands.items(): - completed = subprocess.run(command.split(), cwd=root, text=True, capture_output=True, check=False) - if completed.returncode: - fail(f"test listing failed for {target}:\n{completed.stdout}{completed.stderr}") - outputs[target] = completed.stdout - return outputs - - -def validate_manifest( + if tested != "not recorded before remote review" or run_url != "not recorded before remote review": + fail("candidate G2 scope evidence must use exact remote-review placeholders") + return status, None, None + if not re.fullmatch(r"[0-9a-f]{40}", tested): + fail("passed G2 scope evidence requires a 40-hex tested source") + if not re.fullmatch(r"https://github\.com/OpenCoven/psyche/actions/runs/[0-9]+", run_url): + fail("passed G2 scope evidence requires an immutable Actions run URL") + if re.search(r"not recorded|pending|placeholder|TBD|TODO", markdown, re.IGNORECASE): + fail("passed G2 scope evidence contains a candidate placeholder") + return status, tested, run_url + + +def _load_scope_state( root: pathlib.Path, - manifest: Mapping[str, object], - matrix: list[list[str]], - listed_tests: Mapping[str, str] | None, -) -> None: - atomic_manifest, list_commands = manifest_atomic_commands(manifest) - matrix_commands: list[str] = [] - for row in matrix: - matrix_commands.extend(row[1][1:-1].split(" && ")) - if len(matrix_commands) != len(set(matrix_commands)): - fail("an atomic matrix command is duplicated") - if set(matrix_commands) != set(atomic_manifest): - missing = sorted(set(matrix_commands) - set(atomic_manifest)) - unused = sorted(set(atomic_manifest) - set(matrix_commands)) - fail(f"matrix/manifest mismatch; missing={missing}, unused={unused}") - outputs = dict(listed_tests) if listed_tests is not None else list_tests(root, list_commands) - if set(outputs) != set(list_commands): - fail("test listing output does not cover exactly the manifest targets") - for target, output in outputs.items(): - names = { - line.rsplit(": test", 1)[0] - for line in output.splitlines() - if line.endswith(": test") - } - if not names: - fail(f"test target lists zero tests: {target}") - for name in manifest["targets"][target]["tests"]: # type: ignore[index] - if name not in names: - fail(f"exact manifest test is absent from {target}: {name}") - - -def require_terms(path: str, text: str, terms: tuple[str, ...]) -> None: - missing = [term for term in terms if term not in text] - if missing: - fail(f"{path} is missing required relationships: {missing}") - - -def require_digest_mutation_matrix(path: str, text: str) -> None: - common = ( - "schema_version", "request_id", "graph_id", "node_id", "attempt_id", - "principal_id", "familiar_snapshot_id", "project_id", - "context_manifest_digest", "required_artifact_bindings", "payload_digest", - "created_at", "valid_until", - ) - launch = common + ("project_root", "cwd", "harness", "delegation_digest", "budget_digest") - input_request = common + ("session_id", "input_digest") - artifact = ("artifact_id", "digest", "media_type", "size") - for name in launch + input_request: - if f'"/input/{name}"' not in text: - fail(f"{path} does not stale-digest mutate request field: {name}") - for name in artifact: - if f'"/input/required_artifact_bindings/0/{name}"' not in text: - fail(f"{path} does not stale-digest mutate artifact field: {name}") - if 'mutations.push(("/input", other_input))' not in text: - fail(f"{path} does not stale-digest mutate the request variant") - - -def validate_record_kinds(source: str) -> None: - enum = re.search(r"pub enum RecordKind \{(.*?)\n\}", source, re.DOTALL) - if not enum: - fail("RecordKind declaration is absent") - variants = re.findall(r"^\s{4}([A-Z][A-Za-z0-9]+),$", enum.group(1), re.MULTILINE) - if len(variants) != 15 or "Attempt" not in variants or "ExecutionBinding" in variants: - fail(f"RecordKind must have exactly 15 variants with Attempt only: {variants}") - all_block = re.search(r"pub const ALL: \[RecordKind; 15\] = \[(.*?)\];", source, re.DOTALL) - prefixes = re.search(r"pub const fn prefix\(self\).*?match self \{(.*?)\n\s*\}", source, re.DOTALL) - if not all_block or re.findall(r"RecordKind::([A-Za-z0-9]+)", all_block.group(1)) != variants: - fail("RecordKind::ALL is not exhaustive and declaration-ordered") - if not prefixes: - fail("RecordKind prefix match is absent") - pairs = re.findall(r'RecordKind::([A-Za-z0-9]+) => "([a-z]{3}_)"', prefixes.group(1)) - if [name for name, _ in pairs] != variants: - fail("RecordKind prefix match does not use the exact variant set") - if dict(pairs).get("Attempt") != "att_" or sum(prefix == "att_" for _, prefix in pairs) != 1: - fail("Attempt must be the only att_ record kind") - if source.count("SchemaKind::ExecutionBinding => Some(RecordKind::Attempt)") != 1: - fail("ExecutionBinding must map exactly once to RecordKind::Attempt") - - -def validate_result_fixture(text: str) -> None: - try: - bundle = json.loads(text) - except json.JSONDecodeError as error: - fail(f"result-bundle fixture is invalid JSON: {error}") - if set(bundle) != {"artifacts", "correlation", "result", "session_id"}: - fail("result-bundle fixture is not strict and complete") - if not isinstance(bundle["artifacts"], list) or not bundle["artifacts"]: - fail("result-bundle fixture has no artifact") - references = [bundle["result"]] + [artifact.get("content", {}) for artifact in bundle["artifacts"]] - for reference in references: - if set(reference) != {"digest", "expires_at", "media_type", "size_bytes"}: - fail("every result/artifact content reference needs digest, media_type, size_bytes, expires_at") - if not re.fullmatch(r"sha256:[0-9a-f]{64}", str(reference["digest"])): - fail("content reference digest is not canonical") - if not isinstance(reference["size_bytes"], int) or reference["size_bytes"] <= 0: - fail("content reference size is invalid") - if not re.fullmatch(r"[^/\s]+/[^/\s]+", str(reference["media_type"])): - fail("content reference media type is invalid") - if not re.fullmatch(r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ", str(reference["expires_at"])): - fail("content reference expiry is not canonical RFC3339 UTC") - correlation = bundle["correlation"] - for artifact in bundle["artifacts"]: - if artifact.get("correlation") != correlation or artifact.get("session_id") != bundle["session_id"]: - fail("artifact lifetime/correlation does not match its result bundle") - - -def validate_golden(path: str, raw: bytes, expected_sha: str) -> None: - if raw.endswith(b"\n"): - fail(f"golden fixture has a trailing newline: {path}") - try: - value = json.loads(raw) - except json.JSONDecodeError as error: - fail(f"golden fixture is invalid JSON: {path}: {error}") - for key in ("created_at", "valid_until"): - if not isinstance(value.get(key), str) or not re.fullmatch(r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ", value[key]): - fail(f"golden fixture lacks an RFC3339 string {key}: {path}") - canonical = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() - if canonical != raw: - fail(f"golden fixture is not canonical JSON: {path}") - if hashlib.sha256(raw).hexdigest() != expected_sha: - fail(f"golden fixture SHA-256 drifted: {path}") - - -def validate_sources(root: pathlib.Path, overrides: Mapping[str, str]) -> None: - port_path = "crates/psyche-coven/src/port.rs" - port = read_text(root, port_path, overrides) - require_terms(port_path, port, ( - "pub fn new(input: ExecutionRequestInput)", "let request_digest = digest(&input)?;", - "pub fn recompute_digest", "pub trait CovenPort", "async fn reconcile(", - "pub struct ResultBundle", "CancellationAcknowledgementEvidence", - )) - if re.search(r"(?:struct|enum)\s+\w*Acknowledgement\w*", port): - fail("psyche-coven owns a duplicate acknowledgement wire type") - - suite_path = "crates/psyche-test-support/src/suites/coven.rs" - suite = read_text(root, suite_path, overrides) - for number in range(1, 13): - if suite.count(f"pub async fn assert_c_s{number}_") != 1: - fail(f"reusable C-S{number} function is absent or duplicated") - require_terms(suite_path, suite, ( - "stale_digest_mutations", "request.recompute_digest()", "RequestDigestMismatch", - "RAW_LEDGER_STATES", '"killed"', '"orphaned"', "CancellationAcknowledgementEvidence", - "assert_c_s6_ambiguity_fence", "ReconciliationDisposition::Returned", - "ReconciliationDisposition::Fenced", "DurableDispositionKind::Returned", - "DurableDispositionKind::Fenced", "fixture.restart().await", "require_fault(", - "redispatch_eligibility", "EligibleAfterFence", "assert_c_s10_result_artifact_binding", - "mutate_result_digest", "mutate_result_media_type", "mutate_result_size", - "mutate_result_expiry", "mutate_artifact_digest", "mutate_artifact_media_type", - "mutate_artifact_size", "mutate_artifact_expiry", - )) - require_digest_mutation_matrix(suite_path, suite) - fake_path = "crates/psyche-test-support/src/coven.rs" - fake = read_text(root, fake_path, overrides) - require_terms(fake_path, fake, ("async fn adopt", "request.validate_digest()?;")) - if suite.count("request.validate_digest()?;") != 1: - fail("scripted Coven boundary must recompute the request digest exactly once") - conformance_path = "crates/psyche-test-support/tests/conformance.rs" - conformance = read_text(root, conformance_path, overrides) - for number in range(1, 13): - if len(re.findall(rf"async fn c_s{number}_[a-z0-9_]+\(\)", conformance)) != 1: - fail(f"exact C-S{number} wrapper is absent or duplicated") - - state_path = "crates/psyche-test-support/tests/state_machine.rs" - state = read_text(root, state_path, overrides) - require_terms(state_path, state, ( - "c_s6_model_never_redispatches_without_fence", "request_digest_binds_every_typed_field", - "stale_digest_requests", "MutateRequestFieldRetainDigest", "RequestDigestMismatch", - "ReconciliationDisposition::Returned", "ReconciliationDisposition::Fenced", - "fixture.restart().await", "select_fault(", "RedispatchEligibility::EligibleAfterFence", - )) - require_digest_mutation_matrix(state_path, state) - - core_path = "crates/psyche-core/src/contracts/mod.rs" - core = read_text(root, core_path, overrides) - validate_record_kinds(core) - require_terms(core_path, core, ( - "const ALL: [SchemaKind; 16]", "CanonicalDocument::ExecutionBinding", - "SchemaKind::Error => None", "UnknownSchema", "UnsupportedMajor", "UnknownEnumValue", - )) - error_path = "crates/psyche-core/src/contracts/error.rs" - require_terms(error_path, read_text(root, error_path, overrides), ("pub const ALL: [Self; 36]",)) - contracts_path = "crates/psyche-core/tests/contracts.rs" - require_terms(contracts_path, read_text(root, contracts_path, overrides), ( - "all_canonical_error_codes_decode", "delivery_v1_fixture_round_trips_canonically", - "surface_event_and_effect_fixtures_round_trip", "cancellation_state_vocabulary_requires_matching_o5_evidence", - "graph_and_node_accept_only_the_two_frozen_nullable_bindings", - )) - records_path = "crates/psyche-store/tests/records.rs" - require_terms(records_path, read_text(root, records_path, overrides), ( - "direct_insert_rejects_acknowledged_cancellation_without_evidence", - "CancellationAcknowledgementEvidence", "direct_insert_rejects_mismatched_cancellation_evidence", - )) - result_path = "crates/psyche-coven/tests/fixtures/result-bundle.json" - validate_result_fixture(read_text(root, result_path, overrides)) - bindings_path = "crates/psyche-coven/tests/bindings.rs" - require_terms(bindings_path, read_text(root, bindings_path, overrides), ( - "result_bundle_fixture_round_trips_complete_content_references", - "result_bundle_fixture_uses_launch_request_correlation", - "content_reference_rejects_digest_size_media_type_and_lifetime_mismatch", - '"/artifacts/0/correlation/request_digest"', '"/artifacts/0/correlation/valid_until"', - )) - request_test_path = "crates/psyche-coven/tests/request_digest.rs" - request_tests = read_text(root, request_test_path, overrides) - require_terms(request_test_path, request_tests, ( - "execution_request_launch_matches_golden_bytes_and_digest", - "execution_request_input_matches_golden_bytes_and_digest", - "75d651c5eb7f6e3ccd65631fce08afdcb8ac2a800bc0d8db55eaf9cf43519d04", - "c8c3d0cad99f65d0fdac7b2bb577cf1278412a7ea6255d443e45394109311c61", - )) - for path, digest in ( - ("crates/psyche-coven/tests/fixtures/execution-request-launch.json", "75d651c5eb7f6e3ccd65631fce08afdcb8ac2a800bc0d8db55eaf9cf43519d04"), - ("crates/psyche-coven/tests/fixtures/execution-request-input.json", "c8c3d0cad99f65d0fdac7b2bb577cf1278412a7ea6255d443e45394109311c61"), - ): - raw = overrides[path].encode() if path in overrides else (root / path).read_bytes() - validate_golden(path, raw, digest) + overrides: Mapping[str, str] | None = None, +) -> tuple[dict[str, dict[str, tuple[str, ...]]], str, str | None, str | None]: + scope = _read_scope_manifest(root, overrides) + evidence = read_text(root, SCOPE_EVIDENCE_PATH, overrides or {}) + status, tested, run_url = _validate_scope_evidence(evidence) + return scope, status, tested, run_url + + +def _historic_tested_source(root: pathlib.Path) -> str: + return field(read_text(root, HISTORIC_EVIDENCE_PATH, {}), "Tested source commit") + + +def _entry_map_from_remote_tree(tree: object, expected_tree_sha: str) -> dict[str, tuple[str, str, str]]: + if not isinstance(tree, dict) or tree.get("sha") != expected_tree_sha: + fail("GitHub recursive tree response does not match the requested tree") + if tree.get("truncated") is not False: + fail("GitHub recursive tree response is truncated or missing its truncation proof") + entries = tree.get("tree") + if not isinstance(entries, list): + fail("GitHub recursive tree response has no entry list") + result: dict[str, tuple[str, str, str]] = {} + for raw in entries: + if not isinstance(raw, dict): + fail("GitHub recursive tree contains a non-object entry") + entry_type = raw.get("type") + if entry_type == "tree": + continue + path = _safe_repository_path(raw.get("path")) + mode = raw.get("mode") + sha = raw.get("sha") + if ( + entry_type not in {"blob", "commit"} + or not isinstance(mode, str) + or not re.fullmatch(r"[0-9]{6}", mode) + or not isinstance(sha, str) + or not re.fullmatch(r"[0-9a-f]{40}", sha) + or path in result + ): + fail(f"GitHub recursive tree contains an invalid or duplicate entry: {path}") + result[path] = (mode, entry_type, sha) + return result -def validate_docs(root: pathlib.Path, overrides: Mapping[str, str]) -> None: - architecture = read_text(root, "docs/ARCHITECTURE.md", overrides) - for line in ( - "psyche-core <- psyche-config", "psyche-core <- psyche-store", "psyche-core <- psyche-coven", - "psyche-core <- psyche-surfaces", - "psyche-core + psyche-coven + psyche-surfaces + psyche-store <- psyche-test-support", - "psyche-config + psyche-store <- psyche-runtime <- psyche-cli", - "psyche-config <- psyche-cli", "psyche-store <- psyche-cli", +def _remote_tree_entries(root: pathlib.Path, commit: str) -> dict[str, tuple[str, str, str]]: + commit_data = run_json(["gh", "api", f"repos/OpenCoven/psyche/git/commits/{commit}"], root) + tree = commit_data.get("tree") if isinstance(commit_data, dict) else None + tree_sha = tree.get("sha") if isinstance(tree, dict) else None + if ( + not isinstance(commit_data, dict) + or commit_data.get("sha") != commit + or not isinstance(tree_sha, str) + or not re.fullmatch(r"[0-9a-f]{40}", tree_sha) ): - if architecture.count(line) != 1: - fail(f"architecture dependency direction is absent or duplicated: {line}") - schemas = read_text(root, "docs/SCHEMAS.md", overrides) - require_terms("docs/SCHEMAS.md", schemas, ( - "psyche.identity_snapshot.v1", "psyche.intent.v1", "psyche.surface_event.v1", "psyche.graph.v1", - "psyche.graph_node.v1", "psyche.delegation.v1", "psyche.budget.v1", "psyche.approval.v1", - "psyche.execution_binding.v1", "psyche.evidence.v1", "psyche.verdict.v1", "psyche.recovery.v1", - "psyche.addon.v1", "psyche.surface_effect.v1", "psyche.delivery.v1", "psyche.error.v1", - "unknown kind", "unknown major", "unknown enum", "Transition", "del_", "dlg_", "QuarantineId", - "qua_", "CancellationState", "CancellationAcknowledgementEvidence", "killed", "orphaned", - "ResultBundle", "digest", "media_type", "size_bytes", "expires_at", "Attempt", "att_", - "ExecutionBinding", "retention", "Deferred", - )) - delivery_fields = ( - "schema_version", "delivery_id", "intent_id", "action_class", "account_id", "chat_id", - "topic", "relationship", "effect", "effect_digest", "surface_decision", - "logical_response_id", "logical_part", "state", "attempt_count", "telegram_message_id", + fail("GitHub commit response does not identify the requested commit tree") + response = run_json( + ["gh", "api", f"repos/OpenCoven/psyche/git/trees/{tree_sha}?recursive=1"], + root, ) - ordered_delivery_shape = "The canonical delivery v1 fields are " + ", ".join( - f"`{field}`" for field in delivery_fields[:-1] - ) + f", and `{delivery_fields[-1]}`." - if ordered_delivery_shape not in " ".join(schemas.split()): - fail("docs/SCHEMAS.md does not freeze the exact ordered delivery v1 fields") - testing = read_text(root, "docs/TESTING.md", overrides) - require_terms("docs/TESTING.md", testing, ( - "scripts", "PROPTEST_CASES", "PROPTEST_RNG_SEED", "crash", "fault", "observation", - "full-request digest", "RFC3339", "g2-test-manifest.json", "killed", "orphaned", - "Immutable Coven", "return", "fence", "restart", "no-redispatch", "ExpectedUnsupported", - ) + tuple(f"C-S{number}" for number in range(1, 13))) + return _entry_map_from_remote_tree(response, tree_sha) -def run_json(command: list[str], root: pathlib.Path) -> object: - completed = subprocess.run(command, cwd=root, text=True, capture_output=True, check=False) +def _local_tree_entries(root: pathlib.Path, ref: str) -> dict[str, tuple[str, str, str]]: + completed = subprocess.run( + ["git", "ls-tree", "-r", "-z", "--full-tree", ref], + cwd=root, + capture_output=True, + check=False, + ) if completed.returncode: - fail(f"command failed: {' '.join(command)}\n{completed.stdout}{completed.stderr}") - try: - return json.loads(completed.stdout) - except json.JSONDecodeError as error: - fail(f"command did not return JSON: {' '.join(command)}: {error}") - - -def verify_coven_blob(root: pathlib.Path, url: str, expected_digest: str) -> None: - commit, path = parse_blob_url(url) - response = run_json(["gh", "api", f"repos/OpenCoven/coven/contents/{path}?ref={commit}"], root) - if not isinstance(response, dict) or response.get("type") != "file" or not response.get("sha"): - fail(f"Coven content API did not return a commit-owned blob: {path}") - try: - content = base64.b64decode(str(response["content"]), validate=False) - except (ValueError, TypeError) as error: - fail(f"Coven content API returned invalid base64: {path}: {error}") - if hashlib.sha256(content).hexdigest() != expected_digest: - fail(f"Coven content SHA-256 disagrees with evidence: {path}") - - -def verify_local_source_relationship(root: pathlib.Path, tested: str) -> None: - subprocess.run(["git", "merge-base", "--is-ancestor", tested, "HEAD"], cwd=root, check=True) - changed = subprocess.run( - ["git", "diff", "--name-only", f"{tested}..HEAD"], cwd=root, text=True, capture_output=True, check=True - ).stdout.splitlines() - if changed != ["docs/G2-EVIDENCE.md"]: - fail(f"passed source-to-HEAD diff is not evidence-only: {changed}") + stderr = completed.stderr.decode("utf-8", "replace") if isinstance(completed.stderr, bytes) else str(completed.stderr) + fail(f"cannot read Git tree for retained G2 scope at {ref}: {stderr.strip()}") + output = completed.stdout if isinstance(completed.stdout, bytes) else str(completed.stdout).encode() + result: dict[str, tuple[str, str, str]] = {} + for record in output.split(b"\0"): + if not record: + continue + try: + metadata, raw_path = record.split(b"\t", 1) + mode, entry_type, sha = metadata.decode("ascii").split(" ", 2) + path = raw_path.decode("utf-8") + except (ValueError, UnicodeDecodeError) as error: + fail(f"local Git tree contains an unparseable entry: {error}") + path = _safe_repository_path(path) + if ( + entry_type not in {"blob", "commit"} + or not re.fullmatch(r"[0-9]{6}", mode) + or not re.fullmatch(r"[0-9a-f]{40}", sha) + or path in result + ): + fail(f"local Git tree contains an invalid or duplicate entry: {path}") + result[path] = (mode, entry_type, sha) + return result + + +def _change_status( + before: tuple[str, str, str] | None, + after: tuple[str, str, str] | None, +) -> str: + if before is None: + return "added" + if after is None: + return "deleted" + return "modified" + + +def _validate_attested_changes( + before: Mapping[str, tuple[str, str, str]], + after: Mapping[str, tuple[str, str, str]], + scope: Mapping[str, Mapping[str, tuple[str, ...]]], +) -> list[tuple[str, str]]: + changes: list[tuple[str, str]] = [] + rejected: list[tuple[str, str]] = [] + for path in sorted(set(before) | set(after)): + old = before.get(path) + new = after.get(path) + if old == new: + continue + status = _change_status(old, new) + changes.append((path, status)) + if path == SCOPE_EVIDENCE_PATH: + if ( + status != "modified" + or old is None + or new is None + or old[0] != "100644" + or new[0] != "100644" + or old[1] != "blob" + or new[1] != "blob" + ): + rejected.append((path, status)) + continue + rule = scope.get(path) + if ( + rule is None + or status not in rule["operations"] + or new is None + or new[1] != "blob" + or new[0] not in rule["modes"] + ): + rejected.append((path, status)) + if rejected: + rendered = ", ".join(f"{path} ({status})" for path, status in rejected[:20]) + suffix = "" if len(rejected) <= 20 else f", and {len(rejected) - 20} more" + fail(f"retained G2 attestation does not cover changed paths: {rendered}{suffix}") + return changes -def validate_evidence_only_compare(compare: object, tested: str, terminal: str) -> None: +def _compare_common(compare: object, tested: str, terminal: str) -> tuple[str, list[dict[str, object]]]: if not isinstance(compare, dict): fail("GitHub compare response is invalid") base = compare.get("base_commit") @@ -816,95 +324,98 @@ def validate_evidence_only_compare(compare: object, tested: str, terminal: str) ): fail("GitHub compare response does not end at the pull-request terminal commit") if compare.get("ahead_by") != len(commits) or compare.get("total_commits") != len(commits): - fail("GitHub compare response commit counts are inconsistent") + fail("GitHub compare response commit counts are inconsistent or truncated") + status = compare.get("status") + if status not in {"ahead", "diverged"}: + fail("GitHub compare response has an unsupported source relationship") if ( - compare.get("status") != "ahead" - or not isinstance(merge_base, dict) - or merge_base.get("sha") != tested + not isinstance(merge_base, dict) + or not isinstance(merge_base.get("sha"), str) + or not re.fullmatch(r"[0-9a-f]{40}", merge_base["sha"]) ): - fail("tested source is not the pull-request head's merge-base ancestor") + fail("GitHub compare response has no valid merge base") files = compare.get("files") - if ( - not isinstance(files, list) - or len(files) != 1 - or not isinstance(files[0], dict) - or files[0].get("filename") != "docs/G2-EVIDENCE.md" - or files[0].get("status") != "modified" - ): - fail(f"passed source-to-pull-request diff is not one modified evidence file: {files}") + if not isinstance(files, list) or not files or len(files) >= 300: + fail("GitHub compare file list is absent or potentially truncated") + normalized: list[dict[str, object]] = [] + seen: set[str] = set() + for raw in files: + if not isinstance(raw, dict): + fail("GitHub compare file list contains a non-object entry") + path = _safe_repository_path(raw.get("filename")) + file_status = raw.get("status") + if path in seen or file_status not in {"added", "modified", "removed", "renamed", "copied", "changed"}: + fail(f"GitHub compare file list contains an invalid or duplicate entry: {raw}") + if file_status in {"renamed", "copied"}: + fail(f"GitHub compare uses an unsupported rename/copy status: {path}") + seen.add(path) + normalized.append(raw) + return str(status), normalized + + +def _compare_requires_tree_scope( + compare: object, + tested: str, + terminal: str, + scope: Mapping[str, Mapping[str, tuple[str, ...]]], +) -> bool: + status, files = _compare_common(compare, tested, terminal) + exact_legacy_evidence = ( + len(files) == 1 + and files[0].get("filename") == HISTORIC_EVIDENCE_PATH + and files[0].get("status") == "modified" + ) + merge_base = compare["merge_base_commit"] + is_ancestor = status == "ahead" and isinstance(merge_base, dict) and merge_base.get("sha") == tested + if exact_legacy_evidence: + if is_ancestor: + return False + fail("tested source is not the pull-request head's merge-base ancestor") + for raw in files: + path = str(raw["filename"]) + file_status = str(raw["status"]) + if path == SCOPE_EVIDENCE_PATH: + if file_status != "modified": + fail(f"passed source-to-pull-request diff has an invalid scope evidence change: {raw}") + continue + rule = scope.get(path) + translated = "deleted" if file_status == "removed" else file_status + if rule is None or translated not in rule["operations"]: + if is_ancestor: + fail(f"passed source-to-pull-request diff is not one modified evidence file: {files}") + fail(f"retained G2 attestation does not cover compare path: {path} ({file_status})") + return True -def squash_merge_terminal(root: pathlib.Path, event: dict[str, object]) -> str: - before = event.get("before") - after = event.get("after") - if ( - event.get("ref") != "refs/heads/main" - or not isinstance(before, str) - or not re.fullmatch(r"[0-9a-f]{40}", before) - or not isinstance(after, str) - or not re.fullmatch(r"[0-9a-f]{40}", after) - ): - fail("GitHub Actions main push provenance is invalid") +def validate_evidence_only_compare(compare: object, tested: str, terminal: str) -> None: + """Retain the historic exact-evidence contract used by the v1 mutation tests.""" + _LEGACY_VALIDATE_EVIDENCE_ONLY_COMPARE(compare, tested, terminal) - pulls = run_json( - [ - "gh", "api", "-H", "Accept: application/vnd.github+json", - f"repos/OpenCoven/psyche/commits/{after}/pulls", - ], - root, - ) - if not isinstance(pulls, list) or len(pulls) != 1 or not isinstance(pulls[0], dict): - fail("GitHub squash commit must belong to exactly one pull request") - pull = pulls[0] - base = pull.get("base") - head = pull.get("head") - base_repository = base.get("repo") if isinstance(base, dict) else None - head_repository = head.get("repo") if isinstance(head, dict) else None - terminal = head.get("sha") if isinstance(head, dict) else None - if ( - pull.get("state") != "closed" - or not isinstance(pull.get("merged_at"), str) - or not pull["merged_at"] - or pull.get("merge_commit_sha") != after - or not isinstance(base, dict) - or base.get("ref") != "main" - or not isinstance(base_repository, dict) - or base_repository.get("full_name") != "OpenCoven/psyche" - or not isinstance(head_repository, dict) - or head_repository.get("full_name") != "OpenCoven/psyche" - or not isinstance(terminal, str) - or not re.fullmatch(r"[0-9a-f]{40}", terminal) - ): - fail("GitHub squash pull-request provenance is invalid") - merge_commit = run_json(["gh", "api", f"repos/OpenCoven/psyche/git/commits/{after}"], root) - terminal_commit = run_json( - ["gh", "api", f"repos/OpenCoven/psyche/git/commits/{terminal}"], - root, - ) - merge_tree = merge_commit.get("tree") if isinstance(merge_commit, dict) else None - terminal_tree = terminal_commit.get("tree") if isinstance(terminal_commit, dict) else None - parents = merge_commit.get("parents") if isinstance(merge_commit, dict) else None - if ( - not isinstance(merge_commit, dict) - or merge_commit.get("sha") != after - or not isinstance(parents, list) - or len(parents) != 1 - or not isinstance(parents[0], dict) - or parents[0].get("sha") != before - or not isinstance(merge_tree, dict) - or not isinstance(terminal_commit, dict) - or terminal_commit.get("sha") != terminal - or not isinstance(terminal_tree, dict) - or not isinstance(merge_tree.get("sha"), str) - or not re.fullmatch(r"[0-9a-f]{40}", merge_tree["sha"]) - or merge_tree.get("sha") != terminal_tree.get("sha") - ): - fail("GitHub squash commit tree does not match the reviewed pull-request tree") - return terminal +def _scope_relationship_for_actual_evidence(root: pathlib.Path, tested: str) -> tuple[ + dict[str, dict[str, tuple[str, ...]]], str, str | None +] | None: + if tested != _historic_tested_source(root): + return None + scope, status, scope_tested, _ = _load_scope_state(root) + return scope, status, scope_tested -def verify_actions_source_relationship(root: pathlib.Path, tested: str) -> None: +def verify_local_source_relationship(root: pathlib.Path, tested: str) -> None: + state = _scope_relationship_for_actual_evidence(root, tested) + if state is None: + _LEGACY_VERIFY_LOCAL_SOURCE_RELATIONSHIP(root, tested) + return + scope, status, scope_tested = state + if status == "candidate": + return + assert scope_tested is not None + before = _local_tree_entries(root, scope_tested) + after = _local_tree_entries(root, "HEAD") + _validate_attested_changes(before, after, scope) + + +def _actions_terminal(root: pathlib.Path) -> str: event_path = os.environ.get("GITHUB_EVENT_PATH") if not event_path: fail("GitHub Actions event payload path is absent") @@ -914,46 +425,61 @@ def verify_actions_source_relationship(root: pathlib.Path, tested: str) -> None: fail(f"GitHub Actions event payload is invalid: {error}") if not isinstance(event, dict): fail("GitHub Actions event payload is not an object") - repository = event.get("repository") if not isinstance(repository, dict) or repository.get("full_name") != "OpenCoven/psyche": fail("GitHub Actions repository provenance is invalid") - pull_request = event.get("pull_request") if isinstance(pull_request, dict): + base = pull_request.get("base") head = pull_request.get("head") + base_repository = base.get("repo") if isinstance(base, dict) else None head_repository = head.get("repo") if isinstance(head, dict) else None terminal = head.get("sha") if isinstance(head, dict) else None if ( - not isinstance(head_repository, dict) + not isinstance(base, dict) + or base.get("ref") != "main" + or not isinstance(base_repository, dict) + or base_repository.get("full_name") != "OpenCoven/psyche" + or not isinstance(head_repository, dict) or head_repository.get("full_name") != "OpenCoven/psyche" ): - fail("GitHub Actions pull-request repository provenance is invalid") + fail("GitHub Actions pull-request repository/base provenance is invalid") if not isinstance(terminal, str) or not re.fullmatch(r"[0-9a-f]{40}", terminal): fail("GitHub Actions pull-request head provenance is invalid") - else: - terminal = squash_merge_terminal(root, event) + return terminal + return squash_merge_terminal(root, event) + +def verify_actions_source_relationship(root: pathlib.Path, tested: str) -> None: + state = _scope_relationship_for_actual_evidence(root, tested) + if state is None: + _LEGACY_VERIFY_ACTIONS_SOURCE_RELATIONSHIP(root, tested) + return + scope, status, scope_tested = state + if status == "candidate": + return + assert scope_tested is not None + terminal = _actions_terminal(root) compare = run_json( - ["gh", "api", f"repos/OpenCoven/psyche/compare/{tested}...{terminal}"], + ["gh", "api", f"repos/OpenCoven/psyche/compare/{scope_tested}...{terminal}"], root, ) - validate_evidence_only_compare(compare, tested, terminal) + needs_trees = _compare_requires_tree_scope(compare, scope_tested, terminal, scope) + if not needs_trees: + return + before = _remote_tree_entries(root, scope_tested) + after = _remote_tree_entries(root, terminal) + _validate_attested_changes(before, after, scope) -def verify_passed(root: pathlib.Path, markdown: str, source_rows: list[list[str]]) -> None: - tested = field(markdown, "Tested source commit") - run_url = field(markdown, "CI attestation") - if os.environ.get("GITHUB_ACTIONS") == "true": - verify_actions_source_relationship(root, tested) - else: - verify_local_source_relationship(root, tested) +def _verify_scope_ci_attestation(root: pathlib.Path, tested: str, run_url: str) -> None: match = re.fullmatch(r"https://github\.com/(OpenCoven)/(psyche)/actions/runs/([0-9]+)", run_url) if not match: - fail("CI attestation URL is malformed") + fail("G2 scope CI attestation URL is malformed") owner, repo, run_id = match.groups() run = run_json( - ["gh", "run", "view", run_id, "--repo", f"{owner}/{repo}", "--json", "conclusion,event,headSha,url,workflowName"], + ["gh", "run", "view", run_id, "--repo", f"{owner}/{repo}", "--json", + "conclusion,event,headSha,url,workflowName"], root, ) expected_run = { @@ -964,7 +490,7 @@ def verify_passed(root: pathlib.Path, markdown: str, source_rows: list[list[str] "workflowName": "CI", } if not isinstance(run, dict) or run != expected_run: - fail(f"CI attestation does not match the tested source: {run}") + fail(f"G2 scope CI attestation does not match the tested source: {run}") rest = run_json(["gh", "api", f"repos/OpenCoven/psyche/actions/runs/{run_id}"], root) expected_rest = { "conclusion": "success", @@ -984,8 +510,8 @@ def verify_passed(root: pathlib.Path, markdown: str, source_rows: list[list[str] or not isinstance(rest.get("head_repository"), dict) or rest["head_repository"].get("full_name") != "OpenCoven/psyche" ): - fail(f"CI REST attestation does not match the tested workflow run: {rest}") - workflow = run_json( + fail(f"G2 scope CI REST attestation does not match the tested workflow run: {rest}") + workflow_metadata = run_json( ["gh", "api", f"repos/OpenCoven/psyche/actions/workflows/{CI_WORKFLOW_ID}"], root, ) @@ -995,11 +521,22 @@ def verify_passed(root: pathlib.Path, markdown: str, source_rows: list[list[str] "path": ".github/workflows/ci.yml", "state": "active", } - if not isinstance(workflow, dict) or any(workflow.get(key) != value for key, value in expected_workflow.items()): - fail(f"CI workflow metadata is not the active reviewed workflow: {workflow}") - for _, url, digest in source_rows: - verify_coven_blob(root, url.strip("`"), normalize_grouped(digest.strip("`"), "sha256:")) - verify_coven_blob(root, field(markdown, "Coven plan URL"), APPROVED_PLAN_SHA256) + if ( + not isinstance(workflow_metadata, dict) + or any(workflow_metadata.get(key) != value for key, value in expected_workflow.items()) + ): + fail(f"G2 scope CI workflow metadata is not the active reviewed workflow: {workflow_metadata}") + + +def verify_passed(root: pathlib.Path, markdown: str, source_rows: list[list[str]]) -> None: + tested = field(markdown, "Tested source commit") + _LEGACY_VERIFY_PASSED(root, markdown, source_rows) + if tested != _historic_tested_source(root): + return + _, status, scope_tested, run_url = _load_scope_state(root) + if status == "passed": + assert scope_tested is not None and run_url is not None + _verify_scope_ci_attestation(root, scope_tested, run_url) def validate_repository( @@ -1011,37 +548,26 @@ def validate_repository( source_overrides: Mapping[str, str] | None = None, verify_remote: bool = True, ) -> None: - root = root.resolve() - overrides = source_overrides or {} - workflow = read_text(root, ".github/workflows/ci.yml", overrides) - validate_ci_workflow(workflow) - for path in ("crates/psyche-store/tests/migrations.rs", "crates/psyche-store/tests/crash.rs"): - if not (root / path).is_file(): - fail(f"store evidence target is absent: {path}") - - manifest_data = manifest - if manifest_data is None: - try: - manifest_data = json.loads(read_text(root, "scripts/g2-test-manifest.json", overrides)) - except json.JSONDecodeError as error: - fail(f"G2 manifest is invalid JSON: {error}") - evidence_text = evidence if evidence is not None else read_text(root, "docs/G2-EVIDENCE.md", overrides) - status, source_rows, matrix = validate_evidence(evidence_text) - validate_manifest(root, manifest_data, matrix, listed_tests) - validate_sources(root, overrides) - validate_docs(root, overrides) - if status == "passed" and verify_remote: - verify_passed(root, evidence_text, source_rows) + _load_scope_state(root.resolve(), source_overrides or {}) + _LEGACY_VALIDATE_REPOSITORY( + root, + manifest=manifest, + evidence=evidence, + listed_tests=listed_tests, + source_overrides=source_overrides, + verify_remote=verify_remote, + ) def main() -> int: root = pathlib.Path(__file__).resolve().parents[1] try: validate_repository(root) + _, scope_status, _, _ = _load_scope_state(root) except (EvidenceError, subprocess.CalledProcessError, OSError) as error: print(f"G2 evidence check failed: {error}", file=sys.stderr) return 1 - print("G2 evidence relationships verified") + print(f"G2 evidence relationships verified; scope attestation {scope_status}") return 0 diff --git a/scripts/g2-attestation-scope.json b/scripts/g2-attestation-scope.json new file mode 100644 index 0000000..92be55b --- /dev/null +++ b/scripts/g2-attestation-scope.json @@ -0,0 +1,97 @@ +{ + "default": "protected", + "schema_version": "psyche.g2-attestation-scope/v1", + "scope_evidence_path": "docs/G2-SCOPE-EVIDENCE.md", + "unattested_paths": { + "AGENTS.md": { + "modes": [ + "100644" + ], + "operations": [ + "added", + "modified" + ] + }, + "CONTRIBUTING.md": { + "modes": [ + "100644" + ], + "operations": [ + "added", + "modified" + ] + }, + "LICENSE": { + "modes": [ + "100644" + ], + "operations": [ + "added", + "modified" + ] + }, + "README.md": { + "modes": [ + "100644" + ], + "operations": [ + "added", + "modified" + ] + }, + "SECURITY.md": { + "modes": [ + "100644" + ], + "operations": [ + "added", + "modified" + ] + }, + "agent/manifest.yaml": { + "modes": [ + "100644" + ], + "operations": [ + "added", + "modified" + ] + }, + "docs/PROTOCOL-OWNERSHIP.md": { + "modes": [ + "100644" + ], + "operations": [ + "added", + "modified" + ] + }, + "docs/ROADMAP.md": { + "modes": [ + "100644" + ], + "operations": [ + "added", + "modified" + ] + }, + "scripts/agent-bootstrap": { + "modes": [ + "100755" + ], + "operations": [ + "added", + "modified" + ] + }, + "scripts/agent-check": { + "modes": [ + "100755" + ], + "operations": [ + "added", + "modified" + ] + } + } +}