Gitea-Tools/native_mcp_preference.py

"""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"
            )
        ),
    }