From 9892a69f119a754817ba98e36863c1f0a14bf0eb Mon Sep 17 00:00:00 2001 From: yyobject <48043998+yyobject@users.noreply.github.com> Date: Tue, 20 Oct 2026 19:57:13 +0800 Subject: [PATCH 1/3] fix(anthropic): handle fragmented SSE and log interrupted streams as failures --- .../messages/streaming_iterator.py | 21 +- litellm/proxy/common_request_processing.py | 38 ++- .../anthropic_stream_diagnostics.py | 299 ++++++++++++++++++ litellm/proxy/common_utils/sse_keepalive.py | 8 +- .../pass_through_endpoints/architecture.md | 67 +++- .../anthropic_passthrough_logging_handler.py | 13 +- .../streaming_handler.py | 97 +++++- litellm/router.py | 12 +- ..._messages_bridge_mid_stream_failure_sdk.py | 229 ++++++++++++++ .../test_passthrough_parent_span.py | 24 +- .../messages/test_streaming_iterator.py | 19 +- .../test_anthropic_stream_diagnostics.py | 285 +++++++++++++++++ .../proxy/common_utils/test_sse_keepalive.py | 32 +- .../test_streaming_handler.py | 7 +- .../test_streaming_handler_interrupt.py | 119 +++++-- .../test_success_handler.py | 6 +- .../proxy/test_common_request_processing.py | 145 +++++++++ tests/unit/test_router/test_router.py | 27 ++ 18 files changed, 1376 insertions(+), 72 deletions(-) create mode 100644 litellm/proxy/common_utils/anthropic_stream_diagnostics.py create mode 100644 tests/unit/proxy/common_utils/test_anthropic_stream_diagnostics.py diff --git a/litellm/llms/anthropic/pass_through/messages/streaming_iterator.py b/litellm/llms/anthropic/pass_through/messages/streaming_iterator.py index 44db193ea83..eb7b1f4e62a 100644 --- a/litellm/llms/anthropic/pass_through/messages/streaming_iterator.py +++ b/litellm/llms/anthropic/pass_through/messages/streaming_iterator.py @@ -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, diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index 24a7b46da4b..5d69f39606a 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -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 @@ -90,6 +90,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, @@ -966,9 +967,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: @@ -976,9 +979,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) @@ -1150,6 +1175,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. @@ -1270,6 +1296,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 + ), ) @@ -3110,6 +3143,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 diff --git a/litellm/proxy/common_utils/anthropic_stream_diagnostics.py b/litellm/proxy/common_utils/anthropic_stream_diagnostics.py new file mode 100644 index 00000000000..e0fd120b653 --- /dev/null +++ b/litellm/proxy/common_utils/anthropic_stream_diagnostics.py @@ -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) diff --git a/litellm/proxy/common_utils/sse_keepalive.py b/litellm/proxy/common_utils/sse_keepalive.py index d9685971f52..6241a8a6e5b 100644 --- a/litellm/proxy/common_utils/sse_keepalive.py +++ b/litellm/proxy/common_utils/sse_keepalive.py @@ -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() diff --git a/litellm/proxy/pass_through_endpoints/architecture.md b/litellm/proxy/pass_through_endpoints/architecture.md index f7dd8077ab5..16e015e5047 100644 --- a/litellm/proxy/pass_through_endpoints/architecture.md +++ b/litellm/proxy/pass_through_endpoints/architecture.md @@ -68,4 +68,69 @@ sequenceDiagram - Request body - Response body -- Provider-specific parameters \ No newline at end of file +- 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 diff --git a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py index a7f3a4f84a7..f0352b9f70e 100644 --- a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py +++ b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py @@ -240,11 +240,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: diff --git a/litellm/proxy/pass_through_endpoints/streaming_handler.py b/litellm/proxy/pass_through_endpoints/streaming_handler.py index 1fcfe2e279a..d72f8f93026 100644 --- a/litellm/proxy/pass_through_endpoints/streaming_handler.py +++ b/litellm/proxy/pass_through_endpoints/streaming_handler.py @@ -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 @@ -14,6 +15,11 @@ from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER 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 @@ -37,6 +43,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__( @@ -62,9 +70,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( @@ -152,10 +176,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, @@ -197,7 +229,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 @@ -218,7 +250,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( @@ -227,6 +259,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 @@ -237,14 +270,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( @@ -261,6 +295,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. @@ -268,7 +303,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 @@ -295,10 +334,7 @@ class PassThroughStreamingHandler: @staticmethod def _error_frame_closed_at(endpoint_type: EndpointType, raw_bytes: Sequence[bytes]) -> tuple[str, str, int] | None: - """The provider `event: error` frame of a stream the router closed because a retry or fallback took the - request over. That attempt is logged as its own failure, on a copy taken now so the retry's rebinding of - the shared logging object never reaches it. A stream that ran to its end or that the client closed keeps - logging through the success route, so its usage is still billed.""" + """Capture a superseded attempt's provider error before the retry rebinds the shared logging object.""" from litellm.llms.anthropic.pass_through.messages.streaming_iterator import ( closing_superseded_attempt, parse_anthropic_error_event, @@ -333,6 +369,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 @@ -349,6 +389,41 @@ 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: + await AnthropicPassthroughLoggingHandler.log_error_frame_as_failure( + attempt_logging_obj=litellm_logging_obj, + request_body=request_mapping, + all_chunks=raw_bytes, + error_event=error_event, + ) + 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_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), + ), + "", + prefer_async_handlers=True, + ) + 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): diff --git a/litellm/router.py b/litellm/router.py index aadbdaa8520..a98ee73a75e 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -6270,8 +6270,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. @@ -6280,6 +6278,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 @@ -6337,7 +6337,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( diff --git a/tests/integration/sdk/test_anthropic_messages_bridge_mid_stream_failure_sdk.py b/tests/integration/sdk/test_anthropic_messages_bridge_mid_stream_failure_sdk.py index 98c42220aab..4c7e257d98c 100644 --- a/tests/integration/sdk/test_anthropic_messages_bridge_mid_stream_failure_sdk.py +++ b/tests/integration/sdk/test_anthropic_messages_bridge_mid_stream_failure_sdk.py @@ -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 diff --git a/tests/unit/integrations/open_telemetry/test_passthrough_parent_span.py b/tests/unit/integrations/open_telemetry/test_passthrough_parent_span.py index bd37b3e3c76..acc0e2e193a 100644 --- a/tests/unit/integrations/open_telemetry/test_passthrough_parent_span.py +++ b/tests/unit/integrations/open_telemetry/test_passthrough_parent_span.py @@ -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( diff --git a/tests/unit/llms/anthropic/pass_through/messages/test_streaming_iterator.py b/tests/unit/llms/anthropic/pass_through/messages/test_streaming_iterator.py index b995c6dba9b..acbac74e97a 100644 --- a/tests/unit/llms/anthropic/pass_through/messages/test_streaming_iterator.py +++ b/tests/unit/llms/anthropic/pass_through/messages/test_streaming_iterator.py @@ -854,10 +854,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) -> LiteLLMLoggingObj: logging_obj = LiteLLMLoggingObj( @@ -869,6 +873,7 @@ def _make_priced_logging_obj(call_id: str, recorder: _SuccessRecorder, model: st 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, @@ -908,7 +913,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 @@ -952,8 +957,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"] @@ -1009,8 +1016,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"] diff --git a/tests/unit/proxy/common_utils/test_anthropic_stream_diagnostics.py b/tests/unit/proxy/common_utils/test_anthropic_stream_diagnostics.py new file mode 100644 index 00000000000..4c5ba279b50 --- /dev/null +++ b/tests/unit/proxy/common_utils/test_anthropic_stream_diagnostics.py @@ -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 diff --git a/tests/unit/proxy/common_utils/test_sse_keepalive.py b/tests/unit/proxy/common_utils/test_sse_keepalive.py index 228fd5bcae6..5c383744c9a 100644 --- a/tests/unit/proxy/common_utils/test_sse_keepalive.py +++ b/tests/unit/proxy/common_utils/test_sse_keepalive.py @@ -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 diff --git a/tests/unit/proxy/pass_through_endpoints/test_streaming_handler.py b/tests/unit/proxy/pass_through_endpoints/test_streaming_handler.py index da084146894..694e9b2487d 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_streaming_handler.py +++ b/tests/unit/proxy/pass_through_endpoints/test_streaming_handler.py @@ -177,9 +177,11 @@ 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() _, took, lags = await timed_with_loop_lags( lambda: PassThroughStreamingHandler._route_streaming_logging_to_handler( @@ -195,8 +197,9 @@ 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_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) diff --git a/tests/unit/proxy/pass_through_endpoints/test_streaming_handler_interrupt.py b/tests/unit/proxy/pass_through_endpoints/test_streaming_handler_interrupt.py index 8ce41129aec..b027993392e 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_streaming_handler_interrupt.py +++ b/tests/unit/proxy/pass_through_endpoints/test_streaming_handler_interrupt.py @@ -28,6 +28,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: @@ -663,6 +664,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( @@ -782,7 +784,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"}), ) @@ -802,6 +808,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: @@ -811,11 +818,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"), ], @@ -872,9 +959,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", @@ -900,21 +985,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", @@ -940,12 +1021,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 diff --git a/tests/unit/proxy/pass_through_endpoints/test_success_handler.py b/tests/unit/proxy/pass_through_endpoints/test_success_handler.py index 2cc03949ff1..39ef04cb3e3 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_success_handler.py +++ b/tests/unit/proxy/pass_through_endpoints/test_success_handler.py @@ -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(), ) diff --git a/tests/unit/proxy/test_common_request_processing.py b/tests/unit/proxy/test_common_request_processing.py index f837dc3fb5f..95f3c06f3e4 100644 --- a/tests/unit/proxy/test_common_request_processing.py +++ b/tests/unit/proxy/test_common_request_processing.py @@ -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 @@ -11791,3 +11795,144 @@ 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, + ) diff --git a/tests/unit/test_router/test_router.py b/tests/unit/test_router/test_router.py index 4a9dd199c08..5248e58c219 100644 --- a/tests/unit/test_router/test_router.py +++ b/tests/unit/test_router/test_router.py @@ -16855,6 +16855,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 From 6bd10f8adf7708aefc1278200e907cff10dc7f18 Mon Sep 17 00:00:00 2001 From: yyobject <48043998+yyobject@users.noreply.github.com> Date: Tue, 27 Oct 2026 13:06:01 +0800 Subject: [PATCH 2/3] fix(anthropic): preserve attempt failure logging after main update --- .../pass_through_endpoints/streaming_handler.py | 15 +++++++++------ .../test_streaming_handler_interrupt.py | 1 + 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/litellm/proxy/pass_through_endpoints/streaming_handler.py b/litellm/proxy/pass_through_endpoints/streaming_handler.py index 0feac6dc5ec..afc2d400fa8 100644 --- a/litellm/proxy/pass_through_endpoints/streaming_handler.py +++ b/litellm/proxy/pass_through_endpoints/streaming_handler.py @@ -408,11 +408,15 @@ class PassThroughStreamingHandler: request_mapping: Final = _REQUEST_BODY.validate_python(request_body) error_event: Final = parse_anthropic_error_event(complete_frames) if error_event is not None: - await AnthropicPassthroughLoggingHandler.log_error_frame_as_failure( - attempt_logging_obj=litellm_logging_obj, + 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, - all_chunks=raw_bytes, - error_event=error_event, + 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"]: @@ -424,7 +428,7 @@ class PassThroughStreamingHandler: consumer_closed: Final = ( litellm_logging_obj.model_call_details.get("anthropic_stream_end") == "closed" ) - await litellm_logging_obj.dispatch_failure_handlers( + await litellm_logging_obj.dispatch_attempt_failure_handlers( litellm.APIError( status_code=499 if consumer_closed else 502, message=( @@ -436,7 +440,6 @@ class PassThroughStreamingHandler: model=str(litellm_logging_obj.model), ), "", - prefer_async_handlers=True, ) return # TinyFish billing is owned by the detached poller; the $0 fallback below is only for streams with no run_id diff --git a/tests/unit/proxy/pass_through_endpoints/test_streaming_handler_interrupt.py b/tests/unit/proxy/pass_through_endpoints/test_streaming_handler_interrupt.py index ebdc8ea8326..32d092c2ff7 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_streaming_handler_interrupt.py +++ b/tests/unit/proxy/pass_through_endpoints/test_streaming_handler_interrupt.py @@ -759,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: From 647a962f41a5710e4bd3c2bd6777b9aae18909b5 Mon Sep 17 00:00:00 2001 From: yyobject <48043998+yyobject@users.noreply.github.com> Date: Tue, 27 Oct 2026 13:16:11 +0800 Subject: [PATCH 3/3] test(anthropic): cover attempt-scoped interruption logging --- .../proxy/pass_through_endpoints/test_streaming_handler.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/unit/proxy/pass_through_endpoints/test_streaming_handler.py b/tests/unit/proxy/pass_through_endpoints/test_streaming_handler.py index d943b3cc433..bee4fec4771 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_streaming_handler.py +++ b/tests/unit/proxy/pass_through_endpoints/test_streaming_handler.py @@ -182,6 +182,9 @@ async def test_interrupted_anthropic_stream_recovers_output_tokens_off_the_event 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( @@ -198,6 +201,7 @@ async def test_interrupted_anthropic_stream_recovers_output_tokens_off_the_event ) 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