fix: bind reviewer mutations to active branches worktree (Closes #503)

Reviewer mutation tools now resolve workspace from worktree_path,
GITEA_ACTIVE_WORKTREE, or the session reviewer lease before preflight
and branches-only validation. Preflight rejects metadata-only worktree_path
bindings that would mislead runtime_context while mutations still inspect
the MCP process root.

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

gitea_mcp_server.py

@@ -208,6 +208,71 @@ def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
     )
 
 
+def _reviewer_session_worktree() -> str | None:
+    session = reviewer_pr_lease.get_session_lease()
+    if not session:
+        return None
+    worktree = (session.get("worktree") or "").strip()
+    return worktree or None
+
+
+def _resolve_reviewer_mutation_worktree_path(
+    worktree_path: str | None = None,
+    worktree: str | None = None,
+) -> str:
+    """Resolve the workspace inspected before reviewer mutations (#503)."""
+    return reviewer_mutation_workspace.resolve_reviewer_mutation_workspace(
+        worktree_path=worktree_path,
+        worktree=worktree,
+        process_project_root=PROJECT_ROOT,
+        active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
+        session_lease_worktree=_reviewer_session_worktree(),
+    )
+
+
+def _assess_reviewer_mutation_workspace(
+    worktree_path: str | None = None,
+    worktree: str | None = None,
+) -> dict:
+    git_state = issue_lock_worktree.read_worktree_git_state(
+        _resolve_reviewer_mutation_worktree_path(worktree_path, worktree)
+    )
+    return reviewer_mutation_workspace.assess_reviewer_mutation_workspace(
+        worktree_path=worktree_path,
+        worktree=worktree,
+        process_project_root=PROJECT_ROOT,
+        active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
+        session_lease_worktree=_reviewer_session_worktree(),
+        current_branch=git_state.get("current_branch"),
+    )
+
+
+def _verify_reviewer_mutation_workspace(
+    remote: str | None = None,
+    worktree_path: str | None = None,
+    worktree: str | None = None,
+) -> str:
+    """Bind reviewer mutations to the active branches/ worktree (#503)."""
+    assessment = _assess_reviewer_mutation_workspace(worktree_path, worktree)
+    if assessment["block"]:
+        raise RuntimeError(
+            reviewer_mutation_workspace.format_stale_mcp_workspace_binding_error(
+                process_project_root=assessment["process_project_root"],
+                mutation_workspace=assessment["mutation_workspace"],
+                declared_worktree_path=assessment.get("declared_worktree_path"),
+                metadata_only=assessment.get("metadata_only", False),
+            )
+            + (
+                f" Details: {'; '.join(assessment['reasons'])}"
+                if assessment.get("reasons")
+                else ""
+            )
+        )
+    resolved = assessment["mutation_workspace"]
+    verify_preflight_purity(remote, worktree_path=resolved)
+    return resolved
+
+
 def _get_git_root(path: str) -> str | None:
     try:
         res = subprocess.run(
@@ -332,6 +397,22 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
                 "Active task workspace has tracked file edits before mutation "
                 f"({_format_preflight_workspace_details(workspace_details)})"
             )
