From 6142203bd7af6c5d78f5dd0d58dbe64af5c02345 Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Sat, 11 Jul 2026 05:50:45 -0700 Subject: [PATCH] fix(gateway): ground readiness in live runtime state --- gateway/platforms/api_server.py | 31 +++++++++------- gateway/readiness.py | 37 ++++++------------- tests/gateway/test_api_server.py | 19 +++++++++- tests/gateway/test_readiness.py | 18 ++++----- .../docs/user-guide/features/api-server.md | 10 ++++- .../current/user-guide/features/api-server.md | 4 +- 6 files changed, 67 insertions(+), 52 deletions(-) diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 896efaf3bba3..c71d92a29b52 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -902,10 +902,15 @@ 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.""" + def _readiness_work_counts(self) -> tuple[int, int, int]: + """Return bounded work counts from each subsystem's public state.""" + active_api_runs = sum( + 1 + for status in self._run_statuses.values() + if status.get("status") in {"queued", "running", "waiting_for_approval"} + ) process_depth = 0 - delegation_depth = 0 + active_delegations = 0 try: from tools.process_registry import process_registry @@ -913,16 +918,12 @@ class APIServerAdapter(BasePlatformAdapter): except Exception: pass try: - from tools.async_delegation import list_async_delegations + from tools.async_delegation import active_count - delegation_depth = sum( - 1 - for record in list_async_delegations() - if record.get("status") in {"queued", "running"} - ) + active_delegations = active_count() except Exception: pass - return len(self._run_streams), process_depth, delegation_depth + return active_api_runs, process_depth, active_delegations @staticmethod def _parse_cors_origins(value: Any) -> tuple[str, ...]: @@ -1420,13 +1421,15 @@ 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() + active_api_runs, process_depth, active_delegations = self._readiness_work_counts() + from gateway.run import _resolve_gateway_model + readiness = collect_runtime_readiness( - model_name=self._model_name, + configured_model=_resolve_gateway_model(), runtime_status=runtime, - api_run_queue_depth=api_depth, + active_api_runs=active_api_runs, process_completion_queue_depth=process_depth, - delegation_completion_queue_depth=delegation_depth, + active_delegations=active_delegations, ) return web.json_response({ "status": readiness["status"], diff --git a/gateway/readiness.py b/gateway/readiness.py index 13953d26ad48..379e074bb40d 100644 --- a/gateway/readiness.py +++ b/gateway/readiness.py @@ -28,14 +28,13 @@ def _probe_state_db(home: Path) -> dict[str, Any]: 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" + # A readiness probe must never compete with normal state writers. A + # read-only schema query still catches unreadable/corrupt databases + # without taking a write reservation on every health poll. + uri = f"file:{path.as_posix()}?mode=ro" 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") + conn.execute("PRAGMA query_only = ON") + conn.execute("SELECT name FROM sqlite_master LIMIT 1").fetchone() return _check("ok") except Exception as exc: return _check("degraded", type(exc).__name__) @@ -51,14 +50,6 @@ def _probe_config(home: Path) -> dict[str, Any]: 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__})") @@ -92,11 +83,11 @@ def _probe_gateway(runtime_status: dict[str, Any]) -> dict[str, Any]: def collect_runtime_readiness( *, - model_name: str, + configured_model: str, runtime_status: dict[str, Any] | None, - api_run_queue_depth: int = 0, + active_api_runs: int = 0, process_completion_queue_depth: int = 0, - delegation_completion_queue_depth: int = 0, + active_delegations: int = 0, ) -> dict[str, Any]: """Return bounded readiness diagnostics without mutating runtime state. @@ -109,18 +100,14 @@ def collect_runtime_readiness( checks = { "state_db": _probe_state_db(home), "config": _probe_config(home), - "provider": _check("ok" if str(model_name or "").strip() else "degraded"), + "model": _check("ok" if str(configured_model 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)), + active_api_runs=max(0, int(active_api_runs)), process_completions=max(0, int(process_completion_queue_depth)), - delegation_completions=max(0, int(delegation_completion_queue_depth)), + active_delegations=max(0, int(active_delegations)), ), } overall = "ok" if all(item.get("status") == "ok" for item in checks.values()) else "degraded" diff --git a/tests/gateway/test_api_server.py b/tests/gateway/test_api_server.py index 1ecc065d688c..1d5e9ca91e10 100644 --- a/tests/gateway/test_api_server.py +++ b/tests/gateway/test_api_server.py @@ -738,7 +738,7 @@ class TestHealthDetailedEndpoint: "active_agents": 2, "exit_reason": None, "updated_at": "2026-04-14T00:00:00Z", - }): + }), patch("gateway.run._resolve_gateway_model", return_value="test/model"): async with TestClient(TestServer(app)) as cli: resp = await cli.get("/health/detailed") assert resp.status == 200 @@ -798,7 +798,7 @@ class TestHealthDetailedEndpoint: "status": "degraded", "checks": { "state_db": {"status": "ok"}, - "config": {"status": "degraded", "detail": "using last-known-good"}, + "config": {"status": "degraded", "detail": "invalid config"}, }, } with patch("gateway.status.read_runtime_status", return_value={"gateway_state": "running"}), \ @@ -820,6 +820,21 @@ class TestHealthDetailedEndpoint: assert (await resp.json())["status"] == "ok" probe.assert_not_called() + def test_readiness_work_counts_exclude_retained_completed_runs(self, adapter): + adapter._run_statuses = { + "queued": {"status": "queued"}, + "running": {"status": "running"}, + "approval": {"status": "waiting_for_approval"}, + "done": {"status": "completed"}, + "failed": {"status": "failed"}, + } + # Completed streams may remain attached for replay; they are not work. + adapter._run_streams = {"done": object(), "failed": object()} + + with patch("tools.process_registry.process_registry.completion_queue.qsize", return_value=4), \ + patch("tools.async_delegation.active_count", return_value=2): + assert adapter._readiness_work_counts() == (3, 4, 2) + # --------------------------------------------------------------------------- # /v1/models endpoint diff --git a/tests/gateway/test_readiness.py b/tests/gateway/test_readiness.py index 46c8dad93d37..f379ad41b4ae 100644 --- a/tests/gateway/test_readiness.py +++ b/tests/gateway/test_readiness.py @@ -20,21 +20,21 @@ def test_collect_runtime_readiness_reports_healthy_local_runtime(tmp_path, monke monkeypatch.setenv("HERMES_HOME", str(home)) result = collect_runtime_readiness( - model_name="test/model", + configured_model="test/model", runtime_status={ "gateway_state": "running", "platforms": {"telegram": {"state": "connected"}}, "updated_at": "2026-07-09T00:00:00Z", }, - api_run_queue_depth=2, + active_api_runs=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"]["model"]["status"] == "ok" assert result["checks"]["gateway"]["status"] == "ok" - assert result["checks"]["background_queues"]["api_runs"] == 2 + assert result["checks"]["background_queues"]["active_api_runs"] == 2 assert result["checks"]["disk"]["status"] in {"ok", "degraded"} @@ -47,13 +47,13 @@ def test_collect_runtime_readiness_degrades_on_invalid_config_and_stopped_gatewa monkeypatch.setenv("HERMES_HOME", str(home)) result = collect_runtime_readiness( - model_name="", + configured_model="", 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"]["model"]["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" @@ -66,7 +66,7 @@ def test_collect_runtime_readiness_marks_corrupt_state_db_degraded(tmp_path, mon (home / "state.db").write_bytes(b"not sqlite") monkeypatch.setenv("HERMES_HOME", str(home)) - result = collect_runtime_readiness(model_name="configured-model", runtime_status={}) + result = collect_runtime_readiness(configured_model="configured-model", runtime_status={}) assert result["status"] == "degraded" assert result["checks"]["state_db"]["status"] == "degraded" @@ -83,7 +83,7 @@ def test_collect_runtime_readiness_never_exposes_config_values(tmp_path, monkeyp ) monkeypatch.setenv("HERMES_HOME", str(home)) - result = collect_runtime_readiness(model_name="model", runtime_status={}) + result = collect_runtime_readiness(configured_model="model", runtime_status={}) assert secret not in json.dumps(result) assert str(home) not in json.dumps(result) @@ -96,7 +96,7 @@ def test_collect_runtime_readiness_uses_active_profile_home(tmp_path, monkeypatc (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={}) + result = collect_runtime_readiness(configured_model="model", runtime_status={}) assert result["checks"]["config"]["status"] == "ok" assert not (tmp_path / ".hermes" / "state.db").exists() diff --git a/website/docs/user-guide/features/api-server.md b/website/docs/user-guide/features/api-server.md index 4f1db5ab0c26..3bcb053dbe77 100644 --- a/website/docs/user-guide/features/api-server.md +++ b/website/docs/user-guide/features/api-server.md @@ -227,7 +227,15 @@ Health check. Returns `{"status": "ok"}`. Also available at **GET /v1/health** f ### GET /health/detailed -Extended health check that also reports active sessions, running agents, and resource usage. Useful for monitoring/observability tooling. +Authenticated readiness check for monitoring and control planes. It reports +bounded status for the active profile's config, state database, configured +model, disk space, gateway/platform state, active API runs, pending process +completions, and active delegations. The response exposes status and counts, +not config values, credentials, paths, commands, queue payloads, or raw errors. + +The public `/health` route remains a cheap liveness probe and does not run +readiness checks. A degraded readiness result still uses HTTP 200; inspect the +top-level `status` and `readiness.checks` fields. ## Runs API (streaming-friendly alternative) diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/api-server.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/api-server.md index ec6cf483c51a..7e2b0fa84f7a 100644 --- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/api-server.md +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/api-server.md @@ -227,7 +227,9 @@ OpenAI Responses API 格式。通过 `previous_response_id` 支持服务端对 ### GET /health/detailed -扩展健康检查，同时报告活跃 session、运行中的 agent 和资源使用情况。适用于监控/可观测性工具。 +面向监控和控制平面的已认证就绪检查。它会报告当前 profile 的配置、状态数据库、已配置模型、磁盘空间、gateway/platform 状态、活跃 API run、待处理进程完成通知和活跃 delegation 的有限状态。响应只暴露状态与计数，不包含配置值、凭据、路径、命令、队列载荷或原始错误。 + +公开的 `/health` 路由仍是低开销的存活探针，不运行就绪检查。就绪状态降级时仍返回 HTTP 200；请检查顶层 `status` 和 `readiness.checks` 字段。 ## Runs API（流式友好的替代方案）