From 24221636b27c9570d731d22b9f651c8f3bc1296d Mon Sep 17 00:00:00 2001 From: jenkins Date: Wed, 12 Aug 2026 23:48:06 -0300 Subject: [PATCH] hermes: correct provider status and session lineage --- services/hermes/agent-deployment.yaml | 9 ++ .../auto-router/dashboard/dist/index.js | 8 +- .../plugins/auto-router/provider_status.py | 26 +++-- services/hermes/scripts/codex_broker.py | 107 +++++++++++++++++- .../scripts/migrate_api_session_lineage.py | 26 ++++- testing/tests/test_hermes_chat_quality.py | 84 +++++++++++++- 6 files changed, 240 insertions(+), 20 deletions(-) diff --git a/services/hermes/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index a12147d4d..0209f030f 100644 --- a/services/hermes/agent-deployment.yaml +++ b/services/hermes/agent-deployment.yaml @@ -459,6 +459,9 @@ spec: - {name: HERMES_MEDIA_DELIVERY_STRICT, value: "1"} - {name: HERMES_MEDIA_ALLOW_DIRS, value: /opt/data/workspace} - {name: HERMES_AUTO_ROUTER_PROFILE, value: agent} + - {name: HERMES_ROUTING_CATALOG_PATH, value: /routing-catalog/catalog.json} + - {name: HERMES_CODEX_HEALTH_PATH, value: /opt/data/provider-health/codex.json} + - {name: HERMES_CLAUDE_HEALTH_PATH, value: /opt/data/provider-health/claude.json} volumeMounts: - {name: home, mountPath: /opt/data} - {name: provider-auth, mountPath: /shared-auth} @@ -474,6 +477,7 @@ spec: - {name: api-server-patch, mountPath: /opt/hermes/gateway/platforms/api_server.py, subPath: api_server.py} - {name: kubeconfig, mountPath: /opt/data/home/.kube/config, subPath: config, readOnly: true} - {name: auto-router-plugin, mountPath: /opt/data/plugins/auto-router, readOnly: true} + - {name: routing-catalog, mountPath: /routing-catalog, readOnly: true} - {name: tmp, mountPath: /tmp} startupProbe: exec: @@ -615,6 +619,9 @@ spec: - {name: AGENT_BROWSER_ARGS, value: "--no-sandbox,--disable-dev-shm-usage"} - {name: HERMES_TUI_AGENT_INIT_TIMEOUT_S, value: "180"} - {name: HERMES_AUTO_ROUTER_PROFILE, value: agent} + - {name: HERMES_ROUTING_CATALOG_PATH, value: /routing-catalog/catalog.json} + - {name: HERMES_CODEX_HEALTH_PATH, value: /opt/data/provider-health/codex.json} + - {name: HERMES_CLAUDE_HEALTH_PATH, value: /opt/data/provider-health/claude.json} volumeMounts: - {name: home, mountPath: /opt/data} - {name: provider-auth, mountPath: /shared-auth} @@ -629,6 +636,7 @@ spec: - {name: tui-gateway-patch, mountPath: /opt/hermes/tui_gateway/server.py, subPath: server.py} - {name: kubeconfig, mountPath: /opt/data/home/.kube/config, subPath: config, readOnly: true} - {name: auto-router-plugin, mountPath: /opt/data/plugins/auto-router, readOnly: true} + - {name: routing-catalog, mountPath: /routing-catalog, readOnly: true} - {name: tmp, mountPath: /tmp} - {name: ttyd-index, mountPath: /ttyd-index, readOnly: true} startupProbe: @@ -798,6 +806,7 @@ spec: - {name: CODEX_HOME, value: /opt/data/home/.codex} - {name: PYTHONPATH, value: /opt/hermes} - {name: HERMES_CODEX_BROKER_LISTEN_PORT, value: "9003"} + - {name: HERMES_CODEX_HEALTH_PATH, value: /opt/data/provider-health/codex.json} - {name: HERMES_ROUTING_CATALOG_PATH, value: /routing-catalog/catalog.json} readinessProbe: tcpSocket: {port: codex-broker} diff --git a/services/hermes/plugins/auto-router/dashboard/dist/index.js b/services/hermes/plugins/auto-router/dashboard/dist/index.js index 9cf509c85..29fa4e591 100644 --- a/services/hermes/plugins/auto-router/dashboard/dist/index.js +++ b/services/hermes/plugins/auto-router/dashboard/dist/index.js @@ -48,7 +48,13 @@ h("span", { className: account.authenticated ? "is-good" : "is-bad" }, authLabel), account.rate_limit_tier && account.rate_limit_tier !== "unknown" ? h("span", null, "Tier: " + account.rate_limit_tier) : null, h("span", null, "Access token: " + when(account.token_expires_at)), - nativeHealth && nativeHealth.transport ? h("span", null, "Transport: native Claude Code subscription") : null, + nativeHealth && nativeHealth.transport ? h("span", null, + nativeHealth.transport === "claude-code-cli-subscription" + ? "Transport: native Claude Code subscription" + : nativeHealth.transport === "codex-chatgpt-subscription" + ? "Transport: native ChatGPT Codex subscription" + : "Transport: " + nativeHealth.transport + ) : null, nativeHealth && nativeHealth.rate_limit && nativeHealth.rate_limit.utilization != null ? h("span", null, "Observed utilization: " + Math.round(Number(nativeHealth.rate_limit.utilization) * 100) + "%") : null, diff --git a/services/hermes/plugins/auto-router/provider_status.py b/services/hermes/plugins/auto-router/provider_status.py index cb4dff0fc..c37949c73 100644 --- a/services/hermes/plugins/auto-router/provider_status.py +++ b/services/hermes/plugins/auto-router/provider_status.py @@ -33,6 +33,9 @@ CLAUDE_HEALTH_PATH = Path( "HERMES_CLAUDE_HEALTH_PATH", "/opt/data/provider-health/claude.json" ) ) +CODEX_HEALTH_PATH = Path( + os.environ.get("HERMES_CODEX_HEALTH_PATH", "/opt/data/provider-health/codex.json") +) def _read_json(path: Path) -> dict[str, Any]: @@ -251,6 +254,18 @@ def _fresh_health(path: Path, maximum_age: float = 86400.0) -> dict[str, Any]: return value if age <= maximum_age else {} +def _apply_native_health(provider: dict[str, Any], health: dict[str, Any]) -> None: + """Prefer current native transport health over lifetime router counters.""" + provider["native_health"] = health + native_state = health.get("state") + if native_state == "available": + provider["state"] = "available" + elif native_state in {"capacity-limited", "degraded"}: + provider["state"] = "degraded" + elif native_state == "unavailable": + provider["state"] = "unavailable" + + def provider_status_payload() -> dict[str, Any]: """Build the owner-safe status document shared by dashboard and TUI.""" health = _get_json(f"{SWITCHYARD_ROOT}/health") @@ -270,15 +285,8 @@ def provider_status_payload() -> dict[str, Any]: ) providers["codex"]["account"] = _codex_account() providers["claude"]["account"] = _claude_account() - claude_health = _fresh_health(CLAUDE_HEALTH_PATH) - providers["claude"]["native_health"] = claude_health - native_state = claude_health.get("state") - if native_state == "available": - providers["claude"]["state"] = "available" - elif native_state == "capacity-limited": - providers["claude"]["state"] = "degraded" - elif native_state == "unavailable": - providers["claude"]["state"] = "unavailable" + _apply_native_health(providers["codex"], _fresh_health(CODEX_HEALTH_PATH)) + _apply_native_health(providers["claude"], _fresh_health(CLAUDE_HEALTH_PATH)) classifier = stats.get("classifier") classifier = classifier if isinstance(classifier, dict) else {} fallbacks = stats.get("routing_fallbacks") diff --git a/services/hermes/scripts/codex_broker.py b/services/hermes/scripts/codex_broker.py index f25e99bf7..3d803c1a3 100644 --- a/services/hermes/scripts/codex_broker.py +++ b/services/hermes/scripts/codex_broker.py @@ -11,6 +11,7 @@ import json import os import tempfile import time +from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any, Iterable @@ -44,6 +45,54 @@ ROUTED_MODEL_PREFIX = "route/codex/" TOKEN_REFRESH_SKEW_SECONDS = int( os.environ.get("HERMES_CODEX_BROKER_REFRESH_SKEW_SECONDS", "300") ) +HEALTH_PATH = Path( + os.environ.get("HERMES_CODEX_HEALTH_PATH", "/opt/data/provider-health/codex.json") +) + + +def _previous_health() -> dict[str, Any]: + """Read the previous non-secret native transport health snapshot.""" + try: + value = json.loads(HEALTH_PATH.read_text(encoding="utf-8")) + except (OSError, TypeError, ValueError, json.JSONDecodeError): + return {} + return value if isinstance(value, dict) else {} + + +def _record_health(state: str, **updates: Any) -> None: + """Atomically publish current first-party Codex health without credentials.""" + value = _previous_health() + value.update( + { + "transport": "codex-chatgpt-subscription", + "state": state, + "checked_at": datetime.now(timezone.utc).isoformat(), + "authenticated": state != "unavailable", + **updates, + } + ) + try: + HEALTH_PATH.parent.mkdir(parents=True, exist_ok=True) + temporary = HEALTH_PATH.with_name(f".{HEALTH_PATH.name}.{os.getpid()}.tmp") + temporary.write_text( + json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + os.replace(temporary, HEALTH_PATH) + except OSError: + pass + + +def _requested_effort(payload: dict[str, Any]) -> str: + """Return the bounded Codex reasoning effort carried by a routed request.""" + reasoning = payload.get("reasoning") + if isinstance(reasoning, dict) and reasoning.get("effort") in { + "low", + "medium", + "high", + "xhigh", + }: + return str(reasoning["effort"]) + return "medium" def _real_model(model: str) -> str: @@ -409,10 +458,21 @@ class Handler(BaseHTTPRequestHandler): return if self.path == "/health": try: - _access_token() + token = _access_token() except RuntimeError as exc: + _record_health( + "unavailable", + authenticated=False, + last_error_at=datetime.now(timezone.utc).isoformat(), + error_type=type(exc).__name__, + ) self._json(503, {"ok": False, "error": str(exc)}) return + _record_health( + "available", + authenticated=True, + token_expires_at=_token_expiry(token) or None, + ) self._json(200, {"ok": True, "provider": "openai-codex"}) return if self.path in {"/models", "/v1/models"}: @@ -448,6 +508,9 @@ class Handler(BaseHTTPRequestHandler): payload = json.loads(self.rfile.read(length)) requested_stream = payload.get("stream") is True payload = _validate_payload(payload) + model = str(payload["model"]) + effort = _requested_effort(payload) + started = time.monotonic() token = _access_token() timeout = httpx.Timeout( READ_TIMEOUT_SECONDS, @@ -460,6 +523,20 @@ class Handler(BaseHTTPRequestHandler): with client.stream("POST", f"{UPSTREAM}/responses", json=payload) as response: if response.status_code >= 400: body = response.read() + state = ( + "capacity-limited" + if response.status_code == 429 + else "unavailable" + if response.status_code in {401, 403} + else "degraded" + ) + _record_health( + state, + last_error_at=datetime.now(timezone.utc).isoformat(), + status_code=response.status_code, + model=model, + effort=effort, + ) self.send_response(response.status_code) self.send_header("Content-Type", response.headers.get("Content-Type", "application/json")) self.send_header("Content-Length", str(len(body))) @@ -483,6 +560,14 @@ class Handler(BaseHTTPRequestHandler): completed = _completed_response( body.decode("utf-8", errors="replace").splitlines() ) + _record_health( + "available", + last_success_at=datetime.now(timezone.utc).isoformat(), + latency_ms=int((time.monotonic() - started) * 1000), + model=model, + effort=effort, + token_expires_at=_token_expiry(token) or None, + ) if not requested_stream: self._json(200, completed) return @@ -505,6 +590,11 @@ class Handler(BaseHTTPRequestHandler): except Exception as exc: if response_started: return + _record_health( + "degraded", + last_error_at=datetime.now(timezone.utc).isoformat(), + error_type=type(exc).__name__, + ) self._json( 502, { @@ -524,6 +614,21 @@ def main() -> None: """Serve until Kubernetes terminates the sidecar.""" if not TOKEN: raise SystemExit("HERMES_IMAGE_BROKER_KEY is required") + try: + token = _access_token() + except RuntimeError as exc: + _record_health( + "unavailable", + authenticated=False, + last_error_at=datetime.now(timezone.utc).isoformat(), + error_type=type(exc).__name__, + ) + else: + _record_health( + "available", + authenticated=True, + token_expires_at=_token_expiry(token) or None, + ) ThreadingHTTPServer((HOST, PORT), Handler).serve_forever() diff --git a/services/hermes/scripts/migrate_api_session_lineage.py b/services/hermes/scripts/migrate_api_session_lineage.py index c21ed0420..34c7a220c 100644 --- a/services/hermes/scripts/migrate_api_session_lineage.py +++ b/services/hermes/scripts/migrate_api_session_lineage.py @@ -19,6 +19,9 @@ LEGACY_CASSANDRA_WORKERS = { "api-c288b9f3024a91b8": "Cassandra stale-worktree check", "api-058e51802b58ee3f": "Cassandra verification worker", } +LEGACY_ORPHANED_SMOKE_SESSIONS = { + "e17f2d888689": "Archived agent workspace smoke test", +} def migrate(path: Path = STATE_DB) -> int: @@ -30,19 +33,32 @@ def migrate(path: Path = STATE_DB) -> int: parent = connection.execute( "SELECT id FROM sessions WHERE id = ?", (LEGACY_CASSANDRA_PARENT,) ).fetchone() - if not parent: - return 0 - for session_id, title in LEGACY_CASSANDRA_WORKERS.items(): + if parent: + for session_id, title in LEGACY_CASSANDRA_WORKERS.items(): + cursor = connection.execute( + """ + UPDATE sessions + SET parent_session_id = ?, + title = ? + WHERE id = ? + AND source = 'api_server' + AND parent_session_id IS NULL + """, + (LEGACY_CASSANDRA_PARENT, title, session_id), + ) + changed += cursor.rowcount + for session_id, title in LEGACY_ORPHANED_SMOKE_SESSIONS.items(): cursor = connection.execute( """ UPDATE sessions - SET parent_session_id = ?, + SET archived = 1, title = ? WHERE id = ? AND source = 'api_server' AND parent_session_id IS NULL + AND archived = 0 """, - (LEGACY_CASSANDRA_PARENT, title, session_id), + (title, session_id), ) changed += cursor.rowcount return changed diff --git a/testing/tests/test_hermes_chat_quality.py b/testing/tests/test_hermes_chat_quality.py index 8e26010af..055417e4e 100644 --- a/testing/tests/test_hermes_chat_quality.py +++ b/testing/tests/test_hermes_chat_quality.py @@ -927,6 +927,52 @@ def test_claude_broker_uses_native_subscription_without_api_billing(monkeypatch) assert module.CAPACITY_PATTERN.search("weekly usage limit exhausted") +def test_codex_native_health_overrides_historical_router_errors( + tmp_path: Path, monkeypatch +): + """Fresh first-party health is authoritative over old Switchyard probes.""" + plugin_path = HERMES / "plugins" / "auto-router" / "provider_status.py" + spec = importlib.util.spec_from_file_location("hermes_provider_status", plugin_path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + health_path = tmp_path / "codex.json" + health_path.write_text( + json.dumps( + { + "state": "available", + "authenticated": True, + "transport": "codex-chatgpt-subscription", + } + ) + ) + monkeypatch.setattr(module, "CODEX_HEALTH_PATH", health_path) + monkeypatch.setattr(module, "CLAUDE_HEALTH_PATH", tmp_path / "missing.json") + monkeypatch.setattr( + module, + "_get_json", + lambda url: {"status": "ok"} + if url.endswith("/health") + else { + "models": { + "route/codex/terra/medium": { + "calls": 1, + "errors": 99, + "total_tokens": 12, + } + } + }, + ) + monkeypatch.setattr(module, "_codex_account", lambda: {}) + monkeypatch.setattr(module, "_claude_account", lambda: {}) + + codex = module.provider_status_payload()["providers"]["codex"] + + assert codex["errors"] == 99 + assert codex["state"] == "available" + assert codex["native_health"]["transport"] == "codex-chatgpt-subscription" + + def test_api_session_patch_accepts_parent_lineage(tmp_path: Path): """API-created workers must persist the originating Hermes session.""" module_path = HERMES / "scripts" / "patch_api_server_sessions.py" @@ -957,19 +1003,27 @@ def test_legacy_api_sessions_are_nested_idempotently(tmp_path: Path): with sqlite3.connect(database) as connection: connection.execute( "CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT, " - "parent_session_id TEXT, title TEXT, transcript TEXT)" + "parent_session_id TEXT, title TEXT, transcript TEXT, archived INTEGER DEFAULT 0)" ) connection.execute( - "INSERT INTO sessions VALUES (?, 'tui', NULL, 'Cassandra', 'parent-data')", + "INSERT INTO sessions (id, source, parent_session_id, title, transcript) " + "VALUES (?, 'tui', NULL, 'Cassandra', 'parent-data')", (module.LEGACY_CASSANDRA_PARENT,), ) worker_id = next(iter(module.LEGACY_CASSANDRA_WORKERS)) connection.execute( - "INSERT INTO sessions VALUES (?, 'api_server', NULL, 'old', 'keep-me')", + "INSERT INTO sessions (id, source, parent_session_id, title, transcript) " + "VALUES (?, 'api_server', NULL, 'old', 'keep-me')", (worker_id,), ) + orphan_id = next(iter(module.LEGACY_ORPHANED_SMOKE_SESSIONS)) + connection.execute( + "INSERT INTO sessions (id, source, parent_session_id, title, transcript) " + "VALUES (?, 'api_server', NULL, 'old smoke', 'keep-smoke')", + (orphan_id,), + ) - assert module.migrate(database) == 1 + assert module.migrate(database) == 2 assert module.migrate(database) == 0 with sqlite3.connect(database) as connection: row = connection.execute( @@ -981,6 +1035,16 @@ def test_legacy_api_sessions_are_nested_idempotently(tmp_path: Path): module.LEGACY_CASSANDRA_WORKERS[worker_id], "keep-me", ) + with sqlite3.connect(database) as connection: + orphan = connection.execute( + "SELECT archived, title, transcript FROM sessions WHERE id = ?", + (orphan_id,), + ).fetchone() + assert orphan == ( + 1, + module.LEGACY_ORPHANED_SMOKE_SESSIONS[orphan_id], + "keep-smoke", + ) def test_switchyard_brokers_and_native_claude_lane_use_the_right_images(): @@ -1008,6 +1072,18 @@ def test_switchyard_brokers_and_native_claude_lane_use_the_right_images(): container["name"]: container for container in agent["spec"]["template"]["spec"]["containers"] } + for container_name in ("hermes", "terminal"): + container = agent_containers[container_name] + environment = {item["name"]: item["value"] for item in container["env"]} + mounts = {item["name"]: item for item in container["volumeMounts"]} + assert environment["HERMES_ROUTING_CATALOG_PATH"] == "/routing-catalog/catalog.json" + assert environment["HERMES_CODEX_HEALTH_PATH"] == "/opt/data/provider-health/codex.json" + assert environment["HERMES_CLAUDE_HEALTH_PATH"] == "/opt/data/provider-health/claude.json" + assert mounts["routing-catalog"]["mountPath"] == "/routing-catalog" + assert mounts["routing-catalog"]["readOnly"] is True + codex = agent_containers["codex-broker"] + codex_environment = {item["name"]: item["value"] for item in codex["env"]} + assert codex_environment["HERMES_CODEX_HEALTH_PATH"] == "/opt/data/provider-health/codex.json" claude = agent_containers["claude-broker"] assert claude["image"].startswith("registry.bstein.dev/bstein/hermes-agent@") assert "unset ANTHROPIC_API_KEY CLAUDE_API_KEY" in claude["args"][0]