fix(codex): recover from null terminal stream output

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)

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
 
 
 

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

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"))