fix(a2a): do not forward a nested task hop's finish_task

src/google/adk/a2a/executor/a2a_agent_executor.py

@@ -32,6 +32,7 @@
 from typing_extensions import override
 
 from .. import _compat
+from ...agents.llm.task._finish_task_tool import without_nested_finish_task
 from ...events.event import Event
 from ..agent.interceptors.new_integration_extension import _NEW_A2A_ADK_INTEGRATION_EXTENSION
 from ..converters.request_converter import AgentRunRequest
@@ -244,11 +245,15 @@ async def _handle_request(
 
     async def _on_event(adk_event: Event) -> None:
       nonlocal aborted, last_adk_event
-      last_adk_event = adk_event
-      if adk_event and _is_abort_adk_event(adk_event):
+      # Before last_adk_event: a nested finish_task is not this task.
+      forwarded = without_nested_finish_task(adk_event)
+      if forwarded is None:
+        return
+      last_adk_event = forwarded
+      if _is_abort_adk_event(forwarded):
         aborted = True
       for a2a_event in self._config.event_converter(
-          adk_event,
+          forwarded,
           invocation_context,
           task_id,
           context_id,
@@ -257,7 +262,7 @@ async def _on_event(adk_event: Event) -> None:
         a2a_events = await execute_after_event_interceptors(
             a2a_event,
             executor_context,
-            adk_event,
+            forwarded,
             self._config.execute_interceptors,
         )
         for e in a2a_events:

src/google/adk/a2a/executor/a2a_agent_executor_impl.py

@@ -30,6 +30,7 @@
 from typing_extensions import override
 
 from .. import _compat
+from ...agents.llm.task._finish_task_tool import without_nested_finish_task
 from ...events.event import Event
 from ...runners import Runner
 from ...sessions import base_session_service
@@ -209,7 +210,12 @@ async def _handle_request(
 
     async def _on_event(adk_event: Event) -> None:
       nonlocal canceled_event, error_event
-      if adk_event and _is_abort_adk_event(adk_event):
+      # A nested finish_task is not this task. Other parts in the event stay.
+      forwarded = without_nested_finish_task(adk_event)
+      if forwarded is None:
+        return
+      adk_event = forwarded
+      if _is_abort_adk_event(adk_event):
         canceled_event = create_error_status_event(
             adk_event, task_id, context_id
         )

src/google/adk/agents/llm/task/_finish_task_tool.py

@@ -25,6 +25,7 @@
 from pydantic import ValidationError
 from typing_extensions import override
 
+from ....events._branch_path import _BranchPath
 from ....events.event import Event
 from ....tools.base_tool import BaseTool
 from ....utils._schema_utils import schema_to_json_schema
@@ -206,6 +207,44 @@ async def run_async(
     return FINISH_TASK_SUCCESS_RESULT
 
 
+def _is_nested_finish_part(part: types.Part) -> bool:
+  """True for a finish_task call or a terminal finish_task response."""
+  call = part.function_call
+  if call is not None and call.name == FINISH_TASK_TOOL_NAME:
+    return True
+  response = part.function_response
+  if response is None or response.name != FINISH_TASK_TOOL_NAME:
+    return False
+  body = response.response or {}
+  return body.get('result') in (
+      FINISH_TASK_SUCCESS_RESULT,
+      FINISH_TASK_ERROR_RESULT,
+  )
+
+
+def without_nested_finish_task(event: Event) -> Event | None:
+  """Drop a task-hop finish_task, keeping any other parts.
+
+  A branch with a run id is a nested task hop. ParallelAgent's
+  ``parent.child`` has no run id, and an unset branch is the served
+  root. Returns None when the event has nothing left to forward.
+  """
+  branch = getattr(event, 'branch', None)
+  if not branch or not _BranchPath.from_string(branch).run_ids:
+    return event
+  content = event.content
+  if content is None or not content.parts:
+    return event
+  kept = [part for part in content.parts if not _is_nested_finish_part(part)]
+  if len(kept) == len(content.parts):
+    return event
+  if not kept:
+    return None
+  return event.model_copy(
+      update={'content': content.model_copy(update={'parts': kept})}
+  )
+
+
 def is_finish_task_terminal_fr(event: Event) -> bool:
   """True iff this event is a terminal FR from FinishTaskTool.
 

tests/unittests/a2a/executor/test_a2a_agent_executor.py

@@ -38,6 +38,8 @@
 from google.adk.runners import Runner
 from google.adk.sessions.in_memory_session_service import InMemorySessionService
 from google.genai.types import Content
+from google.genai.types import FunctionCall
+from google.genai.types import FunctionResponse
 from google.genai.types import Part
 import pytest
 
@@ -102,6 +104,35 @@ def _final_events(call_args_list):
   return [e for e in events if getattr(e, "final", False)]
 
 
+def _text_event(author, branch, *extra):
+  return Event(
+      author=author,
+      invocation_id="inv",
+      branch=branch,
+      content=Content(role="model", parts=[Part(text="kept"), *extra]),
+  )
+
+
+def _finish_event(author, branch):
+  return Event(
+      author=author,
+      invocation_id="inv",
+      branch=branch,
+      content=Content(
+          role="user",
+          parts=[
+              Part(
+                  function_response=FunctionResponse(
+                      id="fc",
+                      name="finish_task",
+                      response={"result": "Task completed."},
+                  )
+              )
+          ],
+      ),
+  )
+
+
 class TestA2aAgentExecutor:
   """Test suite for A2aAgentExecutor class."""
 
@@ -1676,3 +1707,52 @@ async def fixed_sig_run_async(
         e.status.state
         for e in _final_events(idle_queue.enqueue_event.call_args_list)
     ] == [_compat.TS_CANCELED]
+
+  @pytest.mark.asyncio
+  async def test_nested_run_id_finish_task_is_stripped(self):
+    """Drop a task-hop finish_task. Keep sibling text and root branches."""
+    self.mock_request_converter.return_value = AgentRunRequest(
+        user_id="test-user",
+        session_id="test-session",
+        new_message=Mock(spec=Content),
+        run_config=RunConfig(),
+    )
+    mock_session = Mock()
+    mock_session.id = "test-session"
+    self.mock_runner.session_service.get_session = AsyncMock(
+        return_value=mock_session
+    )
+    self.mock_runner._new_invocation_context.return_value = Mock()
+    call = Part(
+        function_call=FunctionCall(
+            id="fc", name="finish_task", args={"result": "x"}
+        )
+    )
+    text = _text_event("inner", "inner@fc")
+    mixed = _text_event("inner", "inner@fc", call)
+    nested = _finish_event("inner", "inner@fc")
+    middle = _text_event("middle", None)
+    parallel = _finish_event("child", "par.child")
+    root = _finish_event("reporter", None)
+
+    async def mock_run_async(**kwargs):
+      del kwargs
+      for item in (text, mixed, nested, middle, parallel, root):
+        yield item
+
+    self.mock_runner.run_async = mock_run_async
+    self.mock_event_converter.return_value = []
+    await self.executor.execute(self.mock_context, self.mock_event_queue)
+    forwarded = [
+        call.args[0] for call in self.mock_event_converter.call_args_list
+    ]
+    assert [event.author for event in forwarded] == [
+        "inner",
+        "inner",
+        "middle",
+        "child",
+        "reporter",
+    ]
+    assert forwarded[1].get_function_calls() == []
+    assert forwarded[1].content.parts[0].text == "kept"
+    assert mixed.get_function_calls()

tests/unittests/a2a/executor/test_a2a_agent_executor_impl.py

@@ -35,6 +35,7 @@
 from google.adk.runners import RunConfig
 from google.adk.runners import Runner
 from google.adk.sessions.base_session_service import GetSessionConfig
+from google.genai import types
 from google.genai.types import Content
 import pytest
 
@@ -65,6 +66,37 @@ def _final_events(call_args_list):
   return [e for e in events if getattr(e, "final", False)]
 
 
+def _text_event(author, branch, *extra):
+  return Event(
+      author=author,
+      invocation_id="inv",
+      branch=branch,
+      content=types.Content(
+          role="model", parts=[types.Part(text="kept"), *extra]
+      ),
+  )
+
+
+def _finish_event(author, branch):
+  return Event(
+      author=author,
+      invocation_id="inv",
+      branch=branch,
+      content=types.Content(
+          role="user",
+          parts=[
+              types.Part(
+                  function_response=types.FunctionResponse(
+                      id="fc",
+                      name="finish_task",
+                      response={"result": "Task completed."},
+                  )
+              )
+          ],
+      ),
+  )
+
+
 class TestA2aAgentExecutor:
   """Test suite for A2aAgentExecutor class."""
 
@@ -933,3 +965,51 @@ async def mock_run_async(**kwargs):
     assert self.mock_event_queue.enqueue_event.call_count == 1
     final_event = self.mock_event_queue.enqueue_event.call_args_list[0][0][0]
     assert final_event.status.state == _compat.TS_COMPLETED
+
+  @pytest.mark.asyncio
+  async def test_nested_run_id_finish_task_is_stripped(self):
+    """Drop a task-hop finish_task. Keep sibling text and root branches."""
+    self.mock_request_converter.return_value = AgentRunRequest(
+        user_id="test-user",
+        session_id="test-session",
+        new_message=Mock(spec=Content),
+        run_config=RunConfig(),
+    )
+    mock_session = Mock()
+    mock_session.id = "test-session"
+    self.mock_runner.session_service.get_session = AsyncMock(
+        return_value=mock_session
+    )
+    call = types.Part(
+        function_call=types.FunctionCall(
+            id="fc", name="finish_task", args={"result": "x"}
+        )
+    )
+    text = _text_event("inner", "inner@fc")
+    mixed = _text_event("inner", "inner@fc", call)
+    nested = _finish_event("inner", "inner@fc")
+    middle = _text_event("middle", None)
+    parallel = _finish_event("child", "par.child")
+    root = _finish_event("reporter", None)
+
+    async def mock_run_async(**kwargs):
+      del kwargs
+      for item in (text, mixed, nested, middle, parallel, root):
+        yield item
+
+    self.mock_runner.run_async = mock_run_async
+    self.mock_event_converter.return_value = []
+    await self.executor.execute(self.mock_context, self.mock_event_queue)
+    forwarded = [
+        item.args[0] for item in self.mock_event_converter.call_args_list
+    ]
+    assert [event.author for event in forwarded] == [
+        "inner",
+        "inner",
+        "middle",
+        "child",
+        "reporter",
+    ]
+    assert forwarded[1].get_function_calls() == []
+    assert forwarded[1].content.parts[0].text == "kept"
+    assert mixed.get_function_calls()

tests/unittests/agents/llm/task/test_finish_task_tool.py

@@ -21,6 +21,9 @@
 
 from google.adk.agents.llm.task._finish_task_tool import FINISH_TASK_TOOL_NAME
 from google.adk.agents.llm.task._finish_task_tool import FinishTaskTool
+from google.adk.agents.llm.task._finish_task_tool import without_nested_finish_task
+from google.adk.events.event import Event
+from google.genai import types
 from pydantic import BaseModel
 import pytest
 
@@ -361,3 +364,48 @@ def test_get_declaration_list_basemodel_defs_at_root(self):
     assert 'SampleOutputSchema' in schema['$defs']
     # The nested result schema should not contain $defs
     assert '$defs' not in schema['properties']['result']
+
+
+def test_without_nested_finish_task_keeps_sibling_parts():
+  text = types.Part(text='kept')
+  call = types.Part(
+      function_call=types.FunctionCall(
+          id='fc', name='finish_task', args={'result': 'x'}
+      )
+  )
+  done = types.Part(
+      function_response=types.FunctionResponse(
+          id='fc',
+          name='finish_task',
+          response={'result': 'Task completed.'},
+      )
+  )
+  retry = types.Part(
+      function_response=types.FunctionResponse(
+          id='fc',
+          name='finish_task',
+          response={'error': 'missing result'},
+      )
+  )
+
+  def ev(branch, parts):
+    return Event(
+        author='a',
+        invocation_id='i',
+        branch=branch,
+        content=types.Content(role='model', parts=parts),
+    )
+
+  root = without_nested_finish_task(ev(None, [done]))
+  parallel = without_nested_finish_task(ev('par.c', [done]))
+  plain = without_nested_finish_task(ev('inner@fc', [text]))
+  mixed_event = ev('inner@fc', [text, call])
+  mixed = without_nested_finish_task(mixed_event)
+  assert root is not None and root.content.parts == [done]
+  assert parallel is not None and parallel.content.parts == [done]
+  assert plain is not None and plain.content.parts == [text]
+  retry_event = without_nested_finish_task(ev('inner@fc', [retry]))
+  assert without_nested_finish_task(ev('inner@fc', [done])) is None
+  assert retry_event is not None and retry_event.content.parts == [retry]
+  assert mixed is not None and mixed.content.parts == [text]
+  assert mixed_event.get_function_calls()