feat: enforce branches-only author mutation worktree guard (Closes #274)

Add author_mutation_worktree assessment and integrate it into MCP
preflight so author mutations fail closed outside branches/ session
worktrees, with tests for control-checkout and branches-path cases.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

author_mutation_worktree.py

@@ -0,0 +1,103 @@
+"""Branches-only author mutation worktree guard (#274).
+
+Author/coder mutations must run from a session-owned worktree under the
+project's ``branches/`` directory, never from the stable control checkout.
+"""
+
+from __future__ import annotations
+
+import os
+
+BASE_BRANCHES = frozenset({"master", "main", "dev"})
+
+
+def _normalize_path(path: str) -> str:
+    return (path or "").replace("\\", "/").rstrip("/")
+
+
+def is_path_under_branches(path: str, project_root: str | None = None) -> bool:
+    """True when *path* resolves inside ``<project_root>/branches/``."""
+    normalized = _normalize_path(path)
+    if not normalized:
+        return False
+    if "/branches/" in f"{normalized}/":
+        return True
+    if normalized.endswith("/branches"):
+        return True
+    if project_root:
+        root = _normalize_path(os.path.realpath(project_root))
+        real = _normalize_path(os.path.realpath(path))
+        if real.startswith(f"{root}/"):
+            rel = real[len(root) + 1 :]
+            return rel == "branches" or rel.startswith("branches/")
+    return False
+
+
+def resolve_mutation_workspace(
+    worktree_path: str | None,
+    project_root: str,
+    *,
+    active_worktree_env: str | None = None,
+    author_worktree_env: str | None = None,
+) -> str:
+    """Resolve the workspace path inspected before author mutations."""
+    for candidate in (worktree_path, active_worktree_env, author_worktree_env):
+        text = (candidate or "").strip()
+        if text:
+            return os.path.realpath(os.path.abspath(text))
+    return os.path.realpath(project_root)
+
+
+def assess_author_mutation_worktree(
+    *,
+    workspace_path: str,
+    project_root: str,
+    current_branch: str | None = None,
+    base_branches: frozenset[str] | None = None,
+) -> dict:
+    """Fail closed when author mutations are not rooted under ``branches/``."""
+    bases = base_branches or BASE_BRANCHES
+    reasons: list[str] = []
+    root = os.path.realpath(project_root)
+    workspace = os.path.realpath(workspace_path)
+    branch = (current_branch or "").strip()
+
+    if workspace == root:
+        reasons.append(
+            "author mutation blocked: workspace is the stable control checkout; "
+            "create or switch to a session-owned worktree under branches/"
+        )
+    elif not is_path_under_branches(workspace, root):
+        reasons.append(
+            f"author mutation blocked: workspace '{workspace}' is not under "
+            f"'{root}/branches/'; create a branches/<task> worktree first"
+        )
+
+    if workspace == root and branch and branch not in bases:
+        reasons.append(
+            f"control checkout drift: branch '{branch}' is not a stable base "
+            f"branch ({'/'.join(sorted(bases))})"
+        )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "project_root": root,
+        "workspace_path": workspace,
+        "under_branches": is_path_under_branches(workspace, root),
+        "current_branch": branch or None,
+    }
+
+
+def format_author_mutation_worktree_error(assessment: dict) -> str:
+    """Single RuntimeError message for MCP preflight gates."""
+    workspace = assessment.get("workspace_path") or "(unknown)"
+    root = assessment.get("project_root") or "(unknown)"
+    reasons = "; ".join(assessment.get("reasons") or ["unknown branches-only violation"])
+    return (
+        f"Branches-only mutation guard (#274): {reasons}. "
+        f"project root: {root}; workspace: {workspace}. "
+        "Create a session-owned worktree under branches/ before mutating."
+    )

gitea_mcp_server.py

@@ -397,6 +397,28 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
             _preflight_resolved_role = resolved_role
 
 
