scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-03 07:23:33 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 166

Merge 647a962f41 into cfe0100e1f

Browse source
This commit is contained in:
yyobject 2026-10-27 05:16:24 +00:00 • committed by GitHub
commit 23fcf00a09
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
18 changed files with 1383 additions and 68 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
litellm/llms/anthropic/pass_through/messages/streaming_iterator.py
View file

@ -41,11 +41,17 @@ _UPSTREAM_PUMP_TASKS: Final[set[asyncio.Task[None]]] = set() # mutable-ok: stdl
_DETACHED_STREAM_DRAINS: Final[set[asyncio.Task[None]]] = set() # mutable-ok: bounded strong-ref set, detached drains
def _sse_event_names(chunk: bytes | bytearray) -> tuple[bytes, ...]:
return tuple(
line[len(b"event:") :].removeprefix(b" ") for line in bytes(chunk).splitlines() if line.startswith(b"event:")
)
def is_message_stop_chunk(chunk: object) -> bool:
if isinstance(chunk, dict):
return chunk.get("type") == "message_stop"
if isinstance(chunk, (bytes, bytearray)):
return any(line == b"event: message_stop" for line in chunk.splitlines())
return b"message_stop" in _sse_event_names(chunk)
return False
@ -72,12 +78,12 @@ def is_anthropic_ping_chunk(chunk: object) -> bool:
return chunk.get("type") == "ping"
if not isinstance(chunk, (bytes, bytearray)):
return False
event_lines: Final = tuple(line for line in chunk.splitlines() if line.startswith(b"event:"))
event_names: Final = _sse_event_names(chunk)
return (
bool(event_lines)
and all(line == b"event: ping" for line in event_lines)
and chunk.startswith(b"event: ping")
and chunk.endswith((b"\n\n", b"\r\n\r\n"))
bool(event_names)
and all(name == b"ping" for name in event_names)
and chunk.startswith((b"event: ping", b"event:ping"))
and chunk.endswith((b"\n\n", b"\r\n\r\n", b"\r\r"))
)
@ -92,7 +98,7 @@ def is_anthropic_content_delta_chunk(chunk: object) -> bool:
if isinstance(chunk, dict):
return chunk.get("type") == "content_block_delta"
if isinstance(chunk, (bytes, bytearray)):
return any(line == b"event: content_block_delta" for line in chunk.splitlines())
return b"content_block_delta" in _sse_event_names(chunk)
return False
@ -521,6 +527,7 @@ class BaseAnthropicMessagesStreamingIterator:
if self.completion_start_time is not None:
self.litellm_logging_obj.completion_start_time = self.completion_start_time
self.litellm_logging_obj.model_call_details["completion_start_time"] = self.completion_start_time
self.litellm_logging_obj.model_call_details["anthropic_stream_end"] = "closed" if stream_teardown else "eof"
logging_coroutine: Final = PassThroughStreamingHandler.route_streaming_logging_to_handler(
litellm_logging_obj=self.litellm_logging_obj,
passthrough_success_handler_obj=GLOBAL_PASS_THROUGH_SUCCESS_HANDLER_OBJ,

38
litellm/proxy/common_request_processing.py
View file

@ -28,7 +28,7 @@ import orjson
from fastapi import HTTPException, Request, status
from fastapi.responses import JSONResponse, Response, StreamingResponse
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
from starlette.types import Receive, Scope, Send
from starlette.types import Message, Receive, Scope, Send
from typing_extensions import Never
import litellm
@ -91,6 +91,7 @@ from litellm.proxy.auth.auth_checks import (
)
from litellm.proxy.auth.auth_utils import check_response_size_is_safe, get_request_route
from litellm.proxy.bug_report_config import build_proxy_bug_report
from litellm.proxy.common_utils.anthropic_stream_diagnostics import AnthropicStreamDiagnostics
from litellm.proxy.common_utils.callback_utils import (
get_logging_caching_headers,
get_remaining_tokens_and_requests_from_request_data,
@ -992,9 +993,11 @@ class _UpstreamClosingStreamingResponse(StreamingResponse):
headers: Mapping[str, str] | None = None,
status_code: int = status.HTTP_200_OK,
upstream_generator: AsyncGenerator[str, None] | None = None,
stream_diagnostics: AnthropicStreamDiagnostics | None = None,
) -> None:
super().__init__(content, status_code=status_code, headers=headers, media_type=media_type)
self._upstream_generator = upstream_generator
self._stream_diagnostics = stream_diagnostics
@property
def upstream_generator(self) -> AsyncGenerator[str, None] | None:
@ -1002,9 +1005,31 @@ class _UpstreamClosingStreamingResponse(StreamingResponse):
return self._upstream_generator
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
diagnostics: Final = self._stream_diagnostics
async def observed_send(message: Message) -> None:
assert diagnostics is not None
body: Final = message.get("body")
has_body: Final = message["type"] == "http.response.body" and isinstance(body, bytes) and bool(body)
if has_body:
diagnostics.start_send()
try:
await send(message)
except BaseException:
diagnostics.send_failed = True
raise
if message["type"] == "http.response.start":
diagnostics.headers_sent_ms = diagnostics.elapsed_ms()
if has_body and isinstance(body, bytes):
diagnostics.complete_send(body)
if message["type"] == "http.response.body" and not message.get("more_body", False):
diagnostics.http_complete = True
try:
await super().__call__(scope, receive, send)
await super().__call__(scope, receive, observed_send if diagnostics is not None else send)
finally:
if diagnostics is not None:
diagnostics.finish("eof" if diagnostics.http_complete else "closed")
with anyio.CancelScope(shield=True):
for target in (self.body_iterator, self._upstream_generator):
aclose = getattr(target, "aclose", None)
@ -1176,6 +1201,7 @@ async def create_response(
request: Request | None = None,
refresh_headers: Callable[[], Awaitable[Mapping[str, str]]] | None = None,
general_settings: Mapping[str, object] = _NO_GENERAL_SETTINGS,
diagnose_anthropic_stream: bool = False,
) -> StreamingResponse | JSONResponse:
"""
Create streaming response, checking if the first chunk is an error.
@ -1296,6 +1322,13 @@ async def create_response(
headers=_sse_stream_headers(resolved_headers),
status_code=final_status_code,
upstream_generator=generator,
stream_diagnostics=(
AnthropicStreamDiagnostics.from_env(
"http_send", resolved_headers.get(LITELLM_CALL_ID_HEADER), resolved_headers
)
if diagnose_anthropic_stream
else None
),
)
@ -3148,6 +3181,7 @@ class ProxyBaseLLMRequestProcessing:
request=request,
refresh_headers=refresh_stream_headers,
general_settings=general_settings,
diagnose_anthropic_stream=route_type == "anthropic_messages",
)
### CALL HOOKS ### - modify outgoing data

299
litellm/proxy/common_utils/anthropic_stream_diagnostics.py Normal file
View file

@ -0,0 +1,299 @@
import json
import os
import re
from collections.abc import AsyncGenerator, Callable, Mapping
from contextlib import suppress
from dataclasses import asdict, dataclass, replace
from time import monotonic, time
from typing import Final, Literal, TypeAlias
from pydantic import JsonValue, TypeAdapter
from litellm._logging import verbose_proxy_logger
_FRAME_END: Final = re.compile(rb"\r\n\r\n|\n\n|\r\r")
_MAX_FRAME_BYTES: Final = 64 * 1024
_SAFE_ID: Final = re.compile(r"[A-Za-z0-9_-]{1,160}")
_MODE: Final = frozenset(("errors", "slow", "all"))
_EVENTS: Final = frozenset(
(
"message_start",
"message_stop",
"message_delta",
"content_block_start",
"content_block_delta",
"content_block_stop",
"ping",
"error",
)
)
_REQUEST_ID_HEADERS: Final = (
"x-tc-requestid",
"x-tc-request-id",
"x-tencent-request-id",
"request-id",
"x-request-id",
"x-amzn-requestid",
)
_SLOW_CONTENT_MS: Final = 30_000
_SLOW_SEND_MS: Final = 1_000
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, JsonValue])
StreamEnd: TypeAlias = Literal["eof", "closed", "transport_error"]
StreamStage: TypeAlias = Literal["provider_read", "router_emit", "http_send"]
@dataclass(frozen=True, slots=True)
class _StreamCounts:
chunks: int = 0
bytes: int = 0
events: int = 0
starts: int = 0
stops: int = 0
errors: int = 0
pings: int = 0
text_deltas: int = 0
text_chars: int = 0
thinking_deltas: int = 0
tool_deltas: int = 0
unparsed_frames: int = 0
empty_data_frames: int = 0
first_byte_ms: int | None = None
first_event_ms: int | None = None
first_text_ms: int | None = None
first_content_ms: int | None = None
last_content_ms: int | None = None
max_content_gap_ms: int = 0
first_event_type: str | None = None
last_event_type: str | None = None
last_byte_ms: int | None = None
message_id: str | None = None
def _safe_id(value: object) -> str | None:
return value if isinstance(value, str) and _SAFE_ID.fullmatch(value) and not value.startswith("sk-") else None
class AnthropicStreamDiagnostics:
def __init__(
self,
mode: str,
stage: StreamStage,
call_id: object,
*,
upstream_request_id: object = None,
clock: Callable[[], float] = monotonic,
wall_clock: Callable[[], float] = time,
) -> None:
self.mode = mode
self.stage = stage
self.call_id = _safe_id(call_id)
self.upstream_request_id = _safe_id(upstream_request_id)
self.clock = clock
self.started = clock()
self.started_at_unix_ms = int(wall_clock() * 1000)
self.counts = _StreamCounts()
self.pending = b""
self.discarding_frame = False
self.finished = False
self.headers_sent_ms: int | None = None
self.send_started_ms: int | None = None
self.first_send_attempt_ms: int | None = None
self.last_send_completed_ms: int | None = None
self.max_send_ms = 0
self.send_attempts = 0
self.sends_completed = 0
self.send_failed = False
self.http_complete = False
def elapsed_ms(self) -> int:
return max(0, int((self.clock() - self.started) * 1000))
def start_send(self) -> None:
self.send_started_ms = self.elapsed_ms()
self.send_attempts += 1
if self.first_send_attempt_ms is None:
self.first_send_attempt_ms = self.send_started_ms
def complete_send(self, chunk: bytes) -> None:
self.last_send_completed_ms = self.elapsed_ms()
if self.send_started_ms is not None:
self.max_send_ms = max(self.max_send_ms, self.last_send_completed_ms - self.send_started_ms)
self.send_started_ms = None
self.sends_completed += 1
self.observe(chunk)
@classmethod
def from_env(
cls, stage: StreamStage, call_id: object, headers: Mapping[str, str] | None = None
) -> "AnthropicStreamDiagnostics | None":
mode: Final = os.getenv("LITELLM_ANTHROPIC_STREAM_DIAGNOSTICS", "off").lower()
if mode not in _MODE:
return None
normalized_headers: Final = {name.lower(): value for name, value in (headers or {}).items()}
request_id: Final = next(
(
normalized_headers.get(name)
for name in (
_REQUEST_ID_HEADERS
if stage == "provider_read"
else tuple("llm_provider-" + key for key in _REQUEST_ID_HEADERS)
)
if _safe_id(normalized_headers.get(name)) is not None
),
None,
)
return cls(mode, stage, call_id, upstream_request_id=request_id)
def observe(self, chunk: bytes) -> None:
if self.finished or not chunk:
return
elapsed: Final = self.elapsed_ms()
self.counts = replace(
self.counts,
chunks=self.counts.chunks + 1,
bytes=self.counts.bytes + len(chunk),
first_byte_ms=self.counts.first_byte_ms if self.counts.first_byte_ms is not None else elapsed,
last_byte_ms=elapsed,
)
data: Final = self.pending + chunk
offset = 0 # rebind-ok: scan transport bytes once and retain only a bounded unfinished frame
for boundary in _FRAME_END.finditer(data, max(0, len(self.pending) - 3)):
if not self.discarding_frame:
self._observe_frame(data[offset : boundary.start()], elapsed)
self.discarding_frame = False
offset = boundary.end()
tail: Final = data[offset:]
if len(tail) > _MAX_FRAME_BYTES:
if not self.discarding_frame:
self.counts = replace(self.counts, unparsed_frames=self.counts.unparsed_frames + 1)
self.discarding_frame = True
self.pending = tail[-3:] if self.discarding_frame else tail
def _observe_frame(self, frame: bytes, elapsed: int) -> None:
if len(frame) > _MAX_FRAME_BYTES:
self.counts = replace(self.counts, unparsed_frames=self.counts.unparsed_frames + 1)
return
data_lines: Final = tuple(
line[len(b"data:") :].removeprefix(b" ")
for line in frame.splitlines()
if line.startswith(b"data:") or line == b"data"
)
if not data_lines:
return
data: Final = b"\n".join(data_lines)
if not data.strip():
self.counts = replace(self.counts, empty_data_frames=self.counts.empty_data_frames + 1)
return
try:
payload: Final = _JSON_OBJECT.validate_json(data)
except (ValueError, UnicodeError):
self.counts = replace(self.counts, unparsed_frames=self.counts.unparsed_frames + 1)
return
event: Final = payload.get("type")
delta: Final = payload.get("delta")
delta_type: Final = delta.get("type") if isinstance(delta, dict) and event == "content_block_delta" else None
text: Final = delta.get("text") if isinstance(delta, dict) and delta_type == "text_delta" else None
text_chars: Final = len(text) if isinstance(text, str) else 0
content: Final = (
delta.get(
{"text_delta": "text", "thinking_delta": "thinking", "input_json_delta": "partial_json"}.get(
delta_type, ""
)
)
if isinstance(delta, dict) and isinstance(delta_type, str)
else None
)
has_content: Final = isinstance(content, str) and bool(content)
event_name: Final = event if isinstance(event, str) and event in _EVENTS else "other"
message: Final = payload.get("message")
message_id: Final = _safe_id(message.get("id")) if isinstance(message, dict) else None
self.counts = replace(
self.counts,
events=self.counts.events + 1,
starts=self.counts.starts + (event == "message_start"),
stops=self.counts.stops + (event == "message_stop"),
errors=self.counts.errors + (event == "error"),
pings=self.counts.pings + (event == "ping"),
text_deltas=self.counts.text_deltas + (delta_type == "text_delta"),
text_chars=self.counts.text_chars + text_chars,
thinking_deltas=self.counts.thinking_deltas + (delta_type == "thinking_delta"),
tool_deltas=self.counts.tool_deltas + (delta_type == "input_json_delta"),
first_event_ms=self.counts.first_event_ms if self.counts.first_event_ms is not None else elapsed,
first_text_ms=(
elapsed if text_chars > 0 and self.counts.first_text_ms is None else self.counts.first_text_ms
),
first_content_ms=elapsed
if has_content and self.counts.first_content_ms is None
else self.counts.first_content_ms,
last_content_ms=elapsed if has_content else self.counts.last_content_ms,
max_content_gap_ms=max(self.counts.max_content_gap_ms, elapsed - (self.counts.last_content_ms or 0))
if has_content
else self.counts.max_content_gap_ms,
first_event_type=self.counts.first_event_type or event_name,
last_event_type=event_name,
message_id=message_id if event == "message_start" and message_id is not None else self.counts.message_id,
)
def finish(self, reason: StreamEnd) -> None:
if self.finished:
return
self.finished = True
elapsed: Final = self.elapsed_ms()
content_gap: Final = max(self.counts.max_content_gap_ms, elapsed - (self.counts.last_content_ms or 0))
complete: Final = (
self.counts.stops == 1
and self.counts.errors == 0
and self.counts.starts == 1
and not self.pending.strip()
and not self.discarding_frame
and self.counts.unparsed_frames == 0
and self.counts.empty_data_frames == 0
and not self.send_failed
and (self.stage != "http_send" or self.http_complete)
)
slow: Final = content_gap >= _SLOW_CONTENT_MS or self.max_send_ms >= _SLOW_SEND_MS
if complete and reason == "eof" and (self.mode == "errors" or (self.mode == "slow" and not slow)):
return
summary: Final = {
"stage": self.stage,
"call_id": self.call_id,
"upstream_request_id": self.upstream_request_id,
"end": reason,
"complete": complete,
"slow": slow,
"started_at_unix_ms": self.started_at_unix_ms,
"started_monotonic_ms": int(self.started * 1000),
"elapsed_ms": elapsed,
"pending_bytes": len(self.pending),
"headers_sent_ms": self.headers_sent_ms,
"first_send_attempt_ms": self.first_send_attempt_ms,
"last_send_completed_ms": self.last_send_completed_ms,
"send_attempts": self.send_attempts,
"sends_completed": self.sends_completed,
"send_failed": self.send_failed,
"send_in_flight_ms": elapsed - self.send_started_ms if self.send_started_ms is not None else None,
"max_send_ms": self.max_send_ms,
**asdict(self.counts),
"max_content_gap_ms": content_gap,
}
with suppress(Exception):
verbose_proxy_logger.info("anthropic_stream_summary %s", json.dumps(summary, separators=(",", ":")))
async def relay(self, stream: AsyncGenerator[bytes, None]) -> AsyncGenerator[bytes, None]:
end: StreamEnd = "closed" # rebind-ok: stream teardown distinguishes EOF, errors and consumer close
try:
async for chunk in stream:
self.observe(chunk)
yield chunk
end = "eof" # rebind-ok: upstream iteration ended normally
except Exception:
end = "transport_error" # rebind-ok: preserve the failure for the final summary
raise
finally:
self.finish(end)
await stream.aclose()
def finish_stream_diagnostics(diagnostics: AnthropicStreamDiagnostics | None, reason: StreamEnd) -> None:
if diagnostics is not None:
diagnostics.finish(reason)

8
litellm/proxy/common_utils/sse_keepalive.py
View file

@ -66,16 +66,20 @@ async def _keepalive_ping_stream(
ping_chunk: str,
) -> AsyncGenerator[str, None]:
pending = asyncio.ensure_future(stream.__anext__())
recent_tail = SSE_STREAM_START_TAIL # rebind-ok: keepalives must not split a partially forwarded SSE event
try:
while True:
await asyncio.wait({pending}, timeout=ping_interval_seconds)
if not pending.done():
yield ping_chunk
if recent_tail.endswith(_SSE_FRAME_DELIMITERS):
yield ping_chunk
continue
try:
yield pending.result()
chunk = pending.result()
except StopAsyncIteration:
return
recent_tail = advance_sse_tail(recent_tail, chunk)
yield chunk
pending = asyncio.ensure_future(stream.__anext__())
finally:
pending.cancel()

67
litellm/proxy/pass_through_endpoints/architecture.md
View file

@ -68,4 +68,69 @@ sequenceDiagram
- Request body
- Response body
- Provider-specific parameters
- Provider-specific parameters
## Anthropic stream diagnostics
Set `LITELLM_ANTHROPIC_STREAM_DIAGNOSTICS=slow` to collect an end-of-stream
summary for incomplete or failed native Anthropic streams, content gaps of at
least 30 seconds, or ASGI sends taking at least one second. `errors` omits slow
completed streams. `all` includes every observed stream. The default is `off`.
Remove the variable or set it to `off`
and restart the workers to disable the additional summaries. This does not
disable normal success/failure callbacks or change billing
Each native HTTP attempt can emit one `provider_read` summary. A routed stream
can also emit one `router_emit` summary and one `http_send` summary. Retries have
their own attempt summaries. Each stage emits at most once, with no per-token
or periodic logs. At one million single-attempt routed requests, `all` can emit
approximately three million summary lines. Use `slow` for routine troubleshooting
and `all` for a bounded investigation that needs a complete comparison of stages.
Stage filtering is independent, so the absence of a summary in `slow` or `errors`
mode does not establish that a stage received no content
Keep the proxy logger at INFO rather than enabling detailed DEBUG logging
Correlate summaries by `call_id`, the provider's `message_id` and, when supplied,
an allowlisted `upstream_request_id` response header. The summaries
contain byte and event counts, text character counts, first-byte and first-text timings, thinking and
tool-delta counts, terminal-event counts and the end reason. Tencent's
`x-tc-requestid` and its spelling variants are accepted case-insensitively,
including the proxy's `llm_provider-` prefix. If the provider supplies no ID,
`upstream_request_id` stays null. No synthetic provider ID is substituted
Times are relative to the observation stage starting. Add `started_at_unix_ms`
to an event's elapsed milliseconds to compare wall-clock arrival times.
`started_monotonic_ms` allows comparisons within the same worker without wall-clock
adjustments. A provider observer starts after HTTP headers are available, while
the HTTP-send observer starts after the proxy's first-chunk inspection.
These values are not total request latency or model execution time.
`first_content_ms` includes nonempty text, thinking and tool-argument deltas;
`first_text_ms` includes only nonempty answer text
`provider_read.text_chars > 0` with no corresponding `router_emit.text_chars`
shows that LiteLLM received text before its router emitted it. Zero text characters
at `provider_read` means no nonempty text delta was observed there, not proof
that the provider never generated text. Thinking and ping counts help distinguish
reasoning from an entirely silent stream. `router_emit` means yielded to the
next consumer. `http_send` counts a body chunk only after the ASGI `send` call
returns. `first_send_attempt_ms`, `last_send_completed_ms`, `max_send_ms` and
unfinished or failed send counters distinguish yielding from a blocked send.
ASGI acceptance still does not prove that downstream proxies or clients received those bytes.
Compare ingress and client logs for failures beyond that boundary
These summaries never include prompts, generated text, tool arguments, keys,
raw headers or raw exception messages. The diagnostic parser retains at most
64 KiB of an unfinished event. Oversized or malformed events increase
`unparsed_frames`, making content counts inconclusive for those events without
changing the forwarded bytes. Empty data-bearing events increase
`empty_data_frames` and make the summary incomplete, while comment heartbeats
are ignored. A worker killed before teardown cannot emit its
final summary. A process that stays stuck without teardown also has no final
summary yet. This logging does not add a timeout, retry or cancellation policy
An HTTP 200 with no complete `message_stop`, or an SSE provider error, is logged
as a failed stream. Partial token usage and its cost still reach the failure
callbacks. Initial `message_start.usage.output_tokens` alone does not establish
a successful response. A consumer closing an unfinished stream is recorded as
499 rather than a provider 5xx. A premature upstream EOF is recorded as 502.
These are logging outcomes, not replacements for HTTP headers already sent

13
litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py
View file

@ -276,11 +276,20 @@ class AnthropicPassthroughLoggingHandler:
) -> None:
if not all_chunks:
return
byte_chunks: Final = tuple(chunk for chunk in all_chunks if isinstance(chunk, bytes))
normalized_chunks: Final = (
tuple(b"".join(byte_chunks).decode("utf-8", errors="replace").splitlines())
if len(byte_chunks) == len(all_chunks)
else all_chunks
)
model: Final = AnthropicPassthroughLoggingHandler._resolve_logged_model(
litellm_logging_obj, request_body, all_chunks
litellm_logging_obj, request_body, normalized_chunks
)
partial_response: Final = AnthropicPassthroughLoggingHandler._build_streaming_response_for_logging(
litellm_logging_obj=litellm_logging_obj, request_body=request_body, all_chunks=all_chunks, model=model
litellm_logging_obj=litellm_logging_obj,
request_body=request_body,
all_chunks=normalized_chunks,
model=model,
)
usage: Final = cast(Usage | None, getattr(partial_response, "usage", None))
if partial_response is None or usage is None:

95
litellm/proxy/pass_through_endpoints/streaming_handler.py
View file

@ -5,6 +5,7 @@ from datetime import datetime, timezone
from typing import Final, Protocol
import httpx
from pydantic import TypeAdapter
import litellm
from litellm._logging import verbose_proxy_logger
@ -15,6 +16,11 @@ from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from litellm.llms.anthropic.common_utils import anthropic_error_frame_exception
from litellm.proxy._types import PassThroughEndpointLoggingResultValues
from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
from litellm.proxy.common_utils.anthropic_stream_diagnostics import (
AnthropicStreamDiagnostics,
StreamEnd,
finish_stream_diagnostics,
)
from litellm.proxy.common_utils.sse_keepalive import split_complete_sse_frames
from litellm.types.passthrough_endpoints.pass_through_endpoints import EndpointType
from litellm.types.utils import StandardPassThroughResponseObject, Usage
@ -38,6 +44,8 @@ from .llm_provider_handlers.vertex_passthrough_logging_handler import (
)
from .success_handler import PassThroughEndpointLogging
_REQUEST_BODY: Final = TypeAdapter(dict[str, object])
class RouteStreamingLogging(Protocol):
def __call__(
@ -63,9 +71,25 @@ class PassThroughStreamContext:
class PassThroughStreamingHandler:
@staticmethod
def _stamp_first_chunk_if_needed(litellm_logging_obj: LiteLLMLoggingObj) -> None:
def _stamp_first_chunk_if_needed(
litellm_logging_obj: LiteLLMLoggingObj,
diagnostics: AnthropicStreamDiagnostics | None = None,
chunk: bytes = b"",
) -> None:
if litellm_logging_obj.completion_start_time is None:
litellm_logging_obj.update_completion_start_time(completion_start_time=datetime.now())
if diagnostics is not None:
diagnostics.observe(chunk)
@staticmethod
def _stream_logging_context(
logging_obj: LiteLLMLoggingObj, endpoint_type: EndpointType, stream_end: StreamEnd
) -> LiteLLMLoggingObj:
if endpoint_type != EndpointType.ANTHROPIC:
return logging_obj
attempt: Final = logging_obj.attempt_scoped_copy() if stream_end == "closed" else logging_obj
attempt.model_call_details["anthropic_stream_end"] = stream_end
return attempt
@staticmethod
async def schedule_stream_failure_logging(
@ -172,10 +196,18 @@ class PassThroughStreamingHandler:
)
raw_bytes: Final[list[bytes]] = []
resolved_request_body: Final[dict[str, object]] = request_body or {}
diagnostics: Final = (
AnthropicStreamDiagnostics.from_env("provider_read", litellm_logging_obj.litellm_call_id, response.headers)
if endpoint_type == EndpointType.ANTHROPIC
else None
)
stream_end: StreamEnd = "closed" # rebind-ok: distinguish generator close, EOF and transport failure
def _build_logging_coroutine() -> Coroutine[None, None, None]:
return resolved_route_streaming_logging(
litellm_logging_obj=litellm_logging_obj,
litellm_logging_obj=PassThroughStreamingHandler._stream_logging_context(
litellm_logging_obj, endpoint_type, stream_end
),
passthrough_success_handler_obj=passthrough_success_handler_obj,
url_route=url_route,
request_body=resolved_request_body,
@ -217,7 +249,7 @@ class PassThroughStreamingHandler:
# Hot path: just buffer for end-of-stream logging and forward.
async for chunk in response.aiter_bytes():
raw_bytes.append(chunk)
PassThroughStreamingHandler._stamp_first_chunk_if_needed(litellm_logging_obj)
PassThroughStreamingHandler._stamp_first_chunk_if_needed(litellm_logging_obj, diagnostics, chunk)
if tinyfish_scan_active:
complete_frames, tinyfish_pending = split_complete_sse_frames(tinyfish_pending + chunk)
run_id = run_id_from_sse_frames(complete_frames) if b"run_id" in complete_frames else None
@ -238,7 +270,7 @@ class PassThroughStreamingHandler:
pending = b""
async for chunk in response.aiter_bytes():
raw_bytes.append(chunk)
PassThroughStreamingHandler._stamp_first_chunk_if_needed(litellm_logging_obj)
PassThroughStreamingHandler._stamp_first_chunk_if_needed(litellm_logging_obj, diagnostics, chunk)
complete_frames, pending = split_complete_sse_frames(pending + chunk)
if complete_frames:
yield ProxyBaseLLMRequestProcessing.process_chunk_with_cost_injection(
@ -247,6 +279,7 @@ class PassThroughStreamingHandler:
if pending:
yield pending
stream_ended = True # rebind-ok: the finally reads it to spot a close-at-frame
stream_end = "eof" # rebind-ok: upstream iteration completed without a transport exception
# Stream completed cleanly. When the proxy armed deferred
# dispatch (post-call guardrails active), park the logging
# coroutine on logging_obj instead of enqueueing now, so
@ -257,14 +290,15 @@ class PassThroughStreamingHandler:
# upstream exceptions log a failure instead (LIT-3798).
if (
getattr(litellm_logging_obj, "_on_deferred_stream_complete", None) is not None
and raw_bytes
and (raw_bytes or endpoint_type == EndpointType.ANTHROPIC)
and response.status_code < 400
):
logging_scheduled = True
bind_budget_reservation_to_callbacks(litellm_logging_obj.litellm_params)
litellm_logging_obj._deferred_stream_complete_args = (_build_logging_coroutine(),)
except Exception as e:
verbose_proxy_logger.error("Error in chunk_processor: %s", e)
stream_end = "transport_error" # rebind-ok: preserve the transport failure in the final diagnostic
verbose_proxy_logger.error("Error in chunk_processor (%s)", type(e).__name__)
if response.status_code < 400:
logging_scheduled = True
await PassThroughStreamingHandler.schedule_stream_failure_logging(
@ -281,6 +315,7 @@ class PassThroughStreamingHandler:
)
raise
finally:
finish_stream_diagnostics(diagnostics, stream_end)
# GeneratorExit (raised on client disconnect) is not caught by
# `except Exception`; the finally block ensures partial usage
# still gets logged for spend tracking. See LIT-2642.
@ -288,7 +323,11 @@ class PassThroughStreamingHandler:
# the caller before this generator starts (see
# _log_passthrough_upstream_failure); logging them again here as
# a success would double-log the same request.
if not logging_scheduled and raw_bytes and response.status_code < 400:
if (
not logging_scheduled
and (raw_bytes or endpoint_type == EndpointType.ANTHROPIC)
and response.status_code < 400
):
logging_scheduled = True
superseded_error_event: Final = (
None
@ -345,6 +384,10 @@ class PassThroughStreamingHandler:
from litellm.llms.anthropic.pass_through.messages.streaming_iterator import (
is_message_stop_chunk, # pyright: ignore[reportPrivateUsage] # both native stream paths share terminal-event detection
is_provider_error_chunk, # pyright: ignore[reportPrivateUsage] # provider errors must not become cache evidence
parse_anthropic_error_event,
)
from litellm.llms.anthropic.pass_through.messages.utils import (
INCOMPLETE_STREAM_ERROR_MESSAGE,
)
# Transport reads can split event names and JSON payloads. Recognize terminal
@ -361,6 +404,44 @@ class PassThroughStreamingHandler:
and not is_provider_error_chunk(complete_frames)
)
try:
if endpoint_type == EndpointType.ANTHROPIC:
request_mapping: Final = _REQUEST_BODY.validate_python(request_body)
error_event: Final = parse_anthropic_error_event(complete_frames)
if error_event is not None:
error_type, message, status_code = error_event
await PassThroughStreamingHandler.schedule_stream_failure_logging(
litellm_logging_obj=litellm_logging_obj,
endpoint_type=endpoint_type,
request_body=request_mapping,
raw_bytes=raw_bytes,
exception=anthropic_error_frame_exception(
error_type, message, status_code, str(litellm_logging_obj.model)
),
)
return
if not litellm_logging_obj.model_call_details["prompt_cache_response_complete"]:
await asyncify(AnthropicPassthroughLoggingHandler.record_partial_usage_for_failure)(
litellm_logging_obj=litellm_logging_obj,
request_body=request_mapping,
all_chunks=raw_bytes,
)
consumer_closed: Final = (
litellm_logging_obj.model_call_details.get("anthropic_stream_end") == "closed"
)
await litellm_logging_obj.dispatch_attempt_failure_handlers(
litellm.APIError(
status_code=499 if consumer_closed else 502,
message=(
"Stream consumer closed before message_stop"
if consumer_closed
else INCOMPLETE_STREAM_ERROR_MESSAGE
),
llm_provider="anthropic",
model=str(litellm_logging_obj.model),
),
"",
)
return
# TinyFish billing is owned by the detached poller; the $0 fallback below is only for streams with no run_id
if endpoint_type == EndpointType.TINYFISH:
if sse_poller_spawned(litellm_logging_obj):

12
litellm/router.py
View file

@ -6480,8 +6480,6 @@ class Router:
if _anthropic_stream_forwards_ping_live(chunk, has_generated_content):
yield chunk
continue
if _stream_commits_now(chunk, has_generated_content, len(buffered_lifecycle_chunks)):
has_generated_content = True
# A transport can split one SSE data line across byte chunks, so pre-content
# detection parses the accumulated buffer plus the current chunk, never the
# chunk alone; the buffer is already capped, which bounds this window too.
@ -6490,6 +6488,8 @@ class Router:
if not has_generated_content and isinstance(chunk, (bytes, bytearray)) # pyright: ignore[reportUnnecessaryIsInstance] # bridge-path chunks are not always bytes at runtime
else chunk
)
if _stream_commits_now(parse_window, has_generated_content, len(buffered_lifecycle_chunks)):
has_generated_content = True
error_event = parse_anthropic_error_event(parse_window)
recoverable_frame_error = self._anthropic_messages_recoverable_frame_error(
error_event, chunk, has_generated_content, model, initial_kwargs
@ -6547,7 +6547,13 @@ class Router:
# Referenced by stream_with_fallbacks via closure - assigned here, before
# the generator body ever runs, so the reference resolves fine despite
# being defined textually after the function that captures it.
wrapper: Final = FallbackAwareAnthropicMessagesStream(stream_with_fallbacks(), source_iterator)
from litellm.proxy.common_utils.anthropic_stream_diagnostics import AnthropicStreamDiagnostics
diagnostics: Final = AnthropicStreamDiagnostics.from_env("router_emit", initial_kwargs.get("litellm_call_id"))
stream: Final = stream_with_fallbacks()
wrapper: Final = FallbackAwareAnthropicMessagesStream(
diagnostics.relay(stream) if diagnostics is not None else stream, source_iterator
)
return wrapper
async def _aanthropic_messages_recover_stream_error(

229
tests/integration/sdk/test_anthropic_messages_bridge_mid_stream_failure_sdk.py
View file

@ -3,8 +3,11 @@ import json
import threading
import uuid
from collections.abc import AsyncIterable, Callable
from itertools import chain
from typing import Final
import httpx
import pytest
from integration._support.anthropic_sse import (
SseEvent,
delta_text,
@ -39,12 +42,14 @@ class _CallbackProbe(CustomLogger):
self._lock: Final = threading.Lock()
self._failures: Final[list[object]] = []
self._successes: Final[list[object]] = []
self._failure_logs: Final[list[object]] = []
async def async_log_failure_event(
self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
with self._lock:
self._failures.append(kwargs.get("exception"))
self._failure_logs.append(kwargs.get("standard_logging_object"))
async def async_log_success_event(
self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object
@ -60,6 +65,10 @@ class _CallbackProbe(CustomLogger):
with self._lock:
return tuple(self._successes)
def failure_logs(self) -> tuple[object, ...]:
with self._lock:
return tuple(self._failure_logs)
def _upstream(marker: str) -> Callable[[Request], Reply]:
def respond(request: Request) -> Reply:
@ -156,3 +165,223 @@ async def test_sdk_native_messages_stream_delivers_one_success_callback_with_usa
assert (logging["prompt_tokens"], logging["completion_tokens"], logging["total_tokens"]) == (5, 3, 8)
assert probe.failures() == ()
assert posted_targets(wire) == ("/v1/messages",)
@pytest.mark.parametrize("ending", ("complete", "eof", "drop", "cancel"))
async def test_native_http_stream_partial_usage_has_one_failed_result(
ending: str, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
import logging
monkeypatch.setenv("LITELLM_ANTHROPIC_STREAM_DIAGNOSTICS", "all")
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
probe: Final = _CallbackProbe()
litellm.callbacks.append(probe)
call_id: Final = str(uuid.uuid4())
model: Final = "claude-sonnet-5-5"
frames: Final = message_stream("msg_synthetic_partial", model, _TEXT)
gate: Final = threading.Event()
def respond(request: Request) -> Reply:
assert (request.method, request.target) == ("POST", "/v1/messages")
assert request.headers["x-api-key"] == _PROVIDER_KEY
assert json.loads(request.body) == {
"model": model,
"messages": [{"role": "user", "content": "synthetic prompt"}],
"stream": True,
"max_tokens": 16,
}
if ending == "complete":
return stream_reply(tuple(bytes((byte,)) for byte in b"".join(frames)))
if ending == "cancel":
return Reply(content_type="text/event-stream", chunks=(frames[0], frames[-1]), gate_after_first=gate)
if ending == "drop":
return dropping_reply((frames[0], frames[-1]), abort_after=1)
return stream_reply((frames[0][:13], frames[0][13:]))
with wire_server(respond) as wire:
stream: Final = await litellm.anthropic.messages.acreate(
model=f"anthropic/{model}",
api_base=wire.url,
api_key=_PROVIDER_KEY,
max_tokens=16,
stream=True,
messages=[{"role": "user", "content": "synthetic prompt"}],
litellm_call_id=call_id,
)
from litellm.llms.anthropic.pass_through.messages.streaming_iterator import (
AnthropicMessagesStreamingResponse,
)
assert isinstance(stream, AnthropicMessagesStreamingResponse)
try:
if ending == "drop":
with pytest.raises(
httpx.TransportError,
match=r"(incomplete|closed|payload|transfer)",
):
async for _ in stream:
pass
elif ending == "cancel":
assert b"message_start" in await stream.__anext__()
await stream.aclose()
else:
output: Final = b"".join([chunk async for chunk in stream])
assert (b"message_stop" in output) is (ending == "complete")
finally:
await stream.aclose()
gate.set()
await _settled(probe.successes if ending == "complete" else probe.failures)
await GLOBAL_LOGGING_WORKER.flush()
assert len(probe.successes()) + len(probe.failures()) == 1
if ending == "complete":
assert probe.failures() == ()
else:
assert probe.successes() == ()
log: Final = TypeAdapter(dict[str, object]).validate_python(probe.failure_logs()[0])
assert log["status"] == "failure" and log["prompt_tokens"] == 5
summaries: Final = tuple(
record.getMessage()
for record in caplog.records
if record.getMessage().startswith("anthropic_stream_summary ")
)
assert len(summaries) == 1
assert call_id in summaries[0] and "msg_synthetic_partial" in summaries[0]
assert _PROVIDER_KEY not in summaries[0] and "synthetic prompt" not in summaries[0]
assert posted_targets(wire) == ("/v1/messages",)
@pytest.mark.parametrize("transport", ("text", "bytes"))
@pytest.mark.parametrize("newline", (b"\n", b"\r\n", b"\r"), ids=("lf", "crlf", "cr"))
@pytest.mark.parametrize("split_at", ("header", "json", "delimiter"))
async def test_native_fragmented_http_frames_are_not_corrupted_by_proxy_keepalives(
transport: str, newline: bytes, split_at: str
) -> None:
from collections.abc import AsyncGenerator
from anthropic._streaming import SSEDecoder
from litellm.proxy.common_utils.sse_keepalive import (
ANTHROPIC_PING_SSE_CHUNK,
SSE_COMMENT_PING_BYTES,
wrap_passthrough_sse_bytes_with_keepalive_pings,
wrap_sse_stream_with_keepalive_pings,
)
model: Final = "claude-sonnet-5-5"
frames: Final = tuple(
frame.replace(b"\n", newline) for frame in message_stream("msg_synthetic_keepalive", model, _TEXT)
)
def fragments(frame: bytes) -> tuple[bytes, ...]:
boundary: Final = {
"header": 13,
"json": frame.index(b"data:") + len(b"data:") + 12,
"delimiter": len(frame) - 1,
}[split_at]
return frame[:boundary], frame[boundary:]
chunks: Final = (frames[0], *chain.from_iterable(fragments(frame) for frame in frames[1:]))
gate: Final = threading.Event()
def respond(request: Request) -> Reply:
assert (request.method, request.target) == ("POST", "/v1/messages")
assert json.loads(request.body) == {
"model": model,
"messages": [{"role": "user", "content": "synthetic"}],
"stream": True,
"max_tokens": 16,
}
return Reply(content_type="text/event-stream", chunks=chunks, gate_after_first=gate, pause_between_chunks=0.03)
with wire_server(respond) as wire:
response: Final = await litellm.anthropic.messages.acreate(
model=f"anthropic/{model}",
api_base=wire.url,
api_key=_PROVIDER_KEY,
max_tokens=16,
stream=True,
messages=[{"role": "user", "content": "synthetic"}],
)
assert isinstance(response, AsyncIterable)
async def raw() -> AsyncGenerator[bytes, None]:
async for chunk in response:
assert isinstance(chunk, bytes)
yield chunk
gate.set()
async def serialized() -> AsyncGenerator[str, None]:
async for chunk in raw():
yield chunk.decode()
received: Final[bytes]
if transport == "bytes":
byte_stream: Final = wrap_passthrough_sse_bytes_with_keepalive_pings(
raw(), 0.005, {"content-type": "text/event-stream"}
)
received = b"".join([chunk async for chunk in byte_stream])
else:
text_stream: Final = wrap_sse_stream_with_keepalive_pings(serialized(), 0.005)
received = "".join([chunk async for chunk in text_stream]).encode()
ping: Final = SSE_COMMENT_PING_BYTES if transport == "bytes" else ANTHROPIC_PING_SSE_CHUNK.encode()
assert ping in received
events: Final = tuple(SSEDecoder().iter_bytes(iter((received,))))
for event in events:
assert TypeAdapter(dict[str, object]).validate_json(event.data)["type"] == event.event
expected: Final = tuple((event.event, event.data) for event in SSEDecoder().iter_bytes(iter(frames)))
assert tuple((event.event, event.data) for event in events if event.event != "ping") == expected
assert received.replace(ping, b"") == b"".join(frames)
@pytest.mark.parametrize("mode", ("all", "errors", "off"))
async def test_routed_stream_diagnostics_correlate_provider_and_router_without_content(
mode: str, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
import logging
monkeypatch.setenv("LITELLM_ANTHROPIC_STREAM_DIAGNOSTICS", mode)
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
model: Final = "claude-sonnet-5-5"
call_id: Final = str(uuid.uuid4())
frames: Final = message_stream("msg_synthetic_routed", model, "SENSITIVE_RESPONSE")
def respond(request: Request) -> Reply:
assert request.target == "/v1/messages"
return stream_reply(tuple(piece for frame in frames for piece in (frame[:13], frame[13:])))
with wire_server(respond) as wire:
router: Final = litellm.Router(
model_list=[
{
"model_name": "diagnostic-test",
"litellm_params": {"model": f"anthropic/{model}", "api_base": wire.url, "api_key": _PROVIDER_KEY},
}
],
num_retries=1,
)
response: Final = await router.aanthropic_messages(
model="diagnostic-test",
max_tokens=16,
stream=True,
messages=[{"role": "user", "content": "SENSITIVE_PROMPT"}],
litellm_call_id=call_id,
)
assert isinstance(response, AsyncIterable)
received: Final = b"".join([chunk async for chunk in response])
assert received == b"".join(frames)
await GLOBAL_LOGGING_WORKER.flush()
summaries: Final = tuple(
record.getMessage()
for record in caplog.records
if record.getMessage().startswith("anthropic_stream_summary ")
)
assert len(summaries) == (2 if mode == "all" else 0)
for summary in summaries:
value: Final = TypeAdapter(dict[str, object]).validate_json(summary.split(" ", 1)[1])
assert value["call_id"] == call_id
assert value["text_deltas"] == 1 and value["complete"] is True
assert value["message_id"] == "msg_synthetic_routed"
assert "SENSITIVE_PROMPT" not in summary and "SENSITIVE_RESPONSE" not in summary
assert _PROVIDER_KEY not in summary

24
tests/unit/integrations/open_telemetry/test_passthrough_parent_span.py
View file

@ -130,18 +130,16 @@ def _model_response() -> ModelResponse:
# Real Anthropic SSE stream (single text block) reused for the streaming path.
STREAM_CHUNKS = [
"event: message_start",
'data: {"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","model":"claude-sonnet-4-5-20250929","content":[],"stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":17,"output_tokens":5}}}',
"event: content_block_start",
'data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}',
"event: content_block_delta",
'data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello world"}}',
"event: content_block_stop",
'data: {"type":"content_block_stop","index":0}',
"event: message_delta",
'data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":2}}',
"event: message_stop",
'data: {"type":"message_stop"}',
"event: message_start\n"
'data: {"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","model":"claude-sonnet-4-5-20250929","content":[],"stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":17,"output_tokens":5}}}\n\n',
"event: content_block_start\n"
'data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}\n\n',
"event: content_block_delta\n"
'data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello world"}}\n\n',
'event: content_block_stop\ndata: {"type":"content_block_stop","index":0}\n\n',
"event: message_delta\n"
'data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":2}}\n\n',
'event: message_stop\ndata: {"type":"message_stop"}\n\n',
]
@ -231,7 +229,7 @@ def test_streaming_passthrough_links_to_server_root(
logging_obj, _kwargs, start_time = _build_logging_obj_wired_to_root(
root, stream=True
)
raw_bytes = ["\n".join(STREAM_CHUNKS).encode("utf-8")]
raw_bytes = ["".join(STREAM_CHUNKS).encode("utf-8")]
end_time = datetime.now()
asyncio.run(
PassThroughStreamingHandler._route_streaming_logging_to_handler(

19
tests/unit/llms/anthropic/pass_through/messages/test_streaming_iterator.py
View file

@ -857,10 +857,14 @@ class _SuccessRecorder(CustomLogger):
def __init__(self):
super().__init__()
self.success_kwargs: list = []
self.failure_kwargs: list = []
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
self.success_kwargs.append(kwargs)
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
self.failure_kwargs.append(kwargs)
def _make_priced_logging_obj(
call_id: str, recorder: _SuccessRecorder, model: str, start_time: datetime | None = None
@ -874,6 +878,7 @@ def _make_priced_logging_obj(
litellm_call_id=call_id,
function_id=call_id,
dynamic_async_success_callbacks=[recorder],
dynamic_async_failure_callbacks=[recorder],
)
logging_obj.update_environment_variables(
model=model,
@ -960,7 +965,7 @@ async def test_client_disconnect_partial_billing_prices_recovered_tokens(monkeyp
"""
Regression (LIT-6872): a client disconnect that lands on partial billing
re-tokenizes the buffered text into completion_tokens, but the logged cost
stayed priced at the message_start placeholder (1 output token). The success
stayed priced at the message_start placeholder (1 output token). The failure
row's response_cost must match its recovered completion_tokens.
"""
import litellm
@ -1004,8 +1009,10 @@ async def test_client_disconnect_partial_billing_prices_recovered_tokens(monkeyp
assert len(enqueued) == 1, "client disconnect never reached partial billing"
await enqueued[0]
assert len(recorder.success_kwargs) == 1
logged = recorder.success_kwargs[0]["standard_logging_object"]
assert recorder.success_kwargs == []
assert len(recorder.failure_kwargs) == 1
logged = recorder.failure_kwargs[0]["standard_logging_object"]
assert logged["status"] == "failure"
assert 1 < logged["completion_tokens"] < 1500
prompt_cost, completion_cost = litellm.cost_per_token(
model=model, prompt_tokens=29, completion_tokens=logged["completion_tokens"]
@ -1061,8 +1068,10 @@ async def test_proxy_disconnect_closing_upstream_prices_recovered_tokens():
assert len(enqueued) == 1, "closing the upstream never reached partial billing"
await enqueued[0]
assert len(recorder.success_kwargs) == 1
logged = recorder.success_kwargs[0]["standard_logging_object"]
assert recorder.success_kwargs == []
assert len(recorder.failure_kwargs) == 1
logged = recorder.failure_kwargs[0]["standard_logging_object"]
assert logged["status"] == "failure"
assert logged["completion_tokens"] > 1
prompt_cost, completion_cost = litellm.cost_per_token(
model=model, prompt_tokens=29, completion_tokens=logged["completion_tokens"]

285
tests/unit/proxy/common_utils/test_anthropic_stream_diagnostics.py Normal file
View file

@ -0,0 +1,285 @@
import json
import logging
from collections.abc import AsyncGenerator
from typing import Final
import pytest
from litellm.proxy.common_utils.anthropic_stream_diagnostics import AnthropicStreamDiagnostics, StreamStage
_START: Final = b'data:{"type":"message_start","message":{"id":"msg_synthetic"}}\r\n\r\n'
_TEXT: Final = (
b'event:content_block_delta\r\ndata:{"type":"content_block_delta",'
b'"delta":{"type":"text_delta","text":"PRIVATE_PROMPT sk-synthetic-secret"}}\r\n\r\n'
)
_STOP: Final = b'data:{"type":"message_stop"}\r\n\r\n'
class _Clock:
def __init__(self) -> None:
self.seconds = 100.0
def __call__(self) -> float:
return self.seconds
def advance(self, seconds: float) -> None:
self.seconds += seconds
@pytest.mark.parametrize("mode", ("off", "", "invalid"))
def test_diagnostics_disabled_creates_no_observer(mode: str, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("LITELLM_ANTHROPIC_STREAM_DIAGNOSTICS", mode)
assert AnthropicStreamDiagnostics.from_env("provider_read", "call_synthetic") is None
def test_diagnostics_default_is_off(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("LITELLM_ANTHROPIC_STREAM_DIAGNOSTICS", raising=False)
assert AnthropicStreamDiagnostics.from_env("provider_read", "call_synthetic") is None
def test_only_allowlisted_request_id_header_is_logged(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
monkeypatch.setenv("LITELLM_ANTHROPIC_STREAM_DIAGNOSTICS", "all")
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
observer: Final = AnthropicStreamDiagnostics.from_env(
"provider_read",
"call_synthetic",
{"x-request-id": "upstream_request_1", "authorization": "sk-private", "x-api-key": "secret-value"},
)
assert observer is not None
observer.finish("eof")
assert "upstream_request_1" in caplog.text
assert "sk-private" not in caplog.text and "secret-value" not in caplog.text
@pytest.mark.parametrize("mode,complete,expected", (("errors", True, 0), ("errors", False, 1), ("all", True, 1)))
def test_summary_is_bounded_and_excludes_content(
mode: str, complete: bool, expected: int, caplog: pytest.LogCaptureFixture
) -> None:
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
observer: Final = AnthropicStreamDiagnostics(mode, "provider_read", "call_synthetic", clock=lambda: 1.0)
wire: Final = _START + _TEXT + (_STOP if complete else b"")
for byte in wire:
observer.observe(bytes((byte,)))
assert not caplog.records
observer.finish("eof" if complete else "closed")
assert len(caplog.records) == expected
assert "PRIVATE_PROMPT" not in caplog.text and "sk-synthetic-secret" not in caplog.text
if not expected:
return
summary: Final = json.loads(caplog.records[0].getMessage().split(" ", 1)[1])
assert (summary["stage"], summary["call_id"], summary["message_id"]) == (
"provider_read",
"call_synthetic",
"msg_synthetic",
)
assert (summary["starts"], summary["text_deltas"], summary["stops"]) == (1, 1, int(complete))
assert (summary["chunks"], summary["bytes"], summary["pending_bytes"]) == (len(wire), len(wire), 0)
assert summary["complete"] is complete
assert summary["first_text_ms"] == 0
assert len(caplog.records[0].getMessage()) < 1024
def test_oversize_unfinished_frame_is_bounded_and_next_frame_is_observed(caplog: pytest.LogCaptureFixture) -> None:
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
observer: Final = AnthropicStreamDiagnostics("all", "provider_read", "call_synthetic", clock=lambda: 1.0)
observer.observe(_START)
for _ in range(100):
observer.observe(b"x" * 65536)
assert len(observer.pending) <= 65536
observer.observe(b"\r")
observer.observe(b"\n\r")
observer.observe(b"\n" + _TEXT + _STOP)
observer.finish("eof")
summary: Final = json.loads(caplog.records[0].getMessage().split(" ", 1)[1])
assert (summary["unparsed_frames"], summary["text_deltas"], summary["stops"]) == (1, 1, 1)
assert len(caplog.records) == 1
def test_empty_text_delta_does_not_count_as_body_received(caplog: pytest.LogCaptureFixture) -> None:
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
observer: Final = AnthropicStreamDiagnostics("all", "provider_read", "call_synthetic", clock=lambda: 1.0)
observer.observe(
_START + b'data:{"type":"content_block_delta","delta":{"type":"text_delta","text":""}}\n\n' + _STOP
)
observer.finish("eof")
summary: Final = json.loads(caplog.records[0].getMessage().split(" ", 1)[1])
assert (summary["text_deltas"], summary["text_chars"], summary["first_text_ms"]) == (1, 0, None)
@pytest.mark.parametrize(
"stage,header", (("provider_read", "X-TC-RequestId"), ("http_send", "LLM_Provider-X-TC-RequestId"))
)
def test_tencent_request_id_survives_header_casing_and_proxy_prefix(
stage: StreamStage, header: str, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
monkeypatch.setenv("LITELLM_ANTHROPIC_STREAM_DIAGNOSTICS", "slow")
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
observer: Final = AnthropicStreamDiagnostics.from_env(
stage, "call_synthetic", {header: "tc-request-synthetic", "authorization": "PRIVATE_KEY"}
)
assert observer is not None
observer.observe(_START)
observer.finish("closed")
observer.finish("closed")
assert len(caplog.records) == 1
summary: Final = json.loads(caplog.records[0].getMessage().split(" ", 1)[1])
assert summary["upstream_request_id"] == "tc-request-synthetic"
assert "PRIVATE_KEY" not in caplog.text
def test_event_timestamps_distinguish_reasoning_from_text_without_logging_either(
caplog: pytest.LogCaptureFixture,
) -> None:
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
clock: Final = _Clock()
observer: Final = AnthropicStreamDiagnostics(
"all", "provider_read", "call_synthetic", clock=clock, wall_clock=lambda: 1_000.0
)
clock.advance(2)
observer.observe(_START)
clock.advance(3)
observer.observe(
b'data:{"type":"content_block_delta","delta":{"type":"thinking_delta","thinking":"PRIVATE_REASONING"}}\n\n'
)
clock.advance(7)
observer.observe(_TEXT[:10])
clock.advance(1)
observer.observe(_TEXT[10:])
observer.observe(_STOP)
observer.finish("eof")
summary: Final = json.loads(caplog.records[0].getMessage().split(" ", 1)[1])
assert (summary["started_at_unix_ms"], summary["started_monotonic_ms"]) == (1_000_000, 100_000)
assert (
summary["first_event_ms"],
summary["first_content_ms"],
summary["first_text_ms"],
summary["last_content_ms"],
) == (2_000, 5_000, 13_000, 13_000)
assert (summary["first_event_type"], summary["last_event_type"], summary["max_content_gap_ms"]) == (
"message_start",
"message_stop",
8_000,
)
assert "PRIVATE_REASONING" not in caplog.text and "PRIVATE_PROMPT" not in caplog.text
@pytest.mark.parametrize("gap,expected", ((2, 0), (31, 1)))
def test_slow_mode_logs_content_gaps_not_total_generation_length(
gap: int, expected: int, caplog: pytest.LogCaptureFixture
) -> None:
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
clock: Final = _Clock()
observer: Final = AnthropicStreamDiagnostics("slow", "provider_read", "call_synthetic", clock=clock)
observer.observe(_START)
for _ in range(50):
clock.advance(gap)
observer.observe(_TEXT)
observer.observe(_STOP)
observer.finish("eof")
observer.finish("closed")
assert len(caplog.records) == expected
def test_unknown_events_and_invalid_ids_cannot_leak_payload_into_summary(caplog: pytest.LogCaptureFixture) -> None:
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
observer: Final = AnthropicStreamDiagnostics(
"all", "provider_read", "sk-secret", upstream_request_id="PRIVATE\nPROMPT"
)
observer.observe(
b'data:{"type":"PRIVATE_EVENT","message":{"id":"sk-private"},"error":{"message":"PRIVATE_ERROR"}}\n\n'
)
observer.finish("eof")
summary: Final = json.loads(caplog.records[0].getMessage().split(" ", 1)[1])
assert (summary["call_id"], summary["upstream_request_id"], summary["message_id"]) == (None, None, None)
assert (summary["first_event_type"], summary["last_event_type"]) == ("other", "other")
assert "PRIVATE" not in caplog.text and "sk-secret" not in caplog.text
assert len(caplog.records[0].getMessage()) < 2048
@pytest.mark.parametrize("frame", (b"data:\n\n", b"data: \r\n\r\n", b"data\r\r", b"data: \ndata:\n\n"))
def test_empty_data_is_reported_even_when_the_stream_completes(frame: bytes, caplog: pytest.LogCaptureFixture) -> None:
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
observer: Final = AnthropicStreamDiagnostics("errors", "provider_read", "call_synthetic", clock=lambda: 1.0)
for byte in _START + frame + _STOP:
observer.observe(bytes((byte,)))
observer.finish("eof")
assert len(caplog.records) == 1
summary: Final = json.loads(caplog.records[0].getMessage().split(" ", 1)[1])
assert (summary["empty_data_frames"], summary["unparsed_frames"], summary["complete"]) == (1, 0, False)
assert (summary["starts"], summary["stops"], summary["pending_bytes"]) == (1, 1, 0)
def test_comment_heartbeats_do_not_make_a_completed_stream_fail(caplog: pytest.LogCaptureFixture) -> None:
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
observer: Final = AnthropicStreamDiagnostics("errors", "provider_read", "call_synthetic", clock=lambda: 1.0)
observer.observe(_START + b": ping\n\n" + _TEXT + b": heartbeat\r\n\r\n" + _STOP)
observer.finish("eof")
assert not caplog.records
@pytest.mark.parametrize("delimiter", (b"\n\n", b"\r\n\r\n", b"\r\r"))
def test_large_fragmented_frames_and_split_delimiters_preserve_event_counts(
delimiter: bytes, caplog: pytest.LogCaptureFixture
) -> None:
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
observer: Final = AnthropicStreamDiagnostics("all", "provider_read", "call_synthetic", clock=lambda: 1.0)
prefix: Final = b'data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"'
text: Final = b"x" * 32_768
observer.observe(_START)
for byte in prefix + text + b'"}}' + delimiter:
observer.observe(bytes((byte,)))
observer.observe(_STOP)
observer.finish("eof")
summary: Final = json.loads(caplog.records[0].getMessage().split(" ", 1)[1])
assert (summary["events"], summary["text_chars"], summary["pending_bytes"], summary["complete"]) == (
3,
len(text),
0,
True,
)
@pytest.mark.parametrize("frame", (b"data: broken JSON\n\n", b"data: " + b"x" * 65_536 + b"\n\n"))
def test_unparseable_complete_event_is_reported_without_blocking_later_events(
frame: bytes, caplog: pytest.LogCaptureFixture
) -> None:
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
observer: Final = AnthropicStreamDiagnostics("errors", "provider_read", "call_synthetic", clock=lambda: 1.0)
observer.observe(_START + frame + _STOP)
observer.finish("eof")
summary: Final = json.loads(caplog.records[0].getMessage().split(" ", 1)[1])
assert (summary["unparsed_frames"], summary["stops"], summary["complete"]) == (1, 1, False)
@pytest.mark.asyncio
@pytest.mark.parametrize("transport_error", (False, True))
async def test_relay_preserves_bytes_and_reports_eof_or_transport_failure_once(
transport_error: bool, caplog: pytest.LogCaptureFixture
) -> None:
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
observer: Final = AnthropicStreamDiagnostics("all", "router_emit", "call_synthetic", clock=lambda: 1.0)
async def source() -> AsyncGenerator[bytes, None]:
yield _START
if transport_error:
raise OSError("PRIVATE_ERROR")
yield _STOP
relay: Final = observer.relay(source())
assert await anext(relay) == _START
if transport_error:
with pytest.raises(OSError, match="PRIVATE_ERROR"):
await anext(relay)
else:
assert await anext(relay) == _STOP
with pytest.raises(StopAsyncIteration):
await anext(relay)
observer.finish("closed")
observer.observe(b"ignored after finish")
assert len(caplog.records) == 1
summary: Final = json.loads(caplog.records[0].getMessage().split(" ", 1)[1])
assert summary["end"] == ("transport_error" if transport_error else "eof")
assert summary["complete"] is not transport_error
assert "PRIVATE_ERROR" not in caplog.text

32
tests/unit/proxy/common_utils/test_sse_keepalive.py
View file

@ -6,7 +6,6 @@ import pytest
from fastapi.responses import StreamingResponse
from litellm.proxy.common_request_processing import create_response
from litellm.types.utils import ModelResponse
from litellm.proxy.common_utils.sse_keepalive import (
ANTHROPIC_PING_SSE_CHUNK,
SSE_COMMENT_PING_BYTES,
@ -17,6 +16,7 @@ from litellm.proxy.common_utils.sse_keepalive import (
wrap_passthrough_sse_bytes_with_keepalive_pings,
wrap_sse_stream_with_keepalive_pings,
)
from litellm.types.utils import ModelResponse
MESSAGE_START_CHUNK: Final = 'data: {"type": "message_start"}\n\n'
TEXT_DELTA_CHUNK: Final = 'data: {"type": "content_block_delta"}\n\n'
@ -120,6 +120,32 @@ async def test_aclose_mid_silence_cancels_upstream_and_runs_its_cleanup():
assert upstream_cleaned_up.is_set()
@pytest.mark.asyncio
async def test_cancelling_a_read_inside_partial_json_closes_upstream():
reading: Final = asyncio.Event()
cleaned_up: Final = asyncio.Event()
partial: Final = 'event: content_block_delta\ndata: {"type":"content_block_delta","index"'
async def partial_stream() -> AsyncGenerator[str, None]:
try:
yield partial
reading.set()
await asyncio.Event().wait()
finally:
cleaned_up.set()
wrapped: Final = wrap_sse_stream_with_keepalive_pings(partial_stream(), 1)
assert await wrapped.__anext__() == partial
pending: Final = asyncio.create_task(wrapped.__anext__())
await reading.wait()
pending.cancel()
with pytest.raises(asyncio.CancelledError):
await pending
await wrapped.aclose()
assert cleaned_up.is_set()
@pytest.mark.asyncio
async def test_non_positive_interval_returns_stream_unwrapped():
async def any_stream() -> AsyncGenerator[str, None]:
@ -404,7 +430,5 @@ def _deployment(keepalive_seconds=..., model="openai/gpt-4o"):
([_deployment()], None, None, "nothing anywhere leaves it off"),
],
)
def test_ttft_interval_resolves_through_the_deployments_it_could_land_on(
deployments, global_interval, expected, why
):
def test_ttft_interval_resolves_through_the_deployments_it_could_land_on(deployments, global_interval, expected, why):
assert resolve_ttft_keepalive_interval(deployments, global_interval) == expected, why

11
tests/unit/proxy/pass_through_endpoints/test_streaming_handler.py
View file

@ -177,9 +177,14 @@ async def test_interrupted_anthropic_stream_recovers_output_tokens_off_the_event
warm_tokenizer(model)
logging_obj = _logging_obj()
logging_obj.model_call_details = {"model": model, "stream": True}
logging_obj.model = model
logging_obj.litellm_params = {}
logging_obj.get_router_model_id.return_value = None
logging_obj.dispatch_success_handlers = AsyncMock()
logging_obj.dispatch_failure_handlers = AsyncMock()
logging_obj.dispatch_attempt_failure_handlers = AsyncMock(
side_effect=logging_obj.dispatch_failure_handlers
)
_, took, lags = await timed_with_loop_lags(
lambda: PassThroughStreamingHandler._route_streaming_logging_to_handler(
@ -195,8 +200,10 @@ async def test_interrupted_anthropic_stream_recovers_output_tokens_off_the_event
)
)
logging_obj.dispatch_success_handlers.assert_awaited_once()
logged_usage = logging_obj.dispatch_success_handlers.await_args.kwargs["result"].usage
logging_obj.dispatch_success_handlers.assert_not_awaited()
logging_obj.dispatch_attempt_failure_handlers.assert_awaited_once()
logging_obj.dispatch_failure_handlers.assert_awaited_once()
logged_usage = logging_obj.record_partial_usage_for_failure.call_args.kwargs["usage"]
assert logged_usage.completion_tokens > 100_000
assert_loop_stayed_free(took, lags)

120
tests/unit/proxy/pass_through_endpoints/test_streaming_handler_interrupt.py
View file

@ -32,6 +32,7 @@ from litellm.types.passthrough_endpoints.pass_through_endpoints import EndpointT
def _make_streaming_response(chunks):
mock = MagicMock(spec=httpx.Response)
mock.status_code = 200
mock.headers = httpx.Headers()
async def _aiter_bytes():
for c in chunks:
@ -667,6 +668,7 @@ def _anthropic_sse(event: str, payload: dict) -> bytes:
def _anthropic_stream_that_times_out_mid_stream():
mock = MagicMock(spec=httpx.Response)
mock.status_code = 200
mock.headers = httpx.Headers()
async def _aiter_bytes():
yield _anthropic_sse(
@ -757,6 +759,7 @@ async def test_chunk_processor_logs_failure_not_success_on_mid_stream_exception(
def _anthropic_stream_dropping_after(frames: tuple[bytes, ...]) -> httpx.Response:
mock: Final = MagicMock(spec=httpx.Response)
mock.status_code = 200
mock.headers = httpx.Headers()
async def _aiter_bytes() -> AsyncIterator[bytes]:
for frame in frames:
@ -889,7 +892,11 @@ def _anthropic_completed_stream_tail() -> tuple[bytes, ...]:
_anthropic_sse("content_block_stop", {"type": "content_block_stop", "index": 0}),
_anthropic_sse(
"message_delta",
{"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": None}, "usage": {"output_tokens": 3}},
{
"type": "message_delta",
"delta": {"stop_reason": "end_turn", "stop_sequence": None},
"usage": {"output_tokens": 3},
},
),
_anthropic_sse("message_stop", {"type": "message_stop"}),
)
@ -909,6 +916,7 @@ def _anthropic_overloaded_error_frame() -> bytes:
def _anthropic_stream_of(frames: tuple[bytes, ...]):
mock = MagicMock(spec=httpx.Response)
mock.status_code = 200
mock.headers = httpx.Headers()
async def _aiter_bytes():
for frame in frames:
@ -918,11 +926,91 @@ def _anthropic_stream_of(frames: tuple[bytes, ...]):
return mock
@pytest.mark.asyncio
@pytest.mark.parametrize("ending", ("eof", "close", "cancel", "complete"))
@pytest.mark.parametrize("split_at", (1, 17, 67, -3))
async def test_initial_three_output_tokens_do_not_make_an_interrupted_stream_successful(
ending: str, split_at: int
) -> None:
recorder = _EventRecorder()
logging_obj = LiteLLMLoggingObj(
model="claude-sonnet-5",
messages=[{"role": "user", "content": "synthetic"}],
stream=True,
call_type="anthropic_messages",
start_time=datetime.now(),
litellm_call_id="test-initial-three-tokens",
function_id="test-initial-three-tokens",
dynamic_async_success_callbacks=[recorder],
dynamic_async_failure_callbacks=[recorder],
)
start = _anthropic_message_start_frame().replace(b'"output_tokens": 1', b'"output_tokens": 3')
tail = _anthropic_completed_stream_tail() if ending == "complete" else ()
stream = PassThroughStreamingHandler.chunk_processor(
response=_anthropic_stream_of((start[:split_at], start[split_at:], *tail)),
request_body={"model": "claude-sonnet-5", "stream": True},
litellm_logging_obj=logging_obj,
endpoint_type=EndpointType.ANTHROPIC,
start_time=datetime.now(),
passthrough_success_handler_obj=MagicMock(),
url_route="/v1/messages",
)
assert await stream.__anext__() + await stream.__anext__() == start
if ending == "close":
await stream.aclose()
elif ending == "cancel":
with pytest.raises(asyncio.CancelledError):
await stream.athrow(asyncio.CancelledError())
else:
assert tuple([chunk async for chunk in stream]) == tail
await GLOBAL_LOGGING_WORKER.flush()
events = recorder.success_kwargs if ending == "complete" else recorder.failure_kwargs
assert len(events) == 1
assert len(recorder.success_kwargs) + len(recorder.failure_kwargs) == 1
logged = events[0]["standard_logging_object"]
assert (logged["prompt_tokens"], logged["completion_tokens"]) == (52, 3)
assert logged["status"] == ("success" if ending == "complete" else "failure")
if ending != "complete":
assert events[0]["exception"].status_code == (499 if ending in ("close", "cancel") else 502)
@pytest.mark.asyncio
async def test_anthropic_empty_http_200_stream_is_logged_as_failure() -> None:
recorder = _EventRecorder()
logging_obj = LiteLLMLoggingObj(
model="claude-sonnet-5-5",
messages=[{"role": "user", "content": "synthetic"}],
stream=True,
call_type="anthropic_messages",
start_time=datetime.now(),
litellm_call_id="test-empty-stream",
function_id="test-empty-stream",
dynamic_async_success_callbacks=[recorder],
dynamic_async_failure_callbacks=[recorder],
)
stream = PassThroughStreamingHandler.chunk_processor(
response=_anthropic_stream_of(()),
request_body={"model": "claude-sonnet-5-5", "stream": True},
litellm_logging_obj=logging_obj,
endpoint_type=EndpointType.ANTHROPIC,
start_time=datetime.now(),
passthrough_success_handler_obj=MagicMock(),
url_route="/v1/messages",
)
assert tuple([chunk async for chunk in stream]) == ()
await GLOBAL_LOGGING_WORKER.flush()
assert recorder.success_kwargs == []
assert len(recorder.failure_kwargs) == 1
assert recorder.failure_kwargs[0]["exception"].status_code == 502
@pytest.mark.parametrize(
"frames, billed",
[
pytest.param(
(_anthropic_message_start_frame(), _anthropic_overloaded_error_frame()), True, id="error-after-message-start"
(_anthropic_message_start_frame(), _anthropic_overloaded_error_frame()),
True,
id="error-after-message-start",
),
pytest.param((_anthropic_overloaded_error_frame(),), False, id="error-frame-only"),
],
@ -977,9 +1065,7 @@ async def test_chunk_processor_logs_a_provider_error_frame_as_a_failure_when_clo
@pytest.mark.asyncio
async def test_chunk_processor_logs_a_stream_that_ended_at_the_provider_error_frame_through_the_success_route():
"""A provider error frame that ends a stream nobody closed early is the request's final answer, and that
stream keeps logging the way it always has: once, through the success route, with the usage it carried."""
async def test_chunk_processor_logs_a_stream_that_ended_at_the_provider_error_frame_as_failure():
recorder = _EventRecorder()
logging_obj = LiteLLMLoggingObj(
model="claude-sonnet-5",
@ -1005,21 +1091,17 @@ async def test_chunk_processor_logs_a_stream_that_ended_at_the_provider_error_fr
url_route="/v1/messages",
)
]
await _settle(recorder.success_kwargs)
await _settle(recorder.failure_kwargs)
assert received == list(frames)
assert recorder.failure_kwargs == []
success_payload = recorder.success_kwargs[0]["standard_logging_object"]
assert success_payload["status"] == "success"
assert success_payload["prompt_tokens"] == 52
assert logging_obj.model_call_details["prompt_cache_response_complete"] is False
assert recorder.success_kwargs == []
failure_payload = recorder.failure_kwargs[0]["standard_logging_object"]
assert failure_payload["status"] == "failure"
assert failure_payload["prompt_tokens"] == 52
@pytest.mark.asyncio
async def test_chunk_processor_bills_a_stream_the_client_closed_at_the_provider_error_frame():
"""The Anthropic SDK raises on a provider `event: error` frame and drops the connection, so a non-retried
stream is closed at the frame by the client, not by a retry taking over. Its usage must still reach the
success route, or the tokens the provider billed never reach the key's spend."""
recorder = _EventRecorder()
logging_obj = LiteLLMLoggingObj(
model="claude-sonnet-5",
@ -1045,12 +1127,12 @@ async def test_chunk_processor_bills_a_stream_the_client_closed_at_the_provider_
if b"event: error" in chunk:
break
await stream.aclose()
await _settle(recorder.success_kwargs)
await _settle(recorder.failure_kwargs)
assert recorder.failure_kwargs == []
success_payload = recorder.success_kwargs[0]["standard_logging_object"]
assert success_payload["status"] == "success"
assert success_payload["prompt_tokens"] == 52
assert recorder.success_kwargs == []
failure_payload = recorder.failure_kwargs[0]["standard_logging_object"]
assert failure_payload["status"] == "failure"
assert failure_payload["prompt_tokens"] == 52
@pytest.mark.asyncio

6
tests/unit/proxy/pass_through_endpoints/test_success_handler.py
View file

@ -52,6 +52,7 @@ async def test_chunk_processor_yields_raw_bytes(endpoint_type, url_route):
# Mock inputs
response = AsyncMock(spec=httpx.Response)
response.status_code = 200
response.headers = httpx.Headers()
raw_chunks = [
b'{"id": "1", "content": "Hello"}',
b'{"id": "2", "content": "World"}',
@ -145,7 +146,10 @@ async def test_route_streaming_logging_runs_async_handler_for_sdk_passthrough():
request_body={},
endpoint_type=EndpointType.ANTHROPIC,
start_time=datetime.now(),
raw_bytes=[],
raw_bytes=[
b'event: message_start\ndata: {"type":"message_start","message":{"id":"msg_test"}}\n\n',
b'event: message_stop\ndata: {"type":"message_stop"}\n\n',
],
end_time=datetime.now(),
)

145
tests/unit/proxy/test_common_request_processing.py
View file

@ -2,6 +2,7 @@ import asyncio
import copy
import datetime
import json
import logging
from collections.abc import AsyncIterable, Mapping
from types import MappingProxyType, SimpleNamespace
from typing import AsyncGenerator, Callable, Final, Iterator, Literal, Optional, Sequence
@ -13,6 +14,8 @@ import httpx
import pytest
from fastapi import HTTPException, Request, Response, status
from fastapi.responses import JSONResponse, StreamingResponse
from starlette.requests import ClientDisconnect
from starlette.types import Message, Scope
import litellm
from litellm._uuid import uuid
@ -61,6 +64,7 @@ from litellm.proxy.common_request_processing import (
create_response,
sse_error_payload,
)
from litellm.proxy.common_utils.anthropic_stream_diagnostics import AnthropicStreamDiagnostics
from litellm.proxy.common_utils.callback_utils import add_guardrail_to_applied_guardrails_header
from litellm.proxy.common_utils.proxy_rate_limit_error import ProxyRateLimitError
from litellm.proxy.common_utils.sse_keepalive import ANTHROPIC_PING_SSE_CHUNK
@ -12055,6 +12059,147 @@ class TestResponseCostFromLoggingObjForCacheHits:
assert cost == ""
class _DiagnosticClock:
def __init__(self) -> None:
self.seconds = 100.0
def __call__(self) -> float:
return self.seconds
@pytest.mark.asyncio
@pytest.mark.parametrize("disconnect", (False, True))
async def test_anthropic_stream_diagnostics_measure_asgi_send_not_just_generator_yield(
disconnect: bool, caplog: pytest.LogCaptureFixture
) -> None:
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
clock: Final = _DiagnosticClock()
diagnostics: Final = AnthropicStreamDiagnostics("slow", "http_send", "synthetic-call", clock=clock)
body: Final = (
'event: message_start\ndata: {"type":"message_start","message":{"id":"msg_synthetic"}}\n\n'
'event: content_block_delta\ndata: {"type":"content_block_delta",'
'"delta":{"type":"text_delta","text":"PRIVATE_OUTPUT"}}\n\n'
'event: message_stop\ndata: {"type":"message_stop"}\n\n'
)
closed: Final = asyncio.Event()
async def source() -> AsyncGenerator[str, None]:
try:
yield body
finally:
closed.set()
async def receive() -> Message:
pytest.fail("ASGI 2.4 does not use the receive-side disconnect listener")
async def send(message: Message) -> None:
if message["type"] == "http.response.start":
clock.seconds = 101.0
if message["type"] == "http.response.body" and message.get("body"):
assert diagnostics.counts.text_chars == 0
assert message["body"] == body.encode()
clock.seconds = 104.0
if disconnect:
raise OSError("PRIVATE_ERROR")
response: Final = _UpstreamClosingStreamingResponse(source(), stream_diagnostics=diagnostics)
scope: Final[Scope] = {"type": "http", "asgi": {"spec_version": "2.4"}}
if disconnect:
with pytest.raises(ClientDisconnect):
await response(scope, receive, send)
else:
await response(scope, receive, send)
assert closed.is_set()
assert len(caplog.records) == 1
summary: Final = json.loads(caplog.records[0].getMessage().split(" ", 1)[1])
assert (summary["stage"], summary["send_attempts"], summary["sends_completed"], summary["send_failed"]) == (
"http_send",
1,
int(not disconnect),
disconnect,
)
assert summary["complete"] is not disconnect
assert summary["text_chars"] == (0 if disconnect else len("PRIVATE_OUTPUT"))
assert summary["first_send_attempt_ms"] == 1000
assert summary["last_send_completed_ms"] == (None if disconnect else 4000)
assert summary["max_send_ms"] == (0 if disconnect else 3000)
assert summary["send_in_flight_ms"] == (3000 if disconnect else None)
assert "PRIVATE_OUTPUT" not in caplog.text and "PRIVATE_ERROR" not in caplog.text
@pytest.mark.asyncio
@pytest.mark.parametrize("enabled", (False, True))
async def test_create_response_diagnostics_preserve_sse_and_provider_request_id(
enabled: bool, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
monkeypatch.setenv("LITELLM_ANTHROPIC_STREAM_DIAGNOSTICS", "all")
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
frame: Final = 'event: ping\ndata: {"type":"ping"}\n\n'
async def source() -> AsyncGenerator[str, None]:
yield frame
async def receive() -> Message:
pytest.fail("ASGI 2.4 does not use the receive-side disconnect listener")
async def send(message: Message) -> None:
if message["type"] == "http.response.body" and message.get("body"):
assert message["body"] == frame.encode()
response: Final = await create_response(
source(),
"text/event-stream",
{"x-litellm-call-id": "synthetic-call", "llm_provider-x-tc-requestid": "tc-synthetic"},
diagnose_anthropic_stream=enabled,
)
scope: Final[Scope] = {"type": "http", "asgi": {"spec_version": "2.4"}}
await response(scope, receive, send)
assert len(caplog.records) == int(enabled)
if enabled:
summary: Final = json.loads(caplog.records[0].getMessage().split(" ", 1)[1])
assert (summary["call_id"], summary["upstream_request_id"], summary["pings"]) == (
"synthetic-call",
"tc-synthetic",
1,
)
@pytest.mark.asyncio
async def test_anthropic_stream_diagnostics_log_a_cancelled_send_once(caplog: pytest.LogCaptureFixture) -> None:
caplog.set_level(logging.INFO, logger="LiteLLM Proxy")
sending: Final = asyncio.Event()
closed: Final = asyncio.Event()
diagnostics: Final = AnthropicStreamDiagnostics("slow", "http_send", "synthetic-call", clock=lambda: 100.0)
async def source() -> AsyncGenerator[str, None]:
try:
yield 'event: ping\ndata: {"type":"ping"}\n\n'
finally:
closed.set()
async def receive() -> Message:
await sending.wait()
return {"type": "http.disconnect"}
async def send(message: Message) -> None:
if message["type"] == "http.response.body" and message.get("body"):
sending.set()
await asyncio.Event().wait()
response: Final = _UpstreamClosingStreamingResponse(source(), stream_diagnostics=diagnostics)
scope: Final[Scope] = {"type": "http", "asgi": {"spec_version": "2.3"}}
await response(scope, receive, send)
assert closed.is_set()
assert len(caplog.records) == 1
summary: Final = json.loads(caplog.records[0].getMessage().split(" ", 1)[1])
assert (summary["end"], summary["send_failed"], summary["send_attempts"], summary["sends_completed"]) == (
"closed",
True,
1,
0,
)
def test_maybe_get_model_id_reads_the_chat_metadata_the_router_updated_in_place():
processing_obj: Final = ProxyBaseLLMRequestProcessing(data={"metadata": {"model_info": {"id": "declared-text"}}})

27
tests/unit/test_router/test_router.py
View file

@ -17058,6 +17058,33 @@ async def test_anthropic_messages_streaming_iterator_flushes_buffered_lifecycle_
assert collected == [_anthropic_messages_message_start_chunk(), _anthropic_messages_content_chunk("hi"), message_stop]
@pytest.mark.asyncio
@pytest.mark.parametrize("separator", (b"\n", b"\r\n", b"\r"))
@pytest.mark.parametrize("space", (b" ", b""))
@pytest.mark.parametrize("split_at", range(1, 28))
async def test_anthropic_content_split_across_reads_reaches_client_before_upstream_ends(
separator: bytes, space: bytes, split_at: int
) -> None:
router = _anthropic_messages_make_router()
start = _anthropic_messages_message_start_chunk()
delta = _anthropic_messages_content_chunk("hello").replace(b": ", b":" + space).replace(b"\n", separator)
chunks = (start, delta[:split_at], delta[split_at:])
async def source() -> AsyncIterator[bytes]:
for chunk in chunks:
yield chunk
pytest.fail("the router read past complete content instead of forwarding it")
wrapped = await router._aanthropic_messages_streaming_iterator(
response=source(), initial_kwargs={"model": "primary"}
)
try:
received = tuple([await wrapped.__anext__() for _ in chunks])
assert received == chunks
finally:
await wrapped.aclose()
@pytest.mark.asyncio
async def test_anthropic_messages_streaming_iterator_flushes_buffered_frames_on_stream_end():
"""Regression: if the primary stream ends with only lifecycle frames and