From 4b8c081d1ad9de206749f2021ef93114583fbd39 Mon Sep 17 00:00:00 2001 From: Dan Piths <85949566+danpiths@users.noreply.github.com> Date: Fri, 8 May 2026 19:23:38 +0530 Subject: [PATCH 1/2] fix: wrap Makefile subshell cd commands in parentheses --- Makefile | 47 +++++++++++++++++++++++++++++++++++------------ 1 file changed, 35 insertions(+), 12 deletions(-) diff --git a/Makefile b/Makefile index 767e8c91c8d..11adfa9de97 100644 --- a/Makefile +++ b/Makefile @@ -153,15 +153,33 @@ install-junit-viewer: ## Install junit-viewer for HTML report generation (if not dev: install-ui install-air setup-workspace $(if $(DEBUG),install-delve) ## Start complete development environment (UI + API with proxy) @$(EXPOSE_ENV); \ - set -m; \ + set +m; \ + ui_pid=""; \ + api_pid=""; \ cleanup() { \ + $(ECHO) "$(YELLOW)[make dev] cleanup started; ui_pid=$$ui_pid api_pid=$$api_pid$(NC)"; \ trap - EXIT INT TERM HUP; \ - kill %1 %2 2>/dev/null || true; \ + for pid in "$$ui_pid" "$$api_pid"; do \ + if [ -n "$$pid" ]; then \ + children="$$(pgrep -P "$$pid" 2>/dev/null || true)"; \ + $(ECHO) "$(YELLOW)[make dev] sending TERM to pid $$pid and children: $${children:-none}$(NC)"; \ + kill -TERM $$children "$$pid" 2>/dev/null || true; \ + fi; \ + done; \ sleep 1; \ - kill -KILL %1 %2 2>/dev/null || true; \ + for pid in "$$ui_pid" "$$api_pid"; do \ + if [ -n "$$pid" ]; then \ + children="$$(pgrep -P "$$pid" 2>/dev/null || true)"; \ + $(ECHO) "$(YELLOW)[make dev] sending KILL to pid $$pid and remaining children: $${children:-none}$(NC)"; \ + kill -KILL $$children "$$pid" 2>/dev/null || true; \ + fi; \ + done; \ + $(ECHO) "$(YELLOW)[make dev] waiting for background jobs to exit...$(NC)"; \ wait 2>/dev/null || true; \ + $(ECHO) "$(GREEN)[make dev] cleanup completed.$(NC)"; \ }; \ stop_dev() { \ + $(ECHO) "$(YELLOW)[make dev] received shutdown signal; starting cleanup...$(NC)"; \ cleanup; \ exit 130; \ }; \ @@ -184,33 +202,38 @@ dev: install-ui install-air setup-workspace $(if $(DEBUG),install-delve) ## Star $(ECHO) "$(YELLOW)Starting UI development server...$(NC)"; \ $(USE_NODE); if [ -n "$(DISABLE_PROFILER)" ]; then \ $(ECHO) "$(CYAN)DevProfiler disabled for testing$(NC)"; \ - cd ui && BIFROST_DISABLE_PROFILER=1 npm run dev & \ + (cd ui && BIFROST_DISABLE_PROFILER=1 npm run dev) & \ else \ - cd ui && npm run dev & \ + (cd ui && npm run dev) & \ fi; \ + ui_pid="$$!"; \ + $(ECHO) "$(YELLOW)[make dev] UI dev server started with pid $$ui_pid$(NC)"; \ sleep 3; \ $(ECHO) "$(YELLOW)Starting API server with UI proxy...$(NC)"; \ $(MAKE) setup-workspace >/dev/null; \ if [ -n "$(DEBUG)" ]; then \ $(ECHO) "$(CYAN)Starting with air + delve debugger on port 2345...$(NC)"; \ $(ECHO) "$(YELLOW)Attach your debugger to localhost:2345$(NC)"; \ - cd transports/bifrost-http && BIFROST_UI_DEV=true air -c .air.debug.toml -- \ + (cd transports/bifrost-http && BIFROST_UI_DEV=true air -c .air.debug.toml -- \ -host "$(HOST)" \ -port "$(PORT)" \ -log-style "$(LOG_STYLE)" \ -log-level "$(LOG_LEVEL)" \ $(if $(PROMETHEUS_LABELS),-prometheus-labels "$(PROMETHEUS_LABELS)") \ - $(if $(APP_DIR),-app-dir "$(abspath $(APP_DIR))") & \ + $(if $(APP_DIR),-app-dir "$(abspath $(APP_DIR))")) & \ else \ - cd transports/bifrost-http && BIFROST_UI_DEV=true air -c .air.toml -- \ + (cd transports/bifrost-http && BIFROST_UI_DEV=true air -c .air.toml -- \ -host "$(HOST)" \ -port "$(PORT)" \ -log-style "$(LOG_STYLE)" \ -log-level "$(LOG_LEVEL)" \ $(if $(PROMETHEUS_LABELS),-prometheus-labels "$(PROMETHEUS_LABELS)") \ - $(if $(APP_DIR),-app-dir "$(abspath $(APP_DIR))") & \ + $(if $(APP_DIR),-app-dir "$(abspath $(APP_DIR))")) & \ fi; \ - while [ "$$(jobs -r | wc -l | tr -d ' ')" -eq 2 ]; do sleep 1; done; \ + api_pid="$$!"; \ + $(ECHO) "$(YELLOW)[make dev] API dev server started with pid $$api_pid$(NC)"; \ + while kill -0 "$$ui_pid" 2>/dev/null && kill -0 "$$api_pid" 2>/dev/null; do sleep 1; done; \ + $(ECHO) "$(YELLOW)[make dev] one of the dev processes exited; running cleanup...$(NC)"; \ cleanup; \ exit 1 @@ -247,9 +270,9 @@ dev-pulse: install-ui install-pulse setup-workspace $(if $(DEBUG),install-delve) $(ECHO) "$(YELLOW)Starting UI development server...$(NC)"; \ $(USE_NODE); if [ -n "$(DISABLE_PROFILER)" ]; then \ $(ECHO) "$(CYAN)DevProfiler disabled for testing$(NC)"; \ - cd ui && BIFROST_DISABLE_PROFILER=1 npm run dev & \ + (cd ui && BIFROST_DISABLE_PROFILER=1 npm run dev) & \ else \ - cd ui && npm run dev & \ + (cd ui && npm run dev) & \ fi; \ sleep 3; \ $(ECHO) "$(YELLOW)Starting API server with UI proxy...$(NC)"; \ From de6e290073db2ee6a83d1dda956319cbbfe1941c Mon Sep 17 00:00:00 2001 From: Dan Piths <85949566+danpiths@users.noreply.github.com> Date: Fri, 8 May 2026 19:24:46 +0530 Subject: [PATCH 2/2] feat: add Azure realtime provider and nested model normalization --- core/internal/llmtests/realtime.go | 5 +- core/providers/azure/realtime.go | 383 ++++++++++++++++++ core/providers/elevenlabs/realtime.go | 4 +- core/providers/openai/realtime.go | 92 ++++- core/providers/openai/realtime_test.go | 12 +- core/schemas/realtime.go | 2 +- .../bifrost-http/handlers/wsrealtime.go | 7 +- 7 files changed, 489 insertions(+), 16 deletions(-) create mode 100644 core/providers/azure/realtime.go diff --git a/core/internal/llmtests/realtime.go b/core/internal/llmtests/realtime.go index 400f5f9cda3..e024ffb8c5d 100644 --- a/core/internal/llmtests/realtime.go +++ b/core/internal/llmtests/realtime.go @@ -49,7 +49,10 @@ func RunRealtimeTest(t *testing.T, client *bifrost.Bifrost, ctx context.Context, } wsURL := rtProvider.RealtimeWebSocketURL(key, testConfig.RealtimeModel) - hdrs := rtProvider.RealtimeHeaders(key) + hdrs, headerErr := rtProvider.RealtimeHeaders(bfCtx, key) + if headerErr != nil { + t.Fatalf("failed to build realtime headers for provider %s: %v", testConfig.Provider, headerErr) + } httpHeaders := http.Header{} for k, v := range hdrs { diff --git a/core/providers/azure/realtime.go b/core/providers/azure/realtime.go new file mode 100644 index 00000000000..ae19471a00d --- /dev/null +++ b/core/providers/azure/realtime.go @@ -0,0 +1,383 @@ +package azure + +import ( + "bytes" + "encoding/json" + "fmt" + "mime/multipart" + "net/http" + "net/url" + "strings" + + openaiProvider "github.com/maximhq/bifrost/core/providers/openai" + providerUtils "github.com/maximhq/bifrost/core/providers/utils" + "github.com/maximhq/bifrost/core/schemas" + "github.com/valyala/fasthttp" +) + +// openAIEventHelper is a zero-value OpenAI provider used solely to delegate +// event conversion calls. Azure uses the exact same Realtime wire protocol as +// OpenAI, so all event parsing, serialisation, usage extraction, turn detection, +// and output extraction can be reused without modification. +var openAIEventHelper = &openaiProvider.OpenAIProvider{} + +// --------------------------------------------------------------------------- +// RealtimeProvider interface +// --------------------------------------------------------------------------- + +func (provider *AzureProvider) SupportsRealtimeAPI() bool { + return true +} + +func (provider *AzureProvider) RealtimeWebSocketURL(key schemas.Key, model string) string { + endpoint := strings.TrimRight(key.AzureKeyConfig.Endpoint.GetValue(), "/") + endpoint = strings.Replace(endpoint, "https://", "wss://", 1) + endpoint = strings.Replace(endpoint, "http://", "ws://", 1) + + apiVersion := azureRealtimeAPIVersion(key) + + return fmt.Sprintf("%s/openai/v1/realtime?model=%s&api-version=%s", + endpoint, url.QueryEscape(model), url.QueryEscape(apiVersion)) +} + +func (provider *AzureProvider) RealtimeHeaders(ctx *schemas.BifrostContext, key schemas.Key) (map[string]string, *schemas.BifrostError) { + value := key.Value.GetValue() + + // Ephemeral tokens from /client_secrets use Bearer auth. + if strings.HasPrefix(value, "ek_") { + headers := map[string]string{ + "Authorization": "Bearer " + value, + } + for k, v := range provider.networkConfig.ExtraHeaders { + headers[k] = v + } + return headers, nil + } + + headers, authErr := provider.getAzureAuthHeaders(ctx, key, false) + if authErr != nil { + return nil, authErr + } + for k, v := range provider.networkConfig.ExtraHeaders { + headers[k] = v + } + return headers, nil +} + +func (provider *AzureProvider) SupportsRealtimeWebRTC() bool { + return true +} + +func (provider *AzureProvider) ExchangeRealtimeWebRTCSDP( + ctx *schemas.BifrostContext, + key schemas.Key, + model string, + sdp string, + session json.RawMessage, +) (string, *schemas.BifrostError) { + endpoint := strings.TrimRight(key.AzureKeyConfig.Endpoint.GetValue(), "/") + apiVersion := azureRealtimeAPIVersion(key) + + upstreamURL := fmt.Sprintf("%s/openai/v1/realtime?model=%s&api-version=%s", + endpoint, url.QueryEscape(model), url.QueryEscape(apiVersion)) + + // Build multipart body: sdp + optional session + bodyBuf := &bytes.Buffer{} + writer := multipart.NewWriter(bodyBuf) + if err := writer.WriteField("sdp", sdp); err != nil { + return "", newAzureRealtimeError(fasthttp.StatusInternalServerError, "server_error", "failed to encode upstream SDP body", err) + } + if session != nil { + if err := writer.WriteField("session", string(session)); err != nil { + return "", newAzureRealtimeError(fasthttp.StatusInternalServerError, "server_error", "failed to encode upstream session body", err) + } + } + if err := writer.Close(); err != nil { + return "", newAzureRealtimeError(fasthttp.StatusInternalServerError, "server_error", "failed to finalize upstream SDP body", err) + } + + req := fasthttp.AcquireRequest() + resp := fasthttp.AcquireResponse() + defer fasthttp.ReleaseRequest(req) + defer fasthttp.ReleaseResponse(resp) + + req.SetRequestURI(upstreamURL) + req.Header.SetMethod(http.MethodPost) + req.Header.SetContentType(writer.FormDataContentType()) + + // Ephemeral tokens (ek_*) need Bearer auth; regular API keys use api-key header. + value := key.Value.GetValue() + if strings.HasPrefix(value, "ek_") { + req.Header.Set("Authorization", "Bearer "+value) + } else { + authHeaders, authErr := provider.getAzureAuthHeaders(ctx, key, false) + if authErr != nil { + return "", authErr + } + for k, v := range authHeaders { + req.Header.Set(k, v) + } + } + + for k, v := range provider.networkConfig.ExtraHeaders { + req.Header.Set(k, v) + } + req.SetBody(bodyBuf.Bytes()) + + _, bifrostErr, wait := providerUtils.MakeRequestWithContext(ctx, provider.client, req, resp) + defer wait() + if bifrostErr != nil { + return "", bifrostErr + } + + answerBody := resp.Body() + if resp.StatusCode() < fasthttp.StatusOK || resp.StatusCode() >= fasthttp.StatusMultipleChoices { + return "", provider.realtimeWebRTCUpstreamError(ctx, resp.StatusCode(), answerBody) + } + + return string(answerBody), nil +} + +// --------------------------------------------------------------------------- +// Event conversion — delegates to OpenAI (same wire protocol) +// --------------------------------------------------------------------------- + +func (provider *AzureProvider) ToBifrostRealtimeEvent(providerEvent json.RawMessage) (*schemas.BifrostRealtimeEvent, error) { + return openAIEventHelper.ToBifrostRealtimeEvent(providerEvent) +} + +func (provider *AzureProvider) ToProviderRealtimeEvent(bifrostEvent *schemas.BifrostRealtimeEvent) (json.RawMessage, error) { + return openAIEventHelper.ToProviderRealtimeEvent(bifrostEvent) +} + +// --------------------------------------------------------------------------- +// Turn lifecycle — delegates to OpenAI +// --------------------------------------------------------------------------- + +func (provider *AzureProvider) ShouldStartRealtimeTurn(event *schemas.BifrostRealtimeEvent) bool { + return openAIEventHelper.ShouldStartRealtimeTurn(event) +} + +func (provider *AzureProvider) RealtimeTurnFinalEvent() schemas.RealtimeEventType { + return openAIEventHelper.RealtimeTurnFinalEvent() +} + +func (provider *AzureProvider) ShouldForwardRealtimeEvent(event *schemas.BifrostRealtimeEvent) bool { + return true +} + +func (provider *AzureProvider) ShouldAccumulateRealtimeOutput(eventType schemas.RealtimeEventType) bool { + return openAIEventHelper.ShouldAccumulateRealtimeOutput(eventType) +} + +func (provider *AzureProvider) RealtimeWebRTCDataChannelLabel() string { + return "oai-events" +} + +func (provider *AzureProvider) RealtimeWebSocketSubprotocol() string { + return "realtime" +} + +// --------------------------------------------------------------------------- +// RealtimeUsageExtractor — delegates to OpenAI +// --------------------------------------------------------------------------- + +func (provider *AzureProvider) ExtractRealtimeTurnUsage(terminalEventRaw []byte) *schemas.BifrostLLMUsage { + return openAIEventHelper.ExtractRealtimeTurnUsage(terminalEventRaw) +} + +func (provider *AzureProvider) ExtractRealtimeTurnOutput(terminalEventRaw []byte) *schemas.ChatMessage { + return openAIEventHelper.ExtractRealtimeTurnOutput(terminalEventRaw) +} + +// --------------------------------------------------------------------------- +// RealtimeSessionProvider — client_secrets only (not legacy /sessions) +// --------------------------------------------------------------------------- + +func (provider *AzureProvider) CreateRealtimeClientSecret( + ctx *schemas.BifrostContext, + key schemas.Key, + endpointType schemas.RealtimeSessionEndpointType, + rawRequest json.RawMessage, +) (*schemas.BifrostPassthroughResponse, *schemas.BifrostError) { + // Azure does not support the legacy /sessions endpoint. + if endpointType == schemas.RealtimeSessionEndpointSessions { + return nil, &schemas.BifrostError{ + IsBifrostError: true, + StatusCode: schemas.Ptr(fasthttp.StatusBadRequest), + Error: &schemas.ErrorField{ + Type: schemas.Ptr("invalid_request_error"), + Message: "Azure does not support the legacy /sessions endpoint; use /v1/realtime/client_secrets instead", + }, + ExtraFields: schemas.BifrostErrorExtraFields{ + RequestType: schemas.RealtimeRequest, + Provider: provider.GetProviderKey(), + }, + } + } + + normalizedBody, _, bifrostErr := openaiProvider.NormalizeRealtimeClientSecretRequest(rawRequest, schemas.Azure, endpointType) + if bifrostErr != nil { + return nil, bifrostErr + } + + endpoint := strings.TrimRight(key.AzureKeyConfig.Endpoint.GetValue(), "/") + apiVersion := azureRealtimeAPIVersion(key) + upstreamURL := fmt.Sprintf("%s/openai/v1/realtime/client_secrets?api-version=%s", + endpoint, url.QueryEscape(apiVersion)) + + req := fasthttp.AcquireRequest() + resp := fasthttp.AcquireResponse() + defer fasthttp.ReleaseRequest(req) + defer fasthttp.ReleaseResponse(resp) + + req.SetRequestURI(upstreamURL) + req.Header.SetMethod(http.MethodPost) + req.Header.SetContentType("application/json") + + authHeaders, authErr := provider.getAzureAuthHeaders(ctx, key, false) + if authErr != nil { + return nil, authErr + } + for k, v := range authHeaders { + req.Header.Set(k, v) + } + for k, v := range provider.networkConfig.ExtraHeaders { + req.Header.Set(k, v) + } + req.SetBody(normalizedBody) + + latency, bifrostErr, wait := providerUtils.MakeRequestWithContext(ctx, provider.client, req, resp) + defer wait() + if bifrostErr != nil { + return nil, bifrostErr + } + + headers := providerUtils.ExtractProviderResponseHeaders(resp) + ctx.SetValue(schemas.BifrostContextKeyProviderResponseHeaders, headers) + + if resp.StatusCode() < fasthttp.StatusOK || resp.StatusCode() >= fasthttp.StatusMultipleChoices { + return nil, provider.parseRealtimeClientSecretError(ctx, resp) + } + + body, err := providerUtils.CheckAndDecodeBody(resp) + if err != nil { + return nil, providerUtils.NewBifrostOperationError("failed to decode response body", err) + } + + out := &schemas.BifrostPassthroughResponse{ + StatusCode: resp.StatusCode(), + Headers: headers, + Body: body, + ExtraFields: schemas.BifrostResponseExtraFields{ + Latency: latency.Milliseconds(), + ProviderResponseHeaders: headers, + }, + } + if providerUtils.ShouldSendBackRawRequest(ctx, provider.sendBackRawRequest) { + providerUtils.ParseAndSetRawRequestIfJSON(req, &out.ExtraFields) + } + + return out, nil +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +func (provider *AzureProvider) realtimeWebRTCUpstreamError(ctx *schemas.BifrostContext, statusCode int, body []byte) *schemas.BifrostError { + message := fmt.Sprintf("upstream realtime handshake failed for %s", provider.GetProviderKey()) + var parsed struct { + Error struct { + Message string `json:"message"` + } `json:"error"` + } + if json.Unmarshal(body, &parsed) == nil && parsed.Error.Message != "" { + message = parsed.Error.Message + } + + bifrostErr := &schemas.BifrostError{ + IsBifrostError: false, + StatusCode: schemas.Ptr(statusCode), + Error: &schemas.ErrorField{ + Type: schemas.Ptr("upstream_error"), + Message: message, + }, + ExtraFields: schemas.BifrostErrorExtraFields{ + RequestType: schemas.RealtimeRequest, + Provider: provider.GetProviderKey(), + }, + } + if providerUtils.ShouldSendBackRawResponse(ctx, provider.sendBackRawResponse) { + bifrostErr.ExtraFields.RawResponse = map[string]any{ + "status": statusCode, + "body": string(body), + } + } + return bifrostErr +} + +func newAzureRealtimeError(status int, errorType, message string, err error) *schemas.BifrostError { + bifrostErr := &schemas.BifrostError{ + IsBifrostError: true, + StatusCode: schemas.Ptr(status), + Error: &schemas.ErrorField{ + Type: schemas.Ptr(errorType), + Message: message, + }, + ExtraFields: schemas.BifrostErrorExtraFields{ + RequestType: schemas.RealtimeRequest, + Provider: schemas.Azure, + }, + } + if err != nil { + bifrostErr.Error.Error = err + } + return bifrostErr +} + +// azureRealtimeAPIVersion returns the API version to use for realtime endpoints. +// Realtime requires a preview API version. If the key has an explicit version +// configured we honour it; otherwise we fall back to the preview version rather +// than the stable default (which does not support realtime). +func azureRealtimeAPIVersion(key schemas.Key) string { + if key.AzureKeyConfig != nil && key.AzureKeyConfig.APIVersion != nil { + if apiVersion := key.AzureKeyConfig.APIVersion.GetValue(); apiVersion != "" { + return apiVersion + } + } + return AzureAPIVersionPreview +} + +func (provider *AzureProvider) parseRealtimeClientSecretError(ctx *schemas.BifrostContext, resp *fasthttp.Response) *schemas.BifrostError { + body, _ := providerUtils.CheckAndDecodeBody(resp) + var parsed struct { + Error struct { + Code string `json:"code"` + Message string `json:"message"` + } `json:"error"` + } + msg := string(body) + if json.Unmarshal(body, &parsed) == nil && parsed.Error.Message != "" { + msg = parsed.Error.Message + } + bifrostErr := &schemas.BifrostError{ + IsBifrostError: false, + StatusCode: schemas.Ptr(resp.StatusCode()), + Error: &schemas.ErrorField{ + Type: schemas.Ptr("upstream_error"), + Message: msg, + }, + ExtraFields: schemas.BifrostErrorExtraFields{ + RequestType: schemas.RealtimeRequest, + Provider: provider.GetProviderKey(), + }, + } + if providerUtils.ShouldSendBackRawResponse(ctx, provider.sendBackRawResponse) { + bifrostErr.ExtraFields.RawResponse = map[string]any{ + "status": resp.StatusCode(), + "body": string(body), + } + } + return bifrostErr +} diff --git a/core/providers/elevenlabs/realtime.go b/core/providers/elevenlabs/realtime.go index a18e1cd5143..20ef26da267 100644 --- a/core/providers/elevenlabs/realtime.go +++ b/core/providers/elevenlabs/realtime.go @@ -26,7 +26,7 @@ func (provider *ElevenlabsProvider) RealtimeWebSocketURL(key schemas.Key, model } // RealtimeHeaders returns the headers required for the ElevenLabs Conversational AI WebSocket. -func (provider *ElevenlabsProvider) RealtimeHeaders(key schemas.Key) map[string]string { +func (provider *ElevenlabsProvider) RealtimeHeaders(_ *schemas.BifrostContext, key schemas.Key) (map[string]string, *schemas.BifrostError) { headers := map[string]string{ "xi-api-key": key.Value.GetValue(), } @@ -36,7 +36,7 @@ func (provider *ElevenlabsProvider) RealtimeHeaders(key schemas.Key) map[string] } headers[k] = v } - return headers + return headers, nil } // SupportsRealtimeWebRTC returns false — ElevenLabs WebRTC SDP exchange is not yet implemented. diff --git a/core/providers/openai/realtime.go b/core/providers/openai/realtime.go index 1a2e46bf34a..65cddb4dd30 100644 --- a/core/providers/openai/realtime.go +++ b/core/providers/openai/realtime.go @@ -30,14 +30,14 @@ func (provider *OpenAIProvider) RealtimeWebSocketURL(key schemas.Key, model stri } // RealtimeHeaders returns the headers required for the OpenAI Realtime WebSocket connection. -func (provider *OpenAIProvider) RealtimeHeaders(key schemas.Key) map[string]string { +func (provider *OpenAIProvider) RealtimeHeaders(_ *schemas.BifrostContext, key schemas.Key) (map[string]string, *schemas.BifrostError) { headers := map[string]string{ "Authorization": "Bearer " + key.Value.GetValue(), } for k, v := range provider.networkConfig.ExtraHeaders { headers[k] = v } - return headers + return headers, nil } // SupportsRealtimeWebRTC reports that OpenAI supports WebRTC SDP exchange. @@ -217,7 +217,7 @@ func (provider *OpenAIProvider) CreateRealtimeClientSecret( return nil, err } - normalizedBody, _, bifrostErr := normalizeRealtimeClientSecretRequest(rawRequest, provider.GetProviderKey(), endpointType) + normalizedBody, _, bifrostErr := NormalizeRealtimeClientSecretRequest(rawRequest, provider.GetProviderKey(), endpointType) if bifrostErr != nil { return nil, bifrostErr } @@ -226,7 +226,8 @@ func (provider *OpenAIProvider) CreateRealtimeClientSecret( defer fasthttp.ReleaseRequest(req) defer fasthttp.ReleaseResponse(resp) - req.SetRequestURI(provider.buildRequestURL(ctx, realtimeSessionUpstreamPath(endpointType), schemas.RealtimeRequest)) + upstreamURL := provider.buildRequestURL(ctx, realtimeSessionUpstreamPath(endpointType), schemas.RealtimeRequest) + req.SetRequestURI(upstreamURL) req.Header.SetMethod(http.MethodPost) req.Header.SetContentType("application/json") for k, v := range provider.realtimeSessionHeaders(key, endpointType) { @@ -268,7 +269,11 @@ func (provider *OpenAIProvider) CreateRealtimeClientSecret( return out, nil } -func normalizeRealtimeClientSecretRequest( +// NormalizeRealtimeClientSecretRequest normalizes a realtime client secret request body +// by parsing the model string, resolving the provider, and restructuring the body +// to match the upstream provider's expected format. Exported for reuse by providers +// that share the same OpenAI-compatible Realtime protocol (e.g. Azure). +func NormalizeRealtimeClientSecretRequest( rawRequest json.RawMessage, defaultProvider schemas.ModelProvider, endpointType schemas.RealtimeSessionEndpointType, @@ -316,6 +321,7 @@ func normalizeRealtimeClientSecretsRequest( return nil, "", newRealtimeClientSecretError(fasthttp.StatusInternalServerError, "server_error", "failed to encode normalized model", marshalErr) } session["model"] = modelJSON + StripNestedModelPrefixes(session) if _, ok := session["type"]; !ok { typeJSON, marshalErr := json.Marshal("realtime") if marshalErr != nil { @@ -361,6 +367,7 @@ func normalizeRealtimeSessionsRequest( } root["model"] = modelJSON delete(root, "session") + StripNestedModelPrefixes(root) normalizedBody, marshalErr := json.Marshal(root) if marshalErr != nil { @@ -370,6 +377,68 @@ func normalizeRealtimeSessionsRequest( return normalizedBody, normalizedModel, nil } +// StripNestedModelPrefixes removes provider prefixes (e.g. "openai/whisper-1" → "whisper-1") +// from known nested model fields in the realtime session config. This prevents forwarding +// Bifrost-style "provider/model" strings to upstream providers that expect bare model names. +func StripNestedModelPrefixes(session map[string]json.RawMessage) { + // Old format: input_audio_transcription.model + stripModelInNestedObject(session, "input_audio_transcription") + + // New format: audio.input.transcription.model + if audioRaw, ok := session["audio"]; ok { + var audio map[string]json.RawMessage + if json.Unmarshal(audioRaw, &audio) == nil { + if inputRaw, ok := audio["input"]; ok { + var input map[string]json.RawMessage + if json.Unmarshal(inputRaw, &input) == nil { + if stripModelInNestedObject(input, "transcription") { + if updated, err := json.Marshal(input); err == nil { + audio["input"] = updated + if updatedAudio, err := json.Marshal(audio); err == nil { + session["audio"] = updatedAudio + } + } + } + } + } + } + } +} + +// stripModelInNestedObject strips the provider prefix from a "model" field inside a nested +// object at session[key]. Returns true if any change was made. +func stripModelInNestedObject(parent map[string]json.RawMessage, key string) bool { + objRaw, ok := parent[key] + if !ok || len(objRaw) == 0 || bytes.Equal(objRaw, []byte("null")) { + return false + } + var obj map[string]json.RawMessage + if json.Unmarshal(objRaw, &obj) != nil { + return false + } + modelRaw, ok := obj["model"] + if !ok { + return false + } + var modelStr string + if json.Unmarshal(modelRaw, &modelStr) != nil { + return false + } + // Strip provider prefix if present (e.g. "openai/whisper-1" → "whisper-1") + _, bareModel := schemas.ParseModelString(modelStr, "") + if bareModel == modelStr { + return false // no prefix to strip + } + if updated, err := json.Marshal(bareModel); err == nil { + obj["model"] = updated + if updatedObj, err := json.Marshal(obj); err == nil { + parent[key] = updatedObj + return true + } + } + return false +} + func (provider *OpenAIProvider) realtimeSessionHeaders( key schemas.Key, endpointType schemas.RealtimeSessionEndpointType, @@ -965,3 +1034,16 @@ func isRealtimeDeltaEvent(eventType string) bool { } return false } + +// ExtractNestedVoice digs into the new session.audio.output.voice path. +func ExtractNestedVoice(audioRaw json.RawMessage) string { + var audio struct { + Output struct { + Voice string `json:"voice"` + } `json:"output"` + } + if err := json.Unmarshal(audioRaw, &audio); err == nil && audio.Output.Voice != "" { + return audio.Output.Voice + } + return "" +} diff --git a/core/providers/openai/realtime_test.go b/core/providers/openai/realtime_test.go index 5710230b9b5..9c5d1f0d2c6 100644 --- a/core/providers/openai/realtime_test.go +++ b/core/providers/openai/realtime_test.go @@ -11,13 +11,13 @@ import ( func TestNormalizeRealtimeClientSecretRequest(t *testing.T) { t.Parallel() - body, model, bifrostErr := normalizeRealtimeClientSecretRequest( + body, model, bifrostErr := NormalizeRealtimeClientSecretRequest( json.RawMessage(`{"model":"openai/gpt-4o-realtime-preview","voice":"alloy"}`), schemas.OpenAI, schemas.RealtimeSessionEndpointClientSecrets, ) if bifrostErr != nil { - t.Fatalf("normalizeRealtimeClientSecretRequest() error = %v", bifrostErr) + t.Fatalf("NormalizeRealtimeClientSecretRequest() error = %v", bifrostErr) } if model != "gpt-4o-realtime-preview" { t.Fatalf("model = %q, want %q", model, "gpt-4o-realtime-preview") @@ -46,13 +46,13 @@ func TestNormalizeRealtimeClientSecretRequest(t *testing.T) { func TestNormalizeRealtimeClientSecretRequestUsesDefaultProvider(t *testing.T) { t.Parallel() - body, model, bifrostErr := normalizeRealtimeClientSecretRequest( + body, model, bifrostErr := NormalizeRealtimeClientSecretRequest( json.RawMessage(`{"session":{"model":"gpt-4o-realtime-preview"}}`), schemas.OpenAI, schemas.RealtimeSessionEndpointClientSecrets, ) if bifrostErr != nil { - t.Fatalf("normalizeRealtimeClientSecretRequest() error = %v", bifrostErr) + t.Fatalf("NormalizeRealtimeClientSecretRequest() error = %v", bifrostErr) } if model != "gpt-4o-realtime-preview" { t.Fatalf("model = %q, want %q", model, "gpt-4o-realtime-preview") @@ -78,13 +78,13 @@ func TestNormalizeRealtimeClientSecretRequestUsesDefaultProvider(t *testing.T) { func TestNormalizeRealtimeSessionsRequest(t *testing.T) { t.Parallel() - body, model, bifrostErr := normalizeRealtimeClientSecretRequest( + body, model, bifrostErr := NormalizeRealtimeClientSecretRequest( json.RawMessage(`{"session":{"model":"openai/gpt-4o-realtime-preview","voice":"alloy"}}`), schemas.OpenAI, schemas.RealtimeSessionEndpointSessions, ) if bifrostErr != nil { - t.Fatalf("normalizeRealtimeClientSecretRequest() error = %v", bifrostErr) + t.Fatalf("NormalizeRealtimeClientSecretRequest() error = %v", bifrostErr) } if model != "gpt-4o-realtime-preview" { t.Fatalf("model = %q, want %q", model, "gpt-4o-realtime-preview") diff --git a/core/schemas/realtime.go b/core/schemas/realtime.go index ec4fd6789d8..cb4004582a8 100644 --- a/core/schemas/realtime.go +++ b/core/schemas/realtime.go @@ -181,7 +181,7 @@ type RealtimeSessionRoute struct { type RealtimeProvider interface { SupportsRealtimeAPI() bool RealtimeWebSocketURL(key Key, model string) string - RealtimeHeaders(key Key) map[string]string + RealtimeHeaders(ctx *BifrostContext, key Key) (map[string]string, *BifrostError) // SupportsRealtimeWebRTC reports whether the provider supports WebRTC SDP exchange. SupportsRealtimeWebRTC() bool // ExchangeRealtimeWebRTCSDP performs the provider-specific SDP signaling exchange. diff --git a/transports/bifrost-http/handlers/wsrealtime.go b/transports/bifrost-http/handlers/wsrealtime.go index 6f488103afc..81aec443140 100644 --- a/transports/bifrost-http/handlers/wsrealtime.go +++ b/transports/bifrost-http/handlers/wsrealtime.go @@ -197,11 +197,16 @@ func (h *WSRealtimeHandler) runRealtimeSession( model = key.Aliases.Resolve(model) wsURL := rtProvider.RealtimeWebSocketURL(key, model) + realtimeHeaders, headerErr := rtProvider.RealtimeHeaders(bifrostCtx, key) + if headerErr != nil { + clientConn.writeRealtimeError(headerErr) + return + } upstream, err := h.pool.Get(bfws.PoolKey{ Provider: providerKey, KeyID: key.ID, Endpoint: wsURL, - }, mapToHTTPHeader(rtProvider.RealtimeHeaders(key))) + }, mapToHTTPHeader(realtimeHeaders)) if err != nil { clientConn.writeRealtimeError(newRealtimeWireBifrostError(502, "server_error", err.Error())) return