diff --git a/litellm/llms/a2a/chat/streaming_iterator.py b/litellm/llms/a2a/chat/streaming_iterator.py index 1d00889679c..4cf7237afde 100644 --- a/litellm/llms/a2a/chat/streaming_iterator.py +++ b/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 diff --git a/tests/test_litellm/llms/a2a/chat/test_a2a_chat_transformation.py b/tests/test_litellm/llms/a2a/chat/test_a2a_chat_transformation.py index 636e8ba23c3..9df586b45b1 100644 --- a/tests/test_litellm/llms/a2a/chat/test_a2a_chat_transformation.py +++ b/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}"