From 9a9b670e29f592fdf50bcca1e3e777150522b6b5 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Thu, 30 Jul 2026 08:28:31 -0600 Subject: [PATCH] fix(relay): avoid concurrent turn scope corruption Signed-off-by: Bryan Bednarski --- agent/relay_runtime.py | 38 +++++++++++++--- run_agent.py | 11 ++--- .../test_relay_shared_metrics_runtime.py | 44 ++++++++++++++++++- 3 files changed, 81 insertions(+), 12 deletions(-) diff --git a/agent/relay_runtime.py b/agent/relay_runtime.py index 533604791a865..5655e08ef5810 100644 --- a/agent/relay_runtime.py +++ b/agent/relay_runtime.py @@ -487,6 +487,7 @@ class RelayTurnContext: repr=False, ) _active_registered: bool = field(default=False, repr=False) + relay_enabled: bool = True closed: bool = False @@ -600,7 +601,28 @@ class RelaySessionCoordinator: if lease.released: raise RuntimeError("Hermes Relay conversation lease is released") turn = RelayTurnContext(lease=lease, turn_id=turn_id, task_id=task_id) - if isinstance(lease.host, RelayRuntime) and lease.session is not None: + key = (lease.profile_key, lease.session_id) + with self._active_turns_lock: + active = self._active_turns.get(key) + if active: + # A Relay session owns one physical scope stack. Concurrent + # Hermes turns would create sibling scopes on that stack, but + # their completion order is not guaranteed to be LIFO. + turn.relay_enabled = False + logger.warning( + "Skipping Relay instrumentation for concurrent Hermes turn " + "%s in session %s", + turn_id, + lease.session_id, + ) + else: + self._active_turns[key] = {id(turn)} + turn._active_registered = True + if ( + turn.relay_enabled + and isinstance(lease.host, RelayRuntime) + and lease.session is not None + ): try: turn.handle = lease.host.run_in_session( lease.session, @@ -618,10 +640,6 @@ class RelaySessionCoordinator: except Exception: logger.warning("Hermes Relay turn initialization failed", exc_info=True) turn._token = _CURRENT_TURN.set(turn) - key = (lease.profile_key, lease.session_id) - with self._active_turns_lock: - self._active_turns.setdefault(key, set()).add(id(turn)) - turn._active_registered = True return turn def end_turn( @@ -796,7 +814,12 @@ def current_turn() -> RelayTurnContext | None: def active_turn(session_id: str | None = None) -> RelayTurnContext | None: """Return a live turn only when it belongs to the active profile/session.""" turn = current_turn() - if turn is None or turn.closed or turn.lease.released: + if ( + turn is None + or not turn.relay_enabled + or turn.closed + or turn.lease.released + ): return None if turn.lease.profile_key != current_profile_key(): return None @@ -814,6 +837,9 @@ def resolve_execution_context( session_id: str, ) -> tuple[RelayRuntime | None, RelaySession | None, Any]: """Resolve one active turn/session parent for managed Relay execution.""" + inherited_turn = current_turn() + if inherited_turn is not None and not inherited_turn.relay_enabled: + return None, None, None turn = active_turn(session_id) if ( turn is not None diff --git a/run_agent.py b/run_agent.py index e2bc5f8660bd6..bd2b6e5c29aef 100644 --- a/run_agent.py +++ b/run_agent.py @@ -7624,11 +7624,12 @@ class AIAgent: turn_id=relay_turn_id, task_id=effective_task_id, ) - start_task_run( - **task_context, - parent_session_id=getattr(self, "_parent_session_id", None) or "", - ) - task_started = True + if relay_turn.relay_enabled: + start_task_run( + **task_context, + parent_session_id=getattr(self, "_parent_session_id", None) or "", + ) + task_started = True # Publish the conversation id for ambient Nous Portal tagging. Every # LLM call made inside this turn — main loop, compression, vision, # web_extract, session_search, MoA slots, background-review forks diff --git a/tests/hermes_cli/test_relay_shared_metrics_runtime.py b/tests/hermes_cli/test_relay_shared_metrics_runtime.py index fab88396da77e..6255f01f70235 100644 --- a/tests/hermes_cli/test_relay_shared_metrics_runtime.py +++ b/tests/hermes_cli/test_relay_shared_metrics_runtime.py @@ -792,6 +792,49 @@ def test_sync_session_runner_releases_lock_before_callback(direct_runtime): assert contender.is_alive() is False +def test_concurrent_turn_skips_relay_before_scope_stack_can_interleave( + direct_runtime, +): + coordinator = relay_runtime.SESSION_COORDINATOR + profile_key = relay_runtime.current_profile_key() + lease = coordinator.acquire_conversation( + profile_key=profile_key, + session_id="shared-session", + platform="cli", + ) + first = coordinator.begin_turn(lease, turn_id="first", task_id="first-task") + second = coordinator.begin_turn( + lease, + turn_id="second", + task_id="second-task", + ) + + assert first.relay_enabled is True + assert first.handle is not None + assert second.relay_enabled is False + assert second.handle is None + assert relay_runtime.resolve_execution_context("shared-session") == ( + None, + None, + None, + ) + + coordinator.end_turn(first, outcome="success") + coordinator.end_turn(second, outcome="success") + coordinator.release_conversation(lease) + coordinator.finalize_conversation( + profile_key=profile_key, + session_id="shared-session", + ) + + turn_closes = [ + event + for event in direct_runtime.events + if event[0] == "scope.pop" and event[1] == first.handle + ] + assert len(turn_closes) == 1 + + @@ -974,4 +1017,3 @@ def test_failed_flush_keeps_daily_export_open_for_later_task( -