+def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
+    """#274: author mutations must run from a branches/ session worktree."""
+    if _preflight_resolved_role == "reviewer":
+        return
+    workspace = author_mutation_worktree.resolve_mutation_workspace(
+        worktree_path,
+        PROJECT_ROOT,
+        active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
+        author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
+    )
+    git_state = issue_lock_worktree.read_worktree_git_state(workspace)
+    assessment = author_mutation_worktree.assess_author_mutation_worktree(
+        workspace_path=workspace,
+        project_root=PROJECT_ROOT,
+        current_branch=git_state.get("current_branch"),
+    )
+    if assessment["block"]:
+        raise RuntimeError(
+            author_mutation_worktree.format_author_mutation_worktree_error(assessment)
+        )
+
+
 def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
     """Verify that identity and capability were verified prior to session edits."""
     global _preflight_reviewer_violation_files
@@ -426,8 +448,7 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
                 "file edits before mutation (fail closed). "
                 f"{_format_preflight_workspace_details(details)}"
             )
-        return
+    else:
-
         if _preflight_whoami_violation:
             raise RuntimeError(
                 "Pre-flight order violation: Workspace file edits occurred before "
@@ -453,6 +474,8 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
                     f"{_format_preflight_files(reviewer_delta)}"
                 )
 
+    _enforce_branches_only_author_mutation(worktree_path)
+
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
 from gitea_auth import (  # noqa: E402
@@ -476,6 +499,7 @@ import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
+import author_mutation_worktree  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 
 

tests/test_author_mutation_worktree.py

@@ -0,0 +1,96 @@
+"""Tests for branches-only author mutation worktree guard (#274)."""
+
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import author_mutation_worktree as amw  # noqa: E402
+
+
+class TestPathUnderBranches(unittest.TestCase):
+    ROOT = "/repo/Gitea-Tools"
+
+    def test_branches_subdirectory_passes(self):
+        self.assertTrue(
+            amw.is_path_under_branches(f"{self.ROOT}/branches/issue-274", self.ROOT)
+        )
+
+    def test_control_checkout_fails(self):
+        self.assertFalse(amw.is_path_under_branches(self.ROOT, self.ROOT))
+
+    def test_sibling_directory_fails(self):
+        self.assertFalse(
+            amw.is_path_under_branches("/repo/other-checkout", self.ROOT)
+        )
+
+
+class TestAssessAuthorMutationWorktree(unittest.TestCase):
+    ROOT = "/repo/Gitea-Tools"
+
+    def test_branches_worktree_passes(self):
+        result = amw.assess_author_mutation_worktree(
+            workspace_path=f"{self.ROOT}/branches/issue-274",
+            project_root=self.ROOT,
+            current_branch="feat/issue-274-branches-only-worktrees",
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_control_checkout_blocks(self):
+        result = amw.assess_author_mutation_worktree(
+            workspace_path=self.ROOT,
+            project_root=self.ROOT,
+            current_branch="master",
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertIn("control checkout", result["reasons"][0])
+
+    def test_drifted_control_branch_blocks(self):
+        result = amw.assess_author_mutation_worktree(
+            workspace_path=self.ROOT,
+            project_root=self.ROOT,
+            current_branch="feat/some-task",
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(any("drift" in r for r in result["reasons"]))
+
+
+class TestPreflightIntegration(unittest.TestCase):
+    def test_verify_preflight_blocks_control_checkout_with_test_porcelain(self):
+        import mcp_server
+
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+        mcp_server._preflight_resolved_role = "author"
+        with mock.patch.dict(
+            "os.environ",
+            {"GITEA_TEST_PORCELAIN": ""},
+            clear=False,
+        ):
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server.verify_preflight_purity()
+        self.assertIn("Branches-only mutation guard", str(ctx.exception))
+
+    def test_verify_preflight_allows_branches_worktree(self):
+        import mcp_server
+
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+        mcp_server._preflight_resolved_role = "author"
+        worktree = "/repo/Gitea-Tools/branches/issue-274"
+        with mock.patch.dict(
+            "os.environ",
+            {"GITEA_TEST_PORCELAIN": ""},
+            clear=False,
+        ):
+            mcp_server.verify_preflight_purity(worktree_path=worktree)
+
+
+if __name__ == "__main__":
+    unittest.main()