diff --git a/litellm/exceptions.py b/litellm/exceptions.py index f8c6a50a589..61810f650af 100644 --- a/litellm/exceptions.py +++ b/litellm/exceptions.py @@ -1289,7 +1289,10 @@ class MidStreamFallbackError(ServiceUnavailableError): self.model = model self.llm_provider = llm_provider self.original_exception = original_exception - self.litellm_debug_info = litellm_debug_info + inherited_debug_info: Final[object] = getattr(original_exception, "litellm_debug_info", None) + self.litellm_debug_info = litellm_debug_info or ( + inherited_debug_info if isinstance(inherited_debug_info, str) else None + ) self.max_retries = max_retries self.num_retries = num_retries self.generated_content = generated_content diff --git a/litellm/litellm_core_utils/exception_mapping_utils.py b/litellm/litellm_core_utils/exception_mapping_utils.py index 16cd076db39..338902aa919 100644 --- a/litellm/litellm_core_utils/exception_mapping_utils.py +++ b/litellm/litellm_core_utils/exception_mapping_utils.py @@ -3,6 +3,7 @@ import json import re import traceback from collections.abc import Mapping +from dataclasses import dataclass from types import MappingProxyType from typing import Final, Protocol, cast @@ -2514,15 +2515,113 @@ _HOST_WITH_OPTIONAL_PORT: Final = re.compile( ) +def _router_metadata(params: Mapping[str, object]) -> Mapping[str, object]: + metadata: Final = as_str_mapping(params.get("metadata")) or _NO_LITELLM_PARAMS + litellm_metadata: Final = as_str_mapping(params.get("litellm_metadata")) or _NO_LITELLM_PARAMS + return MappingProxyType({**metadata, **litellm_metadata}) + + +def _deployment_api_base(metadata: Mapping[str, object]) -> str | None: + api_base: Final = metadata.get("api_base") + return api_base if isinstance(api_base, str) and api_base else None + + def with_deployment_api_base(litellm_params: object) -> Mapping[str, object]: params: Final = as_str_mapping(litellm_params) or _NO_LITELLM_PARAMS - metadata: Final = as_str_mapping(params.get("metadata")) - deployment_api_base: Final = None if metadata is None else metadata.get("api_base") - if not isinstance(deployment_api_base, str) or not deployment_api_base: + deployment_api_base: Final = _deployment_api_base(_router_metadata(params)) + if deployment_api_base is None: return params return MappingProxyType({**params, "api_base": deployment_api_base}) +_DEBUG_INFO_ATTRIBUTE: Final = "litellm_debug_info" +_DEBUG_INFO_CALL_ID_ATTRIBUTE: Final = "_litellm_debug_info_call_id" +_RAISED_BY_SEVERAL_CALLS: Final = object() + + +@dataclass(frozen=True, slots=True) +class _RequestDebugInfo: + with_prompt: str + without_prompt: str + call_id: str + + +_NO_REQUEST_DEBUG_INFO: Final = _RequestDebugInfo(with_prompt="", without_prompt="", call_id="") + + +def _request_debug_info(model: object, extra_kwargs: object, completion_kwargs: object) -> _RequestDebugInfo: + params: Final = as_str_mapping(extra_kwargs) or _NO_LITELLM_PARAMS + try: + metadata: Final = _router_metadata(params) + api_base: Final = litellm.get_api_base(model=model, optional_params=params) or _deployment_api_base(metadata) + messages: Final = litellm.get_first_chars_messages( + kwargs=as_str_mapping(completion_kwargs) or _NO_LITELLM_PARAMS + ) + shown_api_base: Final = _api_base_without_credentials(api_base) if api_base else "" + head: Final = f"\nModel: {model}" + (f"\nAPI Base: `{shown_api_base}`" if shown_api_base else "") + prompt: Final = ( + f"\nMessages: `{messages}`" if messages and litellm.redact_messages_in_exceptions is False else "" + ) + labelled_values: Final = ( + ("model_group", metadata.get("model_group")), + ("deployment", metadata.get("deployment")), + ("vertex_project", params.get("vertex_project")), + ("vertex_location", params.get("vertex_location")), + ) + tail: Final = "".join(f"\n{label}: `{value}`\n" for label, value in labelled_values if value is not None) + call_id: Final = params.get("litellm_call_id") + return _RequestDebugInfo( + with_prompt=add_key_name_and_team_to_alert(request_info=head + prompt + tail, metadata=metadata), + without_prompt=add_key_name_and_team_to_alert(request_info=head + tail, metadata=metadata), + call_id=call_id if isinstance(call_id, str) else "", + ) + except Exception: + return _NO_REQUEST_DEBUG_INFO + + +def _attach_debug_info(exception: Exception, request_debug_info: _RequestDebugInfo) -> None: + attached_call_id: Final = getattr(exception, _DEBUG_INFO_CALL_ID_ATTRIBUTE, None) + if attached_call_id is _RAISED_BY_SEVERAL_CALLS: + return + if ( + isinstance(attached_call_id, str) + and attached_call_id + and request_debug_info.call_id + and attached_call_id != request_debug_info.call_id + ): + setattr(exception, _DEBUG_INFO_ATTRIBUTE, None) + setattr(exception, _DEBUG_INFO_CALL_ID_ATTRIBUTE, _RAISED_BY_SEVERAL_CALLS) + return + if request_debug_info.without_prompt and getattr(exception, _DEBUG_INFO_ATTRIBUTE, None) is None: + setattr(exception, _DEBUG_INFO_ATTRIBUTE, request_debug_info.without_prompt) + setattr(exception, _DEBUG_INFO_CALL_ID_ATTRIBUTE, request_debug_info.call_id) + + +def mapped_stream_exception( + original_exception: Exception, model: str, custom_llm_provider: str, litellm_params: object +) -> Exception: + try: + return exception_type( + model=model, + custom_llm_provider=custom_llm_provider, + original_exception=original_exception, + completion_kwargs={}, + extra_kwargs=with_deployment_api_base(litellm_params), + ) + except Exception as mapped_exception: + return mapped_exception + + +def with_deployment_debug_info( + error: Exception, model: str, custom_llm_provider: str, litellm_params: object +) -> Exception: + if not (model or custom_llm_provider): + return error + request_debug_info: Final = _request_debug_info(model, with_deployment_api_base(litellm_params), _NO_LITELLM_PARAMS) + _attach_debug_info(error, request_debug_info) + return error + + def _api_base_without_credentials(api_base: str) -> str: scheme, separator, _ = api_base.partition("://") scheme_prefix: Final = scheme + separator if separator and _URL_SCHEME.fullmatch(scheme) else "" @@ -2546,11 +2645,17 @@ def exception_type( model, original_exception, custom_llm_provider, - completion_kwargs={}, - extra_kwargs={}, + completion_kwargs: object = _NO_LITELLM_PARAMS, + extra_kwargs: object = _NO_LITELLM_PARAMS, ): """Maps an LLM Provider Exception to OpenAI Exception Format""" - if any(isinstance(original_exception, exc_type) for exc_type in litellm.LITELLM_EXCEPTION_TYPES): + request_debug_info: Final = ( + _request_debug_info(model, extra_kwargs, completion_kwargs) + if model or custom_llm_provider + else _NO_REQUEST_DEBUG_INFO + ) + if isinstance(original_exception, tuple(litellm.LITELLM_EXCEPTION_TYPES)): + _attach_debug_info(original_exception, request_debug_info) return original_exception if _is_guardrail_block(original_exception): return original_exception @@ -2572,11 +2677,11 @@ def exception_type( print() # noqa: T201 litellm_response_headers: Final = _get_response_headers(original_exception=original_exception) + extra_information: Final = request_debug_info.with_prompt try: error_str = ( redact_secret_string(str(original_exception)) if _ENABLE_SECRET_REDACTION else str(original_exception) ) - extra_information = "" if model or custom_llm_provider: if hasattr(original_exception, "message"): error_str = ( @@ -2589,47 +2694,8 @@ def exception_type( else: exception_type = "" - ################################################################################ - # Common Extra information needed for all providers - # We pass num retries, api_base, vertex_deployment etc to the exception here - ################################################################################ - try: - _api_base: Final = litellm.get_api_base(model=model, optional_params=extra_kwargs) - messages: Final = litellm.get_first_chars_messages(kwargs=completion_kwargs) - _vertex_project: Final = extra_kwargs.get("vertex_project") - _vertex_location: Final = extra_kwargs.get("vertex_location") - _metadata: Final = extra_kwargs.get("metadata", {}) or {} - _model_group: Final = _metadata.get("model_group") - _deployment: Final = _metadata.get("deployment") - extra_information = f"\nModel: {model}" - - if isinstance(custom_llm_provider, str) and len(custom_llm_provider) > 0: - exception_provider = custom_llm_provider[0].upper() + custom_llm_provider[1:] + "Exception" - - shown_api_base: Final = _api_base_without_credentials(_api_base) if _api_base else "" - if shown_api_base: - extra_information += f"\nAPI Base: `{shown_api_base}`" - if messages and len(messages) > 0 and litellm.redact_messages_in_exceptions is False: - extra_information += f"\nMessages: `{messages}`" - - if _model_group is not None: - extra_information += f"\nmodel_group: `{_model_group}`\n" - if _deployment is not None: - extra_information += f"\ndeployment: `{_deployment}`\n" - if _vertex_project is not None: - extra_information += f"\nvertex_project: `{_vertex_project}`\n" - if _vertex_location is not None: - extra_information += f"\nvertex_location: `{_vertex_location}`\n" - - # on litellm proxy add key name + team to exceptions - extra_information = add_key_name_and_team_to_alert(request_info=extra_information, metadata=_metadata) - except Exception: - # DO NOT LET this Block raising the original exception - pass - - ################################################################################ - # End of Common Extra information Needed for all providers - ################################################################################ + if isinstance(custom_llm_provider, str) and len(custom_llm_provider) > 0: + exception_provider = custom_llm_provider[0].upper() + custom_llm_provider[1:] + "Exception" ################################################################################ #################### Start of Provider Exception mapping #################### @@ -2923,17 +2989,20 @@ def exception_type( # don't let an error with mapping interrupt the user from receiving an error from the llm api calls if exception_mapping_worked: setattr(e, "litellm_response_headers", litellm_response_headers) + _attach_debug_info(e, request_debug_info) raise e else: for error_type in litellm.LITELLM_EXCEPTION_TYPES: if isinstance(e, error_type): setattr(e, "litellm_response_headers", litellm_response_headers) + _attach_debug_info(e, request_debug_info) raise e # it's already mapped raised_exc: Final = APIConnectionError( message=f"{original_exception}\n{redact_string(traceback.format_exc())}", llm_provider="", model="", ) + _attach_debug_info(raised_exc, request_debug_info) setattr(raised_exc, "litellm_response_headers", litellm_response_headers) carry_agentic_followup_mark(source=original_exception, mapped=raised_exc) raise raised_exc diff --git a/litellm/llms/anthropic/messages/native/streaming_iterator.py b/litellm/llms/anthropic/messages/native/streaming_iterator.py index 1195652dfd2..5a7633698ec 100644 --- a/litellm/llms/anthropic/messages/native/streaming_iterator.py +++ b/litellm/llms/anthropic/messages/native/streaming_iterator.py @@ -15,6 +15,7 @@ from litellm.constants import ( ANTHROPIC_MESSAGES_STREAM_RELAY_QUEUE_MAXSIZE, ) from litellm.litellm_core_utils.core_helpers import process_response_headers +from litellm.litellm_core_utils.exception_mapping_utils import with_deployment_debug_info from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER from litellm.llms.anthropic.common_utils import ANTHROPIC_ERROR_STATUS_CODE_MAP @@ -563,6 +564,7 @@ class BaseAnthropicMessagesStreamingIterator: start_time=self.start_time, passthrough_success_handler_obj=GLOBAL_PASS_THROUGH_SUCCESS_HANDLER_OBJ, url_route="/v1/messages", + on_upstream_error=self._with_deployment_debug_info, ) def _convert_chunk_to_sse_format(self, chunk: dict | Any) -> bytes: @@ -610,8 +612,9 @@ class BaseAnthropicMessagesStreamingIterator: An upstream failure (Bedrock read / decode / chunk-conversion error) that happens while the client is still connected is forwarded through the queue and re-raised here, so the original provider exception (and - its status) reaches the proxy's failure handling unchanged rather than - being masked by a generic incomplete-stream event. + its status) reaches the proxy's failure handling with the deployment's + debug info attached, rather than being masked by a generic + incomplete-stream event. This method provides the common logic for both Anthropic and Bedrock implementations. """ @@ -802,18 +805,29 @@ class BaseAnthropicMessagesStreamingIterator: """ from litellm.proxy.pass_through_endpoints.streaming_handler import PassThroughStreamingHandler + described: Final = self._with_deployment_debug_info(exc) await PassThroughStreamingHandler.schedule_stream_failure_logging( litellm_logging_obj=self.litellm_logging_obj, endpoint_type=EndpointType.ANTHROPIC, request_body=self.request_body, raw_bytes=collected_chunks, - exception=exc, + exception=described, ) - if not client_detached.is_set() and await self._enqueue_for_client(queue, client_detached, exc): + if not client_detached.is_set() and await self._enqueue_for_client(queue, client_detached, described): await client_detached.wait() - if not _exception_left_unconsumed(queue, exc): + if not _exception_left_unconsumed(queue, described): return - await self._fire_detached_failure_hook(exc) + await self._fire_detached_failure_hook(described) + + def _with_deployment_debug_info(self, exc: Exception) -> Exception: + details: Final = self.litellm_logging_obj.model_call_details + provider: Final = details.get("custom_llm_provider") + return with_deployment_debug_info( + exc, + self.litellm_logging_obj.model, + provider if isinstance(provider, str) and provider else "anthropic", + details.get("litellm_params"), + ) async def _fire_detached_failure_hook(self, exc: Exception) -> None: from litellm._logging import verbose_proxy_logger diff --git a/litellm/main.py b/litellm/main.py index c86a888aa96..239a85cef4c 100644 --- a/litellm/main.py +++ b/litellm/main.py @@ -90,6 +90,7 @@ from litellm.litellm_core_utils.chat_completion_agentic_loop import ( ) from litellm.litellm_core_utils.completion_timeout import CompletionTimeout, StreamKind from litellm.litellm_core_utils.dd_tracing import tracer +from litellm.litellm_core_utils.exception_mapping_utils import mapped_stream_exception from litellm.litellm_core_utils.get_litellm_params import ( AGENTIC_LOOP_CONTROL_KWARGS_KEYS, AGENTIC_LOOP_TOOL_CHOICE_KEY, @@ -1033,7 +1034,7 @@ def _text_completion_response_from_cached_dict(cached_response_dict: Mapping[str return TextCompletionResponse(stream=bool(cached_response_dict.get("stream", False)), **fields) -async def _async_streaming(response, model, custom_llm_provider, args): +async def _async_streaming(response, model, custom_llm_provider, litellm_params: Mapping[str, object]): try: print_verbose(f"received response in _async_streaming: {response}") if asyncio.iscoroutine(response): @@ -1042,11 +1043,11 @@ async def _async_streaming(response, model, custom_llm_provider, args): print_verbose(f"line in async streaming: {line}") yield line except Exception as e: - custom_llm_provider = custom_llm_provider or "openai" - raise exception_type( - model=model, - custom_llm_provider=custom_llm_provider, - original_exception=e, + raise mapped_stream_exception( + e, + model if isinstance(model, str) else "", + custom_llm_provider if isinstance(custom_llm_provider, str) and custom_llm_provider else "openai", + litellm_params, ) @@ -8090,7 +8091,7 @@ async def atext_completion(*args, **kwargs) -> TextCompletionResponse | TextComp response=response, model=model, custom_llm_provider=custom_llm_provider, - args=args, + litellm_params=_REQUEST_KWARGS_ADAPTER.validate_python(kwargs), ), model=model, custom_llm_provider=custom_llm_provider, diff --git a/litellm/proxy/pass_through_endpoints/streaming_handler.py b/litellm/proxy/pass_through_endpoints/streaming_handler.py index 345e17c435d..722e58211ae 100644 --- a/litellm/proxy/pass_through_endpoints/streaming_handler.py +++ b/litellm/proxy/pass_through_endpoints/streaming_handler.py @@ -173,6 +173,7 @@ class PassThroughStreamingHandler: passthrough_success_handler_obj: PassThroughEndpointLogging, url_route: str, route_streaming_logging: RouteStreamingLogging | None = None, + on_upstream_error: Callable[[Exception], object] = lambda _: None, ): resolved_route_streaming_logging: Final[RouteStreamingLogging] = ( route_streaming_logging or PassThroughStreamingHandler.route_streaming_logging_to_handler @@ -271,6 +272,7 @@ class PassThroughStreamingHandler: litellm_logging_obj._deferred_stream_complete_args = (_build_logging_coroutine(),) except Exception as e: verbose_proxy_logger.error("Error in chunk_processor: %s", e) + on_upstream_error(e) if response.status_code < 400: logging_scheduled = True await PassThroughStreamingHandler.schedule_stream_failure_logging( diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index c2191375cb4..347537277df 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -47,7 +47,7 @@ from litellm.litellm_core_utils.core_helpers import ( set_response_cost_in_hidden_params, stamp_litellm_bill_on_usage, ) -from litellm.litellm_core_utils.exception_mapping_utils import with_deployment_api_base +from litellm.litellm_core_utils.exception_mapping_utils import mapped_stream_exception, with_deployment_debug_info from litellm.litellm_core_utils.internal_call_metadata import ( is_background_response, requests_background, @@ -344,27 +344,21 @@ def map_stream_error_to_exception( return _mapped_exception(provider_exception, model, custom_llm_provider, logging_obj) +def _logged_litellm_params(logging_obj: LiteLLMLoggingObj | None) -> object: + return ( + None + if logging_obj is None + else _typed_gets_litellm_params(logging_obj.model_call_details.get)("litellm_params", {}) + ) + + def _mapped_exception( original_exception: Exception, model: str, custom_llm_provider: str, logging_obj: LiteLLMLoggingObj | None, ) -> Exception: - litellm_params: Final = ( - None - if logging_obj is None - else _typed_gets_litellm_params(logging_obj.model_call_details.get)("litellm_params", {}) - ) - try: - return litellm.exception_type( - model=model, - custom_llm_provider=custom_llm_provider, - original_exception=original_exception, - completion_kwargs={}, - extra_kwargs=with_deployment_api_base(litellm_params), - ) - except Exception as mapped_exception: - return mapped_exception + return mapped_stream_exception(original_exception, model, custom_llm_provider, _logged_litellm_params(logging_obj)) def _event_field(event: object, name: str) -> object: @@ -1219,12 +1213,17 @@ class BaseResponsesAPIStreamingIterator: model=self.model or "", ) + def _with_deployment_debug_info(self, error: Exception) -> Exception: + return with_deployment_debug_info( + error, self.model or "", self.custom_llm_provider or "", _logged_litellm_params(self.logging_obj) + ) + def _raise_if_ended_without_terminal_event(self) -> None: if self.completed_response is not None: return if self._replays_stored_stream and not self._yielded_first_chunk: return - error: Final = self._stream_ended_early_error() + error: Final = self._with_deployment_debug_info(self._stream_ended_early_error()) self._handle_failure(error) if self._output_started: raise error @@ -1236,10 +1235,11 @@ class BaseResponsesAPIStreamingIterator: raise _mapped_exception(error, self.model or "", self.custom_llm_provider or "", self.logging_obj) from error def _raise_for_transport_error(self, error: httpx.ReadError | httpx.RemoteProtocolError) -> NoReturn: - self._handle_failure(error) + described: Final = self._with_deployment_debug_info(error) + self._handle_failure(described) if self._output_started: - raise error - raise self._fallback_error(error) from error + raise described + raise self._fallback_error(described) from described async def call_post_streaming_hooks_for_testing( @@ -1341,14 +1341,9 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): self.finished = True raise StopAsyncIteration from e self._raise_for_transport_timeout(e) - except httpx.HTTPError as e: - # Handle HTTP errors - self.finished = True - self._handle_failure(e) - raise e except Exception as e: self.finished = True - self._handle_failure(e) + self._handle_failure(self._with_deployment_debug_info(e)) raise e def _handle_logging_completed_response(self): @@ -1434,14 +1429,9 @@ class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): self.finished = True raise StopIteration from e self._raise_for_transport_timeout(e) - except httpx.HTTPError as e: - # Handle HTTP errors - self.finished = True - self._handle_failure(e) - raise e except Exception as e: self.finished = True - self._handle_failure(e) + self._handle_failure(self._with_deployment_debug_info(e)) raise e def _handle_logging_completed_response(self): diff --git a/litellm/utils.py b/litellm/utils.py index 7012741148c..2820638cd28 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -6583,13 +6583,10 @@ def get_model_region(litellm_params: LiteLLM_Params, mode: str | None) -> str | return None -def get_first_chars_messages(kwargs: dict) -> str: - try: - _messages = kwargs.get("messages") - _messages = str(_messages)[:100] - return _messages - except Exception: - return "" +def get_first_chars_messages(kwargs: object) -> str: + request: Final = as_str_mapping(kwargs) + messages: Final = None if request is None else request.get("messages") + return "" if messages is None else str(messages)[:100] def count_characters(text: str) -> int: diff --git a/tests/integration/observability/test_stream_error_alert_api_base.py b/tests/integration/observability/test_stream_error_alert_api_base.py index d7f325da8c7..b91cf7e3025 100644 --- a/tests/integration/observability/test_stream_error_alert_api_base.py +++ b/tests/integration/observability/test_stream_error_alert_api_base.py @@ -13,7 +13,15 @@ import openai import pytest import yaml from anthropic.types import RawContentBlockDeltaEvent, TextDelta -from integration._support.anthropic_sse import DROP_PAUSE, dropping_reply, stream_reply +from integration._support.anthropic_sse import ( + CONTENT_BLOCK_START, + DROP_PAUSE, + dropping_reply, + error_body, + message_start, + stream_reply, + text_delta, +) from integration._support.client import ( JSON_OBJECT, Gateway, @@ -30,6 +38,7 @@ from integration._support.responses_ws_peer import PROVIDER_MODEL, Script, event from integration._support.tls import write_self_signed_cert from integration._support.vertex import service_account_json from integration._support.wire import Reply, Request, Wire, wire_server +from openai.types import Completion from openai.types.chat import ChatCompletionChunk from openai.types.responses import ResponseFailedEvent, ResponseTextDeltaEvent from pydantic import JsonValue @@ -39,6 +48,7 @@ pytestmark: Final = [pytest.mark.timeout(2 * graceful_stop_seconds() + 120), pyt AZURE_KEY: Final = "synthetic-azure-key" API_VERSION: Final = "2025-04-01-preview" AZURE_CHAT_BACKEND: Final = "azure/gpt-4o-mini" +ANTHROPIC_BACKEND: Final = "anthropic/claude-opus-5-5" PARTIAL_TEXT: Final = "partial answer" ALERT_SECONDS: Final = 40 NO_RETRY_NO_CACHE: Final[Mapping[str, JsonValue]] = MappingProxyType({"num_retries": 0, "cache": {"no-cache": True}}) @@ -234,6 +244,10 @@ def responses_text(item: object) -> str: return item.delta if isinstance(item, ResponseTextDeltaEvent) else "" +def completion_text(item: object) -> str: + return "".join(choice.text for choice in item.choices) if isinstance(item, Completion) else "" + + def failure_of(item: object) -> str: if isinstance(item, Exception): return str(item) @@ -296,6 +310,19 @@ def responses_outcome(gateway: Gateway, key: str, model: str) -> Outcome: return outcome(items, responses_text) +def completions_outcome(gateway: Gateway, key: str, model: str) -> Outcome: + with openai_client(gateway, key) as client: + items: Final = tuple( + items_or_error( + lambda: client.completions.create( + model=model, prompt="stream an answer", stream=True, extra_body=dict(NO_RETRY_NO_CACHE) + ), + openai.APIError, + ) + ) + return outcome(items, completion_text) + + ENTRYPOINTS: Final[Mapping[str, Callable[[Gateway, str, str], Outcome]]] = MappingProxyType( {"chat": chat_outcome, "messages": messages_outcome, "responses-bridge": responses_outcome} ) @@ -480,3 +507,74 @@ def test_fallback_mid_stream_failure_alert_names_the_fallback_deployment(alertin alert: Final = alert_naming(alerting.alerts, marker) assert_names_deployment(alert, backup_wire.url, backup, AZURE_CHAT_BACKEND, caller) assert primary_wire.url not in alert, alert + + +PING_NEVER_SENT: Final = b"event: ping\ndata: {}\n\n" + + +def anthropic_cut_after_text(marker: str) -> Reply: + return dropping_reply( + ( + message_start(f"msg_{marker}", ANTHROPIC_BACKEND), + CONTENT_BLOCK_START, + text_delta(PARTIAL_TEXT), + PING_NEVER_SENT, + ), + abort_after=3, + ) + + +def responses_cut_after_text(marker: str) -> Reply: + identity: Final = f"resp_{marker}" + return dropping_reply( + (frame(created(identity)), frame(delta(identity, PARTIAL_TEXT)), b": never sent\n\n"), abort_after=2 + ) + + +@dataclass(frozen=True, slots=True) +class CutPath: + backend: str + reply: Callable[[str], Reply] + call: Callable[[Gateway, str, str], Outcome] + + +CUT_PATHS: Final[Mapping[str, CutPath]] = MappingProxyType( + { + "anthropic-chat": CutPath(ANTHROPIC_BACKEND, anthropic_cut_after_text, chat_outcome), + "completions": CutPath(AZURE_CHAT_BACKEND, cut_after_text, completions_outcome), + "native-responses": CutPath(AZURE_CHAT_BACKEND, responses_cut_after_text, responses_outcome), + "native-messages": CutPath(ANTHROPIC_BACKEND, anthropic_cut_after_text, messages_outcome), + } +) + + +def scripted_model(scenario: Scenario, backend: str, api_base: str) -> str: + if backend == AZURE_CHAT_BACKEND: + return azure_chat(scenario, api_base) + return scenario.model(model=backend, api_base=api_base, api_key="synthetic-anthropic-key", num_retries=0) + + +@pytest.mark.parametrize("path", list(CUT_PATHS)) +def test_stream_cut_alert_names_the_deployment(alerting: AlertingProxy, path: str) -> None: + marker: Final = uuid.uuid4().hex + cut: Final = CUT_PATHS[path] + with wire_server(chat_upstream(cut.reply, marker)) as wire, alerting.gateway.scenario() as scenario: + model: Final = scripted_model(scenario, cut.backend, wire.url) + caller: Final = keyed_caller(scenario) + seen: Final = cut.call(alerting.gateway, caller.key, model) + assert seen.text == PARTIAL_TEXT and seen.failure, seen + assert_names_deployment(alert_naming(alerting.alerts, model), wire.url, model, cut.backend, caller) + + +def test_anthropic_rejection_before_stream_alert_names_the_deployment(alerting: AlertingProxy) -> None: + marker: Final = uuid.uuid4().hex + + def rejected(_: str) -> Reply: + return Reply(status=500, body=error_body(500, provider_message(marker))) + + with wire_server(chat_upstream(rejected, marker)) as wire, alerting.gateway.scenario() as scenario: + model: Final = scripted_model(scenario, ANTHROPIC_BACKEND, wire.url) + caller: Final = keyed_caller(scenario) + seen: Final = chat_outcome(alerting.gateway, caller.key, model) + assert provider_message(marker) in seen.failure and not seen.text, seen + assert_names_deployment(alert_naming(alerting.alerts, marker), wire.url, model, ANTHROPIC_BACKEND, caller) diff --git a/tests/unit/litellm_core_utils/test_exception_mapping_utils.py b/tests/unit/litellm_core_utils/test_exception_mapping_utils.py index 32a99a37870..bc7827f4f07 100644 --- a/tests/unit/litellm_core_utils/test_exception_mapping_utils.py +++ b/tests/unit/litellm_core_utils/test_exception_mapping_utils.py @@ -20,6 +20,7 @@ from litellm.litellm_core_utils.exception_mapping_utils import ( exception_type, extract_and_raise_litellm_exception, ) +from litellm.llms.anthropic.common_utils import AnthropicError from litellm.llms.base_llm.chat.transformation import BaseLLMException from litellm.llms.bedrock.common_utils import BedrockError from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler @@ -157,6 +158,265 @@ def test_api_base_line_keeps_an_at_sign_in_the_path(api_base: str) -> None: assert f"API Base: `{api_base}`" in _debug_info_for_api_base(api_base) +@pytest.mark.parametrize( + ("provider_error", "mapped_class"), + [ + ( + httpx.ReadError("peer closed connection", request=httpx.Request("POST", "https://relay.example/v1/messages")), + litellm.APIConnectionError, + ), + (RuntimeError("stream decoder failed"), litellm.APIConnectionError), + (AnthropicError(status_code=500, message="[Errno 104] Connection reset by peer"), litellm.InternalServerError), + (AnthropicError(status_code=429, message="rate_limit_error"), litellm.RateLimitError), + ], + ids=["unmapped_with_request", "unmapped_without_request", "anthropic_500", "anthropic_429"], +) +def test_mapped_provider_error_carries_the_deployment_debug_info( + provider_error: Exception, mapped_class: type[Exception] +) -> None: + with pytest.raises(mapped_class) as caught: + exception_type( + model="claude-opus-5-5", + original_exception=provider_error, + custom_llm_provider="anthropic", + completion_kwargs={"messages": [{"role": "user", "content": "CONFIDENTIAL patient record"}]}, + extra_kwargs={ + "api_base": "https://relay.example", + "metadata": {"model_group": "claude-alias", "deployment": "anthropic/claude-opus-5-5"}, + }, + ) + debug_info: Final = caught.value.litellm_debug_info + assert isinstance(debug_info, str) + assert "API Base: `https://relay.example`" in debug_info + assert "model_group: `claude-alias`" in debug_info + assert "deployment: `anthropic/claude-opus-5-5`" in debug_info + assert "CONFIDENTIAL" not in debug_info + + +_ROUTER_METADATA: Final = { + "api_base": "https://relay.example/v1", + "model_group": "claude-alias", + "deployment": "anthropic/claude-haiku-5-5", +} + + +@pytest.mark.parametrize( + ("provider_error", "custom_llm_provider", "extra_kwargs"), + [ + ( + AnthropicError(status_code=529, message="Overloaded"), + "anthropic", + {"litellm_metadata": _ROUTER_METADATA}, + ), + ( + OpenAIError(status_code=500, message="upstream failed"), + "hosted_vllm", + { + "api_base": "https://relay.example/v1", + "metadata": {"user_id": "set by the caller"}, + "litellm_metadata": _ROUTER_METADATA, + }, + ), + ], + ids=["native_messages_without_api_base", "bridged_responses_with_request_metadata"], +) +def test_router_metadata_under_litellm_metadata_names_the_deployment( + provider_error: Exception, custom_llm_provider: str, extra_kwargs: dict[str, object] +) -> None: + with pytest.raises(litellm.InternalServerError) as caught: + exception_type( + model="claude-haiku-5-5", + original_exception=provider_error, + custom_llm_provider=custom_llm_provider, + extra_kwargs=extra_kwargs, + ) + debug_info: Final = caught.value.litellm_debug_info + assert isinstance(debug_info, str) + assert "API Base: `https://relay.example/v1`" in debug_info + assert "model_group: `claude-alias`" in debug_info + assert "deployment: `anthropic/claude-haiku-5-5`" in debug_info + + +_KEY_METADATA: Final = {"user_api_key_alias": "alias-key", "user_api_key_team_alias": "team-a"} + + +@pytest.mark.parametrize( + "extra_kwargs", + [ + {"metadata": _ROUTER_METADATA, "litellm_metadata": _KEY_METADATA}, + {"metadata": _KEY_METADATA, "litellm_metadata": _ROUTER_METADATA}, + ], + ids=["chat_with_guardrail_litellm_metadata", "generic_route_with_key_fields_in_metadata"], +) +def test_debug_info_reads_lines_from_both_metadata_buckets(extra_kwargs: dict[str, object]) -> None: + with pytest.raises(litellm.InternalServerError) as caught: + exception_type( + model="claude-haiku-5-5", + original_exception=AnthropicError(status_code=529, message="Overloaded"), + custom_llm_provider="anthropic", + extra_kwargs=extra_kwargs, + ) + debug_info: Final = caught.value.litellm_debug_info + assert isinstance(debug_info, str) + assert "Key Name: `alias-key`" in debug_info + assert "Team: `team-a`" in debug_info + assert "API Base: `https://relay.example/v1`" in debug_info + assert "model_group: `claude-alias`" in debug_info + assert "deployment: `anthropic/claude-haiku-5-5`" in debug_info + + +def test_litellm_metadata_wins_over_caller_metadata_for_the_same_key() -> None: + with pytest.raises(litellm.InternalServerError) as caught: + exception_type( + model="claude-haiku-5-5", + original_exception=AnthropicError(status_code=529, message="Overloaded"), + custom_llm_provider="anthropic", + extra_kwargs={ + "metadata": {"deployment": "set by the caller", "api_base": "https://caller.example"}, + "litellm_metadata": _ROUTER_METADATA, + }, + ) + debug_info: Final = caught.value.litellm_debug_info + assert isinstance(debug_info, str) + assert "deployment: `anthropic/claude-haiku-5-5`" in debug_info + assert "API Base: `https://relay.example/v1`" in debug_info + assert "caller" not in debug_info + + +def test_mapper_that_passes_its_debug_info_keeps_the_prompt_line() -> None: + request: Final = httpx.Request("POST", "https://gateway.example/v1/chat/completions") + original: Final = OpenAIError( + status_code=500, message="upstream failed", request=request, response=httpx.Response(500, request=request) + ) + with pytest.raises(litellm.InternalServerError) as caught: + exception_type( + model="gpt-5.4-mini", + original_exception=original, + custom_llm_provider="openai", + completion_kwargs={"messages": [{"role": "user", "content": "hello there"}]}, + extra_kwargs={"metadata": {"deployment": "openai/gpt-5.4-mini"}}, + ) + debug_info: Final = caught.value.litellm_debug_info + assert isinstance(debug_info, str) + assert "Messages: `[{'role': 'user', 'content': 'hello there'}]`" in debug_info + assert "deployment: `openai/gpt-5.4-mini`" in debug_info + + +def test_debug_info_has_no_messages_line_when_the_request_has_no_messages() -> None: + assert "Messages" not in _debug_info_for_api_base("https://gateway.example/v1") + + +@pytest.mark.parametrize( + "already_mapped", + [ + litellm.InternalServerError( + message="Upstream generation aborted (finish_reason=abort)", model="gpt-5.4-mini", llm_provider="openai" + ), + litellm.Timeout( + message="Stream exceeded max streaming duration of 10s", model="gpt-5.4-mini", llm_provider="openai" + ), + ], + ids=["finish_reason_abort", "max_streaming_duration"], +) +def test_already_mapped_error_gains_the_deployment_debug_info(already_mapped: Exception) -> None: + returned: Final = exception_type( + model="gpt-5.4-mini", + original_exception=already_mapped, + custom_llm_provider="openai", + completion_kwargs={"messages": [{"role": "user", "content": "CONFIDENTIAL patient record"}]}, + extra_kwargs={ + "api_base": "https://relay.example/v1", + "metadata": {"model_group": "gpt-alias", "deployment": "openai/gpt-5.4-mini"}, + }, + ) + assert returned is already_mapped + debug_info: Final = returned.litellm_debug_info + assert isinstance(debug_info, str) + assert "API Base: `https://relay.example/v1`" in debug_info + assert "deployment: `openai/gpt-5.4-mini`" in debug_info + assert "CONFIDENTIAL" not in debug_info + + +def test_error_instance_raised_by_several_requests_names_none_of_them() -> None: + shared: Final = litellm.RateLimitError(message="mock rate limit", llm_provider="openai", model="gpt-5.4-mini") + + def debug_info_for(alias: str) -> object: + with pytest.raises(litellm.RateLimitError) as caught: + completion( + model="openai/gpt-5.4-mini", + messages=[{"role": "user", "content": "hi"}], + mock_response=shared, + metadata={"user_api_key_alias": alias, "model_group": f"group-{alias}"}, + ) + assert caught.value is shared + return caught.value.litellm_debug_info + + first: Final = debug_info_for("alias-a") + assert isinstance(first, str) and "Key Name: `alias-a`" in first + assert debug_info_for("alias-b") is None + assert debug_info_for("alias-c") is None + + +_RELAY_DEPLOYMENT_KWARGS: Final = { + "api_base": "https://relay.example/v1", + "metadata": {"deployment": "openai/gpt-5.4-mini"}, +} + + +@pytest.mark.parametrize( + ("first_kwargs", "second_kwargs"), + [ + ({"litellm_call_id": "call-1", **_RELAY_DEPLOYMENT_KWARGS}, {}), + (_RELAY_DEPLOYMENT_KWARGS, {"litellm_call_id": "call-1"}), + ], + ids=["second-mapping-has-no-request", "first-mapping-had-no-call-id"], +) +def test_mapping_again_for_the_same_request_keeps_the_deployment_already_named( + first_kwargs: dict[str, object], second_kwargs: dict[str, object] +) -> None: + already_mapped: Final = litellm.InternalServerError( + message="Upstream generation aborted (finish_reason=abort)", model="gpt-5.4-mini", llm_provider="openai" + ) + exception_type( + model="gpt-5.4-mini", + original_exception=already_mapped, + custom_llm_provider="openai", + extra_kwargs=first_kwargs, + ) + exception_type( + model="gpt-5.4-mini", + original_exception=already_mapped, + custom_llm_provider="openai", + completion_kwargs={}, + extra_kwargs=second_kwargs, + ) + debug_info: Final = already_mapped.litellm_debug_info + assert isinstance(debug_info, str) + assert "API Base: `https://relay.example/v1`" in debug_info + assert "deployment: `openai/gpt-5.4-mini`" in debug_info + + +class _ErrorWhoseMessageCannotBeRead(Exception): + @property + def message(self) -> str: + raise RuntimeError("message unavailable") + + +def test_error_that_breaks_the_mapping_still_names_the_deployment() -> None: + with pytest.raises(litellm.APIConnectionError) as caught: + exception_type( + model="claude-opus-5-5", + original_exception=_ErrorWhoseMessageCannotBeRead("upstream failed"), + custom_llm_provider="anthropic", + completion_kwargs={}, + extra_kwargs={"api_base": "https://relay.example", "metadata": {"deployment": "anthropic/claude-opus-5-5"}}, + ) + debug_info: Final = caught.value.litellm_debug_info + assert isinstance(debug_info, str) + assert "API Base: `https://relay.example`" in debug_info + assert "deployment: `anthropic/claude-opus-5-5`" in debug_info + + # Test cases for is_error_str_context_window_exceeded # Tuple format: (error_message, expected_result) context_window_test_cases = [ diff --git a/tests/unit/llms/anthropic/messages/native/test_streaming_iterator.py b/tests/unit/llms/anthropic/messages/native/test_streaming_iterator.py index dd50f60250a..39f7a4a05cb 100644 --- a/tests/unit/llms/anthropic/messages/native/test_streaming_iterator.py +++ b/tests/unit/llms/anthropic/messages/native/test_streaming_iterator.py @@ -645,6 +645,67 @@ async def test_async_sse_wrapper_reraises_upstream_error_to_connected_client(): assert detached_hook.exceptions == [], "the relay re-raised the error, so the proxy failure hook already ran" +_DEPLOYMENT_PARAMS: Final = { + "metadata": { + "api_base": "https://relay.example/anthropic", + "model_group": "claude-alias", + "deployment": "anthropic/claude-opus-5-5", + } +} + + +def _assert_names_the_deployment(error: Exception) -> None: + debug_info: Final = getattr(error, "litellm_debug_info", None) + assert isinstance(debug_info, str) + assert "API Base: `https://relay.example/anthropic`" in debug_info + assert "model_group: `claude-alias`" in debug_info + assert "deployment: `anthropic/claude-opus-5-5`" in debug_info + + +@pytest.mark.asyncio +async def test_relayed_upstream_error_reaches_the_client_naming_the_deployment(): + async def _cut_stream(): + yield {"type": "message_start", "message": {"id": "msg_1", "usage": {"input_tokens": 52, "output_tokens": 1}}} + raise _ProviderStreamError("Response payload is not completed", status_code=529) + + iterator = _RecordingLoggingIterator( + litellm_logging_obj=_make_logging_obj("test_upstream_error_names_the_deployment"), + request_body={}, + ) + iterator.litellm_logging_obj.model_call_details["litellm_params"] = _DEPLOYMENT_PARAMS + + with pytest.raises(_ProviderStreamError) as excinfo: + async for _ in iterator.async_sse_wrapper(_cut_stream()): + pass + + assert excinfo.value.status_code == 529 + _assert_names_the_deployment(excinfo.value) + + +@pytest.mark.asyncio +async def test_native_anthropic_stream_cut_reaches_the_client_naming_the_deployment() -> None: + logging_obj: Final = _make_logging_obj("native_anthropic_stream_cut") + logging_obj.model_call_details["litellm_params"] = _DEPLOYMENT_PARAMS + cut: Final = httpx.ReadError("Response payload is not completed") + + async def _cut_upstream() -> AsyncIterator[bytes]: + yield b'event: message_start\ndata: {"type": "message_start"}\n\n' + raise cut + + stream: Final = AnthropicMessagesConfig().get_async_streaming_response_iterator( + model=logging_obj.model, + httpx_response=httpx.Response(200, content=_cut_upstream()), + request_body={"model": logging_obj.model}, + litellm_logging_obj=logging_obj, + ) + + with pytest.raises(httpx.ReadError) as excinfo: + _ = [chunk async for chunk in stream] + + assert excinfo.value is cut + _assert_names_the_deployment(cut) + + @pytest.mark.asyncio async def test_async_sse_wrapper_logs_failure_on_upstream_error_after_disconnect(): """ diff --git a/tests/unit/responses/test_streaming_iterator.py b/tests/unit/responses/test_streaming_iterator.py index 17a5112ce6f..ec14f67ce17 100644 --- a/tests/unit/responses/test_streaming_iterator.py +++ b/tests/unit/responses/test_streaming_iterator.py @@ -8,6 +8,7 @@ import functools import gc import json import threading +import time import weakref from collections.abc import Callable, Mapping from datetime import datetime @@ -367,6 +368,136 @@ def test_sync_transport_read_timeout_raises_litellm_timeout(prefix): assert "Request timed out. Error_str: ReadTimeout" in str(exc_info.value) +@pytest.mark.asyncio +@pytest.mark.parametrize("prefix", _READ_TIMEOUT_PREFIXES) +@pytest.mark.parametrize("error_class", [httpx.ReadError, httpx.RemoteProtocolError]) +async def test_transport_cut_error_names_the_deployment(prefix, error_class): + logging_obj: Final = _logging_obj_stub() + logging_obj.model_call_details = { + "litellm_params": { + "metadata": { + "api_base": "https://relay.example/openai/v1", + "model_group": "responses-alias", + "deployment": "openai/gpt-4o-mini", + } + } + } + trailing_error: Final = error_class("peer closed connection without sending complete message body") + iterator: Final = _make_iterator(sse_events=prefix, logging_obj=logging_obj, trailing_error=trailing_error) + + with pytest.raises((error_class, MidStreamFallbackError)) as exc_info: + async for _ in iterator: + pass + + debug_info: Final = exc_info.value.litellm_debug_info + assert isinstance(debug_info, str) + assert "API Base: `https://relay.example/openai/v1`" in debug_info + assert "model_group: `responses-alias`" in debug_info + assert "deployment: `openai/gpt-4o-mini`" in debug_info + + +class _FailureDebugInfoRecorder(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.debug_info: object = None + + async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time): + self.debug_info = getattr(kwargs["exception"], "litellm_debug_info", None) + + +@pytest.mark.parametrize("prefix", _READ_TIMEOUT_PREFIXES) +def test_sync_transport_cut_failure_callbacks_see_the_deployment(prefix): + recorder: Final = _FailureDebugInfoRecorder() + logging_obj: Final = _real_logging_obj( + call_type="responses", + litellm_params={ + "metadata": { + "api_base": "https://relay.example/openai/v1", + "model_group": "responses-alias", + "deployment": "openai/gpt-4o-mini", + } + }, + callbacks=(recorder,), + ) + iterator: Final = _make_sync_iterator( + sse_events=prefix, + logging_obj=logging_obj, + trailing_error=httpx.ReadError("peer closed connection without sending complete message body"), + ) + + with pytest.raises((httpx.ReadError, MidStreamFallbackError)): + for _ in iterator: + pass + + assert isinstance(recorder.debug_info, str) + assert "API Base: `https://relay.example/openai/v1`" in recorder.debug_info + assert "model_group: `responses-alias`" in recorder.debug_info + + +@pytest.mark.asyncio +@pytest.mark.parametrize("prefix", _READ_TIMEOUT_PREFIXES) +async def test_stream_ending_without_terminal_event_names_the_deployment(prefix): + logging_obj: Final = _failure_tracking_logging_obj() + logging_obj.model_call_details = { + "litellm_params": { + "metadata": { + "api_base": "https://relay.example/openai/v1", + "model_group": "responses-alias", + "deployment": "openai/gpt-4o-mini", + } + } + } + iterator: Final = _make_iterator(sse_events=prefix, logging_obj=logging_obj) + + with pytest.raises((litellm.APIConnectionError, MidStreamFallbackError)) as exc_info: + async for _ in iterator: + pass + + debug_info: Final = exc_info.value.litellm_debug_info + assert isinstance(debug_info, str) + assert "API Base: `https://relay.example/openai/v1`" in debug_info + assert "deployment: `openai/gpt-4o-mini`" in debug_info + + +_DEPLOYMENT_LITELLM_PARAMS: Final = { + "metadata": { + "api_base": "https://relay.example/openai/v1", + "model_group": "responses-alias", + "deployment": "openai/gpt-4o-mini", + } +} + + +async def _drain_async(iterator: ResponsesAPIStreamingIterator) -> None: + async for _ in iterator: + pass + + +async def _drain_sync(iterator: SyncResponsesAPIStreamingIterator) -> None: + for _ in iterator: + pass + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("make_iterator", "drain"), [(_make_iterator, _drain_async), (_make_sync_iterator, _drain_sync)] +) +async def test_max_streaming_duration_timeout_names_the_deployment(monkeypatch, make_iterator, drain): + monkeypatch.setattr(streaming_module, "LITELLM_MAX_STREAMING_DURATION_SECONDS", 10.0) + logging_obj: Final = _failure_tracking_logging_obj() + logging_obj.model_call_details = {"litellm_params": _DEPLOYMENT_LITELLM_PARAMS} + iterator: Final = make_iterator(sse_events=_PARTIAL_OUTPUT_EVENTS, logging_obj=logging_obj) + iterator._stream_created_time = time.time() - 20 + + with pytest.raises(litellm.Timeout, match="max streaming duration") as exc_info: + await drain(iterator) + + debug_info: Final = exc_info.value.litellm_debug_info + assert isinstance(debug_info, str) + assert "API Base: `https://relay.example/openai/v1`" in debug_info + assert "deployment: `openai/gpt-4o-mini`" in debug_info + + @pytest.mark.asyncio @pytest.mark.parametrize("trailer", [[], [_DONE_MARKER]], ids=["eof", "done-marker"]) async def test_stream_ending_after_partial_output_without_terminal_event_raises(trailer): diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index a1400f53ef8..b5a5c609dae 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -11850,6 +11850,11 @@ async def test_success_enqueue_is_tracked_without_starting_callbacks_synchronous assert logging_obj.model_call_details["has_logged_async_success"] is True +@pytest.mark.parametrize("kwargs", [None, "not a mapping"], ids=["none", "string"]) +def test_get_first_chars_messages_returns_empty_text_for_input_that_is_not_a_mapping(kwargs: object) -> None: + assert litellm.get_first_chars_messages(kwargs=kwargs) == "" + + def test_get_response_string_joins_the_text_parts_of_a_completed_responses_event() -> None: from litellm.types.llms.openai import ResponseCompletedEvent from litellm.utils import get_response_string diff --git a/tests/unit/text_completions/test_streaming_iterator.py b/tests/unit/text_completions/test_streaming_iterator.py index 571451164d9..097546f7b31 100644 --- a/tests/unit/text_completions/test_streaming_iterator.py +++ b/tests/unit/text_completions/test_streaming_iterator.py @@ -2,12 +2,14 @@ from collections.abc import AsyncIterator, Iterator, Sequence from itertools import chain from typing import Final +import httpx import pytest from openai.types.completion import Completion from litellm.exceptions import APIConnectionError from litellm.litellm_core_utils.litellm_logging import Logging from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper +from litellm.main import _async_streaming from litellm.text_completions.streaming_iterator import TextCompletionStreamWrapper from litellm.types.utils import Choices, Delta, ModelResponseStream, StreamingChoices, TextCompletionResponse, Usage @@ -288,6 +290,33 @@ def test_sync_text_completion_stream_maps_upstream_failure_after_content() -> No assert (error.value.model, error.value.llm_provider) == ("test-model", "openai") +_DEPLOYMENT_PARAMS: Final = { + "api_base": "https://relay.example/v1", + "metadata": {"model_group": "instruct-alias", "deployment": "text-completion-openai/gpt-instruct"}, +} + + +@pytest.mark.asyncio +async def test_async_text_completion_upstream_cut_names_the_deployment() -> None: + async def cut_stream() -> AsyncIterator[ModelResponseStream]: + raise httpx.ReadError("Response payload is not completed") + yield + + wrapper: Final = TextCompletionStreamWrapper( + completion_stream=_async_streaming(cut_stream(), "test-model", None, _DEPLOYMENT_PARAMS), + model="test-model", + ) + + with pytest.raises(APIConnectionError, match="OpenAIException") as error: + await anext(wrapper) + + debug_info: Final = error.value.litellm_debug_info + assert isinstance(debug_info, str) + assert "API Base: `https://relay.example/v1`" in debug_info + assert "model_group: `instruct-alias`" in debug_info + assert "deployment: `text-completion-openai/gpt-instruct`" in debug_info + + def _stream_with_trailing_usage_chunk() -> tuple[ModelResponseStream, ...]: return ( ModelResponseStream(choices=[StreamingChoices(index=0, delta=Delta(content="Hi"))]),