refactor(passthrough): move Responses stream terminal-event parsing into OpenAI provider config

Addresses review feedback: the ResponseCompletedEvent SSE parsing now lives
in OpenAIResponsesAPIConfig next to the other Responses stream event handling,
and the proxy logging handler calls it. Adds coverage for streams that end
without a response.completed event.
This commit is contained in:
mateo-berri 2026-08-11 20:11:53 -07:00
parent c8655c3825
commit dc30e1816d
3 changed files with 35 additions and 11 deletions

View file

@ -353,6 +353,15 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
)
return event_pydantic_model.model_construct(**parsed_chunk)
@staticmethod
def parse_completed_response_from_stream_chunks(all_chunks: list[str]) -> ResponsesAPIResponse | None:
for chunk_str in reversed(all_chunks):
try:
return ResponseCompletedEvent.model_validate_json(chunk_str.removeprefix("data: ")).response
except ValueError:
continue
return None
@staticmethod
def get_event_model_class(event_type: str) -> Any:
"""

View file

@ -26,7 +26,7 @@ from litellm.proxy.pass_through_endpoints.llm_provider_handlers.base_passthrough
from litellm.proxy.pass_through_endpoints.success_handler import (
PassThroughEndpointLogging,
)
from litellm.types.llms.openai import ResponseCompletedEvent, ResponsesAPIResponse
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.passthrough_endpoints.pass_through_endpoints import (
EndpointType,
PassthroughStandardLoggingPayload,
@ -518,15 +518,6 @@ class OpenAIPassthroughLoggingHandler(BasePassthroughLoggingHandler):
verbose_proxy_logger.error("Error building complete streaming response: %s", e)
return None
@staticmethod
def _build_complete_streaming_responses_response(all_chunks: list[str]) -> ResponsesAPIResponse | None:
for chunk_str in reversed(all_chunks):
try:
return ResponseCompletedEvent.model_validate_json(chunk_str.removeprefix("data: ")).response
except ValueError:
continue
return None
@staticmethod
def _handle_logging_openai_collected_chunks(
litellm_logging_obj: LiteLLMLoggingObj,
@ -551,7 +542,7 @@ class OpenAIPassthroughLoggingHandler(BasePassthroughLoggingHandler):
handler: Final = OpenAIPassthroughLoggingHandler()
handler_instance: Final = handler
complete_response: Final = (
handler._build_complete_streaming_responses_response(all_chunks=all_chunks)
OpenAIResponsesAPIConfig.parse_completed_response_from_stream_chunks(all_chunks=all_chunks)
if is_responses
else handler._build_complete_streaming_response(
all_chunks=all_chunks,

View file

@ -593,6 +593,30 @@ class TestOpenAIPassthroughLoggingHandler:
call_type="responses",
)
@patch(f"{OpenAIPassthroughLoggingHandler.__module__}.get_standard_logging_object_payload")
@patch("litellm.completion_cost")
def test_streaming_responses_without_completed_event_returns_none(
self, mock_completion_cost, mock_get_standard_logging
):
logging_obj = self._create_mock_logging_obj()
result = OpenAIPassthroughLoggingHandler._handle_logging_openai_collected_chunks(
litellm_logging_obj=logging_obj,
passthrough_success_handler_obj=MagicMock(),
url_route="https://api.openai.com/v1/responses",
request_body={"model": "gpt-4o-mini", "stream": True},
endpoint_type=MagicMock(),
start_time=self.start_time,
all_chunks=[
'data: {"type": "response.created", "sequence_number": 0}',
'data: {"type": "response.output_text.delta", "sequence_number": 1, "delta": "OK"}',
],
end_time=self.end_time,
)
assert result == {"result": None, "kwargs": {}}
mock_completion_cost.assert_not_called()
@patch("litellm.completion_cost")
@patch(
"litellm.litellm_core_utils.litellm_logging.get_standard_logging_object_payload"