fix(codex): stop draining interrupted streams

Signed-off-by: Alex Fournier <afournier@nvidia.com>

agent/codex_runtime.py

@@ -1350,8 +1350,9 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta
                 # 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.
-                for _ignored in event_stream:
-                    pass
+                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(

tests/run_agent/test_stream_single_writer_65991.py

@@ -208,6 +208,55 @@ class TestCodexSingleWriter:
 
         assert "".join(delivered) == "hello world"
 
+    def test_codex_interrupt_closes_stream_without_draining_provider(self):
+        from agent.codex_runtime import run_codex_stream
+
+        agent = _make_agent()
+        agent.api_mode = "codex_responses"
+        produced = []
+        stream_closed = threading.Event()
+
+        def interrupt_after_first_delta(_text):
+            agent._interrupt_requested = True
+
+        agent.stream_delta_callback = interrupt_after_first_delta
+        agent._stream_callback = None
+
+        def event_gen():
+            try:
+                produced.append("first")
+                yield self._codex_event(
+                    "response.output_text.delta",
+                    delta="first",
+                    item_id="i1",
+                )
+                produced.append("lookahead")
+                yield self._codex_event(
+                    "response.output_text.delta",
+                    delta="-unused",
+                    item_id="i1",
+                )
+                produced.append("terminal")
+                yield self._codex_event(
+                    "response.completed",
+                    response=SimpleNamespace(
+                        id="r1",
+                        status="completed",
+                        output=[],
+                        usage=None,
+                    ),
+                )
+            finally:
+                stream_closed.set()
+
+        mock_client = MagicMock()
+        mock_client.responses.create.return_value = event_gen()
+
+        run_codex_stream(agent, {"model": "gpt-5.3-codex"}, client=mock_client)
+
+        assert produced == ["first", "lookahead"]
+        assert stream_closed.is_set()
+
 
 if __name__ == "__main__":
     raise SystemExit(pytest.main([__file__, "-q"]))