From 1388f72384ab1cd54ce9c039ff426718740ed6a6 Mon Sep 17 00:00:00 2001 From: Simin Yuan <281164480+simin-yuan@users.noreply.github.com> Date: Thu, 5 Nov 2026 04:15:00 +0800 Subject: [PATCH 1/2] fix(passthrough): end an Anthropic stream with event error when the upstream fails late A passthrough stream keeps relaying upstream bytes, so once the SSE keepalive has written the 200 status line an upstream 4xx or 5xx arrives as a bare JSON chunk inside the event stream. Anthropic clients pick events by name, so they read that truncated stream as a normal end and report success. The route now declares that it answers as Anthropic, which reuses the late failure frames from #46063 for both shapes: an exception before the first byte of the upstream body, and an error status that arrives after the keepalive. Routes that do not declare it keep relaying bytes unchanged. The error body is read with a 64 KiB bound, and a read that fails still ends the stream with an event, so a broken upstream connection cannot leave the caller with nothing. --- litellm/proxy/common_request_processing.py | 73 +++++++++- .../llm_passthrough_endpoints.py | 1 + .../pass_through_endpoints.py | 2 + .../test_passthrough_late_error_frames.py | 125 ++++++++++++++++++ 4 files changed, 200 insertions(+), 1 deletion(-) create mode 100644 tests/unit/proxy/test_passthrough_late_error_frames.py diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index e812b03385d..95bb762173c 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -1292,6 +1292,66 @@ def _anthropic_late_failure_frames(exc: Exception) -> tuple[str, ...]: ) +_ANTHROPIC_LATE_ERROR_BODY_LIMIT: Final = 64 * 1024 + + +async def _read_late_error_body(response: Response) -> bytes: + """Read at most one page of an error body. A read that fails yields what it got.""" + read: bytes = b"" + try: + if isinstance(response, StreamingResponse): + async for chunk in response.body_iterator: + read += chunk.encode("utf-8") if isinstance(chunk, str) else bytes(chunk) + if len(read) >= _ANTHROPIC_LATE_ERROR_BODY_LIMIT: + break + else: + read = bytes(response.body or b"") + except Exception: # noqa: BLE001 # the stream is already broken; the caller still needs a frame + verbose_proxy_logger.exception("could not read the body of an upstream late failure") + return read[:_ANTHROPIC_LATE_ERROR_BODY_LIMIT] + + +class _LateErrorDetail(BaseModel): + model_config = ConfigDict(extra="ignore") + + message: str | None = None + + +class _LateErrorEnvelope(BaseModel): + model_config = ConfigDict(extra="ignore") + + error: _LateErrorDetail | None = None + + +def _late_error_message(body: bytes, status_code: int) -> str: + fallback: Final = f"upstream returned HTTP {status_code}" + detail: Final = body.decode("utf-8", "replace") + if not detail: + return fallback + try: + envelope: Final = _LateErrorEnvelope.model_validate_json(detail) + except ValidationError: + return detail + message: Final = envelope.error.message if envelope.error is not None else None + return message or detail + + +async def _anthropic_late_response_frames(response: Response) -> tuple[str, ...]: + """The frame an Anthropic stream ends with when the upstream answered with an error status. + + The status line is already out, so the failure has to travel as an event. Anthropic + clients read `event: error` and skip anything else, which is why the upstream body + cannot simply be relayed. + """ + body: Final = await _read_late_error_body(response) + return ( + anthropic_error_sse_frame( + status_code=response.status_code, + raw_message=_late_error_message(body, response.status_code), + ), + ) + + def _sse_stream_headers(headers: Mapping[str, str]) -> Mapping[str, str]: """`headers` plus the two that stop reverse proxies from buffering SSE (issue #28384).""" return MappingProxyType({**headers, **_TTFT_KEEPALIVE_HEADERS}) @@ -1541,6 +1601,7 @@ async def open_sse_before_first_byte( media_type: str = "text/event-stream", on_late_failure: Callable[[Exception], Awaitable[HTTPException | None]] | None = None, late_failure_frames: Callable[[Exception], tuple[str, ...]] = _openai_late_failure_frames, + answers_as_anthropic: bool = False, ) -> _LateResponseT | StreamingResponse: """Write SSE keepalive comments while the upstream LLM call is still in flight. @@ -1586,9 +1647,19 @@ async def open_sse_before_first_byte( # would never fire and the failure would go unaudited. The hook # also gets to sanitize what reaches the client, by returning or # raising a replacement, so its answer decides the frame. - for frame in late_failure_frames(await _sanitized_late_failure(exc, on_late_failure)): + failure_frames: Final = _anthropic_late_failure_frames if answers_as_anthropic else late_failure_frames + for frame in failure_frames(await _sanitized_late_failure(exc, on_late_failure)): yield frame.encode() return + if answers_as_anthropic and produced.status_code >= 400: + # The upstream answered with an error status after the keepalive had + # already committed a 200. Those bytes cannot be relayed as they are: + # a caller that speaks a different stream protocol reads them as an + # unknown event and treats the truncated stream as a completed turn. + for frame in await _anthropic_late_response_frames(produced): + yield frame.encode() + await _aclose_late_response(produced) + return async for chunk in _relay_late_response(produced): yield chunk finally: diff --git a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py index 627164abb93..606f84025aa 100644 --- a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py @@ -1108,6 +1108,7 @@ async def anthropic_proxy_route( target=str(updated_url), custom_headers=_upstream_headers_for_anthropic_route(request, user_api_key_dict, auth_header), is_streaming_request=is_streaming_request, + answers_as_anthropic=True, ) # dynamically construct pass-through endpoint based on incoming path received_value: Final = await endpoint_func( request, diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index 7f5983d5799..2fe538a3f38 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -2313,6 +2313,7 @@ def create_pass_through_route( cost_per_request: float | None = None, custom_llm_provider: str | None = None, is_streaming_request: bool | None = False, + answers_as_anthropic: bool = False, query_params: dict | None = None, default_query_params: dict | None = None, guardrails: dict[str, object] | None = None, @@ -2484,6 +2485,7 @@ def create_pass_through_route( return await open_sse_before_first_byte( _relay(), ping_interval_seconds=(litellm.sse_keepalive_ping_interval_seconds if is_stream else None), + answers_as_anthropic=answers_as_anthropic, ) setattr(endpoint_func, LITELLM_PASS_THROUGH_ENDPOINT_MARKER, True) diff --git a/tests/unit/proxy/test_passthrough_late_error_frames.py b/tests/unit/proxy/test_passthrough_late_error_frames.py new file mode 100644 index 00000000000..63045221260 --- /dev/null +++ b/tests/unit/proxy/test_passthrough_late_error_frames.py @@ -0,0 +1,125 @@ +"""A passthrough stream that fails after its keepalive has committed a 200. + +The Anthropic `/anthropic/*` passthrough relays upstream bytes. Once the keepalive +has written `HTTP/1.1 200`, an upstream 4xx/5xx can no longer be expressed as a +status, so those bytes would reach the caller as a bare JSON chunk inside an event +stream. Anthropic clients pick events by name and read that as a normal end. +""" + +import asyncio +import json +from collections.abc import AsyncIterator + +import pytest +from starlette.responses import StreamingResponse + +from litellm.proxy.common_request_processing import open_sse_before_first_byte + +PING = ": ping\n\n" + + +def _error_stream(chunks: list[bytes], status_code: int = 500) -> StreamingResponse: + async def _body() -> AsyncIterator[bytes]: + for chunk in chunks: + yield chunk + + return StreamingResponse(_body(), status_code=status_code, media_type="application/json") + + +def _json_error(payload: dict[str, object], status_code: int = 500) -> StreamingResponse: + return _error_stream([json.dumps(payload).encode("utf-8")], status_code=status_code) + + +async def _drain(response: StreamingResponse) -> str: + chunks = [chunk async for chunk in response.body_iterator] + return b"".join(chunks).decode("utf-8") + + +async def _relay(produced: StreamingResponse, *, answers_as_anthropic: bool = False) -> str: + async def deliver() -> StreamingResponse: + await asyncio.sleep(0.05) + return produced + + response = await open_sse_before_first_byte( + deliver(), ping_interval_seconds=0.02, answers_as_anthropic=answers_as_anthropic + ) + assert isinstance(response, StreamingResponse) + return await _drain(response) + + +@pytest.mark.asyncio +async def test_an_anthropic_route_replaces_a_late_error_body_with_an_event(): + body = await _relay( + _json_error({"error": {"message": "upstream said no", "type": "server_error"}}), + answers_as_anthropic=True, + ) + + assert PING in body + assert body.count("event: error") == 1 + assert "upstream said no" in body + assert "[DONE]" not in body + + +@pytest.mark.asyncio +async def test_a_route_without_the_flag_still_relays_the_error_body_unchanged(): + body = await _relay(_json_error({"error": {"message": "relay me as is"}})) + + assert "event: error" not in body + assert '"message": "relay me as is"' in body + + +@pytest.mark.asyncio +async def test_the_upstream_message_is_what_the_client_reads(): + body = await _relay( + _json_error({"error": {"message": "model overloaded", "type": "server_error"}}, status_code=529), + answers_as_anthropic=True, + ) + + assert "model overloaded" in body + assert "overloaded_error" in body + + +@pytest.mark.asyncio +async def test_a_body_that_is_not_json_still_produces_an_error_frame(): + body = await _relay(_error_stream([b"gateway timeout"], status_code=504), answers_as_anthropic=True) + + assert body.count("event: error") == 1 + assert "gateway timeout" in body + + +@pytest.mark.asyncio +async def test_an_exception_before_the_first_byte_uses_an_anthropic_frame(): + async def boom(): + await asyncio.sleep(0.05) + raise RuntimeError("upstream died after accepting the request") + + response = await open_sse_before_first_byte(boom(), ping_interval_seconds=0.02, answers_as_anthropic=True) + + body = await _drain(response) + assert body.count("event: error") == 1 + assert "[DONE]" not in body + + +@pytest.mark.asyncio +async def test_a_body_that_fails_mid_read_still_produces_an_error_frame(): + async def _broken() -> AsyncIterator[bytes]: + raise RuntimeError("upstream connection reset") + yield b"" + + produced = StreamingResponse(_broken(), status_code=500, media_type="application/json") + + body = await _relay(produced, answers_as_anthropic=True) + + assert body.count("event: error") == 1 + assert "upstream returned HTTP 500" in body + + +@pytest.mark.asyncio +async def test_a_large_error_body_is_read_with_a_bound(): + padding = "x" * (256 * 1024) + produced = _error_stream([padding.encode("utf-8")], status_code=500) + + body = await _relay(produced, answers_as_anthropic=True) + + assert body.count("event: error") == 1 + assert len(body) < 128 * 1024 From 82f369090318e1684d753ea9612b9abfecae7ce6 Mon Sep 17 00:00:00 2001 From: Simin Yuan <281164480+simin-yuan@users.noreply.github.com> Date: Thu, 5 Nov 2026 11:36:34 +0800 Subject: [PATCH 2/2] fix(passthrough): redact internal details in late Anthropic failure frames --- litellm/proxy/common_request_processing.py | 5 +- .../test_passthrough_late_error_frames.py | 64 +++++++++++++++++++ 2 files changed, 67 insertions(+), 2 deletions(-) diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index 95bb762173c..48b035e6ce2 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -1287,7 +1287,7 @@ def _anthropic_late_failure_frames(exc: Exception) -> tuple[str, ...]: return ( anthropic_error_sse_frame( status_code=error_status_code(answered, status.HTTP_500_INTERNAL_SERVER_ERROR), - raw_message=answered.message, + raw_message=redact_internal_details_from_client_message(answered.message), ), ) @@ -1344,10 +1344,11 @@ async def _anthropic_late_response_frames(response: Response) -> tuple[str, ...] cannot simply be relayed. """ body: Final = await _read_late_error_body(response) + message: Final = redact_internal_details_from_client_message(_late_error_message(body, response.status_code)) return ( anthropic_error_sse_frame( status_code=response.status_code, - raw_message=_late_error_message(body, response.status_code), + raw_message=message, ), ) diff --git a/tests/unit/proxy/test_passthrough_late_error_frames.py b/tests/unit/proxy/test_passthrough_late_error_frames.py index 63045221260..4c49ffcb0be 100644 --- a/tests/unit/proxy/test_passthrough_late_error_frames.py +++ b/tests/unit/proxy/test_passthrough_late_error_frames.py @@ -13,6 +13,7 @@ from collections.abc import AsyncIterator import pytest from starlette.responses import StreamingResponse +from litellm.proxy._types import ProxyException from litellm.proxy.common_request_processing import open_sse_before_first_byte PING = ": ping\n\n" @@ -123,3 +124,66 @@ async def test_a_large_error_body_is_read_with_a_bound(): assert body.count("event: error") == 1 assert len(body) < 128 * 1024 + + +FAKE_KEY = "sk-test-ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" +INTERNAL_PATH = "/home/runner/litellm/proxy/internal_state.py" + + +@pytest.mark.asyncio +async def test_a_late_failure_exception_is_redacted_before_framing(): + async def boom(): + await asyncio.sleep(0.05) + raise RuntimeError(f"upstream failed with key {FAKE_KEY} at {INTERNAL_PATH}") + + response = await open_sse_before_first_byte(boom(), ping_interval_seconds=0.02, answers_as_anthropic=True) + body = await _drain(response) + + assert body.count("event: error") == 1 + assert FAKE_KEY not in body + assert INTERNAL_PATH not in body + assert "REDACTED" in body + + +@pytest.mark.asyncio +async def test_a_late_failure_that_is_already_a_proxy_exception_is_redacted(): + async def boom(): + await asyncio.sleep(0.05) + raise ProxyException( + message=f"upstream failed with key {FAKE_KEY} at {INTERNAL_PATH}", + type="api_error", + param=None, + code=500, + ) + + response = await open_sse_before_first_byte(boom(), ping_interval_seconds=0.02, answers_as_anthropic=True) + body = await _drain(response) + + assert body.count("event: error") == 1 + assert FAKE_KEY not in body + assert INTERNAL_PATH not in body + assert "REDACTED" in body + + +@pytest.mark.asyncio +async def test_a_late_error_body_is_redacted_before_framing(): + body = await _relay( + _json_error({"error": {"message": f"failed with key {FAKE_KEY} at {INTERNAL_PATH}", "type": "server_error"}}), + answers_as_anthropic=True, + ) + + assert body.count("event: error") == 1 + assert FAKE_KEY not in body + assert INTERNAL_PATH not in body + assert "REDACTED" in body + + +@pytest.mark.asyncio +async def test_an_ordinary_error_message_is_not_redacted(): + body = await _relay( + _json_error({"error": {"message": "upstream said no", "type": "server_error"}}), + answers_as_anthropic=True, + ) + + assert "upstream said no" in body + assert "REDACTED" not in body