diff --git a/agent/relay_runtime.py b/agent/relay_runtime.py index 095fbb8315133..7e4d7de1d779e 100644 --- a/agent/relay_runtime.py +++ b/agent/relay_runtime.py @@ -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: diff --git a/tests/agent/test_relay_runtime_bounded_scope_ops.py b/tests/agent/test_relay_runtime_bounded_scope_ops.py new file mode 100644 index 0000000000000..131dfee8f594f --- /dev/null +++ b/tests/agent/test_relay_runtime_bounded_scope_ops.py @@ -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)" + )