Merge pull request #74348 from JoaoMarcos44/fix/ia-03-codex-post-terminal-retry

fix(codex): stop duplicating billed inferences on post-terminal drain errors

agent/codex_runtime.py

@@ -1365,12 +1365,6 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta
                     on_event=_on_event,
                     interrupt_check=_interrupt_or_superseded,
                 )
-                # The terminal SSE frame is contractually last. Request the
-                # end-of-stream marker so Relay can run its response finalizer
-                # and close the physical attempt scope before Hermes returns.
-                if not agent._interrupt_requested:
-                    for _ignored in event_stream:
-                        pass
             except (_httpx.RemoteProtocolError, _httpx.ReadTimeout, _httpx.ConnectError, ConnectionError) as exc:
                 if attempt < max_stream_retries:
                     logger.debug(
@@ -1386,6 +1380,25 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta
                     return event_stream.final_response
                 raise
 
+            # A terminal response has already been assembled at this point
+            # (``final`` is built), so a transport error while draining the
+            # rest of the iterator — done only to let Relay run its response
+            # finalizer — must NOT discard it or trigger a new physical
+            # request. Record it as a non-fatal finalization warning and
+            # still return the already-completed, already-billed response.
+            if not agent._interrupt_requested:
+                try:
+                    for _ignored in event_stream:
+                        pass
+                except (_httpx.RemoteProtocolError, _httpx.ReadTimeout, _httpx.ConnectError, ConnectionError) as exc:
+                    logger.warning(
+                        "Codex Responses stream transport finalization failed "
+                        "after a terminal response was already received; "
+                        "returning the completed response instead of "
+                        "retrying. %s error=%s",
+                        agent._client_log_context(), exc,
+                    )
+
             if final.status in {"incomplete", "failed"}:
                 logger.warning(
                     "Codex Responses stream terminal status=%s "

tests/run_agent/test_run_agent_codex_responses.py

@@ -607,10 +607,72 @@ def test_run_codex_stream_delivers_redacted_commentary_once(monkeypatch):
 
 
 
+def test_run_codex_stream_returns_terminal_response_when_post_terminal_drain_fails(
+    monkeypatch, caplog
+):
+    """Regression test for issue #74310.
 
+    A transport error while draining the SSE iterator *after* a valid
+    ``response.completed`` has already been observed (and the response
+    object fully assembled) must NOT discard that response and retry with
+    a brand-new physical request -- that would silently duplicate an
+    already-billed inference. Only errors that occur BEFORE a terminal
+    event is captured should trigger the retry-with-new-request path.
+    """
+    import logging
 
+    import httpx
 
+    agent = _build_agent(monkeypatch)
 
+    message_item = SimpleNamespace(
+        type="message",
+        status="completed",
+        content=[SimpleNamespace(type="output_text", text="All done.")],
+    )
+    usage = SimpleNamespace(input_tokens=10, output_tokens=6, total_tokens=16)
+
+    class _PostTerminalDroppingStream(_FakeCreateStream):
+        """Yields events normally, then raises only on the *next* pull --
+        i.e. after ``response.completed`` has already been consumed and the
+        event-driven parser has broken out of its loop."""
+
+        def __iter__(self):
+            yield from super().__iter__()
+            raise httpx.RemoteProtocolError("connection dropped during drain")
+
+    events = [
+        SimpleNamespace(type="response.output_item.done", item=message_item),
+        SimpleNamespace(
+            type="response.completed",
+            response=SimpleNamespace(
+                status="completed",
+                usage=usage,
+                id="resp_post_terminal_1",
+            ),
+        ),
+    ]
+
+    calls = {"count": 0}
+
+    def _fake_create(**kwargs):
+        calls["count"] += 1
+        return _PostTerminalDroppingStream(events)
+
+    agent.client = SimpleNamespace(responses=SimpleNamespace(create=_fake_create))
+
+    with caplog.at_level(logging.WARNING, logger="agent.codex_runtime"):
+        response = agent._run_codex_stream(_codex_request_kwargs())
+
+    # Only ONE physical request was ever opened -- the drain failure did not
+    # trigger a second call to responses.create(stream=True).
+    assert calls["count"] == 1
+    assert response.status == "completed"
+    assert response.usage is usage
+    assert response.id == "resp_post_terminal_1"
+    assert any(
+        "finalization" in record.message for record in caplog.records
+    )
 
 
 def test_run_conversation_codex_plain_text(monkeypatch):