diff --git a/backend/internal/handler/failover_loop.go b/backend/internal/handler/failover_loop.go index 669cfcf365..d349980fc6 100644 --- a/backend/internal/handler/failover_loop.go +++ b/backend/internal/handler/failover_loop.go @@ -76,6 +76,12 @@ func sameAccountRetryDelayFor(failoverErr *service.UpstreamFailoverError, retryC return delay } +// sameAccountRetryDeadlineAllows prevents a retry from starting after the +// service-provided same-account retry window has elapsed. +func sameAccountRetryDeadlineAllows(failoverErr *service.UpstreamFailoverError) bool { + return failoverErr == nil || failoverErr.SameAccountRetryDeadline.IsZero() || time.Now().Before(failoverErr.SameAccountRetryDeadline) +} + // FailoverState 跨循环迭代共享的 failover 状态 type FailoverState struct { SwitchCount int diff --git a/backend/internal/handler/failover_loop_test.go b/backend/internal/handler/failover_loop_test.go index e65b43c172..40602f9df9 100644 --- a/backend/internal/handler/failover_loop_test.go +++ b/backend/internal/handler/failover_loop_test.go @@ -60,6 +60,16 @@ func TestSameAccountRetryDelayFor(t *testing.T) { }) } +func TestSameAccountRetryDeadlineAllows(t *testing.T) { + require.True(t, sameAccountRetryDeadlineAllows(&service.UpstreamFailoverError{})) + require.True(t, sameAccountRetryDeadlineAllows(&service.UpstreamFailoverError{ + SameAccountRetryDeadline: time.Now().Add(time.Second), + })) + require.False(t, sameAccountRetryDeadlineAllows(&service.UpstreamFailoverError{ + SameAccountRetryDeadline: time.Now().Add(-time.Second), + })) +} + // --------------------------------------------------------------------------- // Helper // --------------------------------------------------------------------------- diff --git a/backend/internal/handler/grok_media.go b/backend/internal/handler/grok_media.go index 3cef97ed3f..6e5619b6e5 100644 --- a/backend/internal/handler/grok_media.go +++ b/backend/internal/handler/grok_media.go @@ -362,18 +362,20 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service. } if failoverErr.RetryableOnSameAccount { retryLimit := account.GetPoolModeRetryCount() - if sameAccountRetryCount[account.ID] < retryLimit { + if sameAccountRetryCount[account.ID] < retryLimit && sameAccountRetryDeadlineAllows(failoverErr) { sameAccountRetryCount[account.ID]++ + retryDelay := sameAccountRetryDelayFor(failoverErr, sameAccountRetryCount[account.ID]) reqLog.Warn("grok_media.pool_mode_same_account_retry", zap.Int64("account_id", account.ID), zap.Int("upstream_status", failoverErr.StatusCode), zap.Int("retry_limit", retryLimit), zap.Int("retry_count", sameAccountRetryCount[account.ID]), + zap.Duration("retry_delay", retryDelay), ) select { case <-requestCtx.Done(): return - case <-time.After(sameAccountRetryDelay): + case <-time.After(retryDelay): } continue } diff --git a/backend/internal/handler/openai_chat_completions.go b/backend/internal/handler/openai_chat_completions.go index 79709cabb7..1dd8f63ceb 100644 --- a/backend/internal/handler/openai_chat_completions.go +++ b/backend/internal/handler/openai_chat_completions.go @@ -328,7 +328,7 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) { // Pool mode: retry on the same account if failoverErr.RetryableOnSameAccount { retryLimit := account.GetPoolModeRetryCount() - if sameAccountRetryCount[account.ID] < retryLimit { + if sameAccountRetryCount[account.ID] < retryLimit && sameAccountRetryDeadlineAllows(failoverErr) { sameAccountRetryCount[account.ID]++ retryDelay := sameAccountRetryDelayFor(failoverErr, sameAccountRetryCount[account.ID]) reqLog.Warn("openai_chat_completions.pool_mode_same_account_retry", diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index aebb68f543..1f7c849a6a 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -718,7 +718,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { // 池模式：同账号重试 if failoverErr.RetryableOnSameAccount { retryLimit := account.GetPoolModeRetryCount() - if sameAccountRetryCount[account.ID] < retryLimit { + if sameAccountRetryCount[account.ID] < retryLimit && sameAccountRetryDeadlineAllows(failoverErr) { sameAccountRetryCount[account.ID]++ retryDelay := sameAccountRetryDelayFor(failoverErr, sameAccountRetryCount[account.ID]) reqLog.Warn("openai.pool_mode_same_account_retry", @@ -1267,7 +1267,7 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) { // 池模式：同账号重试 if failoverErr.RetryableOnSameAccount { retryLimit := account.GetPoolModeRetryCount() - if sameAccountRetryCount[account.ID] < retryLimit { + if sameAccountRetryCount[account.ID] < retryLimit && sameAccountRetryDeadlineAllows(failoverErr) { sameAccountRetryCount[account.ID]++ retryDelay := sameAccountRetryDelayFor(failoverErr, sameAccountRetryCount[account.ID]) reqLog.Warn("openai_messages.pool_mode_same_account_retry", diff --git a/backend/internal/handler/openai_images.go b/backend/internal/handler/openai_images.go index da41bf6aa3..a82c7b7193 100644 --- a/backend/internal/handler/openai_images.go +++ b/backend/internal/handler/openai_images.go @@ -305,18 +305,20 @@ func (h *OpenAIGatewayHandler) Images(c *gin.Context) { } if failoverErr.RetryableOnSameAccount { retryLimit := account.GetPoolModeRetryCount() - if sameAccountRetryCount[account.ID] < retryLimit { + if sameAccountRetryCount[account.ID] < retryLimit && sameAccountRetryDeadlineAllows(failoverErr) { sameAccountRetryCount[account.ID]++ + retryDelay := sameAccountRetryDelayFor(failoverErr, sameAccountRetryCount[account.ID]) reqLog.Warn("openai.images.pool_mode_same_account_retry", zap.Int64("account_id", account.ID), zap.Int("upstream_status", failoverErr.StatusCode), zap.Int("retry_limit", retryLimit), zap.Int("retry_count", sameAccountRetryCount[account.ID]), + zap.Duration("retry_delay", retryDelay), ) select { case <-requestCtx.Done(): return - case <-time.After(sameAccountRetryDelay): + case <-time.After(retryDelay): } continue } diff --git a/backend/internal/repository/http_upstream.go b/backend/internal/repository/http_upstream.go index 43be1bc995..bdd6478cff 100644 --- a/backend/internal/repository/http_upstream.go +++ b/backend/internal/repository/http_upstream.go @@ -99,6 +99,7 @@ const ( upstreamProtocolModeOpenAIH1 = "openai_h1" upstreamProtocolModeOpenAIH2 = "openai_h2" upstreamProtocolModeOpenAIH1Fallback = "openai_h1_fallback" + upstreamProtocolModeGrok = "grok" ) var errUpstreamClientLimitReached = errors.New("upstream client cache limit reached") @@ -910,7 +911,7 @@ func (s *httpUpstreamService) applyProfilePoolSettings(settings poolSettings, pr // generic 600s gateway timeout from turning one request into a 10-minute // resource hold; streaming after headers is unaffected. settings.responseHeaderTimeout = 120 * time.Second - if s != nil && s.cfg != nil && s.cfg.Gateway.GrokResponseHeaderTimeout > 0 { + if s != nil && s.cfg != nil { settings.responseHeaderTimeout = time.Duration(s.cfg.Gateway.GrokResponseHeaderTimeout) * time.Second } } @@ -991,6 +992,9 @@ func (s *httpUpstreamService) resolveOpenAIHTTP2Settings() openAIHTTP2Settings { } func (s *httpUpstreamService) resolveProtocolMode(profile service.HTTPUpstreamProfile, proxyKey string, parsedProxy *url.URL) string { + if profile == service.HTTPUpstreamProfileGrok { + return upstreamProtocolModeGrok + } if profile != service.HTTPUpstreamProfileOpenAI { return upstreamProtocolModeDefault } diff --git a/backend/internal/service/grok_stream_idle.go b/backend/internal/service/grok_stream_idle.go index 3bdfa27fc9..d984ae1335 100644 --- a/backend/internal/service/grok_stream_idle.go +++ b/backend/internal/service/grok_stream_idle.go @@ -32,7 +32,11 @@ func grokStreamIdleFailoverError(account *Account, idle time.Duration) *Upstream StatusCode: 502, ResponseBody: []byte(`{"error":{"code":"empty_upstream","message":"` + strings.ReplaceAll(msg, `"`, `'`) + `"}}`), SafeToFailoverAfterWrite: true, - // Allow pool-mode retries; normal OAuth switches account via handler. - RetryableOnSameAccount: account != nil && account.IsPoolMode(), + // An idle upstream stream is transient and should get the configured + // same-account retry budget before switching credentials. This applies + // to both pooled and dedicated Grok accounts; the handler still enforces + // the request's retry limit. + RetryableOnSameAccount: account != nil && account.Platform == PlatformGrok, + RequestScopedTransient: true, } } diff --git a/backend/internal/service/grok_stream_idle_test.go b/backend/internal/service/grok_stream_idle_test.go index e3c6d16943..c8ee0443b6 100644 --- a/backend/internal/service/grok_stream_idle_test.go +++ b/backend/internal/service/grok_stream_idle_test.go @@ -21,5 +21,14 @@ func TestGrokStreamIdleFailoverError(t *testing.T) { require.NotNil(t, err) require.Equal(t, 502, err.StatusCode) require.True(t, err.SafeToFailoverAfterWrite) + require.True(t, err.RetryableOnSameAccount) + require.True(t, err.RequestScopedTransient) require.Contains(t, string(err.ResponseBody), "empty_upstream") } + +func TestGrokStreamIdleFailoverErrorRequiresGrokAccount(t *testing.T) { + openAI := &Account{ID: 2, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + err := grokStreamIdleFailoverError(openAI, time.Second) + require.False(t, err.RetryableOnSameAccount) + require.True(t, err.RequestScopedTransient) +} diff --git a/backend/internal/service/grok_upstream_failure.go b/backend/internal/service/grok_upstream_failure.go index 475c6cff16..b84413dd2b 100644 --- a/backend/internal/service/grok_upstream_failure.go +++ b/backend/internal/service/grok_upstream_failure.go @@ -417,6 +417,19 @@ func grokSameAccountRetryMetadata(account *Account, statusCode int, responseBody return true, 500 * time.Millisecond, time.Now().Add(30 * time.Second) } +// shouldMarkGrokTeamModelRateLimit controls the process-local sibling-account +// overlay. Model-capacity responses are request pressure, not a team quota; +// marking them would hide healthy sibling credentials while the bounded +// same-account retry is still in progress. Ordinary 429s and free-usage +// exhaustion retain the existing quota/team isolation behavior. +func shouldMarkGrokTeamModelRateLimit(statusCode int, responseBody []byte) bool { + decision := classifyGrokUpstreamFailure(statusCode, responseBody, "") + if decision.Class == GrokFailureModelCapacity { + return false + } + return statusCode == http.StatusTooManyRequests || decision.Class == GrokFailureFreeUsage +} + func isGrokCompatibilityError(low, code string) bool { combined := strings.ToLower(strings.TrimSpace(low + " " + code)) // Compaction blobs are account/session-bound and frequently fail with 400 @@ -427,11 +440,6 @@ func isGrokCompatibilityError(low, code string) bool { "decode the compaction blob", "ensure it is unmodified from the compact response", "compaction blob", - "failed to deserialize the json body", - "data did not match any variant", - "untagged enum content", - "invalid content shape", - "invalid response history", } { if strings.Contains(combined, phrase) { return true @@ -440,8 +448,6 @@ func isGrokCompatibilityError(low, code string) bool { for _, marker := range []string{ "invalid_compaction", "compaction_decode_error", - "response_history_incompatible", - "invalid_content_shape", } { if strings.Contains(combined, marker) { return true diff --git a/backend/internal/service/grok_upstream_failure_test.go b/backend/internal/service/grok_upstream_failure_test.go index 358c1595d3..2424eb8dff 100644 --- a/backend/internal/service/grok_upstream_failure_test.go +++ b/backend/internal/service/grok_upstream_failure_test.go @@ -77,6 +77,11 @@ func TestGrokRetryableOnSameAccount_CapacityAndRateLimit(t *testing.T) { []byte(`{"error":{"message":"rate limit exceeded"}}`))) require.False(t, grokRetryableOnSameAccount(account, http.StatusPaymentRequired, []byte(`{"error":{"message":"You have run out of credits or need a Grok subscription"}}`))) + poolAccount := &Account{ID: 9108, Platform: PlatformGrok, Type: AccountTypeOAuth, + Credentials: map[string]any{"pool_mode": true}} + require.False(t, grokRetryableOnSameAccount(poolAccount, http.StatusTooManyRequests, + []byte(`{"error":{"code":"subscription:free-usage-exhausted"}}`)), + "pool free-usage must fail over instead of retrying the exhausted account") require.False(t, grokRetryableOnSameAccount(account, http.StatusBadRequest, []byte(`{"error":{"message":"capacity field is invalid"}}`))) nonGrok := &Account{ID: 9106, Platform: PlatformOpenAI, Type: AccountTypeOAuth} @@ -84,6 +89,17 @@ func TestGrokRetryableOnSameAccount_CapacityAndRateLimit(t *testing.T) { []byte(`{"error":{"message":"model at capacity"}}`))) } +func TestShouldMarkGrokTeamModelRateLimit_ExcludesCapacity(t *testing.T) { + require.False(t, shouldMarkGrokTeamModelRateLimit(http.StatusTooManyRequests, + []byte(`{"error":{"message":"The model is currently at capacity due to high demand"}}`))) + require.True(t, shouldMarkGrokTeamModelRateLimit(http.StatusTooManyRequests, + []byte(`{"error":{"message":"rate limit exceeded"}}`))) + require.True(t, shouldMarkGrokTeamModelRateLimit(http.StatusBadRequest, + []byte(`{"error":{"code":"subscription:free-usage-exhausted"}}`))) + require.False(t, shouldMarkGrokTeamModelRateLimit(http.StatusBadRequest, + []byte(`{"error":{"message":"invalid request"}}`))) +} + func TestGrokSameAccountRetryMetadata_CapacityDeadline(t *testing.T) { account := &Account{ID: 9107, Platform: PlatformGrok, Type: AccountTypeOAuth} retryable, delay, deadline := grokSameAccountRetryMetadata(account, http.StatusTooManyRequests, @@ -116,7 +132,6 @@ func TestClassifyGrokUpstreamFailure_FreeUsageWinsOver5xx(t *testing.T) { func TestClassifyGrokUpstreamFailure_CompatibilityDoesNotCooldown(t *testing.T) { cases := []string{ `{"error":{"message":"Could not decode the compaction blob. Ensure it is unmodified from the compact response"}}`, - `{"error":{"message":"data did not match any variant of the untagged enum content"}}`, `{"code":"compaction_decode_error","message":"invalid response history"}`, } for _, body := range cases { @@ -128,6 +143,13 @@ func TestClassifyGrokUpstreamFailure_CompatibilityDoesNotCooldown(t *testing.T) } } +func TestClassifyGrokUpstreamFailure_GenericShapeErrorDoesNotFailover(t *testing.T) { + d := classifyGrokUpstreamFailure(http.StatusBadRequest, + []byte(`{"error":{"message":"data did not match any variant of the untagged enum content"}}`), "grok-4.6") + require.NotEqual(t, GrokFailureCompatibility, d.Class) + require.False(t, d.ShouldFailover) +} + func TestShouldFailoverGrokUpstreamError_FreeUsageBody(t *testing.T) { svc := &OpenAIGatewayService{} body := []byte(`{"error":{"code":"subscription:free-usage-exhausted","message":"free usage exhausted"}}`) diff --git a/backend/internal/service/openai_gateway_grok.go b/backend/internal/service/openai_gateway_grok.go index 9cb3b01246..589c2844b2 100644 --- a/backend/internal/service/openai_gateway_grok.go +++ b/backend/internal/service/openai_gateway_grok.go @@ -167,9 +167,9 @@ func (s *OpenAIGatewayService) forwardGrokResponses( }) errCtx := withGrokTeamRateLimitModel(ctx, upstreamModel) s.handleGrokAccountUpstreamError(errCtx, account, resp.StatusCode, resp.Header, respBody) - // 429 / free-usage: stamp team+model cool so sibling accounts skip this model. - if resp.StatusCode == http.StatusTooManyRequests || - classifyGrokUpstreamFailure(resp.StatusCode, respBody, upstreamModel).Class == GrokFailureFreeUsage { + // Quota/rate-limit responses stamp the team+model overlay. Capacity is + // request pressure and must not hide sibling accounts. + if shouldMarkGrokTeamModelRateLimit(resp.StatusCode, respBody) { markGrokTeamModelRateLimit(account, upstreamModel, resolveGrokTeamRateLimitUntil(time.Now().Add(grokTeamRateLimitDefaultTTL), time.Now())) } if s.shouldFailoverGrokUpstreamError(resp.StatusCode, respBody) { @@ -179,6 +179,7 @@ func (s *OpenAIGatewayService) forwardGrokResponses( ResponseBody: respBody, ResponseHeaders: resp.Header.Clone(), RetryableOnSameAccount: retryable, + RequestScopedTransient: retryable && resp.StatusCode == http.StatusTooManyRequests, SameAccountRetryDelay: retryDelay, SameAccountRetryDeadline: retryDeadline, } @@ -649,7 +650,7 @@ func grokSupportsXHighReasoningEffort(model string) bool { func grokSupportsReasoningEffort(model string) bool { model = strings.ToLower(xai.StripGrokProviderPrefix(strings.TrimSpace(model))) switch model { - case xai.DefaultTextModel, "grok-4.5-latest", "grok-4.6", "grok-4.6-latest", + case xai.DefaultTextModel, "grok-4.5-latest", "grok-4.6-latest", "grok-4.3", "grok-4.3-latest", "grok-3-mini", "grok-3-mini-fast", "grok-4.20-0309-reasoning", "grok-4.20-reasoning", "grok-4.20-multi-agent-0309": @@ -1189,6 +1190,7 @@ func (s *OpenAIGatewayService) describeGrokComposerImage( ResponseBody: respBody, ResponseHeaders: resp.Header.Clone(), RetryableOnSameAccount: retryable, + RequestScopedTransient: retryable && resp.StatusCode == http.StatusTooManyRequests, SameAccountRetryDelay: retryDelay, SameAccountRetryDeadline: retryDeadline, } @@ -1744,7 +1746,7 @@ func persistGrokTransientModelCooldown(account *Account, decision GrokUpstreamFa return false } model := strings.TrimSpace(decision.Model) - if model == "" || !isGrokHeavyTransientModel(model) { + if model == "" { return false } cooldown := decision.Cooldown diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 5577e90e4b..5558093d11 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -1182,8 +1182,26 @@ func openAIUsageFromGJSON(value gjson.Result) (OpenAIUsage, bool) { inputTokens = value.Get("prompt_tokens").Int() } outputTokens := value.Get("output_tokens").Int() + outputTokensFromResponses := outputTokens != 0 if outputTokens == 0 { outputTokens = value.Get("completion_tokens").Int() + outputTokensFromResponses = false + } + // xAI Chat Completions reports visible completion tokens separately from + // reasoning_tokens. Billing must include both. Responses usage generally + // already folds reasoning into output_tokens; use total_tokens when present + // to avoid double-counting that canonical shape. + reasoningTokens := max(int(firstPositiveGJSONInt( + value.Get("completion_tokens_details.reasoning_tokens"), + value.Get("output_tokens_details.reasoning_tokens"), + )), 0) + if reasoningTokens > 0 { + totalTokens := value.Get("total_tokens").Int() + if !outputTokensFromResponses { + outputTokens += int64(reasoningTokens) + } else if totalTokens > inputTokens && totalTokens >= outputTokens+int64(reasoningTokens) { + outputTokens += int64(reasoningTokens) + } } cacheReadTokens := openAICacheReadTokensFromUsage(value) cacheCreationTokens := openAICacheCreationTokensFromUsage(value) diff --git a/backend/internal/service/openai_gateway_service_test.go b/backend/internal/service/openai_gateway_service_test.go index 1d18d3d70c..f3385b3c4e 100644 --- a/backend/internal/service/openai_gateway_service_test.go +++ b/backend/internal/service/openai_gateway_service_test.go @@ -3453,6 +3453,20 @@ func TestExtractOpenAIUsageFromJSONBytes_AcceptsResponseAndChatUsageShapes(t *te require.Zero(t, usage.CacheReadInputTokens, "官方嵌套缓存读取字段显式为零时仍应优先于兼容顶层别名") } +func TestExtractOpenAIUsageFromJSONBytes_IncludesGrokReasoningTokens(t *testing.T) { + usage, ok := extractOpenAIUsageFromJSONBytes([]byte(`{"usage":{"prompt_tokens":32,"completion_tokens":9,"total_tokens":135,"completion_tokens_details":{"reasoning_tokens":94}}}`)) + require.True(t, ok) + require.Equal(t, 103, usage.OutputTokens, "Grok Chat usage bills visible completion plus reasoning tokens") + + usage, ok = extractOpenAIUsageFromJSONBytes([]byte(`{"usage":{"input_tokens":32,"output_tokens":103,"total_tokens":135,"output_tokens_details":{"reasoning_tokens":94}}}`)) + require.True(t, ok) + require.Equal(t, 103, usage.OutputTokens, "Responses output_tokens already includes reasoning when total confirms it") + + usage, ok = extractOpenAIUsageFromJSONBytes([]byte(`{"usage":{"input_tokens":32,"output_tokens":9,"total_tokens":135,"output_tokens_details":{"reasoning_tokens":94}}}`)) + require.True(t, ok) + require.Equal(t, 103, usage.OutputTokens, "Responses detail-only shape is normalized when total exposes the full output") +} + func TestExtractCodexFinalResponse_SampleReplay(t *testing.T) { body := strings.Join([]string{ `event: message`, diff --git a/deploy/config.example.yaml b/deploy/config.example.yaml index 8f134dd609..38a50d0607 100644 --- a/deploy/config.example.yaml +++ b/deploy/config.example.yaml @@ -214,6 +214,9 @@ gateway: # OpenAI/Codex upstream response header timeout (seconds, 0=disabled) # OpenAI/Codex 等待上游响应头超时时间（秒，0=禁用本地响应头超时） openai_response_header_timeout: 0 + # Grok text Responses first-byte timeout (seconds, 0=disabled; default 120) + # Grok 文本 Responses 首字节超时时间（秒，0=禁用；默认 120） + grok_response_header_timeout: 120 # Native OpenAI HTTP Responses first semantic output timeout (seconds, 0=disabled) # Includes response-header wait; does not apply to passthrough or WebSocket transports. # A timed-out request may already have incurred upstream usage; account failover can therefore duplicate upstream billing.