diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index ab327299243..221bc241999 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -6589,7 +6589,7 @@ class BaseLLMHTTPHandler: custom_llm_provider: str | None = None, first_message: str | None = None, **kwargs: Any, - ): + ) -> Exception | None: """ Handles Responses API WebSocket mode. @@ -6623,7 +6623,7 @@ class BaseLLMHTTPHandler: **kwargs, ) await handler.run() - return + return None import websockets from websockets.asyncio.client import ClientConnection @@ -6744,7 +6744,7 @@ class BaseLLMHTTPHandler: authorized_model=model, custom_llm_provider=custom_llm_provider, ) - await streaming.bidirectional_forward() + return await streaming.bidirectional_forward() except websockets.exceptions.InvalidStatusCode as e: verbose_logger.exception("Error connecting to responses WS backend: %s", e) @@ -6758,6 +6758,7 @@ class BaseLLMHTTPHandler: pass else: raise Exception(f"Unexpected error while closing WebSocket: {close_error}") + return None def image_edit_handler( self, diff --git a/litellm/proxy/_lazy_openapi_snapshot.json b/litellm/proxy/_lazy_openapi_snapshot.json index 40b64160b71..213cd88b6ce 100644 --- a/litellm/proxy/_lazy_openapi_snapshot.json +++ b/litellm/proxy/_lazy_openapi_snapshot.json @@ -19616,7 +19616,7 @@ } } }, - "description": "\nUnified rate-limit error.\n\nEvery rate-limit condition surfaced by litellm \u2014 whether it originated from\nan upstream LLM provider, a vendor batch endpoint, or one of litellm's own\nproxy-side limiters (parallel-requests, dynamic-rate, batch-rate, budget,\nmax-iterations, etc.) \u2014 is raised as an instance of this class.\n\nThe :attr:`category` attribute lets callers distinguish the source. See\n:class:`RateLimitErrorCategory` for the available values.\n" + "description": "\n Unified rate-limit error.\n\n Every rate-limit condition surfaced by litellm \u2014 whether it originated from\n an upstream LLM provider, a vendor batch endpoint, or one of litellm's own\n proxy-side limiters (parallel-requests, dynamic-rate, batch-rate, budget,\n max-iterations, etc.) \u2014 is raised as an instance of this class.\n\n The :attr:`category` attribute lets callers distinguish the source. See\n :class:`RateLimitErrorCategory` for the available values.\n " }, "500": { "content": { diff --git a/litellm/proxy/response_api_endpoints/endpoints.py b/litellm/proxy/response_api_endpoints/endpoints.py index ea6b67fa026..4b178c52de8 100644 --- a/litellm/proxy/response_api_endpoints/endpoints.py +++ b/litellm/proxy/response_api_endpoints/endpoints.py @@ -1567,7 +1567,13 @@ async def responses_websocket_endpoint( llm_router=llm_router, user_model=user_model, ) - await llm_call + failure: Final = await llm_call + if isinstance(failure, Exception): + await proxy_logging_obj.post_call_failure_hook( + user_api_key_dict=user_api_key_dict, + original_exception=failure, + request_data=data, + ) except Exception: verbose_proxy_logger.exception("Responses WebSocket error") await websocket.close(code=1011, reason="Internal server error") diff --git a/litellm/responses/main.py b/litellm/responses/main.py index 9705794d01d..3a4be06add9 100644 --- a/litellm/responses/main.py +++ b/litellm/responses/main.py @@ -2269,11 +2269,11 @@ async def _aresponses_websocket( api_key: str | None = None, timeout: float | None = None, **kwargs, -): +) -> Exception | None: """ Private function to handle the Responses API WebSocket mode. - For PROXY use only. + For PROXY use only. Returns the provider failure that ended the connection, if any. Resolves the LLM provider from ``model``, looks up the matching ``BaseResponsesAPIConfig``, and hands off to @@ -2343,7 +2343,7 @@ async def _aresponses_websocket( } remaining_kwargs: Final = {k: v for k, v in kwargs.items() if k not in _explicit_keys} - await base_llm_http_handler.async_responses_websocket( + return await base_llm_http_handler.async_responses_websocket( model=resolved_model, websocket=websocket, logging_obj=litellm_logging_obj, diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index c99a481db7d..b9abdffce94 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -1857,6 +1857,16 @@ class ResponsesWebSocketStreaming: if self.logging_obj: self.logging_obj.pre_call(input=message, api_key="") + def _failure_exception(self) -> Exception | None: + failed_event: Final = next( + (event for event in self.messages if event.get("type") in _RESPONSES_WS_FAILURE_EVENT_TYPES), None + ) + if failed_event is None: + return None + return _map_stream_error_to_exception( + _ws_event_error(failed_event), self.authorized_model or "", self.custom_llm_provider or "" + ) + async def _log_messages(self) -> None: if not self.logging_obj: return @@ -1864,16 +1874,11 @@ class ResponsesWebSocketStreaming: self.logging_obj.model_call_details["messages"] = self.input_messages if not self.messages: return - failed_event: Final = next( - (event for event in self.messages if event.get("type") in _RESPONSES_WS_FAILURE_EVENT_TYPES), None - ) - if failed_event is None: + exception: Final = self._failure_exception() + if exception is None: asyncio.create_task(self.logging_obj.dispatch_success_handlers(self.messages, prefer_async_handlers=True)) return self._record_usage_for_failure() - exception: Final = _map_stream_error_to_exception( - _ws_event_error(failed_event), self.authorized_model or "", self.custom_llm_provider or "" - ) traceback_exception: Final = "".join(traceback.format_exception(exception)) asyncio.create_task( self.logging_obj.dispatch_failure_handlers(exception, traceback_exception, prefer_async_handlers=True) @@ -2306,8 +2311,8 @@ class ResponsesWebSocketStreaming: except Exception as e: verbose_logger.debug("Responses WS client_to_backend ended: %s", e) - async def bidirectional_forward(self) -> None: - """Run both forwarding directions concurrently.""" + async def bidirectional_forward(self) -> Exception | None: + """Run both forwarding directions concurrently and return the provider failure that ended the connection.""" forward_task: Final = asyncio.create_task(self.backend_to_client()) try: await self.client_to_backend() @@ -2324,6 +2329,7 @@ class ResponsesWebSocketStreaming: await self.backend_ws.close() except Exception: pass + return self._failure_exception() # --------------------------------------------------------------------------- diff --git a/litellm/utils.py b/litellm/utils.py index 2c9200fbad7..298c5471b48 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -2008,7 +2008,7 @@ def client(original_function): result=result, call_type=call_type, ) - elif call_type == CallTypes.arealtime.value: + elif call_type in (CallTypes.arealtime.value, CallTypes.aresponses_websocket.value): return result ### POST-CALL RULES ### post_call_processing( diff --git a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py index 8ce5357dc94..0d2d600a7fc 100644 --- a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py +++ b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py @@ -1068,6 +1068,38 @@ async def test_arealtime_marks_litellm_params_async(monkeypatch): assert LitellmLogging._is_sync_litellm_request(captured["litellm_params"]) is False +@pytest.mark.asyncio +async def test_aresponses_websocket_hands_back_the_provider_failure_without_a_success_log(monkeypatch): + """A native Responses WebSocket connection the provider rejected comes back from the ``@client`` + wrapper as the mapped failure, and the wrapper books no success for it: the relay's own dispatch + is the connection's single log, so the proxy can record the connection as a failed request.""" + from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER + from litellm.responses.main import base_llm_http_handler + + success_events = [] + + class CaptureLogger(CustomLogger): + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): + success_events.append(response_obj) + + monkeypatch.setattr(litellm, "callbacks", [CaptureLogger()]) + monkeypatch.setattr(litellm, "failure_callback", []) + monkeypatch.setattr(litellm, "_async_failure_callback", []) + monkeypatch.setattr(litellm, "success_callback", []) + monkeypatch.setattr(litellm, "_async_success_callback", []) + failure = litellm.BadRequestError(message="invalid_encrypted_content", model="gpt-4o", llm_provider="openai") + with patch.object( # test-quality-ok: the provider socket is the seam; how the wrapper treats the relay's outcome is under test + base_llm_http_handler, "async_responses_websocket", AsyncMock(return_value=failure) + ): + outcome = await litellm._aresponses_websocket(model="openai/gpt-4o", websocket=MagicMock(), api_key="sk-test") + await asyncio.sleep(0) + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for(GLOBAL_LOGGING_WORKER.flush(), timeout=10.0) + + assert outcome is failure + assert success_events == [] + + @pytest.mark.asyncio async def test_agenerate_content_marks_litellm_params_async(): """LIT-4475: the async ``agenerate_content`` entrypoint must plant diff --git a/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py b/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py index 91d688fbacf..45ec529ce7d 100644 --- a/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py +++ b/tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py @@ -570,6 +570,74 @@ class TestResponsesWSFirstFrameModelAuth: assert mock_route_request.await_args.kwargs["route_type"] == "_aresponses_websocket" ws.close.assert_not_awaited() + @pytest.mark.asyncio + @pytest.mark.parametrize("provider_rejected", [True, False]) + async def test_endpoint_books_a_provider_rejected_connection_as_a_failed_request(self, provider_rejected): + from litellm.proxy.response_api_endpoints.endpoints import ( + responses_websocket_endpoint, + ) + + ws = MagicMock() + ws.headers = {} + ws.query_params = {} + ws.scope = {"headers": []} + ws.url = "ws://testserver/v1/responses" + ws.accept = AsyncMock() + ws.receive_text = AsyncMock( + return_value=json.dumps({"type": "response.create", "model": "gpt-4o-mini", "input": []}) + ) + ws.close = AsyncMock() + + processor = MagicMock() + processor.common_processing_pre_call_logic = AsyncMock( + return_value=({"model": "gpt-4o-mini", "litellm_metadata": {}}, MagicMock()) + ) + failure = litellm.BadRequestError( + message="invalid_encrypted_content", model="gpt-4o-mini", llm_provider="openai" + ) + + async def fake_llm_call(): + return failure if provider_rejected else None + + proxy_logging_obj = MagicMock() + proxy_logging_obj.post_call_failure_hook = AsyncMock(return_value=None) + user_api_key_dict = MagicMock() + + with ( + patch( # test-quality-ok: first-frame model auth needs a live router and key table and has its own tests above + "litellm.proxy.response_api_endpoints.endpoints._enforce_responses_ws_first_frame_model_auth", + new_callable=AsyncMock, + ), + patch( # test-quality-ok: the pre-call processor needs a live proxy; what the endpoint does with the relay's outcome is under test + "litellm.proxy.response_api_endpoints.endpoints.ProxyBaseLLMRequestProcessing", + return_value=processor, + ), + patch( # test-quality-ok: routing is the seam that hands back the relay's outcome + "litellm.proxy.route_llm_request.route_request", + new_callable=AsyncMock, + return_value=fake_llm_call(), + ), + patch( # test-quality-ok: the failure hook is the proxy's only path to a failed spend log row + "litellm.proxy.proxy_server.proxy_logging_obj", + proxy_logging_obj, + ), + ): + await responses_websocket_endpoint( + websocket=ws, + model=None, + user_api_key_dict=user_api_key_dict, + ) + + ws.close.assert_not_awaited() + if not provider_rejected: + proxy_logging_obj.post_call_failure_hook.assert_not_awaited() + return + proxy_logging_obj.post_call_failure_hook.assert_awaited_once() + booked = proxy_logging_obj.post_call_failure_hook.await_args.kwargs + assert booked["original_exception"] is failure + assert booked["user_api_key_dict"] is user_api_key_dict + assert booked["request_data"]["model"] == "gpt-4o-mini" + @pytest.mark.asyncio async def test_reruns_model_auth_for_first_frame_model(self): from starlette.requests import Request diff --git a/tests/test_litellm/responses/test_responses_websocket_all_providers.py b/tests/test_litellm/responses/test_responses_websocket_all_providers.py index b671e60438e..43946c8907d 100644 --- a/tests/test_litellm/responses/test_responses_websocket_all_providers.py +++ b/tests/test_litellm/responses/test_responses_websocket_all_providers.py @@ -2894,3 +2894,115 @@ class TestNativeWebSocketEncryptedContentAffinity: assert response_cost == 0.01 logging_obj.dispatch_success_handlers.assert_not_awaited() logging_obj.dispatch_failure_handlers.assert_awaited_once() + + @pytest.mark.asyncio + async def test_bidirectional_forward_returns_the_provider_failure(self): + import asyncio + from unittest.mock import AsyncMock + + import websockets.exceptions # noqa: F401 (lazy submodule must be importable) + + backend_drained = asyncio.Event() + backend_events = [ + json.dumps({"type": "response.created", "response": {"id": "resp_1", "status": "in_progress"}}), + json.dumps( + { + "type": "error", + "status": 400, + "error": { + "type": "invalid_request_error", + "code": "invalid_encrypted_content", + "message": "could not be verified", + }, + } + ), + ] + + async def recv(decode=False): + if backend_events: + return backend_events.pop(0) + backend_drained.set() + raise Exception("stop") + + async def receive_text(): + await backend_drained.wait() + raise Exception("client gone") + + websocket = MagicMock() + websocket.send_text = AsyncMock() + websocket.receive_text = receive_text + backend_ws = MagicMock() + backend_ws.recv = recv + backend_ws.send = AsyncMock() + backend_ws.close = AsyncMock() + logging_obj = MagicMock() + logging_obj.dispatch_success_handlers = AsyncMock() + logging_obj.dispatch_failure_handlers = AsyncMock() + logging_obj._response_cost_calculator = MagicMock(return_value=0.0) + handler = _make_streaming( + websocket=websocket, + backend_ws=backend_ws, + logging_obj=logging_obj, + request_data={}, + authorized_model="gpt-5.6", + custom_llm_provider="openai", + ) + + failure = await handler.bidirectional_forward() + + assert isinstance(failure, Exception) + assert failure.status_code == 400 + assert "could not be verified" in str(failure) + + @pytest.mark.asyncio + async def test_bidirectional_forward_returns_none_after_a_completed_turn(self): + import asyncio + from unittest.mock import AsyncMock + + import websockets.exceptions # noqa: F401 (lazy submodule must be importable) + + backend_drained = asyncio.Event() + backend_events = [ + json.dumps( + { + "type": "response.completed", + "response": { + "id": "resp_1", + "status": "completed", + "output": [], + "usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}, + }, + } + ), + ] + + async def recv(decode=False): + if backend_events: + return backend_events.pop(0) + backend_drained.set() + raise Exception("stop") + + async def receive_text(): + await backend_drained.wait() + raise Exception("client gone") + + websocket = MagicMock() + websocket.send_text = AsyncMock() + websocket.receive_text = receive_text + backend_ws = MagicMock() + backend_ws.recv = recv + backend_ws.send = AsyncMock() + backend_ws.close = AsyncMock() + logging_obj = MagicMock() + logging_obj.dispatch_success_handlers = AsyncMock() + logging_obj.dispatch_failure_handlers = AsyncMock() + handler = _make_streaming( + websocket=websocket, + backend_ws=backend_ws, + logging_obj=logging_obj, + request_data={}, + authorized_model="gpt-5.6", + custom_llm_provider="openai", + ) + + assert await handler.bidirectional_forward() is None