From ec282ce9264c07ae1a5ed040e92a20ab2ddb7d74 Mon Sep 17 00:00:00 2001 From: Jarvis Date: Mon, 6 Jul 2026 16:47:21 +0800 Subject: [PATCH 1/2] feat(realtime): /v1/realtime WebSocket relay (OpenAI protocol) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the OpenAI Realtime endpoint: authenticate on connect (Bearer / x-api-key headers, or the browser-flow openai-insecure-api-key. subprotocol item with the realtime subprotocol echoed), resolve the Model from ?model=, enforce ACL / IP allowlist / budget+rate limits BEFORE the upgrade, then open the provider WebSocket and relay frames bidirectionally. Provider scope v1 is the OpenAI Realtime wire protocol: adapter openai (any OpenAI-compatible api_base) and azure-openai (/openai/realtime?api-version=…&deployment=…, api-key header). Gemini Live / Bedrock use different session/event models and need a cross-protocol translation layer (LiteLLM ships those as dedicated per-provider transform modules) — tracked separately. Usage: the relay harvests response.done and transcription-completed usage frames and emits one aggregated UsageEvent per session (inbound_protocol="realtime"), committing total tokens to the rate-limit reservation (#911 [21] parity). Guardrail chains scan each text frame both directions (passthrough blob precedent); blocks send an error event then close 1011. Optional stream_timeout on the Model acts as the inter-frame idle cap. Fixes #721 --- Cargo.lock | 20 +- Cargo.toml | 1 + crates/aisix-proxy/Cargo.toml | 1 + crates/aisix-proxy/src/auth.rs | 57 +- crates/aisix-proxy/src/lib.rs | 7 + crates/aisix-proxy/src/realtime.rs | 974 ++++++++++++++++++++ tests/e2e/package.json | 4 +- tests/e2e/src/cases/realtime-ws-e2e.test.ts | 185 ++++ 8 files changed, 1223 insertions(+), 26 deletions(-) create mode 100644 crates/aisix-proxy/src/realtime.rs create mode 100644 tests/e2e/src/cases/realtime-ws-e2e.test.ts diff --git a/Cargo.lock b/Cargo.lock index a92097a4..2ce6092e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -355,6 +355,7 @@ dependencies = [ "sha2 0.10.9", "thiserror 1.0.69", "tokio", + "tokio-tungstenite", "tower 0.5.3", "tower-http", "tracing", @@ -2369,7 +2370,7 @@ dependencies = [ "tokio", "tokio-rustls 0.26.4", "tower-service", - "webpki-roots", + "webpki-roots 1.0.7", ] [[package]] @@ -3807,7 +3808,7 @@ dependencies = [ "wasm-bindgen-futures", "wasm-streams 0.4.2", "web-sys", - "webpki-roots", + "webpki-roots 1.0.7", ] [[package]] @@ -4774,8 +4775,12 @@ checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9" dependencies = [ "futures-util", "log", + "rustls 0.23.38", + "rustls-pki-types", "tokio", + "tokio-rustls 0.26.4", "tungstenite", + "webpki-roots 0.26.11", ] [[package]] @@ -5044,6 +5049,8 @@ dependencies = [ "httparse", "log", "rand 0.8.5", + "rustls 0.23.38", + "rustls-pki-types", "sha1", "thiserror 1.0.69", "utf-8", @@ -5350,6 +5357,15 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.7", +] + [[package]] name = "webpki-roots" version = "1.0.7" diff --git a/Cargo.toml b/Cargo.toml index 2d226a9c..fc1d1acf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,6 +33,7 @@ authors = ["moonming"] # Runtime tokio = { version = "1.41", features = ["full"] } futures = "0.3" +tokio-tungstenite = { version = "0.24", features = ["rustls-tls-webpki-roots"] } async-trait = "0.1" # HTTP server diff --git a/crates/aisix-proxy/Cargo.toml b/crates/aisix-proxy/Cargo.toml index 67aaaaea..a35a4719 100644 --- a/crates/aisix-proxy/Cargo.toml +++ b/crates/aisix-proxy/Cargo.toml @@ -42,6 +42,7 @@ bytes.workspace = true serde.workspace = true serde_json.workspace = true futures.workspace = true +tokio-tungstenite.workspace = true async-trait.workspace = true async-stream = "0.3" dashmap.workspace = true diff --git a/crates/aisix-proxy/src/auth.rs b/crates/aisix-proxy/src/auth.rs index 3b6dca19..9dc48c41 100644 --- a/crates/aisix-proxy/src/auth.rs +++ b/crates/aisix-proxy/src/auth.rs @@ -38,30 +38,41 @@ where async fn from_request_parts(parts: &mut Parts, state: &S) -> Result { let token = extract_bearer(parts)?; let proxy_state = ProxyState::from_ref(state); - let snapshot = proxy_state.snapshot.load(); - // Self-hosted CP (prd-09a §9A.7B.4): the snapshot stores - // SHA-256 hashes of the plaintext bearer, never the plaintext - // itself. Hash the incoming token via the canonical helper - // and look up by the hex digest. cp-api hashes with the same - // function before persistence, so the two sides agree byte - // for byte. - let entry = snapshot - .apikeys - .get_by_name(&ApiKey::hash_bearer(&token)) - .ok_or(ProxyError::InvalidApiKey)?; - // Lifecycle enforcement (#933): a known key that is disabled or - // past its expiry deadline must be rejected here, at the single - // auth choke point, so every proxy surface (chat, messages, - // responses, embeddings, audio, passthrough, MCP, …) inherits - // the same 401 without per-handler checks. - if entry.value.disabled { - return Err(ProxyError::ApiKeyDisabled); - } - if entry.value.is_expired_at(chrono::Utc::now()) { - return Err(ProxyError::ApiKeyExpired); - } - Ok(AuthenticatedKey { entry }) + authenticate_token(&proxy_state, &token) + } +} + +/// Look a plaintext bearer up in the snapshot and enforce key lifecycle. +/// The single auth choke point behind the [`AuthenticatedKey`] extractor; +/// also called directly by surfaces whose credentials arrive outside the +/// standard headers (WebSocket subprotocol auth on `/v1/realtime`). +pub(crate) fn authenticate_token( + state: &ProxyState, + token: &str, +) -> Result { + let snapshot = state.snapshot.load(); + // Self-hosted CP (prd-09a §9A.7B.4): the snapshot stores + // SHA-256 hashes of the plaintext bearer, never the plaintext + // itself. Hash the incoming token via the canonical helper + // and look up by the hex digest. cp-api hashes with the same + // function before persistence, so the two sides agree byte + // for byte. + let entry = snapshot + .apikeys + .get_by_name(&ApiKey::hash_bearer(token)) + .ok_or(ProxyError::InvalidApiKey)?; + // Lifecycle enforcement (#933): a known key that is disabled or + // past its expiry deadline must be rejected here, at the single + // auth choke point, so every proxy surface (chat, messages, + // responses, embeddings, audio, passthrough, MCP, …) inherits + // the same 401 without per-handler checks. + if entry.value.disabled { + return Err(ProxyError::ApiKeyDisabled); + } + if entry.value.is_expired_at(chrono::Utc::now()) { + return Err(ProxyError::ApiKeyExpired); } + Ok(AuthenticatedKey { entry }) } fn extract_bearer(parts: &Parts) -> Result { diff --git a/crates/aisix-proxy/src/lib.rs b/crates/aisix-proxy/src/lib.rs index 88f1ed35..88ad245a 100644 --- a/crates/aisix-proxy/src/lib.rs +++ b/crates/aisix-proxy/src/lib.rs @@ -49,6 +49,7 @@ mod model_resolve; mod models; mod passthrough; mod quota; +mod realtime; mod redact; mod render; mod request_id; @@ -105,6 +106,9 @@ pub fn build_router(state: ProxyState) -> Router { .route("/v1/audio/transcriptions", post(audio::transcriptions)) .route("/v1/audio/translations", post(audio::translations)) .route("/v1/audio/speech", post(audio::speech)) + // OpenAI Realtime WebSocket relay (#721). Auth/ACL/quota are + // enforced pre-upgrade inside the handler. + .route("/v1/realtime", get(realtime::realtime)) .route( "/passthrough/:provider/*rest", any(passthrough::passthrough), @@ -196,6 +200,7 @@ fn normalize_endpoint_label(path: &str) -> &'static str { "/v1/audio/translations" => "/v1/audio/translations", "/v1/audio/speech" => "/v1/audio/speech", "/mcp" | "/mcp/" => "/mcp", + "/v1/realtime" => "/v1/realtime", _ if path.starts_with("/a2a/") => "/a2a", _ if path.starts_with("/passthrough/") => "/passthrough/:provider/*rest", _ => "other", @@ -209,6 +214,8 @@ fn inbound_protocol_for_endpoint(endpoint: &str) -> &'static str { "mcp" } else if endpoint == "/a2a" { "a2a" + } else if endpoint == "/v1/realtime" { + "realtime" } else { "openai" } diff --git a/crates/aisix-proxy/src/realtime.rs b/crates/aisix-proxy/src/realtime.rs new file mode 100644 index 00000000..1b8225c7 --- /dev/null +++ b/crates/aisix-proxy/src/realtime.rs @@ -0,0 +1,974 @@ +//! `/v1/realtime` — OpenAI Realtime WebSocket relay (#721, +//! AISIX-Cloud#873 §⑤). +//! +//! Authenticates on connect, resolves the target Model from `?model=`, +//! opens the provider WebSocket and relays frames bidirectionally. +//! +//! ## Protocol scope +//! +//! v1 relays the **OpenAI Realtime wire protocol**: adapter `openai` +//! (api.openai.com and any OpenAI-compatible `api_base`, which covers +//! xAI-style vendors) and `azure-openai` (`{base}/openai/realtime +//! ?api-version=…&deployment=…`, `api-key` header). Gemini Live / AWS +//! Bedrock speak entirely different session/event models and need a +//! cross-protocol translation layer (LiteLLM ships those as dedicated +//! per-provider `transform_realtime_request/response` modules) — that is +//! a separate feature, not part of this endpoint. +//! +//! ## Auth +//! +//! Two credential channels, checked before the upgrade completes: +//! +//! 1. `Authorization: Bearer ` / `x-api-key` headers — server-side +//! clients (LiteLLM parity: `user_api_key_auth_websocket`). +//! 2. The `sec-websocket-protocol` item `openai-insecure-api-key.` +//! — browser clients cannot set headers; this is the documented +//! OpenAI browser flow. The gateway echoes the `realtime` subprotocol +//! when offered. +//! +//! Auth/ACL/quota failures reject the HTTP upgrade itself (401/403/429 +//! envelope) rather than accept-then-close-1008: same enforcement point, +//! observable to every WS client as a failed handshake. +//! +//! ## Usage +//! +//! The relay harvests `response.done` usage frames (and +//! `conversation.item.input_audio_transcription.completed` token usage) +//! from the upstream stream and emits ONE aggregated UsageEvent per +//! session (`inbound_protocol = "realtime"`), committing total tokens to +//! the rate-limit reservation like the other non-chat surfaces (#911 +//! [21]). + +use std::collections::HashMap; +use std::time::{Duration, Instant}; + +use aisix_core::models::model::Adapter; +use aisix_obs::{AccessLog, RequestOutcome, UsageEvent}; +use axum::extract::ws::{CloseFrame, Message as AxMessage, WebSocket, WebSocketUpgrade}; +use axum::extract::{Query, State}; +use axum::http::{HeaderMap, Method}; +use axum::response::{IntoResponse, Response}; +use futures::{SinkExt, StreamExt}; +use serde_json::Value; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tokio_tungstenite::tungstenite::Message as TgMessage; + +use crate::auth::AuthenticatedKey; +use crate::client_ip::ClientContext; +use crate::error::ProxyError; +use crate::request_id::new_request_id; +use crate::state::ProxyState; + +/// Azure Realtime GA api-version (see the jobs surface twin constant). +const AZURE_REALTIME_API_VERSION: &str = "2024-10-01-preview"; + +/// Subprotocol item carrying the caller's API key in the browser flow. +const SUBPROTOCOL_KEY_PREFIX: &str = "openai-insecure-api-key."; + +pub(crate) async fn realtime( + State(state): State, + Query(params): Query>, + headers: HeaderMap, + client: ClientContext, + ws: WebSocketUpgrade, +) -> Response { + let request_id = new_request_id(); + let started = Instant::now(); + + match prepare(&state, ¶ms, &headers, &client).await { + Ok(prep) => { + let state2 = state.clone(); + let client2 = client.clone(); + ws.protocols(["realtime"]) + .on_upgrade(move |socket| async move { + run_session(state2, prep, socket, client2, request_id, started).await; + }) + } + Err(err) => { + let status = err.status().as_u16(); + emit_access_log(&Method::GET, status, started.elapsed(), &request_id, None); + crate::usage_attr::emit_error_usage_event( + &state, + "realtime", + &request_id, + params.get("model").map(String::as_str).unwrap_or(""), + "", + status, + err.kind(), + &client, + ); + err.into_response() + } + } +} + +/// Everything resolved before the upgrade is accepted. +struct Prepared { + auth: AuthenticatedKey, + model_entry: std::sync::Arc>, + pk_id: String, + upstream_request: tokio_tungstenite::tungstenite::handshake::client::Request, + reservation: aisix_ratelimit::MultiReservation, + requested_model: String, + provider_label: String, +} + +async fn prepare( + state: &ProxyState, + params: &HashMap, + headers: &HeaderMap, + client: &ClientContext, +) -> Result { + let auth = authenticate(state, headers)?; + + let requested_model = params + .get("model") + .map(String::as_str) + .unwrap_or("") + .trim() + .to_string(); + if requested_model.is_empty() { + return Err(ProxyError::InvalidRequest( + "`model` query parameter is required on /v1/realtime".into(), + )); + } + + let snapshot = state.snapshot.load(); + let model_entry = crate::model_resolve::resolve_model(&snapshot, &requested_model) + .ok_or_else(|| ProxyError::ModelNotFound(format!("model {requested_model:?} not found")))?; + if !auth.key().can_access(&requested_model) { + return Err(ProxyError::ModelForbidden(format!( + "api key is not authorized for model {requested_model:?}" + ))); + } + let model = &model_entry.value; + if model.is_routing() || model.is_ensemble() || model.is_semantic() { + return Err(ProxyError::InvalidRequest(format!( + "model {requested_model:?} is a virtual router; /v1/realtime requires a direct model" + ))); + } + crate::dispatch::check_ip_access(model, &client.source_ip)?; + + let pk_entry = crate::dispatch::resolve_provider_key(&snapshot, model)?; + let secret = crate::dispatch::require_secret(&pk_entry.value, model)?.to_string(); + let upstream_model = crate::dispatch::require_upstream_model(model)?.to_string(); + + let upstream_request = match pk_entry.value.adapter { + Some(Adapter::Openai) => { + let base = crate::dispatch::resolve_base_url(&pk_entry.value)?; + let url = crate::dispatch::build_v1_url(&base, "/realtime"); + let url = format!( + "{}?model={}", + to_ws_scheme(&url)?, + urlencode(&upstream_model) + ); + let mut req = url.into_client_request().map_err(|e| { + ProxyError::InvalidRequest(format!("invalid upstream realtime URL: {e}")) + })?; + req.headers_mut().insert( + "authorization", + format!("Bearer {secret}").parse().map_err(|_| { + ProxyError::InvalidRequest("provider secret is not header-safe".into()) + })?, + ); + // LiteLLM parity (OpenAIRealtime.async_realtime): the beta + // header is sent unconditionally; GA endpoints ignore it. + req.headers_mut() + .insert("openai-beta", "realtime=v1".parse().unwrap()); + req + } + Some(Adapter::AzureOpenai) => { + let base = pk_entry + .value + .api_base + .as_deref() + .map(str::trim) + .filter(|b| !b.is_empty()) + .ok_or_else(|| { + ProxyError::InvalidRequest(format!( + "azure provider_key {:?} has no api_base", + pk_entry.value.display_name + )) + })? + .trim_end_matches('/') + .to_string(); + let url = format!( + "{}/openai/realtime?api-version={AZURE_REALTIME_API_VERSION}&deployment={}", + to_ws_scheme(&base)?, + urlencode(&upstream_model) + ); + let mut req = url.into_client_request().map_err(|e| { + ProxyError::InvalidRequest(format!("invalid upstream realtime URL: {e}")) + })?; + req.headers_mut().insert( + "api-key", + secret.parse().map_err(|_| { + ProxyError::InvalidRequest("provider secret is not header-safe".into()) + })?, + ); + req + } + _ => { + return Err(ProxyError::InvalidRequest(format!( + "model {requested_model:?} uses provider {:?} which does not speak the OpenAI \ + Realtime protocol; /v1/realtime supports OpenAI-compatible and Azure OpenAI \ + providers", + pk_entry.value.provider + ))); + } + }; + let reservation = crate::quota::enforce( + state, + &auth, + Some(&crate::quota::ModelRateLimit::from_model( + &model_entry.value.display_name, + &model_entry.id, + &model_entry.value, + )), + ) + .await?; + + let provider_label = model.provider.clone().unwrap_or_default(); + Ok(Prepared { + auth, + pk_id: pk_entry.id.to_string(), + model_entry, + upstream_request, + reservation, + requested_model, + provider_label, + }) +} + +/// Header bearer (`Authorization` / `x-api-key`) first, then the browser +/// subprotocol credential. +fn authenticate(state: &ProxyState, headers: &HeaderMap) -> Result { + if let Some(auth) = headers.get(axum::http::header::AUTHORIZATION) { + let s = auth.to_str().map_err(|_| ProxyError::MissingAuth)?; + let token = s.strip_prefix("Bearer ").map(str::trim).unwrap_or(""); + if token.is_empty() { + return Err(ProxyError::MissingAuth); + } + return crate::auth::authenticate_token(state, token); + } + if let Some(raw) = headers.get("x-api-key") { + let token = raw.to_str().map_err(|_| ProxyError::MissingAuth)?.trim(); + if token.is_empty() { + return Err(ProxyError::MissingAuth); + } + return crate::auth::authenticate_token(state, token); + } + if let Some(proto) = headers.get("sec-websocket-protocol") { + let s = proto.to_str().map_err(|_| ProxyError::MissingAuth)?; + for item in s.split(',') { + if let Some(token) = item.trim().strip_prefix(SUBPROTOCOL_KEY_PREFIX) { + if !token.is_empty() { + return crate::auth::authenticate_token(state, token); + } + } + } + } + Err(ProxyError::MissingAuth) +} + +fn to_ws_scheme(url: &str) -> Result { + if let Some(rest) = url.strip_prefix("https://") { + Ok(format!("wss://{rest}")) + } else if let Some(rest) = url.strip_prefix("http://") { + Ok(format!("ws://{rest}")) + } else if url.starts_with("ws://") || url.starts_with("wss://") { + Ok(url.to_string()) + } else { + Err(ProxyError::InvalidRequest(format!( + "api_base {url:?} has no http(s) scheme" + ))) + } +} + +fn urlencode(s: &str) -> String { + // Conservative percent-encoding for the query-value position. + let mut out = String::with_capacity(s.len()); + for b in s.bytes() { + match b { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => { + out.push(b as char) + } + _ => out.push_str(&format!("%{b:02X}")), + } + } + out +} + +/// Accumulated session usage harvested from upstream frames. +#[derive(Default)] +struct SessionUsage { + input_tokens: u64, + output_tokens: u64, + cached_tokens: u64, + responses: u32, +} + +impl SessionUsage { + fn absorb(&mut self, text: &str) { + // Fast path: only parse frames that can carry usage. + if !text.contains("\"response.done\"") + && !text.contains("\"conversation.item.input_audio_transcription.completed\"") + { + return; + } + let Ok(v) = serde_json::from_str::(text) else { + return; + }; + match v.get("type").and_then(Value::as_str) { + Some("response.done") => { + let usage = &v["response"]["usage"]; + self.input_tokens += usage["input_tokens"].as_u64().unwrap_or(0); + self.output_tokens += usage["output_tokens"].as_u64().unwrap_or(0); + self.cached_tokens += usage["input_token_details"]["cached_tokens"] + .as_u64() + .unwrap_or(0); + self.responses += 1; + } + Some("conversation.item.input_audio_transcription.completed") => { + // Transcription-intent sessions bill via this frame + // (LiteLLM `_capture_transcription_usage`). + let usage = &v["usage"]; + if usage.get("type").and_then(Value::as_str) == Some("tokens") { + self.input_tokens += usage["input_tokens"].as_u64().unwrap_or(0); + self.output_tokens += usage["output_tokens"].as_u64().unwrap_or(0); + self.responses += 1; + } + } + _ => {} + } + } +} + +async fn run_session( + state: ProxyState, + prep: Prepared, + client_ws: WebSocket, + client: ClientContext, + request_id: String, + started: Instant, +) { + let Prepared { + auth, + model_entry, + pk_id, + upstream_request, + reservation, + requested_model, + provider_label, + } = prep; + + let guardrail_ctx = aisix_guardrails::RequestContext { + model_id: &model_entry.id, + api_key_id: &auth.entry.id, + team_id: auth.key().team_id.as_deref(), + }; + let chain = state.guardrail_index.resolve(&guardrail_ctx); + + let (mut client_tx, mut client_rx) = client_ws.split(); + + let upstream = match tokio_tungstenite::connect_async(upstream_request).await { + Ok((ws, _resp)) => ws, + Err(e) => { + tracing::warn!(error = %e, model = %requested_model, "realtime upstream connect failed"); + let _ = client_tx + .send(AxMessage::Text( + serde_json::json!({ + "type": "error", + "error": { + "type": "upstream_error", + "message": "failed to connect to the upstream realtime endpoint" + } + }) + .to_string(), + )) + .await; + let _ = client_tx + .send(AxMessage::Close(Some(CloseFrame { + code: 1011, + reason: "upstream connect failed".into(), + }))) + .await; + crate::cooldown::note_failure( + &state.runtime_status, + &model_entry.id, + model_entry.value.cooldown.as_ref(), + aisix_gateway::BridgeError::Transport(e.to_string()), + ); + emit_access_log( + &Method::GET, + 502, + started.elapsed(), + &request_id, + Some((&provider_label, &requested_model)), + ); + crate::usage_attr::emit_error_usage_event( + &state, + "realtime", + &request_id, + &requested_model, + &auth.entry.id, + 502, + "transport", + &client, + ); + return; + } + }; + let (mut up_tx, mut up_rx) = upstream.split(); + + let mut usage = SessionUsage::default(); + let mut close_status: u16 = 200; + // Operator-configured stream idle deadline (stream_timeout on the + // Model). Absent → no idle cap; realtime sessions are long-lived by + // design. + let idle_cap = model_entry.value.stream_timeout_effective(); + + loop { + let next = async { + tokio::select! { + m = client_rx.next() => Dir::FromClient(m), + m = up_rx.next() => Dir::FromUpstream(m), + } + }; + let event = match idle_cap { + Some(cap) => match tokio::time::timeout(cap, next).await { + Ok(r) => r, + Err(_) => { + let _ = client_tx + .send(AxMessage::Close(Some(CloseFrame { + code: 1001, + reason: "idle timeout".into(), + }))) + .await; + close_status = 504; + break; + } + }, + None => next.await, + }; + + match event { + Dir::FromClient(m) => match m { + Some(Ok(AxMessage::Text(text))) => { + if !chain.is_empty() { + if let Some(resp) = guardrail_block_event( + &chain, + &model_entry.value.display_name, + &text, + true, + ) + .await + { + let _ = client_tx.send(AxMessage::Text(resp)).await; + let _ = client_tx + .send(AxMessage::Close(Some(CloseFrame { + code: 1011, + reason: "content policy".into(), + }))) + .await; + close_status = 400; + break; + } + } + if up_tx.send(TgMessage::Text(text)).await.is_err() { + break; + } + } + Some(Ok(AxMessage::Binary(b))) => { + if up_tx.send(TgMessage::Binary(b)).await.is_err() { + break; + } + } + Some(Ok(AxMessage::Close(_))) | None => { + let _ = up_tx.send(TgMessage::Close(None)).await; + break; + } + Some(Ok(_)) => {} // ping/pong handled by the transports + Some(Err(_)) => { + let _ = up_tx.send(TgMessage::Close(None)).await; + break; + } + }, + Dir::FromUpstream(m) => match m { + Some(Ok(TgMessage::Text(text))) => { + usage.absorb(&text); + if !chain.is_empty() { + if let Some(resp) = guardrail_block_event( + &chain, + &model_entry.value.display_name, + &text, + false, + ) + .await + { + let _ = client_tx.send(AxMessage::Text(resp)).await; + let _ = client_tx + .send(AxMessage::Close(Some(CloseFrame { + code: 1011, + reason: "content policy".into(), + }))) + .await; + close_status = 400; + break; + } + } + if client_tx.send(AxMessage::Text(text)).await.is_err() { + break; + } + } + Some(Ok(TgMessage::Binary(b))) => { + if client_tx.send(AxMessage::Binary(b)).await.is_err() { + break; + } + } + Some(Ok(TgMessage::Close(frame))) => { + let _ = client_tx + .send(AxMessage::Close(frame.map(|f| CloseFrame { + code: f.code.into(), + reason: f.reason.to_string().into(), + }))) + .await; + break; + } + Some(Ok(_)) => {} // ping/pong/raw frames + Some(Err(e)) => { + tracing::debug!(error = %e, "realtime upstream stream error"); + let _ = client_tx + .send(AxMessage::Close(Some(CloseFrame { + code: 1011, + reason: "upstream error".into(), + }))) + .await; + close_status = 502; + break; + } + None => { + let _ = client_tx.send(AxMessage::Close(None)).await; + break; + } + }, + } + } + + let elapsed = started.elapsed(); + let total_tokens = usage.input_tokens + usage.output_tokens; + reservation.commit_tokens(total_tokens).await; + + emit_access_log( + &Method::GET, + close_status, + elapsed, + &request_id, + Some((&provider_label, &requested_model)), + ); + state.metrics.record_request( + &provider_label, + &model_entry.value.display_name, + close_status, + RequestOutcome::from_status(close_status), + elapsed, + ); + + let snap = state.snapshot.load(); + let mut event = UsageEvent { + request_id: request_id.clone(), + occurred_at: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true), + model_id: model_entry.id.clone(), + api_key_id: auth.entry.id.clone(), + requested_model: requested_model.clone(), + prompt_tokens: usage.input_tokens.min(u32::MAX as u64) as u32, + completion_tokens: usage.output_tokens.min(u32::MAX as u64) as u32, + cached_prompt_tokens: usage.cached_tokens.min(u32::MAX as u64) as u32, + status_code: close_status, + latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32, + cost_usd: model_entry + .value + .cost + .as_ref() + .map(|c| c.calculate(usage.input_tokens, usage.output_tokens)) + .unwrap_or(0.0), + inbound_protocol: "realtime".to_string(), + client_source_ip: client.source_ip.clone(), + client_user_agent: client.user_agent.clone(), + ..Default::default() + }; + crate::usage_attr::apply_pk_telemetry(&mut event, &snap, &pk_id); + state.usage_sink.try_emit("realtime", event.clone()); + let exporters = snap.observability_exporters.entries(); + state + .otlp_fan_out + .fan_out(&event, None, exporters.iter().map(|e| &e.value)); +} + +enum Dir { + FromClient(Option>), + FromUpstream(Option>), +} + +/// Whole-frame guardrail scan (the `/passthrough` blob precedent applied +/// per WS text frame). Returns the client-facing error event on Block. +async fn guardrail_block_event( + chain: &aisix_guardrails::GuardrailChain, + model_name: &str, + text: &str, + input_side: bool, +) -> Option { + let verdict = if input_side { + let chat = aisix_gateway::ChatFormat::new( + model_name, + vec![aisix_gateway::ChatMessage::user(text.to_string())], + ); + aisix_guardrails::Guardrail::check_input(chain, &chat).await + } else { + let synth = aisix_gateway::ChatResponse { + id: String::new(), + model: model_name.to_string(), + message: aisix_gateway::ChatMessage::assistant(text.to_string()), + finish_reason: aisix_gateway::FinishReason::Stop, + usage: aisix_gateway::UsageStats::default(), + }; + aisix_guardrails::Guardrail::check_output(chain, &synth).await + }; + if let aisix_guardrails::GuardrailVerdict::Block { + reason, + guardrail_name, + } = verdict + { + let side = if input_side { "input" } else { "output" }; + tracing::warn!( + guardrail_hook = side, + reason = %reason, + "guardrail blocked realtime frame", + ); + let msg = crate::error::guardrail_block_message( + if input_side { "request" } else { "response" }, + guardrail_name.as_deref(), + ); + return Some( + serde_json::json!({ + "type": "error", + "error": {"type": "invalid_request_error", "code": "content_filtered", "message": msg} + }) + .to_string(), + ); + } + None +} + +fn emit_access_log( + method: &Method, + status: u16, + elapsed: Duration, + request_id: &str, + target: Option<(&str, &str)>, +) { + AccessLog { + method: method.as_str(), + path: "/v1/realtime", + status, + latency: elapsed, + provider: target.map(|(p, _)| p).filter(|p| !p.is_empty()), + model: target.map(|(_, m)| m), + api_key_id: None, + prompt_tokens: None, + completion_tokens: None, + total_tokens: None, + request_id, + served_by_model: None, + routing_attempt_count: None, + routing_fallback_count: None, + } + .emit(); +} + +#[cfg(test)] +mod tests { + use super::*; + use aisix_core::resource::ResourceEntry; + use aisix_core::snapshot::SnapshotHandle; + use aisix_core::{AisixSnapshot, ApiKey, Model, ProxyConfig}; + use aisix_gateway::Hub; + use aisix_obs::{UsageEvent as ObsUsageEvent, UsageSink}; + use futures::{SinkExt, StreamExt}; + use std::sync::{Arc, Mutex}; + use tokio_tungstenite::tungstenite::client::IntoClientRequest; + + fn cfg() -> ProxyConfig { + ProxyConfig { + addr: "127.0.0.1:0".into(), + request_body_limit_bytes: 1_048_576, + real_ip: Default::default(), + tls: None, + } + } + + const PK_ID: &str = "22222222-2222-2222-2222-222222222222"; + // sha256("sk-caller") — the plaintext used by all tests below. + const CALLER_HASH: &str = "8b6712790a2089c67aa97a2d80022df18cc65c7814350e33baebe79aab508891"; + + fn snapshot(api_base: &str, adapter: &str, provider: &str) -> AisixSnapshot { + let snap = AisixSnapshot::new(); + let pk_json = format!( + r#"{{"display_name":"rt-pk","secret":"sk-up","api_base":"{api_base}","provider":"{provider}","adapter":"{adapter}"}}"# + ); + let pk: aisix_core::ProviderKey = serde_json::from_str(&pk_json).unwrap(); + snap.provider_keys.insert(ResourceEntry::new(PK_ID, pk, 1)); + let m_json = format!( + r#"{{"display_name":"rt-model","provider":"{provider}","model_name":"gpt-realtime","provider_key_id":"{PK_ID}"}}"# + ); + let m: Model = serde_json::from_str(&m_json).unwrap(); + snap.models.insert(ResourceEntry::new("m-rt", m, 1)); + let k_json = format!(r#"{{"key_hash":"{CALLER_HASH}","allowed_models":["*"]}}"#); + let k: ApiKey = serde_json::from_str(&k_json).unwrap(); + snap.apikeys.insert(ResourceEntry::new("k-1", k, 1)); + snap + } + + /// Bind the full proxy router on a real TCP port (WS handshakes need a + /// live connection; `oneshot` can't upgrade). + async fn serve( + snap: AisixSnapshot, + ) -> ( + std::net::SocketAddr, + tokio::sync::mpsc::Receiver, + ) { + let (tx, rx) = tokio::sync::mpsc::channel::(16); + let hub = Arc::new(Hub::new()); + let handle = SnapshotHandle::new(snap); + let state = crate::ProxyState::new(handle, hub, &cfg()) + .without_cache() + .with_usage_sink(UsageSink::new(tx)); + let app = crate::build_router(state); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app.into_make_service()) + .await + .unwrap(); + }); + (addr, rx) + } + + /// Scripted mock upstream: accepts ONE WebSocket, records the request + /// path + auth header, waits for one text frame, replies with a + /// `response.done` usage frame, then closes. + async fn spawn_upstream() -> ( + std::net::SocketAddr, + Arc>>, + Arc>>, + ) { + let seen_handshake: Arc>> = Arc::new(Mutex::new(None)); + let seen_frames: Arc>> = Arc::new(Mutex::new(Vec::new())); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let hs = seen_handshake.clone(); + let frames = seen_frames.clone(); + tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let hs2 = hs.clone(); + let ws = tokio_tungstenite::accept_hdr_async( + stream, + move |req: &tokio_tungstenite::tungstenite::handshake::server::Request, + resp: tokio_tungstenite::tungstenite::handshake::server::Response| { + let auth = req + .headers() + .get("authorization") + .and_then(|v| v.to_str().ok()) + .unwrap_or("") + .to_string(); + *hs2.lock().unwrap() = Some((req.uri().to_string(), auth)); + Ok(resp) + }, + ) + .await + .unwrap(); + let (mut tx, mut rx) = ws.split(); + while let Some(Ok(msg)) = rx.next().await { + if let TgMessage::Text(t) = msg { + frames.lock().unwrap().push(t.clone()); + tx.send(TgMessage::Text( + serde_json::json!({ + "type": "response.done", + "response": {"usage": { + "input_tokens": 7, + "output_tokens": 3, + "input_token_details": {"cached_tokens": 1} + }} + }) + .to_string(), + )) + .await + .unwrap(); + tx.send(TgMessage::Close(None)).await.ok(); + break; + } + } + }); + (addr, seen_handshake, seen_frames) + } + + #[tokio::test] + async fn relays_frames_and_emits_aggregated_usage_event() { + let (up_addr, handshake, frames) = spawn_upstream().await; + let snap = snapshot(&format!("http://{up_addr}/v1"), "openai", "openai"); + let (addr, mut rx) = serve(snap).await; + + let mut req = format!("ws://{addr}/v1/realtime?model=rt-model") + .into_client_request() + .unwrap(); + req.headers_mut() + .insert("authorization", "Bearer sk-caller".parse().unwrap()); + let (ws, _) = tokio_tungstenite::connect_async(req) + .await + .expect("handshake"); + let (mut tx, mut client_rx) = ws.split(); + + tx.send(TgMessage::Text( + serde_json::json!({"type": "session.update", "session": {"instructions": "hi"}}) + .to_string(), + )) + .await + .unwrap(); + + // The upstream's response.done frame must reach the client verbatim. + let mut got_done = false; + while let Some(Ok(msg)) = client_rx.next().await { + match msg { + TgMessage::Text(t) if t.contains("response.done") => { + got_done = true; + } + TgMessage::Close(_) => break, + _ => {} + } + } + assert!(got_done, "client must receive the upstream response.done"); + + // Upstream saw the relayed client frame + the gateway's provider auth. + assert_eq!(frames.lock().unwrap().len(), 1); + assert!(frames.lock().unwrap()[0].contains("session.update")); + let (uri, auth) = handshake + .lock() + .unwrap() + .clone() + .expect("handshake recorded"); + assert!( + uri.contains("/v1/realtime") && uri.contains("model=gpt-realtime"), + "upstream URI must be the realtime path with the UPSTREAM model id, got {uri}" + ); + assert_eq!(auth, "Bearer sk-up"); + + // Session-aggregate usage event. + let ev = tokio::time::timeout(Duration::from_secs(3), rx.recv()) + .await + .expect("usage event expected") + .expect("sink closed"); + assert_eq!(ev.inbound_protocol, "realtime"); + assert_eq!(ev.prompt_tokens, 7); + assert_eq!(ev.completion_tokens, 3); + assert_eq!(ev.cached_prompt_tokens, 1); + assert_eq!(ev.requested_model, "rt-model"); + assert_eq!(ev.api_key_id, "k-1"); + } + + #[tokio::test] + async fn subprotocol_key_authenticates_and_realtime_is_echoed() { + let (up_addr, _handshake, _frames) = spawn_upstream().await; + let snap = snapshot(&format!("http://{up_addr}/v1"), "openai", "openai"); + let (addr, _rx) = serve(snap).await; + + let mut req = format!("ws://{addr}/v1/realtime?model=rt-model") + .into_client_request() + .unwrap(); + // Browser flow: no headers, key rides the subprotocol list. + req.headers_mut().insert( + "sec-websocket-protocol", + "realtime, openai-insecure-api-key.sk-caller, openai-beta.realtime-v1" + .parse() + .unwrap(), + ); + let (ws, resp) = tokio_tungstenite::connect_async(req) + .await + .expect("subprotocol auth must be accepted"); + assert_eq!( + resp.headers() + .get("sec-websocket-protocol") + .and_then(|v| v.to_str().ok()), + Some("realtime"), + "the gateway must echo the `realtime` subprotocol" + ); + drop(ws); + } + + #[tokio::test] + async fn missing_auth_rejects_the_handshake() { + let snap = snapshot("http://127.0.0.1:9/v1", "openai", "openai"); + let (addr, _rx) = serve(snap).await; + + let req = format!("ws://{addr}/v1/realtime?model=rt-model") + .into_client_request() + .unwrap(); + let err = tokio_tungstenite::connect_async(req) + .await + .expect_err("handshake must fail without credentials"); + let msg = err.to_string(); + assert!(msg.contains("401"), "expected 401 rejection, got: {msg}"); + } + + #[tokio::test] + async fn missing_model_param_rejects_with_400() { + let snap = snapshot("http://127.0.0.1:9/v1", "openai", "openai"); + let (addr, _rx) = serve(snap).await; + + let mut req = format!("ws://{addr}/v1/realtime") + .into_client_request() + .unwrap(); + req.headers_mut() + .insert("authorization", "Bearer sk-caller".parse().unwrap()); + let err = tokio_tungstenite::connect_async(req) + .await + .expect_err("handshake must fail without ?model="); + assert!(err.to_string().contains("400"), "got: {err}"); + } + + #[tokio::test] + async fn non_realtime_capable_adapter_is_rejected() { + let snap = snapshot("http://127.0.0.1:9", "anthropic", "anthropic"); + let (addr, _rx) = serve(snap).await; + + let mut req = format!("ws://{addr}/v1/realtime?model=rt-model") + .into_client_request() + .unwrap(); + req.headers_mut() + .insert("authorization", "Bearer sk-caller".parse().unwrap()); + let err = tokio_tungstenite::connect_async(req) + .await + .expect_err("handshake must fail on a non-OpenAI-protocol provider"); + assert!(err.to_string().contains("400"), "got: {err}"); + } + + #[tokio::test] + async fn model_acl_rejects_unauthorized_key() { + let (up_addr, _h, _f) = spawn_upstream().await; + let snap = snapshot(&format!("http://{up_addr}/v1"), "openai", "openai"); + // Restrict the caller key to a different model. + let k_json = format!(r#"{{"key_hash":"{CALLER_HASH}","allowed_models":["other-model"]}}"#); + let k: ApiKey = serde_json::from_str(&k_json).unwrap(); + snap.apikeys.insert(ResourceEntry::new("k-1", k, 2)); + let (addr, _rx) = serve(snap).await; + + let mut req = format!("ws://{addr}/v1/realtime?model=rt-model") + .into_client_request() + .unwrap(); + req.headers_mut() + .insert("authorization", "Bearer sk-caller".parse().unwrap()); + let err = tokio_tungstenite::connect_async(req) + .await + .expect_err("handshake must fail on model ACL"); + assert!(err.to_string().contains("403"), "got: {err}"); + } +} diff --git a/tests/e2e/package.json b/tests/e2e/package.json index 099cc4a6..16b72205 100644 --- a/tests/e2e/package.json +++ b/tests/e2e/package.json @@ -11,9 +11,11 @@ }, "devDependencies": { "@types/node": "20.11.30", + "@types/ws": "8.5.12", "@vitest/coverage-v8": "2.1.9", "typescript": "5.5.4", - "vitest": "2.1.9" + "vitest": "2.1.9", + "ws": "8.18.0" }, "dependencies": { "openai": "4.65.0", diff --git a/tests/e2e/src/cases/realtime-ws-e2e.test.ts b/tests/e2e/src/cases/realtime-ws-e2e.test.ts new file mode 100644 index 00000000..a7d7cc25 --- /dev/null +++ b/tests/e2e/src/cases/realtime-ws-e2e.test.ts @@ -0,0 +1,185 @@ +import { createHash } from "node:crypto"; +import { WebSocketServer, type WebSocket as WsSocket } from "ws"; +import { afterAll, beforeAll, describe, expect, test } from "vitest"; +import { + AdminClient, + EtcdClient, + spawnApp, + type SpawnedApp, +} from "../harness/index.js"; + +// E2E: /v1/realtime WebSocket relay (#721, AISIX-Cloud#873 §⑤) against a +// real `aisix` binary. Verifies with a live WS handshake what unit tests +// can't fully pin: +// +// 1. Browser-flow auth: the caller key rides the +// `openai-insecure-api-key.` subprotocol item (Node's native +// WebSocket client can't set headers — exactly like a browser), and +// the gateway echoes the `realtime` subprotocol. +// 2. Bidirectional frame relay: a client event reaches the mock +// upstream verbatim; the upstream's `response.done` reaches the +// client verbatim. +// 3. The upstream handshake carries the PROVIDER credential and the +// UPSTREAM model id (`?model=gpt-realtime-mock`), not the caller's +// key or the gateway alias. +// 4. Auth failure rejects the HTTP upgrade (native client fires +// an error/close, never `open`). + +const CALLER_PLAINTEXT = "sk-realtime-e2e-caller"; +const CALLER_KEY_HASH = createHash("sha256") + .update(CALLER_PLAINTEXT) + .digest("hex"); + +interface RealtimeUpstream { + port: number; + handshakes: { url: string; authorization: string }[]; + frames: string[]; + close(): Promise; +} + +/** Mock OpenAI Realtime upstream: records the handshake, then answers the + * first client event with a usage-bearing `response.done` frame. */ +async function startRealtimeUpstream(): Promise { + const handshakes: RealtimeUpstream["handshakes"] = []; + const frames: string[] = []; + const wss = new WebSocketServer({ host: "127.0.0.1", port: 0 }); + wss.on("connection", (socket: WsSocket, req) => { + handshakes.push({ + url: req.url ?? "", + authorization: (req.headers.authorization as string) ?? "", + }); + socket.on("message", (data) => { + frames.push(data.toString()); + socket.send( + JSON.stringify({ + type: "response.done", + response: { + usage: { + input_tokens: 9, + output_tokens: 4, + input_token_details: { cached_tokens: 0 }, + }, + }, + }), + ); + }); + }); + await new Promise((resolve) => wss.on("listening", resolve)); + const addr = wss.address(); + if (addr === null || typeof addr === "string") throw new Error("no port"); + return { + port: addr.port, + handshakes, + frames, + close: () => + new Promise((resolve, reject) => + wss.close((e) => (e ? reject(e) : resolve())), + ), + }; +} + +describe("realtime e2e: /v1/realtime WebSocket relay (#721)", () => { + let app: SpawnedApp | undefined; + let admin: AdminClient | undefined; + let upstream: RealtimeUpstream | undefined; + let etcdReachable = false; + + beforeAll(async () => { + etcdReachable = await new EtcdClient().ping(); + if (!etcdReachable) return; + + app = await spawnApp(); + admin = new AdminClient(app.adminUrl, app.adminKey); + upstream = await startRealtimeUpstream(); + + await admin.createApiKey({ + key_hash: CALLER_KEY_HASH, + allowed_models: ["*"], + }); + const pk = await admin.createProviderKey({ + display_name: "realtime-e2e-pk", + secret: "sk-upstream-realtime", + api_base: `http://127.0.0.1:${upstream.port}/v1`, + }); + await admin.createModel({ + display_name: "realtime-e2e-model", + provider: "openai", + model_name: "gpt-realtime-mock", + provider_key_id: pk.id, + }); + }); + + afterAll(async () => { + await app?.exit(); + await upstream?.close(); + }); + + test("browser-flow subprotocol auth + bidirectional relay + upstream credential swap", async (ctx) => { + if (!etcdReachable || !app || !upstream) { + ctx.skip(); + return; + } + + const wsUrl = `${app.proxyUrl.replace("http://", "ws://")}/v1/realtime?model=realtime-e2e-model`; + // Node's NATIVE WebSocket (undici) — cannot set headers, exactly the + // browser constraint the subprotocol flow exists for. + const ws = new WebSocket(wsUrl, [ + "realtime", + `openai-insecure-api-key.${CALLER_PLAINTEXT}`, + "openai-beta.realtime-v1", + ]); + + const opened = new Promise((resolve, reject) => { + ws.addEventListener("open", () => resolve(), { once: true }); + ws.addEventListener("error", () => reject(new Error("handshake failed")), { + once: true, + }); + }); + await opened; + expect(ws.protocol).toBe("realtime"); + + const done = new Promise((resolve) => { + ws.addEventListener("message", (ev) => resolve(String(ev.data)), { + once: true, + }); + }); + ws.send( + JSON.stringify({ type: "session.update", session: { instructions: "hi" } }), + ); + const frame = JSON.parse(await done) as { + type: string; + response: { usage: { input_tokens: number } }; + }; + expect(frame.type).toBe("response.done"); + expect(frame.response.usage.input_tokens).toBe(9); + + // Upstream saw the relayed event, the provider credential, and the + // upstream model id. + expect(upstream.frames.some((f) => f.includes("session.update"))).toBe(true); + expect(upstream.handshakes.length).toBe(1); + expect(upstream.handshakes[0].authorization).toBe( + "Bearer sk-upstream-realtime", + ); + expect(upstream.handshakes[0].url).toContain("model=gpt-realtime-mock"); + expect(upstream.handshakes[0].url).not.toContain(CALLER_PLAINTEXT); + + ws.close(); + }); + + test("bad credentials reject the upgrade handshake", async (ctx) => { + if (!etcdReachable || !app) { + ctx.skip(); + return; + } + const wsUrl = `${app.proxyUrl.replace("http://", "ws://")}/v1/realtime?model=realtime-e2e-model`; + const ws = new WebSocket(wsUrl, [ + "realtime", + "openai-insecure-api-key.sk-wrong", + ]); + const failed = await new Promise((resolve) => { + ws.addEventListener("open", () => resolve(false), { once: true }); + ws.addEventListener("error", () => resolve(true), { once: true }); + }); + expect(failed).toBe(true); + }); +}); From c2fde4d5d873cc9bba4c0a3e36196cd52b4bd2c9 Mon Sep 17 00:00:00 2001 From: Jarvis Date: Mon, 6 Jul 2026 16:55:01 +0800 Subject: [PATCH 2/2] =?UTF-8?q?test(e2e):=20use=20undici=20WebSocket=20?= =?UTF-8?q?=E2=80=94=20Node=2020=20CI=20has=20no=20global=20WebSocket?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/e2e/src/cases/realtime-ws-e2e.test.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/e2e/src/cases/realtime-ws-e2e.test.ts b/tests/e2e/src/cases/realtime-ws-e2e.test.ts index a7d7cc25..5d68b43e 100644 --- a/tests/e2e/src/cases/realtime-ws-e2e.test.ts +++ b/tests/e2e/src/cases/realtime-ws-e2e.test.ts @@ -1,4 +1,5 @@ import { createHash } from "node:crypto"; +import { WebSocket } from "undici"; import { WebSocketServer, type WebSocket as WsSocket } from "ws"; import { afterAll, beforeAll, describe, expect, test } from "vitest"; import { @@ -121,8 +122,9 @@ describe("realtime e2e: /v1/realtime WebSocket relay (#721)", () => { } const wsUrl = `${app.proxyUrl.replace("http://", "ws://")}/v1/realtime?model=realtime-e2e-model`; - // Node's NATIVE WebSocket (undici) — cannot set headers, exactly the - // browser constraint the subprotocol flow exists for. + // undici's browser-style WebSocket — cannot set headers, exactly the + // browser constraint the subprotocol flow exists for (Node 20 CI has + // no global WebSocket, so import it from undici explicitly). const ws = new WebSocket(wsUrl, [ "realtime", `openai-insecure-api-key.${CALLER_PLAINTEXT}`,