From c92aac2a4ec2a91d602851e0cf48dff015f8eda0 Mon Sep 17 00:00:00 2001 From: Jay Prajapati <79649559+jayy-77@users.noreply.github.com> Date: Fri, 18 Sep 2026 23:13:59 +0530 Subject: [PATCH] feat(flows): dedupe identical tool calls within an invocation Models re-emit an identical tool call while a slow tool is still running or repeat it in a later step; the reporter of #3940 saw production tools loop this way, and long-running tools re-fire while the model waits. Wrap the tool execution step of the shared tool-calling pipeline in an invocation-scoped single-flight cache keyed by agent, branch, tool name and canonical arguments: the first call runs the tool and publishes a snapshot that concurrent and later duplicates reuse as their own copy. It is opt-in via RunConfig.dedupe_tool_calls and always on for LongRunningFunctionTool; callbacks still run per call, a failed run is evicted so a later call retries, and a result that transfers, escalates or requests confirmation or auth is never shared. Implemented with Claude Code (Claude Fable 5.1); the author reviewed and tested the change. Closes #3940 --- docs/guides/runners/runner/index.md | 1 + src/google/adk/agents/invocation_context.py | 31 + src/google/adk/agents/run_config.py | 26 + .../flows/llm_flows/tools/_batch_executor.py | 9 + .../adk/flows/llm_flows/tools/_caller.py | 246 +++++- tests/unittests/agents/test_run_config.py | 5 + .../llm_flows/tools/test_batch_executor.py | 25 + .../flows/llm_flows/tools/test_caller.py | 277 +++++++ .../llm_flows/tools/test_functions_dedupe.py | 743 ++++++++++++++++++ 9 files changed, 1360 insertions(+), 3 deletions(-) create mode 100644 tests/unittests/flows/llm_flows/tools/test_functions_dedupe.py diff --git a/docs/guides/runners/runner/index.md b/docs/guides/runners/runner/index.md index 08dfda1907d..1e15514773e 100644 --- a/docs/guides/runners/runner/index.md +++ b/docs/guides/runners/runner/index.md @@ -117,6 +117,7 @@ Passed per-invocation to `runner.run_async(..., run_config=RunConfig(...))`: | `get_session_config` | `GetSessionConfig \| None` | `None` | Fine-grained session retrieval and event window loading configuration. | | `model_input_context` | `list[types.Content] \| None` | `None` | Transient unpersisted context added to model input for the current invocation. | | `max_llm_calls` | `int` | `500` | Maximum limit on LLM calls per run execution. | +| `dedupe_tool_calls` | `bool` | `False` | Opt-in sharing of one execution per invocation between identical tool calls (same agent, branch, tool name and arguments). Callbacks still run per call; a result that transfers, escalates or requests confirmation or auth is never shared; a reused result is marked with `custom_metadata['adk_tool_call_cache_hit']`. `LongRunningFunctionTool` calls are always deduped. | ## Advanced applications diff --git a/src/google/adk/agents/invocation_context.py b/src/google/adk/agents/invocation_context.py index 5a8eb086d2a..b9ad6ddbfa1 100644 --- a/src/google/adk/agents/invocation_context.py +++ b/src/google/adk/agents/invocation_context.py @@ -15,6 +15,7 @@ from __future__ import annotations import asyncio +import dataclasses from typing import Any from google.adk.platform import uuid as platform_uuid @@ -105,6 +106,24 @@ def __init__( self.aborted = False +@dataclasses.dataclass +class _ToolCallCacheEntry: + """One tool execution shared by the identical calls of an invocation. + + Attributes: + future: Resolves to the tool result once the first call for the key has + run the tool, or to the exception that run raised. + shareable: Whether identical calls may reuse the result. False when the + run recorded an action other than a state or artifact delta, such as a + transfer, an escalation or a request for authentication or + confirmation: that action is the effect of the call that made it, and + reusing the result would not replay it. + """ + + future: asyncio.Future[object] + shareable: bool = True + + class InvocationContext(BaseModel): """An invocation context represents the data of a single invocation of an agent. @@ -282,6 +301,18 @@ class InvocationContext(BaseModel): _abort_state: _AbortState = PrivateAttr(default_factory=_AbortState) """Captured abort state (signal, loop, and aborted flag) shared across copies.""" + _tool_call_cache: dict[tuple[Any, ...], _ToolCallCacheEntry] = PrivateAttr( + default_factory=dict + ) + """Tool executions shared by the identical tool calls of this invocation. + + Keyed by agent name, branch, tool name and canonical arguments. Which calls + are deduped, and how one execution is shared, is decided by the tool caller + in the LLM flow; the context only holds the entries. Like ``_abort_state``, the dict is + the very same object in every ``model_copy()`` clone of this context, so an + identical call made by a sub-agent finds what its ancestor already ran. + """ + @override def model_post_init(self, __context: Any) -> None: super().model_post_init(__context) diff --git a/src/google/adk/agents/run_config.py b/src/google/adk/agents/run_config.py index 2bb2588af92..033a9985143 100644 --- a/src/google/adk/agents/run_config.py +++ b/src/google/adk/agents/run_config.py @@ -243,6 +243,32 @@ class RunConfig(BaseModel): - Less than or equal to 0: This allows for unbounded number of llm calls. """ + dedupe_tool_calls: bool = False + """Whether identical tool calls run only once per invocation. + + When enabled, a tool call whose name and arguments match one the same agent + already made in this invocation, on the same agent branch, reuses that + call's result instead of running the tool again. Models sometimes re-emit a + call while a slow or expensive tool is still running, or repeat it in a + later step; deduping saves the repeated execution and keeps the two + responses consistent. The cache lives for one invocation; a resumed + invocation starts with an empty one. + + Only the tool execution is shared. Before-tool and after-tool callbacks still + run for every call, every call gets its own function response event, and a + duplicate receives its own copy of the result as the tool returned it. State + and artifact deltas come from the first execution only. A result is not + shared when its run recorded any other action (a transfer, an escalation, a + request for authentication or confirmation...), since that action is the + effect of the call that made it; a failed execution is not reused either. A + reused result is marked with ``custom_metadata['adk_tool_call_cache_hit'] = + True`` on its function response event (on the merged event when the + responses of one step are merged). + + ``LongRunningFunctionTool`` calls are always deduped, whatever this setting + is. + """ + custom_metadata: Optional[dict[str, Any]] = None """Custom metadata for the current invocation.""" diff --git a/src/google/adk/flows/llm_flows/tools/_batch_executor.py b/src/google/adk/flows/llm_flows/tools/_batch_executor.py index a943fb19cd8..c94dafbf796 100644 --- a/src/google/adk/flows/llm_flows/tools/_batch_executor.py +++ b/src/google/adk/flows/llm_flows/tools/_batch_executor.py @@ -101,6 +101,14 @@ def merge_parallel_function_response_events( merged_actions = EventActions.model_validate(merged_actions_data) + # Metadata is merged too, so what a call's own event says about its response + # (such as that the result was reused from an identical call) is still said + # once the responses of a step are combined. + merged_metadata: dict[str, Any] = {} + for event in function_response_events: + if event.custom_metadata: + merged_metadata.update(event.custom_metadata) + # Create the new merged event merged_event = Event( invocation_id=base_event.invocation_id, @@ -109,6 +117,7 @@ def merge_parallel_function_response_events( content=types.Content(role='user', parts=merged_parts), actions=merged_actions, live_session_id=base_event.live_session_id, + custom_metadata=merged_metadata or None, ) # Use the base_event as the timestamp diff --git a/src/google/adk/flows/llm_flows/tools/_caller.py b/src/google/adk/flows/llm_flows/tools/_caller.py index 4e5f522aa0a..7948582eeb6 100644 --- a/src/google/adk/flows/llm_flows/tools/_caller.py +++ b/src/google/adk/flows/llm_flows/tools/_caller.py @@ -39,13 +39,16 @@ from google.genai import types from . import _error_handler as _tool_error_handler +from ....agents.invocation_context import _ToolCallCacheEntry from ....events.event import Event +from ....events.event_actions import EventActions from ....live._active_streaming_tool import ActiveStreamingTool from ....live.live_request_queue import LiveRequestQueue from ....telemetry import _instrumentation from ....tools.base_tool import BaseTool from ....tools.function_tool import _use_sync_callable_runner from ....tools.function_tool import FunctionTool +from ....tools.long_running_tool import LongRunningFunctionTool from ....tools.tool_confirmation import ToolConfirmation from ....tools.tool_context import ToolContext from ....utils._callback_pipeline import _run_callbacks @@ -80,6 +83,10 @@ _MESSAGE_EVENT_FIELDS = frozenset({'content', 'id', 'timestamp'}) +# Set in the custom metadata of a function response event whose result was +# reused from an earlier identical call instead of running the tool again. +_TOOL_CALL_CACHE_HIT_KEY = 'adk_tool_call_cache_hit' + def _is_live_request_queue_annotation(param: inspect.Parameter) -> bool: """Check whether a parameter is annotated as LiveRequestQueue. @@ -751,6 +758,225 @@ async def _apply_confirmation_gate( return None +def _canonicalize_tool_args(value: object) -> object: + """Returns a stable, hashable form of a tool argument value. + + A dict becomes a tuple of ``(key, value)`` pairs sorted by key and a list or + tuple becomes a tuple, so the order of the keys in a JSON object does not + matter. Every value is paired with its type name: ``1``, ``True`` and + ``1.0`` hash and compare equal in Python while being different arguments to + a tool, and an empty dict would otherwise look like an empty list. Anything + else falls back to its ``repr``. + """ + if isinstance(value, dict): + return ( + 'dict', + tuple( + sorted( + ( + (key, _canonicalize_tool_args(item)) + for key, item in value.items() + ), + key=lambda pair: str(pair[0]), + ) + ), + ) + if isinstance(value, (list, tuple)): + return ('list', tuple(_canonicalize_tool_args(item) for item in value)) + if value is None or isinstance(value, (str, int, float, bool)): + return (type(value).__name__, value) + return repr(value) + + +def _tool_call_cache_key( + invocation_context: InvocationContext, + prepared_call: _PreparedFunctionCall, +) -> tuple[Any, ...]: + """Returns the key under which identical calls share one tool execution. + + The key is the calling agent, its branch, the tool name and the canonical + arguments as the tool receives them (after the before-tool callbacks). + Keying on the agent keeps the same-named tools of two agents apart, since a + tool name is only unique within one agent. Keying on the branch means that + parallel sibling agents, which run on branches of their own, each execute + the call once, while the identical calls one agent makes, in one step or + across its steps, share a single execution. + """ + return ( + _require_agent_name(invocation_context), + invocation_context.branch, + prepared_call.tool.name, + _canonicalize_tool_args(prepared_call.function_args), + ) + + +def _should_dedupe_tool_call( + invocation_context: InvocationContext, + prepared_call: _PreparedFunctionCall, +) -> bool: + """Whether a call may share one tool execution with identical calls. + + Deduping is opted into per run with ``RunConfig.dedupe_tool_calls``. A + ``LongRunningFunctionTool`` is always deduped: its real response arrives + later, and re-firing the call is exactly what a model does while it waits. + Other tools flagged as long-running, such as a workflow node wrapped as a + tool, run to completion and return their real result, so they follow the + run's setting. A call is never deduped when: + + - the tool name resolved to nothing, since that path answers the call on + its own before the tool would run; + - the tool defers its response, since another orchestrator synthesizes the + response and expects one run per function call id; + - it carries the end user's answer to a confirmation request, since the + answer is bound to one function call id and identical calls may have + been answered differently; + - it is the ``stop_streaming`` live control operation; + - the tool is a live streaming tool, since its return value is only a + pending marker and the results stream through the live queue. + """ + tool = prepared_call.tool + if prepared_call.tool_lookup_error is not None or tool._defers_response: + return False + if prepared_call.tool_context.tool_confirmation is not None: + return False + if prepared_call.function_call.name == 'stop_streaming': + return False + if inspect.isasyncgenfunction(getattr(tool, 'func', None)): + return False + if isinstance(tool, LongRunningFunctionTool): + return True + run_config = invocation_context.run_config + return run_config is not None and run_config.dedupe_tool_calls + + +# The actions a tool run may record and still have its result reused: applying +# a state or artifact delta once is the very point of running the tool once. +_SHAREABLE_ACTION_FIELDS = frozenset({'state_delta', 'artifact_delta'}) + + +def _tool_result_is_shareable(actions: EventActions) -> bool: + """Whether identical calls may reuse the result of the run that set `actions`. + + Any action other than a state or artifact delta (a transfer, an escalation, + a request for authentication or confirmation, a UI widget...) is the effect + of the call that recorded it on the flow of that call. A duplicate reusing + the result builds its event from its own, untouched actions and would not + replay it, so such a result is not shared. The built-in control tools, for + instance, return nothing and act through their actions alone. + """ + recorded = actions.model_dump(exclude_none=True, exclude_defaults=True) + return recorded.keys() <= _SHAREABLE_ACTION_FIELDS + + +def _copy_tool_result(result: object) -> object: + """Returns a deep copy of a tool result, or the result itself if it has none. + + A result may hold something that cannot be copied, such as a lock or a + client handle. Sharing it as is then beats failing a duplicate whose tool + run did succeed. + """ + try: + return copy.deepcopy(result) + except Exception: + logger.debug( + 'A tool result cannot be deep-copied; identical calls share it as is.', + exc_info=True, + ) + return result + + +async def _run_tool_single_flight( + invocation_context: InvocationContext, + prepared_call: _PreparedFunctionCall, + *, + tool_runner: Callable[[], Awaitable[Any]], +) -> tuple[object, bool]: + """Runs the tool once on behalf of every identical call of the invocation. + + The first call for a key runs ``tool_runner`` in its own task and publishes + the result on a future. Identical calls running at the same time await that + future; identical calls made later read it once completed. Only the tool + execution is shared: the caller still runs the callbacks and builds the + response event of every call. What is published is a snapshot of the result + taken before any after-tool callback of the first call can alter it, and + every duplicate receives its own copy of that snapshot, so no call's + callbacks reach another call's event. + + A failed execution is evicted so that a later identical call retries, and + every waiter re-raises the failure to its own on-tool-error callbacks. A + result whose run recorded an action other than a state or artifact delta + (a transfer, an escalation, a request for authentication or confirmation) + is the effect of that call alone, so it is evicted and flagged as not + shareable, and waiters run the tool themselves. + + Args: + invocation_context: The invocation whose calls share executions. + prepared_call: The call to run or answer from an earlier run. + tool_runner: An async callable that runs the tool for this call. + + Returns: + The tool result and whether it was reused from an earlier call. + """ + cache = invocation_context._tool_call_cache + key = _tool_call_cache_key(invocation_context, prepared_call) + # No lock is needed: every read and write of the cache happens on the event + # loop thread, and there is no await between looking a key up and claiming + # it, so the check and the insert are atomic. + entry = cache.get(key) + if entry is None: + entry = _ToolCallCacheEntry( + future=asyncio.get_running_loop().create_future() + ) + cache[key] = entry + try: + result = await tool_runner() + except BaseException as error: + if cache.get(key) is entry: + del cache[key] + if isinstance(error, Exception): + entry.future.set_exception(error) + # Reading the exception back marks it as retrieved, so that asyncio + # does not log it as never retrieved when no identical call waits. + entry.future.exception() + else: + entry.future.cancel() + raise + if _tool_result_is_shareable(prepared_call.tool_context.actions): + entry.future.set_result(_copy_tool_result(result)) + else: + entry.shareable = False + # Waiters run the tool themselves because of the flag; the eviction only + # keeps the entry from lingering. + if cache.get(key) is entry: + del cache[key] + entry.future.set_result(None) + return result, False + + # Shielded so that cancelling a waiter cannot cancel the shared future out + # from under the call that is running the tool and the other waiters. + snapshot = await asyncio.shield(entry.future) + if not entry.shareable: + return await tool_runner(), False + logger.debug( + 'Reusing the result of an identical `%s` call for function call %s.', + prepared_call.tool.name, + prepared_call.function_call.id, + ) + return _copy_tool_result(snapshot), True + + +def _mark_tool_call_cache_hit(event: Event) -> None: + """Flags a response event as carrying the result of an earlier identical call. + + The flag is merged into the event's custom metadata, so whatever metadata is + already there is kept. + """ + event.custom_metadata = { + **(event.custom_metadata or {}), + _TOOL_CALL_CACHE_HIT_KEY: True, + } + + async def _execute_single_prepared_call( invocation_context: InvocationContext, prepared_call: _PreparedFunctionCall, @@ -787,6 +1013,7 @@ async def _run_with_trace() -> Event | None: 5. Building the final FunctionResponse Event to be returned. """ nonlocal function_response, detected_error_type + cache_hit = False # Step 1: Check if plugin before_tool_callback overrides the function # response. @@ -836,14 +1063,24 @@ async def _run_with_trace() -> Event | None: # the tool normally. A tool that requires confirmation is answered by the # gate instead, so the gate holds for every tool rather than only the ones # that check it themselves, and a gate that raises is handled like a tool - # that raises. + # that raises. The gate runs before any deduping, so a held call is + # answered per function call id and never shared. Only the tool execution + # itself is shared between identical calls: the callbacks around it are + # per-call hooks (guardrails, logging plugins), and running them once would + # leave plugin telemetry asymmetric with the response events, one of which + # the flow emits per call. if function_response is None: try: function_response = await _apply_confirmation_gate( tool, function_args, tool_context ) if function_response is None: - function_response = await tool_runner() + if _should_dedupe_tool_call(invocation_context, prepared_call): + function_response, cache_hit = await _run_tool_single_flight( + invocation_context, prepared_call, tool_runner=tool_runner + ) + else: + function_response = await tool_runner() except Exception as tool_error: error_response = await _tool_error_handler.run_on_tool_error_callbacks( invocation_context=invocation_context, @@ -906,9 +1143,12 @@ async def _run_with_trace() -> Event | None: # Note: State deltas are not applied here - they are collected in # tool_context.actions.state_delta and applied later when the session # service processes the events - return _build_response_event( + function_response_event = _build_response_event( tool, function_response, tool_context, invocation_context ) + if cache_hit: + _mark_tool_call_cache_hit(function_response_event) + return function_response_event async with _instrumentation.record_tool_execution( tool, agent, function_args, invocation_context=invocation_context diff --git a/tests/unittests/agents/test_run_config.py b/tests/unittests/agents/test_run_config.py index 01a4d916a3a..1e8f306cac6 100644 --- a/tests/unittests/agents/test_run_config.py +++ b/tests/unittests/agents/test_run_config.py @@ -219,3 +219,8 @@ def test_max_llm_calls_invalid_env_var_warning(monkeypatch): assert config.max_llm_calls == 500 mock_warning.assert_called_once() assert "Invalid value for ADK_MAX_LLM_CALLS" in mock_warning.call_args[0][0] + + +def test_dedupe_tool_calls_is_off_by_default(): + """Deduping identical tool calls is opt-in.""" + assert RunConfig().dedupe_tool_calls is False diff --git a/tests/unittests/flows/llm_flows/tools/test_batch_executor.py b/tests/unittests/flows/llm_flows/tools/test_batch_executor.py index 951c93db742..7ae9fb72ddf 100644 --- a/tests/unittests/flows/llm_flows/tools/test_batch_executor.py +++ b/tests/unittests/flows/llm_flows/tools/test_batch_executor.py @@ -96,6 +96,31 @@ def test_merge_parallel_function_response_events_multiple() -> None: assert merged.actions.state_delta == {'key1': 'val1', 'key2': 'val2'} +def test_merged_event_keeps_the_custom_metadata_of_its_events() -> None: + """Metadata set on the events of a step is carried onto their merged event.""" + + def event(text: str, custom_metadata: dict[str, object] | None) -> Event: + return Event( + invocation_id='inv-1', + author='agent', + content=types.Content( + role='user', parts=[types.Part.from_text(text=text)] + ), + custom_metadata=custom_metadata, + ) + + merged = _batch_tool_executor.merge_parallel_function_response_events([ + event('part1', None), + event('part2', {'adk_tool_call_cache_hit': True}), + event('part3', {'source': 'cache'}), + ]) + + assert merged.custom_metadata == { + 'adk_tool_call_cache_hit': True, + 'source': 'cache', + } + + def test_is_non_blocking_tool() -> None: assert not _batch_tool_executor._is_non_blocking_tool(None) diff --git a/tests/unittests/flows/llm_flows/tools/test_caller.py b/tests/unittests/flows/llm_flows/tools/test_caller.py index 637a90534e6..ce1a92de362 100644 --- a/tests/unittests/flows/llm_flows/tools/test_caller.py +++ b/tests/unittests/flows/llm_flows/tools/test_caller.py @@ -17,6 +17,7 @@ from __future__ import annotations import asyncio +from collections.abc import AsyncGenerator from collections.abc import Awaitable import concurrent.futures import contextvars @@ -26,11 +27,14 @@ from google.adk.agents.invocation_context import InvocationContext from google.adk.agents.llm_agent import LlmAgent +from google.adk.agents.run_config import RunConfig from google.adk.events.event_actions import EventActions from google.adk.flows.llm_flows import functions from google.adk.flows.llm_flows.tools import _caller as _tool_caller from google.adk.tools.base_tool import BaseTool from google.adk.tools.function_tool import FunctionTool +from google.adk.tools.long_running_tool import LongRunningFunctionTool +from google.adk.tools.tool_confirmation import ToolConfirmation from google.adk.tools.tool_context import ToolContext from google.genai import types import pytest @@ -85,6 +89,7 @@ async def test_execute_single_prepared_call_runs_tool_runner() -> None: invocation_context = mock.create_autospec(InvocationContext, instance=True) invocation_context.invocation_id = 'inv-1' invocation_context.branch = 'main' + invocation_context.run_config = None invocation_context.agent = mock.Mock() invocation_context.agent.name = 'test_agent' invocation_context.plugin_manager = mock.AsyncMock() @@ -432,3 +437,275 @@ async def main() -> None: assert not _is_shut_down(first) asyncio.run(main()) + + +def _record(x: int) -> dict[str, int]: + """A plain tool for exercising the dedupe key and predicate helpers.""" + return {'x': x} + + +def _prepared_call( + invocation_context: InvocationContext, + tool: BaseTool, + *, + args: dict[str, Any] | None = None, + name: str | None = None, + tool_lookup_error: Exception | None = None, + tool_confirmation: ToolConfirmation | None = None, +) -> _tool_caller._PreparedFunctionCall: + """A prepared call of `tool` with real contexts, as the prepare phase builds.""" + function_call = types.FunctionCall( + name=name or tool.name, id='call-1', args=args or {} + ) + return _tool_caller._PreparedFunctionCall( + function_call=function_call, + tool=tool, + tool_context=_tool_caller._create_tool_context( + invocation_context, function_call, tool_confirmation + ), + function_args=dict(args or {}), + contextvars_snapshot=contextvars.copy_context(), + tool_lookup_error=tool_lookup_error, + ) + + +async def _dedupe_context() -> InvocationContext: + return await testing_utils.create_invocation_context( + LlmAgent(name='test_agent'), + run_config=RunConfig(dedupe_tool_calls=True), + ) + + +async def test_dedupe_is_off_unless_the_run_opts_in() -> None: + """A regular tool is deduped only when RunConfig.dedupe_tool_calls is set.""" + plain = await testing_utils.create_invocation_context( + LlmAgent(name='test_agent') + ) + opted_in = await _dedupe_context() + tool = FunctionTool(_record) + + assert not _tool_caller._should_dedupe_tool_call( + plain, _prepared_call(plain, tool, args={'x': 1}) + ) + assert _tool_caller._should_dedupe_tool_call( + opted_in, _prepared_call(opted_in, tool, args={'x': 1}) + ) + + +async def test_long_running_function_tool_is_a_candidate_by_default() -> None: + """A LongRunningFunctionTool is deduped even when the run did not opt in.""" + plain = await testing_utils.create_invocation_context( + LlmAgent(name='test_agent') + ) + tool = LongRunningFunctionTool(func=_record) + + assert _tool_caller._should_dedupe_tool_call( + plain, _prepared_call(plain, tool, args={'x': 1}) + ) + + +async def test_other_long_running_tools_follow_the_run_setting() -> None: + """A tool merely flagged long-running, like a wrapped workflow node, needs the opt-in.""" + plain = await testing_utils.create_invocation_context( + LlmAgent(name='test_agent') + ) + opted_in = await _dedupe_context() + tool = BaseTool( + name='run_node', description='Runs a workflow node.', is_long_running=True + ) + + assert not _tool_caller._should_dedupe_tool_call( + plain, _prepared_call(plain, tool) + ) + assert _tool_caller._should_dedupe_tool_call( + opted_in, _prepared_call(opted_in, tool) + ) + + +async def test_tool_that_defers_its_response_is_never_deduped() -> None: + """A tool whose response another orchestrator synthesizes is never deduped.""" + invocation_context = await _dedupe_context() + + class _DeferringTool(BaseTool): + + def __init__(self) -> None: + super().__init__(name='delegate', description='Runs a sub-agent.') + self._defers_response = True + + assert not _tool_caller._should_dedupe_tool_call( + invocation_context, _prepared_call(invocation_context, _DeferringTool()) + ) + + +async def test_call_carrying_a_confirmation_answer_is_never_deduped() -> None: + """A call re-run with the user's confirmation answer is never deduped.""" + invocation_context = await _dedupe_context() + tool = FunctionTool(_record, require_confirmation=True) + + assert not _tool_caller._should_dedupe_tool_call( + invocation_context, + _prepared_call( + invocation_context, + tool, + args={'x': 1}, + tool_confirmation=ToolConfirmation(confirmed=True), + ), + ) + + +async def test_stop_streaming_call_is_never_deduped() -> None: + """The stop_streaming live control operation is never deduped.""" + invocation_context = await _dedupe_context() + + def stop_streaming(function_name: str) -> None: + del function_name + + assert not _tool_caller._should_dedupe_tool_call( + invocation_context, + _prepared_call( + invocation_context, + FunctionTool(stop_streaming), + args={'function_name': 'monitor'}, + ), + ) + + +async def test_streaming_tool_is_never_deduped() -> None: + """A live streaming tool, whose results arrive on the live queue, is never deduped.""" + invocation_context = await _dedupe_context() + + async def monitor(x: int) -> AsyncGenerator[dict[str, int], None]: + yield {'x': x} + + assert not _tool_caller._should_dedupe_tool_call( + invocation_context, + _prepared_call(invocation_context, FunctionTool(monitor), args={'x': 1}), + ) + + +async def test_unresolved_tool_is_never_deduped() -> None: + """A call whose tool name resolved to nothing is never deduped.""" + invocation_context = await _dedupe_context() + tool = BaseTool(name='missing_tool', description='Tool not found') + + assert not _tool_caller._should_dedupe_tool_call( + invocation_context, + _prepared_call( + invocation_context, + tool, + tool_lookup_error=ValueError('Tool missing_tool not found'), + ), + ) + + +async def test_cache_key_ignores_argument_order() -> None: + """Calls whose arguments differ only in key order share a cache key.""" + invocation_context = await _dedupe_context() + tool = FunctionTool(_record) + first = _prepared_call( + invocation_context, tool, args={'a': 1, 'b': [1, {'c': 2, 'd': 3}]} + ) + second = _prepared_call( + invocation_context, tool, args={'b': [1, {'d': 3, 'c': 2}], 'a': 1} + ) + + assert _tool_caller._tool_call_cache_key( + invocation_context, first + ) == _tool_caller._tool_call_cache_key(invocation_context, second) + + +@pytest.mark.parametrize('other_x', [True, 1.0, '1']) +async def test_cache_key_tells_equal_values_of_different_types_apart( + other_x: Any, +) -> None: + """``1`` and a value that compares equal to it are different arguments.""" + invocation_context = await _dedupe_context() + tool = FunctionTool(_record) + first = _prepared_call(invocation_context, tool, args={'x': 1}) + second = _prepared_call(invocation_context, tool, args={'x': other_x}) + + assert _tool_caller._tool_call_cache_key( + invocation_context, first + ) != _tool_caller._tool_call_cache_key(invocation_context, second) + + +async def test_cache_key_tells_an_empty_dict_from_an_empty_list() -> None: + """``{}`` and ``[]`` are different arguments although both are empty.""" + invocation_context = await _dedupe_context() + tool = FunctionTool(_record) + first = _prepared_call(invocation_context, tool, args={'x': {}}) + second = _prepared_call(invocation_context, tool, args={'x': []}) + + assert _tool_caller._tool_call_cache_key( + invocation_context, first + ) != _tool_caller._tool_call_cache_key(invocation_context, second) + + +async def test_cache_key_differs_between_branches() -> None: + """The same call on two agent branches has two cache keys.""" + invocation_context = await _dedupe_context() + left = invocation_context.model_copy(update={'branch': 'root.left'}) + right = invocation_context.model_copy(update={'branch': 'root.right'}) + tool = FunctionTool(_record) + + assert _tool_caller._tool_call_cache_key( + left, _prepared_call(left, tool, args={'x': 1}) + ) != _tool_caller._tool_call_cache_key( + right, _prepared_call(right, tool, args={'x': 1}) + ) + + +async def test_cache_key_differs_between_agents() -> None: + """The same call made by two agents on one branch has two cache keys.""" + invocation_context = await _dedupe_context() + other = invocation_context.model_copy( + update={'agent': LlmAgent(name='other_agent')} + ) + tool = FunctionTool(_record) + + assert _tool_caller._tool_call_cache_key( + invocation_context, + _prepared_call(invocation_context, tool, args={'x': 1}), + ) != _tool_caller._tool_call_cache_key( + other, _prepared_call(other, tool, args={'x': 1}) + ) + + +@pytest.mark.parametrize( + ('actions', 'shareable'), + [ + pytest.param(EventActions(), True, id='nothing'), + pytest.param( + EventActions(state_delta={'runs': 1}, artifact_delta={'report': 1}), + True, + id='deltas', + ), + pytest.param( + EventActions(transfer_to_agent='child'), False, id='transfer' + ), + pytest.param( + EventActions(escalate=True, skip_summarization=True), + False, + id='exit_loop', + ), + pytest.param( + EventActions(skip_summarization=True), + False, + id='skip_summarization', + ), + pytest.param( + EventActions( + requested_tool_confirmations={ + 'call-1': ToolConfirmation(hint='Approve?') + } + ), + False, + id='confirmation', + ), + ], +) +def test_result_is_shareable_unless_its_run_acted_beyond_deltas( + actions: EventActions, shareable: bool +) -> None: + """A result is shared unless its run recorded an action beyond state and artifact deltas.""" + assert _tool_caller._tool_result_is_shareable(actions) is shareable diff --git a/tests/unittests/flows/llm_flows/tools/test_functions_dedupe.py b/tests/unittests/flows/llm_flows/tools/test_functions_dedupe.py new file mode 100644 index 00000000000..e8c1864fd17 --- /dev/null +++ b/tests/unittests/flows/llm_flows/tools/test_functions_dedupe.py @@ -0,0 +1,743 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for deduping identical tool calls within one invocation. + +Verifies that, once a run opts in with ``RunConfig.dedupe_tool_calls`` (or the +tool is a long-running function tool), identical tool calls share a single +tool execution while every call keeps its own callbacks, response event and +function call id. +""" + +import asyncio +from typing import Any + +from google.adk.agents.llm_agent import Agent +from google.adk.agents.parallel_agent import ParallelAgent +from google.adk.agents.run_config import RunConfig +from google.adk.events.event import Event +from google.adk.tools.base_tool import BaseTool +from google.adk.tools.function_tool import FunctionTool +from google.adk.tools.long_running_tool import LongRunningFunctionTool +from google.adk.tools.tool_context import ToolContext +from google.genai import types + +from .... import testing_utils + +_CACHE_HIT_KEY = 'adk_tool_call_cache_hit' +_CONFIRMATION_CALL = 'adk_request_confirmation' + + +def _call(args: dict[str, Any], name: str = 'slow_tool') -> types.Part: + """A fresh function call part; each call needs its own id, so never reuse one.""" + return types.Part.from_function_call(name=name, args=args) + + +async def _run( + runner: testing_utils.InMemoryRunner, + *, + dedupe: bool, + new_message: types.Content | None = None, +) -> list[Event]: + """Runs one invocation of the runner's agent, opting into deduping or not.""" + events = [] + async for event in runner.runner.run_async( + user_id=runner.session.user_id, + session_id=runner.session.id, + new_message=new_message or testing_utils.get_user_content('run'), + run_config=RunConfig(dedupe_tool_calls=dedupe), + ): + events.append(event) + return events + + +def _response_events(events: list[Event]) -> list[Event]: + """The events that carry at least one function response.""" + return [event for event in events if event.get_function_responses()] + + +def _responses(events: list[Event]) -> list[types.FunctionResponse]: + """The function responses of the events, in the order they were emitted.""" + return [ + response + for event in events + for response in event.get_function_responses() + ] + + +def _call_ids(events: list[Event], name: str) -> list[str]: + """The ids of the `name` function calls, in emission order.""" + return [ + call.id + for event in events + for call in event.get_function_calls() + if call.name == name and call.id is not None + ] + + +def _is_cache_hit(event: Event) -> bool: + """Whether the event is marked as reusing the result of an earlier call.""" + return bool((event.custom_metadata or {}).get(_CACHE_HIT_KEY)) + + +async def test_identical_call_in_a_later_step_reuses_the_first_result(): + """A call repeated in the next step reuses the result instead of running again.""" + runs = 0 + + def slow_tool(x: int) -> dict[str, int]: + nonlocal runs + runs += 1 + return {'result': runs} + + model = testing_utils.MockModel.create( + responses=[_call({'x': 1}), _call({'x': 1}), 'done'] + ) + agent = Agent(name='root_agent', model=model, tools=[slow_tool]) + runner = testing_utils.InMemoryRunner(agent) + + events = await _run(runner, dedupe=True) + + assert runs == 1 + first, second = _response_events(events) + assert first.get_function_responses()[0].response == {'result': 1} + assert second.get_function_responses()[0].response == {'result': 1} + assert not _is_cache_hit(first) + assert second.custom_metadata is not None + assert second.custom_metadata[_CACHE_HIT_KEY] is True + + +async def test_identical_calls_in_one_step_share_one_execution(): + """Two identical calls in one step run the tool once and each answer their own id.""" + runs = 0 + + async def slow_tool(x: int) -> dict[str, int]: + nonlocal runs + runs += 1 + # Yields so that the second call finds the first one still running. + await asyncio.sleep(0.01) + return {'result': runs} + + model = testing_utils.MockModel.create( + responses=[[_call({'x': 1}), _call({'x': 1})], 'done'] + ) + agent = Agent(name='root_agent', model=model, tools=[slow_tool]) + runner = testing_utils.InMemoryRunner(agent) + + events = await _run(runner, dedupe=True) + + assert runs == 1 + call_ids = _call_ids(events, 'slow_tool') + assert len(set(call_ids)) == 2 + (merged,) = _response_events(events) + responses = merged.get_function_responses() + assert [response.response for response in responses] == [ + {'result': 1}, + {'result': 1}, + ] + assert [response.id for response in responses] == call_ids + assert merged.custom_metadata is not None + assert merged.custom_metadata[_CACHE_HIT_KEY] is True + + +async def test_identical_calls_run_separately_by_default(): + """Without opting in, a repeated call runs the tool again and is not marked.""" + runs = 0 + + def slow_tool(x: int) -> dict[str, int]: + nonlocal runs + runs += 1 + return {'result': runs} + + model = testing_utils.MockModel.create( + responses=[_call({'x': 1}), _call({'x': 1}), 'done'] + ) + agent = Agent(name='root_agent', model=model, tools=[slow_tool]) + runner = testing_utils.InMemoryRunner(agent) + + events = await _run(runner, dedupe=False) + + assert runs == 2 + assert [response.response for response in _responses(events)] == [ + {'result': 1}, + {'result': 2}, + ] + assert not any(_is_cache_hit(event) for event in events) + + +async def test_long_running_tool_is_deduped_without_opting_in(): + """Identical long-running calls in one step run once; both ids stay long-running.""" + runs = 0 + + def start_job(x: int) -> dict[str, str]: + nonlocal runs + runs += 1 + return {'status': 'pending'} + + model = testing_utils.MockModel.create( + responses=[ + [_call({'x': 1}, 'start_job'), _call({'x': 1}, 'start_job')], + 'done', + ] + ) + agent = Agent( + name='root_agent', + model=model, + tools=[LongRunningFunctionTool(func=start_job)], + ) + runner = testing_utils.InMemoryRunner(agent) + + events = await _run(runner, dedupe=False) + + assert runs == 1 + call_ids = _call_ids(events, 'start_job') + assert len(set(call_ids)) == 2 + call_event = next(event for event in events if event.get_function_calls()) + assert call_event.long_running_tool_ids == set(call_ids) + assert [response.response for response in _responses(events)] == [ + {'status': 'pending'}, + {'status': 'pending'}, + ] + + +async def test_calls_with_different_arguments_run_separately(): + """Only calls whose arguments match share an execution.""" + runs = 0 + + def slow_tool(x: int) -> dict[str, int]: + nonlocal runs + runs += 1 + return {'result': runs} + + model = testing_utils.MockModel.create( + responses=[_call({'x': 1}), _call({'x': 2}), 'done'] + ) + agent = Agent(name='root_agent', model=model, tools=[slow_tool]) + runner = testing_utils.InMemoryRunner(agent) + + events = await _run(runner, dedupe=True) + + assert runs == 2 + assert not any(_is_cache_hit(event) for event in events) + + +async def test_argument_type_is_part_of_a_call_identity(): + """``{'x': 1}`` and ``{'x': True}`` are different calls although ``1 == True``.""" + runs = 0 + + def slow_tool(x: Any) -> dict[str, int]: + nonlocal runs + runs += 1 + return {'result': runs} + + model = testing_utils.MockModel.create( + responses=[_call({'x': 1}), _call({'x': True}), 'done'] + ) + agent = Agent(name='root_agent', model=model, tools=[slow_tool]) + runner = testing_utils.InMemoryRunner(agent) + + events = await _run(runner, dedupe=True) + + assert runs == 2 + assert not any(_is_cache_hit(event) for event in events) + + +async def test_argument_order_is_not_part_of_a_call_identity(): + """Calls whose arguments differ only in key order share one execution.""" + runs = 0 + + def slow_tool(a: int, b: int) -> dict[str, int]: + nonlocal runs + runs += 1 + return {'result': runs} + + model = testing_utils.MockModel.create( + responses=[_call({'a': 1, 'b': 2}), _call({'b': 2, 'a': 1}), 'done'] + ) + agent = Agent(name='root_agent', model=model, tools=[slow_tool]) + runner = testing_utils.InMemoryRunner(agent) + + events = await _run(runner, dedupe=True) + + assert runs == 1 + assert [_is_cache_hit(event) for event in _response_events(events)] == [ + False, + True, + ] + + +async def test_side_effects_of_a_reused_result_apply_once(): + """The state the tool writes is applied by the first call only.""" + + def slow_tool(x: int, tool_context: ToolContext) -> dict[str, int]: + tool_context.state['runs'] = tool_context.state.get('runs', 0) + 1 + return {'result': tool_context.state['runs']} + + model = testing_utils.MockModel.create( + responses=[_call({'x': 1}), _call({'x': 1}), 'done'] + ) + agent = Agent(name='root_agent', model=model, tools=[slow_tool]) + runner = testing_utils.InMemoryRunner(agent) + + events = await _run(runner, dedupe=True) + + assert runner.session.state['runs'] == 1 + first, second = _response_events(events) + assert first.actions.state_delta == {'runs': 1} + assert second.actions.state_delta == {} + assert second.get_function_responses()[0].response == {'result': 1} + + +async def test_failed_execution_is_not_reused(): + """A call identical to one whose tool raised runs the tool again.""" + runs = 0 + + def slow_tool(x: int) -> dict[str, int]: + nonlocal runs + runs += 1 + if runs == 1: + raise RuntimeError('transient failure') + return {'result': runs} + + def on_tool_error( + tool: BaseTool, + args: dict[str, Any], + tool_context: ToolContext, + error: Exception, + ) -> dict[str, str]: + return {'error': str(error)} + + model = testing_utils.MockModel.create( + responses=[_call({'x': 1}), _call({'x': 1}), 'done'] + ) + agent = Agent( + name='root_agent', + model=model, + tools=[slow_tool], + on_tool_error_callback=on_tool_error, + ) + runner = testing_utils.InMemoryRunner(agent) + + events = await _run(runner, dedupe=True) + + assert runs == 2 + assert [response.response for response in _responses(events)] == [ + {'error': 'transient failure'}, + {'result': 2}, + ] + assert not any(_is_cache_hit(event) for event in events) + + +async def test_failure_while_identical_calls_wait_reaches_each_call_then_retries(): + """A failure is reported to every call waiting on it; the next identical call retries.""" + runs = 0 + failed_call_ids: list[str] = [] + + async def slow_tool(x: int) -> dict[str, int]: + nonlocal runs + runs += 1 + # Yields so that the second call finds the first one still running. + await asyncio.sleep(0.01) + if runs == 1: + raise RuntimeError('transient failure') + return {'result': runs} + + def on_tool_error( + tool: BaseTool, + args: dict[str, Any], + tool_context: ToolContext, + error: Exception, + ) -> dict[str, str]: + failed_call_ids.append(tool_context.function_call_id) + return {'error': str(error)} + + model = testing_utils.MockModel.create( + responses=[[_call({'x': 1}), _call({'x': 1})], _call({'x': 1}), 'done'] + ) + agent = Agent( + name='root_agent', + model=model, + tools=[slow_tool], + on_tool_error_callback=on_tool_error, + ) + runner = testing_utils.InMemoryRunner(agent) + + events = await _run(runner, dedupe=True) + + assert runs == 2 + assert [response.response for response in _responses(events)] == [ + {'error': 'transient failure'}, + {'error': 'transient failure'}, + {'result': 2}, + ] + assert sorted(failed_call_ids) == sorted(_call_ids(events, 'slow_tool')[:2]) + assert not any(_is_cache_hit(event) for event in events) + + +async def test_callbacks_run_for_every_call_while_the_tool_runs_once(): + """Before- and after-tool callbacks run per call; only the tool run is shared.""" + runs = 0 + before_calls = 0 + after_calls = 0 + + def slow_tool(x: int) -> dict[str, Any]: + nonlocal runs + runs += 1 + return {'result': runs, 'seen_by': []} + + def before_tool( + tool: BaseTool, args: dict[str, Any], tool_context: ToolContext + ) -> None: + nonlocal before_calls + before_calls += 1 + + def after_tool( + tool: BaseTool, + args: dict[str, Any], + tool_context: ToolContext, + tool_response: dict[str, Any], + ) -> None: + nonlocal after_calls + after_calls += 1 + # Alters the response in place rather than returning a new one. + tool_response['seen_by'].append(after_calls) + + model = testing_utils.MockModel.create( + responses=[_call({'x': 1}), _call({'x': 1}), 'done'] + ) + agent = Agent( + name='root_agent', + model=model, + tools=[slow_tool], + before_tool_callback=before_tool, + after_tool_callback=after_tool, + ) + runner = testing_utils.InMemoryRunner(agent) + + events = await _run(runner, dedupe=True) + + assert runs == 1 + assert before_calls == 2 + assert after_calls == 2 + # Each call's callback sees the result as the tool returned it: neither + # call's in-place edit reaches the other call's response. + first, second = _responses(events) + assert first.response == {'result': 1, 'seen_by': [1]} + assert second.response == {'result': 1, 'seen_by': [2]} + + +async def test_result_that_cannot_be_copied_is_shared_as_is(): + """A reused result that cannot be deep-copied is shared rather than failing the call.""" + runs = 0 + + class _Handle(dict): + """A value, such as a live client handle, that refuses to be copied.""" + + def __deepcopy__(self, memo: dict[int, Any]) -> Any: + raise TypeError('cannot copy a live handle') + + def slow_tool(x: int) -> dict[str, Any]: + nonlocal runs + runs += 1 + return {'result': runs, 'handle': _Handle(connection=runs)} + + model = testing_utils.MockModel.create( + responses=[_call({'x': 1}), _call({'x': 1}), 'done'] + ) + agent = Agent(name='root_agent', model=model, tools=[slow_tool]) + runner = testing_utils.InMemoryRunner(agent) + + events = await _run(runner, dedupe=True) + + assert runs == 1 + first, second = _responses(events) + assert second.response['result'] == 1 + assert second.response['handle'] is first.response['handle'] + assert [_is_cache_hit(event) for event in _response_events(events)] == [ + False, + True, + ] + + +async def test_calls_that_need_confirmation_each_request_their_own(): + """Identical calls held for confirmation each request it, deduped or not. + + The flow answers a call whose tool requires confirmation before the tool + would run, so deduping never sees such a call; the events must come out the + same with and without deduping. + """ + runs = 0 + + def guarded(x: int) -> dict[str, int]: + nonlocal runs + runs += 1 + return {'result': x} + + def make_runner() -> testing_utils.InMemoryRunner: + model = testing_utils.MockModel.create( + responses=[ + [_call({'x': 1}, 'guarded'), _call({'x': 1}, 'guarded')], + 'done', + ] + ) + agent = Agent( + name='root_agent', + model=model, + tools=[FunctionTool(guarded, require_confirmation=True)], + ) + return testing_utils.InMemoryRunner(agent) + + def shape(events: list[Event]) -> list[tuple[str, list[str]]]: + """The events as (author, part kinds), ignoring ids and payloads.""" + return [ + ( + event.author, + [ + f'call:{part.function_call.name}' + if part.function_call + else f'response:{part.function_response.name}' + if part.function_response + else 'text' + for part in event.content.parts + ], + ) + for event in events + if event.content and event.content.parts + ] + + def confirmation_targets(events: list[Event]) -> set[str]: + return { + call.args['originalFunctionCall']['id'] + for event in events + for call in event.get_function_calls() + if call.name == _CONFIRMATION_CALL and call.args + } + + deduped = await _run(make_runner(), dedupe=True) + plain = await _run(make_runner(), dedupe=False) + + assert runs == 0 + call_ids = set(_call_ids(deduped, 'guarded')) + assert len(call_ids) == 2 + assert confirmation_targets(deduped) == call_ids + (response_event,) = _response_events(deduped) + assert set(response_event.actions.requested_tool_confirmations) == call_ids + assert shape(deduped) == shape(plain) + assert not any(_is_cache_hit(event) for event in deduped) + + +async def test_calls_whose_tool_asks_for_confirmation_itself_each_run(): + """A tool that requests confirmation in its own body runs for every identical call. + + Setup: an async tool that yields, then records a confirmation request and + returns a pending answer, the way a tool that gates itself does; the model + calls it twice with the same arguments in one step, deduping on. + Assert: the tool runs twice, since the first run's confirmation request is + the effect of that call alone; each call id has its own request; nothing + is marked as reused. + """ + runs = 0 + + async def guarded(x: int, tool_context: ToolContext) -> dict[str, Any]: + nonlocal runs + runs += 1 + # Yields so that the second call finds the first one still running. + await asyncio.sleep(0.01) + if not tool_context.tool_confirmation: + tool_context.request_confirmation(hint='Approve?') + tool_context.actions.skip_summarization = True + return {'error': 'This tool call requires confirmation.'} + return {'result': x} + + model = testing_utils.MockModel.create( + responses=[ + [_call({'x': 1}, 'guarded'), _call({'x': 1}, 'guarded')], + 'done', + ] + ) + agent = Agent(name='root_agent', model=model, tools=[guarded]) + runner = testing_utils.InMemoryRunner(agent) + + events = await _run(runner, dedupe=True) + + assert runs == 2 + call_ids = set(_call_ids(events, 'guarded')) + assert len(call_ids) == 2 + (response_event,) = _response_events(events) + assert set(response_event.actions.requested_tool_confirmations) == call_ids + assert not any(_is_cache_hit(event) for event in events) + + +async def test_confirmed_and_rejected_identical_calls_are_answered_apart(): + """Of two identical calls, the confirmed one runs and the rejected one is refused. + + Setup: a confirmation-gated tool called twice with the same arguments. + Act: + - Turn 1: both calls request confirmation. + - Turn 2: the user confirms the first request and rejects the second. + Assert: the tool runs once, for the confirmed call id; the rejected call id + is answered with the rejection error. + """ + runs = 0 + + def guarded(x: int) -> dict[str, int]: + nonlocal runs + runs += 1 + return {'result': x} + + model = testing_utils.MockModel.create( + responses=[ + [_call({'x': 1}, 'guarded'), _call({'x': 1}, 'guarded')], + 'done', + ] + ) + agent = Agent( + name='root_agent', + model=model, + tools=[FunctionTool(guarded, require_confirmation=True)], + ) + runner = testing_utils.InMemoryRunner(agent) + + first_turn = await _run(runner, dedupe=True) + requests = [ + call + for event in first_turn + for call in event.get_function_calls() + if call.name == _CONFIRMATION_CALL + ] + confirmed_id, rejected_id = [ + call.args['originalFunctionCall']['id'] for call in requests + ] + answers = types.Content( + role='user', + parts=[ + types.Part( + function_response=types.FunctionResponse( + name=_CONFIRMATION_CALL, + id=request.id, + response={'confirmed': confirmed}, + ) + ) + for request, confirmed in zip(requests, [True, False]) + ], + ) + second_turn = await _run(runner, dedupe=True, new_message=answers) + + assert runs == 1 + assert { + response.id: response.response + for response in _responses(second_turn) + if response.name == 'guarded' + } == { + confirmed_id: {'result': 1}, + rejected_id: {'error': 'This tool call is rejected.'}, + } + + +async def test_repeating_a_transfer_hands_off_again(): + """A transfer identical to an earlier one still hands control to its target.""" + + def transfer(agent_name: str) -> types.Part: + return _call({'agent_name': agent_name}, 'transfer_to_agent') + + child = Agent( + name='child', + model=testing_utils.MockModel.create( + responses=[transfer('root_agent'), 'child done'] + ), + ) + root = Agent( + name='root_agent', + model=testing_utils.MockModel.create( + responses=[transfer('child'), transfer('child'), 'root done'] + ), + sub_agents=[child], + ) + runner = testing_utils.InMemoryRunner(root) + + events = await _run(runner, dedupe=True) + + assert [ + event.actions.transfer_to_agent for event in _response_events(events) + ] == ['child', 'root_agent', 'child'] + assert testing_utils.simplify_events(events)[-1] == ('child', 'child done') + assert not any(_is_cache_hit(event) for event in events) + + +async def test_same_named_tools_of_two_agents_run_separately(): + """An agent's tool is not answered from another agent's tool of the same name.""" + + def agent_with_lookup( + name: str, responses: list[Any], sub_agents: list[Agent] | None = None + ) -> Agent: + def lookup(x: int) -> dict[str, str]: + return {'source': name} + + return Agent( + name=name, + model=testing_utils.MockModel.create(responses=responses), + tools=[lookup], + sub_agents=sub_agents or [], + ) + + child = agent_with_lookup('child', [_call({'x': 1}, 'lookup'), 'child done']) + root = agent_with_lookup( + 'root_agent', + [ + _call({'x': 1}, 'lookup'), + _call({'agent_name': 'child'}, 'transfer_to_agent'), + ], + sub_agents=[child], + ) + runner = testing_utils.InMemoryRunner(root) + + events = await _run(runner, dedupe=True) + + assert [ + (event.author, response.response, _is_cache_hit(event)) + for event in _response_events(events) + for response in event.get_function_responses() + if response.name == 'lookup' + ] == [ + ('root_agent', {'source': 'root_agent'}, False), + ('child', {'source': 'child'}, False), + ] + + +async def test_parallel_sub_agents_each_run_the_call_once(): + """Deduping is scoped to an agent branch; sibling branches run the tool themselves.""" + runs = 0 + + def slow_tool(x: int) -> dict[str, int]: + nonlocal runs + runs += 1 + return {'result': runs} + + def sub_agent(name: str) -> Agent: + model = testing_utils.MockModel.create( + responses=[_call({'x': 1}), f'done by {name}'] + ) + return Agent(name=name, model=model, tools=[slow_tool]) + + root = ParallelAgent( + name='root_agent', sub_agents=[sub_agent('left'), sub_agent('right')] + ) + runner = testing_utils.InMemoryRunner(root) + + events = await _run(runner, dedupe=True) + + assert runs == 2 + assert sorted( + response.response['result'] for response in _responses(events) + ) == [1, 2] + assert not any(_is_cache_hit(event) for event in events)