fix: defer parsing incomplete streamed text

src/openai/lib/streaming/responses/_responses.py

@@ -5,6 +5,8 @@
 from typing import Any, List, Generic, Iterable, Awaitable, cast
 from typing_extensions import Self, Callable, Iterator, AsyncIterator
 
+import pydantic
+
 from ._types import ParsedResponseSnapshot
 from ._events import (
     ResponseStreamEvent,
@@ -337,7 +339,7 @@ def handle_event(self, event: RawResponseStreamEvent) -> List[ResponseStreamEven
                     logprobs=event.logprobs,
                     type="response.output_text.done",
                     text=event.text,
-                    parsed=parse_text(event.text, text_format=self._text_format, phase=output.phase),
+                    parsed=self._parse_text_done_event(event.text, phase=output.phase),
                 )
             )
         elif event.type == "response.function_call_arguments.delta":
@@ -522,3 +524,21 @@ def _create_initial_response(self, event: RawResponseStreamEvent) -> ParsedRespo
             snapshot.output = []
         self._output_items = dict(enumerate(snapshot.output))
         return snapshot
+
+    def _parse_text_done_event(self, text: str, *, phase: str | None) -> TextFormatT | None:
+        try:
+            return parse_text(text, text_format=self._text_format, phase=phase)
+        except pydantic.ValidationError as exc:
+            if not _is_json_parse_error(exc):
+                raise
+
+            return None
+
+
+def _is_json_parse_error(exc: pydantic.ValidationError) -> bool:
+    # Deferral only makes sense while the outer document itself fails to
+    # parse, since later chunks may still complete it. A nested Json field
+    # raises the same json_invalid / json_type codes but with a non-empty
+    # loc, and by then the document is already complete, so the error is
+    # deterministic and must keep surfacing at text-done.
+    return any(error.get("type") == "json_invalid" and error.get("loc") == () for error in exc.errors())

tests/lib/responses/test_responses.py

@@ -4,14 +4,24 @@
 from typing_extensions import TypeVar
 
 import pytest
+import pydantic
 from inline_snapshot import snapshot
 
 from openai import OpenAI, AsyncOpenAI
 from tests.respx2 import MockRouter
 from openai._types import omit
 from openai._utils import assert_signatures_in_sync
 from openai._models import construct_type_unchecked
-from openai.types.responses import Response, ResponseCreatedEvent, ResponseOutputItemAddedEvent, response_create_params
+from openai.types.responses import (
+    Response,
+    ResponseCreatedEvent,
+    ResponseTextDoneEvent,
+    ResponseCompletedEvent as RawResponseCompletedEvent,
+    ResponseIncompleteEvent,
+    ResponseOutputItemAddedEvent,
+    ResponseContentPartAddedEvent,
+    response_create_params,
+)
 from openai.lib._parsing._responses import parse_response
 from openai.lib.streaming.responses._responses import ResponseStreamState
 
@@ -20,6 +30,86 @@
 
 _T = TypeVar("_T")
 
+
+class _StructuredText(pydantic.BaseModel):
+    answer: str
+
+
+def _response_payload(*, status: str, output: list[object] | None = None) -> dict[str, object]:
+    return {
+        "id": "resp_test",
+        "object": "response",
+        "created_at": 0,
+        "model": "gpt-4.1",
+        "output": output or [],
+        "parallel_tool_calls": True,
+        "temperature": None,
+        "tool_choice": "auto",
+        "tools": [],
+        "top_p": None,
+        "status": status,
+    }
+
+
+def _message_payload(*, text: str, status: str) -> dict[str, object]:
+    return {
+        "id": "msg_test",
+        "type": "message",
+        "role": "assistant",
+        "status": status,
+        "content": [
+            {
+                "type": "output_text",
+                "text": text,
+                "annotations": [],
+                "logprobs": [],
+            }
+        ],
+    }
+
+
+def _start_response_stream(state: ResponseStreamState[_T]) -> None:
+    state.handle_event(
+        construct_type_unchecked(
+            type_=ResponseCreatedEvent,
+            value={
+                "type": "response.created",
+                "sequence_number": 0,
+                "response": _response_payload(status="in_progress"),
+            },
+        )
+    )
+    state.handle_event(
+        construct_type_unchecked(
+            type_=ResponseOutputItemAddedEvent,
+            value={
+                "type": "response.output_item.added",
+                "sequence_number": 1,
+                "output_index": 0,
+                "item": _message_payload(text="", status="in_progress"),
+            },
+        )
+    )
+    state.handle_event(
+        construct_type_unchecked(
+            type_=ResponseContentPartAddedEvent,
+            value={
+                "type": "response.content_part.added",
+                "sequence_number": 2,
+                "output_index": 0,
+                "content_index": 0,
+                "item_id": "msg_test",
+                "part": {
+                    "type": "output_text",
+                    "text": "",
+                    "annotations": [],
+                    "logprobs": [],
+                },
+            },
+        )
+    )
+
+
 # all the snapshots in this file are auto-generated from the live API
 #
 # you can update them with
