fix: recover codex streams with missing final output

agent/auxiliary_client.py

@@ -811,11 +811,32 @@ def _check_cancelled() -> None:
                     elif "function_call" in _etype:
                         has_function_calls = True
                 _check_cancelled()
-                final = stream.get_final_response()
+                try:
+                    final = stream.get_final_response()
+                except TypeError as exc:
+                    err_text = str(exc)
+                    if (
+                        "NoneType" in err_text
+                        and "iterable" in err_text
+                        and (collected_output_items or collected_text_deltas)
+                    ):
+                        logger.warning(
+                            "Codex auxiliary Responses SDK failed to parse final "
+                            "response after streaming output; synthesizing from "
+                            "stream events. err=%s",
+                            err_text,
+                        )
+                        final = SimpleNamespace(output=None, usage=None)
+                    else:
+                        raise
 
-            # Backfill empty output from collected stream events
+            # Backfill empty/None output from collected stream events.  The
+            # chatgpt.com Codex backend can stream valid output items/text, then
+            # leave response.output as None while the OpenAI SDK parses the
+            # final response, which otherwise breaks title generation and other
+            # auxiliary tasks after the user-facing turn has already succeeded.
             _output = getattr(final, "output", None)
-            if isinstance(_output, list) and not _output:
+            if (isinstance(_output, list) and not _output) or _output is None:
                 if collected_output_items:
                     final.output = list(collected_output_items)
                     logger.debug(

agent/codex_runtime.py

@@ -246,12 +246,49 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta
                             sum(len(p) for p in agent._codex_streamed_text_parts),
                             agent._client_log_context(),
                         )
-                final_response = stream.get_final_response()
+                try:
+                    final_response = stream.get_final_response()
+                except TypeError as exc:
+                    err_text = str(exc)
+                    if (
+                        "NoneType" in err_text
+                        and "iterable" in err_text
+                        and (collected_output_items or agent._codex_streamed_text_parts)
+                    ):
+                        logger.warning(
+                            "Codex Responses SDK failed to parse final response after "
+                            "streaming output items; synthesizing response from stream "
+                            "events. %s err=%s",
+                            agent._client_log_context(),
+                            err_text,
+                        )
+                        if collected_output_items:
+                            final_response = SimpleNamespace(
+                                output=list(collected_output_items),
+                                status="completed",
+                                usage=None,
+                                model=api_kwargs.get("model"),
+                            )
+                        else:
+                            assembled = "".join(agent._codex_streamed_text_parts)
+                            final_response = SimpleNamespace(
+                                output=[SimpleNamespace(
+                                    type="message",
+                                    role="assistant",
+                                    status="completed",
+                                    content=[SimpleNamespace(type="output_text", text=assembled)],
+                                )],
+                                status="completed",
+                                usage=None,
+                                model=api_kwargs.get("model"),
+                            )
+                    else:
+                        raise
                 # PATCH: ChatGPT Codex backend streams valid output items
                 # but get_final_response() can return an empty output list.
                 # Backfill from collected items or synthesize from deltas.
                 _out = getattr(final_response, "output", None)
-                if isinstance(_out, list) and not _out:
+                if (isinstance(_out, list) and not _out) or _out is None:
                     if collected_output_items:
                         final_response.output = list(collected_output_items)
                         logger.debug(
@@ -271,6 +308,40 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta
                             len(agent._codex_streamed_text_parts), len(assembled),
                         )
                 return final_response
