Merge 09b49357cb into 7abff3dd63

.circleci/scripts/run_integration.sh

@@ -70,6 +70,12 @@ trap 'exit 143' TERM
 
 export PATH="$PWD/.venv/bin:$PATH"
 export PYTHONPATH="$PWD:$PWD/tests:$PWD/tests/e2e"
+if [ "$suite" = extensions ]; then
+  retained_openai="$(mktemp -d)/retained-openai"
+  uv venv --quiet --python 3.12 "$retained_openai"
+  uv pip install --quiet --python "$retained_openai/bin/python" --exclude-newer 2026-04-29T00:00:00Z openai==2.33.0
+  export RETAINED_OPENAI_PYTHON="$retained_openai/bin/python"
+fi
 export DATABASE_URL="postgresql://postgres:postgres@127.0.0.1:5432/circle_test"
 export REDIS_HOST=127.0.0.1 REDIS_PORT=6379
 export LITELLM_MASTER_KEY="sk-$(openssl rand -hex 16)" LITELLM_SALT_KEY=sk-integration-salt

.github/workflows/test-unit.yml

@@ -1346,7 +1346,7 @@ jobs:
         if: steps.changes.outputs.decision != 'skip'
         run: |
           .github/scripts/uv_sync_with_retries.sh --frozen --group dev --group proxy-dev --extra proxy --no-install-workspace
-          uv pip install "openai==${OPENAI_VERSION:?}" --exclude-newer-package openai=2026-10-07T00:00:00Z
+          uv pip install "openai==${OPENAI_VERSION:?}" --override <(echo "openai==${OPENAI_VERSION:?}") --exclude-newer-package openai=2026-10-07T00:00:00Z
           uv run --no-sync python -c 'import os, openai; assert openai.__version__ == os.environ["OPENAI_VERSION"], openai.__version__'
 
       - name: Run OpenAI SDK compatibility tests

litellm/litellm_core_utils/exception_mapping_utils.py

@@ -18,6 +18,7 @@ from litellm.litellm_core_utils.bug_report import (
 )
 from litellm.litellm_core_utils.core_helpers import as_str_mapping, attribute_or_none
 from litellm.litellm_core_utils.secret_redaction import redact_string as redact_secret_string
