Merge pull request 'Add explicit close_pr capability resolution and gated PR close path (Issue #216)' (#219) from feat/issue-216-close-pr-capability into master

docs/gitea-execution-profiles.md

@@ -203,6 +203,29 @@ remote/org/repo arguments. Create operations are audit-logged
 redacted, and normal output contains no endpoint URLs
 (`GITEA_MCP_REVEAL_ENDPOINTS=1` is the local admin opt-in for web links).
 
+## PR edits versus PR closure (#216)
+
+Editing a pull request and closing one are different capabilities:
+
+- **PR edits** (`gitea_edit_pr` with `title`/`body`/`base`, or reopening with
+  `state="open"`) stay on the ordinary edit path and need no dedicated
+  capability.
+- **PR closure** (`gitea_edit_pr` with `state="closed"`) requires the
+  distinct `gitea.pr.close` operation. The resolver task is `close_pr`
+  (`gitea_resolve_task_capability(task="close_pr")`, author-side). Without
+  `gitea.pr.close` the close attempt fails closed — no API call, structured
+  `permission_report` — so the broad edit path can never be used as an
+  untracked close fallback.
+- Closures are audited as a distinct `close_pr` action with
+  `required_permission: gitea.pr.close` in the request metadata, so final
+  reports can prove exactly which mutation capability was exercised (#191).
+
+`gitea.pr.close` has no legacy alias; spell it canonically. It is not part of
+any default profile: the operator grants it deliberately (e.g. for an
+explicit operator-directed closure of a contaminated PR). If `close_pr` ever
+resolves as unknown, agents must fail closed rather than fall back to the
+edit path.
+
 ## Identity and fail-closed rules
 
 Before **any** mutating action, a workflow must know both:

mcp_server.py

@@ -977,7 +977,7 @@ def gitea_get_pr_review_feedback(
         'feedback_not_attempted' True, 'reasons', and 'permission_report' —
         deliberately distinct from a successful "no reviews yet" result.
     """
-    reasons = _issue_comment_gate("gitea.read")
+    reasons = _profile_operation_gate("gitea.read")
     if reasons:
         return {
             "success": False,
@@ -1220,11 +1220,20 @@ def gitea_edit_pr(
 ) -> dict:
     """Edit an existing pull request on a Gitea repository.
 
+    Closing a PR (``state='closed'``) is a distinct capability from other
+    edits (#216): it requires the ``gitea.pr.close`` operation (resolver
+    task ``close_pr``), fails closed with a structured permission report
+    when the active profile lacks it, and is audited as a distinct
+    ``close_pr`` action. Title/body/base edits and reopening stay on the
+    ordinary edit path, so the edit tool can never be used as an untracked
+    close fallback.
+
     Args:
         pr_number: The pull request index/number (required).
         title: New PR title.
         body: New PR description.
-        state: New state — 'open' or 'closed'.
+        state: New state — 'open' or 'closed'. 'closed' requires the
+            ``gitea.pr.close`` capability.
         base: Target branch name.
         remote: Known instance — 'dadeschools' or 'prgs'.
         host: Override the Gitea host.
@@ -1232,7 +1241,10 @@ def gitea_edit_pr(
         repo: Override the repository name.
 
     Returns:
-        dict with success status and details of the edited PR.
+        dict with success status and details of the edited PR. A close
+        attempt without ``gitea.pr.close`` returns 'success'/'performed'
+        False with 'reasons' and a structured 'permission_report' and makes
+        no API call.
     """
     # Validate inputs BEFORE any auth/profile resolution or API setup: a
     # no-fields call is a pure validation error and must not depend on
@@ -1243,6 +1255,9 @@ def gitea_edit_pr(
     if body is not None:
         payload["body"] = body
     if state is not None:
+        if state not in ("open", "closed"):
+            raise ValueError(
+                f"Invalid state {state!r}: must be 'open' or 'closed' (fail closed).")
         payload["state"] = state
     if base is not None:
         payload["base"] = base
@@ -1250,12 +1265,33 @@ def gitea_edit_pr(
     if not payload:
         raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
 
+    # PR closure is a first-class capability, distinct from retitling or
+    # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
+    # never touches the network.
+    closing = payload.get("state") == "closed"
+    if closing:
+        gate_reasons = _profile_operation_gate("gitea.pr.close")
+        if gate_reasons:
+            return {
+                "success": False,
+                "performed": False,
+                "pr_number": pr_number,
+                "requested_state": "closed",
+                "required_permission": "gitea.pr.close",
+                "reasons": gate_reasons,
+                "permission_report": _permission_block_report("gitea.pr.close"),
+            }
+
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
 
-    with _audited("edit_pr", host=h, remote=remote, org=o, repo=r,
-                  pr_number=pr_number, request_metadata={"fields": sorted(payload)}):
+    request_metadata = {"fields": sorted(payload)}
+    if closing:
+        request_metadata["required_permission"] = "gitea.pr.close"
+    with _audited("close_pr" if closing else "edit_pr",
+                  host=h, remote=remote, org=o, repo=r,
+                  pr_number=pr_number, request_metadata=request_metadata):
         data = api_request("PATCH", url, auth, payload)
 
     cleanup_status = None
@@ -2027,13 +2063,14 @@ def _permission_block_report(required_operation: str,
     return report
 
 
-def _issue_comment_gate(op: str) -> list[str]:
-    """Profile permission check for issue-comment tools (#126).
+def _profile_operation_gate(op: str) -> list[str]:
+    """Profile permission check for a single gated operation (#126, #216).
 
     Issue discussion comments are gated separately from the gitea.pr.*
     review/merge family: listing requires ``gitea.read``, creating requires
-    ``gitea.issue.comment``. Returns a list of block reasons (empty = allowed);
-    an unreadable profile fails closed.
+    ``gitea.issue.comment``. Closing a PR requires the distinct
+    ``gitea.pr.close`` (#216). Returns a list of block reasons (empty =
+    allowed); an unreadable profile fails closed.
     """
     try:
         profile = get_profile()
@@ -2085,7 +2122,7 @@ def gitea_list_issue_comments(
         'success' False, 'reasons', and a structured 'permission_report'
         (#142) with no API call made.
     """
-    reasons = _issue_comment_gate("gitea.read")
+    reasons = _profile_operation_gate("gitea.read")
     if reasons:
         return {"success": False, "issue_number": issue_number,
                 "reasons": reasons,
@@ -2147,7 +2184,7 @@ def gitea_create_issue_comment(
         (permission blocks also carry a structured 'permission_report',
         #142).
     """
-    gate_reasons = _issue_comment_gate("gitea.issue.comment")
+    gate_reasons = _profile_operation_gate("gitea.issue.comment")
     reasons = list(gate_reasons)
     if not (body or "").strip():
         reasons.append("comment body must be a non-empty string")
@@ -3581,6 +3618,14 @@ def gitea_resolve_task_capability(
             "permission": "gitea.pr.comment",
             "role": "author",
         },
+        # PR closure is a first-class capability (#216): distinct from
+        # comment_pr (gitea.pr.comment) and from ordinary PR edits, which
+        # need no dedicated capability. gitea_edit_pr(state='closed') is
+        # gated on the same operation, so no edit-path fallback exists.
+        "close_pr": {
+            "permission": "gitea.pr.close",
+            "role": "author",
+        },
         "address_pr_change_requests": {
             "permission": "gitea.branch.push",
             "role": "author",

tests/test_close_pr_capability.py

@@ -0,0 +1,154 @@
+"""Tests for the gated PR close path (Issue #216).
+
+``gitea_edit_pr(state='closed')`` requires the distinct ``gitea.pr.close``
+capability: without it the close fails closed (no auth lookup, no API call,
+structured permission report). With it — the explicit operator-directed
+contaminated-PR closure path — the close proceeds and is audited as a
+distinct ``close_pr`` action carrying the required capability, so final
+reports can prove exactly which mutation capability was exercised.
+
+Ordinary edits (title/body/base) and reopening never require the close
+capability: PR comment, PR edit, and PR close remain distinct capabilities.
+"""
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import mcp_server
+from mcp_server import gitea_edit_pr
+
+FAKE_AUTH = "token fake"
+
+AUTHOR_NO_CLOSE = {
+    "profile_name": "prgs-author",
+    "allowed_operations": [
+        "gitea.read", "gitea.pr.create", "gitea.pr.comment",
+        "gitea.branch.push", "gitea.issue.comment",
+    ],
+    "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
+    "audit_label": "prgs-author",
+}
+
+AUTHOR_WITH_CLOSE = {
+    "profile_name": "prgs-author-closer",
+    "allowed_operations": AUTHOR_NO_CLOSE["allowed_operations"] + ["gitea.pr.close"],
+    "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
+    "audit_label": "prgs-author-closer",
+}
+
+
+class TestEditPrCloseGate(unittest.TestCase):
+
+    def setUp(self):
+        self.mock_api = patch("mcp_server.api_request").start()
+        self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
+        # Deterministic permission reports: no real operator config, no switching.
+        patch("gitea_config.load_config", return_value={}).start()
+        patch("gitea_config.is_runtime_switching_enabled", return_value=False).start()
+        patch("gitea_audit.audit_enabled", return_value=False).start()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def tearDown(self):
+        patch.stopall()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def _set_profile(self, profile):
+        patch("mcp_server.get_profile", return_value=profile).start()
+
+    def test_close_blocked_without_close_capability(self):
+        # Author profile without gitea.pr.close: the broad edit path must not
+        # be usable as an untracked close fallback.
+        self._set_profile(AUTHOR_NO_CLOSE)
+        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
+        self.assertFalse(res["success"])
+        self.assertFalse(res["performed"])
+        self.assertEqual(res["requested_state"], "closed")
+        self.assertEqual(res["required_permission"], "gitea.pr.close")
+        self.assertTrue(res["reasons"])
+        report = res["permission_report"]
+        self.assertEqual(report["required_permission"], "gitea.pr.close")
+        self.assertEqual(report["active_profile"], "prgs-author")
+        self.mock_api.assert_not_called()
+        self.mock_auth.assert_not_called()
+
+    def test_close_blocked_when_close_forbidden(self):
+        profile = dict(AUTHOR_WITH_CLOSE)
+        profile["forbidden_operations"] = ["gitea.pr.close"]
+        self._set_profile(profile)
+        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
+        self.assertFalse(res["success"])
+        self.assertIn("profile forbids 'gitea.pr.close'", res["reasons"])
+        self.mock_api.assert_not_called()
+
+    def test_close_blocked_when_profile_unresolvable(self):
+        patch("mcp_server.get_profile", side_effect=RuntimeError("no profile")).start()
+        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
+        self.assertFalse(res["success"])
+        self.assertTrue(any("fail closed" in r for r in res["reasons"]))
+        self.mock_api.assert_not_called()
+
+    def test_operator_directed_close_allowed_and_audited(self):
+        # Explicit operator-directed contaminated-PR closure: the operator
+        # granted gitea.pr.close, so the close proceeds and the audit trail
+        # records a distinct close_pr action with the capability proof.
+        self._set_profile(AUTHOR_WITH_CLOSE)
+        patch("gitea_audit.audit_enabled", return_value=True).start()
+        mock_write = patch("gitea_audit.write_event").start()
+
+        def api_side_effect(method, url, auth, payload=None):
+            if method == "GET" and "/user" in url:
+                return {"login": "jcwalker3"}
+            if method == "PATCH" and "pulls/205" in url:
+                self.assertEqual(payload["state"], "closed")
+                return {
+                    "number": 205,
+                    "title": "Contaminated PR",
+                    "state": "closed",
+                    "html_url": "url",
+                    "body": "No issue link",
+                    "head": {"ref": "feat/invalid-provenance"},
+                }
+            return {}
+        self.mock_api.side_effect = api_side_effect
+
+        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
+        self.assertTrue(res["success"])
+        self.assertEqual(res["state"], "closed")
+        mock_write.assert_called()
+        event = mock_write.call_args[0][0]
+        self.assertEqual(event["action"], "close_pr")
+        self.assertEqual(
+            event["request_metadata"]["required_permission"], "gitea.pr.close")
+
+    def test_title_edit_needs_no_close_capability(self):
+        # PR edit and PR close are distinct capabilities: retitling stays on
+        # the ordinary edit path.
+        self._set_profile(AUTHOR_NO_CLOSE)
+        self.mock_api.return_value = {
+            "number": 7, "title": "Renamed", "state": "open",
+            "body": "", "html_url": "u"}
+        res = gitea_edit_pr(pr_number=7, title="Renamed", remote="prgs")
+        self.assertTrue(res["success"])
+
+    def test_reopen_needs_no_close_capability(self):
+        self._set_profile(AUTHOR_NO_CLOSE)
+        self.mock_api.return_value = {
+            "number": 7, "title": "T", "state": "open",
+            "body": "", "html_url": "u"}
+        res = gitea_edit_pr(pr_number=7, state="open", remote="prgs")
+        self.assertTrue(res["success"])
+
+    def test_invalid_state_fails_closed_before_auth(self):
+        # A case-variant state can neither bypass the close gate nor reach
+        # the API: it is rejected as pure validation, before auth.
+        self._set_profile(AUTHOR_WITH_CLOSE)
+        with self.assertRaises(ValueError):
+            gitea_edit_pr(pr_number=7, state="CLOSED", remote="prgs")
+        self.mock_api.assert_not_called()
+        self.mock_auth.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_server.py

@@ -1623,7 +1623,8 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
         self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
         patch("gitea_audit.audit_enabled", return_value=True).start()
         self.mock_audit = patch("gitea_audit.write_event").start()
-        patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close"], "audit_label": "test", "forbidden_operations": []}).start()
+        # gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216).
+        patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close"], "audit_label": "test", "forbidden_operations": []}).start()
 
     def tearDown(self):
         patch.stopall()

tests/test_resolve_task_capability.py

@@ -203,5 +203,67 @@ class TestResolveTaskCapability(unittest.TestCase):
             reasons = res.get("reasons", [])
             self.assertTrue(any("author" in str(r).lower() for r in reasons) or len(reasons) > 0)
 
+    # Issue #216: close_pr is a first-class resolver task gated on gitea.pr.close.
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_close_pr_author_profile_without_close_blocked(self, _auth, _api):
+        # Default author profile has PR create/comment but not close: the
+        # close capability must never be implied by broader author ops.
+        with patch.dict(os.environ, self._env("author-profile")):
+            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
+            self.assertEqual(res["requested_task"], "close_pr")
+            self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
+            self.assertEqual(res["required_role_kind"], "author")
+            self.assertFalse(res["allowed_in_current_session"])
+            self.assertTrue(res["stop_required"])
+            self.assertEqual(res["matching_configured_profile"], [])
+            self.assertTrue(res["different_mcp_namespace_required"])
+
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_close_pr_author_profile_with_close_allowed(self, _auth, _api):
+        # Operator-granted close capability resolves cleanly under an author profile.
+        config = json.loads(json.dumps(CONFIG_RESOLVER))
+        config["profiles"]["author-closer"] = {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "author-user",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": ["gitea.read", "gitea.pr.close"],
+            "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
+            "execution_profile": "author-closer",
+        }
+        self._write_config(config)
+        with patch.dict(os.environ, self._env("author-closer")):
+            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
+            self.assertTrue(res["allowed_in_current_session"])
+            self.assertFalse(res["stop_required"])
+            self.assertIn("author-closer", res["matching_configured_profile"])
+            self.assertIn("ready for operations", res["exact_safe_next_action"])
+
+    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_resolve_close_pr_reviewer_profile_blocked(self, _auth, _api):
+        # close_pr is author-side: a reviewer profile must be told to stop.
+        with patch.dict(os.environ, self._env("reviewer-profile")):
+            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
+            self.assertFalse(res["allowed_in_current_session"])
+            self.assertTrue(res["stop_required"])
+            self.assertEqual(res["required_role_kind"], "author")
+            self.assertIn("author", res["exact_safe_next_action"].lower())
+
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_close_pr_known_and_lookalike_tasks_still_fail_closed(self, _auth, _api):
+        # close_pr resolves (no Unknown-task error); near-miss spellings keep
+        # failing closed so no untracked close fallback can be rationalized.
+        with patch.dict(os.environ, self._env("author-profile")):
+            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")
+
 if __name__ == "__main__":
     unittest.main()