feat: diagnose live MCP namespace EOF health
This commit is contained in:
@@ -0,0 +1,197 @@
|
||||
"""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"}
|
||||
|
||||
Reference in New Issue
Block a user