fix(agent): bound concurrent tool execution with a wall-clock deadline

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

@@ -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]"

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")