Read the guard's registry and contracts from the evaluated commit

The completion guard's CI mode now compares contract text against the
evaluated commit (#2502), but it still decided which contracts a commit
owed from HEAD's registry and contract index. CI walks a branch with HEAD
at the tip (or at main after a merge), so a later commit could change an
earlier commit's verdict. If a later commit added a Shared Boundaries
reference to a file, an earlier commit that was complete when written got
blocked. If a later commit dropped ownership of a file, an earlier commit
that changed it without its contract passed. The pre-commit hook read
both from the commit's own index, so CI and the hook could still
disagree.

repo_file_io and subsystem_contracts can now read from a revision as well
as the working tree or index. A revision read uses only that commit's
tree and fails closed rather than falling back to the disk. In diff-base
mode the guard reads the registry and contract index from the evaluated
commit, as does the Prettier-only inference that --commit enables.
Replaying CI's per-commit loop over the last 120 commits on main gives
the same verdicts as before.

scripts/release_control/canonical_completion_guard.py

@@ -58,7 +58,7 @@ def prettier_only_contract_neutral_reason(
     if not commit:
         return None
 
-    impacted = infer_impacted_subsystems(staged_files)
+    impacted = infer_impacted_subsystems(staged_files, revision=commit)
     runtime_paths = {
         path
         for data in impacted.values()
@@ -121,8 +121,8 @@ def validate_verification_policy(
             raise ValueError(f"subsystem {subsystem_id} {context} missing {field}")
 
 
-def load_subsystem_rules(*, staged: bool = False) -> List[dict]:
-    payload = load_repo_json(SUBSYSTEM_REGISTRY, staged=staged)
+def load_subsystem_rules(*, staged: bool = False, revision: str | None = None) -> List[dict]:
+    payload = load_repo_json(SUBSYSTEM_REGISTRY, staged=staged, revision=revision)
     rules = list(payload.get("subsystems", []))
     for rule in rules:
         subsystem_id = str(rule.get("id"))
@@ -300,9 +300,10 @@ def infer_impacted_subsystems(
     staged_files: Sequence[str],
     *,
     use_staged_registry: bool = False,
+    revision: str | None = None,
 ) -> Dict[str, dict]:
     impacted: Dict[str, dict] = {}
-    rules = load_subsystem_rules(staged=use_staged_registry)
+    rules = load_subsystem_rules(staged=use_staged_registry, revision=revision)
     rules_by_id = {str(rule["id"]): rule for rule in rules}
 
     for path in staged_files:
@@ -352,10 +353,15 @@ def required_contract_updates(
     impacted: Dict[str, dict] | None = None,
     *,
     use_staged_contract_index: bool = False,
+    revision: str | None = None,
 ) -> Dict[str, dict]:
-    impacted_subsystems = impacted if impacted is not None else infer_impacted_subsystems(staged_files)
+    impacted_subsystems = (
+        impacted
+        if impacted is not None
+        else infer_impacted_subsystems(staged_files, revision=revision)
+    )
     required: Dict[str, dict] = {}
-    contract_index = load_contract_index(staged=use_staged_contract_index)
+    contract_index = load_contract_index(staged=use_staged_contract_index, revision=revision)
 
     for subsystem_id, data in impacted_subsystems.items():
         # Contract update requirement triggers on non-cycle-artifact touches
@@ -670,11 +676,22 @@ def check_staged_contracts(
     inferred_override_reason: str | None = None,
 ) -> int:
     staged_set: Set[str] = set(staged_files)
-    impacted = infer_impacted_subsystems(staged_files, use_staged_registry=True)
+    # Judge the change by the rules it was written under: the staged registry
+    # and contracts before a commit exists, the evaluated commit's own after.
+    # Reading HEAD's in CI let a later commit on the branch (or main merged
+    # under it) add or drop ownership and contract references for an
+    # earlier commit, so its verdict depended on what came after it.
+    governance_revision = revision if diff_base is not None else None
+    impacted = infer_impacted_subsystems(
+        staged_files,
+        use_staged_registry=True,
+        revision=governance_revision,
+    )
     required_contracts = required_contract_updates(
         staged_files,
         impacted,
         use_staged_contract_index=True,
+        revision=governance_revision,
     )
     missing_contracts = {
         contract_path: data

scripts/release_control/canonical_completion_guard_test.py

@@ -1,4 +1,5 @@
 import io
+import json
 import os
 import subprocess
 import tempfile
@@ -3962,6 +3963,148 @@ class EvaluatedCommitContractComparisonTest(unittest.TestCase):
         self.assertEqual(resolve_diff_base(f"{first}^", first), parent)
 
 
+class EvaluatedCommitGovernanceSourceTest(unittest.TestCase):
+    """The registry and contract index decide which contracts a commit owes.
+    In CI they must come from the evaluated commit, as the pre-commit hook
+    read them from that commit's index, not from the branch tip: a later
+    commit (or main merged under the branch) that adds a contract reference
+    blocked an earlier commit that was complete when written, and one that
+    drops ownership let an earlier incomplete commit pass."""
+
+    SUBSYSTEMS = "docs/release-control/v6/internal/subsystems"
+    REGISTRY = f"{SUBSYSTEMS}/registry.json"
+    ALPHA = f"{SUBSYSTEMS}/alpha.md"
+    BETA = f"{SUBSYSTEMS}/beta.md"
+    RUNTIME = "app/alpha/widget.go"
+    PROOF = "app/alpha/widget_test.go"
+
+    def setUp(self) -> None:
+        temp_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(temp_dir.cleanup)
+        self.repo = Path(temp_dir.name)
+        for target in (
+            patch.dict(os.environ, strip_local_git_env(os.environ.copy()), clear=True),
+            patch("canonical_completion_guard.REPO_ROOT", self.repo),
+            patch("canonical_completion_guard.SUBSYSTEM_REGISTRY", self.repo / self.REGISTRY),
+            patch("repo_file_io.REPO_ROOT", self.repo),
+            patch("subsystem_contracts.REPO_ROOT", self.repo),
+            patch("subsystem_contracts.CONTRACTS_DIR", self.repo / self.SUBSYSTEMS),
+        ):
+            target.start()
+            self.addCleanup(target.stop)
+        self.git("init", "-q", "-b", "main")
+        self.git("config", "user.name", "Pulse Test")
+        self.git("config", "user.email", "pulse-test@example.invalid")
+        self.commit(
+            "base",
+            {
+                self.REGISTRY: self.registry(["app/alpha/"]),
+                self.ALPHA: self.contract("alpha", "Alpha owns the widget."),
+                self.BETA: self.contract("beta", "Beta renders summaries."),
+                self.RUNTIME: "package alpha\n",
+            },
+        )
+
+    @staticmethod
+    def registry(alpha_prefixes: list[str]) -> str:
+        def rule(subsystem_id: str, prefixes: list[str]) -> dict:
+            return {
+                "id": subsystem_id,
+                "contract": f"docs/release-control/v6/internal/subsystems/{subsystem_id}.md",
+                "owned_prefixes": prefixes,
+                "owned_files": [],
+                "verification": {
+                    "allow_same_subsystem_tests": False,
+                    "test_prefixes": [f"app/{subsystem_id}/"],
+                    "exact_files": [],
+                    "require_explicit_path_policy_coverage": False,
+                    "path_policies": [],
+                },
+            }
+
+        rules = [rule("alpha", alpha_prefixes), rule("beta", ["app/beta/"])]
+        return json.dumps({"subsystems": rules}, indent=2) + "\n"
+
+    @staticmethod
+    def contract(subsystem_id: str, current_state: str, shared_boundary: str = "") -> str:
+        boundary = f"1. `{shared_boundary}` feeds this subsystem.\n" if shared_boundary else ""
+        return (
+            f"# {subsystem_id} contract\n\n"
+            "## Contract Metadata\n\n"
+            "```json\n"
+            f'{{"subsystem_id": "{subsystem_id}"}}\n'
+            "```\n\n"
+            "## Shared Boundaries\n\n"
+            f"{boundary}\n"
+            "## Current State\n\n"
+            f"{current_state}\n"
+        )
+
+    def git(self, *args: str) -> str:
+        return subprocess.run(
+            ["git", *args],
+            cwd=self.repo,
+            check=True,
+            capture_output=True,
+            text=True,
+        ).stdout.strip()
+
+    def commit(self, message: str, files: dict[str, str]) -> str:
+        for rel, text in files.items():
+            path = self.repo / rel
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(text, encoding="utf-8")
+        self.git("add", *files)
+        self.git("commit", "-q", "-m", message)
+        return self.git("rev-parse", "HEAD")
+
+    def run_guard(self, commit: str) -> tuple[int, str]:
+        files = self.git("diff-tree", "--no-commit-id", "--name-only", "-r", commit)
+        stderr = io.StringIO()
+        with (
+            patch.dict(os.environ, {CONTRACT_NEUTRAL_OVERRIDE_ENV: ""}),
+            patch("sys.stdin", io.StringIO(files + "\n")),
+            redirect_stderr(stderr),
+        ):
+            status = main(["--files-from-stdin", "--diff-base", f"{commit}^", "--commit", commit])
+        return status, stderr.getvalue()
+
+    def test_later_contract_reference_does_not_block_an_earlier_complete_commit(self) -> None:
+        complete = self.commit(
+            "change the widget with its contract",
+            {
+                self.RUNTIME: "package alpha\n\nconst Size = 2\n",
+                self.PROOF: "package alpha\n",
+                self.ALPHA: self.contract("alpha", "Alpha owns the widget and its size."),
+            },
+        )
+        self.commit(
+            "beta starts depending on the widget",
+            {
+                self.BETA: self.contract(
+                    "beta", "Beta renders widget summaries.", shared_boundary=self.RUNTIME
+                ),
+            },
+        )
+
+        status, stderr = self.run_guard(complete)
+        self.assertEqual(status, 0, stderr)
+
+    def test_later_ownership_change_does_not_pass_an_earlier_incomplete_commit(self) -> None:
+        incomplete = self.commit(
+            "change the widget without its contract",
+            {
+                self.RUNTIME: "package alpha\n\nconst Size = 3\n",
+                self.PROOF: "package alpha\n",
+            },
+        )
+        self.commit("retire alpha's ownership of app/alpha", {self.REGISTRY: self.registry([])})
+
+        status, stderr = self.run_guard(incomplete)
+        self.assertEqual(status, 1)
+        self.assertIn(f"- subsystem alpha: missing contract {self.ALPHA}", stderr)
+
+
 class ReleaseCycleArtifactGuardTest(unittest.TestCase):
     """The release_cycle_artifact_globs field in a subsystem rule lets
     per-RC-cycle artifacts (packet drafts, version-pointer docs, regenerated

scripts/release_control/repo_file_io.py

@@ -234,8 +234,31 @@ def repo_relative_path(path: str | Path) -> str:
     return candidate.as_posix()
 
 
-def read_repo_text(path: str | Path, *, staged: bool = False, strict_staged: bool = False) -> str:
+def read_repo_text(
+    path: str | Path,
+    *,
+    staged: bool = False,
+    strict_staged: bool = False,
+    revision: str | None = None,
+) -> str:
+    """Read a repo file from the working tree, the index, or a commit.
+
+    `revision` wins over `staged` and never falls back to the working tree:
+    a caller judging a specific commit must not silently read another tree.
+    """
     rel = repo_relative_path(path)
+    if revision is not None:
+        result = subprocess.run(
+            ["git", "show", f"{revision}:{rel}"],
+            cwd=REPO_ROOT,
+            check=False,
+            capture_output=True,
+            text=True,
+            env=git_env(),
+        )
+        if result.returncode != 0:
+            raise FileNotFoundError(f"missing {rel} at revision {revision}")
+        return result.stdout
     if staged:
         try:
             result = subprocess.run(
@@ -275,5 +298,13 @@ def missing_staged_repo_paths(paths: Iterable[str | Path]) -> list[str]:
     return missing
 
 
-def load_repo_json(path: str | Path, *, staged: bool = False, strict_staged: bool = False) -> dict[str, Any]:
-    return json.loads(read_repo_text(path, staged=staged, strict_staged=strict_staged))
+def load_repo_json(
+    path: str | Path,
+    *,
+    staged: bool = False,
+    strict_staged: bool = False,
+    revision: str | None = None,
+) -> dict[str, Any]:
+    return json.loads(
+        read_repo_text(path, staged=staged, strict_staged=strict_staged, revision=revision)
+    )

scripts/release_control/repo_file_io_test.py

@@ -117,6 +117,37 @@ class RepoFileIoTest(unittest.TestCase):
                     self.assertEqual(read_repo_text(rel, staged=True), '{"version": "temporary-index"}\n')
                     self.assertEqual(load_repo_json(rel, staged=True), {"version": "temporary-index"})
 
+    def test_revision_reads_ignore_the_index_and_working_tree(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            repo_root = Path(tmpdir)
+            rel = "docs/release-control/v6/internal/subsystems/registry.json"
+            path = repo_root / rel
+            path.parent.mkdir(parents=True, exist_ok=True)
+            identity = ("-c", "user.name=Pulse Test", "-c", "user.email=pulse-test@example.invalid")
+
+            self.git(repo_root, "init")
+            path.write_text('{"version": "evaluated"}\n', encoding="utf-8")
+            self.git(repo_root, "add", rel)
+            self.git(repo_root, *identity, "commit", "-q", "-m", "evaluated")
+            evaluated = self.git_stdout(repo_root, "rev-parse", "HEAD")
+            path.write_text('{"version": "later"}\n', encoding="utf-8")
+            self.git(repo_root, "add", rel)
+            self.git(repo_root, *identity, "commit", "-q", "-m", "later")
+            path.write_text('{"version": "staged"}\n', encoding="utf-8")
+            self.git(repo_root, "add", rel)
+            path.write_text('{"version": "working-tree"}\n', encoding="utf-8")
+
+            with patch("repo_file_io.REPO_ROOT", repo_root):
+                self.assertEqual(load_repo_json(rel, revision=evaluated), {"version": "evaluated"})
+                self.assertEqual(
+                    load_repo_json(rel, staged=True, revision="HEAD"),
+                    {"version": "later"},
+                )
+                # A file the commit lacks fails closed rather than reading
+                # another tree's copy.
+                with self.assertRaisesRegex(FileNotFoundError, "missing .* at revision"):
+                    read_repo_text("docs/absent.json", revision=evaluated)
+
     def test_strict_staged_read_rejects_worktree_only_file(self) -> None:
         with tempfile.TemporaryDirectory() as tmpdir:
             repo_root = Path(tmpdir)

scripts/release_control/subsystem_contracts.py

@@ -72,21 +72,20 @@ def git(*args: str, text: bool) -> subprocess.CompletedProcess:
     )
 
 
-def tracked_contract_paths(*, staged: bool = False) -> list[str]:
-    if staged:
-        result = git(
-            "ls-files",
-            "-z",
-            "--",
-            CONTRACTS_DIR.relative_to(REPO_ROOT).as_posix(),
-            text=False,
-        )
+def tracked_contract_paths(*, staged: bool = False, revision: str | None = None) -> list[str]:
+    if staged or revision is not None:
+        contracts_rel = CONTRACTS_DIR.relative_to(REPO_ROOT).as_posix()
+        if revision is not None:
+            result = git("ls-tree", "-r", "-z", "--name-only", revision, "--", contracts_rel, text=False)
+        else:
+            result = git("ls-files", "-z", "--", contracts_rel, text=False)
         git_paths = sorted(
             entry.decode("utf-8")
             for entry in result.stdout.split(b"\x00")
             if entry and entry.decode("utf-8").endswith(".md")
         )
-        if git_paths:
+        if git_paths or revision is not None:
+            # A commit's contracts are what its tree holds, never the disk's.
             return git_paths
         # Governance contract files are filesystem-only (gitignored) — fall back to disk
     return sorted(
@@ -95,20 +94,27 @@ def tracked_contract_paths(*, staged: bool = False) -> list[str]:
     )
 
 
-def staged_contract_text(rel: str) -> str:
-    result = git("show", f":{rel}", text=True)
+def git_contract_text(rel: str, revision: str | None = None) -> str:
+    """Read a contract from `revision`, or from the index when it is None."""
+    result = git("show", f"{revision or ''}:{rel}", text=True)
     return result.stdout
 
 
-def tracked_contract_files(*, staged: bool = False) -> dict[str, str]:
+def tracked_contract_files(*, staged: bool = False, revision: str | None = None) -> dict[str, str]:
+    """Read contracts from the working tree, the index, or a commit.
+
+    `revision` wins over `staged`; a contract the commit tracks is read only
+    from that commit, so a caller judging it never sees another tree's text.
+    """
     payload: dict[str, str] = {}
-    for rel in tracked_contract_paths(staged=staged):
-        if staged:
+    for rel in tracked_contract_paths(staged=staged, revision=revision):
+        if staged or revision is not None:
             try:
-                payload[rel] = staged_contract_text(rel)
+                payload[rel] = git_contract_text(rel, revision)
                 continue
             except subprocess.CalledProcessError:
-                pass
+                if revision is not None:
+                    raise
         payload[rel] = (REPO_ROOT / rel).read_text(encoding="utf-8")
     return payload
 
@@ -219,9 +225,11 @@ def load_contract_index(
     contract_texts: dict[str, str] | None = None,
     *,
     staged: bool = False,
+    revision: str | None = None,
 ) -> dict[str, dict[str, Any]]:
     contract_index: dict[str, dict[str, Any]] = {}
-    for rel, content in (contract_texts or tracked_contract_files(staged=staged)).items():
+    texts = contract_texts or tracked_contract_files(staged=staged, revision=revision)
+    for rel, content in texts.items():
         if rel == TEMPLATE_REL or not rel.endswith(".md"):
             continue
         parsed, _ = parse_contract_text(rel, content)

scripts/release_control/subsystem_contracts_test.py

@@ -251,5 +251,49 @@ Stable.
                 )
 
 
+    def test_tracked_contract_files_can_read_a_commits_contracts(self) -> None:
+        # A CI check of an earlier commit on a branch must see that commit's
+        # contracts, not the ones a later commit, the index or the working
+        # tree hold.
+        with tempfile.TemporaryDirectory() as tmpdir:
+            repo_root = Path(tmpdir)
+            contracts_dir = repo_root / "docs" / "release-control" / "v6" / "internal" / "subsystems"
+            contracts_dir.mkdir(parents=True, exist_ok=True)
+            first_rel = "docs/release-control/v6/internal/subsystems/first.md"
+            later_rel = "docs/release-control/v6/internal/subsystems/later.md"
+            identity = ("-c", "user.name=Pulse Test", "-c", "user.email=pulse-test@example.invalid")
+
+            self.git(repo_root, "init")
+            (repo_root / first_rel).write_text("# committed first\n", encoding="utf-8")
+            self.git(repo_root, "add", first_rel)
+            self.git(repo_root, *identity, "commit", "-q", "-m", "first")
+            evaluated = self.git(repo_root, "rev-parse", "HEAD").stdout.strip()
+            (repo_root / first_rel).write_text("# committed later\n", encoding="utf-8")
+            (repo_root / later_rel).write_text("# later\n", encoding="utf-8")
+            self.git(repo_root, "add", first_rel, later_rel)
+            self.git(repo_root, *identity, "commit", "-q", "-m", "later")
+            (repo_root / first_rel).write_text("# staged\n", encoding="utf-8")
+            self.git(repo_root, "add", first_rel)
+            (repo_root / first_rel).write_text("# working tree\n", encoding="utf-8")
+
+            with (
+                patch("subsystem_contracts.REPO_ROOT", repo_root),
+                patch("subsystem_contracts.CONTRACTS_DIR", contracts_dir),
+            ):
+                self.assertEqual(tracked_contract_paths(revision=evaluated), [first_rel])
+                self.assertEqual(
+                    tracked_contract_files(revision=evaluated),
+                    {first_rel: "# committed first\n"},
+                )
+                self.assertEqual(
+                    tracked_contract_files(staged=True, revision="HEAD"),
+                    {first_rel: "# committed later\n", later_rel: "# later\n"},
+                )
+                # Unlike the index read, a commit without contracts does not
+                # borrow the working tree's.
+                empty_tree = self.git(repo_root, "hash-object", "-t", "tree", "/dev/null").stdout.strip()
+                self.assertEqual(tracked_contract_files(revision=empty_tree), {})
+
+
 if __name__ == "__main__":
     unittest.main()