diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index 0fa0058a5d..048f2d76e9 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -934,6 +934,9 @@ type GatewayConfig struct { // OpenAIResponseHeaderTimeout: OpenAI/Codex 上游等待响应头的超时时间（秒），0表示无超时 // OpenAI/Codex 请求可能在上游排队较久；默认不使用通用响应头超时截断。 OpenAIResponseHeaderTimeout int `mapstructure:"openai_response_header_timeout"` + // GrokResponseHeaderTimeout bounds the pre-first-byte wait for xAI/Grok. + // A zero value uses the provider-safe default instead of the generic gateway timeout. + GrokResponseHeaderTimeout int `mapstructure:"grok_response_header_timeout"` // OpenAIFirstOutputTimeoutSeconds: native HTTP Responses 首个语义输出超时（秒），0表示禁用。 OpenAIFirstOutputTimeoutSeconds int `mapstructure:"openai_first_output_timeout_seconds"` // OpenAIHighEffortFirstOutputTimeoutSeconds: high/xhigh/max 推理的首个语义输出超时（秒）。 @@ -2327,6 +2330,7 @@ func setDefaults() { // Gateway viper.SetDefault("gateway.response_header_timeout", 600) // 600秒(10分钟)等待上游响应头，LLM高负载时可能排队较久 viper.SetDefault("gateway.openai_response_header_timeout", 0) + viper.SetDefault("gateway.grok_response_header_timeout", 120) viper.SetDefault("gateway.openai_first_output_timeout_seconds", 0) viper.SetDefault("gateway.openai_high_effort_first_output_timeout_seconds", 0) viper.SetDefault("gateway.log_upstream_error_body", true) @@ -3241,6 +3245,9 @@ func (c *Config) Validate() error { if c.Gateway.OpenAIResponseHeaderTimeout < 0 { return fmt.Errorf("gateway.openai_response_header_timeout must be non-negative") } + if c.Gateway.GrokResponseHeaderTimeout < 0 || c.Gateway.GrokResponseHeaderTimeout > 1800 { + return fmt.Errorf("gateway.grok_response_header_timeout must be between 0-1800 seconds") + } if c.Gateway.OpenAIFirstOutputTimeoutSeconds < 0 || c.Gateway.OpenAIFirstOutputTimeoutSeconds > 600 || (c.Gateway.OpenAIFirstOutputTimeoutSeconds > 0 && c.Gateway.OpenAIFirstOutputTimeoutSeconds < 30) { return fmt.Errorf("gateway.openai_first_output_timeout_seconds must be 0 or between 30-600 seconds") diff --git a/backend/internal/handler/grok_audio.go b/backend/internal/handler/grok_audio.go index 4ac1ae53bf..a31ec802f6 100644 --- a/backend/internal/handler/grok_audio.go +++ b/backend/internal/handler/grok_audio.go @@ -77,6 +77,20 @@ func (h *OpenAIGatewayHandler) GrokRealtime(c *gin.Context) { h.errorResponse(c, http.StatusBadGateway, "upstream_error", "Grok credential unavailable") return } + model := c.Query("model") + if strings.TrimSpace(model) == "" { + model = "grok-voice-latest" + } + // Complete the upstream handshake before sending HTTP 101. This keeps + // pre-accept failures representable as normal JSON errors to the client. + probeCtx, cancelProbe := context.WithTimeout(c.Request.Context(), 15*time.Second) + probeErr := h.gatewayService.ProbeGrokRealtime(probeCtx, selection.Account, token, model) + cancelProbe() + if probeErr != nil { + reqLog.Warn("grok_realtime.pre_accept_failed", zap.Error(probeErr)) + h.errorResponse(c, http.StatusBadGateway, "upstream_error", "Grok realtime upstream unavailable") + return + } conn, err := coderws.Accept(c.Writer, c.Request, &coderws.AcceptOptions{CompressionMode: coderws.CompressionContextTakeover}) if err != nil { @@ -84,10 +98,6 @@ func (h *OpenAIGatewayHandler) GrokRealtime(c *gin.Context) { } defer func() { _ = conn.CloseNow() }() - model := c.Query("model") - if strings.TrimSpace(model) == "" { - model = "grok-voice-latest" - } started := time.Now() audioObserved, proxyErr := h.gatewayService.ProxyGrokRealtime(c.Request.Context(), c, conn, selection.Account, token, model) elapsed := time.Since(started) diff --git a/backend/internal/repository/http_upstream.go b/backend/internal/repository/http_upstream.go index 904ff09eb9..43be1bc995 100644 --- a/backend/internal/repository/http_upstream.go +++ b/backend/internal/repository/http_upstream.go @@ -899,12 +899,20 @@ func (s *httpUpstreamService) resolvePoolSettings(isolation string, accountConcu } func (s *httpUpstreamService) applyProfilePoolSettings(settings poolSettings, profile service.HTTPUpstreamProfile) poolSettings { - if profile != service.HTTPUpstreamProfileOpenAI { - return settings - } - settings.responseHeaderTimeout = 0 - if s != nil && s.cfg != nil && s.cfg.Gateway.OpenAIResponseHeaderTimeout > 0 { - settings.responseHeaderTimeout = time.Duration(s.cfg.Gateway.OpenAIResponseHeaderTimeout) * time.Second + switch profile { + case service.HTTPUpstreamProfileOpenAI: + settings.responseHeaderTimeout = 0 + if s != nil && s.cfg != nil && s.cfg.Gateway.OpenAIResponseHeaderTimeout > 0 { + settings.responseHeaderTimeout = time.Duration(s.cfg.Gateway.OpenAIResponseHeaderTimeout) * time.Second + } + case service.HTTPUpstreamProfileGrok: + // Grok can stall before its first byte under capacity pressure. Keep the + // 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 { + settings.responseHeaderTimeout = time.Duration(s.cfg.Gateway.GrokResponseHeaderTimeout) * time.Second + } } return settings } diff --git a/backend/internal/service/grok_audio.go b/backend/internal/service/grok_audio.go index 06c8410b8d..63a2e28b00 100644 --- a/backend/internal/service/grok_audio.go +++ b/backend/internal/service/grok_audio.go @@ -63,6 +63,7 @@ func (s *OpenAIGatewayService) ForwardGrokVoice(ctx context.Context, c *gin.Cont } upstreamCtx, release := detachUpstreamContext(ctx) defer release() + upstreamCtx = WithHTTPUpstreamProfile(upstreamCtx, HTTPUpstreamProfileGrok) method := http.MethodPost if c != nil && c.Request != nil && strings.TrimSpace(c.Request.Method) != "" { method = c.Request.Method @@ -207,6 +208,45 @@ func (s *OpenAIGatewayService) ProxyGrokRealtime(ctx context.Context, c *gin.Con return awaitGrokRealtimeAudioObserved(errCh, &audioObserved) } +// ProbeGrokRealtime performs the upstream WebSocket handshake without sending +// any client-visible events. Handlers use it before accepting the downstream +// upgrade so authentication and endpoint failures remain ordinary HTTP errors. +func (s *OpenAIGatewayService) ProbeGrokRealtime(ctx context.Context, account *Account, token, model string) error { + if s == nil || account == nil { + return fmt.Errorf("realtime service and account are required") + } + if account.Platform != PlatformGrok { + return fmt.Errorf("account platform %s is not supported for grok realtime", account.Platform) + } + base, err := buildGrokVoiceURL(account, s.cfg, "realtime") + if err != nil { + return err + } + u, err := url.Parse(base) + if err != nil { + return err + } + u.Scheme = "wss" + q := u.Query() + q.Set("model", firstNonEmpty(model, "grok-voice-latest")) + u.RawQuery = q.Encode() + headers := http.Header{"Authorization": []string{"Bearer " + token}} + if account.IsGrokOAuth() && isGrokCLIProxyTarget(u.String()) { + applyGrokCLIHeaders(headers) + } + account.ApplyHeaderOverrides(headers) + proxyURL := "" + if account.ProxyID != nil && account.Proxy != nil { + proxyURL = account.Proxy.URL() + } + dialer := s.getOpenAIWSPassthroughDialer() + conn, _, _, err := dialer.Dial(ctx, u.String(), headers, proxyURL) + if err != nil { + return err + } + return conn.Close() +} + func awaitGrokRealtimeAudioObserved(errCh <-chan error, audioObserved *atomic.Bool) (bool, error) { err := <-errCh if audioObserved == nil { diff --git a/backend/internal/service/grok_media.go b/backend/internal/service/grok_media.go index 2efb2332f6..2155f7eb99 100644 --- a/backend/internal/service/grok_media.go +++ b/backend/internal/service/grok_media.go @@ -658,6 +658,7 @@ func (s *OpenAIGatewayService) ForwardGrokMedia( } upstreamCtx, releaseUpstreamCtx := detachUpstreamContext(ctx) defer releaseUpstreamCtx() + upstreamCtx = WithHTTPUpstreamProfile(upstreamCtx, HTTPUpstreamProfileGrok) upstreamReq, err := http.NewRequestWithContext(upstreamCtx, endpoint.httpMethod(), targetURL, bodyReader) if err != nil { return nil, err @@ -1288,11 +1289,14 @@ func (s *OpenAIGatewayService) handleGrokMediaErrorResponse( Detail: upstreamDetail, }) if kind == "failover" { + retryable, retryDelay, retryDeadline := grokSameAccountRetryMetadata(account, resp.StatusCode, body) return nil, &UpstreamFailoverError{ - StatusCode: resp.StatusCode, - ResponseBody: body, - ResponseHeaders: resp.Header.Clone(), - RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), + StatusCode: resp.StatusCode, + ResponseBody: body, + ResponseHeaders: resp.Header.Clone(), + RetryableOnSameAccount: retryable, + SameAccountRetryDelay: retryDelay, + SameAccountRetryDeadline: retryDeadline, } } diff --git a/backend/internal/service/http_upstream_profile.go b/backend/internal/service/http_upstream_profile.go index 83bd6e70fc..b2de6002cf 100644 --- a/backend/internal/service/http_upstream_profile.go +++ b/backend/internal/service/http_upstream_profile.go @@ -9,6 +9,7 @@ type HTTPUpstreamProfile string const ( HTTPUpstreamProfileDefault HTTPUpstreamProfile = "" HTTPUpstreamProfileOpenAI HTTPUpstreamProfile = "openai" + HTTPUpstreamProfileGrok HTTPUpstreamProfile = "grok" ) type httpUpstreamProfileContextKey struct{} @@ -35,7 +36,7 @@ func HTTPUpstreamProfileFromContext(ctx context.Context) HTTPUpstreamProfile { return HTTPUpstreamProfileDefault } switch profile { - case HTTPUpstreamProfileOpenAI: + case HTTPUpstreamProfileOpenAI, HTTPUpstreamProfileGrok: return profile default: return HTTPUpstreamProfileDefault