From 201c37c18c80fc38a757112d945228e24cad621b Mon Sep 17 00:00:00 2001 From: Chaitanya Laxman Date: Sat, 31 Oct 2026 01:33:17 +0400 Subject: [PATCH] fix(a2a): do not forward a nested task hop's finish_task A task-mode caller treats any terminal finish_task as the remote task completing. Both executors drop that call or terminal response at the top of _on_event, and only when the branch carries a run id. Text in the same event stays. ParallelAgent's parent.child branch has no run id, so a served parallel root still completes. An unset branch is kept, including a Workflow child with a different author. Fixes #7745 --- .../adk/a2a/executor/a2a_agent_executor.py | 13 ++- .../a2a/executor/a2a_agent_executor_impl.py | 8 +- .../adk/agents/llm/task/_finish_task_tool.py | 39 +++++++++ .../a2a/executor/test_a2a_agent_executor.py | 80 +++++++++++++++++++ .../executor/test_a2a_agent_executor_impl.py | 80 +++++++++++++++++++ .../agents/llm/task/test_finish_task_tool.py | 48 +++++++++++ 6 files changed, 263 insertions(+), 5 deletions(-) diff --git a/src/google/adk/a2a/executor/a2a_agent_executor.py b/src/google/adk/a2a/executor/a2a_agent_executor.py index 2a08ed2860d..60ae60dbe6c 100644 --- a/src/google/adk/a2a/executor/a2a_agent_executor.py +++ b/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: diff --git a/src/google/adk/a2a/executor/a2a_agent_executor_impl.py b/src/google/adk/a2a/executor/a2a_agent_executor_impl.py index 264e30d262f..f591785627f 100644 --- a/src/google/adk/a2a/executor/a2a_agent_executor_impl.py +++ b/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 ) diff --git a/src/google/adk/agents/llm/task/_finish_task_tool.py b/src/google/adk/agents/llm/task/_finish_task_tool.py index cd747e26f0b..0d5e0163fb3 100644 --- a/src/google/adk/agents/llm/task/_finish_task_tool.py +++ b/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. diff --git a/tests/unittests/a2a/executor/test_a2a_agent_executor.py b/tests/unittests/a2a/executor/test_a2a_agent_executor.py index 8395adebcab..90ed6620a53 100644 --- a/tests/unittests/a2a/executor/test_a2a_agent_executor.py +++ b/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() diff --git a/tests/unittests/a2a/executor/test_a2a_agent_executor_impl.py b/tests/unittests/a2a/executor/test_a2a_agent_executor_impl.py index d4c222fc6cc..99cb10c703d 100644 --- a/tests/unittests/a2a/executor/test_a2a_agent_executor_impl.py +++ b/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() diff --git a/tests/unittests/agents/llm/task/test_finish_task_tool.py b/tests/unittests/agents/llm/task/test_finish_task_tool.py index 9ce4025cbe7..2ac4acb7150 100644 --- a/tests/unittests/agents/llm/task/test_finish_task_tool.py +++ b/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()