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)"; \ diff --git a/core/bifrost.go b/core/bifrost.go index 5fc49bef91b..a696c152352 100644 --- a/core/bifrost.go +++ b/core/bifrost.go @@ -4051,6 +4051,31 @@ func (bifrost *Bifrost) SelectKeyForProviderRequestType(ctx *schemas.BifrostCont return bifrost.keySelector(ctx, supportedKeys, providerKey, model) } +// ComputeRawStorageForProvider determines whether raw request/response payloads should be +// captured and stored in log records for the given provider. This is the same computation +// performed inside executeRequest (lines 5675-5713), exported for callers that bypass +// the normal inference path (e.g. realtime WebSocket/WebRTC sessions). +func (bifrost *Bifrost) ComputeRawStorageForProvider(ctx *schemas.BifrostContext, providerKey schemas.ModelProvider) bool { + if ctx == nil { + ctx = bifrost.ctx + } + if ctx == nil { + return false + } + config, err := bifrost.account.GetConfigForProvider(providerKey) + if err != nil || config == nil { + return false + } + effectiveStore := config.StoreRawRequestResponse + allowStorageOverride, _ := ctx.Value(schemas.BifrostContextKeyAllowPerRequestStorageOverride).(bool) + if allowStorageOverride { + if override, ok := ctx.Value(schemas.BifrostContextKeyStoreRawRequestResponse).(bool); ok { + effectiveStore = override + } + } + return effectiveStore +} + // WSStreamHooks holds the post-hook runner and cleanup function returned by RunStreamPreHooks. // Call PostHookRunner for each streaming chunk, setting StreamEndIndicator on the final chunk. // Call Cleanup when done to release the pipeline back to the pool. 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/bifrost.go b/core/schemas/bifrost.go index 8abd0390f03..5d60f586502 100644 --- a/core/schemas/bifrost.go +++ b/core/schemas/bifrost.go @@ -251,6 +251,8 @@ const ( BifrostContextKeyRealtimeProviderSessionID BifrostContextKey = "bifrost-realtime-provider-session-id" // string BifrostContextKeyRealtimeSource BifrostContextKey = "bifrost-realtime-source" // string ("ei" or "lm") BifrostContextKeyRealtimeEventType BifrostContextKey = "bifrost-realtime-event-type" // string + BifrostContextKeyRealtimeTransport BifrostContextKey = "bifrost-realtime-transport" // string ("websocket" or "webrtc") + BifrostContextKeyRealtimeVoice BifrostContextKey = "bifrost-realtime-voice" // string BifrostIsAsyncRequest BifrostContextKey = "bifrost-is-async-request" // bool (set by bifrost - DO NOT SET THIS MANUALLY)) - whether the request is an async request (only used in gateway) BifrostContextKeyRequestHeaders BifrostContextKey = "bifrost-request-headers" // map[string]string (all request headers with lowercased keys) BifrostContextKeyAllowPerRequestStorageOverride BifrostContextKey = "bifrost-allow-per-request-storage-override" // bool (set by transport from config — gates whether x-bf-disable-content-logging and x-bf-store-raw-request-response per-request overrides are honored) 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/framework/modelcatalog/pricing.go b/framework/modelcatalog/pricing.go index 68b6551be5a..f3c13ff907e 100644 --- a/framework/modelcatalog/pricing.go +++ b/framework/modelcatalog/pricing.go @@ -322,7 +322,7 @@ func (mc *ModelCatalog) calculateBaseCost(result *schemas.BifrostResponse, scope // Route to the appropriate compute function switch requestType { - case schemas.ChatCompletionRequest, schemas.TextCompletionRequest, schemas.ResponsesRequest: + case schemas.ChatCompletionRequest, schemas.TextCompletionRequest, schemas.ResponsesRequest, schemas.RealtimeRequest: return computeTextCost(pricing, input.usage, input.tier) case schemas.EmbeddingRequest: return computeEmbeddingCost(pricing, input.usage, input.tier) @@ -457,6 +457,7 @@ func responsesUsageToBifrostUsage(u *schemas.ResponsesResponseUsage) *schemas.Bi if u.OutputTokensDetails != nil { usage.CompletionTokensDetails = &schemas.ChatCompletionTokensDetails{ ReasoningTokens: u.OutputTokensDetails.ReasoningTokens, + AudioTokens: u.OutputTokensDetails.AudioTokens, } if u.OutputTokensDetails.NumSearchQueries != nil { usage.CompletionTokensDetails.NumSearchQueries = u.OutputTokensDetails.NumSearchQueries @@ -561,13 +562,43 @@ func computeTextCost(pricing *configstoreTables.TableModelPricing, usage *schema outputCost := float64(completionTokens) * outputRate + // Audio token cost: when token details include audio tokens, price them + // at the dedicated audio rate and subtract from the text token costs above. + // Realtime and audio-enabled chat models report audio tokens in details. + audioCost := 0.0 + inputAudioTokens := 0 + outputAudioTokens := 0 + if usage.PromptTokensDetails != nil { + inputAudioTokens = usage.PromptTokensDetails.AudioTokens + } + if usage.CompletionTokensDetails != nil { + outputAudioTokens = usage.CompletionTokensDetails.AudioTokens + } + if inputAudioTokens < 0 { + inputAudioTokens = 0 + } else if inputAudioTokens > promptTokens { + inputAudioTokens = promptTokens + } + if outputAudioTokens < 0 { + outputAudioTokens = 0 + } else if outputAudioTokens > completionTokens { + outputAudioTokens = completionTokens + } + if inputAudioTokens > 0 && pricing.InputCostPerAudioToken != nil { + // Subtract audio tokens charged at text rate, add at audio rate. + audioCost += float64(inputAudioTokens) * (*pricing.InputCostPerAudioToken - inputRate) + } + if outputAudioTokens > 0 && pricing.OutputCostPerAudioToken != nil { + audioCost += float64(outputAudioTokens) * (*pricing.OutputCostPerAudioToken - outputRate) + } + // Search query cost searchCost := 0.0 if pricing.SearchContextCostPerQuery != nil && usage.CompletionTokensDetails != nil && usage.CompletionTokensDetails.NumSearchQueries != nil { searchCost = float64(*usage.CompletionTokensDetails.NumSearchQueries) * *pricing.SearchContextCostPerQuery } - return inputCost + outputCost + searchCost + return inputCost + outputCost + audioCost + searchCost } // computeEmbeddingCost handles embedding requests (input-only). diff --git a/plugins/governance/main.go b/plugins/governance/main.go index d87dcb0e932..4374f16d098 100644 --- a/plugins/governance/main.go +++ b/plugins/governance/main.go @@ -355,8 +355,15 @@ func (p *GovernancePlugin) HTTPTransportPreHook(ctx *schemas.BifrostContext, req return nil, nil } - // If no body, check if large payload mode is active for read-only governance + // If no body, check if the request carries a model via query params (e.g. realtime + // WebSocket upgrades: GET /v1/realtime?model=... or Azure preview ?deployment=...) + // or if large payload mode is active. + // For query-param-based models we build a synthetic payload so routing rules and VK + // load-balancing can rewrite provider/model, then propagate changes back to the query. if len(req.Body) == 0 { + if modelParam := realtimeModelQueryParam(req); modelParam != "" { + return p.governRealtimeQueryParam(ctx, req, virtualKeyValue, hasRoutingRules) + } isLargePayload, _ := ctx.Value(schemas.BifrostContextKeyLargePayloadMode).(bool) if !isLargePayload { return nil, nil @@ -572,6 +579,93 @@ func (p *GovernancePlugin) governLargePayload(ctx *schemas.BifrostContext, req * return nil, nil } +// realtimeModelQueryParam returns the query parameter used as the realtime model selector. +// Azure preview realtime uses `deployment`, while GA/OpenAI-compatible paths use `model`. +func realtimeModelQueryParam(req *schemas.HTTPRequest) string { + if req == nil || req.Query == nil { + return "" + } + if modelParam := req.Query["model"]; modelParam != "" { + return modelParam + } + return req.Query["deployment"] +} + +// governRealtimeQueryParam handles governance for bodyless realtime requests +// (e.g. WebSocket upgrade GET /v1/realtime?model=... or Azure preview +// /realtime?deployment=...) where the model lives in a query parameter instead +// of the JSON body. We build a synthetic payload so routing rules and VK +// load-balancing can evaluate normally, then propagate any model rewrite back +// to the original query param for the downstream handler to pick up. +func (p *GovernancePlugin) governRealtimeQueryParam(ctx *schemas.BifrostContext, req *schemas.HTTPRequest, virtualKeyValue *string, hasRoutingRules bool) (*schemas.HTTPResponse, error) { + modelQueryKey := "model" + modelParam := req.Query[modelQueryKey] + if modelParam == "" { + modelQueryKey = "deployment" + modelParam = req.Query[modelQueryKey] + } + if modelParam == "" { + return nil, nil + } + + payload := map[string]any{ + "model": modelParam, + } + originalModel := modelParam + + // Process virtual key if provided + var virtualKey *configstoreTables.TableVirtualKey + if virtualKeyValue != nil { + vk, ok := p.store.GetVirtualKey(ctx, *virtualKeyValue) + if !ok || vk == nil || !vk.IsActiveValue() { + return nil, nil + } + virtualKey = vk + } + + // Attaching team and customer based on the virtual key + if virtualKey != nil { + if virtualKey.TeamID != nil { + ctx.SetValue(schemas.BifrostContextKeyGovernanceTeamID, *virtualKey.TeamID) + } + if virtualKey.Team != nil { + ctx.SetValue(schemas.BifrostContextKeyGovernanceTeamName, virtualKey.Team.Name) + } + if virtualKey.CustomerID != nil { + ctx.SetValue(schemas.BifrostContextKeyGovernanceCustomerID, *virtualKey.CustomerID) + } + if virtualKey.Customer != nil { + ctx.SetValue(schemas.BifrostContextKeyGovernanceCustomerName, virtualKey.Customer.Name) + } + } + + // Apply routing rules + if hasRoutingRules { + var err error + payload, _, err = p.applyRoutingRules(ctx, req, payload, virtualKey) + if err != nil { + return nil, err + } + } + + // Process virtual key: load balance provider + if virtualKey != nil { + var err error + payload, err = p.loadBalanceProvider(ctx, req, payload, virtualKey) + if err != nil { + return nil, err + } + } + + // Propagate model changes back to the original query param so the downstream + // realtime handler sees the routed/load-balanced model. + if newModel, ok := payload["model"].(string); ok && newModel != originalModel { + req.Query[modelQueryKey] = newModel + } + + return nil, nil +} + // HTTPTransportPostHook intercepts requests after they are processed (governance decision point) // It modifies the response in-place and returns nil to continue func (p *GovernancePlugin) HTTPTransportPostHook(ctx *schemas.BifrostContext, req *schemas.HTTPRequest, resp *schemas.HTTPResponse) error { diff --git a/plugins/logging/main.go b/plugins/logging/main.go index 0180ada0b46..02ecc07aef8 100644 --- a/plugins/logging/main.go +++ b/plugins/logging/main.go @@ -528,6 +528,13 @@ func (p *LoggerPlugin) PreLLMHook(ctx *schemas.BifrostContext, req *schemas.Bifr case schemas.RealtimeRequest: if req.ResponsesRequest != nil { initialData.Params = req.ResponsesRequest.Params + if req.ResponsesRequest.Params != nil { + var tools []schemas.ChatTool + for _, tool := range req.ResponsesRequest.Params.Tools { + tools = append(tools, *tool.ToChatTool()) + } + initialData.Tools = tools + } } case schemas.EmbeddingRequest: initialData.Params = req.EmbeddingRequest.Params @@ -790,11 +797,6 @@ func (p *LoggerPlugin) PostLLMHook(ctx *schemas.BifrostContext, result *schemas. } pending := pendingVal.(*PendingLogData) - if requestType == schemas.RealtimeRequest { - if resolvedRealtimeSessionID := bifrost.GetStringFromContext(ctx, schemas.BifrostContextKeyRealtimeSessionID); resolvedRealtimeSessionID != "" { - pending.ParentRequestID = resolvedRealtimeSessionID - } - } // Should never happen, but just in case // Fallback to request type from pending data if request type is not set @@ -827,6 +829,16 @@ func (p *LoggerPlugin) PostLLMHook(ctx *schemas.BifrostContext, result *schemas. } // Extract routing engine logs from context before entering goroutine routingEngineLogs := formatRoutingEngineLogs(ctx.GetRoutingEngineLogs()) + if requestType == schemas.RealtimeRequest { + if resolvedRealtimeSessionID := bifrost.GetStringFromContext(ctx, schemas.BifrostContextKeyRealtimeSessionID); resolvedRealtimeSessionID != "" { + pending.ParentRequestID = resolvedRealtimeSessionID + } + pending.InitialData.Metadata = mergeRealtimeMetadata(pending.InitialData.Metadata, ctx) + if routingEngines, ok := ctx.Value(schemas.BifrostContextKeyRoutingEnginesUsed).([]string); ok { + pending.InitialData.RoutingEngineUsed = routingEngines + pending.RoutingEnginesUsed = routingEngines + } + } // Build the complete log entry with input (from PreLLMHook) + output (from PostLLMHook) entry := buildCompleteLogEntryFromPending(pending) diff --git a/plugins/logging/utils.go b/plugins/logging/utils.go index df9da1e5738..b4a73bde1e8 100644 --- a/plugins/logging/utils.go +++ b/plugins/logging/utils.go @@ -752,6 +752,8 @@ func mergeRealtimeMetadata(metadata map[string]interface{}, ctx *schemas.Bifrost set("provider_session_id", schemas.BifrostContextKeyRealtimeProviderSessionID) set("realtime_source", schemas.BifrostContextKeyRealtimeSource) set("realtime_event_type", schemas.BifrostContextKeyRealtimeEventType) + set("realtime_transport", schemas.BifrostContextKeyRealtimeTransport) + set("realtime_voice", schemas.BifrostContextKeyRealtimeVoice) if bifrost.GetStringFromContext(ctx, schemas.BifrostContextKeyRealtimeSessionID) != "" { if metadata == nil { metadata = make(map[string]interface{}) diff --git a/transports/bifrost-http/handlers/realtime_client_secrets.go b/transports/bifrost-http/handlers/realtime_client_secrets.go index 9fe07dd61e9..6b8f680e154 100644 --- a/transports/bifrost-http/handlers/realtime_client_secrets.go +++ b/transports/bifrost-http/handlers/realtime_client_secrets.go @@ -80,12 +80,15 @@ func (h *RealtimeClientSecretsHandler) handleRequest(ctx *fasthttp.RequestCtx) { return } - providerKey, model, normalizedBody, err := resolveRealtimeClientSecretTarget(route, body) + providerKey, model, normalizedBody, err := resolveRealtimeClientSecretTarget(ctx, h.config, route, body) if err != nil { SendBifrostError(ctx, err) return } + logger.Info("[realtime-client-secrets] request: path=%s provider=%s model=%s endpoint_type=%s", + string(ctx.Path()), providerKey, model, route.EndpointType) + bifrostCtx, cancel := lib.ConvertToBifrostContext(ctx, h.handlerStore) defer cancel() bifrostCtx.SetValue(schemas.BifrostContextKeyHTTPRequestType, schemas.RealtimeRequest) @@ -150,9 +153,14 @@ func (h *RealtimeClientSecretsHandler) handleRequest(ctx *fasthttp.RequestCtx) { resp, bifrostErr := sessionProvider.CreateRealtimeClientSecret(bifrostCtx, key, route.EndpointType, normalizedBody) if bifrostErr != nil { + logger.Error("[realtime-client-secrets] upstream error: provider=%s model=%s error=%s", + providerKey, model, bifrostErr.Error) SendBifrostError(ctx, bifrostErr) return } + + logger.Info("[realtime-client-secrets] upstream success: provider=%s model=%s status=%d", + providerKey, model, resp.StatusCode) cacheRealtimeEphemeralKeyMapping( h.handlerStore.GetKVStore(), resp.Body, @@ -208,7 +216,7 @@ func (h *RealtimeClientSecretsHandler) realtimeSessionRoutes() []schemas.Realtim return routes } -func resolveRealtimeClientSecretTarget(route schemas.RealtimeSessionRoute, body []byte) (schemas.ModelProvider, string, []byte, *schemas.BifrostError) { +func resolveRealtimeClientSecretTarget(ctx *fasthttp.RequestCtx, config *lib.Config, route schemas.RealtimeSessionRoute, body []byte) (schemas.ModelProvider, string, []byte, *schemas.BifrostError) { root, err := schemas.ParseRealtimeClientSecretBody(body) if err != nil { return "", "", nil, err @@ -221,6 +229,18 @@ func resolveRealtimeClientSecretTarget(route schemas.RealtimeSessionRoute, body defaultProvider := route.DefaultProvider providerKey, model := schemas.ParseModelString(rawModel, defaultProvider) + // Model catalog auto-resolution for bare model names on /v1 client secret routes + if defaultProvider == "" && providerKey == "" && model != "" { + providers := config.GetProvidersForModel(model) + if len(providers) > 0 { + ctx.SetUserValue(lib.FastHTTPUserValueModelCatalogResolution, &lib.ModelCatalogResolution{ + Model: model, + ResolvedProvider: providers[0], + AllProviders: providers, + }) + providerKey = providers[0] + } + } if defaultProvider == "" && providerKey == "" { return "", "", nil, newRealtimeClientSecretHandlerError( fasthttp.StatusBadRequest, diff --git a/transports/bifrost-http/handlers/realtime_client_secrets_test.go b/transports/bifrost-http/handlers/realtime_client_secrets_test.go index 8029622921d..8c1b83dfa8b 100644 --- a/transports/bifrost-http/handlers/realtime_client_secrets_test.go +++ b/transports/bifrost-http/handlers/realtime_client_secrets_test.go @@ -65,7 +65,8 @@ func TestResolveRealtimeClientSecretTarget(t *testing.T) { t.Run(tt.name, func(t *testing.T) { t.Parallel() - gotProvider, gotModel, _, err := resolveRealtimeClientSecretTarget(tt.route, tt.body) + var ctx fasthttp.RequestCtx + gotProvider, gotModel, _, err := resolveRealtimeClientSecretTarget(&ctx, &lib.Config{}, tt.route, tt.body) if tt.wantErr { if err == nil { t.Fatal("expected error, got nil") @@ -118,7 +119,8 @@ func TestResolveRealtimeClientSecretTarget_NormalizesModel(t *testing.T) { t.Run(tt.name, func(t *testing.T) { t.Parallel() - _, _, normalizedBody, err := resolveRealtimeClientSecretTarget(tt.route, []byte(tt.body)) + var ctx fasthttp.RequestCtx + _, _, normalizedBody, err := resolveRealtimeClientSecretTarget(&ctx, &lib.Config{}, tt.route, []byte(tt.body)) if err != nil { t.Fatalf("unexpected error: %v", err) } diff --git a/transports/bifrost-http/handlers/realtime_logging_test.go b/transports/bifrost-http/handlers/realtime_logging_test.go index 054f2ea0e9f..b94307664dd 100644 --- a/transports/bifrost-http/handlers/realtime_logging_test.go +++ b/transports/bifrost-http/handlers/realtime_logging_test.go @@ -255,6 +255,27 @@ func TestPendingRealtimeToolOutputUpdate(t *testing.T) { } } +func TestRealtimeSessionDedupeNestedRawEvents(t *testing.T) { + t.Parallel() + + session := bfws.NewSession(nil) + firstRaw := `{"type":"conversation.item.created","item":{"id":"item_tool_123","type":"function_call_output","output":"{\"nextResponse\":\"tool result\"}"}}` + laterRaw := `{"type":"conversation.item.done","item":{"id":"item_tool_123","type":"function_call_output","output":"{\"nextResponse\":\"tool result\"}"}}` + + session.RecordRealtimeToolOutput("item_tool_123", `{"nextResponse":"tool result"}`, firstRaw) + session.RecordRealtimeToolOutput("item_tool_123", `{"nextResponse":"tool result"}`, laterRaw) + + inputs := session.ConsumeRealtimeTurnInputs() + if len(inputs) != 1 { + t.Fatalf("len(inputs) = %d, want 1", len(inputs)) + } + // Same-itemID updates replace raw with the latest event — later events + // (e.g. conversation.item.done) carry the same or more complete data. + if inputs[0].Raw != laterRaw { + t.Fatalf("Raw = %q, want latest raw event", inputs[0].Raw) + } +} + func TestBuildRealtimeTurnPostResponseUsesFullResponseDonePayload(t *testing.T) { rawRequest := `{"type":"conversation.item.input_audio_transcription.completed","transcript":""}` rawResponse := []byte(`{ @@ -314,6 +335,49 @@ func TestBuildRealtimeTurnPostResponseUsesFullResponseDonePayload(t *testing.T) } } +func TestBuildRealtimeTurnPostResponseMergesTextAndToolCalls(t *testing.T) { + rawResponse := []byte(`{ + "type":"response.done", + "response":{ + "output":[ + { + "id":"item_message_123", + "type":"message", + "content":[{"type":"text","text":"assistant text"}] + }, + { + "id":"item_call_123", + "type":"function_call", + "call_id":"call_123", + "name":"lookup_weather", + "arguments":"{\"city\":\"SF\"}" + } + ] + } + }`) + + resp := buildRealtimeTurnPostResponse(&openai.OpenAIProvider{}, schemas.OpenAI, "gpt-realtime", "", rawResponse, "", 123) + if resp == nil || resp.ResponsesResponse == nil { + t.Fatal("expected realtime post response") + } + if len(resp.ResponsesResponse.Output) != 2 { + t.Fatalf("len(Output) = %d, want 2", len(resp.ResponsesResponse.Output)) + } + if resp.ResponsesResponse.Output[0].Type == nil || *resp.ResponsesResponse.Output[0].Type != schemas.ResponsesMessageTypeMessage { + t.Fatalf("Output[0].Type = %#v, want message", resp.ResponsesResponse.Output[0].Type) + } + toolOutput := resp.ResponsesResponse.Output[1] + if toolOutput.Type == nil || *toolOutput.Type != schemas.ResponsesMessageTypeFunctionCall { + t.Fatalf("Output[1].Type = %#v, want function_call", toolOutput.Type) + } + if toolOutput.ResponsesToolMessage == nil || toolOutput.ResponsesToolMessage.Name == nil || *toolOutput.ResponsesToolMessage.Name != "lookup_weather" { + t.Fatalf("tool name = %#v, want lookup_weather", toolOutput.ResponsesToolMessage) + } + if toolOutput.CallID == nil || *toolOutput.CallID != "call_123" { + t.Fatalf("CallID = %#v, want call_123", toolOutput.CallID) + } +} + func TestFinalizeRealtimeTurnHooksWithErrorCompletesActiveHooks(t *testing.T) { t.Parallel() diff --git a/transports/bifrost-http/handlers/realtime_turn_pipeline.go b/transports/bifrost-http/handlers/realtime_turn_pipeline.go index 91095e5843f..947185ee97d 100644 --- a/transports/bifrost-http/handlers/realtime_turn_pipeline.go +++ b/transports/bifrost-http/handlers/realtime_turn_pipeline.go @@ -9,6 +9,7 @@ import ( "github.com/google/uuid" bifrost "github.com/maximhq/bifrost/core" + openaiProvider "github.com/maximhq/bifrost/core/providers/openai" "github.com/maximhq/bifrost/core/schemas" bfws "github.com/maximhq/bifrost/transports/bifrost-http/websocket" ) @@ -71,6 +72,29 @@ func newRealtimeTurnContext( return ctx } +func applyRealtimeRawStorageContext(ctx *schemas.BifrostContext, storeRaw bool) { + if ctx == nil { + return + } + // Realtime turn logging captures raw payloads only for log storage. There is + // no client-facing raw send-back path for synthetic realtime turn responses. + sendBackRawRequest := false + sendBackRawResponse := false + ctx.SetValue(schemas.BifrostContextKeyShouldStoreRawInLogs, storeRaw) + ctx.SetValue(schemas.BifrostContextKeyCaptureRawRequest, storeRaw || sendBackRawRequest) + ctx.SetValue(schemas.BifrostContextKeyCaptureRawResponse, storeRaw || sendBackRawResponse) + ctx.SetValue(schemas.BifrostContextKeyDropRawRequestFromClient, storeRaw && !sendBackRawRequest) + ctx.SetValue(schemas.BifrostContextKeyDropRawResponseFromClient, storeRaw && !sendBackRawResponse) +} + +func shouldStoreRealtimeRawPayloads(ctx *schemas.BifrostContext) bool { + if ctx == nil { + return false + } + storeRaw, _ := ctx.Value(schemas.BifrostContextKeyShouldStoreRawInLogs).(bool) + return storeRaw +} + func applyRealtimeTurnContextValues(ctx *schemas.BifrostContext, values map[any]any) { if ctx == nil || len(values) == 0 { return @@ -93,6 +117,18 @@ func applyRealtimeTurnContextValues(ctx *schemas.BifrostContext, values map[any] } } +func restoreRealtimeTurnTraceContext(ctx *schemas.BifrostContext, traceID string, values map[any]any) { + if ctx == nil { + return + } + if strings.TrimSpace(traceID) != "" { + ctx.SetValue(schemas.BifrostContextKeyTraceID, strings.TrimSpace(traceID)) + } + if tracer, ok := values[schemas.BifrostContextKeyTracer].(schemas.Tracer); ok && tracer != nil { + ctx.SetValue(schemas.BifrostContextKeyTracer, tracer) + } +} + func setRealtimeTurnStreamContext(ctx *schemas.BifrostContext, startedAt time.Time, isFinal bool) { if ctx == nil { return @@ -106,7 +142,52 @@ func setRealtimeTurnStreamContext(ctx *schemas.BifrostContext, startedAt time.Ti } } -func buildRealtimeTurnPreRequest(provider schemas.ModelProvider, model string, turnInputs []bfws.RealtimeTurnInput) *schemas.BifrostRequest { +// 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) { + if event == nil || event.Session == nil { + return + } + switch event.Type { + case schemas.RTEventSessionUpdate, + schemas.RTEventSessionCreated, + schemas.RTEventSessionUpdated: + if event.Session.ExtraParams != nil { + openaiProvider.StripNestedModelPrefixes(event.Session.ExtraParams) + } + } +} + +// updateRealtimeSessionFromEvent updates the session's tracked tool +// definitions and voice whenever a session.update, session.created, or +// session.updated event carries them. +func updateRealtimeSessionFromEvent(session *bfws.Session, event *schemas.BifrostRealtimeEvent) { + if event == nil || event.Session == nil { + return + } + switch event.Type { + case schemas.RTEventSessionUpdate, + schemas.RTEventSessionCreated, + schemas.RTEventSessionUpdated: + // Only update if the event explicitly carries tools (even an empty array + // means "clear tools"). A nil/absent tools field means "not changed". + if event.Session.Tools != nil { + session.SetRealtimeSessionTools(event.Session.Tools) + } + if event.Session.Voice != "" { + session.SetRealtimeVoice(event.Session.Voice) + } else if audioRaw, ok := event.Session.ExtraParams["audio"]; ok { + // New API format nests voice under session.audio.output.voice + // instead of the legacy top-level session.voice. + if voice := openaiProvider.ExtractNestedVoice(audioRaw); voice != "" { + session.SetRealtimeVoice(voice) + } + } + } +} + +func buildRealtimeTurnPreRequest(provider schemas.ModelProvider, model string, turnInputs []bfws.RealtimeTurnInput, sessionTools json.RawMessage) *schemas.BifrostRequest { input := make([]schemas.ResponsesMessage, 0, len(turnInputs)) for _, turnInput := range turnInputs { summary := strings.TrimSpace(turnInput.Summary) @@ -134,12 +215,21 @@ func buildRealtimeTurnPreRequest(provider schemas.ModelProvider, model string, t } } + var params *schemas.ResponsesParameters + if len(sessionTools) > 0 { + var tools []schemas.ResponsesTool + if json.Unmarshal(sessionTools, &tools) == nil && len(tools) > 0 { + params = &schemas.ResponsesParameters{Tools: tools} + } + } + return &schemas.BifrostRequest{ RequestType: schemas.RealtimeRequest, ResponsesRequest: &schemas.BifrostResponsesRequest{ Provider: provider, Model: model, Input: input, + Params: params, }, } } @@ -180,12 +270,15 @@ func buildRealtimeTurnPostResponse( func buildRealtimeTurnOutputMessages(rtProvider schemas.RealtimeProvider, rawResponse []byte, contentOverride string) []schemas.ResponsesMessage { outputs := make([]schemas.ResponsesMessage, 0) + seenFunctionCalls := make(map[string]struct{}) if outputMessage := extractRealtimeTurnOutputMessage(rtProvider, rawResponse, contentOverride); outputMessage != nil { outputs = append(outputs, buildRealtimeResponsesMessagesFromChat(outputMessage, contentOverride)...) - } - - if len(outputs) > 0 { - return outputs + for _, output := range outputs { + if output.Type == nil || *output.Type != schemas.ResponsesMessageTypeFunctionCall { + continue + } + seenFunctionCalls[realtimeResponsesFunctionCallKey(output)] = struct{}{} + } } var parsed realtimeResponseDoneEnvelope @@ -193,6 +286,9 @@ func buildRealtimeTurnOutputMessages(rtProvider schemas.RealtimeProvider, rawRes for _, item := range parsed.Response.Output { switch item.Type { case "message": + if realtimeOutputsContainMessage(outputs) { + continue + } content := strings.TrimSpace(contentOverride) if content == "" { content = extractRealtimeResponseDoneContentText(item.Content) @@ -227,6 +323,11 @@ func buildRealtimeTurnOutputMessages(rtProvider schemas.RealtimeProvider, rawRes if strings.TrimSpace(item.CallID) != "" { msg.CallID = schemas.Ptr(strings.TrimSpace(item.CallID)) } + key := realtimeResponsesFunctionCallKey(msg) + if _, exists := seenFunctionCalls[key]; exists { + continue + } + seenFunctionCalls[key] = struct{}{} outputs = append(outputs, msg) } } @@ -246,6 +347,35 @@ func buildRealtimeTurnOutputMessages(rtProvider schemas.RealtimeProvider, rawRes return outputs } +func realtimeOutputsContainMessage(outputs []schemas.ResponsesMessage) bool { + for _, output := range outputs { + if output.Type != nil && *output.Type == schemas.ResponsesMessageTypeMessage { + return true + } + } + return false +} + +func realtimeResponsesFunctionCallKey(message schemas.ResponsesMessage) string { + if message.CallID != nil && strings.TrimSpace(*message.CallID) != "" { + return "call_id:" + strings.TrimSpace(*message.CallID) + } + if message.ID != nil && strings.TrimSpace(*message.ID) != "" { + return "id:" + strings.TrimSpace(*message.ID) + } + + var parts []string + if message.ResponsesToolMessage != nil { + if message.ResponsesToolMessage.Name != nil { + parts = append(parts, strings.TrimSpace(*message.ResponsesToolMessage.Name)) + } + if message.ResponsesToolMessage.Arguments != nil { + parts = append(parts, strings.TrimSpace(*message.ResponsesToolMessage.Arguments)) + } + } + return strings.Join(parts, "\x00") +} + func buildRealtimeResponsesMessagesFromChat(message *schemas.ChatMessage, contentOverride string) []schemas.ResponsesMessage { if message == nil { return nil @@ -488,9 +618,14 @@ func startRealtimeTurnHooks( }() startedAt := time.Now() + storeRaw := shouldStoreRealtimeRawPayloads(baseCtx) turnCtx := newRealtimeTurnContext(baseCtx, "", session.ID(), session.ProviderSessionID(), realtimeTurnSourceEI, startEventType, key) + applyRealtimeRawStorageContext(turnCtx, storeRaw) + if voice := session.RealtimeVoice(); voice != "" { + turnCtx.SetValue(schemas.BifrostContextKeyRealtimeVoice, voice) + } setRealtimeTurnStreamContext(turnCtx, startedAt, false) - req := buildRealtimeTurnPreRequest(provider, model, session.PeekRealtimeTurnInputs()) + req := buildRealtimeTurnPreRequest(provider, model, session.PeekRealtimeTurnInputs(), session.RealtimeSessionTools()) hooks, bifrostErr := client.RunRealtimeTurnPreHooks(turnCtx, req) if bifrostErr != nil { // RunRealtimeTurnPreHooks already executed post-hooks and flushed the trace @@ -502,12 +637,15 @@ func startRealtimeTurnHooks( } requestID, _ := turnCtx.Value(schemas.BifrostContextKeyRequestID).(string) + traceID, _ := turnCtx.Value(schemas.BifrostContextKeyTraceID).(string) session.SetRealtimeTurnHooks(&bfws.RealtimeTurnPluginState{ PostHookRunner: hooks.PostHookRunner, Cleanup: hooks.Cleanup, RequestID: requestID, StartedAt: startedAt, PreHookValues: turnCtx.GetUserValues(), + TraceID: traceID, + RawStore: storeRaw, }) committed = true return nil @@ -548,6 +686,8 @@ func finalizeRealtimeTurnHooks( ) postCtx := newRealtimeTurnContext(baseCtx, activeHooks.RequestID, session.ID(), session.ProviderSessionID(), realtimeTurnSourceLM, rtProvider.RealtimeTurnFinalEvent(), key) applyRealtimeTurnContextValues(postCtx, activeHooks.PreHookValues) + restoreRealtimeTurnTraceContext(postCtx, activeHooks.TraceID, activeHooks.PreHookValues) + applyRealtimeRawStorageContext(postCtx, activeHooks.RawStore) setRealtimeTurnStreamContext(postCtx, activeHooks.StartedAt, true) _, bifrostErr := activeHooks.PostHookRunner(postCtx, postResponse, nil) completeRealtimeTurnTrace(postCtx) @@ -555,18 +695,22 @@ func finalizeRealtimeTurnHooks( } startedAt := time.Now() + storeRaw := shouldStoreRealtimeRawPayloads(baseCtx) preCtx := newRealtimeTurnContext(baseCtx, "", session.ID(), session.ProviderSessionID(), realtimeTurnSourceEI, "", key) + applyRealtimeRawStorageContext(preCtx, storeRaw) setRealtimeTurnStreamContext(preCtx, startedAt, false) - preReq := buildRealtimeTurnPreRequest(provider, model, turnInputs) + preReq := buildRealtimeTurnPreRequest(provider, model, turnInputs, session.RealtimeSessionTools()) hooks, bifrostErr := client.RunRealtimeTurnPreHooks(preCtx, preReq) if bifrostErr != nil { return bifrostErr } + preHookValues := preCtx.GetUserValues() if hooks.Cleanup != nil { defer hooks.Cleanup() } requestID, _ := preCtx.Value(schemas.BifrostContextKeyRequestID).(string) + traceID, _ := preCtx.Value(schemas.BifrostContextKeyTraceID).(string) postResponse := buildRealtimeTurnPostResponse( rtProvider, provider, @@ -577,7 +721,9 @@ func finalizeRealtimeTurnHooks( time.Since(startedAt).Milliseconds(), ) postCtx := newRealtimeTurnContext(baseCtx, requestID, session.ID(), session.ProviderSessionID(), realtimeTurnSourceLM, rtProvider.RealtimeTurnFinalEvent(), key) - applyRealtimeTurnContextValues(postCtx, preCtx.GetUserValues()) + applyRealtimeTurnContextValues(postCtx, preHookValues) + restoreRealtimeTurnTraceContext(postCtx, traceID, preHookValues) + applyRealtimeRawStorageContext(postCtx, storeRaw) setRealtimeTurnStreamContext(postCtx, startedAt, true) _, bifrostErr = hooks.PostHookRunner(postCtx, postResponse, nil) completeRealtimeTurnTrace(postCtx) @@ -618,6 +764,8 @@ func finalizeRealtimeTurnHooksWithError( ) postCtx := newRealtimeTurnContext(baseCtx, activeHooks.RequestID, session.ID(), session.ProviderSessionID(), realtimeTurnSourceLM, eventType, key) applyRealtimeTurnContextValues(postCtx, activeHooks.PreHookValues) + restoreRealtimeTurnTraceContext(postCtx, activeHooks.TraceID, activeHooks.PreHookValues) + applyRealtimeRawStorageContext(postCtx, activeHooks.RawStore) setRealtimeTurnStreamContext(postCtx, activeHooks.StartedAt, true) _, hookErr := activeHooks.PostHookRunner(postCtx, nil, postErr) completeRealtimeTurnTrace(postCtx) @@ -633,18 +781,22 @@ func finalizeRealtimeTurnHooksWithError( } startedAt := time.Now() + storeRaw := shouldStoreRealtimeRawPayloads(baseCtx) preCtx := newRealtimeTurnContext(baseCtx, "", session.ID(), session.ProviderSessionID(), realtimeTurnSourceEI, "", key) + applyRealtimeRawStorageContext(preCtx, storeRaw) setRealtimeTurnStreamContext(preCtx, startedAt, false) - preReq := buildRealtimeTurnPreRequest(provider, model, turnInputs) + preReq := buildRealtimeTurnPreRequest(provider, model, turnInputs, session.RealtimeSessionTools()) hooks, hookPreErr := client.RunRealtimeTurnPreHooks(preCtx, preReq) if hookPreErr != nil { return hookPreErr } + preHookValues := preCtx.GetUserValues() if hooks.Cleanup != nil { defer hooks.Cleanup() } requestID, _ := preCtx.Value(schemas.BifrostContextKeyRequestID).(string) + traceID, _ := preCtx.Value(schemas.BifrostContextKeyTraceID).(string) postErr := buildRealtimeTurnPostError( provider, model, @@ -653,7 +805,9 @@ func finalizeRealtimeTurnHooksWithError( bifrostErr, ) postCtx := newRealtimeTurnContext(baseCtx, requestID, session.ID(), session.ProviderSessionID(), realtimeTurnSourceLM, eventType, key) - applyRealtimeTurnContextValues(postCtx, preCtx.GetUserValues()) + applyRealtimeTurnContextValues(postCtx, preHookValues) + restoreRealtimeTurnTraceContext(postCtx, traceID, preHookValues) + applyRealtimeRawStorageContext(postCtx, storeRaw) setRealtimeTurnStreamContext(postCtx, startedAt, true) _, hookErr := hooks.PostHookRunner(postCtx, nil, postErr) completeRealtimeTurnTrace(postCtx) diff --git a/transports/bifrost-http/handlers/webrtc_realtime.go b/transports/bifrost-http/handlers/webrtc_realtime.go index da10f0a9cb7..6119ee4a445 100644 --- a/transports/bifrost-http/handlers/webrtc_realtime.go +++ b/transports/bifrost-http/handlers/webrtc_realtime.go @@ -13,6 +13,7 @@ import ( "github.com/fasthttp/router" bifrost "github.com/maximhq/bifrost/core" + "github.com/maximhq/bifrost/core/providers/openai" "github.com/maximhq/bifrost/core/schemas" "github.com/maximhq/bifrost/transports/bifrost-http/integrations" "github.com/maximhq/bifrost/transports/bifrost-http/lib" @@ -62,6 +63,10 @@ func (h *WebRTCRealtimeHandler) RegisterRoutes(r *router.Router, middlewares ... // Base bifrost route — GA /calls format (multipart sdp + session) r.POST("/v1/realtime/calls", handler) + // Base bifrost route — legacy format (raw SDP or multipart on /v1/realtime) + h.legacyRoutes["/v1/realtime"] = "" + r.POST("/v1/realtime", handler) + // OpenAI integration routes — /calls variants (GA format) for _, path := range integrations.OpenAIRealtimeWebRTCCallsPaths("/openai") { r.POST(path, handler) @@ -105,7 +110,7 @@ func (h *WebRTCRealtimeHandler) handleRequest(ctx *fasthttp.RequestCtx) { // Raw SDP bodies (application/sdp) fall back to ?model= for the legacy // raw-SDP path only; the multipart contract has no ?model= fallback. func (h *WebRTCRealtimeHandler) handleCallsRequest(ctx *fasthttp.RequestCtx) { - sdpOffer, providerKey, model, normalizedSession, bifrostErr := parseCallsWebRTCRequest(ctx) + sdpOffer, providerKey, model, normalizedSession, bifrostErr := parseCallsWebRTCRequest(ctx, h.config) if bifrostErr != nil { SendBifrostError(ctx, bifrostErr) return @@ -124,7 +129,7 @@ func (h *WebRTCRealtimeHandler) handleCallsRequest(ctx *fasthttp.RequestCtx) { h.runWebRTCRelay(ctx, rtProvider, providerKey, model, sdpOffer, exchangeSDP) } -func parseCallsWebRTCRequest(ctx *fasthttp.RequestCtx) (string, schemas.ModelProvider, string, []byte, *schemas.BifrostError) { +func parseCallsWebRTCRequest(ctx *fasthttp.RequestCtx, config *lib.Config) (string, schemas.ModelProvider, string, []byte, *schemas.BifrostError) { contentType := strings.ToLower(string(ctx.Request.Header.ContentType())) path := string(ctx.Path()) if strings.HasPrefix(contentType, "multipart/form-data") { @@ -142,7 +147,7 @@ func parseCallsWebRTCRequest(ctx *fasthttp.RequestCtx) (string, schemas.ModelPro if strings.TrimSpace(sessionField) == "" { return "", "", "", nil, newRealtimeWebRTCError(fasthttp.StatusBadRequest, "invalid_request_error", "session form field is required", nil) } - providerKey, model, normalizedSession, bifrostErr := resolveRealtimeSDPTarget(path, []byte(sessionField)) + providerKey, model, normalizedSession, bifrostErr := resolveRealtimeSDPTarget(ctx, config, path, []byte(sessionField)) if bifrostErr != nil { return "", "", "", nil, bifrostErr } @@ -160,6 +165,18 @@ func parseCallsWebRTCRequest(ctx *fasthttp.RequestCtx) (string, schemas.ModelPro } providerKey, model := schemas.ParseModelString(rawModel, realtimeDefaultProviderForPath(path)) + // Model catalog auto-resolution for bare model names on base /v1 routes + if providerKey == "" && strings.TrimSpace(model) != "" { + providers := config.GetProvidersForModel(model) + if len(providers) > 0 { + ctx.SetUserValue(lib.FastHTTPUserValueModelCatalogResolution, &lib.ModelCatalogResolution{ + Model: model, + ResolvedProvider: providers[0], + AllProviders: providers, + }) + providerKey = providers[0] + } + } if providerKey == "" || strings.TrimSpace(model) == "" { if realtimeDefaultProviderForPath(path) == "" { return "", "", "", nil, newRealtimeWebRTCError(fasthttp.StatusBadRequest, "invalid_request_error", "model must use provider/model on /v1 realtime routes", nil) @@ -180,6 +197,18 @@ func (h *WebRTCRealtimeHandler) handleLegacyRequest(ctx *fasthttp.RequestCtx, de } providerKey, model := schemas.ParseModelString(rawModel, defaultProvider) + // Model catalog auto-resolution for bare model names on base /v1 routes + if providerKey == "" && strings.TrimSpace(model) != "" { + providers := h.config.GetProvidersForModel(model) + if len(providers) > 0 { + ctx.SetUserValue(lib.FastHTTPUserValueModelCatalogResolution, &lib.ModelCatalogResolution{ + Model: model, + ResolvedProvider: providers[0], + AllProviders: providers, + }) + providerKey = providers[0] + } + } if providerKey == "" || model == "" { SendBifrostError(ctx, newRealtimeWebRTCError(fasthttp.StatusBadRequest, "invalid_request_error", "invalid model: "+rawModel, nil)) return @@ -197,6 +226,16 @@ func (h *WebRTCRealtimeHandler) handleLegacyRequest(ctx *fasthttp.RequestCtx, de return } + // Strip provider prefixes from nested model fields (e.g. input_audio_transcription.model) + if sessionJSON != nil { + if root, parseErr := schemas.ParseRealtimeClientSecretBody(sessionJSON); parseErr == nil { + openai.StripNestedModelPrefixes(root) + if updated, marshalErr := json.Marshal(root); marshalErr == nil { + sessionJSON = updated + } + } + } + exchangeSDP := func(rCtx *schemas.BifrostContext, key schemas.Key, upstreamOffer string) (string, *schemas.BifrostError) { return legacyProvider.ExchangeLegacyRealtimeWebRTCSDP(rCtx, key, upstreamOffer, sessionJSON, model) } @@ -254,6 +293,10 @@ func (h *WebRTCRealtimeHandler) runWebRTCRelay( ) { bifrostCtx, cancel := lib.ConvertToBifrostContext(ctx, h.handlerStore) defer cancel() + // Apply governance/routing values from the transport middleware. + // ConvertToBifrostContext creates a fresh context that doesn't carry the user + // values the middleware stored on the fasthttp RequestCtx via SetUserValue. + applyRealtimeMiddlewareValues(bifrostCtx, snapshotRealtimeMiddlewareValues(ctx)) bifrostCtx.SetValue(schemas.BifrostContextKeyHTTPRequestType, schemas.RealtimeRequest) if strings.HasPrefix(string(ctx.Path()), "/openai") { bifrostCtx.SetValue(schemas.BifrostContextKeyIntegrationType, "openai") @@ -272,6 +315,11 @@ func (h *WebRTCRealtimeHandler) runWebRTCRelay( model = authKey.Aliases.Resolve(model) } + // Compute raw storage flag from provider config + per-request header overrides. + // Normal inference computes this inside bifrost.executeRequest, which is bypassed + // for realtime WebRTC connections. + applyRealtimeRawStorageContext(bifrostCtx, h.client.ComputeRawStorageForProvider(bifrostCtx, providerKey)) + boundExchange := func(rCtx *schemas.BifrostContext, upstreamOffer string) (string, *schemas.BifrostError) { return exchangeSDP(rCtx, authKey, upstreamOffer) } @@ -792,6 +840,7 @@ func (r *webrtcRealtimeRelay) handleDownstreamMessage(msg webrtc.DataChannelMess } } + sanitizeRealtimeSessionEventForProvider(event) providerEvent, err := r.provider.ToProviderRealtimeEvent(event) if err != nil { if startsTurn { @@ -816,6 +865,9 @@ func (r *webrtcRealtimeRelay) handleDownstreamMessage(msg webrtc.DataChannelMess r.sendUpstream(msg.Data, msg.IsString) return } + // Track session metadata only after provider translation succeeds. Rejected + // session.update events must not affect later turn logs. + updateRealtimeSessionFromEvent(r.session, event) r.sendUpstream(providerEvent, msg.IsString) } @@ -844,6 +896,8 @@ func (r *webrtcRealtimeRelay) handleUpstreamMessage(msg webrtc.DataChannelMessag if event.Session != nil && event.Session.ID != "" { r.session.SetProviderSessionID(event.Session.ID) } + // Track session tool definitions from session.created/session.updated (server→client). + updateRealtimeSessionFromEvent(r.session, event) inputItemID, inputSummary := pendingRealtimeInputUpdate(event) if inputSummary != "" { r.session.RecordRealtimeInput(inputItemID, inputSummary, string(msg.Data)) @@ -1062,12 +1116,26 @@ func newRealtimeRelayContext(requestCtx *schemas.BifrostContext) (*schemas.Bifro schemas.BifrostContextKeySelectedKeyID, schemas.BifrostContextKeySelectedKeyName, schemas.BifrostContextKeyIsEnterprise, + schemas.BifrostContextKeyRoutingEnginesUsed, + schemas.BifrostContextKeyRoutingEngineLogs, + schemas.BifrostContextKeyShouldStoreRawInLogs, + schemas.BifrostContextKeyAllowPerRequestStorageOverride, + schemas.BifrostContextKeyAllowPerRequestRawOverride, + schemas.BifrostContextKeyStoreRawRequestResponse, + schemas.BifrostContextKeyDisableContentLogging, + schemas.BifrostContextKeyCaptureRawRequest, + schemas.BifrostContextKeyCaptureRawResponse, + schemas.BifrostContextKeyDropRawRequestFromClient, + schemas.BifrostContextKeyDropRawResponseFromClient, } { if value := requestCtx.Value(key); value != nil { relayCtx.SetValue(key, value) } } + // Tag the relay context with transport type for downstream logging/metadata. + relayCtx.SetValue(schemas.BifrostContextKeyRealtimeTransport, "webrtc") + return relayCtx, cancel } @@ -1149,7 +1217,7 @@ func sendDataChannelMessage(dc *webrtc.DataChannel, payload []byte, isString boo } } -func resolveRealtimeSDPTarget(path string, sessionJSON []byte) (schemas.ModelProvider, string, []byte, *schemas.BifrostError) { +func resolveRealtimeSDPTarget(ctx *fasthttp.RequestCtx, config *lib.Config, path string, sessionJSON []byte) (schemas.ModelProvider, string, []byte, *schemas.BifrostError) { root, err := schemas.ParseRealtimeClientSecretBody(sessionJSON) if err != nil { return "", "", nil, err @@ -1166,6 +1234,18 @@ func resolveRealtimeSDPTarget(path string, sessionJSON []byte) (schemas.ModelPro } providerKey, model := schemas.ParseModelString(strings.TrimSpace(rawModel), realtimeDefaultProviderForPath(path)) + // Model catalog auto-resolution for bare model names in session body + if providerKey == "" && strings.TrimSpace(model) != "" { + providers := config.GetProvidersForModel(model) + if len(providers) > 0 { + ctx.SetUserValue(lib.FastHTTPUserValueModelCatalogResolution, &lib.ModelCatalogResolution{ + Model: model, + ResolvedProvider: providers[0], + AllProviders: providers, + }) + providerKey = providers[0] + } + } if providerKey == "" || strings.TrimSpace(model) == "" { if realtimeDefaultProviderForPath(path) == "" { return "", "", nil, newRealtimeWebRTCError(fasthttp.StatusBadRequest, "invalid_request_error", "session.model must use provider/model on /v1 realtime routes", nil) @@ -1178,6 +1258,7 @@ func resolveRealtimeSDPTarget(path string, sessionJSON []byte) (schemas.ModelPro return "", "", nil, newRealtimeWebRTCError(fasthttp.StatusInternalServerError, "server_error", "failed to encode normalized session model", marshalErr) } root["model"] = normalizedModel + openai.StripNestedModelPrefixes(root) normalizedSession, marshalErr := json.Marshal(root) if marshalErr != nil { return "", "", nil, newRealtimeWebRTCError(fasthttp.StatusInternalServerError, "server_error", "failed to encode normalized realtime session", marshalErr) diff --git a/transports/bifrost-http/handlers/webrtc_realtime_test.go b/transports/bifrost-http/handlers/webrtc_realtime_test.go index 8ed36bd0407..a2636c9b112 100644 --- a/transports/bifrost-http/handlers/webrtc_realtime_test.go +++ b/transports/bifrost-http/handlers/webrtc_realtime_test.go @@ -33,7 +33,9 @@ func (s testHandlerStore) GetMCPExternalServerURL() string { re func (s testHandlerStore) GetMCPExternalClientURL() string { return "" } func TestResolveRealtimeSDPTarget_BaseRouteRequiresProviderPrefix(t *testing.T) { - _, _, _, err := resolveRealtimeSDPTarget("/v1/realtime", []byte(`{"model":"gpt-4o-realtime-preview"}`)) + var ctx fasthttp.RequestCtx + cfg := &lib.Config{} + _, _, _, err := resolveRealtimeSDPTarget(&ctx, cfg, "/v1/realtime", []byte(`{"model":"gpt-4o-realtime-preview"}`)) if err == nil { t.Fatal("expected provider/model validation error") } @@ -43,7 +45,9 @@ func TestResolveRealtimeSDPTarget_BaseRouteRequiresProviderPrefix(t *testing.T) } func TestResolveRealtimeSDPTarget_BaseRouteNormalizesModel(t *testing.T) { - provider, model, normalized, err := resolveRealtimeSDPTarget("/v1/realtime", []byte(`{"model":"openai/gpt-4o-realtime-preview","voice":"alloy"}`)) + var ctx fasthttp.RequestCtx + cfg := &lib.Config{} + provider, model, normalized, err := resolveRealtimeSDPTarget(&ctx, cfg, "/v1/realtime", []byte(`{"model":"openai/gpt-4o-realtime-preview","voice":"alloy"}`)) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -68,7 +72,9 @@ func TestResolveRealtimeSDPTarget_BaseRouteNormalizesModel(t *testing.T) { } func TestResolveRealtimeSDPTarget_OpenAIRouteDefaultsProvider(t *testing.T) { - provider, model, _, err := resolveRealtimeSDPTarget("/openai/v1/realtime", []byte(`{"model":"gpt-4o-realtime-preview"}`)) + var ctx fasthttp.RequestCtx + cfg := &lib.Config{} + provider, model, _, err := resolveRealtimeSDPTarget(&ctx, cfg, "/openai/v1/realtime", []byte(`{"model":"gpt-4o-realtime-preview"}`)) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -87,7 +93,7 @@ func TestParseCallsWebRTCRequest_RawSDPKeepsGARoute(t *testing.T) { ctx.Request.Header.SetContentType("application/sdp") ctx.Request.SetBodyString("v=0\r\n") - sdpOffer, provider, model, session, err := parseCallsWebRTCRequest(&ctx) + sdpOffer, provider, model, session, err := parseCallsWebRTCRequest(&ctx, &lib.Config{}) if err != nil { t.Fatalf("unexpected error: %v", err) } diff --git a/transports/bifrost-http/handlers/wsrealtime.go b/transports/bifrost-http/handlers/wsrealtime.go index 6f488103afc..81edd1496f0 100644 --- a/transports/bifrost-http/handlers/wsrealtime.go +++ b/transports/bifrost-http/handlers/wsrealtime.go @@ -2,6 +2,7 @@ package handlers import ( "errors" + "fmt" "io" "net" "net/http" @@ -77,7 +78,7 @@ func (h *WSRealtimeHandler) handleUpgrade(ctx *fasthttp.RequestCtx) { } } - providerKey, model, err := resolveRealtimeTarget(path, modelParam, deploymentParam) + providerKey, model, err := resolveRealtimeTarget(ctx, h.config, path, modelParam, deploymentParam) if err != nil { upgrader := h.websocketUpgrader("") upgradeErr := upgrader.Upgrade(ctx, func(conn *ws.Conn) { @@ -106,6 +107,13 @@ func (h *WSRealtimeHandler) handleUpgrade(ctx *fasthttp.RequestCtx) { return } + // Capture governance/routing values set by the transport middleware. + // TransportInterceptorMiddleware copies BifrostContext user values to individual + // fasthttp UserValue slots after HTTPTransportPreHook runs. We snapshot them now + // because the fasthttp RequestCtx is recycled after the handler returns — the + // WebSocket session outlives it. + middlewareContextValues := snapshotRealtimeMiddlewareValues(ctx) + upgrader := h.websocketUpgrader(rtProvider.RealtimeWebSocketSubprotocol()) err = upgrader.Upgrade(ctx, func(conn *ws.Conn) { defer conn.Close() @@ -118,7 +126,7 @@ func (h *WSRealtimeHandler) handleUpgrade(ctx *fasthttp.RequestCtx) { } defer h.sessions.Remove(conn) - h.runRealtimeSession(clientConn, session, auth, path, providerKey, model) + h.runRealtimeSession(clientConn, session, auth, path, providerKey, model, middlewareContextValues) }) if err != nil { logger.Warn("websocket upgrade failed for %s: %v", path, err) @@ -150,6 +158,7 @@ func (h *WSRealtimeHandler) runRealtimeSession( path string, providerKey schemas.ModelProvider, model string, + middlewareValues map[any]any, ) { clientConn.startHeartbeat() defer clientConn.stopHeartbeat() @@ -161,6 +170,12 @@ func (h *WSRealtimeHandler) runRealtimeSession( } defer cancel() + // Restore governance and routing values from the transport middleware context. + // These include routing rule ID/name, virtual key ID/name, routing engines, + // routing engine logs, raw-storage header overrides, and other values set by + // HTTPTransportPreHook plugins (governance, prompts, etc.). + applyRealtimeMiddlewareValues(bifrostCtx, middlewareValues) + // Resolve ephemeral key mapping to restore virtual key context. token := extractRealtimeBearerTokenFromHeader(auth.authorization) if isRealtimeEphemeralToken(token) { @@ -196,12 +211,26 @@ func (h *WSRealtimeHandler) runRealtimeSession( // Resolve model alias so the provider receives the actual model identifier. model = key.Aliases.Resolve(model) + // Compute raw storage flag from provider config + per-request header overrides. + // Normal inference computes this inside bifrost.executeRequest, which is bypassed + // for realtime WebSocket connections. Setting it on the session context ensures + // turn-level hooks can read it via shouldStoreRealtimeRawPayloads(). + applyRealtimeRawStorageContext(bifrostCtx, h.client.ComputeRawStorageForProvider(bifrostCtx, providerKey)) + + // Tag the session context with transport type for downstream logging/metadata. + bifrostCtx.SetValue(schemas.BifrostContextKeyRealtimeTransport, "websocket") + 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 @@ -288,6 +317,7 @@ func (h *WSRealtimeHandler) relayClientToRealtimeProvider( } } + sanitizeRealtimeSessionEventForProvider(event) providerEvent, err := provider.ToProviderRealtimeEvent(event) if err != nil { if startsTurn { @@ -310,6 +340,10 @@ func (h *WSRealtimeHandler) relayClientToRealtimeProvider( continue } + // Track session metadata only after provider translation succeeds. Rejected + // session.update events must not affect later turn logs. + updateRealtimeSessionFromEvent(session, event) + // Record tool output / input only after the event passed validation. if !startsTurn { if toolSummary != "" { @@ -402,6 +436,8 @@ func (h *WSRealtimeHandler) relayRealtimeProviderToClient( if event.Session != nil && event.Session.ID != "" { session.SetProviderSessionID(event.Session.ID) } + // Track session tool definitions from session.created/session.updated. + updateRealtimeSessionFromEvent(session, event) if event.Delta != nil && provider.ShouldAccumulateRealtimeOutput(event.Type) { session.AppendRealtimeOutputText(event.Delta.Text) session.AppendRealtimeOutputText(event.Delta.Transcript) @@ -481,25 +517,41 @@ func (h *WSRealtimeHandler) relayRealtimeProviderToClient( } } -func resolveRealtimeTarget(path, modelParam, deploymentParam string) (schemas.ModelProvider, string, error) { +func resolveRealtimeTarget(ctx *fasthttp.RequestCtx, config *lib.Config, path, modelParam, deploymentParam string) (schemas.ModelProvider, string, error) { defaultProvider := realtimeDefaultProviderForPath(path) + var rawParam string switch { case strings.TrimSpace(modelParam) != "": - provider, model := schemas.ParseModelString(strings.TrimSpace(modelParam), defaultProvider) - if provider == "" || strings.TrimSpace(model) == "" { - return "", "", errRealtimeModelFormat - } - return provider, strings.TrimSpace(model), nil + rawParam = strings.TrimSpace(modelParam) case strings.TrimSpace(deploymentParam) != "": - provider, model := schemas.ParseModelString(strings.TrimSpace(deploymentParam), defaultProvider) - if provider == "" || strings.TrimSpace(model) == "" { - return "", "", errRealtimeDeploymentFormat - } - return provider, strings.TrimSpace(model), nil + rawParam = strings.TrimSpace(deploymentParam) default: return "", "", errRealtimeModelRequired } + + provider, model := schemas.ParseModelString(rawParam, defaultProvider) + if strings.TrimSpace(model) == "" { + return "", "", errRealtimeModelFormat + } + + // Model catalog auto-resolution: when no provider prefix is present and the + // path doesn't imply a default provider, look up the model catalog — same + // logic as resolveModelAndProvider in inference.go. + if provider == "" { + providers := config.GetProvidersForModel(model) + if len(providers) == 0 { + return "", "", errRealtimeModelFormat + } + ctx.SetUserValue(lib.FastHTTPUserValueModelCatalogResolution, &lib.ModelCatalogResolution{ + Model: model, + ResolvedProvider: providers[0], + AllProviders: providers, + }) + provider = providers[0] + } + + return provider, model, nil } func realtimeDefaultProviderForPath(path string) schemas.ModelProvider { @@ -673,3 +725,113 @@ func newRealtimeWireBifrostError(status int, code, message string) *schemas.Bifr }, } } + +// applyRealtimeMiddlewareValues copies governance and routing values from the transport +// middleware BifrostContext (populated by HTTPTransportPreHook plugins) to the long-lived +// WebSocket session context. Without this, values set by the governance plugin during +// the HTTP upgrade (routing rule ID/name, VK ID/name, routing engines, routing engine +// logs, raw-storage overrides) would be lost because the WebSocket handler creates a +// fresh BifrostContext that outlives the fasthttp request. +// +// Values already explicitly set by createBifrostContextFromAuth (VK, parent request ID, +// request headers, extra headers) are preserved — middleware values do not overwrite them +// since createBifrostContextFromAuth runs first. +// realtimeMiddlewareKeys lists the BifrostContext keys that TransportInterceptorMiddleware +// copies from the governance plugin's context onto individual fasthttp UserValue slots. +// We snapshot exactly these keys before the WebSocket upgrade so the long-lived session +// has access to routing rule info, virtual key resolution, routing engine logs, etc. +var realtimeMiddlewareKeys = []any{ + schemas.BifrostContextKeyGovernanceVirtualKeyID, + schemas.BifrostContextKeyGovernanceVirtualKeyName, + schemas.BifrostContextKeyGovernanceRoutingRuleID, + schemas.BifrostContextKeyGovernanceRoutingRuleName, + schemas.BifrostContextKeyGovernanceCustomerID, + schemas.BifrostContextKeyGovernanceCustomerName, + schemas.BifrostContextKeyGovernanceTeamID, + schemas.BifrostContextKeyGovernanceTeamName, + schemas.BifrostContextKeyGovernanceBusinessUnitID, + schemas.BifrostContextKeyGovernanceBusinessUnitName, + schemas.BifrostContextKeyGovernanceIncludeOnlyKeys, + schemas.BifrostContextKeyGovernancePluginName, + schemas.BifrostContextKeyRoutingEnginesUsed, + schemas.BifrostContextKeyRoutingEngineLogs, + schemas.BifrostContextKeyShouldStoreRawInLogs, + schemas.BifrostContextKeyCaptureRawRequest, + schemas.BifrostContextKeyCaptureRawResponse, + schemas.BifrostContextKeyDropRawRequestFromClient, + schemas.BifrostContextKeyDropRawResponseFromClient, + schemas.BifrostContextKeyUserID, + schemas.BifrostContextKeyUserName, + schemas.BifrostContextKeyAPIKeyID, + schemas.BifrostContextKeyAPIKeyName, + schemas.BifrostContextKeySelectedKeyID, + schemas.BifrostContextKeySelectedKeyName, + schemas.BifrostContextKeyTraceID, + schemas.BifrostContextKeyTransportPluginLogs, +} + +// snapshotRealtimeMiddlewareValues reads governance/routing values from the fasthttp +// context's UserValue store. TransportInterceptorMiddleware copies them there as +// individual key-value pairs (not inside a BifrostContext). +// +// It also processes FastHTTPUserValueModelCatalogResolution, which is set by +// resolveRealtimeTarget when a bare model name is auto-resolved via the model +// catalog. ConvertToBifrostContext normally handles this for regular inference, +// but WebSocket handlers use createBifrostContextFromAuth instead, so we do the +// same log/engine enrichment here. +func snapshotRealtimeMiddlewareValues(ctx *fasthttp.RequestCtx) map[any]any { + result := make(map[any]any) + for _, key := range realtimeMiddlewareKeys { + if value := ctx.UserValue(key); value != nil { + result[key] = value + } + } + + // Model catalog auto-resolution: replicate the routing engine log that + // ConvertToBifrostContext would normally emit (see lib/ctx.go). + if res, ok := ctx.UserValue(lib.FastHTTPUserValueModelCatalogResolution).(*lib.ModelCatalogResolution); ok && res != nil { + providerStrs := make([]string, len(res.AllProviders)) + for i, p := range res.AllProviders { + providerStrs[i] = string(p) + } + logEntry := schemas.RoutingEngineLogEntry{ + Engine: schemas.RoutingEngineModelCatalog, + Level: schemas.LogLevelInfo, + Message: fmt.Sprintf("No provider specified for model %s, found %d options in model catalog: [%s], selecting first: %s", res.Model, len(res.AllProviders), strings.Join(providerStrs, ", "), res.ResolvedProvider), + Timestamp: time.Now().UnixMilli(), + } + // Merge with any existing routing engine logs from governance middleware. + if existing, ok := result[schemas.BifrostContextKeyRoutingEngineLogs].([]schemas.RoutingEngineLogEntry); ok { + result[schemas.BifrostContextKeyRoutingEngineLogs] = append(existing, logEntry) + } else { + result[schemas.BifrostContextKeyRoutingEngineLogs] = []schemas.RoutingEngineLogEntry{logEntry} + } + if existing, ok := result[schemas.BifrostContextKeyRoutingEnginesUsed].([]string); ok { + result[schemas.BifrostContextKeyRoutingEnginesUsed] = append(existing, schemas.RoutingEngineModelCatalog) + } else { + result[schemas.BifrostContextKeyRoutingEnginesUsed] = []string{schemas.RoutingEngineModelCatalog} + } + } + + if len(result) == 0 { + return nil + } + return result +} + +func applyRealtimeMiddlewareValues(ctx *schemas.BifrostContext, middlewareValues map[any]any) { + if ctx == nil || len(middlewareValues) == 0 { + return + } + for key, value := range middlewareValues { + if value == nil { + continue + } + // Skip values already set by createBifrostContextFromAuth to avoid overwriting + // auth-resolved values with stale middleware copies. + if existing := ctx.Value(key); existing != nil { + continue + } + ctx.SetValue(key, value) + } +} diff --git a/transports/bifrost-http/websocket/session.go b/transports/bifrost-http/websocket/session.go index e10180280e5..3d95977f839 100644 --- a/transports/bifrost-http/websocket/session.go +++ b/transports/bifrost-http/websocket/session.go @@ -1,6 +1,7 @@ package websocket import ( + "encoding/json" "strings" "sync" "time" @@ -47,6 +48,14 @@ type Session struct { // attached to a persisted turn, so late transcript updates do not pollute later turns. realtimeConsumedTurnItemIDs map[string]struct{} + // realtimeSessionTools holds the latest session tool definitions from + // session.created / session.updated / session.update events, so that + // each turn log can record which tools were available. + realtimeSessionTools json.RawMessage + + // realtimeVoice holds the voice from the latest session configuration. + realtimeVoice string + // realtimeTurnHooks tracks the active turn-scoped plugin pipeline between // response.create and response.done. realtimeTurnHooks *RealtimeTurnPluginState @@ -73,6 +82,8 @@ type RealtimeTurnPluginState struct { RequestID string StartedAt time.Time PreHookValues map[any]any + TraceID string + RawStore bool } // NewSession creates a new session for a client WebSocket connection. @@ -170,6 +181,42 @@ func (s *Session) ProviderSessionID() string { return s.providerSessionID } +// SetRealtimeSessionTools updates the tracked session tool definitions. +// Called when session.created, session.updated, or session.update events +// carry a tools array. +func (s *Session) SetRealtimeSessionTools(tools json.RawMessage) { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return + } + s.realtimeSessionTools = tools +} + +// RealtimeSessionTools returns the latest session tool definitions, or nil. +func (s *Session) RealtimeSessionTools() json.RawMessage { + s.mu.RLock() + defer s.mu.RUnlock() + return s.realtimeSessionTools +} + +// SetRealtimeVoice updates the tracked voice from session configuration. +func (s *Session) SetRealtimeVoice(voice string) { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return + } + s.realtimeVoice = voice +} + +// RealtimeVoice returns the current session voice, or empty string. +func (s *Session) RealtimeVoice() string { + s.mu.RLock() + defer s.mu.RUnlock() + return s.realtimeVoice +} + // AppendRealtimeOutputText appends provider output content for the current realtime turn. func (s *Session) AppendRealtimeOutputText(text string) { if text == "" { @@ -177,6 +224,9 @@ func (s *Session) AppendRealtimeOutputText(text string) { } s.mu.Lock() defer s.mu.Unlock() + if s.closed { + return + } s.realtimeOutputText += text } @@ -236,11 +286,16 @@ func (s *Session) recordRealtimeTurnInput(itemID, role, summary, raw string) { s.mu.Lock() defer s.mu.Unlock() + if s.closed { + return + } itemID = strings.TrimSpace(itemID) if itemID != "" { - if _, consumed := s.realtimeConsumedTurnItemIDs[itemID]; consumed { - return + if s.realtimeConsumedTurnItemIDs != nil { + if _, consumed := s.realtimeConsumedTurnItemIDs[itemID]; consumed { + return + } } for idx := range s.realtimeTurnInputs { if s.realtimeTurnInputs[idx].ItemID != itemID || s.realtimeTurnInputs[idx].Role != role { @@ -250,15 +305,11 @@ func (s *Session) recordRealtimeTurnInput(itemID, role, summary, raw string) { s.realtimeTurnInputs[idx].Summary = summary } if strings.TrimSpace(raw) != "" { - existingRaw := strings.TrimSpace(s.realtimeTurnInputs[idx].Raw) - incomingRaw := strings.TrimSpace(raw) - switch { - case existingRaw == "": - s.realtimeTurnInputs[idx].Raw = raw - case incomingRaw == "" || existingRaw == incomingRaw: - default: - s.realtimeTurnInputs[idx].Raw = existingRaw + "\n\n" + incomingRaw - } + // Same item ID + role: replace raw with the latest event. + // Later events (e.g. conversation.item.created after + // conversation.item.create) carry the same or more complete + // data, so the newest version is always preferred. + s.realtimeTurnInputs[idx].Raw = raw } return } @@ -377,6 +428,15 @@ func (s *Session) Close() { s.realtimeTurnHooks = nil } s.realtimeTurnBusy = false + + // Release accumulated turn data so GC can reclaim memory even if a + // goroutine briefly holds a reference to this session after close. + s.realtimeTurnInputs = nil + s.realtimeConsumedTurnItemIDs = nil + s.realtimeSessionTools = nil + s.realtimeVoice = "" + s.realtimeOutputText = "" + if s.clientConn != nil { _ = s.clientConn.Close() } diff --git a/ui/app/workspace/logs/sheets/logDetailView.tsx b/ui/app/workspace/logs/sheets/logDetailView.tsx index ec581d78100..e6bda63ae6c 100644 --- a/ui/app/workspace/logs/sheets/logDetailView.tsx +++ b/ui/app/workspace/logs/sheets/logDetailView.tsx @@ -47,6 +47,41 @@ import SpeechView from "../views/speechView"; import TranscriptionView from "../views/transcriptionView"; import VideoView from "../views/videoView"; +const formatRealtimeTransport = (value: unknown): string => { + const transport = String(value ?? "").trim(); + switch (transport.toLowerCase()) { + case "websocket": + return "WebSocket"; + case "webrtc": + return "WebRTC"; + default: + return transport || "Unknown"; + } +}; + +const getRealtimeTransportBadgeClass = (value: unknown): string => { + switch (String(value ?? "").toLowerCase()) { + case "websocket": + return "border-indigo-300 bg-indigo-50 text-indigo-700 dark:border-indigo-600 dark:bg-indigo-950 dark:text-indigo-300"; + case "webrtc": + return "border-purple-300 bg-purple-50 text-purple-700 dark:border-purple-600 dark:bg-purple-950 dark:text-purple-300"; + default: + return "border-slate-300 bg-slate-50 text-slate-700 dark:border-slate-600 dark:bg-slate-950 dark:text-slate-300"; + } +}; + +const formatRealtimeSource = (value: unknown): string => { + const source = String(value ?? "").trim(); + switch (source.toLowerCase()) { + case "ei": + return "Event Initiated"; + case "lm": + return "Language Model"; + default: + return source || "Unknown"; + } +}; + const extractResponsesText = (msg: ResponsesMessage): string => { if (msg.type === "reasoning") { const summaryText = (msg.summary ?? []) @@ -503,6 +538,7 @@ export function LogDetailView({ const isContainer = isContainerOperation(log.object); const showTabs = !isContainer; const isPassthrough = isPassthroughOperation(log.object); + const isRealtimeTurn = log.object === "realtime.turn"; const passthroughParams = isPassthrough ? (log.params as { method?: string; @@ -651,6 +687,22 @@ export function LogDetailView({ Large Payload )} + {isRealtimeTurn && log.metadata?.realtime_transport && ( + + {formatRealtimeTransport(log.metadata.realtime_transport)} + + )} + {isRealtimeTurn && log.metadata?.realtime_voice && ( + + {log.metadata.realtime_voice} + + )}
Request
@@ -736,11 +788,19 @@ export function LogDetailView({ } hasRightBorder /> - + {isRealtimeTurn ? ( + + ) : ( + + )}
@@ -971,6 +1031,65 @@ export function LogDetailView({ )} + {isRealtimeTurn && ( + <> + {log.metadata?.realtime_session_id && ( + + {log.metadata.realtime_session_id} + + + } + /> + )} + {log.metadata?.provider_session_id && ( + + {log.metadata.provider_session_id} + + + } + /> + )} + {log.metadata?.realtime_transport && ( + + )} + {log.metadata?.realtime_voice && ( + + )} + {log.metadata?.realtime_source && ( + + )} + {log.metadata?.realtime_event_type && ( + {log.metadata.realtime_event_type}} + /> + )} + + )} + {passthroughParams && ( <> {passthroughParams.method && } @@ -1011,7 +1130,42 @@ export function LogDetailView({ label="Cost" value={log.cost != null ? `$${parseFloat(log.cost.toFixed(6))}` : "-"} /> - {log.token_usage?.prompt_tokens_details && ( + {isRealtimeTurn && ( + <> + + + + + {(log.token_usage?.completion_tokens_details?.reasoning_tokens ?? 0) > 0 && ( + + )} + + )} + {!isRealtimeTurn && log.token_usage?.prompt_tokens_details && ( <> {log.token_usage.prompt_tokens_details.cached_read_tokens && ( )} - {log.token_usage?.completion_tokens_details && ( + {!isRealtimeTurn && log.token_usage?.completion_tokens_details && ( <> {log.token_usage.completion_tokens_details.reasoning_tokens && ( )} - {log.metadata && Object.keys(log.metadata).filter((k) => k !== "isAsyncRequest").length > 0 && ( - <> - -
- -
- {Object.entries(log.metadata) - .filter(([key]) => key !== "isAsyncRequest") - .map(([key, value]) => ( - - ))} + {log.metadata && + Object.keys(log.metadata).filter((k) => { + if (k === "isAsyncRequest") return false; + if ( + isRealtimeTurn && + [ + "realtime_session_id", + "provider_session_id", + "realtime_source", + "realtime_event_type", + "realtime_transport", + "realtime_voice", + "realtime", + ].includes(k) + ) + return false; + return true; + }).length > 0 && ( + <> + +
+ +
+ {Object.entries(log.metadata) + .filter(([key]) => { + if (key === "isAsyncRequest") return false; + if ( + isRealtimeTurn && + [ + "realtime_session_id", + "provider_session_id", + "realtime_source", + "realtime_event_type", + "realtime_transport", + "realtime_voice", + "realtime", + ].includes(key) + ) + return false; + return true; + }) + .map(([key, value]) => ( + + ))} +
-
- - )} + + )} )}