diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index cf7c9e7dda1..bf5ca3d357a 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -17,6 +17,7 @@ import base64 import binascii import concurrent.futures import functools +from collections import deque from dataclasses import dataclass from datetime import datetime, timezone import hashlib @@ -214,10 +215,15 @@ async def _lifespan(app: "FastAPI"): # Reap idle/dead keep-alive PTY sessions in the background (30-min TTL). pty_reaper_task = asyncio.create_task(run_reaper(PTY_REGISTRY)) + # Periodic authenticated self-test (feeds the ``dashboard`` component on + # /api/status). The loop exits immediately when httpx is unavailable. + selftest_task = asyncio.create_task(_dashboard_selftest_loop()) + try: yield finally: pty_reaper_task.cancel() + selftest_task.cancel() await PTY_REGISTRY.close_all() if cron_stop is not None: cron_stop.set() @@ -625,6 +631,137 @@ async def _token_auth_seam(request: Request, call_next): return await token_auth_middleware(request, call_next) +# --------------------------------------------------------------------------- +# Dashboard component health — in-process error/self-test counters that feed +# the ``components`` dict on ``/api/status``. That endpoint is in +# ``PUBLIC_API_PATHS``, so everything exported from here must be counts and +# enums only: no exception messages, no request paths, no tokens. +# --------------------------------------------------------------------------- + +_DASHBOARD_HEALTH_WINDOW_SECONDS = 300.0 + + +class DashboardHealth: + """Module-level holder for dashboard-process health signals. + + Tracks unhandled exceptions / 5xx responses seen by the outermost HTTP + middleware (rolling window) and the result of the periodic authenticated + self-test. ``last_error_path`` and ``last_error_type`` are internal + diagnostics for logs/debuggers — :meth:`snapshot` deliberately exports + neither (public-payload no-secrets contract). + """ + + def __init__(self, window_seconds: float = _DASHBOARD_HEALTH_WINDOW_SECONDS) -> None: + self.window_seconds = window_seconds + self._error_times: "deque[float]" = deque(maxlen=256) + self.last_error_type: Optional[str] = None + self.last_error_path: Optional[str] = None # internal-only, never serialized + self.last_error_at: Optional[float] = None + self.selftest_status: str = "unknown" # unknown | ok | failing + self.selftest_http_status: Optional[int] = None + self.selftest_at: Optional[float] = None + + def record_error(self, exc_type: str, path: str) -> None: + now = time.time() + self._error_times.append(now) + self.last_error_type = exc_type + self.last_error_path = path + self.last_error_at = now + + def record_selftest(self, passed: bool, http_status: Optional[int]) -> None: + self.selftest_status = "ok" if passed else "failing" + self.selftest_http_status = http_status + self.selftest_at = time.time() + + def recent_error_count(self) -> int: + cutoff = time.time() - self.window_seconds + while self._error_times and self._error_times[0] < cutoff: + self._error_times.popleft() + return len(self._error_times) + + def snapshot(self) -> Dict[str, Any]: + """Public component payload: status enum + counts + timestamps only.""" + errors = self.recent_error_count() + status = "degraded" if (errors or self.selftest_status == "failing") else "ok" + return { + "status": status, + "recent_unhandled_errors": errors, + "last_error_at": self.last_error_at, + "selftest": self.selftest_status, + } + + +DASHBOARD_HEALTH = DashboardHealth() + + +@app.middleware("http") +async def _dashboard_health_middleware(request: Request, call_next): + """Outermost middleware: count unhandled exceptions and 5xx responses. + + Registered after ``_token_auth_seam`` so it is the outermost layer + (Starlette middleware is outermost-last) — nothing below can raise past + it unseen. Records into :data:`DASHBOARD_HEALTH` and re-raises; never + swallows or alters the response. + """ + try: + response = await call_next(request) + except Exception as exc: + DASHBOARD_HEALTH.record_error(type(exc).__name__, request.url.path) + raise + if response.status_code >= 500: + DASHBOARD_HEALTH.record_error(f"http_{response.status_code}", request.url.path) + return response + + +# --------------------------------------------------------------------------- +# Authenticated-route self-test: every minute, make one in-process request +# against a cheap DB-touching authenticated route with the real session +# token. Catches the class of failure where liveness looks fine but every +# authenticated request 500s (e.g. wedged state DB). +# --------------------------------------------------------------------------- + +_DASHBOARD_SELFTEST_INTERVAL_SECONDS = 60.0 +_DASHBOARD_SELFTEST_ROUTE = "/api/sessions?limit=1" + + +async def _dashboard_selftest_once() -> None: + """Run one authenticated in-process self-test request and record it.""" + try: + import httpx + except ImportError: + return # optional dependency — skip cleanly, leave status "unknown" + try: + transport = httpx.ASGITransport(app=app) + # base_url uses a loopback name so the Host-header middleware accepts + # the request on loopback binds. + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1" + ) as client: + resp = await client.get( + _DASHBOARD_SELFTEST_ROUTE, + headers={_SESSION_HEADER_NAME: _SESSION_TOKEN}, + ) + DASHBOARD_HEALTH.record_selftest(resp.status_code == 200, resp.status_code) + except Exception: + DASHBOARD_HEALTH.record_selftest(False, None) + + +async def _dashboard_selftest_loop() -> None: + """Periodic self-test driver started from the lifespan.""" + try: + import httpx # noqa: F401 + except ImportError: + _log.debug("httpx unavailable — dashboard self-test disabled") + return + while True: + await asyncio.sleep(_DASHBOARD_SELFTEST_INTERVAL_SECONDS) + # On OAuth-gated binds the legacy session token is not honoured, so + # the probe would false-alarm 401 — skip until the gate is off. + if getattr(app.state, "auth_required", False): + continue + await _dashboard_selftest_once() + + # --------------------------------------------------------------------------- # Config schema — auto-generated from DEFAULT_CONFIG # --------------------------------------------------------------------------- @@ -2995,6 +3132,49 @@ async def get_status(profile: Optional[str] = None): "nous_session_valid": nous_session_valid, } + # Component-level health rollup. Counts and status enums only — this + # payload is public (PUBLIC_API_PATHS), so no messages, paths, or + # other detail that could carry secrets. The storage probe reuses the + # gateway readiness state_db check (read-only, 1s-bounded) in an + # executor so a wedged DB can't stall the event loop. + components: Dict[str, Any] = { + "gateway": { + "status": "ok" if gateway_running and gateway_state in {"running", "draining"} else "degraded", + "state": gateway_state or ("running" if gateway_running else "stopped"), + }, + "dashboard": DASHBOARD_HEALTH.snapshot(), + } + try: + from gateway.readiness import _probe_state_db + + storage_check = await asyncio.get_running_loop().run_in_executor( + None, functools.partial(_probe_state_db, get_hermes_home()) + ) + components["storage"] = {"status": storage_check.get("status", "degraded")} + except Exception: + components["storage"] = {"status": "degraded"} + platform_states = [ + str(value.get("state") or value.get("status") or "").lower() + for value in gateway_platforms.values() + if isinstance(value, dict) + ] + platforms_ok = all( + state in {"connected", "running", "ok"} for state in platform_states + ) + components["platforms"] = { + "status": "ok" if platforms_ok else "degraded", + "configured": len(gateway_platforms), + "connected": sum( + 1 for state in platform_states if state in {"connected", "running", "ok"} + ), + } + status["components"] = components + status["overall"] = ( + "ok" + if all(item.get("status") == "ok" for item in components.values()) + else "degraded" + ) + # Profile + gateway topology: which profiles exist, whether one # multiplexed gateway or several per-profile gateways serve them, and # (gated) which host ports the live gateways' port-binding platforms diff --git a/tests/hermes_cli/test_web_server.py b/tests/hermes_cli/test_web_server.py index b06ec4dd926..1e847ab5dda 100644 --- a/tests/hermes_cli/test_web_server.py +++ b/tests/hermes_cli/test_web_server.py @@ -9276,3 +9276,193 @@ class TestServeIndexMissingIndex: resp = client.get("/chat") assert resp.status_code == 200 assert "SPA-rebuilt" in resp.text + + +class TestDashboardComponentHealth: + """Component-health rollup: error middleware, /api/status components, self-test.""" + + @pytest.fixture(autouse=True) + def _setup(self, monkeypatch, _isolate_hermes_home): + try: + from starlette.testclient import TestClient + except ImportError: + pytest.skip("fastapi/starlette not installed") + + import hermes_state + from hermes_constants import get_hermes_home + import hermes_cli.web_server as ws + + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db") + # Fresh state holder per test so counters don't leak across tests. + monkeypatch.setattr(ws, "DASHBOARD_HEALTH", ws.DashboardHealth()) + self.ws = ws + self.client = TestClient(ws.app, raise_server_exceptions=False) + self.client.headers[ws._SESSION_HEADER_NAME] = ws._SESSION_TOKEN + + # -- middleware ------------------------------------------------------- + + def test_middleware_counts_unhandled_exception(self): + """An exception escaping a route must be recorded (and re-raised).""" + route_path = "/api/_test_boom" + + async def _boom(): + raise RuntimeError("kaboom") + + from fastapi.routing import APIRoute + + self.ws.app.router.routes.insert( + 0, APIRoute(route_path, _boom, methods=["GET"]) + ) + try: + resp = self.client.get(route_path) + assert resp.status_code == 500 + assert self.ws.DASHBOARD_HEALTH.recent_error_count() == 1 + assert self.ws.DASHBOARD_HEALTH.last_error_type == "RuntimeError" + # Path is retained internally only — snapshot must not export it. + assert self.ws.DASHBOARD_HEALTH.last_error_path == route_path + finally: + self.ws.app.router.routes[:] = [ + r for r in self.ws.app.router.routes + if getattr(r, "path", None) != route_path + ] + + def test_middleware_counts_5xx_response(self): + route_path = "/api/_test_teapot_fire" + + async def _fivehundred(): + from fastapi.responses import JSONResponse + return JSONResponse(status_code=503, content={"detail": "down"}) + + from fastapi.routing import APIRoute + + self.ws.app.router.routes.insert( + 0, APIRoute(route_path, _fivehundred, methods=["GET"]) + ) + try: + resp = self.client.get(route_path) + assert resp.status_code == 503 + assert self.ws.DASHBOARD_HEALTH.recent_error_count() == 1 + assert self.ws.DASHBOARD_HEALTH.last_error_type == "http_503" + finally: + self.ws.app.router.routes[:] = [ + r for r in self.ws.app.router.routes + if getattr(r, "path", None) != route_path + ] + + def test_error_window_expires_old_entries(self, monkeypatch): + health = self.ws.DashboardHealth(window_seconds=300) + now = {"t": 1000.0} + monkeypatch.setattr(self.ws.time, "time", lambda: now["t"]) + health.record_error("RuntimeError", "/api/x") + assert health.recent_error_count() == 1 + now["t"] = 1000.0 + 301 + assert health.recent_error_count() == 0 + + # -- /api/status components ------------------------------------------ + + def test_status_includes_components_and_overall(self): + resp = self.client.get("/api/status") + assert resp.status_code == 200 + data = resp.json() + assert data["overall"] in {"ok", "degraded"} + components = data["components"] + assert set(components) == {"gateway", "storage", "dashboard", "platforms"} + for comp in components.values(): + assert comp["status"] in {"ok", "degraded"} + dashboard = components["dashboard"] + assert dashboard["recent_unhandled_errors"] == 0 + assert "last_error_at" in dashboard + assert dashboard["selftest"] in {"unknown", "ok", "failing"} + + def test_storage_degraded_when_state_db_probe_fails(self, monkeypatch): + import gateway.readiness as readiness + + monkeypatch.setattr( + readiness, "_probe_state_db", lambda home: {"status": "degraded", "detail": "OperationalError"} + ) + resp = self.client.get("/api/status") + data = resp.json() + assert data["components"]["storage"] == {"status": "degraded"} + assert data["overall"] == "degraded" + + def test_dashboard_component_degraded_after_error(self): + self.ws.DASHBOARD_HEALTH.record_error("RuntimeError", "/api/x") + resp = self.client.get("/api/status") + data = resp.json() + dashboard = data["components"]["dashboard"] + assert dashboard["status"] == "degraded" + assert dashboard["recent_unhandled_errors"] == 1 + assert data["overall"] == "degraded" + + def test_public_component_payload_carries_no_secret_bearing_fields(self): + """PUBLIC_API_PATHS contract: counts/enums only — no paths/messages.""" + self.ws.DASHBOARD_HEALTH.record_error("RuntimeError", "/api/secret-route?token=abc") + resp = self.client.get("/api/status") + payload = json.dumps(resp.json()["components"]) + assert "secret-route" not in payload + assert "token=abc" not in payload + assert "last_error_path" not in payload + assert "last_error_type" not in payload + assert "kaboom" not in payload + + # -- self-test --------------------------------------------------------- + + def test_selftest_records_failure_on_500(self, monkeypatch): + httpx = pytest.importorskip("httpx") + + class _FakeResponse: + status_code = 500 + + class _FakeClient: + def __init__(self, *args, **kwargs): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def get(self, *args, **kwargs): + return _FakeResponse() + + monkeypatch.setattr(httpx, "AsyncClient", _FakeClient) + asyncio.run(self.ws._dashboard_selftest_once()) + assert self.ws.DASHBOARD_HEALTH.selftest_status == "failing" + assert self.ws.DASHBOARD_HEALTH.selftest_http_status == 500 + assert self.ws.DASHBOARD_HEALTH.snapshot()["status"] == "degraded" + + def test_selftest_records_pass_on_200(self, monkeypatch): + httpx = pytest.importorskip("httpx") + + class _FakeResponse: + status_code = 200 + + class _FakeClient: + def __init__(self, *args, **kwargs): + self.calls = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def get(self, url, headers=None, **kwargs): + # The probe must authenticate with the real session token. + assert headers[self.ws_header] == self.ws_token + return _FakeResponse() + + _FakeClient.ws_header = self.ws._SESSION_HEADER_NAME + _FakeClient.ws_token = self.ws._SESSION_TOKEN + monkeypatch.setattr(httpx, "AsyncClient", _FakeClient) + asyncio.run(self.ws._dashboard_selftest_once()) + assert self.ws.DASHBOARD_HEALTH.selftest_status == "ok" + assert self.ws.DASHBOARD_HEALTH.selftest_http_status == 200 + + def test_selftest_real_asgi_roundtrip(self): + """End-to-end: the in-process ASGI self-test hits the real route.""" + pytest.importorskip("httpx") + asyncio.run(self.ws._dashboard_selftest_once()) + assert self.ws.DASHBOARD_HEALTH.selftest_status in {"ok", "failing"} + assert self.ws.DASHBOARD_HEALTH.selftest_http_status is not None