From 5d9c7abed59a5a53e36cd6cd62257807a3099ab0 Mon Sep 17 00:00:00 2001 From: shaw Date: Sat, 29 Aug 2026 09:45:49 +0800 Subject: [PATCH 1/4] fix(openai): scope Spark quota 429 to model --- backend/internal/service/model_rate_limit.go | 1 + .../openai_account_runtime_block_fastpath.go | 4 ++ ...nai_account_runtime_block_fastpath_test.go | 67 ++++++++++++++++++- .../service/openai_gateway_passthrough.go | 11 ++- .../openai_gateway_response_handling.go | 4 +- .../service/openai_ws_forwarder_ingress.go | 4 +- .../service/openai_ws_forwarder_support.go | 15 +++-- .../service/openai_ws_forwarder_v2.go | 4 +- .../openai_ws_v2_passthrough_adapter.go | 4 +- backend/internal/service/ratelimit_service.go | 32 +++++++++ 10 files changed, 128 insertions(+), 18 deletions(-) diff --git a/backend/internal/service/model_rate_limit.go b/backend/internal/service/model_rate_limit.go index fe962b17c0ab..2ab17923b034 100644 --- a/backend/internal/service/model_rate_limit.go +++ b/backend/internal/service/model_rate_limit.go @@ -12,6 +12,7 @@ const ( modelRateLimitsKey = "model_rate_limits" antigravityGeminiModelRateLimitKey = "antigravity:gemini" openAIImageGenerationRateLimitKey = "openai:image_generation" + openAICodexSparkRateLimitReason = "openai_codex_spark_rate_limit" // anthropicFableRateLimitKey 是 Anthropic 7d_oi（Fable 专属 7d 窗口）限流的 // 家族级 scope：命中后所有 Fable 变体（含 [1m] 等后缀）都不再调度到该账号。 anthropicFableRateLimitKey = "claude-fable-5" diff --git a/backend/internal/service/openai_account_runtime_block_fastpath.go b/backend/internal/service/openai_account_runtime_block_fastpath.go index 7b59c28fd47c..3551d8ecb27d 100644 --- a/backend/internal/service/openai_account_runtime_block_fastpath.go +++ b/backend/internal/service/openai_account_runtime_block_fastpath.go @@ -162,6 +162,10 @@ func (s *OpenAIGatewayService) handleOpenAIAccountUpstreamError(ctx context.Cont s.rateLimitService.HandleTempUnschedulable(stateCtx, account, statusCode, responseBody, canonicalModel[0]) { return true } + if statusCode == http.StatusTooManyRequests && s.rateLimitService != nil && len(canonicalModel) > 0 && + s.rateLimitService.HandleOpenAICodexSparkRateLimit(stateCtx, account, canonicalModel[0], statusCode, headers, responseBody) { + return false + } if statusCode == http.StatusTooManyRequests { s.markOpenAIOAuth429RateLimited(stateCtx, account, headers, responseBody) } diff --git a/backend/internal/service/openai_account_runtime_block_fastpath_test.go b/backend/internal/service/openai_account_runtime_block_fastpath_test.go index 89d8c5b5ec44..9a6475f23e0d 100644 --- a/backend/internal/service/openai_account_runtime_block_fastpath_test.go +++ b/backend/internal/service/openai_account_runtime_block_fastpath_test.go @@ -15,8 +15,11 @@ import ( type oauth429RateLimitRepo struct { mockAccountRepoForGemini - setRateLimitedCalls int - lastRateLimitedUntil time.Time + setRateLimitedCalls int + lastRateLimitedUntil time.Time + setModelRateLimitCalls int + lastModelRateLimitKey string + lastModelRateLimitedUntil time.Time } func (r *oauth429RateLimitRepo) SetRateLimited(_ context.Context, _ int64, until time.Time) error { @@ -25,6 +28,13 @@ func (r *oauth429RateLimitRepo) SetRateLimited(_ context.Context, _ int64, until return nil } +func (r *oauth429RateLimitRepo) SetModelRateLimit(_ context.Context, _ int64, scope string, until time.Time, _ ...string) error { + r.setModelRateLimitCalls++ + r.lastModelRateLimitKey = scope + r.lastModelRateLimitedUntil = until + return nil +} + func TestOpenAI429FastPath_KeepsOAuthAccountSchedulableDuringRetryWindow(t *testing.T) { repo := &oauth429RateLimitRepo{} rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) @@ -85,6 +95,59 @@ func TestOpenAI429FastPath_BlocksOAuthImmediatelyWhenSevenDayQuotaIsExhausted(t require.False(t, svc.ShouldRetryOpenAIOAuth429(account, headers, nil)) } +func TestOpenAI429FastPath_SparkQuotaOnlyBlocksSparkModel(t *testing.T) { + repo := &oauth429RateLimitRepo{} + rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + svc := &OpenAIGatewayService{rateLimitService: rateLimits} + rateLimits.SetAccountRuntimeBlocker(svc) + account := &Account{ID: 425, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + headers := http.Header{} + headers.Set("x-codex-primary-used-percent", "100") + headers.Set("x-codex-primary-reset-after-seconds", "604800") + headers.Set("x-codex-primary-window-minutes", "10080") + headers.Set("x-codex-secondary-used-percent", "20") + headers.Set("x-codex-secondary-reset-after-seconds", "3600") + headers.Set("x-codex-secondary-window-minutes", "300") + + shouldDisable := svc.handleOpenAIAccountUpstreamError( + context.Background(), account, http.StatusTooManyRequests, headers, + []byte(`{"error":{"type":"rate_limit_error","code":"rate_limit_exceeded"}}`), + "gpt-5.3-codex-spark", + ) + + require.False(t, shouldDisable) + require.False(t, svc.isOpenAIAccountRuntimeBlocked(account), "Spark quota must not create an account runtime block") + require.Equal(t, 0, repo.setRateLimitedCalls, "Spark quota must not persist account-level rate limit") + require.Equal(t, 1, repo.setModelRateLimitCalls) + require.Equal(t, "gpt-5.3-codex-spark", repo.lastModelRateLimitKey) + require.Greater(t, time.Until(repo.lastModelRateLimitedUntil), 6*24*time.Hour) +} + +func TestOpenAI429FastPath_SparkShadowQuotaStaysModelScoped(t *testing.T) { + repo := &oauth429RateLimitRepo{} + rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + svc := &OpenAIGatewayService{rateLimitService: rateLimits} + rateLimits.SetAccountRuntimeBlocker(svc) + parentID := int64(426) + shadow := &Account{ID: 427, Platform: PlatformOpenAI, Type: AccountTypeOAuth, ParentAccountID: &parentID, QuotaDimension: QuotaDimensionSpark} + headers := http.Header{} + headers.Set("x-codex-primary-used-percent", "100") + headers.Set("x-codex-primary-reset-after-seconds", "604800") + headers.Set("x-codex-primary-window-minutes", "10080") + + shouldDisable := svc.handleOpenAIAccountUpstreamError( + context.Background(), shadow, http.StatusTooManyRequests, headers, + []byte(`{"error":{"type":"rate_limit_error","code":"rate_limit_exceeded"}}`), + "gpt-5.3-codex-spark", + ) + + require.False(t, shouldDisable) + require.False(t, svc.isOpenAIAccountRuntimeBlocked(shadow)) + require.Equal(t, 0, repo.setRateLimitedCalls) + require.Equal(t, 1, repo.setModelRateLimitCalls) + require.Equal(t, "gpt-5.3-codex-spark", repo.lastModelRateLimitKey) +} + func TestOpenAI429FastPath_RetriesOAuthWhenNoQuotaSignalExists(t *testing.T) { svc := &OpenAIGatewayService{} account := &Account{ID: 424, Platform: PlatformOpenAI, Type: AccountTypeOAuth} diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 0ccb9adcaa67..aaa74e4a7f86 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -1558,6 +1558,7 @@ func (s *OpenAIGatewayService) handleOpenAIStreamTerminalAccountSideEffects( payload []byte, message string, headers http.Header, + canonicalModel ...string, ) (int, bool) { statusCode := openAIStreamFailureStatus(payload, message) switch statusCode { @@ -1578,7 +1579,11 @@ func (s *OpenAIGatewayService) handleOpenAIStreamTerminalAccountSideEffects( // carried by a stream terminal event. accountHeaders = nil } - return statusCode, s.handleOpenAIAccountUpstreamError(ctx, account, statusCode, accountHeaders, payload) + model := firstNonEmpty(canonicalModel...) + if model == "" { + model = firstNonEmpty(gjson.GetBytes(payload, "model").String(), gjson.GetBytes(payload, "response.model").String()) + } + return statusCode, s.handleOpenAIAccountUpstreamError(ctx, account, statusCode, accountHeaders, payload, model) default: return statusCode, false } @@ -1873,7 +1878,7 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( return } if bareErrorAccountSideEffectsPending { - s.handleOpenAIStreamTerminalAccountSideEffects(c, account, bareErrorPayload, failedMessage, resp.Header) + s.handleOpenAIStreamTerminalAccountSideEffects(c, account, bareErrorPayload, failedMessage, resp.Header, mappedModel) bareErrorAccountSideEffectsPending = false } if clientDisconnected || !writePendingLines() { @@ -2005,7 +2010,7 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( // account health; EOF synthesis applies the pending effect. bareErrorAccountSideEffectsPending = true } else { - s.handleOpenAIStreamTerminalAccountSideEffects(c, account, dataBytes, failedMessage, resp.Header) + s.handleOpenAIStreamTerminalAccountSideEffects(c, account, dataBytes, failedMessage, resp.Header, mappedModel) bareErrorAccountSideEffectsPending = false } } diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 30986f3e2132..6b9c6f19cb49 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -372,7 +372,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. completeGuardedEvent(true) } if codexFailureTerminal && sawBareError && !sawResponseFailed && bareErrorAccountSideEffectsPending { - s.handleOpenAIStreamTerminalAccountSideEffects(c, account, bareErrorPayload, failedMessage, resp.Header) + s.handleOpenAIStreamTerminalAccountSideEffects(c, account, bareErrorPayload, failedMessage, resp.Header, mappedModel) bareErrorAccountSideEffectsPending = false } if codexFailureTerminal && sawBareError && !sawResponseFailed && !clientDisconnected { @@ -557,7 +557,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. // Defer account health updates so the pair is applied once. bareErrorAccountSideEffectsPending = true } else { - s.handleOpenAIStreamTerminalAccountSideEffects(c, account, dataBytes, failedMessage, resp.Header) + s.handleOpenAIStreamTerminalAccountSideEffects(c, account, dataBytes, failedMessage, resp.Header, mappedModel) bareErrorAccountSideEffectsPending = false } } diff --git a/backend/internal/service/openai_ws_forwarder_ingress.go b/backend/internal/service/openai_ws_forwarder_ingress.go index aff0a5d06775..09e16dc70480 100644 --- a/backend/internal/service/openai_ws_forwarder_ingress.go +++ b/backend/internal/service/openai_ws_forwarder_ingress.go @@ -866,7 +866,7 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( ) var dialErr *openAIWSDialError if errors.As(acquireErr, &dialErr) && dialErr != nil && dialErr.StatusCode == http.StatusTooManyRequests { - s.persistOpenAIWSRateLimitSignal(ctx, account, dialErr.ResponseHeaders, nil, "rate_limit_exceeded", "rate_limit_error", strings.TrimSpace(acquireErr.Error())) + s.persistOpenAIWSRateLimitSignal(ctx, account, dialErr.ResponseHeaders, nil, "rate_limit_exceeded", "rate_limit_error", strings.TrimSpace(acquireErr.Error()), canonicalModel) return nil, s.newOpenAIWSRateLimitFailoverError(account, dialErr.ResponseHeaders, nil, acquireErr.Error()) } if errors.Is(acquireErr, errOpenAIWSPreferredConnUnavailable) { @@ -1019,7 +1019,7 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( } } } - s.persistOpenAIWSRateLimitSignal(ctx, account, lease.HandshakeHeaders(), upstreamMessage, errCodeRaw, errTypeRaw, errMsgRaw) + s.persistOpenAIWSRateLimitSignal(ctx, account, lease.HandshakeHeaders(), upstreamMessage, errCodeRaw, errTypeRaw, errMsgRaw, mappedModel) fallbackReason, _ := classifyOpenAIWSErrorEventFromRaw(errCodeRaw, errTypeRaw, errMsgRaw) errCode, errType, errMessage := summarizeOpenAIWSErrorEventFieldsFromRaw(errCodeRaw, errTypeRaw, errMsgRaw) recoverablePrevNotFound := fallbackReason == openAIWSIngressStagePreviousResponseNotFound && diff --git a/backend/internal/service/openai_ws_forwarder_support.go b/backend/internal/service/openai_ws_forwarder_support.go index 20ea5eb6b62c..c5d00c8072b0 100644 --- a/backend/internal/service/openai_ws_forwarder_support.go +++ b/backend/internal/service/openai_ws_forwarder_support.go @@ -131,7 +131,8 @@ func (s *OpenAIGatewayService) performOpenAIWSGeneratePrewarm( if eventType == "error" { errCodeRaw, errTypeRaw, errMsgRaw := parseOpenAIWSErrorEventFields(message) - s.persistOpenAIWSRateLimitSignal(ctx, account, lease.HandshakeHeaders(), message, errCodeRaw, errTypeRaw, errMsgRaw) + prewarmModel, _ := reqBody["model"].(string) + s.persistOpenAIWSRateLimitSignal(ctx, account, lease.HandshakeHeaders(), message, errCodeRaw, errTypeRaw, errMsgRaw, prewarmModel) errMsg := strings.TrimSpace(errMsgRaw) if errMsg == "" { errMsg = "OpenAI websocket prewarm error" @@ -336,13 +337,13 @@ func (s *OpenAIGatewayService) handleOpenAIWSFailureAccountSideEffects(ctx conte status := openAIStreamFailureStatus(payload, message) switch status { case http.StatusUnauthorized, http.StatusTooManyRequests, 529: - s.handleOpenAIStreamTerminalAccountSideEffects(nil, account, payload, message, headers) + s.handleOpenAIStreamTerminalAccountSideEffects(nil, account, payload, message, headers, canonicalModel) return true case http.StatusForbidden: if !openAIStream403AccountFailure(payload, message) { return false } - s.handleOpenAIStreamTerminalAccountSideEffects(nil, account, payload, message, headers) + s.handleOpenAIStreamTerminalAccountSideEffects(nil, account, payload, message, headers, canonicalModel) return true } @@ -683,14 +684,18 @@ func isOpenAIWSRateLimitError(codeRaw, errTypeRaw, msgRaw string) bool { return false } -func (s *OpenAIGatewayService) persistOpenAIWSRateLimitSignal(ctx context.Context, account *Account, headers http.Header, responseBody []byte, codeRaw, errTypeRaw, msgRaw string) { +func (s *OpenAIGatewayService) persistOpenAIWSRateLimitSignal(ctx context.Context, account *Account, headers http.Header, responseBody []byte, codeRaw, errTypeRaw, msgRaw string, canonicalModel ...string) { if s == nil || s.rateLimitService == nil || account == nil || account.Platform != PlatformOpenAI { return } if !isOpenAIWSRateLimitError(codeRaw, errTypeRaw, msgRaw) { return } - s.handleOpenAIAccountUpstreamError(ctx, account, http.StatusTooManyRequests, headers, responseBody) + model := firstNonEmpty(canonicalModel...) + if model == "" { + model = firstNonEmpty(gjson.GetBytes(responseBody, "model").String(), gjson.GetBytes(responseBody, "response.model").String()) + } + s.handleOpenAIAccountUpstreamError(ctx, account, http.StatusTooManyRequests, headers, responseBody, model) } func (s *OpenAIGatewayService) newOpenAIWSRateLimitFailoverError(account *Account, headers http.Header, responseBody []byte, message string) *UpstreamFailoverError { diff --git a/backend/internal/service/openai_ws_forwarder_v2.go b/backend/internal/service/openai_ws_forwarder_v2.go index f977edc558eb..81f8631e96ce 100644 --- a/backend/internal/service/openai_ws_forwarder_v2.go +++ b/backend/internal/service/openai_ws_forwarder_v2.go @@ -245,7 +245,7 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( ) var dialErr *openAIWSDialError if errors.As(err, &dialErr) && dialErr != nil && dialErr.StatusCode == http.StatusTooManyRequests { - s.persistOpenAIWSRateLimitSignal(ctx, account, dialErr.ResponseHeaders, nil, "rate_limit_exceeded", "rate_limit_error", strings.TrimSpace(err.Error())) + s.persistOpenAIWSRateLimitSignal(ctx, account, dialErr.ResponseHeaders, nil, "rate_limit_exceeded", "rate_limit_error", strings.TrimSpace(err.Error()), mappedModel) } return nil, wrapOpenAIWSFallback(classifyOpenAIWSAcquireError(err), err) } @@ -601,7 +601,7 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( if eventType == "error" { s.handleOpenAIWSErrorEventTransientFailure(ctx, account, mappedModel, lease.HandshakeHeaders(), message) errCodeRaw, errTypeRaw, errMsgRaw := parseOpenAIWSErrorEventFields(message) - s.persistOpenAIWSRateLimitSignal(ctx, account, lease.HandshakeHeaders(), message, errCodeRaw, errTypeRaw, errMsgRaw) + s.persistOpenAIWSRateLimitSignal(ctx, account, lease.HandshakeHeaders(), message, errCodeRaw, errTypeRaw, errMsgRaw, mappedModel) errMsg := strings.TrimSpace(errMsgRaw) if errMsg == "" { errMsg = "Upstream websocket error" diff --git a/backend/internal/service/openai_ws_v2_passthrough_adapter.go b/backend/internal/service/openai_ws_v2_passthrough_adapter.go index 1e26d0155f74..6bb1be715426 100644 --- a/backend/internal/service/openai_ws_v2_passthrough_adapter.go +++ b/backend/internal/service/openai_ws_v2_passthrough_adapter.go @@ -902,7 +902,7 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( ) s.handleOpenAIWSDialTransientFailure(ctx, account, capturedSessionModel, dialErr) if statusCode == http.StatusTooManyRequests { - s.persistOpenAIWSRateLimitSignal(ctx, account, handshakeHeaders, nil, "rate_limit_exceeded", "rate_limit_error", strings.TrimSpace(err.Error())) + s.persistOpenAIWSRateLimitSignal(ctx, account, handshakeHeaders, nil, "rate_limit_exceeded", "rate_limit_error", strings.TrimSpace(err.Error()), capturedSessionModel) return s.newOpenAIWSRateLimitFailoverError(account, handshakeHeaders, nil, err.Error()) } return s.mapOpenAIWSPassthroughDialError(err, statusCode, handshakeHeaders) @@ -1291,7 +1291,7 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( if wroteDownstream || !isOpenAIWSRateLimitError(errCodeRaw, errTypeRaw, errMsgRaw) { return nil } - s.persistOpenAIWSRateLimitSignal(ctx, account, handshakeHeaders, payload, errCodeRaw, errTypeRaw, errMsgRaw) + s.persistOpenAIWSRateLimitSignal(ctx, account, handshakeHeaders, payload, errCodeRaw, errTypeRaw, errMsgRaw, capturedSessionModel) logOpenAIWSV2Passthrough( "relay_rate_limit_failover account_id=%d err_code=%s err_type=%s err_message=%s", account.ID, diff --git a/backend/internal/service/ratelimit_service.go b/backend/internal/service/ratelimit_service.go index 4d6be56b06fa..e0263e8f3eb5 100644 --- a/backend/internal/service/ratelimit_service.go +++ b/backend/internal/service/ratelimit_service.go @@ -2192,6 +2192,38 @@ func (s *RateLimitService) HandleOpenAIImageRateLimit(ctx context.Context, accou return true } +// HandleOpenAICodexSparkRateLimit 将 Spark 独立配额窗口记录为模型级限流。 +// Spark 的 x-codex-* 使用率和 reset 时间只代表 Spark 模型维度，不能写入账号级 +// RateLimitResetAt，否则同一 OAuth 账号上的其他模型也会被错误停调。 +func (s *RateLimitService) HandleOpenAICodexSparkRateLimit(ctx context.Context, account *Account, requestedModel string, statusCode int, headers http.Header, responseBody []byte) bool { + if s == nil || account == nil || s.accountRepo == nil || statusCode != http.StatusTooManyRequests || !isOpenAIOAuthAccount(account) { + return false + } + if !isCodexSparkModel(requestedModel) || !account.ShouldHandleErrorCode(statusCode) { + return false + } + + modelKey := normalizeCodexModel(modelRateLimitKeyForUpstreamModelNotFound(ctx, account, requestedModel)) + if modelKey == "" { + return false + } + now := time.Now() + _, resetAt := classifyOpenAIOAuth429(headers, responseBody) + if resetAt == nil || !resetAt.After(now) { + cooldown, ok := s.get429FallbackCooldown(ctx, account) + if !ok || cooldown <= 0 { + cooldown = time.Duration(defaultRateLimit429CooldownSeconds) * time.Second + } + reset := now.Add(cooldown) + resetAt = &reset + } + if err := s.accountRepo.SetModelRateLimit(ctx, account.ID, modelKey, *resetAt, openAICodexSparkRateLimitReason); err != nil { + slog.Warn("openai_codex_spark_model_rate_limit_set_failed", "account_id", account.ID, "model", modelKey, "error", err) + } + slog.Info("openai_codex_spark_model_rate_limited", "account_id", account.ID, "model", modelKey, "reset_at", *resetAt) + return true +} + func (s *RateLimitService) HandleOpenAIImageCapabilityLoss(ctx context.Context, account *Account, statusCode int, responseBody []byte) bool { if s == nil || account == nil || s.accountRepo == nil { return false From 3c22e78af084235aac2983925bf115eca464392f Mon Sep 17 00:00:00 2001 From: shaw Date: Sat, 29 Aug 2026 11:08:01 +0800 Subject: [PATCH 2/4] fix(openai): preserve Spark quota reset semantics --- ...nai_account_runtime_block_fastpath_test.go | 53 +++++++++++++++++++ .../service/openai_gateway_passthrough.go | 13 +++-- backend/internal/service/ratelimit_service.go | 7 ++- 3 files changed, 65 insertions(+), 8 deletions(-) diff --git a/backend/internal/service/openai_account_runtime_block_fastpath_test.go b/backend/internal/service/openai_account_runtime_block_fastpath_test.go index 9a6475f23e0d..e013ba8831b4 100644 --- a/backend/internal/service/openai_account_runtime_block_fastpath_test.go +++ b/backend/internal/service/openai_account_runtime_block_fastpath_test.go @@ -123,6 +123,59 @@ func TestOpenAI429FastPath_SparkQuotaOnlyBlocksSparkModel(t *testing.T) { require.Greater(t, time.Until(repo.lastModelRateLimitedUntil), 6*24*time.Hour) } +func TestOpenAI429FastPath_SparkTransient429UsesShortFallback(t *testing.T) { + repo := &oauth429RateLimitRepo{} + rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + svc := &OpenAIGatewayService{rateLimitService: rateLimits} + rateLimits.SetAccountRuntimeBlocker(svc) + account := &Account{ID: 428, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + headers := http.Header{} + headers.Set("x-codex-primary-used-percent", "37") + headers.Set("x-codex-primary-reset-after-seconds", "604800") + headers.Set("x-codex-primary-window-minutes", "10080") + headers.Set("x-codex-secondary-used-percent", "20") + headers.Set("x-codex-secondary-reset-after-seconds", "3600") + headers.Set("x-codex-secondary-window-minutes", "300") + + shouldDisable := svc.handleOpenAIAccountUpstreamError( + context.Background(), account, http.StatusTooManyRequests, headers, + []byte(`{"error":{"type":"rate_limit_error","code":"rate_limit_exceeded"}}`), + "gpt-5.3-codex-spark", + ) + + require.False(t, shouldDisable) + require.Equal(t, 1, repo.setModelRateLimitCalls) + require.Less(t, time.Until(repo.lastModelRateLimitedUntil), time.Minute) + require.Greater(t, time.Until(repo.lastModelRateLimitedUntil), time.Second) +} + +func TestOpenAIStream429_SparkQuotaUsesQuotaHeaders(t *testing.T) { + repo := &oauth429RateLimitRepo{} + rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + svc := &OpenAIGatewayService{rateLimitService: rateLimits} + rateLimits.SetAccountRuntimeBlocker(svc) + account := &Account{ID: 429, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + headers := http.Header{} + headers.Set("x-codex-primary-used-percent", "100") + headers.Set("x-codex-primary-reset-after-seconds", "604800") + headers.Set("x-codex-primary-window-minutes", "10080") + headers.Set("x-codex-secondary-used-percent", "20") + headers.Set("x-codex-secondary-reset-after-seconds", "3600") + headers.Set("x-codex-secondary-window-minutes", "300") + payload := []byte(`{"type":"error","error":{"type":"rate_limit_error","code":"rate_limit_exceeded"}}`) + + status, shouldDisable := svc.handleOpenAIStreamTerminalAccountSideEffects( + nil, account, payload, "quota exhausted", headers, "gpt-5.3-codex-spark", + ) + + require.Equal(t, http.StatusTooManyRequests, status) + require.False(t, shouldDisable) + require.Equal(t, 1, repo.setModelRateLimitCalls) + require.Equal(t, "gpt-5.3-codex-spark", repo.lastModelRateLimitKey) + require.Greater(t, time.Until(repo.lastModelRateLimitedUntil), 6*24*time.Hour) + require.False(t, svc.isOpenAIAccountRuntimeBlocked(account)) +} + func TestOpenAI429FastPath_SparkShadowQuotaStaysModelScoped(t *testing.T) { repo := &oauth429RateLimitRepo{} rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index aaa74e4a7f86..65df226dd953 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -1572,17 +1572,16 @@ func (s *OpenAIGatewayService) handleOpenAIStreamTerminalAccountSideEffects( if c != nil && c.Request != nil { ctx = c.Request.Context() } - accountHeaders := headers - if statusCode == http.StatusTooManyRequests { - // The enclosing HTTP response succeeded. Its quota snapshot describes - // normal account state and must not become the reset for a semantic 429 - // carried by a stream terminal event. - accountHeaders = nil - } model := firstNonEmpty(canonicalModel...) if model == "" { model = firstNonEmpty(gjson.GetBytes(payload, "model").String(), gjson.GetBytes(payload, "response.model").String()) } + accountHeaders := headers + if statusCode == http.StatusTooManyRequests && !(isCodexSparkModel(model) && isOpenAIOAuthAccount(account)) { + // 普通模型的流式 429 不能继承外层 HTTP 200 的全局 quota 快照； + // 只有 OAuth/SetupToken 的 Spark 配额 429 才需要保留 headers 读取明确的 5h/7d reset。 + accountHeaders = nil + } return statusCode, s.handleOpenAIAccountUpstreamError(ctx, account, statusCode, accountHeaders, payload, model) default: return statusCode, false diff --git a/backend/internal/service/ratelimit_service.go b/backend/internal/service/ratelimit_service.go index e0263e8f3eb5..ec3c134e5a4f 100644 --- a/backend/internal/service/ratelimit_service.go +++ b/backend/internal/service/ratelimit_service.go @@ -2208,7 +2208,12 @@ func (s *RateLimitService) HandleOpenAICodexSparkRateLimit(ctx context.Context, return false } now := time.Now() - _, resetAt := classifyOpenAIOAuth429(headers, responseBody) + disposition, resetAt := classifyOpenAIOAuth429(headers, responseBody) + // Spark 只有明确耗尽 5h/7d 窗口时才能使用上游长 reset；普通瞬时 429 + // 即使携带全局 reset 头，也只能使用短时回避，避免错误冷却数天。 + if disposition != openAIOAuth429Quota5h && disposition != openAIOAuth429Quota7d { + resetAt = nil + } if resetAt == nil || !resetAt.After(now) { cooldown, ok := s.get429FallbackCooldown(ctx, account) if !ok || cooldown <= 0 { From 571d1e1d9be93fea4e2d840f0e7d9570a442b263 Mon Sep 17 00:00:00 2001 From: shaw Date: Sat, 29 Aug 2026 11:14:54 +0800 Subject: [PATCH 3/4] fix(openai): isolate websocket semantic rate limits --- ...nai_account_runtime_block_fastpath_test.go | 38 +++++++++++++++++++ .../service/openai_gateway_passthrough.go | 4 +- .../service/openai_ws_forwarder_support.go | 16 ++++++++ 3 files changed, 56 insertions(+), 2 deletions(-) diff --git a/backend/internal/service/openai_account_runtime_block_fastpath_test.go b/backend/internal/service/openai_account_runtime_block_fastpath_test.go index e013ba8831b4..3b72135bec06 100644 --- a/backend/internal/service/openai_account_runtime_block_fastpath_test.go +++ b/backend/internal/service/openai_account_runtime_block_fastpath_test.go @@ -176,6 +176,44 @@ func TestOpenAIStream429_SparkQuotaUsesQuotaHeaders(t *testing.T) { require.False(t, svc.isOpenAIAccountRuntimeBlocked(account)) } +func TestOpenAIWSErrorEvent_OrdinaryModelIgnoresHandshakeQuotaHeaders(t *testing.T) { + repo := &oauth429RateLimitRepo{} + rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + svc := &OpenAIGatewayService{rateLimitService: rateLimits} + rateLimits.SetAccountRuntimeBlocker(svc) + account := &Account{ID: 430, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + headers := http.Header{} + headers.Set("x-codex-primary-used-percent", "100") + headers.Set("x-codex-primary-reset-after-seconds", "604800") + headers.Set("x-codex-primary-window-minutes", "10080") + payload := []byte(`{"type":"error","error":{"type":"rate_limit_error","code":"rate_limit_exceeded"}}`) + + svc.persistOpenAIWSRateLimitSignal(context.Background(), account, headers, payload, "rate_limit_exceeded", "rate_limit_error", "quota exhausted", "gpt-5.3-codex") + + require.False(t, svc.isOpenAIAccountRuntimeBlocked(account)) + require.Zero(t, repo.setRateLimitedCalls) +} + +func TestOpenAIWSErrorEvent_SparkQuotaUsesHandshakeQuotaHeaders(t *testing.T) { + repo := &oauth429RateLimitRepo{} + rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + svc := &OpenAIGatewayService{rateLimitService: rateLimits} + rateLimits.SetAccountRuntimeBlocker(svc) + account := &Account{ID: 431, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + headers := http.Header{} + headers.Set("x-codex-primary-used-percent", "100") + headers.Set("x-codex-primary-reset-after-seconds", "604800") + headers.Set("x-codex-primary-window-minutes", "10080") + payload := []byte(`{"type":"error","error":{"type":"rate_limit_error","code":"rate_limit_exceeded"}}`) + + svc.persistOpenAIWSRateLimitSignal(context.Background(), account, headers, payload, "rate_limit_exceeded", "rate_limit_error", "quota exhausted", "gpt-5.3-codex-spark") + + require.Equal(t, 1, repo.setModelRateLimitCalls) + require.Equal(t, "gpt-5.3-codex-spark", repo.lastModelRateLimitKey) + require.Greater(t, time.Until(repo.lastModelRateLimitedUntil), 6*24*time.Hour) + require.False(t, svc.isOpenAIAccountRuntimeBlocked(account)) +} + func TestOpenAI429FastPath_SparkShadowQuotaStaysModelScoped(t *testing.T) { repo := &oauth429RateLimitRepo{} rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 65df226dd953..142d2e31980a 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -1577,10 +1577,10 @@ func (s *OpenAIGatewayService) handleOpenAIStreamTerminalAccountSideEffects( model = firstNonEmpty(gjson.GetBytes(payload, "model").String(), gjson.GetBytes(payload, "response.model").String()) } accountHeaders := headers - if statusCode == http.StatusTooManyRequests && !(isCodexSparkModel(model) && isOpenAIOAuthAccount(account)) { + if statusCode == http.StatusTooManyRequests { // 普通模型的流式 429 不能继承外层 HTTP 200 的全局 quota 快照； // 只有 OAuth/SetupToken 的 Spark 配额 429 才需要保留 headers 读取明确的 5h/7d reset。 - accountHeaders = nil + accountHeaders = openAIWSSemantic429Headers(account, model, headers) } return statusCode, s.handleOpenAIAccountUpstreamError(ctx, account, statusCode, accountHeaders, payload, model) default: diff --git a/backend/internal/service/openai_ws_forwarder_support.go b/backend/internal/service/openai_ws_forwarder_support.go index c5d00c8072b0..02332f18849f 100644 --- a/backend/internal/service/openai_ws_forwarder_support.go +++ b/backend/internal/service/openai_ws_forwarder_support.go @@ -325,6 +325,9 @@ func (s *OpenAIGatewayService) handleOpenAIWSErrorEventTransientFailure(ctx cont } status := openAIWSPayloadTransientStatus(payload) if status != 0 { + if status == http.StatusTooManyRequests { + headers = openAIWSSemantic429Headers(account, canonicalModel, headers) + } s.handleOpenAIAccountUpstreamError(ctx, account, status, headers, payload, canonicalModel) } } @@ -695,9 +698,22 @@ func (s *OpenAIGatewayService) persistOpenAIWSRateLimitSignal(ctx context.Contex if model == "" { model = firstNonEmpty(gjson.GetBytes(responseBody, "model").String(), gjson.GetBytes(responseBody, "response.model").String()) } + // 非空 responseBody 表示已建立连接后收到的语义错误事件；握手响应头 + // 可能只是成功连接时的全局快照，不能用于普通模型的 429 账号级限流。 + // 实际拨号 HTTP 429 使用 nil responseBody，必须保留响应头。 + if len(responseBody) > 0 { + headers = openAIWSSemantic429Headers(account, model, headers) + } s.handleOpenAIAccountUpstreamError(ctx, account, http.StatusTooManyRequests, headers, responseBody, model) } +func openAIWSSemantic429Headers(account *Account, model string, headers http.Header) http.Header { + if isCodexSparkModel(model) && isOpenAIOAuthAccount(account) { + return headers + } + return nil +} + func (s *OpenAIGatewayService) newOpenAIWSRateLimitFailoverError(account *Account, headers http.Header, responseBody []byte, message string) *UpstreamFailoverError { return s.newOpenAIAccountFailoverError( account, From 804679d9967877418e10b72ca00ca6627626c4a0 Mon Sep 17 00:00:00 2001 From: shaw Date: Sat, 29 Aug 2026 15:25:10 +0800 Subject: [PATCH 4/4] fix(openai): preserve model scope on stream failover --- ...nai_account_runtime_block_fastpath_test.go | 26 +++++++++++++++++++ .../openai_gateway_chat_completions.go | 4 +-- .../service/openai_gateway_messages.go | 4 +-- .../service/openai_gateway_passthrough.go | 22 +++++++++++++--- .../openai_gateway_response_handling.go | 4 +-- .../internal/service/openai_ws_http_bridge.go | 2 +- 6 files changed, 51 insertions(+), 11 deletions(-) diff --git a/backend/internal/service/openai_account_runtime_block_fastpath_test.go b/backend/internal/service/openai_account_runtime_block_fastpath_test.go index 3b72135bec06..543c999359b8 100644 --- a/backend/internal/service/openai_account_runtime_block_fastpath_test.go +++ b/backend/internal/service/openai_account_runtime_block_fastpath_test.go @@ -176,6 +176,32 @@ func TestOpenAIStream429_SparkQuotaUsesQuotaHeaders(t *testing.T) { require.False(t, svc.isOpenAIAccountRuntimeBlocked(account)) } +func TestOpenAIStreamFailover_Spark429KeepsModelScope(t *testing.T) { + repo := &oauth429RateLimitRepo{} + rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + svc := &OpenAIGatewayService{rateLimitService: rateLimits} + rateLimits.SetAccountRuntimeBlocker(svc) + account := &Account{ID: 432, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + headers := http.Header{} + headers.Set("x-codex-primary-used-percent", "100") + headers.Set("x-codex-primary-reset-after-seconds", "604800") + headers.Set("x-codex-primary-window-minutes", "10080") + headers.Set("x-codex-secondary-used-percent", "20") + headers.Set("x-codex-secondary-reset-after-seconds", "3600") + headers.Set("x-codex-secondary-window-minutes", "300") + payload := []byte(`{"type":"error","error":{"type":"rate_limit_error","code":"rate_limit_exceeded"}}`) + + failoverErr := svc.newOpenAIStreamFailoverErrorWithModel( + nil, account, false, "", payload, "quota exhausted", "gpt-5.3-codex-spark", headers, + ) + + require.Equal(t, http.StatusTooManyRequests, failoverErr.StatusCode) + require.Equal(t, 0, repo.setRateLimitedCalls) + require.Equal(t, 1, repo.setModelRateLimitCalls) + require.Equal(t, "gpt-5.3-codex-spark", repo.lastModelRateLimitKey) + require.False(t, svc.isOpenAIAccountRuntimeBlocked(account)) +} + func TestOpenAIWSErrorEvent_OrdinaryModelIgnoresHandshakeQuotaHeaders(t *testing.T) { repo := &oauth429RateLimitRepo{} rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index a3a992175cef..07b32497bd3e 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -511,7 +511,7 @@ func (s *OpenAIGatewayService) handleChatBufferedStreamingResponse( } message := openAICompatFailedResponseMessage(finalResponse) if openAIStreamFailedEventShouldFailover(payload, message) { - return nil, s.newOpenAIStreamFailoverError(c, account, false, requestID, payload, message, resp.Header) + return nil, s.newOpenAIStreamFailoverErrorWithModel(c, account, false, requestID, payload, message, upstreamModel, resp.Header) } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payload, message) // response.failed 到达在 HTTP 200 SSE 流上，无真实 HTTP 错误码；统一走语义 @@ -766,7 +766,7 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse( shouldFailover = openAIStreamErrorEventShouldFailover(payloadBytes, message) } if !clientOutputStarted && shouldFailover { - streamFailoverErr = s.newOpenAIStreamFailoverError(c, account, false, requestID, payloadBytes, message, resp.Header) + streamFailoverErr = s.newOpenAIStreamFailoverErrorWithModel(c, account, false, requestID, payloadBytes, message, upstreamModel, resp.Header) return true } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payloadBytes, message) diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index a494be8adf9d..cfd064915475 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -605,7 +605,7 @@ func (s *OpenAIGatewayService) handleAnthropicBufferedStreamingResponse( } message := openAICompatFailedResponseMessage(finalResponse) if openAIStreamFailedEventShouldFailover(payload, message) { - return nil, s.newOpenAIStreamFailoverError(c, account, false, requestID, payload, message, resp.Header) + return nil, s.newOpenAIStreamFailoverErrorWithModel(c, account, false, requestID, payload, message, upstreamModel, resp.Header) } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payload, message) // 统一走语义状态推断 + body 归一化（与 /v1/responses 路径一致）， @@ -1038,7 +1038,7 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( shouldFailover = openAIStreamErrorEventShouldFailover(payloadBytes, message) } if !clientOutputStarted && shouldFailover { - streamFailoverErr = s.newOpenAIStreamFailoverError(c, account, false, requestID, payloadBytes, message, resp.Header) + streamFailoverErr = s.newOpenAIStreamFailoverErrorWithModel(c, account, false, requestID, payloadBytes, message, upstreamModel, resp.Header) return true } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payloadBytes, message) diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 142d2e31980a..746473ac2949 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -1658,6 +1658,19 @@ func (s *OpenAIGatewayService) newOpenAIStreamFailoverError( payload []byte, message string, responseHeaders ...http.Header, +) *UpstreamFailoverError { + return s.newOpenAIStreamFailoverErrorWithModel(c, account, passthrough, upstreamRequestID, payload, message, "", responseHeaders...) +} + +func (s *OpenAIGatewayService) newOpenAIStreamFailoverErrorWithModel( + c *gin.Context, + account *Account, + passthrough bool, + upstreamRequestID string, + payload []byte, + message string, + canonicalModel string, + responseHeaders ...http.Header, ) *UpstreamFailoverError { message = sanitizeUpstreamErrorMessage(strings.TrimSpace(message)) if message == "" { @@ -1667,7 +1680,7 @@ func (s *OpenAIGatewayService) newOpenAIStreamFailoverError( if len(responseHeaders) > 0 && responseHeaders[0] != nil { headers = responseHeaders[0].Clone() } - statusCode, shouldDisable := s.handleOpenAIStreamTerminalAccountSideEffects(c, account, payload, message, headers) + statusCode, shouldDisable := s.handleOpenAIStreamTerminalAccountSideEffects(c, account, payload, message, headers, canonicalModel) // 流内 failed 事件承载于 HTTP 200；使用事件的语义状态更新账号健康， // 再由 failover 引擎按 StatusCode/RetryableOnSameAccount 决定恢复策略。 message = s.recordOpenAIStreamUpstreamError(c, account, passthrough, upstreamRequestID, "failover", payload, message) @@ -1738,6 +1751,7 @@ func (s *OpenAIGatewayService) nonStreamingTerminalFailureFailover( terminalType string, payload []byte, message string, + canonicalModel ...string, ) *UpstreamFailoverError { if account == nil || IsResponseCommitted(c) { return nil @@ -1755,7 +1769,7 @@ func (s *OpenAIGatewayService) nonStreamingTerminalFailureFailover( headers = resp.Header upstreamRequestID = strings.TrimSpace(resp.Header.Get("x-request-id")) } - return s.newOpenAIStreamFailoverError(c, account, passthrough, upstreamRequestID, payload, message, headers) + return s.newOpenAIStreamFailoverErrorWithModel(c, account, passthrough, upstreamRequestID, payload, message, firstNonEmpty(canonicalModel...), headers) } func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( @@ -2024,7 +2038,7 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( } if shouldFailover { return resultWithUsage(), - s.newOpenAIStreamFailoverError(c, account, true, upstreamRequestID, dataBytes, failedMessage, resp.Header) + s.newOpenAIStreamFailoverErrorWithModel(c, account, true, upstreamRequestID, dataBytes, failedMessage, mappedModel, resp.Header) } if !cyberHit && !sawBareError { if status, errType, errMsg, matched := applyOpenAIStreamFailedErrorPassthroughRule(c, account.Platform, dataBytes, failedMessage); matched { @@ -2277,7 +2291,7 @@ func (s *OpenAIGatewayService) handlePassthroughSSEToJSON(resp *http.Response, c if compactErr := newOpenAICompactFallbackSignal(c, terminalPayload, msg); compactErr != nil { return nil, compactErr } - if failoverErr := s.nonStreamingTerminalFailureFailover(c, resp, account, true, terminalType, terminalPayload, msg); failoverErr != nil { + if failoverErr := s.nonStreamingTerminalFailureFailover(c, resp, account, true, terminalType, terminalPayload, msg, mappedModel); failoverErr != nil { return nil, failoverErr } return nil, s.writeOpenAINonStreamingProtocolError(resp, c, msg) diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 6b9c6f19cb49..7e3dc380207d 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -572,7 +572,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. } if shouldFailover { sawFailedEvent = true - streamEarlyErr = s.newOpenAIStreamFailoverError(c, account, false, upstreamRequestID, dataBytes, failedMessage, resp.Header) + streamEarlyErr = s.newOpenAIStreamFailoverErrorWithModel(c, account, false, upstreamRequestID, dataBytes, failedMessage, mappedModel, resp.Header) return } if !cyberHit && !sawBareError { @@ -1684,7 +1684,7 @@ func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Conte if compactErr := newOpenAICompactFallbackSignal(c, terminalPayload, msg); compactErr != nil { return nil, compactErr } - if failoverErr := s.nonStreamingTerminalFailureFailover(c, resp, account, false, terminalType, terminalPayload, msg); failoverErr != nil { + if failoverErr := s.nonStreamingTerminalFailureFailover(c, resp, account, false, terminalType, terminalPayload, msg, mappedModel); failoverErr != nil { return nil, failoverErr } return nil, s.writeOpenAINonStreamingProtocolError(resp, c, msg) diff --git a/backend/internal/service/openai_ws_http_bridge.go b/backend/internal/service/openai_ws_http_bridge.go index fbf789202689..ef5c73e129a8 100644 --- a/backend/internal/service/openai_ws_http_bridge.go +++ b/backend/internal/service/openai_ws_http_bridge.go @@ -695,7 +695,7 @@ func (s *OpenAIGatewayService) proxyOpenAIWSHTTPBridgeTurn( if account.Platform == PlatformGrok { return nil, newOpenAIUpstreamFailoverError(statusCode, resp.Header, upstreamMessage, errMessage, false) } - return nil, s.newOpenAIStreamFailoverError(c, account, true, resp.Header.Get("x-request-id"), upstreamMessage, errMessage, resp.Header) + return nil, s.newOpenAIStreamFailoverErrorWithModel(c, account, true, resp.Header.Get("x-request-id"), upstreamMessage, errMessage, mappedModel, resp.Header) } if account.Platform != PlatformGrok && !failureAccountSideEffectsApplied { if eventType == "response.failed" || (!officialOpenAIResponses && shouldFailover && !requestScopedCapacity) {