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

@@ -9959,6 +9959,183 @@ 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"),
+        }
+
+    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"
+        )
+    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 {}
+    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,
+        reviewer_identity=identity,
+        profile=profile.get("profile_name") or "unknown",
+        session_id=sid,
+        repo=repo_label,
+        issue_number=issue_number,
+        worktree=worktree,
+        candidate_head=head_pin,
+        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})
+
+    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,
+        "session_id": sid,
+        "reviewer_identity": identity,
+        "profile": profile.get("profile_name"),
+        "worktree": worktree,
+        "phase": "claimed",
+        "candidate_head": head_pin,
+        "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,
+        "candidate_head": head_pin,
+        "repo": repo_label,
+        "reasons": [],
+    }
+
+
 @mcp.tool()
 def gitea_adopt_merger_pr_lease(
     pr_number: int,
@@ -13817,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",
@@ -13829,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",

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

@@ -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,6 +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": {

tests/test_merger_lease_acquire.py

@@ -0,0 +1,389 @@
+"""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 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):
+    def setUp(self):
+        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=("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, mock_profile, _resolve, _auth, mock_api
+    ):
+        mock_profile.return_value = self._merger_profile()
+        mock_api.side_effect = [
+            {"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"], 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=("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, mock_profile, _resolve, _auth, mock_api
+    ):
+        mock_profile.return_value = self._merger_profile()
+        active_body = leases.format_lease_body(
+            repo="Scaled-Tech-Consulting/Gitea-Tools",
+            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=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"}}
+        ]
+        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"]))
+        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")
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    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.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()

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()