From 042a2cf3d70d456128eabd3f375c2f4c9d96d2e0 Mon Sep 17 00:00:00 2001 From: kshitij Date: Thu, 6 Aug 2026 03:34:51 +0530 Subject: [PATCH] fix(agent): keep the start-order gate under the batch deadline and abort abandoned workers Follow-up to the salvaged start-order gate bound. Two gaps remained, both reachable through the same knob. 1. The gate bound ignored the batch deadline it sits under. With HERMES_CONCURRENT_TOOL_TIMEOUT_S below 120s the deadline fired first, so the parked tools were still reported as "timed out" without ever running -- the exact bug the bound exists to fix. The gate now clamps to min(120s, batch_timeout / 2), matching the sibling constant's documented habit of relating the two timeouts. 2. A gate-parked worker released purely by its own timeout could wake up after the batch was abandoned and dispatch its tool anyway: wasted work whose result nobody reads, a duplicate post_tool_call for a tool_call_id the turn already closed as timeout, and agent._current_tool left pointing at a dead tool for the rest of the session (the main thread's reset already ran). Abandonment is now a first-class wakeup: both abandon sites set an event and notify the condition, and a released worker raises _BatchAbandoned instead of dispatching. Parked threads are reclaimed in milliseconds rather than one full gate timeout plus a tool runtime. Also names the tool in the gate-timeout warning. The closure's function_name binds the last-parsed tool, so logging it directly would have printed the wrong name; it is threaded through _begin_in_order instead. Measured, 3-tool batch with the first tool wedged during dispatch: main PR as-is with this commit dispatched in batch 0 0 tool_b, tool_c dispatched after return 0 2 (ghost) 0 _current_tool leaked no "tool_b" no Adds tests/run_agent/test_start_order_gate.py (3 tests). Mutation-checked against the parent commit: the starvation guard passes there (it binds the salvaged fix), while the deadline-clamp and abandonment guards both fail, reproducing the ghost dispatch as "tool(s) dispatched after the batch was abandoned: [tool_a, tool_b]". --- agent/tool_executor.py | 98 ++++++++- tests/run_agent/test_start_order_gate.py | 251 +++++++++++++++++++++++ 2 files changed, 341 insertions(+), 8 deletions(-) create mode 100644 tests/run_agent/test_start_order_gate.py diff --git a/agent/tool_executor.py b/agent/tool_executor.py index 0014686bbc..6b545f9e6f 100644 --- a/agent/tool_executor.py +++ b/agent/tool_executor.py @@ -104,6 +104,14 @@ _DEFAULT_CONCURRENT_TOOL_TIMEOUT_S = 420.0 _START_ORDER_GATE_TIMEOUT_S = 120.0 +class _BatchAbandoned(BaseException): + """Raised inside a worker when the batch was abandoned before dispatch. + + Derives from BaseException so intermediate ``except Exception`` handlers in + the middleware chain cannot swallow it and dispatch the tool anyway. + """ + + def _parse_tool_arguments(raw_arguments: Any) -> tuple[dict, Optional[str]]: """Parse model-emitted arguments without repairing or coercing them.""" try: @@ -806,9 +814,31 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe start_condition = threading.Condition() next_start_order = 0 + # Set once the batch is abandoned (deadline or interrupt) so a worker parked + # at the start-order gate exits immediately instead of waking up minutes + # later and dispatching a tool the turn has already reported as timed out. + batch_abandoned = threading.Event() authorization_gate = _ConcurrentToolAuthorizationGate() - def _begin_in_order(order: int, callback=None) -> None: + def _abandon_batch() -> None: + """Release every gate-parked worker so none dispatches post-abandon.""" + batch_abandoned.set() + with start_condition: + start_condition.notify_all() + + # The gate bound must sit UNDER the batch deadline, otherwise the deadline + # fires first and the parked workers are still falsely reported as timed + # out without ever starting — the very bug this gate timeout fixes. A + # disabled deadline (None) keeps the stock bound rather than waiting forever. + def _start_order_gate_timeout(batch_timeout: float | None) -> float: + if batch_timeout is None: + return _START_ORDER_GATE_TIMEOUT_S + return min(_START_ORDER_GATE_TIMEOUT_S, batch_timeout / 2) + + def _begin_in_order( + order: int, callback=None, *, tool_name: str = "", gate_timeout: float | None = None + ) -> bool: + """Serialize dispatch by submit order. Returns False if abandoned.""" nonlocal next_start_order with start_condition: # Bounded wait: a tool that wedges during its dispatch must not @@ -823,15 +853,24 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe # starvation. The >= predicate (rather than ==) lets one worker's # timeout-jump release every skipped worker immediately instead # of each burning its own full timeout; max() keeps the counter - # monotonic when workers advance out of order. + # monotonic when workers advance out of order. batch_abandoned + # short-circuits the wait so an abandoned batch releases its + # parked workers in milliseconds instead of one gate timeout. in_order = start_condition.wait_for( - lambda: next_start_order >= order, - timeout=_START_ORDER_GATE_TIMEOUT_S, + lambda: next_start_order >= order or batch_abandoned.is_set(), + timeout=( + _START_ORDER_GATE_TIMEOUT_S if gate_timeout is None else gate_timeout + ), ) + if batch_abandoned.is_set(): + # Do not run the callback or advance the counter: the turn has + # already synthesized this tool's result and moved on. + return False if not in_order: logger.warning( - "start-order gate timed out (order=%d next=%d); " + "start-order gate timed out for %s (order=%d next=%d); " "proceeding out of order", + tool_name or "tool", order, next_start_order, ) @@ -841,6 +880,12 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe finally: next_start_order = max(next_start_order, order + 1) start_condition.notify_all() + return True + + # Resolved before the workers are defined so the start-order gate can clamp + # its own bound against the batch deadline it must stay under. + timeout_s = _resolve_concurrent_tool_timeout() + gate_timeout_s = _start_order_gate_timeout(timeout_s) # Touch activity before launching workers so the gateway knows # we're executing tools (not stuck). @@ -894,9 +939,18 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe if start_advanced: return try: - _begin_in_order(start_order, callback) + proceed = _begin_in_order( + start_order, + callback, + tool_name=function_name, + gate_timeout=gate_timeout_s, + ) finally: start_advanced = True + if not proceed: + # Batch already abandoned: the turn synthesized this tool's + # result and moved on. Abort instead of dispatching late. + raise _BatchAbandoned(function_name) try: try: @@ -931,6 +985,17 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe middleware_trace = managed.middleware_trace blocked = managed.blocked dispatched = managed.dispatched + except _BatchAbandoned: + # The batch was abandoned while we were parked at the start-order + # gate. The main thread already synthesized this tool's result + # (timeout/cancelled) and moved on, so write nothing: a late + # results[index] write, post_tool_call emit, or progress print + # would double-report a tool_call_id the turn already closed. + logger.info( + "tool %s abandoned at start-order gate; skipping dispatch", + function_name, + ) + return except KeyboardInterrupt: try: agent.interrupt("keyboard interrupt") @@ -987,7 +1052,13 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe middleware_trace, ) finally: - _advance_start() + # Teardown advance: keep the counter moving for any later-ordered + # worker. Never let the abandonment signal escape from here — the + # worker is already unwinding and the turn owns the result. + try: + _advance_start() + except _BatchAbandoned: + pass # 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 @@ -1019,7 +1090,6 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe 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 = _max_workers_for_tool_batch(runnable_calls) @@ -1138,6 +1208,10 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe ) for f in not_done: f.cancel() + # Release gate-parked workers before the interrupt + # fan-out so none of them wakes up later and dispatches + # a tool this loop just reported as timed out. + _abandon_batch() with agent._tool_worker_threads_lock: worker_tids = list(agent._tool_worker_threads) for tid in worker_tids: @@ -1163,6 +1237,9 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe ) for f in not_done: f.cancel() + # Release gate-parked workers so they abort instead of + # dispatching after the turn was already interrupted. + _abandon_batch() # Give already-running tools a moment to notice the # per-thread interrupt signal and exit gracefully. concurrent.futures.wait(not_done, timeout=3.0) @@ -1181,6 +1258,11 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe f"{len(not_done)} remaining: {', '.join(_still_running[:3])})" ) finally: + # Belt-and-braces: any exit from the wait loop that abandoned + # the batch must release gate-parked workers, including the + # exception path that never reaches the branches above. + if abandon_executor: + _abandon_batch() # 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 diff --git a/tests/run_agent/test_start_order_gate.py b/tests/run_agent/test_start_order_gate.py new file mode 100644 index 0000000000..94dc84d148 --- /dev/null +++ b/tests/run_agent/test_start_order_gate.py @@ -0,0 +1,251 @@ +"""Tests for the concurrent start-order gate (PR #79571 / issue #79569). + +The gate serializes tool dispatch by submit order so approval prompts and +progress output appear in the order the model requested them. A tool that +wedges *during dispatch* must not park every later-ordered worker forever: +before the bound existed, those parked tools never started, the batch deadline +then falsely reported them as "timed out", and the parked threads leaked +permanently (``f.cancel()`` cannot stop a running thread and nothing ever +notified the condition again). +""" + +import json +import threading +import time +from unittest.mock import MagicMock + +import pytest + + +@pytest.fixture(autouse=True) +def _isolate_hermes(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + (tmp_path / ".hermes").mkdir(exist_ok=True) + + +def _make_agent(monkeypatch): + """Minimal AIAgent-like stub, mirroring test_concurrent_interrupt.py.""" + monkeypatch.setenv("OPENROUTER_API_KEY", "") + monkeypatch.setenv("HERMES_INFERENCE_PROVIDER", "") + import run_agent as _ra + + class _Stub: + _interrupt_requested = False + _interrupt_message = None + _execution_thread_id = threading.current_thread().ident + _interrupt_thread_signal_pending = False + log_prefix = "" + quiet_mode = True + verbose_logging = False + log_prefix_chars = 200 + _checkpoint_mgr = MagicMock(enabled=False) + tool_progress_callback = None + tool_start_callback = None + tool_complete_callback = None + tool_progress_mode = "off" + _todo_store = MagicMock() + _session_db = None + valid_tool_names = set() + _turns_since_memory = 0 + _iters_since_skill = 0 + _current_tool = None + _last_activity = 0 + _print_fn = print + session_id = "" + _current_turn_id = "" + _current_api_request_id = "" + _active_children: list = [] + + def __init__(self): + self._tool_worker_threads: set = set() + self._tool_worker_threads_lock = threading.Lock() + self._active_children_lock = threading.Lock() + + def _touch_activity(self, desc): + self._last_activity = time.time() + + def _vprint(self, msg, force=False): + pass + + def _safe_print(self, msg): + pass + + def _should_emit_quiet_tool_messages(self): + return False + + def _should_start_quiet_spinner(self): + return False + + def _has_stream_consumers(self): + return False + + def _tool_result_content_for_active_model(self, name, result): + return result + + def _record_file_mutation_result(self, *a, **kw): + pass + + stub = _Stub() + stub._subdirectory_hints = MagicMock() + stub._subdirectory_hints.check_tool_call = lambda *a, **kw: None + stub._flush_messages_to_session_db = lambda *a, **kw: None + stub._append_guardrail_observation = lambda name, result, *a, **kw: result + stub._execute_tool_calls_concurrent = ( + _ra.AIAgent._execute_tool_calls_concurrent.__get__(stub) + ) + stub.interrupt = _ra.AIAgent.interrupt.__get__(stub) + stub.clear_interrupt = _ra.AIAgent.clear_interrupt.__get__(stub) + stub._apply_pending_steer_to_tool_results = lambda *a, **kw: None + stub._guardrail_block_result = lambda d: json.dumps({"error": "blocked"}) + return stub + + +class _FakeToolCall: + def __init__(self, name, call_id): + self.function = MagicMock(name=name, arguments="{}") + self.function.name = name + self.id = call_id + + +class _FakeAssistantMsg: + def __init__(self, tool_calls): + self.tool_calls = tool_calls + + +def _wedge_first_tool(agent, wedged_name, dispatched, stop): + """Wedge ``wedged_name`` during dispatch; record every real dispatch.""" + + def _before_call(name, args): + if name == wedged_name: + stop.wait(30) # released in test teardown, not by the gate + return MagicMock(allows_execution=True) + + agent._tool_guardrails = MagicMock() + agent._tool_guardrails.before_call = _before_call + + def _invoke(name, *a, **kw): + dispatched.append((name, time.monotonic())) + return json.dumps({"ok": name}) + + agent._invoke_tool = MagicMock(side_effect=_invoke) + + +def test_wedged_dispatch_does_not_starve_later_tools(monkeypatch): + """A tool wedged during dispatch must not block the rest of the batch. + + Before the gate was bounded, tool_b/tool_c never started and were falsely + reported as "timed out" despite doing zero work. + """ + import agent.tool_executor as te + + agent = _make_agent(monkeypatch) + monkeypatch.setattr(te, "_START_ORDER_GATE_TIMEOUT_S", 0.3) + monkeypatch.setattr(te, "_resolve_concurrent_tool_timeout", lambda: 6.0) + + dispatched: list = [] + stop = threading.Event() + _wedge_first_tool(agent, "tool_a", dispatched, stop) + + msg = _FakeAssistantMsg([ + _FakeToolCall("tool_a", "tc_a"), + _FakeToolCall("tool_b", "tc_b"), + _FakeToolCall("tool_c", "tc_c"), + ]) + messages: list = [] + try: + agent._execute_tool_calls_concurrent(msg, messages, "task") + finally: + stop.set() + + names = [n for n, _ in dispatched] + assert "tool_b" in names and "tool_c" in names, ( + f"later-ordered tools were starved by the wedged dispatch: {names}" + ) + + by_tool = {m.get("name"): m["content"] for m in messages} + for late in ("tool_b", "tool_c"): + assert "timed out" not in str(by_tool[late]), ( + f"{late} ran but was still reported as timed out: {by_tool[late]!r}" + ) + + +def test_gate_timeout_stays_under_the_batch_deadline(monkeypatch): + """The gate bound must clamp below the batch deadline it sits under. + + With a batch timeout shorter than the stock 120s gate, an unclamped gate + expires only after the deadline already blamed the parked tools — the exact + bug the bound exists to fix. + """ + import agent.tool_executor as te + + agent = _make_agent(monkeypatch) + monkeypatch.setattr(te, "_resolve_concurrent_tool_timeout", lambda: 2.0) + + dispatched: list = [] + stop = threading.Event() + _wedge_first_tool(agent, "tool_a", dispatched, stop) + + msg = _FakeAssistantMsg([ + _FakeToolCall("tool_a", "tc_a"), + _FakeToolCall("tool_b", "tc_b"), + ]) + messages: list = [] + try: + agent._execute_tool_calls_concurrent(msg, messages, "task") + finally: + stop.set() + + assert "tool_b" in [n for n, _ in dispatched], ( + "gate outlived the batch deadline, so tool_b was blamed without running" + ) + + +def test_abandoned_batch_does_not_dispatch_late(monkeypatch): + """A gate-parked worker must abort once the batch is abandoned. + + Otherwise it wakes up after the turn already synthesized its result and + dispatches the tool anyway — wasted work plus a duplicate post_tool_call + for a tool_call_id the turn already closed. + """ + import agent.tool_executor as te + + agent = _make_agent(monkeypatch) + # Long gate: only the abandonment signal can release the parked workers. + monkeypatch.setattr(te, "_START_ORDER_GATE_TIMEOUT_S", 30.0) + monkeypatch.setattr(te, "_resolve_concurrent_tool_timeout", lambda: 60.0) + + dispatched: list = [] + stop = threading.Event() + _wedge_first_tool(agent, "tool_a", dispatched, stop) + + def _fire_interrupt(): + time.sleep(0.5) + agent.interrupt("user pressed stop") + + threading.Thread(target=_fire_interrupt, daemon=True).start() + + msg = _FakeAssistantMsg([ + _FakeToolCall("tool_a", "tc_a"), + _FakeToolCall("tool_b", "tc_b"), + ]) + messages: list = [] + started = time.monotonic() + try: + agent._execute_tool_calls_concurrent(msg, messages, "task") + returned_at = time.monotonic() + finally: + stop.set() + agent.clear_interrupt() + + assert returned_at - started < 25.0, ( + "batch waited out the full gate timeout instead of releasing parked " + "workers on abandonment" + ) + + # Give a would-be late worker room to misbehave. + time.sleep(1.0) + late = [(n, t) for n, t in dispatched if t > returned_at] + assert not late, f"tool(s) dispatched after the batch was abandoned: {late}" + assert agent._current_tool is None, ( + f"_current_tool left pointing at a dead tool: {agent._current_tool!r}" + )