mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-15 14:22:43 +00:00
fix(gateway): ground readiness in live runtime state
This commit is contained in:
teknium1
Teknium
parent
f9728af5e2
commit
6142203bd7
6 changed files with 67 additions and 52 deletions
|
|
@ -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"],
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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（流式友好的替代方案）
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue