feat: isolate MCP workspace binding across role namespaces (Closes #510)

Namespace-scoped workspace resolution prevents GITEA_AUTHOR_WORKTREE from
poisoning reviewer, merger, and reconciler purity checks. Each role uses
its own worktree env var with actionable binding-source errors and safe
reconnect guidance. Preserves #274/#475 author and control-checkout guards.

gitea_mcp_server.py

@@ -165,7 +165,6 @@ _preflight_capability_called = False
 _preflight_whoami_violation = False
 _preflight_capability_violation = False
 _preflight_resolved_role = None
-_preflight_resolved_task: str | None = None
 _process_start_porcelain: str | None = None
 _preflight_whoami_baseline_porcelain: str | None = None
 _preflight_capability_baseline_porcelain: str | None = None
@@ -452,30 +451,11 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
     }
 
 
-def _clear_preflight_capability_state() -> None:
-    """Drop resolved capability proof (consumed by a mutation or fresh resolve)."""
-    global _preflight_capability_called, _preflight_capability_violation
-    global _preflight_resolved_task
-    global _preflight_capability_baseline_porcelain, _preflight_capability_violation_files
-    global _preflight_reviewer_violation_files
-
-    _preflight_capability_called = False
-    _preflight_capability_violation = False
-    _preflight_capability_violation_files = []
-    _preflight_capability_baseline_porcelain = None
-    _preflight_resolved_task = None
-    _preflight_reviewer_violation_files = []
-
-
-def record_preflight_check(
-    type_name: str,
-    resolved_role: str | None = None,
-    resolved_task: str | None = None,
-):
+def record_preflight_check(type_name: str, resolved_role: str | None = None):
     """Record a pre-flight check (whoami or capability) with session-scoped deltas."""
     global _preflight_whoami_called, _preflight_capability_called
     global _preflight_whoami_violation, _preflight_capability_violation
-    global _preflight_resolved_role, _preflight_resolved_task
+    global _preflight_resolved_role
     global _preflight_whoami_baseline_porcelain, _preflight_capability_baseline_porcelain
     global _preflight_whoami_violation_files, _preflight_capability_violation_files
     global _preflight_reviewer_violation_files
