RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(relay): bound native scope lifecycle operations so a wedged pipeline cannot block the agent 

The NeMo Relay native binding's scope.pop/push are synchronous and
unbounded ('returns after the scope is closed successfully'). When the
native pipeline cannot make progress, the session coordinator's turn and
session finalization block forever inside run_conversation: delegated
children finish their turns but never return, and delegation batches die
on the stall watchdog. Proven live 2026-08-10 on the staging fleet — a
falsification probe (plugin disabled, identical config) completed the
same delegation batch that wedged with the plugin active.

Bound every scope lifecycle operation that gates turn/session completion
(session push, turn push, turn pop, logical-LLM pops, session pop,
subscriber flush) by running the native call on a shared
DaemonThreadPoolExecutor and honoring a 10s result timeout. On breach a
TimeoutError propagates into each call site's existing exception
handling — warn, retain the unclosed-prefix diagnostics, continue — so
the worst case is one lost span, never a blocked agent. timeout=None
preserves byte-identical synchronous behavior for all other callers, and
interpreter-shutdown paths fall back to the synchronous call so the
atexit flush still exports.

Observability must never block the product.
This commit is contained in:
Victor Kyriazakos 2026-08-10 18:54:32 +00:00 committed by Teknium
parent 11310068c6
commit d607f0cafb
2 changed files with 361 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

112
agent/relay_runtime.py
View File

