From df55901611ab3e94db4af463c77a297fd51aa58b Mon Sep 17 00:00:00 2001 From: rcourtman Date: Wed, 13 May 2026 14:45:14 +0100 Subject: [PATCH] Fix registry audit repo identity in worktrees --- scripts/release_control/registry_audit.py | 12 ++-- .../release_control/registry_audit_test.py | 65 ++++++++++++++++++- scripts/release_control/repo_file_io.py | 36 +++++++++- scripts/release_control/repo_file_io_test.py | 36 +++++++++- 4 files changed, 140 insertions(+), 9 deletions(-) diff --git a/scripts/release_control/registry_audit.py b/scripts/release_control/registry_audit.py index 3a1e8a978b..08b0bafba3 100644 --- a/scripts/release_control/registry_audit.py +++ b/scripts/release_control/registry_audit.py @@ -20,7 +20,7 @@ from canonical_completion_guard import ( subsystem_matches_path, ) from control_plane import DEFAULT_CONTROL_PLANE -from repo_file_io import load_repo_json +from repo_file_io import canonical_repo_id, canonical_workspace_repos_root, load_repo_json from status_audit import load_status_payload @@ -62,12 +62,13 @@ def sorted_casefold(values: list[str]) -> list[str]: def tracked_repo_files() -> set[str]: - return tracked_workspace_files(active_repos=[REPO_ROOT.name], local_repo=REPO_ROOT.name) + local_repo = canonical_repo_id(REPO_ROOT) + return tracked_workspace_files(active_repos=[local_repo], local_repo=local_repo) def tracked_workspace_files(*, active_repos: list[str], local_repo: str) -> set[str]: files: set[str] = set() - repos_root = REPO_ROOT.parent + repos_root = canonical_workspace_repos_root(REPO_ROOT) for repo_id in active_repos: repo_root = REPO_ROOT if repo_id == local_repo else repos_root / repo_id if not repo_root.exists(): @@ -633,7 +634,8 @@ def main(argv: list[str] | None = None) -> int: repo_id for repo_id in scope.get("active_repos", []) if isinstance(repo_id, str) and repo_id.strip() - ] or [REPO_ROOT.name] + ] or [canonical_repo_id(REPO_ROOT)] + local_repo = canonical_repo_id(REPO_ROOT) lane_ids = { lane.get("id") for lane in status_payload.get("lanes", []) @@ -641,7 +643,7 @@ def main(argv: list[str] | None = None) -> int: } report = audit_registry_payload( load_registry_payload(staged=args.staged), - tracked_files=tracked_workspace_files(active_repos=active_repos, local_repo=REPO_ROOT.name), + tracked_files=tracked_workspace_files(active_repos=active_repos, local_repo=local_repo), status_lane_ids=lane_ids, schema_contract=registry_schema_contract(staged=args.staged), ) diff --git a/scripts/release_control/registry_audit_test.py b/scripts/release_control/registry_audit_test.py index b074b4ea2b..a79dec84b9 100644 --- a/scripts/release_control/registry_audit_test.py +++ b/scripts/release_control/registry_audit_test.py @@ -1,14 +1,77 @@ +import os +import subprocess +import tempfile import unittest +from pathlib import Path +from unittest.mock import patch -from registry_audit import audit_registry_payload, parse_args +import registry_audit +from repo_file_io import canonical_repo_id +from registry_audit import audit_registry_payload, parse_args, tracked_workspace_files class RegistryAuditTest(unittest.TestCase): + def git(self, repo_root: Path, *args: str) -> subprocess.CompletedProcess: + env = os.environ.copy() + env.pop("GIT_INDEX_FILE", None) + return subprocess.run( + ["git", *args], + cwd=repo_root, + check=True, + capture_output=True, + text=True, + env=env, + ) + def test_parse_args_accepts_staged_flag(self) -> None: args = parse_args(["--check", "--staged"]) self.assertTrue(args.check) self.assertTrue(args.staged) + def test_tracked_workspace_files_uses_linked_worktree_as_canonical_local_repo(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + workspace = Path(tmpdir) / "workspace" + repo_root = workspace / "repos" / "pulse" + linked_worktree = workspace / ".worktrees" / "pulse-first-session-onboarding-parity" + repo_root.mkdir(parents=True) + linked_worktree.parent.mkdir(parents=True) + + self.git(repo_root, "init") + (repo_root / "internal").mkdir() + (repo_root / "internal" / "existing.go").write_text("package internal\n", encoding="utf-8") + self.git(repo_root, "add", "internal/existing.go") + self.git( + repo_root, + "-c", + "user.name=Pulse Test", + "-c", + "user.email=pulse-test@example.invalid", + "commit", + "-m", + "initial", + ) + self.git(repo_root, "worktree", "add", "--detach", str(linked_worktree), "HEAD") + + (linked_worktree / "internal" / "staged.go").write_text("package internal\n", encoding="utf-8") + self.git(linked_worktree, "add", "internal/staged.go") + contracts_dir = linked_worktree / "docs" / "release-control" / "v6" / "internal" / "subsystems" + contracts_dir.mkdir(parents=True) + + with patch("registry_audit.REPO_ROOT", linked_worktree), patch( + "registry_audit.DEFAULT_CONTROL_PLANE", + { + **registry_audit.DEFAULT_CONTROL_PLANE, + "subsystems_dir_path": str(contracts_dir), + }, + ): + files = tracked_workspace_files( + active_repos=["pulse"], + local_repo=canonical_repo_id(linked_worktree), + ) + + self.assertIn("internal/staged.go", files) + self.assertNotIn("pulse:internal/staged.go", files) + def test_audit_registry_payload_accepts_valid_minimal_registry(self) -> None: payload = { "version": 13, diff --git a/scripts/release_control/repo_file_io.py b/scripts/release_control/repo_file_io.py index 19b51a35b0..903ca0e97f 100644 --- a/scripts/release_control/repo_file_io.py +++ b/scripts/release_control/repo_file_io.py @@ -14,13 +14,45 @@ REPO_ROOT = Path(__file__).resolve().parents[2] DEFAULT_REPO_ROOT = REPO_ROOT -def git_env() -> dict[str, str]: +def git_env(repo_root: Path | None = None) -> dict[str, str]: env = os.environ.copy() - if REPO_ROOT != DEFAULT_REPO_ROOT: + root = repo_root or REPO_ROOT + if root != DEFAULT_REPO_ROOT: env.pop("GIT_INDEX_FILE", None) return env +def git_common_dir(repo_root: Path | None = None) -> Path: + root = (repo_root or REPO_ROOT).resolve() + result = subprocess.run( + ["git", "rev-parse", "--path-format=absolute", "--git-common-dir"], + cwd=root, + check=True, + capture_output=True, + text=True, + env=git_env(root), + ) + common_dir = Path(result.stdout.strip()) + if not common_dir.is_absolute(): + common_dir = root / common_dir + return common_dir.resolve() + + +def canonical_repo_root(repo_root: Path | None = None) -> Path: + common_dir = git_common_dir(repo_root) + if common_dir.name == ".git": + return common_dir.parent.resolve() + return (repo_root or REPO_ROOT).resolve() + + +def canonical_repo_id(repo_root: Path | None = None) -> str: + return canonical_repo_root(repo_root).name + + +def canonical_workspace_repos_root(repo_root: Path | None = None) -> Path: + return canonical_repo_root(repo_root).parent + + def repo_relative_path(path: str | Path) -> str: candidate = Path(path) if candidate.is_absolute(): diff --git a/scripts/release_control/repo_file_io_test.py b/scripts/release_control/repo_file_io_test.py index 80033b669a..14da8ab4e0 100644 --- a/scripts/release_control/repo_file_io_test.py +++ b/scripts/release_control/repo_file_io_test.py @@ -6,7 +6,14 @@ import unittest from pathlib import Path from unittest.mock import patch -from repo_file_io import load_repo_json, missing_staged_repo_paths, read_repo_text +from repo_file_io import ( + canonical_repo_id, + canonical_repo_root, + canonical_workspace_repos_root, + load_repo_json, + missing_staged_repo_paths, + read_repo_text, +) class RepoFileIoTest(unittest.TestCase): @@ -72,6 +79,33 @@ class RepoFileIoTest(unittest.TestCase): with patch("repo_file_io.REPO_ROOT", repo_root): self.assertEqual(missing_staged_repo_paths([staged_rel, missing_rel]), [missing_rel]) + def test_canonical_repo_identity_uses_git_common_dir_for_linked_worktree(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + workspace = Path(tmpdir) / "workspace" + repo_root = workspace / "repos" / "pulse" + linked_worktree = workspace / ".worktrees" / "pulse-first-session-onboarding-parity" + repo_root.mkdir(parents=True) + linked_worktree.parent.mkdir(parents=True) + + self.git(repo_root, "init") + (repo_root / "README.md").write_text("pulse\n", encoding="utf-8") + self.git(repo_root, "add", "README.md") + self.git( + repo_root, + "-c", + "user.name=Pulse Test", + "-c", + "user.email=pulse-test@example.invalid", + "commit", + "-m", + "initial", + ) + self.git(repo_root, "worktree", "add", "--detach", str(linked_worktree), "HEAD") + + self.assertEqual(canonical_repo_root(linked_worktree), repo_root.resolve()) + self.assertEqual(canonical_repo_id(linked_worktree), "pulse") + self.assertEqual(canonical_workspace_repos_root(linked_worktree), (workspace / "repos").resolve()) + if __name__ == "__main__": unittest.main()