@@ -483,24 +463,13 @@ def record_preflight_check(
     current = _get_workspace_porcelain()
 
     if type_name == "whoami":
-        # Re-evaluate whoami violations instead of replaying sticky state (#252).
-        # Interleaved read-only whoami must not clear a valid capability proof (#469).
-        saved_capability = None
-        preserve_capability = (
-            _preflight_capability_called
-            and _preflight_whoami_called
-            and not _preflight_whoami_violation
-        )
-        if preserve_capability:
-            saved_capability = (
-                _preflight_capability_violation,
-                list(_preflight_capability_violation_files),
-                _preflight_capability_baseline_porcelain,
-                _preflight_resolved_role,
-                _preflight_resolved_task,
-            )
-        else:
-            _clear_preflight_capability_state()
+        # Fresh whoami restarts the capability step and re-evaluates violations
+        # instead of replaying a sticky record (#252).
+        _preflight_capability_called = False
+        _preflight_capability_violation = False
+        _preflight_capability_violation_files = []
+        _preflight_capability_baseline_porcelain = None
+        _preflight_reviewer_violation_files = []
 
         process_start = _ensure_process_start_porcelain()
         whoami_delta = _new_tracked_changes_since(process_start, current)
@@ -508,20 +477,6 @@ def record_preflight_check(
         _preflight_whoami_violation_files = whoami_delta
         _preflight_whoami_baseline_porcelain = current
         _preflight_whoami_called = True
-
-        if (
-            preserve_capability
-            and saved_capability is not None
-            and not whoami_delta
-        ):
-            (
-                _preflight_capability_violation,
-                _preflight_capability_violation_files,
-                _preflight_capability_baseline_porcelain,
-                _preflight_resolved_role,
-                _preflight_resolved_task,
-            ) = saved_capability
-            _preflight_capability_called = True
     elif type_name == "capability":
         baseline = _preflight_whoami_baseline_porcelain or ""
         capability_delta = _new_tracked_changes_since(baseline, current)
@@ -531,8 +486,6 @@ def record_preflight_check(
         _preflight_capability_called = True
         if resolved_role:
             _preflight_resolved_role = resolved_role
-        if resolved_task:
-            _preflight_resolved_task = resolved_task
 
 
 def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
@@ -559,11 +512,7 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
         )
 
 
-def verify_preflight_purity(
-    remote: str | None = None,
-    worktree_path: str | None = None,
-    task: str | None = None,
-):
+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
 
@@ -582,16 +531,6 @@ def verify_preflight_purity(
         raise RuntimeError(
             "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
         )
-    if (
-        task is not None
-        and _preflight_resolved_task is not None
-        and task != _preflight_resolved_task
-    ):
-        raise RuntimeError(
-            "Pre-flight task mismatch: "
-            f"resolved '{_preflight_resolved_task}' but mutation requires "
-            f"'{task}' (fail closed)"
-        )
 
     ctx = _resolve_namespace_mutation_context(worktree_path)
     workspace = ctx["workspace_path"]
@@ -652,7 +591,6 @@ def verify_preflight_purity(
                 )
 
     _enforce_branches_only_author_mutation(worktree_path)
-    _clear_preflight_capability_state()
 
 
 def _verify_role_mutation_workspace(
@@ -660,7 +598,6 @@ def _verify_role_mutation_workspace(
     *,
     worktree_path: str | None = None,
     worktree: str | None = None,
-    task: str | None = None,
 ) -> str:
     """Bind reviewer/merger mutations to the active namespace workspace (#510)."""
     role = _effective_workspace_role()
@@ -690,7 +627,7 @@ def _verify_role_mutation_workspace(
             )
         )
     resolved = assessment["mutation_workspace"]
-    verify_preflight_purity(remote, worktree_path=resolved, task=task)
+    verify_preflight_purity(remote, worktree_path=resolved)
     return resolved
 
 from mcp.server.fastmcp import FastMCP  # noqa: E402
@@ -1407,7 +1344,7 @@ def gitea_create_issue(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_issue")
+    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(
@@ -1540,7 +1477,7 @@ def gitea_lock_issue(
         )
     else:
         git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
-    verify_preflight_purity(remote, worktree_path=resolved_worktree, task="lock_issue")
+    verify_preflight_purity(remote, worktree_path=resolved_worktree)
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
         worktree_path=resolved_worktree,
         current_branch=git_state.get("current_branch"),
@@ -1764,7 +1701,7 @@ def gitea_create_pr(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr")
+    verify_preflight_purity(remote, worktree_path=worktree_path)
     h, o, r = _resolve(remote, host, org, repo)
 
     # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
@@ -2831,9 +2768,7 @@ def _evaluate_pr_review_submission(
     worktree_path: str | None = None,
 ) -> dict:
     """Shared gate chain for live submit and dry-run review tools."""
-    _verify_role_mutation_workspace(
-        remote, worktree_path=worktree_path, task="review_pr"
-    )
+    _verify_role_mutation_workspace(remote, worktree_path=worktree_path)
     action = (action or "").strip().lower()
     result = {
         "requested_action": action,
@@ -3344,12 +3279,12 @@ def gitea_edit_pr(
     if not payload:
         raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
 
-    closing = payload.get("state") == "closed"
-    verify_preflight_purity(remote, task="close_pr" if closing else None)
+    verify_preflight_purity(remote)
 
     # PR closure is a first-class capability, distinct from retitling or
     # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
     # never touches the network.
+    closing = payload.get("state") == "closed"
     if closing:
         gate_reasons = _profile_operation_gate("gitea.pr.close")
         if gate_reasons:
@@ -3594,7 +3529,7 @@ def gitea_commit_files(
             branch="",
         )
 
-    verify_preflight_purity(remote, task="commit_files")
+    verify_preflight_purity(remote)
     processed_files, source_proofs = _prepare_commit_payload_files(files)
 
     h, o, r = _resolve(remote, host, org, repo)
@@ -3700,9 +3635,7 @@ def gitea_merge_pr(
         reasons/gates passed or blocked, and merge result / merge commit if
         available. Never secrets.
     """
-    _verify_role_mutation_workspace(
-        remote, worktree_path=worktree_path, task="merge_pr"
-    )
+    _verify_role_mutation_workspace(remote, worktree_path=worktree_path)
     do = (do or "").strip().lower()
     result = {
         "performed": False,
@@ -4231,7 +4164,7 @@ def gitea_delete_branch(
             "permission_report": _permission_block_report("gitea.branch.delete"),
         }
 
-    verify_preflight_purity(remote, task="delete_branch")
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     import urllib.parse
@@ -4340,7 +4273,7 @@ def gitea_reconcile_merged_cleanups(
         report["executed"] = False
         return {"success": True, "performed": False, **report}
 
-    verify_preflight_purity(remote, task="reconcile_merged_cleanups")
+    verify_preflight_purity(remote)
     actions: list[dict] = []
     for entry in report.get("entries") or []:
         head_branch = entry.get("head_branch") or ""
@@ -4654,7 +4587,7 @@ def gitea_reconcile_already_landed_pr(
         )
         return result
 
-    verify_preflight_purity(remote, task="reconcile_already_landed_pr")
+    verify_preflight_purity(remote)
 
     if post_comment and comment_body.strip():
         comment_block = _profile_operation_gate("gitea.pr.comment")
@@ -4757,7 +4690,7 @@ def gitea_close_issue(
         task_capability_map.required_permission("close_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="close_issue")
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
@@ -5185,7 +5118,7 @@ def gitea_acquire_reviewer_pr_lease(
             "permission_report": _permission_block_report("gitea.pr.comment"),
         }
 
-    _verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr")
+    _verify_role_mutation_workspace(remote, worktree=worktree)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     profile = get_profile()
@@ -5286,7 +5219,7 @@ def gitea_heartbeat_reviewer_pr_lease(
             ],
         }
 
-    verify_preflight_purity(remote, task="review_pr")
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     body = reviewer_pr_lease.format_lease_body(
@@ -5462,7 +5395,7 @@ def gitea_create_issue_comment(
         (permission blocks also carry a structured 'permission_report',
         #142).
     """
-    verify_preflight_purity(remote, task="comment_issue")
+    verify_preflight_purity(remote)
     gate_reasons = _profile_operation_gate("gitea.issue.comment")
     reasons = list(gate_reasons)
     if not (body or "").strip():
@@ -6996,7 +6929,7 @@ def gitea_mark_issue(
         task_capability_map.required_permission("mark_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="mark_issue")
+    verify_preflight_purity(remote, worktree_path=worktree_path)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -7074,7 +7007,7 @@ def gitea_post_heartbeat(
         task_capability_map.required_permission("post_heartbeat"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="post_heartbeat")
+    verify_preflight_purity(remote)
     active_profile = profile or get_profile().get("profile_name")
     body = issue_claim_heartbeat.format_heartbeat_body(
         kind="progress",
@@ -7317,7 +7250,7 @@ def gitea_cleanup_stale_claims(
         task_capability_map.required_permission("cleanup_stale_claims"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="cleanup_stale_claims")
+    verify_preflight_purity(remote)
 
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -7471,7 +7404,7 @@ def gitea_set_issue_labels(
         task_capability_map.required_permission("set_issue_labels"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="set_issue_labels")
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -7708,7 +7641,7 @@ def gitea_resolve_task_capability(
                 "exact_safe_next_action": next_safe_action,
             }
 
-    record_preflight_check("capability", required_role, resolved_task=task)
+    record_preflight_check("capability", required_role)
 
     # Try automatic dispatch switching
     _ensure_matching_profile(required_permission, required_role, remote, host)

tests/test_preflight_read_survival.py

@@ -1,106 +0,0 @@
-"""#469: capability preflight survives interleaved read-only whoami calls."""
-
-import os
-import unittest
-
-import gitea_mcp_server as mcp_server
-
-
-class TestPreflightReadSurvival(unittest.TestCase):
-    def setUp(self):
-        self.orig_whoami = mcp_server._preflight_whoami_called
-        self.orig_capability = mcp_server._preflight_capability_called
-        self.orig_whoami_violation = mcp_server._preflight_whoami_violation
-        self.orig_capability_violation = mcp_server._preflight_capability_violation
-        self.orig_resolved_role = mcp_server._preflight_resolved_role
-        self.orig_resolved_task = mcp_server._preflight_resolved_task
-        self.orig_process_start = mcp_server._process_start_porcelain
-        self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
-        self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
-        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
-            if key in os.environ:
-                del os.environ[key]
-        os.environ["GITEA_TEST_PORCELAIN"] = ""
-        mcp_server._preflight_whoami_called = False
-        mcp_server._preflight_capability_called = False
-        mcp_server._preflight_whoami_violation = False
-        mcp_server._preflight_capability_violation = False
-        mcp_server._preflight_resolved_role = None
-        mcp_server._preflight_resolved_task = None
-        mcp_server._process_start_porcelain = ""
-        mcp_server._preflight_whoami_baseline_porcelain = None
-        mcp_server._preflight_capability_baseline_porcelain = None
-
-    def tearDown(self):
-        mcp_server._preflight_whoami_called = self.orig_whoami
-        mcp_server._preflight_capability_called = self.orig_capability
-        mcp_server._preflight_whoami_violation = self.orig_whoami_violation
-        mcp_server._preflight_capability_violation = self.orig_capability_violation
-        mcp_server._preflight_resolved_role = self.orig_resolved_role
-        mcp_server._preflight_resolved_task = self.orig_resolved_task
-        mcp_server._process_start_porcelain = self.orig_process_start
-        mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
-        mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
-        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
-            if key in os.environ:
-                del os.environ[key]
-
-    def test_interleaved_whoami_preserves_capability(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        self.assertTrue(mcp_server._preflight_capability_called)
-        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
-
-        mcp_server.record_preflight_check("whoami")
-        self.assertTrue(mcp_server._preflight_capability_called)
-        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
-
-        mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertFalse(mcp_server._preflight_capability_called)
-
-    def test_missing_capability_still_fails_closed(self):
-        mcp_server.record_preflight_check("whoami")
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertIn("has not been resolved", str(ctx.exception))
-
-    def test_task_mismatch_fails_closed(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="author", resolved_task="create_issue"
-        )
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertIn("task mismatch", str(ctx.exception))
-
-    def test_capability_consumed_after_mutation_gate(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="author", resolved_task="create_issue"
-        )
-        mcp_server.verify_preflight_purity(task="create_issue")
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="create_issue")
-        self.assertIn("has not been resolved", str(ctx.exception))
-
-    def test_whoami_recovery_after_violation_clears_capability(self):
-        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
-        mcp_server.record_preflight_check("whoami")
-        self.assertTrue(mcp_server._preflight_whoami_violation)
-
-        del os.environ["GITEA_TEST_FORCE_DIRTY"]
-        os.environ["GITEA_TEST_PORCELAIN"] = ""
-        mcp_server.record_preflight_check("whoami")
-        self.assertFalse(mcp_server._preflight_whoami_violation)
-        self.assertFalse(mcp_server._preflight_capability_called)
-
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reviewer", resolved_task="review_pr"
-        )
-        mcp_server.verify_preflight_purity(task="review_pr")
-
-
-if __name__ == "__main__":
-    unittest.main()