diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index e812b03385d..48b035e6ce2 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -1287,7 +1287,68 @@ 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), + ), + ) + + +_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) + 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=message, ), ) @@ -1541,6 +1602,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 +1648,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..4c49ffcb0be --- /dev/null +++ b/tests/unit/proxy/test_passthrough_late_error_frames.py @@ -0,0 +1,189 @@ +"""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._types import ProxyException +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 + + +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