408 lines
15 KiB
Python
408 lines
15 KiB
Python
"""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.
|
|
|
|
Probe sources
|
|
-------------
|
|
* ``client_namespace`` — evidence from the IDE-managed MCP client path (the
|
|
only source that can prove the workflow namespace is healthy).
|
|
* ``offline_spawn`` — a separate ``subprocess.Popen`` JSON-RPC handshake
|
|
(e.g. ``test_mcp_conn.py``). Useful offline, but **never** proves the
|
|
IDE-managed namespace is callable.
|
|
* ``unknown`` — legacy/unspecified; treated as not IDE-proven.
|
|
"""
|
|
|
|
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)
|
|
|
|
# Namespaces that must be healthy for a given mutation task.
|
|
TASK_REQUIRED_NAMESPACES = {
|
|
"review_pr": "gitea-reviewer",
|
|
"submit_review": "gitea-reviewer",
|
|
"merge_pr": "gitea-merger",
|
|
}
|
|
|
|
PROBE_SOURCE_CLIENT = "client_namespace"
|
|
PROBE_SOURCE_OFFLINE = "offline_spawn"
|
|
PROBE_SOURCE_UNKNOWN = "unknown"
|
|
VALID_PROBE_SOURCES = frozenset(
|
|
{PROBE_SOURCE_CLIENT, PROBE_SOURCE_OFFLINE, PROBE_SOURCE_UNKNOWN}
|
|
)
|
|
|
|
EOF_PATTERNS = (
|
|
"client is closing: eof",
|
|
"transport closed",
|
|
"connection closed",
|
|
"broken pipe",
|
|
"end of file",
|
|
"eof",
|
|
)
|
|
|
|
ERROR_CONNECTED_NAMESPACES_MISSING = "mcp_connected_namespaces_missing"
|
|
|
|
UNSAFE_FALLBACK_WARNING = (
|
|
"Workflow Safety Hard Stop (#708): Connected-but-namespaces-missing recovery must "
|
|
"NEVER use direct imports, Gitea API mutations, profile hopping, session-state "
|
|
"overrides, PID kills, or config mtime touches. Use client reconnect only."
|
|
)
|
|
|
|
SAFE_ENV_KEYS = (
|
|
"GITEA_MCP_PROFILE",
|
|
"GITEA_PROFILE_NAME",
|
|
"GITEA_SERVICE",
|
|
"GITEA_EXECUTION_ROLE",
|
|
"GITEA_MCP_CONFIG",
|
|
)
|
|
|
|
|
|
def assess_connected_namespace_attachment(
|
|
*,
|
|
connected_servers: list[str] | tuple[str, ...] | set[str] | None = None,
|
|
attached_session_namespaces: list[str] | tuple[str, ...] | set[str] | None = None,
|
|
required_namespaces: list[str] | tuple[str, ...] | set[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Assess whether host-connected MCP servers have attached tool namespaces in the active session (#708).
|
|
|
|
Addresses the Connected-but-namespaces-missing defect: CLI/host status may report Connected
|
|
while the active LLM session tool surface exposes 0 attached tool namespaces.
|
|
|
|
Returns structured telemetry and detection details.
|
|
"""
|
|
connected = [str(s).strip() for s in (connected_servers or []) if str(s).strip()]
|
|
attached = set(str(ns).strip() for ns in (attached_session_namespaces or []) if str(ns).strip())
|
|
req = [str(r).strip() for r in (required_namespaces or DEFAULT_NAMESPACES) if str(r).strip()]
|
|
|
|
proof: dict[str, dict[str, bool]] = {}
|
|
missing: list[str] = []
|
|
|
|
for s in connected:
|
|
is_attached = s in attached
|
|
proof[s] = {"connected": True, "attached": is_attached}
|
|
if not is_attached and s in req:
|
|
missing.append(s)
|
|
|
|
for r in req:
|
|
if r not in proof:
|
|
proof[r] = {"connected": r in connected, "attached": r in attached}
|
|
if r in connected and r not in attached and r not in missing:
|
|
missing.append(r)
|
|
|
|
attachment_healthy = len(missing) == 0 and len(connected) > 0
|
|
discovery_status = (
|
|
"namespaces_attached" if attachment_healthy
|
|
else ("connected_but_namespaces_missing" if len(connected) > 0 else "disconnected")
|
|
)
|
|
|
|
reasons: list[str] = []
|
|
remediation: list[str] = []
|
|
if missing:
|
|
reasons.append(
|
|
f"MCP server(s) {missing} report Connected at host/CLI layer but tool namespaces "
|
|
f"are missing from active session attached tools (Connected ≠ attached tools, #708)."
|
|
)
|
|
remediation.append(
|
|
"Reconnect the IDE/client MCP session to attach namespaces to the active session. "
|
|
"Do not use direct imports, CLI API mutations, profile hopping, or session file overrides."
|
|
)
|
|
elif not connected:
|
|
reasons.append("No MCP servers reported Connected.")
|
|
remediation.append("Start or reconnect Gitea MCP servers in client config.")
|
|
else:
|
|
reasons.append("All connected MCP server namespaces are attached to the active session.")
|
|
|
|
return {
|
|
"success": attachment_healthy,
|
|
"attachment_healthy": attachment_healthy,
|
|
"discovery_status": discovery_status,
|
|
"connected_servers": connected,
|
|
"attached_session_namespaces": list(attached),
|
|
"missing_namespaces": missing,
|
|
"proof_of_connected_vs_attached": proof,
|
|
"error_type": None if attachment_healthy else ERROR_CONNECTED_NAMESPACES_MISSING,
|
|
"reasons": reasons,
|
|
"remediation": remediation,
|
|
"exact_next_action": (
|
|
"Reconnect the IDE/client MCP session so tool namespaces attach to the active session. "
|
|
"Do not use direct imports, CLI API mutations, profile hopping, or session-state overrides."
|
|
if not attachment_healthy
|
|
else "None; session tool namespaces attached."
|
|
),
|
|
"unsafe_fallback_policy": UNSAFE_FALLBACK_WARNING,
|
|
}
|
|
|
|
|
|
|
|
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 _normalize_probe_source(probe_source: str | None) -> str:
|
|
raw = (probe_source or PROBE_SOURCE_UNKNOWN).strip().lower()
|
|
if raw in VALID_PROBE_SOURCES:
|
|
return raw
|
|
return PROBE_SOURCE_UNKNOWN
|
|
|
|
|
|
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,
|
|
probe_source: str | None = None,
|
|
) -> 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 invocation evidence. Only ``probe_source=client_namespace`` proves the
|
|
IDE-managed path; ``offline_spawn`` is an offline subprocess check only.
|
|
"""
|
|
ns = (namespace or "").strip()
|
|
tool = required_tool or REQUIRED_NAMESPACE_TOOLS.get(ns) or "gitea_whoami"
|
|
source = _normalize_probe_source(probe_source)
|
|
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)
|
|
# Probe-path health (spawn or client). IDE-proven only for client path.
|
|
healthy = bool(configured and registered is not False and callable_live)
|
|
ide_namespace_proven = bool(healthy and source == PROBE_SOURCE_CLIENT)
|
|
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}'."
|
|
)
|
|
if source == PROBE_SOURCE_OFFLINE:
|
|
reasons.append(
|
|
"Probe source is offline_spawn (subprocess JSON-RPC); this does "
|
|
"not prove the IDE-managed MCP namespace is healthy."
|
|
)
|
|
elif source == PROBE_SOURCE_UNKNOWN and probe_result:
|
|
reasons.append(
|
|
"Probe source unspecified; treat as not IDE-namespace proof unless "
|
|
"re-supplied with probe_source='client_namespace'."
|
|
)
|
|
|
|
remediation = []
|
|
if not healthy or not ide_namespace_proven:
|
|
remediation.append(
|
|
"Reconnect the IDE MCP client namespace (client reconnect / "
|
|
f"relaunch), then invoke '{tool}' through namespace '{ns}' and "
|
|
"record the result with probe_source='client_namespace'."
|
|
)
|
|
remediation.append(
|
|
"Do not treat offline subprocess probes (test_mcp_conn.py) or "
|
|
"shell kill/PID respawn as proof the IDE namespace is repaired."
|
|
)
|
|
if process_pid and source != PROBE_SOURCE_OFFLINE:
|
|
remediation.append(
|
|
f"Diagnostics may include PID {process_pid}; process details "
|
|
"are informational only — recovery is client-layer reconnect."
|
|
)
|
|
else:
|
|
remediation.append(
|
|
f"IDE-managed namespace '{ns}' can invoke '{tool}' "
|
|
f"(probe_source={source})."
|
|
)
|
|
|
|
# Client-namespace broken health blocks review/merge. Offline probes never
|
|
# authorize mutations and only block when they report unhealthy (still
|
|
# fail-closed for known bad spawn evidence).
|
|
blocks = False
|
|
if source == PROBE_SOURCE_CLIENT:
|
|
blocks = namespace_health_blocks_task("merge_pr", healthy)
|
|
elif source == PROBE_SOURCE_OFFLINE:
|
|
# Offline never proves IDE health; never unblock. Unhealthy offline
|
|
# still surfaces as a soft diagnostic, not a mutation-ledger block.
|
|
blocks = False
|
|
else:
|
|
# Unknown source: only block when evidence is unhealthy (fail closed
|
|
# on bad data without treating success as IDE proof).
|
|
blocks = namespace_health_blocks_task("merge_pr", healthy)
|
|
|
|
import gitea_config
|
|
raw_env = process.get("env") if isinstance(process, dict) else None
|
|
unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides(raw_env)
|
|
is_client_managed = bool(
|
|
env_summary.get("GITEA_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed")
|
|
or env_summary.get("GITEA_MCP_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed")
|
|
or env_summary.get("GITEA_SERVER_PROVENANCE") == "client_managed"
|
|
)
|
|
provenance = "client_managed" if is_client_managed else "manual_launch"
|
|
|
|
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,
|
|
"probe_source": source,
|
|
"ide_namespace_proven": ide_namespace_proven,
|
|
"error_type": None if healthy else error_type,
|
|
"error_message": error_message or None,
|
|
"reasons": reasons,
|
|
"remediation": remediation,
|
|
"provenance": provenance,
|
|
"is_client_managed": is_client_managed,
|
|
"unconsumed_gitea_env": unconsumed_env,
|
|
"diagnostics": {
|
|
"namespace": ns,
|
|
"required_tool": tool,
|
|
"process_pid": process_pid,
|
|
"profile": profile_name,
|
|
"env": env_summary,
|
|
"config_path": config_path,
|
|
"probe_source": source,
|
|
"provenance": provenance,
|
|
"is_client_managed": is_client_managed,
|
|
"unconsumed_gitea_env": unconsumed_env,
|
|
},
|
|
"blocks_merge_workflow": blocks,
|
|
}
|
|
|
|
|
|
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"}
|
|
|
|
|
|
def required_namespace_for_task(task: str) -> str | None:
|
|
"""Map a mutation task to the MCP namespace that must be healthy."""
|
|
return TASK_REQUIRED_NAMESPACES.get((task or "").strip())
|
|
|
|
|
|
def mutation_gate_from_session(
|
|
task: str,
|
|
session_health: dict[str, dict[str, Any]] | None,
|
|
) -> list[str]:
|
|
"""Fail-closed gate using recorded client-namespace health assessments.
|
|
|
|
* Missing session entry → no gate (caller has not assessed yet).
|
|
* Client-namespace unhealthy / not IDE-proven → block mutation.
|
|
* Offline-only session entries never authorize mutations.
|
|
"""
|
|
ns = required_namespace_for_task(task)
|
|
if not ns:
|
|
return []
|
|
store = session_health or {}
|
|
entry = store.get(ns)
|
|
if not entry:
|
|
return []
|
|
source = _normalize_probe_source(entry.get("probe_source"))
|
|
if source != PROBE_SOURCE_CLIENT:
|
|
return [
|
|
f"recorded namespace health for '{ns}' is probe_source={source}, "
|
|
"not client_namespace; re-probe through the IDE-managed path "
|
|
f"before {(task or 'mutation')}"
|
|
]
|
|
if entry.get("ide_namespace_proven") and entry.get("healthy"):
|
|
return []
|
|
if entry.get("blocks_merge_workflow") or not entry.get("healthy"):
|
|
detail = entry.get("error_type") or "unhealthy"
|
|
return [
|
|
f"live MCP namespace '{ns}' is recorded {detail} "
|
|
f"(probe_source={source}); repair the IDE namespace before "
|
|
f"{(task or 'mutation')} (fail closed, #543)"
|
|
]
|
|
if not entry.get("ide_namespace_proven"):
|
|
return [
|
|
f"live MCP namespace '{ns}' is not IDE-proven; supply a "
|
|
"client_namespace probe before mutation (fail closed, #543)"
|
|
]
|
|
return []
|