From d88bdaa17ca786848e6bb48933e6773ccc2887b4 Mon Sep 17 00:00:00 2001 From: Peter Boers Date: Thu, 3 Sep 2026 09:43:04 +0200 Subject: [PATCH 01/10] fix(proxy): resolve a2a/ models before router branches can claim them `_is_a2a_agent_model()` was the last `elif` in the routing chain, so two earlier branches shadowed it and the request failed with "no healthy deployments" instead of reaching the agent: - `map_team_model()` claims the request for any team-scoped virtual key, so A2A agents were unreachable for every non-admin key. A key with no team works; the same key attached to a team does not. - the wildcard/default-deployment fallback claims it whenever a pattern model group is configured (fixes #37581). An `a2a/` prefix is unambiguous and is never backed by a router deployment, so resolve it before any router-based branch. The existing test mocked `map_team_model` to None and `pattern_router` to an empty pattern list, which is exactly why neither case was caught; the new parametrized test exercises both shadowing branches. Co-Authored-By: Claude Opus 5 (1M context) --- litellm/proxy/route_llm_request.py | 24 ++++-- .../proxy/test_route_a2a_models.py | 84 +++++++++++++++++++ 2 files changed, 99 insertions(+), 9 deletions(-) diff --git a/litellm/proxy/route_llm_request.py b/litellm/proxy/route_llm_request.py index 3d0bd5e61c9..48fdf29c209 100644 --- a/litellm/proxy/route_llm_request.py +++ b/litellm/proxy/route_llm_request.py @@ -536,6 +536,21 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr return getattr(litellm, f"{route_type}")(**data) elif llm_router is not None: _raise_if_model_fully_blocked(llm_router=llm_router, model_name=data.get("model"), team_id=team_id) + + # A2A agents are addressed by an unambiguous "a2a/" model prefix and are never backed + # by a router deployment, so they must be resolved before any router-based branch. + # Otherwise the branches below swallow the request and it fails with + # "no healthy deployments": `map_team_model` claims it for team-scoped keys, and the + # wildcard/default-deployment fallback claims it whenever a pattern model group exists. + if _is_a2a_agent_model(data.get("model", "")): + from litellm.proxy.agent_endpoints.a2a_routing import ( + route_a2a_agent_request, + ) + + a2a_result: Final = await route_a2a_agent_request(data, route_type, user_api_key_dict=user_api_key_dict) + if a2a_result is not None: + return a2a_result + # Evals API: always route to litellm directly (not through router) # But extract model credentials if a model is provided if route_type in [ @@ -696,15 +711,6 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr except Exception: # If router fails (e.g., model not found in router), fall back to direct call return getattr(litellm, f"{route_type}")(**data) - elif _is_a2a_agent_model(data.get("model", "")): - from litellm.proxy.agent_endpoints.a2a_routing import ( - route_a2a_agent_request, - ) - - result: Final = await route_a2a_agent_request(data, route_type, user_api_key_dict=user_api_key_dict) - if result is not None: - return result - # Fall through to raise exception below if result is None elif user_model is not None or route_type == "allm_passthrough_route": return getattr(litellm, f"{route_type}")(**data) diff --git a/tests/test_litellm/proxy/test_route_a2a_models.py b/tests/test_litellm/proxy/test_route_a2a_models.py index 35308474949..8e5954cf5af 100644 --- a/tests/test_litellm/proxy/test_route_a2a_models.py +++ b/tests/test_litellm/proxy/test_route_a2a_models.py @@ -180,3 +180,87 @@ async def test_route_a2a_model_read_through_recovers_agent_created_on_sibling_re assert call_kwargs["model"] == f"a2a/{agent_name}" assert call_kwargs["api_base"] == "http://sibling-db-agent.example.com" prisma_client.db.litellm_agentstable.find_unique.assert_awaited() + + +def _router_without_a2a_deployment( + *, + team_model: str | None = None, + patterns: tuple[str, ...] = (), + default_deployment: dict | None = None, +) -> Mock: + """A router that serves no A2A deployment, optionally tripping one shadowing branch.""" + router = Mock() + router.model_names = ["gpt-4", "gpt-3.5-turbo"] + router.deployment_names = [] + router.has_model_id = Mock(return_value=False) + router.is_recognized_model = Mock(return_value=False) + router.get_routing_group = Mock(return_value=None) + router.model_group_alias = None + router.router_general_settings = Mock(pass_through_all_models=False) + router.default_deployment = default_deployment + router.pattern_router = Mock(patterns=list(patterns)) + router.map_team_model = Mock(return_value=team_model) + return router + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "router_kwargs, extra_data", + [ + pytest.param({}, {}, id="no_shadowing_branch"), + pytest.param( + {"team_model": "a2a/test-agent"}, + {"metadata": {"user_api_key_team_id": "team-123"}}, + id="team_scoped_key", + ), + pytest.param({"patterns": ("openrouter/*",)}, {}, id="wildcard_model_group"), + pytest.param( + {"default_deployment": {"model_name": "*"}}, {}, id="default_deployment" + ), + ], +) +async def test_a2a_model_resolves_before_router_branches(router_kwargs, extra_data): + """ + Regression: an `a2a/` model must reach A2A routing even when a router branch would + otherwise claim it. + + `map_team_model` claims the request for any team-scoped key, and the + wildcard/default-deployment fallback claims it whenever a pattern model group exists. + Both previously shadowed the A2A branch, so the call failed with + "no healthy deployments" instead of reaching the agent. + """ + from litellm.types.agents import AgentResponse + + data = { + "model": "a2a/test-agent", + "messages": [{"role": "user", "content": "Hello"}], + **extra_data, + } + + mock_agent = AgentResponse( + agent_id="test-agent-id", + agent_name="test-agent", + agent_card_params={"url": "http://agent.example.com"}, + litellm_params=None, + ) + mock_registry = Mock() + mock_registry.get_agent_by_id = Mock(return_value=None) + mock_registry.get_agent_by_name = Mock(return_value=mock_agent) + + mock_acompletion = AsyncMock(return_value={"id": "test-response"}) + + with patch("litellm.acompletion", mock_acompletion), patch( + "litellm.proxy.agent_endpoints.agent_registry.global_agent_registry", + mock_registry, + ): + await route_request( + data=data, + llm_router=_router_without_a2a_deployment(**router_kwargs), + user_model=None, + route_type="acompletion", + ) + + mock_acompletion.assert_called_once() + call_kwargs = mock_acompletion.call_args.kwargs + assert call_kwargs["model"] == "a2a/test-agent" + assert call_kwargs["api_base"] == "http://agent.example.com" From f757c525703e1d3db5568781d4c94f46a3a80a0a Mon Sep 17 00:00:00 2001 From: Peter Boers Date: Thu, 3 Sep 2026 09:43:07 +0200 Subject: [PATCH 02/10] fix(a2a): stop echoing the prompt and repeating the reply when streaming `extract_text_from_a2a_response()` took text from every streaming event regardless of who authored it or whether it was a delta, so a reply of "OK" reached the client as "user: Reply with exactly: OKOKOKOK": - the opening `status-update` (`state: "submitted"`) carries the caller's own message, and A2A marks it `role: "user"`. It was emitted as assistant output. Now skipped: per spec `Message.role` is "user" or "agent", and only agent text belongs in a completion. - the terminal non-partial `status-update` and the `artifact-update` (`append: false`) each repeat the whole reply. Both were forwarded as deltas. The iterator now tracks what it has emitted and forwards only new text, which also lets servers that stream growing snapshots ("O", "OK") collapse to the same output as servers that stream true deltas ("O", "K"). Tests use an event sequence captured from a real kagent agent. Co-Authored-By: Claude Opus 5 (1M context) --- litellm/llms/a2a/chat/streaming_iterator.py | 37 +++++++- litellm/llms/a2a/common_utils.py | 17 ++++ .../chat/test_a2a_chat_streaming_iterator.py | 91 +++++++++++++++++++ 3 files changed, 143 insertions(+), 2 deletions(-) create mode 100644 tests/test_litellm/llms/a2a/chat/test_a2a_chat_streaming_iterator.py diff --git a/litellm/llms/a2a/chat/streaming_iterator.py b/litellm/llms/a2a/chat/streaming_iterator.py index 1983c18a6b3..7be27b2fa0f 100644 --- a/litellm/llms/a2a/chat/streaming_iterator.py +++ b/litellm/llms/a2a/chat/streaming_iterator.py @@ -30,6 +30,9 @@ class A2AModelResponseIterator(BaseModelResponseIterator): json_mode=json_mode, ) self.model = model + # Text already emitted downstream, used to collapse cumulative snapshots. + self._emitted_text: str = "" + self._delta_count: int = 0 def chunk_parser(self, chunk: dict) -> GenericStreamingChunk | ModelResponseStream: """ @@ -57,8 +60,8 @@ class A2AModelResponseIterator(BaseModelResponseIterator): } """ try: - # Extract text from A2A response - text: Final = extract_text_from_a2a_response(chunk) + # Extract text from A2A response, then reduce it to what is actually new. + text: Final = self._to_incremental_text(extract_text_from_a2a_response(chunk)) # Determine finish reason finish_reason: Final = self._get_finish_reason(chunk) @@ -83,6 +86,36 @@ class A2AModelResponseIterator(BaseModelResponseIterator): tool_use=None, ) + def _to_incremental_text(self, text: str) -> str: + """ + Reduce an A2A event's text to the portion not yet emitted. + + A2A servers interleave true deltas with cumulative snapshots of the whole reply: a + terminal non-partial ``status-update`` and an ``artifact-update`` carrying + ``append: false`` both repeat everything produced so far. Forwarding those verbatim + makes the client render the reply two or three times over, so emit only new text. + + Handles both streaming styles: servers that send deltas ("O", "K") and servers that + send growing snapshots ("O", "OK") collapse to the same output. + """ + if not text: + return "" + + emitted: str = self._emitted_text + if emitted and text.startswith(emitted): + suffix: str = text[len(emitted) :] + if suffix: + self._emitted_text = text + return suffix + # text == emitted. Treat as a snapshot repeat, except while only a single delta + # has been emitted, where a genuinely repeated delta is still indistinguishable. + if self._delta_count > 1: + return "" + + self._emitted_text = emitted + text + self._delta_count += 1 + return text + def _get_finish_reason(self, chunk: dict) -> str | None: """Extract finish reason from A2A chunk""" result: Final = chunk.get("result", {}) diff --git a/litellm/llms/a2a/common_utils.py b/litellm/llms/a2a/common_utils.py index 178b4c47a0f..d4a437027e2 100644 --- a/litellm/llms/a2a/common_utils.py +++ b/litellm/llms/a2a/common_utils.py @@ -61,6 +61,17 @@ def convert_messages_to_prompt(messages: list[AllMessageValues]) -> str: return "\n".join(conversation_parts) +def _is_user_authored(message: object) -> bool: + """ + Whether an A2A message was authored by the caller rather than the agent. + + A2A ``Message.role`` is either ``"user"`` or ``"agent"``. Agents commonly echo the + inbound message back on the first ``status-update`` (``state: "submitted"``), and that + text must never surface as assistant output in a chat completion. + """ + return isinstance(message, dict) and message.get("role") == "user" + + def extract_text_from_a2a_message(message: dict[str, Any], depth: int = 0, max_depth: int = 10) -> str: """ Extract text content from A2A message parts. @@ -115,11 +126,15 @@ def extract_text_from_a2a_response(response_dict: dict[str, Any], max_depth: int # Check if result itself has parts (direct message) if "parts" in result: + if _is_user_authored(result): + return "" return extract_text_from_a2a_message(result, depth=0, max_depth=max_depth) # Check for nested message message: Final = result.get("message") if message: + if _is_user_authored(message): + return "" return extract_text_from_a2a_message(message, depth=0, max_depth=max_depth) # Check for streaming artifact-update (singular artifact) @@ -132,6 +147,8 @@ def extract_text_from_a2a_response(response_dict: dict[str, Any], max_depth: int if isinstance(status, dict): status_message: Final = status.get("message") if status_message: + if _is_user_authored(status_message): + return "" return extract_text_from_a2a_message(status_message, depth=0, max_depth=max_depth) # Handle task result with artifacts (plural, array) diff --git a/tests/test_litellm/llms/a2a/chat/test_a2a_chat_streaming_iterator.py b/tests/test_litellm/llms/a2a/chat/test_a2a_chat_streaming_iterator.py new file mode 100644 index 00000000000..c49f989d76d --- /dev/null +++ b/tests/test_litellm/llms/a2a/chat/test_a2a_chat_streaming_iterator.py @@ -0,0 +1,91 @@ +"""Tests for litellm/llms/a2a/chat/streaming_iterator.py delta handling.""" + +import pytest + +from litellm.llms.a2a.chat.streaming_iterator import A2AModelResponseIterator + + +def _iterator() -> A2AModelResponseIterator: + return A2AModelResponseIterator(streaming_response=iter(()), sync_stream=True) + + +def _status_update( + *, + text: str | None = None, + role: str = "agent", + state: str = "working", + final: bool = False, +) -> dict: + status: dict = {"state": state} + if text is not None: + status["message"] = { + "kind": "message", + "role": role, + "parts": [{"kind": "text", "text": text}], + } + return { + "jsonrpc": "2.0", + "id": "1", + "result": {"kind": "status-update", "final": final, "status": status}, + } + + +def _artifact_update(*, text: str, append: bool = False) -> dict: + return { + "jsonrpc": "2.0", + "id": "1", + "result": { + "kind": "artifact-update", + "append": append, + "lastChunk": True, + "artifact": {"parts": [{"kind": "text", "text": text}]}, + }, + } + + +# Event sequence captured from a real kagent A2A agent replying "OK": the submitted +# status-update echoes the caller's own message, then two true deltas are followed by two +# cumulative snapshots of the whole reply. +KAGENT_OK_STREAM = [ + _status_update(text="Reply with exactly: OK", role="user", state="submitted"), + _status_update(), + _status_update(text="O"), + _status_update(text="K"), + _status_update(text="OK"), + _artifact_update(text="OK"), + _status_update(state="completed", final=True), +] + + +def test_kagent_stream_yields_reply_exactly_once(): + """ + Regression: the caller's echoed message must not be emitted as assistant output, and + cumulative snapshots must not repeat the reply. + + Previously this stream rendered as "user: Reply with exactly: OKOKOKOK". + """ + iterator = _iterator() + assert "".join(iterator.chunk_parser(e)["text"] for e in KAGENT_OK_STREAM) == "OK" + + +def test_kagent_stream_finishes_on_completed_state(): + iterator = _iterator() + chunks = [iterator.chunk_parser(e) for e in KAGENT_OK_STREAM] + assert [c["finish_reason"] for c in chunks if c["is_finished"]] == ["stop"] + + +@pytest.mark.parametrize( + "texts, expected", + [ + pytest.param(["Hello", " world"], "Hello world", id="incremental_deltas"), + pytest.param(["O", "OK", "OKAY"], "OKAY", id="cumulative_snapshots"), + pytest.param(["O", "K", "OK"], "OK", id="deltas_then_final_snapshot"), + pytest.param(["O", "K", "OK", "OK"], "OK", id="deltas_then_repeated_snapshots"), + pytest.param(["a", "a", "a"], "aaa", id="genuinely_repeated_deltas"), + pytest.param(["", "OK", ""], "OK", id="empty_events_ignored"), + ], +) +def test_incremental_text_reduction(texts, expected): + """Delta-style and snapshot-style servers must collapse to the same output.""" + iterator = _iterator() + assert "".join(iterator._to_incremental_text(t) for t in texts) == expected From a1dbd1f28caaf4cfa5c7b0f67fc2ff0d46d6753f Mon Sep 17 00:00:00 2001 From: Peter Boers Date: Thu, 3 Sep 2026 10:52:48 +0200 Subject: [PATCH 03/10] test(a2a): record why the routing test patches SDK internals (TQ008) The test-quality gate flags `patch()` on `litellm.` internals, preferring a faked HTTP boundary or an injected collaborator. Neither applies here: the regression is that the request never leaves the router, so which collaborator is called is precisely the assertion, and the agent registry is Prisma-backed with no unit-test injection seam. Both patches now carry a reason, matching the pattern the sibling tests in this file already use. Co-Authored-By: Claude Opus 5 (1M context) --- .../proxy/test_route_a2a_models.py | 24 +++++++++++-------- 1 file changed, 14 insertions(+), 10 deletions(-) diff --git a/tests/test_litellm/proxy/test_route_a2a_models.py b/tests/test_litellm/proxy/test_route_a2a_models.py index 8e5954cf5af..6dc69e566d8 100644 --- a/tests/test_litellm/proxy/test_route_a2a_models.py +++ b/tests/test_litellm/proxy/test_route_a2a_models.py @@ -249,16 +249,20 @@ async def test_a2a_model_resolves_before_router_branches(router_kwargs, extra_da mock_acompletion = AsyncMock(return_value={"id": "test-response"}) - with patch("litellm.acompletion", mock_acompletion), patch( - "litellm.proxy.agent_endpoints.agent_registry.global_agent_registry", - mock_registry, - ): - await route_request( - data=data, - llm_router=_router_without_a2a_deployment(**router_kwargs), - user_model=None, - route_type="acompletion", - ) + # The bug is that the request never leaves the router, so there is no HTTP boundary to + # fake: which collaborator gets called *is* the behaviour under test. The registry is + # Prisma-backed and has no injection seam available to a unit test. + with patch("litellm.acompletion", mock_acompletion): # test-quality-ok: the dispatch target is the assertion + with patch( # test-quality-ok: Prisma-backed registry has no unit-test injection seam + "litellm.proxy.agent_endpoints.agent_registry.global_agent_registry", + mock_registry, + ): + await route_request( + data=data, + llm_router=_router_without_a2a_deployment(**router_kwargs), + user_model=None, + route_type="acompletion", + ) mock_acompletion.assert_called_once() call_kwargs = mock_acompletion.call_args.kwargs From 7ce56c0f80131c0fe93b6e2884a55455946b425d Mon Sep 17 00:00:00 2001 From: Peter Boers Date: Thu, 3 Sep 2026 10:58:53 +0200 Subject: [PATCH 04/10] fix(a2a): detect cumulative snapshots across part-join whitespace Snapshot detection compared raw strings, but A2A joins a multi-part message's text parts with spaces. A server that streams deltas ("Hello", "world") and then sends the whole reply as one two-part artifact produces "Helloworld" accumulated against a "Hello world" snapshot, so the prefix test missed and the reply was emitted twice: "HelloworldHello world". Compare with whitespace removed, and map the match back to a raw offset so the emitted tail keeps the server's own spacing and newlines. Reported by Greptile on #39513. Co-Authored-By: Claude Opus 5 (1M context) --- litellm/llms/a2a/chat/streaming_iterator.py | 43 +++++++++++++++---- .../chat/test_a2a_chat_streaming_iterator.py | 30 +++++++++++-- .../proxy/test_route_a2a_models.py | 6 +-- 3 files changed, 63 insertions(+), 16 deletions(-) diff --git a/litellm/llms/a2a/chat/streaming_iterator.py b/litellm/llms/a2a/chat/streaming_iterator.py index 7be27b2fa0f..f2b073efd73 100644 --- a/litellm/llms/a2a/chat/streaming_iterator.py +++ b/litellm/llms/a2a/chat/streaming_iterator.py @@ -2,6 +2,7 @@ A2A Streaming Response Iterator """ +from itertools import accumulate from typing import Final from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator @@ -10,6 +11,29 @@ from litellm.types.utils import GenericStreamingChunk, ModelResponseStream from ..common_utils import extract_text_from_a2a_response +def _ignoring_whitespace(text: str) -> str: + """ + Comparison key for snapshot detection. + + A2A joins a multi-part message's text parts with spaces, so the same content arrives + with different whitespace depending on how the server chunked it: two delta events + ("Hello", "world") accumulate to "Helloworld" while a single two-part snapshot of the + same content renders "Hello world". Comparing without whitespace makes them equal. + """ + return "".join(text.split()) + + +def _index_after(text: str, non_space_count: int) -> int: + """Index in `text` just past its first `non_space_count` non-whitespace characters.""" + if non_space_count <= 0: + return 0 + running: Final = accumulate(0 if char.isspace() else 1 for char in text) + return next( + (index + 1 for index, total in enumerate(running) if total >= non_space_count), + len(text), + ) + + class A2AModelResponseIterator(BaseModelResponseIterator): """ Iterator for parsing A2A streaming responses. @@ -101,18 +125,21 @@ class A2AModelResponseIterator(BaseModelResponseIterator): if not text: return "" - emitted: str = self._emitted_text - if emitted and text.startswith(emitted): - suffix: str = text[len(emitted) :] - if suffix: - self._emitted_text = text + emitted_key: Final = _ignoring_whitespace(self._emitted_text) + text_key: Final = _ignoring_whitespace(text) + + if emitted_key and text_key.startswith(emitted_key): + suffix: Final = text[_index_after(text, len(emitted_key)) :] + if suffix.strip(): + self._emitted_text += suffix return suffix - # text == emitted. Treat as a snapshot repeat, except while only a single delta - # has been emitted, where a genuinely repeated delta is still indistinguishable. + # Same content as everything emitted so far: a snapshot repeat, except while + # only a single delta has been emitted, where a genuinely repeated delta is + # still indistinguishable from one. if self._delta_count > 1: return "" - self._emitted_text = emitted + text + self._emitted_text += text self._delta_count += 1 return text diff --git a/tests/test_litellm/llms/a2a/chat/test_a2a_chat_streaming_iterator.py b/tests/test_litellm/llms/a2a/chat/test_a2a_chat_streaming_iterator.py index c49f989d76d..4becf808cd9 100644 --- a/tests/test_litellm/llms/a2a/chat/test_a2a_chat_streaming_iterator.py +++ b/tests/test_litellm/llms/a2a/chat/test_a2a_chat_streaming_iterator.py @@ -30,7 +30,7 @@ def _status_update( } -def _artifact_update(*, text: str, append: bool = False) -> dict: +def _artifact_update(*texts: str, append: bool = False) -> dict: return { "jsonrpc": "2.0", "id": "1", @@ -38,7 +38,7 @@ def _artifact_update(*, text: str, append: bool = False) -> dict: "kind": "artifact-update", "append": append, "lastChunk": True, - "artifact": {"parts": [{"kind": "text", "text": text}]}, + "artifact": {"parts": [{"kind": "text", "text": text} for text in texts]}, }, } @@ -52,7 +52,7 @@ KAGENT_OK_STREAM = [ _status_update(text="O"), _status_update(text="K"), _status_update(text="OK"), - _artifact_update(text="OK"), + _artifact_update("OK"), _status_update(state="completed", final=True), ] @@ -82,6 +82,12 @@ def test_kagent_stream_finishes_on_completed_state(): pytest.param(["O", "K", "OK"], "OK", id="deltas_then_final_snapshot"), pytest.param(["O", "K", "OK", "OK"], "OK", id="deltas_then_repeated_snapshots"), pytest.param(["a", "a", "a"], "aaa", id="genuinely_repeated_deltas"), + pytest.param(["Hello", "world", "Hello world"], "Helloworld", id="multipart_snapshot_respaced"), + pytest.param( + ["Hello", "world", "Hello world again"], + "Helloworld again", + id="multipart_snapshot_extends", + ), pytest.param(["", "OK", ""], "OK", id="empty_events_ignored"), ], ) @@ -89,3 +95,21 @@ def test_incremental_text_reduction(texts, expected): """Delta-style and snapshot-style servers must collapse to the same output.""" iterator = _iterator() assert "".join(iterator._to_incremental_text(t) for t in texts) == expected + + +# A server that chunks its reply into separate delta events but sends the final artifact +# as one multi-part message: A2A joins those parts with a space, so the snapshot reads +# "Hello world" while the deltas accumulated to "Helloworld". +MULTIPART_SNAPSHOT_STREAM = [ + _status_update(text="Hello"), + _status_update(text="world"), + _artifact_update("Hello", "world"), + _status_update(state="completed", final=True), +] + + +def test_multipart_snapshot_is_not_re_emitted(): + """Regression: whitespace introduced by part-joining must not defeat snapshot detection.""" + iterator = _iterator() + rendered = "".join(iterator.chunk_parser(e)["text"] for e in MULTIPART_SNAPSHOT_STREAM) + assert rendered == "Helloworld" diff --git a/tests/test_litellm/proxy/test_route_a2a_models.py b/tests/test_litellm/proxy/test_route_a2a_models.py index 6dc69e566d8..d1e85d56817 100644 --- a/tests/test_litellm/proxy/test_route_a2a_models.py +++ b/tests/test_litellm/proxy/test_route_a2a_models.py @@ -4,8 +4,6 @@ Test A2A model routing in proxy. Maps to: litellm/proxy/agent_endpoints/a2a_routing.py """ - - from unittest.mock import AsyncMock, Mock, patch import pytest @@ -214,9 +212,7 @@ def _router_without_a2a_deployment( id="team_scoped_key", ), pytest.param({"patterns": ("openrouter/*",)}, {}, id="wildcard_model_group"), - pytest.param( - {"default_deployment": {"model_name": "*"}}, {}, id="default_deployment" - ), + pytest.param({"default_deployment": {"model_name": "*"}}, {}, id="default_deployment"), ], ) async def test_a2a_model_resolves_before_router_branches(router_kwargs, extra_data): From edb455ee782c1b4177ac841e62658bb7ddd1a886 Mon Sep 17 00:00:00 2001 From: Peter Boers Date: Thu, 3 Sep 2026 11:02:37 +0200 Subject: [PATCH 05/10] test(a2a): fold streaming coverage into the mapped a2a chat test file CLAUDE.md asks bug fixes to extend the existing mapped test file for the directory rather than add a new module. Moves the streaming-iterator regression tests into tests/test_litellm/llms/a2a/chat/test_a2a_chat_transformation.py so the a2a chat bridge keeps one owner, and widens that file's docstring to match. Reported by Greptile on #39513. Co-Authored-By: Claude Opus 5 (1M context) --- .../chat/test_a2a_chat_streaming_iterator.py | 115 ------------------ .../a2a/chat/test_a2a_chat_transformation.py | 115 +++++++++++++++++- 2 files changed, 114 insertions(+), 116 deletions(-) delete mode 100644 tests/test_litellm/llms/a2a/chat/test_a2a_chat_streaming_iterator.py diff --git a/tests/test_litellm/llms/a2a/chat/test_a2a_chat_streaming_iterator.py b/tests/test_litellm/llms/a2a/chat/test_a2a_chat_streaming_iterator.py deleted file mode 100644 index 4becf808cd9..00000000000 --- a/tests/test_litellm/llms/a2a/chat/test_a2a_chat_streaming_iterator.py +++ /dev/null @@ -1,115 +0,0 @@ -"""Tests for litellm/llms/a2a/chat/streaming_iterator.py delta handling.""" - -import pytest - -from litellm.llms.a2a.chat.streaming_iterator import A2AModelResponseIterator - - -def _iterator() -> A2AModelResponseIterator: - return A2AModelResponseIterator(streaming_response=iter(()), sync_stream=True) - - -def _status_update( - *, - text: str | None = None, - role: str = "agent", - state: str = "working", - final: bool = False, -) -> dict: - status: dict = {"state": state} - if text is not None: - status["message"] = { - "kind": "message", - "role": role, - "parts": [{"kind": "text", "text": text}], - } - return { - "jsonrpc": "2.0", - "id": "1", - "result": {"kind": "status-update", "final": final, "status": status}, - } - - -def _artifact_update(*texts: str, append: bool = False) -> dict: - return { - "jsonrpc": "2.0", - "id": "1", - "result": { - "kind": "artifact-update", - "append": append, - "lastChunk": True, - "artifact": {"parts": [{"kind": "text", "text": text} for text in texts]}, - }, - } - - -# Event sequence captured from a real kagent A2A agent replying "OK": the submitted -# status-update echoes the caller's own message, then two true deltas are followed by two -# cumulative snapshots of the whole reply. -KAGENT_OK_STREAM = [ - _status_update(text="Reply with exactly: OK", role="user", state="submitted"), - _status_update(), - _status_update(text="O"), - _status_update(text="K"), - _status_update(text="OK"), - _artifact_update("OK"), - _status_update(state="completed", final=True), -] - - -def test_kagent_stream_yields_reply_exactly_once(): - """ - Regression: the caller's echoed message must not be emitted as assistant output, and - cumulative snapshots must not repeat the reply. - - Previously this stream rendered as "user: Reply with exactly: OKOKOKOK". - """ - iterator = _iterator() - assert "".join(iterator.chunk_parser(e)["text"] for e in KAGENT_OK_STREAM) == "OK" - - -def test_kagent_stream_finishes_on_completed_state(): - iterator = _iterator() - chunks = [iterator.chunk_parser(e) for e in KAGENT_OK_STREAM] - assert [c["finish_reason"] for c in chunks if c["is_finished"]] == ["stop"] - - -@pytest.mark.parametrize( - "texts, expected", - [ - pytest.param(["Hello", " world"], "Hello world", id="incremental_deltas"), - pytest.param(["O", "OK", "OKAY"], "OKAY", id="cumulative_snapshots"), - pytest.param(["O", "K", "OK"], "OK", id="deltas_then_final_snapshot"), - pytest.param(["O", "K", "OK", "OK"], "OK", id="deltas_then_repeated_snapshots"), - pytest.param(["a", "a", "a"], "aaa", id="genuinely_repeated_deltas"), - pytest.param(["Hello", "world", "Hello world"], "Helloworld", id="multipart_snapshot_respaced"), - pytest.param( - ["Hello", "world", "Hello world again"], - "Helloworld again", - id="multipart_snapshot_extends", - ), - pytest.param(["", "OK", ""], "OK", id="empty_events_ignored"), - ], -) -def test_incremental_text_reduction(texts, expected): - """Delta-style and snapshot-style servers must collapse to the same output.""" - iterator = _iterator() - assert "".join(iterator._to_incremental_text(t) for t in texts) == expected - - -# A server that chunks its reply into separate delta events but sends the final artifact -# as one multi-part message: A2A joins those parts with a space, so the snapshot reads -# "Hello world" while the deltas accumulated to "Helloworld". -MULTIPART_SNAPSHOT_STREAM = [ - _status_update(text="Hello"), - _status_update(text="world"), - _artifact_update("Hello", "world"), - _status_update(state="completed", final=True), -] - - -def test_multipart_snapshot_is_not_re_emitted(): - """Regression: whitespace introduced by part-joining must not defeat snapshot detection.""" - iterator = _iterator() - rendered = "".join(iterator.chunk_parser(e)["text"] for e in MULTIPART_SNAPSHOT_STREAM) - assert rendered == "Helloworld" 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 2e11c68244c..e6d31565b29 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 @@ -1,7 +1,10 @@ -"""Tests for litellm/llms/a2a/chat/transformation.py response transform.""" +"""Tests for the A2A chat bridge: litellm/llms/a2a/chat/ response transform and streaming.""" from unittest.mock import MagicMock +import pytest + +from litellm.llms.a2a.chat.streaming_iterator import A2AModelResponseIterator from litellm.llms.a2a.chat.transformation import A2AConfig from litellm.types.utils import ModelResponse @@ -40,3 +43,113 @@ def test_transform_response_sets_usage(): assert result.usage.prompt_tokens > 0 assert result.usage.completion_tokens > 0 assert result.usage.total_tokens == (result.usage.prompt_tokens + result.usage.completion_tokens) + + +def _iterator() -> A2AModelResponseIterator: + return A2AModelResponseIterator(streaming_response=iter(()), sync_stream=True) + + +def _status_update( + *, + text: str | None = None, + role: str = "agent", + state: str = "working", + final: bool = False, +) -> dict: + status: dict = {"state": state} + if text is not None: + status["message"] = { + "kind": "message", + "role": role, + "parts": [{"kind": "text", "text": text}], + } + return { + "jsonrpc": "2.0", + "id": "1", + "result": {"kind": "status-update", "final": final, "status": status}, + } + + +def _artifact_update(*texts: str, append: bool = False) -> dict: + return { + "jsonrpc": "2.0", + "id": "1", + "result": { + "kind": "artifact-update", + "append": append, + "lastChunk": True, + "artifact": {"parts": [{"kind": "text", "text": text} for text in texts]}, + }, + } + + +# Event sequence captured from a real kagent A2A agent replying "OK": the submitted +# status-update echoes the caller's own message, then two true deltas are followed by two +# cumulative snapshots of the whole reply. +KAGENT_OK_STREAM = [ + _status_update(text="Reply with exactly: OK", role="user", state="submitted"), + _status_update(), + _status_update(text="O"), + _status_update(text="K"), + _status_update(text="OK"), + _artifact_update("OK"), + _status_update(state="completed", final=True), +] + + +def test_kagent_stream_yields_reply_exactly_once(): + """ + Regression: the caller's echoed message must not be emitted as assistant output, and + cumulative snapshots must not repeat the reply. + + Previously this stream rendered as "user: Reply with exactly: OKOKOKOK". + """ + iterator = _iterator() + assert "".join(iterator.chunk_parser(e)["text"] for e in KAGENT_OK_STREAM) == "OK" + + +def test_kagent_stream_finishes_on_completed_state(): + iterator = _iterator() + chunks = [iterator.chunk_parser(e) for e in KAGENT_OK_STREAM] + assert [c["finish_reason"] for c in chunks if c["is_finished"]] == ["stop"] + + +@pytest.mark.parametrize( + "texts, expected", + [ + pytest.param(["Hello", " world"], "Hello world", id="incremental_deltas"), + pytest.param(["O", "OK", "OKAY"], "OKAY", id="cumulative_snapshots"), + pytest.param(["O", "K", "OK"], "OK", id="deltas_then_final_snapshot"), + pytest.param(["O", "K", "OK", "OK"], "OK", id="deltas_then_repeated_snapshots"), + pytest.param(["a", "a", "a"], "aaa", id="genuinely_repeated_deltas"), + pytest.param(["Hello", "world", "Hello world"], "Helloworld", id="multipart_snapshot_respaced"), + pytest.param( + ["Hello", "world", "Hello world again"], + "Helloworld again", + id="multipart_snapshot_extends", + ), + pytest.param(["", "OK", ""], "OK", id="empty_events_ignored"), + ], +) +def test_incremental_text_reduction(texts, expected): + """Delta-style and snapshot-style servers must collapse to the same output.""" + iterator = _iterator() + assert "".join(iterator._to_incremental_text(t) for t in texts) == expected + + +# A server that chunks its reply into separate delta events but sends the final artifact +# as one multi-part message: A2A joins those parts with a space, so the snapshot reads +# "Hello world" while the deltas accumulated to "Helloworld". +MULTIPART_SNAPSHOT_STREAM = [ + _status_update(text="Hello"), + _status_update(text="world"), + _artifact_update("Hello", "world"), + _status_update(state="completed", final=True), +] + + +def test_multipart_snapshot_is_not_re_emitted(): + """Regression: whitespace introduced by part-joining must not defeat snapshot detection.""" + iterator = _iterator() + rendered = "".join(iterator.chunk_parser(e)["text"] for e in MULTIPART_SNAPSHOT_STREAM) + assert rendered == "Helloworld" From 13a313987a9c53a045889bdaef868eb3d4893aa8 Mon Sep 17 00:00:00 2001 From: Peter Boers Date: Thu, 3 Sep 2026 11:33:29 +0200 Subject: [PATCH 06/10] fix(a2a): suppress a repeated snapshot after a single delta too Snapshot suppression was gated on more than one delta having been emitted, to avoid collapsing a delta that genuinely repeats the accumulated text. That left the ordinary shape of a short reply broken: one delta "OK" followed by the terminal cumulative snapshot "OK" rendered as "OKOK". A2A marks no event as delta-or-snapshot, so an event whose text equals everything emitted so far is inherently ambiguous. Read it as a snapshot: a server repeating the whole reply at the end of a stream is common, a delta that exactly reproduces the accumulated text is not, and duplicating a reply is far worse for a reader than dropping one repeated fragment. The tradeoff is pinned by the `identical_deltas_collapse` case rather than left implicit. Reported by Greptile on #39513. Co-Authored-By: Claude Opus 5 (1M context) --- litellm/llms/a2a/chat/streaming_iterator.py | 20 ++++++++--------- .../a2a/chat/test_a2a_chat_transformation.py | 22 ++++++++++++++++++- 2 files changed, 31 insertions(+), 11 deletions(-) diff --git a/litellm/llms/a2a/chat/streaming_iterator.py b/litellm/llms/a2a/chat/streaming_iterator.py index f2b073efd73..4c352ca2b6e 100644 --- a/litellm/llms/a2a/chat/streaming_iterator.py +++ b/litellm/llms/a2a/chat/streaming_iterator.py @@ -56,7 +56,6 @@ class A2AModelResponseIterator(BaseModelResponseIterator): self.model = model # Text already emitted downstream, used to collapse cumulative snapshots. self._emitted_text: str = "" - self._delta_count: int = 0 def chunk_parser(self, chunk: dict) -> GenericStreamingChunk | ModelResponseStream: """ @@ -121,6 +120,12 @@ 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. + + 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 + exactly reproduces the accumulated text is not, and duplicating a reply is far + worse for a reader than dropping one repeated fragment. """ if not text: return "" @@ -129,18 +134,13 @@ class A2AModelResponseIterator(BaseModelResponseIterator): text_key: Final = _ignoring_whitespace(text) if emitted_key and text_key.startswith(emitted_key): + # A cumulative snapshot: emit only its tail, which is empty when the snapshot + # just repeats everything sent so far. suffix: Final = text[_index_after(text, len(emitted_key)) :] - if suffix.strip(): - self._emitted_text += suffix - return suffix - # Same content as everything emitted so far: a snapshot repeat, except while - # only a single delta has been emitted, where a genuinely repeated delta is - # still indistinguishable from one. - if self._delta_count > 1: - return "" + self._emitted_text += suffix + return suffix self._emitted_text += text - self._delta_count += 1 return text def _get_finish_reason(self, chunk: dict) -> str | None: 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 e6d31565b29..a17c4bdb440 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 @@ -121,7 +121,11 @@ def test_kagent_stream_finishes_on_completed_state(): pytest.param(["O", "OK", "OKAY"], "OKAY", id="cumulative_snapshots"), pytest.param(["O", "K", "OK"], "OK", id="deltas_then_final_snapshot"), pytest.param(["O", "K", "OK", "OK"], "OK", id="deltas_then_repeated_snapshots"), - pytest.param(["a", "a", "a"], "aaa", id="genuinely_repeated_deltas"), + pytest.param(["OK", "OK"], "OK", id="one_delta_then_equal_snapshot"), + # Known limitation: A2A marks no event as delta-or-snapshot, so a delta that + # exactly reproduces the accumulated text is indistinguishable from a snapshot + # and collapses. Duplicating a whole reply is the worse failure of the two. + pytest.param(["a", "a", "a"], "a", id="identical_deltas_collapse"), pytest.param(["Hello", "world", "Hello world"], "Helloworld", id="multipart_snapshot_respaced"), pytest.param( ["Hello", "world", "Hello world again"], @@ -153,3 +157,19 @@ def test_multipart_snapshot_is_not_re_emitted(): iterator = _iterator() rendered = "".join(iterator.chunk_parser(e)["text"] for e in MULTIPART_SNAPSHOT_STREAM) assert rendered == "Helloworld" + + +# A one-token reply: a single delta followed by the terminal cumulative snapshot. This is +# the ordinary shape of a short A2A answer, so the snapshot must not be forwarded again. +SINGLE_DELTA_STREAM = [ + _status_update(text="Reply with exactly: OK", role="user", state="submitted"), + _status_update(text="OK"), + _artifact_update("OK"), + _status_update(state="completed", final=True), +] + + +def test_single_delta_then_snapshot_is_not_duplicated(): + """Regression: snapshot suppression must not depend on how many deltas preceded it.""" + iterator = _iterator() + assert "".join(iterator.chunk_parser(e)["text"] for e in SINGLE_DELTA_STREAM) == "OK" From cc7107c782b26135e5d0b850c6129e83fb18dfe6 Mon Sep 17 00:00:00 2001 From: Peter Boers Date: Thu, 3 Sep 2026 11:42:40 +0200 Subject: [PATCH 07/10] fix(a2a): stop a snapshot resending whitespace already delivered The raw-offset mapping lands just past the last matched non-whitespace character, so whitespace already emitted at the end of the accumulated text was forwarded a second time by the next cumulative snapshot: ["Hello ", "Hello world"] -> "Hello world" ["OK\n", "OK\n"] -> "OK\n\n" Drop only the overlap between the whitespace already sent and the whitespace the snapshot repeats, rather than stripping the tail outright, so a snapshot that introduces further whitespace keeps it: ["Hello ", "Hello \n\nworld"] -> "Hello \n\nworld" Reported by Greptile on #39513. Co-Authored-By: Claude Opus 5 (1M context) --- litellm/llms/a2a/chat/streaming_iterator.py | 20 +++++++++++++++++-- .../a2a/chat/test_a2a_chat_transformation.py | 7 +++++++ 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/litellm/llms/a2a/chat/streaming_iterator.py b/litellm/llms/a2a/chat/streaming_iterator.py index 4c352ca2b6e..1d00889679c 100644 --- a/litellm/llms/a2a/chat/streaming_iterator.py +++ b/litellm/llms/a2a/chat/streaming_iterator.py @@ -34,6 +34,16 @@ def _index_after(text: str, non_space_count: int) -> int: ) +def _trailing_whitespace(text: str) -> int: + """How many whitespace characters `text` ends with.""" + return len(text) - len(text.rstrip()) + + +def _leading_whitespace(text: str) -> int: + """How many whitespace characters `text` begins with.""" + return len(text) - len(text.lstrip()) + + class A2AModelResponseIterator(BaseModelResponseIterator): """ Iterator for parsing A2A streaming responses. @@ -135,8 +145,14 @@ class A2AModelResponseIterator(BaseModelResponseIterator): if emitted_key and text_key.startswith(emitted_key): # A cumulative snapshot: emit only its tail, which is empty when the snapshot - # just repeats everything sent so far. - suffix: Final = text[_index_after(text, len(emitted_key)) :] + # just repeats everything sent so far. The offset lands just past the last + # matched non-whitespace character, so any whitespace already delivered at the + # end of the emitted text would otherwise be sent a second time. Drop only that + # overlap, so a snapshot introducing further whitespace (a paragraph break, say) + # keeps it. + tail: Final = text[_index_after(text, len(emitted_key)) :] + already_sent: Final = min(_trailing_whitespace(self._emitted_text), _leading_whitespace(tail)) + suffix: Final = tail[already_sent:] self._emitted_text += suffix return suffix 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 a17c4bdb440..36be80db4fa 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 @@ -122,6 +122,13 @@ def test_kagent_stream_finishes_on_completed_state(): pytest.param(["O", "K", "OK"], "OK", id="deltas_then_final_snapshot"), pytest.param(["O", "K", "OK", "OK"], "OK", id="deltas_then_repeated_snapshots"), pytest.param(["OK", "OK"], "OK", id="one_delta_then_equal_snapshot"), + pytest.param(["Hello ", "Hello world"], "Hello world", id="snapshot_keeps_emitted_space_once"), + pytest.param(["OK\n", "OK\n"], "OK\n", id="snapshot_keeps_emitted_newline_once"), + pytest.param( + ["Hello ", "Hello \n\nworld"], + "Hello \n\nworld", + id="snapshot_keeps_its_own_new_whitespace", + ), # Known limitation: A2A marks no event as delta-or-snapshot, so a delta that # exactly reproduces the accumulated text is indistinguishable from a snapshot # and collapses. Duplicating a whole reply is the worse failure of the two. From 85952f9b3d734d445059daae3e5154fba524100d Mon Sep 17 00:00:00 2001 From: Peter Boers Date: Sat, 5 Sep 2026 10:43:49 +0200 Subject: [PATCH 08/10] fix(a2a): don't let a caller-authored message hide the agent's reply Skipping the echoed caller message returned "" instead of continuing to the next extraction branch, so a task that echoes the prompt in status.message while carrying its answer in artifacts came back empty: {"status": {"message": {"role": "user", ...}}, <- echo, skipped "artifacts": [{"parts": [{"text": "..."}]}]} <- never reached Before the role filter this returned the echoed prompt (wrong, but visible); after it the reply was dropped entirely. Skip caller-authored messages and fall through, so extraction still reaches the agent's own output. Co-Authored-By: Claude Opus 5 (1M context) --- litellm/llms/a2a/common_utils.py | 16 +++---- .../a2a/chat/test_a2a_chat_transformation.py | 48 +++++++++++++++++++ 2 files changed, 54 insertions(+), 10 deletions(-) diff --git a/litellm/llms/a2a/common_utils.py b/litellm/llms/a2a/common_utils.py index d4a437027e2..6d98350ddf8 100644 --- a/litellm/llms/a2a/common_utils.py +++ b/litellm/llms/a2a/common_utils.py @@ -124,17 +124,15 @@ def extract_text_from_a2a_response(response_dict: dict[str, Any], max_depth: int # 4. Task with status message: {"result": {"kind": "task", "status": {"message": {"parts": [...]}}}} # 5. Streaming artifact-update: {"result": {"kind": "artifact-update", "artifact": {"parts": [...]}}} - # Check if result itself has parts (direct message) - if "parts" in result: - if _is_user_authored(result): - return "" + # Check if result itself has parts (direct message). A caller-authored message is + # skipped rather than returned, so extraction falls through to the agent's own + # output further down (artifacts, typically) instead of yielding nothing. + if "parts" in result and not _is_user_authored(result): return extract_text_from_a2a_message(result, depth=0, max_depth=max_depth) # Check for nested message message: Final = result.get("message") - if message: - if _is_user_authored(message): - return "" + if message and not _is_user_authored(message): return extract_text_from_a2a_message(message, depth=0, max_depth=max_depth) # Check for streaming artifact-update (singular artifact) @@ -146,9 +144,7 @@ def extract_text_from_a2a_response(response_dict: dict[str, Any], max_depth: int status: Final = result.get("status", {}) if isinstance(status, dict): status_message: Final = status.get("message") - if status_message: - if _is_user_authored(status_message): - return "" + if status_message and not _is_user_authored(status_message): return extract_text_from_a2a_message(status_message, depth=0, max_depth=max_depth) # Handle task result with artifacts (plural, array) 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 36be80db4fa..636e8ba23c3 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 @@ -5,6 +5,7 @@ from unittest.mock import MagicMock import pytest from litellm.llms.a2a.chat.streaming_iterator import A2AModelResponseIterator +from litellm.llms.a2a.common_utils import extract_text_from_a2a_response from litellm.llms.a2a.chat.transformation import A2AConfig from litellm.types.utils import ModelResponse @@ -180,3 +181,50 @@ def test_single_delta_then_snapshot_is_not_duplicated(): """Regression: snapshot suppression must not depend on how many deltas preceded it.""" iterator = _iterator() assert "".join(iterator.chunk_parser(e)["text"] for e in SINGLE_DELTA_STREAM) == "OK" + + +def _task(*, status_role: str | None = None, status_text: str = "", artifact_text: str | None = None) -> dict: + result: dict = {"kind": "task"} + if status_role is not None: + result["status"] = { + "state": "completed", + "message": {"role": status_role, "parts": [{"kind": "text", "text": status_text}]}, + } + if artifact_text is not None: + result["artifacts"] = [{"parts": [{"kind": "text", "text": artifact_text}]}] + return {"jsonrpc": "2.0", "id": "1", "result": result} + + +@pytest.mark.parametrize( + "response, expected", + [ + # Regression: a task echoing the caller in status.message while carrying the real + # answer in artifacts must not come back empty. Skipping a caller-authored message + # has to fall through to the agent's own output, not short-circuit extraction. + pytest.param( + _task(status_role="user", status_text="check active alarms", artifact_text="THE ANSWER"), + "THE ANSWER", + id="user_echo_falls_through_to_artifacts", + ), + pytest.param( + _task(status_role="agent", status_text="agent status", artifact_text="THE ANSWER"), + "agent status", + id="agent_status_message_preferred", + ), + pytest.param(_task(artifact_text="THE ANSWER"), "THE ANSWER", id="artifacts_only"), + pytest.param(_task(status_role="user", status_text="echo"), "", id="user_echo_alone_is_empty"), + pytest.param( + {"result": {"kind": "message", "role": "user", "parts": [{"kind": "text", "text": "echo"}]}}, + "", + id="direct_user_message_is_empty", + ), + pytest.param( + {"result": {"kind": "message", "role": "agent", "parts": [{"kind": "text", "text": "hi"}]}}, + "hi", + id="direct_agent_message", + ), + ], +) +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 From 8348e1585755cc2b0113548744b7553ada9992ee Mon Sep 17 00:00:00 2001 From: Peter Boers Date: Wed, 9 Sep 2026 15:13:41 +0200 Subject: [PATCH 09/10] 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) --- litellm/llms/a2a/chat/streaming_iterator.py | 14 +++++++ .../a2a/chat/test_a2a_chat_transformation.py | 38 +++++++++++++++++++ 2 files changed, 52 insertions(+) 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}" From 1cf002fd2bb427cf3fa8475706f726331716ef0b Mon Sep 17 00:00:00 2001 From: Peter Boers Date: Thu, 10 Sep 2026 12:16:01 +0200 Subject: [PATCH 10/10] Update route_llm_request.py Co-authored-by: veria-ai[bot] <224490171+veria-ai[bot]@users.noreply.github.com> --- litellm/proxy/route_llm_request.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/litellm/proxy/route_llm_request.py b/litellm/proxy/route_llm_request.py index 48fdf29c209..b5c8daa1684 100644 --- a/litellm/proxy/route_llm_request.py +++ b/litellm/proxy/route_llm_request.py @@ -542,7 +542,7 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr # Otherwise the branches below swallow the request and it fails with # "no healthy deployments": `map_team_model` claims it for team-scoped keys, and the # wildcard/default-deployment fallback claims it whenever a pattern model group exists. - if _is_a2a_agent_model(data.get("model", "")): + if route_type == "acompletion" and _is_a2a_agent_model(data.get("model", "")): from litellm.proxy.agent_endpoints.a2a_routing import ( route_a2a_agent_request, )