fix: defer parsing incomplete streamed text

[he-yufeng]

Fixes #3263.

Summary

• defer JSON parse failures from response.output_text.done so an incomplete structured stream can reach the terminal response.incomplete event
• keep schema validation errors on complete-but-invalid structured text unchanged
• add stream state coverage for incomplete JSON, schema validation, and completed-response parsing

To verify

• PYTHONPATH=src python -m pytest tests/lib/responses/test_responses.py -q
• PYTHONPATH=src python -m ruff check src/openai/lib/streaming/responses/_responses.py tests/lib/responses/test_responses.py
• PYTHONPATH=src python -m ruff format --check src/openai/lib/streaming/responses/_responses.py tests/lib/responses/test_responses.py
• git diff --check

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: d60d06b1b6

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Distinguish top-level JSON parse failures

When a structured model contains a JSON-validated field, e.g. a Pydantic Json[...] field, Pydantic reports invalid field contents with a json_* error type even though the top-level response text parsed successfully. This helper treats those schema validation failures as truncation and returns None from response.output_text.done, so an incomplete stream with complete-but-invalid structured text can silently skip the validation error that this change intended to preserve. Check that the JSON error is for the top-level parse (for example by inspecting the error location) before suppressing it.

Useful? React with 👍 / 👎.

[he-yufeng]

Gentle nudge. The failure mode is still live: when streamed text arrives in fragments, the eager parse throws on incomplete JSON even though more chunks are on the way. The patch defers parsing until the text is complete, which matches how the non-streaming path already behaves. Tests cover the split-mid-string and split-mid-unicode cases. CI green.

[jnohclee-rgb]

Additional validation boundary (AI-assisted), reproduced at head 31298b20be5b6b8a8079137aea83057d09007d95:

_is_json_parse_error() currently matches any error type containing "json". That includes nested field validation as well as an incomplete outer JSON document. For a model with payload: pydantic.Json[list[int]], syntactically complete outer JSON with payload="not-json" produces json_invalid at location ('payload',); payload=42 produces json_type at that same location. Both now yield a response.output_text.done event with parsed=None, and raise only later at response.completed. Main 919b6236382f3f9f76d8453efb9f9ad03cff1126 raises these at text-done. This is narrower than data loss—the completed path still raises—but it differs from preserving schema-validation errors at the original boundary.

I checked eight constructed stream cases using the actual ResponseStream: this head correctly defers two outer truncations (json_invalid, location ()) and retains ordinary missing-field/nested-value errors, but delays the two nested Json-field errors above. Valid plain and nested payloads still parse. Could the discriminator distinguish a root JSON parse failure from nested field errors, or explicitly document that broader deferral?

A separate boundary control: after response.incomplete, get_final_response() raises the existing “did not receive response.completed” RuntimeError. I am not treating that as a new defect.

Python 3.14.7 / Pydantic 2.13.5, isolated exact source trees and shared dependencies plus legacy HTTPX compatibility dependencies; network-denied. These are fictional constructed event iterators, not SSE/network tests or a claim that the strict API accepts a nested Pydantic Json-field schema.

Reproducer:

import importlib,json
from types import SimpleNamespace
from pydantic import BaseModel,Json
from openai._models import construct_type_unchecked
from openai.types.responses import ResponseCreatedEvent,ResponseOutputItemAddedEvent,ResponseContentPartAddedEvent,ResponseTextDoneEvent,ResponseIncompleteEvent,ResponseCompletedEvent
from openai.lib.streaming.responses._responses import ResponseStream
source=importlib.import_module('openai.lib.streaming.responses._responses')
class Result(BaseModel):answer:str
class Nested(BaseModel):
 answer:str
 payload:Json[list[int]]
def response(status,output):return {'id':'fictional','object':'response','created_at':0,'model':'fictional','status':status,'output':output,'parallel_tool_calls':True,'tool_choice':'auto','tools':[]}
def message(text,status):return {'id':'fictional-message','type':'message','role':'assistant','status':status,'content':[{'type':'output_text','text':text,'annotations':[],'logprobs':[]}]}
class Raw:
 def __init__(self,events):self.events=events;self.response=SimpleNamespace(close=lambda:None);self.last_type=None
 def __iter__(self):
  for event in self.events:self.last_type=event.type;yield event
