From 1421fa8568518e7149467a50aa6ea712b0e7d872 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 17 Jul 2026 01:18:25 -0400 Subject: [PATCH 1/4] 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 | 134 +++++++++++++++++++++++++++++ merger_lease_adoption.py | 4 + task_capability_map.py | 4 + tests/test_merger_lease_acquire.py | 130 ++++++++++++++++++++++++++++ 4 files changed, 272 insertions(+) create mode 100644 tests/test_merger_lease_acquire.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 6fb55cd..2141b11 100644 --- a/gitea_mcp_server.py +++ b/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, diff --git a/merger_lease_adoption.py b/merger_lease_adoption.py index 821779a..46ae926 100644 --- a/merger_lease_adoption.py +++ b/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*[:=]|" diff --git a/task_capability_map.py b/task_capability_map.py index 02826af..e270009 100644 --- a/task_capability_map.py +++ b/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": { diff --git a/tests/test_merger_lease_acquire.py b/tests/test_merger_lease_acquire.py new file mode 100644 index 0000000..2ff0488 --- /dev/null +++ b/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() From 4f894009f6c49b55d706f0673212e07c1375f7cc Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 17 Jul 2026 10:09:36 -0400 Subject: [PATCH 2/4] fix(mcp): complete merger lease acquire + structured capability resolve (#718, #723) Harden gitea_acquire_merger_pr_lease with required candidate_head, live-head match, and fail-closed POST handling. Add capability-map aliases for reviewer and merger lease acquire/adopt. Make unknown tasks return structured unknown_task (no ValueError escape). Expand tests for roles, head scoping, aliases, and unknown_task. Closes #718. Addresses #723 resolver unknown_task and lease task mapping. --- gitea_mcp_server.py | 104 ++++++- task_capability_map.py | 16 ++ tests/test_merger_lease_acquire.py | 387 +++++++++++++++++++++----- tests/test_resolve_task_capability.py | 18 +- 4 files changed, 451 insertions(+), 74 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 2141b11..74982f3 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -10004,8 +10004,21 @@ def gitea_acquire_merger_pr_lease( "permission_report": _permission_block_report("gitea.pr.merge"), } + head_pin = (candidate_head or "").strip() + if not head_pin: + return { + "success": False, + "acquired": False, + "reasons": [ + "candidate_head is required for merger lease acquisition " + "(exact-head scoping; fail closed)" + ], + } + try: - _verify_role_mutation_workspace(remote, worktree=worktree, task="acquire_merger_pr_lease") + _verify_role_mutation_workspace( + remote, worktree=worktree, task="acquire_merger_pr_lease" + ) except RuntimeError as e: return { "success": False, @@ -10022,13 +10035,31 @@ def gitea_acquire_merger_pr_lease( 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 {} + live_head = ( + (pr_live.get("head") or {}).get("sha") + or pr_live.get("head_sha") + or "" + ) + live_head = str(live_head).strip() + if live_head and live_head != head_pin: + return { + "success": False, + "acquired": False, + "reasons": [ + f"candidate_head {head_pin[:12]}… does not match live PR head " + f"{live_head[:12]}… (exact-head scoping; fail closed)" + ], + "live_head": live_head, + "candidate_head": head_pin, + } + 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, @@ -10038,7 +10069,7 @@ def gitea_acquire_merger_pr_lease( repo=repo_label, issue_number=issue_number, worktree=worktree, - candidate_head=candidate_head, + candidate_head=head_pin, target_branch=target_branch, target_branch_sha=target_branch_sha, pr_merged_or_closed=pr_merged_or_closed, @@ -10065,6 +10096,16 @@ def gitea_acquire_merger_pr_lease( ): posted = api_request("POST", comment_url, auth, {"body": body}) + if not isinstance(posted, dict) or not posted.get("id"): + return { + "success": False, + "acquired": False, + "reasons": [ + "lease comment POST failed or returned no comment id " + "(no partial session lease recorded)" + ], + } + session_lease = reviewer_pr_lease.record_session_lease({ "pr_number": pr_number, "issue_number": issue_number, @@ -10073,7 +10114,7 @@ def gitea_acquire_merger_pr_lease( "profile": profile.get("profile_name"), "worktree": worktree, "phase": "claimed", - "candidate_head": candidate_head, + "candidate_head": head_pin, "target_branch": target_branch, "target_branch_sha": target_branch_sha, "repo": repo_label, @@ -10089,6 +10130,8 @@ def gitea_acquire_merger_pr_lease( "session_id": sid, "comment_id": posted.get("id"), "session_lease": session_lease, + "candidate_head": head_pin, + "repo": repo_label, "reasons": [], } @@ -13951,11 +13994,56 @@ def gitea_resolve_task_capability( TASK_MAP = task_capability_map.TASK_CAPABILITY_MAP if task not in TASK_MAP: - raise ValueError(f"Unknown task/action: '{task}' (fail closed)") + # #723: structured fail-closed unknown_task (never raise into internal_error). + profile = get_profile() + h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None) + username = _authenticated_username(h) if h else None + active_role_kind = _profile_role_kind(profile) + switching = gitea_config.is_runtime_switching_enabled() + result = { + "requested_task": task, + "required_operation_permission": "unknown", + "required_role_kind": "unknown", + "active_profile": profile.get("profile_name", "unknown"), + "active_identity": username, + "active_role_kind": active_role_kind, + "active_profile_allowed_operations": profile.get("allowed_operations") or [], + "active_profile_permission_allowed": False, + "allowed_in_current_session": False, + "available_in_session": False, + "configured": False, + "restart_required": False, + "stop_required": True, + "task_role_guidance": [ + f"STOP: Unknown task/action: '{task}' (fail closed)" + ], + "matching_configured_profile": [], + "runtime_switching_supported": switching, + "different_mcp_namespace_required": False, + "exact_safe_next_action": ( + f"None; task '{task}' is unrecognized by the capability map." + ), + "reason": f"Unknown task/action: '{task}' (fail closed)", + "reason_code": "unknown_task", + "mutation_performed": False, + } + role_session_router.sync_route_from_capability(result) + was_terminal = capability_stop_terminal.is_active() + terminal = capability_stop_terminal.sync_from_capability_result(result) + if terminal: + result["terminal_mode"] = True + result["terminal_report"] = ( + capability_stop_terminal.build_terminal_report(result) + ) + elif was_terminal: + result["cleared_stale_denial"] = True + return result required_permission = task_capability_map.required_permission(task) required_role = task_capability_map.required_role(task) role_exclusive_tasks = { + "acquire_reviewer_pr_lease", + "gitea_acquire_reviewer_pr_lease", "review_pr", "approve_pr", "request_changes_pr", @@ -13963,6 +14051,10 @@ def gitea_resolve_task_capability( "pr_queue_cleanup", "pr-queue-cleanup", "merge_pr", + "acquire_merger_pr_lease", + "gitea_acquire_merger_pr_lease", + "adopt_merger_pr_lease", + "gitea_adopt_merger_pr_lease", "create_branch", "push_branch", "create_pr", diff --git a/task_capability_map.py b/task_capability_map.py index e270009..d9a4789 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -64,6 +64,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.branch.push", "role": "author", }, + "acquire_reviewer_pr_lease": { + "permission": "gitea.pr.comment", + "role": "reviewer", + }, + "gitea_acquire_reviewer_pr_lease": { + "permission": "gitea.pr.comment", + "role": "reviewer", + }, "review_pr": { "permission": "gitea.pr.review", "role": "reviewer", @@ -86,10 +94,18 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.pr.comment", "role": "merger", }, + "gitea_adopt_merger_pr_lease": { + "permission": "gitea.pr.comment", + "role": "merger", + }, "acquire_merger_pr_lease": { "permission": "gitea.pr.comment", "role": "merger", }, + "gitea_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": { diff --git a/tests/test_merger_lease_acquire.py b/tests/test_merger_lease_acquire.py index 2ff0488..3c9da2a 100644 --- a/tests/test_merger_lease_acquire.py +++ b/tests/test_merger_lease_acquire.py @@ -1,15 +1,22 @@ -"""Merger lease acquisition tests (#718).""" +"""Merger lease acquisition + capability resolution tests (#718, #723).""" + +from __future__ import annotations import os import sys import unittest from datetime import datetime, timezone -from unittest.mock import patch +from unittest.mock import MagicMock, 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 +import task_capability_map as tcm + + +HEAD = "a" * 40 +LIVE_HEAD = HEAD class TestMergerLeaseAcquireTool(unittest.TestCase): @@ -17,62 +24,80 @@ class TestMergerLeaseAcquireTool(unittest.TestCase): mcp_server._preflight_whoami_called = True mcp_server._preflight_capability_called = True mcp_server._preflight_resolved_role = "merger" + mcp_server._preflight_resolved_task = "acquire_merger_pr_lease" leases.clear_session_lease() + def tearDown(self): + leases.clear_session_lease() + + def _merger_profile(self, **extra): + p = { + "username": "merger-user", + "profile_name": "prgs-merger", + "role": "merger", + "allowed_operations": [ + "gitea.read", + "gitea.pr.comment", + "gitea.pr.merge", + ], + "forbidden_operations": [ + "gitea.pr.approve", + "gitea.pr.review", + "gitea.pr.request_changes", + ], + } + p.update(extra) + return p + @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._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")) + @patch("gitea_mcp_server.get_profile") @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 + def test_acquire_merger_lease_success( + self, _verify, _comments, mock_profile, _resolve, _auth, mock_api + ): + mock_profile.return_value = self._merger_profile() mock_api.side_effect = [ - {"state": "open"}, # PR is open - {"id": 456}, # Posted comment ID + {"state": "open", "head": {"sha": LIVE_HEAD}}, + {"id": 456}, ] - 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", + candidate_head=HEAD, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", ) - - 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") + self.assertTrue(res["success"], res) + self.assertTrue(res["acquired"]) + self.assertEqual(res["comment_id"], 456) + self.assertEqual(res["session_id"], "merger-session") + self.assertEqual(res.get("repo"), "Scaled-Tech-Consulting/Gitea-Tools") + body = mock_api.call_args_list[-1][0][3]["body"] + self.assertIn("reviewer_identity: merger-user", body) + self.assertIn("profile: prgs-merger", body) + session_lease = leases._SESSION_LEASE + self.assertIsNotNone(session_lease) + 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._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")) + @patch("gitea_mcp_server.get_profile") @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.""" + def test_acquire_merger_lease_fails_if_active_exists( + self, _verify, _comments, mock_profile, _resolve, _auth, mock_api + ): + mock_profile.return_value = self._merger_profile() active_body = leases.format_lease_body( - repo="org/repo", + repo="Scaled-Tech-Consulting/Gitea-Tools", pr_number=718, issue_number=None, reviewer_identity="other-user", @@ -80,51 +105,285 @@ class TestMergerLeaseAcquireTool(unittest.TestCase): session_id="other-session", worktree="branches/review-1", phase="claimed", - candidate_head="a" * 40, + candidate_head=HEAD, 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"} - + _comments.return_value = [ + {"id": 1, "body": active_body, "user": {"login": "other-user"}} + ] + mock_api.return_value = {"state": "open", "head": {"sha": LIVE_HEAD}} 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", + candidate_head=HEAD, ) - - 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) + self.assertFalse(res["success"]) + self.assertFalse(res["acquired"]) + self.assertIn("already has active reviewer lease", " ".join(res["reasons"])) + self.assertEqual( + [c for c in mock_api.call_args_list if c[0][0] == "POST"], [] + ) + 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.get_profile") @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.""" + def test_acquire_merger_lease_fails_workspace_binding(self, _verify, mock_profile): + mock_profile.return_value = self._merger_profile() _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", + candidate_head=HEAD, ) - - self.assertFalse(res["success"]) - self.assertFalse(res["acquired"]) - self.assertIn("Branches-only mutation guard", res["reasons"][0]) - self.assertIsNone(leases._SESSION_LEASE) + self.assertFalse(res["success"]) + self.assertIn("Branches-only mutation guard", res["reasons"][0]) + self.assertIsNone(leases._SESSION_LEASE) + + @patch("gitea_mcp_server.get_profile") + def test_acquire_merger_requires_candidate_head(self, mock_profile): + mock_profile.return_value = self._merger_profile() + res = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + candidate_head=None, + ) + self.assertFalse(res["success"]) + self.assertTrue(any("candidate_head is required" in r for r in res["reasons"])) + + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server._auth", return_value="token pass") + @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")) + @patch("gitea_mcp_server.get_profile") + @patch("gitea_mcp_server._fetch_pr_comments", return_value=[]) + @patch("gitea_mcp_server._verify_role_mutation_workspace") + def test_acquire_merger_head_mismatch( + self, _verify, _comments, mock_profile, _resolve, _auth, mock_api + ): + mock_profile.return_value = self._merger_profile() + mock_api.return_value = {"state": "open", "head": {"sha": "b" * 40}} + res = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + candidate_head=HEAD, + ) + self.assertFalse(res["success"]) + self.assertTrue(any("does not match live PR head" in r for r in res["reasons"])) + self.assertIsNone(leases._SESSION_LEASE) + + @patch("gitea_mcp_server._profile_operation_gate") + def test_author_denied_merge_permission(self, mock_gate): + """Author profile lacks gitea.pr.merge → fail closed before mutation.""" + def gate(op): + if op == "gitea.pr.merge": + return [f"profile lacks {op}"] + return None + mock_gate.side_effect = gate + res = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + candidate_head=HEAD, + ) + self.assertFalse(res["success"]) + self.assertTrue(any("gitea.pr.merge" in r for r in res["reasons"])) + + @patch("gitea_mcp_server._profile_operation_gate") + def test_reviewer_denied_merge_permission(self, mock_gate): + def gate(op): + if op == "gitea.pr.merge": + return [f"profile lacks {op}"] + return None + mock_gate.side_effect = gate + res = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + candidate_head=HEAD, + ) + self.assertFalse(res["success"]) + self.assertTrue(any("gitea.pr.merge" in r for r in res["reasons"])) + + +class TestCapabilityMapLeaseTasks(unittest.TestCase): + def test_merger_acquire_map_entries(self): + for task in ("acquire_merger_pr_lease", "gitea_acquire_merger_pr_lease"): + self.assertEqual(tcm.required_permission(task), "gitea.pr.comment") + self.assertEqual(tcm.required_role(task), "merger") + + def test_reviewer_acquire_map_entries(self): + for task in ("acquire_reviewer_pr_lease", "gitea_acquire_reviewer_pr_lease"): + self.assertEqual(tcm.required_permission(task), "gitea.pr.comment") + self.assertEqual(tcm.required_role(task), "reviewer") + + def test_adopt_merger_aliases(self): + for task in ("adopt_merger_pr_lease", "gitea_adopt_merger_pr_lease"): + self.assertEqual(tcm.required_role(task), "merger") + + +class TestResolveTaskCapabilityLeaseAndUnknown(unittest.TestCase): + def setUp(self): + mcp_server._process_boot_head_sha = None + if hasattr(mcp_server, "capability_stop_terminal"): + mcp_server.capability_stop_terminal.clear() + + @patch.object(mcp_server, "record_mutation_authority", return_value=None) + @patch.object(mcp_server, "init_review_decision_lock", return_value=None) + @patch.object(mcp_server, "record_preflight_check", return_value=None) + @patch.object(mcp_server, "_ensure_matching_profile", return_value=None) + @patch.object(mcp_server, "_authenticated_username", return_value="sysadmin") + @patch.object(mcp_server, "get_profile") + @patch.object(mcp_server.gitea_config, "load_config") + def test_resolve_acquire_reviewer_authorized( + self, mock_cfg, mock_profile, *_mocks + ): + mock_profile.return_value = { + "profile_name": "prgs-reviewer", + "role": "reviewer", + "allowed_operations": ["gitea.pr.comment", "gitea.read", "gitea.pr.review"], + "forbidden_operations": [], + } + mock_cfg.return_value = { + "profiles": { + "prgs-reviewer": { + "role": "reviewer", + "allowed_operations": [ + "gitea.pr.comment", + "gitea.read", + "gitea.pr.review", + ], + "forbidden_operations": [], + } + } + } + with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-reviewer"}, clear=False): + for task in ("acquire_reviewer_pr_lease", "gitea_acquire_reviewer_pr_lease"): + res = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs") + self.assertEqual(res["required_role_kind"], "reviewer", res) + self.assertEqual( + res["required_operation_permission"], "gitea.pr.comment", res + ) + self.assertTrue(res.get("allowed_in_current_session"), res) + + @patch.object(mcp_server, "record_mutation_authority", return_value=None) + @patch.object(mcp_server, "init_review_decision_lock", return_value=None) + @patch.object(mcp_server, "record_preflight_check", return_value=None) + @patch.object(mcp_server, "_ensure_matching_profile", return_value=None) + @patch.object(mcp_server, "_authenticated_username", return_value="jcwalker3") + @patch.object(mcp_server, "get_profile") + @patch.object(mcp_server.gitea_config, "load_config") + def test_resolve_acquire_reviewer_author_denied( + self, mock_cfg, mock_profile, *_mocks + ): + mock_profile.return_value = { + "profile_name": "prgs-author", + "role": "author", + "allowed_operations": ["gitea.pr.comment", "gitea.branch.push", "gitea.read"], + "forbidden_operations": [], + } + mock_cfg.return_value = { + "profiles": { + "prgs-author": { + "role": "author", + "allowed_operations": [ + "gitea.pr.comment", + "gitea.branch.push", + "gitea.read", + ], + "forbidden_operations": [], + }, + "prgs-reviewer": { + "role": "reviewer", + "allowed_operations": ["gitea.pr.comment", "gitea.pr.review"], + "forbidden_operations": [], + }, + } + } + with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-author"}, clear=False): + res = mcp_server.gitea_resolve_task_capability( + task="acquire_reviewer_pr_lease", remote="prgs" + ) + self.assertFalse(res.get("allowed_in_current_session"), res) + self.assertEqual(res["required_role_kind"], "reviewer") + guidance = " ".join(res.get("task_role_guidance") or []) + self.assertTrue( + "cannot perform reviewer" in guidance.lower() + or "reviewer" in guidance.lower() + or res.get("stop_required"), + res, + ) + + @patch.object(mcp_server, "record_mutation_authority", return_value=None) + @patch.object(mcp_server, "init_review_decision_lock", return_value=None) + @patch.object(mcp_server, "record_preflight_check", return_value=None) + @patch.object(mcp_server, "_ensure_matching_profile", return_value=None) + @patch.object(mcp_server, "_authenticated_username", return_value="sysadmin") + @patch.object(mcp_server, "get_profile") + def test_resolve_unknown_task_structured_no_raise(self, mock_profile, *_mocks): + mock_profile.return_value = { + "profile_name": "prgs-reviewer", + "role": "reviewer", + "allowed_operations": ["gitea.pr.comment"], + "forbidden_operations": [], + } + # Must not raise ValueError into the tool boundary. + res = mcp_server.gitea_resolve_task_capability( + task="some_made_up_task", remote="prgs" + ) + self.assertIsInstance(res, dict) + self.assertEqual(res.get("reason_code"), "unknown_task", res) + self.assertFalse(res.get("allowed_in_current_session")) + self.assertTrue(res.get("stop_required")) + self.assertIs(res.get("mutation_performed"), False) + self.assertNotIn("token", str(res).lower()) + self.assertNotIn("password", str(res).lower()) + guidance = " ".join(res.get("task_role_guidance") or []) + self.assertIn("Unknown task/action", guidance) + + @patch.object(mcp_server, "record_mutation_authority", return_value=None) + @patch.object(mcp_server, "init_review_decision_lock", return_value=None) + @patch.object(mcp_server, "record_preflight_check", return_value=None) + @patch.object(mcp_server, "_ensure_matching_profile", return_value=None) + @patch.object(mcp_server, "_authenticated_username", return_value="sysadmin") + @patch.object(mcp_server, "get_profile") + @patch.object(mcp_server.gitea_config, "load_config") + def test_resolve_merger_acquire_aliases( + self, mock_cfg, mock_profile, *_mocks + ): + mock_profile.return_value = { + "profile_name": "prgs-merger", + "role": "merger", + "allowed_operations": [ + "gitea.read", + "gitea.pr.comment", + "gitea.pr.merge", + ], + "forbidden_operations": [], + } + mock_cfg.return_value = { + "profiles": { + "prgs-merger": { + "role": "merger", + "allowed_operations": [ + "gitea.read", + "gitea.pr.comment", + "gitea.pr.merge", + ], + "forbidden_operations": [], + } + } + } + with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-merger"}, clear=False): + for task in ("acquire_merger_pr_lease", "gitea_acquire_merger_pr_lease"): + res = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs") + self.assertEqual(res["required_role_kind"], "merger", res) + self.assertEqual( + res["required_operation_permission"], "gitea.pr.comment", res + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_resolve_task_capability.py b/tests/test_resolve_task_capability.py index 23f05bc..07e5769 100644 --- a/tests/test_resolve_task_capability.py +++ b/tests/test_resolve_task_capability.py @@ -231,9 +231,16 @@ class TestResolveTaskCapability(unittest.TestCase): self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"]) def test_resolve_unknown_task_fails_closed(self): + # #723: unknown tasks return structured unknown_task (no ValueError escape). with patch.dict(os.environ, self._env("author-profile")): - with self.assertRaises(ValueError): - mcp_server.gitea_resolve_task_capability(task="invalid_task_xyz", remote="prgs") + res = mcp_server.gitea_resolve_task_capability( + task="invalid_task_xyz", remote="prgs" + ) + self.assertIsInstance(res, dict) + self.assertEqual(res.get("reason_code"), "unknown_task", res) + self.assertFalse(res.get("allowed_in_current_session")) + self.assertTrue(res.get("stop_required")) + self.assertIs(res.get("mutation_performed"), False) # Additional regression tests per #145 for permission boundaries and structured guidance def test_issue_comment_does_not_imply_close(self): @@ -439,8 +446,11 @@ class TestResolveTaskCapability(unittest.TestCase): res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs") self.assertEqual(res["required_operation_permission"], "gitea.pr.close") for unknown in ("close_pull_request", "close", "pr_close"): - with self.assertRaises(ValueError): - mcp_server.gitea_resolve_task_capability(task=unknown, remote="prgs") + unk = mcp_server.gitea_resolve_task_capability( + task=unknown, remote="prgs" + ) + self.assertEqual(unk.get("reason_code"), "unknown_task", unk) + self.assertFalse(unk.get("allowed_in_current_session")) if __name__ == "__main__": unittest.main() From 29c2cf2ef6bc427417045a7474a67fcd40ea25bd Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 17 Jul 2026 13:25:34 -0400 Subject: [PATCH 3/4] fix(mcp): fail closed when merger lease cannot resolve live PR head gitea_acquire_merger_pr_lease previously continued when GET /pulls returned an empty head SHA. Fail closed so exact-head scoping never proceeds with an unresolvable live head. Add regression coverage for empty head payloads. Fixes #718 Refs #723 --- gitea_mcp_server.py | 15 +++++++++- tests/test_merger_lease_acquire.py | 46 ++++++++++++++++++++++++++++++ 2 files changed, 60 insertions(+), 1 deletion(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 3ebf71a..e7551db 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -10821,7 +10821,20 @@ def gitea_acquire_merger_pr_lease( or "" ) live_head = str(live_head).strip() - if live_head and live_head != head_pin: + # Fail closed when live PR head cannot be resolved — never proceed with + # an unpinned/unknown head even if candidate_head was supplied (#718). + if not live_head: + return { + "success": False, + "acquired": False, + "reasons": [ + "live PR head could not be resolved from GET /pulls response " + "(exact-head scoping; fail closed)" + ], + "live_head": None, + "candidate_head": head_pin, + } + if live_head != head_pin: return { "success": False, "acquired": False, diff --git a/tests/test_merger_lease_acquire.py b/tests/test_merger_lease_acquire.py index 3c9da2a..88fa94b 100644 --- a/tests/test_merger_lease_acquire.py +++ b/tests/test_merger_lease_acquire.py @@ -176,6 +176,52 @@ class TestMergerLeaseAcquireTool(unittest.TestCase): self.assertTrue(any("does not match live PR head" in r for r in res["reasons"])) self.assertIsNone(leases._SESSION_LEASE) + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server._auth", return_value="token pass") + @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")) + @patch("gitea_mcp_server.get_profile") + @patch("gitea_mcp_server._fetch_pr_comments", return_value=[]) + @patch("gitea_mcp_server._verify_role_mutation_workspace") + def test_acquire_merger_fails_closed_when_live_head_unresolvable( + self, _verify, _comments, mock_profile, _resolve, _auth, mock_api + ): + """Empty/missing live head must not silently proceed past exact-head gate.""" + mock_profile.return_value = self._merger_profile() + # PR payload present but head.sha missing / empty / non-dict. + mock_api.return_value = {"state": "open", "head": {}} + res = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + session_id="merger-session", + candidate_head=HEAD, + ) + self.assertFalse(res["success"], res) + self.assertFalse(res.get("acquired"), res) + self.assertTrue( + any("live PR head could not be resolved" in r for r in res["reasons"]), + res, + ) + self.assertIsNone(res.get("live_head")) + self.assertEqual(res.get("candidate_head"), HEAD) + self.assertEqual( + [c for c in mock_api.call_args_list if c[0][0] == "POST"], [] + ) + self.assertIsNone(leases._SESSION_LEASE) + + # Completely empty GET /pulls body also fails closed. + mock_api.return_value = {} + res2 = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + candidate_head=HEAD, + ) + self.assertFalse(res2["success"], res2) + self.assertTrue( + any("live PR head could not be resolved" in r for r in res2["reasons"]), + res2, + ) + self.assertIsNone(leases._SESSION_LEASE) + @patch("gitea_mcp_server._profile_operation_gate") def test_author_denied_merge_permission(self, mock_gate): """Author profile lacks gitea.pr.merge → fail closed before mutation.""" From 83bc7246ffe0bba45904d4bbb321f1322ab93d6f Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 17 Jul 2026 13:30:52 -0400 Subject: [PATCH 4/4] fix(mcp): forward explicit org/repo into lease/review anti-stomp _verify_role_mutation_workspace dropped org/repo, so anti-stomp resolved bare remote=prgs to Timesheet and failed closed wrong_repo on Gitea-Tools. Forward explicit org/repo for merger/reviewer lease acquire, adopt, review, merge, and lease release. Add regression for merger acquire forwarding. Fixes #718 Refs #723 --- gitea_mcp_server.py | 52 ++++++++++++++++++++++++++---- tests/test_merger_lease_acquire.py | 31 ++++++++++++++++++ 2 files changed, 76 insertions(+), 7 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index e7551db..810e546 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1282,12 +1282,18 @@ def _verify_role_mutation_workspace( worktree_path: str | None = None, worktree: str | None = None, task: str | None = None, + org: str | None = None, + repo: str | None = None, ) -> str: """Bind reviewer/merger mutations to the active namespace workspace (#510). #683: must NOT early-return solely because pytest/unittest is loaded. Production workspace binding always runs; test isolation uses explicit env fixtures / force-on flags, never a production short-circuit here. + + org/repo are forwarded into anti-stomp (#604) so callers that pass explicit + repository targets (e.g. prgs + Gitea-Tools when REMOTES defaults to + Timesheet) do not fail closed with wrong_repo after a successful resolve. """ # Check running runtimes to prevent stale mutations @@ -1350,7 +1356,13 @@ 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, + task=task, + org=org, + repo=repo, + ) return resolved @@ -4521,7 +4533,11 @@ def _evaluate_pr_review_submission( ) -> dict: """Shared gate chain for live submit and dry-run review tools.""" _verify_role_mutation_workspace( - remote, worktree_path=worktree_path, task="review_pr" + remote, + worktree_path=worktree_path, + task="review_pr", + org=org, + repo=repo, ) action = (action or "").strip().lower() workflow_blockers = _review_workflow_load_gate_reasons() if live else [] @@ -7619,7 +7635,11 @@ def gitea_merge_pr( available. Never secrets. """ _verify_role_mutation_workspace( - remote, worktree_path=worktree_path, task="merge_pr" + remote, + worktree_path=worktree_path, + task="merge_pr", + org=org, + repo=repo, ) allowed = get_profile()["allowed_operations"] forbidden = get_profile()["forbidden_operations"] @@ -10652,8 +10672,14 @@ def gitea_acquire_reviewer_pr_lease( # task=acquire_reviewer_pr_lease so verify_preflight_purity runs shared #604 # anti-stomp for the declared lease-acquire mutation inventory entry. + # Forward explicit org/repo so anti-stomp does not default bare remote=prgs + # to Timesheet when the local git remote is Gitea-Tools (#604 wrong_repo). _verify_role_mutation_workspace( - remote, worktree=worktree, task="acquire_reviewer_pr_lease" + remote, + worktree=worktree, + task="acquire_reviewer_pr_lease", + org=org, + repo=repo, ) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) @@ -10794,7 +10820,11 @@ def gitea_acquire_merger_pr_lease( try: _verify_role_mutation_workspace( - remote, worktree=worktree, task="acquire_merger_pr_lease" + remote, + worktree=worktree, + task="acquire_merger_pr_lease", + org=org, + repo=repo, ) except RuntimeError as e: return { @@ -10987,7 +11017,11 @@ def gitea_adopt_merger_pr_lease( # task=adopt_merger_pr_lease so verify_preflight_purity runs shared #604 anti-stomp. _verify_role_mutation_workspace( - remote, worktree=worktree, task="adopt_merger_pr_lease" + remote, + worktree=worktree, + task="adopt_merger_pr_lease", + org=org, + repo=repo, ) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) @@ -11709,7 +11743,11 @@ def gitea_release_reviewer_pr_lease( dict reporting release status. """ _verify_role_mutation_workspace( - remote, worktree=worktree, task="review_pr" + remote, + worktree=worktree, + task="review_pr", + org=org, + repo=repo, ) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) diff --git a/tests/test_merger_lease_acquire.py b/tests/test_merger_lease_acquire.py index 88fa94b..77bc118 100644 --- a/tests/test_merger_lease_acquire.py +++ b/tests/test_merger_lease_acquire.py @@ -145,6 +145,37 @@ class TestMergerLeaseAcquireTool(unittest.TestCase): self.assertIn("Branches-only mutation guard", res["reasons"][0]) self.assertIsNone(leases._SESSION_LEASE) + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server._auth", return_value="token pass") + @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")) + @patch("gitea_mcp_server.get_profile") + @patch("gitea_mcp_server._fetch_pr_comments", return_value=[]) + @patch("gitea_mcp_server._verify_role_mutation_workspace") + def test_acquire_merger_forwards_explicit_org_repo_to_workspace_verify( + self, mock_verify, _comments, mock_profile, _resolve, _auth, mock_api + ): + """Explicit org/repo must reach anti-stomp (prgs bare default is Timesheet).""" + mock_profile.return_value = self._merger_profile() + mock_api.side_effect = [ + {"state": "open", "head": {"sha": LIVE_HEAD}}, + {"id": 789}, + ] + 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", + candidate_head=HEAD, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + ) + self.assertTrue(res["success"], res) + kwargs = mock_verify.call_args.kwargs + self.assertEqual(kwargs.get("org"), "Scaled-Tech-Consulting") + self.assertEqual(kwargs.get("repo"), "Gitea-Tools") + self.assertEqual(kwargs.get("task"), "acquire_merger_pr_lease") + @patch("gitea_mcp_server.get_profile") def test_acquire_merger_requires_candidate_head(self, mock_profile): mock_profile.return_value = self._merger_profile()