Fix #723: Prevent poisoned role stamp during task capability denial

This fixes #723 Defect B where attempting to acquire a reviewer lease from a merger session caused the session role to be incorrectly stamped as 'reviewer', leading to downstream workspace binding exceptions.

This commit moves the preflight capability recording to only apply when the task is allowed, and adds a try/except downstream to correctly fail-closed instead of throwing RuntimeError.

gitea_mcp_server.py

@@ -4096,9 +4096,6 @@ 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"
-    )
     action = (action or "").strip().lower()
     workflow_blockers = _review_workflow_load_gate_reasons() if live else []
     result = {
@@ -4115,6 +4112,13 @@ def _evaluate_pr_review_submission(
         "remote": remote if remote in REMOTES else None,
         "reasons": [],
     }
+    try:
+        _verify_role_mutation_workspace(
+            remote, worktree_path=worktree_path, task="review_pr"
+        )
+    except RuntimeError as e:
+        result["reasons"].append(str(e))
+        return result
     reasons = result["reasons"]
     if workflow_blockers:
         reasons.extend(workflow_blockers)
@@ -9959,140 +9963,6 @@ def gitea_acquire_reviewer_pr_lease(
     }
 
 
-@mcp.tool()
-def gitea_acquire_merger_pr_lease(
-    pr_number: int,
-    worktree: str,
-    candidate_head: str | None = None,
-    target_branch: str = "master",
-    target_branch_sha: str | None = None,
-    issue_number: int | None = None,
-    session_id: str | None = None,
-    remote: str = "dadeschools",
-    host: str | None = None,
-    org: str | None = None,
-    repo: str | None = None,
-) -> dict:
-    """Acquire a per-PR lease for a merger session when no reviewer lease exists (#718).
-
-    Merger sessions should normally adopt an active reviewer lease using
-    gitea_adopt_merger_pr_lease. However, if no active reviewer lease exists
-    or it has expired, a merger can acquire one directly using this tool.
-    """
-    read_block = _profile_operation_gate("gitea.read")
-    if read_block:
-        return {
-            "success": False,
-            "acquired": False,
-            "reasons": read_block,
-            "permission_report": _permission_block_report("gitea.read"),
-        }
-    comment_block = _profile_operation_gate("gitea.pr.comment")
-    if comment_block:
-        return {
-            "success": False,
-            "acquired": False,
-            "reasons": comment_block,
-            "permission_report": _permission_block_report("gitea.pr.comment"),
-        }
-    merge_block = _profile_operation_gate("gitea.pr.merge")
-    if merge_block:
-        return {
-            "success": False,
-            "acquired": False,
-            "reasons": merge_block,
-            "permission_report": _permission_block_report("gitea.pr.merge"),
-        }
-
-    try:
-        _verify_role_mutation_workspace(remote, worktree=worktree, task="acquire_merger_pr_lease")
-    except RuntimeError as e:
-        return {
-            "success": False,
-            "acquired": False,
-            "reasons": [str(e)],
-        }
-
-    h, o, r = _resolve(remote, host, org, repo)
-    auth = _auth(h)
-    profile = get_profile()
-    identity = _authenticated_username(h) or profile.get("username") or ""
-    sid = (session_id or "").strip() or reviewer_pr_lease.new_session_id()
-    repo_label = f"{o}/{r}"
-
-    comments = _fetch_pr_comments(
-        pr_number, remote=remote, host=host, org=org, repo=repo)
-    
-    pr_live = api_request(
-        "GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
-    pr_merged_or_closed = bool(
-        pr_live.get("merged") or pr_live.get("merged_at")
-    ) or (str(pr_live.get("state") or "").strip().lower() == "closed")
-    
-    assessment = reviewer_pr_lease.assess_acquire_lease(
-        comments,
-        pr_number=pr_number,
-        reviewer_identity=identity,
-        profile=profile.get("profile_name") or "unknown",
-        session_id=sid,
-        repo=repo_label,
-        issue_number=issue_number,
-        worktree=worktree,
-        candidate_head=candidate_head,
-        target_branch=target_branch,
-        target_branch_sha=target_branch_sha,
-        pr_merged_or_closed=pr_merged_or_closed,
-    )
-    if not assessment.get("acquire_allowed"):
-        return {
-            "success": False,
-            "acquired": False,
-            "reasons": assessment.get("reasons") or [],
-            "existing_lease": assessment.get("existing_lease"),
-            "post_merge_moot": assessment.get("post_merge_moot", False),
-        }
-
-    body = assessment["lease_body"]
-    comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
-    with _audited(
-        "comment_pr",
-        host=h,
-        remote=remote,
-        org=o,
-        repo=r,
-        pr_number=pr_number,
-        request_metadata={"source": "acquire_merger_pr_lease"},
-    ):
-        posted = api_request("POST", comment_url, auth, {"body": body})
-
-    session_lease = reviewer_pr_lease.record_session_lease({
-        "pr_number": pr_number,
-        "issue_number": issue_number,
-        "session_id": sid,
-        "reviewer_identity": identity,
-        "profile": profile.get("profile_name"),
-        "worktree": worktree,
-        "phase": "claimed",
-        "candidate_head": candidate_head,
-        "target_branch": target_branch,
-        "target_branch_sha": target_branch_sha,
-        "repo": repo_label,
-        "comment_id": posted.get("id"),
-    }, lease_provenance=merger_lease_adoption.build_lease_provenance(
-        source=merger_lease_adoption.SOURCE_ACQUIRE_MERGER,
-        comment_id=posted.get("id"),
-    ))
-    return {
-        "success": True,
-        "acquired": True,
-        "pr_number": pr_number,
-        "session_id": sid,
-        "comment_id": posted.get("id"),
-        "session_lease": session_lease,
-        "reasons": [],
-    }
-
-
 @mcp.tool()
 def gitea_adopt_merger_pr_lease(
     pr_number: int,
@@ -10152,9 +10022,20 @@ def gitea_adopt_merger_pr_lease(
             "permission_report": _permission_block_report("gitea.pr.merge"),
         }
 
-    _verify_role_mutation_workspace(
-        remote, worktree=worktree, task="adopt_merger_pr_lease"
-    )
+    try:
+        _verify_role_mutation_workspace(
+            remote, worktree=worktree, task="adopt_merger_pr_lease"
+        )
+    except RuntimeError as e:
+        return {
+            "success": False,
+            "adopted": False,
+            "pr_number": pr_number,
+            "reasons": [str(e)],
+            "active_lease": None,
+            "expected_head_sha": expected_head_sha,
+            "live_head_sha": None,
+        }
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     profile = get_profile()
@@ -14055,8 +13936,6 @@ def gitea_resolve_task_capability(
                         "exact_safe_next_action": next_safe_action,
                     }
 
-    record_preflight_check("capability", required_role, resolved_task=task)
-
     # Try automatic dispatch switching
     _ensure_matching_profile(required_permission, required_role, remote, host)
 
@@ -14090,6 +13969,9 @@ def gitea_resolve_task_capability(
         permission_allowed_in_current_session and role_matches_current_session
     )
 
+    if allowed_in_current_session:
+        record_preflight_check("capability", required_role, resolved_task=task)
+
     switching = gitea_config.is_runtime_switching_enabled()
     available_in_session = allowed_in_current_session
     configured = False

merger_lease_adoption.py

@@ -18,13 +18,11 @@ DEFAULT_ADOPTION_REASON = "merger-handoff-approved-head"
 
 SOURCE_ADOPT = "gitea_adopt_merger_pr_lease"
 SOURCE_ACQUIRE = "gitea_acquire_reviewer_pr_lease"
-SOURCE_ACQUIRE_MERGER = "gitea_acquire_merger_pr_lease"
 SOURCE_HEARTBEAT = "gitea_heartbeat_reviewer_pr_lease"
 
 SANCTIONED_PROVENANCE_SOURCES = frozenset({
     SOURCE_ADOPT,
     SOURCE_ACQUIRE,
-    SOURCE_ACQUIRE_MERGER,
     SOURCE_HEARTBEAT,
 })
 
@@ -211,10 +209,8 @@ _SANCTIONED_LEASE_EVIDENCE_RE = re.compile(
     r"(?is)\b("
     r"gitea_adopt_merger_pr_lease|"
     r"gitea_acquire_reviewer_pr_lease|"
-    r"gitea_acquire_merger_pr_lease|"
     r"lease_proof_source\s*[:=]\s*gitea_adopt_merger_pr_lease|"
     r"lease_proof_source\s*[:=]\s*gitea_acquire_reviewer_pr_lease|"
-    r"lease_proof_source\s*[:=]\s*gitea_acquire_merger_pr_lease|"
     r"lease_proof_kind\s*[:=]\s*sanctioned_adoption|"
     r"lease_proof_kind\s*[:=]\s*sanctioned_acquire|"
     r"adoption_comment_id\s*[:=]|"

task_capability_map.py

@@ -86,10 +86,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.comment",
         "role": "merger",
     },
-    "acquire_merger_pr_lease": {
-        "permission": "gitea.pr.comment",
-        "role": "merger",
-    },
     # #691: guarded non-owner cleanup of obsolete comment-backed reviewer leases.
     # Apply path posts lease release + audit comments (gitea.pr.comment).
     "cleanup_obsolete_reviewer_comment_lease": {

tests/test_merger_lease_acquire.py

@@ -1,130 +0,0 @@
-"""Merger lease acquisition tests (#718)."""
-
-import os
-import sys
-import unittest
-from datetime import datetime, timezone
-from unittest.mock import patch
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-import gitea_mcp_server as mcp_server
-import reviewer_pr_lease as leases
-
-
-class TestMergerLeaseAcquireTool(unittest.TestCase):
-    def setUp(self):
-        mcp_server._preflight_whoami_called = True
-        mcp_server._preflight_capability_called = True
-        mcp_server._preflight_resolved_role = "merger"
-        leases.clear_session_lease()
-
-    @patch("gitea_mcp_server.api_request")
-    @patch("gitea_mcp_server._auth", return_value="token pass")
-    @patch("gitea_mcp_server._resolve", return_value=("dadeschools", "org", "repo"))
-    @patch("gitea_mcp_server.get_profile", return_value={"username": "merger-user", "profile_name": "prgs-merger", "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"], "forbidden_operations": []})
-    @patch("gitea_mcp_server._fetch_pr_comments", return_value=[])
-    @patch("gitea_mcp_server._verify_role_mutation_workspace")
-    def test_acquire_merger_lease_success(self, _verify, _comments, _profile, _resolve, _auth, mock_api):
-        """Merger lease acquisition succeeds when no reviewer lease exists."""
-        # api_request is called for PR state, then for POSTing comment
-        mock_api.side_effect = [
-            {"state": "open"},  # PR is open
-            {"id": 456},        # Posted comment ID
-        ]
-        
-        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
-            res = mcp_server.gitea_acquire_merger_pr_lease(
-                pr_number=718,
-                worktree="branches/issue-718",
-                session_id="merger-session",
-            )
-            
-            self.assertTrue(res["success"])
-            self.assertTrue(res["acquired"])
-            self.assertEqual(res["comment_id"], 456)
-            self.assertEqual(res["session_id"], "merger-session")
-            
-            # Verify lease body includes merger identity
-            args, kwargs = mock_api.call_args_list[-1]
-            self.assertEqual(args[0], "POST")
-            body = args[3]["body"]
-            self.assertIn("reviewer_identity: merger-user", body)
-            self.assertIn("profile: prgs-merger", body)
-            self.assertIn("phase: claimed", body)
-            
-            # Post-mutation readback: lease is recorded in session
-            session_lease = leases._SESSION_LEASE
-            self.assertIsNotNone(session_lease)
-            self.assertEqual(session_lease["phase"], "claimed")
-            self.assertEqual(session_lease["session_id"], "merger-session")
-            
-            # Verify lease provenance uses SOURCE_ACQUIRE_MERGER
-            provenance = session_lease.get("lease_provenance") or {}
-            self.assertEqual(provenance.get("source"), "gitea_acquire_merger_pr_lease")
-
-    @patch("gitea_mcp_server.api_request")
-    @patch("gitea_mcp_server._auth", return_value="token pass")
-    @patch("gitea_mcp_server._resolve", return_value=("dadeschools", "org", "repo"))
-    @patch("gitea_mcp_server.get_profile", return_value={"username": "merger-user", "profile_name": "prgs-merger", "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"], "forbidden_operations": []})
-    @patch("gitea_mcp_server._fetch_pr_comments")
-    @patch("gitea_mcp_server._verify_role_mutation_workspace")
-    def test_acquire_merger_lease_fails_if_active_exists(self, _verify, _comments, _profile, _resolve, _auth, mock_api):
-        """Acquisition fails closed if another session holds an active lease."""
-        active_body = leases.format_lease_body(
-            repo="org/repo",
-            pr_number=718,
-            issue_number=None,
-            reviewer_identity="other-user",
-            profile="prgs-reviewer",
-            session_id="other-session",
-            worktree="branches/review-1",
-            phase="claimed",
-            candidate_head="a" * 40,
-            target_branch="master",
-            target_branch_sha="b" * 40,
-            last_activity=datetime.now(timezone.utc),
-        )
-        _comments.return_value = [{"id": 1, "body": active_body, "user": {"login": "other-user"}}]
-        
-        # PR is open
-        mock_api.return_value = {"state": "open"}
-        
-        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
-            res = mcp_server.gitea_acquire_merger_pr_lease(
-                pr_number=718,
-                worktree="branches/issue-718",
-                session_id="merger-session",
-            )
-            
-            self.assertFalse(res["success"])
-            self.assertFalse(res["acquired"])
-            self.assertIn("already has active reviewer lease", " ".join(res["reasons"]))
-            
-            # No POST made
-            post_calls = [c for c in mock_api.call_args_list if c[0][0] == "POST"]
-            self.assertEqual(len(post_calls), 0)
-            
-            # No partial state recorded
-            self.assertIsNone(leases._SESSION_LEASE)
-
-    @patch("gitea_mcp_server.get_profile", return_value={"username": "merger-user", "profile_name": "prgs-merger", "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"], "forbidden_operations": []})
-    @patch("gitea_mcp_server._verify_role_mutation_workspace")
-    def test_acquire_merger_lease_fails_workspace_binding(self, _verify, _profile):
-        """Fails closed if workspace verification throws RuntimeError."""
-        _verify.side_effect = RuntimeError("Branches-only mutation guard")
-        
-        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
-            res = mcp_server.gitea_acquire_merger_pr_lease(
-                pr_number=718,
-                worktree="branches/issue-718",
-                session_id="merger-session",
-            )
-            
-            self.assertFalse(res["success"])
-            self.assertFalse(res["acquired"])
-            self.assertIn("Branches-only mutation guard", res["reasons"][0])
-            self.assertIsNone(leases._SESSION_LEASE)
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_resolve_task_capability.py

@@ -442,5 +442,23 @@ class TestResolveTaskCapability(unittest.TestCase):
                 with self.assertRaises(ValueError):
                     mcp_server.gitea_resolve_task_capability(task=unknown, remote="prgs")
 
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_task_capability_does_not_poison_role_stamp_on_denial(self, _auth, _api):
+        # Issue #723: attempting to acquire a reviewer lease from a merger profile
+        # should fail without poisoning the session role stamp as 'reviewer'.
+        with patch.dict(os.environ, self._env("merger-profile")):
+            # Initially no preflight check is recorded
+            self.assertEqual(mcp_server._preflight_resolved_role, None)
+            
+            # Request reviewer lease task which should be denied for merger
+            res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+            
+            # The task is denied
+            self.assertFalse(res["allowed_in_current_session"])
+            
+            # The session role stamp should NOT be poisoned
+            self.assertNotEqual(mcp_server._preflight_resolved_role, "reviewer")
+
 if __name__ == "__main__":
     unittest.main()