diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 5ba09d67492..896efaf3bba 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -61,6 +61,7 @@ from gateway.platforms.base import ( validate_media_delivery_path, ) from agent.redact import redact_sensitive_text +from gateway.readiness import collect_runtime_readiness logger = logging.getLogger(__name__) @@ -901,6 +902,28 @@ class APIServerAdapter(BasePlatformAdapter): # (the /v1/runs path tracks its own in-flight set via _run_streams). self._inflight_agent_runs: int = 0 + def _readiness_queue_depths(self) -> tuple[int, int, int]: + """Return queue counts only; readiness must never inspect payloads.""" + process_depth = 0 + delegation_depth = 0 + try: + from tools.process_registry import process_registry + + process_depth = process_registry.completion_queue.qsize() + except Exception: + pass + try: + from tools.async_delegation import list_async_delegations + + delegation_depth = sum( + 1 + for record in list_async_delegations() + if record.get("status") in {"queued", "running"} + ) + except Exception: + pass + return len(self._run_streams), process_depth, delegation_depth + @staticmethod def _parse_cors_origins(value: Any) -> tuple[str, ...]: """Normalize configured CORS origins into a stable tuple.""" @@ -1397,8 +1420,17 @@ class APIServerAdapter(BasePlatformAdapter): # This endpoint is served BY the gateway process, so it is by definition # alive — gateway_running is True. Derive busy/drainable from the same # shared contract /api/status uses so the two surfaces never disagree. + api_depth, process_depth, delegation_depth = self._readiness_queue_depths() + readiness = collect_runtime_readiness( + model_name=self._model_name, + runtime_status=runtime, + api_run_queue_depth=api_depth, + process_completion_queue_depth=process_depth, + delegation_completion_queue_depth=delegation_depth, + ) return web.json_response({ - "status": "ok", + "status": readiness["status"], + "readiness": readiness, "platform": "hermes-agent", "version": _hermes_version(), "gateway_state": gw_state, diff --git a/gateway/readiness.py b/gateway/readiness.py new file mode 100644 index 00000000000..13953d26ad4 --- /dev/null +++ b/gateway/readiness.py @@ -0,0 +1,130 @@ +"""Bounded, non-destructive readiness probes for authenticated health surfaces.""" + +from __future__ import annotations + +import shutil +import sqlite3 +from pathlib import Path +from typing import Any + +import yaml + +from hermes_constants import get_hermes_home + + +_DISK_DEGRADED_PERCENT = 90.0 + + +def _check(status: str, detail: str | None = None, **extra: Any) -> dict[str, Any]: + result: dict[str, Any] = {"status": status} + if detail: + result["detail"] = detail + result.update(extra) + return result + + +def _probe_state_db(home: Path) -> dict[str, Any]: + path = home / "state.db" + if not path.exists(): + return _check("ok", "not initialized") + try: + # mode=rw prevents a health probe from creating or repairing state. + uri = f"file:{path.as_posix()}?mode=rw" + with sqlite3.connect(uri, uri=True, timeout=1.0) as conn: + row = conn.execute("PRAGMA quick_check(1)").fetchone() + conn.execute("BEGIN IMMEDIATE") + conn.execute("ROLLBACK") + if not row or str(row[0]).lower() != "ok": + return _check("degraded", "integrity check failed") + return _check("ok") + except Exception as exc: + return _check("degraded", type(exc).__name__) + + +def _probe_config(home: Path) -> dict[str, Any]: + path = home / "config.yaml" + if not path.exists(): + return _check("ok", "using defaults") + try: + raw = yaml.safe_load(path.read_text(encoding="utf-8")) + if raw is not None and not isinstance(raw, dict): + return _check("degraded", "top level is not a mapping") + return _check("ok") + except Exception as exc: + # Recommendation 15 may provide a durable snapshot on a sibling branch. + # Recognize it without depending on its implementation or exposing paths. + lkg_candidates = ( + home / "config.last-known-good.yaml", + home / ".config.last-known-good.yaml", + ) + if any(candidate.is_file() for candidate in lkg_candidates): + return _check("degraded", "invalid config; last-known-good available") + return _check("degraded", f"invalid config ({type(exc).__name__})") + + +def _probe_disk(home: Path) -> dict[str, Any]: + try: + usage = shutil.disk_usage(home) + used_pct = round((usage.used / usage.total) * 100, 1) if usage.total else 0.0 + status = "degraded" if used_pct >= _DISK_DEGRADED_PERCENT else "ok" + return _check(status, used_percent=used_pct, free_bytes=usage.free) + except Exception as exc: + return _check("degraded", type(exc).__name__) + + +def _probe_gateway(runtime_status: dict[str, Any]) -> dict[str, Any]: + state = str(runtime_status.get("gateway_state") or "unknown") + platforms = runtime_status.get("platforms") + connected = 0 + configured = 0 + if isinstance(platforms, dict): + configured = len(platforms) + connected = sum( + 1 + for value in platforms.values() + if isinstance(value, dict) + and str(value.get("state") or value.get("status") or "").lower() + in {"connected", "running", "ok"} + ) + status = "ok" if state in {"running", "draining"} else "degraded" + return _check(status, state=state, connected_platforms=connected, platforms=configured) + + +def collect_runtime_readiness( + *, + model_name: str, + runtime_status: dict[str, Any] | None, + api_run_queue_depth: int = 0, + process_completion_queue_depth: int = 0, + delegation_completion_queue_depth: int = 0, +) -> dict[str, Any]: + """Return bounded readiness diagnostics without mutating runtime state. + + The detailed health endpoint is authenticated. Even there, probes expose + status and counts only: never config values, credentials, paths, commands, + queue payloads, or exception messages. + """ + home = get_hermes_home() + runtime = runtime_status if isinstance(runtime_status, dict) else {} + checks = { + "state_db": _probe_state_db(home), + "config": _probe_config(home), + "provider": _check("ok" if str(model_name or "").strip() else "degraded"), + "disk": _probe_disk(home), + "gateway": _probe_gateway(runtime), + "scheduler": _check( + "ok" if runtime.get("updated_at") and runtime.get("gateway_state") == "running" else "degraded", + "gateway ticker status unavailable" if not runtime.get("updated_at") else None, + ), + "background_queues": _check( + "ok", + api_runs=max(0, int(api_run_queue_depth)), + process_completions=max(0, int(process_completion_queue_depth)), + delegation_completions=max(0, int(delegation_completion_queue_depth)), + ), + } + overall = "ok" if all(item.get("status") == "ok" for item in checks.values()) else "degraded" + return {"status": overall, "checks": checks} + + +__all__ = ["collect_runtime_readiness"] diff --git a/tests/gateway/test_api_server.py b/tests/gateway/test_api_server.py index 1aed7455eef..1ecc065d688 100644 --- a/tests/gateway/test_api_server.py +++ b/tests/gateway/test_api_server.py @@ -764,7 +764,8 @@ class TestHealthDetailedEndpoint: resp = await cli.get("/health/detailed") assert resp.status == 200 data = await resp.json() - assert data["status"] == "ok" + assert data["status"] == "degraded" + assert data["readiness"]["checks"]["gateway"]["status"] == "degraded" assert data["gateway_state"] is None assert data["platforms"] == {} # No runtime file ⇒ state None ⇒ not busy, not drainable. @@ -789,6 +790,36 @@ class TestHealthDetailedEndpoint: resp = await cli.get("/health/detailed", headers=headers) assert resp.status == 200 + @pytest.mark.asyncio + async def test_health_detailed_reports_runtime_readiness(self, adapter): + """Detailed health exposes bounded readiness probes without changing /health.""" + app = _create_app(adapter) + expected = { + "status": "degraded", + "checks": { + "state_db": {"status": "ok"}, + "config": {"status": "degraded", "detail": "using last-known-good"}, + }, + } + with patch("gateway.status.read_runtime_status", return_value={"gateway_state": "running"}), \ + patch("gateway.platforms.api_server.collect_runtime_readiness", return_value=expected): + async with TestClient(TestServer(app)) as cli: + resp = await cli.get("/health/detailed") + assert resp.status == 200 + data = await resp.json() + assert data["status"] == "degraded" + assert data["readiness"] == expected + + @pytest.mark.asyncio + async def test_public_health_does_not_run_readiness_probes(self, adapter): + app = _create_app(adapter) + with patch("gateway.platforms.api_server.collect_runtime_readiness") as probe: + async with TestClient(TestServer(app)) as cli: + resp = await cli.get("/health") + assert resp.status == 200 + assert (await resp.json())["status"] == "ok" + probe.assert_not_called() + # --------------------------------------------------------------------------- # /v1/models endpoint diff --git a/tests/gateway/test_readiness.py b/tests/gateway/test_readiness.py new file mode 100644 index 00000000000..46c8dad93d3 --- /dev/null +++ b/tests/gateway/test_readiness.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +import json +import os +import sqlite3 +from pathlib import Path + +from gateway.readiness import collect_runtime_readiness + + +def test_collect_runtime_readiness_reports_healthy_local_runtime(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + home.mkdir() + (home / "config.yaml").write_text( + "model:\n provider: openrouter\n model: test/model\n", + encoding="utf-8", + ) + with sqlite3.connect(home / "state.db") as conn: + conn.execute("CREATE TABLE probe (id INTEGER PRIMARY KEY)") + monkeypatch.setenv("HERMES_HOME", str(home)) + + result = collect_runtime_readiness( + model_name="test/model", + runtime_status={ + "gateway_state": "running", + "platforms": {"telegram": {"state": "connected"}}, + "updated_at": "2026-07-09T00:00:00Z", + }, + api_run_queue_depth=2, + ) + + assert result["status"] == "ok" + assert result["checks"]["state_db"]["status"] == "ok" + assert result["checks"]["config"]["status"] == "ok" + assert result["checks"]["provider"]["status"] == "ok" + assert result["checks"]["gateway"]["status"] == "ok" + assert result["checks"]["background_queues"]["api_runs"] == 2 + assert result["checks"]["disk"]["status"] in {"ok", "degraded"} + + +def test_collect_runtime_readiness_degrades_on_invalid_config_and_stopped_gateway( + tmp_path, monkeypatch +): + home = tmp_path / ".hermes" + home.mkdir() + (home / "config.yaml").write_text("model: [unterminated", encoding="utf-8") + monkeypatch.setenv("HERMES_HOME", str(home)) + + result = collect_runtime_readiness( + model_name="", + runtime_status={"gateway_state": "stopped", "platforms": {}}, + ) + + assert result["status"] == "degraded" + assert result["checks"]["config"]["status"] == "degraded" + assert result["checks"]["provider"]["status"] == "degraded" + assert result["checks"]["gateway"]["status"] == "degraded" + # Readiness is diagnostic data, not an exception or a destructive repair. + assert (home / "config.yaml").read_text(encoding="utf-8") == "model: [unterminated" + + +def test_collect_runtime_readiness_marks_corrupt_state_db_degraded(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + home.mkdir() + (home / "config.yaml").write_text("{}\n", encoding="utf-8") + (home / "state.db").write_bytes(b"not sqlite") + monkeypatch.setenv("HERMES_HOME", str(home)) + + result = collect_runtime_readiness(model_name="configured-model", runtime_status={}) + + assert result["status"] == "degraded" + assert result["checks"]["state_db"]["status"] == "degraded" + assert "detail" in result["checks"]["state_db"] + + +def test_collect_runtime_readiness_never_exposes_config_values(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + home.mkdir() + secret = "do-not-return-this-value" + (home / "config.yaml").write_text( + f"model:\n provider: openrouter\nprivate_value: {secret}\n", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(home)) + + result = collect_runtime_readiness(model_name="model", runtime_status={}) + + assert secret not in json.dumps(result) + assert str(home) not in json.dumps(result) + assert result["checks"]["config"]["status"] == "ok" + + +def test_collect_runtime_readiness_uses_active_profile_home(tmp_path, monkeypatch): + profile_home = tmp_path / "profiles" / "coder" + profile_home.mkdir(parents=True) + (profile_home / "config.yaml").write_text("{}\n", encoding="utf-8") + monkeypatch.setenv("HERMES_HOME", str(profile_home)) + + result = collect_runtime_readiness(model_name="model", runtime_status={}) + + assert result["checks"]["config"]["status"] == "ok" + assert not (tmp_path / ".hermes" / "state.db").exists() + assert os.environ["HERMES_HOME"] == str(profile_home)