+from litellm.llms.custom_httpx.transport_errors import TransportErrorKind, transport_error_classes
 from litellm.types.utils import LlmProviders
 
 from ..exceptions import (
@@ -271,6 +272,10 @@ class _ProviderHTTPException(Protocol):
 
 
 _ERROR_BODY_ADAPTER: Final = TypeAdapter(dict[str, object])
+_OPENAI_CLIENT_MISSING_CREDENTIALS_MESSAGES: Final = (
+    "The api_key client option must be set either by passing api_key to the client or by setting the OPENAI_API_KEY environment variable",
+    "Missing credentials. Please pass an `api_key`",
+)
 
 
 def _openai_cyber_policy_body(original_exception: _ProviderHTTPException) -> Mapping[str, object] | None:
@@ -439,10 +444,7 @@ def _map_openai_exception(
             litellm_debug_info=extra_information,
             body=getattr(original_exception, "body", None),
         )
-    elif (
-        "The api_key client option must be set either by passing api_key to the client or by setting the OPENAI_API_KEY environment variable"
-        in error_str
-    ):
+    elif any(marker in error_str for marker in _OPENAI_CLIENT_MISSING_CREDENTIALS_MESSAGES):
         raise AuthenticationError(
             message=f"AuthenticationError: {exception_provider} - {message}",
             llm_provider=custom_llm_provider,
@@ -2556,7 +2558,7 @@ def exception_type(
             ################################################################################
 
             if (
-                isinstance(original_exception, httpx.TimeoutException)
+                isinstance(original_exception, transport_error_classes(TransportErrorKind.TIMEOUT))
                 or "Request Timeout Error" in error_str
                 or "Request timed out" in error_str
                 or "Timed out generating response" in error_str

litellm/litellm_core_utils/streaming_handler.py

@@ -14,6 +14,7 @@ from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypeVar, cast
 
 import anyio
 import httpx
+from openai.types import Completion
 from openai.types.chat import ChatCompletionChunk
 from openai.types.chat.chat_completion_chunk import Choice as ChunkChoice
 from openai.types.chat.chat_completion_chunk import ChoiceDelta
@@ -36,6 +37,7 @@ from litellm.litellm_core_utils.model_response_utils import (
 )
 from litellm.litellm_core_utils.redact_messages import LiteLLMLoggingObject
 from litellm.litellm_core_utils.thread_pool_executor import executor
+from litellm.llms.custom_httpx.transport_errors import TransportErrorKind, chained_transport_error_kind
 from litellm.llms.ollama.common_utils import OllamaResponseIdNormalizer
 from litellm.types.llms.base import LiteLLMBaseModel
 from litellm.types.router import GenericLiteLLMParams
@@ -54,7 +56,7 @@ from litellm.types.utils import (
 )
 from litellm.types.utils import GenericStreamingChunk as GChunk
 
-from ..exceptions import OpenAIError, Timeout
+from ..exceptions import OpenAIError
 from .core_helpers import map_finish_reason, process_response_headers
 from .exception_mapping_utils import exception_type, with_deployment_api_base
 from .llm_response_utils.get_api_base import get_api_base
@@ -92,6 +94,21 @@ def _next_sync_or_exhausted(it: Iterator[object]) -> object:
         return _SYNC_ITER_EXHAUSTED
 
 
+_CHOICE_CHUNK_TYPES: Final = (ChatCompletionChunk, ModelResponseStream, Completion)
+
+
+def _first_choice_index(chunk: ChatCompletionChunk | ModelResponseStream | Completion | None) -> int:
+    match chunk:
+        case ChatCompletionChunk(choices=[first, *_]):
+            return first.index
+        case ModelResponseStream(choices=[first, *_]):
+            return first.index
+        case Completion(choices=[first, *_]):
+            return first.index
+        case _:
+            return 0
+
+
 def _stamp_served_service_tier(response: ModelResponseStream, complete_streaming_response: ModelResponse) -> None:
     served_tier: Final = complete_streaming_response.model_dump().get("service_tier")
     if isinstance(served_tier, str) and served_tier:
@@ -128,12 +145,6 @@ def _delta_from_upstream(upstream_delta: "BaseModel | _UpstreamDeltaLike | None"
         return None
 
 
-def _is_transport_timeout(error: BaseException) -> bool:
-    return isinstance(error, httpx.TimeoutException) or (
-        isinstance(error, Timeout) and isinstance(error.__cause__, httpx.TimeoutException)
-    )
-
-
 def _skips_stream_fallback(status_code: int | None) -> bool:
     return status_code is not None and 400 <= status_code < 500 and status_code != 429
 
@@ -338,6 +349,7 @@ class CustomStreamWrapper:
         self._provider_response_model: str | None = None
         self.received_finish_reason: str | None = None
         self.intermittent_finish_reason: str | None = None  # finish reasons that show up mid-stream
+        self._finished_choice_indices: frozenset[int] | None = None
         self.special_tokens = [
             "<|assistant|>",
             "<|system|>",
@@ -1096,11 +1108,16 @@ class CustomStreamWrapper:
         """
         return delta is not None and stored_attribute(delta, attribute_name) is not None
 
-    def _copy_delta_attribute(self, source_delta, target_delta, attribute_name: str) -> None:
+    def _copy_delta_attribute(self, source_delta: object, target_delta: object, attribute_name: str) -> None:
         """
         Copy a specific attribute from source delta to target delta.
         """
-        setattr(target_delta, attribute_name, getattr(source_delta, attribute_name))
+        value: Final = stored_attribute(source_delta, attribute_name)
+        setattr(
+            target_delta,
+            attribute_name,
+            value.model_dump(exclude_unset=True) if isinstance(value, BaseModel) else value,
+        )
 
     def _has_any_special_delta_attributes(self, delta: object) -> bool:
         """
@@ -1492,6 +1509,7 @@ class CustomStreamWrapper:
             print_verbose(f"completion obj content: {completion_obj['content']}")
             if response_obj["is_finished"]:
                 self.received_finish_reason = response_obj["finish_reason"]
+                self._record_finished_choice(chunk if isinstance(chunk, _CHOICE_CHUNK_TYPES) else None)
             if response_obj["usage"] is not None:
                 _text_completion_usage: Final[Usage] = response_obj["usage"]
                 setattr(
@@ -1531,6 +1549,7 @@ class CustomStreamWrapper:
             print_verbose(f"completion obj content: {completion_obj['content']}")
             if response_obj["is_finished"]:
                 self.received_finish_reason = response_obj["finish_reason"]
+                self._record_finished_choice(chunk if isinstance(chunk, _CHOICE_CHUNK_TYPES) else None)
         elif self.custom_llm_provider == "cached_response":
             cached_chunk: Final = cast(ModelResponseStream, chunk)
             cached_choice: Final = cached_chunk.choices[0] if cached_chunk.choices else None
@@ -1575,6 +1594,7 @@ class CustomStreamWrapper:
                         f"{self.custom_llm_provider} raised a streaming error - finish_reason: error, no content string given. Received Chunk={response_obj}"
                     )
                 self.received_finish_reason = response_obj["finish_reason"]
+                self._record_finished_choice(chunk if isinstance(chunk, _CHOICE_CHUNK_TYPES) else None)
             if response_obj.get("original_chunk", None) is not None:
                 if hasattr(response_obj["original_chunk"], "id"):
                     model_response = self.set_model_id(response_obj["original_chunk"].id, model_response)
@@ -2036,7 +2056,7 @@ class CustomStreamWrapper:
                 ):
                     chunk = self.completion_stream
                 else:
-                    chunk = next(self.completion_stream)
+                    chunk = self._next_sync_chunk()
                 if chunk is not None and chunk != b"":
                     print_verbose(
                         f"PROCESSED CHUNK PRE CHUNK CREATOR: {chunk.decode('utf-8', errors='replace') if isinstance(chunk, bytes) else chunk}; custom_llm_provider: {self.custom_llm_provider}"
@@ -2200,6 +2220,37 @@ class CustomStreamWrapper:
             self.close()
             self._handle_stream_fallback_error(e)
 
+    def _next_sync_chunk(self) -> object:
+        try:
+            return next(self.completion_stream)
+        except Exception as e:
+            if (
+                not self._every_choice_finished()
+                or chained_transport_error_kind(e) is not TransportErrorKind.DROPPED_CONNECTION
+            ):
+                raise
+            raise StopIteration from e
+
+    def _record_finished_choice(self, chunk: ChatCompletionChunk | ModelResponseStream | Completion | None) -> None:
+        self._finished_choice_indices = (self._finished_choice_indices or frozenset()) | {_first_choice_index(chunk)}
+
+    def _every_choice_finished(self) -> bool:
+        if self.received_finish_reason is None:
+            return False
+        requested_choices: Final = self._requested_choice_count()
+        match self._finished_choice_indices:
+            case None:
+                return True
+            case finished_choices:
+                return requested_choices <= 1 or len(finished_choices) >= requested_choices
+
+    def _requested_choice_count(self) -> int:
+        match getattr(self.logging_obj, "optional_params", None):
+            case {"n": int() as requested}:
+                return requested
+            case _:
+                return 1
+
     def fetch_sync_stream(self):
         if self.completion_stream is None and self.make_call is not None:
             # Call make_call to get the completion stream
@@ -2308,21 +2359,20 @@ class CustomStreamWrapper:
                         return usage_free_chunk
         except (StopAsyncIteration, StopIteration):
             return await self._finalize_completed_stream(cache_hit=cache_hit)
-        except httpx.TimeoutException as e:  # if httpx read timeout error occues
-            traceback_exception = traceback.format_exc()
-            ## ADD DEBUG INFORMATION - E.G. LITELLM REQUEST TIMEOUT
-            traceback_exception += f"\nLiteLLM Default Request Timeout - {litellm.request_timeout}"
-            self._dispatch_stream_failure(e, traceback_exception)
-            await self.aclose()
-            self._handle_stream_fallback_error(e)
-        except (httpx.ReadError, httpx.RemoteProtocolError) as e:
-            if self.received_finish_reason is None:
-                await self.aclose()
-                self._log_stream_failure_and_raise(e)
-            return await self._finalize_completed_stream(cache_hit=cache_hit)
         except Exception as e:
-            await self.aclose()
-            self._log_stream_failure_and_raise(e)
+            match chained_transport_error_kind(e):
+                case TransportErrorKind.DROPPED_CONNECTION if self._every_choice_finished():
+                    return await self._finalize_completed_stream(cache_hit=cache_hit)
+                case TransportErrorKind.TIMEOUT:
+                    traceback_exception: Final = (
+                        f"{traceback.format_exc()}\nLiteLLM Default Request Timeout - {litellm.request_timeout}"
+                    )
+                    self._dispatch_stream_failure(e, traceback_exception)
+                    await self.aclose()
+                    self._handle_stream_fallback_error(e)
+                case _:
+                    await self.aclose()
+                    self._log_stream_failure_and_raise(e)
 
     async def _finalize_completed_stream(self, cache_hit: bool) -> "ModelResponseStream":
         if self.sent_last_chunk is True:
@@ -2604,7 +2654,7 @@ class CustomStreamWrapper:
         # Raise non-retriable client errors directly (skip fallback).
         # Exception: 429 (rate-limit) IS retriable/transient — allow it
         # through so the Router can switch to a different model group.
-        if not _is_transport_timeout(e) and (
+        if chained_transport_error_kind(e) is not TransportErrorKind.TIMEOUT and (
             _skips_stream_fallback(mapped_status_code) or _skips_stream_fallback(original_status_code)
         ):
             raise mapped_exception

litellm/llms/azure/azure.py

@@ -1,7 +1,7 @@
 import asyncio
 import json
 import time
-from collections.abc import Callable, Coroutine
+from collections.abc import Callable, Coroutine, Mapping
 from typing import Final
 
 import httpx
@@ -13,7 +13,7 @@ from openai import (
     BadRequestError,
     OpenAI,
 )
-from pydantic import BaseModel, TypeAdapter
+from pydantic import BaseModel, ConfigDict, TypeAdapter
 
 import litellm
 from litellm.constants import AZURE_OPERATION_POLLING_TIMEOUT, DEFAULT_MAX_RETRIES
@@ -140,6 +140,16 @@ def check_dynamic_azure_params(
 _check_dynamic_azure_params = check_dynamic_azure_params
 
 
+class _AzureDefaultQuery(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    default_query: Mapping[str, str] | None = None
+
+
+def _with_default_query(url: str, default_query: Mapping[str, str] | None) -> str:
+    return str(httpx.URL(url).copy_merge_params(default_query)) if default_query else url
+
+
 class AzureChatCompletion(BaseAzureLLM, BaseLLM):
     def __init__(self) -> None:
         super().__init__()
@@ -1110,6 +1120,7 @@ class AzureChatCompletion(BaseAzureLLM, BaseLLM):
             "/"
         )
         api_version: Final[str] = azure_client_params.get("api_version", "")
+        default_query: Final = _AzureDefaultQuery.model_validate(azure_client_params).default_query
         if model is None:
             model = ""
 
@@ -1134,17 +1145,12 @@ class AzureChatCompletion(BaseAzureLLM, BaseLLM):
             route="/openai/images/generations",
         )
         if v1_url is not None:
-            return v1_url
+            return _with_default_query(v1_url, default_query)
 
-        if "/openai/deployments/" in api_base:
-            base_url_with_deployment = api_base
-        else:
-            base_url_with_deployment = api_base + "/openai/deployments/" + model
-
-        base_url_with_deployment += "/images/generations"
-        base_url_with_deployment += "?api-version=" + api_version
-
-        return base_url_with_deployment
+        deployment_url: Final = (
+            api_base if "/openai/deployments/" in api_base else f"{api_base}/openai/deployments/{model}"
+        )
+        return _with_default_query(f"{deployment_url}/images/generations?api-version={api_version}", default_query)
 
     async def aimage_generation(
         self,

litellm/llms/azure/common_utils.py

@@ -8,6 +8,7 @@ from contextlib import closing
 from functools import lru_cache
 from types import MappingProxyType
 from typing import Final, Literal, NamedTuple, cast
+from urllib.parse import SplitResult, parse_qsl, urlsplit, urlunsplit
 
 import httpx
 from openai import AsyncAzureOpenAI, AsyncOpenAI, AzureOpenAI, OpenAI
@@ -327,16 +328,34 @@ def get_azure_ad_token_from_oidc(
     return azure_ad_token_access_token
 
 
-def select_azure_base_url_or_endpoint(azure_client_params: dict):
-    azure_endpoint: Final = azure_client_params.get("azure_endpoint", None)
-    if azure_endpoint is not None:
-        # see : https://github.com/openai/openai-python/blob/3d61ed42aba652b547029095a7eb269ad4e1e957/src/openai/lib/azure.py#L192
-        if "/openai/deployments" in azure_endpoint:
-            # this is base_url, not an azure_endpoint
-            azure_client_params["base_url"] = azure_endpoint
-            azure_client_params.pop("azure_endpoint")
+_DEPLOYMENT_OPERATION_PATH: Final = re.compile(
+    r"(?P<deployment>.*/openai/deployments/[^/]+)"
+    r"/(?:chat/completions|completions|embeddings|audio/(?:transcriptions|translations|speech)|images/(?:generations|edits))/?"
+)
 
-    return azure_client_params
+
+def _azure_deployment_base_url(deployment_url: SplitResult) -> str:
+    operation_path: Final = _DEPLOYMENT_OPERATION_PATH.fullmatch(deployment_url.path)
+    deployment_path: Final = deployment_url.path if operation_path is None else operation_path["deployment"]
+    return urlunsplit(deployment_url._replace(path=deployment_path, query="", fragment=""))
+
+
+def _azure_deployment_extra_query(deployment_url: SplitResult) -> Mapping[str, str]:
+    return {name: value for name, value in parse_qsl(deployment_url.query) if name != "api-version"}
+
+
+def select_azure_base_url_or_endpoint(azure_client_params: Mapping[str, object]):
+    azure_endpoint: Final = azure_client_params.get("azure_endpoint", None)
+    # see : https://github.com/openai/openai-python/blob/3d61ed42aba652b547029095a7eb269ad4e1e957/src/openai/lib/azure.py#L192
+    if not isinstance(azure_endpoint, str) or "/openai/deployments" not in azure_endpoint:
+        return dict(azure_client_params)
+    deployment_url: Final = urlsplit(azure_endpoint)
+    extra_query: Final = _azure_deployment_extra_query(deployment_url)
+    return {
+        **{name: value for name, value in azure_client_params.items() if name != "azure_endpoint"},
+        "base_url": _azure_deployment_base_url(deployment_url),
+        **({"default_query": extra_query} if extra_query else {}),
+    }
 
 
 def get_azure_ad_token(

litellm/llms/custom_httpx/aiohttp_transport.py

@@ -109,6 +109,32 @@ def map_aiohttp_exceptions() -> Generator[None, None, None]:
         raise mapped_exc(message) from exc
 
 
+@contextlib.contextmanager
+def _map_stream_read_errors() -> Generator[None, None, None]:
+    try:
+        yield
+    except RuntimeError as e:
+        if "Connection closed" not in str(e):
+            raise
+        raise httpx.ReadError(str(e)) from e
+    except aiohttp.http_exceptions.TransferEncodingError as e:
+        raise httpx.ReadError(str(e)) from e
+    except Exception:
+        with map_aiohttp_exceptions():
+            raise
+
+
+def _received_unread_bytes(content: aiohttp.StreamReader) -> bytes:
+    read_buffer: Final[Callable[[int], object] | None] = getattr(content, "_read_nowait", None)
+    if read_buffer is None:
+        return b""
+    try:
+        unread: Final = read_buffer(-1)
+    except (AttributeError, TypeError, TimeoutError):
+        return b""
+    return unread if isinstance(unread, bytes) else b""
+
+
 class AiohttpResponseStream(httpx.AsyncByteStream):
     CHUNK_SIZE = 1024 * 16
 
@@ -116,18 +142,15 @@ class AiohttpResponseStream(httpx.AsyncByteStream):
         self._aiohttp_response = aiohttp_response
 
     async def __aiter__(self) -> typing.AsyncIterator[bytes]:
+        content: Final = self._aiohttp_response.content
         try:
-            async for chunk in self._aiohttp_response.content.iter_chunked(self.CHUNK_SIZE):
+            async for chunk in content.iter_chunked(self.CHUNK_SIZE):
                 yield chunk
-        except RuntimeError as e:
-            if "Connection closed" not in str(e):
-                raise
-            raise httpx.ReadError(str(e)) from e
-        except aiohttp.http_exceptions.TransferEncodingError as e:
-            raise httpx.ReadError(str(e)) from e
         except Exception:
-            # For other exceptions, use the normal mapping
-            with map_aiohttp_exceptions():
+            unread: Final = _received_unread_bytes(content)
+            if unread:
+                yield unread
+            with _map_stream_read_errors():
                 raise
         finally:
             # Release the aiohttp connection when iteration ends for any

litellm/llms/custom_httpx/transport_errors.py

@@ -0,0 +1,59 @@
+import sys
+from collections.abc import Iterator
+from enum import Enum
+from functools import partial
+from itertools import accumulate, takewhile
+from operator import is_not
+from typing import Final, TypeAlias
+
+import httpx
+
+
+class TransportErrorKind(Enum):
+    TIMEOUT = "timeout"
+    DROPPED_CONNECTION = "dropped_connection"
+
+
+_ErrorTable: TypeAlias = tuple[tuple[type[BaseException], TransportErrorKind], ...]
+
+_HTTPX_ERRORS: Final[_ErrorTable] = (
+    (httpx.TimeoutException, TransportErrorKind.TIMEOUT),
+    (httpx.ReadError, TransportErrorKind.DROPPED_CONNECTION),
+    (httpx.RemoteProtocolError, TransportErrorKind.DROPPED_CONNECTION),
+)
+_MAX_CAUSE_DEPTH: Final = 16
+
+
+def _transport_errors() -> _ErrorTable:
+    if "httpx2" not in sys.modules:
+        return _HTTPX_ERRORS
+    import httpx2
+
+    return (
+        *_HTTPX_ERRORS,
+        (httpx2.TimeoutException, TransportErrorKind.TIMEOUT),
+        (httpx2.ReadError, TransportErrorKind.DROPPED_CONNECTION),
+        (httpx2.RemoteProtocolError, TransportErrorKind.DROPPED_CONNECTION),
+    )
+
+
+def transport_error_kind(error: object) -> TransportErrorKind | None:
+    return next((kind for error_class, kind in _transport_errors() if isinstance(error, error_class)), None)
+
+
+def transport_error_classes(kind: TransportErrorKind) -> tuple[type[BaseException], ...]:
+    return tuple(error_class for error_class, error_kind in _transport_errors() if error_kind is kind)
+
+
+def _next_cause(error: BaseException | None, _depth: int) -> BaseException | None:
+    return None if error is None else error.__cause__
+
+
+def _explicit_causes(error: BaseException) -> Iterator[BaseException | None]:
+    causes: Final = accumulate(range(1, _MAX_CAUSE_DEPTH), _next_cause, initial=error)
+    return takewhile(partial(is_not, None), causes)
+
+
+def chained_transport_error_kind(error: BaseException) -> TransportErrorKind | None:
+    kinds: Final = (transport_error_kind(cause) for cause in _explicit_causes(error))
+    return next((kind for kind in kinds if kind is not None), None)

litellm/proxy/common_utils/responses_stream_errors.py

@@ -72,7 +72,7 @@ def _failure_details(original: Exception) -> _FailureDetails:
     upstream: Final = _FailureDetails.model_validate(body)
     return _FailureDetails(
         message=upstream.message or mapped.message,
-        code=upstream.code if upstream.code is not None else mapped.code,
+        code=upstream.code if "code" in body and body["code"] is not None else mapped.code,
         type=upstream.type or mapped.type,
         status_code=mapped.status_code,
     )

litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py

@@ -22,6 +22,7 @@ from types import MappingProxyType
 from typing import TYPE_CHECKING, Annotated, Final, Literal, Protocol, cast
 
 import httpx
+import httpx2
 import openai
 from fastapi import APIRouter, Depends, HTTPException, Request, Response, WebSocket
 from fastapi.responses import StreamingResponse
@@ -3411,7 +3412,7 @@ async def openai_websocket_proxy_route(
     base_target_url: Final = os.getenv("OPENAI_API_BASE") or "https://api.openai.com/"
     try:
         openai_api_key: Final = await _openai_passthrough_credential(base_target_url)
-    except (openai.OpenAIError, httpx.HTTPError, LiteLLMOpenAIError):
+    except (openai.OpenAIError, httpx.HTTPError, httpx2.HTTPError, LiteLLMOpenAIError):
         verbose_proxy_logger.exception("OpenAI workload identity token exchange failed for websocket passthrough")
         await websocket.close(code=1011, reason=_OPENAI_WS_TOKEN_EXCHANGE_FAILED_REASON)
         return

litellm/responses/streaming_iterator.py

@@ -228,6 +228,9 @@ _ERROR_CODE_HTTP_STATUS: Final[Mapping[str, int]] = MappingProxyType(
         "insufficient_quota": 429,
         "vector_store_timeout": 504,
         "invalid_prompt": 400,
+        "data_residency_mismatch": 500,
+        "bio_policy": 500,
+        "misalignment_policy_violation": 500,
         "invalid_image": 400,
         "invalid_image_format": 400,
         "invalid_base64_image": 400,

pyproject.toml

@@ -297,7 +297,7 @@ ci = [
     "blockbuster==1.5.26",
     "beautifulsoup4==4.14.3",
     "pylint==4.0.9",
-    "langchain-openai==1.1.14",
+    "langchain-openai==1.7.2",
     "langgraph>=1.2.4,<1.3.0",
     "langgraph-prebuilt>=1.1.0,<1.3.0",
     "claude-agent-sdk==0.2.165",
@@ -369,6 +369,8 @@ constraint-dependencies = [
 override-dependencies = [
     # a2a-sdk 1.x requires packaging>=24.0; lunary 1.4.x still caps at <24.0.
     "packaging>=24.0",
+    # semantic-router (through 1.1.0) caps openai at <3.0.0, yet its OpenAI encoder for auto router files runs on SDK 3.
+    "openai>=2.20.0,<4.0.0",
     "cryptography>=50.0.0,<51.0",
     "mcp>=2.3.0,<3",
 ]

tests/integration/compatibility/_retained_openai_consumer.py

@@ -0,0 +1,56 @@
+import asyncio
+import json
+import sys
+from importlib.metadata import version
+from typing import Final
+
+import httpx
+from openai import AsyncOpenAI, OpenAI
+from openai.types.chat import ChatCompletion
+
+
+def _summary(response: ChatCompletion) -> dict[str, object]:
+    choice: Final = response.choices[0]
+    calls: Final = choice.message.tool_calls or ()
+    return {
+        "id": response.id,
+        "object": response.object,
+        "usage": [response.usage.prompt_tokens, response.usage.completion_tokens, response.usage.total_tokens]
+        if response.usage
+        else None,
+        "finish_reason": choice.finish_reason,
+        "content": choice.message.content,
+        "tool_calls": [[call.id, call.function.name, json.loads(call.function.arguments)] for call in calls],
+    }
+
+
+async def _asynchronous(base_url: str, key: str, parameters: dict[str, object]) -> tuple[ChatCompletion, ...]:
+    plain: Final = {name: value for name, value in parameters.items() if name != "tools"}
+    async with AsyncOpenAI(
+        api_key=key, base_url=base_url, max_retries=0, http_client=httpx.AsyncClient(timeout=10, trust_env=False)
+    ) as client:
+        return (await client.chat.completions.create(**parameters), await client.chat.completions.create(**plain))
+
+
+def main(base_url: str, key: str, parameters_json: str) -> None:
+    parameters: Final = json.loads(parameters_json)
+    plain: Final = {name: value for name, value in parameters.items() if name != "tools"}
+    with OpenAI(
+        api_key=key, base_url=base_url, max_retries=0, http_client=httpx.Client(timeout=10, trust_env=False)
+    ) as client:
+        sync_tools: Final[ChatCompletion] = client.chat.completions.create(**parameters)
+        sync_text: Final[ChatCompletion] = client.chat.completions.create(**plain)
+    async_tools, async_text = asyncio.run(_asynchronous(base_url, key, parameters))
+    print(
+        json.dumps(
+            {
+                "openai": version("openai"),
+                "tools": [_summary(sync_tools), _summary(async_tools)],
+                "text": [_summary(sync_text), _summary(async_text)],
+            }
+        )
+    )
+
+
+if __name__ == "__main__":
+    main(*sys.argv[1:])

tests/integration/compatibility/test_openai_consumer.py

@@ -1,7 +1,9 @@
 import json
+import os
+import subprocess
 import uuid
 from collections.abc import Callable
-from importlib.metadata import version
+from pathlib import Path
 from typing import Final
 
 import httpx
@@ -11,12 +13,22 @@ from integration._support.wire import Reply, Request, wire_server
 from openai import AsyncAzureOpenAI, AsyncOpenAI, AzureOpenAI, OpenAI
 
 
-@pytest.mark.covers("other.compatibility.openai.retained_client_parses_tools_and_usage")
-async def test_retained_openai_clients_parse_real_proxy_tool_and_usage_responses(gateway: Gateway) -> None:
-    assert version("openai") == "2.33.0", (
-        "Retain this consumer version independently before upgrading the candidate lock"
-    )
+_RETAINED_OPENAI_VERSION: Final = "2.33.0"
+_RETAINED_CONSUMER: Final = Path(__file__).with_name("_retained_openai_consumer.py")
 
+
+def _retained_openai_python() -> str:
+    python: Final = os.environ.get("RETAINED_OPENAI_PYTHON")
+    if not python:
+        pytest.fail(
+            f"Set RETAINED_OPENAI_PYTHON to a Python with openai=={_RETAINED_OPENAI_VERSION} installed, "
+            "as .circleci/scripts/run_integration.sh does for the extensions suite"
+        )
+    return python
+
+
+@pytest.mark.covers("other.compatibility.openai.retained_client_parses_tools_and_usage")
+def test_retained_openai_clients_parse_real_proxy_tool_and_usage_responses(gateway: Gateway) -> None:
     def provider(request: Request) -> Reply:
         assert request.method == "POST" and request.target == "/v1/chat/completions"
         body: Final = json.loads(request.body)
@@ -72,40 +84,35 @@ async def test_retained_openai_clients_parse_real_proxy_tool_and_usage_responses
             ],
             "extra_body": {"cache": {"no-cache": True}},
         }
-        plain: Final = {name: value for name, value in parameters.items() if name != "tools"}
-        with OpenAI(
-            api_key=key,
-            base_url=str(gateway.client.base_url).rstrip("/") + "/v1",
-            max_retries=0,
-            http_client=httpx.Client(timeout=10, trust_env=False),
-        ) as sync:
-            first: Final = sync.chat.completions.create(**parameters)
-            first_text: Final = sync.chat.completions.create(**plain)
-        async with AsyncOpenAI(
-            api_key=key,
-            base_url=str(gateway.client.base_url).rstrip("/") + "/v1",
-            max_retries=0,
-            http_client=httpx.AsyncClient(timeout=10, trust_env=False),
-        ) as asynchronous:
-            second: Final = await asynchronous.chat.completions.create(**parameters)
-            second_text: Final = await asynchronous.chat.completions.create(**plain)
-        assert len({response.id for response in (first, second, first_text, second_text)}) == 4
-        for response in (first, second, first_text, second_text):
-            assert response.object == "chat.completion"
-            assert (
-                response.usage.prompt_tokens == 11
-                and response.usage.completion_tokens == 4
-                and response.usage.total_tokens == 15
-            )
-        for response in (first, second):
-            assert response.choices[0].finish_reason == "tool_calls"
-            call: Final = response.choices[0].message.tool_calls[0]
-            assert call.id == "synthetic-call" and call.function.name == "add"
-            assert json.loads(call.function.arguments) == {"a": 3, "b": 5}
-        for response in (first_text, second_text):
-            assert response.choices[0].finish_reason == "stop"
-            assert response.choices[0].message.content == "Synthetic answer: 8"
-            assert not response.choices[0].message.tool_calls
+        consumer: Final = subprocess.run(
+            (
+                _retained_openai_python(),
+                "-I",
+                str(_RETAINED_CONSUMER),
+                str(gateway.client.base_url).rstrip("/") + "/v1",
+                key,
+                json.dumps(parameters),
+            ),
+            capture_output=True,
+            text=True,
+            timeout=60,
+            check=False,
+        )
+        assert consumer.returncode == 0, consumer.stderr
+        observed: Final = json.loads(consumer.stdout)
+        assert observed["openai"] == _RETAINED_OPENAI_VERSION
+        responses: Final = (*observed["tools"], *observed["text"])
+        assert len({response["id"] for response in responses}) == 4
+        for response in responses:
+            assert response["object"] == "chat.completion"
+            assert response["usage"] == [11, 4, 15]
+        for response in observed["tools"]:
+            assert response["finish_reason"] == "tool_calls"
+            assert response["tool_calls"] == [["synthetic-call", "add", {"a": 3, "b": 5}]]
+        for response in observed["text"]:
+            assert response["finish_reason"] == "stop"
+            assert response["content"] == "Synthetic answer: 8"
+            assert response["tool_calls"] == []
         assert len(wire.drain()) == 4
 
 

tests/unit/conftest.py

@@ -2,8 +2,11 @@ import asyncio
 import base64
 import importlib
 import os
+import sys
 from collections.abc import Callable, Coroutine, Iterator
 from dataclasses import dataclass, field
+from functools import partial
+from importlib.metadata import version
 from pathlib import Path
 from types import MappingProxyType
 from typing import Final, Literal, TypeAlias
@@ -431,6 +434,33 @@ def fake_provider_credentials(monkeypatch: pytest.MonkeyPatch) -> None:
         monkeypatch.setenv(name, value)
 
 
+@pytest.fixture
+def openai_token_exchange_via_httpx(monkeypatch: pytest.MonkeyPatch) -> None:
+    if int(version("openai").split(".")[0]) < 3:
+        return
+    import httpx2
+    import openai.auth._workload as workload_module
+
+    native_client_factory: Final = workload_module.DefaultHttpx2Client
+
+    def send_through_httpx(request: httpx2.Request) -> httpx2.Response:
+        with httpx.Client(trust_env=False) as client:
+            try:
+                response: Final = client.send(
+                    httpx.Request(request.method, str(request.url), headers=dict(request.headers), content=request.content)
+                )
+            except httpx.TransportError as error:
+                native_error: Final = getattr(httpx2, type(error).__name__, httpx2.TransportError)
+                raise native_error(str(error), request=request) from error
+        return httpx2.Response(response.status_code, headers=dict(response.headers), content=response.content)
+
+    monkeypatch.setattr(
+        workload_module,
+        "DefaultHttpx2Client",
+        partial(native_client_factory, transport=httpx2.MockTransport(send_through_httpx), trust_env=False),
+    )
+
+
 @pytest.fixture
 async def drained_logging_worker() -> None:
     await asyncio.wait_for(GLOBAL_LOGGING_WORKER.clear_queue(), timeout=10)

tests/unit/litellm_core_utils/test_exception_mapping_utils.py

@@ -2241,10 +2241,8 @@ def test_completion_perplexity_exception_on_openai_client(monkeypatch):
             model="perplexity/mistral-7b-instruct",
             messages=[{"role": "user", "content": "hello"}],
         )
-    assert (
-        "The api_key client option must be set either by passing api_key to the client or by setting the PERPLEXITY_API_KEY environment variable"
-        in str(exc_info.value)
-    )
+    assert "PERPLEXITY_API_KEY" in str(exc_info.value)
+    assert "OPENAI_API_KEY" not in str(exc_info.value)
 
 
 @pytest.mark.asyncio

tests/unit/litellm_core_utils/test_streaming_handler.py

@@ -12,7 +12,9 @@ from typing import AsyncIterator, Callable, Final, Iterator, Literal, NoReturn,
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import httpx
+import openai
 import pytest
+from openai.types import Completion, CompletionChoice
 from openai.types.chat import ChatCompletionChunk
 from pydantic import BaseModel, TypeAdapter
 import litellm
@@ -4759,7 +4761,9 @@ class TransportErrorAfterChunksIterator:
         raise StopAsyncIteration
 
 
-def _reset_test_chunk(content: Optional[str] = None, finish_reason: Optional[str] = None) -> ModelResponseStream:
+def _reset_test_chunk(
+    content: Optional[str] = None, finish_reason: Optional[str] = None, index: int = 0
+) -> ModelResponseStream:
     return ModelResponseStream(
         id="chatcmpl-reset-test",
         created=1783458104,
@@ -4767,7 +4771,7 @@ def _reset_test_chunk(content: Optional[str] = None, finish_reason: Optional[str
         object="chat.completion.chunk",
         choices=[
             StreamingChoices(
-                index=0,
+                index=index,
                 delta=Delta(content=content),
                 finish_reason=finish_reason,
             )
@@ -4851,6 +4855,216 @@ async def test_transport_read_error_before_finish_reason_raises(logging_obj: Log
     assert fabricated_finish_reasons == []
 
 
+_SDK3_CHAT_REQUEST_URL: Final = "https://api.openai.com/v1/chat/completions"
+
+
+def _raised_from(error: Exception, cause: Exception) -> Exception:
+    try:
+        raise error from cause
+    except Exception as raised:
+        return raised
+
+
+def _httpx2_read_error() -> Exception:
+    return pytest.importorskip("httpx2").ReadError("peer closed connection")
+
+
+def _httpx2_protocol_error() -> Exception:
+    return pytest.importorskip("httpx2").RemoteProtocolError("peer closed connection without sending complete body")
+
+
+def _sdk_wrapped_read_error() -> Exception:
+    httpx2: Final = pytest.importorskip("httpx2")
+    return _raised_from(
+        openai.APIConnectionError(request=httpx2.Request("POST", _SDK3_CHAT_REQUEST_URL)),
+        httpx2.ReadError("peer closed connection"),
+    )
+
+
+def _httpx2_read_timeout() -> Exception:
+    return pytest.importorskip("httpx2").ReadTimeout("timed out")
+
+
+def _sdk_wrapped_read_timeout() -> Exception:
+    httpx2: Final = pytest.importorskip("httpx2")
+    return _raised_from(
+        openai.APITimeoutError(request=httpx2.Request("POST", _SDK3_CHAT_REQUEST_URL)),
+        httpx2.ReadTimeout("timed out"),
+    )
+
+
+_SDK3_DROPPED_CONNECTIONS: Final = pytest.mark.parametrize(
+    "dropped_connection",
+    (_httpx2_read_error, _httpx2_protocol_error, _sdk_wrapped_read_error),
+    ids=("httpx2_read_error", "httpx2_protocol_error", "sdk_wrapped_read_error"),
+)
+_SYNC_AND_ASYNC: Final = pytest.mark.parametrize("sync_mode", (True, False), ids=("sync", "async"))
+
+
+def _chunks_then_failure(chunks: Sequence[ModelResponseStream], failure: Exception) -> Iterator[ModelResponseStream]:
+    yield from chunks
+    raise failure
+
+
+def _openai_stream_over(
+    chunks: Sequence[ModelResponseStream], failure: Exception, sync_mode: bool, logging_obj: Logging
+) -> CustomStreamWrapper:
+    stream: Final = _chunks_then_failure(chunks, failure)
+    return CustomStreamWrapper(
+        completion_stream=stream if sync_mode else _as_async(stream),
+        model="openai/stub-model",
+        custom_llm_provider="openai",
+        logging_obj=logging_obj,
+    )
+
+
+async def _collect(wrapper: CustomStreamWrapper, sync_mode: bool) -> list[ModelResponseStream]:
+    if sync_mode:
+        return list(wrapper)
+    return [chunk async for chunk in wrapper]
+
+
+@pytest.mark.asyncio
+@_SYNC_AND_ASYNC
+@_SDK3_DROPPED_CONNECTIONS
+async def test_sdk3_dropped_connection_after_finish_reason_ends_the_stream_cleanly(
+    logging_obj: Logging, sync_mode: bool, dropped_connection: Callable[[], Exception]
+):
+    wrapper: Final = _openai_stream_over(
+        (_reset_test_chunk(content="Hello"), _reset_test_chunk(finish_reason="stop")),
+        dropped_connection(),
+        sync_mode,
+        logging_obj,
+    )
+
+    chunks: Final = await _collect(wrapper, sync_mode)
+
+    assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices) == "Hello"
+    assert [chunk.choices[0].finish_reason for chunk in chunks if chunk.choices[0].finish_reason] == ["stop"]
+
+
+def _two_choice_stream(
+    finished_choices: Sequence[int], sync_mode: bool, logging_obj: Logging
+) -> CustomStreamWrapper:
+    logging_obj.update_environment_variables(litellm_params={}, optional_params={"n": 2})
+    return _openai_stream_over(
+        (
+            _reset_test_chunk(content="Hi", index=0),
+            _reset_test_chunk(content="Hel", index=1),
+            *(_reset_test_chunk(finish_reason="stop", index=index) for index in finished_choices),
+        ),
+        _httpx2_read_error(),
+        sync_mode,
+        logging_obj,
+    )
+
+
+@pytest.mark.asyncio
+@_SYNC_AND_ASYNC
+async def test_sdk3_dropped_connection_while_another_requested_choice_streams_raises(
+    logging_obj: Logging, sync_mode: bool
+):
+    wrapper: Final = _two_choice_stream((0,), sync_mode, logging_obj)
+
+    with pytest.raises(MidStreamFallbackError) as excinfo:
+        await _collect(wrapper, sync_mode)
+
+    assert isinstance(excinfo.value.original_exception, litellm.APIConnectionError)
+
+
+@pytest.mark.asyncio
+@_SYNC_AND_ASYNC
+async def test_sdk3_dropped_connection_after_every_requested_choice_finished_ends_cleanly(
+    logging_obj: Logging, sync_mode: bool
+):
+    wrapper: Final = _two_choice_stream((0, 1), sync_mode, logging_obj)
+
+    chunks: Final = await _collect(wrapper, sync_mode)
+
+    assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices) == "HiHel"
+
+
+def _text_completion_chunk(index: int, text: str = "", finish_reason: Optional[str] = None) -> Completion:
+    return Completion.model_construct(
+        id="cmpl-two-choices",
+        object="text_completion",
+        created=1,
+        model="gpt-3.5-turbo-instruct",
+        choices=[CompletionChoice.model_construct(index=index, text=text, finish_reason=finish_reason, logprobs=None)],
+    )
+
+
+@pytest.mark.asyncio
+@_SYNC_AND_ASYNC
+@pytest.mark.parametrize("provider", ("text-completion-openai", "azure_text"))
+@pytest.mark.parametrize(
+    ("finished_choices", "ends_cleanly"), (((0, 1), True), ((1,), False)), ids=("both_finished", "choice_0_open")
+)
+async def test_two_choice_text_completion_stream_dropped_before_done_ends_cleanly_only_once_both_finish(
+    logging_obj: Logging, sync_mode: bool, provider: str, finished_choices: Sequence[int], ends_cleanly: bool
+):
+    logging_obj.update_environment_variables(litellm_params={}, optional_params={"n": 2})
+    stream: Final = _chunks_then_failure(
+        (
+            _text_completion_chunk(0, "Hi"),
+            _text_completion_chunk(1, "Hel"),
+            *(_text_completion_chunk(index, finish_reason="stop") for index in finished_choices),
+        ),
+        _httpx2_read_error(),
+    )
+    wrapper: Final = CustomStreamWrapper(
+        completion_stream=stream if sync_mode else _as_async(stream),
+        model="gpt-3.5-turbo-instruct",
+        custom_llm_provider=provider,
+        logging_obj=logging_obj,
+    )
+
+    if not ends_cleanly:
+        with pytest.raises(MidStreamFallbackError):
+            await _collect(wrapper, sync_mode)
+        return
+    chunks: Final = await _collect(wrapper, sync_mode)
+    assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices) == "HiHel"
+
+
+@pytest.mark.asyncio
+@_SYNC_AND_ASYNC
+@_SDK3_DROPPED_CONNECTIONS
+async def test_sdk3_dropped_connection_before_finish_reason_raises_mid_stream_fallback(
+    logging_obj: Logging, sync_mode: bool, dropped_connection: Callable[[], Exception]
+):
+    wrapper: Final = _openai_stream_over(
+        (_reset_test_chunk(content="Hel"),), dropped_connection(), sync_mode, logging_obj
+    )
+
+    with pytest.raises(MidStreamFallbackError) as excinfo:
+        await _collect(wrapper, sync_mode)
+
+    assert excinfo.value.is_pre_first_chunk is False
+    assert excinfo.value.generated_content == "Hel"
+    assert isinstance(excinfo.value.original_exception, litellm.APIConnectionError)
+
+
+@pytest.mark.asyncio
+@_SYNC_AND_ASYNC
+@pytest.mark.parametrize(
+    "read_timeout", (_httpx2_read_timeout, _sdk_wrapped_read_timeout), ids=("httpx2", "sdk_wrapped_httpx2")
+)
+@pytest.mark.parametrize("finished", (False, True), ids=("before_finish_reason", "after_finish_reason"))
+async def test_sdk3_read_timeout_mid_stream_falls_back_as_a_timeout(
+    logging_obj: Logging, sync_mode: bool, read_timeout: Callable[[], Exception], finished: bool
+):
+    chunks: Final = (_reset_test_chunk(content="Hel"), _reset_test_chunk(finish_reason="stop"))
+    wrapper: Final = _openai_stream_over(chunks if finished else chunks[:1], read_timeout(), sync_mode, logging_obj)
+
+    with pytest.raises(MidStreamFallbackError) as excinfo:
+        await _collect(wrapper, sync_mode)
+
+    assert excinfo.value.is_pre_first_chunk is False
+    assert isinstance(excinfo.value.original_exception, litellm.Timeout)
+    assert excinfo.value.original_exception.status_code == 408
+
+
 def test_openai_custom_tool_call_stream_deltas_survive_conversion(logging_obj: Logging):
     """
     Regression test: OpenAI chat completions custom tool calls stream as

tests/unit/llms/azure/test_azure.py

@@ -271,6 +271,27 @@ def test_process_azure_endpoint_url(api_base, model, expected_endpoint):
     assert result == expected_endpoint, "Unexpected endpoint"
 
 
+@respx.mock
+@pytest.mark.parametrize("api_version", ("2025-04-01-preview", "v1"))
+def test_azure_image_generation_keeps_deployment_url_query_params(api_version: str):
+    route: Final = respx.post(url__regex=r"https://azure-gateway\.example\.com/.*").mock(
+        return_value=httpx.Response(200, json={"created": 1, "data": [{"b64_json": "aGk="}]})
+    )
+
+    litellm.image_generation(
+        model="azure/gpt-image-1",
+        prompt="a red circle",
+        api_base="https://azure-gateway.example.com/openai/deployments/gpt-image-1/images/generations?subscription-key=gateway-key",
+        api_version=api_version,
+        api_key="test-azure-key",
+    )
+
+    request_url: Final = route.calls.last.request.url
+    assert request_url.path.endswith("/images/generations")
+    assert request_url.params["subscription-key"] == "gateway-key"
+    assert request_url.params["api-version"] == api_version
+
+
 @patch("azure.identity.UsernamePasswordCredential")
 @patch("azure.identity.get_bearer_token_provider")
 def test_get_azure_ad_token_from_username_password(mock_get_bearer_token_provider, mock_credential):

tests/unit/llms/azure/test_azure_common_utils.py

@@ -2360,3 +2360,35 @@ def test_an_azure_client_litellm_built_its_own_http_client_for_is_still_closed(m
     closer.reap()
 
     assert wrapper.is_closed() is True
+
+
+def test_full_deployment_url_keeps_its_extra_query_params_on_the_sdk_request(monkeypatch: pytest.MonkeyPatch) -> None:
+    def echo_url(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200,
+            json={
+                "id": "chatcmpl-url",
+                "object": "chat.completion",
+                "created": 0,
+                "model": "gpt",
+                "choices": [
+                    {"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": str(request.url)}}
+                ],
+            },
+        )
+
+    monkeypatch.setattr(litellm, "client_session", httpx.Client(transport=httpx.MockTransport(echo_url)))
+    client: Final = BaseAzureLLM().get_azure_openai_client(
+        api_key="azure-key",
+        api_base="https://gateway.example.com/openai/deployments/gpt/chat/completions"
+        "?api-version=2024-10-21&subscription-key=abc",
+        api_version="2025-02-01-preview",
+        litellm_params={},
+        _is_async=False,
+    )
+
+    response: Final = client.chat.completions.create(model="gpt", messages=[{"role": "user", "content": "hi"}])
+
+    sent: Final = httpx.URL(response.choices[0].message.content or "")
+    assert sent.path == "/openai/deployments/gpt/chat/completions"
+    assert dict(sent.params) == {"api-version": "2025-02-01-preview", "subscription-key": "abc"}

tests/unit/llms/azure_ai/chat/test_azure_ai_transformation.py

@@ -160,6 +160,11 @@ async def test_azure_ai_acompletion_posts_the_conversation_to_the_models_endpoin
             "/openai/deployments/gpt-4.1-mini/chat/completions",
             id="full-deployment-url",
         ),
+        pytest.param(
+            "https://example-resource.services.ai.azure.com/openai/deployments/gpt-4.1-mini/chat/completions",
+            "/openai/deployments/gpt-4.1-mini/chat/completions",
+            id="full-deployment-url-without-query",
+        ),
     ],
 )
 def test_azure_ai_completion_routes_flexible_api_base(

tests/unit/llms/custom_httpx/test_aiohttp_transport.py

@@ -1278,6 +1278,20 @@ async def test_refused_connection_maps_to_httpx_connect_error(
     assert (cause.host, cause.port) == ("127.0.0.1", port)
 
 
+async def _read_request_headers(reader: asyncio.StreamReader) -> bytes:
+    headers: Final = await reader.readuntil(b"\r\n\r\n")
+    content_length: Final = next(
+        (
+            int(header.split(b":", 1)[1])
+            for header in headers.split(b"\r\n")
+            if header.lower().startswith(b"content-length:")
+        ),
+        0,
+    )
+    await reader.readexactly(content_length)
+    return headers
+
+
 @pytest.mark.asyncio
 async def test_reset_after_request_is_not_replayed(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("NO_PROXY", "127.0.0.1")
@@ -1285,17 +1299,7 @@ async def test_reset_after_request_is_not_replayed(monkeypatch: pytest.MonkeyPat
     received_requests: Final[list[bytes]] = []  # mutable-ok: local server captures requests
 
     async def reset_after_request(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
-        headers: Final = await reader.readuntil(b"\r\n\r\n")
-        content_length: Final = next(
-            (
-                int(header.split(b":", 1)[1])
-                for header in headers.split(b"\r\n")
-                if header.lower().startswith(b"content-length:")
-            ),
-            0,
-        )
-        await reader.readexactly(content_length)
-        received_requests.append(headers)
+        received_requests.append(await _read_request_headers(reader))
         transport_socket: Final = writer.get_extra_info("socket")
         assert transport_socket is not None
         transport_socket.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
@@ -1314,6 +1318,107 @@ async def test_reset_after_request_is_not_replayed(monkeypatch: pytest.MonkeyPat
         await handler.close()
 
 
+_SSE_EVENTS_BEFORE_DROP: Final = (
+    b'data: {"id":"chatcmpl-drop","object":"chat.completion.chunk","created":1,"model":"gpt-5.6",'
+    b'"choices":[{"index":0,"delta":{"role":"assistant","content":"1 2 "},"finish_reason":null}]}\n\n',
+    b'data: {"id":"chatcmpl-drop","object":"chat.completion.chunk","created":1,"model":"gpt-5.6",'
+    b'"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\n',
+)
+
+
+async def _stream_events_then_drop_before_done(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+    await _read_request_headers(reader)
+    chunked_events: Final = b"".join(b"%x\r\n%s\r\n" % (len(event), event) for event in _SSE_EVENTS_BEFORE_DROP)
+    writer.write(
+        b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\n\r\n" + chunked_events
+    )
+    await writer.drain()
+    writer.close()
+    await writer.wait_closed()
+
+
+async def _until_connection_drop_is_received(content: aiohttp.StreamReader) -> None:
+    while content.exception() is None:
+        await asyncio.sleep(0)
+
+
+async def _read_until_error(stream: httpx.AsyncByteStream) -> tuple[bytes, Exception | None]:
+    chunks: Final[list[bytes]] = []  # mutable-ok: keeps the chunks read before the stream raises
+    try:
+        async for chunk in stream:
+            chunks.append(chunk)
+    except Exception as error:
+        return b"".join(chunks), error
+    return b"".join(chunks), None
+
+
+@pytest.mark.asyncio
+async def test_bytes_received_before_a_connection_drop_are_yielded_before_read_error() -> None:
+    server: Final = await asyncio.start_server(_stream_events_then_drop_before_done, "127.0.0.1", 0)
+    port: Final = server.sockets[0].getsockname()[1]
+    async with server, aiohttp.ClientSession() as session:
+        aiohttp_response: Final = await session.post(f"http://127.0.0.1:{port}/v1/chat/completions", data=b"{}")
+        await _until_connection_drop_is_received(aiohttp_response.content)
+        received, error = await _read_until_error(AiohttpResponseStream(aiohttp_response))
+
+    assert received == b"".join(_SSE_EVENTS_BEFORE_DROP)
+    assert isinstance(error, httpx.ReadError)
+
+
+@pytest.mark.asyncio
+async def test_finished_stream_dropped_before_done_ends_cleanly_through_acompletion(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("NO_PROXY", "127.0.0.1")
+    monkeypatch.setenv("no_proxy", "127.0.0.1")
+    upstream_responses: Final[list[aiohttp.ClientResponse]] = []  # mutable-ok: trace hook captures the response
+
+    async def capture_response(
+        _session: ClientSession, _context: object, params: aiohttp.TraceRequestEndParams
+    ) -> None:
+        upstream_responses.append(params.response)
+
+    trace_config: Final = aiohttp.TraceConfig()
+    trace_config.on_request_end.append(capture_response)
+    server: Final = await asyncio.start_server(_stream_events_then_drop_before_done, "127.0.0.1", 0)
+    port: Final = server.sockets[0].getsockname()[1]
+    async with server, aiohttp.ClientSession(trace_configs=[trace_config]) as session:
+        stream: Final = await litellm.acompletion(
+            model="openai/gpt-5.6",
+            api_base=f"http://127.0.0.1:{port}/v1",
+            api_key="sk-local-test",
+            messages=[{"role": "user", "content": "count to two"}],
+            stream=True,
+            shared_session=session,
+        )
+        await _until_connection_drop_is_received(upstream_responses[0].content)
+        chunks: Final = [chunk async for chunk in stream]
+
+    assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks) == "1 2 "
+    assert chunks[-1].choices[0].finish_reason == "stop"
+
+
+class _ContentWithChangedBufferInternals(MockContent):
+    def _read_nowait(self, n: int) -> bytes:
+        raise TypeError(f"_read_nowait({n}) no longer drains the buffer")
+
+
+@pytest.mark.asyncio
+async def test_payload_error_still_maps_to_read_error_when_buffer_drain_breaks() -> None:
+    mock_response: Final = MockAiohttpResponse()
+    mock_response.content = _ContentWithChangedBufferInternals(
+        chunks=[b"chunk1", b"chunk2"],
+        exception_to_raise=aiohttp.ClientPayloadError("Response payload is not completed"),
+        exception_at_chunk=1,
+    )
+
+    received, error = await _read_until_error(AiohttpResponseStream(mock_response))  # pyright: ignore[reportArgumentType]  # duck-typed aiohttp response double
+
+    assert received == b"chunk1"
+    assert isinstance(error, httpx.ReadError)
+    assert mock_response.closed is True
+
+
 async def _new_session() -> aiohttp.ClientSession:
     return aiohttp.ClientSession()
 

tests/unit/llms/custom_httpx/test_transport_errors.py

@@ -0,0 +1,86 @@
+from typing import Final
+
+import httpx
+import openai
+import pytest
+
+from litellm.llms.custom_httpx.transport_errors import (
+    TransportErrorKind,
+    chained_transport_error_kind,
+    transport_error_classes,
+    transport_error_kind,
+)
+
+REQUEST: Final = httpx.Request("POST", "https://api.openai.com/v1/chat/completions")
+
+
+def _raised_from(error: Exception, cause: Exception) -> Exception:
+    try:
+        raise error from cause
+    except Exception as raised:
+        return raised
+
+
+def _raised_while_handling(error: Exception, handled: Exception) -> Exception:
+    try:
+        try:
+            raise handled
+        except Exception:
+            raise error
+    except Exception as raised:
+        return raised
+
+
+@pytest.mark.parametrize("module_name", ("httpx", "httpx2"))
+@pytest.mark.parametrize(
+    ("class_name", "expected"),
+    (
+        ("ReadTimeout", TransportErrorKind.TIMEOUT),
+        ("ConnectTimeout", TransportErrorKind.TIMEOUT),
+        ("ReadError", TransportErrorKind.DROPPED_CONNECTION),
+        ("RemoteProtocolError", TransportErrorKind.DROPPED_CONNECTION),
+        ("ConnectError", None),
+        ("WriteError", None),
+    ),
+)
+def test_transport_error_kind_matches_httpx_and_httpx2_alike(
+    module_name: str, class_name: str, expected: TransportErrorKind | None
+):
+    module: Final = pytest.importorskip(module_name)
+    error: Final = getattr(module, class_name)("transport failure")
+
+    assert transport_error_kind(error) is expected
+    assert chained_transport_error_kind(error) is expected
+    assert all(isinstance(error, transport_error_classes(kind)) is (kind is expected) for kind in TransportErrorKind)
+
+
+def test_transport_error_kind_ignores_values_that_are_not_transport_errors():
+    assert transport_error_kind(ValueError("decoder failure")) is None
+    assert transport_error_kind({"error": "not an exception"}) is None
+
+
+def test_chained_kind_reads_the_transport_error_an_sdk_error_was_raised_from():
+    httpx2: Final = pytest.importorskip("httpx2")
+    dropped: Final = _raised_from(openai.APIConnectionError(request=REQUEST), httpx2.ReadError("peer closed"))
+    timed_out: Final = _raised_from(openai.APITimeoutError(request=REQUEST), httpx2.ReadTimeout("timed out"))
+    twice_wrapped: Final = _raised_from(ValueError("outer"), timed_out)
+
+    assert transport_error_kind(dropped) is None
+    assert chained_transport_error_kind(dropped) is TransportErrorKind.DROPPED_CONNECTION
+    assert chained_transport_error_kind(timed_out) is TransportErrorKind.TIMEOUT
+    assert chained_transport_error_kind(twice_wrapped) is TransportErrorKind.TIMEOUT
+    assert chained_transport_error_kind(openai.APIConnectionError(request=REQUEST)) is None
+
+
+def test_chained_kind_ignores_an_error_raised_while_handling_a_transport_error():
+    contextual: Final = _raised_while_handling(ValueError("bug in a handler"), httpx.ReadError("connection reset"))
+
+    assert contextual.__context__ is not None
+    assert chained_transport_error_kind(contextual) is None
+
+
+def test_chained_kind_stops_on_a_cause_cycle():
+    cyclic: Final = ValueError("caused by itself")
+    cyclic.__cause__ = cyclic
+
+    assert chained_transport_error_kind(cyclic) is None

tests/unit/llms/openai/test_openai.py

@@ -2,7 +2,7 @@ import asyncio
 import importlib
 import json
 import os
-from collections.abc import Mapping
+from collections.abc import AsyncIterator, Iterator, Mapping
 from types import ModuleType
 from typing import Final, Literal, Protocol
 from unittest.mock import MagicMock, Mock, patch
@@ -32,6 +32,7 @@ from litellm.llms.openai.openai import (
 from litellm.types.utils import (
     ImageResponse,
     ModelResponse,
+    ModelResponseStream,
     PromptTokensDetails,
 )
 from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
@@ -441,6 +442,84 @@ async def test_acompletion_streams_text_deltas_over_injected_transport(
         assert last_finish == "stop"
 
 
+def _sse_event(delta: Mapping[str, str], finish: str | None) -> bytes:
+    body: Final = {
+        "id": "chatcmpl-dropped",
+        "object": "chat.completion.chunk",
+        "created": 0,
+        "model": "gpt-5.6",
+        "choices": [{"index": 0, "delta": delta, "finish_reason": finish}],
+    }
+    return f"data: {json.dumps(body)}\n\n".encode()
+
+
+_FINISHED_EVENTS_WITHOUT_DONE: Final = (
+    _sse_event({"role": "assistant", "content": "Hel"}, None),
+    _sse_event({"content": "lo"}, "stop"),
+)
+
+
+def _finished_then_dropped_transport(http_backend: ModuleType) -> object:
+    class FinishedThenDropped(http_backend.SyncByteStream, http_backend.AsyncByteStream):
+        def __iter__(self) -> Iterator[bytes]:
+            yield from _FINISHED_EVENTS_WITHOUT_DONE
+            raise http_backend.ReadError("peer closed the connection before [DONE]")
+
+        async def __aiter__(self) -> AsyncIterator[bytes]:
+            for event in _FINISHED_EVENTS_WITHOUT_DONE:
+                yield event
+            raise http_backend.ReadError("peer closed the connection before [DONE]")
+
+    return http_backend.MockTransport(
+        lambda _request: http_backend.Response(
+            200, headers={"content-type": "text/event-stream"}, stream=FinishedThenDropped()
+        )
+    )
+
+
+def _streamed_text_and_finish_reasons(chunks: list[ModelResponseStream]) -> tuple[str, list[str]]:
+    text: Final = "".join(part.choices[0].delta.content or "" for part in chunks if part.choices)
+    finish_reasons: Final = [part.choices[0].finish_reason for part in chunks if part.choices and part.choices[0].finish_reason]
+    return text, finish_reasons
+
+
+@pytest.mark.asyncio
+async def test_acompletion_ends_a_finished_stream_cleanly_when_the_connection_drops_before_done(
+    http_backend: ModuleType,
+):
+    async with http_backend.AsyncClient(transport=_finished_then_dropped_transport(http_backend)) as http_client:
+        stream: Final = await litellm.acompletion(
+            model="openai/gpt-5.6",
+            client=AsyncOpenAI(api_key="transport-only", http_client=http_client),
+            messages=[{"role": "user", "content": "dropped-stream-request"}],
+            stream=True,
+            num_retries=0,
+            max_retries=0,
+        )
+
+        async def drain() -> list[ModelResponseStream]:
+            return [part async for part in stream]
+
+        chunks: Final = await asyncio.wait_for(drain(), timeout=10)
+
+    assert _streamed_text_and_finish_reasons(chunks) == ("Hello", ["stop"])
+
+
+def test_completion_ends_a_finished_stream_cleanly_when_the_connection_drops_before_done(http_backend: ModuleType):
+    with http_backend.Client(transport=_finished_then_dropped_transport(http_backend)) as http_client:
+        stream: Final = litellm.completion(
+            model="openai/gpt-5.6",
+            client=OpenAI(api_key="transport-only", http_client=http_client),
+            messages=[{"role": "user", "content": "dropped-stream-request"}],
+            stream=True,
+            num_retries=0,
+            max_retries=0,
+        )
+        chunks: Final = list(stream)
+
+    assert _streamed_text_and_finish_reasons(chunks) == ("Hello", ["stop"])
+
+
 @pytest.mark.asyncio
 async def test_acompletion_streams_tool_call_arguments_over_injected_transport(http_backend: ModuleType):
     outbound: Final = asyncio.Queue()

tests/unit/llms/openai/test_openai_workload_identity.py

@@ -56,22 +56,7 @@ CHAT_COMPLETION_BODY: Final = {
 }
 
 
-@pytest.fixture(autouse=True)
-def mock_sdk_token_exchange_transport(monkeypatch: pytest.MonkeyPatch) -> None:
-    native_client_factory: Final = getattr(sys.modules.get("openai.auth._workload"), "DefaultHttpx2Client", None)
-    if native_client_factory is not None:
-        import httpx2
-
-        def handle_exchange(request: httpx2.Request) -> httpx2.Response:
-            response: Final = respx.mock.handler(
-                httpx.Request(request.method, str(request.url), headers=dict(request.headers), content=request.content)
-            )
-            return httpx2.Response(response.status_code, headers=dict(response.headers), content=response.content)
-
-        monkeypatch.setattr(
-            "openai.auth._workload.DefaultHttpx2Client",
-            partial(native_client_factory, transport=httpx2.MockTransport(handle_exchange), trust_env=False),
-        )
+pytestmark = pytest.mark.usefixtures("openai_token_exchange_via_httpx")
 
 
 @pytest.fixture

tests/unit/proxy/guardrails/guardrail_hooks/openai/test_moderations.py

@@ -1119,6 +1119,7 @@ class _ModerationsEndpointThatRecordsTheAuthorization:
 
 
 @pytest.mark.asyncio
+@pytest.mark.usefixtures("openai_token_exchange_via_httpx")
 @respx.mock
 async def test_openai_moderation_sends_the_federated_bearer_when_only_workload_identity_is_set(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path
@@ -1149,6 +1150,7 @@ async def test_openai_moderation_sends_the_federated_bearer_when_only_workload_i
 
 
 @pytest.mark.asyncio
+@pytest.mark.usefixtures("openai_token_exchange_via_httpx")
 @respx.mock
 async def test_openai_moderation_federates_with_the_identity_in_its_own_guardrail_config(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path

tests/unit/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py

@@ -4058,7 +4058,7 @@ def test_openai_passthrough_forwards_verbatim_to_openai(
 
 
 @pytest.fixture
-def openai_wif_env(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
+def openai_wif_env(monkeypatch: pytest.MonkeyPatch, tmp_path, openai_token_exchange_via_httpx: None) -> None:
     from litellm.llms.openai.workload_identity import _workload_identity_auth
 
     token_file: Final = tmp_path / "subject_token.jwt"

tests/unit/proxy/test_openai_ws_passthrough_routes.py

@@ -182,7 +182,7 @@ TOKEN_EXCHANGE_URL: Final = "https://auth.openai.com/oauth/token"
 
 
 @pytest.fixture
-def openai_wif_token_file(monkeypatch, tmp_path):
+def openai_wif_token_file(monkeypatch, tmp_path, openai_token_exchange_via_httpx: None):
     token_file = tmp_path / "subject_token.jwt"
     token_file.write_text("subject-token-from-file")
     monkeypatch.delenv("OPENAI_API_BASE", raising=False)

tests/unit/realtime_api/test_main.py

@@ -1081,7 +1081,9 @@ async def test_voice_transcript_blocked_by_guardrail():
 
 
 @pytest.fixture
-def federated_deployment(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> dict[str, str]:
+def federated_deployment(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, openai_token_exchange_via_httpx: None
+) -> dict[str, str]:
     token_file: Final = tmp_path / "subject_token.jwt"
     token_file.write_text("subject-token")
     for name in (

tests/unit/responses/test_streaming_iterator_error_events.py

@@ -687,6 +687,9 @@ def test_every_openai_sdk_response_error_code_has_explicit_status_mapping():
         ("empty_image_file", 400),
         ("failed_to_download_image", 400),
         ("image_file_not_found", 400),
+        ("data_residency_mismatch", 500),
+        ("bio_policy", 500),
+        ("misalignment_policy_violation", 500),
         ("totally_unknown_future_code", 500),
         ("429", 429),
         ("503", 503),

tests/unit/test_router/test_router.py

@@ -27229,7 +27229,7 @@ def test_router_dynamic_cooldown_correct_retry_after_time():
         ]
     )
 
-    openai_client = openai.OpenAI(api_key="")
+    openai_client = openai.OpenAI(api_key="sk-fake")
 
     cooldown_time = 30
 

tests/unit/test_router/test_router_provider_endpoints.py

@@ -95,11 +95,16 @@ def _ratelimit_headers(response: ModelResponse | CustomStreamWrapper) -> dict[st
 
 
 def _openai_router_client() -> AsyncOpenAI:
-    return AsyncOpenAI(api_key="sk-fake")
+    return AsyncOpenAI(api_key="sk-fake", http_client=httpx.AsyncClient())
 
 
 def _azure_router_client() -> AsyncAzureOpenAI:
-    return AsyncAzureOpenAI(api_key="sk-fake", azure_endpoint="https://azure.test", api_version="2025-02-01-preview")
+    return AsyncAzureOpenAI(
+        api_key="sk-fake",
+        azure_endpoint="https://azure.test",
+        api_version="2025-02-01-preview",
+        http_client=httpx.AsyncClient(),
+    )
 
 
 @pytest.mark.asyncio