+        if _preflight_resolved_role == "reviewer":
+            binding = reviewer_mutation_workspace.assess_metadata_only_worktree_binding(
+                declared_worktree_path=worktree_path,
+                mutation_workspace=_resolve_reviewer_mutation_worktree_path(),
+                process_project_root=PROJECT_ROOT,
+            )
+            if binding.get("block"):
+                reasons.append(binding["reasons"][0])
+                reasons.append(
+                    reviewer_mutation_workspace.format_stale_mcp_workspace_binding_error(
+                        process_project_root=PROJECT_ROOT,
+                        mutation_workspace=binding["mutation_workspace"],
+                        declared_worktree_path=binding.get("declared_worktree_path"),
+                        metadata_only=True,
+                    )
+                )
         return {
             "preflight_ready": not reasons,
             "preflight_block_reasons": reasons,
@@ -542,6 +623,7 @@ import author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
 import issue_work_duplicate_gate  # noqa: E402
 import reviewer_pr_lease  # noqa: E402
+import reviewer_mutation_workspace  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import reconciler_profile  # noqa: E402
 import reconciliation_workflow  # noqa: E402
@@ -2639,9 +2721,28 @@ def _evaluate_pr_review_submission(
     *,
     live: bool,
     final_review_decision_ready: bool = False,
+    worktree_path: str | None = None,
 ) -> dict:
     """Shared gate chain for live submit and dry-run review tools."""
-    verify_preflight_purity(remote)
+    try:
+        _verify_reviewer_mutation_workspace(
+            remote, worktree_path=worktree_path
+        )
+    except RuntimeError as exc:
+        return {
+            "requested_action": (action or "").strip().lower(),
+            "performed": False,
+            "dry_run": not live,
+            "would_perform": False,
+            "authenticated_user": None,
+            "profile_name": get_profile()["profile_name"],
+            "pr_author": None,
+            "pr_number": pr_number,
+            "head_sha": None,
+            "expected_head_sha": expected_head_sha,
+            "remote": remote if remote in REMOTES else None,
+            "reasons": [str(exc)],
+        }
     action = (action or "").strip().lower()
     result = {
         "requested_action": action,
@@ -2832,9 +2933,16 @@ def gitea_mark_final_review_decision(
     remote: str = "dadeschools",
     org: str | None = None,
     repo: str | None = None,
+    worktree_path: str | None = None,
 ) -> dict:
     """Mark validation complete; the final review decision is ready to submit."""
     action = (action or "").strip().lower()
+    try:
+        _verify_reviewer_mutation_workspace(
+            remote, worktree_path=worktree_path
+        )
+    except RuntimeError as exc:
+        return {"marked_ready": False, "reasons": [str(exc)]}
     lock = _load_review_decision_lock()
     if lock is None:
         return {
@@ -3041,6 +3149,7 @@ def gitea_dry_run_pr_review(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
+    worktree_path: str | None = None,
 ) -> dict:
     """Validate review submission mechanics without a live PR mutation."""
     return _evaluate_pr_review_submission(
@@ -3053,6 +3162,7 @@ def gitea_dry_run_pr_review(
         org=org,
         repo=repo,
         live=False,
+        worktree_path=worktree_path,
     )
 
 
@@ -3068,6 +3178,7 @@ def gitea_submit_pr_review(
     org: str | None = None,
     repo: str | None = None,
     final_review_decision_ready: bool = False,
+    worktree_path: str | None = None,
 ) -> dict:
     """Gated PR review mutation: comment findings, request changes, or approve.
 
@@ -3086,6 +3197,7 @@ def gitea_submit_pr_review(
         repo=repo,
         live=True,
         final_review_decision_ready=final_review_decision_ready,
+        worktree_path=worktree_path,
     )
 
 
@@ -4982,7 +5094,17 @@ def gitea_acquire_reviewer_pr_lease(
             "permission_report": _permission_block_report("gitea.pr.comment"),
         }
 
-    verify_preflight_purity(remote)
+    try:
+        resolved_worktree = _verify_reviewer_mutation_workspace(
+            remote, worktree=worktree
+        )
+    except RuntimeError as exc:
+        return {
+            "success": False,
+            "acquired": False,
+            "reasons": [str(exc)],
+            "permission_report": None,
+        }
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     profile = get_profile()
@@ -5000,7 +5122,7 @@ def gitea_acquire_reviewer_pr_lease(
         session_id=sid,
         repo=repo_label,
         issue_number=issue_number,
-        worktree=worktree,
+        worktree=resolved_worktree,
         candidate_head=candidate_head,
         target_branch=target_branch,
         target_branch_sha=target_branch_sha,
@@ -5032,7 +5154,7 @@ def gitea_acquire_reviewer_pr_lease(
         "session_id": sid,
         "reviewer_identity": identity,
         "profile": profile.get("profile_name"),
-        "worktree": worktree,
+        "worktree": resolved_worktree,
         "phase": "claimed",
         "candidate_head": candidate_head,
         "target_branch": target_branch,
@@ -5083,7 +5205,14 @@ def gitea_heartbeat_reviewer_pr_lease(
             ],
         }
 
-    verify_preflight_purity(remote)
+    try:
+        _verify_reviewer_mutation_workspace(
+            remote,
+            worktree=worktree,
+            worktree_path=session.get("worktree"),
+        )
+    except RuntimeError as exc:
+        return {"success": False, "posted": False, "reasons": [str(exc)]}
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     body = reviewer_pr_lease.format_lease_body(

reviewer_mutation_workspace.py

@@ -0,0 +1,157 @@
+"""Reviewer mutation workspace binding (#503).
+
+Reviewer mutation tools must validate the same branches/ worktree that
+read-only preflight and runtime_context inspect — not only the MCP server
+process root.
+"""
+
+from __future__ import annotations
+
+import os
+
+import author_mutation_worktree as amw
+
+ACTIVE_WORKTREE_ENV = amw.ACTIVE_WORKTREE_ENV
+
+
+def resolve_reviewer_mutation_workspace(
+    *,
+    worktree_path: str | None,
+    worktree: str | None,
+    process_project_root: str,
+    active_worktree_env: str | None = None,
+    session_lease_worktree: str | None = None,
+) -> str:
+    """Resolve the workspace inspected before reviewer mutations."""
+    for candidate in (
+        worktree_path,
+        worktree,
+        active_worktree_env,
+        session_lease_worktree,
+    ):
+        text = (candidate or "").strip()
+        if text:
+            return os.path.realpath(os.path.abspath(text))
+    return os.path.realpath(process_project_root)
+
+
+def assess_metadata_only_worktree_binding(
+    *,
+    declared_worktree_path: str | None,
+    mutation_workspace: str,
+    process_project_root: str,
+) -> dict:
+    """Fail closed when declared worktree_path would not redirect mutations."""
+    declared = (declared_worktree_path or "").strip()
+    process_root = os.path.realpath(process_project_root)
+    mutation_root = os.path.realpath(mutation_workspace)
+    if not declared:
+        return {"block": False, "reasons": [], "metadata_only": False}
+
+    declared_root = os.path.realpath(os.path.abspath(declared))
+    if declared_root == mutation_root:
+        return {"block": False, "reasons": [], "metadata_only": False}
+
+    if declared_root != process_root and mutation_root == process_root:
+        return {
+            "block": True,
+            "metadata_only": True,
+            "reasons": [
+                "worktree_path is metadata-only for reviewer mutations: preflight "
+                f"inspected '{declared_root}' but mutation tools would still "
+                f"validate MCP server process root '{process_root}'"
+            ],
+            "declared_worktree_path": declared_root,
+            "mutation_workspace": mutation_root,
+            "process_project_root": process_root,
+        }
+
+    return {"block": False, "reasons": [], "metadata_only": False}
+
+
+def format_stale_mcp_workspace_binding_error(
+    *,
+    process_project_root: str,
+    mutation_workspace: str,
+    declared_worktree_path: str | None = None,
+    metadata_only: bool = False,
+) -> str:
+    """Canonical handoff when reviewer mutations bind to the wrong workspace."""
+    process_root = os.path.realpath(process_project_root)
+    mutation_root = os.path.realpath(mutation_workspace)
+    declared = (declared_worktree_path or "").strip()
+    declared_root = (
+        os.path.realpath(os.path.abspath(declared)) if declared else None
+    )
+
+    if metadata_only and declared_root:
+        lead = (
+            "Reviewer mutation workspace mismatch (#503): worktree_path affected "
+            f"read-only preflight only ('{declared_root}') while mutation tools "
+            f"still inspect MCP server process root '{process_root}'."
+        )
+    elif mutation_root == process_root:
+        lead = (
+            "Reviewer mutation workspace mismatch (#503): mutation tools are "
+            f"bound to the stable control checkout '{process_root}' instead of "
+            "an active branches/ review worktree."
+        )
+    else:
+        lead = (
+            "Reviewer mutation workspace mismatch (#503): mutation workspace "
+            f"'{mutation_root}' does not match the declared active task worktree."
+        )
+
+    fixes = (
+        "Fix: relaunch or reconnect the MCP server from the branches/ review "
+        "worktree, set GITEA_ACTIVE_WORKTREE to that path, pass worktree_path "
+        "to reviewer mutation tools, or acquire a reviewer PR lease from the "
+        "same branches/ worktree so mutations inherit the validated workspace."
+    )
+    return f"{lead} {fixes}"
+
+
+def assess_reviewer_mutation_workspace(
+    *,
+    worktree_path: str | None,
+    worktree: str | None,
+    process_project_root: str,
+    active_worktree_env: str | None = None,
+    session_lease_worktree: str | None = None,
+    current_branch: str | None = None,
+) -> dict:
+    """Evaluate reviewer mutation workspace binding before preflight/mutation."""
+    mutation_workspace = resolve_reviewer_mutation_workspace(
+        worktree_path=worktree_path,
+        worktree=worktree,
+        process_project_root=process_project_root,
+        active_worktree_env=active_worktree_env,
+        session_lease_worktree=session_lease_worktree,
+    )
+    process_root = os.path.realpath(process_project_root)
+    metadata = assess_metadata_only_worktree_binding(
+        declared_worktree_path=worktree_path,
+        mutation_workspace=mutation_workspace,
+        process_project_root=process_root,
+    )
+    reasons = list(metadata.get("reasons") or [])
+    canonical_root = amw.resolve_canonical_repo_root(process_root, process_root)
+    branches = amw.assess_author_mutation_worktree(
+        workspace_path=mutation_workspace,
+        project_root=canonical_root,
+        current_branch=current_branch,
+    )
+    if branches["block"]:
+        reasons.extend(branches["reasons"])
+
+    block = bool(reasons)
+    return {
+        "block": block,
+        "reasons": reasons,
+        "mutation_workspace": mutation_workspace,
+        "process_project_root": process_root,
+        "canonical_repo_root": canonical_root,
+        "metadata_only": metadata.get("metadata_only", False),
+        "declared_worktree_path": metadata.get("declared_worktree_path"),
+        "under_branches": branches.get("under_branches", False),
+    }

tests/test_reviewer_mutation_workspace.py

@@ -0,0 +1,217 @@
+"""Tests for reviewer mutation workspace binding (#503)."""
+
+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 gitea_mcp_server as srv  # noqa: E402
+import reviewer_mutation_workspace as rmw  # noqa: E402
+import reviewer_pr_lease  # noqa: E402
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+if REPO_ROOT.parent.name == "branches":
+    CONTROL_ROOT = str(REPO_ROOT.parent.parent)
+else:
+    CONTROL_ROOT = str(REPO_ROOT)
+BRANCHES_WORKTREE = f"{CONTROL_ROOT}/branches/review-pr503-submit"
+MCP_PROCESS_ROOT = CONTROL_ROOT
+
+
+class TestReviewerMutationWorkspaceModule(unittest.TestCase):
+    def test_resolve_prefers_explicit_worktree_path(self):
+        resolved = rmw.resolve_reviewer_mutation_workspace(
+            worktree_path=BRANCHES_WORKTREE,
+            worktree="/ignored",
+            process_project_root=MCP_PROCESS_ROOT,
+            active_worktree_env="/env/path",
+            session_lease_worktree="/lease/path",
+        )
+        self.assertEqual(resolved, os.path.realpath(BRANCHES_WORKTREE))
+
+    def test_resolve_falls_back_to_session_lease(self):
+        resolved = rmw.resolve_reviewer_mutation_workspace(
+            worktree_path=None,
+            worktree=None,
+            process_project_root=MCP_PROCESS_ROOT,
+            session_lease_worktree=BRANCHES_WORKTREE,
+        )
+        self.assertEqual(resolved, os.path.realpath(BRANCHES_WORKTREE))
+
+    def test_metadata_only_binding_detected(self):
+        assessment = rmw.assess_metadata_only_worktree_binding(
+            declared_worktree_path=BRANCHES_WORKTREE,
+            mutation_workspace=MCP_PROCESS_ROOT,
+            process_project_root=MCP_PROCESS_ROOT,
+        )
+        self.assertTrue(assessment["block"])
+        self.assertTrue(assessment["metadata_only"])
+        self.assertIn("metadata-only", assessment["reasons"][0])
+
+    def test_canonical_error_mentions_fix_paths(self):
+        msg = rmw.format_stale_mcp_workspace_binding_error(
+            process_project_root=MCP_PROCESS_ROOT,
+            mutation_workspace=MCP_PROCESS_ROOT,
+            declared_worktree_path=BRANCHES_WORKTREE,
+            metadata_only=True,
+        )
+        self.assertIn("GITEA_ACTIVE_WORKTREE", msg)
+        self.assertIn("worktree_path", msg)
+        self.assertIn("branches/", msg)
+
+
+class TestReviewerMutationWorkspaceIntegration(unittest.TestCase):
+    def setUp(self):
+        reviewer_pr_lease.clear_session_lease()
+        self._saved = {
+            "whoami_called": srv._preflight_whoami_called,
+            "capability_called": srv._preflight_capability_called,
+            "resolved_role": srv._preflight_resolved_role,
+            "whoami_violation": srv._preflight_whoami_violation,
+            "whoami_files": list(srv._preflight_whoami_violation_files),
+            "capability_violation": srv._preflight_capability_violation,
+            "capability_files": list(srv._preflight_capability_violation_files),
+            "reviewer_files": list(srv._preflight_reviewer_violation_files),
+            "in_test": srv._preflight_in_test_mode,
+        }
+        srv._preflight_whoami_called = True
+        srv._preflight_capability_called = True
+        srv._preflight_resolved_role = "reviewer"
+        srv._preflight_whoami_violation = True
+        srv._preflight_whoami_violation_files = ["gitea_mcp_server.py"]
+        srv._preflight_capability_violation = False
+        srv._preflight_capability_violation_files = []
+        srv._preflight_reviewer_violation_files = []
+        srv._preflight_in_test_mode = lambda: False
+        self._env_patch = mock.patch.dict(os.environ, {"GITEA_TEST_PORCELAIN": ""}, clear=False)
+        self._env_patch.start()
+        os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
+        os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
+
+    def tearDown(self):
+        reviewer_pr_lease.clear_session_lease()
+        srv._preflight_whoami_called = self._saved["whoami_called"]
+        srv._preflight_capability_called = self._saved["capability_called"]
+        srv._preflight_resolved_role = self._saved["resolved_role"]
+        srv._preflight_whoami_violation = self._saved["whoami_violation"]
+        srv._preflight_whoami_violation_files = self._saved["whoami_files"]
+        srv._preflight_capability_violation = self._saved["capability_violation"]
+        srv._preflight_capability_violation_files = self._saved["capability_files"]
+        srv._preflight_reviewer_violation_files = self._saved["reviewer_files"]
+        srv._preflight_in_test_mode = self._saved["in_test"]
+        self._env_patch.stop()
+
+    @mock.patch("subprocess.run")
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    def test_reviewer_mutation_uses_declared_branches_worktree(
+        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):
+            resolved = srv._verify_reviewer_mutation_workspace(
+                "prgs", worktree_path=BRANCHES_WORKTREE
+            )
+        self.assertEqual(resolved, os.path.realpath(BRANCHES_WORKTREE))
+
+    def test_control_checkout_blocks_reviewer_mutations(self):
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            with self.assertRaises(RuntimeError) as ctx:
+                srv._verify_reviewer_mutation_workspace("prgs")
+        self.assertIn("stable control checkout", str(ctx.exception))
+        self.assertIn("GITEA_ACTIVE_WORKTREE", str(ctx.exception))
+
+    def test_preflight_and_mutation_agree_on_workspace(self):
+        os.environ["GITEA_ACTIVE_WORKTREE"] = BRANCHES_WORKTREE
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            with mock.patch.object(
+                srv,
+                "_get_git_root",
+                return_value=BRANCHES_WORKTREE,
+            ):
+                status = srv.assess_preflight_status(worktree_path=BRANCHES_WORKTREE)
+            resolved = srv._resolve_reviewer_mutation_worktree_path(
+                worktree_path=BRANCHES_WORKTREE
+            )
+        self.assertTrue(status["preflight_ready"])
+        self.assertEqual(
+            status["preflight_workspace"]["active_task_workspace_root"],
+            os.path.realpath(BRANCHES_WORKTREE),
+        )
+        self.assertEqual(resolved, os.path.realpath(BRANCHES_WORKTREE))
+
+    def test_metadata_only_worktree_path_blocks_preflight_ready(self):
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            status = srv.assess_preflight_status(worktree_path=BRANCHES_WORKTREE)
+        self.assertFalse(status["preflight_ready"])
+        self.assertTrue(
+            any("metadata-only" in reason for reason in status["preflight_block_reasons"])
+        )
+
+    @mock.patch("subprocess.run")
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    def test_dry_run_inherits_session_lease_worktree(
+        self, _exists, _isdir, mock_run
+    ):
+        mock_run.return_value = MagicMock(
+            returncode=0,
+            stdout=f"{CONTROL_ROOT}/.git\n",
+        )
+        reviewer_pr_lease.record_session_lease({
+            "pr_number": 503,
+            "session_id": "sess-503",
+            "worktree": BRANCHES_WORKTREE,
+        })
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            with mock.patch.object(
+                srv,
+                "gitea_check_pr_eligibility",
+                return_value={
+                    "eligible": False,
+                    "reasons": ["stub"],
+                    "authenticated_user": "reviewer",
+                    "profile_name": "prgs-reviewer",
+                    "pr_author": "author",
+                    "head_sha": "abc123",
+                },
+            ):
+                result = srv.gitea_dry_run_pr_review(
+                    pr_number=503,
+                    action="comment",
+                    remote="prgs",
+                )
+        self.assertNotIn("stable control checkout", " ".join(result.get("reasons") or []))
+        self.assertNotIn("metadata-only", " ".join(result.get("reasons") or []))
+
+    @mock.patch("subprocess.run")
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    def test_dry_run_fails_closed_from_control_checkout_without_binding(
+        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):
+            result = srv.gitea_dry_run_pr_review(
+                pr_number=503,
+                action="comment",
+                remote="prgs",
+            )
+        reasons = " ".join(result.get("reasons") or [])
+        self.assertIn("stable control checkout", reasons)
+
+
+if __name__ == "__main__":
+    unittest.main()