sysadminandClaude Opus 4.8
be592d6d4d
Introduce root_checkout_guard helper and enforce it in verify_preflight_purity so author, reviewer, and merger mutations fail closed when the stable control checkout is on the wrong branch, detached, dirty, or diverged from prgs/master. Isolated branches/... worktrees and reconciler paths remain allowed. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
133 lines
4.3 KiB
Python
133 lines
4.3 KiB
Python
"""Root checkout guard (#475).
|
|
|
|
The project root checkout is the stable control checkout on master/prgs/master.
|
|
Author/reviewer/merge flows must fail closed when the control checkout is
|
|
contaminated (wrong branch, detached HEAD, dirty, or HEAD behind/ahead of
|
|
prgs/master). Isolated ``branches/...`` worktrees remain allowed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
|
|
from author_mutation_worktree import is_path_under_branches
|
|
from reviewer_worktree import parse_dirty_tracked_files
|
|
|
|
REMEDIATION = (
|
|
"Root checkout is not on master. Preserve state, switch root back to master, "
|
|
"and use scripts/worktree-review or the sanctioned issue worktree flow."
|
|
)
|
|
|
|
BASE_BRANCHES = frozenset({"master", "main", "dev"})
|
|
REMOTE_MASTER_REFS = ("prgs/master", "refs/remotes/prgs/master")
|
|
|
|
|
|
def resolve_remote_master_sha(
|
|
canonical_repo_root: str,
|
|
*,
|
|
remote_refs: tuple[str, ...] | None = None,
|
|
) -> str | None:
|
|
"""Return the commit SHA for the tracking master ref when available."""
|
|
root = (canonical_repo_root or "").strip()
|
|
if not root:
|
|
return None
|
|
for ref in remote_refs or REMOTE_MASTER_REFS:
|
|
res = subprocess.run(
|
|
["git", "-C", root, "rev-parse", "--verify", ref],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if res.returncode == 0:
|
|
sha = (res.stdout or "").strip()
|
|
if sha:
|
|
return sha
|
|
return None
|
|
|
|
|
|
resolve_tracking_master_sha = resolve_remote_master_sha
|
|
|
|
|
|
def assess_root_checkout_guard(
|
|
*,
|
|
workspace_path: str,
|
|
canonical_repo_root: str,
|
|
current_branch: str | None,
|
|
head_sha: str | None,
|
|
porcelain_status: str,
|
|
remote_master_sha: str | None,
|
|
resolved_role: str | None = None,
|
|
) -> dict:
|
|
"""Fail closed when the control checkout is not clean master/prgs/master."""
|
|
reasons: list[str] = []
|
|
root = os.path.realpath(canonical_repo_root)
|
|
workspace = os.path.realpath(workspace_path)
|
|
branch = (current_branch or "").strip()
|
|
dirty_files = parse_dirty_tracked_files(porcelain_status)
|
|
|
|
if resolved_role == "reconciler":
|
|
return _assessment(True, [], root, workspace, branch, head_sha, dirty_files)
|
|
|
|
if resolved_role != "merger" and is_path_under_branches(workspace, root):
|
|
return _assessment(True, [], root, workspace, branch, head_sha, dirty_files)
|
|
|
|
if dirty_files:
|
|
reasons.append(
|
|
"control checkout has tracked local edits before role work "
|
|
f"(dirty files: {', '.join(dirty_files)})"
|
|
)
|
|
|
|
if not branch:
|
|
reasons.append("control checkout is detached HEAD; expected branch 'master'")
|
|
elif branch not in BASE_BRANCHES:
|
|
reasons.append(
|
|
f"control checkout branch '{branch}' is not a stable base branch "
|
|
f"({'/'.join(sorted(BASE_BRANCHES))})"
|
|
)
|
|
|
|
if remote_master_sha and head_sha and head_sha != remote_master_sha:
|
|
reasons.append(
|
|
"control checkout HEAD does not match prgs/master "
|
|
f"(HEAD {head_sha[:12]}, prgs/master {remote_master_sha[:12]})"
|
|
)
|
|
|
|
proven = not reasons
|
|
return _assessment(proven, reasons, root, workspace, branch or None, head_sha, dirty_files)
|
|
|
|
|
|
def format_root_checkout_guard_error(assessment: dict) -> str:
|
|
"""Single RuntimeError message for MCP preflight gates."""
|
|
root = assessment.get("canonical_repo_root") or "(unknown)"
|
|
workspace = assessment.get("workspace_path") or "(unknown)"
|
|
reasons = "; ".join(assessment.get("reasons") or ["unknown root checkout violation"])
|
|
return (
|
|
f"Root checkout guard (#475): {reasons}. "
|
|
f"canonical repository root: {root}; workspace: {workspace}. "
|
|
f"{REMEDIATION}"
|
|
)
|
|
|
|
|
|
def _assessment(
|
|
proven: bool,
|
|
reasons: list[str],
|
|
canonical_repo_root: str,
|
|
workspace_path: str,
|
|
current_branch: str | None,
|
|
head_sha: str | None,
|
|
dirty_files: list[str],
|
|
) -> dict:
|
|
return {
|
|
"proven": proven,
|
|
"block": not proven,
|
|
"reasons": reasons,
|
|
"canonical_repo_root": canonical_repo_root,
|
|
"workspace_path": workspace_path,
|
|
"current_branch": current_branch,
|
|
"head_sha": head_sha,
|
|
"dirty_files": dirty_files,
|
|
"remediation": REMEDIATION,
|
|
}
|
|
|
|
|
|
assess_root_checkout = assess_root_checkout_guard |