fix(providers): retry transient Codex response failures (#1334)

Backport of #1332 to dev — retry transient Codex failures

cmd/gateway_cron.go

@@ -139,6 +139,7 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
 		var modelOverride string
 		if job.Model != nil {
 			modelOverride = *job.Model
+
 		}
 
 		// Schedule through cron lane — scheduler handles agent resolution and concurrency

internal/providers/codex.go

@@ -5,9 +5,9 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"net/http"
 	"strings"
+	"time"
 )
 
 type CodexRoutingDefaults struct {
@@ -131,6 +131,42 @@ func (p *CodexProvider) middlewareConfig(req ChatRequest) MiddlewareConfig {
 }
 
 func (p *CodexProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk func(StreamChunk)) (*ChatResponse, error) {
+	cfg := p.retryConfig
+	if cfg.Attempts <= 0 {
+		cfg.Attempts = 1
+	}
+
+	var lastErr error
+	for attempt := 1; attempt <= cfg.Attempts; attempt++ {
+		result, emitted, err := p.chatStreamOnce(ctx, req, onChunk)
+		if err == nil {
+			return result, nil
+		}
+		lastErr = err
+
+		// If any user-visible chunk already escaped, do not replay the run: retrying
+		// could duplicate streamed text/tool calls/images. Pre-output Codex backend
+		// failures (e.g. response.failed: "processing your request…retry") are safe
+		// to retry here and are exactly the flaky 429-ish case this guard targets.
+		if emitted || !IsRetryableError(err) || attempt == cfg.Attempts {
+			return result, err
+		}
+
+		delay := computeDelay(cfg, attempt, err)
+		if hook := retryHookFromContext(ctx); hook != nil {
+			hook(attempt, cfg.Attempts, err)
+		}
+		select {
+		case <-ctx.Done():
+			return result, ctx.Err()
+		case <-time.After(delay):
+		}
+	}
+
+	return nil, lastErr
+}
+
+func (p *CodexProvider) chatStreamOnce(ctx context.Context, req ChatRequest, onChunk func(StreamChunk)) (*ChatResponse, bool, error) {
 	// stripThinking: drop reasoning summaries from ChatResponse.Thinking and
 	// onChunk callbacks. Usage.ThinkingTokens is still populated from the
 	// final response.usage payload (Phase 1 billing accuracy).
@@ -138,11 +174,9 @@ func (p *CodexProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk
 	body := p.buildRequestBody(req, true)
 	body = ApplyMiddlewares(body, p.middlewares, p.middlewareConfig(req))
 
-	respBody, err := RetryDo(ctx, p.retryConfig, func() (io.ReadCloser, error) {
-		return p.doRequest(ctx, body)
-	})
+	respBody, err := p.doRequest(ctx, body)
 	if err != nil {
-		return nil, err
+		return nil, false, err
 	}
 	// Wrap respBody so ctx cancellation closes the socket, unblocking bufio.Scanner.
 	cb := NewCtxBody(ctx, respBody)
@@ -152,6 +186,16 @@ func (p *CodexProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk
 	toolCalls := make(map[string]*codexToolCallAcc) // keyed by item_id
 	streamState := newCodexMessageStreamState()
 	imageState := newCodexImageState()
+	emitted := false
+	wrappedOnChunk := onChunk
+	if onChunk != nil {
+		wrappedOnChunk = func(chunk StreamChunk) {
+			if chunk.Content != "" || chunk.Thinking != "" || len(chunk.Images) > 0 {
+				emitted = true
+			}
+			onChunk(chunk)
+		}
+	}
 
 	sse := NewSSEScanner(cb)
 	for sse.Next() {
@@ -162,13 +206,13 @@ func (p *CodexProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk
 			continue
 		}
 
-		if err := p.processSSEEvent(&event, result, toolCalls, streamState, imageState, onChunk, stripThinking); err != nil {
-			return nil, err
+		if err := p.processSSEEvent(&event, result, toolCalls, streamState, imageState, wrappedOnChunk, stripThinking); err != nil {
+			return result, emitted || codexResultHasVisibleOutput(result), err
 		}
 	}
 
 	if err := sse.Err(); err != nil {
-		return result, fmt.Errorf("%s: stream read error: %w", p.name, err)
+		return result, emitted || codexResultHasVisibleOutput(result), fmt.Errorf("%s: stream read error: %w", p.name, err)
 	}
 
 	// Assemble generated images from image accumulator into ChatResponse.
@@ -202,7 +246,14 @@ func (p *CodexProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk
 		onChunk(StreamChunk{Done: true})
 	}
 
-	return result, nil
+	return result, emitted || codexResultHasVisibleOutput(result), nil
+}
+
+func codexResultHasVisibleOutput(result *ChatResponse) bool {
+	if result == nil {
+		return false
+	}
+	return result.Content != "" || result.Thinking != "" || len(result.ToolCalls) > 0 || len(result.Images) > 0
 }
 
 // processSSEEvent handles a single SSE event during streaming.

internal/providers/codex_test.go

@@ -8,6 +8,7 @@ import (
 	"net/http/httptest"
 	"strings"
 	"testing"
+	"time"
 )
 
 // staticTokenSource implements TokenSource for testing.
@@ -1154,3 +1155,78 @@ func TestCodexProviderCapabilitiesCacheControl(t *testing.T) {
 		t.Fatal("CodexProvider Capabilities().CacheControl = false, want true")
 	}
 }
+
+func TestCodexProviderRetriesPreOutputResponseFailed(t *testing.T) {
+	var calls int
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls++
+		w.Header().Set("Content-Type", "text/event-stream")
+		flusher := w.(http.Flusher)
+		if calls == 1 {
+			fmt.Fprintf(w, "data: %s\n\n", mustJSON(codexSSEEvent{
+				Type:     "response.failed",
+				Response: &codexAPIResponse{Error: &codexErrorDetail{Message: "An error occurred while processing your request. You can retry your request."}},
+			}))
+			flusher.Flush()
+			return
+		}
+		fmt.Fprintf(w, "data: %s\n\n", mustJSON(codexSSEEvent{Type: "response.output_text.delta", Delta: "ok"}))
+		fmt.Fprintf(w, "data: %s\n\n", mustJSON(codexSSEEvent{Type: "response.completed", Response: &codexAPIResponse{Status: "completed"}}))
+		fmt.Fprint(w, "data: [DONE]\n\n")
+		flusher.Flush()
+	}))
+	defer server.Close()
+
+	p := NewCodexProvider("test", &staticTokenSource{token: "tok"}, server.URL, "gpt-4o")
+	p.retryConfig = RetryConfig{Attempts: 2, MinDelay: time.Millisecond, MaxDelay: time.Millisecond}
+
+	result, err := p.ChatStream(context.Background(), ChatRequest{Messages: []Message{{Role: "user", Content: "hi"}}}, nil)
+	if err != nil {
+		t.Fatalf("ChatStream: %v", err)
+	}
+	if calls != 2 {
+		t.Fatalf("calls = %d, want 2", calls)
+	}
+	if result.Content != "ok" {
+		t.Fatalf("content = %q, want ok", result.Content)
+	}
+}
+
+func TestCodexProviderDoesNotRetryAfterVisibleOutput(t *testing.T) {
+	var calls int
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls++
+		w.Header().Set("Content-Type", "text/event-stream")
+		flusher := w.(http.Flusher)
+		fmt.Fprintf(w, "data: %s\n\n", mustJSON(codexSSEEvent{Type: "response.output_text.delta", Delta: "partial"}))
+		fmt.Fprintf(w, "data: %s\n\n", mustJSON(codexSSEEvent{
+			Type:     "response.failed",
+			Response: &codexAPIResponse{Error: &codexErrorDetail{Message: "An error occurred while processing your request. You can retry your request."}},
+		}))
+		flusher.Flush()
+	}))
+	defer server.Close()
+
+	p := NewCodexProvider("test", &staticTokenSource{token: "tok"}, server.URL, "gpt-4o")
+	p.retryConfig = RetryConfig{Attempts: 2, MinDelay: time.Millisecond, MaxDelay: time.Millisecond}
+
+	var chunks []string
+	result, err := p.ChatStream(context.Background(), ChatRequest{Messages: []Message{{Role: "user", Content: "hi"}}}, func(chunk StreamChunk) {
+		if chunk.Content != "" {
+			chunks = append(chunks, chunk.Content)
+		}
+	})
+	if err == nil {
+		t.Fatal("expected error")
+	}
+	if calls != 1 {
+		t.Fatalf("calls = %d, want 1", calls)
+	}
+	if result == nil || result.Content != "partial" {
+		t.Fatalf("result content = %#v, want partial", result)
+	}
+	if len(chunks) != 1 || chunks[0] != "partial" {
+		t.Fatalf("chunks = %#v, want [partial]", chunks)
+
+	}
+}

