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__)
 
@@ -898,10 +899,31 @@ class APIServerAdapter(BasePlatformAdapter):
         # from a request flood (#7483).
         self._max_concurrent_runs: int = self._resolve_max_concurrent_runs()
         # Number of in-flight runs on the non-streaming chat/responses paths
-        # (the /v1/runs path tracks its own in-flight set via
-        # _active_run_tasks).
+        # (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."""
@@ -1398,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,
@@ -3967,18 +3998,14 @@ class APIServerAdapter(BasePlatformAdapter):
 
         The cap bounds total in-flight agent activity across every
         agent-serving endpoint: the non-streaming chat/responses paths
-        (tracked by ``_inflight_agent_runs``) plus the ``/v1/runs`` path
-        (tracked by live entries in ``_active_run_tasks``).  Stream queues are
-        transport state and may disappear while their underlying run remains
-        active, so they must not define run concurrency. A configured value of
-        0 disables the cap entirely.
+        (tracked by ``_inflight_agent_runs``) plus the ``/v1/runs`` streaming
+        path (tracked by ``_run_streams``). A configured value of 0 disables
+        the cap entirely.
         """
         limit = self._max_concurrent_runs
         if limit <= 0:
             return None
-        inflight = self._inflight_agent_runs + sum(
-            not task.done() for task in self._active_run_tasks.values()
-        )
+        inflight = self._inflight_agent_runs + len(self._run_streams)
         if inflight >= limit:
             return web.json_response(
                 _openai_error(
@@ -4675,7 +4702,7 @@ class APIServerAdapter(BasePlatformAdapter):
         return web.json_response({"run_id": run_id, "status": "stopping"})
 
     async def _sweep_orphaned_runs(self) -> None:
-        """Periodically clean up expired transport state for inactive runs."""
+        """Periodically clean up run streams that were never consumed."""
         while True:
             await asyncio.sleep(60)
             now = time.time()
@@ -4683,10 +4710,6 @@ class APIServerAdapter(BasePlatformAdapter):
                 run_id
                 for run_id, created_at in list(self._run_streams_created.items())
                 if now - created_at > self._RUN_STREAM_TTL
-                and (
-                    (task := self._active_run_tasks.get(run_id)) is None
-                    or task.done()
-                )
             ]
             for run_id in stale:
                 logger.debug("[api_server] sweeping orphaned run %s", run_id)

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

@@ -568,29 +568,14 @@ class TestConcurrencyCap:
         assert resp.headers.get("Retry-After")
 
     def test_cap_counts_both_buckets(self):
-        # /v1/runs (tracked by live tasks) + chat/responses (inflight)
+        # /v1/runs (tracked by _run_streams) + chat/responses (inflight)
         adapter = _make_adapter()
         adapter._max_concurrent_runs = 4
         adapter._inflight_agent_runs = 2
-
-        async def _assert_live_tasks_are_counted_without_streams():
-            blocker = asyncio.Event()
-
-            async def _live_run():
-                await blocker.wait()
-
-            tasks = [asyncio.create_task(_live_run()) for _ in range(2)]
-            adapter._active_run_tasks = {f"r{i}": task for i, task in enumerate(tasks)}
-            adapter._run_streams = {}
-            try:
-                resp = adapter._concurrency_limited_response()
-                assert resp is not None
-                assert resp.status == 429
-            finally:
-                blocker.set()
-                await asyncio.gather(*tasks)
-
-        asyncio.run(_assert_live_tasks_are_counted_without_streams())
+        adapter._run_streams = {"r1": object(), "r2": object()}
+        resp = adapter._concurrency_limited_response()
+        assert resp is not None
+        assert resp.status == 429
 
     def test_zero_disables_cap(self):
         adapter = _make_adapter()
@@ -779,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.
@@ -804,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_api_server_runs.py

@@ -442,100 +442,6 @@ class TestRunEvents:
         assert resp.status == 401
 
 
-# ---------------------------------------------------------------------------
-# Run lifecycle TTL sweeping
-# ---------------------------------------------------------------------------
-
-
-class TestRunLifecycleSweep:
-    @pytest.mark.asyncio
-    async def test_expired_live_run_remains_counted_approvable_and_stoppable(self, adapter):
-        """Stream TTL must not detach control state from a still-running task."""
-        app = _create_runs_app(adapter)
-        adapter._max_concurrent_runs = 1
-
-        async with TestClient(TestServer(app)) as cli:
-            with patch.object(adapter, "_create_agent") as mock_create:
-                mock_agent, agent_ready, _ = _make_slow_agent()
-                mock_create.return_value = mock_agent
-
-                start_resp = await cli.post("/v1/runs", json={"input": "hello"})
-                assert start_resp.status == 202
-                run_id = (await start_resp.json())["run_id"]
-                assert agent_ready.wait(timeout=3.0)
-
-                task = adapter._active_run_tasks[run_id]
-                assert isinstance(task, asyncio.Task)
-                assert not task.done()
-
-                pending = approval_mod._ApprovalEntry({
-                    "command": "bash -c long-running",
-                    "description": "approval after stream TTL",
-                    "pattern_keys": ["shell-c"],
-                })
-                with approval_mod._lock:
-                    approval_mod._gateway_queues[run_id] = [pending]
-
-                adapter._run_streams_created[run_id] -= adapter._RUN_STREAM_TTL + 1
-                # Exercise one real sweeper iteration without waiting 60 seconds.
-                with patch(
-                    "gateway.platforms.api_server.asyncio.sleep",
-                    side_effect=[None, asyncio.CancelledError()],
-                ):
-                    with pytest.raises(asyncio.CancelledError):
-                        await adapter._sweep_orphaned_runs()
-
-                assert adapter._active_run_tasks[run_id] is task
-                assert adapter._active_run_agents[run_id] is mock_agent
-                assert run_id in adapter._run_streams
-                assert adapter._run_approval_sessions[run_id] == run_id
-
-                limited = adapter._concurrency_limited_response()
-                assert limited is not None
-                assert limited.status == 429
-
-                approval_resp = await cli.post(
-                    f"/v1/runs/{run_id}/approval",
-                    json={"choice": "once"},
-                )
-                assert approval_resp.status == 200
-                assert pending.event.is_set()
-                assert pending.result == "once"
-
-                stop_resp = await cli.post(f"/v1/runs/{run_id}/stop")
-                assert stop_resp.status == 200
-                mock_agent.interrupt.assert_called_once_with("Stop requested via API")
-
-    @pytest.mark.asyncio
-    async def test_expired_orphan_run_state_is_reaped(self, adapter):
-        run_id = "run_expired_orphan"
-        adapter._run_streams[run_id] = asyncio.Queue()
-        adapter._run_streams_created[run_id] = 0
-        adapter._run_approval_sessions[run_id] = run_id
-
-        pending = approval_mod._ApprovalEntry({
-            "command": "bash -c orphaned",
-            "description": "orphaned approval",
-            "pattern_keys": ["shell-c"],
-        })
-        with approval_mod._lock:
-            approval_mod._gateway_queues[run_id] = [pending]
-
-        with patch(
-            "gateway.platforms.api_server.asyncio.sleep",
-            side_effect=[None, asyncio.CancelledError()],
-        ):
-            with pytest.raises(asyncio.CancelledError):
-                await adapter._sweep_orphaned_runs()
-
-        assert run_id not in adapter._run_streams
-        assert run_id not in adapter._run_streams_created
-        assert run_id not in adapter._run_approval_sessions
-        assert pending.event.is_set()
-        with approval_mod._lock:
-            assert run_id not in approval_mod._gateway_queues
-
-
 # ---------------------------------------------------------------------------
 # POST /v1/runs/{run_id}/stop — interrupt a running agent
 # ---------------------------------------------------------------------------

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)