fix(a2a): suppress a terminal snapshot that repeats only the reply's tail

Snapshot detection assumed a terminal snapshot extends everything already
emitted. That holds for an agent that only ever speaks for itself, but not for
one that delegates: it reports sub-agent progress into the same task before
answering, so its closing status-update and artifact-update carry just the
answer. That extends nothing, escapes the prefix test, and the reply renders
three times over - once from the deltas, once from the status-update and once
from the artifact.

Read text that merely repeats the tail of what has been delivered as a snapshot
as well. A whitespace-only event has an empty comparison key and would match any
tail, so it is excluded and still emitted as a delta.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

litellm/llms/a2a/chat/streaming_iterator.py

@@ -131,6 +131,11 @@ class A2AModelResponseIterator(BaseModelResponseIterator):
         Handles both streaming styles: servers that send deltas ("O", "K") and servers that
         send growing snapshots ("O", "OK") collapse to the same output.
 
+        A snapshot need not repeat the whole stream. An agent that delegates reports
+        sub-agent progress into the same task before answering, so its closing events carry
+        only its own answer, extending nothing. Text that merely repeats the tail of what has
+        been delivered is therefore read as a snapshot as well.
+
         A2A marks no event as delta-or-snapshot, so an event whose text equals everything
         emitted so far is necessarily ambiguous. It is read as a snapshot, because servers
         repeating the whole reply at the end of a stream are common while a delta that
@@ -156,6 +161,15 @@ class A2AModelResponseIterator(BaseModelResponseIterator):
             self._emitted_text += suffix
             return suffix
 
+        if emitted_key and text_key and emitted_key.endswith(text_key):
+            # A terminal snapshot that repeats only the agent's own answer instead of the
+            # whole stream. A delegating agent reports sub-agent progress into the same task
+            # before answering, so its closing ``status-update`` and ``artifact-update`` carry
+            # just the answer: that extends nothing already emitted and so escapes the
+            # prefix test above, and would be forwarded twice more. Everything such a
+            # snapshot carries has already been delivered, so emit nothing.
+            return ""
+
         self._emitted_text += text
         return text
 

tests/test_litellm/llms/a2a/chat/test_a2a_chat_transformation.py

@@ -141,6 +141,10 @@ def test_kagent_stream_finishes_on_completed_state():
             id="multipart_snapshot_extends",
         ),
         pytest.param(["", "OK", ""], "OK", id="empty_events_ignored"),
+        # A whitespace-only delta carries no non-whitespace text, so it must not be
+        # mistaken for a snapshot repeating the tail of the stream.
+        pytest.param(["Hello", " "], "Hello ", id="whitespace_only_delta_kept"),
+        pytest.param(["chatter", "answer", "answer"], "chatteranswer", id="snapshot_repeats_only_tail"),
     ],
 )
 def test_incremental_text_reduction(texts, expected):
@@ -228,3 +232,37 @@ def _task(*, status_role: str | None = None, status_text: str = "", artifact_tex
 def test_extract_text_skips_caller_authored_messages(response, expected):
     """Caller-authored parts are skipped, never returned, and never hide the agent's reply."""
     assert extract_text_from_a2a_response(response) == expected
+
+
+# A delegating agent reports sub-agent progress into the same task before producing its own
+# answer. Modelled on kagent's ADK executor, which enqueues a status-update for every ADK event
+# and then re-sends the aggregated answer as BOTH a terminal status-update and a final artifact
+# (kagent-adk/src/kagent/adk/_agent_executor.py: run loop, then task result publication).
+ANSWER = "Diagnosis: optic degraded on emm001a-jnx-01."
+
+DELEGATING_AGENT_STREAM = [
+    _status_update(text="Diagnose the optical alarm on emm001a-jnx-01.", role="user", state="submitted"),
+    _status_update(text="Calling telemetry-agent..."),
+    _status_update(text="telemetry-agent: no anomalies found."),
+    _status_update(text="Diagnosis: optic degraded"),
+    _status_update(text=" on emm001a-jnx-01."),
+    _status_update(text=ANSWER),
+    _artifact_update(ANSWER),
+    _status_update(state="completed", final=True),
+]
+
+
+def test_delegating_agent_answer_is_not_repeated_after_tool_chatter():
+    """
+    Regression: a terminal snapshot repeats only the agent's answer, not the progress text
+    streamed ahead of it, so snapshot suppression cannot depend on the snapshot extending
+    everything emitted so far.
+
+    Without this, a delegating agent renders its answer three times: once from the deltas,
+    once from the terminal status-update and once from the final artifact.
+    """
+    iterator = _iterator()
+    rendered = "".join(iterator.chunk_parser(e)["text"] for e in DELEGATING_AGENT_STREAM)
+
+    assert rendered.count(ANSWER) == 1
+    assert rendered == f"Calling telemetry-agent...telemetry-agent: no anomalies found.{ANSWER}"