From c9747106b683428d2c7cc910ae20a943cfae960c Mon Sep 17 00:00:00 2001
From: Alex Fournier <afournier@nvidia.com>
Date: Thu, 23 Jul 2026 07:22:38 -0700
Subject: [PATCH] fix(runtime): retry Relay stream construction failures

Signed-off-by: Alex Fournier <afournier@nvidia.com>
---
 agent/codex_runtime.py | 72 ++++++++++++++++++++++++++----------------
 1 file changed, 45 insertions(+), 27 deletions(-)

diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py
index 525e58e4e1d..f6862e7a05f 100644
--- a/agent/codex_runtime.py
+++ b/agent/codex_runtime.py
@@ -1278,34 +1278,52 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta
                 model=api_kwargs.get("model"),
             )
 
-        event_stream = relay_llm.stream(
-            dict(api_kwargs),
-            _open_codex_stream,
-            session_id=str(getattr(agent, "session_id", "") or ""),
-            name=str(getattr(agent, "provider", "") or "codex"),
-            model_name=str(api_kwargs.get("model") or ""),
-            finalizer=_finalize_codex_stream,
-            on_stream_created=_codex_stream_created,
-            on_chunk=intercepted_events.append,
-            chunk_adapter=lambda chunk: chunk,
-            accept_chunk=_accept_codex_chunk,
-            completed_response_predicate=lambda response: bool(
-                hasattr(response, "output") and not hasattr(response, "__iter__")
-            ),
-            metadata={
-                "api_mode": "codex_responses",
-                "api_request_id": getattr(agent, "_current_api_request_id", None),
-                "call_role": (
-                    "delegated"
-                    if getattr(agent, "is_subagent", False)
-                    else "fallback"
-                    if int(getattr(agent, "_fallback_index", 0) or 0) > 0
-                    else "primary"
+        try:
+            event_stream = relay_llm.stream(
+                dict(api_kwargs),
+                _open_codex_stream,
+                session_id=str(getattr(agent, "session_id", "") or ""),
+                name=str(getattr(agent, "provider", "") or "codex"),
+                model_name=str(api_kwargs.get("model") or ""),
+                finalizer=_finalize_codex_stream,
+                on_stream_created=_codex_stream_created,
+                on_chunk=intercepted_events.append,
+                chunk_adapter=lambda chunk: chunk,
+                accept_chunk=_accept_codex_chunk,
+                completed_response_predicate=lambda response: bool(
+                    hasattr(response, "output") and not hasattr(response, "__iter__")
                 ),
-                "retry_count": attempt,
-            },
-            defer_logical_completion=True,
-        )
+                metadata={
+                    "api_mode": "codex_responses",
+                    "api_request_id": getattr(agent, "_current_api_request_id", None),
+                    "call_role": (
+                        "delegated"
+                        if getattr(agent, "is_subagent", False)
+                        else "fallback"
+                        if int(getattr(agent, "_fallback_index", 0) or 0) > 0
+                        else "primary"
+                    ),
+                    "retry_count": attempt,
+                },
+                defer_logical_completion=True,
+            )
+        except (
+            _httpx.RemoteProtocolError,
+            _httpx.ReadTimeout,
+            _httpx.ConnectError,
+            ConnectionError,
+        ) as exc:
+            if attempt < max_stream_retries:
+                logger.debug(
+                    "Codex Responses stream connect failed (attempt %s/%s); "
+                    "retrying. %s error=%s",
+                    attempt + 1,
+                    max_stream_retries + 1,
+                    agent._client_log_context(),
+                    exc,
+                )
+                continue
+            raise
 
         def _interrupt_or_superseded() -> bool:
             return bool(agent._interrupt_requested)