Interrupt state leak fixes and recovery handling

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #35382 fix(tools): wrap _run_tool cleanup in finally to prevent interrupt state leak (#35309)

  merged to main

  files: agent/tool_executor.py, tests/tools/test_interrupt.py

  Show the diff

  • agent/tool_executor.py

    @@ -306,33 +306,38 @@ def _run_tool(index, tool_call, function_name, function_args):
             # submit site below (GHSA-qg5c-hvr5-hjgr, #13617).
             start = time.time()
             try:
    -            result = agent._invoke_tool(
    -                function_name,
    -                function_args,
    -                effective_task_id,
    -                tool_call.id,
    -                messages=messages,
    -                pre_tool_block_checked=True,
    -            )
    -        except Exception as tool_error:
    -            result = f"Error executing tool '{function_name}': {tool_error}"
    -            logger.error("_invoke_tool raised for %s: %s", function_name, tool_error, exc_info=True)
    -        duration = time.time() - start
    -        is_error, _ = _detect_tool_failure(function_name, result)
    -        if is_error:
    -            logger.info("tool %s failed (%.2fs): %s", function_name, duration, result[:200])
    -        else:
    -            logger.info("tool %s completed (%.2fs, %d chars)", function_name, duration, len(result))
    -        results[index] = (function_name, function_args, result, duration, is_error, False)
    -        # Tear down worker-tid tracking.  Clear any interrupt bit we may
    -        # have set so the next task scheduled onto this recycled tid
    -        # starts with a clean slate.
    -        with agent._tool_worker_threads_lock:
    -            agent._tool_worker_threads.discard(_worker_tid)
    -        try:
    -            _ra()._set_interrupt(False, _worker_tid)
    -        except Exception:
    -            pass
    +            try:
    +                result = agent._invoke_tool(
    +                    function_name,
    +                    function_args,
    +                    effective_task_id,
    +                    tool_call.id,
    +                    messages=messages,
    +                    pre_tool_block_checked=True,
    +                )
    +            except Exception as tool_error:
    +                result = f"Error executing tool '{function_name}': {tool_error}"
    +                logger.error("_invoke_tool raised for %s: %s", function_name, tool_error, exc_info=True)
    +            duration = time.time() - start
    +            is_error, _ = _detect_tool_failure(function_name, result)
    +            if is_error:
    +                logger.info("tool %s failed (%.2fs): %s", function_name, duration, result[:200])
    +            else:
    +                logger.info("tool %s completed (%.2fs, %d chars)", function_name, duration, len(result))
    +            results[index] = (function_name, function_args, result, duration, is_error, False)
    +        finally:
    +            # Tear down worker-tid tracking.  Clear any interrupt bit we may
    +            # have set so the next task scheduled onto this recycled tid
    +            # starts with a clean slate.  This MUST be in a finally block
    +            # because BaseException subclasses (CancelledError, KeyboardInterrupt)
    +            # bypass ``except Exception`` and would otherwise leak the tid
    +            # into _interrupted_threads, poisoning the recycled thread.
    +            with agent._tool_worker_threads_lock:
    +                agent._tool_worker_threads.discard(_worker_tid)
    +            try:
    +                _ra()._set_interrupt(False, _worker_tid)
    +            except Exception:
    +                pass
     
         # Start spinner for CLI mode (skip when TUI handles tool progress)
         spinner = None

  • tests/tools/test_interrupt.py

    @@ -203,6 +203,83 @@ def _run():
             assert "interrupted" in result_holder["value"]["output"].lower()
     
     
    +# ---------------------------------------------------------------------------
    +# Regression: _run_tool cleanup on BaseException (issue #35309)
    +# ---------------------------------------------------------------------------
    +
    +class TestRunToolCleanupOnBaseException:
    +    """Verify that _run_tool cleans up _interrupted_threads even when
    +    _invoke_tool raises a BaseException (e.g. CancelledError).
    +
    +    Regression test for #35309: without the finally block, a BaseException
    +    bypasses ``except Exception``, leaking the worker tid into
    +    _interrupted_threads.  ThreadPoolExecutor recycles tids, so the next
    +    tool scheduled on the same thread is instantly "interrupted".
    +    """
    +
    +    def test_cleanup_on_base_exception(self):
    +        from unittest.mock import MagicMock, patch
    +        import types
    +        from tools.interrupt import set_interrupt, is_interrupted, _interrupted_threads, _lock
    +
    +        # Clear global state
    +        with _lock:
    +            _interrupted_threads.clear()
    +
    +        # Build a minimal mock agent with the attributes _run_tool needs
    +        agent = MagicMock()
    +        agent._interrupt_requested = False
    +        agent._tool_worker_threads = set()
    +        agent._tool_worker_threads_lock = threading.Lock()
    +
    +        # _set_interrupt delegates to the real module
    +        def _mock_set_interrupt(active, tid=None):
    +            set_interrupt(active, tid)
    +        agent._set_interrupt = _mock_set_interrupt
    +
    +        # _invoke_tool raises BaseException (simulating CancelledError)
    +        agent._invoke_tool = MagicMock(side_effect=BaseException("simulated CancelledError"))
    +
    +        # Bind the real concurrent method so we get _run_tool
    +        from run_agent import AIAgent
    +        agent._execute_tool_calls_concurrent = types.MethodType(
    +            AIAgent._execute_tool_calls_concurrent, agent
    +        )
    +
    +        # Build a single tool call
    +        tc = MagicMock()
    +        tc.id = "tc_base_exc"
    +        tc.function.name = "dummy_tool"
    +        tc.function.arguments = "{}"
    +
    +        assistant_msg = MagicMock()
    +        assistant_msg.tool_calls = [tc]
    +
    +        # _execute_tool_calls_concurrent will submit _run_tool to a
    +        # ThreadPoolExecutor.  The BaseException propagates out of the
    +        # worker, but the finally block should still clean up.
    +        try:
    +            agent._execute_tool_calls_concurrent(assistant_msg, [], "default")
    +        except Exception:
    +            pass  # ThreadPoolExecutor may re-raise
    +
    +        # After the worker finishes (even with BaseException), the worker
    +        # tid should have been removed from _interrupted_threads and
    +        # _tool_worker_threads.
    +        assert len(agent._tool_worker_threads) == 0, (
    +            f"_tool_worker_threads not cleaned up: {agent._tool_worker_threads}"
    +        )
    +
    +        # Verify no stale tid is left in the global interrupt set.  The
    +        # worker thread is recycled by ThreadPoolExecutor, so a leaked tid
    +        # would poison the next task on that thread.  We cleared the set at
    +        # the start and never set any interrupt ourselves, so a leak from
    +        # _run_tool is the only way an entry could land here.
    +        with _lock:
    +            leaked = set(_interrupted_threads)
    +        assert leaked == set(), f"leaked tids in _interrupted_threads: {leaked}"
    +
    +
     # ---------------------------------------------------------------------------
     # Manual smoke test checklist (not automated)
     # ---------------------------------------------------------------------------

• #35381 fix(gateway): recover model on post-interrupt recovery turn; gate fallback status

  merged to main

  files: agent/conversation_loop.py, gateway/run.py, run_agent.py, tests/gateway/test_empty_model_recovery.py