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