diff --git a/backend/internal/handler/grok_audio.go b/backend/internal/handler/grok_audio.go index a31ec802f6..bbe7a4210c 100644 --- a/backend/internal/handler/grok_audio.go +++ b/backend/internal/handler/grok_audio.go @@ -44,53 +44,65 @@ func (h *OpenAIGatewayHandler) GrokRealtime(c *gin.Context) { return } - selection, _, err := h.gatewayService.SelectAccountWithSchedulerForCapability( - c.Request.Context(), - apiKey.GroupID, - "", - "", - "grok-4.5", - nil, - service.OpenAIUpstreamTransportHTTPSSE, - // Grok only advertises chat_completions + media capabilities on HEAD. - service.OpenAIEndpointCapabilityChatCompletions, - false, - false, - false, - service.PlatformGrok, - ) - if err != nil || selection == nil || selection.Account == nil { - h.errorResponse(c, http.StatusServiceUnavailable, "api_error", "No available Grok accounts") - return - } - - var streamStarted bool reqLog := requestLogger(c, "handler.openai_gateway.grok_realtime") - release, slotStatus := h.acquireResponsesAccountSlot(c, apiKey.GroupID, "", selection, true, &streamStarted, reqLog) - if slotStatus != openAISlotAcquireOK { - return - } - defer release() - - token, _, err := h.gatewayService.GetRequestCredential(c.Request.Context(), c, selection.Account) - if err != nil { - 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)) + // Keep the HTTP response uncommitted while selecting and probing an account. + // Realtime is not an HTTP streaming response; using reqStream=true here would + // let the wait queue flush an SSE ping before the WebSocket handshake succeeds. + failed := map[int64]struct{}{} + var selection *service.AccountSelectionResult + var release func() + var token string + for attempts := 0; attempts < 4; attempts++ { + candidate, _, selectErr := h.gatewayService.SelectAccountWithSchedulerForCapability( + c.Request.Context(), apiKey.GroupID, "", "", "grok-4.6", failed, + service.OpenAIUpstreamTransportHTTPSSE, + service.OpenAIEndpointCapabilityChatCompletions, + false, false, false, service.PlatformGrok, + ) + if selectErr != nil || candidate == nil || candidate.Account == nil { + break + } + account := candidate.Account + var streamStarted bool + var slotStatus openAISlotAcquireResult + release, slotStatus = h.acquireResponsesAccountSlot(c, apiKey.GroupID, "", candidate, false, &streamStarted, reqLog) + if slotStatus != openAISlotAcquireOK { + if slotStatus == openAISlotAcquireFailed { + return + } + failed[account.ID] = struct{}{} + continue + } + var credErr error + token, _, credErr = h.gatewayService.GetRequestCredential(c.Request.Context(), c, account) + if credErr != nil { + release() + release = nil + failed[account.ID] = struct{}{} + continue + } + probeCtx, cancelProbe := context.WithTimeout(c.Request.Context(), 15*time.Second) + probeErr := h.gatewayService.ProbeGrokRealtime(probeCtx, account, token, model) + cancelProbe() + if probeErr != nil { + reqLog.Warn("grok_realtime.pre_accept_failed", zap.Int64("account_id", account.ID), zap.Error(probeErr)) + release() + release = nil + failed[account.ID] = struct{}{} + continue + } + selection = candidate + break + } + if selection == nil || selection.Account == nil || release == nil { h.errorResponse(c, http.StatusBadGateway, "upstream_error", "Grok realtime upstream unavailable") return } + defer release() conn, err := coderws.Accept(c.Writer, c.Request, &coderws.AcceptOptions{CompressionMode: coderws.CompressionContextTakeover}) if err != nil { diff --git a/backend/internal/service/grok_audio.go b/backend/internal/service/grok_audio.go index 63a2e28b00..9b769a5fa4 100644 --- a/backend/internal/service/grok_audio.go +++ b/backend/internal/service/grok_audio.go @@ -63,7 +63,6 @@ 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