fix(flows): re-execute orphaned sibling calls on confirmation resume

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,28 +465,54 @@ 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,
             events,
             confirmation_fc_ids,
             confirmations_by_fc_id,
             tools_dict,
+            dynamically_requested_fc_ids,
         )
     )
 
     if not tools_to_resume_with_confirmation:
       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