feat(gateway): add authenticated runtime readiness checks

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,

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"]

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

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)