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:

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"])