From 2100c7eb9f61c35f4deb228cd60faa82bd4c0502 Mon Sep 17 00:00:00 2001 From: CC#1 Kora Substrate Date: Thu, 21 May 2026 23:37:32 -0700 Subject: [PATCH] =?UTF-8?q?feat(kora):=20KR-MCP-CONSUMPTION=20ST2=20?= =?UTF-8?q?=E2=80=94=20active=20health-check=20+=20payload=20extension?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the KR-MCP-CONSUMPTION bucket. After this PR, the heartbeat scheduler periodically polls each pool endpoint + the `/api/mcp/clients/list` panel surfaces live `connected` status + real `tools_count` + per-cycle `last_check_at` / `last_error`. # Periodic health-check (kora_cli/listeners/mcp_consumption.py) - `run_health_check()` — one pass: for each endpoint in the current pool, call `pool.list_tools(prefix)`; cache result as `HealthSnapshot(connected, tools_count, last_check_at, last_error)`. - Per-endpoint failure (MCPCallFailed or any other exception) captured on snapshot — does NOT crash the scheduler. - Per-endpoint 30s timeout via the endpoint's configured `timeout_seconds` (catalog defaults already set; matches §4 Q1). - `current_health_snapshots()` returns a defensive copy — caller mutations don't leak into the internal cache. - Shutdown clears the cache (avoid stale pre-restart data bleeding into the post-restart panel view). - Module-level `register_periodic_task("mcp.health_check", interval_seconds=_read_health_check_interval(), callable=run_health_check)` fires at import time. # Cadence (§4 Q1: 5min ruling) - DEFAULT_HEALTH_CHECK_INTERVAL_SEC = 300.0 - KORA_MCP_HEALTH_CHECK_INTERVAL_SEC env override - Invalid (non-numeric / ≤0) values WARN-log + fall back to default # Additive pool API `kora_mcp/pool.py:MCPClientPool.list_tools(prefix)` — per-endpoint list_tools with full error surfacing. Mirrors `call_tool`'s lazy-open + per-call timeout + cache-drop-on-error pattern; raises MCPCallFailed on transport / protocol failure (vs `list_tools_all` which catches per-endpoint errors + maps to empty lists). Needed by the health-check task to populate `last_error` per prefix. # Endpoint payload extension (kora_cli/web_server.py) `/api/mcp/clients/list` now reads snapshots from `current_health_snapshots()` and derives status: - 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 (compares `(now - last_check_at) > timedelta(seconds=cadence)`). Missed heartbeat cycles surface as `configured_but_unconnected` rather than falsely reporting `connected`. Additive payload fields (no breaking change for FE): - last_check_at: ISO string when snapshot was taken (or null) - last_error: operator-readable failure string (or null) - tools_count: from snapshot (null when not connected; preserves last-known count when stale + previously-connected) # TS interface (web/src/lib/api.ts) `MCPClient` interface extended with `last_check_at: string | null` and `last_error: string | null`. Additive — existing consumers keep working; new FE bucket (KR-MCP-CLIENTS-HEALTH-DISPLAY) lands the operator-visible rendering on CC#2's lane. # Security contract preserved (carried from KR-MCP-3 + FLIP) Both CC#2 security tests stay green: - Regex pin on auth_token_env shape (UPPER_SNAKE) - Walk-all-keys guard against token-value-shaped fields Neither last_check_at nor last_error introduces a token-bearing field. last_error is the str(exception) from MCPCallFailed which is constructed deliberately to NOT include sensitive substrate state. # Tests 29 new tests across 2 files: `tests/kora_cli/test_listeners/test_mcp_consumption_health.py` (17 tests): - run_health_check skips cleanly when no pool - One pass populates snapshot per endpoint - Successful list_tools → connected=True + tools_count=N - MCPCallFailed → connected=False + last_error preserved - Unknown exception type wrapped with ExceptionType prefix - Per-endpoint failure doesn't block siblings - current_health_snapshots returns a copy - Shutdown clears cache (with + without prior startup) - _read_health_check_interval: default / env override / invalid / non-positive / negative fallback - Periodic task registered in PERIODIC_TASK_REGISTRY at import `tests/kora_cli/test_web_server_mcp_clients.py` (5 new tests + 1 updated shape test): - Updated `required = {...}` set to include last_check_at + last_error - Connected snapshot → status=connected + tools_count + ISO last_check_at - Failed snapshot → status=configured_but_unconnected + last_error surfaced - Stale snapshot (older than cadence) → status=configured_but_ unconnected even if snapshot.connected=True - No snapshot → status=configured_but_unconnected + all additive fields null - Auth env unset → status=unhealthy overrides any snapshot; additive fields null 102/102 cross-bucket regression (test_listeners/ + test_web_server_mcp_clients.py + tests/kora_mcp/) clean. 1 expected skip (integration test gated behind KORA_INTEGRATION_TEST=1). Ruff clean. Manual smoke: `curl localhost:9119/api/mcp/clients/list` returns the new shape with last_check_at + last_error fields (both null when no daemon/snapshots present, which matches the unhealthy-because-no-auth-env path). # After ST2 merges External MCP calls wired end-to-end. The MCP-clients UI surfaces live connection status + tools_count once the daemon's heartbeat scheduler completes its first cycle (~5min after start). FE visualization of last_check_at / last_error is a small CC#2 follow-on (KR-MCP-CLIENTS-HEALTH-DISPLAY). Co-Authored-By: Claude Opus 4.7 (1M context) --- kora_cli/listeners/mcp_consumption.py | 176 +++++++- kora_cli/web_server.py | 76 +++- kora_mcp/pool.py | 28 ++ .../test_mcp_consumption_health.py | 377 ++++++++++++++++++ tests/kora_cli/test_web_server_mcp_clients.py | 167 ++++++++ web/src/lib/api.ts | 12 + 6 files changed, 820 insertions(+), 16 deletions(-) create mode 100644 tests/kora_cli/test_listeners/test_mcp_consumption_health.py diff --git a/kora_cli/listeners/mcp_consumption.py b/kora_cli/listeners/mcp_consumption.py index 20a24ab25b8b..26755c59d24f 100644 --- a/kora_cli/listeners/mcp_consumption.py +++ b/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, +) diff --git a/kora_cli/web_server.py b/kora_cli/web_server.py index 3c214f2d7488..0a5196901653 100644 --- a/kora_cli/web_server.py +++ b/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"), } diff --git a/kora_mcp/pool.py b/kora_mcp/pool.py index 2ba5bea83d51..541845f94b65 100644 --- a/kora_mcp/pool.py +++ b/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. diff --git a/tests/kora_cli/test_listeners/test_mcp_consumption_health.py b/tests/kora_cli/test_listeners/test_mcp_consumption_health.py new file mode 100644 index 000000000000..adc87ebbb316 --- /dev/null +++ b/tests/kora_cli/test_listeners/test_mcp_consumption_health.py @@ -0,0 +1,377 @@ +"""Tests for KR-MCP-CONSUMPTION ST2 — active health-check. + +Covers: + - ``run_health_check`` skips cleanly when no pool is registered + - One pass populates ``_health_cache`` with one snapshot per + pool endpoint + - Successful ``list_tools`` → snapshot.connected=True + + tools_count=N + last_error=None + - ``list_tools`` raises MCPCallFailed → snapshot.connected=False + + tools_count=None + last_error preserved + - Unknown exception type (defense in depth) → snapshot.connected=False + + last_error includes ExceptionType prefix + - Per-endpoint failure does NOT block siblings + - ``current_health_snapshots`` returns a copy (mutations don't + leak into cache) + - Shutdown clears the cache + - Listener startup-without-shutdown leaves cache empty (no + auto-populate) + - ``_read_health_check_interval``: default / env override / + invalid value fallback / non-positive value fallback + - Periodic task ``mcp.health_check`` registered in + PERIODIC_TASK_REGISTRY at module-import time with correct + cadence + callable +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + +import pytest + +from kora_cli.listeners import heartbeat as heartbeat_module +from kora_cli.listeners.mcp_consumption import ( + DEFAULT_HEALTH_CHECK_INTERVAL_SEC, + HEALTH_CHECK_INTERVAL_ENV, + HealthSnapshot, + MCPConsumptionListener, + _clear_health_cache, + _clear_singleton, + _read_health_check_interval, + current_health_snapshots, + run_health_check, +) +from kora_mcp.catalog import DEFAULT_CATALOG +from kora_mcp.pool import MCPCallFailed, ToolDescriptor +from kora_mcp.registry import MCPRegistryConfig + + +@pytest.fixture(autouse=True) +def _reset_state(): + _clear_singleton() + _clear_health_cache() + yield + _clear_singleton() + _clear_health_cache() + + +# --------------------------------------------------------------------------- +# run_health_check — no-pool branch +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_run_health_check_skips_when_no_pool(caplog): + """No active pool (daemon not started / already stopped) → + snapshot cache untouched + DEBUG log only. Scheduler keeps + firing; doesn't crash.""" + import logging + + with caplog.at_level(logging.DEBUG, logger="kora_cli.listeners.mcp_consumption"): + await run_health_check() + assert current_health_snapshots() == {} + assert any("no active pool" in r.message for r in caplog.records) + + +# --------------------------------------------------------------------------- +# run_health_check — happy path + per-endpoint outcomes +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_one_pass_populates_snapshot_per_endpoint(): + """Successful list_tools per endpoint → cache contains one + HealthSnapshot per endpoint name; all marked connected with + matching tools_count.""" + default_registry = MCPRegistryConfig(endpoints=list(DEFAULT_CATALOG)) + with patch( + "kora_cli.listeners.mcp_consumption.load_effective_catalog", + return_value=default_registry, + ): + listener = MCPConsumptionListener() + await listener.startup() + + # Stub list_tools to return different counts per endpoint + async def _fake_list_tools(prefix): + if prefix == "github": + return [ + ToolDescriptor(name="create_issue", description=None, input_schema=None), + ToolDescriptor(name="list_repos", description=None, input_schema=None), + ] + return [ToolDescriptor(name="d1_query", description=None, input_schema=None)] + + with patch.object(listener.pool, "list_tools", new=_fake_list_tools): + await run_health_check() + + snapshots = current_health_snapshots() + assert set(snapshots.keys()) == {"github", "cloudflare"} + assert snapshots["github"].connected is True + assert snapshots["github"].tools_count == 2 + assert snapshots["github"].last_error is None + assert snapshots["cloudflare"].connected is True + assert snapshots["cloudflare"].tools_count == 1 + assert snapshots["cloudflare"].last_error is None + + await listener.shutdown() + + +@pytest.mark.asyncio +async def test_mcp_call_failed_recorded_on_snapshot(): + """MCPCallFailed → snapshot.connected=False + tools_count=None + + last_error captures the failure message.""" + default_registry = MCPRegistryConfig(endpoints=list(DEFAULT_CATALOG)) + with patch( + "kora_cli.listeners.mcp_consumption.load_effective_catalog", + return_value=default_registry, + ): + listener = MCPConsumptionListener() + await listener.startup() + + async def _failing_list_tools(prefix): + raise MCPCallFailed(f"list_tools({prefix}) failed: transport boom") + + with patch.object(listener.pool, "list_tools", new=_failing_list_tools): + await run_health_check() + + snapshots = current_health_snapshots() + for prefix in ("github", "cloudflare"): + assert snapshots[prefix].connected is False + assert snapshots[prefix].tools_count is None + assert "transport boom" in snapshots[prefix].last_error + + await listener.shutdown() + + +@pytest.mark.asyncio +async def test_unknown_exception_type_wrapped_with_type_prefix(): + """Defense in depth: a non-MCPCallFailed exception still gets + captured, with ExceptionType prefix so operator can diagnose.""" + default_registry = MCPRegistryConfig(endpoints=list(DEFAULT_CATALOG)) + with patch( + "kora_cli.listeners.mcp_consumption.load_effective_catalog", + return_value=default_registry, + ): + listener = MCPConsumptionListener() + await listener.startup() + + async def _broken_list_tools(prefix): + raise ValueError("malformed response shape") + + with patch.object(listener.pool, "list_tools", new=_broken_list_tools): + await run_health_check() + + snapshots = current_health_snapshots() + for prefix in ("github", "cloudflare"): + assert snapshots[prefix].connected is False + assert "ValueError" in snapshots[prefix].last_error + assert "malformed response shape" in snapshots[prefix].last_error + + await listener.shutdown() + + +@pytest.mark.asyncio +async def test_per_endpoint_failure_does_not_block_siblings(): + """Mixed outcomes: github fails, cloudflare succeeds. Both + snapshots present, each reflecting its own outcome.""" + default_registry = MCPRegistryConfig(endpoints=list(DEFAULT_CATALOG)) + with patch( + "kora_cli.listeners.mcp_consumption.load_effective_catalog", + return_value=default_registry, + ): + listener = MCPConsumptionListener() + await listener.startup() + + async def _mixed_list_tools(prefix): + if prefix == "github": + raise MCPCallFailed("github down") + return [ + ToolDescriptor(name="d1_query", description=None, input_schema=None) + ] + + with patch.object(listener.pool, "list_tools", new=_mixed_list_tools): + await run_health_check() + + snapshots = current_health_snapshots() + assert snapshots["github"].connected is False + assert snapshots["github"].last_error is not None + assert snapshots["cloudflare"].connected is True + assert snapshots["cloudflare"].tools_count == 1 + + await listener.shutdown() + + +# --------------------------------------------------------------------------- +# current_health_snapshots is a copy (cache isolation) +# --------------------------------------------------------------------------- + + +def test_current_health_snapshots_returns_copy(): + """Mutating the returned dict must not affect the internal + cache — callers can freely modify their view.""" + now = datetime.now(timezone.utc) + from kora_cli.listeners.mcp_consumption import _health_cache + + _health_cache["test"] = HealthSnapshot( + connected=True, tools_count=5, last_check_at=now, last_error=None + ) + view = current_health_snapshots() + view["test"] = HealthSnapshot( + connected=False, tools_count=0, last_check_at=now, last_error="tampered" + ) + # Internal cache unchanged + assert current_health_snapshots()["test"].connected is True + assert current_health_snapshots()["test"].last_error is None + + +# --------------------------------------------------------------------------- +# Listener shutdown clears cache +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_shutdown_clears_health_cache(): + """Cache persistence across daemon restarts would surface stale + pre-restart data on the panel. Cleared on shutdown.""" + default_registry = MCPRegistryConfig(endpoints=list(DEFAULT_CATALOG)) + with patch( + "kora_cli.listeners.mcp_consumption.load_effective_catalog", + return_value=default_registry, + ): + listener = MCPConsumptionListener() + await listener.startup() + + async def _ok(prefix): + return [ToolDescriptor(name="x", description=None, input_schema=None)] + + with patch.object(listener.pool, "list_tools", new=_ok): + await run_health_check() + assert len(current_health_snapshots()) == 2 + + await listener.shutdown() + assert current_health_snapshots() == {} + + +@pytest.mark.asyncio +async def test_shutdown_without_startup_still_clears_cache(): + """If a prior daemon left a snapshot behind + the new daemon + starts but immediately shuts down without populating, cache is + still cleared. (No-pool shutdown short-circuit must still clear + the cache.)""" + now = datetime.now(timezone.utc) + from kora_cli.listeners.mcp_consumption import _health_cache + + _health_cache["stale_from_prior_daemon"] = HealthSnapshot( + connected=True, tools_count=99, last_check_at=now, last_error=None + ) + assert current_health_snapshots() != {} + + listener = MCPConsumptionListener() + # No startup; immediate shutdown + await listener.shutdown() + assert current_health_snapshots() == {} + + +# --------------------------------------------------------------------------- +# _read_health_check_interval +# --------------------------------------------------------------------------- + + +def test_interval_default_when_env_unset(monkeypatch): + monkeypatch.delenv(HEALTH_CHECK_INTERVAL_ENV, raising=False) + assert _read_health_check_interval() == DEFAULT_HEALTH_CHECK_INTERVAL_SEC + + +def test_interval_env_override(monkeypatch): + monkeypatch.setenv(HEALTH_CHECK_INTERVAL_ENV, "60") + assert _read_health_check_interval() == 60.0 + + +def test_interval_env_override_floats(monkeypatch): + monkeypatch.setenv(HEALTH_CHECK_INTERVAL_ENV, "45.5") + assert _read_health_check_interval() == 45.5 + + +def test_interval_invalid_value_falls_back(monkeypatch, caplog): + import logging + + monkeypatch.setenv(HEALTH_CHECK_INTERVAL_ENV, "not-a-number") + with caplog.at_level( + logging.WARNING, logger="kora_cli.listeners.mcp_consumption" + ): + result = _read_health_check_interval() + assert result == DEFAULT_HEALTH_CHECK_INTERVAL_SEC + assert any("not numeric" in r.message for r in caplog.records) + + +def test_interval_zero_falls_back(monkeypatch, caplog): + import logging + + monkeypatch.setenv(HEALTH_CHECK_INTERVAL_ENV, "0") + with caplog.at_level( + logging.WARNING, logger="kora_cli.listeners.mcp_consumption" + ): + result = _read_health_check_interval() + assert result == DEFAULT_HEALTH_CHECK_INTERVAL_SEC + assert any("must be > 0" in r.message for r in caplog.records) + + +def test_interval_negative_falls_back(monkeypatch): + monkeypatch.setenv(HEALTH_CHECK_INTERVAL_ENV, "-30") + assert _read_health_check_interval() == DEFAULT_HEALTH_CHECK_INTERVAL_SEC + + +# --------------------------------------------------------------------------- +# Periodic task registered at module-import time +# --------------------------------------------------------------------------- + + +def test_periodic_task_registered_in_registry(): + """Import-time side effect: ``register_periodic_task("mcp.health_check", ...)`` + fires when kora_cli.listeners.mcp_consumption imports. The + HeartbeatScheduler picks it up at daemon start.""" + # Force import so registration fires (already imported, but + # safe in this test path). + from kora_cli.listeners import mcp_consumption # noqa: F401 + + names = {t.name for t in heartbeat_module.PERIODIC_TASK_REGISTRY} + assert "mcp.health_check" in names + + task = next( + t + for t in heartbeat_module.PERIODIC_TASK_REGISTRY + if t.name == "mcp.health_check" + ) + # Cadence honors env (or default); the callable is run_health_check + assert task.interval_seconds > 0 + assert task.callable is run_health_check + + +# --------------------------------------------------------------------------- +# Stale-snapshot derivation (mirrors what the panel endpoint computes) +# --------------------------------------------------------------------------- + + +def test_stale_snapshot_detection_via_threshold(): + """The endpoint compares (now - snapshot.last_check_at) against + the cadence to decide stale. A snapshot taken longer ago than + the cadence is "stale" → panel reports + configured_but_unconnected instead of connected. This test pins + the timedelta-based comparison the endpoint uses.""" + now = datetime.now(timezone.utc) + cadence = timedelta(seconds=DEFAULT_HEALTH_CHECK_INTERVAL_SEC) + + fresh = HealthSnapshot( + connected=True, + tools_count=5, + last_check_at=now - timedelta(seconds=60), + last_error=None, + ) + stale = HealthSnapshot( + connected=True, + tools_count=5, + last_check_at=now - timedelta(seconds=DEFAULT_HEALTH_CHECK_INTERVAL_SEC + 60), + last_error=None, + ) + assert (now - fresh.last_check_at) <= cadence + assert (now - stale.last_check_at) > cadence diff --git a/tests/kora_cli/test_web_server_mcp_clients.py b/tests/kora_cli/test_web_server_mcp_clients.py index 5c9ff31fe0f5..48403d988c04 100644 --- a/tests/kora_cli/test_web_server_mcp_clients.py +++ b/tests/kora_cli/test_web_server_mcp_clients.py @@ -112,6 +112,9 @@ async def test_each_client_entry_has_required_keys_and_valid_enums(_isolate_conf "auth_token_present", "allowed_tools_regex", "tools_count", + # KR-MCP-CONSUMPTION ST2 additive fields + "last_check_at", + "last_error", } for client in result["clients"]: assert set(client.keys()) == required @@ -128,6 +131,13 @@ async def test_each_client_entry_has_required_keys_and_valid_enums(_isolate_conf assert isinstance(client["tools_count"], int) else: assert client["tools_count"] is None + # ST2 additive fields: null or string + assert client["last_check_at"] is None or isinstance( + client["last_check_at"], str + ) + assert client["last_error"] is None or isinstance( + client["last_error"], str + ) # ---- 5. SECURITY: no token-value shapes ---------------------------- @@ -347,3 +357,160 @@ async def test_cron_endpoint_still_works_with_mcp_clients_registered(_isolate_co jobs = await web_server.list_cron_jobs(profile="all") assert isinstance(jobs, list) + + +# ---- 9. KR-MCP-CONSUMPTION ST2 snapshot wiring ---------------------- + + +@pytest.fixture +def _clear_consumption_state(): + """Reset the listener's singletons between snapshot tests.""" + from kora_cli.listeners.mcp_consumption import ( + _clear_health_cache, + _clear_singleton, + ) + + _clear_singleton() + _clear_health_cache() + yield + _clear_singleton() + _clear_health_cache() + + +def _seed_snapshot( + prefix: str, + *, + connected: bool, + tools_count, + last_error, + age_seconds: int = 0, +) -> None: + from kora_cli.listeners.mcp_consumption import HealthSnapshot, _health_cache + from datetime import datetime, timedelta, timezone + + _health_cache[prefix] = HealthSnapshot( + connected=connected, + tools_count=tools_count, + last_check_at=datetime.now(timezone.utc) - timedelta(seconds=age_seconds), + last_error=last_error, + ) + + +@pytest.mark.asyncio +async def test_snapshot_connected_surfaces_status_and_tools_count( + _isolate_config, _clear_consumption_state, monkeypatch +): + """Fresh connected snapshot + auth env set → status=connected + + tools_count=N + last_check_at populated + last_error=None.""" + monkeypatch.setenv("KORA_MCP_GITHUB_TOKEN", "ghp_real") + _seed_snapshot("github", connected=True, tools_count=7, last_error=None) + + from kora_cli import web_server + + result = await web_server.list_mcp_clients() + github = next(c for c in result["clients"] if c["name"] == "github") + assert github["status"] == "connected" + assert github["tools_count"] == 7 + assert github["last_check_at"] is not None + assert github["last_error"] is None + + +@pytest.mark.asyncio +async def test_snapshot_failed_surfaces_configured_but_unconnected_and_last_error( + _isolate_config, _clear_consumption_state, monkeypatch +): + """Failed snapshot (connected=False) + auth env set → + status=configured_but_unconnected + tools_count=null + last_error + surfaced.""" + monkeypatch.setenv("KORA_MCP_GITHUB_TOKEN", "ghp_real") + _seed_snapshot( + "github", + connected=False, + tools_count=None, + last_error="MCPCallFailed: transport timeout", + ) + + from kora_cli import web_server + + result = await web_server.list_mcp_clients() + github = next(c for c in result["clients"] if c["name"] == "github") + assert github["status"] == "configured_but_unconnected" + assert github["tools_count"] is None + assert github["last_error"] == "MCPCallFailed: transport timeout" + assert github["last_check_at"] is not None + + +@pytest.mark.asyncio +async def test_stale_snapshot_degrades_to_configured_but_unconnected( + _isolate_config, _clear_consumption_state, monkeypatch +): + """Connected snapshot OLDER than the cadence → treated as stale + → status=configured_but_unconnected even though + snapshot.connected=True. The heartbeat scheduler missed a cycle; + operator sees the gap rather than a false "connected" badge.""" + from kora_cli.listeners.mcp_consumption import ( + DEFAULT_HEALTH_CHECK_INTERVAL_SEC, + ) + + monkeypatch.setenv("KORA_MCP_GITHUB_TOKEN", "ghp_real") + # Snapshot taken 2x the cadence ago — definitively stale + _seed_snapshot( + "github", + connected=True, + tools_count=7, + last_error=None, + age_seconds=int(DEFAULT_HEALTH_CHECK_INTERVAL_SEC * 2), + ) + + from kora_cli import web_server + + result = await web_server.list_mcp_clients() + github = next(c for c in result["clients"] if c["name"] == "github") + assert github["status"] == "configured_but_unconnected" + # tools_count still surfaces (operator can see the last-known + # count even when stale) — verify the policy: snapshot.connected= + # True means tools_count IS the snapshot's count; staleness only + # affects status string. + assert github["tools_count"] == 7 + assert github["last_check_at"] is not None + + +@pytest.mark.asyncio +async def test_no_snapshot_surfaces_configured_but_unconnected_with_null_fields( + _isolate_config, _clear_consumption_state, monkeypatch +): + """Auth env set + no snapshot in cache (daemon just started; no + heartbeat cycle yet) → status=configured_but_unconnected + + tools_count/last_check_at/last_error all null.""" + monkeypatch.setenv("KORA_MCP_GITHUB_TOKEN", "ghp_real") + monkeypatch.setenv("KORA_MCP_CLOUDFLARE_TOKEN", "cf_real") + # Don't seed any snapshot + + from kora_cli import web_server + + result = await web_server.list_mcp_clients() + for client in result["clients"]: + assert client["status"] == "configured_but_unconnected" + assert client["tools_count"] is None + assert client["last_check_at"] is None + assert client["last_error"] is None + + +@pytest.mark.asyncio +async def test_unhealthy_overrides_snapshot_when_auth_env_unset( + _isolate_config, _clear_consumption_state, monkeypatch +): + """Auth env unset → status=unhealthy + ALL snapshot fields null + (snapshot ignored — the auth gap is the gating signal).""" + monkeypatch.delenv("KORA_MCP_GITHUB_TOKEN", raising=False) + # Seed a connected snapshot — it should NOT override unhealthy + _seed_snapshot("github", connected=True, tools_count=99, last_error=None) + + from kora_cli import web_server + + result = await web_server.list_mcp_clients() + github = next(c for c in result["clients"] if c["name"] == "github") + assert github["status"] == "unhealthy" + assert github["tools_count"] is None + assert github["last_check_at"] is None + assert github["last_error"] is None diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 709c4caa3f17..174d2adb2140 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -1454,6 +1454,18 @@ export interface MCPClient { auth_token_present: boolean; allowed_tools_regex: string | null; tools_count: number | null; + // KR-MCP-CONSUMPTION ST2 additive fields. The daemon's heartbeat + // scheduler probes each endpoint every + // KORA_MCP_HEALTH_CHECK_INTERVAL_SEC (default 300s); these + // capture the last cycle's result. + // last_check_at: ISO string when the snapshot was taken + // (null if no cycle has run yet for this endpoint). + // last_error: operator-readable failure string from the last + // cycle (null on success / no snapshot). + // FE rendering of these fields lands as a small follow-on bucket + // (KR-MCP-CLIENTS-HEALTH-DISPLAY) on CC#2's lane. + last_check_at: string | null; + last_error: string | null; } export interface MCPClientsListResponse {