From 8ea166f2e5a368b04c84efac2f6f4948050cea5c Mon Sep 17 00:00:00 2001 From: Zezae Oh Date: Fri, 3 Jul 2026 11:23:48 +0900 Subject: [PATCH] fix(providers): retry transient Codex response failures (#1334) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backport of #1332 to dev — retry transient Codex failures --- cmd/gateway_cron.go | 1 + internal/providers/codex.go | 69 +++++++++++++++++++++++++---- internal/providers/codex_test.go | 76 ++++++++++++++++++++++++++++++++ internal/providers/retry.go | 12 ++++- internal/providers/retry_test.go | 2 + 5 files changed, 149 insertions(+), 11 deletions(-) diff --git a/cmd/gateway_cron.go b/cmd/gateway_cron.go index ad83b423..5fc4ea0a 100644 --- a/cmd/gateway_cron.go +++ b/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 diff --git a/internal/providers/codex.go b/internal/providers/codex.go index 717a7128..5fe417fb 100644 --- a/internal/providers/codex.go +++ b/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. diff --git a/internal/providers/codex_test.go b/internal/providers/codex_test.go index 8d69aa67..f2a1d806 100644 --- a/internal/providers/codex_test.go +++ b/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) + + } +} diff --git a/internal/providers/retry.go b/internal/providers/retry.go index 9935c82f..de06e045 100644 --- a/internal/providers/retry.go +++ b/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 } diff --git a/internal/providers/retry_test.go b/internal/providers/retry_test.go index bd361d29..806a7ae2 100644 --- a/internal/providers/retry_test.go +++ b/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},