From 75ed0632b43620afc3072f98972da47f77e33664 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 11:27:39 -0400 Subject: [PATCH 1/3] feat(mcp): implement operation-scoped role selection and automatic dispatch switching (#228) --- gitea_config.py | 10 +- tests/test_operation_scoped_roles.py | 181 +++++++++++++++++++++++++++ 2 files changed, 190 insertions(+), 1 deletion(-) create mode 100644 tests/test_operation_scoped_roles.py diff --git a/gitea_config.py b/gitea_config.py index a4b0821..ffbf7ea 100644 --- a/gitea_config.py +++ b/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 diff --git a/tests/test_operation_scoped_roles.py b/tests/test_operation_scoped_roles.py new file mode 100644 index 0000000..c382085 --- /dev/null +++ b/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() From 7dde2f54057139b751775b0e6cac54be43a16216 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 12:20:37 -0400 Subject: [PATCH 2/3] fix(mcp): implement operation-scoped role selection and auto profile switching --- gitea_mcp_server.py | 344 ++++++++++++++++++++++++------ role_session_router.py | 51 ++++- tests/test_role_session_router.py | 19 ++ 3 files changed, 346 insertions(+), 68 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index f173581..1c5c186 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -164,81 +164,194 @@ _preflight_capability_called = False _preflight_whoami_violation = False _preflight_capability_violation = False _preflight_resolved_role = None +_process_start_porcelain: str | None = None +_preflight_whoami_baseline_porcelain: str | None = None +_preflight_capability_baseline_porcelain: str | None = None +_preflight_whoami_violation_files: list[str] = [] +_preflight_capability_violation_files: list[str] = [] +_preflight_reviewer_violation_files: list[str] = [] + + +def _preflight_in_test_mode() -> bool: + return "pytest" in sys.modules or "unittest" in sys.modules + + +def _ensure_process_start_porcelain() -> str: + """Capture the shared-worktree baseline once per MCP process (#252).""" + global _process_start_porcelain + if _process_start_porcelain is None: + _process_start_porcelain = _get_workspace_porcelain() + return _process_start_porcelain + + +def _get_workspace_porcelain() -> str: + """Return tracked-workspace porcelain for pre-flight attribution.""" + if os.environ.get("GITEA_TEST_FORCE_DIRTY"): + return " M __gitea_test_force_dirty__.py\n" + override = os.environ.get("GITEA_TEST_PORCELAIN") + if override is not None: + return override + try: + res = subprocess.run( + ["git", "status", "--porcelain"], + capture_output=True, + text=True, + cwd=PROJECT_ROOT, + ) + return res.stdout or "" + except Exception: + return "" + + +def _parse_porcelain_entries(porcelain: str) -> dict[str, str]: + """Map tracked path -> full porcelain line (untracked ``??`` ignored).""" + entries: dict[str, str] = {} + for line in (porcelain or "").splitlines(): + if not line or len(line) < 4 or line.startswith("??"): + continue + path = line[3:].strip() + if " -> " in path: + path = path.split(" -> ", 1)[1].strip() + if path: + entries[path] = line + return entries + + +def _new_tracked_changes_since(baseline: str, current: str) -> list[str]: + """Tracked paths that are new or changed since *baseline* porcelain.""" + base = _parse_porcelain_entries(baseline) + cur = _parse_porcelain_entries(current) + changed = [ + path for path, line in cur.items() + if path not in base or base[path] != line + ] + return sorted(changed) + + +def _format_preflight_files(files: list[str]) -> str: + if not files: + return "(none)" + return ", ".join(files) + + +def assess_preflight_status() -> dict: + """Non-throwing pre-flight readiness for runtime-context alignment (#252).""" + reasons: list[str] = [] + if not _preflight_whoami_called: + reasons.append( + "Identity (gitea_whoami) has not been verified" + ) + if not _preflight_capability_called: + reasons.append( + "Task capability (gitea_resolve_task_capability) has not been resolved" + ) + if _preflight_whoami_violation: + reasons.append( + "Workspace file edits occurred before gitea_whoami verification " + f"(offending files: {_format_preflight_files(_preflight_whoami_violation_files)})" + ) + if _preflight_capability_violation: + reasons.append( + "Workspace file edits occurred before gitea_resolve_task_capability verification " + f"(offending files: {_format_preflight_files(_preflight_capability_violation_files)})" + ) + if _preflight_reviewer_violation_files: + reasons.append( + "Reviewer profile modified tracked workspace files after capability resolution " + f"(offending files: {_format_preflight_files(_preflight_reviewer_violation_files)})" + ) + return { + "preflight_ready": not reasons, + "preflight_block_reasons": reasons, + "preflight_whoami_verified": _preflight_whoami_called, + "preflight_capability_resolved": _preflight_capability_called, + "preflight_whoami_violation_files": list(_preflight_whoami_violation_files), + "preflight_capability_violation_files": list(_preflight_capability_violation_files), + "preflight_reviewer_violation_files": list(_preflight_reviewer_violation_files), + } + def record_preflight_check(type_name: str, resolved_role: str | None = None): - """Record a pre-flight check (whoami or capability) and check for workspace edits.""" + """Record a pre-flight check (whoami or capability) with session-scoped deltas.""" global _preflight_whoami_called, _preflight_capability_called global _preflight_whoami_violation, _preflight_capability_violation global _preflight_resolved_role + global _preflight_whoami_baseline_porcelain, _preflight_capability_baseline_porcelain + global _preflight_whoami_violation_files, _preflight_capability_violation_files + global _preflight_reviewer_violation_files - in_test = "pytest" in sys.modules or "unittest" in sys.modules - if os.environ.get("GITEA_TEST_FORCE_DIRTY"): - is_dirty = True - elif in_test: - is_dirty = False - else: - is_dirty = False - try: - res = subprocess.run( - ["git", "status", "--porcelain"], - capture_output=True, text=True, cwd=PROJECT_ROOT - ) - for line in res.stdout.splitlines(): - if line and not line.startswith("??"): - is_dirty = True - break - except Exception: - pass - - if is_dirty: - if type_name == "whoami" and not _preflight_whoami_called: - _preflight_whoami_violation = True - if type_name == "capability" and not _preflight_capability_called: - _preflight_capability_violation = True + current = _get_workspace_porcelain() if type_name == "whoami": + # Fresh whoami restarts the capability step and re-evaluates violations + # instead of replaying a sticky record (#252). + _preflight_capability_called = False + _preflight_capability_violation = False + _preflight_capability_violation_files = [] + _preflight_capability_baseline_porcelain = None + _preflight_reviewer_violation_files = [] + + process_start = _ensure_process_start_porcelain() + whoami_delta = _new_tracked_changes_since(process_start, current) + _preflight_whoami_violation = bool(whoami_delta) + _preflight_whoami_violation_files = whoami_delta + _preflight_whoami_baseline_porcelain = current _preflight_whoami_called = True elif type_name == "capability": + baseline = _preflight_whoami_baseline_porcelain or "" + capability_delta = _new_tracked_changes_since(baseline, current) + _preflight_capability_violation = bool(capability_delta) + _preflight_capability_violation_files = capability_delta + _preflight_capability_baseline_porcelain = current _preflight_capability_called = True if resolved_role: _preflight_resolved_role = resolved_role + def verify_preflight_purity(remote: str | None = None): - """Verify that identity and capability were verified prior to edits, and that reviewers made no edits.""" - in_test = "pytest" in sys.modules or "unittest" in sys.modules - if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"): + """Verify that identity and capability were verified prior to session edits.""" + global _preflight_reviewer_violation_files + + in_test = _preflight_in_test_mode() + if in_test and not ( + os.environ.get("GITEA_TEST_FORCE_DIRTY") + or os.environ.get("GITEA_TEST_PORCELAIN") is not None + ): return if not _preflight_whoami_called: - raise RuntimeError("Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)") + raise RuntimeError( + "Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)" + ) if not _preflight_capability_called: - raise RuntimeError("Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)") + raise RuntimeError( + "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)" + ) if _preflight_whoami_violation: - raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_whoami verification (fail closed)") + raise RuntimeError( + "Pre-flight order violation: Workspace file edits occurred before " + f"gitea_whoami verification (fail closed). Offending files: " + f"{_format_preflight_files(_preflight_whoami_violation_files)}" + ) if _preflight_capability_violation: - raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_resolve_task_capability verification (fail closed)") + raise RuntimeError( + "Pre-flight order violation: Workspace file edits occurred before " + f"gitea_resolve_task_capability verification (fail closed). Offending files: " + f"{_format_preflight_files(_preflight_capability_violation_files)}" + ) - if os.environ.get("GITEA_TEST_FORCE_DIRTY"): - is_dirty = True - elif in_test: - is_dirty = False - else: - is_dirty = False - try: - res = subprocess.run( - ["git", "status", "--porcelain"], - capture_output=True, text=True, cwd=PROJECT_ROOT + if _preflight_resolved_role == "reviewer": + current = _get_workspace_porcelain() + baseline = _preflight_capability_baseline_porcelain or "" + reviewer_delta = _new_tracked_changes_since(baseline, current) + _preflight_reviewer_violation_files = reviewer_delta + if reviewer_delta: + raise RuntimeError( + "Reviewer role violation: Reviewer profile is forbidden from modifying " + "tracked workspace files (fail closed). Offending files: " + f"{_format_preflight_files(reviewer_delta)}" ) - for line in res.stdout.splitlines(): - if line and not line.startswith("??"): - is_dirty = True - break - except Exception: - pass - - if _preflight_resolved_role == "reviewer" and is_dirty: - raise RuntimeError("Reviewer role violation: Reviewer profile is forbidden from modifying tracked workspace files (fail closed)") from mcp.server.fastmcp import FastMCP # noqa: E402 @@ -2948,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). @@ -2965,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": @@ -3967,6 +4161,14 @@ def gitea_get_runtime_context( "or ask the operator to update GITEA_MCP_PROFILE to a reviewer profile." ) + preflight = assess_preflight_status() + if not preflight["preflight_ready"]: + safe_next_action = ( + "Complete pre-flight verification before mutating: call gitea_whoami, then " + "gitea_resolve_task_capability for the intended task. " + f"Blocked: {'; '.join(preflight['preflight_block_reasons'])}" + ) + result = { "active_profile": profile["profile_name"], "authenticated_username": username, @@ -3981,6 +4183,8 @@ def gitea_get_runtime_context( "review_merge_blocked_reasons": blocked_reasons, "suggested_fix": suggested_fix, "safe_next_action": safe_next_action, + "preflight_ready": preflight["preflight_ready"], + "preflight_block_reasons": preflight["preflight_block_reasons"], } if reveal and h: @@ -4628,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: @@ -4652,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." @@ -4740,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: @@ -4758,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 diff --git a/role_session_router.py b/role_session_router.py index 23eb945..d0422a0 100644 --- a/role_session_router.py +++ b/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, [] diff --git a/tests/test_role_session_router.py b/tests/test_role_session_router.py index 7aedd53..a2cdffb 100644 --- a/tests/test_role_session_router.py +++ b/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") From 4faf839dabd570f01e528c70fec33b51e546fb96 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 12:39:01 -0400 Subject: [PATCH 3/3] test: align preflight tests with session-scoped porcelain baselines Update TestPreflightVerification for #252 delta-based preflight checks on the rebased #228 branch. Refs #228 Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_mcp_server.py | 49 ++++++++++++++++++++++++++++++---------- 1 file changed, 37 insertions(+), 12 deletions(-) diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 328e749..044d76b 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -3158,6 +3158,12 @@ class TestPreflightVerification(unittest.TestCase): self.orig_whoami_violation = mcp_server._preflight_whoami_violation self.orig_capability_violation = mcp_server._preflight_capability_violation self.orig_resolved_role = mcp_server._preflight_resolved_role + self.orig_process_start = mcp_server._process_start_porcelain + self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain + self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain + self.orig_whoami_files = mcp_server._preflight_whoami_violation_files + self.orig_capability_files = mcp_server._preflight_capability_violation_files + self.orig_reviewer_files = mcp_server._preflight_reviewer_violation_files # Reset state for each test mcp_server._preflight_whoami_called = False @@ -3165,6 +3171,13 @@ class TestPreflightVerification(unittest.TestCase): mcp_server._preflight_whoami_violation = False mcp_server._preflight_capability_violation = False mcp_server._preflight_resolved_role = None + mcp_server._process_start_porcelain = "" + mcp_server._preflight_whoami_baseline_porcelain = None + mcp_server._preflight_capability_baseline_porcelain = None + mcp_server._preflight_whoami_violation_files = [] + mcp_server._preflight_capability_violation_files = [] + mcp_server._preflight_reviewer_violation_files = [] + os.environ["GITEA_TEST_PORCELAIN"] = "" def tearDown(self): # Restore real global variables @@ -3174,30 +3187,42 @@ class TestPreflightVerification(unittest.TestCase): mcp_server._preflight_whoami_violation = self.orig_whoami_violation mcp_server._preflight_capability_violation = self.orig_capability_violation mcp_server._preflight_resolved_role = self.orig_resolved_role - if "GITEA_TEST_FORCE_DIRTY" in os.environ: - del os.environ["GITEA_TEST_FORCE_DIRTY"] + mcp_server._process_start_porcelain = self.orig_process_start + mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline + mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline + mcp_server._preflight_whoami_violation_files = self.orig_whoami_files + mcp_server._preflight_capability_violation_files = self.orig_capability_files + mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files + for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"): + if key in os.environ: + del os.environ[key] def test_preflight_whoami_violation(self): import mcp_server os.environ["GITEA_TEST_FORCE_DIRTY"] = "1" - mcp_server._preflight_capability_called = True mcp_server.record_preflight_check("whoami") self.assertTrue(mcp_server._preflight_whoami_violation) + mcp_server.record_preflight_check("capability", resolved_role="author") with self.assertRaises(RuntimeError) as ctx: mcp_server.verify_preflight_purity() self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception)) + self.assertIn("Offending files:", str(ctx.exception)) def test_preflight_capability_violation(self): import mcp_server + mcp_server.record_preflight_check("whoami") os.environ["GITEA_TEST_FORCE_DIRTY"] = "1" - mcp_server._preflight_whoami_called = True mcp_server.record_preflight_check("capability", resolved_role="author") self.assertTrue(mcp_server._preflight_capability_violation) with self.assertRaises(RuntimeError) as ctx: mcp_server.verify_preflight_purity() - self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability verification", str(ctx.exception)) + self.assertIn( + "Workspace file edits occurred before gitea_resolve_task_capability verification", + str(ctx.exception), + ) + self.assertIn("Offending files:", str(ctx.exception)) def test_preflight_not_called_fails_closed(self): import mcp_server @@ -3213,16 +3238,16 @@ class TestPreflightVerification(unittest.TestCase): def test_preflight_reviewer_mutation_violation(self): import mcp_server - mcp_server._preflight_whoami_called = True - mcp_server._preflight_capability_called = True - mcp_server._preflight_resolved_role = "reviewer" + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check("capability", resolved_role="reviewer") - # When dirty, reviewer edits are blocked - os.environ["GITEA_TEST_FORCE_DIRTY"] = "1" + # Session-owned reviewer edits after capability are blocked. + os.environ["GITEA_TEST_PORCELAIN"] = " M reviewer_edit.py\n" with self.assertRaises(RuntimeError) as ctx: mcp_server.verify_preflight_purity() self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception)) + self.assertIn("reviewer_edit.py", str(ctx.exception)) - # When clean, reviewer is allowed - del os.environ["GITEA_TEST_FORCE_DIRTY"] + # Foreign pre-existing dirty state does not block when unchanged. + os.environ["GITEA_TEST_PORCELAIN"] = "" mcp_server.verify_preflight_purity()