mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-30 19:09:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(monitoring): drain terminal lifecycle events

Browse source
This commit is contained in:
Victor Kyriazakos 2026-07-22 22:08:51 +00:00
parent 4dc4f4302a
commit 18b4664543
7 changed files with 139 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

33
agent/monitoring/emitter.py
View file

@ -67,6 +67,7 @@ class MonitoringEmitter:
# Drop oldest to make room — bounded memory, newest-wins.
try:
self._q.get_nowait()
self._q.task_done()
self._dropped += 1
self._q.put_nowait(payload)
except Exception:
@ -99,7 +100,11 @@ class MonitoringEmitter:
batch.append(self._q.get_nowait())
except queue.Empty:
break
self._dispatch(batch)
try:
self._dispatch(batch)
finally:
for _ in batch:
self._q.task_done()
def _dispatch(self, batch) -> None:
# Fan-out to subscribers (OTLP streamers) — fully fail-isolated.
@ -126,15 +131,23 @@ class MonitoringEmitter:
# ── introspection / shutdown (tests, CLI) ───────────────────────────────
def flush(self, timeout: float = 2.0) -> None:
"""Block until the queue drains (test/CLI helper, NOT the hot path)."""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if self._q.empty():
# give the dispatcher a tick to finish the in-flight batch
time.sleep(0.05)
if self._q.empty():
return
time.sleep(0.02)
"""Wait boundedly for queued and in-flight batches to finish dispatch."""
if timeout <= 0:
return
finished = threading.Event()
def _wait_for_completion() -> None:
self._q.join()
finished.set()
waiter = threading.Thread(
target=_wait_for_completion,
name="hermes-monitoring-flush",
daemon=True,
)
waiter.start()
finished.wait(timeout=timeout)
def stats(self) -> Dict[str, int]:
return {

9
agent/monitoring/gateway_health_export.py
View file

@ -118,12 +118,13 @@ class GatewayHealthExportRuntime:
except Exception:
pass
# Detach synchronously so no new records are collected during a slow
# exporter shutdown. Network flush/close then runs under one bounded
# daemon-thread deadline and can never delay gateway teardown.
# All producers above are now stopped. Drain queued and in-flight
# events before detaching subscribers so the terminal lifecycle event
# cannot race exporter shutdown. The barrier is bounded and fail-open.
try:
from agent.monitoring.emitter import get_emitter
emitter = get_emitter()
emitter.flush(timeout=1.0)
if self.streamer is not None:
emitter.unsubscribe(self.streamer)
if self.log_streamer is not None:
@ -131,6 +132,8 @@ class GatewayHealthExportRuntime:
except Exception:
pass
# Network flush/close runs under one bounded daemon-thread deadline and
# can never delay gateway teardown indefinitely.
closeables = [
item for item in (self.streamer, self.log_streamer, self.metric_provider)
if item is not None

54
gateway/run.py
View file

@ -9770,12 +9770,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
self._update_runtime_status("running", self._exit_reason)
else:
self._update_runtime_status("stopped", self._exit_reason)
try:
_gh_runtime = getattr(self, "_gateway_health_export_runtime", None)
if _gh_runtime is not None:
_gh_runtime.shutdown()
except Exception:
logger.debug("gateway health OTLP export shutdown failed", exc_info=True)
_shutdown_gateway_health_export(self)
logger.info("Gateway stopped (total teardown %.2fs)", _phase_elapsed())
self._stop_task = asyncio.create_task(_stop_impl())
@ -23552,6 +23547,18 @@ async def _await_thread_exit(
return not thread.is_alive()
def _shutdown_gateway_health_export(runner: Any) -> None:
"""Idempotently drain and detach Gateway Health OTLP export."""
runtime = getattr(runner, "_gateway_health_export_runtime", None)
if runtime is None:
return
runner._gateway_health_export_runtime = None
try:
runtime.shutdown()
except Exception:
logger.debug("gateway health OTLP export shutdown failed", exc_info=True)
async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool = False, verbosity: Optional[int] = 0) -> bool:
"""
Start the gateway and run until interrupted.
@ -23999,10 +24006,16 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
logger.debug("MCP tool discovery failed: %s", e)
# Start the gateway
success = await runner.start()
try:
success = await runner.start()
except BaseException:
_shutdown_gateway_health_export(runner)
raise
if not success:
_shutdown_gateway_health_export(runner)
return False
if runner.should_exit_cleanly:
_shutdown_gateway_health_export(runner)
if runner.exit_reason:
logger.error("Gateway exiting cleanly: %s", runner.exit_reason)
# A clean exit that carries an explicit exit code (e.g. a fatal
@ -24018,19 +24031,22 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
if not runner._running:
# Startup was intentionally aborted by restart/shutdown before entering
# running mode; preserve that lifecycle path without starting cron.
await runner.wait_for_shutdown()
if runner.should_exit_with_failure:
if runner.exit_reason:
logger.error("Gateway exiting with failure: %s", runner.exit_reason)
return False
try:
from tools.mcp_tool import shutdown_mcp_servers
shutdown_mcp_servers()
except Exception:
pass
if runner.exit_code is not None:
raise SystemExit(runner.exit_code)
return True
await runner.wait_for_shutdown()
if runner.should_exit_with_failure:
if runner.exit_reason:
logger.error("Gateway exiting with failure: %s", runner.exit_reason)
return False
try:
from tools.mcp_tool import shutdown_mcp_servers
shutdown_mcp_servers()
except Exception:
pass
if runner.exit_code is not None:
raise SystemExit(runner.exit_code)
return True
finally:
_shutdown_gateway_health_export(runner)
# Start the background cron scheduler via the resolved provider so
# scheduled jobs fire automatically. The built-in provider is the

1
scripts/observability/gateway_health_export_probe.py
View file

@ -72,6 +72,7 @@ def main() -> None:
logging.getLogger("gateway.platforms.slack").warning("Slack token *** rejected for smoke@example.com")
emitter.get_emitter().flush(timeout=2.0)
time.sleep(args.wait)
write_runtime_status(gateway_state="stopped", active_agents=0)
runtime.shutdown()
emitter.get_emitter().flush(timeout=2.0)

10
tests/gateway/test_startup_restart_race.py
View file

@ -250,16 +250,23 @@ async def test_startup_aborts_after_registered_adapter_restart(tmp_path, monkeyp
async def test_start_gateway_does_not_start_cron_after_aborted_startup(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
cron_started = False
export_shutdown_calls = 0
class ExportRuntime:
def shutdown(self):
nonlocal export_shutdown_calls
export_shutdown_calls += 1
class AbortedStartupRunner:
def __init__(self, config):
self.config = config
self.adapters = {}
self._running = False
self.should_exit_cleanly = False
self.should_exit_cleanly = True
self.should_exit_with_failure = False
self.exit_reason = None
self.exit_code = GATEWAY_SERVICE_RESTART_EXIT_CODE
self._gateway_health_export_runtime = ExportRuntime()
async def start(self):
return True
@ -287,3 +294,4 @@ async def test_start_gateway_does_not_start_cron_after_aborted_startup(tmp_path,
assert exc.value.code == GATEWAY_SERVICE_RESTART_EXIT_CODE
assert cron_started is False
assert export_shutdown_calls == 1

30
tests/monitoring/test_emitter.py
View file

@ -3,6 +3,7 @@
from __future__ import annotations
import time
import threading
from agent.monitoring.emitter import MonitoringEmitter
from agent.monitoring.events import GatewayHealthEvent
@ -65,6 +66,35 @@ def test_subscriber_failure_is_isolated():
assert len(good) == 1 # the raising subscriber did not break fan-out
def test_flush_waits_for_in_flight_subscriber_delivery():
em = MonitoringEmitter()
subscriber_started = threading.Event()
release_subscriber = threading.Event()
flush_finished = threading.Event()
def blocking_subscriber(_batch):
subscriber_started.set()
release_subscriber.wait(timeout=2.0)
em.subscribe(blocking_subscriber)
em.emit({"event": "gateway_health", "name": "gateway.exit"})
assert subscriber_started.wait(timeout=1.0)
flush_thread = threading.Thread(
target=lambda: (em.flush(timeout=1.0), flush_finished.set()),
daemon=True,
)
flush_thread.start()
try:
assert not flush_finished.wait(timeout=0.1)
finally:
release_subscriber.set()
assert flush_finished.wait(timeout=1.0)
em.close()
def test_unsubscribe_stops_delivery():
em = MonitoringEmitter()
seen: list = []

35
tests/monitoring/test_gateway_health_export.py
View file

@ -388,6 +388,41 @@ def test_gateway_health_export_start_is_fail_open_when_otlp_missing(monkeypatch)
assert runtime.reason == "otlp_unavailable"
def test_gateway_health_export_shutdown_flushes_before_unsubscribe(monkeypatch):
from agent.monitoring import emitter
from agent.monitoring.gateway_health_export import GatewayHealthExportRuntime
calls = []
class FakeEmitter:
def flush(self, timeout):
calls.append(("flush", timeout))
def unsubscribe(self, subscriber):
calls.append(("unsubscribe", subscriber))
class Streamer:
def shutdown(self):
calls.append(("shutdown", self))
streamer = Streamer()
log_streamer = Streamer()
monkeypatch.setattr(emitter, "get_emitter", lambda: FakeEmitter())
runtime = GatewayHealthExportRuntime(
enabled=True,
streamer=streamer,
log_streamer=log_streamer,
)
runtime.shutdown()
assert calls[0][0] == "flush"
assert calls[1:3] == [
("unsubscribe", streamer),
("unsubscribe", log_streamer),
]
def test_gateway_health_export_streams_only_gateway_events(monkeypatch):
from agent.monitoring import gateway_health_export