diff --git a/agent/tool_executor.py b/agent/tool_executor.py index cb28c81ea65ca..130acd5c8a91b 100644 --- a/agent/tool_executor.py +++ b/agent/tool_executor.py @@ -33,6 +33,7 @@ _detect_tool_failure, ) from agent.tool_dispatch_helpers import ( + _NEVER_PARALLEL_TOOLS, _is_destructive_command, _is_multimodal_tool_result, _multimodal_text_summary, @@ -387,6 +388,10 @@ class _ManagedToolResult: dispatched: bool +class _ToolTimeoutResult(str): + """Marker for a synthesized sequential-tool timeout result.""" + + class _ConcurrentToolAuthorizationGate: """Serialize policy prompts and exclude human approval waits from batch deadlines. @@ -661,6 +666,146 @@ def _hermes_pipeline(relay_args: dict[str, Any]) -> Any: ) +def _resolve_sequential_tool_timeout() -> float | None: + """Deadline for one sequential tool call (#85125 Phase 2a). + + ``timeouts.tools.sequential_call`` in config.yaml wins; when unset, the + sequential path inherits the concurrent batch deadline (same value, same + ``HERMES_CONCURRENT_TOOL_TIMEOUT_S`` legacy bridge) so the two executor + paths cannot drift apart by default. ``0``/negative disables the bound. + + NOTE: this path deliberately does NOT use ``agent.deadline.run_bounded_sync``. + The sequential/concurrent executors extend their deadline dynamically while + a human approval prompt is open (``_ConcurrentToolAuthorizationGate`` + excluded seconds — a MUST-preserve invariant) and touch agent activity + mid-wait; the shared primitive is fixed-deadline by design. Simpler call + sites migrate onto the primitive; these two stay symmetric with each other. + """ + from agent.deadline import resolve_timeout + + return resolve_timeout( + "tools.sequential_call", + default=_resolve_concurrent_tool_timeout(), + ) + + +def _run_sequential_tool_execution_middleware( + agent, + *, + function_name: str, + function_args: dict, + effective_task_id: str, + tool_call_id: str, + execute, + scope_block: str | None = None, + display_index: int | None = None, + middleware_trace: list[dict[str, Any]] | None = None, +) -> _ManagedToolResult: + """Run one sequential call with the concurrent executor's deadline. + + Interactive input tools such as ``clarify`` wait on a human. Their own + timeout (``agent.clarify_timeout``: default 3600s, or unlimited when + ``<= 0``) owns that wait. Applying the generic tool deadline here would + return ``tool_timeout`` while the prompt and worker stay active. + """ + timeout_s = _resolve_sequential_tool_timeout() + kwargs = { + "function_name": function_name, + "function_args": function_args, + "effective_task_id": effective_task_id, + "tool_call_id": tool_call_id, + "execute": execute, + "scope_block": scope_block, + "display_index": display_index, + "middleware_trace": middleware_trace, + } + if timeout_s is None or function_name in _NEVER_PARALLEL_TOOLS: + return _run_agent_tool_execution_middleware(agent, **kwargs) + + from tools.daemon_pool import DaemonThreadPoolExecutor + + authorization_gate = _ConcurrentToolAuthorizationGate() + worker_tid: list[int] = [] + + def _run() -> _ManagedToolResult: + tid = threading.current_thread().ident + worker_tid.append(tid) + with agent._tool_worker_threads_lock: + agent._tool_worker_threads.add(tid) + try: + return _run_agent_tool_execution_middleware( + agent, authorization_gate=authorization_gate, **kwargs + ) + finally: + with agent._tool_worker_threads_lock: + agent._tool_worker_threads.discard(tid) + try: + _ra()._set_interrupt(False, tid) + except Exception: + pass + + executor = DaemonThreadPoolExecutor(max_workers=1) + future = executor.submit(propagate_context_to_thread(_run)) + deadline = time.monotonic() + timeout_s + started = time.monotonic() + timed_out = False + try: + while True: + remaining = ( + deadline + authorization_gate.excluded_seconds() - time.monotonic() + ) + if remaining <= 0: + timed_out = True + break + try: + return future.result(timeout=min(5.0, remaining)) + except concurrent.futures.TimeoutError: + elapsed = int(time.monotonic() - started) + if elapsed > 0 and elapsed % 30 < 5: + agent._touch_activity( + f"sequential tool running ({elapsed}s): {function_name}" + ) + + message = ( + f"Error executing tool '{function_name}': " + f"timed out after {timeout_s:.1f}s" + ) + logger.warning( + "sequential tool %s timed out after %.1fs", function_name, timeout_s + ) + future.cancel() + for tid in worker_tid: + try: + _ra()._set_interrupt(True, tid) + except Exception: + pass + trace = middleware_trace if middleware_trace is not None else [] + _emit_terminal_post_tool_call( + agent, + function_name=function_name, + function_args=function_args, + result=message, + effective_task_id=effective_task_id, + tool_call_id=tool_call_id, + duration_ms=int(timeout_s * 1000), + status="timeout", + error_type="tool_timeout", + error_message=message, + middleware_trace=list(trace), + ) + return _ManagedToolResult( + result=_ToolTimeoutResult(message), + args=function_args, + middleware_trace=trace, + blocked=False, + dispatched=True, + ) + finally: + # Never join a wedged worker. DaemonThreadPoolExecutor also keeps it out + # of the stdlib atexit join, matching the concurrent timeout path. + executor.shutdown(wait=not timed_out, cancel_futures=timed_out) + + def _begin_tool_execution( agent, *, @@ -1608,6 +1753,12 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe """ # Resolve the context-scaled tool-output budget once per turn. _tool_budget = _budget_for_agent(agent) + + # Keep every runtime-tool branch on one bounded execution funnel without + # duplicating timeout policy across the branch-specific callbacks below. + def _run_agent_tool_execution_middleware(agent, **kwargs): + return _run_sequential_tool_execution_middleware(agent, **kwargs) + for i, tool_call in enumerate(assistant_message.tool_calls, 1): if getattr(agent, "_incremental_persistence_failed", False): return @@ -2046,27 +2197,30 @@ def _execute(next_args: dict) -> Any: _spinner_result = None try: def _execute(next_args: dict) -> Any: - return _ra().handle_function_call( - function_name, - next_args, - effective_task_id, - tool_call_id=tool_call.id, - session_id=agent.session_id or "", - turn_id=getattr(agent, "_current_turn_id", "") or "", - api_request_id=getattr(agent, "_current_api_request_id", "") - or "", - enabled_tools=( - list(agent.valid_tool_names) - if agent.valid_tool_names - else None - ), - skip_pre_tool_call_hook=True, - skip_tool_request_middleware=True, - skip_tool_execution_middleware=True, - tool_request_middleware_trace=list(middleware_trace), - enabled_toolsets=getattr(agent, "enabled_toolsets", None), - disabled_toolsets=getattr(agent, "disabled_toolsets", None), - ) + from model_tools import suppress_post_tool_call_hook + + with suppress_post_tool_call_hook(): + return _ra().handle_function_call( + function_name, + next_args, + effective_task_id, + tool_call_id=tool_call.id, + session_id=agent.session_id or "", + turn_id=getattr(agent, "_current_turn_id", "") or "", + api_request_id=getattr(agent, "_current_api_request_id", "") + or "", + enabled_tools=( + list(agent.valid_tool_names) + if agent.valid_tool_names + else None + ), + skip_pre_tool_call_hook=True, + skip_tool_request_middleware=True, + skip_tool_execution_middleware=True, + tool_request_middleware_trace=list(middleware_trace), + enabled_toolsets=getattr(agent, "enabled_toolsets", None), + disabled_toolsets=getattr(agent, "disabled_toolsets", None), + ) ( function_result, @@ -2125,27 +2279,30 @@ def _execute(next_args: dict) -> Any: else: try: def _execute(next_args: dict) -> Any: - return _ra().handle_function_call( - function_name, - next_args, - effective_task_id, - tool_call_id=tool_call.id, - session_id=agent.session_id or "", - turn_id=getattr(agent, "_current_turn_id", "") or "", - api_request_id=getattr(agent, "_current_api_request_id", "") - or "", - enabled_tools=( - list(agent.valid_tool_names) - if agent.valid_tool_names - else None - ), - skip_pre_tool_call_hook=True, - skip_tool_request_middleware=True, - skip_tool_execution_middleware=True, - tool_request_middleware_trace=list(middleware_trace), - enabled_toolsets=getattr(agent, "enabled_toolsets", None), - disabled_toolsets=getattr(agent, "disabled_toolsets", None), - ) + from model_tools import suppress_post_tool_call_hook + + with suppress_post_tool_call_hook(): + return _ra().handle_function_call( + function_name, + next_args, + effective_task_id, + tool_call_id=tool_call.id, + session_id=agent.session_id or "", + turn_id=getattr(agent, "_current_turn_id", "") or "", + api_request_id=getattr(agent, "_current_api_request_id", "") + or "", + enabled_tools=( + list(agent.valid_tool_names) + if agent.valid_tool_names + else None + ), + skip_pre_tool_call_hook=True, + skip_tool_request_middleware=True, + skip_tool_execution_middleware=True, + tool_request_middleware_trace=list(middleware_trace), + enabled_toolsets=getattr(agent, "enabled_toolsets", None), + disabled_toolsets=getattr(agent, "disabled_toolsets", None), + ) ( function_result, @@ -2193,6 +2350,7 @@ def _execute(next_args: dict) -> Any: logger.error("handle_function_call raised for %s: %s", function_name, tool_error, exc_info=True) tool_duration = time.time() - tool_start_time + _execution_timed_out = isinstance(function_result, _ToolTimeoutResult) if isinstance(function_result, str): result_preview = function_result if agent.verbose_logging else ( function_result[:200] if len(function_result) > 200 else function_result @@ -2210,15 +2368,12 @@ def _execute(next_args: dict) -> Any: # context-engine, memory-manager, clarify, delegate_task) are # dispatched inline — they never reach handle_function_call, so the # executor is the one that has to fire post_tool_call. For - # registry-dispatched tools the else-branch above invoked - # handle_function_call, which already fires the hook. - from agent.agent_runtime_helpers import agent_runtime_owns_post_tool_hook + # Every dispatch suppresses the inner handle_function_call observer so + # the executor owns one terminal event for this tool_call_id. This also + # prevents an abandoned timeout worker from reporting late success. _executor_must_emit_post_hook = ( not _execution_blocked - and ( - not _execution_dispatched - or agent_runtime_owns_post_tool_hook(agent, function_name) - ) + and not _execution_timed_out ) if _executor_must_emit_post_hook: _emit_terminal_post_tool_call( @@ -2287,7 +2442,12 @@ def _execute(next_args: dict) -> Any: # Unwrap _multimodal dicts to an OpenAI-style content list # (see parallel path for rationale). String results pass through. _tool_content = agent._tool_result_content_for_active_model(function_name, function_result) - tool_message = make_tool_result_message(function_name, _tool_content, tool_call.id) + tool_message = make_tool_result_message( + function_name, + _tool_content, + tool_call.id, + effect_disposition="unknown" if _execution_timed_out else None, + ) messages.append(tool_message) risk_metadata = tool_message.get("_tool_output_risk") if not _flush_session_db_after_tool_progress( diff --git a/cli-config.yaml.example b/cli-config.yaml.example index 5f80fb23ca597..37a4a0f66d0fb 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -181,6 +181,10 @@ model: # tools: # concurrent_batch: 420 # Deadline for a parallel tool-call batch # # (legacy env: HERMES_CONCURRENT_TOOL_TIMEOUT_S) +# sequential_call: 420 # Deadline for one sequentially-executed tool call. +# # Defaults to concurrent_batch's value so the two +# # executor paths stay in sync; human waits +# # (approval prompts, clarify) never count against it. # ============================================================================= # OpenRouter Provider Routing (only applies when using OpenRouter) diff --git a/model_tools.py b/model_tools.py index 8fe4ffd34c13e..a14b28e345cc4 100644 --- a/model_tools.py +++ b/model_tools.py @@ -24,6 +24,8 @@ import json import re import asyncio +from contextlib import contextmanager +from contextvars import ContextVar import logging import threading import time @@ -40,6 +42,20 @@ logger = logging.getLogger(__name__) +_post_tool_call_hook_suppressed: ContextVar[bool] = ContextVar( + "post_tool_call_hook_suppressed", default=False +) + + +@contextmanager +def suppress_post_tool_call_hook(): + """Let an outer executor own the terminal post-tool event.""" + token = _post_tool_call_hook_suppressed.set(True) + try: + yield + finally: + _post_tool_call_hook_suppressed.reset(token) + # Tracks platform-bundle names already flagged in disabled_toolsets so the # advisory (#33924) is logged once per name, not on every tool recompute. _WARNED_DISABLED_BUNDLES: set = set() @@ -1138,6 +1154,8 @@ def _emit_post_tool_call_hook( result *after* the gate (parsing the result is only worth it when a listener will actually consume it). """ + if _post_tool_call_hook_suppressed.get(): + return try: from hermes_cli.lifecycle import has_hook, invoke_hook if not has_hook("post_tool_call"): diff --git a/tests/agent/test_deadline.py b/tests/agent/test_deadline.py index 9e5238b9afc8c..378a5e2c10569 100644 --- a/tests/agent/test_deadline.py +++ b/tests/agent/test_deadline.py @@ -476,3 +476,46 @@ def test_new_config_key_wins(self, monkeypatch): ) monkeypatch.setenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "60") assert self._resolver()() == 300.0 + + +class TestSequentialToolTimeoutResolver: + """_resolve_sequential_tool_timeout: own key, inherits concurrent default.""" + + def _resolver(self): + from agent import tool_executor + + return tool_executor._resolve_sequential_tool_timeout + + def test_inherits_concurrent_default(self, monkeypatch): + monkeypatch.setattr("agent.deadline._timeouts_section", lambda: {}) + monkeypatch.delenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", raising=False) + assert self._resolver()() == 420.0 + + def test_inherits_concurrent_env_bridge(self, monkeypatch): + # No sequential-specific setting -> concurrent env var flows through. + monkeypatch.setattr("agent.deadline._timeouts_section", lambda: {}) + monkeypatch.setenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "60") + assert self._resolver()() == 60.0 + + def test_own_config_key_wins_over_concurrent(self, monkeypatch): + monkeypatch.setattr( + "agent.deadline._timeouts_section", + lambda: {"tools": {"concurrent_batch": 300, "sequential_call": 90}}, + ) + assert self._resolver()() == 90.0 + + def test_zero_disables_independently(self, monkeypatch): + # Sequential bound can be disabled while the concurrent one stays on. + monkeypatch.setattr( + "agent.deadline._timeouts_section", + lambda: {"tools": {"concurrent_batch": 300, "sequential_call": 0}}, + ) + assert self._resolver()() is None + + def test_concurrent_disabled_flows_through(self, monkeypatch): + # concurrent disabled (None default) + no sequential key -> unbounded. + monkeypatch.setattr( + "agent.deadline._timeouts_section", + lambda: {"tools": {"concurrent_batch": 0}}, + ) + assert self._resolver()() is None diff --git a/tests/run_agent/test_sequential_tool_timeout.py b/tests/run_agent/test_sequential_tool_timeout.py new file mode 100644 index 0000000000000..fbcb46a133b7d --- /dev/null +++ b/tests/run_agent/test_sequential_tool_timeout.py @@ -0,0 +1,228 @@ +"""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") + finally: + release_first.set() + + assert first_started.is_set() + assert time.monotonic() - started < 1.0 + assert 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)