fix: enforce merger role boundaries (Closes #539)

tests/test_resolve_task_capability.py

@@ -287,6 +287,49 @@ class TestResolveTaskCapability(unittest.TestCase):
             self.assertEqual(res["required_operation_permission"], "gitea.pr.merge")
             self.assertEqual(res["required_role_kind"], "merger")
 
+    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_reviewer_role_with_merge_permission_still_cannot_merge(self, _auth, _api):
+        # #539: exact permission alone is insufficient. The active profile
+        # role must also be merger for merge_pr.
+        config = json.loads(json.dumps(CONFIG_RESOLVER))
+        config["profiles"]["reviewer-with-merge"] = {
+            "enabled": True,
+            "context": "ctx",
+            "role": "reviewer",
+            "username": "reviewer-user",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
+            "allowed_operations": [
+                "gitea.read", "gitea.pr.review", "gitea.pr.approve",
+                "gitea.pr.merge",
+            ],
+            "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
+            "execution_profile": "reviewer-with-merge",
+        }
+        self._write_config(config)
+        with patch.dict(os.environ, self._env("reviewer-with-merge")):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="merge_pr", remote="prgs"
+            )
+        self.assertEqual(res["required_role_kind"], "merger")
+        self.assertEqual(res["active_role_kind"], "reviewer")
+        self.assertTrue(res["active_profile_permission_allowed"])
+        self.assertFalse(res["allowed_in_current_session"])
+        self.assertTrue(res["stop_required"])
+        self.assertIn("cannot perform merger task", " ".join(res["task_role_guidance"]))
+
+    @patch("mcp_server.api_request", return_value={"login": "merger-user"})
+    @patch("mcp_server.get_auth_header", return_value="token merger-pass")
+    def test_merger_profile_can_resolve_merge_task(self, _auth, _api):
+        with patch.dict(os.environ, self._env("merger-profile")):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="merge_pr", remote="prgs"
+            )
+        self.assertEqual(res["required_role_kind"], "merger")
+        self.assertEqual(res["active_role_kind"], "merger")
+        self.assertTrue(res["allowed_in_current_session"])
+        self.assertFalse(res["stop_required"])
+
     @patch("mcp_server.api_request")
     def test_self_review_blocked_structured(self, mock_api):
         # Test that self-approve (self-review gate) is blocked and returns structured reasons

tests/test_role_session_router.py

@@ -52,6 +52,20 @@ CONFIG = {
             ],
             "execution_profile": "prgs-reviewer",
         },
+        "prgs-merger": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "merger",
+            "username": "sysadmin",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_MERGER"},
+            "allowed_operations": [
+                "gitea.read", "gitea.pr.merge", "gitea.issue.comment",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.create", "gitea.branch.push", "gitea.pr.approve",
+            ],
+            "execution_profile": "prgs-merger",
+        },
         "prgs-reconciler": {
             "enabled": True,
             "context": "ctx",
@@ -104,6 +118,7 @@ class TestRoleSessionRouter(unittest.TestCase):
             "GITEA_MCP_PROFILE": profile,
             "GITEA_TOKEN_AUTHOR": "author-pass",
             "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+            "GITEA_TOKEN_MERGER": "merger-pass",
             "GITEA_TOKEN_RECONCILER": "reconciler-pass",
         }
 
@@ -227,6 +242,31 @@ class TestRoleSessionRouter(unittest.TestCase):
         self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
         self.assertTrue(route["downstream_allowed"])
 
+    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_merge_task_under_reviewer_profile_wrong_role_stop(self, _auth, _api):
+        # #539: a reviewer session must not pass the pre-task merge route
+        # even if its config accidentally includes gitea.pr.merge.
+        with patch.dict(os.environ, self._env("prgs-reviewer")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="merge_pr", remote="prgs"
+            )
+        self.assertEqual(route["required_role"], "merger")
+        self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE)
+        self.assertFalse(route["downstream_allowed"])
+        self.assertIn("merger", route["message"].lower())
+
+    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
+    @patch("mcp_server.get_auth_header", return_value="token merger-pass")
+    def test_merge_task_under_merger_profile_allowed(self, _auth, _api):
+        with patch.dict(os.environ, self._env("prgs-merger")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="merge_pr", remote="prgs"
+            )
+        self.assertEqual(route["required_role"], "merger")
+        self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
+        self.assertTrue(route["downstream_allowed"])
+
     @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
     @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
     def test_author_task_under_reviewer_profile_routes_to_author(self, _auth, _api):
@@ -334,4 +374,4 @@ class TestCheckMidMerge(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()