gitea_mcp_server.py

@@ -440,10 +440,48 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
             "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
         )
 
-    if worktree_path:
-        dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path)))
+    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),
+    )
+    real_workspace = os.path.realpath(workspace)
+    real_root = os.path.realpath(PROJECT_ROOT)
+
+    if real_workspace != real_root:
+        if not _preflight_in_test_mode():
+            if not os.path.exists(real_workspace):
+                raise RuntimeError(
+                    f"Branches-only mutation guard (#274): worktree path '{workspace}' does not exist (fail closed)"
+                )
+            if not os.path.isdir(real_workspace):
+                raise RuntimeError(
+                    f"Branches-only mutation guard (#274): worktree path '{workspace}' is not a directory (fail closed)"
+                )
+            try:
+                res = subprocess.run(
+                    ["git", "-C", real_workspace, "rev-parse", "--git-common-dir"],
+                    capture_output=True,
+                    text=True,
+                    check=True,
+                )
+                common_dir = os.path.realpath(res.stdout.strip())
+                expected_dir = os.path.realpath(os.path.join(real_root, ".git"))
+                if common_dir != expected_dir:
+                    raise RuntimeError(
+                        f"Branches-only mutation guard (#274): worktree '{workspace}' does not belong to the target repository '{PROJECT_ROOT}' (fail closed)"
+                    )
+            except Exception as e:
+                if isinstance(e, RuntimeError):
+                    raise e
+                raise RuntimeError(
+                    f"Branches-only mutation guard (#274): worktree '{workspace}' is not a valid git repository (fail closed)"
+                )
+
+        dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace)))
         if dirty_files:
-            details = _preflight_workspace_details(worktree_path, dirty_files)
+            details = _preflight_workspace_details(workspace, dirty_files)
             raise RuntimeError(
                 "Pre-flight order violation: Active task workspace has tracked "
                 "file edits before mutation (fail closed). "
@@ -972,6 +1010,7 @@ def gitea_create_issue(
     repo: str | None = None,
     allow_duplicate_override: bool = False,
     split_from_issue: int | None = None,
+    worktree_path: str | None = None,
 ) -> dict:
     """Create a new issue on a Gitea repository.
 
@@ -984,6 +1023,7 @@ def gitea_create_issue(
         repo: Override the repository name.
         allow_duplicate_override: Operator-approved split after duplicate found.
         split_from_issue: Existing duplicate issue number when overriding.
+        worktree_path: Optional path to verify branches-only guard.
 
     Returns:
         dict with 'number' of the created issue ('url' only with the reveal opt-in).
@@ -1010,7 +1050,7 @@ def gitea_create_issue(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, worktree_path=worktree_path)
     base = repo_api_url(h, o, r)
     open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
     closed_issues = api_get_all(

tests/test_create_issue_workspace_guard.py

@@ -0,0 +1,145 @@
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+
+# Ensure we import from the repo root
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server as srv
+
+FAKE_AUTH = {"Authorization": "token test-token"}
+# Stable control checkout (parent of branches/), not the MCP server worktree root.
+CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
+PROJECT_ROOT = srv.PROJECT_ROOT
+
+
+class TestCreateIssueWorkspaceGuard(unittest.TestCase):
+
+    def setUp(self):
+        # Reset preflight flags
+        srv._preflight_whoami_called = True
+        srv._preflight_capability_called = True
+        srv._preflight_resolved_role = "author"
+        srv._preflight_whoami_violation = False
+        srv._preflight_capability_violation = False
+        
+        # Disable early return in verify_preflight_purity for testing
+        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._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @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):
+        # 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):
+            with self.assertRaises(RuntimeError) as ctx:
+                srv.gitea_create_issue(title="Test issue", body="body text")
+        self.assertIn("stable control checkout", str(ctx.exception))
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @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"})
+    @patch("os.path.exists", return_value=True)
+    @patch("os.path.isdir", return_value=True)
+    @patch("subprocess.run")
+    def test_create_issue_valid_worktree_succeeds(self, mock_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
+        # Mock subprocess.run for git --git-common-dir to return PROJECT_ROOT/.git
+        mock_res = MagicMock()
+        mock_res.stdout = f"{CONTROL_CHECKOUT_ROOT}/.git\n"
+        mock_run.return_value = mock_res
+
+        mock_api.return_value = {"number": 42, "html_url": "https://gitea.example.com/issues/42"}
+
+        # Provide a valid branches path under the control checkout root
+        valid_path = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "feat-issue-1")
+        
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
+                res = srv.gitea_create_issue(
+                    title="Test issue", body="body", worktree_path=valid_path
+                )
+            
+        self.assertEqual(res["number"], 42)
+        mock_api.assert_called_once()
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @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_missing_worktree_fails_closed(self, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
+        # Path under branches/ but doesn't exist
+        missing_path = os.path.join(
+            CONTROL_CHECKOUT_ROOT, "branches", "nonexistent-worktree-path-999"
+        )
+        
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with self.assertRaises(RuntimeError) as ctx:
+                srv.gitea_create_issue(
+                    title="Test issue", body="body", worktree_path=missing_path
+                )
+        self.assertIn("does not exist (fail closed)", str(ctx.exception))
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @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"})
+    @patch("os.path.exists", return_value=True)
+    @patch("os.path.isdir", return_value=True)
+    @patch("subprocess.run")
+    def test_create_issue_wrong_repo_fails_closed(self, mock_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
+        # Mock subprocess.run for git --git-common-dir to return a different path
+        mock_res = MagicMock()
+        mock_res.stdout = "/Users/jasonwalker/Development/some-other-repo/.git\n"
+        mock_run.return_value = mock_res
+
+        wrong_repo_path = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "feat-issue-1")
+
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with self.assertRaises(RuntimeError) as ctx:
+                srv.gitea_create_issue(
+                    title="Test issue", body="body", worktree_path=wrong_repo_path
+                )
+        self.assertIn("does not belong to the target repository", str(ctx.exception))
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    def test_create_issue_fails_without_whoami_preflight(self, _ns, _prof, _auth):
+        srv._preflight_whoami_called = False
+        with self.assertRaises(RuntimeError) as ctx:
+            srv.gitea_create_issue(title="Test issue", body="body")
+        self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    def test_create_issue_fails_without_capability_preflight(self, _ns, _prof, _auth):
+        srv._preflight_capability_called = False
+        with self.assertRaises(RuntimeError) as ctx:
+            srv.gitea_create_issue(title="Test issue", body="body")
+        self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()