Merge 82f3690903 into 5c60824024

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:

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,

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)

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"<html>gateway timeout</html>"], 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