@ -10,6 +10,7 @@ import inspect
import logging
import threading
import uuid
from concurrent.futures import TimeoutError as FuturesTimeoutError
from dataclasses import dataclass, field
from typing import Any, Callable
@ -25,6 +26,39 @@ RUNTIME_SCHEMA_VERSION = "hermes.relay.runtime.v1"
RUNTIME_INSTANCE_KEY = "hermes.relay.runtime_instance"
_PROFILE_KEY_CACHE: dict[str, str] = {}
# Bound for native scope lifecycle operations (push/pop/flush) that gate
# turn/session completion. Healthy operations complete in microseconds;
# only a wedged native pipeline breaches this, and the correct trade there
# is one lost span, never a blocked agent (2026-08-10 delegation stall).
_SCOPE_OP_TIMEOUT = 10.0
_SCOPE_OP_EXECUTOR: Any = None
_SCOPE_OP_EXECUTOR_LOCK = threading.Lock()
def _scope_op_executor():
"""Shared daemon executor for bounded native scope operations.
Daemon workers (tools.daemon_pool) so a wedged native call abandoned at
timeout cannot block interpreter exit. Sized generously: workers are
only consumed for the duration of healthy (microsecond) operations plus
any wedged calls, and ``Future.result(timeout=...)`` bounds callers even
when every worker is consumed by wedged calls — an unstarted future
still honors the result timeout, so exhaustion degrades to fast
timeouts, never a new hang.
"""
global _SCOPE_OP_EXECUTOR
if _SCOPE_OP_EXECUTOR is None:
with _SCOPE_OP_EXECUTOR_LOCK:
if _SCOPE_OP_EXECUTOR is None:
from tools.daemon_pool import DaemonThreadPoolExecutor
_SCOPE_OP_EXECUTOR = DaemonThreadPoolExecutor(
max_workers=8,
thread_name_prefix="relay-scope-op",
)
return _SCOPE_OP_EXECUTOR
@dataclass
class RelaySession:
@ -113,15 +147,29 @@ class RelayRuntime:
scope_metadata["nemo_relay_scope_role"] = "subagent"
context = contextvars.Context()
try:
session.handle = context.run(
self.relay.scope.push,
SESSION_SCOPE,
self.relay.ScopeType.Agent,
handle=parent_handle,
data=data,
input={},
metadata=scope_metadata,
)
try:
session.handle = _scope_op_executor().submit(
context.run,
self.relay.scope.push,
SESSION_SCOPE,
self.relay.ScopeType.Agent,
handle=parent_handle,
data=data,
input={},
metadata=scope_metadata,
).result(timeout=_SCOPE_OP_TIMEOUT)
except RuntimeError:
# Interpreter shutdown: executor refuses new futures;
# push synchronously (no agent turn waits at exit).
session.handle = context.run(
self.relay.scope.push,
SESSION_SCOPE,
self.relay.ScopeType.Agent,
handle=parent_handle,
data=data,
input={},
metadata=scope_metadata,
)
except Exception:
session.context = None
raise
@ -194,9 +242,22 @@ class RelayRuntime:
callback: Callable[..., Any],
*args: Any,
allow_closing: bool = False,
timeout: float | None = None,
**kwargs: Any,
) -> Any:
"""Run a Relay operation against a session's isolated scope stack."""
"""Run a Relay operation against a session's isolated scope stack.
``timeout`` (seconds) bounds the native call by running it on a
shared daemon executor; ``TimeoutError`` propagates to the caller's
existing exception handling on breach. ``None`` (default) preserves
the historical synchronous behavior. Scope lifecycle operations
that gate turn/session completion pass ``_SCOPE_OP_TIMEOUT``: the
native binding's ``scope.pop`` "returns after the scope is closed
successfully" — unbounded — and a wedged native pipeline (proven
live 2026-08-10 in the delegation topology) must cost at most one
span, never the agent. The abandoned daemon worker cannot block
process exit (tools.daemon_pool contract).
"""
with session.lock:
if session.closing and not allow_closing:
raise RuntimeError("Hermes Relay session is closing")
@ -214,7 +275,23 @@ class RelayRuntime:
# A copy permits a helper called by an existing Relay callback to
# re-enter the same logical session without re-entering Context.
return context.run(invoke)
if timeout is None:
return context.run(invoke)
try:
future = _scope_op_executor().submit(context.run, invoke)
except RuntimeError:
# Interpreter shutdown: the executor refuses new futures, but
# the atexit close path must still flush cleanly. No agent
# turn is waiting at shutdown, so the unbounded call is safe.
return context.run(invoke)
try:
return future.result(timeout=timeout)
except FuturesTimeoutError as exc:
raise TimeoutError(
f"Relay scope operation exceeded {timeout}s "
f"(session={session.session_id}); abandoning the native call "
"so the agent can continue — the span for this scope is lost"
) from exc
async def run_in_session_async(
self,
@ -323,11 +400,19 @@ class RelayRuntime:
RUNTIME_INSTANCE_KEY: self.runtime_id,
},
allow_closing=True,
timeout=_SCOPE_OP_TIMEOUT,
)
except Exception as exc:
failures.append(f"session scope close failed: {exc}")
try:
self.relay.subscribers.flush()
try:
_scope_op_executor().submit(
self.relay.subscribers.flush
).result(timeout=_SCOPE_OP_TIMEOUT)
except RuntimeError:
# Interpreter shutdown: executor refuses new futures; flush
# synchronously so the atexit close path still exports.
self.relay.subscribers.flush()
except Exception as exc:
failures.append(f"subscriber flush failed: {exc}")
with self._sessions_lock:
@ -661,6 +746,7 @@ class RelaySessionCoordinator:
RUNTIME_INSTANCE_KEY: lease.host.runtime_id,
"hermes.execution_surface": lease.platform or "unknown",
},
timeout=_SCOPE_OP_TIMEOUT,
)
except Exception:
logger.warning("Hermes Relay turn initialization failed", exc_info=True)
@ -694,6 +780,7 @@ class RelaySessionCoordinator:
RUNTIME_SCHEMA_KEY: RUNTIME_SCHEMA_VERSION,
RUNTIME_INSTANCE_KEY: lease.host.runtime_id,
},
timeout=_SCOPE_OP_TIMEOUT,
)
except Exception:
logger.warning(
@ -778,6 +865,7 @@ class RelaySessionCoordinator:
RUNTIME_SCHEMA_KEY: RUNTIME_SCHEMA_VERSION,
RUNTIME_INSTANCE_KEY: lease.host.runtime_id,
},
timeout=_SCOPE_OP_TIMEOUT,
)
except Exception:
with turn.logical_llm_lock:

261
tests/agent/test_relay_runtime_bounded_scope_ops.py Normal file
View File

