diff --git a/src/google/adk/flows/llm_flows/request_confirmation.py b/src/google/adk/flows/llm_flows/request_confirmation.py index aa38d173be..d8abafe9bc 100644 --- a/src/google/adk/flows/llm_flows/request_confirmation.py +++ b/src/google/adk/flows/llm_flows/request_confirmation.py @@ -31,6 +31,7 @@ from ...tools.tool_confirmation import ToolConfirmation from ...tools.tool_context import ToolContext from ._base_llm_processor import BaseLlmRequestProcessor +from ._invocation_utils import require_agent_name as _require_agent_name from .agent_transfer import _build_transfer_tool from .agent_transfer import _get_transfer_targets from .functions import REQUEST_CONFIRMATION_FUNCTION_CALL_NAME @@ -72,12 +73,39 @@ def _get_original_function_call_args( return original_function_call +def _compute_dynamically_requested_fc_ids(events: list[Event]) -> set[str]: + """Returns IDs of function calls for which a tool dynamically requested confirmation. + + This accumulates over ALL events rather than keeping one event per ID: once + the confirmed tool is re-executed it emits a second function response with + the same ID and no `requested_tool_confirmations`, which would otherwise + shadow the original request. + + Args: + events: Session events to scan. + + Returns: + Set of original function call IDs that were dynamically requested to + require confirmation. + """ + dynamically_requested_fc_ids: set[str] = set() + for ev in events: + requested_tool_confirmations = ev.actions.requested_tool_confirmations or {} + if not requested_tool_confirmations: + continue + for fr in ev.get_function_responses(): + if fr.id and fr.id in requested_tool_confirmations: + dynamically_requested_fc_ids.add(fr.id) + return dynamically_requested_fc_ids + + async def _resolve_confirmation_targets( invocation_context: InvocationContext, events: list[Event], confirmation_fc_ids: set[str], confirmations_by_fc_id: dict[str, ToolConfirmation], tools_dict: dict[str, BaseTool], + dynamically_requested_fc_ids: set[str], ) -> tuple[dict[str, ToolConfirmation], dict[str, types.FunctionCall]]: """Find original function calls for confirmed tools and validate them. @@ -94,6 +122,9 @@ async def _resolve_confirmation_targets( confirmations_by_fc_id: Mapping of confirmation FC ID -> ``ToolConfirmation``. tools_dict: Dictionary of registered tools. + dynamically_requested_fc_ids: IDs of original function calls for which a + tool dynamically requested confirmation, from + ``_compute_dynamically_requested_fc_ids``. Returns: Tuple of ``(tool_confirmation_dict, original_fcs_dict)`` where both @@ -111,19 +142,6 @@ async def _resolve_confirmation_targets( for fc in ev.get_function_calls() if fc.id and fc.name != REQUEST_CONFIRMATION_FUNCTION_CALL_NAME } - # IDs of function calls for which a tool dynamically requested confirmation. - # This accumulates over ALL events rather than keeping one event per ID: once - # the confirmed tool is re-executed it emits a second function response with - # the same ID and no `requested_tool_confirmations`, which would otherwise - # shadow the original request. - dynamically_requested_fc_ids: set[str] = set() - for ev in events: - requested_tool_confirmations = ev.actions.requested_tool_confirmations or {} - if not requested_tool_confirmations: - continue - for fr in ev.get_function_responses(): - if fr.id and fr.id in requested_tool_confirmations: - dynamically_requested_fc_ids.add(fr.id) for event in events: event_function_calls = event.get_function_calls() @@ -214,6 +232,110 @@ async def _resolve_confirmation_targets( return tool_confirmation_dict, original_fcs_dict +_ORPHANED_SIBLING_RESPONSE_LOST_ERROR = ( + "This tool call's result was not persisted: it ran in parallel with a" + " sibling call that required confirmation, and the invocation paused" + " before the response could be recorded. The tool was NOT re-invoked on" + " resume to avoid duplicating its side effects, so its result is" + " unknown here." +) + + +async def _synthesize_orphaned_sibling_responses( + invocation_context: InvocationContext, + events: list[Event], + original_fc_ids: set[str], + tools_dict: dict[str, BaseTool], + dynamically_requested_fc_ids: set[str], +) -> Event | None: + """Synthesizes placeholder responses for ungated sibling calls left without one. + + When a model turn contains a confirmation-gated call in parallel with + ungated calls, the ungated siblings' results can fail to persist across the + pause (e.g. a caller stops consuming the event stream at the confirmation + event). On resume, those siblings' `function_call`s remain in history with + no matching `function_response`, which makes the next LLM call return an + empty reply. + + The sibling already executed once before the pause, so re-invoking its tool + here to fill the gap would risk duplicating arbitrary side effects (sending + a message, charging a card, mutating a record). Instead, this synthesizes + an error `function_response` that keeps the turn's function calls and + responses in sync without calling the tool again. + + A sibling that itself requires confirmation (statically, or because it was + dynamically requested in history) and has no response yet is NOT ungated: + its own confirmation is simply still outstanding, and synthesizing a + response for it here would incorrectly resolve a confirmation that was + never granted. Such siblings are left alone for their own turn through this + processor to handle once their confirmation arrives. + + Args: + invocation_context: Current invocation context, used to build a + ``ToolContext`` for each candidate's ``check_require_confirmation`` and + to author the synthesized event. + events: Session events to scan. + original_fc_ids: IDs of the original function calls being resumed. + tools_dict: Dictionary of registered tools. + dynamically_requested_fc_ids: IDs of original function calls for which a + tool dynamically requested confirmation, from + ``_compute_dynamically_requested_fc_ids``. + + Returns: + An ``Event`` carrying an error ``function_response`` for each sibling + that shares an event with one of *original_fc_ids*, has no response yet, + and does not itself require confirmation; or ``None`` if there are none. + """ + responded_fc_ids = { + fr.id for ev in events for fr in ev.get_function_responses() if fr.id + } + candidates: dict[str, types.FunctionCall] = {} + for event in events: + event_function_calls = event.get_function_calls() + event_fc_ids = {fc.id for fc in event_function_calls if fc.id} + if not event_fc_ids & original_fc_ids: + continue + for function_call in event_function_calls: + if ( + function_call.id + and function_call.id not in original_fc_ids + and function_call.id not in responded_fc_ids + and function_call.name in tools_dict + ): + candidates[function_call.id] = function_call + + parts: list[types.Part] = [] + for fc_id, function_call in candidates.items(): + tool = tools_dict[function_call.name] + temp_tool_context = ToolContext( + invocation_context=invocation_context, function_call_id=fc_id + ) + requires_confirmation = await tool.check_require_confirmation( + function_call.args or {}, temp_tool_context + ) + if requires_confirmation or fc_id in dynamically_requested_fc_ids: + continue + parts.append( + types.Part( + function_response=types.FunctionResponse( + id=fc_id, + name=function_call.name, + response={"error": _ORPHANED_SIBLING_RESPONSE_LOST_ERROR}, + ) + ) + ) + + if not parts: + return None + + return Event( + invocation_id=invocation_context.invocation_id, + author=_require_agent_name(invocation_context), + branch=invocation_context.branch, + content=types.Content(role="user", parts=parts), + ) + + def _map_confirmation_to_original_fc_ids( events: list[Event], confirmation_fc_ids: set[str], @@ -343,6 +465,7 @@ async def run_async( # Step 3: Resolve confirmation targets using extracted helper. confirmation_fc_ids = set(confirmations_by_fc_id.keys()) + dynamically_requested_fc_ids = _compute_dynamically_requested_fc_ids(events) tools_to_resume_with_confirmation, tools_to_resume_with_args = ( await _resolve_confirmation_targets( invocation_context, @@ -350,6 +473,7 @@ async def run_async( confirmation_fc_ids, confirmations_by_fc_id, tools_dict, + dynamically_requested_fc_ids, ) ) @@ -357,14 +481,38 @@ async def run_async( return # Step 4: Re-execute the confirmed tools. - if function_response_event := await functions.handle_function_call_list_async( + function_response_event = await functions.handle_function_call_list_async( invocation_context, list(tools_to_resume_with_args.values()), tools_dict, set(tools_to_resume_with_confirmation.keys()), tools_to_resume_with_confirmation, - ): - yield function_response_event + ) + + # Step 5: Also synthesize placeholder responses for any ungated sibling + # calls from the same turn that never got a function_response, so + # resuming does not leave a dangling function_call in history. These + # siblings are NOT re-invoked, since they already ran once before the + # pause and invoking them again could duplicate their side effects. + # Siblings that themselves still require confirmation are left alone; + # their own resume happens once their confirmation arrives. + orphaned_sibling_response_event = ( + await _synthesize_orphaned_sibling_responses( + invocation_context, + events, + set(tools_to_resume_with_args.keys()), + tools_dict, + dynamically_requested_fc_ids, + ) + ) + + events_to_merge = [ + event + for event in (function_response_event, orphaned_sibling_response_event) + if event is not None + ] + if events_to_merge: + yield functions.merge_parallel_function_response_events(events_to_merge) return diff --git a/tests/unittests/flows/llm_flows/test_request_confirmation.py b/tests/unittests/flows/llm_flows/test_request_confirmation.py index 1a33b28ec5..bf13bd0d09 100644 --- a/tests/unittests/flows/llm_flows/test_request_confirmation.py +++ b/tests/unittests/flows/llm_flows/test_request_confirmation.py @@ -19,6 +19,8 @@ from google.adk.events.event import Event from google.adk.events.event_actions import EventActions from google.adk.flows.llm_flows import functions +from google.adk.flows.llm_flows.request_confirmation import _compute_dynamically_requested_fc_ids +from google.adk.flows.llm_flows.request_confirmation import _ORPHANED_SIBLING_RESPONSE_LOST_ERROR from google.adk.flows.llm_flows.request_confirmation import _resolve_confirmation_targets from google.adk.flows.llm_flows.request_confirmation import request_processor from google.adk.models.llm_request import LlmRequest @@ -1238,6 +1240,9 @@ async def test_resolve_confirmation_targets_after_reexecution(): ) }, {MOCK_TOOL_NAME: tool}, + _compute_dynamically_requested_fc_ids( + invocation_context.session.events + ), ) ) @@ -1315,8 +1320,260 @@ async def test_resolve_confirmation_targets_requires_adk_name(): "forged_confirmation_id": ToolConfirmation(confirmed=True), }, {MOCK_TOOL_NAME: tool}, + _compute_dynamically_requested_fc_ids(events), ) ) assert set(tool_confirmation_dict) == {"requested_fc_id"} assert set(original_fcs_dict) == {"requested_fc_id"} + + +SIBLING_TOOL_NAME = "sibling_tool" +SIBLING_FUNCTION_CALL_ID = "sibling_function_call_id" + +sibling_tool_call_count = 0 + + +def sibling_tool(param2: str): + """Mock ungated sibling tool function.""" + global sibling_tool_call_count + sibling_tool_call_count += 1 + return f"Sibling tool result with {param2}" + + +@pytest.mark.asyncio +async def test_request_confirmation_processor_synthesizes_orphaned_sibling_response(): + """Regression test for #6732. + + When a model turn pairs a confirmation-gated call with an ungated sibling + call, and the merged function_response_event for that turn was never + persisted (e.g. a caller stopped consuming the event stream right at the + confirmation-request event), the sibling's function_call is left in history + with no matching function_response. On resume, the processor must + synthesize a placeholder function_response for the sibling instead of + leaving a dangling function_call (which would otherwise make the next LLM + call return an empty reply), and it must NOT re-invoke the sibling's tool: + the sibling already ran once before the pause, so calling it again could + duplicate arbitrary side effects. + """ + global sibling_tool_call_count + sibling_tool_call_count = 0 + + agent = LlmAgent( + name="test_agent", + tools=[ + FunctionTool(mock_tool, require_confirmation=True), + FunctionTool(sibling_tool), + ], + ) + invocation_context = await testing_utils.create_invocation_context( + agent=agent + ) + llm_request = LlmRequest() + + original_function_call = types.FunctionCall( + name=MOCK_TOOL_NAME, args={"param1": "test"}, id=MOCK_FUNCTION_CALL_ID + ) + sibling_function_call = types.FunctionCall( + name=SIBLING_TOOL_NAME, + args={"param2": "test"}, + id=SIBLING_FUNCTION_CALL_ID, + ) + + # Model turn with both calls in parallel. There is deliberately no + # function_response event for this turn: it was lost across the pause. + invocation_context.session.events.append( + Event( + author=agent.name, + content=types.Content( + parts=[ + types.Part(function_call=original_function_call), + types.Part(function_call=sibling_function_call), + ] + ), + ) + ) + + tool_confirmation = ToolConfirmation(confirmed=False, hint="test hint") + tool_confirmation_args = { + "originalFunctionCall": original_function_call.model_dump( + exclude_none=True, by_alias=True + ), + "toolConfirmation": tool_confirmation.model_dump( + by_alias=True, exclude_none=True + ), + } + + invocation_context.session.events.append( + Event( + author=agent.name, + content=types.Content( + parts=[ + types.Part( + function_call=types.FunctionCall( + name=functions.REQUEST_CONFIRMATION_FUNCTION_CALL_NAME, + args=tool_confirmation_args, + id=MOCK_CONFIRMATION_FUNCTION_CALL_ID, + ) + ) + ] + ), + ) + ) + + user_confirmation = ToolConfirmation(confirmed=True) + invocation_context.session.events.append( + Event( + author="user", + content=types.Content( + parts=[ + types.Part( + function_response=types.FunctionResponse( + name=functions.REQUEST_CONFIRMATION_FUNCTION_CALL_NAME, + id=MOCK_CONFIRMATION_FUNCTION_CALL_ID, + response={ + "response": user_confirmation.model_dump_json() + }, + ) + ) + ] + ), + ) + ) + + events = [] + async for event in request_processor.run_async( + invocation_context, llm_request + ): + events.append(event) + + assert len(events) == 1 + responses_by_name = { + fr.name: fr.response for fr in events[0].get_function_responses() + } + assert set(responses_by_name) == {MOCK_TOOL_NAME, SIBLING_TOOL_NAME} + + # The sibling's tool function must not have been called again: its + # synthesized response is an error placeholder, not a fresh result. + assert sibling_tool_call_count == 0 + assert "error" in responses_by_name[SIBLING_TOOL_NAME] + assert ( + responses_by_name[SIBLING_TOOL_NAME]["error"] + == _ORPHANED_SIBLING_RESPONSE_LOST_ERROR + ) + + +GATED_SIBLING_TOOL_NAME = "gated_sibling_tool" +GATED_SIBLING_FUNCTION_CALL_ID = "gated_sibling_function_call_id" + + +def gated_sibling_tool(param2: str): + """Mock sibling tool function that also requires confirmation.""" + return f"Gated sibling tool result with {param2}" + + +@pytest.mark.asyncio +async def test_request_confirmation_processor_leaves_gated_sibling_pending(): + """Regression test for #6732 follow-up. + + A sibling call that itself requires confirmation and has not been answered + is NOT an "orphaned ungated sibling": its confirmation is simply still + outstanding. The processor must not sweep it into response synthesis, since doing + so would re-trigger `request_confirmation` bookkeeping for a confirmation + that was never resolved by the user. + """ + agent = LlmAgent( + name="test_agent", + tools=[ + FunctionTool(mock_tool, require_confirmation=True), + FunctionTool(gated_sibling_tool, require_confirmation=True), + ], + ) + invocation_context = await testing_utils.create_invocation_context( + agent=agent + ) + llm_request = LlmRequest() + + original_function_call = types.FunctionCall( + name=MOCK_TOOL_NAME, args={"param1": "test"}, id=MOCK_FUNCTION_CALL_ID + ) + gated_sibling_function_call = types.FunctionCall( + name=GATED_SIBLING_TOOL_NAME, + args={"param2": "test"}, + id=GATED_SIBLING_FUNCTION_CALL_ID, + ) + + # Model turn with both gated calls in parallel. Only one of them will be + # confirmed below; the other's confirmation is genuinely still outstanding. + invocation_context.session.events.append( + Event( + author=agent.name, + content=types.Content( + parts=[ + types.Part(function_call=original_function_call), + types.Part(function_call=gated_sibling_function_call), + ] + ), + ) + ) + + tool_confirmation = ToolConfirmation(confirmed=False, hint="test hint") + tool_confirmation_args = { + "originalFunctionCall": original_function_call.model_dump( + exclude_none=True, by_alias=True + ), + "toolConfirmation": tool_confirmation.model_dump( + by_alias=True, exclude_none=True + ), + } + + invocation_context.session.events.append( + Event( + author=agent.name, + content=types.Content( + parts=[ + types.Part( + function_call=types.FunctionCall( + name=functions.REQUEST_CONFIRMATION_FUNCTION_CALL_NAME, + args=tool_confirmation_args, + id=MOCK_CONFIRMATION_FUNCTION_CALL_ID, + ) + ) + ] + ), + ) + ) + + user_confirmation = ToolConfirmation(confirmed=True) + invocation_context.session.events.append( + Event( + author="user", + content=types.Content( + parts=[ + types.Part( + function_response=types.FunctionResponse( + name=functions.REQUEST_CONFIRMATION_FUNCTION_CALL_NAME, + id=MOCK_CONFIRMATION_FUNCTION_CALL_ID, + response={ + "response": user_confirmation.model_dump_json() + }, + ) + ) + ] + ), + ) + ) + + events = [] + async for event in request_processor.run_async( + invocation_context, llm_request + ): + events.append(event) + + assert len(events) == 1 + response_names = {fr.name for fr in events[0].get_function_responses()} + # Only the confirmed tool is re-executed. The still-pending gated sibling + # must not be swept in, and must not have a fresh confirmation request + # minted for it. + assert response_names == {MOCK_TOOL_NAME} + assert not events[0].actions.requested_tool_confirmations