Gitea-Tools/mcp_namespace_health.py

"""Assess live MCP namespace health without trusting static registration.

The IDE/client namespace can fail with EOF even when this Python process still
registers the Gitea tools with FastMCP.  These helpers keep that distinction
explicit so reviewer/merger flows can fail closed on the live path.
"""

from __future__ import annotations

from typing import Any


REQUIRED_NAMESPACE_TOOLS = {
    "gitea-author": "gitea_whoami",
    "gitea-reviewer": "gitea_whoami",
    "gitea-merger": "gitea_whoami",
    "gitea-tools": "gitea_list_profiles",
}

DEFAULT_NAMESPACES = tuple(REQUIRED_NAMESPACE_TOOLS)

EOF_PATTERNS = (
    "client is closing: eof",
    "transport closed",
    "connection closed",
    "broken pipe",
    "end of file",
    "eof",
)

SAFE_ENV_KEYS = (
    "GITEA_MCP_PROFILE",
    "GITEA_PROFILE_NAME",
    "GITEA_SERVICE",
    "GITEA_EXECUTION_ROLE",
    "GITEA_MCP_CONFIG",
)


def _as_list(value: Any) -> list[str] | None:
    if value is None:
        return None
    if isinstance(value, (list, tuple, set)):
        return [str(v) for v in value]
    return [str(value)]


def _contains_eof(text: str | None) -> bool:
    lowered = (text or "").lower()
    return any(pattern in lowered for pattern in EOF_PATTERNS)


def _safe_env_summary(process: dict[str, Any] | None) -> dict[str, str]:
    if not process:
        return {}
    env = process.get("env") or process.get("environment") or {}
    if not isinstance(env, dict):
        return {}
    return {
        key: str(env[key])
        for key in SAFE_ENV_KEYS
        if key in env and env[key] not in (None, "")
    }


def classify_namespace_probe(
    namespace: str,
    *,
    required_tool: str | None = None,
    registered_tools: list[str] | tuple[str, ...] | set[str] | None = None,
    probe_result: dict[str, Any] | None = None,
    process: dict[str, Any] | None = None,
    config_path: str | None = None,
    profile: str | None = None,
    configured: bool = True,
) -> dict[str, Any]:
    """Classify whether a required tool is callable through a live namespace.

    ``registered_tools`` is static/server-side evidence. ``probe_result`` is
    live client evidence from a real namespace invocation. The assessment is
    healthy only when the required tool is registered and callable.
    """
    ns = (namespace or "").strip()
    tool = required_tool or REQUIRED_NAMESPACE_TOOLS.get(ns) or "gitea_whoami"
    registered_list = _as_list(registered_tools)
    registered = None if registered_list is None else tool in registered_list

    probe = probe_result or {}
    probe_success = bool(probe.get("success"))
    error_message = str(
        probe.get("error")
        or probe.get("message")
        or probe.get("stderr")
        or probe.get("exception")
        or ""
    )
    error_type = str(probe.get("error_type") or "").strip()
    if not error_type and error_message:
        if _contains_eof(error_message):
            error_type = "namespace_eof"
        elif "timeout" in error_message.lower():
            error_type = "namespace_timeout"
        else:
            error_type = "namespace_call_failed"

    if not configured:
        error_type = "namespace_not_configured"
    elif registered is False:
        error_type = "tool_missing"
    elif not probe_result:
        error_type = "live_probe_missing"
    elif not probe_success and not error_type:
        error_type = "namespace_call_failed"

    callable_live = bool(configured and probe_result and probe_success)
    healthy = bool(configured and registered is not False and callable_live)
    process_pid = process.get("pid") if isinstance(process, dict) else None
    profile_name = profile or (
        process.get("profile") if isinstance(process, dict) else None
    )
    env_summary = _safe_env_summary(process)

    reasons: list[str] = []
    if not configured:
        reasons.append(f"MCP namespace '{ns}' is not configured.")
    if registered is False:
        reasons.append(
            f"Required tool '{tool}' is not registered in namespace '{ns}'."
        )
    if error_type == "live_probe_missing":
        reasons.append(
            f"No live client invocation proof was supplied for '{ns}.{tool}'."
        )
    elif error_type == "namespace_eof":
        reasons.append(
            f"Live MCP namespace '{ns}' returned EOF while invoking '{tool}'."
        )
    elif error_type == "namespace_timeout":
        reasons.append(
            f"Live MCP namespace '{ns}' timed out while invoking '{tool}'."
        )
    elif error_type == "namespace_call_failed":
        reasons.append(
            f"Live MCP namespace '{ns}' failed while invoking '{tool}'."
        )

    remediation = []
    if not healthy:
        remediation.append(
            "Restart or reconnect the IDE MCP client namespace, then retry "
            f"'{tool}' through namespace '{ns}'."
        )
        remediation.append(
            "Reload the MCP config and verify the configured command, args, "
            "profile environment, and working tree point at the current checkout."
        )
        if process_pid:
            remediation.append(
                f"If PID {process_pid} belongs to a stale MCP process, stop it "
                "before reconnecting the namespace."
            )
    else:
        remediation.append(f"Namespace '{ns}' can invoke '{tool}'.")

    return {
        "success": healthy,
        "healthy": healthy,
        "namespace": ns,
        "required_tool": tool,
        "configured": configured,
        "registered_tools_checked": registered_list is not None,
        "required_tool_registered": registered,
        "required_tool_callable": callable_live,
        "error_type": None if healthy else error_type,
        "error_message": error_message or None,
        "reasons": reasons,
        "remediation": remediation,
        "diagnostics": {
            "namespace": ns,
            "required_tool": tool,
            "process_pid": process_pid,
            "profile": profile_name,
            "env": env_summary,
            "config_path": config_path,
        },
        "blocks_merge_workflow": namespace_health_blocks_task(
            "merge_pr", healthy
        ),
    }


def namespace_health_blocks_task(task: str, healthy: bool) -> bool:
    """Return whether a broken namespace must block a workflow task."""
    if healthy:
        return False
    return (task or "").strip() in {"merge_pr", "review_pr", "submit_review"}