From b23b705f5929fab71e5f19b466a6f0104c7e77bc Mon Sep 17 00:00:00 2001 From: Yatsuiii Date: Sun, 18 Oct 2026 09:46:45 +0530 Subject: [PATCH] fix(a2a): do not convert a resumed task's earlier artifacts again A non-streaming A2A response is the whole task, and a task's artifacts accumulate over its life. When a task that paused for input, such as a tool confirmation or a long-running tool, is resumed, the response carries every earlier turn's artifacts again. Both task converters flatten all of them, so RemoteA2aAgent emitted the earlier turns again: after a confirmation, the original tool call and its "requires confirmation" response arrived again with the result. Before converting a task, RemoteA2aAgent now drops the artifact parts it already converted for that task. It reads them from the raw task snapshots recorded on its earlier events, which are written after the after-request hooks run, so overridden hooks and message converters keep the fix; only a custom task converter still receives the whole task. Parts are compared by content, so an artifact replaced under the same id is still converted, and only the new parts of an appended one are. When every artifact was already converted, the task history is dropped from the copy too, so the legacy converter cannot fall back to an earlier turn's message. --- src/google/adk/a2a/_compat.py | 54 ++++++ src/google/adk/a2a/agent/_remote_a2a_agent.py | 59 ++++++- .../a2a/agent/test_remote_a2a_agent.py | 155 ++++++++++++++++++ .../a2a/integration/test_client_server.py | 153 +++++++++++++++++ .../a2a/integration/test_client_server_v1.py | 153 +++++++++++++++++ tests/unittests/a2a/test_compat.py | 77 +++++++++ 6 files changed, 649 insertions(+), 2 deletions(-) diff --git a/src/google/adk/a2a/_compat.py b/src/google/adk/a2a/_compat.py index 51850a92d6d..faf17f6272a 100644 --- a/src/google/adk/a2a/_compat.py +++ b/src/google/adk/a2a/_compat.py @@ -33,7 +33,9 @@ from typing import AsyncGenerator from typing import Callable from typing import cast +from typing import Mapping from typing import Optional +from typing import Sequence from typing import TYPE_CHECKING from typing import TypeAlias from typing import TypeVar @@ -824,6 +826,58 @@ def normalize(item: Any) -> Any: return normalize +def task_without_seen_artifact_parts( + task: Task, seen: Mapping[str, Sequence[Any]] +) -> Task: + """Returns a copy of ``task`` without the artifact parts already seen. + + A task's artifacts accumulate over its whole life, so a non-streaming response + to a resumed task carries every earlier turn's artifacts again. ``seen`` maps + an artifact id to the parts last received for it, in ``a2a_to_dict`` form. + Only the leading parts equal to those are dropped: an artifact that was + appended to keeps its new parts, one replaced under the same id is kept whole, + and one with nothing new is dropped. If every artifact was seen, the history + goes too, so a converter cannot fall back to an earlier turn's message. ``task`` + itself is not modified. + """ + + def new_parts_from(artifact: Artifact) -> int: + earlier = seen.get(artifact.artifact_id) or () + index = 0 + for part, earlier_part in zip(artifact.parts, earlier): + if a2a_to_dict(part) != earlier_part: + break + index += 1 + return index + + kept = [ + (artifact, start) + for artifact in task.artifacts + if (start := new_parts_from(artifact)) < len(artifact.parts) + ] + drop_history = bool(task.artifacts) and not kept + if IS_A2A_V1: + copy = Task() + copy.CopyFrom(task) + del copy.artifacts[:] + for artifact, start in kept: + new = copy.artifacts.add() + new.CopyFrom(artifact) + del new.parts[:start] + if drop_history: + del copy.history[:] + return copy + update: dict[str, Any] = { + "artifacts": [ + artifact.model_copy(update={"parts": artifact.parts[start:]}) + for artifact, start in kept + ] + } + if drop_history: + update["history"] = [] + return task.model_copy(update=update) + + # ----------------------------------------------------------------------------- # send_message adapter # ----------------------------------------------------------------------------- diff --git a/src/google/adk/a2a/agent/_remote_a2a_agent.py b/src/google/adk/a2a/agent/_remote_a2a_agent.py index ad733413066..f404fd40faf 100644 --- a/src/google/adk/a2a/agent/_remote_a2a_agent.py +++ b/src/google/adk/a2a/agent/_remote_a2a_agent.py @@ -32,6 +32,7 @@ from a2a.types import AgentCard from a2a.types import Message as A2AMessage from a2a.types import Part as A2APart +from a2a.types import Task as A2ATask from a2a.types import TaskArtifactUpdateEvent as A2ATaskArtifactUpdateEvent from a2a.types import TaskState from a2a.types import TaskStatusUpdateEvent as A2ATaskStatusUpdateEvent @@ -1560,6 +1561,46 @@ def _get_resumable_task_id( return task_id if isinstance(task_id, str) else None return None + def _without_surfaced_artifacts( + self, task: A2ATask, ctx: InvocationContext + ) -> A2ATask: + """Returns ``task`` without the artifacts this agent already converted. + + A non-streaming response is the whole task, and a task's artifacts + accumulate over its life, so a task resumed after ``input-required`` brings + back every earlier turn's output. Each event this agent yields records the + raw task as it was then, so what was already converted is read from those + snapshots. They are recorded after the after-request hooks run, so a hook + cannot alter them, and an event that a converter or hook suppressed records + nothing, so its artifacts are converted when they come again. + """ + if not getattr(task, "artifacts", None): + return task + seen: dict[str, list[Any]] = {} + for event in ctx.session.events: + metadata = event.custom_metadata or {} + if ( + event.author != self.name + or metadata.get(A2A_METADATA_PREFIX + "task_id") != task.id + ): + continue + response = metadata.get(A2A_METADATA_PREFIX + "response") + if not isinstance(response, dict): + continue + for artifact in response.get("artifacts") or []: + if not isinstance(artifact, dict): + continue + # The recorded snapshot is the wire form: camelCase on both versions. + # Later snapshots win, so a replaced artifact is compared to what was + # received last. + artifact_id = artifact.get("artifactId") or artifact.get("artifact_id") + parts = artifact.get("parts") + if artifact_id and isinstance(parts, list): + seen[artifact_id] = parts + if not seen: + return task + return _compat.task_without_seen_artifact_parts(task, seen) + def _has_default_response_persistence(self) -> bool: """Whether recorded responses can safely determine task continuation. @@ -1616,7 +1657,10 @@ async def _handle_a2a_response( # response for a non-streaming task, which is the full task state. # We process this to get the initial message. event = convert_a2a_task_to_event( - task, self.name, ctx, self._a2a_part_converter + self._without_surfaced_artifacts(task, ctx), + self.name, + ctx, + self._a2a_part_converter, ) if not event: return None @@ -1770,8 +1814,19 @@ async def _handle_a2a_response_v2( if update is None: # This is the initial response for a streaming task or the complete # response for a non-streaming task. + # A custom task converter gets the whole task, as it always has, and + # decides for itself what a resumed task's earlier artifacts mean. + default_task_converter = ( + self._config.a2a_task_converter + is A2aRemoteAgentConfig.model_fields["a2a_task_converter"].default + ) event = self._config.a2a_task_converter( - task, self.name, ctx, self._config.a2a_part_converter + self._without_surfaced_artifacts(task, ctx) + if default_task_converter + else task, + self.name, + ctx, + self._config.a2a_part_converter, ) elif isinstance(update, A2ATaskStatusUpdateEvent): # This is a streaming task status update. diff --git a/tests/unittests/a2a/agent/test_remote_a2a_agent.py b/tests/unittests/a2a/agent/test_remote_a2a_agent.py index 1dcafc603c8..85a30c7349b 100644 --- a/tests/unittests/a2a/agent/test_remote_a2a_agent.py +++ b/tests/unittests/a2a/agent/test_remote_a2a_agent.py @@ -8748,3 +8748,158 @@ async def override_metadata(ctx, response, event): assert turn[-1].custom_metadata["a2a:task_id"] == "caller-task" assert turn[-1].custom_metadata["a2a:context_id"] == "caller-context" assert turn[-1].custom_metadata["a2a:response"]["id"] == "currency-task" + + +def _task_with_artifacts(*artifacts: tuple[str, list[str]], history=None): + return _compat.make_task( + id="task-1", + context_id="context-1", + status=_compat.make_task_status(_compat.TS_COMPLETED), + artifacts=[ + _compat.make_artifact( + artifact_id=artifact_id, + parts=[_compat.make_text_part(text) for text in texts], + ) + for artifact_id, texts in artifacts + ], + **({"history": history} if history else {}), + ) + + +def _context_after(*seen: tuple[str, list[str]]): + """A context whose session holds one earlier response of task-1.""" + earlier = Event( + author="test_agent", + custom_metadata={ + A2A_METADATA_PREFIX + "task_id": "task-1", + A2A_METADATA_PREFIX + + "response": _compat.a2a_to_dict(_task_with_artifacts(*seen)), + }, + ) + session = Mock(spec=Session) + session.events = [earlier] + context = Mock(spec=InvocationContext) + context.session = session + context.invocation_id = "invocation-1" + context.branch = "main" + return context + + +def _earlier_message(): + return _compat.make_message( + role=_compat.ROLE_AGENT, + parts=[_compat.make_text_part("earlier: please confirm")], + message_id="message-1", + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "handler", ["_handle_a2a_response", "_handle_a2a_response_v2"] +) +@pytest.mark.parametrize( + ("seen", "response", "expected"), + [ + pytest.param( + [("artifact-1", ["turn one"])], + [("artifact-1", ["turn one"]), ("artifact-2", ["turn two"])], + ["turn two"], + id="earlier_turn_skipped", + ), + pytest.param( + [("artifact-1", ["first answer"])], + [("artifact-1", ["second answer"])], + ["second answer"], + id="replaced_artifact_kept", + ), + pytest.param( + [("artifact-1", ["turn one"])], + [("artifact-1", ["turn one"])], + [], + id="nothing_new_and_no_history_fallback", + ), + ], +) +async def test_a_resumed_task_converts_only_artifacts_not_seen_before( + handler, seen, response, expected +): + """A resumed task's earlier artifacts are not converted a second time. + + A non-streaming response is the whole task, so after a pause it carries what + the agent already received. The session records that in the earlier event's + task snapshot, and both handlers skip it, comparing content so an artifact + replaced under the same id still comes through. With nothing new, an earlier + message in the task history must not come back either. + """ + agent = RemoteA2aAgent(name="test_agent", agent_card=create_test_agent_card()) + + event = await getattr(agent, handler)( + (_task_with_artifacts(*response, history=[_earlier_message()]), None), + _context_after(*seen), + ) + + texts = ( + [part.text for part in event.content.parts] + if event and event.content and event.content.parts + else [] + ) + assert texts == expected + + +@pytest.mark.asyncio +async def test_a_resumed_task_is_deduplicated_with_other_overrides(): + """Custom message converters and after-request hooks keep the fix. + + Snapshots are recorded after the hooks run, so neither can alter what was + already received. Only a custom task converter opts out; see below. + """ + + async def after_request(ctx, response, event): + return event + + agent = RemoteA2aAgent( + name="test_agent", + agent_card=create_test_agent_card(), + config=A2aRemoteAgentConfig( + a2a_message_converter=lambda *args, **kwargs: None, + request_interceptors=[ + RequestInterceptor(after_request=after_request) + ], + ), + ) + + event = await agent._handle_a2a_response_v2( + ( + _task_with_artifacts( + ("artifact-1", ["turn one"]), ("artifact-2", ["turn two"]) + ), + None, + ), + _context_after(("artifact-1", ["turn one"])), + ) + + assert [part.text for part in event.content.parts] == ["turn two"] + + +@pytest.mark.asyncio +async def test_a_custom_task_converter_still_gets_the_whole_task(): + received = [] + + def task_converter(task, *args, **kwargs): + received.append(task) + return None + + agent = RemoteA2aAgent( + name="test_agent", + agent_card=create_test_agent_card(), + config=A2aRemoteAgentConfig(a2a_task_converter=task_converter), + ) + task = _task_with_artifacts( + ("artifact-1", ["turn one"]), ("artifact-2", ["turn two"]) + ) + + await agent._handle_a2a_response_v2( + (task, None), _context_after(("artifact-1", ["turn one"])) + ) + + assert received == [task] and received[0] is task diff --git a/tests/unittests/a2a/integration/test_client_server.py b/tests/unittests/a2a/integration/test_client_server.py index 690140d9fc5..b782058a0df 100644 --- a/tests/unittests/a2a/integration/test_client_server.py +++ b/tests/unittests/a2a/integration/test_client_server.py @@ -1035,3 +1035,156 @@ async def test_user_follow_up_sends_task_id_with_input_required_legacy_impl(): call_args_2 = mock_handler.on_message_send.call_args_list[1] params_2 = call_args_2[0][0] assert params_2.message.task_id == task_id + + +def _confirmation_run_async(received_requests: list): + """The events a tool with ``require_confirmation=True`` produces, as a script. + + Turn 1 is the tool call, its "requires confirmation" response, and the + ``adk_request_confirmation`` call the client answers; turn 2 runs the tool and + answers. The first two are ordinary events, so they become task artifacts. + """ + + async def run_async(**kwargs): + received_requests.append(kwargs) + if len(received_requests) == 1: + yield Event( + author="FakeAgent", + content=types.Content( + parts=[ + types.Part( + function_call=types.FunctionCall( + name="delete_file", + args={"path": "/tmp/x"}, + id="call_1", + ) + ) + ], + role="model", + ), + ) + yield Event( + author="FakeAgent", + content=types.Content( + parts=[ + types.Part( + function_response=types.FunctionResponse( + name="delete_file", + response={"error": "requires confirmation"}, + id="call_1", + ) + ) + ], + role="user", + ), + ) + yield Event( + author="FakeAgent", + content=types.Content( + parts=[ + types.Part( + function_call=types.FunctionCall( + name="adk_request_confirmation", + args={"originalFunctionCall": {"id": "call_1"}}, + id="confirm_1", + ) + ) + ], + role="model", + ), + long_running_tool_ids={"confirm_1"}, + ) + else: + yield Event( + author="FakeAgent", + content=types.Content( + parts=[ + types.Part( + function_response=types.FunctionResponse( + name="delete_file", + response={"deleted": "/tmp/x"}, + id="call_1", + ) + ) + ], + role="user", + ), + ) + yield Event( + author="FakeAgent", + content=types.Content( + parts=[types.Part(text="Deleted it.")], role="model" + ), + ) + + return run_async + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streaming", [False, True]) +async def test_a_resumed_task_returns_only_its_new_turn(streaming): + """A task resumed after input-required does not bring its earlier turns back. + + A non-streaming response is the whole task, and the task's artifacts + accumulate, so turn 1's tool call and its "requires confirmation" response + used to arrive again with the turn-2 answer. Each turn carries only what it + produced, as when streaming. + """ + received_requests = [] + app = create_server_app( + _confirmation_run_async(received_requests), + force_new_version=True, + ) + + agent = create_client(app, streaming=streaming) + session_service = InMemorySessionService() + await session_service.create_session( + app_name="ClientApp", user_id="test_user", session_id="test_session" + ) + client_runner = Runner( + app_name="ClientApp", agent=agent, session_service=session_service + ) + + async def run_turn(message: types.Content) -> list[Event]: + return [ + event + async for event in client_runner.run_async( + user_id="test_user", + session_id="test_session", + new_message=message, + ) + if not event.partial + ] + + turn_1 = await run_turn( + types.Content(parts=[types.Part(text="Hi")], role="user") + ) + turn_2 = await run_turn( + types.Content( + parts=[ + types.Part( + function_response=types.FunctionResponse( + name="adk_request_confirmation", + response={"confirmed": True}, + id="confirm_1", + ) + ) + ], + role="user", + ) + ) + + assert "adk_request_confirmation" in [ + c.name for e in turn_1 for c in e.get_function_calls() + ] + assert [c.name for e in turn_2 for c in e.get_function_calls()] == [] + assert [r.response for e in turn_2 for r in e.get_function_responses()] == [ + {"deleted": "/tmp/x"} + ] + assert [ + p.text + for e in turn_2 + if e.content and e.content.parts + for p in e.content.parts + if p.text + ] == ["Deleted it."] diff --git a/tests/unittests/a2a/integration/test_client_server_v1.py b/tests/unittests/a2a/integration/test_client_server_v1.py index 37d1c32726a..2d01e4b2aec 100644 --- a/tests/unittests/a2a/integration/test_client_server_v1.py +++ b/tests/unittests/a2a/integration/test_client_server_v1.py @@ -847,3 +847,156 @@ def _status_update(text: str, state: int) -> pb.StreamResponse: # the latest status is applied to the running task. final_task, _ = results[2] assert final_task.status.state == pb.TASK_STATE_COMPLETED + + +def _confirmation_run_async(received_requests: list): + """The events a tool with ``require_confirmation=True`` produces, as a script. + + Turn 1 is the tool call, its "requires confirmation" response, and the + ``adk_request_confirmation`` call the client answers; turn 2 runs the tool and + answers. The first two are ordinary events, so they become task artifacts. + """ + + async def run_async(**kwargs): + received_requests.append(kwargs) + if len(received_requests) == 1: + yield Event( + author="FakeAgent", + content=types.Content( + parts=[ + types.Part( + function_call=types.FunctionCall( + name="delete_file", + args={"path": "/tmp/x"}, + id="call_1", + ) + ) + ], + role="model", + ), + ) + yield Event( + author="FakeAgent", + content=types.Content( + parts=[ + types.Part( + function_response=types.FunctionResponse( + name="delete_file", + response={"error": "requires confirmation"}, + id="call_1", + ) + ) + ], + role="user", + ), + ) + yield Event( + author="FakeAgent", + content=types.Content( + parts=[ + types.Part( + function_call=types.FunctionCall( + name="adk_request_confirmation", + args={"originalFunctionCall": {"id": "call_1"}}, + id="confirm_1", + ) + ) + ], + role="model", + ), + long_running_tool_ids={"confirm_1"}, + ) + else: + yield Event( + author="FakeAgent", + content=types.Content( + parts=[ + types.Part( + function_response=types.FunctionResponse( + name="delete_file", + response={"deleted": "/tmp/x"}, + id="call_1", + ) + ) + ], + role="user", + ), + ) + yield Event( + author="FakeAgent", + content=types.Content( + parts=[types.Part(text="Deleted it.")], role="model" + ), + ) + + return run_async + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streaming", [False, True]) +async def test_a_resumed_task_returns_only_its_new_turn(streaming): + """A task resumed after input-required does not bring its earlier turns back. + + A non-streaming response is the whole task, and the task's artifacts + accumulate, so turn 1's tool call and its "requires confirmation" response + used to arrive again with the turn-2 answer. Each turn carries only what it + produced, as when streaming. + """ + received_requests = [] + app = create_server_app_v1( + _confirmation_run_async(received_requests), force_new_version=True + ) + + async with app.router.lifespan_context(app): + agent = create_client(app, streaming=streaming) + session_service = InMemorySessionService() + await session_service.create_session( + app_name="ClientApp", user_id="test_user", session_id="test_session" + ) + client_runner = Runner( + app_name="ClientApp", agent=agent, session_service=session_service + ) + + async def run_turn(message: types.Content) -> list[Event]: + return [ + event + async for event in client_runner.run_async( + user_id="test_user", + session_id="test_session", + new_message=message, + ) + if not event.partial + ] + + turn_1 = await run_turn( + types.Content(parts=[types.Part(text="Hi")], role="user") + ) + turn_2 = await run_turn( + types.Content( + parts=[ + types.Part( + function_response=types.FunctionResponse( + name="adk_request_confirmation", + response={"confirmed": True}, + id="confirm_1", + ) + ) + ], + role="user", + ) + ) + + assert "adk_request_confirmation" in [ + c.name for e in turn_1 for c in e.get_function_calls() + ] + assert [c.name for e in turn_2 for c in e.get_function_calls()] == [] + assert [r.response for e in turn_2 for r in e.get_function_responses()] == [ + {"deleted": "/tmp/x"} + ] + assert [ + p.text + for e in turn_2 + if e.content and e.content.parts + for p in e.content.parts + if p.text + ] == ["Deleted it."] diff --git a/tests/unittests/a2a/test_compat.py b/tests/unittests/a2a/test_compat.py index 1fd7a2b6040..6092949f247 100644 --- a/tests/unittests/a2a/test_compat.py +++ b/tests/unittests/a2a/test_compat.py @@ -705,3 +705,80 @@ def test_stream_normalizer_task_snapshot_with_artifacts_is_authoritative(): task, _ = normalize(_v1_artifact_chunk('art-2', ' appended', append=True)) by_id = {a.artifact_id: [p.text for p in a.parts] for a in task.artifacts} assert by_id == {'art-1': ['fresh'], 'art-2': ['other', ' appended']} + + +def _task_with_artifacts(*artifacts: tuple[str, list[str]], history=None): + return _compat.make_task( + id='task-1', + context_id='context-1', + status=_compat.make_task_status(_compat.TS_COMPLETED), + artifacts=[ + _compat.make_artifact( + artifact_id=artifact_id, + parts=[_compat.make_text_part(text) for text in texts], + ) + for artifact_id, texts in artifacts + ], + **({'history': history} if history else {}), + ) + + +def _seen(*artifacts: tuple[str, list[str]]): + """The parts already received, as RemoteA2aAgent reads them from a snapshot.""" + return { + artifact['artifactId']: artifact['parts'] + for artifact in _compat.a2a_to_dict(_task_with_artifacts(*artifacts))[ + 'artifacts' + ] + } + + +def _artifact_texts(task) -> list[list[str]]: + return [ + [part.get('text') for part in artifact['parts']] + for artifact in _compat.a2a_to_dict(task).get('artifacts') or [] + ] + + +def test_task_without_seen_artifact_parts_drops_what_was_seen(): + task = _task_with_artifacts( + ('turn-1', ['call']), ('grown', ['a', 'b']), ('turn-2', ['answer']) + ) + + unseen = _compat.task_without_seen_artifact_parts( + task, _seen(('turn-1', ['call']), ('grown', ['a'])) + ) + + # A fully seen artifact goes; one that grew keeps only its new parts. + assert _artifact_texts(unseen) == [['b'], ['answer']] + assert unseen.id == task.id + assert _artifact_texts(task) == [['call'], ['a', 'b'], ['answer']] + + +def test_task_without_seen_artifact_parts_keeps_a_replaced_artifact(): + # A2A lets an update replace an artifact under the same id. The same id and + # part count with different content is new output, not a repeat. + task = _task_with_artifacts(('result', ['second answer'])) + + unseen = _compat.task_without_seen_artifact_parts( + task, _seen(('result', ['first answer'])) + ) + + assert _artifact_texts(unseen) == [['second answer']] + + +def test_task_without_seen_artifact_parts_drops_history_once_all_was_seen(): + earlier = _compat.make_message( + role=_compat.ROLE_AGENT, + parts=[_compat.make_text_part('earlier')], + message_id='message-1', + ) + task = _task_with_artifacts(('a', ['x']), history=[earlier]) + + all_seen = _compat.task_without_seen_artifact_parts(task, _seen(('a', ['x']))) + some_new = _compat.task_without_seen_artifact_parts(task, {}) + + assert not _compat.a2a_to_dict(all_seen).get('artifacts') + assert not _compat.a2a_to_dict(all_seen).get('history') + assert _artifact_texts(some_new) == [['x']] + assert _compat.a2a_to_dict(some_new).get('history')