diff --git a/nanobot/agent/runner.py b/nanobot/agent/runner.py index c9a8e7fcc..258d2b802 100644 --- a/nanobot/agent/runner.py +++ b/nanobot/agent/runner.py @@ -19,7 +19,8 @@ from nanobot.agent.context_governance import ( ContextGovernor, ) from nanobot.agent.hook import AgentHook, AgentHookContext, AgentRunHookContext -from nanobot.agent.tools.registry import ToolRegistry, is_tool_error_result +from nanobot.agent.tools.execution import execute_tool_calls +from nanobot.agent.tools.registry import ToolRegistry from nanobot.llm_usage.context import ( LLMUsageSource, bind_llm_usage_source, @@ -32,7 +33,6 @@ from nanobot.providers.base import ( LLMUsage, ProviderCallContext, ProviderConversationState, - ToolCallRequest, ) from nanobot.providers.conversation_state import ( ProviderConversationStateController, @@ -60,8 +60,6 @@ from nanobot.utils.runtime import ( build_finalization_retry_message, build_length_recovery_message, is_blank_text, - repeated_external_lookup_error, - repeated_workspace_violation_error, ) ContinuationCallback = Callable[[], str | None] @@ -586,13 +584,14 @@ class AgentRunner: await hook.before_execute_tools(context) - results, new_events = await self._execute_tools( - spec, + results, new_events = await execute_tool_calls( + spec.tools, response.tool_calls, - external_lookup_counts, - workspace_violation_counts, - hook, - context, + concurrent=spec.concurrent_tools, + external_lookup_counts=external_lookup_counts, + workspace_violation_counts=workspace_violation_counts, + hook=hook, + context=context, ) tool_events.extend(new_events) tools_used.extend( @@ -1385,253 +1384,6 @@ class AgentRunner: return left return left + right - async def _execute_tools( - self, - spec: AgentRunSpec, - tool_calls: list[ToolCallRequest], - external_lookup_counts: dict[str, int], - workspace_violation_counts: dict[str, int], - hook: AgentHook | None = None, - context: AgentHookContext | None = None, - ) -> tuple[list[Any], list[dict[str, str]]]: - hook = hook or AgentHook() - context = context or AgentHookContext(iteration=0, messages=[]) - batches = self._partition_tool_batches(spec, tool_calls) - tool_results: list[tuple[Any, dict[str, str]]] = [] - for batch in batches: - if spec.concurrent_tools and len(batch) > 1: - batch_results = await asyncio.gather(*( - self._run_tool( - spec, - tool_call, - external_lookup_counts, - workspace_violation_counts, - hook, - context, - ) - for tool_call in batch - )) - tool_results.extend(batch_results) - else: - batch_results: list[tuple[Any, dict[str, str]]] = [] - for tool_call in batch: - result = await self._run_tool( - spec, - tool_call, - external_lookup_counts, - workspace_violation_counts, - hook, - context, - ) - tool_results.append(result) - batch_results.append(result) - - results: list[Any] = [] - events: list[dict[str, str]] = [] - for result, event in tool_results: - results.append(result) - events.append(event) - return results, events - - async def _run_tool( - self, - spec: AgentRunSpec, - tool_call: ToolCallRequest, - external_lookup_counts: dict[str, int], - workspace_violation_counts: dict[str, int], - hook: AgentHook | None = None, - context: AgentHookContext | None = None, - ) -> tuple[Any, dict[str, str]]: - hook = hook or AgentHook() - context = context or AgentHookContext(iteration=0, messages=[]) - hint = "\n\n[Analyze the error above and try a different approach.]" - lookup_error = repeated_external_lookup_error( - tool_call.name, - tool_call.arguments, - external_lookup_counts, - ) - if lookup_error: - event = { - "name": tool_call.name, - "status": "error", - "detail": "repeated external lookup blocked", - } - return lookup_error + hint, event - prepare_call = cast( - Callable[[str, Any], object] | None, - getattr(spec.tools, "prepare_call", None), - ) - tool, params, prep_error = None, tool_call.arguments, None - if callable(prepare_call): - prepared = prepare_call(tool_call.name, tool_call.arguments) - if isinstance(prepared, tuple): - prepared_tuple = cast(tuple[object, ...], prepared) - if len(prepared_tuple) == 3: - tool, params, prep_error = cast(tuple[Any, Any, str | None], prepared_tuple) - if prep_error: - event = { - "name": tool_call.name, - "status": "error", - "detail": prep_error.split(": ", 1)[-1][:120], - } - handled = self._classify_violation( - raw_text=prep_error, - soft_payload=prep_error + hint, - event=event, - tool_call=tool_call, - workspace_violation_counts=workspace_violation_counts, - ) - if handled is not None: - return handled - return prep_error + hint, event - await hook.before_execute_tool(context, tool_call, tool, params) - try: - if tool is not None: - result = await tool.execute(**params) - else: - result = await spec.tools.execute(tool_call.name, params) - except asyncio.CancelledError: - raise - except Exception as exc: - await hook.on_execute_tool_error(context, tool_call, tool, params, exc) - event = { - "name": tool_call.name, - "status": "error", - "detail": str(exc), - } - payload = f"Error: {type(exc).__name__}: {exc}" - handled = self._classify_violation( - raw_text=str(exc), - # Preserve legacy exception payloads without the retry hint. - soft_payload=payload, - event=event, - tool_call=tool_call, - workspace_violation_counts=workspace_violation_counts, - ) - if handled is not None: - return handled - return payload, event - - if is_tool_error_result(result): - await hook.on_execute_tool_error(context, tool_call, tool, params, result) - event = { - "name": tool_call.name, - "status": "error", - "detail": result.replace("\n", " ").strip()[:120], - } - handled = self._classify_violation( - raw_text=result, - soft_payload=result + hint, - event=event, - tool_call=tool_call, - workspace_violation_counts=workspace_violation_counts, - ) - if handled is not None: - return handled - return result + hint, event - - await hook.after_execute_tool(context, tool_call, tool, params, result) - - detail = "" if result is None else str(result) - detail = detail.replace("\n", " ").strip() - if not detail: - detail = "(empty)" - elif len(detail) > 120: - detail = detail[:120] + "..." - return result, {"name": tool_call.name, "status": "ok", "detail": detail} - - # SSRF is a hard security block at the tool boundary, but the agent turn - # should recover conversationally instead of aborting the runtime. - _SSRF_MARKERS: tuple[str, ...] = ( - "internal/private url detected", - "private/internal address", - "private address", - ) - _SSRF_BOUNDARY_NOTE: str = ( - "This is a non-bypassable security boundary. Stop trying to access " - "private/internal URLs. Do not retry with curl, wget, encoded IPs, " - "alternate DNS, redirects, proxies, or another tool. Ask the user for " - "local files, logs, screenshots, or an explicit safe public URL instead. " - "If the user explicitly trusts this private URL, ask them to whitelist " - "the exact IP/CIDR via tools.ssrfWhitelist." - ) - - # Non-SSRF boundary markers returned to the LLM as recoverable tool errors. - _WORKSPACE_VIOLATION_MARKERS: tuple[str, ...] = ( - "outside the configured workspace", - "outside allowed directory", - "working_dir is outside", - "working_dir could not be resolved", - "path outside working dir", - "path traversal detected", - ) - - @classmethod - def _is_ssrf_violation(cls, text: str) -> bool: - if not text: - return False - lowered = text.lower() - return any(marker in lowered for marker in cls._SSRF_MARKERS) - - @classmethod - def _is_workspace_violation(cls, text: str) -> bool: - """True when *text* looks like any policy boundary rejection.""" - if not text: - return False - lowered = text.lower() - if cls._is_ssrf_violation(lowered): - return True - return any(marker in lowered for marker in cls._WORKSPACE_VIOLATION_MARKERS) - - def _classify_violation( - self, - *, - raw_text: str, - soft_payload: str, - event: dict[str, str], - tool_call: ToolCallRequest, - workspace_violation_counts: dict[str, int], - ) -> tuple[Any, dict[str, str]] | None: - """Classify safety-boundary failures, or return ``None`` to pass through.""" - if self._is_ssrf_violation(raw_text): - logger.warning( - "Tool {} blocked by SSRF guard; returning non-retryable tool error: {}", - tool_call.name, - raw_text.replace("\n", " ").strip()[:200], - ) - event["detail"] = self._event_detail("ssrf_violation: ", raw_text) - return self._ssrf_soft_payload(raw_text), event - - if self._is_workspace_violation(raw_text): - escalation = repeated_workspace_violation_error( - tool_call.name, - tool_call.arguments, - workspace_violation_counts, - ) - event["detail"] = self._event_detail("workspace_violation: ", raw_text) - if escalation is not None: - logger.warning( - "Tool {} hit workspace boundary repeatedly; escalating hint", - tool_call.name, - ) - event["detail"] = self._event_detail( - "workspace_violation_escalated: ", - raw_text, - ) - return escalation, event - return soft_payload, event - - return None - - @classmethod - def _ssrf_soft_payload(cls, raw_text: str) -> str: - text = raw_text.strip() or "Error: request blocked by SSRF guard" - return f"{text}\n\n{cls._SSRF_BOUNDARY_NOTE}" - - @staticmethod - def _event_detail(prefix: str, text: str, limit: int = 160) -> str: - return (prefix + text.replace("\n", " ").strip())[:limit] - async def _emit_checkpoint( self, spec: AgentRunSpec, @@ -1661,28 +1413,3 @@ class AgentRunner: if messages and messages[-1].get("role") == "assistant" and not messages[-1].get("tool_calls"): return messages.append(build_assistant_message(_PERSISTED_MODEL_ERROR_PLACEHOLDER)) - - def _partition_tool_batches( - self, - spec: AgentRunSpec, - tool_calls: list[ToolCallRequest], - ) -> list[list[ToolCallRequest]]: - if not spec.concurrent_tools: - return [[tool_call] for tool_call in tool_calls] - - batches: list[list[ToolCallRequest]] = [] - current: list[ToolCallRequest] = [] - for tool_call in tool_calls: - get_tool = cast(Callable[[str], Any] | None, getattr(spec.tools, "get", None)) - tool = get_tool(tool_call.name) if callable(get_tool) else None - can_batch = bool(tool and tool.concurrency_safe) - if can_batch: - current.append(tool_call) - continue - if current: - batches.append(current) - current = [] - batches.append([tool_call]) - if current: - batches.append(current) - return batches diff --git a/nanobot/agent/tools/execution.py b/nanobot/agent/tools/execution.py new file mode 100644 index 000000000..25bde4ae5 --- /dev/null +++ b/nanobot/agent/tools/execution.py @@ -0,0 +1,285 @@ +"""Execute tool calls and turn their outcomes into model observations.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Callable +from typing import Any, cast + +from loguru import logger + +from nanobot.agent.hook import AgentHook, AgentHookContext +from nanobot.agent.tools.registry import ToolRegistry, is_tool_error_result +from nanobot.providers.base import ToolCallRequest +from nanobot.utils.runtime import ( + repeated_external_lookup_error, + repeated_workspace_violation_error, +) + +_RETRY_HINT = "\n\n[Analyze the error above and try a different approach.]" +# SSRF is a hard security block at the tool boundary, but the agent turn +# should recover conversationally instead of aborting the runtime. +_SSRF_MARKERS: tuple[str, ...] = ( + "internal/private url detected", + "private/internal address", + "private address", +) +_SSRF_BOUNDARY_NOTE = ( + "This is a non-bypassable security boundary. Stop trying to access " + "private/internal URLs. Do not retry with curl, wget, encoded IPs, " + "alternate DNS, redirects, proxies, or another tool. Ask the user for " + "local files, logs, screenshots, or an explicit safe public URL instead. " + "If the user explicitly trusts this private URL, ask them to whitelist " + "the exact IP/CIDR via tools.ssrfWhitelist." +) +# Non-SSRF boundary markers returned to the model as recoverable tool errors. +_WORKSPACE_VIOLATION_MARKERS: tuple[str, ...] = ( + "outside the configured workspace", + "outside allowed directory", + "working_dir is outside", + "working_dir could not be resolved", + "path outside working dir", + "path traversal detected", +) + + +async def execute_tool_calls( + tools: ToolRegistry, + tool_calls: list[ToolCallRequest], + *, + concurrent: bool, + external_lookup_counts: dict[str, int], + workspace_violation_counts: dict[str, int], + hook: AgentHook, + context: AgentHookContext, +) -> tuple[list[Any], list[dict[str, str]]]: + """Execute one model response's tool calls in stable result order.""" + tool_results: list[tuple[Any, dict[str, str]]] = [] + for batch in _partition_tool_batches(tools, tool_calls, concurrent=concurrent): + if concurrent and len(batch) > 1: + batch_results = await asyncio.gather(*( + _execute_tool_call( + tools, + tool_call, + external_lookup_counts, + workspace_violation_counts, + hook, + context, + ) + for tool_call in batch + )) + tool_results.extend(batch_results) + else: + for tool_call in batch: + result = await _execute_tool_call( + tools, + tool_call, + external_lookup_counts, + workspace_violation_counts, + hook, + context, + ) + tool_results.append(result) + + results = [result for result, _event in tool_results] + events = [event for _result, event in tool_results] + return results, events + + +async def _execute_tool_call( + tools: ToolRegistry, + tool_call: ToolCallRequest, + external_lookup_counts: dict[str, int], + workspace_violation_counts: dict[str, int], + hook: AgentHook, + context: AgentHookContext, +) -> tuple[Any, dict[str, str]]: + lookup_error = repeated_external_lookup_error( + tool_call.name, + tool_call.arguments, + external_lookup_counts, + ) + if lookup_error: + event = { + "name": tool_call.name, + "status": "error", + "detail": "repeated external lookup blocked", + } + return lookup_error + _RETRY_HINT, event + + prepare_call = cast( + Callable[[str, Any], object] | None, + getattr(tools, "prepare_call", None), + ) + tool, params, prep_error = None, tool_call.arguments, None + if callable(prepare_call): + prepared = prepare_call(tool_call.name, tool_call.arguments) + if isinstance(prepared, tuple): + prepared_tuple = cast(tuple[object, ...], prepared) + if len(prepared_tuple) == 3: + tool, params, prep_error = cast(tuple[Any, Any, str | None], prepared_tuple) + if prep_error: + event = { + "name": tool_call.name, + "status": "error", + "detail": prep_error.split(": ", 1)[-1][:120], + } + handled = _classify_violation( + raw_text=prep_error, + soft_payload=prep_error + _RETRY_HINT, + event=event, + tool_call=tool_call, + workspace_violation_counts=workspace_violation_counts, + ) + if handled is not None: + return handled + return prep_error + _RETRY_HINT, event + + await hook.before_execute_tool(context, tool_call, tool, params) + try: + if tool is not None: + result = await tool.execute(**params) + else: + result = await tools.execute(tool_call.name, params) + except asyncio.CancelledError: + raise + except Exception as exc: + await hook.on_execute_tool_error(context, tool_call, tool, params, exc) + event = { + "name": tool_call.name, + "status": "error", + "detail": str(exc), + } + payload = f"Error: {type(exc).__name__}: {exc}" + handled = _classify_violation( + raw_text=str(exc), + # Preserve legacy exception payloads without the retry hint. + soft_payload=payload, + event=event, + tool_call=tool_call, + workspace_violation_counts=workspace_violation_counts, + ) + if handled is not None: + return handled + return payload, event + + if is_tool_error_result(result): + await hook.on_execute_tool_error(context, tool_call, tool, params, result) + event = { + "name": tool_call.name, + "status": "error", + "detail": result.replace("\n", " ").strip()[:120], + } + handled = _classify_violation( + raw_text=result, + soft_payload=result + _RETRY_HINT, + event=event, + tool_call=tool_call, + workspace_violation_counts=workspace_violation_counts, + ) + if handled is not None: + return handled + return result + _RETRY_HINT, event + + await hook.after_execute_tool(context, tool_call, tool, params, result) + + detail = "" if result is None else str(result) + detail = detail.replace("\n", " ").strip() + if not detail: + detail = "(empty)" + elif len(detail) > 120: + detail = detail[:120] + "..." + return result, {"name": tool_call.name, "status": "ok", "detail": detail} + + +def is_ssrf_violation(text: str) -> bool: + """Return whether a tool error describes a blocked private-network request.""" + if not text: + return False + lowered = text.lower() + return any(marker in lowered for marker in _SSRF_MARKERS) + + +def _is_workspace_violation(text: str) -> bool: + """Return whether text describes any workspace or network boundary rejection.""" + if not text: + return False + lowered = text.lower() + if is_ssrf_violation(lowered): + return True + return any(marker in lowered for marker in _WORKSPACE_VIOLATION_MARKERS) + + +def _classify_violation( + *, + raw_text: str, + soft_payload: str, + event: dict[str, str], + tool_call: ToolCallRequest, + workspace_violation_counts: dict[str, int], +) -> tuple[Any, dict[str, str]] | None: + if is_ssrf_violation(raw_text): + logger.warning( + "Tool {} blocked by SSRF guard; returning non-retryable tool error: {}", + tool_call.name, + raw_text.replace("\n", " ").strip()[:200], + ) + event["detail"] = _event_detail("ssrf_violation: ", raw_text) + return _ssrf_soft_payload(raw_text), event + + if _is_workspace_violation(raw_text): + escalation = repeated_workspace_violation_error( + tool_call.name, + tool_call.arguments, + workspace_violation_counts, + ) + event["detail"] = _event_detail("workspace_violation: ", raw_text) + if escalation is not None: + logger.warning( + "Tool {} hit workspace boundary repeatedly; escalating hint", + tool_call.name, + ) + event["detail"] = _event_detail( + "workspace_violation_escalated: ", + raw_text, + ) + return escalation, event + return soft_payload, event + + return None + + +def _ssrf_soft_payload(raw_text: str) -> str: + text = raw_text.strip() or "Error: request blocked by SSRF guard" + return f"{text}\n\n{_SSRF_BOUNDARY_NOTE}" + + +def _event_detail(prefix: str, text: str, limit: int = 160) -> str: + return (prefix + text.replace("\n", " ").strip())[:limit] + + +def _partition_tool_batches( + tools: ToolRegistry, + tool_calls: list[ToolCallRequest], + *, + concurrent: bool, +) -> list[list[ToolCallRequest]]: + if not concurrent: + return [[tool_call] for tool_call in tool_calls] + + batches: list[list[ToolCallRequest]] = [] + current: list[ToolCallRequest] = [] + for tool_call in tool_calls: + get_tool = cast(Callable[[str], Any] | None, getattr(tools, "get", None)) + tool = get_tool(tool_call.name) if callable(get_tool) else None + can_batch = bool(tool and tool.concurrency_safe) + if can_batch: + current.append(tool_call) + continue + if current: + batches.append(current) + current = [] + batches.append([tool_call]) + if current: + batches.append(current) + return batches diff --git a/tests/agent/test_runner_errors.py b/tests/agent/test_runner_errors.py index 635e09980..f4836e578 100644 --- a/tests/agent/test_runner_errors.py +++ b/tests/agent/test_runner_errors.py @@ -9,7 +9,9 @@ from unittest.mock import AsyncMock, MagicMock import pytest from agent.runner_helpers import make_run_spec +from nanobot.agent.hook import AgentHook, AgentHookContext from nanobot.agent.tools import ToolResult +from nanobot.agent.tools.execution import execute_tool_calls from nanobot.config.schema import AgentDefaults from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest @@ -55,11 +57,7 @@ async def test_runner_returns_tool_exception_to_model_for_recovery(): @pytest.mark.asyncio @pytest.mark.parametrize("control_error", [KeyboardInterrupt, SystemExit]) -async def test_runner_propagates_tool_control_flow_exceptions(control_error: type[BaseException]): - from nanobot.agent.runner import AgentRunner - - provider = MagicMock(spec=LLMProvider) - +async def test_tool_execution_propagates_control_flow_exceptions(control_error: type[BaseException]): async def execute(_name, _args): raise control_error("stop") @@ -67,22 +65,15 @@ async def test_runner_propagates_tool_control_flow_exceptions(control_error: typ get_definitions=lambda: [], execute=execute, ) - runner = AgentRunner() - spec = make_run_spec( - provider, - initial_messages=[], - tools=tools, - model="test-model", - max_iterations=1, - max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, - ) - with pytest.raises(control_error): - await runner._run_tool( - spec, - ToolCallRequest(id="call_1", name="list_dir", arguments={}), + await execute_tool_calls( + tools, + [ToolCallRequest(id="call_1", name="list_dir", arguments={})], + concurrent=False, external_lookup_counts={}, workspace_violation_counts={}, + hook=AgentHook(), + context=AgentHookContext(iteration=0, messages=[]), ) diff --git a/tests/agent/test_runner_safety.py b/tests/agent/test_runner_safety.py index 3819a3e5a..18063412e 100644 --- a/tests/agent/test_runner_safety.py +++ b/tests/agent/test_runner_safety.py @@ -9,6 +9,7 @@ import pytest from agent.runner_helpers import make_run_spec from nanobot.agent.runner import AgentRunner from nanobot.agent.tools import ToolResult +from nanobot.agent.tools.execution import is_ssrf_violation from nanobot.config.schema import AgentDefaults from nanobot.providers.base import LLMResponse, ToolCallRequest @@ -66,20 +67,20 @@ async def test_runner_does_not_abort_on_workspace_violation_anymore(): def test_is_ssrf_violation_recognizes_private_url_blocks(): """SSRF rejections are classified separately from workspace boundaries.""" ssrf_msg = "Error: Command blocked by safety guard (internal/private URL detected)" - assert AgentRunner._is_ssrf_violation(ssrf_msg) is True - assert AgentRunner._is_ssrf_violation( + assert is_ssrf_violation(ssrf_msg) is True + assert is_ssrf_violation( "URL validation failed: Blocked: host resolves to private/internal address 192.168.1.2" ) is True # Workspace-bound markers are NOT classified as SSRF. - assert AgentRunner._is_ssrf_violation( + assert is_ssrf_violation( "Error: Command blocked by safety guard (path outside working dir)" ) is False - assert AgentRunner._is_ssrf_violation( + assert is_ssrf_violation( "Path /tmp/x is outside allowed directory /ws" ) is False # Deny / allowlist filter messages stay non-fatal too. - assert AgentRunner._is_ssrf_violation( + assert is_ssrf_violation( "Error: Command blocked by deny pattern filter" ) is False diff --git a/tests/agent/test_runner_tool_execution.py b/tests/agent/test_runner_tool_execution.py index b3312c99d..8ceaf0866 100644 --- a/tests/agent/test_runner_tool_execution.py +++ b/tests/agent/test_runner_tool_execution.py @@ -3,14 +3,17 @@ from __future__ import annotations import asyncio +from typing import Any from unittest.mock import AsyncMock, MagicMock, patch import pytest from agent.runner_helpers import make_run_spec +from nanobot.agent.hook import AgentHook, AgentHookContext from nanobot.agent.runner import AgentRunner from nanobot.agent.tools.base import Tool, ToolResult from nanobot.agent.tools.context import ToolContext +from nanobot.agent.tools.execution import execute_tool_calls from nanobot.agent.tools.loader import ToolLoader from nanobot.agent.tools.registry import ToolRegistry from nanobot.config.schema import AgentDefaults @@ -150,31 +153,69 @@ def _tool_message(result, tool_call_id: str) -> dict: @pytest.mark.asyncio -async def test_runner_propagates_tool_preparation_failure(): +async def test_tool_execution_propagates_preparation_failure(): tools = MagicMock() tools.prepare_call.side_effect = RuntimeError("tool preparation failed") tools.execute = AsyncMock() with pytest.raises(RuntimeError, match="tool preparation failed"): - await AgentRunner()._run_tool( - make_run_spec( - MagicMock(), - initial_messages=[], - tools=tools, - model="test-model", - max_iterations=1, - max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, - ), - ToolCallRequest(id="call-1", name="demo", arguments={}), - {}, - {}, + await execute_tool_calls( + tools, + [ToolCallRequest(id="call-1", name="demo", arguments={})], + concurrent=False, + external_lookup_counts={}, + workspace_violation_counts={}, + hook=AgentHook(), + context=AgentHookContext(iteration=0, messages=[]), ) tools.execute.assert_not_awaited() @pytest.mark.asyncio -async def test_runner_batches_read_only_tools_before_exclusive_work(): +async def test_tool_execution_propagates_cancellation_without_error_hook(): + tools = MagicMock() + tools.prepare_call.return_value = (None, {}, None) + tools.execute = AsyncMock(side_effect=asyncio.CancelledError) + + events: list[str] = [] + + class RecordingHook(AgentHook): + async def before_execute_tool( + self, + context: AgentHookContext, + tool_call: ToolCallRequest, + tool: Any, + params: Any, + ) -> None: + events.append("before") + + async def on_execute_tool_error( + self, + context: AgentHookContext, + tool_call: ToolCallRequest, + tool: Any, + params: Any, + error: Any, + ) -> None: + events.append("error") + + with pytest.raises(asyncio.CancelledError): + await execute_tool_calls( + tools, + [ToolCallRequest(id="call-1", name="demo", arguments={})], + concurrent=False, + external_lookup_counts={}, + workspace_violation_counts={}, + hook=RecordingHook(), + context=AgentHookContext(iteration=0, messages=[]), + ) + + assert events == ["before"] + + +@pytest.mark.asyncio +async def test_tool_execution_batches_read_only_tools_before_exclusive_work(): tools = ToolRegistry() shared_events: list[str] = [] read_a = _DelayTool("read_a", delay=0.05, read_only=True, shared_events=shared_events) @@ -184,24 +225,18 @@ async def test_runner_batches_read_only_tools_before_exclusive_work(): tools.register(read_b) tools.register(write_a) - provider = MagicMock() - runner = AgentRunner() - await runner._execute_tools( - make_run_spec(provider, - initial_messages=[], - tools=tools, - model="test-model", - max_iterations=1, - max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, - concurrent_tools=True, - ), + await execute_tool_calls( + tools, [ ToolCallRequest(id="ro1", name="read_a", arguments={}), ToolCallRequest(id="ro2", name="read_b", arguments={}), ToolCallRequest(id="rw1", name="write_a", arguments={}), ], - {}, - {}, + concurrent=True, + external_lookup_counts={}, + workspace_violation_counts={}, + hook=AgentHook(), + context=AgentHookContext(iteration=0, messages=[]), ) assert shared_events[0:2] == ["start:read_a", "start:read_b"] @@ -212,7 +247,7 @@ async def test_runner_batches_read_only_tools_before_exclusive_work(): @pytest.mark.asyncio -async def test_runner_does_not_batch_exclusive_read_only_tools(): +async def test_tool_execution_does_not_batch_exclusive_read_only_tools(): tools = ToolRegistry() shared_events: list[str] = [] read_a = _DelayTool("read_a", delay=0.03, read_only=True, shared_events=shared_events) @@ -228,24 +263,18 @@ async def test_runner_does_not_batch_exclusive_read_only_tools(): tools.register(ddg_like) tools.register(read_b) - provider = MagicMock() - runner = AgentRunner() - await runner._execute_tools( - make_run_spec(provider, - initial_messages=[], - tools=tools, - model="test-model", - max_iterations=1, - max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, - concurrent_tools=True, - ), + await execute_tool_calls( + tools, [ ToolCallRequest(id="ro1", name="read_a", arguments={}), ToolCallRequest(id="ddg1", name="ddg_like", arguments={}), ToolCallRequest(id="ro2", name="read_b", arguments={}), ], - {}, - {}, + concurrent=True, + external_lookup_counts={}, + workspace_violation_counts={}, + hook=AgentHook(), + context=AgentHookContext(iteration=0, messages=[]), ) assert shared_events[0] == "start:read_a" diff --git a/tests/agent/tools/test_subagent_tools.py b/tests/agent/tools/test_subagent_tools.py index 92ecc451b..ded12e721 100644 --- a/tests/agent/tools/test_subagent_tools.py +++ b/tests/agent/tools/test_subagent_tools.py @@ -376,9 +376,10 @@ async def test_inline_spawn_waits_for_concurrency_capacity(tmp_path): @pytest.mark.asyncio async def test_runner_executes_inline_spawn_batch_concurrently(tmp_path): """Adjacent blocking consultations should share one concurrent tool batch.""" - from nanobot.agent.runner import AgentRunner, AgentRunSpec + from nanobot.agent.hook import AgentHook, AgentHookContext from nanobot.agent.subagent import SubagentManager from nanobot.agent.tools.context import RequestContext, request_context + from nanobot.agent.tools.execution import execute_tool_calls from nanobot.agent.tools.registry import ToolRegistry from nanobot.agent.tools.spawn import SpawnTool from nanobot.bus.queue import MessageBus @@ -410,14 +411,6 @@ async def test_runner_executes_inline_spawn_batch_concurrently(tmp_path): tools = ToolRegistry() tools.register(SpawnTool(manager)) runtime = _runtime(MagicMock()) - spec = AgentRunSpec( - initial_messages=[], - tools=tools, - runtime=runtime, - max_iterations=1, - max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, - concurrent_tools=True, - ) calls = [ ToolCallRequest( id="spawn-1", @@ -437,7 +430,15 @@ async def test_runner_executes_inline_spawn_batch_concurrently(tmp_path): session_key="test:c1", runtime=runtime, )): - execution = asyncio.create_task(AgentRunner()._execute_tools(spec, calls, {}, {})) + execution = asyncio.create_task(execute_tool_calls( + tools, + calls, + concurrent=True, + external_lookup_counts={}, + workspace_violation_counts={}, + hook=AgentHook(), + context=AgentHookContext(iteration=0, messages=[], session_key="test:c1"), + )) await asyncio.wait_for(both_entered.wait(), timeout=1.0) release.set() results, events = await execution