diff --git a/internal/runtime/executor/openai_compat_executor.go b/internal/runtime/executor/openai_compat_executor.go index ee679d6d8f..385058d303 100644 --- a/internal/runtime/executor/openai_compat_executor.go +++ b/internal/runtime/executor/openai_compat_executor.go @@ -193,7 +193,7 @@ func (e *OpenAICompatExecutor) Execute(ctx context.Context, auth *cliproxyauth.A b, _ := io.ReadAll(httpResp.Body) helps.AppendAPIResponseChunk(ctx, e.cfg, b) helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), b)) - err = statusErr{code: httpResp.StatusCode, msg: string(b)} + err = newOpenAICompatStatusError(httpResp.StatusCode, httpResp.Header, b) return resp, err } body, err := io.ReadAll(httpResp.Body) @@ -294,7 +294,7 @@ func (e *OpenAICompatExecutor) executeImages(ctx context.Context, auth *cliproxy if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), body)) - err = statusErr{code: httpResp.StatusCode, msg: string(body)} + err = newOpenAICompatStatusError(httpResp.StatusCode, httpResp.Header, body) return resp, err } @@ -405,7 +405,7 @@ func (e *OpenAICompatExecutor) ExecuteStream(ctx context.Context, auth *cliproxy if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("openai compat executor: close response body error: %v", errClose) } - err = statusErr{code: httpResp.StatusCode, msg: string(b)} + err = newOpenAICompatStatusError(httpResp.StatusCode, httpResp.Header, b) return nil, err } out := make(chan cliproxyexecutor.StreamChunk) @@ -1024,3 +1024,45 @@ func (e statusErr) Error() string { } func (e statusErr) StatusCode() int { return e.code } func (e statusErr) RetryAfter() *time.Duration { return e.retryAfter } + +const openAICompatTPMFallbackRetryAfter = time.Minute + +func newOpenAICompatStatusError(status int, headers http.Header, body []byte) statusErr { + return statusErr{ + code: status, + msg: string(body), + retryAfter: openAICompatRetryAfter(status, headers, body, time.Now()), + } +} + +// openAICompatRetryAfter preserves the provider's standard Retry-After signal. +// Some OpenAI-compatible providers omit that header for explicit per-minute +// token limits; in that narrow case a one-minute fallback prevents immediate +// replay of the same large request while keeping the retry wait bounded. +func openAICompatRetryAfter(status int, headers http.Header, body []byte, now time.Time) *time.Duration { + if status != http.StatusTooManyRequests { + return nil + } + if raw := strings.TrimSpace(headers.Get("Retry-After")); raw != "" { + if seconds, errParse := strconv.ParseInt(raw, 10, 64); errParse == nil && seconds >= 0 { + delay := time.Duration(seconds) * time.Second + return &delay + } + if deadline, errParse := http.ParseTime(raw); errParse == nil { + delay := deadline.Sub(now) + if delay < 0 { + delay = 0 + } + return &delay + } + } + + code := strings.ToLower(strings.TrimSpace(gjson.GetBytes(body, "error.code").String())) + message := strings.ToLower(strings.TrimSpace(gjson.GetBytes(body, "error.message").String())) + if strings.Contains(code, "tpmratelimitexceeded") || + (strings.Contains(message, "tokens per minute") && strings.Contains(message, "limit") && strings.Contains(message, "exceeded")) { + delay := openAICompatTPMFallbackRetryAfter + return &delay + } + return nil +} diff --git a/internal/runtime/executor/openai_compat_executor_retry_test.go b/internal/runtime/executor/openai_compat_executor_retry_test.go new file mode 100644 index 0000000000..8362c1634e --- /dev/null +++ b/internal/runtime/executor/openai_compat_executor_retry_test.go @@ -0,0 +1,143 @@ +package executor + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/router-for-me/CLIProxyAPI/v7/internal/config" + cliproxyauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth" + cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor" + sdktranslator "github.com/router-for-me/CLIProxyAPI/v7/sdk/translator" +) + +func TestOpenAICompatRetryAfter(t *testing.T) { + now := time.Date(2026, time.September, 3, 12, 0, 0, 0, time.UTC) + tests := []struct { + name string + status int + headers http.Header + body string + want *time.Duration + }{ + { + name: "delta seconds header", + status: http.StatusTooManyRequests, + headers: http.Header{"Retry-After": {"17"}}, + want: durationPointer(17 * time.Second), + }, + { + name: "http date header", + status: http.StatusTooManyRequests, + headers: http.Header{"Retry-After": {now.Add(23 * time.Second).Format(http.TimeFormat)}}, + want: durationPointer(23 * time.Second), + }, + { + name: "explicit TPM code fallback", + status: http.StatusTooManyRequests, + body: `{"error":{"code":"ModelAccountTpmRateLimitExceeded","message":"TPM limit exceeded"}}`, + want: durationPointer(time.Minute), + }, + { + name: "TPM message fallback", + status: http.StatusTooManyRequests, + body: `{"error":{"message":"TPM (Tokens Per Minute) limit of this model is exceeded"}}`, + want: durationPointer(time.Minute), + }, + { + name: "provider header wins over fallback", + status: http.StatusTooManyRequests, + headers: http.Header{"Retry-After": {"5"}}, + body: `{"error":{"code":"ModelAccountTpmRateLimitExceeded"}}`, + want: durationPointer(5 * time.Second), + }, + { + name: "generic 429 has no invented deadline", + status: http.StatusTooManyRequests, + body: `{"error":{"code":"rate_limit"}}`, + }, + { + name: "non-429 ignores header", + status: http.StatusServiceUnavailable, + headers: http.Header{"Retry-After": {"30"}}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got := openAICompatRetryAfter(test.status, test.headers, []byte(test.body), now) + if test.want == nil { + if got != nil { + t.Fatalf("retry-after = %v, want nil", *got) + } + return + } + if got == nil || *got != *test.want { + t.Fatalf("retry-after = %v, want %v", got, *test.want) + } + }) + } +} + +func TestOpenAICompatExecutorPropagatesRetryAfter(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Retry-After", "7") + w.WriteHeader(http.StatusTooManyRequests) + _, _ = w.Write([]byte(`{"error":{"code":"rate_limit","message":"try later"}}`)) + })) + t.Cleanup(server.Close) + + executor := NewOpenAICompatExecutor("openai-compatibility", &config.Config{}) + auth := &cliproxyauth.Auth{Attributes: map[string]string{ + "base_url": server.URL + "/v1", + "api_key": "test", + }} + request := cliproxyexecutor.Request{ + Model: "compatible-model", + Payload: []byte(`{"model":"compatible-model","messages":[{"role":"user","content":"hi"}]}`), + } + tests := []struct { + name string + invoke func() error + }{ + { + name: "nonstream", + invoke: func() error { + _, errExecute := executor.Execute(context.Background(), auth, request, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FromString("openai"), + }) + return errExecute + }, + }, + { + name: "stream bootstrap", + invoke: func() error { + _, errExecute := executor.ExecuteStream(context.Background(), auth, request, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FromString("openai"), + Stream: true, + }) + return errExecute + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + errExecute := test.invoke() + if errExecute == nil { + t.Fatal("expected rate-limit error") + } + retryable, ok := errExecute.(interface{ RetryAfter() *time.Duration }) + if !ok || retryable.RetryAfter() == nil || *retryable.RetryAfter() != 7*time.Second { + t.Fatalf("retry-after = %v, want 7s", retryable) + } + }) + } +} + +func durationPointer(value time.Duration) *time.Duration { + return &value +}