gitea_mcp_server.py

@@ -211,6 +211,21 @@ def _effective_workspace_role() -> str:
     )
 
 
+def _profile_role_kind(profile: dict) -> str:
+    """Resolve a profile's declared role before inferring from permissions."""
+    role = (profile.get("role") or profile.get("role_kind") or "").strip()
+    if role:
+        return role
+    profile_name = (profile.get("profile_name") or "").strip().lower()
+    for candidate in ("reconciler", "merger", "reviewer", "author"):
+        if candidate in profile_name:
+            return candidate
+    return _role_kind(
+        profile.get("allowed_operations") or [],
+        profile.get("forbidden_operations") or [],
+    )
+
+
 def _actual_profile_role() -> str:
     """Resolve the workspace role from the ACTIVE PROFILE alone (#540).
 
@@ -223,10 +238,7 @@ def _actual_profile_role() -> str:
     ``author`` here, so author blocking is preserved.
     """
     profile = get_profile()
-    role = _role_kind(
+    role = _profile_role_kind(profile)
-        profile.get("allowed_operations") or [],
-        profile.get("forbidden_operations") or [],
-    )
     return nwb.normalize_role_kind(
         role,
         profile_name=profile.get("profile_name"),
@@ -9917,6 +9929,24 @@ def gitea_resolve_task_capability(
 
     required_permission = task_capability_map.required_permission(task)
     required_role = task_capability_map.required_role(task)
+    role_exclusive_tasks = {
+        "review_pr",
+        "approve_pr",
+        "request_changes_pr",
+        "blind_pr_queue_review",
+        "pr_queue_cleanup",
+        "pr-queue-cleanup",
+        "merge_pr",
+        "create_branch",
+        "push_branch",
+        "create_pr",
+        "commit_files",
+        "gitea_commit_files",
+        "address_pr_change_requests",
+        "delete_branch",
+        "work_issue",
+        "work-issue",
+    }
 
     infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT)
     if required_role == "reviewer":
@@ -10011,11 +10041,26 @@ def gitea_resolve_task_capability(
     # Load active permissions
     active_allowed = profile.get("allowed_operations") or []
     active_forbidden = profile.get("forbidden_operations") or []
+    active_role_kind = _profile_role_kind(profile)
 
     # Check if allowed in current session
-    allowed_in_current_session, _ = gitea_config.check_operation(
+    permission_allowed_in_current_session, _ = gitea_config.check_operation(
         required_permission, active_allowed, active_forbidden
     )
+    role_matches_current_session = (
+        task not in role_exclusive_tasks
+        or active_role_kind == required_role
+    )
+    role_mismatch_reason = None
+    if not role_matches_current_session:
+        role_mismatch_reason = (
+            f"Active profile role '{active_role_kind}' cannot perform "
+            f"{required_role} task '{task}' even if nearby permissions are "
+            "present (fail closed)."
+        )
+    allowed_in_current_session = (
+        permission_allowed_in_current_session and role_matches_current_session
+    )
 
     switching = gitea_config.is_runtime_switching_enabled()
     available_in_session = allowed_in_current_session
@@ -10042,7 +10087,13 @@ def gitea_resolve_task_capability(
                 except Exception:
                     pass
             ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n)
-            if ok:
+            p_role = (p_data.get("role") or "").strip()
+            p_role_ok = (
+                task not in role_exclusive_tasks
+                or not p_role
+                or p_role == required_role
+            )
+            if ok and p_role_ok:
                 matching_profiles.append(p_name)
 
     configured = len(matching_profiles) > 0
@@ -10070,6 +10121,8 @@ def gitea_resolve_task_capability(
             reason_msg = (
                 f"No profile configured with permission '{required_permission}'."
             )
+        elif role_mismatch_reason:
+            reason_msg = role_mismatch_reason
     different_namespace_required = False
     next_safe_action = "None; ready for operations."
 
@@ -10101,6 +10154,8 @@ def gitea_resolve_task_capability(
     # into author-side mutations.
     stop_required = not allowed_in_current_session
     task_role_guidance = []
+    if role_mismatch_reason:
+        task_role_guidance.append(f"STOP: {role_mismatch_reason}")
     if required_role == "reviewer":
         if allowed_in_current_session:
             task_role_guidance.append(
@@ -10146,7 +10201,9 @@ def gitea_resolve_task_capability(
         "required_role_kind": required_role,
         "active_profile": profile["profile_name"],
         "active_identity": username,
+        "active_role_kind": active_role_kind,
         "active_profile_allowed_operations": active_allowed,
+        "active_profile_permission_allowed": permission_allowed_in_current_session,
         "allowed_in_current_session": allowed_in_current_session,
         "available_in_session": available_in_session,
         "configured": configured,

role_session_router.py

@@ -55,7 +55,6 @@ def skip_python_scan_walk_root(project_root: str, walk_root: str) -> bool:
 
 REVIEWER_TASKS = frozenset({
     "review_pr",
-    "merge_pr",
     "blind_pr_queue_review",
     "pr_queue_cleanup",
     "pr-queue-cleanup",
@@ -63,6 +62,10 @@ REVIEWER_TASKS = frozenset({
     "approve_pr",
 })
 
+MERGER_TASKS = frozenset({
+    "merge_pr",
+})
+
 AUTHOR_TASKS = frozenset({
     "create_issue",
     "comment_issue",
@@ -98,7 +101,7 @@ TASK_REQUIRED_ROLE = {
     "address_pr_change_requests": "author",
     "delete_branch": "author",
     "review_pr": "reviewer",
-    "merge_pr": "reviewer",
+    "merge_pr": "merger",
     "blind_pr_queue_review": "reviewer",
     "pr_queue_cleanup": "reviewer",
     "pr-queue-cleanup": "reviewer",
@@ -128,6 +131,10 @@ WRONG_ROLE_RECONCILER_MSG = (
     "MCP namespace/profile with exact close capability."
 )
 
+WRONG_ROLE_MERGER_MSG = (
+    "Wrong role/session for merger task. Launch merger MCP namespace."
+)
+
 _session_last_route: dict | None = None
 
 
@@ -243,6 +250,25 @@ def route_task_session(
         _record_route(result)
         return result
 
+    if required_role == "merger":
+        result = {
+            "task_type": task_type,
+            "required_role": required_role,
+            "active_role": active_role_kind,
+            "active_profile": active_profile,
+            "route_result": ROUTE_WRONG_ROLE,
+            "downstream_allowed": False,
+            "reasons": [
+                WRONG_ROLE_MERGER_MSG,
+                "Merger tasks cannot run in author or reviewer sessions.",
+            ],
+            "message": WRONG_ROLE_MERGER_MSG,
+            "runtime_switching_supported": runtime_switching_supported,
+            "profile_switch_blocked": not runtime_switching_supported,
+        }
+        _record_route(result)
+        return result
+
     if required_role == "author":
         route = ROUTE_TO_AUTHOR
         message = (

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