From a256caaae7dfc528dc7777f90bac80da29316de4 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Sun, 5 Jul 2026 17:13:18 -0400 Subject: [PATCH] feat: add pre-task role/session router for reviewer tasks (Issue #206) Add gitea_route_task_session and role_session_router to fail closed when reviewer tasks start under author-bound MCP sessions. Block author-side issue creation fallback after wrong_role_stop. Add handoff proof helper and tests. Closes #206 Co-Authored-By: Claude Opus 4.8 (1M context) --- mcp_server.py | 77 ++++++++++++ review_proofs.py | 37 ++++++ role_session_router.py | 195 ++++++++++++++++++++++++++++++ tests/test_role_session_router.py | 188 ++++++++++++++++++++++++++++ 4 files changed, 497 insertions(+) create mode 100644 role_session_router.py create mode 100644 tests/test_role_session_router.py diff --git a/mcp_server.py b/mcp_server.py index 5f26704..1698f03 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -45,6 +45,7 @@ from gitea_auth import ( # noqa: E402 ) import gitea_audit # noqa: E402 import gitea_config # noqa: E402 +import role_session_router # noqa: E402 def _reveal_endpoints() -> bool: @@ -324,6 +325,16 @@ def gitea_create_issue( Returns: dict with 'number' of the created issue ('url' only with the reveal opt-in). """ + ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop( + "create_issue" + ) + if not ok: + return { + "success": False, + "performed": False, + "number": None, + "reasons": block_reasons, + } h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) url = f"{repo_api_url(h, o, r)}/issues" @@ -3304,6 +3315,60 @@ def build_validation_report(commands: list[dict]) -> dict: } +@mcp.tool() +def gitea_route_task_session( + task_type: str, + remote: str = "dadeschools", + host: str | None = None, +) -> dict: + """Pre-task role/session router (#206). + + Classify *task_type* against the active MCP profile before any mutation. + Returns ``route_result`` — only ``allowed_current_session`` permits + downstream tool use. Reviewer tasks under an author-bound session return + ``wrong_role_stop`` with no fallback to author mutations. + """ + task_type = (task_type or "").strip() + profile = get_profile() + allowed = profile.get("allowed_operations") or [] + forbidden = profile.get("forbidden_operations") or [] + active_role = _role_kind(allowed, forbidden) + + if not task_type: + return role_session_router.route_task_session( + "", + active_profile=profile["profile_name"], + active_role_kind=active_role, + allowed_in_current_session=False, + ) + + capability = None + try: + capability = gitea_resolve_task_capability( + task=task_type, + remote=remote, + host=host, + ) + except ValueError: + capability = None + + if capability is not None: + return role_session_router.route_task_session( + task_type, + active_profile=capability["active_profile"], + active_role_kind=active_role, + allowed_in_current_session=capability["allowed_in_current_session"], + runtime_switching_supported=capability["runtime_switching_supported"], + ) + + return role_session_router.route_task_session( + task_type, + active_profile=profile["profile_name"], + active_role_kind=active_role, + allowed_in_current_session=False, + ) + + @mcp.tool() def gitea_resolve_task_capability( task: str, @@ -3365,6 +3430,18 @@ def gitea_resolve_task_capability( "permission": "gitea.pr.merge", "role": "reviewer", }, + "blind_pr_queue_review": { + "permission": "gitea.pr.review", + "role": "reviewer", + }, + "request_changes_pr": { + "permission": "gitea.pr.request_changes", + "role": "reviewer", + }, + "approve_pr": { + "permission": "gitea.pr.approve", + "role": "reviewer", + }, "delete_branch": { "permission": "gitea.branch.delete", "role": "author", diff --git a/review_proofs.py b/review_proofs.py index b031a43..c08bb35 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -635,6 +635,43 @@ def assess_controller_handoff(report_text, role=None): } +ROUTE_HANDOFF_FIELDS = ( + ("Task type", ("task type", "task_type")), + ("Required role", ("required role", "required_role")), + ("Active role", ("active role", "active_role")), + ("Route result", ("route result", "route_result")), +) + + +def assess_role_route_handoff(report_text, route_result=None): + """Issue #206: final handoff must record role routing verdict.""" + text = report_text or "" + lower = text.lower() + missing = [] + for name, aliases in ROUTE_HANDOFF_FIELDS: + if not any(alias in lower for alias in aliases): + missing.append(name) + if route_result is not None: + expected = str(route_result.get("route_result", "")).lower() + if expected and expected not in lower: + missing.append("route result value") + if missing: + return { + "complete": False, + "downgraded": True, + "missing_fields": missing, + "reasons": [ + f"handoff missing role-route field: {field}" for field in missing + ], + } + return { + "complete": True, + "downgraded": False, + "missing_fields": [], + "reasons": [], + } + + # ── PR Inventory Trust Gate (Issue #194) ────────────────────────────────────── # # A reviewer agent may not convert an empty PR list response into a definitive diff --git a/role_session_router.py b/role_session_router.py new file mode 100644 index 0000000..b9a0406 --- /dev/null +++ b/role_session_router.py @@ -0,0 +1,195 @@ +"""Pre-task role/session router (#206). + +Classifies a declared task type against the active MCP profile/session and +returns a route result before any downstream mutation tools run. +""" + +from __future__ import annotations + +ROUTE_ALLOWED = "allowed_current_session" +ROUTE_WRONG_ROLE = "wrong_role_stop" +ROUTE_TO_AUTHOR = "route_to_author_session" +ROUTE_TO_REVIEWER = "route_to_reviewer_session" +ROUTE_AMBIGUOUS = "ambiguous_task_stop" + +REVIEWER_TASKS = frozenset({ + "review_pr", + "merge_pr", + "blind_pr_queue_review", + "request_changes_pr", + "approve_pr", +}) + +AUTHOR_TASKS = frozenset({ + "create_issue", + "comment_issue", + "close_issue", + "claim_issue", + "create_branch", + "push_branch", + "create_pr", + "comment_pr", + "address_pr_change_requests", + "delete_branch", +}) + +TASK_REQUIRED_ROLE = { + "create_issue": "author", + "comment_issue": "author", + "close_issue": "author", + "claim_issue": "author", + "create_branch": "author", + "push_branch": "author", + "create_pr": "author", + "comment_pr": "author", + "address_pr_change_requests": "author", + "delete_branch": "author", + "review_pr": "reviewer", + "merge_pr": "reviewer", + "blind_pr_queue_review": "reviewer", + "request_changes_pr": "reviewer", + "approve_pr": "reviewer", +} + +WRONG_ROLE_REVIEWER_MSG = ( + "Wrong role/session for reviewer task. Launch reviewer MCP namespace." +) + +_session_last_route: dict | None = None + + +def required_role_for_task(task_type: str) -> str | None: + return TASK_REQUIRED_ROLE.get((task_type or "").strip()) + + +def route_task_session( + task_type: str, + *, + active_profile: str, + active_role_kind: str, + allowed_in_current_session: bool, + runtime_switching_supported: bool = False, +) -> dict: + """Return routing verdict for *task_type* under the active session.""" + task_type = (task_type or "").strip() + required_role = required_role_for_task(task_type) + if required_role is None: + result = { + "task_type": task_type, + "required_role": None, + "active_role": active_role_kind, + "active_profile": active_profile, + "route_result": ROUTE_AMBIGUOUS, + "downstream_allowed": False, + "reasons": [ + f"unknown task type '{task_type}'; cannot route session " + "(fail closed)" + ], + "message": ( + "Ambiguous task type; relaunch with an explicit task before " + "any tool use." + ), + } + _record_route(result) + return result + + if allowed_in_current_session: + result = { + "task_type": task_type, + "required_role": required_role, + "active_role": active_role_kind, + "active_profile": active_profile, + "route_result": ROUTE_ALLOWED, + "downstream_allowed": True, + "reasons": [], + "message": "Task role matches active session; proceed.", + } + _record_route(result) + return result + + if required_role == "reviewer": + 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_REVIEWER_MSG, + "Reviewer tasks cannot run in author-bound sessions.", + "Static-profile mode does not permit in-place role switching.", + ], + "message": WRONG_ROLE_REVIEWER_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 = ( + "Wrong role/session for author task. Launch author MCP namespace." + ) + result = { + "task_type": task_type, + "required_role": required_role, + "active_role": active_role_kind, + "active_profile": active_profile, + "route_result": route, + "downstream_allowed": False, + "reasons": [message], + "message": message, + "runtime_switching_supported": runtime_switching_supported, + "profile_switch_blocked": not runtime_switching_supported, + } + _record_route(result) + return result + + result = { + "task_type": task_type, + "required_role": required_role, + "active_role": active_role_kind, + "active_profile": active_profile, + "route_result": ROUTE_AMBIGUOUS, + "downstream_allowed": False, + "reasons": ["unable to classify task role (fail closed)"], + "message": "Ambiguous task type; stop before any mutation.", + } + _record_route(result) + return result + + +def last_route() -> dict | None: + return _session_last_route + + +def clear_route_state(): + global _session_last_route + _session_last_route = None + + +def _record_route(result: dict): + global _session_last_route + _session_last_route = dict(result) + + +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).""" + last = _session_last_route + if not last: + return True, [] + if last.get("route_result") != ROUTE_WRONG_ROLE: + return True, [] + if last.get("required_role") != "reviewer": + return True, [] + if mutation_task in AUTHOR_TASKS: + 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.", + f"Attempted fallback mutation: {mutation_task}", + ] + return True, [] \ No newline at end of file diff --git a/tests/test_role_session_router.py b/tests/test_role_session_router.py new file mode 100644 index 0000000..3ba3db6 --- /dev/null +++ b/tests/test_role_session_router.py @@ -0,0 +1,188 @@ +"""Tests for pre-task role/session router (#206).""" +import json +import os +import sys +import tempfile +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import gitea_config +import mcp_server +import role_session_router +from review_proofs import assess_role_route_handoff + +CONFIG = { + "version": 2, + "contexts": { + "ctx": { + "enabled": True, + "gitea": {"enabled": True, "base_url": "https://gitea.example.com"}, + } + }, + "profiles": { + "prgs-author": { + "enabled": True, + "context": "ctx", + "role": "author", + "username": "jcwalker3", + "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", "gitea.pr.review", + ], + "execution_profile": "prgs-author", + }, + "prgs-reviewer": { + "enabled": True, + "context": "ctx", + "role": "reviewer", + "username": "sysadmin", + "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", "gitea.issue.create", + ], + "execution_profile": "prgs-reviewer", + }, + }, + "rules": {"allow_runtime_switching": False}, +} + + +class TestRoleSessionRouter(unittest.TestCase): + def setUp(self): + role_session_router.clear_route_state() + self._remotes = patch.dict(mcp_server.REMOTES, { + "prgs": { + "host": "gitea.example.com", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools", + }, + }) + self._remotes.start() + mcp_server._IDENTITY_CACHE.clear() + gitea_config._active_profile_override = None + self._dir = tempfile.TemporaryDirectory() + self.config_path = os.path.join(self._dir.name, "profiles.json") + with open(self.config_path, "w", encoding="utf-8") as fh: + fh.write(json.dumps(CONFIG)) + + def tearDown(self): + self._remotes.stop() + role_session_router.clear_route_state() + mcp_server._IDENTITY_CACHE.clear() + gitea_config._active_profile_override = None + self._dir.cleanup() + + def _env(self, profile): + return { + "GITEA_MCP_CONFIG": self.config_path, + "GITEA_MCP_PROFILE": profile, + "GITEA_TOKEN_AUTHOR": "author-pass", + "GITEA_TOKEN_REVIEWER": "reviewer-pass", + } + + def test_reviewer_task_under_author_profile_wrong_role_stop(self): + with patch.dict(os.environ, self._env("prgs-author")): + route = mcp_server.gitea_route_task_session( + task_type="review_pr", remote="prgs" + ) + self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE) + self.assertFalse(route["downstream_allowed"]) + self.assertIn("Wrong role/session for reviewer task", route["message"]) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_issue_creation_blocked_after_reviewer_wrong_role_stop( + self, _auth, mock_api + ): + with patch.dict(os.environ, self._env("prgs-author")): + mcp_server.gitea_route_task_session(task_type="review_pr", remote="prgs") + result = mcp_server.gitea_create_issue( + title="process issue fallback", + body="should not post", + remote="prgs", + ) + self.assertFalse(result.get("success", True)) + self.assertFalse(result.get("performed", True)) + issue_posts = [ + c for c in mock_api.call_args_list + if c.args[0] == "POST" and str(c.args[1]).endswith("/issues") + ] + self.assertEqual(issue_posts, []) + + @patch("mcp_server.api_request", return_value={"number": 999}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_author_task_allowed_after_explicit_author_route( + self, _auth, _api + ): + with patch.dict(os.environ, self._env("prgs-author")): + route = mcp_server.gitea_route_task_session( + task_type="create_issue", remote="prgs" + ) + self.assertEqual( + route["route_result"], role_session_router.ROUTE_ALLOWED + ) + result = mcp_server.gitea_create_issue( + title="legit author task", + remote="prgs", + ) + self.assertEqual(result.get("number"), 999) + + @patch("mcp_server.api_request", return_value={"login": "sysadmin"}) + @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") + def test_reviewer_task_under_reviewer_profile_allowed(self, _auth, _api): + with patch.dict(os.environ, self._env("prgs-reviewer")): + 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"]) + + @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): + with patch.dict(os.environ, self._env("prgs-reviewer")): + route = mcp_server.gitea_route_task_session( + task_type="create_issue", remote="prgs" + ) + self.assertEqual( + route["route_result"], role_session_router.ROUTE_TO_AUTHOR + ) + self.assertFalse(route["downstream_allowed"]) + + def test_activate_profile_blocked_in_static_mode(self): + with patch.dict(os.environ, self._env("prgs-author")): + result = mcp_server.gitea_activate_profile( + profile_name="prgs-reviewer", remote="prgs" + ) + self.assertFalse(result["success"]) + self.assertIn("switching is disabled", result["message"]) + + def test_handoff_requires_route_fields(self): + incomplete = assess_role_route_handoff("Task done.") + self.assertFalse(incomplete["complete"]) + self.assertIn("Task type", incomplete["missing_fields"]) + + complete = assess_role_route_handoff( + "\n".join([ + "Task type: review_pr", + "Required role: reviewer", + "Active role: author", + "Route result: wrong_role_stop", + ]), + route_result={"route_result": "wrong_role_stop"}, + ) + self.assertTrue(complete["complete"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file -- 2.43.7