Merge pull request 'feat: diagnose live MCP namespace EOF health and block merge (#543)' (#587) from feat/issue-543-mcp-namespace-health-check into master

docs/mcp-namespace-eof-recovery.md

@@ -0,0 +1,118 @@
+# Recovering from `client is closing: EOF` on a Gitea MCP namespace (#543)
+
+## Symptom
+
+A tool call through a Gitea MCP namespace — `gitea-author`, `gitea-reviewer`,
+`gitea-merger`, or the shared `gitea-tools` namespace — fails immediately with:
+
+```
+client is closing: EOF
+```
+
+Every subsequent call to that same namespace returns the same error, including
+cheap read tools such as `gitea_whoami` and `gitea_list_profiles`. Other MCP
+servers registered with the same client (for example `context7`) keep working,
+so this is **not** a global MCP-client outage.
+
+## Why this is not a code defect
+
+This failure is a **transport-level** condition in the IDE / MCP client manager,
+not a missing or broken tool:
+
+- The tool can be present and registered in the Python `FastMCP` tool manager.
+- Direct Python inspection of the server confirms the tool exists.
+- Running the server manually and sending JSON-RPC over stdio works fine
+  (offline spawn) — that path does **not** prove the IDE namespace is healthy.
+
+The client manager entered a closed state after the backing subprocess for that
+namespace terminated (or was killed) behind its back. Once closed, the client
+does **not** re-spawn the child on the next tool call — it just replays
+`client is closing: EOF`. The OS process may even still be alive if a parent
+language-server process is holding the stdio pipes open.
+
+This is the canonical "registered in FastMCP ≠ callable through the namespace"
+false-ready state. It is distinct from the **stale-runtime** family in #531 /
+#544, where the process is reachable but running behind `master`; that case is
+detected by the `ps`-based `_check_mcp_runtimes_diagnostics` in
+`gitea_mcp_server.py`. The EOF case is a dead/closed transport, not a stale one,
+so the `ps` check alone will not surface it.
+
+## Recovery path (canonical — client reconnect only)
+
+Do the steps in order. Stop as soon as a live **client-namespace** call succeeds.
+
+1. **Confirm the blast radius.** Call a cheap read tool on the failing namespace
+   (`gitea_whoami` or `gitea_list_profiles`). Then call the same tool on a
+   different MCP server (e.g. `context7`).
+   - Only the Gitea namespace fails → single-namespace transport close. Continue.
+   - Every server fails → restart the whole MCP client, not just one namespace.
+
+2. **Reconnect the namespace through the client, not the shell.** Use the IDE /
+   client MCP-reconnect action for that server entry (in Claude Code:
+   `/mcp` → reconnect the affected `gitea-*` server). Reconnecting forces the
+   client to spawn a fresh subprocess and re-open the pipe. This clears the
+   closed-client state that a bare `kill`/respawn from a terminal does **not**.
+
+3. **Do not "fix" it by importing the server or poking the process.** Reaching
+   for `python -c 'import gitea_mcp_server ...'`, raw JSON-RPC from a shell,
+   killing PIDs to force a respawn, or touching MCP config mtimes does **not**
+   restore the *client's* view of the namespace and violates the daemon-import
+   guard (#558, `docs/mcp-daemon-import-guard.md`). The only sanctioned repair
+   is a **client reconnect / relaunch**.
+
+4. **Verify through the same path the workflow will use.** After reconnect, call
+   the specific tool the blocked workflow needs — not just any tool — through
+   the target namespace. For a merge that means calling the merger-authorized
+   adoption/merge tool through `gitea-merger`. A green `gitea_whoami` on one
+   namespace does **not** prove another namespace or another tool is callable.
+   Record success with:
+
+   ```text
+   gitea_assess_mcp_namespace_health(..., probe_source="client_namespace")
+   ```
+
+5. **If reconnect does not clear it,** relaunch the client entirely, then repeat
+   step 4. If EOF persists after a full relaunch, the backing subprocess is
+   failing to start — inspect its stderr / launch config (command path, venv,
+   `*_MCP_CONFIG`, `*_MCP_PROFILE` env) rather than retrying the call. Still
+   do not use PID kill or config-touch as the primary recovery.
+
+## Diagnostics to capture when reporting EOF
+
+Include all of these so the failure is actionable and reproducible:
+
+- **Namespace name** that returned EOF (`gitea-author` / `gitea-reviewer` /
+  `gitea-merger` / `gitea-tools`).
+- **Tool** that was called and the **exact** error string.
+- **PID** of the backing process (if any) and whether it was still alive
+  (informational only — not a recovery action).
+- **Profile / env** for that namespace (execution profile, `*_MCP_PROFILE`,
+  worktree binding such as `GITEA_AUTHOR_WORKTREE`).
+- **Config path** the client launched the server from.
+- Result of the **cross-server control** call (did `context7` succeed?).
+
+## Offline spawn probe (non-authoritative)
+
+`test_mcp_conn.py` performs a full JSON-RPC handshake against a **fresh
+subprocess** (`initialize` → `initialized` → `tools/list` → `tools/call`) and
+classifies with `probe_source=offline_spawn`. That is useful for offline
+launch/registration debugging. It is **not** proof the IDE-managed namespace is
+healthy. See `docs/mcp-namespace-health.md`.
+
+## Do-not list during EOF recovery
+
+- Do **not** retry a blocked merge/adoption until the required tool is confirmed
+  callable through the merger-authorized **client** namespace (see #543).
+- Do **not** clean, reset, or rebind a **foreign** worktree to work around the
+  error.
+- Do **not** bypass the namespace with direct imports, raw API/curl, or
+  in-memory state restoration.
+- Do **not** kill MCP PIDs or touch config mtimes as a substitute for client
+  reconnect.
+
+## Related
+
+- #531 / #544 — stale-runtime detection (`ps`-based); sibling failure mode.
+- #558 / `docs/mcp-daemon-import-guard.md` — why shell imports are not a repair.
+- `docs/mcp-client-registration.md` — per-server registration contract.
+- `docs/mcp-namespace-health.md` — probe sources and mutation enforcement.

docs/mcp-namespace-health.md

@@ -0,0 +1,88 @@
+# MCP namespace health diagnostics (#543)
+
+Gitea MCP tools can be registered in the Python FastMCP server while the IDE's
+live MCP namespace is still unusable. The failure usually appears as
+`client is closing: EOF`, `transport closed`, or an empty response when calling
+a tool such as `gitea_whoami`.
+
+Do not treat static tool registration as proof that review or merge workflows
+can proceed. A reviewer or merger flow must have **client-namespace** evidence
+that the required tool is callable through the configured IDE MCP namespace.
+
+## Probe sources (do not confuse them)
+
+| Source | How obtained | Proves IDE namespace? |
+| --- | --- | --- |
+| `client_namespace` | Tool call through the IDE-managed MCP client | **Yes** |
+| `offline_spawn` | `test_mcp_conn.py` subprocess JSON-RPC handshake | **No** (offline only) |
+
+`gitea_assess_mcp_namespace_health` accepts `probe_source` and only sets
+`ide_namespace_proven=true` for `client_namespace` success. Offline spawn
+success never clears review/merge mutation gates.
+
+## Client-namespace health check (canonical)
+
+1. Through the IDE client, call a cheap tool on the target namespace
+   (`gitea_whoami` or `gitea_list_profiles`).
+2. Feed the live result into:
+
+```text
+gitea_assess_mcp_namespace_health(
+    namespace="gitea-merger",          # or reviewer / author / tools
+    registered_tools=[...],            # optional static list
+    probe_result={"success": true, "result": {...}},
+    probe_source="client_namespace",
+)
+```
+
+3. A healthy client-namespace assessment is recorded in the MCP session and
+   consulted by **live** `gitea_submit_pr_review` / `gitea_merge_pr` gates.
+4. If the probe fails with EOF, recover via **client reconnect only** — see
+   `docs/mcp-namespace-eof-recovery.md`. Do **not** kill PIDs or touch MCP
+   config mtimes as a recovery procedure.
+
+## Offline spawn probe (non-authoritative)
+
+```bash
+python3 test_mcp_conn.py --config ~/.gemini/config/mcp_config.json
+```
+
+This script spawns a **separate** server process from config, performs
+JSON-RPC `initialize` → `tools/list` → `tools/call`, and classifies the
+result with `probe_source=offline_spawn`. Use it for offline debugging of
+launch command / registration. It does **not** prove the IDE-managed
+namespace is healthy.
+
+By default the script checks:
+
+| Namespace | Required tool |
+| --- | --- |
+| `gitea-author` | `gitea_whoami` |
+| `gitea-reviewer` | `gitea_whoami` |
+| `gitea-merger` | `gitea_whoami` |
+| `gitea-tools` | `gitea_list_profiles` |
+
+## Recovery (canonical)
+
+When a namespace returns EOF, follow
+`docs/mcp-namespace-eof-recovery.md` in order:
+
+1. Confirm blast radius (Gitea namespace vs all MCP servers).
+2. **Reconnect the namespace through the client** (IDE reconnect / relaunch).
+3. Do not repair via shell imports, raw JSON-RPC, PID kills, or config mtime
+   touches — those do not restore the client's closed transport.
+4. Re-verify the **specific** required tool through the target namespace.
+5. Resume review/merge only after a successful `client_namespace` assessment.
+
+## Enforcement
+
+1. **State machine (read-only):** feed `blocks_merge_workflow` from a
+   `client_namespace` assessment into
+   `gitea_assess_review_merge_state_machine(live_namespace_broken=...)`.
+2. **Live mutations:** `gitea_submit_pr_review` and `gitea_merge_pr` call
+   `_live_namespace_health_gate` and fail closed when the session has a
+   recorded unhealthy or non-client probe for the required namespace
+   (`gitea-reviewer` for review, `gitea-merger` for merge).
+
+When blocked, repair the IDE namespace and re-record a healthy
+`client_namespace` assessment before retrying the mutation.

gitea_mcp_server.py

@@ -191,6 +191,7 @@ MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
 RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
 
 import namespace_workspace_binding as nwb  # noqa: E402
+import mcp_namespace_health  # noqa: E402
 
 
 def _preflight_in_test_mode() -> bool:
@@ -2824,6 +2825,35 @@ _TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})
 # remote + profile identity with TTL.
 _REVIEW_DECISION_LOCK: dict | None = None
 
+# Session-scoped live MCP namespace health assessments (#543).
+# Keyed by namespace name. Only client_namespace entries authorize mutations.
+_LIVE_NAMESPACE_HEALTH: dict[str, dict] = {}
+
+
+def _record_live_namespace_health(assessment: dict | None) -> None:
+    """Store a namespace health assessment for mutation gates."""
+    if not isinstance(assessment, dict):
+        return
+    ns = str(assessment.get("namespace") or "").strip()
+    if not ns:
+        return
+    _LIVE_NAMESPACE_HEALTH[ns] = {
+        "namespace": ns,
+        "healthy": bool(assessment.get("healthy")),
+        "ide_namespace_proven": bool(assessment.get("ide_namespace_proven")),
+        "probe_source": assessment.get("probe_source"),
+        "blocks_merge_workflow": bool(assessment.get("blocks_merge_workflow")),
+        "error_type": assessment.get("error_type"),
+        "required_tool": assessment.get("required_tool"),
+    }
+
+
+def _live_namespace_health_gate(task: str) -> list[str]:
+    """Fail closed on recorded broken/non-client IDE namespace health (#543)."""
+    return mcp_namespace_health.mutation_gate_from_session(
+        task, _LIVE_NAMESPACE_HEALTH
+    )
+
 
 def _decision_lock_binding(lock: dict | None = None) -> dict:
     """Resolve key fields for durable decision-lock storage."""
@@ -3514,6 +3544,11 @@ def _evaluate_pr_review_submission(
         reasons.extend(workflow_blockers)
         reasons.extend(review_workflow_load.recovery_handoff_without_replay())
         return result
+    if live:
+        ns_gate = _live_namespace_health_gate("review_pr")
+        if ns_gate:
+            reasons.extend(ns_gate)
+            return result
 
     if action not in _REVIEW_ACTIONS:
         reasons.append(
@@ -4944,6 +4979,12 @@ def gitea_merge_pr(
         reasons.extend(review_workflow_load.recovery_handoff_without_replay())
         return result
 
+    # Gate 0b — recorded live MCP namespace health must not be broken (#543).
+    ns_gate = _live_namespace_health_gate("merge_pr")
+    if ns_gate:
+        reasons.extend(ns_gate)
+        return result
+
     # Gate 1 — valid merge method (no API call on a bad method).
     if do not in _MERGE_METHODS:
         reasons.append(
@@ -8953,40 +8994,45 @@ def gitea_assess_review_merge_state_machine(
     pre_merge_gates: dict[str, bool] | None = None,
     infra_stop: bool = False,
     capability_blocked: bool = False,
+    live_namespace_broken: bool = False,
     recovery_handoff_text: str | None = None,
     final_report_text: str | None = None,
 ) -> dict:
-    """Read-only: assess enforced PR review/merge workflow state (#290)."""
+    """Read-only: assess enforced PR review/merge workflow state (#290).
+
+    ``live_namespace_broken`` fails review/merge closed when the live MCP
+    namespace call path is unusable even though the tool is registered in
+    FastMCP (#543 AC5). Supply the ``blocks_merge_workflow`` verdict from
+    ``gitea_assess_mcp_namespace_health``.
+    """
     completion = state_completion or {}
-    blockers = review_merge_state_machine.assess_workflow_blockers(
-        infra_stop=infra_stop,
-        capability_blocked=capability_blocked,
-    )
+    blocker_kwargs = {
+        "infra_stop": infra_stop,
+        "capability_blocked": capability_blocked,
+        "live_namespace_broken": live_namespace_broken,
+    }
+    blockers = review_merge_state_machine.assess_workflow_blockers(**blocker_kwargs)
     result = {
         "workflow": review_merge_state_machine.workflow_status(
             completion,
-            infra_stop=infra_stop,
-            capability_blocked=capability_blocked,
+            **blocker_kwargs,
         ),
         "blockers": blockers,
         "approve": review_merge_state_machine.can_approve(
             completion,
-            infra_stop=infra_stop,
-            capability_blocked=capability_blocked,
+            **blocker_kwargs,
         ),
         "merge": review_merge_state_machine.can_merge(
             completion,
             pre_merge_gates=pre_merge_gates,
-            infra_stop=infra_stop,
-            capability_blocked=capability_blocked,
+            **blocker_kwargs,
         ),
     }
     if target_state:
         result["advancement"] = review_merge_state_machine.assess_state_advancement(
             completion,
             target_state=target_state,
-            infra_stop=infra_stop,
-            capability_blocked=capability_blocked,
+            **blocker_kwargs,
         )
     if recovery_handoff_text is not None:
         result["recovery_handoff"] = (
@@ -9516,6 +9562,46 @@ def gitea_list_profiles() -> dict:
     return {"profiles": profiles_out}
 
 
+@mcp.tool()
+def gitea_assess_mcp_namespace_health(
+    namespace: str,
+    required_tool: str | None = None,
+    registered_tools: list[str] | None = None,
+    probe_result: dict | None = None,
+    process: dict | None = None,
+    config_path: str | None = None,
+    profile: str | None = None,
+    configured: bool = True,
+    probe_source: str | None = None,
+) -> dict:
+    """Classify MCP namespace health for required Gitea tools (#543).
+
+    Static FastMCP registration is not enough to prove a namespace works: IDE
+    clients can keep a registered tool list while live calls fail with
+    ``client is closing: EOF``. Pass live IDE invocation evidence with
+    ``probe_source='client_namespace'``. Offline subprocess probes
+    (``test_mcp_conn.py``) must use ``probe_source='offline_spawn'`` and never
+    count as IDE proof.
+
+    Assessments are recorded in the session so live
+    ``gitea_submit_pr_review`` / ``gitea_merge_pr`` can fail closed when a
+    client-namespace probe reported unhealthy.
+    """
+    result = mcp_namespace_health.classify_namespace_probe(
+        namespace,
+        required_tool=required_tool,
+        registered_tools=registered_tools,
+        probe_result=probe_result,
+        process=process,
+        config_path=config_path,
+        profile=profile,
+        configured=configured,
+        probe_source=probe_source,
+    )
+    _record_live_namespace_health(result)
+    return result
+
+
 @mcp.tool()
 def gitea_activate_profile(
     profile_name: str,

mcp_namespace_health.py

@@ -0,0 +1,306 @@
+"""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",
+)
+
+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 _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)
+
+    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,
+        "diagnostics": {
+            "namespace": ns,
+            "required_tool": tool,
+            "process_pid": process_pid,
+            "profile": profile_name,
+            "env": env_summary,
+            "config_path": config_path,
+            "probe_source": source,
+        },
+        "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 []

review_merge_state_machine.py

@@ -93,8 +93,16 @@ def assess_workflow_blockers(
     capability_blocked: bool = False,
     mcp_reconnect_failed: bool = False,
     stale_capability_state: bool = False,
+    live_namespace_broken: bool = False,
 ) -> dict[str, Any]:
-    """Return hard blockers that forbid all PR queue work (#290 AC3)."""
+    """Return hard blockers that forbid all PR queue work (#290 AC3).
+
+    ``live_namespace_broken`` fails the merge/review path closed when the live
+    MCP namespace call path is unusable (e.g. ``client is closing: EOF``) even
+    though the tool is registered in FastMCP (#543 AC5). Feed it the
+    ``blocks_merge_workflow`` verdict from
+    ``mcp_namespace_health.classify_namespace_probe``.
+    """
     reasons: list[str] = []
     if infra_stop:
         reasons.append("infra_stop is active; PR selection/review/merge is forbidden")
@@ -104,6 +112,12 @@ def assess_workflow_blockers(
         reasons.append("MCP reconnect failed; stale session state cannot be reused")
     if stale_capability_state:
         reasons.append("stale MCP capability state detected after reconnect failure")
+    if live_namespace_broken:
+        reasons.append(
+            "live MCP namespace call path is broken (registered in FastMCP but "
+            "not callable through the namespace); repair the namespace before "
+            "review/merge"
+        )
     return {
         "block": bool(reasons),
         "reasons": reasons,
@@ -118,16 +132,19 @@ def assess_state_advancement(
     state_completion: dict[str, bool] | None,
     *,
     target_state: str,
-    infra_stop: bool = False,
-    capability_blocked: bool = False,
+    **blocker_kwargs,
 ) -> dict[str, Any]:
-    """Fail closed when *target_state* is requested before upstream gates pass."""
+    """Fail closed when *target_state* is requested before upstream gates pass.
+
+    Forwards every blocker flag (``infra_stop``, ``capability_blocked``,
+    ``mcp_reconnect_failed``, ``stale_capability_state``,
+    ``live_namespace_broken``) to :func:`assess_workflow_blockers` so
+    ``can_approve``/``can_merge`` honor the full blocker set, not just
+    infra_stop/capability_blocked (#543 AC5).
+    """
     completion = dict(state_completion or {})
     target = _clean(target_state).upper()
-    blockers = assess_workflow_blockers(
-        infra_stop=infra_stop,
-        capability_blocked=capability_blocked,
-    )
+    blockers = assess_workflow_blockers(**blocker_kwargs)
     reasons = list(blockers["reasons"])
 
     try:

test_mcp_conn.py

@@ -1,20 +1,45 @@
 #!/usr/bin/env python3
-"""Live health-check script to verify Gitea MCP namespace connections.
+"""Offline-only MCP namespace spawn probe (NOT IDE-namespace proof).
 
-Spawns the MCP server processes as defined in the IDE's global config,
-performs the JSON-RPC handshake, and queries the tools list to verify
-that the connection is fully operational and doesn't return EOF.
+Spawns a *separate* MCP server process from config via subprocess.Popen,
+performs the JSON-RPC handshake, verifies the required tool is registered,
+and invokes that tool. Results are classified with
+``probe_source=offline_spawn``.
+
+This path is useful for offline launch/registration debugging. It does
+**not** prove the IDE-managed MCP client namespace is healthy (#543). For
+workflow gates, pass live IDE call evidence with
+``probe_source=client_namespace`` to ``gitea_assess_mcp_namespace_health``.
 """
+
+import argparse
 import json
 import os
 import subprocess
 import sys
 
-def run_connection_test(name, config):
+from mcp_namespace_health import REQUIRED_NAMESPACE_TOOLS, classify_namespace_probe
+
+
+def _read_json_line(proc):
+    line = proc.stdout.readline()
+    if not line:
+        stderr_content = proc.stderr.read()
+        return None, stderr_content
+    return json.loads(line), None
+
+
+def _write_message(proc, payload):
+    proc.stdin.write(json.dumps(payload) + "\n")
+    proc.stdin.flush()
+
+
+def run_connection_test(name, config, *, required_tool=None, config_path=None):
     print(f"Testing MCP connection for '{name}'...")
     command = config.get("command")
     args = config.get("args", [])
     env = config.get("env", {})
+    tool_name = required_tool or REQUIRED_NAMESPACE_TOOLS.get(name) or "gitea_whoami"
     
     # Merge current environment
     run_env = os.environ.copy()
@@ -31,8 +56,17 @@ def run_connection_test(name, config):
             bufsize=1,
             env=run_env
         )
-    except Exception as e:
-        print(f"  [FAIL] Failed to spawn process: {e}")
+    except Exception as exc:
+        print(f"  [FAIL] Failed to spawn process: {exc}")
+        assessment = classify_namespace_probe(
+            name,
+            required_tool=tool_name,
+            probe_result={"success": False, "error": str(exc)},
+            process={"profile": env.get("GITEA_MCP_PROFILE"), "env": env},
+            config_path=config_path,
+                probe_source="offline_spawn",
+        )
+        print(f"  diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}")
         return False
 
     # Send initialize request
@@ -48,26 +82,36 @@ def run_connection_test(name, config):
     }
     
     try:
-        proc.stdin.write(json.dumps(init_req) + "\n")
-        proc.stdin.flush()
+        _write_message(proc, init_req)
         
         # Read response
-        line = proc.stdout.readline()
-        if not line:
-            stderr_content = proc.stderr.read()
+        res, stderr_content = _read_json_line(proc)
+        if res is None:
             print(f"  [FAIL] Received EOF from process. Stderr:\n{stderr_content}")
+            assessment = classify_namespace_probe(
+                name,
+                required_tool=tool_name,
+                probe_result={"success": False, "error": stderr_content or "EOF"},
+                process={
+                    "pid": proc.pid,
+                    "profile": env.get("GITEA_MCP_PROFILE"),
+                    "env": env,
+                },
+                config_path=config_path,
+                probe_source="offline_spawn",
+            )
+            print(f"  remediation: {' '.join(assessment['remediation'])}")
             proc.terminate()
             return False
             
-        print(f"  [OK] Received initialize response: {line.strip()[:150]}...")
+        print(f"  [OK] Received initialize response: {str(res)[:150]}...")
         
         # Send initialized notification
         init_notif = {
             "jsonrpc": "2.0",
             "method": "notifications/initialized"
         }
-        proc.stdin.write(json.dumps(init_notif) + "\n")
-        proc.stdin.flush()
+        _write_message(proc, init_notif)
         
         # Send tools/list request
         list_req = {
@@ -76,16 +120,27 @@ def run_connection_test(name, config):
             "params": {},
             "id": 2
         }
-        proc.stdin.write(json.dumps(list_req) + "\n")
-        proc.stdin.flush()
+        _write_message(proc, list_req)
         
-        line = proc.stdout.readline()
-        if not line:
+        res, stderr_content = _read_json_line(proc)
+        if res is None:
             print("  [FAIL] Received EOF on tools/list request.")
+            assessment = classify_namespace_probe(
+                name,
+                required_tool=tool_name,
+                probe_result={"success": False, "error": stderr_content or "EOF"},
+                process={
+                    "pid": proc.pid,
+                    "profile": env.get("GITEA_MCP_PROFILE"),
+                    "env": env,
+                },
+                config_path=config_path,
+                probe_source="offline_spawn",
+            )
+            print(f"  remediation: {' '.join(assessment['remediation'])}")
             proc.terminate()
             return False
             
-        res = json.loads(line)
         if "error" in res:
             print(f"  [FAIL] Server returned error: {res['error']}")
             proc.terminate()
@@ -94,16 +149,115 @@ def run_connection_test(name, config):
         tools = res.get("result", {}).get("tools", [])
         tool_names = [t.get("name") for t in tools]
         print(f"  [OK] Successfully retrieved {len(tool_names)} tools: {tool_names[:5]}...")
+        if tool_name not in tool_names:
+            assessment = classify_namespace_probe(
+                name,
+                required_tool=tool_name,
+                registered_tools=tool_names,
+                probe_result={"success": False, "error": "required tool missing"},
+                process={
+                    "pid": proc.pid,
+                    "profile": env.get("GITEA_MCP_PROFILE"),
+                    "env": env,
+                },
+                config_path=config_path,
+                probe_source="offline_spawn",
+            )
+            print(f"  [FAIL] Required tool '{tool_name}' is not registered.")
+            print(f"  remediation: {' '.join(assessment['remediation'])}")
+            proc.terminate()
+            return False
+
+        call_req = {
+            "jsonrpc": "2.0",
+            "method": "tools/call",
+            "params": {"name": tool_name, "arguments": {}},
+            "id": 3,
+        }
+        _write_message(proc, call_req)
+
+        call_res, stderr_content = _read_json_line(proc)
+        if call_res is None:
+            assessment = classify_namespace_probe(
+                name,
+                required_tool=tool_name,
+                registered_tools=tool_names,
+                probe_result={"success": False, "error": stderr_content or "EOF"},
+                process={
+                    "pid": proc.pid,
+                    "profile": env.get("GITEA_MCP_PROFILE"),
+                    "env": env,
+                },
+                config_path=config_path,
+                probe_source="offline_spawn",
+            )
+            print(f"  [FAIL] Received EOF on {tool_name} invocation.")
+            print(f"  diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}")
+            print(f"  remediation: {' '.join(assessment['remediation'])}")
+            proc.terminate()
+            return False
+        if "error" in call_res:
+            assessment = classify_namespace_probe(
+                name,
+                required_tool=tool_name,
+                registered_tools=tool_names,
+                probe_result={"success": False, "error": call_res["error"]},
+                process={
+                    "pid": proc.pid,
+                    "profile": env.get("GITEA_MCP_PROFILE"),
+                    "env": env,
+                },
+                config_path=config_path,
+                probe_source="offline_spawn",
+            )
+            print(f"  [FAIL] {tool_name} invocation returned error: {call_res['error']}")
+            print(f"  remediation: {' '.join(assessment['remediation'])}")
+            proc.terminate()
+            return False
+
+        assessment = classify_namespace_probe(
+            name,
+            required_tool=tool_name,
+            registered_tools=tool_names,
+            probe_result={"success": True, "result": call_res.get("result")},
+            process={
+                "pid": proc.pid,
+                "profile": env.get("GITEA_MCP_PROFILE"),
+                "env": env,
+            },
+            config_path=config_path,
+                probe_source="offline_spawn",
+        )
+        print(f"  [OK] Successfully invoked required tool '{tool_name}'.")
+        print(f"  diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}")
         
         proc.terminate()
         return True
-    except Exception as e:
-        print(f"  [FAIL] Error during handshake: {e}")
+    except Exception as exc:
+        print(f"  [FAIL] Error during handshake: {exc}")
         proc.terminate()
         return False
 
+
 def main():
-    config_path = "/Users/jasonwalker/.gemini/config/mcp_config.json"
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--config",
+        default=os.environ.get(
+            "MCP_CONFIG_PATH",
+            os.path.expanduser("~/.gemini/config/mcp_config.json"),
+        ),
+        help="Path to MCP config JSON.",
+    )
+    parser.add_argument(
+        "--namespace",
+        action="append",
+        dest="namespaces",
+        help="Namespace to test. May be repeated.",
+    )
+    args = parser.parse_args()
+
+    config_path = args.config
     try:
         with open(config_path) as f:
             mcp_config = json.load(f)
@@ -112,10 +266,17 @@ def main():
         sys.exit(1)
         
     servers = mcp_config.get("mcpServers", {})
+    namespaces = args.namespaces or list(REQUIRED_NAMESPACE_TOOLS)
     failed = False
-    for name in ["gitea-author", "gitea-reviewer"]:
+    for name in namespaces:
         if name in servers:
-            if not run_connection_test(name, servers[name]):
+            if not run_connection_test(
+                name,
+                servers[name],
+                required_tool=REQUIRED_NAMESPACE_TOOLS.get(name),
+                config_path=config_path,
+                probe_source="offline_spawn",
+            ):
                 failed = True
         else:
             print(f"Server '{name}' not found in mcp_config.json")

tests/conftest.py

@@ -66,6 +66,7 @@ def _reset_mutation_authority(monkeypatch):
     monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
     monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
     monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
+    monkeypatch.setattr(mcp_server, "_LIVE_NAMESPACE_HEALTH", {})
     monkeypatch.setattr(mcp_server, "_preflight_whoami_called", False)
     monkeypatch.setattr(mcp_server, "_preflight_capability_called", False)
     monkeypatch.setattr(mcp_server, "_preflight_resolved_role", None)

tests/test_mcp_namespace_health.py

@@ -0,0 +1,208 @@
+import unittest
+
+import gitea_mcp_server
+import mcp_namespace_health
+import review_merge_state_machine as rmsm
+
+
+class TestMcpNamespaceHealth(unittest.TestCase):
+    def setUp(self):
+        gitea_mcp_server._LIVE_NAMESPACE_HEALTH.clear()
+
+    def test_registered_tool_but_live_eof_blocks_merge(self):
+        result = mcp_namespace_health.classify_namespace_probe(
+            "gitea-reviewer",
+            required_tool="gitea_whoami",
+            registered_tools=["gitea_whoami", "gitea_list_profiles"],
+            probe_result={
+                "success": False,
+                "error": "client is closing: EOF",
+            },
+            process={
+                "pid": 4321,
+                "profile": "prgs-reviewer",
+                "env": {
+                    "GITEA_MCP_PROFILE": "prgs-reviewer",
+                    "GITEA_TOKEN": "must-not-leak",
+                },
+            },
+            config_path="/Users/jasonwalker/.gemini/config/mcp_config.json",
+            probe_source="client_namespace",
+        )
+
+        self.assertFalse(result["healthy"])
+        self.assertEqual(result["error_type"], "namespace_eof")
+        self.assertTrue(result["required_tool_registered"])
+        self.assertFalse(result["required_tool_callable"])
+        self.assertTrue(result["blocks_merge_workflow"])
+        self.assertFalse(result["ide_namespace_proven"])
+        self.assertEqual(result["probe_source"], "client_namespace")
+        self.assertEqual(result["diagnostics"]["process_pid"], 4321)
+        self.assertEqual(result["diagnostics"]["profile"], "prgs-reviewer")
+        self.assertEqual(
+            result["diagnostics"]["env"]["GITEA_MCP_PROFILE"],
+            "prgs-reviewer",
+        )
+        self.assertNotIn("GITEA_TOKEN", result["diagnostics"]["env"])
+        self.assertIn("gitea-reviewer", result["reasons"][0])
+        self.assertIn("gitea_whoami", result["reasons"][0])
+
+    def test_successful_client_namespace_invocation_is_ide_proven(self):
+        result = mcp_namespace_health.classify_namespace_probe(
+            "gitea-author",
+            registered_tools=["gitea_whoami"],
+            probe_result={"success": True, "result": {"authenticated": True}},
+            process={"pid": 1234, "profile": "prgs-author"},
+            config_path="/tmp/mcp_config.json",
+            probe_source="client_namespace",
+        )
+
+        self.assertTrue(result["healthy"])
+        self.assertTrue(result["required_tool_registered"])
+        self.assertTrue(result["required_tool_callable"])
+        self.assertTrue(result["ide_namespace_proven"])
+        self.assertFalse(result["blocks_merge_workflow"])
+        self.assertIsNone(result["error_type"])
+
+    def test_offline_spawn_success_is_not_ide_proof(self):
+        result = mcp_namespace_health.classify_namespace_probe(
+            "gitea-merger",
+            registered_tools=["gitea_whoami"],
+            probe_result={"success": True, "result": {}},
+            process={"pid": 99, "profile": "prgs-merger"},
+            probe_source="offline_spawn",
+        )
+        self.assertTrue(result["healthy"])
+        self.assertFalse(result["ide_namespace_proven"])
+        self.assertFalse(result["blocks_merge_workflow"])
+        self.assertTrue(
+            any("offline_spawn" in r for r in result["reasons"])
+        )
+
+    def test_registered_missing_required_tool_fails_before_probe_success(self):
+        result = mcp_namespace_health.classify_namespace_probe(
+            "gitea-tools",
+            registered_tools=["gitea_whoami"],
+            probe_result={"success": True, "result": {}},
+            probe_source="client_namespace",
+        )
+
+        self.assertFalse(result["healthy"])
+        self.assertEqual(result["required_tool"], "gitea_list_profiles")
+        self.assertFalse(result["required_tool_registered"])
+        self.assertEqual(result["error_type"], "tool_missing")
+
+    def test_server_tool_exposes_same_assessment_and_records_session(self):
+        result = gitea_mcp_server.gitea_assess_mcp_namespace_health(
+            "gitea-merger",
+            registered_tools=["gitea_whoami"],
+            probe_result={"success": False, "error": "transport closed"},
+            process={"pid": 9876, "profile": "prgs-merger"},
+            probe_source="client_namespace",
+        )
+
+        self.assertFalse(result["success"])
+        self.assertEqual(result["error_type"], "namespace_eof")
+        self.assertEqual(result["namespace"], "gitea-merger")
+        self.assertEqual(result["diagnostics"]["process_pid"], 9876)
+        self.assertIn("gitea-merger", gitea_mcp_server._LIVE_NAMESPACE_HEALTH)
+        gate = gitea_mcp_server._live_namespace_health_gate("merge_pr")
+        self.assertTrue(gate)
+        self.assertTrue(any("gitea-merger" in r for r in gate))
+
+
+class TestLiveNamespaceBlocksMerge(unittest.TestCase):
+    """AC5: a broken live namespace must hard-block the review/merge state machine."""
+
+    def setUp(self):
+        gitea_mcp_server._LIVE_NAMESPACE_HEALTH.clear()
+
+    def _merge_ready_completion(self):
+        # Completion through PRE_MERGE_RECHECK is the state where a clean merge
+        # is otherwise allowed (see test_merge_allowed_with_pre_merge_gates).
+        idx = rmsm.REVIEW_MERGE_STATES.index("PRE_MERGE_RECHECK")
+        return {state: True for state in rmsm.REVIEW_MERGE_STATES[: idx + 1]}
+
+    def _all_gates(self):
+        return {gate: True for gate in rmsm._PRE_MERGE_REQUIRED_GATES}
+
+    def test_assess_workflow_blockers_flags_live_namespace_broken(self):
+        clean = rmsm.assess_workflow_blockers()
+        self.assertFalse(clean["block"])
+
+        broken = rmsm.assess_workflow_blockers(live_namespace_broken=True)
+        self.assertTrue(broken["block"])
+        self.assertTrue(any("namespace" in r.lower() for r in broken["reasons"]))
+
+    def test_can_merge_blocks_even_when_all_gates_pass(self):
+        # Without the namespace blocker a fully-complete workflow can merge...
+        allowed = rmsm.can_merge(
+            self._merge_ready_completion(), pre_merge_gates=self._all_gates()
+        )
+        self.assertTrue(allowed["allowed"])
+
+        # ...but a broken live namespace overrides every satisfied gate.
+        blocked = rmsm.can_merge(
+            self._merge_ready_completion(),
+            pre_merge_gates=self._all_gates(),
+            live_namespace_broken=True,
+        )
+        self.assertTrue(blocked["block"])
+        self.assertFalse(blocked["allowed"])
+
+    def test_workflow_status_forwards_namespace_blocker(self):
+        status = rmsm.workflow_status(
+            self._merge_ready_completion(), live_namespace_broken=True
+        )
+        self.assertFalse(status["merge_allowed"])
+        self.assertFalse(status["approve_allowed"])
+
+    def test_server_tool_blocks_merge_on_live_namespace_broken(self):
+        result = gitea_mcp_server.gitea_assess_review_merge_state_machine(
+            state_completion=self._merge_ready_completion(),
+            pre_merge_gates=self._all_gates(),
+            live_namespace_broken=True,
+        )
+        self.assertTrue(result["blockers"]["block"])
+        self.assertFalse(result["merge"]["allowed"])
+
+    def test_classify_verdict_bridges_into_merge_block(self):
+        # The classify verdict is the intended feed for live_namespace_broken.
+        verdict = mcp_namespace_health.classify_namespace_probe(
+            "gitea-merger",
+            registered_tools=["gitea_whoami", "gitea_adopt_merger_pr_lease"],
+            probe_result={"success": False, "error": "client is closing: EOF"},
+            process={"pid": 555, "profile": "prgs-merger"},
+            probe_source="client_namespace",
+        )
+        self.assertTrue(verdict["blocks_merge_workflow"])
+        blocked = rmsm.can_merge(
+            self._merge_ready_completion(),
+            pre_merge_gates=self._all_gates(),
+            live_namespace_broken=verdict["blocks_merge_workflow"],
+        )
+        self.assertFalse(blocked["allowed"])
+
+    def test_offline_spawn_does_not_authorize_mutation_gate(self):
+        gitea_mcp_server.gitea_assess_mcp_namespace_health(
+            "gitea-merger",
+            registered_tools=["gitea_whoami"],
+            probe_result={"success": True, "result": {}},
+            probe_source="offline_spawn",
+        )
+        gate = gitea_mcp_server._live_namespace_health_gate("merge_pr")
+        self.assertTrue(gate)
+        self.assertTrue(any("offline_spawn" in r or "client_namespace" in r for r in gate))
+
+    def test_client_namespace_healthy_clears_mutation_gate(self):
+        gitea_mcp_server.gitea_assess_mcp_namespace_health(
+            "gitea-merger",
+            registered_tools=["gitea_whoami"],
+            probe_result={"success": True, "result": {}},
+            probe_source="client_namespace",
+        )
+        self.assertEqual(gitea_mcp_server._live_namespace_health_gate("merge_pr"), [])
+
+
+if __name__ == "__main__":
+    unittest.main()