+        except TypeError as exc:
+            err_text = str(exc)
+            if (
+                "NoneType" in err_text
+                and "iterable" in err_text
+                and (collected_output_items or agent._codex_streamed_text_parts)
+            ):
+                logger.warning(
+                    "Codex Responses SDK failed while parsing stream after "
+                    "streaming output items; synthesizing response from stream "
+                    "events. %s err=%s",
+                    agent._client_log_context(),
+                    err_text,
+                )
+                if collected_output_items:
+                    return SimpleNamespace(
+                        output=list(collected_output_items),
+                        status="completed",
+                        usage=None,
+                        model=api_kwargs.get("model"),
+                    )
+                assembled = "".join(agent._codex_streamed_text_parts)
+                return SimpleNamespace(
+                    output=[SimpleNamespace(
+                        type="message",
+                        role="assistant",
+                        status="completed",
+                        content=[SimpleNamespace(type="output_text", text=assembled)],
+                    )],
+                    status="completed",
+                    usage=None,
+                    model=api_kwargs.get("model"),
+                )
+            raise
         except (_httpx.RemoteProtocolError, _httpx.ReadTimeout, _httpx.ConnectError, ConnectionError) as exc:
             if attempt < max_stream_retries:
                 logger.debug(
@@ -414,7 +485,7 @@ def run_codex_create_stream_fallback(agent, api_kwargs: dict, client: Any = None
             if terminal_response is not None:
                 # Backfill empty output from collected stream events
                 _out = getattr(terminal_response, "output", None)
-                if isinstance(_out, list) and not _out:
+                if (isinstance(_out, list) and not _out) or _out is None:
                     if collected_output_items:
                         terminal_response.output = list(collected_output_items)
                         logger.debug(
@@ -433,6 +504,37 @@ def run_codex_create_stream_fallback(agent, api_kwargs: dict, client: Any = None
                             len(collected_text_deltas), len(assembled),
                         )
                 return terminal_response
+    except TypeError as exc:
+        err_text = str(exc)
+        if (
+            "NoneType" in err_text
+            and "iterable" in err_text
+            and (collected_output_items or collected_text_deltas)
+        ):
+            logger.warning(
+                "Codex create(stream=True) SDK parse failed after streaming "
+                "output items; synthesizing response from stream events. err=%s",
+                err_text,
+            )
+            if collected_output_items:
+                return SimpleNamespace(
+                    output=list(collected_output_items),
+                    status="completed",
+                    usage=None,
+                    model=fallback_kwargs.get("model"),
+                )
+            assembled = "".join(collected_text_deltas)
+            return SimpleNamespace(
+                output=[SimpleNamespace(
+                    type="message", role="assistant",
+                    status="completed",
+                    content=[SimpleNamespace(type="output_text", text=assembled)],
+                )],
+                status="completed",
+                usage=None,
+                model=fallback_kwargs.get("model"),
+            )
+        raise
     finally:
         close_fn = getattr(stream_or_response, "close", None)
         if callable(close_fn):

tests/agent/test_auxiliary_client.py

@@ -2475,6 +2475,65 @@ def test_skip_set_covers_exactly_known_entries(self):
         })
 
 
+class TestCodexAuxiliaryAdapterStreamRecovery:
+    def test_synthesizes_response_when_final_parse_crashes_after_streamed_item(self):
+        message_item = SimpleNamespace(
+            type="message",
+            content=[SimpleNamespace(type="output_text", text="title ok")],
+        )
+
+        class FakeStream:
+            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=message_item)
+
+            def get_final_response(self):
+                raise TypeError("'NoneType' object is not iterable")
+
+        class FakeResponses:
+            def stream(self, **kwargs):
+                return FakeStream()
+
+        fake_client = SimpleNamespace(responses=FakeResponses())
+        adapter = _CodexCompletionsAdapter(fake_client, "gpt-5.5")
+
+        response = adapter.create(messages=[{"role": "user", "content": "title me"}])
+
+        assert response.choices[0].message.content == "title ok"
+        assert response.choices[0].finish_reason == "stop"
+
+    def test_synthesizes_response_when_final_output_is_none_after_text_deltas(self):
+        class FakeStream:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, exc_type, exc, tb):
+                return False
+
+            def __iter__(self):
+                yield SimpleNamespace(type="response.output_text.delta", delta="delta title")
+
+            def get_final_response(self):
+                return SimpleNamespace(output=None, usage=None)
+
+        class FakeResponses:
+            def stream(self, **kwargs):
+                return FakeStream()
+
+        fake_client = SimpleNamespace(responses=FakeResponses())
+        adapter = _CodexCompletionsAdapter(fake_client, "gpt-5.5")
+
+        response = adapter.create(messages=[{"role": "user", "content": "title me"}])
+
+        assert response.choices[0].message.content == "delta title"
+        assert response.choices[0].finish_reason == "stop"
+
+
 class TestCodexAuxiliaryAdapterTimeout:
     def test_forwards_timeout_to_responses_stream(self):
         class FakeStream:

tests/run_agent/test_run_agent_codex_responses.py

@@ -155,9 +155,10 @@ def _codex_ack_message_response(text: str):
 
 
 class _FakeResponsesStream:
-    def __init__(self, *, final_response=None, final_error=None):
+    def __init__(self, *, final_response=None, final_error=None, events=None):
         self._final_response = final_response
         self._final_error = final_error
+        self._events = list(events or [])
 
     def __enter__(self):
         return self
@@ -166,7 +167,7 @@ def __exit__(self, exc_type, exc, tb):
         return False
 
     def __iter__(self):
-        return iter(())
+        return iter(self._events)
 
     def get_final_response(self):
         if self._final_error is not None:
@@ -484,6 +485,61 @@ def _fake_create(**kwargs):
     assert response.output[0].content[0].text == "streamed create ok"
 
 
+def test_run_codex_stream_synthesizes_response_after_sdk_none_output_parse_error(monkeypatch):
+    agent = _build_agent(monkeypatch)
+    message_item = SimpleNamespace(
+        type="message",
+        status="completed",
+        content=[SimpleNamespace(type="output_text", text="stream item ok")],
+    )
+    stream = _FakeResponsesStream(
+        events=[
+            SimpleNamespace(type="response.created"),
+            SimpleNamespace(type="response.output_text.delta", delta="stream item ok"),
+            SimpleNamespace(type="response.output_item.done", item=message_item),
+        ],
+        final_error=TypeError("'NoneType' object is not iterable"),
+    )
+
+    agent.client = SimpleNamespace(
+        responses=SimpleNamespace(
+            stream=lambda **kwargs: stream,
+            create=lambda **kwargs: _codex_message_response("fallback should not run"),
+        )
+    )
+
+    response = agent._run_codex_stream(_codex_request_kwargs())
+    assert response.status == "completed"
+    assert response.output == [message_item]
+
+
+def test_run_codex_stream_synthesizes_response_when_sdk_parse_error_interrupts_iteration(monkeypatch):
+    agent = _build_agent(monkeypatch)
+    message_item = SimpleNamespace(
+        type="message",
+        status="completed",
+        content=[SimpleNamespace(type="output_text", text="iteration ok")],
+    )
+
+    class _ParseErrorStream(_FakeResponsesStream):
+        def __iter__(self):
+            yield SimpleNamespace(type="response.created")
+            yield SimpleNamespace(type="response.output_text.delta", delta="iteration ok")
+            yield SimpleNamespace(type="response.output_item.done", item=message_item)
+            raise TypeError("'NoneType' object is not iterable")
+
+    agent.client = SimpleNamespace(
+        responses=SimpleNamespace(
+            stream=lambda **kwargs: _ParseErrorStream(),
+            create=lambda **kwargs: _codex_message_response("fallback should not run"),
+        )
+    )
+
+    response = agent._run_codex_stream(_codex_request_kwargs())
+    assert response.status == "completed"
+    assert response.output == [message_item]
+
+
 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"))