From c1784e909326971d485ac6435d1f2d6c0948704b Mon Sep 17 00:00:00 2001 From: Gustavo Mendes <87918773+gustavosmendes@users.noreply.github.com> Date: Wed, 1 Jul 2026 14:35:36 +0530 Subject: [PATCH] fix(agent): bound concurrent tool execution with a wall-clock deadline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A tool with no internal interrupt check (read_file, web_search, or a wedged terminal backend) that never returns keeps the concurrent-tool poll loop alive forever: the loop only breaks when all futures finish or an interrupt is requested, and the 30s heartbeat resets the gateway idle monitor so idle-kill never fires. The ThreadPoolExecutor was also used as a context manager, so its __exit__ joined the hung worker with wait=True. Add a wall-clock batch deadline (HERMES_CONCURRENT_TOOL_TIMEOUT_S, default 420s — above the 360s web_extract timeout; 0/negative disables). When it fires: cancel pending futures, signal an interrupt to the worker threads, abandon the executor (shutdown wait=False, cancel_futures=True) so hung threads aren't joined, and return a per-tool 'timed out' result for the unfinished calls while still surfacing the finished ones. Also fixes the latent futures.index(f) lookup (ambiguous with duplicate futures) by tracking a future->index map. Salvaged from #54562. Co-authored-by: Gustavo Mendes <87918773+gustavosmendes@users.noreply.github.com> --- agent/tool_executor.py | 106 ++++++++++++++++++++++++++++-- tests/run_agent/test_run_agent.py | 44 +++++++++++++ 2 files changed, 143 insertions(+), 7 deletions(-) diff --git a/agent/tool_executor.py b/agent/tool_executor.py index 167a60946b2c..a42e8f3240d3 100644 --- a/agent/tool_executor.py +++ b/agent/tool_executor.py @@ -69,6 +69,27 @@ def _budget_for_agent(agent) -> BudgetConfig: # Maximum number of concurrent worker threads for parallel tool execution. # Mirrors the constant in ``run_agent`` for tests/imports that look here. _MAX_TOOL_WORKERS = 8 +# Keep this above the stock auxiliary.web_extract timeout (360s) so the batch +# guard does not preempt a slow-but-valid summarization attempt. +_DEFAULT_CONCURRENT_TOOL_TIMEOUT_S = 420.0 + + +def _resolve_concurrent_tool_timeout() -> float | None: + raw = os.getenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "").strip() + if not raw: + return _DEFAULT_CONCURRENT_TOOL_TIMEOUT_S + try: + value = float(raw) + except ValueError: + logger.warning( + "invalid HERMES_CONCURRENT_TOOL_TIMEOUT_S=%r; using %.0fs", + raw, + _DEFAULT_CONCURRENT_TOOL_TIMEOUT_S, + ) + return _DEFAULT_CONCURRENT_TOOL_TIMEOUT_S + if value <= 0: + return None + return value def _flush_session_db_after_tool_progress( @@ -611,9 +632,15 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe if block_result is None ] futures = [] + future_to_index = {} + timed_out_indices: set[int] = set() + timeout_s = _resolve_concurrent_tool_timeout() + deadline = time.monotonic() + timeout_s if timeout_s is not None else None if runnable_calls: max_workers = min(len(runnable_calls), _MAX_TOOL_WORKERS) - with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: + executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) + abandon_executor = False + try: for submit_index, (i, tc, name, args) in enumerate(runnable_calls): # Propagate the agent turn's ContextVars (e.g. # _approval_session_key) AND thread-local approval/sudo @@ -649,6 +676,7 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe ) break futures.append(f) + future_to_index[f] = i # Wait for all to complete with periodic heartbeats so the # gateway's inactivity monitor doesn't kill us during long @@ -658,18 +686,61 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe _conc_start = time.time() _interrupt_logged = False while True: - done, not_done = concurrent.futures.wait( - futures, timeout=5.0, - ) + wait_timeout = 5.0 + if deadline is not None: + remaining = deadline - time.monotonic() + if remaining <= 0: + done, not_done = set(), { + f for f in futures if not f.done() + } + else: + wait_timeout = min(wait_timeout, remaining) + done, not_done = concurrent.futures.wait( + futures, timeout=wait_timeout, + ) + else: + done, not_done = concurrent.futures.wait( + futures, timeout=wait_timeout, + ) if not not_done: break + if deadline is not None and time.monotonic() >= deadline: + abandon_executor = True + timed_out_indices = { + future_to_index[f] + for f in not_done + if f in future_to_index + } + _still_running = [ + parsed_calls[i][1] + for i in timed_out_indices + ] + logger.warning( + "concurrent tool batch timed out after %.1fs; " + "%d tool(s) still running: %s", + timeout_s, + len(timed_out_indices), + ", ".join(_still_running[:5]), + ) + for f in not_done: + f.cancel() + with agent._tool_worker_threads_lock: + worker_tids = list(agent._tool_worker_threads) + for tid in worker_tids: + try: + _ra()._set_interrupt(True, tid) + except Exception: + pass + break + # Check for interrupt — the per-thread interrupt signal # already causes individual tools (terminal, execute_code) # to abort, but tools without interrupt checks (web_search, # read_file) will run to completion. Cancel any futures # that haven't started yet so we don't block on them. if agent._interrupt_requested: + abandon_executor = True if not _interrupt_logged: _interrupt_logged = True agent._vprint( @@ -688,14 +759,19 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe # Heartbeat every ~30s (6 × 5s poll intervals) if _conc_elapsed > 0 and _conc_elapsed % 30 < 6: _still_running = [ - parsed_calls[futures.index(f)][1] + parsed_calls[future_to_index[f]][1] for f in not_done - if f in futures + if f in future_to_index ] agent._touch_activity( f"concurrent tools running ({_conc_elapsed}s, " f"{len(not_done)} remaining: {', '.join(_still_running[:3])})" ) + finally: + executor.shutdown( + wait=not abandon_executor, + cancel_futures=abandon_executor, + ) finally: if spinner: # Build a summary message for the spinner stop @@ -707,7 +783,23 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe for i, (tc, name, args, middleware_trace, block_result, blocked_by_guardrail) in enumerate(parsed_calls): r = results[i] blocked = False - if r is None: + if i in timed_out_indices: + suffix = f"{timeout_s:.1f}s" if timeout_s is not None else "the configured timeout" + function_result = f"Error executing tool '{name}': timed out after {suffix}" + _emit_terminal_post_tool_call( + agent, + function_name=name, + function_args=args, + result=function_result, + effective_task_id=effective_task_id, + tool_call_id=getattr(tc, "id", "") or "", + status="timeout", + error_type="tool_timeout", + error_message=function_result, + middleware_trace=list(middleware_trace), + ) + tool_duration = float(timeout_s or 0.0) + elif r is None: # Tool was cancelled (interrupt) or thread didn't return if agent._interrupt_requested: function_result = f"[Tool execution cancelled — {name} was skipped due to user interrupt]" diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index 1ed8e2a731d3..e97e902b0471 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -2723,6 +2723,9 @@ class TestConcurrentToolExecution: def submit(self, *args, **kwargs): raise RuntimeError("cannot schedule new futures after interpreter shutdown") + def shutdown(self, *args, **kwargs): + pass + tc1 = _mock_tool_call(name="web_search", arguments='{"q": "alpha"}', call_id="c1") tc2 = _mock_tool_call(name="web_search", arguments='{"q": "beta"}', call_id="c2") mock_msg = _mock_assistant_msg(content="", tool_calls=[tc1, tc2]) @@ -2736,6 +2739,47 @@ class TestConcurrentToolExecution: assert messages[1]["tool_call_id"] == "c2" assert all("Python interpreter is shutting down" in m["content"] for m in messages) + def test_concurrent_timeout_returns_finished_tools_without_hanging(self, agent, monkeypatch): + """A wedged worker must not freeze the whole concurrent tool batch.""" + import threading + import time as _time + + monkeypatch.setenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "0.1") + blocker = threading.Event() + tc1 = _mock_tool_call(name="web_search", arguments='{"q": "fast"}', call_id="c1") + tc2 = _mock_tool_call(name="web_search", arguments='{"q": "slow"}', call_id="c2") + mock_msg = _mock_assistant_msg(content="", tool_calls=[tc1, tc2]) + messages = [] + flushed = [] + + def fake_handle(name, args, task_id, **kwargs): + if args.get("q") == "slow": + blocker.wait(5) + return "late" + return "fast-result" + + def record_flush(flush_messages, conversation_history=None): + flushed.append([m.copy() for m in flush_messages if m.get("role") == "tool"]) + + agent._flush_messages_to_session_db = MagicMock(side_effect=record_flush) + + start = _time.monotonic() + try: + with patch("run_agent.handle_function_call", side_effect=fake_handle): + agent._execute_tool_calls_concurrent(mock_msg, messages, "task-1") + finally: + blocker.set() + + assert _time.monotonic() - start < 1.0 + assert len(messages) == 2 + assert messages[0]["tool_call_id"] == "c1" + assert "fast-result" in messages[0]["content"] + assert messages[1]["tool_call_id"] == "c2" + assert "timed out after" in messages[1]["content"] + assert [batch[-1]["tool_call_id"] for batch in flushed] == ["c1", "c2"] + assert "fast-result" in flushed[0][-1]["content"] + assert "timed out after" in flushed[1][-1]["content"] + def test_concurrent_interrupt_before_start(self, agent): """If interrupt is requested before concurrent execution, all tools are skipped.""" tc1 = _mock_tool_call(name="web_search", arguments='{}', call_id="c1")