fix(gateway): publish API server heartbeat and metrics after bind

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

@@ -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)

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)

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:

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.

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: