feat(kora): KR-MCP-CONSUMPTION ST2 — active health-check + payload extension

kora_cli/listeners/mcp_consumption.py

@@ -44,15 +44,163 @@
 from __future__ import annotations
 
 import logging
+import os
+from dataclasses import dataclass
+from datetime import datetime, timezone
 from typing import Optional
 
 from kora_cli.daemon import DEFAULT_SHUTDOWN_TIMEOUT, register_daemon_listener
+from kora_cli.listeners.heartbeat import register_periodic_task
 from kora_mcp.catalog import load_effective_catalog
-from kora_mcp.pool import MCPClientPool
+from kora_mcp.pool import MCPCallFailed, MCPClientPool
 
 logger = logging.getLogger(__name__)
 
 
+# ---------------------------------------------------------------------------
+# ST2 — Health-check task config
+# ---------------------------------------------------------------------------
+
+
+DEFAULT_HEALTH_CHECK_INTERVAL_SEC: float = 300.0  # 5 min per §4 Q1 ruling
+HEALTH_CHECK_INTERVAL_ENV: str = "KORA_MCP_HEALTH_CHECK_INTERVAL_SEC"
+
+
+def _read_health_check_interval() -> float:
+    """Read the per-cycle interval from env or fall back to default.
+
+    Operator override via ``KORA_MCP_HEALTH_CHECK_INTERVAL_SEC``
+    (Doppler-injectable). Invalid values (non-numeric, <=0) log
+    WARN + fall back to the 300s default.
+    """
+    raw = os.environ.get(HEALTH_CHECK_INTERVAL_ENV, "").strip()
+    if not raw:
+        return DEFAULT_HEALTH_CHECK_INTERVAL_SEC
+    try:
+        value = float(raw)
+    except ValueError:
+        logger.warning(
+            "[kora.mcp_consumption] %s=%r is not numeric; using "
+            "default %ss",
+            HEALTH_CHECK_INTERVAL_ENV,
+            raw,
+            DEFAULT_HEALTH_CHECK_INTERVAL_SEC,
+        )
+        return DEFAULT_HEALTH_CHECK_INTERVAL_SEC
+    if value <= 0:
+        logger.warning(
+            "[kora.mcp_consumption] %s=%s must be > 0; using "
+            "default %ss",
+            HEALTH_CHECK_INTERVAL_ENV,
+            value,
+            DEFAULT_HEALTH_CHECK_INTERVAL_SEC,
+        )
+        return DEFAULT_HEALTH_CHECK_INTERVAL_SEC
+    return value
+
+
+# ---------------------------------------------------------------------------
+# ST2 — Health snapshot
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True, slots=True)
+class HealthSnapshot:
+    """One per-endpoint health observation.
+
+    Populated by :func:`run_health_check` per scheduled cycle.
+    Cached in :data:`_health_cache` and surfaced through
+    :func:`current_health_snapshots`.
+
+    Attributes:
+        connected: ``True`` if the endpoint's MCP session opened
+            cleanly + ``list_tools()`` returned a result. ``False``
+            on any transport / protocol failure.
+        tools_count: Number of tools the endpoint exposed (when
+            ``connected``). ``None`` when not connected.
+        last_check_at: UTC time of the cycle that produced this
+            snapshot. Used by the operator panel to detect stale
+            data (snapshot older than the cadence → status visibly
+            degraded).
+        last_error: Operator-readable failure message (when not
+            connected). ``None`` on success.
+    """
+
+    connected: bool
+    tools_count: Optional[int]
+    last_check_at: datetime
+    last_error: Optional[str]
+
+
+_health_cache: dict[str, HealthSnapshot] = {}
+
+
+def current_health_snapshots() -> dict[str, HealthSnapshot]:
+    """Return a copy of the per-endpoint snapshot cache.
+
+    Read by the ``/api/mcp/clients/list`` endpoint to populate
+    ``last_check_at`` / ``last_error`` / ``tools_count`` /
+    ``status`` in the panel payload. Snapshot freshness is the
+    panel's responsibility — compare ``last_check_at`` against the
+    cadence to detect stale data.
+    """
+    return dict(_health_cache)
+
+
+def _clear_health_cache() -> None:
+    """Test hook + shutdown helper."""
+    global _health_cache
+    _health_cache = {}
+
+
+async def run_health_check() -> None:
+    """One pass: poll every endpoint in the current pool.
+
+    Per-endpoint failure (timeout, transport, protocol) is captured
+    on the snapshot — does NOT crash the heartbeat scheduler.
+    Per-endpoint timeout is enforced by
+    :meth:`MCPClientPool.list_tools` via the endpoint's configured
+    ``timeout_seconds`` (default 30s — matches §4 Q1 ruling).
+
+    Skips cleanly when no pool is running (daemon not yet started
+    or already shut down). The scheduler keeps firing; cache stays
+    empty until the listener is up.
+    """
+    pool = current_pool()
+    if pool is None:
+        logger.debug(
+            "[kora.mcp_consumption] health-check skipped: no active pool"
+        )
+        return
+    now = datetime.now(timezone.utc)
+    for prefix in pool.endpoint_names():
+        try:
+            tools = await pool.list_tools(prefix)
+        except MCPCallFailed as exc:
+            _health_cache[prefix] = HealthSnapshot(
+                connected=False,
+                tools_count=None,
+                last_check_at=now,
+                last_error=str(exc),
+            )
+        except Exception as exc:
+            # Defense in depth — unknown error type. Wrap so
+            # operator sees the type prefix.
+            _health_cache[prefix] = HealthSnapshot(
+                connected=False,
+                tools_count=None,
+                last_check_at=now,
+                last_error=f"{type(exc).__name__}: {exc}",
+            )
+        else:
+            _health_cache[prefix] = HealthSnapshot(
+                connected=True,
+                tools_count=len(tools),
+                last_check_at=now,
+                last_error=None,
+            )
+
+
 # ---------------------------------------------------------------------------
 # Listener
 # ---------------------------------------------------------------------------
