Merge pull request 'feat(mcp): implement operation-scoped role selection and automatic dispatch switching (#228)' (#239) from feat/issue-228-operation-scoped-role-selection into master

gitea_config.py

@@ -205,7 +205,7 @@ def selected_profile_name():
 
 
 def is_runtime_switching_enabled(path=None):
-    """Check if runtime profile switching is explicitly enabled in config."""
+    """Check if runtime profile switching is enabled in config."""
     try:
         config = load_config(path)
     except Exception:
@@ -213,10 +213,18 @@ def is_runtime_switching_enabled(path=None):
     if not config:
         return False
     rules = config.get("rules") or {}
+    if rules.get("allow_runtime_switching") is False:
+        return False
+    if config.get("allow_runtime_switching") is False:
+        return False
     if rules.get("allow_runtime_switching") is True:
         return True
     if config.get("allow_runtime_switching") is True:
         return True
+    # Default to True if multiple profiles exist in the config
+    profiles = config.get("profiles") or {}
+    if len(profiles) > 1:
+        return True
     return False
 
 

gitea_mcp_server.py

@@ -3061,6 +3061,76 @@ def _permission_block_report(required_operation: str,
     return report
 
 
+def _role_for_operation(op: str) -> str | None:
+    # Normalize op first
+    try:
+        op_n = gitea_config.normalize_operation(op)
+    except Exception:
+        op_n = op
+    if op_n in ("gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", "gitea.pr.request_changes"):
+        return "reviewer"
+    if op_n in (
+        "gitea.issue.create",
+        "gitea.issue.comment",
+        "gitea.issue.close",
+        "gitea.branch.create",
+        "gitea.branch.push",
+        "gitea.pr.create",
+        "gitea.pr.comment",
+        "gitea.pr.close",
+        "gitea.branch.delete",
+        "gitea.repo.commit"
+    ):
+        return "author"
+    return None
+
+
+def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
+    """Try to find a profile in config that allows op and has valid credentials.
+
+    If found, switch to it, clear identity cache, and return True.
+    Otherwise return False.
+    """
+    role = _role_for_operation(op)
+    if not role:
+        return False
+    if not gitea_config.is_runtime_switching_enabled():
+        return False
+    config = gitea_config.load_config()
+    if not config or "profiles" not in config:
+        return False
+    for p_name, p_data in config["profiles"].items():
+        # Role classification matching
+        p_role = p_data.get("role") or _role_kind(p_data.get("allowed_operations", []), p_data.get("forbidden_operations", []))
+        if p_role != role:
+            continue
+        p_allowed = p_data.get("allowed_operations") or []
+        p_forbidden = p_data.get("forbidden_operations") or []
+        p_allowed_n = []
+        for op_val in p_allowed:
+            try:
+                p_allowed_n.append(gitea_config.normalize_operation(op_val))
+            except Exception:
+                pass
+        p_forbidden_n = []
+        for op_val in p_forbidden:
+            try:
+                p_forbidden_n.append(gitea_config.normalize_operation(op_val))
+            except Exception:
+                pass
+        ok, _ = gitea_config.check_operation(op, p_allowed_n, p_forbidden_n)
+        if ok:
+            try:
+                tok = gitea_config.resolve_token(p_data)
+                if tok:
+                    gitea_config._active_profile_override = p_name
+                    _IDENTITY_CACHE.clear()
+                    return True
+            except Exception:
+                pass
+    return False
+
+
 def _profile_operation_gate(op: str) -> list[str]:
     """Profile permission check for a single gated operation (#126, #216).
 
@@ -3078,6 +3148,17 @@ def _profile_operation_gate(op: str) -> list[str]:
         op, profile["allowed_operations"], profile["forbidden_operations"])
     if op_ok:
         return []
+
+    if _try_auto_switch_for_operation(op):
+        try:
+            profile = get_profile()
+            op_ok, op_reason = gitea_config.check_operation(
+                op, profile["allowed_operations"], profile["forbidden_operations"])
+            if op_ok:
+                return []
+        except Exception as exc:
+            return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"]
+
     if op_reason == "no-allowed-operations":
         return ["profile has no configured allowed operations (fail closed)"]
     if op_reason == "forbidden":
@@ -4751,6 +4832,12 @@ def gitea_resolve_task_capability(
         required_permission, active_allowed, active_forbidden
     )
 
+    switching = gitea_config.is_runtime_switching_enabled()
+    available_in_session = allowed_in_current_session
+    configured = False
+    restart_required = False
+    reason_msg = None
+
     # Find matching configured profiles
     matching_profiles = []
     if config and "profiles" in config:
@@ -4775,17 +4862,19 @@ def gitea_resolve_task_capability(
 
     configured = len(matching_profiles) > 0
     available_in_session = allowed_in_current_session
-    restart_required = False
-    reason = ""
 
     if not allowed_in_current_session:
-        if configured:
+        if configured and switching:
             restart_required = True
-            reason = f"Reviewer profile exists but MCP server was added after session startup and is not attached."
-        else:
-            reason = f"No profile configured with permission '{required_permission}'."
-
-    switching = gitea_config.is_runtime_switching_enabled()
+            available_in_session = False
+            reason_msg = (
+                f"{required_role.capitalize()} profile exists but MCP server "
+                "was added after session startup and is not attached."
+            )
+        elif not configured:
+            reason_msg = (
+                f"No profile configured with permission '{required_permission}'."
+            )
     different_namespace_required = False
     next_safe_action = "None; ready for operations."
 
@@ -4863,17 +4952,19 @@ def gitea_resolve_task_capability(
         "active_identity": username,
         "active_profile_allowed_operations": active_allowed,
         "allowed_in_current_session": allowed_in_current_session,
-        "stop_required": stop_required,
+        "available_in_session": available_in_session,
+        "configured": configured,
+        "restart_required": restart_required,
+        "stop_required": stop_required or restart_required,
         "task_role_guidance": task_role_guidance,
         "matching_configured_profile": matching_profiles,
         "runtime_switching_supported": switching,
         "different_mcp_namespace_required": different_namespace_required,
         "exact_safe_next_action": next_safe_action,
-        "available_in_session": available_in_session,
-        "configured": configured,
-        "restart_required": restart_required,
-        "reason": reason,
     }
+    if reason_msg:
+        result["reason"] = reason_msg
+    role_session_router.sync_route_from_capability(result)
     was_terminal = capability_stop_terminal.is_active()
     terminal = capability_stop_terminal.sync_from_capability_result(result)
     if terminal:
@@ -4881,7 +4972,7 @@ def gitea_resolve_task_capability(
         result["terminal_report"] = (
             capability_stop_terminal.build_terminal_report(result)
         )
-    elif was_terminal and not stop_required:
+    elif was_terminal and not (stop_required or restart_required):
         result["cleared_stale_denial"] = True
     return result
 

role_session_router.py

@@ -194,11 +194,56 @@ def _record_route(result: dict):
     _session_last_route = dict(result)
 
 
+def sync_route_from_capability(capability: dict) -> None:
+    """Align sticky route state with operation-scoped capability resolution (#228)."""
+    capability = capability or {}
+    task = (capability.get("requested_task") or "").strip()
+    required_role = capability.get("required_role_kind")
+    if not task or not required_role:
+        return
+    if capability.get("allowed_in_current_session"):
+        _record_route({
+            "task_type": task,
+            "required_role": required_role,
+            "active_role": required_role,
+            "active_profile": capability.get("active_profile"),
+            "route_result": ROUTE_ALLOWED,
+            "downstream_allowed": True,
+            "reasons": [],
+            "message": (
+                f"Operation-scoped task '{task}' resolved for current session; "
+                "proceed."
+            ),
+        })
+        return
+    if required_role == "reviewer" and capability.get("stop_required"):
+        _record_route({
+            "task_type": task,
+            "required_role": required_role,
+            "active_role": capability.get("required_role_kind"),
+            "active_profile": capability.get("active_profile"),
+            "route_result": ROUTE_WRONG_ROLE,
+            "downstream_allowed": False,
+            "reasons": [WRONG_ROLE_REVIEWER_MSG],
+            "message": WRONG_ROLE_REVIEWER_MSG,
+        })
+
+
 def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool, list[str]]:
-    """Block author-side fallback after a reviewer wrong_role_stop (#206)."""
+    """Block author-side fallback after a reviewer wrong_role_stop (#206).
+
+    An explicit operation-scoped author capability resolution for the same
+    *mutation_task* clears the sticky reviewer denial (#228).
+    """
     last = _session_last_route
     if not last:
         return True, []
+    if (
+        last.get("route_result") == ROUTE_ALLOWED
+        and last.get("task_type") == mutation_task
+        and last.get("required_role") == "author"
+    ):
+        return True, []
     if last.get("route_result") != ROUTE_WRONG_ROLE:
         return True, []
     if last.get("required_role") != "reviewer":
@@ -207,8 +252,8 @@ def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool,
         return False, [
             WRONG_ROLE_REVIEWER_MSG,
             "Author-side mutations are blocked after a reviewer-task "
-            "wrong_role_stop unless the operator explicitly changes the "
-            "task and relaunches an author MCP session.",
+            "wrong_role_stop unless the operator explicitly resolves the "
+            "author task via gitea_resolve_task_capability.",
             f"Attempted fallback mutation: {mutation_task}",
         ]
     return True, []

tests/test_operation_scoped_roles.py

@@ -0,0 +1,181 @@
+"""Tests for operation-scoped role selection and automatic dispatch switching (#228)."""
+import os
+import sys
+import json
+import tempfile
+import unittest
+from unittest.mock import patch, MagicMock
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import gitea_config
+import gitea_auth
+import mcp_server
+from reviewer_worktree import assess_author_worktree_continuity
+
+CONFIG_TEST = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {
+                "enabled": True,
+                "base_url": "https://gitea.example.com"
+            }
+        }
+    },
+    "profiles": {
+        "author-profile": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "author-user",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": ["gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push", "gitea.issue.comment"],
+            "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
+            "execution_profile": "author-profile"
+        },
+        "reviewer-profile": {
+            "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", "gitea.issue.comment"],
+            "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
+            "execution_profile": "reviewer-profile"
+        }
+    }
+}
+
+
+class TestOperationScopedRoles(unittest.TestCase):
+    def setUp(self):
+        self._remotes_patch = patch.dict(mcp_server.REMOTES, {
+            "dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
+            "prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}
+        })
+        self._remotes_patch.start()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        mcp_server._MUTATION_AUTHORITY = None
+        self._dir = tempfile.TemporaryDirectory()
+        self.config_path = os.path.join(self._dir.name, "profiles.json")
+        self._write_config(CONFIG_TEST)
+
+    def tearDown(self):
+        self._remotes_patch.stop()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        mcp_server._MUTATION_AUTHORITY = None
+        self._dir.cleanup()
+
+    def _write_config(self, obj):
+        with open(self.config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(obj))
+
+    def _env(self, profile="author-profile", with_reviewer_token=True):
+        env = {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+        }
+        if with_reviewer_token:
+            env["GITEA_TOKEN_REVIEWER"] = "reviewer-pass"
+        return env
+
+    @patch("mcp_server.api_request")
+    def test_auto_switch_to_reviewer(self, mock_api):
+        # mock identity resolution to return username matching profile
+        mock_api.side_effect = lambda method, url, header: (
+            {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
+        )
+        with patch.dict(os.environ, self._env("author-profile")):
+            # initially we are author-profile
+            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
+            self.assertEqual(mcp_server.get_profile()["profile_name"], "author-profile")
+
+            # resolve a reviewer task (review_pr)
+            res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+            
+            # verify it automatically switched to reviewer-profile
+            self.assertTrue(res["allowed_in_current_session"])
+            self.assertTrue(res["available_in_session"])
+            self.assertEqual(res["active_profile"], "reviewer-profile")
+            self.assertEqual(res["active_identity"], "reviewer-user")
+            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
+
+    @patch("mcp_server.api_request")
+    def test_auto_switch_to_author(self, mock_api):
+        mock_api.side_effect = lambda method, url, header: (
+            {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
+        )
+        with patch.dict(os.environ, self._env("reviewer-profile")):
+            # initially we are reviewer-profile
+            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
+
+            # resolve an author task (create_issue)
+            res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
+            
+            # verify it automatically switched to author-profile
+            self.assertTrue(res["allowed_in_current_session"])
+            self.assertTrue(res["available_in_session"])
+            self.assertEqual(res["active_profile"], "author-profile")
+            self.assertEqual(res["active_identity"], "author-user")
+            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
+
+    @patch("mcp_server.api_request")
+    def test_restart_required_when_unattached(self, mock_api):
+        mock_api.side_effect = lambda method, url, header: {"login": "author-user"}
+        # launch without reviewer token in env
+        with patch.dict(os.environ, self._env("author-profile", with_reviewer_token=False)):
+            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
+
+            # resolve a reviewer task (review_pr)
+            res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+            
+            # verify it did NOT switch and reports restart_required
+            self.assertFalse(res["allowed_in_current_session"])
+            self.assertFalse(res["available_in_session"])
+            self.assertTrue(res["configured"])
+            self.assertTrue(res["restart_required"])
+            self.assertTrue(res["stop_required"])
+            self.assertIn("Reviewer profile exists but MCP server was added after session startup and is not attached", res["reason"])
+
+    def test_author_continuity_dirty_worktree(self):
+        # author is allowed to keep dirty worktree
+        res = assess_author_worktree_continuity({
+            "task_role": "author",
+            "dirty_files": ["review_proofs.py"],
+        })
+        self.assertTrue(res["allowed"])
+
+        # reviewer is blocked by reviewer worktree proof
+        res2 = assess_author_worktree_continuity({
+            "task_role": "reviewer",
+            "worktree_path": "/repo",
+            "dirty_files": ["review_proofs.py"],
+            "pr_scope_files": ["docs/wiki/Repositories.md"],
+            "scratch_used": False,
+        })
+        self.assertFalse(res2["proven"])
+
+    @patch("mcp_server.api_request")
+    def test_mutating_actions_auto_switch(self, mock_api):
+        mock_api.side_effect = lambda method, url, header: (
+            {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
+        )
+        with patch.dict(os.environ, self._env("author-profile")):
+            # verify we are author-profile
+            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
+
+            # call a reviewer mutation check helper (like _profile_permission_block with reviewer permission)
+            blocked = mcp_server._profile_permission_block("gitea.pr.merge", remote="prgs")
+            self.assertIsNone(blocked) # should switch to reviewer-profile and allow it (no permission block)
+
+            # verify we dynamically switched to reviewer-profile
+            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_role_session_router.py

@@ -119,6 +119,25 @@ class TestRoleSessionRouter(unittest.TestCase):
         ]
         self.assertEqual(issue_posts, [])
 
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request", return_value={"number": 888})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_author_task_allowed_after_capability_resolve(
+        self, _auth, _api, _get_all
+    ):
+        """#228: explicit create_issue capability clears reviewer wrong_role_stop."""
+        with patch.dict(os.environ, self._env("prgs-author")):
+            mcp_server.gitea_route_task_session(task_type="review_pr", remote="prgs")
+            cap = mcp_server.gitea_resolve_task_capability(
+                task="create_issue", remote="prgs"
+            )
+            self.assertTrue(cap["allowed_in_current_session"])
+            result = mcp_server.gitea_create_issue(
+                title="operation-scoped author recovery",
+                remote="prgs",
+            )
+        self.assertEqual(result.get("number"), 888)
+
     @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.api_request", return_value={"number": 999})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")