fix(interrupt): per-agent interrupt event to prevent cross-session contamination

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.

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

tests/run_agent/test_interrupt_propagation.py

@@ -29,13 +29,15 @@ 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
 
         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,13 +140,15 @@ 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
 
         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

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

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:' "