fix(governance): isolate release-control worktree roots

This commit is contained in:
rcourtman
2026-08-08 05:12:07 +01:00
parent 790a7d8ce1
commit 8e8edbf7de
23 changed files with 297 additions and 68 deletions
@@ -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
+5 -9
View File
@@ -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
@@ -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
@@ -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:
@@ -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:
@@ -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:
@@ -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:
@@ -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:
@@ -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:
@@ -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:
@@ -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:
@@ -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:
@@ -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"))
@@ -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]]:
@@ -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"),
@@ -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
@@ -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):
+3 -3
View File
@@ -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(
+55 -5
View File
@@ -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:
+60 -3
View File
@@ -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
+2 -2
View File
@@ -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]:
+6 -6
View File
@@ -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()
@@ -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"], [])