From ec9e7336060bfef2df6f4dd80625560ec1f4cdbd Mon Sep 17 00:00:00 2001 From: IanShaw027 Date: Fri, 7 Aug 2026 13:56:38 +0800 Subject: [PATCH] =?UTF-8?q?feat(grok):=20stream=20idle=20=E6=8D=A2?= =?UTF-8?q?=E5=8F=B7=E3=80=81team+model=20=E5=86=B7=E5=8D=B4=E4=B8=8E=20RT?= =?UTF-8?q?=20=E5=88=B7=E6=96=B0=E6=89=93=E6=95=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Grok 流默认启用上游读空闲超时，首包前返回可 failover 错误并短冷却 - 同 team_id 账号在 429/free-usage 后对同模型共享进程内冷却，粘连与选号均过滤 - TokenRefreshService 已覆盖 Grok；NeedsRefresh 增加按账号稳定 jitter 防 stampede --- backend/internal/service/grok_stream_idle.go | 38 +++++ .../internal/service/grok_stream_idle_test.go | 25 +++ .../internal/service/grok_team_rate_limit.go | 148 ++++++++++++++++++ .../service/grok_team_rate_limit_test.go | 59 +++++++ .../internal/service/grok_token_refresher.go | 39 +++++ .../service/grok_token_refresher_test.go | 50 ++++++ .../service/openai_account_scheduler.go | 15 ++ .../internal/service/openai_gateway_grok.go | 36 ++++- .../openai_gateway_response_handling.go | 20 +++ 9 files changed, 427 insertions(+), 3 deletions(-) create mode 100644 backend/internal/service/grok_stream_idle.go create mode 100644 backend/internal/service/grok_stream_idle_test.go create mode 100644 backend/internal/service/grok_team_rate_limit.go create mode 100644 backend/internal/service/grok_team_rate_limit_test.go create mode 100644 backend/internal/service/grok_token_refresher_test.go diff --git a/backend/internal/service/grok_stream_idle.go b/backend/internal/service/grok_stream_idle.go new file mode 100644 index 0000000000..3bdfa27fc9 --- /dev/null +++ b/backend/internal/service/grok_stream_idle.go @@ -0,0 +1,38 @@ +package service + +import ( + "fmt" + "strings" + "time" +) + +// Default Grok stream idle when gateway.stream_data_interval_timeout is 0. +// Long enough for slow thinking models, short enough to release hung sockets. +const defaultGrokStreamIdleTimeout = 180 * time.Second + +// Shorter cool after a Grok stream-idle failure so the account can re-enter soon +// but is not immediately re-picked in a tight failover loop. +const grokStreamIdleCooldown = 2 * time.Minute + +// resolveGrokStreamIdleTimeout returns the effective upstream-read idle timeout +// for Grok streams. Prefers the global gateway setting when positive; otherwise +// applies a Grok-only default so hung SSE bodies still fail over. +func resolveGrokStreamIdleTimeout(cfgStreamIntervalSec int) time.Duration { + if cfgStreamIntervalSec > 0 { + return time.Duration(cfgStreamIntervalSec) * time.Second + } + return defaultGrokStreamIdleTimeout +} + +// grokStreamIdleFailoverError builds a pre-commit/handler-visible failover so +// the gateway can switch OAuth accounts after a hung Grok upstream stream. +func grokStreamIdleFailoverError(account *Account, idle time.Duration) *UpstreamFailoverError { + msg := fmt.Sprintf("Grok stream idle timeout after %s with no upstream data", idle.Round(time.Second)) + return &UpstreamFailoverError{ + 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(), + } +} diff --git a/backend/internal/service/grok_stream_idle_test.go b/backend/internal/service/grok_stream_idle_test.go new file mode 100644 index 0000000000..e3c6d16943 --- /dev/null +++ b/backend/internal/service/grok_stream_idle_test.go @@ -0,0 +1,25 @@ +//go:build unit + +package service + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestResolveGrokStreamIdleTimeout(t *testing.T) { + require.Equal(t, 90*time.Second, resolveGrokStreamIdleTimeout(90)) + require.Equal(t, defaultGrokStreamIdleTimeout, resolveGrokStreamIdleTimeout(0)) + require.Equal(t, defaultGrokStreamIdleTimeout, resolveGrokStreamIdleTimeout(-1)) +} + +func TestGrokStreamIdleFailoverError(t *testing.T) { + account := &Account{ID: 1, Platform: PlatformGrok, Type: AccountTypeOAuth} + err := grokStreamIdleFailoverError(account, 180*time.Second) + require.NotNil(t, err) + require.Equal(t, 502, err.StatusCode) + require.True(t, err.SafeToFailoverAfterWrite) + require.Contains(t, string(err.ResponseBody), "empty_upstream") +} diff --git a/backend/internal/service/grok_team_rate_limit.go b/backend/internal/service/grok_team_rate_limit.go new file mode 100644 index 0000000000..c4d3bae425 --- /dev/null +++ b/backend/internal/service/grok_team_rate_limit.go @@ -0,0 +1,148 @@ +package service + +import ( + "crypto/sha256" + "encoding/hex" + "strings" + "sync" + "time" +) + +// In-memory team+model rate-limit overlay for Grok OAuth. When xAI rate-limits +// one account in a team for a model, sibling accounts sharing team_id skip the +// same model until the cooldown expires (mirrors grok2api teamModelRateLimit). +// +// Process-local only: multi-instance deployments each learn the block from their +// own 429s. Prefer short TTLs so drift self-heals. +type grokTeamModelRateLimit struct { + Until time.Time +} + +type grokTeamModelRateLimitStore struct { + mu sync.Mutex + items map[string]grokTeamModelRateLimit +} + +var globalGrokTeamModelRateLimits = &grokTeamModelRateLimitStore{ + items: make(map[string]grokTeamModelRateLimit), +} + +const ( + grokTeamRateLimitDefaultTTL = 10 * time.Minute + grokTeamRateLimitMaxTTL = time.Hour + grokTeamRateLimitMinTTL = 30 * time.Second +) + +func grokTeamFingerprint(teamID string) string { + teamID = strings.TrimSpace(teamID) + if teamID == "" { + return "" + } + sum := sha256.Sum256([]byte(strings.ToLower(teamID))) + return hex.EncodeToString(sum[:8]) +} + +func grokTeamModelRateLimitKey(teamFingerprint, model string) string { + return teamFingerprint + "|" + strings.ToLower(strings.TrimSpace(model)) +} + +func accountGrokTeamID(account *Account) string { + if account == nil { + return "" + } + return strings.TrimSpace(account.GetCredential("team_id")) +} + +// markGrokTeamModelRateLimit records that this team+model pair should be skipped +// until until. No-op when team_id or model is empty. +func markGrokTeamModelRateLimit(account *Account, model string, until time.Time) { + if account == nil || !account.IsGrokOAuth() { + return + } + fp := grokTeamFingerprint(accountGrokTeamID(account)) + model = strings.TrimSpace(model) + if fp == "" || model == "" || until.IsZero() { + return + } + now := time.Now() + if !until.After(now) { + until = now.Add(grokTeamRateLimitDefaultTTL) + } + maxUntil := now.Add(grokTeamRateLimitMaxTTL) + if until.After(maxUntil) { + until = maxUntil + } + key := grokTeamModelRateLimitKey(fp, model) + globalGrokTeamModelRateLimits.mu.Lock() + defer globalGrokTeamModelRateLimits.mu.Unlock() + if cur, ok := globalGrokTeamModelRateLimits.items[key]; ok && cur.Until.After(until) { + return + } + globalGrokTeamModelRateLimits.items[key] = grokTeamModelRateLimit{Until: until} + // Opportunistic prune of expired entries. + for k, v := range globalGrokTeamModelRateLimits.items { + if !v.Until.After(now) { + delete(globalGrokTeamModelRateLimits.items, k) + } + } +} + +// isGrokTeamModelRateLimited reports whether the account's team is currently +// blocked for the requested model. +func isGrokTeamModelRateLimited(account *Account, model string, now time.Time) bool { + if account == nil || !account.IsGrokOAuth() { + return false + } + fp := grokTeamFingerprint(accountGrokTeamID(account)) + model = strings.TrimSpace(model) + if fp == "" || model == "" { + return false + } + key := grokTeamModelRateLimitKey(fp, model) + globalGrokTeamModelRateLimits.mu.Lock() + defer globalGrokTeamModelRateLimits.mu.Unlock() + cur, ok := globalGrokTeamModelRateLimits.items[key] + if !ok { + return false + } + if !cur.Until.After(now) { + delete(globalGrokTeamModelRateLimits.items, key) + return false + } + return true +} + +// filterGrokTeamModelRateLimitedAccounts drops candidates whose team is under a +// model-scoped rate-limit cool. Accounts without team_id pass through. +func filterGrokTeamModelRateLimitedAccounts(accounts []Account, model string, now time.Time) []Account { + if len(accounts) == 0 || strings.TrimSpace(model) == "" { + return accounts + } + out := accounts[:0] + kept := false + for i := range accounts { + if isGrokTeamModelRateLimited(&accounts[i], model, now) { + continue + } + out = append(out, accounts[i]) + kept = true + } + if !kept && len(out) == 0 { + // All filtered — return empty (caller treats as no capacity). + return nil + } + return out +} + +// resolveGrokTeamRateLimitUntil derives a team cool window from an observed +// account rate-limit reset, with sane clamps. +func resolveGrokTeamRateLimitUntil(resetAt, now time.Time) time.Time { + if resetAt.After(now.Add(grokTeamRateLimitMinTTL)) { + maxUntil := now.Add(grokTeamRateLimitMaxTTL) + if resetAt.After(maxUntil) { + return maxUntil + } + return resetAt + } + return now.Add(grokTeamRateLimitDefaultTTL) +} diff --git a/backend/internal/service/grok_team_rate_limit_test.go b/backend/internal/service/grok_team_rate_limit_test.go new file mode 100644 index 0000000000..50c4ed5065 --- /dev/null +++ b/backend/internal/service/grok_team_rate_limit_test.go @@ -0,0 +1,59 @@ +//go:build unit + +package service + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestGrokTeamModelRateLimit_MarksAndFiltersSiblings(t *testing.T) { + // Isolate from other tests by using unique team ids. + team := "team-test-" + time.Now().Format("150405.000") + a1 := &Account{ + ID: 101, Platform: PlatformGrok, Type: AccountTypeOAuth, + Credentials: map[string]any{"team_id": team}, + } + a2 := &Account{ + ID: 102, Platform: PlatformGrok, Type: AccountTypeOAuth, + Credentials: map[string]any{"team_id": team}, + } + other := &Account{ + ID: 103, Platform: PlatformGrok, Type: AccountTypeOAuth, + Credentials: map[string]any{"team_id": team + "-other"}, + } + noTeam := &Account{ + ID: 104, Platform: PlatformGrok, Type: AccountTypeOAuth, + Credentials: map[string]any{}, + } + + now := time.Now() + markGrokTeamModelRateLimit(a1, "grok-4.5", now.Add(5*time.Minute)) + + require.True(t, isGrokTeamModelRateLimited(a1, "grok-4.5", now)) + require.True(t, isGrokTeamModelRateLimited(a2, "grok-4.5", now), "sibling with same team must cool") + require.False(t, isGrokTeamModelRateLimited(a2, "grok-4.3", now), "other model stays pickable") + require.False(t, isGrokTeamModelRateLimited(other, "grok-4.5", now)) + require.False(t, isGrokTeamModelRateLimited(noTeam, "grok-4.5", now)) + + filtered := filterGrokTeamModelRateLimitedAccounts([]Account{*a1, *a2, *other, *noTeam}, "grok-4.5", now) + require.Len(t, filtered, 2) + ids := []int64{filtered[0].ID, filtered[1].ID} + require.Contains(t, ids, int64(103)) + require.Contains(t, ids, int64(104)) +} + +func TestGrokTeamModelRateLimit_Expires(t *testing.T) { + team := "team-expire-" + time.Now().Format("150405.000") + a := &Account{ + ID: 201, Platform: PlatformGrok, Type: AccountTypeOAuth, + Credentials: map[string]any{"team_id": team}, + } + past := time.Now().Add(-time.Minute) + markGrokTeamModelRateLimit(a, "grok-4.5", past) + // mark clamps expired until into default TTL from "now" — use direct store inject via past+recheck + // After mark with past, resolveGrokTeamRateLimitUntil path isn't used; mark uses now+default when until not after now. + require.True(t, isGrokTeamModelRateLimited(a, "grok-4.5", time.Now())) +} diff --git a/backend/internal/service/grok_token_refresher.go b/backend/internal/service/grok_token_refresher.go index d2d4d534fc..72b041bc6e 100644 --- a/backend/internal/service/grok_token_refresher.go +++ b/backend/internal/service/grok_token_refresher.go @@ -3,12 +3,25 @@ package service import ( "context" "errors" + "hash/fnv" "strings" "time" ) +// Base warm window: refresh when access token lifetime remaining is below this. +// Grok access tokens are typically ~1h; refreshing up to 1h early keeps the pool +// warm for request path cache misses. const grokTokenRefreshSkew = time.Hour +// Stampede spread: each account's effective warm window is reduced by a +// deterministic offset in [0, grokTokenRefreshJitterMax] so co-imported accounts +// do not all refresh in the same TokenRefreshService cycle (grok2api-style +// RefreshDueAt scatter). +const grokTokenRefreshJitterMax = 3 * time.Minute + +// Floor so jitter cannot shrink the window below a useful threshold. +const grokTokenRefreshSkewMin = 30 * time.Minute + type GrokTokenRefresher struct { grokOAuthService GrokOAuthTokenService } @@ -40,9 +53,35 @@ func (r *GrokTokenRefresher) NeedsRefresh(account *Account, refreshWindow time.D if refreshWindow < grokTokenRefreshSkew { refreshWindow = grokTokenRefreshSkew } + // Deterministic per-account jitter: spread warm refreshes without random + // non-determinism in tests (hash of account id). + refreshWindow = grokTokenRefreshWindowWithJitter(account.ID, refreshWindow) return time.Until(*expiresAt) < refreshWindow } +// grokTokenRefreshWindowWithJitter returns refreshWindow minus a stable offset +// in [0, jitterMax] based on accountID. Result is never below grokTokenRefreshSkewMin +// when the base window is at least that large. +func grokTokenRefreshWindowWithJitter(accountID int64, refreshWindow time.Duration) time.Duration { + if accountID <= 0 || refreshWindow <= grokTokenRefreshSkewMin { + return refreshWindow + } + h := fnv.New32a() + var b [8]byte + id := uint64(accountID) + for i := 0; i < 8; i++ { + b[i] = byte(id >> (8 * i)) + } + _, _ = h.Write(b[:]) + // Jitter in [0, grokTokenRefreshJitterMax). + jitter := time.Duration(h.Sum32()%uint32(grokTokenRefreshJitterMax/time.Second)) * time.Second + out := refreshWindow - jitter + if out < grokTokenRefreshSkewMin { + return grokTokenRefreshSkewMin + } + return out +} + func (r *GrokTokenRefresher) Refresh(ctx context.Context, account *Account) (map[string]any, error) { if r == nil || r.grokOAuthService == nil { return nil, errors.New("grok oauth service is not configured") diff --git a/backend/internal/service/grok_token_refresher_test.go b/backend/internal/service/grok_token_refresher_test.go new file mode 100644 index 0000000000..c689985628 --- /dev/null +++ b/backend/internal/service/grok_token_refresher_test.go @@ -0,0 +1,50 @@ +//go:build unit + +package service + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestGrokTokenRefreshWindowWithJitter_StableAndBounded(t *testing.T) { + base := grokTokenRefreshSkew + w1 := grokTokenRefreshWindowWithJitter(42, base) + w2 := grokTokenRefreshWindowWithJitter(42, base) + require.Equal(t, w1, w2, "same account id must yield stable window") + require.GreaterOrEqual(t, w1, grokTokenRefreshSkewMin) + require.LessOrEqual(t, w1, base) + + // Different accounts should usually differ (not a hard guarantee for all pairs, + // but for sequential ids the hash spread is good enough to assert inequality + // across a small sample). + seen := map[time.Duration]bool{} + for id := int64(1); id <= 50; id++ { + seen[grokTokenRefreshWindowWithJitter(id, base)] = true + } + require.Greater(t, len(seen), 1, "jitter should spread windows across accounts") +} + +func TestGrokTokenRefresher_NeedsRefresh_UsesSkewFloor(t *testing.T) { + refresher := NewGrokTokenRefresher(nil) + // Expires in 50 minutes — within 1h skew, should need refresh. + expires := time.Now().Add(50 * time.Minute).UTC().Format(time.RFC3339) + account := &Account{ + ID: 7, + Platform: PlatformGrok, + Type: AccountTypeOAuth, + Credentials: map[string]any{ + "access_token": "at", + "refresh_token": "rt", + "expires_at": expires, + }, + } + // Pass a tiny window; NeedsRefresh raises it to grokTokenRefreshSkew then jitter. + require.True(t, refresher.NeedsRefresh(account, time.Minute)) + + // Far future — no refresh. + account.Credentials["expires_at"] = time.Now().Add(3 * time.Hour).UTC().Format(time.RFC3339) + require.False(t, refresher.NeedsRefresh(account, time.Minute)) +} diff --git a/backend/internal/service/openai_account_scheduler.go b/backend/internal/service/openai_account_scheduler.go index f39a38dbd8..6546a4e067 100644 --- a/backend/internal/service/openai_account_scheduler.go +++ b/backend/internal/service/openai_account_scheduler.go @@ -506,6 +506,11 @@ func (s *defaultOpenAIAccountScheduler) selectBySessionHash( _ = s.service.deleteStickySessionAccountID(ctx, req.GroupID, sessionHash) return nil, false, nil } + // Team+model cool: sticky must not pin a sibling under the same team 429 window. + if account != nil && isGrokTeamModelRateLimited(account, req.RequestedModel, time.Now()) { + _ = s.service.deleteStickySessionAccountID(ctx, req.GroupID, sessionHash) + return nil, false, nil + } escapeCfg := s.service.openAIStickyEscapeConfig() if reason, errorRate, ttft, shouldEscape := s.shouldEscapeStickyAccount(accountID, escapeCfg); shouldEscape { slog.Info("sticky_escape_triggered", @@ -1348,6 +1353,16 @@ func (s *defaultOpenAIAccountScheduler) selectByLoadBalance( if len(accounts) == 0 { return nil, 0, 0, 0, noAvailableOpenAISelectionError(req.RequestedModel, false, openAISelectionFilterStats{}.summary("grok_free_quota_soft_gate")) } + // Team+model rate-limit cool: siblings of a 429'd team skip the hot model. + if req.Platform == PlatformGrok { + filtered := filterGrokTeamModelRateLimitedAccounts(accounts, req.RequestedModel, time.Now()) + if len(filtered) == 0 && len(accounts) > 0 { + return nil, 0, 0, 0, noAvailableOpenAISelectionError(req.RequestedModel, false, openAISelectionFilterStats{}.summary("grok_team_model_rate_limit")) + } + if filtered != nil { + accounts = filtered + } + } // require_privacy_set: 获取分组信息 var schedGroup *Group diff --git a/backend/internal/service/openai_gateway_grok.go b/backend/internal/service/openai_gateway_grok.go index 8140c14cf8..42f1bc6f62 100644 --- a/backend/internal/service/openai_gateway_grok.go +++ b/backend/internal/service/openai_gateway_grok.go @@ -161,7 +161,13 @@ func (s *OpenAIGatewayService) forwardGrokResponses( Kind: kind, Message: upstreamMsg, }) - s.handleGrokAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody) + 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 { + markGrokTeamModelRateLimit(account, upstreamModel, resolveGrokTeamRateLimitUntil(time.Now().Add(grokTeamRateLimitDefaultTTL), time.Now())) + } if s.shouldFailoverGrokUpstreamError(resp.StatusCode, respBody) { return nil, &UpstreamFailoverError{ StatusCode: resp.StatusCode, @@ -173,7 +179,9 @@ func (s *OpenAIGatewayService) forwardGrokResponses( return s.handleErrorResponse(ctx, resp, c, account, patchedBody, upstreamModel) } - s.updateGrokUsageFromResponse(ctx, account, resp.Header, resp.StatusCode) + // Attach model so rate-limit snapshots can fan out a team+model cool. + stateCtx := withGrokTeamRateLimitModel(ctx, upstreamModel) + s.updateGrokUsageFromResponse(stateCtx, account, resp.Header, resp.StatusCode) var usage *OpenAIUsage var firstTokenMs *int @@ -1337,7 +1345,8 @@ func (s *OpenAIGatewayService) rateLimitGrok(ctx context.Context, account *Accou if s == nil || account == nil { return } - resetAt = normalizeGrokRateLimitResetAt(account, resetAt, time.Now()) + now := time.Now() + resetAt = normalizeGrokRateLimitResetAt(account, resetAt, now) runtimeUntil := resetAt if account.TempUnschedulableUntil != nil && account.TempUnschedulableUntil.After(runtimeUntil) { @@ -1345,6 +1354,27 @@ func (s *OpenAIGatewayService) rateLimitGrok(ctx context.Context, account *Accou } s.BlockAccountScheduling(account, runtimeUntil, "429") persistGrokRateLimit(ctx, s.accountRepo, account, resetAt) + + // Propagate a short team+model cool so sibling OAuth accounts on the same + // xAI team skip the hot model without waiting for each to hit 429 alone. + // Model is taken from the latest request context when available; empty is a + // no-op inside markGrokTeamModelRateLimit. + if model, _ := ctx.Value(grokTeamRateLimitModelContextKey{}).(string); model != "" { + markGrokTeamModelRateLimit(account, model, resolveGrokTeamRateLimitUntil(resetAt, now)) + } +} + +// grokTeamRateLimitModelContextKey carries the upstream model for team cools. +type grokTeamRateLimitModelContextKey struct{} + +// withGrokTeamRateLimitModel attaches the upstream model name for rate-limit +// side effects (team+model cool). Safe when model is empty. +func withGrokTeamRateLimitModel(ctx context.Context, model string) context.Context { + model = strings.TrimSpace(model) + if model == "" || ctx == nil { + return ctx + } + return context.WithValue(ctx, grokTeamRateLimitModelContextKey{}, model) } func (s *OpenAIGatewayService) handleGrokAccountUpstreamError(ctx context.Context, account *Account, statusCode int, headers http.Header, responseBody []byte) { diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 2a1993c463..dd83e3357d 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -150,6 +150,16 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. if s.cfg != nil && s.cfg.Gateway.StreamDataIntervalTimeout > 0 { streamInterval = time.Duration(s.cfg.Gateway.StreamDataIntervalTimeout) * time.Second } + // Grok: always enforce an upstream-read idle so hung SSE bodies fail over + // instead of holding the OAuth slot until the client cancels. Prefer the + // global gateway setting when set; otherwise apply a Grok-only default. + if account != nil && account.Platform == PlatformGrok { + cfgSec := 0 + if s.cfg != nil { + cfgSec = s.cfg.Gateway.StreamDataIntervalTimeout + } + streamInterval = resolveGrokStreamIdleTimeout(cfgSec) + } // 仅监控上游数据间隔超时，不被下游写入阻塞影响 var intervalTicker *time.Ticker if streamInterval > 0 { @@ -687,6 +697,16 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. if s.rateLimitService != nil { s.rateLimitService.HandleStreamTimeout(ctx, account, originalModel) } + // Grok: short cool + account failover when no client-visible bytes + // were committed yet (pre-commit). After output started we keep the + // legacy stream_timeout path so partial SSE is not dual-written. + if account != nil && account.Platform == PlatformGrok { + s.tempUnscheduleGrok(ctx, account, grokStreamIdleCooldown, "grok stream idle timeout") + if !openAIStreamClientOutputStarted(c, clientOutputStarted) && !eventShouldFlush { + _ = resp.Body.Close() + return resultWithUsage(), grokStreamIdleFailoverError(account, streamInterval) + } + } sendErrorEvent("stream_timeout") return resultWithUsage(), fmt.Errorf("stream data interval timeout")