From 25bf31ad97bdb4a6c8e383c9a44fb21d030a7793 Mon Sep 17 00:00:00 2001
From: Billy Thakid <billy@srv1182948>
Date: Wed, 27 May 2026 01:54:43 +0200
Subject: [PATCH] Fix Codex stream None output recovery


diff --git a/agent/codex_responses_adapter.py b/agent/codex_responses_adapter.py
index 07ae5cc95..89ee0dbfa 100644
--- a/agent/codex_responses_adapter.py
+++ b/agent/codex_responses_adapter.py
@@ -872,6 +872,16 @@ def _extract_responses_reasoning_text(item: Any) -> str:
     return ""
 
 
+def _safe_response_output_text(response: Any) -> str:
+    """Read ``response.output_text`` without letting SDK convenience accessors crash."""
+    try:
+        out_text = getattr(response, "output_text", None)
+    except Exception as exc:
+        logger.debug("Codex response.output_text access failed: %s", exc)
+        return ""
+    return out_text.strip() if isinstance(out_text, str) else ""
+
+
 # ---------------------------------------------------------------------------
 # Full response normalization
 # ---------------------------------------------------------------------------
@@ -883,15 +893,15 @@ def _normalize_codex_response(response: Any) -> tuple[Any, str]:
         # The Codex backend can return empty output when the answer was
         # delivered entirely via stream events. Check output_text as a
         # last-resort fallback before raising.
-        out_text = getattr(response, "output_text", None)
-        if isinstance(out_text, str) and out_text.strip():
+        out_text = _safe_response_output_text(response)
+        if out_text:
             logger.debug(
                 "Codex response has empty output but output_text is present (%d chars); "
-                "synthesizing output item.", len(out_text.strip()),
+                "synthesizing output item.", len(out_text),
             )
             output = [SimpleNamespace(
                 type="message", role="assistant", status="completed",
-                content=[SimpleNamespace(type="output_text", text=out_text.strip())],
+                content=[SimpleNamespace(type="output_text", text=out_text)],
             )]
             response.output = output
         else:
@@ -1024,10 +1034,8 @@ def _normalize_codex_response(response: Any) -> tuple[Any, str]:
             ))
 
     final_text = "\n".join([p for p in content_parts if p]).strip()
-    if not final_text and hasattr(response, "output_text"):
-        out_text = getattr(response, "output_text", "")
-        if isinstance(out_text, str):
-            final_text = out_text.strip()
+    if not final_text:
+        final_text = _safe_response_output_text(response)
 
     # ── Tool-call leak recovery ──────────────────────────────────
     # gpt-5.x on the Codex Responses API sometimes degenerates and emits
@@ -1076,7 +1084,7 @@ def _normalize_codex_response(response: Any) -> tuple[Any, str]:
         finish_reason = "incomplete"
     elif has_incomplete_items or (saw_commentary_phase and not saw_final_answer_phase):
         finish_reason = "incomplete"
-    elif reasoning_items_raw and not final_text:
+    elif (reasoning_items_raw or reasoning_parts) and not final_text:
         # Response contains only reasoning (encrypted thinking state) with
         # no visible content or tool calls.  The model is still thinking and
         # needs another turn to produce the actual answer.  Marking this as
diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py
index 8c5dff39b..2c3d131d7 100644
--- a/agent/codex_runtime.py
+++ b/agent/codex_runtime.py
@@ -176,6 +176,91 @@ def run_codex_app_server_turn(
 
 
 
+def _backfill_codex_response_output(
+    response: Any,
+    *,
+    collected_output_items: list,
+    text_deltas: list,
+    reasoning_deltas: list,
+    has_tool_calls: bool,
+    log_label: str,
+) -> bool:
+    """Patch missing/empty Responses output from stream events already seen."""
+    _out = getattr(response, "output", None)
+    if isinstance(_out, list) and _out:
+        return False
+
+    if collected_output_items:
+        response.output = list(collected_output_items)
+        logger.debug(
+            "%s: backfilled %d output items from stream events",
+            log_label,
+            len(collected_output_items),
+        )
+        return True
+
+    if text_deltas and not has_tool_calls:
+        assembled = "".join(text_deltas)
+        response.output = [SimpleNamespace(
+            type="message",
+            role="assistant",
+            status="completed",
+            content=[SimpleNamespace(type="output_text", text=assembled)],
+        )]
+        logger.debug(
+            "%s: synthesized output from %d text deltas (%d chars)",
+            log_label,
+            len(text_deltas),
+            len(assembled),
+        )
+        return True
+
+    reasoning_text = "".join(reasoning_deltas).strip()
+    if reasoning_text:
+        response.output = [SimpleNamespace(
+            type="reasoning",
+            status="completed",
+            summary=[SimpleNamespace(type="summary_text", text=reasoning_text)],
+        )]
+        if not getattr(response, "status", None):
+            response.status = "incomplete"
+        logger.debug(
+            "%s: synthesized reasoning-only output from %d deltas (%d chars)",
+            log_label,
+            len(reasoning_deltas),
+            len(reasoning_text),
+        )
+        return True
+
+    return False
+
+
+def _synthesize_codex_response_from_stream(
+    *,
+    collected_output_items: list,
+    text_deltas: list,
+    reasoning_deltas: list,
+    has_tool_calls: bool,
+    reason: str,
+) -> Any | None:
+    response = SimpleNamespace(
+        output=[],
+        status="incomplete",
+        incomplete_details=SimpleNamespace(reason=reason),
+        error=None,
+    )
+    if _backfill_codex_response_output(
+        response,
+        collected_output_items=collected_output_items,
+        text_deltas=text_deltas,
+        reasoning_deltas=reasoning_deltas,
+        has_tool_calls=has_tool_calls,
+        log_label="Codex stream recovery",
+    ):
+        return response
+    return None
+
+
 def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta: callable = None):
     """Execute one streaming Responses API request and return the final response."""
     import httpx as _httpx
@@ -192,6 +277,7 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta
         if agent._interrupt_requested:
             raise InterruptedError("Agent interrupted before Codex stream retry")
         collected_output_items: list = []
+        collected_reasoning_deltas: list = []
         try:
             with active_client.responses.stream(**api_kwargs) as stream:
                 for event in stream:
@@ -225,6 +311,7 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta
                     elif "reasoning" in event_type and "delta" in event_type:
                         reasoning_text = getattr(event, "delta", "")
                         if reasoning_text:
+                            collected_reasoning_deltas.append(reasoning_text)
                             agent._fire_reasoning_delta(reasoning_text)
                     # Collect completed output items — some backends
                     # (chatgpt.com/backend-api/codex) stream valid items
@@ -248,28 +335,16 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta
                         )
                 final_response = stream.get_final_response()
                 # PATCH: ChatGPT Codex backend streams valid output items
-                # but get_final_response() can return an empty output list.
+                # but get_final_response() can return missing/empty output.
                 # Backfill from collected items or synthesize from deltas.
-                _out = getattr(final_response, "output", None)
-                if isinstance(_out, list) and not _out:
-                    if collected_output_items:
-                        final_response.output = list(collected_output_items)
-                        logger.debug(
-                            "Codex stream: backfilled %d output items from stream events",
-                            len(collected_output_items),
-                        )
-                    elif agent._codex_streamed_text_parts and not has_tool_calls:
-                        assembled = "".join(agent._codex_streamed_text_parts)
-                        final_response.output = [SimpleNamespace(
-                            type="message",
-                            role="assistant",
-                            status="completed",
-                            content=[SimpleNamespace(type="output_text", text=assembled)],
-                        )]
-                        logger.debug(
-                            "Codex stream: synthesized output from %d text deltas (%d chars)",
-                            len(agent._codex_streamed_text_parts), len(assembled),
-                        )
+                _backfill_codex_response_output(
+                    final_response,
+                    collected_output_items=collected_output_items,
+                    text_deltas=agent._codex_streamed_text_parts,
+                    reasoning_deltas=collected_reasoning_deltas,
+                    has_tool_calls=has_tool_calls,
+                    log_label="Codex stream",
+                )
                 return final_response
         except (_httpx.RemoteProtocolError, _httpx.ReadTimeout, _httpx.ConnectError, ConnectionError) as exc:
             if attempt < max_stream_retries:
@@ -335,6 +410,35 @@ 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:
+            err_text = str(exc)
+            sdk_none_output = "'NoneType' object is not iterable" in err_text
+            if not sdk_none_output:
+                raise
+            logger.warning(
+                "Codex Responses stream parser failed on missing output; "
+                "falling back to create(stream=True). %s err=%s",
+                agent._client_log_context(),
+                err_text,
+            )
+            try:
+                return agent._run_codex_create_stream_fallback(api_kwargs, client=active_client)
+            except Exception:
+                recovered = _synthesize_codex_response_from_stream(
+                    collected_output_items=collected_output_items,
+                    text_deltas=agent._codex_streamed_text_parts,
+                    reasoning_deltas=collected_reasoning_deltas,
+                    has_tool_calls=has_tool_calls,
+                    reason="stream_parser_recovered_none_output",
+                )
+                if recovered is not None:
+                    logger.warning(
+                        "Codex Responses stream parser failed and fallback failed; "
+                        "returning synthesized incomplete response. %s",
+                        agent._client_log_context(),
+                    )
+                    return recovered
+                raise
 
 
 
@@ -355,6 +459,8 @@ def run_codex_create_stream_fallback(agent, api_kwargs: dict, client: Any = None
     terminal_response = None
     collected_output_items: list = []
     collected_text_deltas: list = []
+    collected_reasoning_deltas: list = []
+    has_tool_calls = False
     try:
         for event in stream_or_response:
             agent._touch_activity("receiving stream response")
@@ -404,6 +510,14 @@ def run_codex_create_stream_fallback(agent, api_kwargs: dict, client: Any = None
                     delta = event.get("delta", "")
                 if delta:
                     collected_text_deltas.append(delta)
+            elif event_type and "function_call" in event_type:
+                has_tool_calls = True
+            elif event_type and "reasoning" in event_type and "delta" in event_type:
+                delta = getattr(event, "delta", "")
+                if not delta and isinstance(event, dict):
+                    delta = event.get("delta", "")
+                if delta:
+                    collected_reasoning_deltas.append(delta)
 
             if event_type not in {"response.completed", "response.incomplete", "response.failed"}:
                 continue
@@ -413,25 +527,14 @@ def run_codex_create_stream_fallback(agent, api_kwargs: dict, client: Any = None
                 terminal_response = event.get("response")
             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 collected_output_items:
-                        terminal_response.output = list(collected_output_items)
-                        logger.debug(
-                            "Codex fallback stream: backfilled %d output items",
-                            len(collected_output_items),
-                        )
-                    elif collected_text_deltas:
-                        assembled = "".join(collected_text_deltas)
-                        terminal_response.output = [SimpleNamespace(
-                            type="message", role="assistant",
-                            status="completed",
-                            content=[SimpleNamespace(type="output_text", text=assembled)],
-                        )]
-                        logger.debug(
-                            "Codex fallback stream: synthesized from %d deltas (%d chars)",
-                            len(collected_text_deltas), len(assembled),
-                        )
+                _backfill_codex_response_output(
+                    terminal_response,
+                    collected_output_items=collected_output_items,
+                    text_deltas=collected_text_deltas,
+                    reasoning_deltas=collected_reasoning_deltas,
+                    has_tool_calls=has_tool_calls,
+                    log_label="Codex fallback stream",
+                )
                 return terminal_response
     finally:
         close_fn = getattr(stream_or_response, "close", None)
@@ -443,6 +546,15 @@ def run_codex_create_stream_fallback(agent, api_kwargs: dict, client: Any = None
 
     if terminal_response is not None:
         return terminal_response
+    recovered = _synthesize_codex_response_from_stream(
+        collected_output_items=collected_output_items,
+        text_deltas=collected_text_deltas,
+        reasoning_deltas=collected_reasoning_deltas,
+        has_tool_calls=has_tool_calls,
+        reason="stream_ended_without_terminal_response",
+    )
+    if recovered is not None:
+        return recovered
     raise RuntimeError("Responses create(stream=True) fallback did not emit a terminal response.")
 
 
diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py
index 35a64df48..dc860898a 100644
--- a/agent/conversation_loop.py
+++ b/agent/conversation_loop.py
@@ -1224,7 +1224,14 @@ def run_conversation(
                             else:
                                 # output_text fallback: stream backfill may have failed
                                 # but normalize can still recover from output_text
-                                _out_text = getattr(response, "output_text", None)
+                                try:
+                                    _out_text = getattr(response, "output_text", None)
+                                except Exception as exc:
+                                    logger.debug(
+                                        "Codex response.output_text access failed during validation: %s",
+                                        exc,
+                                    )
+                                    _out_text = None
                                 _out_text_stripped = _out_text.strip() if isinstance(_out_text, str) else ""
                                 if _out_text_stripped:
                                     logger.debug(
diff --git a/tests/agent/transports/test_codex_transport.py b/tests/agent/transports/test_codex_transport.py
index 96a808272..b8a5ae5e5 100644
--- a/tests/agent/transports/test_codex_transport.py
+++ b/tests/agent/transports/test_codex_transport.py
@@ -453,6 +453,34 @@ class TestCodexNormalizeResponse:
         assert tc.name == "terminal"
         assert '"command"' in tc.arguments
 
+    def test_reasoning_only_response_with_broken_output_text_is_incomplete(self, transport):
+        """SDK output_text access can fail when output contains no final message."""
+
+        class BrokenOutputTextResponse(SimpleNamespace):
+            @property
+            def output_text(self):
+                raise TypeError("'NoneType' object is not iterable")
+
+        r = BrokenOutputTextResponse(
+            output=[
+                SimpleNamespace(
+                    type="reasoning",
+                    summary=[SimpleNamespace(type="summary_text", text="Still thinking")],
+                    status="completed",
+                ),
+            ],
+            status="completed",
+            incomplete_details=None,
+            usage=SimpleNamespace(input_tokens=10, output_tokens=5,
+                                  input_tokens_details=None, output_tokens_details=None),
+        )
+
+        nr = transport.normalize_response(r)
+
+        assert nr.content == ""
+        assert nr.reasoning == "Still thinking"
+        assert nr.finish_reason == "incomplete"
+
 
 
 class TestCodexTransportTimeout:
diff --git a/tests/run_agent/test_run_agent_codex_responses.py b/tests/run_agent/test_run_agent_codex_responses.py
index bc575cc67..df6ea0d60 100644
--- a/tests/run_agent/test_run_agent_codex_responses.py
+++ b/tests/run_agent/test_run_agent_codex_responses.py
@@ -484,6 +484,58 @@ def test_run_codex_stream_fallback_parses_create_stream_events(monkeypatch):
     assert response.output[0].content[0].text == "streamed create ok"
 
 
+def test_run_codex_stream_parser_none_output_falls_back_to_create_stream(monkeypatch):
+    agent = _build_agent(monkeypatch)
+    calls = {"stream": 0, "create": 0}
+
+    class _BrokenParseStream:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def __iter__(self):
+            yield SimpleNamespace(type="response.reasoning.delta", delta="Primary reasoning")
+            raise TypeError("'NoneType' object is not iterable")
+
+        def get_final_response(self):
+            raise AssertionError("stream parser failure should happen during iteration")
+
+    create_stream = _FakeCreateStream(
+        [
+            SimpleNamespace(type="response.created"),
+            SimpleNamespace(type="response.reasoning.delta", delta="Fallback reasoning"),
+            SimpleNamespace(
+                type="response.completed",
+                response=SimpleNamespace(output=None, status="completed"),
+            ),
+        ]
+    )
+
+    def _fake_stream(**kwargs):
+        calls["stream"] += 1
+        return _BrokenParseStream()
+
+    def _fake_create(**kwargs):
+        calls["create"] += 1
+        assert kwargs.get("stream") is True
+        return create_stream
+
+    agent.client = SimpleNamespace(
+        responses=SimpleNamespace(
+            stream=_fake_stream,
+            create=_fake_create,
+        )
+    )
+
+    response = agent._run_codex_stream(_codex_request_kwargs())
+
+    assert calls == {"stream": 1, "create": 1}
+    assert response.output[0].type == "reasoning"
+    assert response.output[0].summary[0].text == "Fallback reasoning"
+
+
 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"))
-- 
2.43.0