author_mutation_worktree.py

@@ -7,8 +7,11 @@ project's ``branches/`` directory, never from the stable control checkout.
 from __future__ import annotations
 
 import os
+import subprocess
 
 BASE_BRANCHES = frozenset({"master", "main", "dev"})
+ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
+AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
 
 
 def _normalize_path(path: str) -> str:
@@ -48,6 +51,130 @@ def resolve_mutation_workspace(
     return os.path.realpath(project_root)
 
 
+def _realpath_git_common_dir(workspace_path: str, common_dir: str) -> str:
+    """Resolve ``git rev-parse --git-common-dir`` relative to *workspace_path*."""
+    raw = (common_dir or "").strip()
+    if not raw:
+        return raw
+    if os.path.isabs(raw):
+        return os.path.realpath(raw)
+    return os.path.realpath(os.path.join(workspace_path, raw))
+
+
+def resolve_canonical_repo_root(workspace_path: str, fallback_project_root: str) -> str:
+    """Return the stable repository root for *workspace_path* via git metadata (#460)."""
+    path = (workspace_path or "").strip()
+    fallback = os.path.realpath(fallback_project_root)
+    if not path:
+        return fallback
+    try:
+        res = subprocess.run(
+            ["git", "-C", path, "rev-parse", "--git-common-dir"],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        common = _realpath_git_common_dir(path, res.stdout)
+    except Exception:
+        return fallback
+    if common.endswith(f"{os.sep}.git"):
+        return os.path.dirname(common)
+    if os.path.basename(common) == ".git":
+        return os.path.dirname(common)
+    return fallback
+
+
+def resolve_author_mutation_context(
+    worktree_path: str | None,
+    process_project_root: str,
+    *,
+    active_worktree_env: str | None = None,
+    author_worktree_env: str | None = None,
+) -> dict:
+    """Shared workspace resolution for runtime_context and mutation guards (#460)."""
+    workspace = resolve_mutation_workspace(
+        worktree_path,
+        process_project_root,
+        active_worktree_env=active_worktree_env,
+        author_worktree_env=author_worktree_env,
+    )
+    process_root = os.path.realpath(process_project_root)
+    # Canonical repository identity comes from the MCP process checkout (#460),
+    # not from the declared task workspace being validated.
+    canonical_root = resolve_canonical_repo_root(process_root, process_root)
+    return {
+        "workspace_path": workspace,
+        "process_project_root": process_root,
+        "canonical_repo_root": canonical_root,
+        "roots_aligned": canonical_root == process_root,
+    }
+
+
+def assess_workspace_repo_membership(
+    *,
+    workspace_path: str,
+    canonical_repo_root: str,
+) -> dict:
+    """Fail closed when *workspace_path* is not a git worktree of *canonical_repo_root*."""
+    workspace = os.path.realpath(workspace_path)
+    root = os.path.realpath(canonical_repo_root)
+    reasons: list[str] = []
+
+    if not os.path.exists(workspace):
+        reasons.append(f"worktree path '{workspace}' does not exist")
+        return _membership_assessment(False, reasons, workspace, root, None)
+
+    if not os.path.isdir(workspace):
+        reasons.append(f"worktree path '{workspace}' is not a directory")
+        return _membership_assessment(False, reasons, workspace, root, None)
+
+    try:
+        res = subprocess.run(
+            ["git", "-C", workspace, "rev-parse", "--git-common-dir"],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        common_dir = _realpath_git_common_dir(workspace, res.stdout)
+    except Exception:
+        reasons.append(f"worktree '{workspace}' is not a valid git repository")
+        return _membership_assessment(False, reasons, workspace, root, None)
+
+    expected_dir = os.path.realpath(os.path.join(root, ".git"))
+    if common_dir != expected_dir:
+        reasons.append(
+            f"worktree '{workspace}' does not belong to the target repository '{root}'"
+        )
+    return _membership_assessment(not reasons, reasons, workspace, root, common_dir)
+
+
+def _membership_assessment(
+    proven: bool,
+    reasons: list[str],
+    workspace: str,
+    root: str,
+    common_dir: str | None,
+) -> dict:
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "workspace_path": workspace,
+        "canonical_repo_root": root,
+        "git_common_dir": common_dir,
+    }
+
+
+def format_workspace_repo_membership_error(assessment: dict) -> str:
+    workspace = assessment.get("workspace_path") or "(unknown)"
+    root = assessment.get("canonical_repo_root") or "(unknown)"
+    reasons = "; ".join(assessment.get("reasons") or ["unknown repository membership violation"])
+    return (
+        f"Branches-only mutation guard (#274): {reasons} (fail closed). "
+        f"canonical repository root: {root}; workspace: {workspace}."
+    )
+
+
 def assess_author_mutation_worktree(
     *,
     workspace_path: str,

gitea_mcp_server.py

@@ -190,14 +190,22 @@ def _ensure_process_start_porcelain() -> str:
 
 def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
     """Resolve the workspace root inspected by pre-flight guards."""
-    path = (worktree_path or "").strip()
+    return author_mutation_worktree.resolve_mutation_workspace(
-    if not path:
+        worktree_path,
-        path = (os.environ.get(ACTIVE_WORKTREE_ENV) or "").strip()
+        PROJECT_ROOT,
-    if not path:
+        active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
-        path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip()
+        author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
-    if not path:
+    )
-        path = PROJECT_ROOT
+
-    return os.path.realpath(os.path.abspath(path))
+
+def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
+    """Canonical workspace + repository root for runtime_context and guards (#460)."""
+    return author_mutation_worktree.resolve_author_mutation_context(
+        worktree_path,
+        PROJECT_ROOT,
+        active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
+        author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
+    )
 
 
 def _get_git_root(path: str) -> str | None:
@@ -267,21 +275,31 @@ def _format_preflight_files(files: list[str]) -> str:
 
 
 def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict:
-    workspace = _resolve_preflight_workspace_path(worktree_path)
+    ctx = _resolve_author_mutation_context(worktree_path)
+    workspace = ctx["workspace_path"]
     inspected_root = _get_git_root(workspace)
-    control_root = os.path.realpath(PROJECT_ROOT)
+    process_root = ctx["process_project_root"]
+    canonical_root = ctx["canonical_repo_root"]
     active_root = os.path.realpath(inspected_root or workspace)
-    if active_root == control_root:
+    if active_root == canonical_root:
         dirty_scope = "control checkout"
     else:
         dirty_scope = "active task workspace"
-    return {
+    details = {
-        "mcp_server_process_root": control_root,
+        "mcp_server_process_root": process_root,
+        "canonical_repository_root": canonical_root,
         "active_task_workspace_root": active_root,
         "inspected_git_root": inspected_root,
         "dirty_files": list(dirty_files),
         "dirty_scope": dirty_scope,
+        "workspace_roots_aligned": ctx["roots_aligned"],
     }
+    if not ctx["roots_aligned"]:
+        details["workspace_root_mismatch"] = (
+            "runtime_context and mutation guard use canonical repository root "
+            f"'{canonical_root}' instead of MCP process root '{process_root}'"
+        )
+    return details
 
 
 def _format_preflight_workspace_details(details: dict) -> str:
@@ -402,16 +420,12 @@ def _enforce_branches_only_author_mutation(worktree_path: str | 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(
+    ctx = _resolve_author_mutation_context(worktree_path)
-        worktree_path,
+    workspace = ctx["workspace_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,
+        project_root=ctx["canonical_repo_root"],
         current_branch=git_state.get("current_branch"),
     )
     if assessment["block"]:
@@ -440,43 +454,23 @@ 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)"
         )
 
-    workspace = author_mutation_worktree.resolve_mutation_workspace(
+    ctx = _resolve_author_mutation_context(worktree_path)
-        worktree_path,
+    workspace = ctx["workspace_path"]
-        PROJECT_ROOT,
+    canonical_root = ctx["canonical_repo_root"]
-        active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
+    process_root = ctx["process_project_root"]
-        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 real_workspace != process_root:
         if not _preflight_in_test_mode():
-            if not os.path.exists(real_workspace):
+            membership = author_mutation_worktree.assess_workspace_repo_membership(
-                raise RuntimeError(
+                workspace_path=workspace,
-                    f"Branches-only mutation guard (#274): worktree path '{workspace}' does not exist (fail closed)"
+                canonical_repo_root=canonical_root,
             )
-            if not os.path.isdir(real_workspace):
+            if membership["block"]:
                 raise RuntimeError(
-                    f"Branches-only mutation guard (#274): worktree path '{workspace}' is not a directory (fail closed)"
+                    author_mutation_worktree.format_workspace_repo_membership_error(
+                        membership
                     )
-            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)))

tests/test_create_issue_workspace_guard.py

@@ -106,16 +106,28 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
     @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("author_mutation_worktree.subprocess.run")
     @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):
+    def test_create_issue_wrong_repo_fails_closed(self, mock_run, mock_amw_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")
 
+        def _subprocess_side_effect(cmd, *args, **kwargs):
+            mock_res = MagicMock(returncode=0)
+            if "--git-common-dir" in cmd:
+                cwd = cmd[cmd.index("-C") + 1] if "-C" in cmd else ""
+                if cwd == wrong_repo_path:
+                    mock_res.stdout = "/Users/jasonwalker/Development/some-other-repo/.git\n"
+                else:
+                    mock_res.stdout = f"{CONTROL_CHECKOUT_ROOT}/.git\n"
+            else:
+                mock_res.stdout = ""
+            return mock_res
+
+        mock_run.side_effect = _subprocess_side_effect
+        mock_amw_run.side_effect = _subprocess_side_effect
+
         with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
                 with self.assertRaises(RuntimeError) as ctx:
                     srv.gitea_create_issue(
                         title="Test issue", body="body", worktree_path=wrong_repo_path

tests/test_workspace_guard_alignment.py

@@ -0,0 +1,153 @@
+"""Tests for runtime_context / mutation-guard workspace alignment (#460)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+from unittest.mock import MagicMock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import author_mutation_worktree as amw  # noqa: E402
+import gitea_mcp_server as srv  # noqa: E402
+
+CONTROL_ROOT = str(Path(__file__).resolve().parents[3])
+BRANCHES_WORKTREE = str(Path(__file__).resolve().parents[1])
+MCP_PROCESS_ROOT = BRANCHES_WORKTREE
+
+
+class TestCanonicalRepoRoot(unittest.TestCase):
+    @mock.patch("subprocess.run")
+    def test_resolves_main_repo_from_branches_worktree(self, mock_run):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout=f"{CONTROL_ROOT}/.git\n",
+        )
+        root = amw.resolve_canonical_repo_root(BRANCHES_WORKTREE, MCP_PROCESS_ROOT)
+        self.assertEqual(root, CONTROL_ROOT)
+
+    def test_falls_back_when_git_unavailable(self):
+        root = amw.resolve_canonical_repo_root("/missing/path", MCP_PROCESS_ROOT)
+        self.assertEqual(root, os.path.realpath(MCP_PROCESS_ROOT))
+
+
+class TestWorkspaceRepoMembership(unittest.TestCase):
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    @mock.patch("subprocess.run")
+    def test_valid_branches_worktree_accepted(self, mock_run, *_exists):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout=f"{CONTROL_ROOT}/.git\n",
+        )
+        result = amw.assess_workspace_repo_membership(
+            workspace_path=BRANCHES_WORKTREE,
+            canonical_repo_root=CONTROL_ROOT,
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    @mock.patch("subprocess.run")
+    def test_wrong_repo_rejected(self, mock_run, *_exists):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout="/other/repo/.git\n",
+        )
+        result = amw.assess_workspace_repo_membership(
+            workspace_path=BRANCHES_WORKTREE,
+            canonical_repo_root=CONTROL_ROOT,
+        )
+        self.assertTrue(result["block"])
+        self.assertIn("does not belong", result["reasons"][0])
+
+    @mock.patch("os.path.exists", return_value=False)
+    def test_missing_worktree_rejected(self, *_exists):
+        result = amw.assess_workspace_repo_membership(
+            workspace_path=f"{CONTROL_ROOT}/branches/missing-worktree",
+            canonical_repo_root=CONTROL_ROOT,
+        )
+        self.assertTrue(result["block"])
+        self.assertIn("does not exist", result["reasons"][0])
+
+
+class TestRuntimeContextGuardAlignment(unittest.TestCase):
+    def setUp(self):
+        srv._preflight_whoami_called = True
+        srv._preflight_capability_called = True
+        srv._preflight_resolved_role = "author"
+        self._orig_in_test = srv._preflight_in_test_mode
+        srv._preflight_in_test_mode = lambda: False
+        self._env_patch = mock.patch.dict(
+            os.environ,
+            {},
+            clear=False,
+        )
+        self._env_patch.start()
+        os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
+        os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
+
+    def tearDown(self):
+        srv._preflight_in_test_mode = self._orig_in_test
+        self._env_patch.stop()
+
+    def test_runtime_context_and_guard_share_resolved_workspace(self):
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            ctx = srv._resolve_author_mutation_context(BRANCHES_WORKTREE)
+            status = srv.assess_preflight_status(worktree_path=BRANCHES_WORKTREE)
+        self.assertEqual(ctx["workspace_path"], os.path.realpath(BRANCHES_WORKTREE))
+        self.assertEqual(ctx["canonical_repo_root"], CONTROL_ROOT)
+        self.assertFalse(ctx["roots_aligned"])
+        self.assertEqual(
+            status["preflight_workspace"]["active_task_workspace_root"],
+            os.path.realpath(BRANCHES_WORKTREE),
+        )
+        self.assertEqual(
+            status["preflight_workspace"]["canonical_repository_root"],
+            CONTROL_ROOT,
+        )
+        self.assertIn("workspace_root_mismatch", status["preflight_workspace"])
+
+    @mock.patch("subprocess.run")
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    def test_declared_branches_worktree_passes_when_mcp_root_differs(
+        self, _exists, _isdir, mock_run
+    ):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout=f"{CONTROL_ROOT}/.git\n",
+        )
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            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):
+        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("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):
+        outside = "/tmp/outside-repo-checkout"
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout=f"{CONTROL_ROOT}/.git\n",
+        )
+        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(worktree_path=outside)
+        self.assertIn("not under", str(ctx.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()