diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 7b8bed6137bb9..c01084c4a4499 100644 --- a/agent/codex_runtime.py +++ b/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 " diff --git a/tests/run_agent/test_run_agent_codex_responses.py b/tests/run_agent/test_run_agent_codex_responses.py index 464537f0b4de5..071dc5100fe1a 100644 --- a/tests/run_agent/test_run_agent_codex_responses.py +++ b/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):