@@ -92,13 +240,20 @@ async def startup(self) -> None:
     async def shutdown(self) -> None:
         """Close every cached pool session under the coordinator's
         per-listener shutdown timeout. Fail-soft on close errors —
-        the pool's close_all logs WARN + continues."""
+        the pool's close_all logs WARN + continues.
+
+        Also clears the ST2 health-snapshot cache so a subsequent
+        daemon start sees a clean slate (avoids stale snapshots
+        bleeding across restarts).
+        """
         if self._pool is None:
+            _clear_health_cache()
             return
         try:
             await self._pool.close_all()
         finally:
             _clear_singleton()
+            _clear_health_cache()
             self._pool = None
         logger.info("[kora.mcp_consumption] pool closed")
 
@@ -147,3 +302,20 @@ def _factory():
 
 
 register_daemon_listener("mcp_consumption", _factory)
+
+
+# ---------------------------------------------------------------------------
+# ST2 — Periodic health-check registration (import-time side effect)
+# ---------------------------------------------------------------------------
+#
+# The heartbeat scheduler owns the asyncio.Task; we just register
+# the callable + cadence. Cadence is read once at module-import
+# time per the §4 Q1 ruling (5min default; KORA_MCP_HEALTH_CHECK_INTERVAL_SEC
+# override). Restart-driven cadence-config refresh — matches the
+# Q3 restart-driven contract for the pool itself.
+
+register_periodic_task(
+    "mcp.health_check",
+    interval_seconds=_read_health_check_interval(),
+    callable=run_health_check,
+)

kora_cli/web_server.py

@@ -4772,40 +4772,88 @@ async def list_mcp_clients():
     security test in ``test_web_server_mcp_clients.py`` pins this
     invariant.
 
-    Status mapping (current — pre-KR-MCP-CONSUMPTION):
-
-      auth env unset / empty → ``unhealthy``
-      auth env set           → ``configured_but_unconnected``
-
-    The ``connected`` status surfaces once the daemon's MCP-client-
-    pool listener is wired (deferred per spec §3).
-    ``tools_count`` likewise remains ``null`` until that lands.
+    Status mapping (post-KR-MCP-CONSUMPTION ST2):
+
+      auth env unset / empty                              → ``unhealthy``
+      auth env set, no snapshot OR stale OR not connected → ``configured_but_unconnected``
+      auth env set + fresh snapshot + connected           → ``connected``
+
+    A snapshot is "stale" when ``last_check_at`` is older than the
+    health-check cadence (``KORA_MCP_HEALTH_CHECK_INTERVAL_SEC``,
+    default 300s). A missed heartbeat cycle surfaces as
+    ``configured_but_unconnected`` instead of falsely reporting
+    ``connected``.
+
+    ST2 additive payload fields:
+      last_check_at  — ISO string when snapshot was taken (or null)
+      last_error     — operator-readable failure string (or null)
+      tools_count    — real value from snapshot (null when not connected)
     """