@ -0,0 +1,261 @@
"""Bounded native scope operations in the Relay session coordinator.
The NeMo Relay native binding's ``scope.pop``/``scope.push`` are synchronous
and unbounded ("returns after the scope is closed successfully"). When the
native pipeline cannot make progress — proven live 2026-08-10 in the
delegation topology, where child sessions register scopes under the parent's
handle on a shared runtime — the coordinator's turn/session finalization
blocks forever inside ``run_conversation``. Children finish their turns but
never return; delegation batches die on the stall watchdog.
Contract under test: observability must never block the product. Scope
lifecycle operations that gate turn/session completion are bounded; on
breach the existing per-site exception handling degrades gracefully (warn,
retain diagnostics, continue) and the agent lives. A lost span is always
the right trade against a dead agent.
These tests use a fake relay whose ``pop`` blocks on a never-set Event —
the minimal stand-in for a wedged native pipeline. They assert the
END-TO-END contract (the coordinator method RETURNS) rather than any
helper's internal shape.
"""
from __future__ import annotations
import threading
import time
import types
from typing import Any
import pytest
from agent import relay_runtime
from agent.relay_runtime import (
RelayRuntime,
RelaySessionCoordinator,
)
# ---------------------------------------------------------------------------
# Fake relay: minimal surface the coordinator touches, with a wedgeable pop.
# ---------------------------------------------------------------------------
class _ScopeHandle:
def __init__(self, name: str) -> None:
self.name = name
class _FakeScopeModule:
"""Stands in for ``nemo_relay.scope`` with a controllable pop."""
def __init__(self, wedge_event: threading.Event | None = None) -> None:
self._wedge = wedge_event
self.pushed: list[str] = []
self.popped: list[str] = []
def push(self, name: str, scope_type: Any, **kwargs: Any) -> _ScopeHandle:
self.pushed.append(name)
return _ScopeHandle(name)
def pop(self, handle: _ScopeHandle, **kwargs: Any) -> None:
if self._wedge is not None:
# Simulates the wedged native pipeline: blocks until the event
# is set — which the tests never do.
self._wedge.wait()
self.popped.append(handle.name)
def event(self, *args: Any, **kwargs: Any) -> None:
return None
class _FakeSubscribers:
def __init__(self, wedge_event: threading.Event | None = None) -> None:
self._wedge = wedge_event
self.flushed = 0
def flush(self) -> None:
if self._wedge is not None:
self._wedge.wait()
self.flushed += 1
class _FakeScopeType:
Function = "function"
Agent = "agent"
class _FakeRelay:
def __init__(
self,
*,
wedge_pop: threading.Event | None = None,
wedge_flush: threading.Event | None = None,
) -> None:
self.scope = _FakeScopeModule(wedge_pop)
self.subscribers = _FakeSubscribers(wedge_flush)
self.ScopeType = _FakeScopeType()
def get_scope_stack(self) -> None:
return None
def _make_runtime(fake_relay: _FakeRelay) -> RelayRuntime:
"""Build a RelayRuntime around the fake relay without native imports."""
return RelayRuntime(relay=fake_relay, profile_key="/tmp/test-profile")
def _run_with_join(fn, timeout: float = 5.0) -> tuple[bool, list[Any]]:
"""Run ``fn`` on a thread; return (returned_within_timeout, result)."""
result: list[Any] = []
def _target() -> None:
result.append(fn())
t = threading.Thread(target=_target, daemon=True)
t.start()
t.join(timeout)
return (not t.is_alive(), result)
@pytest.fixture()
def coordinator() -> RelaySessionCoordinator:
return RelaySessionCoordinator()
@pytest.fixture(autouse=True)
def _fast_scope_timeout(monkeypatch):
"""Shrink the scope-op bound so wedge tests run in seconds.
The production constant is generous (healthy ops are microseconds);
tests only need 'bounded', not the specific bound.
"""
monkeypatch.setattr(relay_runtime, "_SCOPE_OP_TIMEOUT", 1.0)
def _acquire(coordinator, runtime, session_id="sess-1", monkeypatch=None):
"""Acquire a conversation lease against the fake runtime."""
class _Registry:
def for_profile(self, key):
return runtime
coordinator.registry = _Registry()
# _prepare_session invokes plugin initializers; make it inert for the
# scope-op tests (plugin behavior is covered in tests/plugins/).
coordinator._prepare_session = lambda host, ctx: None
return coordinator.acquire_conversation(
profile_key=runtime.profile_key,
session_id=session_id,
platform="test",
)
# ---------------------------------------------------------------------------
# RED tests: today these HANG (the fake pop blocks forever) and the join
# times out. Post-fix the coordinator bounds the native call and returns.
# ---------------------------------------------------------------------------
class TestBoundedScopeFinalization:
def test_end_turn_returns_when_native_pop_wedges(self, coordinator):
wedge = threading.Event() # never set
runtime = _make_runtime(_FakeRelay(wedge_pop=wedge))
lease = _acquire(coordinator, runtime)
assert lease.session is not None, "fake session must initialize"
turn = coordinator.begin_turn(lease, turn_id="t1", task_id="task1")
assert turn.handle is not None, "turn scope must push on fake relay"
returned, _ = _run_with_join(
lambda: coordinator.end_turn(turn, outcome="success")
)
assert returned, (
"end_turn must return even when the native scope.pop wedges — "
"observability must never block turn completion"
)
def test_close_session_returns_when_native_pop_wedges(self, coordinator):
wedge = threading.Event()
runtime = _make_runtime(_FakeRelay(wedge_pop=wedge))
lease = _acquire(coordinator, runtime)
assert lease.session is not None
returned, _ = _run_with_join(
lambda: runtime.close_session({"session_id": "sess-1"})
)
assert returned, (
"close_session must return even when the native scope.pop wedges"
)
def test_close_session_returns_when_subscriber_flush_wedges(
self, coordinator
):
wedge = threading.Event()
runtime = _make_runtime(_FakeRelay(wedge_flush=wedge))
lease = _acquire(coordinator, runtime)
assert lease.session is not None
returned, _ = _run_with_join(
lambda: runtime.close_session({"session_id": "sess-1"})
)
assert returned, (
"close_session must return even when subscribers.flush wedges"
)
def test_finish_logical_calls_returns_and_retains_prefix(
self, coordinator
):
wedge = threading.Event()
runtime = _make_runtime(_FakeRelay(wedge_pop=wedge))
lease = _acquire(coordinator, runtime)
turn = coordinator.begin_turn(lease, turn_id="t1", task_id="task1")
# Register two logical LLM scopes the way relay_llm does.
h1, h2 = _ScopeHandle("llm-1"), _ScopeHandle("llm-2")
with turn.logical_llm_lock:
turn.logical_llm_calls["req-1"] = h1
turn.logical_llm_calls["req-2"] = h2
returned, _ = _run_with_join(
lambda: coordinator.finish_logical_calls(turn, outcome="success")
)
assert returned, (
"finish_logical_calls must return even when native pops wedge"
)
# Existing diagnostic contract: the unclosed prefix is retained.
with turn.logical_llm_lock:
assert turn.logical_llm_calls, (
"wedged logical scopes must be retained for diagnostics, "
"not silently dropped"
)
class TestHealthyPathUnchanged:
"""The bound must be invisible when the native pipeline is healthy."""
def test_full_turn_lifecycle_healthy(self, coordinator):
runtime = _make_runtime(_FakeRelay()) # no wedges
lease = _acquire(coordinator, runtime)
turn = coordinator.begin_turn(lease, turn_id="t1", task_id="task1")
assert turn.handle is not None
coordinator.finish_logical_calls(turn, outcome="success")
coordinator.end_turn(turn, outcome="success")
coordinator.release_conversation(lease)
runtime.close_session({"session_id": "sess-1"})
fake = runtime.relay
# Turn scope and session scope both pushed and popped exactly once.
assert fake.scope.pushed.count(relay_runtime.TURN_SCOPE) == 1
assert relay_runtime.TURN_SCOPE in fake.scope.popped
assert fake.subscribers.flushed >= 1
def test_healthy_pop_result_propagates_synchronously(self, coordinator):
"""A healthy pop completes and is observed before end_turn returns."""
runtime = _make_runtime(_FakeRelay())
lease = _acquire(coordinator, runtime)
turn = coordinator.begin_turn(lease, turn_id="t1", task_id="task1")
coordinator.end_turn(turn, outcome="success")
assert relay_runtime.TURN_SCOPE in runtime.relay.scope.popped, (
"healthy-path pop must complete before end_turn returns "
"(no fire-and-forget on the default lane)"
)