feat: add native MCP preference gate and shell circuit breaker (Closes #270)
Enforce native MCP-first paths for Gitea mutations with fail-closed assessment helpers, shell spawn circuit breaker tracking, and MCP tools for operation-path assessment and shell health reporting. Closes #270
This commit is contained in:
@@ -477,6 +477,7 @@ import review_proofs # noqa: E402
|
||||
import agent_temp_artifacts
|
||||
import issue_lock_worktree # noqa: E402
|
||||
import merged_cleanup_reconcile # noqa: E402
|
||||
import native_mcp_preference # noqa: E402
|
||||
|
||||
|
||||
# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
|
||||
@@ -4082,6 +4083,14 @@ _GUIDE_RULES = {
|
||||
"inventory, summarize) needs no explicit authorization. Enforced "
|
||||
"fail closed via subagent_gate.assess_subagent_delegation and "
|
||||
"validate_subagent_report (#266)."),
|
||||
"native_mcp_preference": (
|
||||
"Gitea mutations must use native MCP tools first. When MCP is "
|
||||
"available, block shell scripts, direct API calls, WebFetch, "
|
||||
"Playwright, helper encoders, and profile-secret reads unless "
|
||||
"explicit recovery-mode proof is complete. If MCP transport is "
|
||||
"broken or the shell circuit breaker trips, emit a terminal recovery "
|
||||
"report instead of improvising unsafe fallback. Reviewers must not "
|
||||
"touch or restart MCP server files/processes (#270)."),
|
||||
}
|
||||
|
||||
_COMMON_WORKFLOWS = [
|
||||
@@ -4786,6 +4795,69 @@ def _build_runtime_task_capabilities(
|
||||
}
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
@mcp.tool()
|
||||
def gitea_assess_gitea_operation_path(
|
||||
task: str,
|
||||
path_kind: str | None = None,
|
||||
command_or_detail: str | None = None,
|
||||
mcp_available: bool = True,
|
||||
mcp_tool_visible: bool = True,
|
||||
recovery_mode: bool = False,
|
||||
recovery_proof_complete: bool = False,
|
||||
remote: str = "dadeschools",
|
||||
host: str | None = None,
|
||||
) -> dict:
|
||||
"""Read-only: assess whether a proposed Gitea mutation path is allowed (#270).
|
||||
|
||||
Native MCP must be preferred when available. Shell scripts, direct API
|
||||
calls, WebFetch, Playwright, and helper encoders are blocked unless
|
||||
explicit recovery-mode proof is supplied.
|
||||
"""
|
||||
profile = get_profile()
|
||||
role = _role_kind(
|
||||
profile.get("allowed_operations") or [],
|
||||
profile.get("forbidden_operations") or [],
|
||||
)
|
||||
return native_mcp_preference.assess_gitea_operation_path(
|
||||
task=task,
|
||||
path_kind=path_kind,
|
||||
command_or_detail=command_or_detail,
|
||||
mcp_available=mcp_available,
|
||||
mcp_tool_visible=mcp_tool_visible,
|
||||
recovery_mode=recovery_mode,
|
||||
recovery_proof_complete=recovery_proof_complete,
|
||||
role=role,
|
||||
active_profile=profile.get("profile_name"),
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_record_shell_spawn_outcome(
|
||||
exit_code: int | None = None,
|
||||
stdout: str = "",
|
||||
stderr: str = "",
|
||||
spawn_failure: bool | None = None,
|
||||
probe_attempted: bool = False,
|
||||
probe_succeeded: bool | None = None,
|
||||
) -> dict:
|
||||
"""Record shell spawn outcome and update the session circuit breaker (#270)."""
|
||||
return native_mcp_preference.record_shell_spawn_outcome(
|
||||
exit_code=exit_code,
|
||||
stdout=stdout,
|
||||
stderr=stderr,
|
||||
spawn_failure=spawn_failure,
|
||||
probe_attempted=probe_attempted,
|
||||
probe_succeeded=probe_succeeded,
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_get_shell_health() -> dict:
|
||||
"""Read-only: return shell spawn circuit-breaker state for this MCP session (#270)."""
|
||||
return native_mcp_preference.shell_health_status()
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_get_runtime_context(
|
||||
remote: str = "dadeschools",
|
||||
@@ -4904,6 +4976,7 @@ def gitea_get_runtime_context(
|
||||
"preflight_block_reasons": preflight["preflight_block_reasons"],
|
||||
"preflight_workspace": preflight.get("preflight_workspace"),
|
||||
"session_capabilities": session_capabilities,
|
||||
"shell_health": native_mcp_preference.shell_health_status(),
|
||||
}
|
||||
|
||||
if reveal and h:
|
||||
|
||||
@@ -0,0 +1,369 @@
|
||||
"""Native MCP preference gate and shell health circuit breaker (#270).
|
||||
|
||||
Gitea mutations must use native MCP tools first. Shell scripts, direct API
|
||||
calls, browser helpers, and improvised encoders are blocked when MCP is
|
||||
available unless explicit recovery-mode proof is supplied.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from reviewer_fallback import LOCAL_GITEA_SCRIPT_NAMES
|
||||
|
||||
# Tasks that must prefer native MCP over shell/API/helper fallbacks.
|
||||
GITEA_MUTATION_TASKS = frozenset({
|
||||
"comment_issue",
|
||||
"mark_issue",
|
||||
"lock_issue",
|
||||
"set_issue_labels",
|
||||
"create_issue",
|
||||
"close_issue",
|
||||
"create_branch",
|
||||
"push_branch",
|
||||
"create_pr",
|
||||
"close_pr",
|
||||
"comment_pr",
|
||||
"review_pr",
|
||||
"merge_pr",
|
||||
"approve_pr",
|
||||
"request_changes_pr",
|
||||
"commit_files",
|
||||
"gitea_commit_files",
|
||||
"delete_branch",
|
||||
"address_pr_change_requests",
|
||||
})
|
||||
|
||||
ALLOWED_PATH_KINDS = frozenset({
|
||||
"mcp_native",
|
||||
"recovery_fallback",
|
||||
})
|
||||
|
||||
BLOCKED_PATH_KINDS = frozenset({
|
||||
"shell_script",
|
||||
"direct_api",
|
||||
"webfetch",
|
||||
"playwright",
|
||||
"helper_script",
|
||||
"unsafe_helper",
|
||||
"mcp_server_touch",
|
||||
})
|
||||
|
||||
SHELL_SPAWN_FAILURE_THRESHOLD = 2
|
||||
|
||||
TERMINAL_REPORT_HEADING = (
|
||||
"MCP transport unavailable or shell circuit breaker tripped. "
|
||||
"No unsafe Gitea fallback performed."
|
||||
)
|
||||
|
||||
_RECOVERY_MODE_RE = re.compile(
|
||||
r"\b(?:recovery mode|explicit recovery|mcp unavailable|mcp not available|"
|
||||
r"mcp tools unavailable|no mcp path)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_LOCAL_SCRIPT_RE = re.compile(
|
||||
r"(?:^|[\s\"'`/])(?:python3?|bash|sh)?\s*(?:"
|
||||
+ "|".join(re.escape(name) for name in LOCAL_GITEA_SCRIPT_NAMES)
|
||||
+ r")",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_WEBFETCH_RE = re.compile(r"\b(?:webfetch|mcp_web_fetch|fetch\s+url)\b", re.IGNORECASE)
|
||||
_PLAYWRIGHT_RE = re.compile(r"\b(?:playwright|browser_navigate|browser_click)\b", re.IGNORECASE)
|
||||
_HELPER_SCRIPT_RE = re.compile(
|
||||
r"(?:^|[\s\"'`/])(?:_encode_|_emit_|_inline_)[\w-]+\.py",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_MANUAL_BASE64_RE = re.compile(
|
||||
r"\b(?:manual\s+base64|llm-generated\s+base64|base64-encode\s+in\s+chat)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_DIRECT_API_RE = re.compile(
|
||||
r"\b(?:api_request|urllib\.request|requests\.(?:post|patch|put)|curl\s+-X\s+POST)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_MCP_SERVER_TOUCH_RE = re.compile(
|
||||
r"\b(?:kill|pkill|restart|reload|touch|edit|modify|write)\b.{0,40}\b(?:mcp_server|mcp-server|gitea_mcp_server)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_CLI_AUTH_DIVERGENCE_RE = re.compile(
|
||||
r"GITEA_MCP_PROFILE\s*=\s*['\"]?([^\s'\";]+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
_session_shell: dict[str, Any] = {
|
||||
"consecutive_spawn_failures": 0,
|
||||
"shell_unavailable": False,
|
||||
"hard_stopped": False,
|
||||
"last_exit_code": None,
|
||||
}
|
||||
|
||||
|
||||
def _clean(value: str | None) -> str:
|
||||
return (value or "").strip()
|
||||
|
||||
|
||||
def is_spawn_failure(
|
||||
*,
|
||||
exit_code: int | None = None,
|
||||
stdout: str | None = None,
|
||||
stderr: str | None = None,
|
||||
spawn_failure: bool | None = None,
|
||||
) -> bool:
|
||||
"""True for executor spawn failures (exit_code -1, empty output)."""
|
||||
if spawn_failure is True:
|
||||
return True
|
||||
if spawn_failure is False:
|
||||
return False
|
||||
if exit_code != -1:
|
||||
return False
|
||||
return not (_clean(stdout) or _clean(stderr))
|
||||
|
||||
|
||||
def record_shell_spawn_outcome(
|
||||
*,
|
||||
exit_code: int | None = None,
|
||||
stdout: str | None = None,
|
||||
stderr: str | None = None,
|
||||
spawn_failure: bool | None = None,
|
||||
probe_attempted: bool = False,
|
||||
probe_succeeded: bool | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Track shell spawn outcomes and trip the session circuit breaker (#270 AC4)."""
|
||||
global _session_shell
|
||||
failed = is_spawn_failure(
|
||||
exit_code=exit_code,
|
||||
stdout=stdout,
|
||||
stderr=stderr,
|
||||
spawn_failure=spawn_failure,
|
||||
)
|
||||
|
||||
if failed:
|
||||
_session_shell["consecutive_spawn_failures"] = (
|
||||
int(_session_shell.get("consecutive_spawn_failures") or 0) + 1
|
||||
)
|
||||
_session_shell["last_exit_code"] = exit_code
|
||||
if probe_attempted and probe_succeeded is False:
|
||||
_session_shell["shell_unavailable"] = True
|
||||
else:
|
||||
_session_shell["consecutive_spawn_failures"] = 0
|
||||
_session_shell["shell_unavailable"] = False
|
||||
_session_shell["hard_stopped"] = False
|
||||
_session_shell["last_exit_code"] = exit_code
|
||||
|
||||
failures = int(_session_shell["consecutive_spawn_failures"])
|
||||
if failures >= SHELL_SPAWN_FAILURE_THRESHOLD:
|
||||
_session_shell["shell_unavailable"] = True
|
||||
_session_shell["hard_stopped"] = True
|
||||
|
||||
return shell_health_status()
|
||||
|
||||
|
||||
def shell_health_status() -> dict[str, Any]:
|
||||
"""Return the current shell health / circuit-breaker state."""
|
||||
failures = int(_session_shell.get("consecutive_spawn_failures") or 0)
|
||||
hard_stopped = bool(_session_shell.get("hard_stopped"))
|
||||
shell_unavailable = bool(_session_shell.get("shell_unavailable"))
|
||||
return {
|
||||
"consecutive_spawn_failures": failures,
|
||||
"shell_unavailable": shell_unavailable,
|
||||
"hard_stopped": hard_stopped,
|
||||
"threshold": SHELL_SPAWN_FAILURE_THRESHOLD,
|
||||
"shell_use_allowed": not hard_stopped,
|
||||
"last_exit_code": _session_shell.get("last_exit_code"),
|
||||
"safe_next_action": (
|
||||
"emit terminal recovery report; prefer native MCP for remaining Gitea mutations"
|
||||
if hard_stopped
|
||||
else (
|
||||
"probe shell once with echo/pwd; if probe fails mark shell unavailable"
|
||||
if failures == 1
|
||||
else "shell healthy"
|
||||
)
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def clear_shell_health_for_tests() -> None:
|
||||
"""Reset session shell state (tests only)."""
|
||||
global _session_shell
|
||||
_session_shell = {
|
||||
"consecutive_spawn_failures": 0,
|
||||
"shell_unavailable": False,
|
||||
"hard_stopped": False,
|
||||
"last_exit_code": None,
|
||||
}
|
||||
|
||||
|
||||
def classify_command_path(command_or_detail: str | None) -> str:
|
||||
"""Classify a proposed command/detail into a path kind."""
|
||||
text = command_or_detail or ""
|
||||
if _MCP_SERVER_TOUCH_RE.search(text):
|
||||
return "mcp_server_touch"
|
||||
if _WEBFETCH_RE.search(text):
|
||||
return "webfetch"
|
||||
if _PLAYWRIGHT_RE.search(text):
|
||||
return "playwright"
|
||||
if _HELPER_SCRIPT_RE.search(text) or _MANUAL_BASE64_RE.search(text):
|
||||
return "unsafe_helper"
|
||||
if _LOCAL_SCRIPT_RE.search(text):
|
||||
return "shell_script"
|
||||
if _DIRECT_API_RE.search(text):
|
||||
return "direct_api"
|
||||
return "mcp_native"
|
||||
|
||||
|
||||
def detect_cli_auth_divergence(
|
||||
command_or_detail: str | None,
|
||||
*,
|
||||
active_profile: str | None,
|
||||
) -> list[str]:
|
||||
"""Flag shell commands that override GITEA_MCP_PROFILE away from the session."""
|
||||
reasons: list[str] = []
|
||||
text = command_or_detail or ""
|
||||
active = _clean(active_profile)
|
||||
match = _CLI_AUTH_DIVERGENCE_RE.search(text)
|
||||
if not match or not active:
|
||||
return reasons
|
||||
requested = _clean(match.group(1))
|
||||
if requested and requested != active:
|
||||
reasons.append(
|
||||
f"CLI auth divergence: command sets GITEA_MCP_PROFILE='{requested}' "
|
||||
f"but active session profile is '{active}' (fail closed)"
|
||||
)
|
||||
return reasons
|
||||
|
||||
|
||||
def format_mcp_unavailable_terminal_report(
|
||||
*,
|
||||
task: str | None = None,
|
||||
reasons: list[str] | None = None,
|
||||
) -> str:
|
||||
"""Terminal report when MCP is broken and unsafe recovery is forbidden (#270 AC3)."""
|
||||
lines = [TERMINAL_REPORT_HEADING, ""]
|
||||
if task:
|
||||
lines.append(f"Blocked task: {task}")
|
||||
if reasons:
|
||||
lines.extend(["", "Reasons:"])
|
||||
lines.extend(f"- {reason}" for reason in reasons)
|
||||
lines.extend([
|
||||
"",
|
||||
"Required recovery (no improvised fallback):",
|
||||
"- Restart the MCP session",
|
||||
"- Kill hung background terminals holding the shell executor",
|
||||
"- Retry the native MCP tool once after reconnect",
|
||||
"- If MCP remains unavailable, stop and hand off to the operator",
|
||||
"- Do not run local Gitea scripts, WebFetch, Playwright, or manual base64 encoders",
|
||||
])
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def assess_gitea_operation_path(
|
||||
*,
|
||||
task: str,
|
||||
path_kind: str | None = None,
|
||||
command_or_detail: str | None = None,
|
||||
mcp_available: bool = True,
|
||||
mcp_tool_visible: bool = True,
|
||||
recovery_mode: bool = False,
|
||||
recovery_proof_complete: bool = False,
|
||||
role: str | None = None,
|
||||
active_profile: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Fail closed when a Gitea mutation would bypass native MCP (#270)."""
|
||||
task_name = _clean(task)
|
||||
resolved_kind = _clean(path_kind) or classify_command_path(command_or_detail)
|
||||
reasons: list[str] = []
|
||||
|
||||
if task_name and task_name not in GITEA_MUTATION_TASKS:
|
||||
reasons.append(
|
||||
f"unknown or non-mutation Gitea task '{task_name}'; "
|
||||
"native MCP preference gate applies only to Gitea mutations"
|
||||
)
|
||||
|
||||
shell = shell_health_status()
|
||||
if shell["hard_stopped"] and resolved_kind != "mcp_native":
|
||||
reasons.append(
|
||||
"shell circuit breaker is hard-stopped after consecutive spawn failures; "
|
||||
"use native MCP or emit terminal recovery report"
|
||||
)
|
||||
|
||||
recovery_declared = recovery_mode or bool(
|
||||
_RECOVERY_MODE_RE.search(command_or_detail or "")
|
||||
)
|
||||
recovery_allowed = (
|
||||
recovery_declared
|
||||
and recovery_proof_complete
|
||||
and not mcp_available
|
||||
and resolved_kind in BLOCKED_PATH_KINDS - {"mcp_server_touch"}
|
||||
)
|
||||
|
||||
if resolved_kind in BLOCKED_PATH_KINDS and not recovery_allowed:
|
||||
reasons.append(f"path kind '{resolved_kind}' is not an approved Gitea mutation path")
|
||||
|
||||
if resolved_kind == "mcp_server_touch":
|
||||
if _clean(role) == "reviewer" or (
|
||||
active_profile and "reviewer" in active_profile.lower()
|
||||
):
|
||||
reasons.append(
|
||||
"reviewer workflows must not touch, restart, or kill MCP server files/processes"
|
||||
)
|
||||
else:
|
||||
reasons.append(
|
||||
"MCP server files/processes must not be modified during normal Gitea workflows"
|
||||
)
|
||||
|
||||
reasons.extend(
|
||||
detect_cli_auth_divergence(command_or_detail, active_profile=active_profile)
|
||||
)
|
||||
|
||||
if (
|
||||
mcp_available
|
||||
and mcp_tool_visible
|
||||
and resolved_kind != "mcp_native"
|
||||
and not recovery_allowed
|
||||
):
|
||||
if not recovery_declared:
|
||||
reasons.append(
|
||||
"native MCP tools are available; shell/API/helper fallback is forbidden"
|
||||
)
|
||||
elif not recovery_proof_complete:
|
||||
reasons.append(
|
||||
"recovery-mode fallback requires complete recovery proof before proceeding"
|
||||
)
|
||||
|
||||
if not mcp_available and resolved_kind != "mcp_native" and not recovery_allowed:
|
||||
if not recovery_declared:
|
||||
reasons.append(
|
||||
"MCP transport unavailable; produce terminal recovery report instead of "
|
||||
"improvising unsafe fallback"
|
||||
)
|
||||
|
||||
block = bool(reasons)
|
||||
terminal_report = None
|
||||
if block and (not mcp_available or shell["hard_stopped"]):
|
||||
terminal_report = format_mcp_unavailable_terminal_report(
|
||||
task=task_name or None,
|
||||
reasons=reasons,
|
||||
)
|
||||
|
||||
return {
|
||||
"task": task_name or None,
|
||||
"path_kind": resolved_kind,
|
||||
"mcp_available": mcp_available,
|
||||
"mcp_tool_visible": mcp_tool_visible,
|
||||
"recovery_mode": recovery_declared,
|
||||
"shell_health": shell,
|
||||
"block": block,
|
||||
"allowed": not block,
|
||||
"reasons": reasons,
|
||||
"terminal_report": terminal_report,
|
||||
"safe_next_action": (
|
||||
"use the native MCP tool for this task"
|
||||
if mcp_available and mcp_tool_visible and block
|
||||
else (
|
||||
terminal_report or "proceed with native MCP"
|
||||
if block
|
||||
else "proceed with native MCP"
|
||||
)
|
||||
),
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
"""Tests for native MCP preference gate and shell circuit breaker (#270)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import native_mcp_preference as nmp # noqa: E402
|
||||
|
||||
|
||||
class TestShellHealthCircuitBreaker(unittest.TestCase):
|
||||
def setUp(self):
|
||||
nmp.clear_shell_health_for_tests()
|
||||
|
||||
def test_two_spawn_failures_hard_stop(self):
|
||||
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
||||
status = nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
||||
self.assertTrue(status["hard_stopped"])
|
||||
self.assertFalse(status["shell_use_allowed"])
|
||||
self.assertEqual(status["consecutive_spawn_failures"], 2)
|
||||
|
||||
def test_success_resets_breaker(self):
|
||||
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
||||
status = nmp.record_shell_spawn_outcome(exit_code=0, stdout="ok", stderr="")
|
||||
self.assertFalse(status["hard_stopped"])
|
||||
self.assertEqual(status["consecutive_spawn_failures"], 0)
|
||||
|
||||
|
||||
class TestNativeMcpPreferenceGate(unittest.TestCase):
|
||||
def setUp(self):
|
||||
nmp.clear_shell_health_for_tests()
|
||||
|
||||
def test_mcp_available_blocks_local_script(self):
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="create_pr",
|
||||
command_or_detail="python create_pr.py --remote prgs --title T --head h",
|
||||
mcp_available=True,
|
||||
mcp_tool_visible=True,
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertEqual(result["path_kind"], "shell_script")
|
||||
|
||||
def test_mcp_native_path_allowed(self):
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="comment_issue",
|
||||
path_kind="mcp_native",
|
||||
mcp_available=True,
|
||||
mcp_tool_visible=True,
|
||||
)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["allowed"])
|
||||
|
||||
def test_mcp_unavailable_requires_terminal_report(self):
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="merge_pr",
|
||||
path_kind="shell_script",
|
||||
mcp_available=False,
|
||||
mcp_tool_visible=False,
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(
|
||||
any("terminal recovery report" in r for r in result["reasons"])
|
||||
)
|
||||
self.assertIn(nmp.TERMINAL_REPORT_HEADING, result["terminal_report"])
|
||||
|
||||
def test_recovery_fallback_allowed_with_proof(self):
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="merge_pr",
|
||||
path_kind="shell_script",
|
||||
command_or_detail="Recovery mode: MCP tools unavailable in this session.",
|
||||
mcp_available=False,
|
||||
recovery_mode=True,
|
||||
recovery_proof_complete=True,
|
||||
)
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_cli_auth_divergence_blocked(self):
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="create_pr",
|
||||
command_or_detail="GITEA_MCP_PROFILE=prgs-reviewer python create_pr.py",
|
||||
mcp_available=True,
|
||||
active_profile="prgs-author",
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("CLI auth divergence" in r for r in result["reasons"]))
|
||||
|
||||
def test_unsafe_helper_fallback_denied(self):
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="commit_files",
|
||||
command_or_detail="python _encode_payload.py",
|
||||
mcp_available=True,
|
||||
mcp_tool_visible=True,
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertEqual(result["path_kind"], "unsafe_helper")
|
||||
|
||||
def test_reviewer_mcp_server_touch_blocked(self):
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="review_pr",
|
||||
command_or_detail="pkill -f gitea_mcp_server.py",
|
||||
mcp_available=True,
|
||||
role="reviewer",
|
||||
active_profile="prgs-reviewer",
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertEqual(result["path_kind"], "mcp_server_touch")
|
||||
|
||||
def test_hard_stopped_shell_blocks_non_mcp_path(self):
|
||||
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
||||
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="lock_issue",
|
||||
path_kind="shell_script",
|
||||
mcp_available=True,
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("circuit breaker" in r for r in result["reasons"]))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user