From 32a19b11c76c05bddc77fc975fcb7f4bb1940797 Mon Sep 17 00:00:00 2001 From: Gabriel Lesperance Date: Tue, 26 May 2026 20:13:04 -0400 Subject: [PATCH] fix(codex): recover from null terminal stream output --- agent/auxiliary_client.py | 57 ++++++++++++++----- agent/codex_runtime.py | 43 ++++++++++++++ tests/agent/test_auxiliary_client.py | 33 +++++++++++ .../test_run_agent_codex_responses.py | 40 +++++++++++++ 4 files changed, 158 insertions(+), 15 deletions(-) diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index e6d42dd21654b..c191435e660f4 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -791,27 +791,54 @@ def _check_cancelled() -> None: collected_output_items: List[Any] = [] collected_text_deltas: List[str] = [] has_function_calls = False + + def _response_from_collected_stream(): + output = list(collected_output_items) + if not output and collected_text_deltas and not has_function_calls: + assembled = "".join(collected_text_deltas) + output = [SimpleNamespace( + type="message", role="assistant", status="completed", + content=[SimpleNamespace(type="output_text", text=assembled)], + )] + if not output: + return None + return SimpleNamespace(output=output, usage=None) + if total_timeout: timeout_timer = threading.Timer(float(total_timeout), _close_client_on_timeout) timeout_timer.daemon = True timeout_timer.start() _check_cancelled() - with self._client.responses.stream(**resp_kwargs) as stream: - for _event in stream: + try: + with self._client.responses.stream(**resp_kwargs) as stream: + for _event in stream: + _check_cancelled() + _etype = getattr(_event, "type", "") + if _etype == "response.output_item.done": + _done = getattr(_event, "item", None) + if _done is not None: + collected_output_items.append(_done) + elif "output_text.delta" in _etype: + _delta = getattr(_event, "delta", "") + if _delta: + collected_text_deltas.append(_delta) + elif "function_call" in _etype: + has_function_calls = True _check_cancelled() - _etype = getattr(_event, "type", "") - if _etype == "response.output_item.done": - _done = getattr(_event, "item", None) - if _done is not None: - collected_output_items.append(_done) - elif "output_text.delta" in _etype: - _delta = getattr(_event, "delta", "") - if _delta: - collected_text_deltas.append(_delta) - elif "function_call" in _etype: - has_function_calls = True - _check_cancelled() - final = stream.get_final_response() + final = stream.get_final_response() + except TypeError as exc: + if "'NoneType' object is not iterable" not in str(exc): + raise + recovered = _response_from_collected_stream() + if recovered is None: + raise + logger.debug( + "Codex auxiliary: recovered from SDK terminal response parse error " + "using %d collected output items and %d text deltas", + len(collected_output_items), + len(collected_text_deltas), + ) + final = recovered # Backfill empty output from collected stream events _output = getattr(final, "output", None) diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 8c5dff39bffee..9d737bc7d6765 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -188,6 +188,30 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta # returns empty output (e.g. chatgpt.com backend-api sends # response.incomplete instead of response.completed). agent._codex_streamed_text_parts: list = [] + + def _response_from_collected_stream( + output_items: list, + *, + status: str = "completed", + ): + output = list(output_items) + if not output and agent._codex_streamed_text_parts and not has_tool_calls: + assembled = "".join(agent._codex_streamed_text_parts) + output = [SimpleNamespace( + type="message", + role="assistant", + status="completed", + content=[SimpleNamespace(type="output_text", text=assembled)], + )] + if not output: + return None + return SimpleNamespace( + output=output, + status=status, + model=api_kwargs.get("model"), + usage=None, + ) + for attempt in range(max_stream_retries + 1): if agent._interrupt_requested: raise InterruptedError("Agent interrupted before Codex stream retry") @@ -335,6 +359,25 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta ) return agent._run_codex_create_stream_fallback(api_kwargs, client=active_client) raise + except TypeError as exc: + if "'NoneType' object is not iterable" in str(exc): + recovered = _response_from_collected_stream(collected_output_items) + if recovered is not None: + logger.debug( + "Codex stream: recovered from SDK terminal response parse error " + "using %d collected output items and %d text deltas", + len(collected_output_items), + len(agent._codex_streamed_text_parts), + ) + return recovered + logger.debug( + "Responses stream failed while parsing terminal response; " + "falling back to create(stream=True). %s err=%s", + agent._client_log_context(), + exc, + ) + return agent._run_codex_create_stream_fallback(api_kwargs, client=active_client) + raise diff --git a/tests/agent/test_auxiliary_client.py b/tests/agent/test_auxiliary_client.py index 20c30c7ea9ed8..b12bb4b7ef974 100644 --- a/tests/agent/test_auxiliary_client.py +++ b/tests/agent/test_auxiliary_client.py @@ -2371,6 +2371,39 @@ def test_reasoning_effort_zero_falls_back_to_medium(self): assert captured.get("reasoning") == {"effort": "medium", "summary": "auto"} assert captured.get("include") == ["reasoning.encrypted_content"] + def test_recovers_when_terminal_completed_response_has_null_output(self): + """chatgpt.com Codex can stream output_item.done and then send a + terminal response.completed object whose response.output is null. + The SDK raises while parsing that final event; auxiliary callers + should still recover the already-streamed item.""" + done_item = SimpleNamespace( + type="message", + status="completed", + content=[SimpleNamespace(type="output_text", text="aux title ok")], + ) + + class _Stream: + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def __iter__(self): + yield SimpleNamespace(type="response.output_item.done", item=done_item) + raise TypeError("'NoneType' object is not iterable") + + def get_final_response(self): + raise AssertionError("terminal parse error should recover before final response") + + real_client = MagicMock() + real_client.responses.stream = lambda **kwargs: _Stream() + adapter = _CodexCompletionsAdapter(real_client, "gpt-5.5") + + response = adapter.create(messages=[{"role": "user", "content": "title this"}]) + + assert response.choices[0].message.content == "aux title ok" + class TestVisionAutoSkipsKimiCoding: """_resolve_auto vision branch skips providers that have no vision on diff --git a/tests/run_agent/test_run_agent_codex_responses.py b/tests/run_agent/test_run_agent_codex_responses.py index bc575cc676f9b..965d6a6240e44 100644 --- a/tests/run_agent/test_run_agent_codex_responses.py +++ b/tests/run_agent/test_run_agent_codex_responses.py @@ -484,6 +484,46 @@ def _fake_create(**kwargs): assert response.output[0].content[0].text == "streamed create ok" +def test_run_codex_stream_recovers_from_terminal_response_without_output(monkeypatch): + agent = _build_agent(monkeypatch) + calls = {"stream": 0, "create": 0} + done_item = SimpleNamespace( + type="message", + status="completed", + content=[SimpleNamespace(type="output_text", text="streamed item ok")], + ) + + class StreamRaisesOnCompleted: + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def __iter__(self): + yield SimpleNamespace(type="response.output_item.done", item=done_item) + raise TypeError("'NoneType' object is not iterable") + + def _fake_stream(**kwargs): + calls["stream"] += 1 + return StreamRaisesOnCompleted() + + def _fake_create(**kwargs): + calls["create"] += 1 + return _codex_message_response("unexpected fallback") + + agent.client = SimpleNamespace( + responses=SimpleNamespace( + stream=_fake_stream, + create=_fake_create, + ) + ) + + response = agent._run_codex_stream(_codex_request_kwargs()) + assert calls == {"stream": 1, "create": 0} + assert response.output[0].content[0].text == "streamed item ok" + + def test_run_conversation_codex_plain_text(monkeypatch): agent = _build_agent(monkeypatch) monkeypatch.setattr(agent, "_interruptible_api_call", lambda api_kwargs: _codex_message_response("OK"))