Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 21 additions & 3 deletions gateway/platforms/api_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -1103,16 +1103,34 @@ async def _handle_health_detailed(self, request: "web.Request") -> "web.Response
dashboard can display full status without needing a shared PID file or
/proc access. No authentication required.
"""
from gateway.status import read_runtime_status
from gateway.status import (
derive_gateway_busy,
derive_gateway_drainable,
read_runtime_status,
)

runtime = read_runtime_status() or {}
gw_state = runtime.get("gateway_state")
gw_active = runtime.get("active_agents", 0)
# This endpoint is served BY the gateway process, so it is by definition
# alive — gateway_running is True. Derive busy/drainable from the same
# shared contract /api/status uses so the two surfaces never disagree.
return web.json_response({
"status": "ok",
"platform": "hermes-agent",
"version": _hermes_version(),
"gateway_state": runtime.get("gateway_state"),
"gateway_state": gw_state,
"platforms": runtime.get("platforms", {}),
"active_agents": runtime.get("active_agents", 0),
"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(
gateway_running=True,
gateway_state=gw_state,
),
"exit_reason": runtime.get("exit_reason"),
"updated_at": runtime.get("updated_at"),
"pid": os.getpid(),
Expand Down
29 changes: 29 additions & 0 deletions gateway/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -3665,6 +3665,28 @@ def _update_runtime_status(self, gateway_state: Optional[str] = None, exit_reaso
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,
Expand Down Expand Up @@ -5187,6 +5209,7 @@ def _schedule_resume_pending_sessions(self, platform=None) -> int:
# 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
Expand Down Expand Up @@ -8364,6 +8387,7 @@ async def _do_undo():
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:
Expand Down Expand Up @@ -13476,6 +13500,11 @@ def _release_running_agent_state(
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:
Expand Down
43 changes: 43 additions & 0 deletions gateway/status.py
Original file line number Diff line number Diff line change
Expand Up @@ -621,6 +621,49 @@ def read_runtime_status() -> Optional[dict[str, Any]]:
return _read_json_file(_get_runtime_status_path())


# States in which the gateway is alive and could be asked to drain. Anything
# else (draining already, stopping, stopped, startup_failed, None) is NOT a
# valid begin-drain target.
_DRAINABLE_GATEWAY_STATES = frozenset({"running"})


def derive_gateway_busy(
*, gateway_running: bool, gateway_state: Any, active_agents: Any
) -> bool:
"""Whether the gateway is actively processing in-flight turns.

The contract NAS gates lifecycle actions on. Busy iff the gateway is live
(``gateway_running``), in the ``running`` state, AND at least one agent is
mid-turn (``active_agents > 0``). Degrades to ``False`` whenever liveness
is unknown, the state is anything but ``running``, or the count is
absent/unparseable — i.e. a down or file-absent gateway reads "not busy",
never a spurious "busy".

NOTE: liveness keys off ``gateway_running`` (a live PID / health probe),
NEVER ``updated_at`` — a healthy idle gateway never advances that timestamp.
"""
if not gateway_running:
return False
if gateway_state not in _DRAINABLE_GATEWAY_STATES:
return False
try:
return int(active_agents) > 0
except (TypeError, ValueError):
return False


def derive_gateway_drainable(*, gateway_running: bool, gateway_state: Any) -> bool:
"""Whether the gateway can accept a begin-drain request right now.

