a2cc867870
test_sequential_tool_timeout_emits_result_and_continues asserted first_started.is_set() immediately after the executor returned — but on a starved CI runner the 0.05s tool timeout can expire before the hung call's worker THREAD is even scheduled, and the timeout path's future.cancel() then legitimately prevents that dispatch entirely. That is the exact behavior under test (executor refuses to wait), not a failure. The test now asserts the real contract: executor returns fast, the NEXT tool actually dispatched, and the message stream carries both call ids in order with the timeout result first. Verified both ways with a thread-start-delay sabotage run: old assert fails under delay, new assertions pass under delay and still fail if the executor stops continuing past a hung call.
244 lines
8.4 KiB
Python
244 lines
8.4 KiB
Python
"""Sequential tool calls recover when one dispatch never returns."""
|
|
|
|
import json
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from agent.tool_executor import execute_tool_calls_sequential
|
|
from run_agent import AIAgent
|
|
from tools.clarify_gateway import resolve_clarify_timeout
|
|
|
|
|
|
def _make_agent(tmp_path: Path) -> AIAgent:
|
|
with (
|
|
patch(
|
|
"run_agent.get_tool_definitions",
|
|
return_value=[
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "web_extract",
|
|
"description": "test tool",
|
|
"parameters": {"type": "object", "properties": {}},
|
|
},
|
|
}
|
|
],
|
|
),
|
|
patch("run_agent.check_toolset_requirements", return_value={}),
|
|
patch("run_agent.OpenAI"),
|
|
patch("run_agent._hermes_home", tmp_path),
|
|
patch("agent.model_metadata.fetch_model_metadata", return_value={}),
|
|
):
|
|
agent = AIAgent(
|
|
api_key="test-key",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
quiet_mode=True,
|
|
skip_context_files=True,
|
|
skip_memory=True,
|
|
)
|
|
agent._flush_messages_to_session_db = MagicMock(return_value=True)
|
|
agent._append_guardrail_observation = MagicMock(
|
|
side_effect=lambda _name, _args, result, **_kwargs: result
|
|
)
|
|
agent._record_file_mutation_result = MagicMock()
|
|
agent._subdirectory_hints.check_tool_call = MagicMock(return_value="")
|
|
agent._tool_result_content_for_active_model = MagicMock(
|
|
side_effect=lambda _name, result: result
|
|
)
|
|
return agent
|
|
|
|
|
|
def _tool_call(call_id: str):
|
|
return SimpleNamespace(
|
|
id=call_id,
|
|
type="function",
|
|
function=SimpleNamespace(name="web_extract", arguments="{}"),
|
|
)
|
|
|
|
|
|
def _clarify_call(call_id: str = "clarify-1"):
|
|
return SimpleNamespace(
|
|
id=call_id,
|
|
type="function",
|
|
function=SimpleNamespace(
|
|
name="clarify",
|
|
arguments='{"question": "Pick one?", "choices": ["A", "B"]}',
|
|
),
|
|
)
|
|
|
|
|
|
def test_sequential_tool_timeout_emits_result_and_continues(tmp_path, monkeypatch):
|
|
agent = _make_agent(tmp_path)
|
|
first_started = threading.Event()
|
|
release_first = threading.Event()
|
|
dispatched: list[str] = []
|
|
terminal_events: list[dict] = []
|
|
|
|
def _dispatch(_name, _args, _task_id, *, tool_call_id, **_kwargs):
|
|
dispatched.append(tool_call_id)
|
|
if tool_call_id == "hung":
|
|
first_started.set()
|
|
release_first.wait()
|
|
return "late result"
|
|
return "second result"
|
|
|
|
def _capture_terminal_event(*_args, **kwargs):
|
|
terminal_events.append(kwargs)
|
|
|
|
calls = [_tool_call("hung"), _tool_call("next")]
|
|
assistant = SimpleNamespace(tool_calls=calls)
|
|
messages: list[dict] = []
|
|
monkeypatch.setenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "0.05")
|
|
|
|
started = time.monotonic()
|
|
try:
|
|
with (
|
|
patch("run_agent.handle_function_call", side_effect=_dispatch),
|
|
patch(
|
|
"agent.tool_executor._emit_terminal_post_tool_call",
|
|
side_effect=_capture_terminal_event,
|
|
),
|
|
):
|
|
execute_tool_calls_sequential(agent, assistant, messages, "task")
|
|
elapsed = time.monotonic() - started
|
|
release_first.set()
|
|
# Give a late-scheduled worker a moment to drain inside the patch
|
|
# context (if the with block exits first, a late worker would hit
|
|
# the REAL handle_function_call).
|
|
for _ in range(100):
|
|
if "next" in dispatched:
|
|
break
|
|
time.sleep(0.05)
|
|
finally:
|
|
release_first.set()
|
|
|
|
assert elapsed < 1.0 # the EXECUTOR must not have waited for the hung call
|
|
# The hung call's dispatch is NOT guaranteed: on a loaded runner the
|
|
# 0.05s timeout can expire before its worker thread starts, in which case
|
|
# future.cancel() legitimately prevents the dispatch entirely
|
|
# (tool_executor timeout path). The contract is: the executor emitted a
|
|
# timeout result for it and CONTINUED — the next tool must have actually
|
|
# run, and the message stream must carry both call ids in order.
|
|
assert "next" in dispatched
|
|
assert set(dispatched) <= {"hung", "next"}
|
|
assert [message["tool_call_id"] for message in messages] == ["hung", "next"]
|
|
assert "timed out after 0.1s" in messages[0]["content"]
|
|
assert messages[0]["effect_disposition"] == "unknown"
|
|
assert messages[1]["content"] == "second result"
|
|
timeout_events = [event for event in terminal_events if event.get("error_type") == "tool_timeout"]
|
|
assert len(timeout_events) == 1
|
|
assert timeout_events[0]["status"] == "timeout"
|
|
agent._flush_messages_to_session_db.assert_called()
|
|
|
|
|
|
def test_sequential_tool_timeout_suppresses_late_terminal_event(tmp_path, monkeypatch):
|
|
import hermes_cli.lifecycle as lifecycle
|
|
import model_tools
|
|
|
|
agent = _make_agent(tmp_path)
|
|
release_first = threading.Event()
|
|
first_returned = threading.Event()
|
|
dispatch_count = 0
|
|
terminal_events: list[dict] = []
|
|
|
|
def _dispatch(_name, _args, **_kwargs):
|
|
nonlocal dispatch_count
|
|
dispatch_count += 1
|
|
if dispatch_count == 1:
|
|
release_first.wait()
|
|
first_returned.set()
|
|
return "late result"
|
|
return "second result"
|
|
|
|
calls = [_tool_call("hung"), _tool_call("next")]
|
|
messages: list[dict] = []
|
|
monkeypatch.setenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "0.05")
|
|
|
|
try:
|
|
with (
|
|
patch.object(model_tools.registry, "dispatch", side_effect=_dispatch),
|
|
patch.object(lifecycle, "has_hook", return_value=True),
|
|
patch.object(
|
|
lifecycle,
|
|
"invoke_hook",
|
|
side_effect=lambda hook, **kwargs: (
|
|
terminal_events.append(kwargs) if hook == "post_tool_call" else []
|
|
),
|
|
),
|
|
):
|
|
execute_tool_calls_sequential(
|
|
agent, SimpleNamespace(tool_calls=calls), messages, "task"
|
|
)
|
|
release_first.set()
|
|
assert first_returned.wait(timeout=1)
|
|
finally:
|
|
release_first.set()
|
|
|
|
assert [(event["tool_call_id"], event.get("error_type")) for event in terminal_events] == [
|
|
("hung", "tool_timeout"),
|
|
("next", None),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"clarify_timeout",
|
|
[resolve_clarify_timeout({}), 0],
|
|
ids=["default-3600s", "unlimited"],
|
|
)
|
|
def test_sequential_timeout_does_not_cut_clarify_human_wait(
|
|
tmp_path, monkeypatch, clarify_timeout
|
|
):
|
|
"""Clarify waits on a human; the generic sequential deadline must not fire.
|
|
|
|
Default ``agent.clarify_timeout`` is 3600s; ``<= 0`` is unlimited. Both
|
|
outlast ``HERMES_CONCURRENT_TOOL_TIMEOUT_S`` (default 420s).
|
|
"""
|
|
agent = _make_agent(tmp_path)
|
|
monkeypatch.setenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "0.05")
|
|
monkeypatch.setattr(
|
|
"tools.clarify_gateway.get_clarify_timeout",
|
|
lambda: clarify_timeout,
|
|
)
|
|
|
|
def _callback(question, choices, multi_select=False):
|
|
time.sleep(0.15)
|
|
return "A"
|
|
|
|
agent.clarify_callback = _callback
|
|
terminal_events: list[dict] = []
|
|
|
|
def _dispatch(_name, _args, _task_id, *, tool_call_id, **_kwargs):
|
|
return "second result"
|
|
|
|
def _capture_terminal_event(*_args, **kwargs):
|
|
terminal_events.append(kwargs)
|
|
|
|
messages: list[dict] = []
|
|
started = time.monotonic()
|
|
with (
|
|
patch("run_agent.handle_function_call", side_effect=_dispatch),
|
|
patch(
|
|
"agent.tool_executor._emit_terminal_post_tool_call",
|
|
side_effect=_capture_terminal_event,
|
|
),
|
|
):
|
|
execute_tool_calls_sequential(
|
|
agent,
|
|
SimpleNamespace(tool_calls=[_clarify_call(), _tool_call("next")]),
|
|
messages,
|
|
"task",
|
|
)
|
|
|
|
assert time.monotonic() - started < 1.0
|
|
assert [message["tool_call_id"] for message in messages] == ["clarify-1", "next"]
|
|
payload = json.loads(messages[0]["content"])
|
|
assert payload["user_response"] == "A"
|
|
assert "timed out" not in messages[0]["content"]
|
|
assert messages[1]["content"] == "second result"
|
|
assert not any(event.get("error_type") == "tool_timeout" for event in terminal_events)
|