@@ -110,6 +200,135 @@ def test_parse_response_preserves_program_items(item: dict[str, object]) -> None
     assert parsed.output[0].to_dict() == item
 
 
+def test_stream_output_text_done_defers_invalid_structured_parse() -> None:
+    state = ResponseStreamState(text_format=_StructuredText, input_tools=[])
+    _start_response_stream(state)
+
+    events = state.handle_event(
+        construct_type_unchecked(
+            type_=ResponseTextDoneEvent,
+            value={
+                "type": "response.output_text.done",
+                "sequence_number": 3,
+                "output_index": 0,
+                "content_index": 0,
+                "item_id": "msg_test",
+                "logprobs": [],
+                "text": '{"answer":',
+            },
+        )
+    )
+
+    assert len(events) == 1
+    assert events[0].type == "response.output_text.done"
+    assert events[0].parsed is None
+
+    incomplete_events = state.handle_event(
+        construct_type_unchecked(
+            type_=ResponseIncompleteEvent,
+            value={
+                "type": "response.incomplete",
+                "sequence_number": 4,
+                "response": {
+                    **_response_payload(
+                        status="incomplete",
+                        output=[_message_payload(text='{"answer":', status="incomplete")],
+                    ),
+                    "incomplete_details": {"reason": "max_output_tokens"},
+                },
+            },
+        )
+    )
+    assert incomplete_events[0].type == "response.incomplete"
+
+
+def test_stream_output_text_done_preserves_structured_validation_errors() -> None:
+    state = ResponseStreamState(text_format=_StructuredText, input_tools=[])
+    _start_response_stream(state)
+
+    with pytest.raises(pydantic.ValidationError):
+        state.handle_event(
+            construct_type_unchecked(
+                type_=ResponseTextDoneEvent,
+                value={
+                    "type": "response.output_text.done",
+                    "sequence_number": 3,
+                    "output_index": 0,
+                    "content_index": 0,
+                    "item_id": "msg_test",
+                    "logprobs": [],
+                    "text": "{}",
+                },
+            )
+        )
+
+
+class _NestedJsonText(pydantic.BaseModel):
+    answer: str
+    payload: pydantic.Json[list[int]]
+
+
+@pytest.mark.parametrize("payload", ['"not-json"', "42"])
+def test_stream_output_text_done_preserves_nested_json_field_errors(payload: str) -> None:
+    # A nested Json field raises the same json_invalid / json_type codes as
+    # a truncated outer document, but with a non-empty loc: the document is
+    # already complete, so the error must surface at text-done instead of
+    # being deferred.
+    state = ResponseStreamState(text_format=_NestedJsonText, input_tools=[])
+    _start_response_stream(state)
+
+    with pytest.raises(pydantic.ValidationError):
+        state.handle_event(
+            construct_type_unchecked(
+                type_=ResponseTextDoneEvent,
+                value={
+                    "type": "response.output_text.done",
+                    "sequence_number": 3,
+                    "output_index": 0,
+                    "content_index": 0,
+                    "item_id": "msg_test",
+                    "logprobs": [],
+                    "text": f'{{"answer":"fictional","payload":{payload}}}',
+                },
+            )
+        )
+
+
+def test_stream_completed_response_still_raises_invalid_structured_parse() -> None:
+    state = ResponseStreamState(text_format=_StructuredText, input_tools=[])
+    _start_response_stream(state)
+
+    state.handle_event(
+        construct_type_unchecked(
+            type_=ResponseTextDoneEvent,
+            value={
+                "type": "response.output_text.done",
+                "sequence_number": 3,
+                "output_index": 0,
+                "content_index": 0,
+                "item_id": "msg_test",
+                "logprobs": [],
+                "text": '{"answer":',
+            },
+        )
+    )
+
+    with pytest.raises(pydantic.ValidationError):
+        state.handle_event(
+            construct_type_unchecked(
+                type_=RawResponseCompletedEvent,
+                value={
+                    "type": "response.completed",
+                    "sequence_number": 4,
+                    "response": _response_payload(
+                        status="completed",
+                        output=[_message_payload(text='{"answer":', status="completed")],
+                    ),
+                },
+            )
+        )
+
+
 @pytest.mark.respx2(base_url=base_url)
 @pytest.mark.parametrize("client,async_client", [(False, False)], indirect=True)
 @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])