feat(routing): enforce role alignment in automatic dispatch profile switching

gitea_mcp_server.py

@@ -438,6 +438,9 @@ def _ensure_matching_profile(required_permission: str, required_role: str, remot
             for p_name, p_data in config["profiles"].items():
                 p_allowed = p_data.get("allowed_operations") or []
                 p_forbidden = p_data.get("forbidden_operations") or []
+                p_role = p_data.get("role") or _role_kind(p_allowed, p_forbidden)
+                if required_role and p_role != required_role:
+                    continue
                 p_allowed_n = []
                 for op in p_allowed:
                     try:
@@ -3880,90 +3883,6 @@ def gitea_get_profile(
     return result
 
 
-_RUNTIME_CAPABILITY_TASKS = (
-    "create_issue",
-    "comment_issue",
-    "create_pr",
-    "review_pr",
-    "merge_pr",
-    "close_issue",
-)
-
-
-def _matching_configured_profiles(
-    config: dict | None,
-    required_permission: str,
-) -> list[str]:
-    """Profile names that allow *required_permission* (redacted metadata only)."""
-    if not config or "profiles" not in config:
-        return []
-    matches: list[str] = []
-    for p_name, p_data in config["profiles"].items():
-        if not p_data.get("enabled", True):
-            continue
-        p_allowed = p_data.get("allowed_operations") or []
-        p_forbidden = p_data.get("forbidden_operations") or []
-        p_allowed_n = []
-        for op in p_allowed:
-            try:
-                p_allowed_n.append(gitea_config.normalize_operation(op))
-            except Exception:
-                pass
-        p_forbidden_n = []
-        for op in p_forbidden:
-            try:
-                p_forbidden_n.append(gitea_config.normalize_operation(op))
-            except Exception:
-                pass
-        ok, _ = gitea_config.check_operation(
-            required_permission, p_allowed_n, p_forbidden_n
-        )
-        if ok:
-            matches.append(p_name)
-    return sorted(matches)
-
-
-def _build_runtime_task_capabilities(
-    allowed: list[str],
-    forbidden: list[str],
-    config: dict | None,
-) -> dict:
-    """Per-task capability summary for role-aware runtime context (#139)."""
-    task_entries = []
-    flags: dict[str, bool] = {}
-    flag_keys = {
-        "create_issue": "can_create_issues",
-        "comment_issue": "can_comment_on_issues",
-        "create_pr": "can_author_prs",
-        "review_pr": "can_review_prs",
-        "merge_pr": "can_merge_prs",
-        "close_issue": "can_close_issues",
-    }
-    for task in _RUNTIME_CAPABILITY_TASKS:
-        permission = task_capability_map.required_permission(task)
-        allowed_here, _ = gitea_config.check_operation(
-            permission, allowed, forbidden
-        )
-        entry = {
-            "task": task,
-            "required_permission": permission,
-            "required_role_kind": task_capability_map.required_role(task),
-            "allowed_in_current_session": allowed_here,
-            "matching_configured_profiles": _matching_configured_profiles(
-                config, permission
-            ),
-        }
-        task_entries.append(entry)
-        flag_name = flag_keys.get(task)
-        if flag_name:
-            flags[flag_name] = allowed_here
-    return {
-        **flags,
-        "issue_comment_not_implied_by_pr_comment": True,
-        "task_capabilities": task_entries,
-    }
-
-
 @mcp.tool()
 def gitea_get_runtime_context(
     remote: str = "dadeschools",
@@ -4051,10 +3970,6 @@ def gitea_get_runtime_context(
             "or ask the operator to update GITEA_MCP_PROFILE to a reviewer profile."
         )
 
-    session_capabilities = _build_runtime_task_capabilities(
-        allowed, forbidden, config
-    )
-
     result = {
         "active_profile": profile["profile_name"],
         "authenticated_username": username,
@@ -4069,7 +3984,6 @@ def gitea_get_runtime_context(
         "review_merge_blocked_reasons": blocked_reasons,
         "suggested_fix": suggested_fix,
         "safe_next_action": safe_next_action,
-        "session_capabilities": session_capabilities,
     }
 
     if reveal and h:

tests/test_role_session_router.py

@@ -168,6 +168,74 @@ class TestRoleSessionRouter(unittest.TestCase):
         self.assertFalse(result["success"])
         self.assertIn("switching is disabled", result["message"])
 
+    def _switching_env(self, profile):
+        switching_config = dict(CONFIG)
+        switching_config["rules"] = {"allow_runtime_switching": True}
+        switching_config_path = os.path.join(self._dir.name, "profiles_switching.json")
+        with open(switching_config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(switching_config))
+        return {
+            "GITEA_MCP_CONFIG": switching_config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+        }
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_dynamic_switch_to_reviewer_profile(self, _auth, mock_api):
+        mock_api.side_effect = lambda method, url, header: (
+            {"login": "sysadmin"} if "reviewer-pass" in str(header) else {"login": "jcwalker3"}
+        )
+        with patch.dict(os.environ, self._switching_env("prgs-author")):
+            self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
+            route = mcp_server.gitea_route_task_session(
+                task_type="review_pr", remote="prgs"
+            )
+            self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
+            self.assertTrue(route["downstream_allowed"])
+            self.assertEqual(gitea_config.selected_profile_name(), "prgs-reviewer")
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_dynamic_switch_to_author_profile(self, _auth, mock_api):
+        mock_api.side_effect = lambda method, url, header: (
+            {"login": "jcwalker3"} if "author-pass" in str(header) else {"login": "sysadmin"}
+        )
+        with patch.dict(os.environ, self._switching_env("prgs-reviewer")):
+            self.assertEqual(gitea_config.selected_profile_name(), "prgs-reviewer")
+            route = mcp_server.gitea_route_task_session(
+                task_type="create_issue", remote="prgs"
+            )
+            self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
+            self.assertTrue(route["downstream_allowed"])
+            self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
+
+    def test_dynamic_switch_blocked_for_wrong_role(self):
+        with patch.dict(os.environ, self._switching_env("prgs-author")):
+            config_data = dict(CONFIG)
+            config_data["rules"] = {"allow_runtime_switching": True}
+            config_data["profiles"]["prgs-author"]["allowed_operations"] = [
+                "gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push"
+            ]
+            custom_path = os.path.join(self._dir.name, "profiles_wrong_role.json")
+            with open(custom_path, "w", encoding="utf-8") as fh:
+                fh.write(json.dumps(config_data))
+
+            env = {
+                "GITEA_MCP_CONFIG": custom_path,
+                "GITEA_MCP_PROFILE": "prgs-author",
+                "GITEA_TOKEN_AUTHOR": "author-pass",
+                "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+            }
+            with patch.dict(os.environ, env):
+                route = mcp_server.gitea_route_task_session(
+                    task_type="comment_issue", remote="prgs"
+                )
+                self.assertEqual(route["route_result"], role_session_router.ROUTE_TO_AUTHOR)
+                self.assertFalse(route["downstream_allowed"])
+                self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
+
     def test_handoff_requires_route_fields(self):
         incomplete = assess_role_route_handoff("Task done.")
         self.assertFalse(incomplete["complete"])

tests/test_runtime_clarity.py

@@ -110,18 +110,6 @@ class TestRuntimeClarity(unittest.TestCase):
         self.assertEqual(ctx["suggested_fix"], "reviewer namespace")
         self.assertIn("does not permit review or merge", ctx["review_merge_blocked_reasons"][0])
         self.assertIn("Switch to the reviewer MCP session", ctx["safe_next_action"])
-        caps = ctx["session_capabilities"]
-        self.assertTrue(caps["can_author_prs"])
-        self.assertFalse(caps["can_create_issues"])
-        self.assertFalse(caps["can_comment_on_issues"])
-        self.assertFalse(caps["can_review_prs"])
-        self.assertFalse(caps["can_merge_prs"])
-        self.assertTrue(caps["issue_comment_not_implied_by_pr_comment"])
-        merge_entry = next(
-            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
-        )
-        self.assertFalse(merge_entry["allowed_in_current_session"])
-        self.assertIn("reviewer-profile", merge_entry["matching_configured_profiles"])
 
     @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
     @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
@@ -133,15 +121,6 @@ class TestRuntimeClarity(unittest.TestCase):
         self.assertTrue(ctx["review_merge_allowed"])
         self.assertEqual(ctx["suggested_fix"], "none")
         self.assertEqual(ctx["safe_next_action"], "None; ready for operations.")
-        caps = ctx["session_capabilities"]
-        self.assertFalse(caps["can_author_prs"])
-        self.assertFalse(caps["can_create_issues"])
-        self.assertFalse(caps["can_review_prs"])
-        self.assertTrue(caps["can_merge_prs"])
-        merge_entry = next(
-            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
-        )
-        self.assertTrue(merge_entry["allowed_in_current_session"])
 
     # -------------------------------------------------------------------------
     # gitea_list_profiles