From 22a137ed407a5549dd00d24e419a8527e92f5137 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Wed, 1 Jul 2026 14:43:28 +0530 Subject: [PATCH] fix(agent): prefer late-completing real result over timeout message (review) Review follow-up on the concurrent-tool deadline salvage. timed_out_indices is snapshotted from not_done at the deadline; a worker can still finish and write results[i] in the window before the post-execution result loop reads it. The loop unconditionally replaced results[i] with a fabricated 'timed out' message for any snapshotted index, discarding a genuinely-successful (just-late) result. Gate the timeout message on 'and r is None' so a real result always wins. Add a regression test that forces the snapshot-vs-result-loop race deterministically (mutation-checked: reverting the guard fails it). Also document the intentional detached-worker leak at the executor abandon site. --- agent/tool_executor.py | 11 ++++++++- tests/run_agent/test_run_agent.py | 39 +++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/agent/tool_executor.py b/agent/tool_executor.py index a42e8f3240d..2b9b5598dac 100644 --- a/agent/tool_executor.py +++ b/agent/tool_executor.py @@ -768,6 +768,11 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe f"{len(not_done)} remaining: {', '.join(_still_running[:3])})" ) finally: + # On abandon (interrupt or deadline) we intentionally do NOT + # join hung workers: wait=False returns immediately and + # cancel_futures drops queued-but-unstarted work. A wedged tool + # thread is left running detached — the deliberate tradeoff vs. + # deadlocking the whole batch. Normal completion joins (wait=True). executor.shutdown( wait=not abandon_executor, cancel_futures=abandon_executor, @@ -783,7 +788,11 @@ 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 i in timed_out_indices: + # A worker can finish and write results[i] in the window between the + # deadline snapshot (timed_out_indices, taken from not_done) and this + # loop. Prefer that real result over a fabricated timeout message — the + # tool genuinely succeeded, just slightly late. + if i in timed_out_indices and r is None: 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( diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index e97e902b047..2f334ec87fa 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -2780,6 +2780,45 @@ class TestConcurrentToolExecution: assert "fast-result" in flushed[0][-1]["content"] assert "timed out after" in flushed[1][-1]["content"] + def test_concurrent_timeout_prefers_late_real_result_over_timeout_message(self, agent, monkeypatch): + """A worker that finishes in the window between the deadline snapshot + and the result loop must keep its real result, not be overwritten with + a fabricated 'timed out' message (late-completion race).""" + import concurrent.futures as _cf + + monkeypatch.setenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "0.1") + tc1 = _mock_tool_call(name="web_search", arguments='{"q": "a"}', call_id="c1") + tc2 = _mock_tool_call(name="web_search", arguments='{"q": "b"}', call_id="c2") + mock_msg = _mock_assistant_msg(content="", tool_calls=[tc1, tc2]) + messages = [] + + # Both tools return instantly, so results[*] are populated almost + # immediately. We still force the deadline path by making the FIRST + # wait() report everything as not-done (after crossing the deadline), + # so the loop snapshots both as timed-out even though the workers have + # in fact already written their results. The fix must surface the real + # results, not the timeout message. + real_wait = _cf.wait + calls = {"n": 0} + + def fake_wait(fs, timeout=None): + calls["n"] += 1 + if calls["n"] == 1: + import time as _t + _t.sleep(0.15) # ensure monotonic() >= deadline + return set(), set(fs) + return real_wait(fs, timeout=timeout) + + with patch("agent.tool_executor.concurrent.futures.wait", side_effect=fake_wait), \ + patch("run_agent.handle_function_call", side_effect=lambda name, args, task_id, **k: f"real-{args.get('q')}"): + agent._execute_tool_calls_concurrent(mock_msg, messages, "task-1") + + assert len(messages) == 2 + joined = " ".join(m["content"] for m in messages) + assert "timed out after" not in joined, "late-completing real results must not be discarded" + assert "real-a" in messages[0]["content"] + assert "real-b" in messages[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")