diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index cb904c4..154faf3 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -377,6 +377,37 @@ explicit control-checkout repair. Portable wording: [`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md). +### Root checkout guard (#475) + +The MCP server enforces a fail-closed **root checkout guard** before author, +reviewer, and merger mutations when the active workspace is not an isolated +`branches/...` worktree (reconciler close paths remain exempt per #468). + +The guard blocks when the **control checkout** (repository root) is: + +- on a non-stable branch (`master` / `main` / `dev` expected), +- detached HEAD, +- dirty (tracked edits), +- or its `HEAD` does not match `prgs/master` when that ref is available. + +**Remediation (never auto-reset or stash):** + +> Root checkout is not on master. Preserve state, switch root back to master, +> and use `scripts/worktree-review` or the sanctioned issue worktree flow. + +**Recovery after root hijack:** + +1. Preserve any in-progress edits (copy paths, note branch name, or commit on a + rescue branch from a `branches/...` worktree). +2. From the repository root: `git checkout master` (or `main` / `dev` per repo + policy) and `git fetch prgs && git merge --ff-only prgs/master` when safe. +3. Confirm `git status` is clean and `git branch --show-current` is `master`. +4. Resume work only inside `branches/issue--` via `gitea_lock_issue` / + `git worktree add`. + +`branches/...` directories are disposable role worktrees; the root checkout is +the stable orchestration surface only. + ## Shell Spawn Hard-Stop Rule Symptom: a shell tool call returns `exit_code: -1` with empty stdout/stderr. diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 8f303fd..db3220a 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -651,6 +651,7 @@ def verify_preflight_purity( f"{_format_preflight_files(reviewer_delta)}" ) + _enforce_root_checkout_guard(worktree_path) _enforce_branches_only_author_mutation(worktree_path) _clear_preflight_capability_state() @@ -693,6 +694,27 @@ def _verify_role_mutation_workspace( verify_preflight_purity(remote, worktree_path=resolved, task=task) return resolved + +def _enforce_root_checkout_guard(worktree_path: str | None = None) -> None: + """#475: fail closed when the stable control checkout is contaminated.""" + ctx = _resolve_author_mutation_context(worktree_path) + canonical_root = ctx["canonical_repo_root"] + workspace = ctx["workspace_path"] + git_state = issue_lock_worktree.read_worktree_git_state(canonical_root) + remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root) + assessment = root_checkout_guard.assess_root_checkout_guard( + workspace_path=workspace, + canonical_repo_root=canonical_root, + current_branch=git_state.get("current_branch"), + head_sha=git_state.get("head_sha"), + porcelain_status=git_state.get("porcelain_status") or "", + remote_master_sha=remote_master_sha, + resolved_role=_preflight_resolved_role, + ) + if assessment["block"]: + raise RuntimeError(root_checkout_guard.format_root_checkout_guard_error(assessment)) + + from mcp.server.fastmcp import FastMCP # noqa: E402 from gitea_auth import ( # noqa: E402 @@ -724,6 +746,7 @@ import stacked_pr_support # noqa: E402 import merge_approval_gate # noqa: E402 import already_landed_reconcile # noqa: E402 import author_mutation_worktree # noqa: E402 +import root_checkout_guard # noqa: E402 import issue_claim_heartbeat # noqa: E402 import issue_work_duplicate_gate # noqa: E402 import reviewer_pr_lease # noqa: E402 diff --git a/root_checkout_guard.py b/root_checkout_guard.py new file mode 100644 index 0000000..8cd29e4 --- /dev/null +++ b/root_checkout_guard.py @@ -0,0 +1,133 @@ +"""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 \ No newline at end of file diff --git a/tests/test_create_issue_workspace_guard.py b/tests/test_create_issue_workspace_guard.py index 8d1d1f6..7c8a2e8 100644 --- a/tests/test_create_issue_workspace_guard.py +++ b/tests/test_create_issue_workspace_guard.py @@ -38,8 +38,18 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase): @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) @patch("gitea_mcp_server.api_request") @patch("gitea_mcp_server.api_get_all", return_value=[]) - @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"}) - def test_create_issue_stable_checkout_rejected(self, _git, _get_all, mock_api, _role, _ns, _prof, _auth): + @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40) + @patch( + "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={ + "current_branch": "master", + "head_sha": "a" * 40, + "porcelain_status": "", + }, + ) + def test_create_issue_stable_checkout_rejected( + self, _git, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth, + ): # Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that # path is the stable control checkout (not under branches/), mutation must fail. with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): diff --git a/tests/test_root_checkout_guard.py b/tests/test_root_checkout_guard.py new file mode 100644 index 0000000..86dc064 --- /dev/null +++ b/tests/test_root_checkout_guard.py @@ -0,0 +1,161 @@ +"""Tests for root checkout guard (#475).""" + +from __future__ import annotations + +import os +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import gitea_mcp_server as srv # noqa: E402 +import root_checkout_guard as rcg # noqa: E402 + +CONTROL_ROOT = str(Path(__file__).resolve().parents[3]) +BRANCHES_WORKTREE = str(Path(__file__).resolve().parents[1]) +MASTER_SHA = "a" * 40 +OTHER_SHA = "b" * 40 + + +class TestAssessRootCheckoutGuard(unittest.TestCase): + def _assess(self, **kwargs): + defaults = { + "workspace_path": CONTROL_ROOT, + "canonical_repo_root": CONTROL_ROOT, + "current_branch": "master", + "head_sha": MASTER_SHA, + "porcelain_status": "", + "remote_master_sha": MASTER_SHA, + "resolved_role": "author", + } + defaults.update(kwargs) + return rcg.assess_root_checkout_guard(**defaults) + + def test_clean_master_control_checkout_allowed(self): + result = self._assess() + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + + def test_branches_worktree_allowed_for_author(self): + result = self._assess( + workspace_path=BRANCHES_WORKTREE, + current_branch="feat/issue-475-root-checkout-guard", + head_sha=OTHER_SHA, + resolved_role="author", + ) + self.assertTrue(result["proven"]) + + def test_branches_worktree_allowed_for_reviewer(self): + result = self._assess( + workspace_path=f"{CONTROL_ROOT}/branches/review-pr-1", + current_branch="review-pr-1", + resolved_role="reviewer", + ) + self.assertTrue(result["proven"]) + + def test_reconciler_always_allowed(self): + result = self._assess( + current_branch="feat/some-branch", + head_sha=OTHER_SHA, + porcelain_status=" M gitea_mcp_server.py\n", + resolved_role="reconciler", + ) + self.assertTrue(result["proven"]) + + def test_feature_branch_on_control_checkout_blocked(self): + result = self._assess( + current_branch="feat/issue-99-example", + head_sha=OTHER_SHA, + ) + self.assertTrue(result["block"]) + self.assertIn("not a stable base branch", result["reasons"][0]) + + def test_detached_head_blocked(self): + result = self._assess(current_branch=None) + self.assertTrue(result["block"]) + self.assertIn("detached HEAD", result["reasons"][0]) + + def test_dirty_control_checkout_blocked(self): + result = self._assess(porcelain_status=" M gitea_mcp_server.py\n") + self.assertTrue(result["block"]) + self.assertIn("tracked local edits", result["reasons"][0]) + + def test_head_behind_prgs_master_blocked(self): + result = self._assess( + head_sha=OTHER_SHA, + remote_master_sha=MASTER_SHA, + ) + self.assertTrue(result["block"]) + self.assertIn("does not match prgs/master", result["reasons"][0]) + + def test_merger_requires_clean_control_checkout(self): + result = self._assess( + workspace_path=BRANCHES_WORKTREE, + current_branch="feat/issue-475-root-checkout-guard", + head_sha=OTHER_SHA, + resolved_role="merger", + ) + self.assertTrue(result["block"]) + + +class TestVerifyPreflightRootGuardIntegration(unittest.TestCase): + def setUp(self): + srv._preflight_whoami_called = True + srv._preflight_capability_called = True + srv._preflight_resolved_role = "reviewer" + srv._preflight_whoami_violation = False + srv._preflight_capability_violation = False + self._orig_in_test = srv._preflight_in_test_mode + srv._preflight_in_test_mode = lambda: False + + def tearDown(self): + srv._preflight_in_test_mode = self._orig_in_test + + @patch("gitea_mcp_server._get_workspace_porcelain", return_value="") + @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value=MASTER_SHA) + @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state") + @patch("gitea_mcp_server._resolve_author_mutation_context") + def test_reviewer_from_contaminated_root_blocked( + self, mock_ctx, mock_git, _remote_sha, _porcelain, + ): + srv._preflight_capability_baseline_porcelain = "" + mock_ctx.return_value = { + "workspace_path": CONTROL_ROOT, + "canonical_repo_root": CONTROL_ROOT, + "process_project_root": CONTROL_ROOT, + } + mock_git.return_value = { + "current_branch": "feat/hijacked-root", + "head_sha": OTHER_SHA, + "porcelain_status": "", + } + with self.assertRaises(RuntimeError) as ctx: + srv.verify_preflight_purity("prgs", worktree_path=CONTROL_ROOT) + self.assertIn("Root checkout guard (#475)", str(ctx.exception)) + self.assertIn(rcg.REMEDIATION, str(ctx.exception)) + + @patch("gitea_mcp_server._get_workspace_porcelain", return_value="") + @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value=MASTER_SHA) + @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state") + @patch("gitea_mcp_server._resolve_author_mutation_context") + def test_reviewer_from_branches_worktree_allowed( + self, mock_ctx, mock_git, _remote_sha, _porcelain, + ): + srv._preflight_capability_baseline_porcelain = "" + mock_ctx.return_value = { + "workspace_path": BRANCHES_WORKTREE, + "canonical_repo_root": CONTROL_ROOT, + "process_project_root": BRANCHES_WORKTREE, + } + mock_git.return_value = { + "current_branch": "feat/issue-475-root-checkout-guard", + "head_sha": OTHER_SHA, + "porcelain_status": "", + } + srv.verify_preflight_purity("prgs", worktree_path=BRANCHES_WORKTREE) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_workspace_guard_alignment.py b/tests/test_workspace_guard_alignment.py index ddcb67f..4454351 100644 --- a/tests/test_workspace_guard_alignment.py +++ b/tests/test_workspace_guard_alignment.py @@ -126,17 +126,37 @@ class TestRuntimeContextGuardAlignment(unittest.TestCase): with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False): srv.verify_preflight_purity(worktree_path=BRANCHES_WORKTREE) - def test_stable_checkout_still_rejected(self): + @mock.patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40) + @mock.patch( + "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={ + "current_branch": "master", + "head_sha": "a" * 40, + "porcelain_status": "", + }, + ) + def test_stable_checkout_still_rejected(self, _git, _remote_sha): with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT): with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False): with self.assertRaises(RuntimeError) as ctx: srv.verify_preflight_purity() self.assertIn("stable control checkout", str(ctx.exception)) + @mock.patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40) + @mock.patch( + "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={ + "current_branch": "master", + "head_sha": "a" * 40, + "porcelain_status": "", + }, + ) @mock.patch("os.path.isdir", return_value=True) @mock.patch("os.path.exists", return_value=True) @mock.patch("subprocess.run") - def test_non_branches_worktree_rejected(self, mock_run, *_exists): + def test_non_branches_worktree_rejected( + self, mock_run, mock_exists, mock_isdir, _git, _remote_sha, + ): outside = "/tmp/outside-repo-checkout" mock_run.return_value = MagicMock( returncode=0,