"""Tool-call execution — sequential and concurrent dispatch. Both AIAgent methods (``_execute_tool_calls_sequential`` and ``_execute_tool_calls_concurrent``) live here as module-level functions that take the parent ``AIAgent`` as their first argument. ``run_agent`` keeps thin wrappers so existing call sites work; tests that patch ``run_agent._set_interrupt`` are honored because the extracted functions reach back through the ``run_agent`` module via ``_ra()`` for that symbol. """ from __future__ import annotations import concurrent.futures import json from pathlib import Path import logging import os import random import threading import time from typing import Any, Optional from agent.display import ( KawaiiSpinner, build_tool_preview as _build_tool_preview, build_tool_label as _build_tool_label, get_cute_tool_message as _get_cute_tool_message_impl, get_tool_emoji as _get_tool_emoji, redact_tool_args_for_display as _redact_tool_args_for_display, _detect_tool_failure, ) from agent.tool_guardrails import ToolGuardrailDecision from agent.tool_dispatch_helpers import ( _is_destructive_command, _is_multimodal_tool_result, _multimodal_text_summary, _append_subdir_hint_to_multimodal, _plan_tool_batch_segments, make_tool_result_message, ) from tools.terminal_tool import ( get_active_env, ) from tools.thread_context import propagate_context_to_thread from tools.tool_result_storage import ( maybe_persist_tool_result, enforce_turn_budget, ) from tools.budget_config import BudgetConfig, DEFAULT_BUDGET, budget_for_context_window logger = logging.getLogger(__name__) def _ensure_file_checkpoint( agent, function_name: str, function_args: dict, effective_task_id: str, ) -> None: """Checkpoint the same workspace path that the file tool will mutate.""" file_path = function_args.get("path", "") if not file_path: return # File tools resolve relative paths against the task's live/session cwd, # which can differ from the Hermes process cwd (notably in Docker). Resolve # through that same path pipeline before asking the checkpoint manager to # discover the project root. from tools.file_tools import _resolve_path_for_task resolved_path = _resolve_path_for_task(file_path, effective_task_id or "default") work_dir = agent._checkpoint_mgr.get_working_dir_for_path(str(resolved_path)) agent._checkpoint_mgr.ensure_checkpoint(work_dir, f"before {function_name}") def _budget_for_agent(agent) -> BudgetConfig: """Resolve a tool-result BudgetConfig scaled to the agent's context window. Large-context models keep the historical 100K/200K char defaults; small models (e.g. a 65K-token local model switched into mid-session) get a budget proportional to their window so a single large tool result can't push the request past the model's limit (#23767). Falls back to the default budget when the context length isn't resolvable. """ try: ctx = getattr(getattr(agent, "context_compressor", None), "context_length", None) return budget_for_context_window(int(ctx)) if ctx else DEFAULT_BUDGET except Exception: return DEFAULT_BUDGET # Maximum number of concurrent worker threads for parallel tool execution. # Mirrors the constant in ``run_agent`` for tests/imports that look here. _MAX_TOOL_WORKERS = 8 # Keep this above the stock auxiliary.web_extract timeout (360s) so the batch # guard does not preempt a slow-but-valid summarization attempt. _DEFAULT_CONCURRENT_TOOL_TIMEOUT_S = 420.0 def _parse_tool_arguments(raw_arguments: Any) -> tuple[dict, Optional[str]]: """Parse model-emitted arguments without repairing or coercing them.""" try: arguments = json.loads(raw_arguments) except (json.JSONDecodeError, TypeError): arguments = None if isinstance(arguments, dict): return arguments, None return {}, json.dumps( { "error": "Invalid tool arguments", "message": ( "Tool arguments must be a valid JSON object; tool was not executed." ), }, ensure_ascii=False, ) def _resolve_concurrent_tool_timeout() -> float | None: raw = os.getenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "").strip() if not raw: return _DEFAULT_CONCURRENT_TOOL_TIMEOUT_S try: value = float(raw) except ValueError: logger.warning( "invalid HERMES_CONCURRENT_TOOL_TIMEOUT_S=%r; using %.0fs", raw, _DEFAULT_CONCURRENT_TOOL_TIMEOUT_S, ) return _DEFAULT_CONCURRENT_TOOL_TIMEOUT_S if value <= 0: return None return value def _flush_session_db_after_tool_progress( agent, messages: list, *, stage: str, ) -> None: """Best-effort incremental SessionDB flush for tool-call progress. Tool execution can perform side effects that terminate or restart the current Hermes process before the normal turn-end persistence path runs. Flush the already-appended assistant/tool messages immediately so the transcript survives destructive-but-valid tool calls. """ try: agent._flush_messages_to_session_db(messages) except Exception as exc: logger.warning("Incremental tool-call persistence failed after %s: %s", stage, exc) def _ra(): """Lazy reference to ``run_agent`` so patches like ``run_agent._set_interrupt`` work.""" import run_agent return run_agent def _is_interpreter_shutdown_submit_error(exc: RuntimeError) -> bool: return "cannot schedule new futures after interpreter shutdown" in str(exc) def _emit_terminal_post_tool_call( agent, *, function_name: str, function_args: dict, result: Any, effective_task_id: str, tool_call_id: str, duration_ms: int = 0, status: str | None = None, error_type: str | None = None, error_message: str | None = None, middleware_trace: Optional[list[dict[str, Any]]] = None, ) -> None: try: from model_tools import _emit_post_tool_call_hook _emit_post_tool_call_hook( function_name=function_name, function_args=function_args, result=result, task_id=effective_task_id or "", session_id=getattr(agent, "session_id", "") or "", tool_call_id=tool_call_id or "", turn_id=getattr(agent, "_current_turn_id", "") or "", api_request_id=getattr(agent, "_current_api_request_id", "") or "", duration_ms=duration_ms, status=status, error_type=error_type, error_message=error_message, middleware_trace=list(middleware_trace or []), ) except Exception: pass def _cancelled_tool_result(reason: str = "user interrupt") -> str: return json.dumps( { "error": f"Tool execution cancelled by {reason}", "status": "cancelled", }, ensure_ascii=False, ) def _emit_cancelled_terminal_post_tool_call( agent, *, function_name: str, function_args: dict, effective_task_id: str, tool_call_id: str, start_time: float, reason: str = "user interrupt", error_type: str = "keyboard_interrupt", middleware_trace: Optional[list[dict[str, Any]]] = None, ) -> str: result = _cancelled_tool_result(reason) _emit_terminal_post_tool_call( agent, function_name=function_name, function_args=function_args, result=result, effective_task_id=effective_task_id, tool_call_id=tool_call_id, duration_ms=int((time.time() - start_time) * 1000), status="cancelled", error_type=error_type, error_message=f"Tool execution cancelled by {reason}", middleware_trace=list(middleware_trace or []), ) return result def _tool_search_scoped_names(agent) -> frozenset: """Return the deferrable tool names the session may invoke via tool_call. The Tool Search unwrap dispatches the underlying tool directly, bypassing the bridge branch (and its scope check) in ``model_tools.handle_function_call``. To keep a restricted-toolset session (subagent, kanban worker, curated gateway session) from reaching tools it was never granted, the unwrap validates the underlying name against this set: the deferrable subset of the session's own enabled/disabled toolset scope. Result is cached on the agent and refreshed when the tool registry's generation changes (e.g. an MCP server reconnects), so the common case is a dict lookup, not a full tool-defs rebuild on every tool call. """ try: import model_tools from tools import tool_search as _ts from tools.registry import registry as _registry except Exception: return frozenset() enabled = getattr(agent, "enabled_toolsets", None) disabled = getattr(agent, "disabled_toolsets", None) cache_key = ( getattr(_registry, "_generation", 0), frozenset(enabled) if enabled is not None else None, frozenset(disabled) if disabled is not None else None, ) cached = getattr(agent, "_tool_search_scope_cache", None) if cached is not None and cached[0] == cache_key: return cached[1] try: scoped_defs = model_tools.get_tool_definitions( enabled_toolsets=enabled, disabled_toolsets=disabled, quiet_mode=True, skip_tool_search_assembly=True, ) or [] names = _ts.scoped_deferrable_names(scoped_defs) except Exception: names = frozenset() try: agent._tool_search_scope_cache = (cache_key, names) except Exception: pass return names def _apply_tool_request_middleware_for_agent( agent, *, function_name: str, function_args: dict, effective_task_id: str, tool_call_id: str, ) -> tuple[dict, list[dict[str, Any]]]: try: from hermes_cli.middleware import apply_tool_request_middleware result = apply_tool_request_middleware( function_name, function_args, task_id=effective_task_id or "", session_id=getattr(agent, "session_id", "") or "", tool_call_id=tool_call_id or "", turn_id=getattr(agent, "_current_turn_id", "") or "", api_request_id=getattr(agent, "_current_api_request_id", "") or "", ) payload = result.payload if isinstance(result.payload, dict) else function_args return payload, list(result.trace) except Exception as exc: logger.debug("tool_request middleware error: %s", exc) return function_args, [] def _run_agent_tool_execution_middleware( agent, *, function_name: str, function_args: dict, effective_task_id: str, tool_call_id: str, execute, ) -> tuple[Any, dict]: observed_args = function_args def _execute(next_args: dict) -> Any: nonlocal observed_args observed_args = next_args if isinstance(next_args, dict) else function_args return execute(observed_args) from hermes_cli.middleware import run_tool_execution_middleware result = run_tool_execution_middleware( function_name, function_args, _execute, original_args=function_args, task_id=effective_task_id or "", session_id=getattr(agent, "session_id", "") or "", tool_call_id=tool_call_id or "", turn_id=getattr(agent, "_current_turn_id", "") or "", api_request_id=getattr(agent, "_current_api_request_id", "") or "", ) return result, observed_args def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effective_task_id: str, api_call_count: int = 0, *, finalize: bool = True) -> None: """Execute multiple tool calls concurrently using a thread pool. Results are collected in the original tool-call order and appended to messages so the API sees them in the expected sequence. ``finalize=False`` skips the end-of-batch aggregate budget enforcement and /steer injection — used when this call is one segment of a larger mixed batch and the segmented dispatcher owns the turn-end work. """ tool_calls = assistant_message.tool_calls num_tools = len(tool_calls) # Resolve the context-scaled tool-output budget once per turn (cheap, but # avoids rebuilding it per result inside the loop below). _tool_budget = _budget_for_agent(agent) # ── Pre-flight: interrupt check ────────────────────────────────── if agent._interrupt_requested: print(f"{agent.log_prefix}⚡ Interrupt: skipping {num_tools} tool call(s)") for tc in tool_calls: messages.append(make_tool_result_message( tc.function.name, f"[Tool execution cancelled — {tc.function.name} was skipped due to user interrupt]", tc.id, effect_disposition="none", )) _flush_session_db_after_tool_progress( agent, messages, stage=f"cancelled tool result {tc.function.name}", ) return # ── Parse args + pre-execution bookkeeping ─────────────────────── parsed_calls = [] # list of (tool_call, function_name, function_args, middleware_trace, block_result, blocked_by_guardrail) for tool_call in tool_calls: function_name = tool_call.function.name function_args, malformed_args_result = _parse_tool_arguments( tool_call.function.arguments ) if malformed_args_result is not None: parsed_calls.append( ( tool_call, function_name, function_args, [], malformed_args_result, False, ) ) continue # Reset nudge counters only for a structurally valid invocation. if function_name == "memory": agent._turns_since_memory = 0 elif function_name == "skill_manage": agent._iters_since_skill = 0 # ── Tool Search unwrap ──────────────────────────────────────── # When the model invokes the tool_call bridge, peel it open so # every downstream check (checkpointing, guardrails, plugin # pre-tool-call hooks, the display/activity feed, the post-call # callback) sees the underlying tool — not the bridge. This is # the OpenClaw lesson: hooks must observe the real tool name. # # The original tool_call entry on ``tool_call.function`` is left # untouched so the conversation transcript and the matching # tool_call_id are preserved exactly as the model emitted them. # # Scope gate: the unwrap dispatches the underlying tool directly # (bypassing the bridge branch in handle_function_call and its # scope check), so we enforce session toolset scope HERE. A tool # the session was not granted is rejected before any checkpoint, # hook, or dispatch fires. _ts_scope_block = None try: from tools import tool_search as _ts if function_name == _ts.TOOL_CALL_NAME: _underlying, _underlying_args, _err = _ts.resolve_underlying_call(function_args) if not _err and _underlying: if _underlying in _tool_search_scoped_names(agent): function_name = _underlying function_args = _underlying_args else: _ts_scope_block = json.dumps({ "error": ( f"'{_underlying}' is not available in this session. " "Use tool_search to find tools you can call." ), }, ensure_ascii=False) except Exception: pass function_args, middleware_trace = _apply_tool_request_middleware_for_agent( agent, function_name=function_name, function_args=function_args, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", ) # ── Block evaluation (BEFORE checkpoint preflight) ─────────── # We must know whether the tool will execute before touching # checkpoint state (dedup slot, real snapshots). block_result = None blocked_by_guardrail = False if _ts_scope_block is not None: # Out-of-scope tool_call: reject before hooks/guardrails/dispatch. block_result = _ts_scope_block _emit_terminal_post_tool_call( agent, function_name=function_name, function_args=function_args, result=block_result, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", status="blocked", error_type="tool_scope_block", error_message=_ts_scope_block, middleware_trace=list(middleware_trace), ) else: try: from hermes_cli.plugins import resolve_pre_tool_block block_message = resolve_pre_tool_block( function_name, function_args, task_id=effective_task_id or "", session_id=getattr(agent, "session_id", "") or "", tool_call_id=getattr(tool_call, "id", "") or "", turn_id=getattr(agent, "_current_turn_id", "") or "", api_request_id=getattr(agent, "_current_api_request_id", "") or "", middleware_trace=list(middleware_trace), ) except Exception: block_message = None if block_message is not None: block_result = json.dumps({"error": block_message}, ensure_ascii=False) _emit_terminal_post_tool_call( agent, function_name=function_name, function_args=function_args, result=block_result, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", status="blocked", error_type="plugin_block", error_message=block_message, middleware_trace=list(middleware_trace), ) else: guardrail_decision = agent._tool_guardrails.before_call(function_name, function_args) if not guardrail_decision.allows_execution: block_result = agent._guardrail_block_result(guardrail_decision) blocked_by_guardrail = True _emit_terminal_post_tool_call( agent, function_name=function_name, function_args=function_args, result=block_result, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", status="blocked", error_type="guardrail_block", error_message=getattr(guardrail_decision, "message", None) or "Tool blocked by guardrail policy", middleware_trace=list(middleware_trace), ) # ── Checkpoint preflight (only for tools that will execute) ── if block_result is None: # Checkpoint for file-mutating tools if function_name in {"write_file", "patch"} and agent._checkpoint_mgr.enabled: try: _ensure_file_checkpoint( agent, function_name, function_args, effective_task_id, ) except Exception: pass # Checkpoint before destructive terminal commands if function_name == "terminal" and agent._checkpoint_mgr.enabled: try: cmd = function_args.get("command", "") if _is_destructive_command(cmd): cwd = function_args.get("workdir") or os.getenv("TERMINAL_CWD", os.getcwd()) agent._checkpoint_mgr.ensure_checkpoint( cwd, f"before terminal: {cmd[:60]}" ) except Exception: pass parsed_calls.append((tool_call, function_name, function_args, middleware_trace, block_result, blocked_by_guardrail)) # ── Logging / callbacks ────────────────────────────────────────── tool_names_str = ", ".join(name for _, name, _, _, _, _ in parsed_calls) if not agent.quiet_mode and getattr(agent, "tool_progress_mode", "all") != "off": print(f" ⚡ Concurrent: {num_tools} tool calls — {tool_names_str}") for i, (tc, name, args, middleware_trace, block_result, blocked_by_guardrail) in enumerate(parsed_calls, 1): display_args = _redact_tool_args_for_display(name, args) or args args_str = json.dumps(display_args, ensure_ascii=False) if agent.verbose_logging: print(f" 📞 Tool {i}: {name}({list(display_args.keys())})") print(agent._wrap_verbose("Args: ", json.dumps(display_args, indent=2, ensure_ascii=False))) else: args_preview = args_str[:agent.log_prefix_chars] + "..." if len(args_str) > agent.log_prefix_chars else args_str print(f" 📞 Tool {i}: {name}({list(args.keys())}) - {args_preview}") for tc, name, args, middleware_trace, block_result, blocked_by_guardrail in parsed_calls: if block_result is not None: continue if agent.tool_progress_callback: try: display_args = _redact_tool_args_for_display(name, args) or args preview = _build_tool_preview(name, display_args) agent.tool_progress_callback("tool.started", name, preview, display_args) except Exception as cb_err: logging.debug(f"Tool progress callback error: {cb_err}") for tc, name, args, middleware_trace, block_result, blocked_by_guardrail in parsed_calls: if block_result is not None: continue if agent.tool_start_callback: try: display_args = _redact_tool_args_for_display(name, args) or args agent.tool_start_callback(tc.id, name, display_args) except Exception as cb_err: logging.debug(f"Tool start callback error: {cb_err}") # ── Concurrent execution ───────────────────────────────────────── # Each slot holds (function_name, function_args, function_result, duration, error_flag, blocked_flag, middleware_trace) results = [None] * num_tools for i, (tc, name, args, middleware_trace, block_result, blocked_by_guardrail) in enumerate(parsed_calls): if block_result is not None: results[i] = (name, args, block_result, 0.0, True, True, middleware_trace) # Touch activity before launching workers so the gateway knows # we're executing tools (not stuck). agent._current_tool = tool_names_str agent._touch_activity(f"executing {num_tools} tools concurrently: {tool_names_str}") def _run_tool(index, tool_call, function_name, function_args, middleware_trace): """Worker function executed in a thread.""" # Register this worker tid so the agent can fan out an interrupt # to it — see AIAgent.interrupt(). Must happen first thing, and # must be paired with discard + clear in the finally block. _worker_tid = threading.current_thread().ident with agent._tool_worker_threads_lock: agent._tool_worker_threads.add(_worker_tid) # Race: if the agent was interrupted between fan-out (which # snapshotted an empty/earlier set) and our registration, apply # the interrupt to our own tid now so is_interrupted() inside # the tool returns True on the next poll. if agent._interrupt_requested: try: _ra()._set_interrupt(True, _worker_tid) except Exception: pass # Set the activity callback on THIS worker thread so # _wait_for_process (terminal commands) can fire heartbeats. # The callback is thread-local; the main thread's callback # is invisible to worker threads. try: from tools.environments.base import set_activity_callback set_activity_callback(agent._touch_activity) except Exception: pass # Approval/sudo callbacks (thread-local) and the agent turn's # ContextVars are propagated by propagate_context_to_thread() at the # submit site below (GHSA-qg5c-hvr5-hjgr, #13617). start = time.time() try: try: result = agent._invoke_tool( function_name, function_args, effective_task_id, tool_call.id, messages=messages, pre_tool_block_checked=True, skip_tool_request_middleware=True, tool_request_middleware_trace=list(middleware_trace), ) except KeyboardInterrupt: try: agent.interrupt("keyboard interrupt") except Exception: pass result = _emit_cancelled_terminal_post_tool_call( agent, function_name=function_name, function_args=function_args, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", start_time=start, middleware_trace=list(middleware_trace), ) duration = time.time() - start logger.info("tool %s cancelled (%.2fs)", function_name, duration) results[index] = (function_name, function_args, result, duration, True, False, middleware_trace) return except Exception as tool_error: result = f"Error executing tool '{function_name}': {tool_error}" logger.error("_invoke_tool raised for %s: %s", function_name, tool_error, exc_info=True) duration = time.time() - start is_error, _ = _detect_tool_failure(function_name, result) if is_error: logger.info("tool %s failed (%.2fs): %s", function_name, duration, result[:200]) else: logger.info("tool %s completed (%.2fs, %d chars)", function_name, duration, len(result)) results[index] = (function_name, function_args, result, duration, is_error, False, middleware_trace) finally: # 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 # because BaseException subclasses (CancelledError, KeyboardInterrupt) # bypass ``except Exception`` and would otherwise leak the tid # into _interrupted_threads, poisoning the recycled thread. with agent._tool_worker_threads_lock: agent._tool_worker_threads.discard(_worker_tid) try: _ra()._set_interrupt(False, _worker_tid) except Exception: pass # Start spinner for CLI mode (skip when TUI handles tool progress) spinner = None if agent._should_emit_quiet_tool_messages() and agent._should_start_quiet_spinner(): face = random.choice(KawaiiSpinner.get_waiting_faces()) spinner = KawaiiSpinner(f"{face} ⚡ running {num_tools} tools concurrently", spinner_type='dots', print_fn=agent._print_fn) spinner.start() try: runnable_calls = [ (i, tc, name, args) for i, (tc, name, args, middleware_trace, block_result, blocked_by_guardrail) in enumerate(parsed_calls) if block_result is None ] 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 = min(len(runnable_calls), _MAX_TOOL_WORKERS) # Daemon workers: an interrupted/timed-out batch is abandoned with # shutdown(wait=False), but stdlib ThreadPoolExecutor workers are # non-daemon and registered in concurrent.futures' atexit hook, # which joins them unconditionally — so one wedged tool thread # would block interpreter exit forever (multi-minute CLI exits). from tools.daemon_pool import DaemonThreadPoolExecutor executor = DaemonThreadPoolExecutor(max_workers=max_workers) abandon_executor = False try: for submit_index, (i, tc, name, args) in enumerate(runnable_calls): # Propagate the agent turn's ContextVars (e.g. # _approval_session_key) AND thread-local approval/sudo # callbacks into the worker thread; clears callbacks on exit. try: f = executor.submit( propagate_context_to_thread(_run_tool), i, tc, name, args, parsed_calls[i][3] ) except RuntimeError as submit_error: if not _is_interpreter_shutdown_submit_error(submit_error): raise skipped_calls = runnable_calls[submit_index:] logger.warning( "interpreter shutdown while scheduling concurrent tools; " "skipping %d unsubmitted tool(s)", len(skipped_calls), ) for skipped_i, _tc, skipped_name, skipped_args in skipped_calls: if results[skipped_i] is None: middleware_trace = parsed_calls[skipped_i][3] result = ( f"Error executing tool '{skipped_name}': " "Python interpreter is shutting down; tool was not started" ) results[skipped_i] = ( skipped_name, skipped_args, result, 0.0, True, False, middleware_trace, ) break futures.append(f) future_to_index[f] = i # Wait for all to complete with periodic heartbeats so the # gateway's inactivity monitor doesn't kill us during long # concurrent tool batches. Also check for user interrupts # so we don't block indefinitely when the user sends /stop # or a new message during concurrent tool execution. _conc_start = time.time() _interrupt_logged = False while True: wait_timeout = 5.0 if deadline is not None: remaining = deadline - time.monotonic() if remaining <= 0: done, not_done = set(), { f for f in futures if not f.done() } else: wait_timeout = min(wait_timeout, remaining) done, not_done = concurrent.futures.wait( futures, timeout=wait_timeout, ) else: done, not_done = concurrent.futures.wait( futures, timeout=wait_timeout, ) if not not_done: break if deadline is not None and time.monotonic() >= deadline: abandon_executor = True timed_out_indices = { future_to_index[f] for f in not_done if f in future_to_index } _still_running = [ parsed_calls[i][1] for i in timed_out_indices ] logger.warning( "concurrent tool batch timed out after %.1fs; " "%d tool(s) still running: %s", timeout_s, len(timed_out_indices), ", ".join(_still_running[:5]), ) for f in not_done: f.cancel() with agent._tool_worker_threads_lock: worker_tids = list(agent._tool_worker_threads) for tid in worker_tids: try: _ra()._set_interrupt(True, tid) except Exception: pass break # Check for interrupt — the per-thread interrupt signal # already causes individual tools (terminal, execute_code) # to abort, but tools without interrupt checks (web_search, # read_file) will run to completion. Cancel any futures # that haven't started yet so we don't block on them. if agent._interrupt_requested: abandon_executor = True if not _interrupt_logged: _interrupt_logged = True agent._vprint( f"{agent.log_prefix}⚡ Interrupt: cancelling " f"{len(not_done)} pending concurrent tool(s)", force=True, ) for f in not_done: f.cancel() # Give already-running tools a moment to notice the # per-thread interrupt signal and exit gracefully. concurrent.futures.wait(not_done, timeout=3.0) break _conc_elapsed = int(time.time() - _conc_start) # Heartbeat every ~30s (6 × 5s poll intervals) if _conc_elapsed > 0 and _conc_elapsed % 30 < 6: _still_running = [ parsed_calls[future_to_index[f]][1] for f in not_done if f in future_to_index ] agent._touch_activity( f"concurrent tools running ({_conc_elapsed}s, " 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, ) finally: if spinner: # Build a summary message for the spinner stop completed = sum(1 for r in results if r is not None) total_dur = sum(r[3] for r in results if r is not None) spinner.stop(f"⚡ {completed}/{num_tools} tools completed in {total_dur:.1f}s total") # ── Post-execution: display per-tool results ───────────────────── for i, (tc, name, args, middleware_trace, block_result, blocked_by_guardrail) in enumerate(parsed_calls): r = results[i] blocked = False # 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. effect_disposition = None 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}" effect_disposition = "unknown" _emit_terminal_post_tool_call( agent, function_name=name, function_args=args, result=function_result, effective_task_id=effective_task_id, tool_call_id=getattr(tc, "id", "") or "", status="timeout", error_type="tool_timeout", error_message=function_result, middleware_trace=list(middleware_trace), ) tool_duration = float(timeout_s or 0.0) elif r is None: # Tool was cancelled (interrupt) or thread didn't return if agent._interrupt_requested: function_result = f"[Tool execution cancelled — {name} was skipped due to user interrupt]" _emit_terminal_post_tool_call( agent, function_name=name, function_args=args, result=function_result, effective_task_id=effective_task_id, tool_call_id=getattr(tc, "id", "") or "", status="cancelled", error_type="keyboard_interrupt", error_message="Tool execution cancelled by user interrupt", middleware_trace=list(middleware_trace), ) else: function_result = f"Error executing tool '{name}': thread did not return a result" _emit_terminal_post_tool_call( agent, function_name=name, function_args=args, result=function_result, effective_task_id=effective_task_id, tool_call_id=getattr(tc, "id", "") or "", status="error", error_type="thread_missing_result", error_message=function_result, middleware_trace=list(middleware_trace), ) tool_duration = 0.0 else: function_name, function_args, function_result, tool_duration, is_error, blocked, middleware_trace = r if blocked: effect_disposition = "none" if not blocked: function_result = agent._append_guardrail_observation( function_name, function_args, function_result, failed=is_error, ) if is_error: _err_text = _multimodal_text_summary(function_result) result_preview = _err_text[:200] if len(_err_text) > 200 else _err_text logger.warning("Tool %s returned error (%.2fs): %s", function_name, tool_duration, result_preview) # Track file-mutation outcome for the turn-end verifier. # `blocked` calls never actually ran — don't let a guardrail # block count as either a failure or a success. if not blocked: try: agent._record_file_mutation_result( function_name, function_args, function_result, is_error, ) except Exception as _ver_err: logging.debug("file-mutation verifier record failed: %s", _ver_err) if not blocked and agent.tool_progress_callback: try: agent.tool_progress_callback( "tool.completed", function_name, None, None, duration=tool_duration, is_error=is_error, result=function_result, ) except Exception as cb_err: logging.debug(f"Tool progress callback error: {cb_err}") if agent.verbose_logging: logging.debug(f"Tool {function_name} completed in {tool_duration:.2f}s") logging.debug(f"Tool result ({len(function_result)} chars): {function_result}") # Print cute message per tool if agent._should_emit_quiet_tool_messages(): cute_msg = _get_cute_tool_message_impl(name, args, tool_duration, result=function_result) agent._safe_print(f" {cute_msg}") elif not agent.quiet_mode and getattr(agent, "tool_progress_mode", "all") != "off": _preview_str = _multimodal_text_summary(function_result) if agent.verbose_logging: print(f" ✅ Tool {i+1} completed in {tool_duration:.2f}s") print(agent._wrap_verbose("Result: ", _preview_str)) else: response_preview = _preview_str[:agent.log_prefix_chars] + "..." if len(_preview_str) > agent.log_prefix_chars else _preview_str print(f" ✅ Tool {i+1} completed in {tool_duration:.2f}s - {response_preview}") agent._current_tool = None agent._touch_activity(f"tool completed: {name} ({tool_duration:.1f}s)") if not blocked and agent.tool_complete_callback: try: display_args = _redact_tool_args_for_display(name, args) or args agent.tool_complete_callback(tc.id, name, display_args, function_result) except Exception as cb_err: logging.debug(f"Tool complete callback error: {cb_err}") function_result = maybe_persist_tool_result( content=function_result, tool_name=name, tool_use_id=tc.id, env=get_active_env(effective_task_id), config=_tool_budget, ) if not _is_multimodal_tool_result(function_result) else function_result subdir_hints = agent._subdirectory_hints.check_tool_call(name, args) if subdir_hints: if _is_multimodal_tool_result(function_result): # Append the hint to the text summary part so the model # still sees it; don't touch the image blocks. _append_subdir_hint_to_multimodal(function_result, subdir_hints) else: function_result += subdir_hints # Unwrap _multimodal dicts to an OpenAI-style content list so any # vision-capable provider receives [{type:text},{type:image_url}] # rather than a raw Python dict. The Anthropic adapter already # accepts content lists; vision-capable OpenAI-compatible servers # (mlx-vlm, GPT-4o, …) accept image_url in tool messages natively. # Text-only servers get a string-safe fallback here so a rejected # image tool result never poisons canonical session history. # String results pass through unchanged. _tool_content = agent._tool_result_content_for_active_model(name, function_result) tool_message = make_tool_result_message( name, _tool_content, tc.id, effect_disposition=effect_disposition, ) messages.append(tool_message) risk_metadata = tool_message.get("_tool_output_risk") if ( risk_metadata is not None and risk_metadata.get("risk") != "low" and agent.tool_progress_callback ): try: agent.tool_progress_callback( "tool.output_risk", name, None, None, tool_call_id=tc.id, risk_metadata=risk_metadata, ) except Exception as cb_err: logging.debug("Tool output risk callback error: %s", cb_err) _flush_session_db_after_tool_progress( agent, messages, stage=f"tool result {name}", ) # ── Per-tool /steer drain ─────────────────────────────────── # Same as the sequential path: drain between each collected # result so the steer lands as early as possible. agent._apply_pending_steer_to_tool_results(messages, 1) # ── Per-turn aggregate budget enforcement ───────────────────────── num_tools = len(parsed_calls) if finalize and num_tools > 0: turn_tool_msgs = messages[-num_tools:] enforce_turn_budget(turn_tool_msgs, env=get_active_env(effective_task_id), config=_tool_budget) # ── /steer injection ────────────────────────────────────────────── # Append any pending user steer text to the last tool result so the # agent sees it on its next iteration. Runs AFTER budget enforcement # so the steer marker is never truncated. See steer() for details. if finalize and num_tools > 0: agent._apply_pending_steer_to_tool_results(messages, num_tools) def execute_tool_calls_sequential(agent, assistant_message, messages: list, effective_task_id: str, api_call_count: int = 0, *, finalize: bool = True) -> None: """Execute tool calls sequentially (original behavior). Used for single calls or interactive tools. ``finalize=False`` skips the end-of-batch aggregate budget enforcement and /steer injection — used when this call is one segment of a larger mixed batch and the segmented dispatcher owns the turn-end work. """ # Resolve the context-scaled tool-output budget once per turn. _tool_budget = _budget_for_agent(agent) for i, tool_call in enumerate(assistant_message.tool_calls, 1): # SAFETY: check interrupt BEFORE starting each tool. # If the user sent "stop" during a previous tool's execution, # do NOT start any more tools -- skip them all immediately. if agent._interrupt_requested: remaining_calls = assistant_message.tool_calls[i-1:] if remaining_calls: agent._vprint(f"{agent.log_prefix}⚡ Interrupt: skipping {len(remaining_calls)} tool call(s)", force=True) for skipped_tc in remaining_calls: skipped_name = skipped_tc.function.name messages.append(make_tool_result_message( skipped_name, f"[Tool execution cancelled — {skipped_name} was skipped due to user interrupt]", skipped_tc.id, effect_disposition="none", )) _flush_session_db_after_tool_progress( agent, messages, stage=f"cancelled tool result {skipped_name}", ) break function_name = tool_call.function.name function_args, malformed_args_result = _parse_tool_arguments( tool_call.function.arguments ) if malformed_args_result is not None: messages.append( make_tool_result_message( function_name, malformed_args_result, tool_call.id, ) ) _flush_session_db_after_tool_progress( agent, messages, stage=f"invalid tool arguments {function_name}", ) agent._apply_pending_steer_to_tool_results(messages, 1) continue # Tool Search unwrap — see execute_tool_calls_concurrent for full # rationale, including the scope gate (the unwrap dispatches the # underlying tool directly, so session toolset scope is enforced here). _ts_scope_block: Optional[str] = None try: from tools import tool_search as _ts if function_name == _ts.TOOL_CALL_NAME: _underlying, _underlying_args, _err = _ts.resolve_underlying_call(function_args) if not _err and _underlying: if _underlying in _tool_search_scoped_names(agent): function_name = _underlying function_args = _underlying_args else: _ts_scope_block = ( f"'{_underlying}' is not available in this session. " "Use tool_search to find tools you can call." ) except Exception: pass function_args, middleware_trace = _apply_tool_request_middleware_for_agent( agent, function_name=function_name, function_args=function_args, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", ) # Check plugin hooks for a block directive before executing. _block_msg: Optional[str] = None _block_error_type = "plugin_block" if _ts_scope_block is not None: _block_msg = _ts_scope_block _block_error_type = "tool_scope_block" else: try: from hermes_cli.plugins import resolve_pre_tool_block _block_msg = resolve_pre_tool_block( function_name, function_args, task_id=effective_task_id or "", session_id=getattr(agent, "session_id", "") or "", tool_call_id=getattr(tool_call, "id", "") or "", turn_id=getattr(agent, "_current_turn_id", "") or "", api_request_id=getattr(agent, "_current_api_request_id", "") or "", middleware_trace=list(middleware_trace), ) except Exception: pass _guardrail_block_decision: ToolGuardrailDecision | None = None if _block_msg is None: guardrail_decision = agent._tool_guardrails.before_call(function_name, function_args) if not guardrail_decision.allows_execution: _guardrail_block_decision = guardrail_decision _execution_blocked = _block_msg is not None or _guardrail_block_decision is not None if _execution_blocked: # Tool blocked by plugin or guardrail policy — skip counters, # callbacks, checkpointing, activity mutation, and real execution. pass # Reset nudge counters when the relevant tool is actually used elif function_name == "memory": agent._turns_since_memory = 0 elif function_name == "skill_manage": agent._iters_since_skill = 0 if not agent.quiet_mode and getattr(agent, "tool_progress_mode", "all") != "off": display_args = _redact_tool_args_for_display(function_name, function_args) or function_args args_str = json.dumps(display_args, ensure_ascii=False) if agent.verbose_logging: print(f" 📞 Tool {i}: {function_name}({list(display_args.keys())})") print(agent._wrap_verbose("Args: ", json.dumps(display_args, indent=2, ensure_ascii=False))) else: args_preview = args_str[:agent.log_prefix_chars] + "..." if len(args_str) > agent.log_prefix_chars else args_str print(f" 📞 Tool {i}: {function_name}({list(function_args.keys())}) - {args_preview}") if not _execution_blocked: agent._current_tool = function_name agent._touch_activity(f"executing tool: {function_name}") # Set activity callback for long-running tool execution (terminal # commands, etc.) so the gateway's inactivity monitor doesn't kill # the agent while a command is running. if not _execution_blocked: try: from tools.environments.base import set_activity_callback set_activity_callback(agent._touch_activity) except Exception: pass if not _execution_blocked and agent.tool_progress_callback: try: display_args = _redact_tool_args_for_display(function_name, function_args) or function_args preview = _build_tool_preview(function_name, display_args) agent.tool_progress_callback("tool.started", function_name, preview, display_args) except Exception as cb_err: logging.debug(f"Tool progress callback error: {cb_err}") if not _execution_blocked and agent.tool_start_callback: try: display_args = _redact_tool_args_for_display(function_name, function_args) or function_args agent.tool_start_callback(tool_call.id, function_name, display_args) except Exception as cb_err: logging.debug(f"Tool start callback error: {cb_err}") # Checkpoint: snapshot working dir before file-mutating tools if not _execution_blocked and function_name in {"write_file", "patch"} and agent._checkpoint_mgr.enabled: try: _ensure_file_checkpoint( agent, function_name, function_args, effective_task_id, ) except Exception: pass # never block tool execution # Checkpoint before destructive terminal commands if not _execution_blocked and function_name == "terminal" and agent._checkpoint_mgr.enabled: try: cmd = function_args.get("command", "") if _is_destructive_command(cmd): cwd = function_args.get("workdir") or os.getenv("TERMINAL_CWD", os.getcwd()) agent._checkpoint_mgr.ensure_checkpoint( cwd, f"before terminal: {cmd[:60]}" ) except Exception: pass # never block tool execution tool_start_time = time.time() if _block_msg is not None: # Tool blocked by plugin policy — return error without executing. function_result = json.dumps({"error": _block_msg}, ensure_ascii=False) tool_duration = 0.0 _emit_terminal_post_tool_call( agent, function_name=function_name, function_args=function_args, result=function_result, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", status="blocked", error_type=_block_error_type, error_message=_block_msg, middleware_trace=list(middleware_trace), ) elif _guardrail_block_decision is not None: # Tool blocked by tool-loop guardrail — synthesize exactly one # tool result for the original tool_call_id without executing. function_result = agent._guardrail_block_result(_guardrail_block_decision) tool_duration = 0.0 _emit_terminal_post_tool_call( agent, function_name=function_name, function_args=function_args, result=function_result, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", status="blocked", error_type="guardrail_block", error_message=getattr(_guardrail_block_decision, "message", None) or "Tool blocked by guardrail policy", middleware_trace=list(middleware_trace), ) elif function_name == "todo": def _execute(next_args: dict) -> Any: from tools.todo_tool import todo_tool as _todo_tool return _todo_tool( todos=next_args.get("todos"), merge=next_args.get("merge", False), store=agent._todo_store, ) function_result, function_args = _run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", execute=_execute, ) tool_duration = time.time() - tool_start_time if agent._should_emit_quiet_tool_messages(): agent._vprint(f" {_get_cute_tool_message_impl('todo', function_args, tool_duration, result=function_result)}") elif function_name == "session_search": def _execute(next_args: dict) -> Any: session_db = agent._get_session_db_for_recall() if not session_db: from hermes_state import format_session_db_unavailable return json.dumps({"success": False, "error": format_session_db_unavailable()}) from tools.session_search_tool import session_search as _session_search return _session_search( query=next_args.get("query", ""), role_filter=next_args.get("role_filter"), limit=next_args.get("limit", 3), session_id=next_args.get("session_id"), around_message_id=next_args.get("around_message_id"), window=next_args.get("window", 5), sort=next_args.get("sort"), db=session_db, current_session_id=agent.session_id, ) function_result, function_args = _run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", execute=_execute, ) tool_duration = time.time() - tool_start_time if agent._should_emit_quiet_tool_messages(): agent._vprint(f" {_get_cute_tool_message_impl('session_search', function_args, tool_duration, result=function_result)}") elif function_name == "memory": def _execute(next_args: dict) -> Any: target = next_args.get("target", "memory") operations = next_args.get("operations") from tools.memory_tool import memory_tool as _memory_tool result = _memory_tool( action=next_args.get("action"), target=target, content=next_args.get("content"), old_text=next_args.get("old_text"), operations=operations, store=agent._memory_store, ) # Mirror successful built-in memory writes to external # providers. All gating/op-expansion lives behind the manager # interface (MemoryManager.notify_memory_tool_write). if agent._memory_manager: agent._memory_manager.notify_memory_tool_write( result, next_args, build_metadata=lambda: agent._build_memory_write_metadata( task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", None), ), ) return result function_result, function_args = _run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", execute=_execute, ) tool_duration = time.time() - tool_start_time if agent._should_emit_quiet_tool_messages(): agent._vprint(f" {_get_cute_tool_message_impl('memory', function_args, tool_duration, result=function_result)}") elif function_name == "clarify": def _execute(next_args: dict) -> Any: from tools.clarify_tool import clarify_tool as _clarify_tool return _clarify_tool( question=next_args.get("question", ""), choices=next_args.get("choices"), callback=agent.clarify_callback, ) function_result, function_args = _run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", execute=_execute, ) tool_duration = time.time() - tool_start_time if agent._should_emit_quiet_tool_messages(): agent._vprint(f" {_get_cute_tool_message_impl('clarify', function_args, tool_duration, result=function_result)}") elif function_name == "read_terminal": def _execute(next_args: dict) -> Any: from tools.read_terminal_tool import read_terminal_tool as _read_terminal_tool return _read_terminal_tool( start_line=next_args.get("start_line"), count=next_args.get("count"), callback=getattr(agent, "read_terminal_callback", None), ) function_result, function_args = _run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", execute=_execute, ) tool_duration = time.time() - tool_start_time if agent._should_emit_quiet_tool_messages(): agent._vprint(f" {_get_cute_tool_message_impl('read_terminal', function_args, tool_duration, result=function_result)}") elif function_name == "delegate_task": tasks_arg = function_args.get("tasks") if tasks_arg and isinstance(tasks_arg, list): spinner_label = f"🔀 delegating {len(tasks_arg)} tasks · (/agents to monitor)" else: goal_preview = (function_args.get("goal") or "")[:30] spinner_label = ( f"🔀 {goal_preview} · (/agents to monitor)" if goal_preview else "🔀 delegating · (/agents to monitor)" ) spinner = None if agent._should_emit_quiet_tool_messages() and agent._should_start_quiet_spinner(): face = random.choice(KawaiiSpinner.get_waiting_faces()) spinner = KawaiiSpinner(f"{face} {spinner_label}", spinner_type='dots', print_fn=agent._print_fn) spinner.start() agent._delegate_spinner = spinner _delegate_result = None try: def _execute(next_args: dict) -> Any: return agent._dispatch_delegate_task(next_args) function_result, function_args = _run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", execute=_execute, ) _delegate_result = function_result finally: agent._delegate_spinner = None tool_duration = time.time() - tool_start_time cute_msg = _get_cute_tool_message_impl('delegate_task', function_args, tool_duration, result=_delegate_result) if spinner: spinner.stop(cute_msg) elif agent._should_emit_quiet_tool_messages(): agent._vprint(f" {cute_msg}") elif agent._context_engine_tool_names and function_name in agent._context_engine_tool_names: # Context engine tools (lcm_grep, lcm_describe, lcm_expand, etc.) spinner = None if agent._should_emit_quiet_tool_messages(): face = random.choice(KawaiiSpinner.get_waiting_faces()) emoji = _get_tool_emoji(function_name) display_args = _redact_tool_args_for_display(function_name, function_args) or function_args preview = _build_tool_label(function_name, display_args) or function_name spinner = KawaiiSpinner(f"{face} {emoji} {preview}", spinner_type='dots', print_fn=agent._print_fn) spinner.start() _ce_result = None try: def _execute(next_args: dict) -> Any: return agent.context_compressor.handle_tool_call(function_name, next_args, messages=messages) function_result, function_args = _run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", execute=_execute, ) _ce_result = function_result except Exception as tool_error: function_result = json.dumps({"error": f"Context engine tool '{function_name}' failed: {tool_error}"}) logger.error("context_engine.handle_tool_call raised for %s: %s", function_name, tool_error, exc_info=True) finally: tool_duration = time.time() - tool_start_time cute_msg = _get_cute_tool_message_impl(function_name, function_args, tool_duration, result=_ce_result) if spinner: spinner.stop(cute_msg) elif agent._should_emit_quiet_tool_messages(): agent._vprint(f" {cute_msg}") elif agent._memory_manager and agent._memory_manager.has_tool(function_name): # Memory provider tools (hindsight_retain, honcho_search, etc.) # These are not in the tool registry — route through MemoryManager. spinner = None if agent._should_emit_quiet_tool_messages() and agent._should_start_quiet_spinner(): face = random.choice(KawaiiSpinner.get_waiting_faces()) emoji = _get_tool_emoji(function_name) display_args = _redact_tool_args_for_display(function_name, function_args) or function_args preview = _build_tool_label(function_name, display_args) or function_name spinner = KawaiiSpinner(f"{face} {emoji} {preview}", spinner_type='dots', print_fn=agent._print_fn) spinner.start() _mem_result = None try: def _execute(next_args: dict) -> Any: return agent._memory_manager.handle_tool_call(function_name, next_args) function_result, function_args = _run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", execute=_execute, ) _mem_result = function_result except Exception as tool_error: function_result = json.dumps({"error": f"Memory tool '{function_name}' failed: {tool_error}"}) logger.error("memory_manager.handle_tool_call raised for %s: %s", function_name, tool_error, exc_info=True) finally: tool_duration = time.time() - tool_start_time cute_msg = _get_cute_tool_message_impl(function_name, function_args, tool_duration, result=_mem_result) if spinner: spinner.stop(cute_msg) elif agent._should_emit_quiet_tool_messages(): agent._vprint(f" {cute_msg}") elif agent.quiet_mode: spinner = None if agent._should_emit_quiet_tool_messages() and agent._should_start_quiet_spinner(): face = random.choice(KawaiiSpinner.get_waiting_faces()) emoji = _get_tool_emoji(function_name) display_args = _redact_tool_args_for_display(function_name, function_args) or function_args preview = _build_tool_label(function_name, display_args) or function_name spinner = KawaiiSpinner(f"{face} {emoji} {preview}", spinner_type='dots', print_fn=agent._print_fn) spinner.start() _spinner_result = None try: function_result = _ra().handle_function_call( function_name, function_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, enabled_toolsets=getattr(agent, "enabled_toolsets", None), disabled_toolsets=getattr(agent, "disabled_toolsets", None), tool_request_middleware_trace=list(middleware_trace), ) _spinner_result = function_result except KeyboardInterrupt: function_result = _emit_cancelled_terminal_post_tool_call( agent, function_name=function_name, function_args=function_args, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", start_time=tool_start_time, middleware_trace=list(middleware_trace), ) _spinner_result = function_result try: agent.interrupt("keyboard interrupt") except Exception: pass raise except Exception as tool_error: function_result = f"Error executing tool '{function_name}': {tool_error}" logger.error("handle_function_call raised for %s: %s", function_name, tool_error, exc_info=True) finally: tool_duration = time.time() - tool_start_time cute_msg = _get_cute_tool_message_impl(function_name, function_args, tool_duration, result=_spinner_result) if spinner: spinner.stop(cute_msg) elif agent._should_emit_quiet_tool_messages(): agent._vprint(f" {cute_msg}") else: try: function_result = _ra().handle_function_call( function_name, function_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, enabled_toolsets=getattr(agent, "enabled_toolsets", None), disabled_toolsets=getattr(agent, "disabled_toolsets", None), tool_request_middleware_trace=list(middleware_trace), ) except KeyboardInterrupt: _emit_cancelled_terminal_post_tool_call( agent, function_name=function_name, function_args=function_args, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", start_time=tool_start_time, middleware_trace=list(middleware_trace), ) try: agent.interrupt("keyboard interrupt") except Exception: pass raise except Exception as tool_error: function_result = f"Error executing tool '{function_name}': {tool_error}" logger.error("handle_function_call raised for %s: %s", function_name, tool_error, exc_info=True) tool_duration = time.time() - tool_start_time 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 ) _result_len = len(function_result) else: # Multimodal dict result (_multimodal=True) — not sliceable as string result_preview = function_result _result_len = len(str(function_result)) # Log tool errors to the persistent error log so [error] tags # in the UI always have a corresponding detailed entry on disk. _is_error_result, _ = _detect_tool_failure(function_name, function_result) # The agent-runtime tools above (todo, session_search, memory, # 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 _executor_must_emit_post_hook = ( not _execution_blocked and agent_runtime_owns_post_tool_hook(agent, function_name) ) if _executor_must_emit_post_hook: _emit_terminal_post_tool_call( agent, function_name=function_name, function_args=function_args, result=function_result, effective_task_id=effective_task_id, tool_call_id=getattr(tool_call, "id", "") or "", duration_ms=int(tool_duration * 1000), middleware_trace=list(middleware_trace), ) if not _execution_blocked: function_result = agent._append_guardrail_observation( function_name, function_args, function_result, failed=_is_error_result, ) result_preview = function_result if agent.verbose_logging else ( function_result[:200] if len(function_result) > 200 else function_result ) if _is_error_result: logger.warning("Tool %s returned error (%.2fs): %s", function_name, tool_duration, result_preview) else: logger.info("tool %s completed (%.2fs, %d chars)", function_name, tool_duration, _result_len) # Track file-mutation outcome for the turn-end verifier. See # the concurrent path for the rationale; both paths must feed # the same state so the footer reflects every tool call in the # turn, not just the parallel ones. if not _execution_blocked: try: agent._record_file_mutation_result( function_name, function_args, function_result, _is_error_result, ) except Exception as _ver_err: logging.debug("file-mutation verifier record failed: %s", _ver_err) if not _execution_blocked and agent.tool_progress_callback: try: agent.tool_progress_callback( "tool.completed", function_name, None, None, duration=tool_duration, is_error=_is_error_result, result=function_result, ) except Exception as cb_err: logging.debug(f"Tool progress callback error: {cb_err}") agent._current_tool = None agent._touch_activity(f"tool completed: {function_name} ({tool_duration:.1f}s)") if agent.verbose_logging: logging.debug(f"Tool {function_name} completed in {tool_duration:.2f}s") _log_result = _multimodal_text_summary(function_result) logging.debug(f"Tool result ({len(_log_result)} chars): {_log_result}") if not _execution_blocked and agent.tool_complete_callback: try: display_args = _redact_tool_args_for_display(function_name, function_args) or function_args agent.tool_complete_callback(tool_call.id, function_name, display_args, function_result) except Exception as cb_err: logging.debug(f"Tool complete callback error: {cb_err}") function_result = maybe_persist_tool_result( content=function_result, tool_name=function_name, tool_use_id=tool_call.id, env=get_active_env(effective_task_id), config=_tool_budget, ) if not _is_multimodal_tool_result(function_result) else function_result # Discover subdirectory context files from tool arguments subdir_hints = agent._subdirectory_hints.check_tool_call(function_name, function_args) if subdir_hints: if _is_multimodal_tool_result(function_result): _append_subdir_hint_to_multimodal(function_result, subdir_hints) else: function_result += subdir_hints # 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) messages.append(tool_message) risk_metadata = tool_message.get("_tool_output_risk") if ( risk_metadata is not None and risk_metadata.get("risk") != "low" and agent.tool_progress_callback ): try: agent.tool_progress_callback( "tool.output_risk", function_name, None, None, tool_call_id=tool_call.id, risk_metadata=risk_metadata, ) except Exception as cb_err: logging.debug("Tool output risk callback error: %s", cb_err) _flush_session_db_after_tool_progress( agent, messages, stage=f"tool result {function_name}", ) # ── Per-tool /steer drain ─────────────────────────────────── # Drain pending steer BETWEEN individual tool calls so the # injection lands as soon as a tool finishes — not after the # entire batch. The model sees it on the next API iteration. agent._apply_pending_steer_to_tool_results(messages, 1) if not agent.quiet_mode and getattr(agent, "tool_progress_mode", "all") != "off": if agent.verbose_logging: print(f" ✅ Tool {i} completed in {tool_duration:.2f}s") print(agent._wrap_verbose("Result: ", function_result)) else: _fr_str = function_result if isinstance(function_result, str) else str(function_result) response_preview = _fr_str[:agent.log_prefix_chars] + "..." if len(_fr_str) > agent.log_prefix_chars else _fr_str print(f" ✅ Tool {i} completed in {tool_duration:.2f}s - {response_preview}") if agent._interrupt_requested and i < len(assistant_message.tool_calls): remaining = len(assistant_message.tool_calls) - i agent._vprint(f"{agent.log_prefix}⚡ Interrupt: skipping {remaining} remaining tool call(s)", force=True) for skipped_tc in assistant_message.tool_calls[i:]: skipped_name = skipped_tc.function.name messages.append(make_tool_result_message( skipped_name, f"[Tool execution skipped — {skipped_name} was not started. User sent a new message]", skipped_tc.id, effect_disposition="none", )) _flush_session_db_after_tool_progress( agent, messages, stage=f"skipped tool result {skipped_name}", ) break if agent.tool_delay > 0 and i < len(assistant_message.tool_calls): time.sleep(agent.tool_delay) # ── Per-turn aggregate budget enforcement ───────────────────────── num_tools_seq = len(assistant_message.tool_calls) if finalize and num_tools_seq > 0: enforce_turn_budget(messages[-num_tools_seq:], env=get_active_env(effective_task_id), config=_tool_budget) # ── /steer injection ────────────────────────────────────────────── # See _execute_tool_calls_parallel for the rationale. Same hook, # applied to sequential execution as well. if finalize and num_tools_seq > 0: agent._apply_pending_steer_to_tool_results(messages, num_tools_seq) def execute_tool_calls_segmented(agent, assistant_message, messages: list, effective_task_id: str, api_call_count: int = 0, segments=None) -> None: """Execute a mixed tool-call batch as ordered parallel/sequential segments. ``segments`` is the ``(kind, calls)`` plan from ``_plan_tool_batch_segments``: maximal contiguous runs of parallel-safe calls execute on the concurrent path, barrier calls on the sequential path, strictly in the model's original call order. Because segments are contiguous, every tool result is still appended one-per-call in emission order and no call ever starts before an earlier barrier finishes — identical ordering and side-effect boundaries to fully-sequential execution, with I/O parallelism recovered inside the safe runs. Turn-end work (aggregate budget enforcement + /steer injection) is done once here for the WHOLE batch; the per-segment executor calls run with ``finalize=False`` so a multi-segment turn cannot multiply the budget or truncate a steer marker. Interrupt semantics: each segment executor already checks ``agent._interrupt_requested`` up front and appends a cancelled/skipped result per call, so an interrupt during segment *k* drains segments *k+1..n* without executing them while preserving one result per tool_call_id. """ from types import SimpleNamespace if segments is None: _active_env = get_active_env(effective_task_id) _exec_cwd = Path(_active_env.cwd) if _active_env is not None and _active_env.cwd else None segments = _plan_tool_batch_segments(assistant_message.tool_calls, execution_cwd=_exec_cwd) for kind, calls in segments: segment_message = SimpleNamespace(tool_calls=list(calls)) if kind == "parallel": execute_tool_calls_concurrent( agent, segment_message, messages, effective_task_id, api_call_count, finalize=False, ) else: execute_tool_calls_sequential( agent, segment_message, messages, effective_task_id, api_call_count, finalize=False, ) # ── Whole-turn finalize (budget + /steer) ───────────────────────── total_tools = len(assistant_message.tool_calls) if total_tools > 0: _tool_budget = _budget_for_agent(agent) enforce_turn_budget( messages[-total_tools:], env=get_active_env(effective_task_id), config=_tool_budget, ) agent._apply_pending_steer_to_tool_results(messages, total_tools) __all__ = [ "execute_tool_calls_concurrent", "execute_tool_calls_sequential", "execute_tool_calls_segmented", ]