cases=[('truncated_colon',Result,'{"answer":','incomplete','final','RuntimeError'),('truncated_unicode',Result,'{"answer":"\\u12','incomplete','final','RuntimeError'),('valid_completed',Result,'{"answer":"fictional"}','completed','success',None),('schema_missing_field',Result,'{}','completed','response.output_text.done','ValidationError'),('nested_json_invalid',Nested,'{"answer":"fictional","payload":"not-json"}','completed','response.output_text.done','ValidationError'),('nested_json_wrong_type',Nested,'{"answer":"fictional","payload":42}','completed','response.output_text.done','ValidationError'),('nested_value_invalid',Nested,'{"answer":"fictional","payload":"[1,\\"wrong\\"]"}','completed','response.output_text.done','ValidationError'),('nested_valid',Nested,'{"answer":"fictional","payload":"[1,2]"}','completed','success',None)]
rows=[]
for name,model,text,status,expected_stage,expected_error in cases:
 payloads=[(ResponseCreatedEvent,{'type':'response.created','sequence_number':0,'response':response('in_progress',[])}),(ResponseOutputItemAddedEvent,{'type':'response.output_item.added','sequence_number':1,'output_index':0,'item':message('','in_progress')}),(ResponseContentPartAddedEvent,{'type':'response.content_part.added','sequence_number':2,'output_index':0,'content_index':0,'item_id':'fictional-message','part':{'type':'output_text','text':'','annotations':[],'logprobs':[]}}),(ResponseTextDoneEvent,{'type':'response.output_text.done','sequence_number':3,'output_index':0,'content_index':0,'item_id':'fictional-message','text':text,'logprobs':[]}),(ResponseIncompleteEvent if status=='incomplete' else ResponseCompletedEvent,{'type':'response.'+status,'sequence_number':4,'response':response(status,[message(text,status)])})]
 raw=Raw([construct_type_unchecked(type_=t,value=p) for t,p in payloads]);stream=ResponseStream(raw_stream=raw,text_format=model,input_tools=[],starting_after=None);seen=[];stage='iteration';error=None;done_parsed=None;final_parsed=None;error_locations=[]
 try:
  for event in stream:
   seen.append(event.type)
   if event.type=='response.output_text.done':done_parsed=event.parsed.model_dump() if event.parsed else None
  stage='final';final=stream.get_final_response();final_parsed=final.output_parsed.model_dump() if final.output_parsed else None;stage='success'
 except Exception as exc:
  if stage=='iteration':stage=raw.last_type
  error=type(exc).__name__
  if hasattr(exc,'errors'):error_locations=[{'type':e['type'],'loc':e['loc']} for e in exc.errors()]
 rows.append({'case':name,'stage':stage,'error':error,'error_locations':error_locations,'seen_events':seen,'done_parsed':done_parsed,'final_parsed':final_parsed,'expected_stage':expected_stage,'expected_error':expected_error,'pass':stage==expected_stage and error==expected_error})
print(json.dumps({'source_module':source.__file__,'cases':rows,'passed':sum(r['pass'] for r in rows),'failed':sum(not r['pass'] for r in rows),'scope':'Actual ResponseStream over constructed event iterator; no SSE transport/provider acceptance claim for nested Pydantic Json fields. Final accessor requires response.completed: incomplete RuntimeError is an existing contract control.'}))

[he-yufeng]

Good catch, and thanks for the harness. Reproduced it exactly as described: on the old head your eight cases go 6/8, with nested_json_invalid and nested_json_wrong_type deferred to response.completed instead of raising at response.output_text.done.

The discriminator was too coarse. "json" in type matches the nested Json[...] codes (json_invalid, json_type) just as well as a truncated outer document, but only the outer case can still be repaired by chunks that have not arrived yet. Fixed in 4eff760: deferral now requires a json_invalid error at the root loc, i.e. loc == (). A nested field error has a non-empty loc, which means the outer document parsed completely and the failure is deterministic, so it surfaces at text-done again, same boundary as main.

Also rebased onto current main (4e152cd). The branch predates the phase argument on parse_text, so the deferred path now forwards phase=output.phase and keeps the "only final_answer phases get parsed" rule.

Verification: your harness is 8/8 on this head (the two nested cases went from deferred to raising ValidationError at response.output_text.done; the two outer truncations still defer and still raise only at get_final_response()). tests/lib/responses/test_responses.py is 28 passed, including new parametrized coverage for the two nested cases you constructed, and ruff is clean.