-    from datetime import datetime, timezone
+    from datetime import datetime, timedelta, timezone
 
+    from kora_cli.listeners.mcp_consumption import (
+        DEFAULT_HEALTH_CHECK_INTERVAL_SEC,
+        _read_health_check_interval,
+        current_health_snapshots,
+    )
     from kora_mcp.catalog import check_endpoint_health, load_effective_catalog
 
     registry = load_effective_catalog()
+    snapshots = current_health_snapshots()
+    now = datetime.now(timezone.utc)
+    try:
+        cadence_seconds = _read_health_check_interval()
+    except Exception:
+        cadence_seconds = DEFAULT_HEALTH_CHECK_INTERVAL_SEC
+    staleness_threshold = timedelta(seconds=cadence_seconds)
+
     clients: list[dict[str, Any]] = []
     for endpoint in registry.endpoints:
         health = check_endpoint_health(endpoint)
+        snapshot = snapshots.get(endpoint.name)
+
+        # Status derivation per KR-MCP-CONSUMPTION ST2.
+        if not health.healthy:
+            status = "unhealthy"
+            tools_count = None
+            last_check_at = None
+            last_error = None
+        elif snapshot is None:
+            status = "configured_but_unconnected"
+            tools_count = None
+            last_check_at = None
+            last_error = None
+        else:
+            is_stale = (now - snapshot.last_check_at) > staleness_threshold
+            if snapshot.connected and not is_stale:
+                status = "connected"
+            else:
+                status = "configured_but_unconnected"
+            tools_count = (
+                snapshot.tools_count if snapshot.connected else None
+            )
+            last_check_at = snapshot.last_check_at.strftime(
+                "%Y-%m-%dT%H:%M:%SZ"
+            )
+            last_error = snapshot.last_error
+
         clients.append({
             "name": endpoint.name,
             "transport": endpoint.transport,
             "endpoint": endpoint.endpoint,
-            "status": (
-                "configured_but_unconnected" if health.healthy else "unhealthy"
-            ),
+            "status": status,
             "auth_token_env": endpoint.auth_token_env or "",
             "auth_token_present": health.auth_env_set
             and endpoint.auth_token_env is not None,
             "allowed_tools_regex": endpoint.allowed_tools_regex,
-            "tools_count": None,
+            "tools_count": tools_count,
+            "last_check_at": last_check_at,
+            "last_error": last_error,
         })
     return {
         "clients": clients,
         "stub": False,
-        "generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
+        "generated_at": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
     }
 
 

kora_mcp/pool.py

@@ -211,6 +211,34 @@ async def call_tool(
             ) from exc
         return _project_call_result(result)
 
+    async def list_tools(self, prefix: str) -> list[ToolDescriptor]:
+        """Per-endpoint ``list_tools`` with full error surfacing.
+
+        Mirrors :meth:`call_tool`'s lazy-open + per-call timeout +
+        cache-drop-on-error pattern. Returns the projected tool
+        descriptors on success; raises :exc:`MCPCallFailed` on any
+        transport / protocol error.
+
+        Used by callers that need per-endpoint health detail (e.g.
+        KR-MCP-CONSUMPTION ST2's health-check task populating
+        ``last_error`` per prefix). Distinct from
+        :meth:`list_tools_all` which is best-effort across the
+        whole catalog + maps failed endpoints to empty lists.
+        """
+        client = await self._ensure_open(prefix)
+        try:
+            timeout = self._by_name[prefix].timeout_seconds
+            result = await asyncio.wait_for(
+                client.session.list_tools(), timeout=timeout
+            )
+        except Exception as exc:
+            await self._close_one(prefix)
+            raise MCPCallFailed(
+                f"list_tools({prefix}) failed: "
+                f"{type(exc).__name__}: {exc}"
+            ) from exc
+        return _project_tools_result(result)
+
     async def list_tools_all(self) -> dict[str, list[ToolDescriptor]]:
         """Open EVERY configured endpoint + return its tool catalog.