From fc86e6e1516ab8a7f65adfa1ead8f5e9e0730d01 Mon Sep 17 00:00:00 2001 From: Shaik-Sirajuddin Date: Tue, 7 Jul 2026 14:49:35 +0530 Subject: [PATCH 1/4] feat: add Gemini Live API (BidiGenerateContent) realtime support Implements schemas.RealtimeProvider for Gemini, closing #3736. Current Gemini Live models are audio-output-only (TEXT responseModalities rejected at setup), so T->T/S->T are achieved via the outputAudioTranscription side-channel alongside native T->S/S->S audio support. Also fixes a shared-transport gap where session.update omitting a model would fail providers whose wire protocol only learns the model from the session payload. --- core/internal/llmtests/realtime.go | 125 ++++- core/providers/gemini/gemini_test.go | 2 + core/providers/gemini/realtime.go | 523 ++++++++++++++++++ core/providers/gemini/realtime_test.go | 403 ++++++++++++++ docs/openapi/paths/inference/realtime.yaml | 12 +- docs/providers/supported-providers/gemini.mdx | 63 +++ .../handlers/realtime_turn_pipeline.go | 12 +- .../bifrost-http/handlers/webrtc_realtime.go | 27 +- .../bifrost-http/handlers/wsrealtime.go | 33 +- 9 files changed, 1192 insertions(+), 8 deletions(-) create mode 100644 core/providers/gemini/realtime.go create mode 100644 core/providers/gemini/realtime_test.go diff --git a/core/internal/llmtests/realtime.go b/core/internal/llmtests/realtime.go index e024ffb8c5d..e5fb0425ac9 100644 --- a/core/internal/llmtests/realtime.go +++ b/core/internal/llmtests/realtime.go @@ -78,9 +78,12 @@ func RunRealtimeTest(t *testing.T, client *bifrost.Bifrost, ctx context.Context, t.Logf("connected to Realtime endpoint: %s", wsURL) - if testConfig.Provider == schemas.Elevenlabs { + switch testConfig.Provider { + case schemas.Elevenlabs: runElevenLabsRealtimeTest(t, conn, testConfig) - } else { + case schemas.Gemini: + runGeminiRealtimeTest(t, conn, testConfig) + default: runOpenAIRealtimeTest(t, conn, testConfig) } }) @@ -259,6 +262,124 @@ func runElevenLabsRealtimeTest(t *testing.T, conn *ws.Conn, testConfig Comprehen t.Logf("ElevenLabs Realtime test passed (%d events)", eventCount) } +// runGeminiRealtimeTest drives a Gemini Live (BidiGenerateContent) session using +// its native wire protocol directly — this harness dials the provider endpoint +// itself (bypassing Bifrost's translation layer), so it speaks Gemini's raw +// setup/clientContent/serverContent shape, not the canonical Bifrost envelope. +// Gemini's current live models only support AUDIO output (TEXT responseModalities +// is rejected at setup time), so this enables outputAudioTranscription to also +// assert on the text side-channel — confirmed live to ride in the same +// serverContent message as the audio parts, not a separate message. +func runGeminiRealtimeTest(t *testing.T, conn *ws.Conn, testConfig ComprehensiveTestConfig) { + model := testConfig.RealtimeModel + if !strings.HasPrefix(model, "models/") { + model = "models/" + model + } + + setup := map[string]interface{}{ + "setup": map[string]interface{}{ + "model": model, + "generationConfig": map[string]interface{}{"responseModalities": []string{"AUDIO"}}, + "outputAudioTranscription": map[string]interface{}{}, + }, + } + writeJSON(t, conn, setup) + + eventCount := 0 + conn.SetReadDeadline(time.Now().Add(30 * time.Second)) + + var gotSetupComplete bool + for i := 0; i < 5 && !gotSetupComplete; i++ { + _, msg, err := conn.ReadMessage() + if err != nil { + t.Fatalf("error reading setup response: %v", err) + } + eventCount++ + if geminiHasTopLevelKey(msg, "setupComplete") { + gotSetupComplete = true + } + if geminiHasTopLevelKey(msg, "error") { + t.Fatalf("received error during setup: %s", string(msg)) + } + } + if !gotSetupComplete { + t.Fatal("did not receive setupComplete event") + } + t.Logf("Gemini Live setup complete (%d events)", eventCount) + + clientContent := map[string]interface{}{ + "clientContent": map[string]interface{}{ + "turns": []map[string]interface{}{{"role": "user", "parts": []map[string]interface{}{{"text": "Say hello in exactly two words."}}}}, + "turnComplete": true, + }, + } + writeJSON(t, conn, clientContent) + + var ( + gotAudioOrTranscript bool + gotTurnComplete bool + ) + + conn.SetReadDeadline(time.Now().Add(30 * time.Second)) + for i := 0; i < 100; i++ { + _, msg, err := conn.ReadMessage() + if err != nil { + if !gotTurnComplete { + t.Fatalf("WS read error before turnComplete (events=%d): %v", eventCount, err) + } + break + } + eventCount++ + + if geminiHasTopLevelKey(msg, "error") { + t.Fatalf("received error event: %s", string(msg)) + } + if strings.Contains(string(msg), "inlineData") || strings.Contains(string(msg), "outputTranscription") { + gotAudioOrTranscript = true + } + if geminiServerContentTurnComplete(msg) { + gotTurnComplete = true + t.Logf("received serverContent.turnComplete (total events: %d)", eventCount) + break + } + } + + if !gotAudioOrTranscript { + t.Error("expected at least one audio (inlineData) or outputTranscription event") + } + if !gotTurnComplete { + t.Error("expected a serverContent.turnComplete event") + } + t.Logf("Gemini Realtime test passed (%d events)", eventCount) +} + +// geminiHasTopLevelKey reports whether a raw Gemini Live message has the given +// top-level key present (Gemini's BidiGenerateContent messages are a oneof of +// top-level keys — setupComplete/serverContent/toolCall/error/etc. — with no +// shared "type" discriminator field, unlike OpenAI/ElevenLabs). +func geminiHasTopLevelKey(msg []byte, key string) bool { + var raw map[string]json.RawMessage + if err := json.Unmarshal(msg, &raw); err != nil { + return false + } + _, ok := raw[key] + return ok +} + +// geminiServerContentTurnComplete reports whether a message is a +// serverContent.turnComplete terminal event. +func geminiServerContentTurnComplete(msg []byte) bool { + var raw struct { + ServerContent *struct { + TurnComplete bool `json:"turnComplete"` + } `json:"serverContent"` + } + if err := json.Unmarshal(msg, &raw); err != nil { + return false + } + return raw.ServerContent != nil && raw.ServerContent.TurnComplete +} + func extractEventType(msg []byte) string { var raw map[string]json.RawMessage if err := json.Unmarshal(msg, &raw); err != nil { diff --git a/core/providers/gemini/gemini_test.go b/core/providers/gemini/gemini_test.go index 8ad2066b938..3a57356155f 100644 --- a/core/providers/gemini/gemini_test.go +++ b/core/providers/gemini/gemini_test.go @@ -48,6 +48,7 @@ func TestGemini(t *testing.T) { ReasoningModel: "gemini-3-pro-preview", VideoGenerationModel: "veo-3.1-generate-preview", PassthroughModel: "gemini-2.5-flash", + RealtimeModel: "gemini-3.1-flash-live-preview", Scenarios: llmtests.TestScenarios{ TextCompletion: false, // Not supported SimpleChat: true, @@ -93,6 +94,7 @@ func TestGemini(t *testing.T) { CountTokens: true, StructuredOutputs: true, // Structured outputs with nullable enum support PassthroughAPI: true, + Realtime: true, }, } diff --git a/core/providers/gemini/realtime.go b/core/providers/gemini/realtime.go new file mode 100644 index 00000000000..837ccb2b343 --- /dev/null +++ b/core/providers/gemini/realtime.go @@ -0,0 +1,523 @@ +package gemini + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "net/url" + "strings" + + providerUtils "github.com/maximhq/bifrost/core/providers/utils" + schemas "github.com/maximhq/bifrost/core/schemas" +) + +// SupportsRealtimeAPI returns true — Gemini Live (BidiGenerateContent) is supported +// for the T<->T, S<->T, T<->S and S<->S modality combinations. Video input is not +// yet supported (no canonical Bifrost schema slot exists for it). +func (provider *GeminiProvider) SupportsRealtimeAPI() bool { + return true +} + +// RealtimeWebSocketURL returns the Gemini Live BidiGenerateContent WS endpoint. +// Auth is via the `key` query parameter — Gemini's websocket handshake does not +// accept the API key as a request header, confirmed against the live endpoint. +func (provider *GeminiProvider) RealtimeWebSocketURL(key schemas.Key, model string) string { + base := provider.networkConfig.BaseURL + base = strings.Replace(base, "https://", "wss://", 1) + base = strings.Replace(base, "http://", "ws://", 1) + base = strings.TrimSuffix(base, "/v1beta") + return base + "/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent?key=" + url.QueryEscape(key.Value.GetValue()) +} + +// RealtimeHeaders returns no headers — Gemini Live auth rides on the URL's `key` +// query parameter, not a request header. +func (provider *GeminiProvider) RealtimeHeaders(_ *schemas.BifrostContext, _ schemas.Key) (map[string]string, *schemas.BifrostError) { + return map[string]string{}, nil +} + +// SupportsRealtimeWebRTC returns false — Gemini Live has no public WebRTC SDP-exchange spec. +func (provider *GeminiProvider) SupportsRealtimeWebRTC() bool { + return false +} + +// ExchangeRealtimeWebRTCSDP is not implemented for Gemini. +func (provider *GeminiProvider) ExchangeRealtimeWebRTCSDP(_ *schemas.BifrostContext, _ schemas.Key, _ string, _ string, _ json.RawMessage) (string, *schemas.BifrostError) { + return "", &schemas.BifrostError{ + IsBifrostError: true, + StatusCode: schemas.Ptr(400), + Error: &schemas.ErrorField{Type: schemas.Ptr("invalid_request_error"), Message: "WebRTC SDP exchange is not implemented for Gemini"}, + } +} + +func (provider *GeminiProvider) RealtimeWebRTCDataChannelLabel() string { + return "" +} + +func (provider *GeminiProvider) RealtimeWebSocketSubprotocol() string { + return "" +} + +// ShouldStartRealtimeTurn starts a Bifrost turn when the client finalizes its +// content (clientContent with turnComplete, mapped to response.create) or commits +// buffered audio input (mirrors OpenAI's input_audio_buffer.committed trigger). +func (provider *GeminiProvider) ShouldStartRealtimeTurn(event *schemas.BifrostRealtimeEvent) bool { + switch event.Type { + case schemas.RTEventResponseCreate, schemas.RTEventInputAudioCommit: + return true + default: + return false + } +} + +// RealtimeTurnFinalEvent — Gemini signals turn completion via +// serverContent.turnComplete, which we map onto the canonical response.done. +func (provider *GeminiProvider) RealtimeTurnFinalEvent() schemas.RealtimeEventType { + return schemas.RTEventResponseDone +} + +func (provider *GeminiProvider) ShouldForwardRealtimeEvent(_ *schemas.BifrostRealtimeEvent) bool { + return true +} + +// ShouldAccumulateRealtimeOutput accumulates text/audio-transcript deltas so the +// full assistant turn text can be reconstructed for logging even though it only +// streams as deltas. +func (provider *GeminiProvider) ShouldAccumulateRealtimeOutput(eventType schemas.RealtimeEventType) bool { + switch eventType { + // RTEventResponseAudioDelta is included because Gemini bundles outputTranscription + // into the *same* serverContent message as the audio parts (confirmed live: both + // fields present together) — unlike OpenAI, which emits them as separate messages. + // The transport appends whichever of Delta.Text/Delta.Transcript is non-empty, so + // this is what makes the transcript actually get logged for audio-out turns. + // RTEventResponseDone is included because Gemini's final serverContent message + // can carry BOTH turnComplete AND the last modelTurn/outputTranscription chunk + // together — that final chunk's Delta must still get accumulated for logging. + case schemas.RTEventResponseTextDelta, schemas.RTEventResponseAudioDelta, schemas.RTEventResponseAudioTransDelta, schemas.RTEventInputAudioTransCompleted, schemas.RTEventResponseDone: + return true + default: + return false + } +} + +// Gemini Live (BidiGenerateContent) wire-protocol pass-through event types. +// These have no clean 1:1 canonical equivalent, so they're cast rather than +// added to the shared core/schemas/realtime.go enum (same technique ElevenLabs +// uses for "ping"/"client_tool_call"). +const ( + geminiEventToolCall = schemas.RealtimeEventType("tool_call") + geminiEventToolCallCancellation = schemas.RealtimeEventType("tool_call_cancellation") + geminiEventToolResponse = schemas.RealtimeEventType("tool_response") + geminiEventInterrupted = schemas.RealtimeEventType("interrupted") + geminiEventGoAway = schemas.RealtimeEventType("go_away") +) + +// --- Gemini Live wire-format structs --- + +type geminiRealtimeServerMessage struct { + SetupComplete json.RawMessage `json:"setupComplete,omitempty"` + ServerContent *geminiServerContent `json:"serverContent,omitempty"` + ToolCall *geminiToolCall `json:"toolCall,omitempty"` + ToolCallCancellation json.RawMessage `json:"toolCallCancellation,omitempty"` + GoAway json.RawMessage `json:"goAway,omitempty"` + UsageMetadata *geminiUsageMetadata `json:"usageMetadata,omitempty"` +} + +type geminiServerContent struct { + ModelTurn *geminiContent `json:"modelTurn,omitempty"` + TurnComplete bool `json:"turnComplete,omitempty"` + Interrupted bool `json:"interrupted,omitempty"` + InputTranscription *geminiTranscription `json:"inputTranscription,omitempty"` + OutputTranscription *geminiTranscription `json:"outputTranscription,omitempty"` +} + +type geminiContent struct { + Role string `json:"role,omitempty"` + Parts []geminiPart `json:"parts,omitempty"` +} + +type geminiPart struct { + Text string `json:"text,omitempty"` + InlineData *geminiInlineData `json:"inlineData,omitempty"` +} + +type geminiInlineData struct { + MimeType string `json:"mimeType,omitempty"` + Data string `json:"data,omitempty"` // base64 +} + +type geminiTranscription struct { + Text string `json:"text,omitempty"` + Finished bool `json:"finished,omitempty"` +} + +type geminiToolCall struct { + FunctionCalls []geminiFunctionCall `json:"functionCalls,omitempty"` +} + +type geminiFunctionCall struct { + ID string `json:"id,omitempty"` + Name string `json:"name,omitempty"` + Args json.RawMessage `json:"args,omitempty"` +} + +type geminiUsageMetadata struct { + PromptTokenCount int `json:"promptTokenCount,omitempty"` + ResponseTokenCount int `json:"responseTokenCount,omitempty"` + TotalTokenCount int `json:"totalTokenCount,omitempty"` +} + +// Client-sent (Bifrost -> Gemini) message shapes. + +type geminiSetupMessage struct { + Setup *geminiSetup `json:"setup"` +} + +type geminiSetup struct { + Model string `json:"model,omitempty"` + GenerationConfig json.RawMessage `json:"generationConfig,omitempty"` + SystemInstruction json.RawMessage `json:"systemInstruction,omitempty"` + Tools json.RawMessage `json:"tools,omitempty"` + // OutputAudioTranscription: top-level per BidiGenerateContentSetup's protobuf shape + // (nesting it under generationConfig, as one Google doc example for a different + // model shows, was rejected outright: "Cannot find field" on gemini-3.1-flash-live-preview). + // Confirmed live: enables a text transcript of the audio response, delivered inside + // the SAME serverContent message as the audio parts — this is what makes T->T/S->T + // achievable today despite Gemini Live having no TEXT-only responseModalities option. + OutputAudioTranscription json.RawMessage `json:"outputAudioTranscription,omitempty"` + InputAudioTranscription json.RawMessage `json:"inputAudioTranscription,omitempty"` +} + +type geminiGenerationConfig struct { + ResponseModalities []string `json:"responseModalities,omitempty"` +} + +type geminiClientContentMessage struct { + ClientContent *geminiClientContent `json:"clientContent"` +} + +type geminiClientContent struct { + Turns []geminiContent `json:"turns,omitempty"` + TurnComplete bool `json:"turnComplete"` +} + +type geminiRealtimeInputMessage struct { + RealtimeInput *geminiRealtimeInput `json:"realtimeInput"` +} + +type geminiRealtimeInput struct { + Audio *geminiInlineData `json:"audio,omitempty"` +} + +// ToBifrostRealtimeEvent converts a raw Gemini Live server message into the +// canonical Bifrost realtime envelope. +func (provider *GeminiProvider) ToBifrostRealtimeEvent(providerEvent json.RawMessage) (*schemas.BifrostRealtimeEvent, error) { + var raw geminiRealtimeServerMessage + if err := json.Unmarshal(providerEvent, &raw); err != nil { + return nil, fmt.Errorf("failed to unmarshal Gemini realtime event: %w", err) + } + + event := &schemas.BifrostRealtimeEvent{ + RawData: providerEvent, + } + + switch { + case raw.SetupComplete != nil: + event.Type = schemas.RTEventSessionCreated + event.Session = &schemas.RealtimeSession{} + + case raw.ServerContent != nil: + sc := raw.ServerContent + // Gemini's terminal serverContent message can carry turnComplete AND the last + // modelTurn/transcription chunk TOGETHER in the same frame (confirmed live) — + // content extraction must not be gated behind an early turnComplete/interrupted + // branch, or that final chunk is silently dropped from the event stream and from + // turn-output accumulation. + delta := &schemas.RealtimeDelta{} + var text strings.Builder + var audio string + if sc.ModelTurn != nil { + for _, part := range sc.ModelTurn.Parts { + if part.Text != "" { + text.WriteString(part.Text) + } + if part.InlineData != nil && strings.HasPrefix(part.InlineData.MimeType, "audio/") && audio == "" { + audio = part.InlineData.Data + } + } + } + if audio != "" { + delta.Audio = audio + } + if text.Len() > 0 { + delta.Text = text.String() + } + switch { + case sc.OutputTranscription != nil: + delta.Transcript = sc.OutputTranscription.Text + case sc.InputTranscription != nil: + delta.Transcript = sc.InputTranscription.Text + } + hasContent := audio != "" || text.Len() > 0 || delta.Transcript != "" + if hasContent { + event.Delta = delta + } + + switch { + case sc.TurnComplete: + event.Type = schemas.RTEventResponseDone + case sc.Interrupted: + event.Type = geminiEventInterrupted + case audio != "": + event.Type = schemas.RTEventResponseAudioDelta + case text.Len() > 0: + event.Type = schemas.RTEventResponseTextDelta + case sc.InputTranscription != nil: + event.Type = schemas.RTEventInputAudioTransCompleted + case sc.OutputTranscription != nil: + event.Type = schemas.RTEventResponseAudioTransDelta + default: + event.Type = schemas.RealtimeEventType("server_content") + } + + case raw.ToolCall != nil: + event.Type = geminiEventToolCall + if len(raw.ToolCall.FunctionCalls) > 0 { + call := raw.ToolCall.FunctionCalls[0] + args := "" + if len(call.Args) > 0 { + if sorted, err := providerUtils.MarshalSorted(json.RawMessage(call.Args)); err == nil { + args = string(sorted) + } else { + args = string(call.Args) + } + } + event.Item = &schemas.RealtimeItem{ + Type: "function_call", + Name: call.Name, + CallID: call.ID, + Arguments: args, + } + // Gemini supports parallel function calling — a single toolCall message can + // carry more than one functionCall, but the canonical Item shape only has room + // for one. Preserve the full list in ExtraParams (and RawData already has the + // untouched original) rather than silently dropping calls beyond the first; + // callers that need multi-call handling can read ExtraParams["function_calls"]. + if len(raw.ToolCall.FunctionCalls) > 1 { + if allCalls, err := providerUtils.MarshalSorted(raw.ToolCall.FunctionCalls); err == nil { + event.Item.ExtraParams = map[string]json.RawMessage{"function_calls": allCalls} + } + } + } + + case raw.ToolCallCancellation != nil: + event.Type = geminiEventToolCallCancellation + + case raw.GoAway != nil: + event.Type = geminiEventGoAway + + default: + event.Type = schemas.RealtimeEventType("unknown") + } + + return event, nil +} + +// extractTextFromItemContent pulls concatenated text out of a canonical +// RealtimeItem.Content payload, which mirrors OpenAI's content-part shape: +// either a plain string, or an array of {"type":"input_text","text":"..."}. +func extractTextFromItemContent(content json.RawMessage) string { + if len(content) == 0 { + return "" + } + + var asString string + if err := json.Unmarshal(content, &asString); err == nil { + return asString + } + + var parts []struct { + Text string `json:"text"` + } + if err := json.Unmarshal(content, &parts); err == nil { + var b strings.Builder + for _, p := range parts { + b.WriteString(p.Text) + } + return b.String() + } + + return "" +} + +// ToProviderRealtimeEvent converts a canonical Bifrost realtime event into +// Gemini Live's native BidiGenerateContent client-message JSON. +func (provider *GeminiProvider) ToProviderRealtimeEvent(bifrostEvent *schemas.BifrostRealtimeEvent) (json.RawMessage, error) { + switch bifrostEvent.Type { + + case schemas.RTEventSessionUpdate: + setup := &geminiSetup{} + if bifrostEvent.Session != nil { + setup.Model = toGeminiModelResourceName(bifrostEvent.Session.Model) + if bifrostEvent.Session.Instructions != "" { + if instr, err := providerUtils.MarshalSorted(map[string]interface{}{ + "parts": []map[string]string{{"text": bifrostEvent.Session.Instructions}}, + }); err == nil { + setup.SystemInstruction = instr + } + } + setup.Tools = bifrostEvent.Session.Tools + } + // Current Gemini Live models only support AUDIO output (TEXT responseModalities + // is rejected at setup time, confirmed against the live endpoint) — always request + // AUDIO and always enable outputAudioTranscription so a text delta is still + // available as a side-channel. This is what makes T->T and S->T achievable today: + // the client gets text via the transcript, audio via the delta, and can ignore + // whichever it doesn't need. + genCfg, err := providerUtils.MarshalSorted(geminiGenerationConfig{ + ResponseModalities: []string{"AUDIO"}, + }) + if err == nil { + setup.GenerationConfig = genCfg + } + setup.OutputAudioTranscription = json.RawMessage("{}") + setup.InputAudioTranscription = json.RawMessage("{}") + return providerUtils.MarshalSorted(geminiSetupMessage{Setup: setup}) + + case schemas.RTEventConversationItemCreate: + // The canonical client protocol represents a submitted tool result as a + // conversation.item.create with item.type="function_call_output" (see + // schemas.IsRealtimeToolOutputEvent) — this must translate to Gemini's + // toolResponse message, not a plain user text turn, or tool-calling round + // trips break silently (the client-sent result never reaches Gemini). + if bifrostEvent.Item != nil && bifrostEvent.Item.Type == "function_call_output" { + return providerUtils.MarshalSorted(map[string]interface{}{ + "toolResponse": buildGeminiToolResponse(bifrostEvent.Item), + }) + } + // Otherwise: Gemini has no standalone "add item" concept — buffer this as a + // non-final clientContent turn; the following response.create finalizes it. + // Role is hardcoded "user" rather than read from bifrostEvent.Item.Role: Gemini's + // Content.role only accepts "user"/"model", and function_call_output (the other + // item type this path could see) is already routed above — so every remaining + // item here is genuinely a user turn. + content := geminiContent{Role: "user"} + if bifrostEvent.Item != nil { + text := extractTextFromItemContent(bifrostEvent.Item.Content) + if text != "" { + content.Parts = append(content.Parts, geminiPart{Text: text}) + } + } + return providerUtils.MarshalSorted(geminiClientContentMessage{ + ClientContent: &geminiClientContent{Turns: []geminiContent{content}, TurnComplete: false}, + }) + + case schemas.RTEventResponseCreate: + return providerUtils.MarshalSorted(geminiClientContentMessage{ + ClientContent: &geminiClientContent{TurnComplete: true}, + }) + + case schemas.RTEventInputAudioAppend: + // Clients populate the top-level Audio field (raw bytes, base64-encoded on the + // wire) for input_audio_buffer.append; Delta.Audio is a fallback for callers that + // pre-encoded it there instead (mirrors OpenAI's dual-check). + audioB64 := "" + if len(bifrostEvent.Audio) > 0 { + audioB64 = base64.StdEncoding.EncodeToString(bifrostEvent.Audio) + } else if bifrostEvent.Delta != nil { + audioB64 = bifrostEvent.Delta.Audio + } + if audioB64 == "" { + return nil, fmt.Errorf("audio must be set for input_audio_buffer.append events") + } + return providerUtils.MarshalSorted(geminiRealtimeInputMessage{ + RealtimeInput: &geminiRealtimeInput{ + Audio: &geminiInlineData{MimeType: "audio/pcm;rate=16000", Data: audioB64}, + }, + }) + + case schemas.RTEventInputAudioCommit: + return providerUtils.MarshalSorted(geminiClientContentMessage{ + ClientContent: &geminiClientContent{TurnComplete: true}, + }) + + case geminiEventToolResponse: + if bifrostEvent.Item == nil { + return nil, nil + } + return providerUtils.MarshalSorted(map[string]interface{}{ + "toolResponse": buildGeminiToolResponse(bifrostEvent.Item), + }) + + default: + // Unmapped client event types are intentionally dropped — returning a nil + // payload here (checked by the transport before writing upstream) rather than + // an empty JSON object, since Gemini's wire protocol is strictly typed and an + // unrecognized top-level key can terminate the connection. + return nil, nil + } +} + +// buildGeminiToolResponse builds a single functionResponses entry from a canonical +// tool-output RealtimeItem (item.type="function_call_output"). Gemini requires +// "response" to be a JSON object — Item.Output is a plain string on the canonical +// schema, so it's parsed and used only when it decodes to an object; any other +// JSON shape (bare number/bool/array/null) or invalid JSON falls back to being +// wrapped, since forwarding a non-object would violate Gemini's contract. +func buildGeminiToolResponse(item *schemas.RealtimeItem) map[string]interface{} { + response := map[string]interface{}{"result": item.Output} + if item.Output != "" { + var parsed map[string]interface{} + // json.Unmarshal of the JSON literal "null" into a map succeeds with err==nil + // but leaves parsed as a nil map — guard against that too, or a literal "null" + // output would silently become {"response":null} instead of the wrapped fallback. + if err := json.Unmarshal([]byte(item.Output), &parsed); err == nil && parsed != nil { + response = parsed + } + } + return map[string]interface{}{ + "functionResponses": []map[string]interface{}{ + {"id": item.CallID, "name": item.Name, "response": response}, + }, + } +} + +// ExtractRealtimeTurnUsage parses usageMetadata from the terminal serverContent +// event into Bifrost's canonical usage shape. +func (provider *GeminiProvider) ExtractRealtimeTurnUsage(terminalEventRaw []byte) *schemas.BifrostLLMUsage { + if len(terminalEventRaw) == 0 { + return nil + } + var raw geminiRealtimeServerMessage + if err := json.Unmarshal(terminalEventRaw, &raw); err != nil || raw.UsageMetadata == nil { + return nil + } + u := raw.UsageMetadata + return &schemas.BifrostLLMUsage{ + PromptTokens: u.PromptTokenCount, + CompletionTokens: u.ResponseTokenCount, + TotalTokens: u.TotalTokenCount, + } +} + +// ExtractRealtimeTurnOutput synthesizes an assistant ChatMessage from the final +// modelTurn content, for turn-level logging. +func (provider *GeminiProvider) ExtractRealtimeTurnOutput(terminalEventRaw []byte) *schemas.ChatMessage { + if len(terminalEventRaw) == 0 { + return nil + } + var raw geminiRealtimeServerMessage + if err := json.Unmarshal(terminalEventRaw, &raw); err != nil || raw.ServerContent == nil || raw.ServerContent.ModelTurn == nil { + return nil + } + var text strings.Builder + for _, part := range raw.ServerContent.ModelTurn.Parts { + text.WriteString(part.Text) + } + if text.Len() == 0 { + return nil + } + return &schemas.ChatMessage{ + Role: schemas.ChatMessageRoleAssistant, + Content: &schemas.ChatMessageContent{ContentStr: schemas.Ptr(text.String())}, + } +} diff --git a/core/providers/gemini/realtime_test.go b/core/providers/gemini/realtime_test.go new file mode 100644 index 00000000000..11ac40d5771 --- /dev/null +++ b/core/providers/gemini/realtime_test.go @@ -0,0 +1,403 @@ +package gemini + +import ( + "encoding/json" + "strings" + "testing" + + schemas "github.com/maximhq/bifrost/core/schemas" +) + +// Fixtures below are trimmed captures from a live Gemini Live +// (BidiGenerateContent) session against models/gemini-3.1-flash-live-preview. + +func TestToBifrostRealtimeEvent_SetupComplete(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + event, err := provider.ToBifrostRealtimeEvent(json.RawMessage(`{"setupComplete": {}}`)) + if err != nil { + t.Fatalf("ToBifrostRealtimeEvent() error = %v", err) + } + if event.Type != schemas.RTEventSessionCreated { + t.Fatalf("Type = %q, want %q", event.Type, schemas.RTEventSessionCreated) + } +} + +func TestToBifrostRealtimeEvent_AudioDeltaOnly(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + raw := json.RawMessage(`{"serverContent":{"modelTurn":{"parts":[{"inlineData":{"mimeType":"audio/pcm;rate=24000","data":"AQADAAQAEgA="}}]}}}`) + event, err := provider.ToBifrostRealtimeEvent(raw) + if err != nil { + t.Fatalf("ToBifrostRealtimeEvent() error = %v", err) + } + if event.Type != schemas.RTEventResponseAudioDelta { + t.Fatalf("Type = %q, want %q", event.Type, schemas.RTEventResponseAudioDelta) + } + if event.Delta == nil || event.Delta.Audio != "AQADAAQAEgA=" { + t.Fatalf("Delta.Audio = %+v, want AQADAAQAEgA=", event.Delta) + } + if event.Delta.Transcript != "" { + t.Fatalf("Delta.Transcript = %q, want empty (no transcription in this fixture)", event.Delta.Transcript) + } +} + +// Confirmed live: Gemini bundles the audio parts AND the outputTranscription of +// that same audio into ONE serverContent message. An earlier implementation +// treated these as mutually exclusive and silently dropped the audio whenever a +// transcript was present — this test guards that regression. +func TestToBifrostRealtimeEvent_AudioDeltaWithTranscriptBundled(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + raw := json.RawMessage(`{"serverContent":{"modelTurn":{"parts":[{"inlineData":{"mimeType":"audio/pcm;rate=24000","data":"AQADAAQAEgA="}}]},"outputTranscription":{"text":"One, two,"}}}`) + event, err := provider.ToBifrostRealtimeEvent(raw) + if err != nil { + t.Fatalf("ToBifrostRealtimeEvent() error = %v", err) + } + if event.Type != schemas.RTEventResponseAudioDelta { + t.Fatalf("Type = %q, want %q", event.Type, schemas.RTEventResponseAudioDelta) + } + if event.Delta == nil || event.Delta.Audio != "AQADAAQAEgA=" { + t.Fatalf("Delta.Audio = %+v, want AQADAAQAEgA=", event.Delta) + } + if event.Delta.Transcript != "One, two," { + t.Fatalf("Delta.Transcript = %q, want %q", event.Delta.Transcript, "One, two,") + } + if !provider.ShouldAccumulateRealtimeOutput(event.Type) { + t.Fatal("ShouldAccumulateRealtimeOutput() = false, want true so the transcript still gets logged") + } +} + +func TestToBifrostRealtimeEvent_InputTranscriptionOnly(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + raw := json.RawMessage(`{"serverContent":{"inputTranscription":{"text":"hello there"}}}`) + event, err := provider.ToBifrostRealtimeEvent(raw) + if err != nil { + t.Fatalf("ToBifrostRealtimeEvent() error = %v", err) + } + if event.Type != schemas.RTEventInputAudioTransCompleted { + t.Fatalf("Type = %q, want %q", event.Type, schemas.RTEventInputAudioTransCompleted) + } + if event.Delta == nil || event.Delta.Transcript != "hello there" { + t.Fatalf("Delta.Transcript = %+v, want %q", event.Delta, "hello there") + } +} + +func TestToBifrostRealtimeEvent_TurnComplete(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + raw := json.RawMessage(`{"serverContent":{"turnComplete":true},"usageMetadata":{"promptTokenCount":145,"responseTokenCount":45,"totalTokenCount":190}}`) + event, err := provider.ToBifrostRealtimeEvent(raw) + if err != nil { + t.Fatalf("ToBifrostRealtimeEvent() error = %v", err) + } + if event.Type != schemas.RTEventResponseDone { + t.Fatalf("Type = %q, want %q", event.Type, schemas.RTEventResponseDone) + } + if event.Type != provider.RealtimeTurnFinalEvent() { + t.Fatalf("turnComplete event type must equal RealtimeTurnFinalEvent()") + } +} + +// Regression test for a bug found via codex review: the terminal serverContent +// message can carry turnComplete AND the final modelTurn chunk together in the +// SAME frame. An earlier implementation checked turnComplete first and never +// built a Delta in that case, silently dropping the last chunk. +func TestToBifrostRealtimeEvent_TurnCompleteWithBundledContent(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + raw := json.RawMessage(`{"serverContent":{"turnComplete":true,"modelTurn":{"parts":[{"text":"final words"}]},"outputTranscription":{"text":"final words"}}}`) + event, err := provider.ToBifrostRealtimeEvent(raw) + if err != nil { + t.Fatalf("ToBifrostRealtimeEvent() error = %v", err) + } + if event.Type != schemas.RTEventResponseDone { + t.Fatalf("Type = %q, want %q", event.Type, schemas.RTEventResponseDone) + } + if event.Delta == nil || event.Delta.Text != "final words" { + t.Fatalf("Delta = %+v, want Text=%q (must not be dropped just because turnComplete is also set)", event.Delta, "final words") + } + if !provider.ShouldAccumulateRealtimeOutput(event.Type) { + t.Fatal("ShouldAccumulateRealtimeOutput(RTEventResponseDone) = false, want true so this bundled final chunk still gets logged") + } +} + +// Regression test: Gemini supports parallel function calling (multiple entries +// in a single toolCall.functionCalls array) — an earlier implementation only +// ever read index 0 and silently dropped the rest. +func TestToBifrostRealtimeEvent_MultipleToolCalls(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + raw := json.RawMessage(`{"toolCall":{"functionCalls":[{"id":"call-1","name":"get_weather","args":{"city":"SF"}},{"id":"call-2","name":"get_time","args":{"tz":"UTC"}}]}}`) + event, err := provider.ToBifrostRealtimeEvent(raw) + if err != nil { + t.Fatalf("ToBifrostRealtimeEvent() error = %v", err) + } + if event.Item == nil || event.Item.Name != "get_weather" { + t.Fatalf("Item = %+v, want first call get_weather preserved", event.Item) + } + allCalls, ok := event.Item.ExtraParams["function_calls"] + if !ok { + t.Fatal("expected ExtraParams[\"function_calls\"] to preserve all calls, not just the first") + } + if !strings.Contains(string(allCalls), "get_time") { + t.Fatalf("ExtraParams[\"function_calls\"] = %s, want it to contain the second call get_time", allCalls) + } +} + +func TestToBifrostRealtimeEvent_ToolCall(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + raw := json.RawMessage(`{"toolCall":{"functionCalls":[{"id":"call-1","name":"get_weather","args":{"city":"SF"}}]}}`) + event, err := provider.ToBifrostRealtimeEvent(raw) + if err != nil { + t.Fatalf("ToBifrostRealtimeEvent() error = %v", err) + } + if event.Type != geminiEventToolCall { + t.Fatalf("Type = %q, want %q", event.Type, geminiEventToolCall) + } + if event.Item == nil || event.Item.Name != "get_weather" || event.Item.CallID != "call-1" { + t.Fatalf("Item = %+v, want function_call for get_weather/call-1", event.Item) + } +} + +func TestExtractRealtimeTurnUsage(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + raw := []byte(`{"serverContent":{"turnComplete":true},"usageMetadata":{"promptTokenCount":145,"responseTokenCount":45,"totalTokenCount":190}}`) + usage := provider.ExtractRealtimeTurnUsage(raw) + if usage == nil { + t.Fatal("ExtractRealtimeTurnUsage() = nil") + } + if usage.PromptTokens != 145 || usage.CompletionTokens != 45 || usage.TotalTokens != 190 { + t.Fatalf("usage = %+v, want {145 45 190}", usage) + } +} + +func TestExtractRealtimeTurnOutput(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + raw := []byte(`{"serverContent":{"modelTurn":{"parts":[{"text":"Hi there, hello!"}]}}}`) + msg := provider.ExtractRealtimeTurnOutput(raw) + if msg == nil { + t.Fatal("ExtractRealtimeTurnOutput() = nil") + } + if msg.Content == nil || msg.Content.ContentStr == nil || *msg.Content.ContentStr != "Hi there, hello!" { + t.Fatalf("Content = %+v, want %q", msg.Content, "Hi there, hello!") + } +} + +func TestToProviderRealtimeEvent_SessionUpdate(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + event := &schemas.BifrostRealtimeEvent{ + Type: schemas.RTEventSessionUpdate, + Session: &schemas.RealtimeSession{Model: "gemini-3.1-flash-live-preview", Instructions: "be terse"}, + } + raw, err := provider.ToProviderRealtimeEvent(event) + if err != nil { + t.Fatalf("ToProviderRealtimeEvent() error = %v", err) + } + + var msg geminiSetupMessage + if err := json.Unmarshal(raw, &msg); err != nil { + t.Fatalf("failed to unmarshal setup message: %v", err) + } + if msg.Setup == nil { + t.Fatal("Setup is nil") + } + // Model must carry the "models/" resource-name prefix Gemini requires — + // missing this causes a live "model not found" close (1008) at connect time. + if msg.Setup.Model != "models/gemini-3.1-flash-live-preview" { + t.Fatalf("Setup.Model = %q, want %q", msg.Setup.Model, "models/gemini-3.1-flash-live-preview") + } + if len(msg.Setup.OutputAudioTranscription) == 0 || len(msg.Setup.InputAudioTranscription) == 0 { + t.Fatal("expected transcription toggles to be enabled by default (this is what makes T->T/S->T achievable)") + } + + var genCfg geminiGenerationConfig + if err := json.Unmarshal(msg.Setup.GenerationConfig, &genCfg); err != nil { + t.Fatalf("failed to unmarshal generationConfig: %v", err) + } + if len(genCfg.ResponseModalities) != 1 || genCfg.ResponseModalities[0] != "AUDIO" { + t.Fatalf("ResponseModalities = %v, want [AUDIO] (current Gemini Live models reject TEXT)", genCfg.ResponseModalities) + } +} + +func TestToProviderRealtimeEvent_ResponseCreate(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + raw, err := provider.ToProviderRealtimeEvent(&schemas.BifrostRealtimeEvent{Type: schemas.RTEventResponseCreate}) + if err != nil { + t.Fatalf("ToProviderRealtimeEvent() error = %v", err) + } + var msg geminiClientContentMessage + if err := json.Unmarshal(raw, &msg); err != nil { + t.Fatalf("failed to unmarshal clientContent message: %v", err) + } + if msg.ClientContent == nil || !msg.ClientContent.TurnComplete { + t.Fatalf("ClientContent = %+v, want TurnComplete=true", msg.ClientContent) + } +} + +func TestToProviderRealtimeEvent_InputAudioAppend(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + // Client audio arrives on the top-level Audio field, not Delta.Audio — this + // was a real bug caught live: Delta-only reads silently rejected every + // input_audio_buffer.append event with "audio must be set". + event := &schemas.BifrostRealtimeEvent{Type: schemas.RTEventInputAudioAppend, Audio: []byte{0x00, 0x01, 0x02}} + raw, err := provider.ToProviderRealtimeEvent(event) + if err != nil { + t.Fatalf("ToProviderRealtimeEvent() error = %v", err) + } + var msg geminiRealtimeInputMessage + if err := json.Unmarshal(raw, &msg); err != nil { + t.Fatalf("failed to unmarshal realtimeInput message: %v", err) + } + if msg.RealtimeInput == nil || msg.RealtimeInput.Audio == nil || msg.RealtimeInput.Audio.Data == "" { + t.Fatalf("RealtimeInput = %+v, want non-empty audio data", msg.RealtimeInput) + } +} + +func TestToProviderRealtimeEvent_InputAudioAppendRequiresAudio(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + _, err := provider.ToProviderRealtimeEvent(&schemas.BifrostRealtimeEvent{Type: schemas.RTEventInputAudioAppend}) + if err == nil { + t.Fatal("expected an error when no audio is set") + } +} + +// Regression test: a client submitting a tool result via the canonical +// conversation.item.create + item.type="function_call_output" shape (see +// schemas.IsRealtimeToolOutputEvent) must translate to Gemini's toolResponse +// message. An earlier implementation only handled Gemini's own private +// "tool_response" event type, which no standard client ever produces — so +// tool-calling round trips were silently broken. +func TestToProviderRealtimeEvent_ConversationItemCreateToolOutput(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + event := &schemas.BifrostRealtimeEvent{ + Type: schemas.RTEventConversationItemCreate, + Item: &schemas.RealtimeItem{ + Type: "function_call_output", + CallID: "call-1", + Name: "get_weather", + Output: `{"tempF":72}`, + }, + } + raw, err := provider.ToProviderRealtimeEvent(event) + if err != nil { + t.Fatalf("ToProviderRealtimeEvent() error = %v", err) + } + + var msg struct { + ToolResponse struct { + FunctionResponses []struct { + ID string `json:"id"` + Name string `json:"name"` + Response json.RawMessage `json:"response"` + } `json:"functionResponses"` + } `json:"toolResponse"` + } + if err := json.Unmarshal(raw, &msg); err != nil { + t.Fatalf("failed to unmarshal toolResponse message: %v", err) + } + if len(msg.ToolResponse.FunctionResponses) != 1 { + t.Fatalf("FunctionResponses = %+v, want exactly 1 entry", msg.ToolResponse.FunctionResponses) + } + got := msg.ToolResponse.FunctionResponses[0] + if got.ID != "call-1" || got.Name != "get_weather" { + t.Fatalf("FunctionResponses[0] = %+v, want id=call-1 name=get_weather", got) + } + if !strings.Contains(string(got.Response), "72") { + t.Fatalf("Response = %s, want parsed JSON output containing 72", got.Response) + } +} + +// Confirms a plain (non-JSON) tool output string is still wrapped into an object, +// since Gemini requires "response" to be a JSON object, not a bare string. +func TestToProviderRealtimeEvent_ToolOutputPlainStringWrapped(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + event := &schemas.BifrostRealtimeEvent{ + Type: schemas.RTEventConversationItemCreate, + Item: &schemas.RealtimeItem{Type: "function_call_output", CallID: "call-1", Name: "echo", Output: "plain text result"}, + } + raw, err := provider.ToProviderRealtimeEvent(event) + if err != nil { + t.Fatalf("ToProviderRealtimeEvent() error = %v", err) + } + if !strings.Contains(string(raw), "plain text result") { + t.Fatalf("raw = %s, want it to contain the wrapped plain-text output", raw) + } +} + +// Regression test (found in round-2 review): a bare JSON value that isn't an +// object (number/bool/array/null) must also be wrapped, not forwarded as-is — +// Gemini requires functionResponse.response to be an object, and an earlier +// version accepted any successfully-parsed JSON value, including non-objects. +func TestToProviderRealtimeEvent_ToolOutputNonObjectJSONWrapped(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + for _, output := range []string{"42", "true", "[1,2,3]", "null", `"a quoted string"`} { + event := &schemas.BifrostRealtimeEvent{ + Type: schemas.RTEventConversationItemCreate, + Item: &schemas.RealtimeItem{Type: "function_call_output", CallID: "call-1", Name: "echo", Output: output}, + } + raw, err := provider.ToProviderRealtimeEvent(event) + if err != nil { + t.Fatalf("ToProviderRealtimeEvent() output=%q error = %v", output, err) + } + var msg struct { + ToolResponse struct { + FunctionResponses []struct { + Response map[string]interface{} `json:"response"` + } `json:"functionResponses"` + } `json:"toolResponse"` + } + if err := json.Unmarshal(raw, &msg); err != nil { + t.Fatalf("output=%q: failed to unmarshal: %v", output, err) + } + if len(msg.ToolResponse.FunctionResponses) != 1 || msg.ToolResponse.FunctionResponses[0].Response == nil { + t.Fatalf("output=%q: response = %+v, want a non-nil wrapped object (response field must decode as a JSON object)", output, msg.ToolResponse.FunctionResponses) + } + } +} + +// Regression test: unmapped client event types must produce a nil payload (so +// the transport skips the upstream write entirely), not a literal "{}" — an +// earlier implementation returned an empty-but-real JSON object, which the +// transport wrote straight to Gemini, risking a connection close on an +// unrecognized top-level key. +func TestToProviderRealtimeEvent_UnmappedEventReturnsNil(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + raw, err := provider.ToProviderRealtimeEvent(&schemas.BifrostRealtimeEvent{Type: schemas.RTEventResponseCancel}) + if err != nil { + t.Fatalf("ToProviderRealtimeEvent() error = %v", err) + } + if len(raw) != 0 { + t.Fatalf("raw = %s, want nil/empty for an unmapped event type", raw) + } +} diff --git a/docs/openapi/paths/inference/realtime.yaml b/docs/openapi/paths/inference/realtime.yaml index 2c734ad7aa4..69e8b000290 100644 --- a/docs/openapi/paths/inference/realtime.yaml +++ b/docs/openapi/paths/inference/realtime.yaml @@ -4,8 +4,9 @@ realtime: summary: Realtime API WebSocket description: | Opens a bidirectional WebSocket session to a realtime-capable provider - (e.g. OpenAI Realtime, Azure Realtime preview). Bifrost proxies the upstream - socket and applies governance, observability, and key selection on connect. + (e.g. OpenAI Realtime, Azure Realtime preview, Gemini Live). Bifrost proxies + the upstream socket and applies governance, observability, and key selection + on connect. The target model is provided via the `model` query parameter (or `deployment` for Azure-style routes). The OpenAI SDK sends the API key over the @@ -74,6 +75,8 @@ realtime-calls: lifetime of the session. Inference auth applies (Bearer/Basic/Virtual Key/API Key). + + Not supported for Gemini — Gemini Live has no public WebRTC SDP-exchange spec. tags: - Realtime requestBody: @@ -133,6 +136,9 @@ realtime-client-secrets: Request body must be JSON. `session.model` (or top-level `model`) must use `provider/model` form. + + Not supported for Gemini — its Ephemeral Tokens API is shaped differently + from OpenAI/Azure's `client_secrets`/`sessions` contract. tags: - Realtime requestBody: @@ -170,7 +176,7 @@ realtime-sessions: Legacy alias for the realtime client-secret minting endpoint. Behaves identically to `createRealtimeClientSecret` but uses the `sessions` route shape; provided for compatibility with older OpenAI Realtime client - libraries. + libraries. Not supported for Gemini (see `createRealtimeClientSecret`). tags: - Realtime requestBody: diff --git a/docs/providers/supported-providers/gemini.mdx b/docs/providers/supported-providers/gemini.mdx index e2c910ee761..c791f33656b 100644 --- a/docs/providers/supported-providers/gemini.mdx +++ b/docs/providers/supported-providers/gemini.mdx @@ -30,6 +30,7 @@ Google Gemini's API has different structure from OpenAI. Bifrost performs extens | Files | ✅ | - | `/upload/storage/v1beta/files` | | Batch | ✅ | - | `/v1beta/batchJobs` | | List Models | ✅ | - | `/v1beta/models` | +| Realtime (Live API) | ✅ | N/A (bidirectional) | `/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent` | --- @@ -859,6 +860,68 @@ Video Delete, List, and Remix are not supported. --- +# 12. Realtime (Live API) + +Bifrost proxies Gemini Live (`BidiGenerateContent`) through the same generic realtime WebSocket endpoint used for OpenAI/Azure Realtime — connect with `model=gemini/` and Bifrost handles translation transparently. + + + + +```js +const ws = new WebSocket( + "ws://localhost:8080/v1/realtime?model=gemini/gemini-3.1-flash-live-preview" +); + +ws.onopen = () => { + ws.send(JSON.stringify({ type: "session.update", session: { modalities: ["audio"] } })); +}; + +ws.onmessage = (event) => { + const msg = JSON.parse(event.data); + if (msg.setupComplete) { + ws.send(JSON.stringify({ + type: "conversation.item.create", + item: { type: "message", role: "user", content: [{ type: "input_text", text: "Say hello in three words." }] }, + })); + ws.send(JSON.stringify({ type: "response.create" })); + } +}; +``` + + + + +### Supported Modalities + +| Combination | Support | Notes | +|-------------|---------|-------| +| Text in → Audio out | ✅ | Native | +| Audio in → Audio out | ✅ | Native | +| Text in → Text out | ✅ | Via the `outputAudioTranscription` side-channel (see below) | +| Audio in → Text out | ✅ | Via the `outputAudioTranscription` side-channel | +| Video in | ❌ | Not supported — no canonical schema slot exists for video in Bifrost's realtime event model | + + +Current Gemini Live models only support `AUDIO` as a response modality — requesting `TEXT` is rejected at setup time. Bifrost always requests `AUDIO` output and enables `inputAudioTranscription`/`outputAudioTranscription`, so a text transcript is delivered alongside the audio response for clients that only need text. + + +### Not Supported + +- **WebRTC** (`/realtime/calls` SDP exchange) — Gemini Live has no public WebRTC SDP-exchange spec. +- **Ephemeral client secrets** (`/realtime/client_secrets`, `/realtime/sessions`) — Gemini's Ephemeral Tokens API is shaped differently from OpenAI/Azure's; not yet implemented. + +### Wire Translation + +| Gemini message | Bifrost canonical event | +|-----------------|--------------------------| +| `setup` / `setupComplete` | `session.update` / `session.created` | +| `clientContent` / `realtimeInput` | Client audio/text input | +| `serverContent.modelTurn` (+ `outputTranscription`, if present in the same message) | `response.audio.delta` / `response.text.delta`, combined into one delta | +| `serverContent.turnComplete` | `response.done` | +| `toolCall` / `toolResponse` | Function-call item, standard tool-calling flow | + +--- + ## Content Type Support Bifrost supports the following content modalities through Gemini: diff --git a/transports/bifrost-http/handlers/realtime_turn_pipeline.go b/transports/bifrost-http/handlers/realtime_turn_pipeline.go index 7286dcc6382..13ea1a248f9 100644 --- a/transports/bifrost-http/handlers/realtime_turn_pipeline.go +++ b/transports/bifrost-http/handlers/realtime_turn_pipeline.go @@ -153,7 +153,14 @@ func setRealtimeTurnStreamContext(ctx *schemas.BifrostContext, startedAt time.Ti // sanitizeRealtimeSessionEventForProvider mutates outbound session events before provider // serialization. It must not persist session state; rejected session.update events should // not affect later turn logs. -func sanitizeRealtimeSessionEventForProvider(event *schemas.BifrostRealtimeEvent) { +// +// connectionModel is the model resolved from the connection's own URL/query param. Some +// clients (notably OpenAI-style SDKs) omit session.model on session.update since it's +// redundant with the URL for providers whose connect URL carries the model (OpenAI, Azure). +// Providers whose wire protocol only learns the model from the session payload itself +// (Gemini Live's `setup.model`) would otherwise silently receive an empty model and get +// their connection rejected — fill it in from the connection when the client didn't send one. +func sanitizeRealtimeSessionEventForProvider(event *schemas.BifrostRealtimeEvent, connectionModel string) { if event == nil || event.Session == nil { return } @@ -164,6 +171,9 @@ func sanitizeRealtimeSessionEventForProvider(event *schemas.BifrostRealtimeEvent if event.Session.ExtraParams != nil { openaiProvider.StripNestedModelPrefixes(event.Session.ExtraParams) } + if event.Type == schemas.RTEventSessionUpdate && event.Session.Model == "" { + event.Session.Model = connectionModel + } } } diff --git a/transports/bifrost-http/handlers/webrtc_realtime.go b/transports/bifrost-http/handlers/webrtc_realtime.go index 36700171751..7f82ac9a4bf 100644 --- a/transports/bifrost-http/handlers/webrtc_realtime.go +++ b/transports/bifrost-http/handlers/webrtc_realtime.go @@ -841,7 +841,7 @@ func (r *webrtcRealtimeRelay) handleDownstreamMessage(msg webrtc.DataChannelMess } } - sanitizeRealtimeSessionEventForProvider(event) + sanitizeRealtimeSessionEventForProvider(event, r.model) providerEvent, err := r.provider.ToProviderRealtimeEvent(event) if err != nil { if startsTurn { @@ -869,6 +869,31 @@ func (r *webrtcRealtimeRelay) handleDownstreamMessage(msg webrtc.DataChannelMess // Track session metadata only after provider translation succeeds. Rejected // session.update events must not affect later turn logs. updateRealtimeSessionFromEvent(r.session, event) + // A nil/empty providerEvent means the provider intentionally dropped this event + // (see wsrealtime.go's identical guard for the WS relay path). Defensive: + // finalize turn hooks if this was a turn-starting event, or the turn-hooks slot + // would stay occupied for the rest of the connection (see wsrealtime.go for the + // full rationale — no provider triggers this today). + if len(providerEvent) == 0 { + if startsTurn { + if finalizeErr := finalizeRealtimeTurnHooksOnTransportError( + r.client, + r.bifrostCtx, + r.session, + r.providerKey, + r.model, + r.key, + 400, + "invalid_request_error", + "provider dropped a turn-starting event", + ); finalizeErr != nil { + r.closeWithErrorEvent(newRealtimeTurnErrorEventPayload(finalizeErr)) + return + } + r.closeWithErrorEvent(newRealtimeTurnErrorEventPayload(newRealtimeWireBifrostError(400, "invalid_request_error", "provider dropped a turn-starting event"))) + } + return + } r.sendUpstream(providerEvent, msg.IsString) } diff --git a/transports/bifrost-http/handlers/wsrealtime.go b/transports/bifrost-http/handlers/wsrealtime.go index 7f9da1a7279..b5aee2e9424 100644 --- a/transports/bifrost-http/handlers/wsrealtime.go +++ b/transports/bifrost-http/handlers/wsrealtime.go @@ -394,7 +394,7 @@ func (h *WSRealtimeHandler) relayClientToRealtimeProvider( } } - sanitizeRealtimeSessionEventForProvider(event) + sanitizeRealtimeSessionEventForProvider(event, model) providerEvent, err := provider.ToProviderRealtimeEvent(event) if err != nil { if startsTurn { @@ -431,6 +431,37 @@ func (h *WSRealtimeHandler) relayClientToRealtimeProvider( } } + // A nil/empty providerEvent means the provider intentionally dropped this + // client event (e.g. Gemini has no wire equivalent for it, and forwarding an + // empty/unrecognized payload risks the upstream closing the connection). + if len(providerEvent) == 0 { + // Defensive: no provider does this today (every ShouldStartRealtimeTurn + // trigger currently translates to a non-nil payload), but if one ever did, + // silently dropping a turn-starting event here would leave + // TryBeginRealtimeTurnHooks's turn-hooks slot occupied forever — every + // later turn on this connection would then be rejected as "already has an + // active response in progress" until the socket disconnects. Finalize with + // an error instead of leaking the slot. + if startsTurn { + if finalizeErr := finalizeRealtimeTurnHooksWithError( + h.client, + bifrostCtx, + session, + providerKey, + model, + &key, + schemas.RTEventError, + nil, + newRealtimeWireBifrostError(400, "invalid_request_error", "provider dropped a turn-starting event"), + ); finalizeErr != nil { + clientConn.writeRealtimeError(finalizeErr) + return nil + } + clientConn.writeRealtimeError(newRealtimeWireBifrostError(400, "invalid_request_error", "provider dropped a turn-starting event")) + } + continue + } + if err := upstream.WriteMessage(ws.TextMessage, providerEvent); err != nil { finalizeRealtimeTurnHooksWithError( h.client, From 5943b4dab3091081f82d1f5c891a59bcb875f8a9 Mon Sep 17 00:00:00 2001 From: Shaik-Sirajuddin Date: Tue, 7 Jul 2026 15:07:15 +0530 Subject: [PATCH 2/4] fix: address Greptile PR review findings for Gemini Live support - Redact query-string secrets (Gemini's API key) from dial-error messages in the websocket pool/connection layer, since Gemini authenticates via a URL query param instead of headers like every other realtime provider. - Concatenate multiple audio inlineData parts in a single serverContent message instead of keeping only the first. - Document the intentional WS-vs-WebRTC error-handling asymmetry for dropped turn-starting events (pre-existing convention, not something to unify here). --- core/providers/gemini/realtime.go | 16 +++++-- core/providers/gemini/realtime_test.go | 34 +++++++++++++ .../bifrost-http/handlers/wsrealtime.go | 8 ++++ .../bifrost-http/websocket/connection.go | 19 +++++++- .../websocket/connection_redact_test.go | 48 +++++++++++++++++++ transports/bifrost-http/websocket/pool.go | 2 +- 6 files changed, 121 insertions(+), 6 deletions(-) create mode 100644 transports/bifrost-http/websocket/connection_redact_test.go diff --git a/core/providers/gemini/realtime.go b/core/providers/gemini/realtime.go index 837ccb2b343..c75d28ef94e 100644 --- a/core/providers/gemini/realtime.go +++ b/core/providers/gemini/realtime.go @@ -234,18 +234,26 @@ func (provider *GeminiProvider) ToBifrostRealtimeEvent(providerEvent json.RawMes // turn-output accumulation. delta := &schemas.RealtimeDelta{} var text strings.Builder - var audio string + var audioBytes []byte if sc.ModelTurn != nil { for _, part := range sc.ModelTurn.Parts { if part.Text != "" { text.WriteString(part.Text) } - if part.InlineData != nil && strings.HasPrefix(part.InlineData.MimeType, "audio/") && audio == "" { - audio = part.InlineData.Data + // A single modelTurn can carry more than one audio part (the + // BidiGenerateContent protobuf allows it) — decode and concatenate the + // raw PCM from every audio part rather than keeping only the first, or + // any part beyond the first is silently lost with no error or log. + if part.InlineData != nil && strings.HasPrefix(part.InlineData.MimeType, "audio/") { + if decoded, err := base64.StdEncoding.DecodeString(part.InlineData.Data); err == nil { + audioBytes = append(audioBytes, decoded...) + } } } } - if audio != "" { + audio := "" + if len(audioBytes) > 0 { + audio = base64.StdEncoding.EncodeToString(audioBytes) delta.Audio = audio } if text.Len() > 0 { diff --git a/core/providers/gemini/realtime_test.go b/core/providers/gemini/realtime_test.go index 11ac40d5771..7c69285a051 100644 --- a/core/providers/gemini/realtime_test.go +++ b/core/providers/gemini/realtime_test.go @@ -1,7 +1,10 @@ package gemini import ( + "bytes" + "encoding/base64" "encoding/json" + "fmt" "strings" "testing" @@ -44,6 +47,37 @@ func TestToBifrostRealtimeEvent_AudioDeltaOnly(t *testing.T) { } } +// Regression test (found in Greptile PR review): a single modelTurn can carry +// more than one audio inlineData part — the BidiGenerateContent protobuf schema +// allows it. An earlier implementation kept only the first part and silently +// dropped the rest; parts must be decoded and concatenated instead. +func TestToBifrostRealtimeEvent_MultipleAudioPartsConcatenated(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + chunk1 := []byte{0x01, 0x02, 0x03} + chunk2 := []byte{0x04, 0x05, 0x06} + raw := json.RawMessage(fmt.Sprintf( + `{"serverContent":{"modelTurn":{"parts":[{"inlineData":{"mimeType":"audio/pcm;rate=24000","data":%q}},{"inlineData":{"mimeType":"audio/pcm;rate=24000","data":%q}}]}}}`, + base64.StdEncoding.EncodeToString(chunk1), base64.StdEncoding.EncodeToString(chunk2), + )) + event, err := provider.ToBifrostRealtimeEvent(raw) + if err != nil { + t.Fatalf("ToBifrostRealtimeEvent() error = %v", err) + } + if event.Delta == nil { + t.Fatal("Delta = nil, want combined audio from both parts") + } + got, err := base64.StdEncoding.DecodeString(event.Delta.Audio) + if err != nil { + t.Fatalf("Delta.Audio is not valid base64: %v", err) + } + want := append(append([]byte{}, chunk1...), chunk2...) + if !bytes.Equal(got, want) { + t.Fatalf("decoded audio = %v, want %v (both parts concatenated, not just the first)", got, want) + } +} + // Confirmed live: Gemini bundles the audio parts AND the outputTranscription of // that same audio into ONE serverContent message. An earlier implementation // treated these as mutually exclusive and silently dropped the audio whenever a diff --git a/transports/bifrost-http/handlers/wsrealtime.go b/transports/bifrost-http/handlers/wsrealtime.go index b5aee2e9424..608e702445c 100644 --- a/transports/bifrost-http/handlers/wsrealtime.go +++ b/transports/bifrost-http/handlers/wsrealtime.go @@ -442,6 +442,14 @@ func (h *WSRealtimeHandler) relayClientToRealtimeProvider( // later turn on this connection would then be rejected as "already has an // active response in progress" until the socket disconnects. Finalize with // an error instead of leaking the slot. + // + // This writes the error and keeps the connection open (continue below), + // matching the existing convention for the `err != nil` branch above in this + // same function — the WS relay path tolerates a bad turn-starting event and + // lets the client retry, whereas webrtc_realtime.go's equivalent branch closes + // the whole relay. That's a pre-existing per-transport difference (present in + // the ToProviderRealtimeEvent-error branch too, unrelated to this guard), not + // something to "fix" into consistency here. if startsTurn { if finalizeErr := finalizeRealtimeTurnHooksWithError( h.client, diff --git a/transports/bifrost-http/websocket/connection.go b/transports/bifrost-http/websocket/connection.go index d894e517ee7..5b383ae62a6 100644 --- a/transports/bifrost-http/websocket/connection.go +++ b/transports/bifrost-http/websocket/connection.go @@ -8,6 +8,7 @@ import ( "fmt" "net" "net/http" + "net/url" "sync" "sync/atomic" "time" @@ -16,6 +17,22 @@ import ( "github.com/maximhq/bifrost/core/schemas" ) +// redactURLForLog strips query params (and any userinfo) before a dial URL is +// placed into an error message. Most realtime providers authenticate via +// request headers, but Gemini Live's protocol requires the API key on the +// URL's `key` query parameter — without this, that key would leak in plain +// text into any error surfaced to the client or written to server logs. +func redactURLForLog(rawURL string) string { + u, err := url.Parse(rawURL) + if err != nil { + return "" + } + u.User = nil + u.RawQuery = "" + u.Fragment = "" + return u.String() +} + // UpstreamConn wraps a WebSocket connection to an upstream provider. // Thread-safe for concurrent read/write via separate mutexes. type UpstreamConn struct { @@ -256,7 +273,7 @@ func isConnectionDead(err error) bool { func DialUpstream(url string, headers http.Header, provider schemas.ModelProvider, keyID string) (*UpstreamConn, error) { wsConn, resp, err := Dial(url, headers) if err != nil { - return nil, fmt.Errorf("failed to dial upstream websocket %s: %w", url, wrapHandshakeError(resp, err)) + return nil, fmt.Errorf("failed to dial upstream websocket %s: %w", redactURLForLog(url), wrapHandshakeError(resp, err)) } return newUpstreamConn(wsConn, provider, keyID, url), nil } diff --git a/transports/bifrost-http/websocket/connection_redact_test.go b/transports/bifrost-http/websocket/connection_redact_test.go new file mode 100644 index 00000000000..4859708858f --- /dev/null +++ b/transports/bifrost-http/websocket/connection_redact_test.go @@ -0,0 +1,48 @@ +package websocket + +import "testing" + +// Regression test (found in Greptile PR review): dial-error messages must not +// leak query-string secrets. Gemini Live authenticates via a `?key=` query +// param on the dial URL (unlike OpenAI/Azure/ElevenLabs, which use headers), +// so a raw dial URL in an error message would expose the API key to clients +// and server logs. +func TestRedactURLForLog(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + in string + want string + }{ + { + name: "strips API key query param", + in: "wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent?key=AIzaSySECRET", + want: "wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent", + }, + { + name: "strips userinfo", + in: "wss://user:pass@example.com/realtime", + want: "wss://example.com/realtime", + }, + { + name: "no query param is a no-op", + in: "wss://api.openai.com/v1/realtime", + want: "wss://api.openai.com/v1/realtime", + }, + { + name: "invalid URL", + in: "://not a url", + want: "", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := redactURLForLog(tc.in) + if got != tc.want { + t.Fatalf("redactURLForLog(%q) = %q, want %q", tc.in, got, tc.want) + } + }) + } +} diff --git a/transports/bifrost-http/websocket/pool.go b/transports/bifrost-http/websocket/pool.go index 3d9e2515631..9dfd57fb7de 100644 --- a/transports/bifrost-http/websocket/pool.go +++ b/transports/bifrost-http/websocket/pool.go @@ -184,7 +184,7 @@ func (p *Pool) Close() { func (p *Pool) dial(key PoolKey, headers http.Header) (*UpstreamConn, error) { wsConn, resp, err := Dial(key.Endpoint, headers) if err != nil { - return nil, fmt.Errorf("failed to dial upstream websocket %s: %w", key.Endpoint, wrapHandshakeError(resp, err)) + return nil, fmt.Errorf("failed to dial upstream websocket %s: %w", redactURLForLog(key.Endpoint), wrapHandshakeError(resp, err)) } return newUpstreamConn(wsConn, key.Provider, key.KeyID, key.Endpoint), nil } From c26d0b9c14b5c3866b1e5b804f920aa139511570 Mon Sep 17 00:00:00 2001 From: Shaik-Sirajuddin Date: Tue, 7 Jul 2026 15:17:12 +0530 Subject: [PATCH 3/4] fix: address CodeRabbit PR review findings for Gemini Live support - Convert session.update tools through the existing Gemini tool mapper (convertBifrostToolsToGemini) instead of forwarding the canonical OpenAI-shaped tool array verbatim, which Gemini's wire protocol rejects. - Extract the duplicated finalize-turn-and-close sequence in webrtc_realtime.go into a shared helper. - Document the setup-must-be-first-frame limitation as a deliberately deferred gap (would require a RealtimeProvider interface change shared by all providers) rather than leaving it unexplained. --- core/providers/gemini/realtime.go | 25 ++++++++- core/providers/gemini/realtime_test.go | 39 ++++++++++++++ .../bifrost-http/handlers/webrtc_realtime.go | 54 +++++++++---------- 3 files changed, 87 insertions(+), 31 deletions(-) diff --git a/core/providers/gemini/realtime.go b/core/providers/gemini/realtime.go index c75d28ef94e..10abe663391 100644 --- a/core/providers/gemini/realtime.go +++ b/core/providers/gemini/realtime.go @@ -359,6 +359,16 @@ func extractTextFromItemContent(content json.RawMessage) string { // ToProviderRealtimeEvent converts a canonical Bifrost realtime event into // Gemini Live's native BidiGenerateContent client-message JSON. +// +// Known limitation (flagged in review, deliberately deferred): Gemini requires +// its `setup` message to be the connection's first frame — if a client sends +// conversation.item.create/input_audio_buffer.append/response.create before any +// session.update, this function has no way to inject a synthesized setup first, +// since translation is a pure per-event function with no connection-level state +// (the RealtimeProvider interface doesn't pass one in, and adding it would touch +// the shared contract used by OpenAI/Azure/ElevenLabs too). A compliant client — +// including Google's own SDKs — always configures the session before sending +// content, matching how every other realtime provider here is used in practice. func (provider *GeminiProvider) ToProviderRealtimeEvent(bifrostEvent *schemas.BifrostRealtimeEvent) (json.RawMessage, error) { switch bifrostEvent.Type { @@ -373,7 +383,20 @@ func (provider *GeminiProvider) ToProviderRealtimeEvent(bifrostEvent *schemas.Bi setup.SystemInstruction = instr } } - setup.Tools = bifrostEvent.Session.Tools + // Session.Tools carries the client's canonical (OpenAI-shaped) tool array + // as raw JSON — Gemini expects tools: [{functionDeclarations: [...]}], a + // different wire shape, so it must go through the same converter the + // non-realtime chat/responses paths use rather than being forwarded verbatim. + if len(bifrostEvent.Session.Tools) > 0 { + var chatTools []schemas.ChatTool + if err := json.Unmarshal(bifrostEvent.Session.Tools, &chatTools); err == nil { + if geminiTools, err := convertBifrostToolsToGemini(chatTools); err == nil && len(geminiTools) > 0 { + if toolsJSON, err := providerUtils.MarshalSorted(geminiTools); err == nil { + setup.Tools = toolsJSON + } + } + } + } } // Current Gemini Live models only support AUDIO output (TEXT responseModalities // is rejected at setup time, confirmed against the live endpoint) — always request diff --git a/core/providers/gemini/realtime_test.go b/core/providers/gemini/realtime_test.go index 7c69285a051..39ff9b8b695 100644 --- a/core/providers/gemini/realtime_test.go +++ b/core/providers/gemini/realtime_test.go @@ -270,6 +270,45 @@ func TestToProviderRealtimeEvent_SessionUpdate(t *testing.T) { } } +// Regression test (found in CodeRabbit PR review): Session.Tools carries the +// client's canonical OpenAI-shaped tool array as raw JSON. An earlier +// implementation forwarded it to Gemini verbatim, but Gemini expects +// tools: [{functionDeclarations: [...]}] — a completely different shape — so +// realtime tool-calling would have silently failed to register any tools. +func TestToProviderRealtimeEvent_SessionUpdateConvertsTools(t *testing.T) { + t.Parallel() + provider := &GeminiProvider{} + + toolsJSON := json.RawMessage(`[{"type":"function","function":{"name":"get_weather","description":"Get the weather","parameters":{"type":"object","properties":{"city":{"type":"string"}}}}}]`) + event := &schemas.BifrostRealtimeEvent{ + Type: schemas.RTEventSessionUpdate, + Session: &schemas.RealtimeSession{Model: "gemini-3.1-flash-live-preview", Tools: toolsJSON}, + } + raw, err := provider.ToProviderRealtimeEvent(event) + if err != nil { + t.Fatalf("ToProviderRealtimeEvent() error = %v", err) + } + + var msg geminiSetupMessage + if err := json.Unmarshal(raw, &msg); err != nil { + t.Fatalf("failed to unmarshal setup message: %v", err) + } + if msg.Setup == nil || len(msg.Setup.Tools) == 0 { + t.Fatal("Setup.Tools is empty, want converted Gemini tool declarations") + } + + var geminiTools []Tool + if err := json.Unmarshal(msg.Setup.Tools, &geminiTools); err != nil { + t.Fatalf("Setup.Tools is not valid Gemini tool JSON: %v", err) + } + if len(geminiTools) != 1 || len(geminiTools[0].FunctionDeclarations) != 1 { + t.Fatalf("geminiTools = %+v, want exactly 1 tool with 1 functionDeclaration", geminiTools) + } + if geminiTools[0].FunctionDeclarations[0].Name != "get_weather" { + t.Fatalf("FunctionDeclarations[0].Name = %q, want %q", geminiTools[0].FunctionDeclarations[0].Name, "get_weather") + } +} + func TestToProviderRealtimeEvent_ResponseCreate(t *testing.T) { t.Parallel() provider := &GeminiProvider{} diff --git a/transports/bifrost-http/handlers/webrtc_realtime.go b/transports/bifrost-http/handlers/webrtc_realtime.go index 7f82ac9a4bf..d0268e838d2 100644 --- a/transports/bifrost-http/handlers/webrtc_realtime.go +++ b/transports/bifrost-http/handlers/webrtc_realtime.go @@ -814,6 +814,28 @@ func (r *webrtcRealtimeRelay) forwardRTCP(sender *webrtc.RTPSender, target *webr } } +// finalizeTurnAndClose finalizes any in-progress turn hooks (recording the given +// error) and closes the relay with that same error event. Shared by every +// handleDownstreamMessage failure path that must both finalize and terminate the +// connection, so the finalize→close sequence stays in one place. +func (r *webrtcRealtimeRelay) finalizeTurnAndClose(status int, code, msg string) { + if finalizeErr := finalizeRealtimeTurnHooksOnTransportError( + r.client, + r.bifrostCtx, + r.session, + r.providerKey, + r.model, + r.key, + status, + code, + msg, + ); finalizeErr != nil { + r.closeWithErrorEvent(newRealtimeTurnErrorEventPayload(finalizeErr)) + return + } + r.closeWithErrorEvent(newRealtimeTurnErrorEventPayload(newRealtimeWireBifrostError(status, code, msg))) +} + func (r *webrtcRealtimeRelay) handleDownstreamMessage(msg webrtc.DataChannelMessage) { event, err := schemas.ParseRealtimeEvent(msg.Data) if err != nil { @@ -845,21 +867,7 @@ func (r *webrtcRealtimeRelay) handleDownstreamMessage(msg webrtc.DataChannelMess providerEvent, err := r.provider.ToProviderRealtimeEvent(event) if err != nil { if startsTurn { - if finalizeErr := finalizeRealtimeTurnHooksOnTransportError( - r.client, - r.bifrostCtx, - r.session, - r.providerKey, - r.model, - r.key, - 400, - "invalid_request_error", - err.Error(), - ); finalizeErr != nil { - r.closeWithErrorEvent(newRealtimeTurnErrorEventPayload(finalizeErr)) - return - } - r.closeWithErrorEvent(newRealtimeTurnErrorEventPayload(newRealtimeWireBifrostError(400, "invalid_request_error", err.Error()))) + r.finalizeTurnAndClose(400, "invalid_request_error", err.Error()) return } logger.Warn("failed to translate browser realtime event: %v", err) @@ -876,21 +884,7 @@ func (r *webrtcRealtimeRelay) handleDownstreamMessage(msg webrtc.DataChannelMess // full rationale — no provider triggers this today). if len(providerEvent) == 0 { if startsTurn { - if finalizeErr := finalizeRealtimeTurnHooksOnTransportError( - r.client, - r.bifrostCtx, - r.session, - r.providerKey, - r.model, - r.key, - 400, - "invalid_request_error", - "provider dropped a turn-starting event", - ); finalizeErr != nil { - r.closeWithErrorEvent(newRealtimeTurnErrorEventPayload(finalizeErr)) - return - } - r.closeWithErrorEvent(newRealtimeTurnErrorEventPayload(newRealtimeWireBifrostError(400, "invalid_request_error", "provider dropped a turn-starting event"))) + r.finalizeTurnAndClose(400, "invalid_request_error", "provider dropped a turn-starting event") } return } From 24d00fbf34e1a00ab87db011563d2169e7a5743c Mon Sep 17 00:00:00 2001 From: Shaik-Sirajuddin Date: Tue, 7 Jul 2026 16:25:03 +0530 Subject: [PATCH 4/4] fix: keep Gemini's API key out of the realtime pool's identity/log key MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the Greptile security finding — the earlier fix only redacted dial-error messages, but the raw key still lived in PoolKey.Endpoint itself: the Go map key held in memory for the pool's lifetime and stored on UpstreamConn for diagnostics. Adds PoolKey.DialURL: an optional field carrying the real (possibly credential-bearing) dial target, decoupled from Endpoint, which now stays a sanitized identity value via the new SanitizeEndpointForPoolKey helper. Only strips credential-shaped query param names (key/token/secret/auth), so OpenAI's ?model= and Azure's ?deployment= params — needed for correct per-model pool bucketing — are left untouched. Confirmed via the realtime path's existing Discard-only usage (never Return) that this can't cause a pool-key mismatch for reuse, and all existing pool/websocket/handler tests pass unmodified. --- .../bifrost-http/handlers/wsrealtime.go | 10 +++- .../bifrost-http/websocket/connection.go | 34 +++++++++++ .../websocket/connection_redact_test.go | 57 +++++++++++++++++++ transports/bifrost-http/websocket/pool.go | 22 ++++++- 4 files changed, 120 insertions(+), 3 deletions(-) diff --git a/transports/bifrost-http/handlers/wsrealtime.go b/transports/bifrost-http/handlers/wsrealtime.go index 608e702445c..c831fbce95d 100644 --- a/transports/bifrost-http/handlers/wsrealtime.go +++ b/transports/bifrost-http/handlers/wsrealtime.go @@ -303,10 +303,18 @@ func (h *WSRealtimeHandler) runRealtimeSession( clientConn.writeRealtimeError(headerErr) return } + // Some providers' RealtimeWebSocketURL embeds the credential itself (Gemini + // Live's `?key=` query param, since its protocol has no header-based auth + // option) — sanitize before it becomes the pool's Go map key / diagnostic + // value, and keep the real URL only as the dial target. This is a no-op for + // OpenAI/Azure (their `?model=`/`?deployment=` params aren't credential-shaped) + // and ElevenLabs (header auth, no query params at all). + poolEndpoint := bfws.SanitizeEndpointForPoolKey(wsURL) upstream, err := h.pool.Get(bfws.PoolKey{ Provider: providerKey, KeyID: key.ID, - Endpoint: wsURL, + Endpoint: poolEndpoint, + DialURL: wsURL, }, mapToHTTPHeader(realtimeHeaders)) if err != nil { clientConn.writeRealtimeError(newRealtimeWireBifrostError(502, "server_error", err.Error())) diff --git a/transports/bifrost-http/websocket/connection.go b/transports/bifrost-http/websocket/connection.go index 5b383ae62a6..3b0263d8a7f 100644 --- a/transports/bifrost-http/websocket/connection.go +++ b/transports/bifrost-http/websocket/connection.go @@ -9,6 +9,7 @@ import ( "net" "net/http" "net/url" + "strings" "sync" "sync/atomic" "time" @@ -33,6 +34,39 @@ func redactURLForLog(rawURL string) string { return u.String() } +// sensitiveQueryParamSubstrings names the query-parameter name fragments +// SanitizeEndpointForPoolKey strips. Deliberately name-based (not provider- +// specific): today only Gemini Live puts a credential on the URL, but this +// keeps the pool-identity path safe for any future provider that does the same, +// without needing a per-provider special case in the caller. +var sensitiveQueryParamSubstrings = []string{"key", "token", "secret", "auth"} + +// SanitizeEndpointForPoolKey strips credential-shaped query parameters from a +// dial URL before it's used as PoolKey.Endpoint — the pool's Go map key, held +// in memory for the life of every idle/in-flight connection and stored on +// UpstreamConn for diagnostics. Non-credential query params (e.g. OpenAI's +// `?model=`, needed to keep different models in separate pool buckets) are +// preserved untouched. Callers whose URL contains a stripped param must pass +// the original URL separately as PoolKey.DialURL. +func SanitizeEndpointForPoolKey(rawURL string) string { + u, err := url.Parse(rawURL) + if err != nil { + return rawURL + } + q := u.Query() + for name := range q { + lower := strings.ToLower(name) + for _, substr := range sensitiveQueryParamSubstrings { + if strings.Contains(lower, substr) { + q.Del(name) + break + } + } + } + u.RawQuery = q.Encode() + return u.String() +} + // UpstreamConn wraps a WebSocket connection to an upstream provider. // Thread-safe for concurrent read/write via separate mutexes. type UpstreamConn struct { diff --git a/transports/bifrost-http/websocket/connection_redact_test.go b/transports/bifrost-http/websocket/connection_redact_test.go index 4859708858f..ef446d97083 100644 --- a/transports/bifrost-http/websocket/connection_redact_test.go +++ b/transports/bifrost-http/websocket/connection_redact_test.go @@ -46,3 +46,60 @@ func TestRedactURLForLog(t *testing.T) { }) } } + +// Regression test (follow-up to the Greptile PR review finding): the Gemini +// API key must never become part of PoolKey.Endpoint — the Go map key held in +// memory for the pool's lifetime and stored on UpstreamConn for diagnostics — +// while non-credential query params other providers rely on for correct pool +// bucketing (OpenAI's `?model=`, Azure-style `?deployment=`) must be preserved +// untouched, or connections for different models would collapse into the same +// pool bucket. +func TestSanitizeEndpointForPoolKey(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + in string + want string + }{ + { + name: "strips Gemini API key query param", + in: "wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent?key=AIzaSySECRET", + want: "wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent", + }, + { + name: "preserves OpenAI's model query param", + in: "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview", + want: "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview", + }, + { + name: "preserves Azure's deployment query param", + in: "wss://my-resource.openai.azure.com/openai/v1/realtime?deployment=gpt-4o-realtime", + want: "wss://my-resource.openai.azure.com/openai/v1/realtime?deployment=gpt-4o-realtime", + }, + { + name: "ElevenLabs has no query-param secret, unchanged", + in: "wss://api.elevenlabs.io/v1/convai/conversation?agent_id=agent-123", + want: "wss://api.elevenlabs.io/v1/convai/conversation?agent_id=agent-123", + }, + { + name: "no query params is a no-op", + in: "wss://api.openai.com/v1/realtime", + want: "wss://api.openai.com/v1/realtime", + }, + { + name: "strips a token-named param alongside a preserved one", + in: "wss://example.com/realtime?model=foo&access_token=SECRET", + want: "wss://example.com/realtime?model=foo", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := SanitizeEndpointForPoolKey(tc.in) + if got != tc.want { + t.Fatalf("SanitizeEndpointForPoolKey(%q) = %q, want %q", tc.in, got, tc.want) + } + }) + } +} diff --git a/transports/bifrost-http/websocket/pool.go b/transports/bifrost-http/websocket/pool.go index 9dfd57fb7de..ff75e28c740 100644 --- a/transports/bifrost-http/websocket/pool.go +++ b/transports/bifrost-http/websocket/pool.go @@ -21,6 +21,16 @@ type PoolKey struct { Provider schemas.ModelProvider KeyID string Endpoint string + + // DialURL is the actual URL to connect to, when it differs from Endpoint. + // Optional — leave empty and Endpoint is used for both identity and dialing + // (true for every provider that authenticates via headers: OpenAI, Azure, + // ElevenLabs). Set this when a provider's auth must ride on the URL itself + // (Gemini Live's `?key=` query param) so Endpoint can stay a sanitized + // identity/log value — see SanitizeEndpointForPoolKey — while the real + // credential-bearing URL is only ever used for the dial itself, never as a + // Go map key held for the pool's lifetime or in any diagnostic output. + DialURL string } // Pool manages a pool of upstream WebSocket connections keyed by (provider, keyID, endpoint). @@ -181,11 +191,19 @@ func (p *Pool) Close() { // dial establishes a new WebSocket connection to the upstream endpoint // identified by key, forwarding the supplied HTTP headers during the handshake. +// Dials key.DialURL when set (the real, possibly credential-bearing URL); +// otherwise dials key.Endpoint directly, unchanged from before DialURL existed. func (p *Pool) dial(key PoolKey, headers http.Header) (*UpstreamConn, error) { - wsConn, resp, err := Dial(key.Endpoint, headers) + dialTarget := key.Endpoint + if key.DialURL != "" { + dialTarget = key.DialURL + } + wsConn, resp, err := Dial(dialTarget, headers) if err != nil { - return nil, fmt.Errorf("failed to dial upstream websocket %s: %w", redactURLForLog(key.Endpoint), wrapHandshakeError(resp, err)) + return nil, fmt.Errorf("failed to dial upstream websocket %s: %w", redactURLForLog(dialTarget), wrapHandshakeError(resp, err)) } + // UpstreamConn stores the sanitized Endpoint, not the (potentially + // credential-bearing) dialTarget — see PoolKey.DialURL's doc comment. return newUpstreamConn(wsConn, key.Provider, key.KeyID, key.Endpoint), nil }