fix(a2a): do not convert a resumed task's earlier artifacts again

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
 # -----------------------------------------------------------------------------

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.

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