diff --git a/run_agent.py b/run_agent.py index a0ae15a16226e..8f1344cd38603 100644 --- a/run_agent.py +++ b/run_agent.py @@ -23,6 +23,7 @@ import asyncio import base64 import concurrent.futures +import contextvars import copy import hashlib import json @@ -68,7 +69,10 @@ ) from tools.terminal_tool import cleanup_vm, get_active_env from tools.tool_result_storage import maybe_persist_tool_result, enforce_turn_budget -from tools.interrupt import set_interrupt as _set_interrupt +from tools.interrupt import ( + bind_event as _bind_interrupt_event, + unbind_event as _unbind_interrupt_event, +) from tools.browser_tool import cleanup_browser @@ -606,9 +610,15 @@ def __init__( # even when stream consumers are registered (no tokens streaming then) self._executing_tools = False - # Interrupt mechanism for breaking out of tool loops + # Interrupt mechanism for breaking out of tool loops. + # ``_interrupt_event`` is the per-agent signal that long-running + # tools poll via ``tools.interrupt.is_interrupted``. It is bound + # to a ``contextvars.ContextVar`` for the duration of + # ``run_conversation()`` so concurrent agents in the same process + # cannot interrupt each other's tool calls. self._interrupt_requested = False self._interrupt_message = None # Optional message that triggered interrupt + self._interrupt_event = threading.Event() self._client_lock = threading.RLock() # Subagent delegation state @@ -2474,8 +2484,14 @@ def interrupt(self, message: str = None) -> None: """ self._interrupt_requested = True self._interrupt_message = message - # Signal all tools to abort any in-flight operations immediately - _set_interrupt(True) + # Signal this agent's own tools to abort in-flight operations. + # Writing to ``self._interrupt_event`` directly (rather than the + # ``set_interrupt`` helper) bypasses the contextvar lookup, which + # is critical because ``interrupt()`` is typically called from a + # different thread (gateway message handler, SSE disconnect + # handler) where the contextvar may be unbound or bound to a + # different agent. + self._interrupt_event.set() # Propagate interrupt to any running child agents (subagent delegation) with self._active_children_lock: children_copy = list(self._active_children) @@ -2488,10 +2504,17 @@ def interrupt(self, message: str = None) -> None: print("\nāš” Interrupt requested" + (f": '{message[:40]}...'" if message and len(message) > 40 else f": '{message}'" if message else "")) def clear_interrupt(self) -> None: - """Clear any pending interrupt request and the global tool interrupt signal.""" + """Clear any pending interrupt request for this agent only. + + Writes to ``self._interrupt_event`` directly so that clearing + one agent's interrupt does not silently un-interrupt another + agent running concurrently in the same process ā€” for example + a fresh ``run_conversation()`` starting on the API server while + a sibling request is mid-tool. + """ self._interrupt_requested = False self._interrupt_message = None - _set_interrupt(False) + self._interrupt_event.clear() def _touch_activity(self, desc: str) -> None: """Update the last-activity timestamp and description (thread-safe).""" @@ -2563,7 +2586,10 @@ def _hydrate_todo_store(self, history: List[Dict[str, Any]]) -> None: self._todo_store.write(last_todo_response, merge=False) if not self.quiet_mode: self._vprint(f"{self.log_prefix}šŸ“‹ Restored {len(last_todo_response)} todo item(s) from history") - _set_interrupt(False) + # Clear stale interrupt state on this agent only ā€” historical + # behaviour reset the process-wide flag here, which would race + # with concurrent agents in the same process. + self._interrupt_event.clear() @property def is_interrupted(self) -> bool: @@ -6143,10 +6169,20 @@ def _run_tool(index, tool_call, function_name, function_args): try: max_workers = min(num_tools, _MAX_TOOL_WORKERS) + # Snapshot the calling thread's context so each worker + # inherits the per-agent interrupt event bound by + # ``run_conversation`` via ``tools.interrupt``. + # ``ThreadPoolExecutor.submit`` does not propagate + # contextvars automatically, and a single ``Context`` + # object cannot be ``run()`` from more than one thread, so + # we hand each worker a fresh ``copy()`` of the snapshot. + _parent_ctx = contextvars.copy_context() with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: futures = [] for i, (tc, name, args) in enumerate(parsed_calls): - f = executor.submit(_run_tool, i, tc, name, args) + f = executor.submit( + _parent_ctx.copy().run, _run_tool, i, tc, name, args + ) futures.append(f) # Wait for all to complete (exceptions are captured inside _run_tool) @@ -6805,6 +6841,40 @@ def run_conversation( task_id: str = None, stream_callback: Optional[callable] = None, persist_user_message: Optional[str] = None, + ) -> Dict[str, Any]: + """Public entry point that binds this agent's interrupt event. + + The actual conversation logic lives in + :meth:`_run_conversation_locked`. This wrapper exists solely so + each turn binds ``self._interrupt_event`` to the + ``tools.interrupt`` context variable for its entire duration and + unbinds it on exit. Without the binding, tools polling + :func:`tools.interrupt.is_interrupted` would observe the + process-wide fallback event and could see interrupts intended + for a different concurrent agent (or fail to see this agent's + interrupt at all). + """ + _interrupt_token = _bind_interrupt_event(self._interrupt_event) + try: + return self._run_conversation_locked( + user_message=user_message, + system_message=system_message, + conversation_history=conversation_history, + task_id=task_id, + stream_callback=stream_callback, + persist_user_message=persist_user_message, + ) + finally: + _unbind_interrupt_event(_interrupt_token) + + def _run_conversation_locked( + self, + user_message: str, + system_message: str = None, + conversation_history: List[Dict[str, Any]] = None, + task_id: str = None, + stream_callback: Optional[callable] = None, + persist_user_message: Optional[str] = None, ) -> Dict[str, Any]: """ Run a complete conversation with tool calling until completion. diff --git a/tests/cli/test_cli_interrupt_subagent.py b/tests/cli/test_cli_interrupt_subagent.py index f4322ea6b9604..99121e75ddc0a 100644 --- a/tests/cli/test_cli_interrupt_subagent.py +++ b/tests/cli/test_cli_interrupt_subagent.py @@ -42,6 +42,7 @@ def test_full_delegate_interrupt_flow(self): parent = AIAgent.__new__(AIAgent) parent._interrupt_requested = False parent._interrupt_message = None + parent._interrupt_event = threading.Event() parent._active_children = [] parent._active_children_lock = threading.Lock() parent.quiet_mode = True @@ -112,6 +113,7 @@ def run_delegate(): mock_instance = MagicMock() mock_instance._interrupt_requested = False mock_instance._interrupt_message = None + mock_instance._interrupt_event = threading.Event() mock_instance._active_children = [] mock_instance._active_children_lock = threading.Lock() mock_instance.quiet_mode = True diff --git a/tests/run_agent/test_interrupt_propagation.py b/tests/run_agent/test_interrupt_propagation.py index 7f8cb01c35b99..2051eb2c54d11 100644 --- a/tests/run_agent/test_interrupt_propagation.py +++ b/tests/run_agent/test_interrupt_propagation.py @@ -29,6 +29,7 @@ def test_parent_interrupt_sets_child_flag(self): parent = AIAgent.__new__(AIAgent) parent._interrupt_requested = False parent._interrupt_message = None + parent._interrupt_event = threading.Event() parent._active_children = [] parent._active_children_lock = threading.Lock() parent.quiet_mode = True @@ -36,6 +37,7 @@ def test_parent_interrupt_sets_child_flag(self): child = AIAgent.__new__(AIAgent) child._interrupt_requested = False child._interrupt_message = None + child._interrupt_event = threading.Event() child._active_children = [] child._active_children_lock = threading.Lock() child.quiet_mode = True @@ -47,31 +49,46 @@ def test_parent_interrupt_sets_child_flag(self): assert parent._interrupt_requested is True assert child._interrupt_requested is True assert child._interrupt_message == "new user message" - assert is_interrupted() is True - - def test_child_clear_interrupt_at_start_clears_global(self): - """child.clear_interrupt() at start of run_conversation clears the GLOBAL event. - - This is the intended behavior at startup, but verify it doesn't - accidentally clear an interrupt intended for a running child. + # Both per-agent events are set; the global fallback remains + # untouched because per-agent isolation is now in effect. + assert parent._interrupt_event.is_set() is True + assert child._interrupt_event.is_set() is True + assert _interrupt_event.is_set() is False + + def test_child_clear_interrupt_does_not_affect_global(self): + """child.clear_interrupt() must clear only the child's per-agent event. + + Historical behaviour reset the process-wide ``_interrupt_event`` + too, which silently un-interrupted any other agent running + concurrently in the same process. The fix isolates each + agent's interrupt state. """ from run_agent import AIAgent child = AIAgent.__new__(AIAgent) child._interrupt_requested = True child._interrupt_message = "msg" + child._interrupt_event = threading.Event() + child._interrupt_event.set() child.quiet_mode = True child._active_children = [] child._active_children_lock = threading.Lock() - # Global is set + # Independently, the global fallback is set (representing some + # other code path or an unrelated agent that has not yet been + # migrated to per-agent events). set_interrupt(True) - assert is_interrupted() is True + assert _interrupt_event.is_set() is True - # child.clear_interrupt() clears both + # Clearing the child only clears the child's own event. child.clear_interrupt() assert child._interrupt_requested is False - assert is_interrupted() is False + assert child._interrupt_event.is_set() is False + # Global fallback must remain untouched. + assert _interrupt_event.is_set() is True + # Manual cleanup so tearDown's set_interrupt(False) is a no-op + # equivalent. + set_interrupt(False) def test_interrupt_during_child_api_call_detected(self): """Interrupt set during _interruptible_api_call is detected within 0.5s.""" @@ -80,6 +97,7 @@ def test_interrupt_during_child_api_call_detected(self): child = AIAgent.__new__(AIAgent) child._interrupt_requested = False child._interrupt_message = None + child._interrupt_event = threading.Event() child._active_children = [] child._active_children_lock = threading.Lock() child.quiet_mode = True @@ -122,6 +140,7 @@ def test_concurrent_interrupt_propagation(self): parent = AIAgent.__new__(AIAgent) parent._interrupt_requested = False parent._interrupt_message = None + parent._interrupt_event = threading.Event() parent._active_children = [] parent._active_children_lock = threading.Lock() parent.quiet_mode = True @@ -129,6 +148,7 @@ def test_concurrent_interrupt_propagation(self): child = AIAgent.__new__(AIAgent) child._interrupt_requested = False child._interrupt_message = None + child._interrupt_event = threading.Event() child._active_children = [] child._active_children_lock = threading.Lock() child.quiet_mode = True diff --git a/tests/run_agent/test_real_interrupt_subagent.py b/tests/run_agent/test_real_interrupt_subagent.py index e0e681cdf4058..60d2a808950e4 100644 --- a/tests/run_agent/test_real_interrupt_subagent.py +++ b/tests/run_agent/test_real_interrupt_subagent.py @@ -54,6 +54,7 @@ def test_interrupt_child_during_api_call(self): parent = AIAgent.__new__(AIAgent) parent._interrupt_requested = False parent._interrupt_message = None + parent._interrupt_event = threading.Event() parent._active_children = [] parent._active_children_lock = threading.Lock() parent.quiet_mode = True diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index 104881a03dd05..b17c8febf2d0d 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -514,27 +514,23 @@ def test_session_id_auto_generated(self): class TestInterrupt: def test_interrupt_sets_flag(self, agent): - with patch("run_agent._set_interrupt"): - agent.interrupt() - assert agent._interrupt_requested is True + agent.interrupt() + assert agent._interrupt_requested is True def test_interrupt_with_message(self, agent): - with patch("run_agent._set_interrupt"): - agent.interrupt("new question") - assert agent._interrupt_message == "new question" + agent.interrupt("new question") + assert agent._interrupt_message == "new question" def test_clear_interrupt(self, agent): - with patch("run_agent._set_interrupt"): - agent.interrupt("msg") - agent.clear_interrupt() - assert agent._interrupt_requested is False - assert agent._interrupt_message is None + agent.interrupt("msg") + agent.clear_interrupt() + assert agent._interrupt_requested is False + assert agent._interrupt_message is None def test_is_interrupted_property(self, agent): assert agent.is_interrupted is False - with patch("run_agent._set_interrupt"): - agent.interrupt() - assert agent.is_interrupted is True + agent.interrupt() + assert agent.is_interrupted is True class TestHydrateTodoStore: @@ -543,8 +539,7 @@ def test_no_todo_in_history(self, agent): {"role": "user", "content": "hello"}, {"role": "assistant", "content": "hi"}, ] - with patch("run_agent._set_interrupt"): - agent._hydrate_todo_store(history) + agent._hydrate_todo_store(history) assert not agent._todo_store.has_items() def test_recovers_from_history(self, agent): @@ -558,8 +553,7 @@ def test_recovers_from_history(self, agent): "tool_call_id": "c1", }, ] - with patch("run_agent._set_interrupt"): - agent._hydrate_todo_store(history) + agent._hydrate_todo_store(history) assert agent._todo_store.has_items() def test_skips_non_todo_tools(self, agent): @@ -570,8 +564,7 @@ def test_skips_non_todo_tools(self, agent): "tool_call_id": "c1", }, ] - with patch("run_agent._set_interrupt"): - agent._hydrate_todo_store(history) + agent._hydrate_todo_store(history) assert not agent._todo_store.has_items() def test_invalid_json_skipped(self, agent): @@ -582,8 +575,7 @@ def test_invalid_json_skipped(self, agent): "tool_call_id": "c1", }, ] - with patch("run_agent._set_interrupt"): - agent._hydrate_todo_store(history) + agent._hydrate_todo_store(history) assert not agent._todo_store.has_items() @@ -975,8 +967,7 @@ def test_interrupt_skips_remaining(self, agent): mock_msg = _mock_assistant_msg(content="", tool_calls=[tc1, tc2]) messages = [] - with patch("run_agent._set_interrupt"): - agent.interrupt() + agent.interrupt() agent._execute_tool_calls(mock_msg, messages, "task-1") # Both calls should be skipped with cancellation messages @@ -1224,8 +1215,7 @@ def test_concurrent_interrupt_before_start(self, agent): mock_msg = _mock_assistant_msg(content="", tool_calls=[tc1, tc2]) messages = [] - with patch("run_agent._set_interrupt"): - agent.interrupt() + agent.interrupt() agent._execute_tool_calls_concurrent(mock_msg, messages, "task-1") assert len(messages) == 2 @@ -1496,7 +1486,6 @@ def interrupt_side_effect(api_kwargs): patch.object(agent, "_persist_session"), patch.object(agent, "_save_trajectory"), patch.object(agent, "_cleanup_task_resources"), - patch("run_agent._set_interrupt"), patch.object( agent, "_interruptible_api_call", side_effect=interrupt_side_effect ), @@ -3449,7 +3438,10 @@ def test_counters_initialized_in_init(self): def test_counters_not_reset_in_preamble(self): """The run_conversation preamble must not zero the nudge counters.""" import inspect - src = inspect.getsource(AIAgent.run_conversation) + # ``run_conversation`` is now a thin wrapper that binds the + # per-agent interrupt event; the actual conversation logic + # lives in ``_run_conversation_locked``. + src = inspect.getsource(AIAgent._run_conversation_locked) # The preamble resets many fields (retry counts, budget, etc.) # before the main loop. Find that reset block and verify our # counters aren't in it. The reset block ends at iteration_budget. @@ -3464,7 +3456,7 @@ class TestDeadRetryCode: def test_no_unreachable_max_retries_after_backoff(self): import inspect - source = inspect.getsource(AIAgent.run_conversation) + source = inspect.getsource(AIAgent._run_conversation_locked) occurrences = source.count("if retry_count >= max_retries:") assert occurrences == 2, ( f"Expected 2 occurrences of 'if retry_count >= max_retries:' " diff --git a/tests/tools/test_interrupt_isolation.py b/tests/tools/test_interrupt_isolation.py new file mode 100644 index 0000000000000..5d680b7294107 --- /dev/null +++ b/tests/tools/test_interrupt_isolation.py @@ -0,0 +1,244 @@ +"""Regression tests for cross-agent interrupt isolation. + +Two AIAgent instances running concurrently in the same process must not +share interrupt state. The historical implementation used a single +process-wide ``threading.Event`` in ``tools/interrupt.py``, which caused +one agent's ``interrupt()`` (e.g. an SSE client disconnecting on the API +server) to abort in-flight tool calls in every other concurrently-running +agent ā€” producing truncated mid-task completions. + +These tests pin the per-agent isolation contract: + +1. Each ``AIAgent`` owns its own ``threading.Event``. +2. ``run_conversation()`` binds that event to a context variable so tools + called via ``is_interrupted()`` observe only the current agent's signal. +3. ``ThreadPoolExecutor`` workers spawned by concurrent tool execution + inherit the bound context. +4. When no event is bound (legacy CLI / test usage), + ``is_interrupted()`` and ``set_interrupt()`` fall through to a + module-level fallback so the existing single-agent API keeps working. + +Run with: + python -m pytest tests/tools/test_interrupt_isolation.py -v +""" + +import concurrent.futures +import contextvars +import threading +import time + +import pytest + +from tools.interrupt import ( + _interrupt_event, + bind_event, + is_interrupted, + set_interrupt, + unbind_event, +) + + +@pytest.fixture(autouse=True) +def _reset_global_event(): + """Each test starts and ends with a clean global fallback event.""" + _interrupt_event.clear() + yield + _interrupt_event.clear() + + +class TestPerAgentBinding: + """Direct tests of the bind_event / unbind_event API.""" + + def test_bound_event_isolated_from_global(self): + agent_event = threading.Event() + token = bind_event(agent_event) + try: + assert not is_interrupted() + agent_event.set() + assert is_interrupted() + # The module-level fallback must remain untouched. + assert not _interrupt_event.is_set() + finally: + unbind_event(token) + + def test_unbind_restores_global_fallback(self): + agent_event = threading.Event() + token = bind_event(agent_event) + unbind_event(token) + # After unbinding, is_interrupted() reflects the global event. + assert not is_interrupted() + _interrupt_event.set() + assert is_interrupted() + + def test_set_interrupt_writes_to_bound_event(self): + agent_event = threading.Event() + token = bind_event(agent_event) + try: + set_interrupt(True) + assert agent_event.is_set() + assert not _interrupt_event.is_set() + set_interrupt(False) + assert not agent_event.is_set() + finally: + unbind_event(token) + + def test_set_interrupt_falls_back_to_global_when_unbound(self): + # No bind_event() in this context. + set_interrupt(True) + assert _interrupt_event.is_set() + set_interrupt(False) + assert not _interrupt_event.is_set() + + +class TestCrossContextIsolation: + """Two independent contexts must not see each other's interrupts. + + These reproduce the API server scenario where two concurrent + /v1/chat/completions requests run agents simultaneously and one + client disconnects. + """ + + def test_two_bound_events_independent(self): + event_a = threading.Event() + event_b = threading.Event() + + results = {} + + def _agent_a(): + token = bind_event(event_a) + try: + # A is interrupted externally. + event_a.set() + results["a_sees_interrupt"] = is_interrupted() + finally: + unbind_event(token) + + def _agent_b(): + token = bind_event(event_b) + try: + # B should NOT observe A's interrupt. + results["b_sees_interrupt"] = is_interrupted() + finally: + unbind_event(token) + + # Run each in its own contextvars.copy_context() so the bindings + # cannot leak between threads via shared context references. + ctx_a = contextvars.copy_context() + ctx_b = contextvars.copy_context() + ta = threading.Thread(target=ctx_a.run, args=(_agent_a,)) + tb = threading.Thread(target=ctx_b.run, args=(_agent_b,)) + ta.start() + ta.join(timeout=2) + tb.start() + tb.join(timeout=2) + + assert results["a_sees_interrupt"] is True + assert results["b_sees_interrupt"] is False + + def test_concurrent_threadpool_workers_inherit_correct_event(self): + """Worker threads spawned via copy_context().run() see the bound event.""" + agent_event = threading.Event() + token = bind_event(agent_event) + observations = [] + + def _worker(): + observations.append(is_interrupted()) + + try: + with concurrent.futures.ThreadPoolExecutor(max_workers=4) as ex: + # Submit a worker that should NOT see an interrupt. + ctx = contextvars.copy_context() + ex.submit(ctx.run, _worker).result(timeout=2) + + # Now signal the agent's event and submit another worker. + agent_event.set() + ctx2 = contextvars.copy_context() + ex.submit(ctx2.run, _worker).result(timeout=2) + finally: + unbind_event(token) + agent_event.clear() + + assert observations == [False, True] + + +class TestAIAgentInstanceIsolation: + """End-to-end test using real AIAgent instances. + + This is the regression test for the API server multi-client scenario: + interrupting agent A must not cause is_interrupted() to return True + inside agent B's tool execution context. + """ + + def _make_minimal_agent(self): + """Build the smallest viable AIAgent without touching the network.""" + from run_agent import AIAgent + + agent = AIAgent.__new__(AIAgent) + agent._interrupt_requested = False + agent._interrupt_message = None + agent._active_children = [] + agent._active_children_lock = threading.Lock() + agent.quiet_mode = True + # The fix gives every agent its own event. + agent._interrupt_event = threading.Event() + return agent + + def test_interrupting_agent_a_does_not_affect_agent_b(self): + from run_agent import AIAgent + + agent_a = self._make_minimal_agent() + agent_b = self._make_minimal_agent() + + b_observations = {} + + def _run_agent_b(): + # Simulate run_conversation()'s contextvar binding. + token = bind_event(agent_b._interrupt_event) + try: + b_observations["before"] = is_interrupted() + # While B is "running", another thread interrupts A. + interrupter = threading.Thread( + target=AIAgent.interrupt, args=(agent_a, "client disconnect") + ) + interrupter.start() + interrupter.join(timeout=2) + b_observations["after"] = is_interrupted() + finally: + unbind_event(token) + + # Run agent B in its own copied context so the binding is local. + ctx = contextvars.copy_context() + tb = threading.Thread(target=ctx.run, args=(_run_agent_b,)) + tb.start() + tb.join(timeout=3) + + assert b_observations["before"] is False + assert b_observations["after"] is False, ( + "Agent B's tools observed agent A's interrupt ā€” cross-contamination " + "regression. The per-agent threading.Event must isolate signals." + ) + # Agent A's own state should reflect the interrupt. + assert agent_a._interrupt_requested is True + assert agent_a._interrupt_event.is_set() is True + + def test_clear_interrupt_on_one_agent_does_not_clear_other(self): + from run_agent import AIAgent + + agent_a = self._make_minimal_agent() + agent_b = self._make_minimal_agent() + + # Both agents are interrupted. + AIAgent.interrupt(agent_a, "stop A") + AIAgent.interrupt(agent_b, "stop B") + assert agent_a._interrupt_event.is_set() + assert agent_b._interrupt_event.is_set() + + # Clearing A must not clear B ā€” historically ``clear_interrupt`` + # reset the process-wide event, so a fresh agent starting up + # would silently un-interrupt every other concurrent agent. + AIAgent.clear_interrupt(agent_a) + assert not agent_a._interrupt_event.is_set() + assert agent_b._interrupt_event.is_set(), ( + "clear_interrupt() leaked across agents ā€” the per-agent event " + "is not isolated from the global fallback." + ) diff --git a/tools/interrupt.py b/tools/interrupt.py index e5c9b1e27e7c0..7b4eafdd5e786 100644 --- a/tools/interrupt.py +++ b/tools/interrupt.py @@ -1,28 +1,101 @@ -"""Shared interrupt signaling for all tools. +"""Shared interrupt signaling for tools. -Provides a global threading.Event that any tool can check to determine -if the user has requested an interrupt. The agent's interrupt() method -sets this event, and tools poll it during long-running operations. +Each ``AIAgent`` owns its own ``threading.Event`` and binds it to a +context variable for the duration of ``run_conversation()``. Tools +that poll :func:`is_interrupted` observe only the currently-bound +agent's signal, so two agents running concurrently in the same process +ā€” for example two ``/v1/chat/completions`` requests on the API server, +or a gateway runner with multiple active sessions ā€” cannot interrupt +each other's in-flight tool calls. + +A module-level fallback ``threading.Event`` is retained for backwards +compatibility with code paths that have not bound a per-agent event: +single-agent CLI usage, the existing pre-tool interrupt tests, and any +caller that imports ``_interrupt_event`` directly. + +Tools should poll the interrupt like this:: -Usage in tools: from tools.interrupt import is_interrupted if is_interrupted(): return {"output": "[interrupted]", "returncode": 130} + +Threads spawned by an agent (e.g. ``ThreadPoolExecutor`` workers used by +``_execute_tool_calls_concurrent``) must inherit the calling thread's +context via ``contextvars.copy_context().run(...)`` ā€” neither +``asyncio.loop.run_in_executor`` nor bare ``threading.Thread`` propagates +context automatically. """ +import contextvars import threading +from typing import Optional + +# Module-level fallback event. Used when no per-agent event is bound to +# ``_current_event`` below ā€” preserves the original single-agent +# semantics for CLI usage, tests that import ``_interrupt_event`` +# directly, and any code path running outside of +# ``AIAgent.run_conversation()``. _interrupt_event = threading.Event() +# ContextVar holding the interrupt event for the currently active agent. +# ``AIAgent.run_conversation()`` binds the agent's own +# ``threading.Event`` here at the start of each turn so tools observe its +# per-instance signal rather than the process-wide fallback. +_current_event: "contextvars.ContextVar[Optional[threading.Event]]" = ( + contextvars.ContextVar("hermes_current_interrupt_event", default=None) +) + + +def _active_event() -> threading.Event: + """Return the per-agent event if one is bound, else the global fallback.""" + bound = _current_event.get() + return bound if bound is not None else _interrupt_event + + +def bind_event(event: threading.Event) -> "contextvars.Token": + """Bind a per-agent interrupt event to the current context. + + Returns a token that must be passed to :func:`unbind_event` (or to + ``_current_event.reset()``) to restore the previous binding when the + agent's turn ends. + """ + return _current_event.set(event) + + +def unbind_event(token: "contextvars.Token") -> None: + """Restore the previous binding established before :func:`bind_event`. + + Safe to call on a token from a different context ā€” the failure modes + of ``ContextVar.reset`` are swallowed so cleanup paths can run + unconditionally. + """ + try: + _current_event.reset(token) + except (LookupError, ValueError): + pass + + def set_interrupt(active: bool) -> None: - """Called by the agent to signal or clear the interrupt.""" + """Signal or clear the interrupt for the currently active agent. + + Operates on the per-agent event when one is bound, otherwise on the + module-level fallback so legacy callers and tests continue to work + without modification. + """ + event = _active_event() if active: - _interrupt_event.set() + event.set() else: - _interrupt_event.clear() + event.clear() def is_interrupted() -> bool: - """Check if an interrupt has been requested. Safe to call from any thread.""" - return _interrupt_event.is_set() + """Check if an interrupt has been requested for the current agent. + + Safe to call from any thread that has inherited the agent's + contextvars (the agent's own thread, or ``ThreadPoolExecutor`` + workers wrapped in ``contextvars.copy_context().run(...)``). + """ + return _active_event().is_set()