From 8e8edbf7de53aed87d8dce6a8295d641b619737c Mon Sep 17 00:00:00 2001 From: rcourtman Date: Sat, 8 Aug 2026 05:12:07 +0100 Subject: [PATCH] fix(governance): isolate release-control worktree roots --- .../canonical_completion_guard_test.py | 15 +++-- scripts/release_control/contract_audit.py | 14 ++--- .../release_control/contract_audit_test.py | 8 ++- ...mercial_cancellation_reactivation_proof.py | 9 ++- ...al_cancellation_reactivation_proof_test.py | 4 +- .../mobile_relay_auth_approvals_proof.py | 11 +++- .../mobile_relay_auth_approvals_proof_test.py | 4 +- .../internal/paid_feature_claims_proof.py | 13 +++- .../paid_feature_claims_proof_test.py | 10 ++- ...elay_registration_reconnect_drain_proof.py | 9 ++- ...registration_reconnect_drain_proof_test.py | 2 +- .../release_control/internal/worktree_base.py | 14 ++++- .../internal/worktree_base_test.py | 17 +++-- .../internal/worktree_claim.py | 15 +++-- .../internal/worktree_claim_test.py | 16 +++-- .../internal/worktree_finish.py | 4 ++ .../mobile_compatibility_test.py | 3 +- scripts/release_control/registry_audit.py | 6 +- scripts/release_control/repo_file_io.py | 60 ++++++++++++++++-- scripts/release_control/repo_file_io_test.py | 63 ++++++++++++++++++- scripts/release_control/status_audit.py | 4 +- scripts/release_control/subsystem_lookup.py | 12 ++-- .../release_control/subsystem_lookup_test.py | 52 ++++++++++++++- 23 files changed, 297 insertions(+), 68 deletions(-) diff --git a/scripts/release_control/canonical_completion_guard_test.py b/scripts/release_control/canonical_completion_guard_test.py index e454d60a7..767d780dc 100644 --- a/scripts/release_control/canonical_completion_guard_test.py +++ b/scripts/release_control/canonical_completion_guard_test.py @@ -26,8 +26,9 @@ from canonical_completion_guard import ( stdin_files, subsystem_matches_path, ) +from repo_file_io import canonical_workspace_repos_root -WORKSPACE_REPOS_ROOT = REPO_ROOT.parent +WORKSPACE_REPOS_ROOT = canonical_workspace_repos_root(REPO_ROOT) PLATFORM_CONNECTIONS_WORKSPACE_EXACT_FILES = [ "frontend-modern/src/components/Settings/ConnectionEditor/__tests__/ConnectionEditor.test.tsx", @@ -103,11 +104,15 @@ def owned_runtime_files(rule: dict) -> list[str]: search_roots.add(parent) for root in search_roots: - repo_root = root - while repo_root != WORKSPACE_REPOS_ROOT and repo_root.parent != WORKSPACE_REPOS_ROOT: - repo_root = repo_root.parent - if repo_root.parent != WORKSPACE_REPOS_ROOT: + try: + root.relative_to(REPO_ROOT) repo_root = REPO_ROOT + except ValueError: + repo_root = root + while repo_root.parent != repo_root and repo_root.parent != WORKSPACE_REPOS_ROOT: + repo_root = repo_root.parent + if repo_root.parent != WORKSPACE_REPOS_ROOT: + repo_root = REPO_ROOT for path in root.rglob("*"): if not path.is_file(): continue diff --git a/scripts/release_control/contract_audit.py b/scripts/release_control/contract_audit.py index 0fd6423b8..72fb9ac5c 100644 --- a/scripts/release_control/contract_audit.py +++ b/scripts/release_control/contract_audit.py @@ -14,9 +14,8 @@ from canonical_completion_guard import REPO_ROOT, subsystem_matches_path from control_plane import DEFAULT_CONTROL_PLANE from repo_file_io import ( canonical_repo_id, - canonical_repo_root, - canonical_workspace_repos_root, load_repo_json, + repo_root_context, ) from subsystem_contracts import tracked_contract_files @@ -115,20 +114,17 @@ def looks_like_repo_path(token: str) -> bool: def repo_roots_for_status(status_payload: dict[str, Any], *, repo_root: Path | None = None) -> dict[str, Path]: - root = repo_root or REPO_ROOT - local_repo_id = canonical_repo_id(root) - local_repo_root = canonical_repo_root(root) - workspace_repos_root = canonical_workspace_repos_root(root) + context = repo_root_context(repo_root or REPO_ROOT) active_repos = [ repo_id for repo_id in status_payload.get("scope", {}).get("active_repos", []) if isinstance(repo_id, str) and repo_id.strip() ] - repo_roots = {local_repo_id: local_repo_root} + repo_roots = {context.repo_id: context.execution_root} for repo_id in active_repos: - if repo_id == local_repo_id: + if repo_id == context.repo_id: continue - repo_roots[repo_id] = workspace_repos_root / repo_id + repo_roots[repo_id] = context.workspace_repos_root / repo_id return repo_roots diff --git a/scripts/release_control/contract_audit_test.py b/scripts/release_control/contract_audit_test.py index 5efa767eb..f7ee2090e 100644 --- a/scripts/release_control/contract_audit_test.py +++ b/scripts/release_control/contract_audit_test.py @@ -383,7 +383,7 @@ Cross-repo relay ownership is explicit. self.assertEqual(report["errors"], []) - def test_audit_contract_payload_resolves_sibling_repos_from_linked_worktree(self) -> None: + def test_audit_contract_payload_isolates_local_paths_and_resolves_siblings_from_linked_worktree(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: workspace = Path(tmpdir) / "workspace" repo_root = workspace / "repos" / "pulse" @@ -408,6 +408,7 @@ Cross-repo relay ownership is explicit. "initial", ) self.git(repo_root, "worktree", "add", "--detach", str(linked_worktree), "HEAD") + (repo_root / "README.md").unlink() status_payload = { "scope": { @@ -416,7 +417,7 @@ Cross-repo relay ownership is explicit. "lanes": [{"id": "L7"}], } repo_roots = repo_roots_for_status(status_payload, repo_root=linked_worktree) - self.assertEqual(repo_roots["pulse"], repo_root.resolve()) + self.assertEqual(repo_roots["pulse"], linked_worktree.resolve()) self.assertEqual(repo_roots["pulse-mobile"], sibling_root.resolve()) registry_payload = { @@ -450,7 +451,8 @@ Own relay runtime truth. ## Canonical Files -1. `pulse-mobile:src/relay/client.ts` +1. `README.md` +2. `pulse-mobile:src/relay/client.ts` ## Shared Boundaries diff --git a/scripts/release_control/internal/commercial_cancellation_reactivation_proof.py b/scripts/release_control/internal/commercial_cancellation_reactivation_proof.py index 2defb7d06..a8b350b56 100644 --- a/scripts/release_control/internal/commercial_cancellation_reactivation_proof.py +++ b/scripts/release_control/internal/commercial_cancellation_reactivation_proof.py @@ -8,6 +8,13 @@ import json import subprocess from dataclasses import asdict, dataclass from pathlib import Path +import sys + +RELEASE_CONTROL_DIR = Path(__file__).resolve().parents[1] +if str(RELEASE_CONTROL_DIR) not in sys.path: + sys.path.append(str(RELEASE_CONTROL_DIR)) + +from repo_file_io import canonical_workspace_repos_root @dataclass @@ -70,7 +77,7 @@ def default_pulse_dir() -> Path: def default_pulse_pro_license_server_dir() -> Path: - return default_pulse_dir().parent / "pulse-pro" / "license-server" + return canonical_workspace_repos_root(default_pulse_dir()) / "pulse-pro" / "license-server" def frontend_dir_from_args(args: argparse.Namespace) -> Path: diff --git a/scripts/release_control/internal/commercial_cancellation_reactivation_proof_test.py b/scripts/release_control/internal/commercial_cancellation_reactivation_proof_test.py index 074dfd7fc..24f606ef0 100644 --- a/scripts/release_control/internal/commercial_cancellation_reactivation_proof_test.py +++ b/scripts/release_control/internal/commercial_cancellation_reactivation_proof_test.py @@ -28,7 +28,9 @@ class CommercialCancellationReactivationProofTest(unittest.TestCase): self.assertEqual(proof.default_pulse_dir(), Path(proof.__file__).resolve().parents[3]) self.assertEqual( proof.default_pulse_pro_license_server_dir(), - proof.default_pulse_dir().parent / "pulse-pro" / "license-server", + proof.canonical_workspace_repos_root(proof.default_pulse_dir()) + / "pulse-pro" + / "license-server", ) def test_build_command_specs_uses_expected_directories(self) -> None: diff --git a/scripts/release_control/internal/mobile_relay_auth_approvals_proof.py b/scripts/release_control/internal/mobile_relay_auth_approvals_proof.py index dc6f22887..3c5b24976 100644 --- a/scripts/release_control/internal/mobile_relay_auth_approvals_proof.py +++ b/scripts/release_control/internal/mobile_relay_auth_approvals_proof.py @@ -8,6 +8,13 @@ import json import subprocess from dataclasses import asdict, dataclass from pathlib import Path +import sys + +RELEASE_CONTROL_DIR = Path(__file__).resolve().parents[1] +if str(RELEASE_CONTROL_DIR) not in sys.path: + sys.path.append(str(RELEASE_CONTROL_DIR)) + +from repo_file_io import canonical_workspace_repos_root @dataclass @@ -32,11 +39,11 @@ def default_pulse_dir() -> Path: def default_pulse_mobile_dir() -> Path: - return default_pulse_dir().parent / "pulse-mobile" + return canonical_workspace_repos_root(default_pulse_dir()) / "pulse-mobile" def default_pulse_enterprise_dir() -> Path: - return default_pulse_dir().parent / "pulse-enterprise" + return canonical_workspace_repos_root(default_pulse_dir()) / "pulse-enterprise" def parse_args(argv: list[str] | None = None) -> argparse.Namespace: diff --git a/scripts/release_control/internal/mobile_relay_auth_approvals_proof_test.py b/scripts/release_control/internal/mobile_relay_auth_approvals_proof_test.py index 2fc1101b8..428712178 100644 --- a/scripts/release_control/internal/mobile_relay_auth_approvals_proof_test.py +++ b/scripts/release_control/internal/mobile_relay_auth_approvals_proof_test.py @@ -20,11 +20,11 @@ class MobileRelayAuthApprovalsProofTest(unittest.TestCase): self.assertEqual(proof.default_pulse_dir(), Path(proof.__file__).resolve().parents[3]) self.assertEqual( proof.default_pulse_mobile_dir(), - proof.default_pulse_dir().parent / "pulse-mobile", + proof.canonical_workspace_repos_root(proof.default_pulse_dir()) / "pulse-mobile", ) self.assertEqual( proof.default_pulse_enterprise_dir(), - proof.default_pulse_dir().parent / "pulse-enterprise", + proof.canonical_workspace_repos_root(proof.default_pulse_dir()) / "pulse-enterprise", ) def test_build_command_specs_are_sorted_and_cross_repo(self) -> None: diff --git a/scripts/release_control/internal/paid_feature_claims_proof.py b/scripts/release_control/internal/paid_feature_claims_proof.py index e1fb4a93b..f29ead608 100644 --- a/scripts/release_control/internal/paid_feature_claims_proof.py +++ b/scripts/release_control/internal/paid_feature_claims_proof.py @@ -9,6 +9,13 @@ import re import subprocess from dataclasses import asdict, dataclass from pathlib import Path +import sys + +RELEASE_CONTROL_DIR = Path(__file__).resolve().parents[1] +if str(RELEASE_CONTROL_DIR) not in sys.path: + sys.path.append(str(RELEASE_CONTROL_DIR)) + +from repo_file_io import canonical_workspace_repos_root @dataclass @@ -157,15 +164,15 @@ def default_pulse_dir() -> Path: def default_pulse_pro_license_server_dir() -> Path: - return default_pulse_dir().parent / "pulse-pro" / "license-server" + return canonical_workspace_repos_root(default_pulse_dir()) / "pulse-pro" / "license-server" def default_pulse_pro_relay_server_dir() -> Path: - return default_pulse_dir().parent / "pulse-pro" / "relay-server" + return canonical_workspace_repos_root(default_pulse_dir()) / "pulse-pro" / "relay-server" def default_pulse_enterprise_dir() -> Path: - return default_pulse_dir().parent / "pulse-enterprise" + return canonical_workspace_repos_root(default_pulse_dir()) / "pulse-enterprise" def default_pulse_pro_dir(args: argparse.Namespace) -> Path: diff --git a/scripts/release_control/internal/paid_feature_claims_proof_test.py b/scripts/release_control/internal/paid_feature_claims_proof_test.py index bc966ee7c..58adb7c27 100644 --- a/scripts/release_control/internal/paid_feature_claims_proof_test.py +++ b/scripts/release_control/internal/paid_feature_claims_proof_test.py @@ -126,15 +126,19 @@ class PaidFeatureClaimsProofTest(unittest.TestCase): self.assertEqual(proof.default_pulse_dir(), Path(proof.__file__).resolve().parents[3]) self.assertEqual( proof.default_pulse_pro_license_server_dir(), - proof.default_pulse_dir().parent / "pulse-pro" / "license-server", + proof.canonical_workspace_repos_root(proof.default_pulse_dir()) + / "pulse-pro" + / "license-server", ) self.assertEqual( proof.default_pulse_pro_relay_server_dir(), - proof.default_pulse_dir().parent / "pulse-pro" / "relay-server", + proof.canonical_workspace_repos_root(proof.default_pulse_dir()) + / "pulse-pro" + / "relay-server", ) self.assertEqual( proof.default_pulse_enterprise_dir(), - proof.default_pulse_dir().parent / "pulse-enterprise", + proof.canonical_workspace_repos_root(proof.default_pulse_dir()) / "pulse-enterprise", ) def test_build_command_specs_cover_paid_claim_layers(self) -> None: diff --git a/scripts/release_control/internal/relay_registration_reconnect_drain_proof.py b/scripts/release_control/internal/relay_registration_reconnect_drain_proof.py index 47b373c19..ff4293722 100644 --- a/scripts/release_control/internal/relay_registration_reconnect_drain_proof.py +++ b/scripts/release_control/internal/relay_registration_reconnect_drain_proof.py @@ -8,6 +8,13 @@ import json import subprocess from dataclasses import asdict, dataclass from pathlib import Path +import sys + +RELEASE_CONTROL_DIR = Path(__file__).resolve().parents[1] +if str(RELEASE_CONTROL_DIR) not in sys.path: + sys.path.append(str(RELEASE_CONTROL_DIR)) + +from repo_file_io import canonical_workspace_repos_root @dataclass @@ -32,7 +39,7 @@ def default_pulse_dir() -> Path: def default_pulse_mobile_dir() -> Path: - return default_pulse_dir().parent / "pulse-mobile" + return canonical_workspace_repos_root(default_pulse_dir()) / "pulse-mobile" def parse_args(argv: list[str] | None = None) -> argparse.Namespace: diff --git a/scripts/release_control/internal/relay_registration_reconnect_drain_proof_test.py b/scripts/release_control/internal/relay_registration_reconnect_drain_proof_test.py index f4539d4e2..59e597771 100644 --- a/scripts/release_control/internal/relay_registration_reconnect_drain_proof_test.py +++ b/scripts/release_control/internal/relay_registration_reconnect_drain_proof_test.py @@ -20,7 +20,7 @@ class RelayRegistrationReconnectDrainProofTest(unittest.TestCase): self.assertEqual(proof.default_pulse_dir(), Path(proof.__file__).resolve().parents[3]) self.assertEqual( proof.default_pulse_mobile_dir(), - proof.default_pulse_dir().parent / "pulse-mobile", + proof.canonical_workspace_repos_root(proof.default_pulse_dir()) / "pulse-mobile", ) def test_build_command_specs_are_sorted_and_cover_expected_workspaces(self) -> None: diff --git a/scripts/release_control/internal/worktree_base.py b/scripts/release_control/internal/worktree_base.py index 8184880e1..c3ebe923a 100644 --- a/scripts/release_control/internal/worktree_base.py +++ b/scripts/release_control/internal/worktree_base.py @@ -9,8 +9,12 @@ from pathlib import Path import subprocess import sys -from repo_file_io import REPO_ROOT -from worktree_claim import WORKTREES_ROOT, list_worktrees +RELEASE_CONTROL_DIR = Path(__file__).resolve().parents[1] +if str(RELEASE_CONTROL_DIR) not in sys.path: + sys.path.append(str(RELEASE_CONTROL_DIR)) + +from repo_file_io import REPO_ROOT, repo_root_context +from worktree_claim import canonical_worktrees_root, list_worktrees def git(*args: str, cwd: Path, check: bool = True) -> subprocess.CompletedProcess[str]: @@ -28,7 +32,11 @@ def base_slug(branch_name: str) -> str: def canonical_base_worktree_path(*, repo_root: Path, branch_name: str) -> Path: - return WORKTREES_ROOT / repo_root.name / base_slug(branch_name) + return ( + canonical_worktrees_root(repo_root=repo_root) + / repo_root_context(repo_root).repo_id + / base_slug(branch_name) + ) def find_worktree_by_path(*, repo_root: Path, path: Path) -> dict[str, str] | None: diff --git a/scripts/release_control/internal/worktree_base_test.py b/scripts/release_control/internal/worktree_base_test.py index 66cbabf43..e5d784a4c 100644 --- a/scripts/release_control/internal/worktree_base_test.py +++ b/scripts/release_control/internal/worktree_base_test.py @@ -1,6 +1,8 @@ import tempfile import unittest from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch from worktree_base import base_slug, canonical_base_worktree_path, parse_args @@ -16,10 +18,17 @@ class WorktreeBaseTest(unittest.TestCase): self.assertEqual(base_slug("pulse/v6"), "base__pulse__v6") def test_canonical_base_worktree_path_uses_workspace_root(self) -> None: - path = canonical_base_worktree_path( - repo_root=Path("/Volumes/Development/pulse/repos/pulse"), - branch_name="pulse/v6", - ) + with patch( + "worktree_base.canonical_worktrees_root", + return_value=Path("/Volumes/Development/pulse/worktrees"), + ), patch( + "worktree_base.repo_root_context", + return_value=SimpleNamespace(repo_id="pulse"), + ): + path = canonical_base_worktree_path( + repo_root=Path("/isolated/worktrees/l1/pulse"), + branch_name="pulse/v6", + ) self.assertEqual(path, Path("/Volumes/Development/pulse/worktrees/pulse/base__pulse__v6")) diff --git a/scripts/release_control/internal/worktree_claim.py b/scripts/release_control/internal/worktree_claim.py index a526ecf6e..466423fff 100644 --- a/scripts/release_control/internal/worktree_claim.py +++ b/scripts/release_control/internal/worktree_claim.py @@ -10,12 +10,14 @@ import subprocess import sys from typing import Any -from repo_file_io import REPO_ROOT +RELEASE_CONTROL_DIR = Path(__file__).resolve().parents[1] +if str(RELEASE_CONTROL_DIR) not in sys.path: + sys.path.append(str(RELEASE_CONTROL_DIR)) + +from repo_file_io import REPO_ROOT, repo_root_context from work_claim import reserve_claim, write_status_payload, _slug -WORKSPACE_ROOT = REPO_ROOT.parents[1] -WORKTREES_ROOT = WORKSPACE_ROOT / "worktrees" DEFAULT_BASE_BRANCH = "pulse/v6" @@ -29,8 +31,13 @@ def branch_path_slug(branch_name: str) -> str: return branch_name.replace("/", "__") +def canonical_worktrees_root(*, repo_root: Path) -> Path: + return repo_root_context(repo_root).workspace_repos_root.parent / "worktrees" + + def build_worktree_path(*, repo_root: Path, branch_name: str) -> Path: - return WORKTREES_ROOT / repo_root.name / branch_path_slug(branch_name) + context = repo_root_context(repo_root) + return canonical_worktrees_root(repo_root=repo_root) / context.repo_id / branch_path_slug(branch_name) def parse_worktree_list(output: str) -> list[dict[str, str]]: diff --git a/scripts/release_control/internal/worktree_claim_test.py b/scripts/release_control/internal/worktree_claim_test.py index fe161a480..ae15025f3 100644 --- a/scripts/release_control/internal/worktree_claim_test.py +++ b/scripts/release_control/internal/worktree_claim_test.py @@ -1,6 +1,7 @@ import tempfile import unittest from pathlib import Path +from types import SimpleNamespace from unittest.mock import patch from worktree_claim import ( @@ -47,10 +48,17 @@ class WorktreeClaimTest(unittest.TestCase): ) def test_build_worktree_path_uses_workspace_root(self) -> None: - path = build_worktree_path( - repo_root=Path("/Volumes/Development/pulse/repos/pulse"), - branch_name="pulse/claude-code/lane-l15", - ) + with patch( + "worktree_claim.repo_root_context", + return_value=SimpleNamespace( + repo_id="pulse", + workspace_repos_root=Path("/Volumes/Development/pulse/repos"), + ), + ): + path = build_worktree_path( + repo_root=Path("/isolated/worktrees/l15/pulse"), + branch_name="pulse/claude-code/lane-l15", + ) self.assertEqual( path, Path("/Volumes/Development/pulse/worktrees/pulse/claude-code__lane-l15"), diff --git a/scripts/release_control/internal/worktree_finish.py b/scripts/release_control/internal/worktree_finish.py index 3d5898c33..9b36f78cd 100644 --- a/scripts/release_control/internal/worktree_finish.py +++ b/scripts/release_control/internal/worktree_finish.py @@ -10,6 +10,10 @@ import subprocess import sys from typing import Any +RELEASE_CONTROL_DIR = Path(__file__).resolve().parents[1] +if str(RELEASE_CONTROL_DIR) not in sys.path: + sys.path.append(str(RELEASE_CONTROL_DIR)) + from repo_file_io import REPO_ROOT from worktree_base import canonical_base_worktree_path from worktree_claim import list_worktrees diff --git a/scripts/release_control/mobile_compatibility_test.py b/scripts/release_control/mobile_compatibility_test.py index cc9d133a7..03498ed1e 100644 --- a/scripts/release_control/mobile_compatibility_test.py +++ b/scripts/release_control/mobile_compatibility_test.py @@ -8,9 +8,10 @@ import unittest from generate_mobile_compatibility import REPO_ROOT, load_manifest from mobile_compatibility import compare_contracts, load_consumer +from repo_file_io import canonical_workspace_repos_root -MOBILE_REPO = REPO_ROOT.parent / "pulse-mobile" +MOBILE_REPO = canonical_workspace_repos_root(REPO_ROOT) / "pulse-mobile" class MobileCompatibilityTest(unittest.TestCase): diff --git a/scripts/release_control/registry_audit.py b/scripts/release_control/registry_audit.py index 3781ec22f..925755e4a 100644 --- a/scripts/release_control/registry_audit.py +++ b/scripts/release_control/registry_audit.py @@ -19,7 +19,7 @@ from canonical_completion_guard import ( subsystem_matches_path, ) from control_plane import DEFAULT_CONTROL_PLANE -from repo_file_io import canonical_repo_id, canonical_workspace_repos_root, git_env, load_repo_json +from repo_file_io import canonical_repo_id, git_env, load_repo_json, repo_root_context from status_audit import load_status_payload @@ -67,9 +67,9 @@ def tracked_repo_files() -> set[str]: def tracked_workspace_files(*, active_repos: list[str], local_repo: str) -> set[str]: files: set[str] = set() - repos_root = canonical_workspace_repos_root(REPO_ROOT) + context = repo_root_context(REPO_ROOT) for repo_id in active_repos: - repo_root = REPO_ROOT if repo_id == local_repo else repos_root / repo_id + repo_root = context.execution_root if repo_id == local_repo else context.workspace_repos_root / repo_id if not repo_root.exists(): continue result = subprocess.run( diff --git a/scripts/release_control/repo_file_io.py b/scripts/release_control/repo_file_io.py index 307cedd6f..a7a58a740 100644 --- a/scripts/release_control/repo_file_io.py +++ b/scripts/release_control/repo_file_io.py @@ -5,6 +5,8 @@ from __future__ import annotations import json import os +from dataclasses import dataclass +from functools import lru_cache from pathlib import Path import subprocess from typing import Any, Iterable @@ -45,6 +47,16 @@ REPO_TARGET_GIT_ENV_VARS = ( ) +@dataclass(frozen=True) +class RepoRootContext: + """Keep local worktree I/O separate from shared repository identity.""" + + execution_root: Path + shared_repo_root: Path + repo_id: str + workspace_repos_root: Path + + def strip_local_git_env(env: dict[str, str]) -> dict[str, str]: for name in LOCAL_GIT_ENV_VARS: env.pop(name, None) @@ -162,19 +174,57 @@ def git_common_dir(repo_root: Path | None = None) -> Path: return common_dir.resolve() -def canonical_repo_root(repo_root: Path | None = None) -> Path: - common_dir = git_common_dir(repo_root) +def execution_repo_root(repo_root: Path | None = None) -> Path: + """Return the worktree whose files and index the current operation owns.""" + + root = (repo_root or REPO_ROOT).resolve() + result = subprocess.run( + ["git", "rev-parse", "--path-format=absolute", "--show-toplevel"], + cwd=root, + check=True, + capture_output=True, + text=True, + env=git_env(root), + ) + return Path(result.stdout.strip()).resolve() + + +def shared_repo_root(repo_root: Path | None = None) -> Path: + """Return the primary checkout used only for shared Git/workspace identity.""" + + worktree_root = execution_repo_root(repo_root) + return _shared_repo_root_for_execution(worktree_root) + + +def _shared_repo_root_for_execution(worktree_root: Path) -> Path: + common_dir = git_common_dir(worktree_root) if common_dir.name == ".git": return common_dir.parent.resolve() - return (repo_root or REPO_ROOT).resolve() + return worktree_root + + +@lru_cache(maxsize=None) +def _repo_root_context(root: Path) -> RepoRootContext: + execution_root = execution_repo_root(root) + identity_root = _shared_repo_root_for_execution(execution_root) + return RepoRootContext( + execution_root=execution_root, + shared_repo_root=identity_root, + repo_id=identity_root.name, + workspace_repos_root=identity_root.parent, + ) + + +def repo_root_context(repo_root: Path | None = None) -> RepoRootContext: + return _repo_root_context((repo_root or REPO_ROOT).resolve()) def canonical_repo_id(repo_root: Path | None = None) -> str: - return canonical_repo_root(repo_root).name + return repo_root_context(repo_root).repo_id def canonical_workspace_repos_root(repo_root: Path | None = None) -> Path: - return canonical_repo_root(repo_root).parent + return repo_root_context(repo_root).workspace_repos_root def repo_relative_path(path: str | Path) -> str: diff --git a/scripts/release_control/repo_file_io_test.py b/scripts/release_control/repo_file_io_test.py index abf102756..4ce8805b5 100644 --- a/scripts/release_control/repo_file_io_test.py +++ b/scripts/release_control/repo_file_io_test.py @@ -9,12 +9,14 @@ from unittest.mock import patch from repo_file_io import ( canonical_repo_id, - canonical_repo_root, canonical_workspace_repos_root, + execution_repo_root, git_env, load_repo_json, missing_staged_repo_paths, read_repo_text, + repo_root_context, + shared_repo_root, strip_local_git_env, ) @@ -148,6 +150,48 @@ 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_linked_worktree_reads_ignore_conflicting_primary_checkout_state(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + workspace = Path(tmpdir) / "workspace" + repo_root = workspace / "repos" / "pulse" + linked_worktree = workspace / ".worktrees" / "pulse-release-control-isolation" + rel = "docs/release-control/v6/internal/status.json" + primary_path = repo_root / rel + repo_root.mkdir(parents=True) + primary_path.parent.mkdir(parents=True) + + self.git(repo_root, "init") + primary_path.write_text('{"version": "head"}\n', encoding="utf-8") + self.git(repo_root, "add", rel) + 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") + + primary_path.unlink() + linked_path = linked_worktree / rel + linked_path.write_text('{"version": "linked-index"}\n', encoding="utf-8") + self.git(linked_worktree, "add", rel) + linked_path.write_text('{"version": "linked-worktree"}\n', encoding="utf-8") + + with ( + patch("repo_file_io.REPO_ROOT", linked_worktree), + patch("repo_file_io.DEFAULT_REPO_ROOT", linked_worktree), + patch.dict(os.environ, self.hook_env_for_worktree(linked_worktree), clear=False), + ): + self.assertEqual(read_repo_text(rel), '{"version": "linked-worktree"}\n') + self.assertEqual(read_repo_text(rel, staged=True), '{"version": "linked-index"}\n') + self.assertEqual(load_repo_json(rel, staged=True), {"version": "linked-index"}) + + self.assertFalse(primary_path.exists()) + def test_git_env_preserves_local_hook_env_and_scrubs_other_repos(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: workspace = Path(tmpdir) / "workspace" @@ -223,7 +267,7 @@ class RepoFileIoTest(unittest.TestCase): self.assertNotIn("GIT_INDEX_FILE", env) self.assertNotIn("GIT_COMMON_DIR", env) - def test_canonical_repo_identity_uses_git_common_dir_for_linked_worktree(self) -> None: + def test_repo_root_context_separates_linked_worktree_from_shared_identity(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: workspace = Path(tmpdir) / "workspace" repo_root = workspace / "repos" / "pulse" @@ -246,10 +290,23 @@ class RepoFileIoTest(unittest.TestCase): ) self.git(repo_root, "worktree", "add", "--detach", str(linked_worktree), "HEAD") - self.assertEqual(canonical_repo_root(linked_worktree), repo_root.resolve()) + context = repo_root_context(linked_worktree) + + self.assertEqual(execution_repo_root(linked_worktree), linked_worktree.resolve()) + self.assertEqual(shared_repo_root(linked_worktree), repo_root.resolve()) + self.assertEqual(context.execution_root, linked_worktree.resolve()) + self.assertEqual(context.shared_repo_root, repo_root.resolve()) + self.assertEqual(context.repo_id, "pulse") + self.assertEqual(context.workspace_repos_root, (workspace / "repos").resolve()) self.assertEqual(canonical_repo_id(linked_worktree), "pulse") self.assertEqual(canonical_workspace_repos_root(linked_worktree), (workspace / "repos").resolve()) + primary_context = repo_root_context(repo_root) + self.assertEqual(primary_context.execution_root, repo_root.resolve()) + self.assertEqual(primary_context.shared_repo_root, repo_root.resolve()) + self.assertEqual(primary_context.repo_id, "pulse") + self.assertEqual(primary_context.workspace_repos_root, (workspace / "repos").resolve()) + def test_scratch_git_init_tests_scrub_env_through_shared_helper(self) -> None: # Running "git init" in a scratch directory while the pre-commit hook # environment from a linked worktree (absolute GIT_DIR et al.) is still diff --git a/scripts/release_control/status_audit.py b/scripts/release_control/status_audit.py index e166696fe..4d30e1019 100644 --- a/scripts/release_control/status_audit.py +++ b/scripts/release_control/status_audit.py @@ -18,7 +18,7 @@ from typing import Any from canonical_completion_guard import load_subsystem_rules from control_plane import DEFAULT_CONTROL_PLANE, active_target_blocking_levels, is_prerelease_version -from repo_file_io import canonical_workspace_repos_root, load_repo_json, read_repo_text +from repo_file_io import load_repo_json, read_repo_text, repo_root_context REPO_ROOT = Path(__file__).resolve().parents[2] @@ -298,7 +298,7 @@ def repo_root_for_name(repo_name: str) -> Path: return Path(raw).expanduser().resolve() if repo_name == "pulse": return REPO_ROOT - return (canonical_workspace_repos_root(REPO_ROOT) / repo_name).resolve() + return (repo_root_context(REPO_ROOT).workspace_repos_root / repo_name).resolve() def load_status_payload(*, staged: bool = False) -> dict[str, Any]: diff --git a/scripts/release_control/subsystem_lookup.py b/scripts/release_control/subsystem_lookup.py index 7ca9dfe31..0b42a670d 100644 --- a/scripts/release_control/subsystem_lookup.py +++ b/scripts/release_control/subsystem_lookup.py @@ -23,33 +23,33 @@ from canonical_completion_guard import ( from subsystem_contracts import load_contract_index, referenced_contracts_for_path from status_audit import audit_status_payload, load_status_payload from registry_audit import load_registry_payload - -WORKSPACE_REPOS_ROOT = REPO_ROOT.parent +from repo_file_io import repo_root_context def normalize_input_path(raw: str) -> str: + context = repo_root_context(REPO_ROOT) candidate = Path(raw.strip()) parts = candidate.parts if len(parts) >= 3 and parts[0] == "repos": repo_id = parts[1] rel = Path(*parts[2:]).as_posix() - if repo_id == REPO_ROOT.name: + if repo_id == context.repo_id: return rel return f"{repo_id}:{rel}" if candidate.is_absolute(): candidate = candidate.resolve() try: - candidate = candidate.relative_to(REPO_ROOT) + candidate = candidate.relative_to(context.execution_root) except ValueError: try: - repo_relative = candidate.relative_to(WORKSPACE_REPOS_ROOT) + repo_relative = candidate.relative_to(context.workspace_repos_root) except ValueError: return candidate.as_posix() parts = repo_relative.parts if len(parts) >= 2: repo_id = parts[0] rel = Path(*parts[1:]).as_posix() - if repo_id == REPO_ROOT.name: + if repo_id == context.repo_id: return rel return f"{repo_id}:{rel}" return candidate.as_posix() diff --git a/scripts/release_control/subsystem_lookup_test.py b/scripts/release_control/subsystem_lookup_test.py index 3f02a9edf..87db454f5 100644 --- a/scripts/release_control/subsystem_lookup_test.py +++ b/scripts/release_control/subsystem_lookup_test.py @@ -1,8 +1,13 @@ +import os +import subprocess +import tempfile import unittest from pathlib import Path +from unittest.mock import patch from canonical_completion_guard import REPO_ROOT -from subsystem_lookup import lookup_paths, parse_args, render_pretty +from repo_file_io import canonical_workspace_repos_root, strip_local_git_env +from subsystem_lookup import lookup_paths, normalize_input_path, parse_args, render_pretty RECOVERY_PRODUCT_SURFACE_EXACT_FILES = [ @@ -75,6 +80,16 @@ def _contract_reference(contract_path: str, needle: str, runtime_path: str) -> d class SubsystemLookupTest(unittest.TestCase): + def git(self, repo_root: Path, *args: str) -> None: + subprocess.run( + ["git", *args], + cwd=repo_root, + check=True, + capture_output=True, + text=True, + env=strip_local_git_env(os.environ.copy()), + ) + def test_parse_args_accepts_lean_flag(self) -> None: args = parse_args(["internal/api/ai_handler.go", "--pretty", "--lean"]) self.assertEqual(args.paths, ["internal/api/ai_handler.go"]) @@ -139,7 +154,9 @@ class SubsystemLookupTest(unittest.TestCase): ) def test_lookup_paths_normalizes_cross_repo_absolute_runtime_paths(self) -> None: - result = lookup_paths([str(REPO_ROOT.parent / "pulse-mobile" / "src/relay/client.ts")]) + result = lookup_paths( + [str(canonical_workspace_repos_root(REPO_ROOT) / "pulse-mobile" / "src/relay/client.ts")] + ) self.assertEqual(result["unowned_runtime_files"], []) self.assertEqual( {item["subsystem"] for item in result["impacted_subsystems"]}, @@ -156,6 +173,37 @@ class SubsystemLookupTest(unittest.TestCase): self.assertEqual(match["lane_context"]["lane_id"], "L7") self.assertEqual(match["verification_requirement"]["id"], "mobile-relay-runtime") + def test_normalize_input_path_resolves_canonical_sibling_from_linked_worktree(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + workspace = Path(tmpdir) / "workspace" + repo_root = workspace / "repos" / "pulse" + sibling_path = workspace / "repos" / "pulse-mobile" / "src" / "relay" / "client.ts" + linked_worktree = workspace / ".worktrees" / "pulse-subsystem-lookup" + repo_root.mkdir(parents=True) + sibling_path.parent.mkdir(parents=True) + sibling_path.write_text("export const relayClient = true;\n", encoding="utf-8") + + 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") + + with patch("subsystem_lookup.REPO_ROOT", linked_worktree): + self.assertEqual( + normalize_input_path(str(sibling_path)), + "pulse-mobile:src/relay/client.ts", + ) + def test_lookup_paths_normalizes_workspace_relative_cross_repo_runtime_paths(self) -> None: result = lookup_paths(["repos/pulse-pro/scripts/grandfathered_recurring_cutover_preview.py"]) self.assertEqual(result["unowned_runtime_files"], [])