internal/providers/retry.go

@@ -83,12 +83,20 @@ func IsRetryableError(err error) bool {
 		return true // includes timeouts
 	}
 
-	// Check for connection reset / broken pipe / EOF in error string
+	// Check for connection reset / broken pipe / EOF in error string.
+	// Some streaming providers (notably Codex/Responses) surface transient
+	// backend failures as SSE `response.failed` events instead of HTTP 429/5xx.
+	// Treat explicit retry guidance/rate-limit wording as retryable too.
 	errStr := err.Error()
+	lowerErr := strings.ToLower(errStr)
 	if strings.Contains(errStr, "connection reset") ||
 		strings.Contains(errStr, "broken pipe") ||
 		strings.Contains(errStr, "EOF") ||
-		strings.Contains(errStr, "timeout") {
+		strings.Contains(lowerErr, "timeout") ||
+		strings.Contains(lowerErr, "rate limit") ||
+		strings.Contains(lowerErr, "429") ||
+		(strings.Contains(lowerErr, "response failed") && strings.Contains(lowerErr, "retry")) ||
+		strings.Contains(lowerErr, "while processing your request") {
 		return true
 	}
 

internal/providers/retry_test.go

@@ -31,6 +31,8 @@ func TestIsRetryableError(t *testing.T) {
 		{"broken_pipe", errors.New("write: broken pipe"), true},
 		{"eof", errors.New("unexpected EOF"), true},
 		{"timeout_string", errors.New("i/o timeout"), true},
+		{"codex_retry_guidance", errors.New("codex: response failed: An error occurred while processing your request. You can retry your request, or contact us through our help center."), true},
+		{"codex_rate_limit_wording", errors.New("codex: response failed: rate limit exceeded"), true},
 		{"generic_error", errors.New("something went wrong"), false},
 		{"wrapped_retryable", fmt.Errorf("provider: %w", &HTTPError{Status: 429}), true},
 		{"wrapped_non_retryable", fmt.Errorf("provider: %w", &HTTPError{Status: 400}), false},