feat(gateway): track active_agents in runtime status on turn boundaries
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.
This commit is contained in:
parent
44d552ea5a
commit
51a338a1b6
2 changed files with 71 additions and 0 deletions
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue