ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

fix(agent): keep the start-order gate under the batch deadline and abort abandoned workers

Browse Source 
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]".
This commit is contained in:
kshitij
2026-08-06 03:34:51 +05:30
committed by kshitij
parent 5d83400f89
commit 042a2cf3d7
2 changed files with 341 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/tool_executor.py
+90 -8
View File
@@ -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
tests/run_agent/test_start_order_gate.py
+251
View File
@@ -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}"
)