fix(chatgpt): accumulate response.output_item.done events for non-streaming response assembly

The ChatGPT Codex backend (chatgpt.com/backend-api/codex/responses) streams
output items via `response.output_item.done` events and emits a terminal
`response.completed` event whose `response.output` is empty -- it only
carries metadata (id, status, usage).

The existing non-streaming path in `transform_response_api_response` only
reads `response.completed.response.output`, so callers of `/v1/responses`
(and the `/v1/chat/completions` bridge that depends on it) get a response
with `status: completed` and correct token usage but `output: []`.

Fix: accumulate items from `response.output_item.done` while iterating the
SSE stream and inject them into `response.output` when the terminal event
does not populate it. When `response.completed` does carry a populated
output (e.g. standard OpenAI Responses API), it wins over the accumulator.

This mirrors the upstream Codex CLI client, which maintains an
`items_added: Vec<ResponseItem>` and fills it from `OutputItemDone` events
before shipping the `Completed` event (see `codex-rs/core/src/client.rs`,
the `map_response_stream` function).

Tests:
- `test_chatgpt_accumulates_output_item_done_when_completed_output_empty`
  reproduces the ChatGPT backend behavior and asserts items are recovered.
- `test_chatgpt_prefers_nonempty_completed_output_over_accumulated`
  guarantees standard OpenAI Responses API stays unaffected.
This commit is contained in:
Leon 2026-04-20 11:55:58 +08:00
parent 2f22a1293e
commit a8cb9d58f6
2 changed files with 155 additions and 11 deletions

View file

@ -134,6 +134,14 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
completed_response = None
error_message = None
# The ChatGPT Codex backend streams output items via
# `response.output_item.done` events and emits a final
# `response.completed` event whose `response.output` is empty — it
# only carries metadata (id, status, usage). Accumulate the items
# while iterating so the assembled non-streaming response is not
# empty. See `codex-rs/core/src/client.rs` (OutputItemDone handler)
# in the upstream Codex CLI for the reference implementation.
accumulated_output_items: list = []
for chunk in body_text.splitlines():
stripped_chunk = CustomStreamWrapper._strip_sse_data_from_chunk(chunk)
if not stripped_chunk:
@ -150,20 +158,17 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
if not isinstance(parsed_chunk, dict):
continue
event_type = parsed_chunk.get("type")
if event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE:
item = parsed_chunk.get("item")
if isinstance(item, dict):
accumulated_output_items.append(item)
continue
if event_type == ResponsesAPIStreamEvents.RESPONSE_COMPLETED:
response_payload = parsed_chunk.get("response")
if isinstance(response_payload, dict):
response_payload = dict(response_payload)
if "created_at" in response_payload:
response_payload["created_at"] = _safe_convert_created_field(
response_payload["created_at"]
)
try:
completed_response = ResponsesAPIResponse(**response_payload)
except Exception:
completed_response = ResponsesAPIResponse.model_construct(
**response_payload
)
completed_response = self._build_completed_response(
response_payload, accumulated_output_items
)
break
if event_type in (
ResponsesAPIStreamEvents.RESPONSE_FAILED,
@ -192,6 +197,22 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
completed_response._hidden_params["headers"] = raw_headers
return completed_response
@staticmethod
def _build_completed_response(
response_payload: dict, accumulated_output_items: list
) -> ResponsesAPIResponse:
response_payload = dict(response_payload)
if "created_at" in response_payload:
response_payload["created_at"] = _safe_convert_created_field(
response_payload["created_at"]
)
if not response_payload.get("output") and accumulated_output_items:
response_payload["output"] = accumulated_output_items
try:
return ResponsesAPIResponse(**response_payload)
except Exception:
return ResponsesAPIResponse.model_construct(**response_payload)
def get_complete_url(
self,
api_base: Optional[str],

View file

@ -201,3 +201,126 @@ class TestChatGPTResponsesAPITransformation:
)
assert parsed.output_text == "Hello!"
def test_chatgpt_accumulates_output_item_done_when_completed_output_empty(
self,
):
"""
The ChatGPT Codex backend streams output items via
`response.output_item.done` events and emits a terminal
`response.completed` event with an empty `response.output`
(only carrying id/status/usage). The transformation must
accumulate those items so the assembled non-streaming response
is not empty.
"""
config = ChatGPTResponsesAPIConfig()
reasoning_item = {
"id": "rs_test",
"type": "reasoning",
"summary": [],
"encrypted_content": "ENCRYPTED",
}
message_item = {
"id": "msg_test",
"type": "message",
"role": "assistant",
"status": "completed",
"content": [
{
"type": "output_text",
"text": "hello world",
"annotations": [],
}
],
}
completed_payload_without_output = {
"id": "resp_test",
"object": "response",
"created_at": 1700000000,
"status": "completed",
"model": "gpt-5.3-codex",
"output": [],
"usage": {
"input_tokens": 10,
"output_tokens": 5,
"total_tokens": 15,
},
}
sse_body = "\n".join(
[
f"data: {json.dumps({'type': 'response.output_item.done', 'output_index': 0, 'item': reasoning_item})}",
f"data: {json.dumps({'type': 'response.output_item.done', 'output_index': 1, 'item': message_item})}",
f"data: {json.dumps({'type': 'response.completed', 'response': completed_payload_without_output})}",
"data: [DONE]",
"",
]
)
raw_response = httpx.Response(
200, headers={"content-type": "text/event-stream"}, text=sse_body
)
logging_obj = MagicMock()
parsed = config.transform_response_api_response(
model="chatgpt/gpt-5.3-codex",
raw_response=raw_response,
logging_obj=logging_obj,
)
assert len(parsed.output) == 2
assert parsed.output[0].type == "reasoning"
assert parsed.output[1].type == "message"
assert parsed.output_text == "hello world"
def test_chatgpt_prefers_nonempty_completed_output_over_accumulated(self):
"""
If a `response.completed` event already carries a populated
`response.output`, it should win over any accumulated
`output_item.done` items the backend is the source of truth
when it chooses to populate the terminal event.
"""
config = ChatGPTResponsesAPIConfig()
stray_item = {
"id": "msg_stray",
"type": "message",
"role": "assistant",
"status": "completed",
"content": [{"type": "output_text", "text": "stray", "annotations": []}],
}
canonical_item = {
"id": "msg_canonical",
"type": "message",
"role": "assistant",
"status": "completed",
"content": [
{"type": "output_text", "text": "canonical", "annotations": []}
],
}
completed_payload_with_output = {
"id": "resp_test",
"object": "response",
"created_at": 1700000000,
"status": "completed",
"model": "gpt-5.3-codex",
"output": [canonical_item],
}
sse_body = "\n".join(
[
f"data: {json.dumps({'type': 'response.output_item.done', 'output_index': 0, 'item': stray_item})}",
f"data: {json.dumps({'type': 'response.completed', 'response': completed_payload_with_output})}",
"data: [DONE]",
"",
]
)
raw_response = httpx.Response(
200, headers={"content-type": "text/event-stream"}, text=sse_body
)
logging_obj = MagicMock()
parsed = config.transform_response_api_response(
model="chatgpt/gpt-5.3-codex",
raw_response=raw_response,
logging_obj=logging_obj,
)
assert len(parsed.output) == 1
assert parsed.output_text == "canonical"