From 9d81deb0f3b81409bbcb164e4c5e8e17b07a2f36 Mon Sep 17 00:00:00 2001 From: Flownium <157689911+itsflownium@users.noreply.github.com> Date: Thu, 25 Jun 2026 15:45:33 +1000 Subject: [PATCH] fix(gateway): publish API server heartbeat and metrics after bind MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The api_server platform wrote runtime status exactly once at bind (the _connected mark) and never again: last_heartbeat stayed at boot time and metrics_today froze at zero, so the dashboard showed stale API Server activity until a full app restart (#52323). The adapter now keeps daily request/message/token counters and a bounded latency sample, publishes a metrics-bearing snapshot at bind, records metrics after each completed _run_agent turn and /v1/runs run, and a 30-second heartbeat loop re-publishes while connected. gateway.status gains a platform_metrics field on the platform payload, and /health/detailed serves the live adapter metrics alongside the persisted platform map. Salvaged from #52345 by @itsflownium (Flownium) — reworked onto current main (run-worker submission, bind-retry loop, readiness work counts). Fixes #52323 Co-authored-by: Flownium <157689911+itsflownium@users.noreply.github.com> --- gateway/platforms/api_server.py | 149 ++++++++++++++++++++++++--- gateway/platforms/api_server_runs.py | 3 + gateway/status.py | 3 +- tests/gateway/test_api_server.py | 54 +++++++++- tests/gateway/test_status.py | 18 ++++ 5 files changed, 213 insertions(+), 14 deletions(-) diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 763da8b42e..5bd5e91fa2 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -24,6 +24,7 @@ import sys import threading import time import uuid +from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Tuple @@ -231,6 +232,8 @@ MAX_REQUEST_BYTES = 10_000_000 # 10 MB — accommodates long agent conversation # Send a comment before remote API clients' common 20-second idle deadline. # This constant is shared by OpenAI chat/Responses and native session SSE. CHAT_COMPLETIONS_SSE_KEEPALIVE_SECONDS = 10.0 +API_SERVER_HEARTBEAT_SECONDS = 30.0 +API_SERVER_LATENCY_SAMPLE_LIMIT = 512 MAX_NORMALIZED_TEXT_LENGTH = 65_536 # 64 KB cap for normalized content parts MAX_CONTENT_LIST_SIZE = 1_000 # Max items when content is an array RESPONSES_AUTO_TRUNCATION_HISTORY_LIMIT = 100 @@ -1240,7 +1243,7 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter): self.gateway_runner: Optional[Any] = None # set by gateway/run.py # Admitted requests not yet in agent bookkeeping, so shutdown drain counts them. self._pending_agent_requests: int = 0 - # Shared broker; this adapter maps HTTP registration + controller WS onto it. +# Shared broker; this adapter maps HTTP registration + controller WS onto it. self._browser_control_broker = get_browser_control_broker() # One-shot artifact transport: lazy per-profile stores + limiter (tests inject). self._browser_control_artifacts: Dict[str, ArtifactStore] = {} @@ -1248,6 +1251,14 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter): # Per-profile single-flight locks for the off-loop store construction in # _artifact_store_for_async(); a lost race would strand receipts (in-memory index). self._browser_control_artifact_locks: Dict[str, asyncio.Lock] = {} + # Daily API metrics + heartbeat stamps published to gateway runtime status (#52323). + self._metrics_day: str = self._metrics_day_key() + self._metrics_requests_today: int = 0 + self._metrics_messages_today: int = 0 + self._metrics_tokens_today: int = 0 + self._metrics_latency_ms: List[float] = [] + self._metrics_last_request_at: Optional[float] = None + self._metrics_last_heartbeat_at: Optional[float] = None def active_agent_work_count(self) -> int: """All live agent work: pending admissions + in-flight turns + live /v1/runs tasks @@ -1310,10 +1321,7 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter): def _readiness_work_counts(self) -> tuple[int, int, int]: """Return bounded work counts from each subsystem's public state.""" - # "stopping" is not terminal: executor work continues until the agent notices. - active_api_runs = sum( - 1 for status in self._run_statuses.values() - if status.get("status") in {"queued", "running", "waiting_for_approval", "stopping"}) + active_api_runs = self._active_structured_run_count() process_depth = 0 active_delegations = 0 with suppress(Exception): @@ -1324,6 +1332,15 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter): active_delegations = active_count() return active_api_runs, process_depth, active_delegations + def _active_structured_run_count(self) -> int: + """Count structured runs that still have executable work.""" + # "stopping" is not terminal: executor work continues until the agent notices. + return sum( + 1 + for status in self._run_statuses.values() + if status.get("status") in {"queued", "running", "waiting_for_approval", "stopping"} + ) + @staticmethod def _parse_cors_origins(value: Any) -> tuple[str, ...]: """Normalize configured CORS origins into a stable tuple.""" @@ -1349,6 +1366,89 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter): return default return max(0, value) + @staticmethod + def _metrics_day_key(timestamp: Optional[float] = None) -> str: + """Return the UTC day bucket for daily API metrics.""" + return time.strftime("%Y-%m-%d", time.gmtime(timestamp or time.time())) + + @staticmethod + def _iso_timestamp(timestamp: Optional[float]) -> Optional[str]: + if timestamp is None: + return None + return datetime.fromtimestamp(timestamp, timezone.utc).isoformat() + + def _reset_metrics_if_needed(self) -> None: + current_day = self._metrics_day_key() + if self._metrics_day == current_day: + return + self._metrics_day = current_day + self._metrics_requests_today = 0 + self._metrics_messages_today = 0 + self._metrics_tokens_today = 0 + self._metrics_latency_ms.clear() + + @staticmethod + def _total_tokens_from_usage(usage: Optional[Dict[str, Any]]) -> int: + if not isinstance(usage, dict): + return 0 + try: + return max(0, int(usage.get("total_tokens") or 0)) + except (TypeError, ValueError): + return 0 + + @staticmethod + def _latency_p95_ms(samples: List[float]) -> Optional[float]: + if not samples: + return None + ordered = sorted(samples) + index = max(0, min(len(ordered) - 1, int(len(ordered) * 0.95 + 0.999999) - 1)) + return round(ordered[index], 2) + + def _api_server_status_payload(self, *, heartbeat_at: Optional[float] = None) -> Dict[str, Any]: + self._reset_metrics_if_needed() + heartbeat = self._metrics_last_heartbeat_at if heartbeat_at is None else heartbeat_at + return { + "host": self._host, + "port": self._port, + "active_runs": self._active_structured_run_count() + self._inflight_agent_runs, + "stored_runs": len(self._run_statuses), + "last_request_at": self._iso_timestamp(self._metrics_last_request_at), + "last_heartbeat": self._iso_timestamp(heartbeat), + "metrics_today": { + "day": self._metrics_day, + "requests": self._metrics_requests_today, + "messages": self._metrics_messages_today, + "tokens": self._metrics_tokens_today, + "latency_p95_ms": self._latency_p95_ms(self._metrics_latency_ms), + }, + } + + def _publish_runtime_status(self) -> None: + self._metrics_last_heartbeat_at = time.time() + self._write_runtime_status_safe( + "api_server_heartbeat", + platform_state="connected" if self.is_connected else "disconnected", + platform_metrics=self._api_server_status_payload(), + ) + + def _record_api_metrics(self, usage: Optional[Dict[str, Any]], latency_seconds: float) -> None: + self._reset_metrics_if_needed() + self._metrics_requests_today += 1 + self._metrics_messages_today += 1 + self._metrics_tokens_today += self._total_tokens_from_usage(usage) + self._metrics_last_request_at = time.time() + self._metrics_latency_ms.append(max(0.0, latency_seconds * 1000.0)) + if len(self._metrics_latency_ms) > API_SERVER_LATENCY_SAMPLE_LIMIT: + del self._metrics_latency_ms[:-API_SERVER_LATENCY_SAMPLE_LIMIT] + self._publish_runtime_status() + + async def _heartbeat_loop(self) -> None: + while self.is_connected: + self._publish_runtime_status() + if not self.is_connected: + break + await asyncio.sleep(API_SERVER_HEARTBEAT_SECONDS) + @staticmethod def _resolve_model_name(explicit: str) -> str: """Advertised /v1/models name: explicit override > active profile name > "hermes-agent" @@ -2319,6 +2419,18 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter): runtime = read_runtime_status() or {} gw_state = runtime.get("gateway_state") gw_active = parse_active_agents(runtime.get("active_agents", 0)) +# Serve the live adapter's own metrics alongside the persisted platform map: the + # heartbeat loop keeps the file fresh, but a just-booted or wedged writer would + # otherwise show boot-time values here too (#52323). + platforms = runtime.get("platforms", {}) + if not isinstance(platforms, dict): + platforms = {} + platforms = dict(platforms) + api_status = self._api_server_status_payload(heartbeat_at=time.time()) + api_platform = dict(platforms.get("api_server", {})) + api_platform.setdefault("state", "connected" if self.is_connected else "disconnected") + api_platform["metrics"] = api_status + platforms["api_server"] = api_platform # Served BY the gateway process, so gateway_running is True by definition; busy/ # drainable use the same shared contract as /api/status so the two never disagree. active_api_runs, process_depth, active_delegations = self._readiness_work_counts() @@ -2328,9 +2440,13 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter): active_api_runs=active_api_runs, process_completion_queue_depth=process_depth, active_delegations=active_delegations) return web.json_response({ - "status": readiness["status"], "readiness": readiness, "platform": "hermes-agent", +"status": readiness["status"], "readiness": readiness, "platform": "hermes-agent", "version": _hermes_version(), "gateway_state": gw_state, - "platforms": runtime.get("platforms", {}), "active_agents": gw_active, + "platforms": platforms, + "api_server": api_status, + "metrics_today": api_status["metrics_today"], + "last_heartbeat": api_status["last_heartbeat"], + "active_agents": gw_active, "gateway_busy": derive_gateway_busy( gateway_running=True, gateway_state=gw_state, active_agents=gw_active), "gateway_drainable": derive_gateway_drainable( @@ -4166,12 +4282,17 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter): clear_session_vars(tokens) self._activate_admitted_request() self._inflight_agent_runs += 1 + started_at = time.perf_counter() + usage: Optional[Dict[str, Any]] = None try: - # Worker-scoped count rides along so the shutdown close gate still sees the thread +# Worker-scoped count rides along so the shutdown close gate still sees the thread # after this handler task is cancelled (#116535); released in the worker's finally. - return await _api_runs._submit_api_worker(loop, _run) + result, usage = await _api_runs._submit_api_worker(loop, _run) + return result, usage finally: self._inflight_agent_runs -= 1 + if usage is not None: + self._record_api_metrics(usage, time.perf_counter() - started_at) # -- /v1/runs, room grants, room dispatch: thin delegators (real methods: tests assert # __dict__ membership and patch the module-level implementations) --------------------- @@ -4338,7 +4459,7 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter): self._wire_plugin_handlers(self._app) self._runner = web.AppRunner(self._app) await self._runner.setup() - # Bind directly instead of probing 127.0.0.1 first — the old single-family pre-probe raced the +# Bind directly instead of probing 127.0.0.1 first — the old single-family pre-probe raced the # real bind and reported a TIME_WAIT socket as "in use" (#10297), failing gateway restarts for # up to ~60s. Platform-dependent SO_REUSEADDR and the macOS TIME_WAIT rebind live in # start_tcp_site; the loop below covers a predecessor still holding the port for a moment. @@ -4366,8 +4487,8 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter): # A port conflict is a configuration error, not a transient blip — another process # holds the port for its lifetime. A bare ``return False`` makes the reconnect # watcher in gateway.run treat it as retryable and loop forever at the backoff cap - # (observed: 1568+ retries over 5 days across multi-profile setups all defaulting to - # the same port, #52132), filling errors.log and leaking the adapter's ResponseStore + # (observed: 1568+ retries over 5 days across multi-profile setups all defaulting + # to the same port, #52132), filling errors.log and leaking the adapter's ResponseStore # fds each retry. Non-retryable drops it from the reconnect queue; the operator # recovers with ``/platform resume api_server`` after changing the port. "api_server_port_in_use", @@ -4382,6 +4503,10 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter): return False from gateway.platforms.shared_ingress import listener_base_url self._mark_connected(listener_base=listener_base_url(self._host, self._port)) + # Publish a metrics-bearing snapshot at bind and keep it fresh: the + # heartbeat loop updates last_heartbeat/metrics_today (#52323). + self._publish_runtime_status() + self._track_background_task(asyncio.create_task(self._heartbeat_loop())) logger.info( "[%s] API server listening on http://%s:%d (model: %s)", self.name, self._host, self._port, self._model_name) diff --git a/gateway/platforms/api_server_runs.py b/gateway/platforms/api_server_runs.py index da94629f66..8b662cacbd 100644 --- a/gateway/platforms/api_server_runs.py +++ b/gateway/platforms/api_server_runs.py @@ -916,6 +916,7 @@ async def _execute_run(self, run: _RunLaunch, *, _api_server) -> None: """Drive one admitted run, publish its terminal event/status, release live state.""" _redact_api_error_text = _api_server._redact_api_error_text run_id, loop = run.run_id, asyncio.get_running_loop() + _run_started_at = time.perf_counter() def _text_cb(delta: Optional[str]) -> None: if delta is None or run_id not in self._run_streams: @@ -964,6 +965,8 @@ async def _execute_run(self, run: _RunLaunch, *, _api_server) -> None: approval_notify = _make_approval_notify(self, run, _api_server=_api_server) result, usage, served_runtime = await _submit_api_worker( loop, lambda: _run_agent_sync(self, run, agent, approval_notify, _api_server=_api_server)) + # Publish request metrics (daily counters + latency) with each completed run (#52323). + self._record_api_metrics(usage, time.perf_counter() - _run_started_at) if not isinstance(result, dict): result = {} status, fields = terminal_run_status(result) diff --git a/gateway/status.py b/gateway/status.py index 60c4529345..a8e14bd995 100644 --- a/gateway/status.py +++ b/gateway/status.py @@ -1190,6 +1190,7 @@ def _prepare_runtime_status_update( error_code: Any = _UNSET, error_message: Any = _UNSET, needs_attention: Any = _UNSET, retrying_since: Any = _UNSET, served_profiles: Any = _UNSET, session_store: Any = _UNSET, multiplex_standalone_reason: Any = _UNSET, + platform_metrics: Any = _UNSET, ingress_url: Any = _UNSET, listener_base: Any = _UNSET, clear_profile_platforms: bool = False, drop_profile_platforms: Optional[str] = None, load_existing: bool = True, reload_existing: bool = False, @@ -1241,6 +1242,7 @@ def _prepare_runtime_status_update( ("error_message", error_message, None), ("needs_attention", needs_attention, bool), ("retrying_since", retrying_since, None), + ("metrics", platform_metrics, None), ("ingress_url", ingress_url, None), ("listener_base", listener_base, None), )) @@ -1259,7 +1261,6 @@ def _emit_runtime_status_transition( from agent.monitoring.gateway_health import emit_runtime_status_transition emit_runtime_status_transition(previous_payload, payload) - def write_runtime_status( *, reload_existing: bool = False, wait_timeout: Optional[float] = None, **fields: Any, ) -> bool: diff --git a/tests/gateway/test_api_server.py b/tests/gateway/test_api_server.py index 9031c4c215..2ba16401ac 100644 --- a/tests/gateway/test_api_server.py +++ b/tests/gateway/test_api_server.py @@ -329,6 +329,52 @@ class TestConcurrencyCap: assert mock_run.await_count == 0, "the turn must not start once the cap is reached" +class TestRuntimeStatusMetrics: + def test_completed_buffered_runs_are_not_reported_active(self, adapter): + adapter._run_statuses = { + "done": {"status": "completed"}, + "failed": {"status": "failed"}, + } + adapter._run_streams = {"done": object(), "failed": object()} + adapter._inflight_agent_runs = 0 + + assert adapter._api_server_status_payload()["active_runs"] == 0 + + def test_record_api_metrics_publishes_status(self, adapter): + adapter._running = True + + with patch.object(adapter, "_write_runtime_status_safe") as mock_write: + adapter._record_api_metrics({"total_tokens": 17}, 0.125) + + assert adapter._metrics_requests_today == 1 + assert adapter._metrics_messages_today == 1 + assert adapter._metrics_tokens_today == 17 + mock_write.assert_called_once() + _, kwargs = mock_write.call_args + assert kwargs["platform_state"] == "connected" + metrics = kwargs["platform_metrics"] + assert metrics["metrics_today"]["requests"] == 1 + assert metrics["metrics_today"]["messages"] == 1 + assert metrics["metrics_today"]["tokens"] == 17 + assert metrics["metrics_today"]["latency_p95_ms"] == 125.0 + assert metrics["last_request_at"] is not None + assert metrics["last_heartbeat"] is not None + + @pytest.mark.asyncio + async def test_heartbeat_loop_publishes_once_when_stopped(self, adapter): + adapter._running = True + calls = [] + + def publish_once(): + calls.append(True) + adapter._running = False + + with patch.object(adapter, "_publish_runtime_status", side_effect=publish_once): + await adapter._heartbeat_loop() + + assert calls == [True] + + # --------------------------------------------------------------------------- # Helpers for HTTP tests # --------------------------------------------------------------------------- @@ -780,6 +826,9 @@ class TestHealthDetailedEndpoint: @pytest.mark.asyncio async def test_health_detailed_returns_ok(self, adapter): """GET /health/detailed returns status, platform, and runtime fields.""" + adapter._running = True + with patch.object(adapter, "_write_runtime_status_safe"): + adapter._record_api_metrics({"total_tokens": 9}, 0.02) app = _create_app(adapter) with patch("gateway.status.read_runtime_status", return_value={ "gateway_state": "running", @@ -798,7 +847,10 @@ class TestHealthDetailedEndpoint: assert data["status"] == "ok" assert data["platform"] == "hermes-agent" assert data["gateway_state"] == "running" - assert data["platforms"] == {"telegram": {"state": "connected"}} + assert data["platforms"]["telegram"] == {"state": "connected"} + assert data["platforms"]["api_server"]["metrics"]["metrics_today"]["requests"] == 1 + assert data["metrics_today"]["tokens"] == 9 + assert data["last_heartbeat"] is not None assert data["active_agents"] == 2 # Derived busy/drainable: this endpoint is served BY the live # gateway, so running + 2 agents ⇒ busy and drainable. diff --git a/tests/gateway/test_status.py b/tests/gateway/test_status.py index f904e67ded..e74c078901 100644 --- a/tests/gateway/test_status.py +++ b/tests/gateway/test_status.py @@ -559,6 +559,24 @@ class TestRuntimeStatusBackgroundWriter: finally: release_write.set() assert writer.flush(timeout=2.0) + def test_write_runtime_status_records_platform_metrics(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + status.write_runtime_status( + platform="api_server", + platform_state="connected", + platform_metrics={ + "last_heartbeat": "2026-06-25T00:00:00+00:00", + "metrics_today": {"requests": 3, "tokens": 42}, + }, + ) + + payload = status.read_runtime_status() + api_status = payload["platforms"]["api_server"] + assert api_status["state"] == "connected" + assert api_status["metrics"]["last_heartbeat"] == "2026-06-25T00:00:00+00:00" + assert api_status["metrics"]["metrics_today"]["requests"] == 3 + assert api_status["metrics"]["metrics_today"]["tokens"] == 42 class TestGetProcessStartTime: