diff --git a/internal/runtime/executor/antigravity_executor_auth.go b/internal/runtime/executor/antigravity_executor_auth.go index 108eb914e7a..87da1632969 100644 --- a/internal/runtime/executor/antigravity_executor_auth.go +++ b/internal/runtime/executor/antigravity_executor_auth.go @@ -183,6 +183,8 @@ func (e *AntigravityExecutor) refreshTokenSingleFlight(ctx context.Context, auth if retryAfter, parseErr := helps.ParseRetryDelay(bodyBytes); parseErr == nil && retryAfter != nil { sErr.retryAfter = retryAfter } + // oauth2.googleapis.com 429 is a token-endpoint throttle, not model quota. + sErr.transientRateLimit = true } return nil, sErr } diff --git a/internal/runtime/executor/antigravity_executor_cooldown_transient_test.go b/internal/runtime/executor/antigravity_executor_cooldown_transient_test.go new file mode 100644 index 00000000000..c5770308158 --- /dev/null +++ b/internal/runtime/executor/antigravity_executor_cooldown_transient_test.go @@ -0,0 +1,90 @@ +package executor + +import ( + "context" + "errors" + "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" +) + +// TestAntigravityShortCooldownErrorIsTransient pins the classification of the +// synthetic 429 the executor raises while an auth sits in a short cooldown. +// The cooldown is a local, self-imposed pause of at most a few minutes, so the +// conductor has to read it as a transient rate limit and rotate to the next +// auth. Unclassified, the same error looks like an exhausted quota carrying a +// retry hint, and the conductor escalates BackoffLevel toward the 30 minute +// ceiling — parking an account that was never actually throttled upstream. +func TestAntigravityShortCooldownErrorIsTransient(t *testing.T) { + resetAntigravityCreditsRetryState() + t.Cleanup(resetAntigravityCreditsRetryState) + client := newFakeAntigravityKVClient() + useFakeAntigravityKVClient(t, client, true, nil) + + exec := NewAntigravityExecutor(&config.Config{}) + opts := cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FormatGemini, + ResponseFormat: sdktranslator.FormatGemini, + } + payload := []byte(`{"contents":[{"role":"user","parts":[{"text":"hi"}]}]}`) + + for _, tc := range []struct { + name string + model string + call func(auth *cliproxyauth.Auth, model string) error + }{ + { + name: "execute", + model: "gemini-3.6-flash", + call: func(auth *cliproxyauth.Auth, model string) error { + _, err := exec.Execute(context.Background(), auth, cliproxyexecutor.Request{Model: model, Payload: payload}, opts) + return err + }, + }, + { + name: "execute-claude", + model: "claude-sonnet-4-5", + call: func(auth *cliproxyauth.Auth, model string) error { + _, err := exec.Execute(context.Background(), auth, cliproxyexecutor.Request{Model: model, Payload: payload}, opts) + return err + }, + }, + { + name: "execute-stream", + model: "gemini-3.6-flash", + call: func(auth *cliproxyauth.Auth, model string) error { + _, err := exec.ExecuteStream(context.Background(), auth, cliproxyexecutor.Request{Model: model, Payload: payload}, opts) + return err + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + auth := &cliproxyauth.Auth{ID: "cooldown-transient-" + tc.name} + if errMark := markAntigravityShortCooldownRequired(context.Background(), auth, tc.model, time.Now(), 30*time.Second); errMark != nil { + t.Fatalf("markAntigravityShortCooldownRequired() error = %v", errMark) + } + + err := tc.call(auth, tc.model) + if err == nil { + t.Fatal("expected the short cooldown to surface a 429") + } + + var classified interface{ TransientRateLimit() bool } + if !errors.As(err, &classified) { + t.Fatalf("short-cooldown error carries no 429 classification: %T", err) + } + if !classified.TransientRateLimit() { + t.Fatal("expected the synthetic short-cooldown 429 to be transient so the conductor rotates instead of escalating backoff") + } + + var hinted interface{ RetryAfter() *time.Duration } + if !errors.As(err, &hinted) || hinted.RetryAfter() == nil || *hinted.RetryAfter() <= 0 { + t.Fatalf("expected a positive retry hint on the short-cooldown 429, got %v", err) + } + }) + } +} diff --git a/internal/runtime/executor/antigravity_executor_credits.go b/internal/runtime/executor/antigravity_executor_credits.go index 9cf15e7fcc0..bd1108adec8 100644 --- a/internal/runtime/executor/antigravity_executor_credits.go +++ b/internal/runtime/executor/antigravity_executor_credits.go @@ -339,6 +339,15 @@ func newAntigravityStatusErr(statusCode int, body []byte) statusErr { if retryAfter, parseErr := helps.ParseRetryDelay(body); parseErr == nil && retryAfter != nil { err.retryAfter = retryAfter } + // Only a decisively rate-limited 429 may keep its raw retry hint downstream; + // exhausted quota and unclassified bodies stay on the escalating cooldown ladder. + // A RATE_LIMIT_EXCEEDED reason without a RetryInfo hint is still a short-lived + // throttle, not an exhausted quota, so it is transient too — but only when the + // classification comes from the ErrorInfo reason, not from the bare + // "too many requests" message heuristic. + category := classifyAntigravity429(body) + err.transientRateLimit = category == antigravity429RateLimited || + (category == antigravity429SoftRateLimit && strings.EqualFold(decideAntigravity429(body).reason, "RATE_LIMIT_EXCEEDED")) } return err } diff --git a/internal/runtime/executor/antigravity_executor_credits_test.go b/internal/runtime/executor/antigravity_executor_credits_test.go index bba63a64b9e..4a7c93d597b 100644 --- a/internal/runtime/executor/antigravity_executor_credits_test.go +++ b/internal/runtime/executor/antigravity_executor_credits_test.go @@ -225,6 +225,79 @@ func TestClassifyAntigravity429(t *testing.T) { }) } +func TestNewAntigravityStatusErrMarksTransientRateLimit(t *testing.T) { + rateLimited := []byte(`{ + "error": { + "code": 429, + "message": "You have exhausted your capacity on this model. Your quota will reset after 0s.", + "status": "RESOURCE_EXHAUSTED", + "details": [ + { + "@type": "type.googleapis.com/google.rpc.ErrorInfo", + "reason": "RATE_LIMIT_EXCEEDED", + "domain": "cloudcode-pa.googleapis.com" + }, + { + "@type": "type.googleapis.com/google.rpc.RetryInfo", + "retryDelay": "0.479417207s" + } + ] + } + }`) + transient := newAntigravityStatusErr(http.StatusTooManyRequests, rateLimited) + if !transient.TransientRateLimit() { + t.Fatal("expected a RATE_LIMIT_EXCEEDED 429 with a sub-second hint to be marked transient") + } + if transient.RetryAfter() == nil { + t.Fatal("expected the provider retry hint to be preserved on a transient rate limit") + } + + exhausted := []byte(`{ + "error": { + "code": 429, + "status": "RESOURCE_EXHAUSTED", + "details": [ + {"@type": "type.googleapis.com/google.rpc.ErrorInfo", "reason": "QUOTA_EXHAUSTED"}, + {"@type": "type.googleapis.com/google.rpc.RetryInfo", "retryDelay": "0.479417207s"} + ] + } + }`) + quotaErr := newAntigravityStatusErr(http.StatusTooManyRequests, exhausted) + if quotaErr.TransientRateLimit() { + t.Fatal("expected a QUOTA_EXHAUSTED 429 not to be marked transient") + } + if quotaErr.RetryAfter() == nil { + t.Fatal("expected the provider retry hint to be preserved on an exhausted quota") + } + + if soft := newAntigravityStatusErr(http.StatusTooManyRequests, []byte(`{"error":{"message":"too many requests"}}`)); soft.TransientRateLimit() { + t.Fatal("expected an unclassified 429 to stay on the escalating cooldown ladder") + } + + // A reasoned RATE_LIMIT_EXCEEDED without a RetryInfo hint is still a + // short-lived throttle, not an exhausted quota. + noHint := []byte(`{ + "error": { + "code": 429, + "status": "RESOURCE_EXHAUSTED", + "details": [ + {"@type": "type.googleapis.com/google.rpc.ErrorInfo", "reason": "RATE_LIMIT_EXCEEDED", "domain": "cloudcode-pa.googleapis.com"} + ] + } + }`) + noHintErr := newAntigravityStatusErr(http.StatusTooManyRequests, noHint) + if !noHintErr.TransientRateLimit() { + t.Fatal("expected a RATE_LIMIT_EXCEEDED 429 without RetryInfo to be marked transient") + } + if noHintErr.RetryAfter() != nil { + t.Fatal("expected no retry hint when Google omits RetryInfo") + } + + if nonRateLimit := newAntigravityStatusErr(http.StatusServiceUnavailable, rateLimited); nonRateLimit.TransientRateLimit() { + t.Fatal("expected a non-429 status not to be marked transient") + } +} + func TestInjectEnabledCreditTypes(t *testing.T) { body := []byte(`{"model":"claude-sonnet-4-6","request":{}}`) got := injectEnabledCreditTypes(body) @@ -734,3 +807,47 @@ func TestParseMetaFloat(t *testing.T) { }) } } + +func TestAntigravityCountTokensClassifiesTransient429(t *testing.T) { + body := `{ + "error": { + "code": 429, + "status": "RESOURCE_EXHAUSTED", + "details": [ + {"@type": "type.googleapis.com/google.rpc.ErrorInfo", "reason": "RATE_LIMIT_EXCEEDED"}, + {"@type": "type.googleapis.com/google.rpc.RetryInfo", "retryDelay": "0.479417207s"} + ] + } + }` + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusTooManyRequests) + _, _ = w.Write([]byte(body)) + })) + defer server.Close() + + exec := NewAntigravityExecutor(&config.Config{RequestRetry: 1}) + _, errCount := exec.CountTokens(context.Background(), testAntigravityAuth(server.URL), cliproxyexecutor.Request{ + Model: "gemini-3.6-flash-high", + Payload: []byte(`{"contents":[{"role":"user","parts":[{"text":"hi"}]}]}`), + }, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FormatGemini, + ResponseFormat: sdktranslator.FormatGemini, + }) + if errCount == nil { + t.Fatal("expected CountTokens to fail on an upstream 429") + } + + var classified interface{ TransientRateLimit() bool } + if !errors.As(errCount, &classified) { + t.Fatalf("CountTokens error carries no 429 classification: %T", errCount) + } + if !classified.TransientRateLimit() { + t.Fatal("expected a RATE_LIMIT_EXCEEDED token-count 429 to be marked transient") + } + + var hinted interface{ RetryAfter() *time.Duration } + if !errors.As(errCount, &hinted) || hinted.RetryAfter() == nil { + t.Fatal("expected the provider retry hint to survive the token-count path") + } +} diff --git a/internal/runtime/executor/antigravity_executor_execute.go b/internal/runtime/executor/antigravity_executor_execute.go index 192a701dfe0..358e9f6013f 100644 --- a/internal/runtime/executor/antigravity_executor_execute.go +++ b/internal/runtime/executor/antigravity_executor_execute.go @@ -33,7 +33,7 @@ func (e *AntigravityExecutor) Execute(ctx context.Context, auth *cliproxyauth.Au } else if inCooldown && !antigravityShouldBypassShortCooldown(ctx, e.cfg) { log.Debugf("antigravity executor: auth %s in short cooldown for model %s (%s remaining), returning 429 to switch auth", auth.ID, baseModel, remaining) d := remaining - return resp, statusErr{code: http.StatusTooManyRequests, msg: fmt.Sprintf("auth in short cooldown, %s remaining", remaining), retryAfter: &d} + return resp, statusErr{code: http.StatusTooManyRequests, msg: fmt.Sprintf("auth in short cooldown, %s remaining", remaining), retryAfter: &d, transientRateLimit: true} } isClaude := strings.Contains(strings.ToLower(baseModel), "claude") @@ -187,7 +187,7 @@ func (e *AntigravityExecutor) executeClaudeNonStream(ctx context.Context, auth * } else if inCooldown && !antigravityShouldBypassShortCooldown(ctx, e.cfg) { log.Debugf("antigravity executor: auth %s in short cooldown for model %s (%s remaining), returning 429 to switch auth", auth.ID, baseModel, remaining) d := remaining - return resp, statusErr{code: http.StatusTooManyRequests, msg: fmt.Sprintf("auth in short cooldown, %s remaining", remaining), retryAfter: &d} + return resp, statusErr{code: http.StatusTooManyRequests, msg: fmt.Sprintf("auth in short cooldown, %s remaining", remaining), retryAfter: &d, transientRateLimit: true} } reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth) diff --git a/internal/runtime/executor/antigravity_executor_stream.go b/internal/runtime/executor/antigravity_executor_stream.go index c8f24175d73..9286bc58127 100644 --- a/internal/runtime/executor/antigravity_executor_stream.go +++ b/internal/runtime/executor/antigravity_executor_stream.go @@ -32,7 +32,7 @@ func (e *AntigravityExecutor) ExecuteStream(ctx context.Context, auth *cliproxya } else if inCooldown && !antigravityShouldBypassShortCooldown(ctx, e.cfg) { log.Debugf("antigravity executor: auth %s in short cooldown for model %s (%s remaining), returning 429 to switch auth", auth.ID, baseModel, remaining) d := remaining - return nil, statusErr{code: http.StatusTooManyRequests, msg: fmt.Sprintf("auth in short cooldown, %s remaining", remaining), retryAfter: &d} + return nil, statusErr{code: http.StatusTooManyRequests, msg: fmt.Sprintf("auth in short cooldown, %s remaining", remaining), retryAfter: &d, transientRateLimit: true} } reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth) diff --git a/internal/runtime/executor/antigravity_executor_tokens.go b/internal/runtime/executor/antigravity_executor_tokens.go index 65412441776..60be91f293d 100644 --- a/internal/runtime/executor/antigravity_executor_tokens.go +++ b/internal/runtime/executor/antigravity_executor_tokens.go @@ -136,11 +136,5 @@ func (e *AntigravityExecutor) CountTokens(ctx context.Context, auth *cliproxyaut return cliproxyexecutor.Response{Payload: translated, Headers: httpResp.Header.Clone()}, nil } - sErr := statusErr{code: httpResp.StatusCode, msg: string(bodyBytes)} - if httpResp.StatusCode == http.StatusTooManyRequests { - if retryAfter, parseErr := helps.ParseRetryDelay(bodyBytes); parseErr == nil && retryAfter != nil { - sErr.retryAfter = retryAfter - } - } - return cliproxyexecutor.Response{}, sErr + return cliproxyexecutor.Response{}, newAntigravityStatusErr(httpResp.StatusCode, bodyBytes) } diff --git a/internal/runtime/executor/antigravity_refresh_test.go b/internal/runtime/executor/antigravity_refresh_test.go index 647b6996032..7b8c804219b 100644 --- a/internal/runtime/executor/antigravity_refresh_test.go +++ b/internal/runtime/executor/antigravity_refresh_test.go @@ -3,6 +3,7 @@ package executor import ( "context" "crypto/tls" + "errors" "io" "net" "net/http" @@ -145,3 +146,181 @@ func TestAntigravityRefresh_DeduplicatesConcurrentRefresh(t *testing.T) { t.Fatalf("expected both refresh callers to share a single upstream token call, got %d", got) } } + +const antigravityOAuth429RetryDelay = "0.479s" + +func antigravityOAuth429Body() string { + return `{ + "error": { + "code": 429, + "message": "Resource has been exhausted (e.g. check quota).", + "details": [{ + "@type": "type.googleapis.com/google.rpc.RetryInfo", + "retryDelay": "` + antigravityOAuth429RetryDelay + `" + }] + } + }` +} + +func startAntigravityOAuthStatusServer(t *testing.T, status int, body string) *httptest.Server { + t.Helper() + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/token" { + t.Errorf("unexpected antigravity test request path: %s", r.URL.Path) + http.Error(w, "unexpected path", http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = io.WriteString(w, body) + })) + t.Cleanup(server.Close) + serverURL, errParse := url.Parse(server.URL) + if errParse != nil { + t.Fatalf("parse test server URL: %v", errParse) + } + useAntigravityRefreshTestTransport(t, serverURL.Host) + return server +} + +func expiredAntigravityRefreshAuth(id string) *cliproxyauth.Auth { + return &cliproxyauth.Auth{ + ID: id, + Provider: "antigravity", + Metadata: map[string]any{ + "refresh_token": "oauth-refresh-token", + "access_token": "expired-access", + "expired": time.Now().Add(-time.Hour).Format(time.RFC3339), + "project_id": "project-refresh", + "type": "antigravity", + }, + } +} + +// TestAntigravityRefresh429IsTransientRateLimit pins oauth2.googleapis.com 429s +// as a token-endpoint throttle. Without TransientRateLimit, the short Retry-After +// looks like an exhausted model quota and the conductor climbs BackoffLevel +// toward the 30-minute ceiling. +func TestAntigravityRefresh429IsTransientRateLimit(t *testing.T) { + resetAntigravityRefreshGroupForTest() + t.Cleanup(resetAntigravityRefreshGroupForTest) + resetAntigravityCreditsRetryState() + t.Cleanup(resetAntigravityCreditsRetryState) + cliproxyauth.SetQuotaCooldownDisabled(false) + t.Cleanup(func() { cliproxyauth.SetQuotaCooldownDisabled(false) }) + + startAntigravityOAuthStatusServer(t, http.StatusTooManyRequests, antigravityOAuth429Body()) + + executor := &AntigravityExecutor{} + auth := expiredAntigravityRefreshAuth("auth-antigravity-refresh-429") + _, errRefresh := executor.Refresh(context.Background(), auth) + if errRefresh == nil { + t.Fatal("expected token refresh 429") + } + + var classified interface{ TransientRateLimit() bool } + if !errors.As(errRefresh, &classified) { + t.Fatalf("refresh 429 carries no classification: %T", errRefresh) + } + if !classified.TransientRateLimit() { + t.Fatal("expected oauth token-refresh 429 to be transient so the conductor skips the quota ladder") + } + + var hinted interface{ RetryAfter() *time.Duration } + if !errors.As(errRefresh, &hinted) || hinted.RetryAfter() == nil { + t.Fatalf("expected Retry-After on the refresh 429, got %v", errRefresh) + } + wantHint, errHint := time.ParseDuration(antigravityOAuth429RetryDelay) + if errHint != nil { + t.Fatalf("parse fixture retry delay: %v", errHint) + } + if got := *hinted.RetryAfter(); got != wantHint { + t.Fatalf("retryAfter = %v, want provider hint %v", got, wantHint) + } + + var status interface{ StatusCode() int } + if !errors.As(errRefresh, &status) || status.StatusCode() != http.StatusTooManyRequests { + t.Fatalf("status = %#v, want 429", errRefresh) + } + + manager := cliproxyauth.NewManager(nil, nil, nil) + registered := &cliproxyauth.Auth{ + ID: auth.ID, + Provider: "antigravity", + Metadata: map[string]any{"type": "antigravity"}, + ModelStates: map[string]*cliproxyauth.ModelState{ + "gemini-3.6-flash": { + Status: cliproxyauth.StatusActive, + Quota: cliproxyauth.QuotaState{BackoffLevel: 0}, + }, + }, + } + if _, errRegister := manager.Register(cliproxyauth.WithSkipPersist(context.Background()), registered); errRegister != nil { + t.Fatalf("Register returned error: %v", errRegister) + } + + hint := *hinted.RetryAfter() + result := cliproxyauth.Result{ + AuthID: registered.ID, + Provider: "antigravity", + Model: "gemini-3.6-flash", + Success: false, + RetryAfter: &hint, + TransientRateLimit: classified.TransientRateLimit(), + Error: &cliproxyauth.Error{ + Code: "rate_limit", + Message: errRefresh.Error(), + Retryable: true, + HTTPStatus: http.StatusTooManyRequests, + }, + } + + now := time.Now() + manager.MarkResult(context.Background(), result) + + updated, ok := manager.GetByID(registered.ID) + if !ok || updated == nil || updated.ModelStates["gemini-3.6-flash"] == nil { + t.Fatalf("expected model state after refresh 429") + } + state := updated.ModelStates["gemini-3.6-flash"] + if state.Quota.BackoffLevel != 0 { + t.Fatalf("expected BackoffLevel 0 on the transient refresh-429 path, got %d", state.Quota.BackoffLevel) + } + if state.Quota.Exceeded { + t.Fatal("transient refresh 429 must not set Quota.Exceeded") + } + // Quota first step is 1s; transient floor is 10s. A 479ms hint on the quota + // ladder would recover in ~1s and increment BackoffLevel. + if got := state.NextRetryAfter.Sub(now); got < 9*time.Second { + t.Fatalf("refresh 429 took the quota ladder: NextRetryAfter delta %v, want at least the 10s transient floor", got) + } + + manager.MarkResult(context.Background(), result) + updated, ok = manager.GetByID(registered.ID) + if !ok || updated == nil || updated.ModelStates["gemini-3.6-flash"] == nil { + t.Fatalf("expected model state after repeated refresh 429") + } + if got := updated.ModelStates["gemini-3.6-flash"].Quota.BackoffLevel; got != 0 { + t.Fatalf("repeated refresh 429 advanced quota BackoffLevel to %d", got) + } +} + +func TestAntigravityRefresh401IsNotTransientRateLimit(t *testing.T) { + resetAntigravityRefreshGroupForTest() + t.Cleanup(resetAntigravityRefreshGroupForTest) + + startAntigravityOAuthStatusServer(t, http.StatusUnauthorized, `{"error":"invalid_grant"}`) + + _, errRefresh := (&AntigravityExecutor{}).Refresh(context.Background(), expiredAntigravityRefreshAuth("auth-antigravity-refresh-401")) + if errRefresh == nil { + t.Fatal("expected token refresh 401") + } + var status interface{ StatusCode() int } + if !errors.As(errRefresh, &status) || status.StatusCode() != http.StatusUnauthorized { + t.Fatalf("status = %#v, want 401", errRefresh) + } + var classified interface{ TransientRateLimit() bool } + if errors.As(errRefresh, &classified) && classified.TransientRateLimit() { + t.Fatal("refresh 401 classified as transient rate limit") + } +} diff --git a/internal/runtime/executor/claude_executor_beta_policy_test.go b/internal/runtime/executor/claude_executor_beta_policy_test.go index 18686f36c27..0a0982e4c16 100644 --- a/internal/runtime/executor/claude_executor_beta_policy_test.go +++ b/internal/runtime/executor/claude_executor_beta_policy_test.go @@ -293,3 +293,31 @@ func TestClassifyClaudeUpstreamError_OtherStatusesUnaffected(t *testing.T) { t.Fatal("non-429 status was misclassified as request-scoped") } } + +// An ordinary model-level Claude 429 (no unified 5h/7d rejection headers) is +// a transient throttle: the conductor must rotate to the next credential +// instead of escalating BackoffLevel as if quota were exhausted. +func TestClassifyClaudeUpstreamError_OrdinaryRateLimitIsTransient(t *testing.T) { + body := []byte(`{"type":"error","error":{"type":"rate_limit_error","message":"Number of requests has exceeded your rate limit."}}`) + err := classifyClaudeUpstreamError(http.StatusTooManyRequests, nil, body) + + var transient interface{ TransientRateLimit() bool } + if !errors.As(err, &transient) || !transient.TransientRateLimit() { + t.Fatalf("ordinary Claude 429 = %v, want a transient rate limit", err) + } +} + +// The unified 5h/7d rejection stays on the quota ladder: it must NOT be +// reported as a transient rate limit. +func TestClassifyClaudeUpstreamError_UnifiedRejectionNotTransient(t *testing.T) { + headers := http.Header{ + "Anthropic-Ratelimit-Unified-5h-Status": []string{"rejected"}, + "Anthropic-Ratelimit-Unified-7d-Status": []string{"allowed"}, + } + err := classifyClaudeUpstreamError(http.StatusTooManyRequests, headers, []byte(`{"type":"error","error":{"type":"rate_limit_error","message":"Shared usage window rejected."}}`)) + + var transient interface{ TransientRateLimit() bool } + if errors.As(err, &transient) && transient.TransientRateLimit() { + t.Fatal("unified 5h/7d rejection must stay on the quota ladder, not be marked transient") + } +} diff --git a/internal/runtime/executor/claude_executor_request.go b/internal/runtime/executor/claude_executor_request.go index 1bc2d6daab0..8235f95046c 100644 --- a/internal/runtime/executor/claude_executor_request.go +++ b/internal/runtime/executor/claude_executor_request.go @@ -304,7 +304,10 @@ func classifyClaudeUpstreamError(statusCode int, headers http.Header, body []byt if claudeBodyIndicatesFastModeCredits(body) { return claudeEntitlementError{err} } - // Ordinary model-level Claude 429 (not a unified 5h/7d rejection) + // Ordinary model-level Claude 429 (not a unified 5h/7d rejection): a + // transient throttle, so the conductor rotates instead of escalating + // BackoffLevel as if quota were exhausted. + err.transientRateLimit = true return claudeRateLimitError{statusErr: err, credentialScoped: false} } return err diff --git a/internal/runtime/executor/codex_websockets_errors.go b/internal/runtime/executor/codex_websockets_errors.go index eae0706a3a9..972a7427dd7 100644 --- a/internal/runtime/executor/codex_websockets_errors.go +++ b/internal/runtime/executor/codex_websockets_errors.go @@ -49,6 +49,7 @@ func parseCodexWebsocketError(payload []byte) (error, bool) { } else if isCodexWebsocketConnectionLimitError(payload) { retryAfter := time.Duration(0) statusError.retryAfter = &retryAfter + statusError.transientRateLimit = true } return statusErrWithHeaders{ statusErr: statusError, diff --git a/internal/runtime/executor/codex_websockets_executor_test.go b/internal/runtime/executor/codex_websockets_executor_test.go index 755bf1fbf9d..a95102110fd 100644 --- a/internal/runtime/executor/codex_websockets_executor_test.go +++ b/internal/runtime/executor/codex_websockets_executor_test.go @@ -1529,6 +1529,10 @@ func TestParseCodexWebsocketErrorMarksConnectionLimitRetryable(t *testing.T) { if got := *retryable.RetryAfter(); got != 0 { t.Fatalf("retryAfter = %v, want connection-limit fallback 0", got) } + transient, ok := err.(interface{ TransientRateLimit() bool }) + if !ok || !transient.TransientRateLimit() { + t.Fatalf("transientRateLimit = %#v, want true for websocket_connection_limit_reached", err) + } withHeaders, ok := err.(interface{ Headers() http.Header }) if !ok || withHeaders.Headers().Get("retry-after") != "1" { t.Fatalf("headers = %#v, want retry-after", err) @@ -1548,6 +1552,9 @@ func TestParseCodexWebsocketErrorUsesUsageLimitRetryMetadata(t *testing.T) { if got := *retryable.RetryAfter(); got != 7*time.Second { t.Fatalf("retryAfter = %v, want 7s", got) } + if transient, ok := err.(interface{ TransientRateLimit() bool }); ok && transient.TransientRateLimit() { + t.Fatal("usage_limit_reached classified as transient rate limit") + } } func TestParseCodexWebsocketErrorPreservesWrappedBodyAndHeaders(t *testing.T) { @@ -1570,12 +1577,90 @@ func TestParseCodexWebsocketErrorPreservesWrappedBodyAndHeaders(t *testing.T) { if !ok || retryable.RetryAfter() == nil { t.Fatalf("expected body.error.code websocket connection limit to be retryable") } + transient, ok := err.(interface{ TransientRateLimit() bool }) + if !ok || !transient.TransientRateLimit() { + t.Fatalf("transientRateLimit = %#v, want true for body.error.code websocket_connection_limit_reached", err) + } withHeaders, ok := err.(interface{ Headers() http.Header }) if !ok || withHeaders.Headers().Get("x-request-id") != "req-1" { t.Fatalf("headers = %#v, want x-request-id", err) } } +func TestParseCodexWebsocketErrorConnectionLimitDrivesTransientZeroDelay(t *testing.T) { + cliproxyauth.SetQuotaCooldownDisabled(false) + t.Cleanup(func() { cliproxyauth.SetQuotaCooldownDisabled(false) }) + + parsed, ok := parseCodexWebsocketError([]byte(`{"type":"error","status":429,"error":{"code":"websocket_connection_limit_reached","message":"too many websockets"}}`)) + if !ok { + t.Fatal("expected websocket error") + } + transient, ok := parsed.(interface{ TransientRateLimit() bool }) + if !ok || !transient.TransientRateLimit() { + t.Fatalf("transientRateLimit = %#v, want true so conductor takes the zero-delay transient path", parsed) + } + retryable, ok := parsed.(interface{ RetryAfter() *time.Duration }) + if !ok || retryable.RetryAfter() == nil || *retryable.RetryAfter() != 0 { + t.Fatalf("retryAfter = %#v, want 0", parsed) + } + + manager := cliproxyauth.NewManager(nil, nil, nil) + auth := &cliproxyauth.Auth{ + ID: "auth-codex-ws-conn-limit", + Provider: "codex", + Metadata: map[string]any{"type": "codex"}, + ModelStates: map[string]*cliproxyauth.ModelState{ + "gpt-5": { + Status: cliproxyauth.StatusActive, + Quota: cliproxyauth.QuotaState{BackoffLevel: 0}, + }, + }, + } + if _, errRegister := manager.Register(cliproxyauth.WithSkipPersist(context.Background()), auth); errRegister != nil { + t.Fatalf("Register returned error: %v", errRegister) + } + + hint := *retryable.RetryAfter() + result := cliproxyauth.Result{ + AuthID: auth.ID, + Provider: "codex", + Model: "gpt-5", + Success: false, + RetryAfter: &hint, + TransientRateLimit: transient.TransientRateLimit(), + Error: &cliproxyauth.Error{ + Code: "rate_limit", + Message: parsed.Error(), + Retryable: true, + HTTPStatus: http.StatusTooManyRequests, + }, + } + + now := time.Now() + manager.MarkResult(context.Background(), result) + + updated, ok := manager.GetByID(auth.ID) + if !ok || updated == nil || updated.ModelStates["gpt-5"] == nil { + t.Fatalf("expected model state after connection-limit failure") + } + state := updated.ModelStates["gpt-5"] + if state.Quota.BackoffLevel != 0 { + t.Fatalf("expected BackoffLevel 0 on the zero-delay transient path, got %d", state.Quota.BackoffLevel) + } + if state.Quota.NextRecoverAt.After(now.Add(500 * time.Millisecond)) { + t.Fatalf("connection-limit took the quota ladder: NextRecoverAt=%v, want <= %v", state.Quota.NextRecoverAt, now) + } + + manager.MarkResult(context.Background(), result) + updated, ok = manager.GetByID(auth.ID) + if !ok || updated == nil || updated.ModelStates["gpt-5"] == nil { + t.Fatalf("expected model state after repeated connection-limit failure") + } + if got := updated.ModelStates["gpt-5"].Quota.BackoffLevel; got != 0 { + t.Fatalf("repeated connection-limit advanced quota BackoffLevel to %d", got) + } +} + func TestApplyCodexHeadersUsesConfigUserAgentForOAuth(t *testing.T) { req, err := http.NewRequest(http.MethodPost, "https://example.com/responses", nil) if err != nil { diff --git a/internal/runtime/executor/openai_compat_executor.go b/internal/runtime/executor/openai_compat_executor.go index ee679d6d8fd..f27a2429b69 100644 --- a/internal/runtime/executor/openai_compat_executor.go +++ b/internal/runtime/executor/openai_compat_executor.go @@ -1011,9 +1011,10 @@ func openAICompatStreamDataError(payload []byte, eventName string) (statusErr, b } type statusErr struct { - code int - msg string - retryAfter *time.Duration + code int + msg string + retryAfter *time.Duration + transientRateLimit bool } func (e statusErr) Error() string { @@ -1024,3 +1025,7 @@ func (e statusErr) Error() string { } func (e statusErr) StatusCode() int { return e.code } func (e statusErr) RetryAfter() *time.Duration { return e.retryAfter } + +// TransientRateLimit reports whether the upstream 429 was classified as a +// short-lived rate limit rather than an exhausted quota window. +func (e statusErr) TransientRateLimit() bool { return e.transientRateLimit } diff --git a/sdk/cliproxy/auth/conductor.go b/sdk/cliproxy/auth/conductor.go index 87e5307859d..faabe3cbce3 100644 --- a/sdk/cliproxy/auth/conductor.go +++ b/sdk/cliproxy/auth/conductor.go @@ -54,6 +54,10 @@ type Result struct { Success bool // RetryAfter carries a provider supplied retry hint (e.g. 429 retryDelay). RetryAfter *time.Duration + // TransientRateLimit marks a 429 the provider classified as a short-lived rate + // limit rather than an exhausted quota window. Such failures use transient + // cooldown handling without advancing the escalating quota cooldown ladder. + TransientRateLimit bool // CredentialScope indicates that the failure affects the whole credential across models (e.g. Anthropic 5h/7d unified limits). CredentialScope bool // Error describes the failure when Success is false. diff --git a/sdk/cliproxy/auth/conductor_cooldown.go b/sdk/cliproxy/auth/conductor_cooldown.go index 0263e08d0f4..2cda382fb86 100644 --- a/sdk/cliproxy/auth/conductor_cooldown.go +++ b/sdk/cliproxy/auth/conductor_cooldown.go @@ -25,6 +25,8 @@ var quotaCooldownDisabled atomic.Bool var transientErrorCooldownSeconds atomic.Int64 +const transientRateLimitMinimum = 10 * time.Second + // SetQuotaCooldownDisabled toggles auth/model cooldown scheduling globally. func SetQuotaCooldownDisabled(disable bool) { quotaCooldownDisabled.Store(disable) @@ -762,6 +764,8 @@ func (m *Manager) MarkResult(ctx context.Context, result Result) { } state := ensureModelState(auth, modelKey) modelState = state + prevUnavailable := state.Unavailable + prevNextRetry := state.NextRetryAfter state.Unavailable = true state.Status = StatusError state.UpdatedAt = now @@ -831,17 +835,56 @@ func (m *Manager) MarkResult(ctx context.Context, result Result) { case 429: var next time.Time backoffLevel := state.Quota.BackoffLevel + transientCooldownOff := false + transientCooldown := false if !disableCooling { - if result.RetryAfter != nil { + switch { + case result.TransientRateLimit && result.RetryAfter != nil && *result.RetryAfter <= 0: + // A deliberate zero-delay hint requests rotation without waiting. next = now.Add(*result.RetryAfter) - } else { + transientCooldown = true + case result.TransientRateLimit: + // Keep short-lived throttles out of the exhausted-quota window, but + // never let a tiny hint repeatedly select the same credential. + if result.RetryAfter == nil && transientErrorCooldownSeconds.Load() < 0 { + next = time.Time{} + transientCooldownOff = true + } else { + next, backoffLevel = transientRateLimitCooldownAfterFailure(state.Quota, result.RetryAfter, now) + } + transientCooldown = true + default: next, backoffLevel = quotaCooldownAfterFailure(state.Quota, now) + if result.RetryAfter != nil { + // An exhausted-quota hint can be sub-second even when the quota is gone + // for the whole day, so never let it undercut the escalating ladder. + if hinted := now.Add(*result.RetryAfter); hinted.After(next) { + next = hinted + } + } } if state.Quota.Exceeded && state.Quota.NextRecoverAt.After(next) { next = state.Quota.NextRecoverAt } } - state.NextRetryAfter = next + if transientCooldownOff && !(state.Quota.Exceeded && state.Quota.NextRecoverAt.After(now)) { + // Transient cooldowns are disabled for this auth: keep the model + // available instead of recording a zero-time quota block. Restore + // any pre-existing non-quota cooldown (401/403/404/5xx) instead of + // clearing it. An active (unexpired) quota window still vetoes skip; + // an Exceeded flag whose NextRecoverAt is already past does not. + state.Unavailable = prevUnavailable + state.NextRetryAfter = prevNextRetry + break + } + nextRetry := next + if !transientCooldown && state.NextRetryAfter.After(nextRetry) { + nextRetry = state.NextRetryAfter + } + state.NextRetryAfter = nextRetry + if transientCooldown { + break + } applyCooldownFields(&state.Quota, QuotaState{ Exceeded: true, Reason: "quota", @@ -907,7 +950,7 @@ func (m *Manager) MarkResult(ctx context.Context, result Result) { if result.Error != nil && result.Error.Code == ErrorCodeForceCooldown { disableCooling = false } - applyAuthFailureState(auth, result.Error, result.RetryAfter, now, disableCooling) + applyAuthFailureState(auth, result.Error, result.RetryAfter, now, disableCooling, result.TransientRateLimit) } } @@ -1467,6 +1510,19 @@ func retryAfterFromError(err error) *time.Duration { return &value } +// isTransientRateLimitError reports whether the executor classified the failure +// as a short-lived provider rate limit rather than an exhausted quota window. +func isTransientRateLimitError(err error) bool { + if err == nil { + return false + } + type transientRateLimitProvider interface { + TransientRateLimit() bool + } + var trp transientRateLimitProvider + return errors.As(err, &trp) && trp != nil && trp.TransientRateLimit() +} + func isCredentialScopedError(err error) bool { if err == nil { return false @@ -1891,13 +1947,15 @@ func isRequestInvalidError(err error) bool { return false } -func applyAuthFailureState(auth *Auth, resultErr *Error, retryAfter *time.Duration, now time.Time, disableCooling bool) { +func applyAuthFailureState(auth *Auth, resultErr *Error, retryAfter *time.Duration, now time.Time, disableCooling bool, transientRateLimit bool) { if auth == nil { return } if shouldSkipCredentialCooldown(resultErr) { return } + prevUnavailable := auth.Unavailable + prevNextRetry := auth.NextRetryAfter defer func() { if disableCooling && auth.NextRetryAfter.IsZero() && auth.Quota.NextRecoverAt.IsZero() { auth.Unavailable = false @@ -1958,22 +2016,61 @@ func applyAuthFailureState(auth *Auth, resultErr *Error, retryAfter *time.Durati auth.NextRetryAfter = now.Add(12 * time.Hour) } case 429: - auth.StatusMessage = "quota exhausted" - auth.Quota.Exceeded = true - auth.Quota.Reason = "quota" + prevStatusMessage := auth.StatusMessage + prevExceeded, prevReason := auth.Quota.Exceeded, auth.Quota.Reason var next time.Time + transientCooldownOff := false if !disableCooling { - if retryAfter != nil { + switch { + case transientRateLimit && retryAfter != nil && *retryAfter <= 0: + // A deliberate zero-delay hint requests rotation without waiting. next = now.Add(*retryAfter) - } else { + case transientRateLimit: + // Keep short-lived throttles out of the exhausted-quota window, but + // never let a tiny hint repeatedly select the same credential. + if retryAfter == nil && transientErrorCooldownSeconds.Load() < 0 { + next = time.Time{} + transientCooldownOff = true + } else { + next, _ = transientRateLimitCooldownAfterFailure(auth.Quota, retryAfter, now) + } + default: next, auth.Quota.BackoffLevel = quotaCooldownAfterFailure(auth.Quota, now) + if retryAfter != nil { + // An exhausted-quota hint can be sub-second even when the quota is gone for + // the whole day, so never let it undercut the escalating quota ladder. + if hinted := now.Add(*retryAfter); hinted.After(next) { + next = hinted + } + } } if auth.Quota.Exceeded && auth.Quota.NextRecoverAt.After(next) { next = auth.Quota.NextRecoverAt } } + if transientCooldownOff && !(prevExceeded && auth.Quota.NextRecoverAt.After(now)) { + // Transient cooldowns are disabled: keep the credential available + // instead of recording a zero-time quota block. An active (unexpired) + // quota window still vetoes skip; an expired Exceeded record does not. + auth.StatusMessage = prevStatusMessage + auth.Quota.Exceeded = prevExceeded + auth.Quota.Reason = prevReason + auth.Unavailable = prevUnavailable + auth.NextRetryAfter = prevNextRetry + break + } + nextRetry := next + if !transientRateLimit && auth.NextRetryAfter.After(nextRetry) { + nextRetry = auth.NextRetryAfter + } + auth.NextRetryAfter = nextRetry + if transientRateLimit { + break + } + auth.StatusMessage = "quota exhausted" + auth.Quota.Exceeded = true + auth.Quota.Reason = "quota" auth.Quota.NextRecoverAt = next - auth.NextRetryAfter = next case 408, 500, 502, 503, 504: auth.StatusMessage = "transient upstream error" auth.NextRetryAfter = recoverableFailureRetryAfter(now, disableCooling) @@ -1991,6 +2088,17 @@ func applyAuthFailureState(auth *Auth, resultErr *Error, retryAfter *time.Durati } } +func transientRateLimitCooldownAfterFailure(quota QuotaState, retryAfter *time.Duration, now time.Time) (time.Time, int) { + next := now.Add(transientRateLimitMinimum) + if retryAfter != nil { + next = now.Add(*retryAfter) + } + if minimum := now.Add(transientRateLimitMinimum); minimum.After(next) { + next = minimum + } + return next, quota.BackoffLevel +} + // quotaCooldownAfterFailure returns the recovery deadline and backoff level for // a quota failure observed at now. Failures that land while a previous quota // window is still open reuse that window instead of escalating, so a burst of diff --git a/sdk/cliproxy/auth/conductor_execution.go b/sdk/cliproxy/auth/conductor_execution.go index c7dcebe2459..e5f8eb3833c 100644 --- a/sdk/cliproxy/auth/conductor_execution.go +++ b/sdk/cliproxy/auth/conductor_execution.go @@ -367,6 +367,8 @@ func (m *Manager) executeMixedOnce(ctx context.Context, providers []string, req return cliproxyexecutor.Response{}, errCancel } result := Result{AuthID: auth.ID, Provider: provider, Model: routeModel, Success: false, Error: resultErrorFromError(errPrepare), Options: pickOpts} + result.RetryAfter = retryAfterFromError(errPrepare) + result.TransientRateLimit = isTransientRateLimitError(errPrepare) m.MarkResult(execCtx, result) lastErr = errPrepare continue @@ -424,6 +426,9 @@ func (m *Manager) executeMixedOnce(ctx context.Context, providers []string, req if ra := retryAfterFromError(errExec); ra != nil { result.RetryAfter = ra } + if isTransientRateLimitError(errExec) { + result.TransientRateLimit = true + } if isCredentialScopedError(errExec) { result.CredentialScope = true } @@ -545,6 +550,8 @@ func (m *Manager) executeCountMixedOnce(ctx context.Context, providers []string, return cliproxyexecutor.Response{}, errCancel } result := Result{AuthID: auth.ID, Provider: provider, Model: routeModel, Success: false, Error: resultErrorFromError(errPrepare), Options: pickOpts, SkipQuotaObservation: true} + result.RetryAfter = retryAfterFromError(errPrepare) + result.TransientRateLimit = isTransientRateLimitError(errPrepare) m.MarkResult(execCtx, result) lastErr = errPrepare continue @@ -602,6 +609,9 @@ func (m *Manager) executeCountMixedOnce(ctx context.Context, providers []string, if ra := retryAfterFromError(errExec); ra != nil { result.RetryAfter = ra } + if isTransientRateLimitError(errExec) { + result.TransientRateLimit = true + } action, okAction := matchRequestScopedErrorAction(auth, errExec, m.runtimeConfigSnapshot()) applyRequestScopedActionToResult(action, okAction, &result) // Some Anthropic-compatible upstreams do not implement the @@ -863,6 +873,8 @@ func (m *Manager) executeStreamMixedOnce(ctx context.Context, providers []string } } result := Result{AuthID: auth.ID, Provider: provider, Model: routeModel, Success: false, Error: resultErrorFromError(errPrepare), Options: pickOpts} + result.RetryAfter = retryAfterFromError(errPrepare) + result.TransientRateLimit = isTransientRateLimitError(errPrepare) if selection != nil { m.reportHomeResult(execCtx, result, auth) releaseAttempt() diff --git a/sdk/cliproxy/auth/conductor_home.go b/sdk/cliproxy/auth/conductor_home.go index ce956a1706a..ba7d0d3733f 100644 --- a/sdk/cliproxy/auth/conductor_home.go +++ b/sdk/cliproxy/auth/conductor_home.go @@ -1353,6 +1353,9 @@ func (m *Manager) tryAntigravityCreditsExecute(ctx context.Context, req cliproxy if ra := retryAfterFromError(errExec); ra != nil { result.RetryAfter = ra } + if isTransientRateLimitError(errExec) { + result.TransientRateLimit = true + } if isCredentialScopedError(errExec) { result.CredentialScope = true } diff --git a/sdk/cliproxy/auth/conductor_home_execution.go b/sdk/cliproxy/auth/conductor_home_execution.go index 9be8f52fe08..9428d09c3db 100644 --- a/sdk/cliproxy/auth/conductor_home_execution.go +++ b/sdk/cliproxy/auth/conductor_home_execution.go @@ -146,7 +146,10 @@ func (m *Manager) executeHomeOnce(ctx context.Context, providers []string, req c } preparedAuth, errPrepare := m.prepareHomeRequestAuth(execCtx, selection.Executor, selection) if errPrepare != nil { - m.reportHomeResult(execCtx, Result{AuthID: auth.ID, Provider: selection.Provider, Model: routeModel, Success: false, Error: resultErrorFromError(errPrepare), Options: opts}, auth) + result := Result{AuthID: auth.ID, Provider: selection.Provider, Model: routeModel, Success: false, Error: resultErrorFromError(errPrepare), Options: opts} + result.RetryAfter = retryAfterFromError(errPrepare) + result.TransientRateLimit = isTransientRateLimitError(errPrepare) + m.reportHomeResult(execCtx, result, auth) releaseAttempt() if errEnd := m.endHomeSelectionBeforeRedispatch(ctx, selection, "prepare_failed"); errEnd != nil { return cliproxyexecutor.Response{}, errEnd @@ -266,6 +269,7 @@ func (m *Manager) executeHomeOnce(ctx context.Context, providers []string, req c } result.Error = resultErrorFromError(errExecute) result.RetryAfter = retryAfterFromError(errExecute) + result.TransientRateLimit = isTransientRateLimitError(errExecute) if isCredentialScopedError(errExecute) { result.CredentialScope = true } diff --git a/sdk/cliproxy/auth/conductor_overrides_test.go b/sdk/cliproxy/auth/conductor_overrides_test.go index e8e635afb56..f6d9b0be8e1 100644 --- a/sdk/cliproxy/auth/conductor_overrides_test.go +++ b/sdk/cliproxy/auth/conductor_overrides_test.go @@ -1048,6 +1048,401 @@ func TestManager_MarkResult_TransientErrorCooldownDefault(t *testing.T) { } } +// A transient 429 without any parseable retry hint must use transient cooldown +// handling on both the per-model and the credential level, without advancing +// the quota ladder. +func TestManager_MarkResult_Transient429WithoutHintUsesTransientFloor(t *testing.T) { + prevQuota := quotaCooldownDisabled.Load() + quotaCooldownDisabled.Store(false) + prevTransient := transientErrorCooldownSeconds.Load() + SetTransientErrorCooldownSeconds(0) + t.Cleanup(func() { + quotaCooldownDisabled.Store(prevQuota) + transientErrorCooldownSeconds.Store(prevTransient) + }) + + m := NewManager(nil, nil, nil) + auth := &Auth{ID: "auth-transient-429-nohint", Provider: "claude"} + if _, errRegister := m.Register(context.Background(), auth); errRegister != nil { + t.Fatalf("register auth: %v", errRegister) + } + + model := "test-model-transient-429-nohint" + m.MarkResult(context.Background(), Result{ + AuthID: auth.ID, + Provider: auth.Provider, + Model: model, + Success: false, + Error: &Error{ + HTTPStatus: http.StatusTooManyRequests, + Message: "rate limited", + }, + TransientRateLimit: true, + }) + + updated, ok := m.GetByID(auth.ID) + if !ok || updated == nil { + t.Fatalf("auth %s missing after MarkResult", auth.ID) + } + + if updated.Quota.BackoffLevel != 0 { + t.Fatalf("expected credential quota ladder level to stay 0 for a transient 429 without hint, got %d", updated.Quota.BackoffLevel) + } + diff := time.Until(updated.NextRetryAfter) + if diff < transientRateLimitMinimum-time.Second || diff > transientRateLimitMinimum+time.Second { + t.Fatalf("expected credential NextRetryAfter ~%v transient floor, got %v", transientRateLimitMinimum, diff) + } + + state := updated.ModelStates[model] + if state == nil || state.NextRetryAfter.IsZero() { + t.Fatalf("expected per-model cooldown state for %s, got %+v", model, state) + } + if state.Quota.BackoffLevel != 0 { + t.Fatalf("expected per-model quota ladder level to stay 0, got %d", state.Quota.BackoffLevel) + } + modelDiff := time.Until(state.NextRetryAfter) + if modelDiff < transientRateLimitMinimum-time.Second || modelDiff > transientRateLimitMinimum+time.Second { + t.Fatalf("expected per-model NextRetryAfter ~%v transient floor, got %v", transientRateLimitMinimum, modelDiff) + } +} + +// With transient cooldowns disabled (transientErrorCooldownSeconds < 0) the +// transient-429 fallback yields a zero retry time. That zero must not be +// stored as state: an unavailable/quota-exceeded flag with an empty +// NextRetryAfter would read as an indefinite block and hide the credential +// forever. +func TestManager_MarkResult_Transient429WithoutHintRespectsDisabledCooldown(t *testing.T) { + prevQuota := quotaCooldownDisabled.Load() + quotaCooldownDisabled.Store(false) + prevTransient := transientErrorCooldownSeconds.Load() + SetTransientErrorCooldownSeconds(-1) + t.Cleanup(func() { + quotaCooldownDisabled.Store(prevQuota) + transientErrorCooldownSeconds.Store(prevTransient) + }) + + m := NewManager(nil, nil, nil) + auth := &Auth{ID: "auth-transient-429-disabled", Provider: "claude"} + if _, errRegister := m.Register(context.Background(), auth); errRegister != nil { + t.Fatalf("register auth: %v", errRegister) + } + + model := "test-model-transient-429-disabled" + m.MarkResult(context.Background(), Result{ + AuthID: auth.ID, + Provider: auth.Provider, + Model: model, + Success: false, + Error: &Error{ + HTTPStatus: http.StatusTooManyRequests, + Message: "rate limited", + }, + TransientRateLimit: true, + }) + + updated, ok := m.GetByID(auth.ID) + if !ok || updated == nil { + t.Fatalf("auth %s missing after MarkResult", auth.ID) + } + + if !updated.NextRetryAfter.IsZero() { + t.Fatalf("expected credential NextRetryAfter to stay zero with transient cooldowns disabled, got %v", updated.NextRetryAfter) + } + if updated.Quota.Exceeded { + t.Fatal("expected the credential quota state to stay clear with transient cooldowns disabled") + } + + state := updated.ModelStates[model] + if state == nil { + t.Fatalf("expected per-model state for %s", model) + } + if state.Unavailable { + t.Fatal("expected the model to stay available with transient cooldowns disabled") + } + if !state.NextRetryAfter.IsZero() { + t.Fatalf("expected per-model NextRetryAfter to stay zero, got %v", state.NextRetryAfter) + } + if state.Quota.Exceeded { + t.Fatal("expected the per-model quota state to stay clear with transient cooldowns disabled") + } +} + +// An expired quota record keeps Exceeded=true after NextRecoverAt has passed. +// That stale flag must not veto the skip path: a hintless transient 429 with +// transient cooldowns disabled must leave the model available and must not +// record a new quota failure. +func TestManager_MarkResult_Transient429DisabledTreatsExpiredQuotaAsInactive(t *testing.T) { + prevQuota := quotaCooldownDisabled.Load() + quotaCooldownDisabled.Store(false) + prevTransient := transientErrorCooldownSeconds.Load() + SetTransientErrorCooldownSeconds(-1) + t.Cleanup(func() { + quotaCooldownDisabled.Store(prevQuota) + transientErrorCooldownSeconds.Store(prevTransient) + }) + + m := NewManager(nil, nil, nil) + model := "test-model-expired-quota-skip" + expired := time.Now().Add(-time.Minute) + auth := &Auth{ + ID: "auth-transient-429-expired-quota", + Provider: "claude", + ModelStates: map[string]*ModelState{ + model: { + Status: StatusActive, + Quota: QuotaState{Exceeded: true, Reason: "quota", NextRecoverAt: expired, BackoffLevel: 3}, + }, + }, + } + if _, errRegister := m.Register(context.Background(), auth); errRegister != nil { + t.Fatalf("register auth: %v", errRegister) + } + + reg := registry.GetGlobalRegistry() + reg.RegisterClient(auth.ID, "claude", []*registry.ModelInfo{{ID: model}}) + t.Cleanup(func() { reg.UnregisterClient(auth.ID) }) + if count := reg.GetModelCount(model); count != 1 { + t.Fatalf("expected model count 1 before MarkResult, got %d", count) + } + + m.MarkResult(context.Background(), Result{ + AuthID: auth.ID, + Provider: auth.Provider, + Model: model, + Success: false, + Error: &Error{HTTPStatus: http.StatusTooManyRequests, Message: "rate limited"}, + TransientRateLimit: true, + }) + + updated, ok := m.GetByID(auth.ID) + if !ok || updated == nil { + t.Fatalf("auth %s missing after MarkResult", auth.ID) + } + state := updated.ModelStates[model] + if state == nil { + t.Fatalf("expected per-model state for %s", model) + } + if state.Unavailable { + t.Fatal("expected the model to stay available when the quota deadline has expired") + } + if !state.NextRetryAfter.IsZero() { + t.Fatalf("expected per-model NextRetryAfter to stay zero, got %v", state.NextRetryAfter) + } + if state.Quota.NextRecoverAt.After(time.Now()) { + t.Fatalf("expired quota was replaced with a new window: %v", state.Quota.NextRecoverAt) + } + if state.Quota.BackoffLevel != 3 { + t.Fatalf("expected BackoffLevel to stay 3, got %d", state.Quota.BackoffLevel) + } + if count := reg.GetModelCount(model); count != 1 { + t.Fatalf("expected registry model count 1 (quota failure not recorded), got %d", count) + } +} + +// An unexpired quota window must still veto the skip path, so a hintless +// transient 429 with cooldowns disabled does not shrink or clear it. +func TestManager_MarkResult_Transient429DisabledKeepsActiveQuota(t *testing.T) { + prevQuota := quotaCooldownDisabled.Load() + quotaCooldownDisabled.Store(false) + prevTransient := transientErrorCooldownSeconds.Load() + SetTransientErrorCooldownSeconds(-1) + t.Cleanup(func() { + quotaCooldownDisabled.Store(prevQuota) + transientErrorCooldownSeconds.Store(prevTransient) + }) + + m := NewManager(nil, nil, nil) + model := "test-model-active-quota-skip" + deadline := time.Now().Add(time.Hour) + auth := &Auth{ + ID: "auth-transient-429-active-quota", + Provider: "claude", + ModelStates: map[string]*ModelState{ + model: { + Status: StatusError, + Unavailable: true, + NextRetryAfter: deadline, + Quota: QuotaState{Exceeded: true, Reason: "quota", NextRecoverAt: deadline, BackoffLevel: 3}, + }, + }, + } + if _, errRegister := m.Register(context.Background(), auth); errRegister != nil { + t.Fatalf("register auth: %v", errRegister) + } + + m.MarkResult(context.Background(), Result{ + AuthID: auth.ID, + Provider: auth.Provider, + Model: model, + Success: false, + Error: &Error{HTTPStatus: http.StatusTooManyRequests, Message: "rate limited"}, + TransientRateLimit: true, + }) + + updated, ok := m.GetByID(auth.ID) + if !ok || updated == nil { + t.Fatalf("auth %s missing after MarkResult", auth.ID) + } + state := updated.ModelStates[model] + if state == nil { + t.Fatalf("expected per-model state for %s", model) + } + if !state.Unavailable { + t.Fatal("expected the active quota window to keep the model unavailable") + } + if !state.Quota.Exceeded { + t.Fatal("expected the active quota record to remain exceeded") + } + if !state.Quota.NextRecoverAt.Equal(deadline) { + t.Fatalf("active quota deadline changed: got %v, want %v", state.Quota.NextRecoverAt, deadline) + } + if state.Quota.BackoffLevel != 3 { + t.Fatalf("expected BackoffLevel to stay 3, got %d", state.Quota.BackoffLevel) + } +} + +// A hintless transient 429 with cooldowns disabled must not wipe a more +// serious pre-existing model cooldown (401/403/404/5xx). The auth-level +// path already restores prevUnavailable/prevNextRetry; the per-model +// path must do the same. +func TestManager_MarkResult_Transient429DisabledPreservesExistingModelCooldown(t *testing.T) { + prevQuota := quotaCooldownDisabled.Load() + quotaCooldownDisabled.Store(false) + prevTransient := transientErrorCooldownSeconds.Load() + SetTransientErrorCooldownSeconds(-1) + t.Cleanup(func() { + quotaCooldownDisabled.Store(prevQuota) + transientErrorCooldownSeconds.Store(prevTransient) + }) + + m := NewManager(nil, nil, nil) + auth := &Auth{ID: "auth-transient-429-preserve-model", Provider: "claude"} + if _, errRegister := m.Register(context.Background(), auth); errRegister != nil { + t.Fatalf("register auth: %v", errRegister) + } + + model := "test-model-preserve-cooldown" + m.MarkResult(context.Background(), Result{ + AuthID: auth.ID, + Provider: auth.Provider, + Model: model, + Success: false, + Error: &Error{HTTPStatus: http.StatusForbidden, Message: "forbidden"}, + }) + + afterForbidden, okForbidden := m.GetByID(auth.ID) + if !okForbidden || afterForbidden == nil || afterForbidden.ModelStates[model] == nil { + t.Fatal("expected model state after 403") + } + forbiddenDeadline := afterForbidden.ModelStates[model].NextRetryAfter + if forbiddenDeadline.IsZero() { + t.Fatal("expected 403 to schedule a model cooldown") + } + + m.MarkResult(context.Background(), Result{ + AuthID: auth.ID, + Provider: auth.Provider, + Model: model, + Success: false, + Error: &Error{HTTPStatus: http.StatusTooManyRequests, Message: "rate limited"}, + TransientRateLimit: true, + }) + + updated, ok := m.GetByID(auth.ID) + if !ok || updated == nil { + t.Fatalf("auth %s missing after transient 429", auth.ID) + } + state := updated.ModelStates[model] + if state == nil { + t.Fatalf("expected per-model state for %s", model) + } + if !state.Unavailable { + t.Fatal("expected the existing 403 cooldown to keep the model unavailable") + } + if !state.NextRetryAfter.Equal(forbiddenDeadline) { + t.Fatalf("hintless transient 429 wiped the 403 deadline: got %v, want %v", state.NextRetryAfter, forbiddenDeadline) + } + if state.Quota.Exceeded { + t.Fatal("expected the 403 cooldown not to be rewritten as quota exhausted") + } +} + +// Same as TestManager_MarkResult_Transient429WithoutHintRespectsDisabledCooldown +// but for an auth-level Result (empty Model), which drives applyAuthFailureState +// instead of the per-model branch: the credential must stay available. +func TestManager_MarkResult_Transient429WithoutHintRespectsDisabledCooldownAuthLevel(t *testing.T) { + prevQuota := quotaCooldownDisabled.Load() + quotaCooldownDisabled.Store(false) + prevTransient := transientErrorCooldownSeconds.Load() + SetTransientErrorCooldownSeconds(-1) + t.Cleanup(func() { + quotaCooldownDisabled.Store(prevQuota) + transientErrorCooldownSeconds.Store(prevTransient) + }) + + m := NewManager(nil, nil, nil) + auth := &Auth{ID: "auth-transient-429-disabled-authlevel", Provider: "claude"} + if _, errRegister := m.Register(context.Background(), auth); errRegister != nil { + t.Fatalf("register auth: %v", errRegister) + } + + m.MarkResult(context.Background(), Result{ + AuthID: auth.ID, + Provider: auth.Provider, + Success: false, + Error: &Error{ + HTTPStatus: http.StatusTooManyRequests, + Message: "rate limited", + }, + TransientRateLimit: true, + }) + + updated, ok := m.GetByID(auth.ID) + if !ok || updated == nil { + t.Fatalf("auth %s missing after MarkResult", auth.ID) + } + if updated.Unavailable { + t.Fatal("expected the credential to stay available with transient cooldowns disabled") + } + if !updated.NextRetryAfter.IsZero() { + t.Fatalf("expected credential NextRetryAfter to stay zero with transient cooldowns disabled, got %v", updated.NextRetryAfter) + } + if updated.Quota.Exceeded { + t.Fatal("expected the credential quota state to stay clear with transient cooldowns disabled") + } +} + +func TestApplyAuthFailureState_Transient429DisabledTreatsExpiredQuotaAsInactive(t *testing.T) { + previousTransient := transientErrorCooldownSeconds.Load() + SetTransientErrorCooldownSeconds(-1) + t.Cleanup(func() { transientErrorCooldownSeconds.Store(previousTransient) }) + + now := time.Now() + expired := now.Add(-time.Minute) + auth := &Auth{ + ID: "auth-level-expired-quota-skip", + Quota: QuotaState{Exceeded: true, Reason: "quota", NextRecoverAt: expired, BackoffLevel: 3}, + } + rateLimitErr := &Error{Code: "rate_limit", Message: "RATE_LIMIT_EXCEEDED", HTTPStatus: http.StatusTooManyRequests} + applyAuthFailureState(auth, rateLimitErr, nil, now, false, true) + + if auth.Unavailable { + t.Fatal("expected the credential to stay available when the quota deadline has expired") + } + if !auth.NextRetryAfter.IsZero() { + t.Fatalf("expected NextRetryAfter to stay zero, got %v", auth.NextRetryAfter) + } + if auth.Quota.NextRecoverAt.After(now) { + t.Fatalf("expired quota was replaced with a new window: %v", auth.Quota.NextRecoverAt) + } + if auth.Quota.BackoffLevel != 3 { + t.Fatalf("expected BackoffLevel to stay 3, got %d", auth.Quota.BackoffLevel) + } + if auth.StatusMessage == "quota exhausted" { + t.Fatal("expected skip path to restore the previous status message") + } +} + func TestManager_MarkResult_TransientErrorCooldownDisabled(t *testing.T) { prevQuota := quotaCooldownDisabled.Load() quotaCooldownDisabled.Store(false) diff --git a/sdk/cliproxy/auth/conductor_prepare_classification_test.go b/sdk/cliproxy/auth/conductor_prepare_classification_test.go new file mode 100644 index 00000000000..95555322f75 --- /dev/null +++ b/sdk/cliproxy/auth/conductor_prepare_classification_test.go @@ -0,0 +1,245 @@ +package auth + +import ( + "context" + "errors" + "net/http" + "testing" + "time" + + internalconfig "github.com/router-for-me/CLIProxyAPI/v7/internal/config" + "github.com/router-for-me/CLIProxyAPI/v7/internal/registry" + "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executionregistry" + cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor" +) + +func TestPrepareErrorPropagatesTransientRateLimit(t *testing.T) { + withQuotaCooldownEnabled(t) + + hint := time.Duration(observedExhaustedQuotaHint) + paths := []struct { + name string + run func(context.Context, *Manager, string) error + }{ + { + name: "execute", + run: func(ctx context.Context, manager *Manager, model string) error { + _, errExecute := manager.Execute(ctx, []string{"claude"}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{}) + return errExecute + }, + }, + { + name: "count tokens", + run: func(ctx context.Context, manager *Manager, model string) error { + _, errCount := manager.ExecuteCount(ctx, []string{"claude"}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{}) + return errCount + }, + }, + { + name: "stream", + run: func(ctx context.Context, manager *Manager, model string) error { + _, errStream := manager.ExecuteStream(ctx, []string{"claude"}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{Stream: true}) + return errStream + }, + }, + } + + for _, path := range paths { + t.Run(path.name, func(t *testing.T) { + hook := &resultCaptureHook{} + executor := &claudeCancellationTestExecutor{ + prepareFn: func(context.Context, *Auth) (*Auth, error) { + return nil, streamTransientRateLimitError{retryAfter: hint} + }, + } + manager, auth, model := newClaudeCancellationTestManager(t, executor, hook) + + before := time.Now() + if errRun := path.run(context.Background(), manager, model); errRun == nil { + t.Fatal("expected prepare 429 to fail the request") + } + if executor.executeCalls.Load()+executor.countCalls.Load()+executor.streamCalls.Load() != 0 { + t.Fatal("executor ran after request preparation failed") + } + + results := hook.Results() + if len(results) != 1 { + t.Fatalf("hook results = %d, want 1", len(results)) + } + got := results[0] + if !got.TransientRateLimit { + t.Fatal("expected oauth token-refresh 429 during prepare to be transient so the conductor skips the quota ladder") + } + if got.RetryAfter == nil || *got.RetryAfter != hint { + t.Fatalf("RetryAfter = %v, want provider hint %v", got.RetryAfter, hint) + } + + updated, ok := manager.GetByID(auth.ID) + if !ok || updated == nil || updated.ModelStates[model] == nil { + t.Fatalf("expected model state after prepare 429") + } + state := updated.ModelStates[model] + if state.Quota.BackoffLevel != 0 { + t.Fatalf("expected BackoffLevel 0 on the transient prepare-429 path, got %d", state.Quota.BackoffLevel) + } + if state.Quota.Exceeded { + t.Fatal("transient prepare 429 must not set Quota.Exceeded") + } + if gotWindow := state.NextRetryAfter.Sub(before); gotWindow < transientRateLimitMinimum { + t.Fatalf("prepare 429 took the quota ladder: NextRetryAfter delta %v, want at least the 10s transient floor", gotWindow) + } + }) + } +} + +func TestPrepareErrorWithoutTransientKeepsQuotaLadder(t *testing.T) { + withQuotaCooldownEnabled(t) + + hint := time.Duration(observedExhaustedQuotaHint) + hook := &resultCaptureHook{} + executor := &claudeCancellationTestExecutor{ + prepareFn: func(context.Context, *Auth) (*Auth, error) { + return nil, &retryAfterStatusError{ + status: http.StatusTooManyRequests, + message: "quota", + retryAfter: hint, + } + }, + } + manager, auth, model := newClaudeCancellationTestManager(t, executor, hook) + + if _, errExecute := manager.Execute(context.Background(), []string{"claude"}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{}); errExecute == nil { + t.Fatal("expected unclassified prepare 429 to fail the request") + } + + results := hook.Results() + if len(results) != 1 { + t.Fatalf("hook results = %d, want 1", len(results)) + } + if results[0].TransientRateLimit { + t.Fatal("unclassified prepare 429 must not become transient") + } + if results[0].RetryAfter == nil || *results[0].RetryAfter != hint { + t.Fatalf("RetryAfter = %v, want provider hint %v", results[0].RetryAfter, hint) + } + + updated, ok := manager.GetByID(auth.ID) + if !ok || updated == nil || updated.ModelStates[model] == nil { + t.Fatal("expected model state after unclassified prepare 429") + } + state := updated.ModelStates[model] + if state.Quota.BackoffLevel != 1 { + t.Fatalf("expected BackoffLevel 1 for unclassified prepare 429, got %d", state.Quota.BackoffLevel) + } +} + +func TestPrepareGenericErrorIsNotTransient(t *testing.T) { + hook := &resultCaptureHook{} + executor := &claudeCancellationTestExecutor{ + prepareFn: func(context.Context, *Auth) (*Auth, error) { + return nil, errors.New("missing project_id") + }, + } + manager, _, model := newClaudeCancellationTestManager(t, executor, hook) + + if _, errExecute := manager.Execute(context.Background(), []string{"claude"}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{}); errExecute == nil { + t.Fatal("expected generic prepare error to fail the request") + } + + results := hook.Results() + if len(results) != 1 { + t.Fatalf("hook results = %d, want 1", len(results)) + } + if results[0].TransientRateLimit { + t.Fatal("generic prepare error must not be classified as transient") + } + if results[0].RetryAfter != nil { + t.Fatalf("RetryAfter = %v, want nil", results[0].RetryAfter) + } +} + +func TestHomePrepareErrorPropagatesTransientRateLimit(t *testing.T) { + hint := time.Duration(observedExhaustedQuotaHint) + paths := []struct { + name string + run func(*Manager, context.Context) error + }{ + { + name: "Execute", + run: func(manager *Manager, ctx context.Context) error { + _, errExecute := manager.Execute(ctx, []string{"antigravity"}, cliproxyexecutor.Request{Model: "test-model"}, cliproxyexecutor.Options{}) + return errExecute + }, + }, + { + name: "Count", + run: func(manager *Manager, ctx context.Context) error { + _, errCount := manager.ExecuteCount(ctx, []string{"antigravity"}, cliproxyexecutor.Request{Model: "test-model"}, cliproxyexecutor.Options{}) + return errCount + }, + }, + { + name: "Stream", + run: func(manager *Manager, ctx context.Context) error { + result, errStream := manager.ExecuteStream(ctx, []string{"antigravity"}, cliproxyexecutor.Request{Model: "test-model"}, cliproxyexecutor.Options{Stream: true}) + if errStream != nil { + return errStream + } + if result != nil { + for range result.Chunks { + } + } + return nil + }, + }, + } + + for _, path := range paths { + t.Run(path.name, func(t *testing.T) { + store := &requestPrepareStore{} + hook := &resultCaptureHook{} + executor := &requestPrepareExecutor{prepareErr: streamTransientRateLimitError{retryAfter: hint}} + manager := NewManager(store, nil, hook) + manager.SetConfig(&internalconfig.Config{Home: internalconfig.HomeConfig{Enabled: true}}) + manager.PublishHomeDispatch(&homeRequestPrepareDispatcher{}, executionregistry.New(), 1) + manager.RegisterExecutor(executor) + localAuth := &Auth{ + ID: "same-id", + Provider: "antigravity", + Status: StatusActive, + Metadata: map[string]any{"access_token": "local-token", "source": "local"}, + } + if _, errRegister := manager.Register(WithSkipPersist(context.Background()), localAuth); errRegister != nil { + t.Fatalf("register local auth: %v", errRegister) + } + registry.GetGlobalRegistry().RegisterClient(localAuth.ID, localAuth.Provider, []*registry.ModelInfo{{ID: "test-model"}}) + t.Cleanup(func() { registry.GetGlobalRegistry().UnregisterClient(localAuth.ID) }) + + if errRun := path.run(manager, context.Background()); errRun == nil { + t.Fatal("expected home prepare 429 to fail the request") + } + + results := hook.Results() + if len(results) != 1 { + t.Fatalf("hook results = %#v, want exactly one ephemeral result", results) + } + got := results[0] + if !got.TransientRateLimit { + t.Fatal("expected home prepare 429 to reach the conductor as a transient rate limit") + } + if got.RetryAfter == nil || *got.RetryAfter != hint { + t.Fatalf("RetryAfter = %v, want provider hint %v", got.RetryAfter, hint) + } + + current, ok := manager.GetByID(localAuth.ID) + if !ok || current == nil { + t.Fatal("local auth disappeared") + } + if current.ModelStates != nil { + if state := current.ModelStates["test-model"]; state != nil && state.Quota.BackoffLevel != 0 { + t.Fatalf("Home prepare must not mutate local BackoffLevel, got %d", state.Quota.BackoffLevel) + } + } + }) + } +} diff --git a/sdk/cliproxy/auth/conductor_stream.go b/sdk/cliproxy/auth/conductor_stream.go index cca3c3da74e..4ad7053f72d 100644 --- a/sdk/cliproxy/auth/conductor_stream.go +++ b/sdk/cliproxy/auth/conductor_stream.go @@ -133,6 +133,11 @@ func (m *Manager) wrapStreamResult(ctx context.Context, auth *Auth, provider, re rerr := resultErrorFromError(chunk.Err) action, okAction := matchRequestScopedErrorAction(auth, chunk.Err, m.runtimeConfigSnapshot()) result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: false, Error: rerr, Options: opts} + result.RetryAfter = retryAfterFromError(chunk.Err) + result.TransientRateLimit = isTransientRateLimitError(chunk.Err) + if isCredentialScopedError(chunk.Err) { + result.CredentialScope = true + } applyRequestScopedActionToResult(action, okAction, &result) m.recordExecutionResult(ctx, result, auth, ephemeralResult) } @@ -290,6 +295,7 @@ func (m *Manager) executeStreamWithModelPool(ctx context.Context, executor Provi action, okAction := matchRequestScopedErrorAction(auth, errStream, m.runtimeConfigSnapshot()) result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: false, Error: rerr, Options: execOpts} result.RetryAfter = retryAfterFromError(errStream) + result.TransientRateLimit = isTransientRateLimitError(errStream) if isCredentialScopedError(errStream) { result.CredentialScope = true } @@ -380,6 +386,7 @@ func (m *Manager) executeStreamWithModelPool(ctx context.Context, executor Provi rerr := resultErrorFromError(bootstrapErr) result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: false, Error: rerr, Options: execOpts} result.RetryAfter = retryAfterFromError(bootstrapErr) + result.TransientRateLimit = isTransientRateLimitError(bootstrapErr) if isCredentialScopedError(bootstrapErr) { result.CredentialScope = true } @@ -399,6 +406,7 @@ func (m *Manager) executeStreamWithModelPool(ctx context.Context, executor Provi rerr := resultErrorFromError(bootstrapErr) result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: false, Error: rerr, Options: execOpts} result.RetryAfter = retryAfterFromError(bootstrapErr) + result.TransientRateLimit = isTransientRateLimitError(bootstrapErr) if isCredentialScopedError(bootstrapErr) { result.CredentialScope = true } @@ -410,6 +418,7 @@ func (m *Manager) executeStreamWithModelPool(ctx context.Context, executor Provi rerr := resultErrorFromError(bootstrapErr) result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: false, Error: rerr, Options: execOpts} result.RetryAfter = retryAfterFromError(bootstrapErr) + result.TransientRateLimit = isTransientRateLimitError(bootstrapErr) if isCredentialScopedError(bootstrapErr) { result.CredentialScope = true } @@ -424,6 +433,7 @@ func (m *Manager) executeStreamWithModelPool(ctx context.Context, executor Provi rerr := resultErrorFromError(bootstrapErr) result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: false, Error: rerr, Options: execOpts} result.RetryAfter = retryAfterFromError(bootstrapErr) + result.TransientRateLimit = isTransientRateLimitError(bootstrapErr) if isCredentialScopedError(bootstrapErr) { result.CredentialScope = true } diff --git a/sdk/cliproxy/auth/conductor_stream_classification_test.go b/sdk/cliproxy/auth/conductor_stream_classification_test.go new file mode 100644 index 00000000000..2a8c2e620a5 --- /dev/null +++ b/sdk/cliproxy/auth/conductor_stream_classification_test.go @@ -0,0 +1,115 @@ +package auth + +import ( + "context" + "net/http" + "testing" + "time" + + cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor" +) + +type streamTransientRateLimitError struct { + retryAfter time.Duration +} + +func (e streamTransientRateLimitError) Error() string { return "429 rate limited" } +func (e streamTransientRateLimitError) StatusCode() int { return http.StatusTooManyRequests } +func (e streamTransientRateLimitError) TransientRateLimit() bool { return true } + +func (e streamTransientRateLimitError) RetryAfter() *time.Duration { + hint := e.retryAfter + return &hint +} + +// TestExecuteStreamFloorsTransientRateLimitHint covers the streaming failure +// path: a tiny provider hint must not repeatedly select the same credential. +func TestExecuteStreamFloorsTransientRateLimitHint(t *testing.T) { + withQuotaCooldownEnabled(t) + + hint := time.Duration(observedExhaustedQuotaHint) + executor := &claudeCancellationTestExecutor{ + streamFn: func(context.Context, *Auth) (*cliproxyexecutor.StreamResult, error) { + return nil, streamTransientRateLimitError{retryAfter: hint} + }, + } + manager, auth, model := newClaudeCancellationTestManager(t, executor, nil) + + before := time.Now() + _, errStream := manager.ExecuteStream(context.Background(), []string{"claude"}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{Stream: true}) + if errStream == nil { + t.Fatal("expected the stream request to fail with the upstream 429") + } + + updated, ok := manager.GetByID(auth.ID) + if !ok || updated == nil { + t.Fatalf("GetByID(%q) did not return auth", auth.ID) + } + state := updated.ModelStates[model] + if state == nil { + t.Fatalf("expected model state for %q after the failure", model) + } + if state.Quota.BackoffLevel != 0 { + t.Fatalf("expected BackoffLevel to stay 0 for a transient rate limit, got %d", state.Quota.BackoffLevel) + } + if state.Quota.Exceeded { + t.Fatal("transient stream 429 must not set Quota.Exceeded") + } + if got := state.NextRetryAfter.Sub(before); got < transientRateLimitMinimum { + t.Fatalf("sub-second transient stream hint was not floored: got %v, want at least %v", got, transientRateLimitMinimum) + } +} + +// TestExecuteStreamFloorsLateTransientRateLimitHint covers wrapStreamResult: +// a classified 429 after the first payload must not take the quota ladder. +func TestExecuteStreamFloorsLateTransientRateLimitHint(t *testing.T) { + withQuotaCooldownEnabled(t) + + hint := time.Duration(observedExhaustedQuotaHint) + executor := &claudeCancellationTestExecutor{ + streamFn: func(context.Context, *Auth) (*cliproxyexecutor.StreamResult, error) { + ch := make(chan cliproxyexecutor.StreamChunk, 2) + ch <- cliproxyexecutor.StreamChunk{Payload: []byte(`data: {"type":"response.created"}`)} + ch <- cliproxyexecutor.StreamChunk{Err: streamTransientRateLimitError{retryAfter: hint}} + close(ch) + return &cliproxyexecutor.StreamResult{Chunks: ch}, nil + }, + } + manager, auth, model := newClaudeCancellationTestManager(t, executor, nil) + + before := time.Now() + result, errStream := manager.ExecuteStream(context.Background(), []string{"claude"}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{Stream: true}) + if errStream != nil { + t.Fatalf("expected a committed stream after the first payload, got error: %v", errStream) + } + if result == nil { + t.Fatal("expected a committed stream result") + } + var sawErr bool + for chunk := range result.Chunks { + if chunk.Err != nil { + sawErr = true + } + } + if !sawErr { + t.Fatal("expected the classified 429 to arrive after the first payload") + } + + updated, ok := manager.GetByID(auth.ID) + if !ok || updated == nil { + t.Fatalf("GetByID(%q) did not return auth", auth.ID) + } + state := updated.ModelStates[model] + if state == nil { + t.Fatalf("expected model state for %q after the late stream failure", model) + } + if state.Quota.BackoffLevel != 0 { + t.Fatalf("expected BackoffLevel to stay 0 for a late transient rate limit, got %d", state.Quota.BackoffLevel) + } + if state.Quota.Exceeded { + t.Fatal("late transient stream 429 must not set Quota.Exceeded") + } + if got := state.NextRetryAfter.Sub(before); got < transientRateLimitMinimum { + t.Fatalf("sub-second late stream hint was not floored: got %v, want at least %v", got, transientRateLimitMinimum) + } +} diff --git a/sdk/cliproxy/auth/cooldown_backoff_test.go b/sdk/cliproxy/auth/cooldown_backoff_test.go index 73a7bdcf392..1a55279821d 100644 --- a/sdk/cliproxy/auth/cooldown_backoff_test.go +++ b/sdk/cliproxy/auth/cooldown_backoff_test.go @@ -2,6 +2,8 @@ package auth import ( "context" + "errors" + "fmt" "net/http" "testing" "time" @@ -117,7 +119,7 @@ func TestApplyAuthFailureStateQuotaBackoffOncePerWindow(t *testing.T) { quotaErr := &Error{Code: "rate_limit", Message: "quota", HTTPStatus: http.StatusTooManyRequests} auth := &Auth{ID: "auth-level-quota"} - applyAuthFailureState(auth, quotaErr, nil, now, false) + applyAuthFailureState(auth, quotaErr, nil, now, false, false) if auth.Quota.BackoffLevel != 1 { t.Fatalf("expected BackoffLevel 1 after first failure, got %d", auth.Quota.BackoffLevel) } @@ -127,7 +129,7 @@ func TestApplyAuthFailureStateQuotaBackoffOncePerWindow(t *testing.T) { } // In-window failure keeps the current window and level. - applyAuthFailureState(auth, quotaErr, nil, now.Add(100*time.Millisecond), false) + applyAuthFailureState(auth, quotaErr, nil, now.Add(100*time.Millisecond), false, false) if auth.Quota.BackoffLevel != 1 { t.Fatalf("expected BackoffLevel to stay 1 for in-window failure, got %d", auth.Quota.BackoffLevel) } @@ -136,7 +138,7 @@ func TestApplyAuthFailureStateQuotaBackoffOncePerWindow(t *testing.T) { } // A failure after the window expired escalates to the next level. - applyAuthFailureState(auth, quotaErr, nil, now.Add(2*time.Second), false) + applyAuthFailureState(auth, quotaErr, nil, now.Add(2*time.Second), false, false) if auth.Quota.BackoffLevel != 2 { t.Fatalf("expected BackoffLevel 2 after post-window failure, got %d", auth.Quota.BackoffLevel) } @@ -146,7 +148,7 @@ func TestApplyAuthFailureStateQuotaBackoffOncePerWindow(t *testing.T) { // A provider supplied retry hint always takes effect, even in-window. retryAfter := 10 * time.Second - applyAuthFailureState(auth, quotaErr, &retryAfter, now.Add(3*time.Second), false) + applyAuthFailureState(auth, quotaErr, &retryAfter, now.Add(3*time.Second), false, false) if auth.Quota.BackoffLevel != 2 { t.Fatalf("expected BackoffLevel to stay 2 with retry hint, got %d", auth.Quota.BackoffLevel) } @@ -308,3 +310,493 @@ func TestJitteredCooldownWaitBounds(t *testing.T) { t.Fatalf("expected sub-4ns wait to stay unchanged, got %v", got) } } + +// Gemini and Antigravity answer an exhausted daily quota with a RetryInfo hint of +// well under a second (see internal/runtime/executor/helps/json_retry_helpers.go). +// Honouring such a hint verbatim returned dead credentials to the pool a few hundred +// milliseconds later and pinned the backoff ladder at its current level forever, so +// every request kept walking the whole exhausted pool before failing. +const observedExhaustedQuotaHint = 479417207 * time.Nanosecond + +func TestMarkResultSubSecondQuotaHintStillEscalates(t *testing.T) { + withQuotaCooldownEnabled(t) + + expired := time.Now().Add(-time.Second) + manager := NewManager(nil, nil, nil) + auth := &Auth{ + ID: "auth-quota-subsecond-hint", + Provider: "codex", + Metadata: map[string]any{"type": "codex"}, + ModelStates: map[string]*ModelState{ + "gpt-5": { + Status: StatusError, + Unavailable: true, + NextRetryAfter: expired, + Quota: QuotaState{Exceeded: true, Reason: "quota", NextRecoverAt: expired, BackoffLevel: 3}, + }, + }, + } + if _, errRegister := manager.Register(WithSkipPersist(context.Background()), auth); errRegister != nil { + t.Fatalf("Register returned error: %v", errRegister) + } + + hint := observedExhaustedQuotaHint + result := quotaResult(auth.ID, "gpt-5") + result.RetryAfter = &hint + + before := time.Now() + manager.MarkResult(context.Background(), result) + + updated, ok := manager.GetByID(auth.ID) + if !ok || updated == nil || updated.ModelStates["gpt-5"] == nil { + t.Fatalf("expected model state after failure") + } + state := updated.ModelStates["gpt-5"] + if state.Quota.BackoffLevel != 4 { + t.Fatalf("expected BackoffLevel 4 after hinted post-window failure, got %d", state.Quota.BackoffLevel) + } + if !state.Quota.NextRecoverAt.After(before.Add(hint)) { + t.Fatalf("sub-second hint was not floored: window closes at %v, the hint alone would close it at %v", state.Quota.NextRecoverAt, before.Add(hint)) + } + if got := state.Quota.NextRecoverAt.Sub(before); got < 8*quotaBackoffBase { + t.Fatalf("expected at least the level-3 ladder step (%v), got %v", 8*quotaBackoffBase, got) + } +} + +func TestApplyAuthFailureStateSubSecondQuotaHintStillEscalates(t *testing.T) { + now := time.Now() + quotaErr := &Error{Code: "rate_limit", Message: "quota", HTTPStatus: http.StatusTooManyRequests} + hint := observedExhaustedQuotaHint + auth := &Auth{ID: "auth-subsecond-hint"} + + applyAuthFailureState(auth, quotaErr, &hint, now, false, false) + if auth.Quota.BackoffLevel != 1 { + t.Fatalf("expected BackoffLevel 1 after the first hinted failure, got %d", auth.Quota.BackoffLevel) + } + if !auth.Quota.NextRecoverAt.Equal(now.Add(quotaBackoffBase)) { + t.Fatalf("expected the sub-second hint to be floored at %v, got %v", now.Add(quotaBackoffBase), auth.Quota.NextRecoverAt) + } + + // A later failure, once the first window has closed, must climb the ladder even + // though the provider keeps repeating the same sub-second hint. + after := now.Add(20 * time.Second) + applyAuthFailureState(auth, quotaErr, &hint, after, false, false) + if auth.Quota.BackoffLevel != 2 { + t.Fatalf("expected BackoffLevel 2 after the repeated hinted failure, got %d", auth.Quota.BackoffLevel) + } + if !auth.Quota.NextRecoverAt.Equal(after.Add(2 * quotaBackoffBase)) { + t.Fatalf("expected the escalated window to close at %v, got %v", after.Add(2*quotaBackoffBase), auth.Quota.NextRecoverAt) + } +} + +func TestMarkResultZeroRetryAfterDoesNotApplyLadderFloor(t *testing.T) { + withQuotaCooldownEnabled(t) + + manager := NewManager(nil, nil, nil) + auth := &Auth{ + ID: "auth-zero-retry-after", + Provider: "codex", + Metadata: map[string]any{"type": "codex"}, + ModelStates: map[string]*ModelState{ + "gpt-5": { + Status: StatusActive, + Quota: QuotaState{BackoffLevel: 0}, + }, + }, + } + if _, errRegister := manager.Register(WithSkipPersist(context.Background()), auth); errRegister != nil { + t.Fatalf("Register returned error: %v", errRegister) + } + + zeroHint := time.Duration(0) + result := quotaResult(auth.ID, "gpt-5") + result.RetryAfter = &zeroHint + result.TransientRateLimit = true + + now := time.Now() + manager.MarkResult(context.Background(), result) + + updated, ok := manager.GetByID(auth.ID) + if !ok || updated == nil || updated.ModelStates["gpt-5"] == nil { + t.Fatalf("expected model state after failure") + } + state := updated.ModelStates["gpt-5"] + if state.Quota.BackoffLevel != 0 { + t.Fatalf("expected BackoffLevel to remain 0 for zero RetryAfter, got %d", state.Quota.BackoffLevel) + } + if state.Quota.NextRecoverAt.After(now.Add(500 * time.Millisecond)) { + t.Fatalf("zero RetryAfter was given ladder floor: NextRecoverAt=%v, want <= %v", state.Quota.NextRecoverAt, now) + } +} + +func TestApplyAuthFailureStateZeroRetryAfterDoesNotApplyLadderFloor(t *testing.T) { + now := time.Now() + err := &Error{Code: "rate_limit", Message: "websocket_connection_limit_reached", HTTPStatus: http.StatusTooManyRequests} + zeroHint := time.Duration(0) + auth := &Auth{ID: "auth-zero-hint"} + + applyAuthFailureState(auth, err, &zeroHint, now, false, true) + if auth.Quota.BackoffLevel != 0 { + t.Fatalf("expected BackoffLevel 0 for zero RetryAfter, got %d", auth.Quota.BackoffLevel) + } + if auth.Quota.NextRecoverAt.After(now) { + t.Fatalf("expected zero RetryAfter not to receive ladder floor, NextRecoverAt=%v, want %v", auth.Quota.NextRecoverAt, now) + } + if auth.NextRetryAfter.After(now) { + t.Fatalf("expected NextRetryAfter not to receive ladder floor, NextRetryAfter=%v, want %v", auth.NextRetryAfter, now) + } +} + +func TestMarkResultZeroRetryAfterExhaustedQuotaUsesLadderFloor(t *testing.T) { + withQuotaCooldownEnabled(t) + + manager := NewManager(nil, nil, nil) + auth := &Auth{ + ID: "auth-zero-retry-after-exhausted", + Provider: "codex", + Metadata: map[string]any{"type": "codex"}, + ModelStates: map[string]*ModelState{ + "gpt-5": { + Status: StatusActive, + Quota: QuotaState{BackoffLevel: 0}, + }, + }, + } + if _, errRegister := manager.Register(WithSkipPersist(context.Background()), auth); errRegister != nil { + t.Fatalf("Register returned error: %v", errRegister) + } + + zeroHint := time.Duration(0) + result := quotaResult(auth.ID, "gpt-5") + result.RetryAfter = &zeroHint + + before := time.Now() + manager.MarkResult(context.Background(), result) + + updated, ok := manager.GetByID(auth.ID) + if !ok || updated == nil || updated.ModelStates["gpt-5"] == nil { + t.Fatalf("expected model state after failure") + } + state := updated.ModelStates["gpt-5"] + if state.Quota.BackoffLevel != 1 { + t.Fatalf("expected BackoffLevel 1 after exhausted-quota 429 with 0s hint, got %d", state.Quota.BackoffLevel) + } + if !state.Quota.NextRecoverAt.After(before) { + t.Fatalf("0s exhausted-quota hint bypassed the ladder: NextRecoverAt=%v, want after %v", state.Quota.NextRecoverAt, before) + } + if got := state.Quota.NextRecoverAt.Sub(before); got < quotaBackoffBase { + t.Fatalf("expected at least the first ladder step (%v), got %v", quotaBackoffBase, got) + } +} + +func TestApplyAuthFailureStateZeroRetryAfterExhaustedQuotaUsesLadderFloor(t *testing.T) { + now := time.Now() + quotaErr := &Error{Code: "rate_limit", Message: "quota", HTTPStatus: http.StatusTooManyRequests} + zeroHint := time.Duration(0) + auth := &Auth{ID: "auth-zero-hint-exhausted"} + + applyAuthFailureState(auth, quotaErr, &zeroHint, now, false, false) + if auth.Quota.BackoffLevel != 1 { + t.Fatalf("expected BackoffLevel 1 after exhausted-quota 429 with 0s hint, got %d", auth.Quota.BackoffLevel) + } + if !auth.Quota.NextRecoverAt.Equal(now.Add(quotaBackoffBase)) { + t.Fatalf("expected 0s exhausted-quota hint to be floored at %v, got %v", now.Add(quotaBackoffBase), auth.Quota.NextRecoverAt) + } + if !auth.NextRetryAfter.Equal(now.Add(quotaBackoffBase)) { + t.Fatalf("expected NextRetryAfter to use the ladder floor at %v, got %v", now.Add(quotaBackoffBase), auth.NextRetryAfter) + } +} + +func TestMarkResultTransientRateLimitFloorsProviderHint(t *testing.T) { + withQuotaCooldownEnabled(t) + + manager := NewManager(nil, nil, nil) + auth := &Auth{ + ID: "auth-transient-rate-limit", + Provider: "codex", + Metadata: map[string]any{"type": "codex"}, + ModelStates: map[string]*ModelState{ + "gpt-5": { + Status: StatusActive, + Quota: QuotaState{BackoffLevel: 0}, + }, + }, + } + if _, errRegister := manager.Register(WithSkipPersist(context.Background()), auth); errRegister != nil { + t.Fatalf("Register returned error: %v", errRegister) + } + + hint := observedExhaustedQuotaHint + result := quotaResult(auth.ID, "gpt-5") + result.RetryAfter = &hint + result.TransientRateLimit = true + + before := time.Now() + manager.MarkResult(context.Background(), result) + + updated, ok := manager.GetByID(auth.ID) + if !ok || updated == nil || updated.ModelStates["gpt-5"] == nil { + t.Fatalf("expected model state after failure") + } + state := updated.ModelStates["gpt-5"] + if state.Quota.BackoffLevel != 0 { + t.Fatalf("expected BackoffLevel to stay 0 for a transient rate limit, got %d", state.Quota.BackoffLevel) + } + if state.Quota.Exceeded { + t.Fatal("transient 429 must not set Quota.Exceeded") + } + if got := state.NextRetryAfter.Sub(before); got < transientRateLimitMinimum { + t.Fatalf("sub-second transient hint was not floored: got %v, want at least %v", got, transientRateLimitMinimum) + } +} + +func TestApplyAuthFailureStateTransientRateLimitFloorsProviderHint(t *testing.T) { + now := time.Now() + rateLimitErr := &Error{Code: "rate_limit", Message: "RATE_LIMIT_EXCEEDED", HTTPStatus: http.StatusTooManyRequests} + hint := observedExhaustedQuotaHint + + transient := &Auth{ID: "auth-transient-hint"} + applyAuthFailureState(transient, rateLimitErr, &hint, now, false, true) + if transient.Quota.BackoffLevel != 0 { + t.Fatalf("expected BackoffLevel to stay 0 for a transient rate limit, got %d", transient.Quota.BackoffLevel) + } + if transient.Quota.Exceeded { + t.Fatal("transient 429 must not set Quota.Exceeded") + } + if !transient.Quota.NextRecoverAt.IsZero() { + t.Fatalf("transient 429 must not write Quota.NextRecoverAt, got %v", transient.Quota.NextRecoverAt) + } + if !transient.NextRetryAfter.Equal(now.Add(transientRateLimitMinimum)) { + t.Fatalf("expected NextRetryAfter to use the transient floor at %v, got %v", now.Add(transientRateLimitMinimum), transient.NextRetryAfter) + } + + longHint := 2 * transientRateLimitMinimum + longHintAuth := &Auth{ID: "auth-transient-long-hint"} + applyAuthFailureState(longHintAuth, rateLimitErr, &longHint, now, false, true) + if !longHintAuth.NextRetryAfter.Equal(now.Add(longHint)) { + t.Fatalf("expected a longer transient hint to extend cooldown to %v, got %v", now.Add(longHint), longHintAuth.NextRetryAfter) + } + + // The same sub-second hint on an exhausted quota must still use the ladder. + exhausted := &Auth{ID: "auth-exhausted-hint"} + applyAuthFailureState(exhausted, rateLimitErr, &hint, now, false, false) + if exhausted.Quota.BackoffLevel != 1 { + t.Fatalf("expected BackoffLevel 1 for an exhausted-quota failure, got %d", exhausted.Quota.BackoffLevel) + } + if !exhausted.Quota.NextRecoverAt.Equal(now.Add(quotaBackoffBase)) { + t.Fatalf("expected the exhausted-quota hint to stay floored at %v, got %v", now.Add(quotaBackoffBase), exhausted.Quota.NextRecoverAt) + } +} + +func TestApplyAuthFailureStateTransientRateLimitDoesNotEscalate(t *testing.T) { + previousTransient := transientErrorCooldownSeconds.Load() + SetTransientErrorCooldownSeconds(0) + t.Cleanup(func() { transientErrorCooldownSeconds.Store(previousTransient) }) + + now := time.Now() + rateLimitErr := &Error{Code: "rate_limit", Message: "RATE_LIMIT_EXCEEDED", HTTPStatus: http.StatusTooManyRequests} + auth := &Auth{ID: "auth-transient-repeat"} + + applyAuthFailureState(auth, rateLimitErr, nil, now, false, true) + if auth.Quota.BackoffLevel != 0 { + t.Fatalf("expected BackoffLevel 0 after the first transient rate limit, got %d", auth.Quota.BackoffLevel) + } + if auth.Quota.Exceeded { + t.Fatal("transient 429 must not set Quota.Exceeded") + } + firstRecover := auth.NextRetryAfter + if !firstRecover.Equal(now.Add(transientRateLimitMinimum)) { + t.Fatalf("expected first transient cooldown to close at %v, got %v", now.Add(transientRateLimitMinimum), firstRecover) + } + + after := firstRecover.Add(time.Second) + applyAuthFailureState(auth, rateLimitErr, nil, after, false, true) + if auth.Quota.BackoffLevel != 0 { + t.Fatalf("expected repeated transient rate limit not to advance BackoffLevel, got %d", auth.Quota.BackoffLevel) + } + if want := after.Add(transientRateLimitMinimum); !auth.NextRetryAfter.Equal(want) { + t.Fatalf("expected repeated transient cooldown to close at %v, got %v", want, auth.NextRetryAfter) + } +} + +type classifiedRateLimitError struct { + transient bool +} + +func (e classifiedRateLimitError) Error() string { return "429 rate limited" } +func (e classifiedRateLimitError) StatusCode() int { return http.StatusTooManyRequests } +func (e classifiedRateLimitError) TransientRateLimit() bool { return e.transient } + +func TestIsTransientRateLimitErrorDetectsWrappedProviderClassification(t *testing.T) { + if isTransientRateLimitError(nil) { + t.Fatal("expected a nil error not to be transient") + } + if isTransientRateLimitError(errors.New("boom")) { + t.Fatal("expected an unclassified error not to be transient") + } + if isTransientRateLimitError(classifiedRateLimitError{}) { + t.Fatal("expected an exhausted-quota classification not to be transient") + } + if !isTransientRateLimitError(fmt.Errorf("upstream: %w", classifiedRateLimitError{transient: true})) { + t.Fatal("expected a wrapped transient rate limit classification to be detected") + } +} + +func TestMarkResultTransientRateLimitDoesNotSetQuotaExceeded(t *testing.T) { + withQuotaCooldownEnabled(t) + + manager := NewManager(nil, nil, nil) + authID := "auth-transient-no-quota-marker" + model := "gpt-5-transient-no-quota-marker" + auth := &Auth{ + ID: authID, + Provider: "codex", + Metadata: map[string]any{"type": "codex"}, + ModelStates: map[string]*ModelState{ + model: {Status: StatusActive, Quota: QuotaState{BackoffLevel: 0}}, + }, + } + if _, errRegister := manager.Register(WithSkipPersist(context.Background()), auth); errRegister != nil { + t.Fatalf("Register returned error: %v", errRegister) + } + + reg := registry.GetGlobalRegistry() + reg.RegisterClient(authID, "codex", []*registry.ModelInfo{{ID: model}}) + t.Cleanup(func() { reg.UnregisterClient(authID) }) + + hint := observedExhaustedQuotaHint + result := quotaResult(authID, model) + result.RetryAfter = &hint + result.TransientRateLimit = true + + before := time.Now() + manager.MarkResult(context.Background(), result) + + updated, ok := manager.GetByID(authID) + if !ok || updated == nil || updated.ModelStates[model] == nil { + t.Fatalf("expected model state after failure") + } + state := updated.ModelStates[model] + if state.Quota.Exceeded { + t.Fatal("transient 429 must not set Quota.Exceeded") + } + if state.Quota.Reason == "quota" { + t.Fatalf("transient 429 must not set Reason %q", state.Quota.Reason) + } + if !state.Quota.NextRecoverAt.IsZero() { + t.Fatalf("transient 429 must not write Quota.NextRecoverAt, got %v", state.Quota.NextRecoverAt) + } + if got := state.NextRetryAfter.Sub(before); got < transientRateLimitMinimum { + t.Fatalf("expected NextRetryAfter at least the transient floor %v, got %v", transientRateLimitMinimum, got) + } + + reg.ResumeClientModel(authID, model) + if count := reg.GetModelCount(model); count != 1 { + t.Fatalf("SetModelQuotaExceeded recorded a transient 429: GetModelCount=%d, want 1", count) + } +} + +func TestMarkResultQuotaAfterTransientUsesLadderFloor(t *testing.T) { + withQuotaCooldownEnabled(t) + + manager := NewManager(nil, nil, nil) + auth := &Auth{ + ID: "auth-quota-after-transient", + Provider: "codex", + Metadata: map[string]any{"type": "codex"}, + ModelStates: map[string]*ModelState{ + "gpt-5": {Status: StatusActive, Quota: QuotaState{BackoffLevel: 0}}, + }, + } + if _, errRegister := manager.Register(WithSkipPersist(context.Background()), auth); errRegister != nil { + t.Fatalf("Register returned error: %v", errRegister) + } + + hint := observedExhaustedQuotaHint + transient := quotaResult(auth.ID, "gpt-5") + transient.RetryAfter = &hint + transient.TransientRateLimit = true + manager.MarkResult(context.Background(), transient) + + updated, ok := manager.GetByID(auth.ID) + if !ok || updated == nil || updated.ModelStates["gpt-5"] == nil { + t.Fatalf("expected model state after transient failure") + } + transientDeadline := updated.ModelStates["gpt-5"].NextRetryAfter + if !transientDeadline.After(time.Now()) { + t.Fatal("expected an unexpired transient NextRetryAfter before the quota 429") + } + + inside := time.Now() + manager.MarkResult(context.Background(), quotaResult(auth.ID, "gpt-5")) + + updated, ok = manager.GetByID(auth.ID) + if !ok || updated == nil || updated.ModelStates["gpt-5"] == nil { + t.Fatalf("expected model state after quota failure") + } + state := updated.ModelStates["gpt-5"] + if !state.Quota.Exceeded || state.Quota.Reason != "quota" { + t.Fatalf("expected the quota 429 to record exhausted quota, got exceeded=%v reason=%q", state.Quota.Exceeded, state.Quota.Reason) + } + if state.Quota.BackoffLevel != 1 { + t.Fatalf("quota path reused the transient window: BackoffLevel=%d, want 1", state.Quota.BackoffLevel) + } + got := state.Quota.NextRecoverAt.Sub(inside) + if got < quotaBackoffBase-50*time.Millisecond { + t.Fatalf("expected at least the first ladder step (%v), got %v", quotaBackoffBase, got) + } + if got >= transientRateLimitMinimum-time.Second { + t.Fatalf("quota path reused the transient window: quota window %v, transient floor %v", got, transientRateLimitMinimum) + } + if state.NextRetryAfter.Before(transientDeadline) { + t.Fatalf("quota path dropped the unexpired transient NextRetryAfter: got %v, want at least %v", state.NextRetryAfter, transientDeadline) + } +} + +func TestApplyAuthFailureStateTransientRateLimitDoesNotSetQuotaExceeded(t *testing.T) { + now := time.Now() + rateLimitErr := &Error{Code: "rate_limit", Message: "RATE_LIMIT_EXCEEDED", HTTPStatus: http.StatusTooManyRequests} + hint := observedExhaustedQuotaHint + auth := &Auth{ID: "auth-transient-no-quota"} + applyAuthFailureState(auth, rateLimitErr, &hint, now, false, true) + + if auth.Quota.Exceeded { + t.Fatal("transient 429 must not set Quota.Exceeded") + } + if auth.Quota.Reason == "quota" { + t.Fatalf("transient 429 must not set Reason %q", auth.Quota.Reason) + } + if !auth.Quota.NextRecoverAt.IsZero() { + t.Fatalf("transient 429 must not write Quota.NextRecoverAt, got %v", auth.Quota.NextRecoverAt) + } + if !auth.NextRetryAfter.Equal(now.Add(transientRateLimitMinimum)) { + t.Fatalf("expected NextRetryAfter at the transient floor %v, got %v", now.Add(transientRateLimitMinimum), auth.NextRetryAfter) + } +} + +func TestApplyAuthFailureStateQuotaAfterTransientUsesLadderFloor(t *testing.T) { + now := time.Now() + rateLimitErr := &Error{Code: "rate_limit", Message: "RATE_LIMIT_EXCEEDED", HTTPStatus: http.StatusTooManyRequests} + hint := observedExhaustedQuotaHint + auth := &Auth{ID: "auth-quota-after-transient-level"} + applyAuthFailureState(auth, rateLimitErr, &hint, now, false, true) + if auth.Quota.Exceeded { + t.Fatal("transient 429 must not set Quota.Exceeded") + } + transientDeadline := auth.NextRetryAfter + + inside := now.Add(time.Second) + applyAuthFailureState(auth, rateLimitErr, nil, inside, false, false) + if !auth.Quota.Exceeded || auth.Quota.Reason != "quota" { + t.Fatalf("expected the quota 429 to record exhausted quota, got exceeded=%v reason=%q", auth.Quota.Exceeded, auth.Quota.Reason) + } + if auth.Quota.BackoffLevel != 1 { + t.Fatalf("quota path reused the transient window: BackoffLevel=%d, want 1", auth.Quota.BackoffLevel) + } + if want := inside.Add(quotaBackoffBase); !auth.Quota.NextRecoverAt.Equal(want) { + t.Fatalf("expected quota ladder floor at %v, got %v", want, auth.Quota.NextRecoverAt) + } + if !auth.NextRetryAfter.Equal(transientDeadline) { + t.Fatalf("quota path dropped the unexpired transient NextRetryAfter: got %v, want %v", auth.NextRetryAfter, transientDeadline) + } +}