From 51a338a1b6ca267f7efc474621d0691488f7e620 Mon Sep 17 00:00:00 2001 From: Ben Date: Sun, 21 Jun 2026 20:17:28 +1000 Subject: [PATCH] feat(gateway): track active_agents in runtime status on turn boundaries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The gateway only rewrote gateway_state.json on lifecycle transitions (start/connect/drain/stop), never on turn start/end. Live-verified on a hosted agent: a confirmed end-to-end turn ran while gateway_updated_at stayed frozen at boot and active_agents was absent — so any active_agents read from the file between transitions is stale. That makes it unusable as a busy/idle signal for an external consumer (NAS deciding whether it's safe to restart/migrate/auto-update an agent mid-turn). Add _persist_active_agents(), called at every turn boundary: - turn start: both running-agent sentinel-claim sites (normal inbound message path + startup-resume path) - turn end: the central _release_running_agent_state() choke point (covers normal completion, /stop, /reset, sentinel cleanup, stale-eviction — every path that ends a running turn) It passes ONLY active_agents to write_runtime_status, leaving gateway_state (and every other field) _UNSET so the read-merge-write preserves the current lifecycle state. Passing gateway_state=None would clobber it — hence a dedicated helper rather than reusing _update_runtime_status. The write is the same cheap JSON write done on lifecycle transitions today; best-effort (a failed status write never disrupts a turn). Behaviour-contract test: an active_agents-only write preserves both running and draining gateway_state, and the count clamps non-negative. --- gateway/run.py | 29 +++++++++++++++++++++++++ tests/gateway/test_status.py | 42 ++++++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+) diff --git a/gateway/run.py b/gateway/run.py index bd991efeb..e5df08d82 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -3665,6 +3665,28 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception: pass + def _persist_active_agents(self) -> None: + """Persist the live in-flight agent count to ``gateway_state.json``. + + Called at every turn boundary (a running-agent slot is claimed or + released) so the dashboard ``/api/status`` readout reflects in-flight + gateway turns in near-real-time. Without this the file is only + rewritten on lifecycle transitions, so any ``active_agents`` read + between transitions is stale (a turn could start and finish without the + file ever moving). + + Deliberately passes ONLY ``active_agents`` — ``gateway_state`` and the + other fields stay ``_UNSET`` so ``write_runtime_status``'s + read-merge-write preserves the current lifecycle state (``running`` / + ``draining`` / …). Passing ``gateway_state=None`` here would clobber it. + Best-effort: a failed status write must never disrupt a turn. + """ + try: + from gateway.status import write_runtime_status + write_runtime_status(active_agents=self._running_agent_count()) + except Exception: + pass + def _update_platform_runtime_status( self, platform: str, @@ -5187,6 +5209,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # instead of spinning up a duplicate AIAgent (#45456). self._running_agents[entry.session_key] = _AGENT_PENDING_SENTINEL self._running_agents_ts[entry.session_key] = time.time() + self._persist_active_agents() # Empty-text internal event — the _is_resume_pending branch in # _handle_message_with_agent prepends the proper reason-aware @@ -8364,6 +8387,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew self._active_session_leases[_quick_key] = _active_session_lease self._running_agents[_quick_key] = _AGENT_PENDING_SENTINEL self._running_agents_ts[_quick_key] = time.time() + self._persist_active_agents() _run_generation = self._begin_session_run_generation(_quick_key) try: @@ -13476,6 +13500,11 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew self._running_agents_ts.pop(session_key, None) if hasattr(self, "_busy_ack_ts"): self._busy_ack_ts.pop(session_key, None) + # Turn boundary: a running-agent slot was just released. Persist the + # new (lower) in-flight count so the dashboard readout stays current + # between lifecycle transitions. Preserves gateway_state (see + # _persist_active_agents). + self._persist_active_agents() return True def _clear_session_boundary_security_state(self, session_key: str) -> None: diff --git a/tests/gateway/test_status.py b/tests/gateway/test_status.py index e8d2f5748..6cfc1dbf7 100644 --- a/tests/gateway/test_status.py +++ b/tests/gateway/test_status.py @@ -1091,3 +1091,45 @@ class TestCorruptStatusFiles: p = tmp_path / "gateway.pid" p.write_text("4242", encoding="utf-8") assert status._read_pid_record(p) == {"pid": 4242} + + +class TestActiveAgentsTurnBoundaryWrite: + """The load-bearing Phase 1a contract: writing the in-flight count at a + turn boundary must PRESERVE the lifecycle gateway_state. The whole readout + depends on active_agents being refreshed per-turn while gateway_state is + only touched by lifecycle transitions — so an active_agents-only write must + not clobber it.""" + + def test_active_agents_only_write_preserves_gateway_state(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + # Lifecycle transition sets running. + status.write_runtime_status(gateway_state="running", active_agents=0) + assert status.read_runtime_status()["gateway_state"] == "running" + + # Turn-boundary write: ONLY active_agents (gateway_state left _UNSET). + status.write_runtime_status(active_agents=2) + + rec = status.read_runtime_status() + assert rec["active_agents"] == 2 + # The state must survive the per-turn write — this is what makes the + # _persist_active_agents helper safe to call on every turn. + assert rec["gateway_state"] == "running" + + def test_active_agents_only_write_preserves_draining_state(self, tmp_path, monkeypatch): + """Same invariant while draining — a turn finishing mid-drain (count + falling) must not flip the state back to running.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + status.write_runtime_status(gateway_state="draining", active_agents=3) + status.write_runtime_status(active_agents=2) + + rec = status.read_runtime_status() + assert rec["active_agents"] == 2 + assert rec["gateway_state"] == "draining" + + def test_active_agents_clamped_non_negative(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + status.write_runtime_status(gateway_state="running", active_agents=-5) + assert status.read_runtime_status()["active_agents"] == 0 +