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 3b72135bec..543c999359 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 a3a992175c..07b32497bd 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 a494be8adf..cfd0649154 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 142d2e3198..746473ac29 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 6b9c6f19cb..7e3dc38020 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 fbf7892026..ef5c73e129 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) {