Fix #718: Implement gitea_acquire_merger_pr_lease

This implements the native merger lease acquisition tool to allow a merger session to acquire its own lease when a reviewer lease does not exist or has expired, unblocking #679 adoption.

gitea_mcp_server.py

@@ -9959,6 +9959,140 @@ 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,

merger_lease_adoption.py

@@ -18,11 +18,13 @@ 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,
 })
 
@@ -209,8 +211,10 @@ _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,6 +86,10 @@ 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

@@ -0,0 +1,130 @@
+"""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()