diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index c4e72cb7dc5..00b877ba569 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -240,6 +240,9 @@ class BaseResponsesAPIStreamingIterator: # Store the completed response (also for incomplete/failed so logging still fires) _chunk_type = getattr(openai_responses_api_chunk, "type", None) openai_types = _get_openai_response_types() + if _chunk_type == openai_types.ResponsesAPIStreamEvents.ERROR: + raise self._exception_from_error_event(openai_responses_api_chunk) + if openai_responses_api_chunk and _chunk_type in ( openai_types.ResponsesAPIStreamEvents.RESPONSE_COMPLETED, openai_types.ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, @@ -291,6 +294,50 @@ class BaseResponsesAPIStreamingIterator: self._handle_failure(e) raise + def _exception_from_error_event(self, error_event: Any) -> Exception: + error = getattr(error_event, "error", None) + message = getattr(error, "message", None) or "Responses API stream error" + code = getattr(error, "code", None) + error_type = getattr(error, "type", None) + body = {"error": {"message": message, "code": code, "type": error_type}} + provider = self.custom_llm_provider or "" + model = self.model or "" + + if code == "context_length_exceeded": + return litellm.ContextWindowExceededError( + message=message, + model=model, + llm_provider=provider, + response=self.response, + ) + if code == "rate_limit_exceeded" or error_type in { + "rate_limit_error", + "rate_limit_exceeded", + }: + return litellm.RateLimitError( + message=message, + model=model, + llm_provider=provider, + response=self.response, + ) + if code in {"invalid_api_key", "authentication_error"} or error_type in { + "authentication_error", + "invalid_api_key", + }: + return litellm.AuthenticationError( + message=message, + model=model, + llm_provider=provider, + response=self.response, + ) + return litellm.BadRequestError( + message=message, + model=model, + llm_provider=provider, + response=self.response, + body=body, + ) + def _log_completed_response(self, *, is_async: bool) -> None: if self._completed_response_logged: return diff --git a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py index 37fcc602d37..95d251fa9c4 100644 --- a/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py +++ b/tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py @@ -2,12 +2,12 @@ Unit tests for BaseResponsesAPIStreamingIterator Tests core functionality including: -1. Processing chunks and handling ResponseCompletedEvent +1. Processing chunks and handling ResponseCompletedEvent 2. Ensuring _update_responses_api_response_id_with_model_id is called for final chunk 3. Verifying ID update is NOT called for non-final chunks (delta events) 4. Edge case handling for invalid JSON, empty chunks, and [DONE] markers -These tests ensure the streaming iterator correctly processes response chunks +These tests ensure the streaming iterator correctly processes response chunks and applies model ID updates only to completed responses, as required for proper response tracking and logging. """ @@ -19,16 +19,20 @@ from datetime import datetime from typing import Any, Dict, Optional from unittest.mock import Mock, patch +import httpx import pytest sys.path.insert(0, os.path.abspath("../..")) +import litellm from litellm.constants import STREAM_SSE_DONE_STRING from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator from litellm.responses.utils import ResponsesAPIRequestUtils from litellm.types.llms.openai import ( + ErrorEvent, + ErrorEventError, ResponseCompletedEvent, ResponseFailedEvent, ResponseIncompleteEvent, @@ -574,6 +578,181 @@ class TestBaseResponsesAPIStreamingIterator: submit_args = mock_executor.submit.call_args assert submit_args[0][0] == mock_logging_obj.failure_handler + @pytest.mark.asyncio + async def test_streaming_error_event_raises_litellm_exception(self): + """ + OpenAI can send a top-level `error` event before response.failed. The + iterator should raise that as a LiteLLM exception instead of yielding it + as a normal stream chunk. + """ + from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator + + error_chunk = { + "type": "error", + "sequence_number": 2, + "error": { + "type": "invalid_request_error", + "code": "context_length_exceeded", + "message": "Input exceeds the model context window.", + "param": "input", + }, + } + + async def mock_aiter_bytes(): + yield f"data: {json.dumps(error_chunk)}\n\n".encode("utf-8") + + mock_response = Mock() + mock_response.headers = {} + mock_response.aiter_bytes = mock_aiter_bytes + + mock_logging_obj = Mock(spec=LiteLLMLoggingObj) + mock_logging_obj.model_call_details = {"litellm_params": {}} + mock_logging_obj.async_failure_handler = Mock() + mock_logging_obj.failure_handler = Mock() + mock_config = Mock(spec=BaseResponsesAPIConfig) + mock_config.transform_streaming_response.return_value = ErrorEvent( + type=ResponsesAPIStreamEvents.ERROR, + sequence_number=2, + error=ErrorEventError( + type="invalid_request_error", + code="context_length_exceeded", + message="Input exceeds the model context window.", + param="input", + ), + ) + + iterator = ResponsesAPIStreamingIterator( + response=mock_response, + model="gpt-5.4-mini", + responses_api_provider_config=mock_config, + logging_obj=mock_logging_obj, + custom_llm_provider="openai", + ) + + with ( + pytest.raises(litellm.ContextWindowExceededError) as exc_info, + patch( + "litellm.responses.streaming_iterator.run_async_function" + ) as mock_run_async, + patch("litellm.responses.streaming_iterator.executor") as mock_executor, + ): + await iterator.__anext__() + + assert "context window" in str(exc_info.value) + assert iterator.finished is True + mock_run_async.assert_called_once() + assert ( + mock_run_async.call_args.kwargs["async_function"] + == mock_logging_obj.async_failure_handler + ) + mock_executor.submit.assert_called_once() + + def test_sync_streaming_error_event_raises_litellm_exception(self): + """ + Sync Responses API streams should raise top-level `error` events too. + """ + from litellm.responses.streaming_iterator import ( + SyncResponsesAPIStreamingIterator, + ) + + error_chunk = { + "type": "error", + "sequence_number": 2, + "error": { + "type": "rate_limit_error", + "code": "rate_limit_exceeded", + "message": "Too many requests.", + "param": None, + }, + } + + mock_response = Mock() + mock_response.headers = {} + mock_response.iter_bytes.return_value = [ + f"data: {json.dumps(error_chunk)}\n\n".encode("utf-8") + ] + + mock_logging_obj = Mock(spec=LiteLLMLoggingObj) + mock_logging_obj.model_call_details = {"litellm_params": {}} + mock_logging_obj.async_failure_handler = Mock() + mock_logging_obj.failure_handler = Mock() + mock_config = Mock(spec=BaseResponsesAPIConfig) + mock_config.transform_streaming_response.return_value = ErrorEvent( + type=ResponsesAPIStreamEvents.ERROR, + sequence_number=2, + error=ErrorEventError( + type="rate_limit_error", + code="rate_limit_exceeded", + message="Too many requests.", + param=None, + ), + ) + + iterator = SyncResponsesAPIStreamingIterator( + response=mock_response, + model="gpt-5.4-mini", + responses_api_provider_config=mock_config, + logging_obj=mock_logging_obj, + custom_llm_provider="openai", + ) + + with ( + pytest.raises(litellm.RateLimitError), + patch( + "litellm.responses.streaming_iterator.run_async_function" + ) as mock_run_async, + patch("litellm.responses.streaming_iterator.executor") as mock_executor, + ): + next(iterator) + + assert iterator.finished is True + mock_run_async.assert_called_once() + mock_executor.submit.assert_called_once() + + def test_error_event_exception_mapping(self): + """Provider error metadata should keep useful LiteLLM exception types.""" + mock_logging_obj = Mock(spec=LiteLLMLoggingObj) + mock_logging_obj.model_call_details = {"litellm_params": {}} + iterator = BaseResponsesAPIStreamingIterator( + response=httpx.Response( + 400, request=httpx.Request("POST", "https://api.example.test") + ), + model="gpt-5.4-mini", + responses_api_provider_config=Mock(spec=BaseResponsesAPIConfig), + logging_obj=mock_logging_obj, + custom_llm_provider="openai", + ) + + auth_event = ErrorEvent( + type=ResponsesAPIStreamEvents.ERROR, + sequence_number=1, + error=ErrorEventError( + type="authentication_error", + code="invalid_api_key", + message="Invalid API key.", + param=None, + ), + ) + default_event = ErrorEvent( + type=ResponsesAPIStreamEvents.ERROR, + sequence_number=2, + error=ErrorEventError( + type="invalid_request_error", + code="bad_request", + message="Bad request.", + param="input", + ), + ) + + assert isinstance( + iterator._exception_from_error_event(auth_event), + litellm.AuthenticationError, + ) + assert isinstance( + iterator._exception_from_error_event(default_event), + litellm.BadRequestError, + ) + def test_process_chunk_response_incomplete_calls_success_handler(self): """ Test that a RESPONSE_INCOMPLETE event routes to success handlers.