fix(mcp): implement operation-scoped role selection and auto profile switching

This commit is contained in:
2026-07-06 12:38:20 -04:00
parent 75ed0632b4
commit 7dde2f5405
3 changed files with 346 additions and 68 deletions
+48 -3
View File
@@ -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, []