True iff the gateway is live and in the ``running`` state — i.e. not already
draining/stopping/stopped and not in a failed-start state. This is
independent of ``active_agents``: an idle running gateway is drainable (the
drain just completes immediately). Degrades to ``False`` for a down or
non-running gateway.
"""
return bool(gateway_running) and gateway_state in _DRAINABLE_GATEWAY_STATES


def get_runtime_status_running_pid(
runtime: Optional[dict[str, Any]] = None,
) -> Optional[int]:
Expand Down
42 changes: 42 additions & 0 deletions hermes_cli/web_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,8 @@
get_memory_provider,
)
from gateway.status import (
derive_gateway_busy,
derive_gateway_drainable,
get_running_pid,
get_runtime_status_running_pid,
read_runtime_status,
Expand Down Expand Up @@ -1835,6 +1837,42 @@ async def get_status(profile: Optional[str] = None):
except Exception:
pass

# Busy/drainable readout (NAS lifecycle-safety gate). active_agents is
# the in-flight gateway-turn count the gateway now persists at every
# turn boundary; gateway_busy/gateway_drainable are derived from it +
# liveness via the single shared contract in gateway.status. Liveness
# keys off gateway_running (a live PID/health probe), NEVER
# gateway_updated_at — a healthy idle gateway never advances that.
active_agents = 0
if runtime:
try:
active_agents = max(0, int(runtime.get("active_agents", 0) or 0))
except (TypeError, ValueError):
active_agents = 0
gateway_busy = derive_gateway_busy(
gateway_running=gateway_running,
gateway_state=gateway_state,
active_agents=active_agents,
)
gateway_drainable = derive_gateway_drainable(
gateway_running=gateway_running,
gateway_state=gateway_state,
)
# Resolved drain timeout (seconds) so NAS can size its poll deadline
# without out-of-band knowledge. Mirrors gateway/restart.py precedence:
# HERMES_RESTART_DRAIN_TIMEOUT env override → config agent.* → default.
from gateway.restart import parse_restart_drain_timeout

_drain_timeout_raw = os.environ.get("HERMES_RESTART_DRAIN_TIMEOUT")
if _drain_timeout_raw is None:
try:
_drain_timeout_raw = cfg_get(
load_config(), "agent", "restart_drain_timeout", default=None
)
except Exception:
_drain_timeout_raw = None
restart_drain_timeout = parse_restart_drain_timeout(_drain_timeout_raw)

# Dashboard auth gate (Phase 7): surface whether the gate is engaged
# and which providers are registered so ``hermes status`` and the
# SPA's StatusPage can show "OAuth gate ON via Nous Research" or
Expand Down Expand Up @@ -1863,6 +1901,10 @@ async def get_status(profile: Optional[str] = None):
"gateway_platforms": gateway_platforms,
"gateway_exit_reason": gateway_exit_reason,
"gateway_updated_at": gateway_updated_at,
"active_agents": active_agents,
"gateway_busy": gateway_busy,
"gateway_drainable": gateway_drainable,
"restart_drain_timeout": restart_drain_timeout,
"active_sessions": active_sessions,
"auth_required": auth_required,
"auth_providers": auth_providers,
Expand Down
7 changes: 7 additions & 0 deletions tests/gateway/test_api_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -584,6 +584,10 @@ async def test_health_detailed_returns_ok(self, adapter):
assert data["gateway_state"] == "running"
assert data["platforms"] == {"telegram": {"state": "connected"}}
assert data["active_agents"] == 2
# Derived busy/drainable: this endpoint is served BY the live
# gateway, so running + 2 agents ⇒ busy and drainable.
assert data["gateway_busy"] is True
assert data["gateway_drainable"] is True
assert isinstance(data["pid"], int)
assert "updated_at" in data

Expand All @@ -599,6 +603,9 @@ async def test_health_detailed_no_runtime_status(self, adapter):
assert data["status"] == "ok"
assert data["gateway_state"] is None
assert data["platforms"] == {}
# No runtime file ⇒ state None ⇒ not busy, not drainable.
assert data["gateway_busy"] is False
assert data["gateway_drainable"] is False

@pytest.mark.asyncio
async def test_health_detailed_does_not_require_auth(self, auth_adapter):
Expand Down
88 changes: 88 additions & 0 deletions tests/gateway/test_status.py
Original file line number Diff line number Diff line change
Expand Up @@ -1091,3 +1091,91 @@ def test_read_pid_record_still_parses_bare_pid(self, tmp_path):
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
class TestGatewayBusyDerivation:
"""Pure contract for derive_gateway_busy / derive_gateway_drainable — the
single shared definition both /api/status and /health/detailed consume."""

def test_busy_requires_running_state_and_positive_count(self):
assert status.derive_gateway_busy(
gateway_running=True, gateway_state="running", active_agents=1
) is True
assert status.derive_gateway_busy(
gateway_running=True, gateway_state="running", active_agents=0
) is False

def test_busy_false_when_not_live_even_if_file_says_active(self):
# Liveness wins: gateway_running False ⇒ never busy, regardless of count.
assert status.derive_gateway_busy(
gateway_running=False, gateway_state="running", active_agents=9
) is False

def test_busy_false_for_non_running_states(self):
for state in ("draining", "stopping", "stopped", "startup_failed", None):
assert status.derive_gateway_busy(
gateway_running=True, gateway_state=state, active_agents=5
) is False, state

def test_busy_degrades_on_unparseable_count(self):
for bad in (None, "garbage", object()):
assert status.derive_gateway_busy(
gateway_running=True, gateway_state="running", active_agents=bad
) is False

def test_drainable_is_running_and_live_independent_of_count(self):
# Idle running gateway is drainable but NOT busy.
assert status.derive_gateway_drainable(
gateway_running=True, gateway_state="running"
) is True
assert status.derive_gateway_busy(
gateway_running=True, gateway_state="running", active_agents=0
) is False

def test_drainable_false_when_down_or_not_running(self):
assert status.derive_gateway_drainable(
gateway_running=False, gateway_state="running"
) is False
for state in ("draining", "stopped", None):
assert status.derive_gateway_drainable(
gateway_running=True, gateway_state=state
) is False, state
Loading
Loading