diff --git a/src/openai/lib/_parsing/_responses.py b/src/openai/lib/_parsing/_responses.py index c607587ec1..bbaa061716 100644 --- a/src/openai/lib/_parsing/_responses.py +++ b/src/openai/lib/_parsing/_responses.py @@ -1,7 +1,7 @@ from __future__ import annotations import json -from typing import TYPE_CHECKING, List, Iterable, cast +from typing import TYPE_CHECKING, List, Iterable, Optional, cast from typing_extensions import TypeVar, assert_never import pydantic @@ -19,6 +19,7 @@ ParsedContent, ParsedResponse, FunctionToolParam, + ResponseOutputItem, ParsedResponseOutputItem, ParsedResponseOutputText, ResponseFunctionToolCall, @@ -58,7 +59,17 @@ def parse_response( ) -> ParsedResponse[TextFormatT]: output_list: List[ParsedResponseOutputItem[TextFormatT]] = [] - for output in response.output: + # Guard against `response.output` being `None` (observed in the chatgpt.com + # Codex backend's consolidated `response.completed` event — see issue #3325). + # The type model declares `output` as non-nullable, but the wire value can + # violate that contract. We cast to `Optional[List[...]]` at the boundary + # so both Pyright and Mypy accept the None check without weakening the + # model contract. When the streaming accumulator has already collected + # output items, it injects them into the response before calling this + # function, so reaching here with `None` means the stream genuinely had + # no output items and an empty list is the correct result. + output_items = cast(Optional[List[ResponseOutputItem]], response.output) or [] + for output in output_items: if output.type == "message": content_list: List[ParsedContent[TextFormatT]] = [] for item in output.content: diff --git a/src/openai/lib/streaming/responses/_responses.py b/src/openai/lib/streaming/responses/_responses.py index 6975a9260d..4f24d1edf1 100644 --- a/src/openai/lib/streaming/responses/_responses.py +++ b/src/openai/lib/streaming/responses/_responses.py @@ -2,7 +2,7 @@ import inspect from types import TracebackType -from typing import Any, List, Generic, Iterable, Awaitable, cast +from typing import Any, List, Generic, Iterable, Optional, Awaitable, cast from typing_extensions import Self, Callable, Iterator, AsyncIterator from ._types import ParsedResponseSnapshot @@ -15,9 +15,15 @@ ) from ...._types import Omit, omit from ...._utils import is_given, consume_sync_iterator, consume_async_iterator +from ...._compat import PYDANTIC_V1 from ...._models import build, construct_type_unchecked from ...._streaming import Stream, AsyncStream -from ....types.responses import ParsedResponse, ResponseStreamEvent as RawResponseStreamEvent +from ....types.responses import ( + Response, + ParsedResponse, + ResponseOutputItem, + ResponseStreamEvent as RawResponseStreamEvent, +) from ..._parsing._responses import TextFormatT, parse_text, parse_response from ....types.responses.tool_param import ToolParam from ....types.responses.parsed_response import ( @@ -330,22 +336,18 @@ def accumulate_event(self, event: RawResponseStreamEvent) -> ParsedResponseSnaps if event.type == "response.output_item.added": if event.item.type == "function_call": snapshot.output.append( - construct_type_unchecked( - type_=cast(Any, ParsedResponseFunctionToolCall), value=event.item.to_dict() - ) + construct_type_unchecked(type_=cast(Any, ParsedResponseFunctionToolCall), value=event.item) ) elif event.item.type == "message": snapshot.output.append( - construct_type_unchecked(type_=cast(Any, ParsedResponseOutputMessage), value=event.item.to_dict()) + construct_type_unchecked(type_=cast(Any, ParsedResponseOutputMessage), value=event.item) ) else: snapshot.output.append(event.item) elif event.type == "response.content_part.added": output = snapshot.output[event.output_index] if output.type == "message": - output.content.append( - construct_type_unchecked(type_=cast(Any, ParsedContent), value=event.part.to_dict()) - ) + output.content.append(construct_type_unchecked(type_=cast(Any, ParsedContent), value=event.part)) elif event.type == "response.output_text.delta": output = snapshot.output[event.output_index] if output.type == "message": @@ -356,12 +358,120 @@ def accumulate_event(self, event: RawResponseStreamEvent) -> ParsedResponseSnaps output = snapshot.output[event.output_index] if output.type == "function_call": output.arguments += event.delta + elif event.type == "response.output_text.done": + output = snapshot.output[event.output_index] + if output.type == "message": + content = output.content[event.content_index] + assert content.type == "output_text" + content.text = event.text + elif event.type == "response.output_item.done": + # Replace the item in the snapshot with the finalized item from the + # done event. The server sends the authoritative item payload here, + # which may include final fields like `results`/`outputs` on tool + # items, or a final `status` of `failed`/`incomplete`. Simply + # setting `status = "completed"` would discard those fields and + # produce a stale final response in the null-output fallback path. + # + # Use event.item directly instead of event.item.to_dict() — + # construct_type_unchecked is shallow, so round-tripping through + # to_dict() would leave nested content parts as plain dicts, + # causing parse_response() to crash on output.content[].type. + if event.output_index < len(snapshot.output): + snapshot.output[event.output_index] = construct_type_unchecked( + type_=type(snapshot.output[event.output_index]), + value=event.item, + ) + elif event.type == "response.content_part.done": + # Replace the content part in the snapshot with the finalized part + # from the done event. The server sends the authoritative part + # payload here, which may include metadata like annotations, + # logprobs, or finalized text/refusal content that the delta + # accumulation may not fully capture. + # + # Use event.part directly instead of event.part.to_dict() for the + # same shallow-construction reason as output_item.done above. + output = snapshot.output[event.output_index] + if output.type == "message" and event.content_index < len(output.content): + output.content[event.content_index] = construct_type_unchecked( + type_=type(output.content[event.content_index]), + value=event.part, + ) + elif event.type == "response.function_call_arguments.done": + # Apply the finalized arguments string from the done event. + # The server sends the authoritative arguments payload here, which + # may differ from the accumulated deltas. Using the finalized + # arguments ensures `parse_response()` can correctly parse + # `parsed_arguments` in the null-output fallback path. + output = snapshot.output[event.output_index] + if output.type == "function_call": + output.arguments = event.arguments + if hasattr(output, "status"): + output.status = "completed" elif event.type == "response.completed": - self._completed_response = parse_response( - text_format=self._text_format, - response=event.response, - input_tools=self._input_tools, - ) + # The chatgpt.com Codex backend sometimes sends `response.output: null` + # in the consolidated `response.completed` event even when valid + # `output_item.done` events were streamed earlier (see issue #3325). + # `parse_response()` guards against `None` with `response.output or []`, + # but that would discard the already-accumulated `snapshot.output` and + # emit an empty final response. When the completed event has no + # output but the snapshot has accumulated items, inject the streamed + # items into a shallow copy of the response so `parse_response()` can + # still run its text_format / parsed_arguments logic on them. + # + # `output` is typed as non-nullable but the wire value can violate + # that contract; cast to `Optional[List[...]]` at the boundary so + # both Pyright and Mypy accept the None check without weakening + # the model contract. + # + # In the default streaming path, SSE data is converted through + # `construct_type(...)`. For a `response.completed` payload whose + # nested `response.output` is `null`, validation of the nested + # `Response` can fail and the discriminator fallback + # shallow-constructs the event, leaving `event.response` as the + # raw dict. Normalize it to a `Response` model before + # dereferencing `.output` so the guard doesn't raise + # `AttributeError` before the fallback can run. + response = event.response + if not isinstance(response, Response): + response = construct_type_unchecked(type_=Response, value=response) + output = cast(Optional[List[ResponseOutputItem]], response.output) + if output is None and snapshot.output: + # Build a copy of the response with the accumulated output + # items injected. Use warnings=False on Pydantic v2 to suppress + # the serializer warning from dumping the invalid null output + # field (the repo's pytest config treats warnings as errors). + # On Pydantic v1, warnings=False is not supported, so we build + # the dict without dumping the invalid field — exclude_unset + # skips the null output entirely. + if PYDANTIC_V1: + base_dict = response.to_dict() + else: + base_dict = response.to_dict(warnings=False) # type: ignore[call-arg] + response_with_output = construct_type_unchecked( + type_=type(response), + value=cast( + Any, + { + **base_dict, + # Preserve the accumulated model objects directly + # instead of converting to dicts — construct_type_unchecked + # is shallow, so dicts would stay dicts and + # parse_response() would crash on `output.type`. + "output": list(snapshot.output), + }, + ), + ) + self._completed_response = parse_response( + text_format=self._text_format, + response=response_with_output, + input_tools=self._input_tools, + ) + else: + self._completed_response = parse_response( + text_format=self._text_format, + response=response, + input_tools=self._input_tools, + ) return snapshot diff --git a/tests/lib/responses/test_null_output_fallback.py b/tests/lib/responses/test_null_output_fallback.py new file mode 100644 index 0000000000..8011fe0f7a --- /dev/null +++ b/tests/lib/responses/test_null_output_fallback.py @@ -0,0 +1,402 @@ +"""Regression tests for ResponseStreamState null-output handling (issue #3325). + +The chatgpt.com Codex backend sometimes sends `response.output: null` in the +consolidated `response.completed` event even when valid `output_item.done` events +were streamed earlier. These tests verify that: + +1. Accumulated text/items survive when `response.completed.output` is `None`. +2. Authoritative done-event fields (status, text, arguments) are applied. +3. Parsed function arguments/text formats still run in the fallback path. +4. The no-prior-items case returns an empty output. +""" + +from __future__ import annotations + +from openai._types import omit +from openai._models import construct_type_unchecked +from openai.types.responses import ( + Response, + ResponseStreamEvent as RawResponseStreamEvent, +) +from openai.lib.streaming.responses._responses import ResponseStreamState +from openai.types.responses.response_created_event import ResponseCreatedEvent +from openai.types.responses.response_completed_event import ResponseCompletedEvent +from openai.types.responses.response_output_item_done_event import ( + ResponseOutputItemDoneEvent, +) +from openai.types.responses.response_output_item_added_event import ( + ResponseOutputItemAddedEvent, +) +from openai.types.responses.response_function_call_arguments_done_event import ( + ResponseFunctionCallArgumentsDoneEvent, +) + + +def _make_created_event() -> RawResponseStreamEvent: + """Create a minimal `response.created` event to seed the stream state. + + The ``response`` field is passed as the *model object* (not a dict) so that + ``construct_type_unchecked`` preserves it as a ``Response`` instance. + ``ResponseStreamState._create_initial_response`` calls + ``event.response.to_dict()``, which would raise ``AttributeError`` on a + plain dict. + """ + response = construct_type_unchecked( + type_=Response, + value={ + "id": "resp_test", + "object": "response", + "created_at": 1754925861, + "status": "in_progress", + "model": "gpt-4o", + "output": [], + }, + ) + return construct_type_unchecked( + type_=ResponseCreatedEvent, + value={ + "type": "response.created", + "sequence_number": 0, + "response": response, + }, + ) + + +def _make_output_item_added_message() -> RawResponseStreamEvent: + """Create a `response.output_item.added` event for a message item.""" + return construct_type_unchecked( + type_=ResponseOutputItemAddedEvent, + value={ + "type": "response.output_item.added", + "sequence_number": 1, + "output_index": 0, + "item": { + "type": "message", + "id": "msg_001", + "status": "in_progress", + "role": "assistant", + "content": [], + }, + }, + ) + + +def _make_output_item_done_message(text: str = "Hello world") -> RawResponseStreamEvent: + """Create a `response.output_item.done` event for a message.""" + return construct_type_unchecked( + type_=ResponseOutputItemDoneEvent, + value={ + "type": "response.output_item.done", + "sequence_number": 4, + "output_index": 0, + "item": { + "type": "message", + "id": "msg_001", + "status": "completed", + "role": "assistant", + "content": [ + { + "type": "output_text", + "text": text, + "annotations": [], + "logprobs": [], + } + ], + }, + }, + ) + + +def _make_completed_event_null_output() -> RawResponseStreamEvent: + """Create a `response.completed` event with `output: null`. + + Build the event from a raw dict to avoid calling ``to_dict()`` on a + ``Response(output=None)``, which would trigger a Pydantic serializer + warning (and the repo's pytest config treats warnings as errors). + """ + return construct_type_unchecked( + type_=ResponseCompletedEvent, + value={ + "type": "response.completed", + "sequence_number": 5, + "response": { + "id": "resp_test", + "object": "response", + "created_at": 1754925861, + "status": "completed", + "model": "gpt-4o", + "output": None, # The bug: output is null + }, + }, + ) + + +def _make_completed_event_with_output() -> RawResponseStreamEvent: + """Create a `response.completed` event with normal output.""" + response = construct_type_unchecked( + type_=Response, + value={ + "id": "resp_test", + "object": "response", + "created_at": 1754925861, + "status": "completed", + "model": "gpt-4o", + "output": [ + { + "type": "message", + "id": "msg_001", + "status": "completed", + "role": "assistant", + "content": [ + { + "type": "output_text", + "text": "Hello world", + "annotations": [], + "logprobs": [], + } + ], + } + ], + }, + ) + return construct_type_unchecked( + type_=ResponseCompletedEvent, + value={ + "type": "response.completed", + "sequence_number": 5, + "response": response, + }, + ) + + +def _make_completed_event_empty_output() -> RawResponseStreamEvent: + """Create a `response.completed` event with empty output list.""" + response = construct_type_unchecked( + type_=Response, + value={ + "id": "resp_test", + "object": "response", + "created_at": 1754925861, + "status": "completed", + "model": "gpt-4o", + "output": [], + }, + ) + return construct_type_unchecked( + type_=ResponseCompletedEvent, + value={ + "type": "response.completed", + "sequence_number": 5, + "response": response, + }, + ) + + +def _make_state() -> ResponseStreamState: + """Create a ResponseStreamState with no text format or tools.""" + return ResponseStreamState( + input_tools=omit, + text_format=omit, + ) + + +class TestNullOutputFallback: + """Tests for the null-output fallback path in ResponseStreamState.""" + + def test_accumulated_text_survives_null_output(self): + """When response.completed has output=None, the accumulated text + from done events must survive in the final parsed response.""" + state = _make_state() + state.handle_event(_make_created_event()) + state.handle_event(_make_output_item_added_message()) + state.handle_event(_make_output_item_done_message("Hello world")) + + events = state.handle_event(_make_completed_event_null_output()) + + # The completed event should produce a ResponseCompletedEvent + assert len(events) == 1 + assert events[0].type == "response.completed" + + response = events[0].response + # The output should contain the accumulated message, not be empty + assert len(response.output) == 1 + assert response.output[0].type == "message" + assert response.output[0].content[0].text == "Hello world" + + def test_done_event_status_survives_null_output(self): + """The authoritative status from output_item.done must survive + in the null-output fallback path.""" + state = _make_state() + state.handle_event(_make_created_event()) + state.handle_event(_make_output_item_added_message()) + state.handle_event(_make_output_item_done_message("Hello world")) + + events = state.handle_event(_make_completed_event_null_output()) + + response = events[0].response + # The status should be "completed" from the done event, not "in_progress" + assert response.output[0].status == "completed" + + def test_no_prior_items_returns_empty_output(self): + """When response.completed has output=None and no items were + accumulated, the result should be an empty output list.""" + state = _make_state() + state.handle_event(_make_created_event()) + + events = state.handle_event(_make_completed_event_null_output()) + + assert len(events) == 1 + assert events[0].type == "response.completed" + # No items were accumulated, so output should be empty + assert len(events[0].response.output) == 0 + + def test_normal_completed_with_output_still_works(self): + """The normal path (output is not None) should still work correctly.""" + state = _make_state() + state.handle_event(_make_created_event()) + state.handle_event(_make_output_item_added_message()) + state.handle_event(_make_output_item_done_message("Hello world")) + + events = state.handle_event(_make_completed_event_with_output()) + + assert len(events) == 1 + assert events[0].type == "response.completed" + response = events[0].response + assert len(response.output) == 1 + assert response.output[0].type == "message" + assert response.output[0].content[0].text == "Hello world" + + def test_empty_output_completed_still_works(self): + """An empty output list (not None) should also produce empty output.""" + state = _make_state() + state.handle_event(_make_created_event()) + + events = state.handle_event(_make_completed_event_empty_output()) + + assert len(events) == 1 + assert events[0].type == "response.completed" + assert len(events[0].response.output) == 0 + + def test_dict_response_coerced_before_null_output_check(self): + """When the discriminator fallback leaves event.response as a raw dict + (because validation of Response(output=None) failed), the null-output + guard must coerce it to a Response model before dereferencing .output + instead of raising AttributeError.""" + state = _make_state() + state.handle_event(_make_created_event()) + state.handle_event(_make_output_item_added_message()) + state.handle_event(_make_output_item_done_message("Hello world")) + + # Build a completed event where `response` is a plain dict (simulating + # the discriminator fallback from construct_type on invalid null output) + completed_with_dict_response = construct_type_unchecked( + type_=ResponseCompletedEvent, + value={ + "type": "response.completed", + "sequence_number": 5, + "response": { + "id": "resp_test", + "object": "response", + "created_at": 1754925861, + "status": "completed", + "model": "gpt-4o", + "output": None, + }, + }, + ) + + events = state.handle_event(completed_with_dict_response) + + assert len(events) == 1 + assert events[0].type == "response.completed" + response = events[0].response + assert len(response.output) == 1 + assert response.output[0].type == "message" + assert response.output[0].content[0].text == "Hello world" + + +class TestFunctionCallArgumentsDone: + """Tests for the function_call_arguments.done event handling.""" + + def _make_function_call_added(self) -> RawResponseStreamEvent: + return construct_type_unchecked( + type_=ResponseOutputItemAddedEvent, + value={ + "type": "response.output_item.added", + "sequence_number": 1, + "output_index": 0, + "item": { + "type": "function_call", + "id": "fc_001", + "call_id": "call_001", + "name": "get_weather", + "arguments": "", + "status": "in_progress", + }, + }, + ) + + def _make_function_call_arguments_done(self, args: str = '{"city": "SF"}') -> RawResponseStreamEvent: + return construct_type_unchecked( + type_=ResponseFunctionCallArgumentsDoneEvent, + value={ + "type": "response.function_call_arguments.done", + "sequence_number": 2, + "output_index": 0, + "item_id": "fc_001", + "arguments": args, + }, + ) + + def _make_function_call_item_done(self, args: str = '{"city": "SF"}') -> RawResponseStreamEvent: + return construct_type_unchecked( + type_=ResponseOutputItemDoneEvent, + value={ + "type": "response.output_item.done", + "sequence_number": 3, + "output_index": 0, + "item": { + "type": "function_call", + "id": "fc_001", + "call_id": "call_001", + "name": "get_weather", + "arguments": args, + "status": "completed", + }, + }, + ) + + def _make_completed_null_output(self) -> RawResponseStreamEvent: + """Build from a raw dict to avoid serializing the invalid null output.""" + return construct_type_unchecked( + type_=ResponseCompletedEvent, + value={ + "type": "response.completed", + "sequence_number": 4, + "response": { + "id": "resp_test", + "object": "response", + "created_at": 1754925861, + "status": "completed", + "model": "gpt-4o", + "output": None, + }, + }, + ) + + def test_function_call_arguments_survive_null_output(self): + """Finalized function call arguments from the done event must + survive in the null-output fallback path.""" + state = _make_state() + state.handle_event(_make_created_event()) + state.handle_event(self._make_function_call_added()) + state.handle_event(self._make_function_call_arguments_done('{"city": "SF"}')) + state.handle_event(self._make_function_call_item_done('{"city": "SF"}')) + + events = state.handle_event(self._make_completed_null_output()) + + response = events[0].response + assert len(response.output) == 1 + assert response.output[0].type == "function_call" + assert response.output[0].arguments == '{"city": "SF"}' + assert response.output[0].name == "get_weather"