fix(workflow): keep finished parallel tool results when a sibling raises

src/google/adk/agents/invocation_context.py

@@ -100,6 +100,12 @@ class _AbortState:
   ``asyncio.Event`` and can only be awaited on ``loop``, so AgentTool does not
   pass it when the tool runs on another event loop (e.g. RunConfig's tool
   thread pool); such a sub-run is not cancelled by a caller abort.
+
+  ``unpersisted_function_responses`` holds the merged response events of
+  parallel tool calls that finished before their batch ended early because a
+  sibling call raised. The batch executor cannot persist them itself, and the
+  error that ends the batch unwinds the agent; keeping them on this shared
+  instance lets the node runner persist them once the agent has stopped.
   """
 
   def __init__(
@@ -111,6 +117,13 @@ def __init__(
     self.loop = loop
     self.aborted = False
     self.event_synthesized = False
+    self.unpersisted_function_responses: list[Event] = []
+
+  def pop_unpersisted_function_responses(self) -> list[Event]:
+    """Returns and clears the kept function responses not yet persisted."""
+    responses = self.unpersisted_function_responses
+    self.unpersisted_function_responses = []
+    return responses
 
   def __deepcopy__(self, memo: dict[int, Any] | None) -> _AbortState:
     # Preserve single-instance sharing across deepcopies and avoid traversing

src/google/adk/flows/llm_flows/tools/_batch_executor.py

@@ -27,6 +27,7 @@
 
 from google.genai import types
 
+from ....agents.llm.task._finish_task_tool import FINISH_TASK_TOOL_NAME
 from ....events.event import Event
 from ....events.event_actions import EventActions
 from ....telemetry.tracing import trace_merged_tool_calls
@@ -251,7 +252,14 @@ async def _execute_prepared_function_calls(
   ]
 
   # Wait for all tasks to complete
-  maybe_function_response_events = await _gather_or_cancel(tasks)
+  try:
+    maybe_function_response_events = await _gather_or_cancel(tasks)
+  except Exception:
+    # A call raised and its unfinished siblings were cancelled. Calls that had
+    # already finished may have had side effects, so keep their real results
+    # for the node runner to persist once the error has unwound the agent.
+    _keep_completed_responses(invocation_context, tasks)
+    raise
 
   # Filter out None results
   function_response_events = [
@@ -270,6 +278,59 @@ async def _execute_prepared_function_calls(
   )
 
 
+def _keep_completed_responses(
+    invocation_context: InvocationContext,
+    tasks: list[asyncio.Task[Optional[Event]]],
+) -> None:
+  """Keeps the responses of calls that finished before their batch ended early.
+
+  Only plain tool results are kept: a call that was cancelled, raised or
+  returned no event gets nothing. They are merged as a complete batch would
+  be and left on the invocation's shared state for the node runner to
+  persist. Live mode is skipped.
+  """
+  if invocation_context.live_request_queue is not None:
+    return
+  finished = [
+      t.result()
+      for t in tasks
+      if t.done() and not t.cancelled() and t.exception() is None
+  ]
+  completed = [
+      event
+      for event in finished
+      if event is not None and _is_plain_result(event)
+  ]
+  if completed:
+    invocation_context._abort_state.unpersisted_function_responses.append(  # pylint: disable=protected-access
+        _merge_and_trace_function_response_events(invocation_context, completed)
+    )
+
+
+def _is_plain_result(event: Event) -> bool:
+  """True if the event is a tool result with no control-flow actions.
+
+  Besides the tool's state and artifact deltas, any action (a transfer, an
+  escalation, a confirmation or credential request whose tool has not run)
+  is an instruction the failed step never carried out, so it is not kept.
+  finish_task sets no action and is detected by name.
+  """
+  actions = event.actions.model_copy(
+      update={
+          'state_delta': {},
+          'artifact_delta': {},
+          'skip_summarization': None,
+          'render_ui_widgets': None,
+      }
+  )
+  if actions != EventActions():
+    return False
+  return not any(
+      response.name == FINISH_TASK_TOOL_NAME
+      for response in event.get_function_responses()
+  )
+
+
 async def _execute_prepared_function_calls_async(
     invocation_context: InvocationContext,
     prepared_calls: list[_PreparedFunctionCall],

src/google/adk/workflow/_node_runner.py

@@ -150,6 +150,10 @@ async def run(
         logger.debug("node %s cancelled via signal.", ctx.node_path)
         raise
       except Exception as e:
+        # Before the DynamicNodeFailError return: a tool that runs a failing
+        # child node raises it out of this node's own tool batch.
+        await self._enqueue_kept_function_responses(ctx)
+
         if isinstance(e, DynamicNodeFailError):
           # TODO: consider to retry upon dynamic node failures later. This may
           # require thorough design to consider a workflow dynamic node and a
@@ -185,6 +189,30 @@ async def run(
       )
       attempt_count += 1
 
+  async def _enqueue_kept_function_responses(self, ctx: Context) -> None:
+    """Enqueues the results kept from a parallel tool batch that failed.
+
+    The node has stopped, so nothing in it can act on them. They go out before
+    the error event, and before a retry builds its next request. Once the
+    invocation is aborted, abort sealing decides how its calls are answered.
+    """
+    ic = ctx._invocation_context
+    if ic.is_aborted:
+      return
+    kept = ic._abort_state.pop_unpersisted_function_responses()  # pylint: disable=protected-access
+    if not kept:
+      return
+    # ctx.event_author is the author of the node's last yielded event, which
+    # in a ParallelAgent node can be another branch. Keep each result's own
+    # author, as BaseAgent._run_impl does for the events it yields.
+    event_author = ctx.event_author
+    try:
+      for event in kept:
+        ctx.event_author = event.author
+        await self._enqueue_event(event, ctx)
+    finally:
+      ctx.event_author = event_author
+
   async def _attempt_retry(self, e: Exception, attempt_count: int) -> bool:
     """Checks if node should retry and sleeps if so."""
     from ._node_state import NodeState

tests/unittests/flows/llm_flows/tools/test_batch_executor.py

@@ -20,11 +20,14 @@
 import contextvars
 from unittest import mock
 
+from google.adk.agents.invocation_context import _AbortState
+from google.adk.agents.llm.task._finish_task_tool import FINISH_TASK_TOOL_NAME
 from google.adk.events.event import Event
 from google.adk.events.event_actions import EventActions
 from google.adk.flows.llm_flows.tools import _batch_executor as _batch_tool_executor
 from google.adk.flows.llm_flows.tools._caller import _PreparedFunctionCall
 from google.adk.tools.base_tool import BaseTool
+from google.adk.tools.tool_confirmation import ToolConfirmation
 from google.adk.tools.tool_context import ToolContext
 from google.genai import types
 import pytest
@@ -227,6 +230,74 @@ async def test_execute_prepared_function_calls_sets_live_session_id_and_merges()
   assert [p.text for p in merged.content.parts] == ['r1', 'r2']
 
 
+def _response_event(name: str, **actions) -> Event:
+  return Event(
+      invocation_id='inv-1',
+      author='agent',
+      content=types.Content(
+          role='user',
+          parts=[types.Part.from_function_response(name=name, response={})],
+      ),
+      actions=EventActions(**actions),
+  )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'name, actions, kept',
+    [
+        ('plain', {}, True),
+        ('state_delta', {'state_delta': {'k': 'v'}}, True),
+        ('transfer', {'transfer_to_agent': 'other'}, False),
+        ('escalate', {'escalate': True}, False),
+        (
+            'confirmation',
+            {'requested_tool_confirmations': {'id': ToolConfirmation()}},
+            False,
+        ),
+        ('set_model_response', {'set_model_response': {'a': 1}}, False),
+        (FINISH_TASK_TOOL_NAME, {}, False),
+    ],
+)
+async def test_keep_completed_responses_keeps_only_finished_plain_results(
+    name: str, actions: dict, kept: bool
+) -> None:
+  """Finished plain results are merged and kept; control actions are not."""
+  mock_ic = mock.MagicMock()
+  mock_ic.session.state = {}
+  mock_ic.live_request_queue = None
+  mock_ic._abort_state = _AbortState()
+
+  async def returns(event):
+    return event
+
+  async def raises():
+    raise RuntimeError('boom')
+
+  tasks = [
+      asyncio.create_task(returns(_response_event('fast'))),
+      asyncio.create_task(asyncio.Event().wait()),
+      asyncio.create_task(raises()),
+      asyncio.create_task(returns(None)),
+      asyncio.create_task(returns(_response_event(name, **actions))),
+  ]
+  await asyncio.sleep(0)
+  tasks[1].cancel()
+  await asyncio.gather(*tasks, return_exceptions=True)
+
+  # The cancelled call, the call that raised and the None result get nothing.
+  _batch_tool_executor._keep_completed_responses(mock_ic, tasks)
+  [merged] = mock_ic._abort_state.pop_unpersisted_function_responses()
+  assert [r.name for r in merged.get_function_responses()] == (
+      ['fast', name] if kept else ['fast']
+  )
+
+  # Live mode keeps nothing: there is nothing that would persist it.
+  mock_ic.live_request_queue = mock.MagicMock()
+  _batch_tool_executor._keep_completed_responses(mock_ic, tasks)
+  assert not mock_ic._abort_state.unpersisted_function_responses
+
+
 @pytest.mark.asyncio
 async def test_handle_function_calls_live_compat_wrapper() -> None:
   mock_context = mock.MagicMock()