From 22acbd1d10a32b8fab79bd8f8765fdfc8239ee2d Mon Sep 17 00:00:00 2001 From: 2han9wen71an <2han9wen71an@gmail.com> Date: Mon, 5 Oct 2026 23:48:55 +0800 Subject: [PATCH] feat(openai): add Codex live voice call proxying (#308) Add scoped call bootstrap, pinned-account sideband relay and hangup. Preserve model metadata, use standard WebSocket handshakes, and release session claims safely on cancellation. Validated with regression tests, CI and a real OAuth/WebRTC call. Co-authored-by: Codex --- backend/Cargo.lock | 18 + backend/crates/gateway-api/Cargo.toml | 2 +- .../crates/gateway-api/src/openai/endpoint.rs | 4 +- .../gateway-api/src/openai/live/http.rs | 383 +++++++++ .../crates/gateway-api/src/openai/live/mod.rs | 350 ++++++++ .../gateway-api/src/openai/live/websocket.rs | 296 +++++++ backend/crates/gateway-api/src/openai/mod.rs | 1 + .../crates/gateway-api/src/openai/router.rs | 2 + .../crates/gateway-api/src/openai/service.rs | 11 +- .../crates/gateway-api/tests/architecture.rs | 5 + .../gateway-api/tests/openai/live/mod.rs | 324 ++++++++ .../tests/openai/live/websocket.rs | 208 +++++ .../crates/gateway-api/tests/openai/mod.rs | 1 + .../gateway-core/src/engine/execution.rs | 12 + .../gateway-core/src/engine/provider.rs | 11 + backend/crates/gateway-core/src/lib.rs | 1 + backend/crates/gateway-core/src/live.rs | 327 ++++++++ .../gateway-core/src/routing/snapshot.rs | 9 +- backend/crates/gateway-core/tests/live.rs | 45 ++ backend/crates/gateway-core/tests/main.rs | 1 + .../crates/gateway-core/tests/routing/mod.rs | 57 +- .../openai/src/credential/selector.rs | 17 +- backend/crates/providers/openai/src/lib.rs | 3 +- .../openai/src/provider/execution.rs | 3 + .../providers/openai/src/provider/live.rs | 750 ++++++++++++++++++ .../providers/openai/src/provider/mod.rs | 50 +- .../openai/src/transport/client_live.rs | 231 ++++++ .../openai/src/transport/endpoints.rs | 2 + .../providers/openai/src/transport/mod.rs | 8 +- .../src/transport/websocket/handshake.rs | 7 + .../openai/src/transport/websocket/live.rs | 162 ++++ .../openai/src/transport/websocket/mod.rs | 2 + .../crates/providers/openai/tests/admin.rs | 15 + .../openai/tests/provider/contract/mod.rs | 11 +- .../providers/openai/tests/provider/live.rs | 234 ++++++ .../providers/openai/tests/provider/mod.rs | 1 + docs/api.md | 39 + 37 files changed, 3563 insertions(+), 40 deletions(-) create mode 100644 backend/crates/gateway-api/src/openai/live/http.rs create mode 100644 backend/crates/gateway-api/src/openai/live/mod.rs create mode 100644 backend/crates/gateway-api/src/openai/live/websocket.rs create mode 100644 backend/crates/gateway-api/tests/openai/live/mod.rs create mode 100644 backend/crates/gateway-api/tests/openai/live/websocket.rs create mode 100644 backend/crates/gateway-core/src/live.rs create mode 100644 backend/crates/gateway-core/tests/live.rs create mode 100644 backend/crates/providers/openai/src/provider/live.rs create mode 100644 backend/crates/providers/openai/src/transport/client_live.rs create mode 100644 backend/crates/providers/openai/src/transport/websocket/live.rs create mode 100644 backend/crates/providers/openai/tests/provider/live.rs diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 5bc9950f8..8c516173d 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -573,6 +573,7 @@ dependencies = [ "matchit", "memchr", "mime", + "multer", "percent-encoding", "pin-project-lite", "serde_core", @@ -2698,6 +2699,23 @@ dependencies = [ "pxfm", ] +[[package]] +name = "multer" +version = "3.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "83e87776546dc87511aa5ee218730c92b666d7264ab6ed41f9d215af9cd5224b" +dependencies = [ + "bytes", + "encoding_rs", + "futures-util", + "http 1.5.0", + "httparse", + "memchr", + "mime", + "spin 0.9.9", + "version_check", +] + [[package]] name = "native-tls" version = "0.2.14" diff --git a/backend/crates/gateway-api/Cargo.toml b/backend/crates/gateway-api/Cargo.toml index d3c5a4a00..9606754fd 100644 --- a/backend/crates/gateway-api/Cargo.toml +++ b/backend/crates/gateway-api/Cargo.toml @@ -12,7 +12,7 @@ name = "main" path = "tests/main.rs" [dependencies] -axum = { version = "0.8.9", default-features = false, features = ["json", "query", "ws"] } +axum = { version = "0.8.9", default-features = false, features = ["json", "multipart", "query", "ws"] } base64 = "0.23.0" bytes = "1.11.1" chrono = { version = "0.4.45", features = ["serde"] } diff --git a/backend/crates/gateway-api/src/openai/endpoint.rs b/backend/crates/gateway-api/src/openai/endpoint.rs index 4b2fb526b..b36a43fe2 100644 --- a/backend/crates/gateway-api/src/openai/endpoint.rs +++ b/backend/crates/gateway-api/src/openai/endpoint.rs @@ -61,7 +61,9 @@ where let operation = request.apply_capabilities(decode(body, &headers)?)?; let started = match service .start_prepared_provider_endpoint( - prepared, operation, client_ip, user_agent, endpoint, + prepared, operation, + // 图像/搜索等原生端点没有对应的上游模型权限。 + None, client_ip, user_agent, endpoint, ) .await { diff --git a/backend/crates/gateway-api/src/openai/live/http.rs b/backend/crates/gateway-api/src/openai/live/http.rs new file mode 100644 index 000000000..7a4a4b290 --- /dev/null +++ b/backend/crates/gateway-api/src/openai/live/http.rs @@ -0,0 +1,383 @@ +//! Live 通话引导（`POST /v1/live`）与 hangup 的 HTTP adapter。 + +use std::net::SocketAddr; + +use axum::{ + body::Bytes, + extract::{ConnectInfo, Extension, FromRequest, Multipart, Request, State}, + http::{HeaderMap, StatusCode, Uri, header}, + response::{IntoResponse, Response}, +}; +use bytes::{BytesMut, buf::BufMut}; +use gateway_core::engine::execution::StartedExecution; +use gateway_core::event::ProviderEvent; +use gateway_core::live::LiveHangupRequest; +use gateway_core::operation::Operation; +use serde_json::Value; + +use super::{ + LiveErrorShape, MAX_LIVE_BODY_BYTES, codex_realtime_model, filter_protocol_headers, + live_call_operation, live_error_response, live_forwarded_response_headers, + normalize_call_request, rewrite_call_request_model, +}; +use crate::{ + ApiState, + openai::{ + auth::{authenticate_client, client_access_error_response}, + error::{engine_error_response, gateway_error_response}, + middleware::PendingExecution, + responses::request_client_context, + with_model_request_id, + }, +}; + +/// 引导 wire 事件使用的协议名；与 Provider 侧 `PROVIDER_NAME` 一致。 +const LIVE_PROTOCOL: &str = "openai"; + +pub(crate) async fn live_call( + State(state): State, + connect_info: Option>>, + uri: Uri, + headers: HeaderMap, + request: Request, +) -> Response { + let service = state.openai(); + let client = match authenticate_client(service, &headers).await { + Ok(client) => client, + Err(error) => return client_access_error_response(error), + }; + let content_type = headers + .get(header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let (body, multipart) = match read_call_body(request, content_type.as_deref()).await { + Ok(parts) => parts, + Err(response) => return *response, + }; + let (upstream_body, upstream_content_type, model) = + match normalize_call_request(&body, content_type.as_deref(), multipart) { + Ok(parts) => parts, + Err(response) => return *response, + }; + let upstream_model = codex_realtime_model(model); + let upstream_body = rewrite_call_request_model(upstream_body, &upstream_model); + // 实际语音模型参与路由与账号权限检查：账号范围不含该模型时在路由层拒绝， + // 而不是写进正文后仍由未授权账号服务。 + let upstream_model_id = + match gateway_core::routing::UpstreamModelId::from_client_wire(upstream_model.clone()) { + Ok(model) => model, + Err(_) => { + return live_error_response( + LiveErrorShape::Live, + StatusCode::BAD_REQUEST, + "Codex live call request has an invalid model", + "invalid_request", + ); + } + }; + let operation = match live_call_operation( + upstream_body, + upstream_content_type.as_deref(), + filter_protocol_headers(&headers), + ) { + Ok(operation) => operation, + Err(response) => return *response, + }; + let (client_ip, user_agent) = request_client_context( + &headers, + connect_info.map(|Extension(ConnectInfo(address))| address), + ); + let prepared = match service.execution().prepare_execution(client).await { + Ok(prepared) => prepared, + Err(error) => return gateway_error_response(&error), + }; + let started = match service + .start_prepared_provider_endpoint( + prepared, + Operation::ProviderHttp(operation), + Some(upstream_model_id), + client_ip, + user_agent, + uri.path().to_owned(), + ) + .await + { + Ok(started) => started, + Err(error) => return gateway_error_response(&error), + }; + collect_live_call_response(started, uri.path().to_owned()).await +} + +pub(crate) async fn hangup( + State(state): State, + axum::extract::Path(call_id): axum::extract::Path, + headers: HeaderMap, + body: Bytes, +) -> Response { + let service = state.openai(); + let client = match authenticate_client(service, &headers).await { + Ok(client) => client, + Err(error) => return client_access_error_response(error), + }; + let Some(gateway) = service.live_gateway() else { + return super::realtime_unsupported_response("Realtime hangup"); + }; + let content_type = headers + .get(header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let protocol_headers = protocol_header_pairs(&headers); + let outcome = gateway + .hangup(LiveHangupRequest { + call_id: &call_id, + client_api_key_id: client.policy().key_id(), + content_type, + body, + protocol_headers, + }) + .await; + match outcome { + Ok(outcome) => { + let status = StatusCode::from_u16(outcome.status).unwrap_or(StatusCode::OK); + let mut response = Response::new(axum::body::Body::from(outcome.body)); + *response.status_mut() = status; + for (name, value) in outcome.headers { + if let (Ok(name), Ok(value)) = ( + header::HeaderName::from_bytes(name.as_bytes()), + header::HeaderValue::from_bytes(value.as_bytes()), + ) { + response.headers_mut().insert(name, value); + } + } + response + } + Err(error) => live_gateway_error_response(LiveErrorShape::Realtime, &error), + } +} + +/// LiveGateway 错误投影：上游显式状态与正文优先，其余用稳定错误码。 +pub(crate) fn live_gateway_error_response( + shape: LiveErrorShape, + error: &gateway_core::live::LiveGatewayError, +) -> Response { + let status = StatusCode::from_u16(error.status()).unwrap_or(StatusCode::SERVICE_UNAVAILABLE); + if let Some(body) = error.body() { + let mut response = Response::new(axum::body::Body::from(body.clone())); + *response.status_mut() = status; + response.headers_mut().insert( + header::CONTENT_TYPE, + header::HeaderValue::from_static("application/json"), + ); + return response; + } + live_error_response(shape, status, error.message(), error.kind().code()) +} + +/// 读取引导正文；multipart 入口在此拆成 `(sdp, session)` 字段。 +async fn read_call_body( + request: Request, + content_type: Option<&str>, +) -> Result<(Bytes, Option<(String, Option)>), Box> { + let is_multipart = content_type + .and_then(|value| value.split(';').next()) + .map(|media| media.trim().eq_ignore_ascii_case("multipart/form-data")) + .unwrap_or(false); + if !is_multipart { + let body = axum::body::to_bytes(request.into_body(), MAX_LIVE_BODY_BYTES) + .await + .map_err(|_| { + Box::new(live_error_response( + LiveErrorShape::Live, + StatusCode::PAYLOAD_TOO_LARGE, + "Codex live request body too large", + "invalid_request", + )) + })?; + return Ok((body, None)); + } + let mut multipart = Multipart::from_request(request, &()).await.map_err(|_| { + Box::new(live_error_response( + LiveErrorShape::Live, + StatusCode::BAD_REQUEST, + "Codex live multipart body is invalid", + "invalid_request", + )) + })?; + let mut sdp = None; + let mut session = None; + loop { + let Some(field) = multipart.next_field().await.map_err(|_| { + Box::new(live_error_response( + LiveErrorShape::Live, + StatusCode::BAD_REQUEST, + "Codex live multipart body is invalid", + "invalid_request", + )) + })? + else { + break; + }; + match field.name() { + Some("sdp") => { + let text = field.text().await.map_err(|_| { + Box::new(live_error_response( + LiveErrorShape::Live, + StatusCode::BAD_REQUEST, + "Codex live multipart body is invalid", + "invalid_request", + )) + })?; + sdp = Some(text); + } + Some("session") => { + let bytes = field.bytes().await.map_err(|_| { + Box::new(live_error_response( + LiveErrorShape::Live, + StatusCode::BAD_REQUEST, + "Codex live multipart body is invalid", + "invalid_request", + )) + })?; + let parsed = serde_json::from_slice::(&bytes).map_err(|_| { + Box::new(live_error_response( + LiveErrorShape::Live, + StatusCode::BAD_REQUEST, + "Codex live session field must contain valid JSON", + "invalid_request", + )) + })?; + session = Some(parsed); + } + _ => {} + } + } + match sdp { + Some(sdp) => Ok((Bytes::new(), Some((sdp, session)))), + None => Err(Box::new(live_error_response( + LiveErrorShape::Live, + StatusCode::BAD_REQUEST, + "Codex live multipart body requires an sdp field", + "invalid_request", + ))), + } +} + +/// realtime 协议头的 `(name, value)` 形态；名称小写化以稳定上游顺序。 +pub(crate) fn protocol_header_pairs(headers: &HeaderMap) -> Vec<(String, String)> { + super::filter_protocol_headers(headers) + .into_iter() + .map(|header| { + ( + header.name().to_ascii_lowercase(), + String::from_utf8_lossy(header.value()).into_owned(), + ) + }) + .collect() +} + +/// 汇集引导响应；只放行白名单响应头，正文按字节透传。 +async fn collect_live_call_response(started: StartedExecution, path: String) -> Response { + let request_id = started.request_id; + let mut execution = PendingExecution::new(started.session); + let Some(session) = execution.session_mut() else { + return invalid_upstream_response(); + }; + let events = match session.collect_uncommitted().await { + Ok(events) => events, + Err(error) => { + let response = engine_error_response(&error); + return execution.record_response_status(response).await; + } + }; + let Some(body) = live_call_body(events) else { + let response = execution + .record_response_status(invalid_upstream_response()) + .await; + execution.cancel_and_finalize().await; + return response; + }; + let status = match session.response_status_code() { + None => StatusCode::OK, + Some(status) => match StatusCode::from_u16(status) { + Ok(status) if status.is_success() => status, + _ => { + let response = execution + .record_response_status(invalid_upstream_response()) + .await; + execution.cancel_and_finalize().await; + return response; + } + }, + }; + let mut response = Response::new(axum::body::Body::from(body)); + *response.status_mut() = status; + for (name, value) in live_forwarded_response_headers(session.response_headers()) { + if name == header::CONTENT_TYPE { + response.headers_mut().insert(name, value); + } else { + response.headers_mut().append(name, value); + } + } + rewrite_realtime_location(&mut response, &path); + let Some(session) = execution.session_mut() else { + return invalid_upstream_response(); + }; + if let Err(error) = session.commit_downstream(Some(status.as_u16())).await { + execution.cancel_and_finalize().await; + let response = engine_error_response(&error); + return execution.record_response_status(response).await; + } + if !session.is_finalized() { + execution.cancel_and_finalize().await; + return invalid_upstream_response(); + } + execution.disarm(); + with_model_request_id(response, &request_id) +} + +/// `/v1/realtime*` 家族的 `Location` 改写为网关自身路径，保持客户端可回环。 +fn rewrite_realtime_location(response: &mut Response, path: &str) { + if !path.starts_with("/v1/realtime") { + return; + } + let Some(location) = response + .headers() + .get(header::LOCATION) + .and_then(|value| value.to_str().ok()) + .map(ToOwned::to_owned) + else { + return; + }; + let Some(call_id) = gateway_core::live::call_id_from_location(&location) else { + return; + }; + if let Ok(value) = header::HeaderValue::from_str(&format!("/v1/realtime/calls/{call_id}")) { + response.headers_mut().insert(header::LOCATION, value); + } +} + +fn live_call_body(events: Vec) -> Option { + let mut body = BytesMut::new(); + let mut raw_events = 0usize; + for event in events { + let (_, wire) = event.into_parts(); + let Some(wire) = wire.filter(|wire| wire.protocol() == LIVE_PROTOCOL) else { + continue; + }; + let raw = wire.into_raw_http_body()?; + raw_events += 1; + if raw_events > 1 || body.len() + raw.len() > MAX_LIVE_BODY_BYTES { + return None; + } + body.put_slice(&raw); + } + (raw_events == 1).then(|| body.freeze()) +} + +fn invalid_upstream_response() -> Response { + ( + StatusCode::BAD_GATEWAY, + "The gateway could not forward the Codex live response.", + ) + .into_response() +} diff --git a/backend/crates/gateway-api/src/openai/live/mod.rs b/backend/crates/gateway-api/src/openai/live/mod.rs new file mode 100644 index 000000000..e1b35b284 --- /dev/null +++ b/backend/crates/gateway-api/src/openai/live/mod.rs @@ -0,0 +1,350 @@ +//! Codex Live / Realtime 语音通话的协议 adapter。 +//! +//! 通话引导走 Provider HTTP 端点通道（保留客户端账号范围、准入与请求记录）； +//! 通话建立后的 sideband 中继走 [`gateway_core::live::LiveGateway`] 的钉住 +//! 账号拨号。上游合同对应官方 Codex 的 realtime_call 与 realtime_websocket： +//! `POST /v1/live` 引导 WebRTC SDP，`GET /v1/live/{call_id}` 提供事件 sideband。 + +mod http; +mod websocket; + +use axum::{ + Router, + body::Bytes, + extract::DefaultBodyLimit, + http::{HeaderMap, HeaderName, HeaderValue, StatusCode, header}, + response::{IntoResponse, Response}, + routing::{any, get, post}, +}; +use gateway_core::event::ProviderResponseHeader; +use gateway_core::operation::{ + ProviderHttpHeader, ProviderHttpMethod, ProviderHttpRequest, RawHttpPayload, +}; +use serde_json::{Value, json}; + +use crate::openai::error::openai_error_response; + +pub(crate) use http::{hangup, live_call}; +pub(crate) use websocket::{realtime_get, sideband}; + +/// 语音引导正文上限，同时约束 JSON、SDP 与 multipart 入口。 +pub(crate) const MAX_LIVE_BODY_BYTES: usize = 16 * 1024 * 1024; +/// Provider 侧识别 realtime calls 端点的符号名；endpoint 字段只允许 +/// 单段安全符号，真实上游路径 `/codex/realtime/calls` 由 Provider 映射。 +pub(crate) const LIVE_CALLS_ENDPOINT: &str = "realtime-calls"; +/// realtime calls 的官方查询参数；上游按 `architecture=avas` 返回 WebRTC answer。 +pub(crate) const LIVE_CALLS_QUERY: &str = "intent=quicksilver&architecture=avas"; +/// 客户端协议头允许清单；语音会话语义字段，转发给上游引导请求。 +pub(crate) const LIVE_PROTOCOL_HEADERS: [&str; 9] = [ + "openai-alpha", + "x-session-id", + "session-id", + "thread-id", + "originator", + "openai-safety-identifier", + "openai-organization", + "openai-project", + "x-oai-attestation", +]; + +pub(crate) fn router() -> Router { + Router::new() + .route("/v1/live", post(live_call)) + .route("/v1/live/{call_id}", get(sideband)) + .route("/v1/realtime", post(live_call).get(realtime_get)) + .route("/v1/realtime/calls", post(live_call)) + .route("/v1/realtime/calls/{call_id}", get(sideband)) + .route("/v1/realtime/calls/{call_id}/hangup", post(hangup)) + // Codex OAuth 上游与本网关都不提供这些 realtime 能力；显式 501 优于神秘 404。 + .route("/v1/realtime/translations", any(translation_unsupported)) + .route( + "/v1/realtime/translations/client_secrets", + any(translation_unsupported), + ) + .route( + "/v1/realtime/transcription_sessions", + post(transcription_unsupported), + ) + .route( + "/v1/realtime/client_secrets", + post(client_secrets_unsupported), + ) + .route("/v1/realtime/sessions", post(legacy_sessions_unsupported)) + .route("/v1/realtime/calls/{call_id}/accept", post(sip_unsupported)) + .route("/v1/realtime/calls/{call_id}/reject", post(sip_unsupported)) + .route("/v1/realtime/calls/{call_id}/refer", post(sip_unsupported)) + .layer(DefaultBodyLimit::max(MAX_LIVE_BODY_BYTES)) +} + +/// `/v1/live` 与 `/v1/realtime*` 的错误投影差异：前者是普通 JSON 错误对象， +/// realtime 家族沿用 OpenAI realtime 的结构化错误合同。 +#[derive(Clone, Copy, PartialEq, Eq)] +pub(crate) enum LiveErrorShape { + Live, + Realtime, +} + +pub(crate) fn live_error_response( + shape: LiveErrorShape, + status: StatusCode, + message: &str, + code: &str, +) -> Response { + match shape { + LiveErrorShape::Live => { + let body = json!({ "error": message }); + (status, axum::Json(body)).into_response() + } + LiveErrorShape::Realtime => { + openai_error_response(status, message, error_type_for_status(status), code) + .into_response() + } + } +} + +fn error_type_for_status(status: StatusCode) -> &'static str { + if status.is_client_error() { + "invalid_request_error" + } else { + "api_error" + } +} + +async fn translation_unsupported() -> Response { + realtime_unsupported_response("Realtime translation sessions") +} + +async fn transcription_unsupported() -> Response { + realtime_unsupported_response("Realtime transcription-only sessions") +} + +async fn client_secrets_unsupported() -> Response { + realtime_unsupported_response("Realtime client secrets") +} + +async fn legacy_sessions_unsupported() -> Response { + realtime_unsupported_response("Legacy realtime session credentials") +} + +async fn sip_unsupported() -> Response { + realtime_unsupported_response("Realtime SIP control") +} + +pub(crate) fn realtime_unsupported_response(capability: &str) -> Response { + openai_error_response( + StatusCode::NOT_IMPLEMENTED, + &format!("{capability} are not supported by the Codex OAuth upstream"), + "not_supported_error", + "realtime_capability_not_supported", + ) + .into_response() +} + +/// 请求侧协议头过滤；语音语义字段按允许清单转发，其余一律不透传。 +pub(crate) fn filter_protocol_headers(headers: &HeaderMap) -> Vec { + LIVE_PROTOCOL_HEADERS + .iter() + .flat_map(|name| { + headers + .get_all(*name) + .into_iter() + .map(|value| (*name, value)) + }) + .filter_map(|(name, value)| { + let value = value.to_str().ok()?; + Some(ProviderHttpHeader::new( + name, + Bytes::copy_from_slice(value.as_bytes()), + )) + }) + .collect() +} + +/// 客户端 offer 的 WebSocket 子协议，按出现顺序透传给上游协商。 +pub(crate) fn offered_subprotocols(headers: &HeaderMap) -> Vec { + headers + .get_all(header::SEC_WEBSOCKET_PROTOCOL) + .into_iter() + .filter_map(|value| value.to_str().ok()) + .flat_map(|value| value.split(',')) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToOwned::to_owned) + .collect() +} + +/// 把协议头与正文装配为 realtime calls 引导 operation。 +pub(crate) fn live_call_operation( + body: Bytes, + content_type: Option<&str>, + protocol_headers: Vec, +) -> Result> { + let mut headers = protocol_headers; + if let Some(content_type) = content_type { + headers.push(ProviderHttpHeader::new( + "content-type", + Bytes::copy_from_slice(content_type.as_bytes()), + )); + } + let invalid = || { + Box::new(live_error_response( + LiveErrorShape::Live, + StatusCode::BAD_REQUEST, + "Codex live call request is invalid", + "invalid_request", + )) + }; + let payload = RawHttpPayload::new("openai", body).map_err(|_| invalid())?; + ProviderHttpRequest::new( + LIVE_CALLS_ENDPOINT, + ProviderHttpMethod::Post, + Some(LIVE_CALLS_QUERY.to_owned()), + headers, + payload, + ) + .map_err(|_| invalid()) +} + +/// 从 `{"sdp": …, "session"?: …}` 引导体提取模型名；`session.model` 优先。 +pub(crate) fn live_requested_model(body: &[u8]) -> Option { + let payload = serde_json::from_slice::(body).ok()?; + let extract = |value: Option<&Value>| -> Option { + value + .and_then(Value::as_str) + .map(str::trim) + .filter(|model| !model.is_empty()) + .map(ToOwned::to_owned) + }; + extract(payload.pointer("/session/model")).or_else(|| extract(payload.get("model"))) +} + +/// realtime 系模型名归一到 Codex 语音模型；其余模型原样透传。 +pub(crate) fn codex_realtime_model(model: Option) -> String { + const DEFAULT_LIVE_MODEL: &str = "gpt-live-1-codex"; + let Some(model) = model else { + return DEFAULT_LIVE_MODEL.to_owned(); + }; + let trimmed = model.trim(); + let lowered = trimmed.to_ascii_lowercase(); + if lowered.is_empty() + || lowered == "gpt-realtime" + || lowered.starts_with("gpt-realtime-") + || lowered.contains("realtime-preview") + { + return DEFAULT_LIVE_MODEL.to_owned(); + } + trimmed.to_owned() +} + +/// 按官方合同的观测，把归一后的模型写回引导体；无模型字段时不改写。 +pub(crate) fn rewrite_call_request_model(body: Bytes, upstream_model: &str) -> Bytes { + let Ok(mut payload) = serde_json::from_slice::>(&body) else { + return body; + }; + let mut changed = false; + if let Some(session) = payload.get_mut("session").filter(|s| !s.is_null()) + && let Some(session) = session.as_object_mut() + { + session.insert("model".to_owned(), Value::String(upstream_model.to_owned())); + changed = true; + } else if payload.contains_key("model") { + payload.insert("model".to_owned(), Value::String(upstream_model.to_owned())); + changed = true; + } + if !changed { + return body; + } + serde_json::to_vec(&payload) + .map(Bytes::from) + .unwrap_or(body) +} + +/// 把 SDP 文本包装为上游要求的 JSON 引导体。 +pub(crate) fn encode_call_request( + sdp: &str, + session: Option, +) -> Result> { + let invalid = |message: &'static str| { + Box::new(live_error_response( + LiveErrorShape::Live, + StatusCode::BAD_REQUEST, + message, + "invalid_request", + )) + }; + if sdp.trim().is_empty() { + return Err(invalid("Codex live call request requires an SDP offer")); + } + let mut payload = serde_json::Map::new(); + payload.insert("sdp".to_owned(), Value::String(sdp.to_owned())); + if let Some(session) = session.filter(|session| !session.is_null()) { + payload.insert("session".to_owned(), session); + } + serde_json::to_vec(&payload) + .map(Bytes::from) + .map_err(|_| invalid("Codex live call request is invalid")) +} + +/// 响应头白名单；`Location` 是客户端获知 call id 的合同。 +pub(crate) fn live_forwarded_response_headers( + headers: &[ProviderResponseHeader], +) -> Vec<(HeaderName, HeaderValue)> { + const FORWARDABLE: [&str; 5] = [ + "location", + "content-type", + "retry-after", + "x-request-id", + "openai-request-id", + ]; + headers + .iter() + .filter(|header| FORWARDABLE.contains(&header.name())) + .filter_map(|header| { + Some(( + HeaderName::from_bytes(header.name().as_bytes()).ok()?, + HeaderValue::from_bytes(header.value()).ok()?, + )) + }) + .collect() +} + +/// 引导体标准化：JSON 原样、SDP 文本包装、multipart 字段（sdp+session）重组。 +pub(crate) fn normalize_call_request( + body: &Bytes, + content_type: Option<&str>, + multipart: Option<(String, Option)>, +) -> Result<(Bytes, Option, Option), Box> { + if let Some((sdp, session)) = multipart { + let model = session + .as_ref() + .and_then(|session| serde_json::to_vec(session).ok()) + .and_then(|encoded| live_requested_model(&encoded)); + let encoded = encode_call_request(&sdp, session)?; + return Ok((encoded, Some("application/json".to_owned()), model)); + } + let media_type = content_type + .and_then(|value| value.split(';').next()) + .map(str::trim) + .map(str::to_ascii_lowercase); + match media_type.as_deref() { + Some("application/sdp") | Some("text/plain") => { + let sdp = std::str::from_utf8(body).map_err(|_| { + Box::new(live_error_response( + LiveErrorShape::Live, + StatusCode::BAD_REQUEST, + "Codex live call request requires a UTF-8 SDP offer", + "invalid_request", + )) + })?; + let encoded = encode_call_request(sdp, None)?; + Ok((encoded, Some("application/json".to_owned()), None)) + } + _ => { + let model = live_requested_model(body); + let content_type = content_type + .map(ToOwned::to_owned) + .or_else(|| Some("application/json".to_owned())); + Ok((body.clone(), content_type, model)) + } + } +} diff --git a/backend/crates/gateway-api/src/openai/live/websocket.rs b/backend/crates/gateway-api/src/openai/live/websocket.rs new file mode 100644 index 000000000..b22721641 --- /dev/null +++ b/backend/crates/gateway-api/src/openai/live/websocket.rs @@ -0,0 +1,296 @@ +//! Live sideband 的 WebSocket 中继 adapter。 +//! +//! 上游连接由 [`gateway_core::live::LiveGateway`] 用钉住账号的凭据建立； +//! 本模块只做帧形态映射（Text/Binary 双向、Ping 本地应答、Close 投影）， +//! 不解释帧内容。 + +use std::time::Duration; + +use axum::{ + extract::{ + State, WebSocketUpgrade, + ws::{CloseFrame, Message, WebSocket, rejection::WebSocketUpgradeRejection}, + }, + http::{HeaderMap, StatusCode, Uri, header}, + response::{IntoResponse, Response}, +}; +use bytes::Bytes; +use gateway_core::lifecycle::{CancellationToken, ConnectionGuard}; +use gateway_core::live::{LiveClose, LiveFrame, LiveRelay, LiveSidebandRequest, LiveSidebandStyle}; + +use super::LiveErrorShape; +use crate::{ + ApiState, + openai::{ + auth::{authenticate_client, client_access_error_response}, + error::runtime_unavailable_response, + live::{ + filter_protocol_headers, live_error_response, offered_subprotocols, + realtime_unsupported_response, + }, + }, +}; + +const SIDEBAND_CLOSE_TIMEOUT: Duration = Duration::from_secs(5); + +pub(crate) async fn sideband( + State(state): State, + axum::extract::Path(call_id): axum::extract::Path, + uri: Uri, + headers: HeaderMap, + upgrade: Result, +) -> Response { + let style = if uri.path().contains("/realtime/calls/") { + LiveSidebandStyle::RealtimeCalls + } else { + LiveSidebandStyle::Live + }; + open_sideband(state, call_id, style, headers, upgrade).await +} + +/// `GET /v1/realtime`：携带 `call_id` 时加入既有通话；标准 realtime +/// WebSocket（`?model=…`）在本网关的 Codex OAuth 上游上不可用，返回 501。 +pub(crate) async fn realtime_get( + State(state): State, + uri: Uri, + headers: HeaderMap, + upgrade: Result, +) -> Response { + let call_id = uri + .query() + .and_then(|query| { + query + .split('&') + .find_map(|pair| pair.strip_prefix("call_id=")) + }) + .map(percent_decode); + match call_id.filter(|call_id| !call_id.is_empty()) { + Some(call_id) => { + open_sideband( + state, + call_id, + LiveSidebandStyle::RealtimeQuery, + headers, + upgrade, + ) + .await + } + None => realtime_unsupported_response("Direct Realtime WebSocket"), + } +} + +fn percent_decode(value: &str) -> String { + let mut output = Vec::with_capacity(value.len()); + let bytes = value.as_bytes(); + let mut index = 0; + while index < bytes.len() { + if bytes[index] == b'%' + && let Some(byte) = bytes + .get(index + 1..index + 3) + .and_then(|hex| u8::from_str_radix(std::str::from_utf8(hex).ok()?, 16).ok()) + { + output.push(byte); + index += 3; + } else { + output.push(bytes[index]); + index += 1; + } + } + String::from_utf8_lossy(&output).into_owned() +} + +async fn open_sideband( + state: ApiState, + call_id: String, + style: LiveSidebandStyle, + headers: HeaderMap, + upgrade: Result, +) -> Response { + let Ok(upgrade) = upgrade else { + return upgrade_required_response(); + }; + let service = state.openai(); + let client = match authenticate_client(service, &headers).await { + Ok(client) => client, + Err(error) => return client_access_error_response(error), + }; + let Some(gateway) = service.live_gateway() else { + return realtime_unsupported_response("Codex live sideband"); + }; + // sideband 是长连接：在拨号前注册连接 guard，drain 才能统计并等待它关闭。 + let connection_guard = match service.try_register_connection() { + Ok(guard) => guard, + Err(_) => return runtime_unavailable_response().into_response(), + }; + let cancellation = service.lifecycle().cancellation(); + let opening = gateway.open_sideband(LiveSidebandRequest { + call_id: &call_id, + client_api_key_id: client.policy().key_id(), + account_scope: client.policy().account_scope(), + style, + protocol_headers: filter_protocol_headers(&headers) + .into_iter() + .map(|header| { + ( + header.name().to_ascii_lowercase(), + String::from_utf8_lossy(header.value()).into_owned(), + ) + }) + .collect(), + subprotocols: offered_subprotocols(&headers), + }); + let result = tokio::select! { + biased; + _ = cancellation.cancelled() => return runtime_unavailable_response().into_response(), + result = opening => result, + }; + let relay = match result { + Ok(relay) => relay, + Err(error) => { + return super::http::live_gateway_error_response(LiveErrorShape::Realtime, &error); + } + }; + serve_sideband(upgrade, relay, connection_guard, cancellation) +} + +fn upgrade_required_response() -> Response { + let mut response = live_error_response( + LiveErrorShape::Realtime, + StatusCode::UPGRADE_REQUIRED, + "WebSocket upgrade required", + "websocket_upgrade_required", + ); + response.headers_mut().insert( + header::UPGRADE, + header::HeaderValue::from_static("websocket"), + ); + response +} + +fn serve_sideband( + mut upgrade: WebSocketUpgrade, + relay: LiveRelay, + connection_guard: Box, + cancellation: CancellationToken, +) -> Response { + // 上游已按客户端 offer 协商子协议；只有客户端确实 offer 过才会命中。 + if let Some(subprotocol) = relay.subprotocol.clone() { + upgrade = upgrade.protocols([subprotocol]); + } + upgrade + .max_message_size(usize::MAX) + .max_frame_size(usize::MAX) + .on_upgrade(move |socket| async move { + relay_sideband(socket, relay, connection_guard, cancellation).await; + }) +} + +/// 双向帧中继；任一侧结束即投影关闭。 +/// +/// 传输中断只结束中继：call 绑定的释放由中继携带的 guard 完成，绑定本身 +/// 保留到会话 TTL，官方 FramelessBidi 客户端会重连同一 call。 +async fn relay_sideband( + mut client: WebSocket, + mut relay: LiveRelay, + connection_guard: Box, + cancellation: CancellationToken, +) { + // guard 覆盖整个中继生命周期，drain 等待其释放。 + let _connection_guard = connection_guard; + tokio::select! { + biased; + _ = cancellation.cancelled() => {}, + () = relay_frames(&mut client, &mut relay) => {}, + } + // 写入或关闭被取消时直接丢弃传输端，避免清理再次阻塞 guard 释放 + drop(relay); + drop(_connection_guard); +} + +async fn relay_frames(client: &mut WebSocket, relay: &mut LiveRelay) { + let mut close_forwarded = false; + loop { + tokio::select! { + client_message = client.recv() => match client_message { + Some(Ok(message)) => { + match message { + Message::Text(text) => { + let frame = LiveFrame::Text(Bytes::copy_from_slice(text.as_bytes())); + if relay.send_frame(frame).await.is_err() { + break; + } + } + Message::Binary(binary) => { + if relay.send_frame(LiveFrame::Binary(binary)).await.is_err() { + break; + } + } + // Ping 由本端应答（上游 Ping 同理），不进入对端。 + Message::Ping(payload) => { + if client + .send(Message::Pong(payload)) + .await + .is_err() + { + break; + } + } + Message::Pong(_) => {} + Message::Close(close) => { + let close = close.map(|frame| LiveClose { + code: frame.code, + reason: frame.reason.to_string(), + }); + let _ = tokio::time::timeout(SIDEBAND_CLOSE_TIMEOUT, relay.close(close)).await; + return; + } + } + } + Some(Err(_)) | None => break, + }, + upstream_frame = relay.next_frame() => match upstream_frame { + Some(LiveFrame::Text(payload)) => { + if client + .send(Message::Text(String::from_utf8_lossy(&payload).into_owned().into())) + .await + .is_err() + { + break; + } + } + Some(LiveFrame::Binary(payload)) => { + if client.send(Message::Binary(payload)).await.is_err() { + break; + } + } + Some(LiveFrame::Ping(payload)) => { + if relay.send_frame(LiveFrame::Pong(payload)).await.is_err() { + break; + } + } + Some(LiveFrame::Pong(_)) => {} + Some(LiveFrame::Close(close)) => { + let _ = tokio::time::timeout(SIDEBAND_CLOSE_TIMEOUT, client.send(Message::Close(close.map(|close| CloseFrame { + code: close.code, + reason: close.reason.into(), + })))).await; + close_forwarded = true; + break; + } + None => break, + }, + } + } + if !close_forwarded { + // 异常中断按观测到的方向投影为正常关闭；原因不透出给客户端。 + let _ = tokio::time::timeout( + SIDEBAND_CLOSE_TIMEOUT, + client.send(Message::Close(Some(CloseFrame { + code: 1000, + reason: "".into(), + }))), + ) + .await; + } + let _ = tokio::time::timeout(SIDEBAND_CLOSE_TIMEOUT, relay.close(None)).await; +} diff --git a/backend/crates/gateway-api/src/openai/mod.rs b/backend/crates/gateway-api/src/openai/mod.rs index e55afc6d8..d2245a92a 100644 --- a/backend/crates/gateway-api/src/openai/mod.rs +++ b/backend/crates/gateway-api/src/openai/mod.rs @@ -4,6 +4,7 @@ pub mod auth; mod endpoint; pub mod error; pub mod images; +pub(crate) mod live; pub(crate) mod middleware; pub mod models; pub mod responses; diff --git a/backend/crates/gateway-api/src/openai/router.rs b/backend/crates/gateway-api/src/openai/router.rs index ea096e595..d753851ff 100644 --- a/backend/crates/gateway-api/src/openai/router.rs +++ b/backend/crates/gateway-api/src/openai/router.rs @@ -8,6 +8,7 @@ use axum::{ use super::{ images::{image_edits, image_generations}, + live, models::{model_detail, models}, responses::{responses, responses_websocket}, search::standalone_search, @@ -36,6 +37,7 @@ pub(crate) fn router() -> Router { // 官方 OpenAI 模型详情合同使用 path ID；它不属于 Admin API 约束 .route(MODEL_DETAIL_PATH, get(model_detail)) .merge(usage::router()) + .merge(live::router()) // OpenAI 数据面正文属于客户端/上游协议；代理不能用私有大小上限提前拒绝 // 上游本可接受的未来 payload .layer(DefaultBodyLimit::disable()) diff --git a/backend/crates/gateway-api/src/openai/service.rs b/backend/crates/gateway-api/src/openai/service.rs index a6b1b53fb..f3d79d2e7 100644 --- a/backend/crates/gateway-api/src/openai/service.rs +++ b/backend/crates/gateway-api/src/openai/service.rs @@ -13,7 +13,7 @@ use gateway_core::engine::execution::{ use gateway_core::error::{GatewayError, GatewayErrorKind}; use gateway_core::lifecycle::{ConnectionDraining, ConnectionGuard, ConnectionLifecycle}; use gateway_core::routing::{ - ProviderCatalogUnavailable, ProviderKind, PublicModelDescriptor, PublicModelId, + ProviderCatalogUnavailable, ProviderKind, PublicModelDescriptor, PublicModelId, UpstreamModelId, }; use uuid::Uuid; @@ -134,6 +134,7 @@ impl OpenAiService { &self, prepared: PreparedRootExecution, operation: gateway_core::operation::Operation, + upstream_model: Option, client_ip: Option, user_agent: Option, endpoint: String, @@ -148,7 +149,7 @@ impl OpenAiService { .start_prepared_provider_endpoint( prepared, provider, - None, + upstream_model, operation, ExecutionRequestMetadata { protocol: "openai".to_owned(), @@ -170,6 +171,12 @@ impl OpenAiService { self.execution.start_provider_endpoint(request).await } + /// Live 语音 sideband 能力；组合未提供时协议层回退到稳定 501。 + #[must_use] + pub(crate) fn live_gateway(&self) -> Option> { + self.execution.live_gateway() + } + pub(crate) fn try_register_connection( &self, ) -> Result, ConnectionDraining> { diff --git a/backend/crates/gateway-api/tests/architecture.rs b/backend/crates/gateway-api/tests/architecture.rs index 2ecbdc903..5a08380cf 100644 --- a/backend/crates/gateway-api/tests/architecture.rs +++ b/backend/crates/gateway-api/tests/architecture.rs @@ -79,6 +79,9 @@ fn source_tree_should_match_frozen_machine_manifest() { "src/openai/error.rs", "src/openai/images/http.rs", "src/openai/images/mod.rs", + "src/openai/live/http.rs", + "src/openai/live/mod.rs", + "src/openai/live/websocket.rs", "src/openai/mod.rs", "src/openai/models.rs", "src/openai/endpoint.rs", @@ -156,6 +159,8 @@ fn test_tree_should_match_frozen_rust_mirror() { "tests/openai/responses/websocket/interrupt.rs", "tests/openai/responses/websocket/mod.rs", "tests/openai/responses/websocket/protocol.rs", + "tests/openai/live/mod.rs", + "tests/openai/live/websocket.rs", "tests/openai/router.rs", "tests/openai/search/mod.rs", "tests/openai/usage.rs", diff --git a/backend/crates/gateway-api/tests/openai/live/mod.rs b/backend/crates/gateway-api/tests/openai/live/mod.rs new file mode 100644 index 000000000..39bb1b70d --- /dev/null +++ b/backend/crates/gateway-api/tests/openai/live/mod.rs @@ -0,0 +1,324 @@ +//! Live HTTP 入参与鉴权验证 + +mod websocket; + +use std::sync::{Arc, Mutex}; + +use axum::{ + body::Body, + http::{Request, StatusCode, header::AUTHORIZATION}, +}; +use futures::future::BoxFuture; +use gateway_core::engine::execution::{ + AuthenticatedClient, ClientAuthenticationError, ExecutionRequestMetadata, ExecutionService, + StartExecution, StartProviderExecution, StartedExecution, +}; +use gateway_core::error::{GatewayError, GatewayErrorKind}; +use gateway_core::operation::Operation; +use serde_json::{Value, json}; +use tower::ServiceExt; + +use super::api_router; +use super::authenticated_client; +use super::models::ModelsExecution; + +const LIVE_KEY: &str = "sk_live_test"; + +/// 捕获 start_provider_endpoint 请求的执行器；语音引导路径的行为断言入口。 +#[derive(Clone, Default)] +struct LiveCapture { + captured: Arc>>, + gateway: Option>, +} + +impl LiveCapture { + fn new() -> Arc { + Arc::new(Self::default()) + } + + fn operation(&self) -> Option<(Operation, ExecutionRequestMetadata)> { + let captured = self.captured.lock().expect("capture").take()?; + Some((captured.operation, captured.metadata)) + } +} + +impl ExecutionService for LiveCapture { + fn live_gateway(&self) -> Option> { + self.gateway.clone() + } + + fn authenticate( + &self, + plaintext: &str, + ) -> Result { + if plaintext == LIVE_KEY { + Ok(authenticated_client(LIVE_KEY)) + } else { + Err(ClientAuthenticationError::InvalidKey) + } + } + + fn public_models(&self, _: &AuthenticatedClient) -> Vec { + Vec::new() + } + + fn contains_public_model( + &self, + _: &AuthenticatedClient, + _: &gateway_core::routing::PublicModelId, + ) -> bool { + false + } + + fn start(&self, _: StartExecution) -> BoxFuture<'_, Result> { + Box::pin(async { + Err(GatewayError::new( + GatewayErrorKind::Internal, + "live test must not start a response", + )) + }) + } + + fn start_provider_endpoint( + &self, + request: StartProviderExecution, + ) -> BoxFuture<'_, Result> { + *self.captured.lock().expect("capture") = Some(request); + Box::pin(async { + Err(GatewayError::new( + GatewayErrorKind::Internal, + "live test captured the provider endpoint request", + )) + }) + } +} + +async fn capture_live_call( + body: Body, + content_type: &str, +) -> (StatusCode, Option<(Operation, ExecutionRequestMetadata)>) { + let capture = LiveCapture::new(); + let response = api_router(Arc::clone(&capture) as Arc) + .await + .oneshot( + Request::post("/v1/live") + .header(AUTHORIZATION, format!("Bearer {LIVE_KEY}")) + .header("content-type", content_type) + .body(body) + .expect("build live request"), + ) + .await + .expect("route live request"); + (response.status(), capture.operation()) +} + +#[tokio::test] +async fn live_call_requires_a_client_key() { + let response = api_router(ModelsExecution::new()) + .await + .oneshot( + Request::post("/v1/live") + .header("content-type", "application/json") + .body(Body::from(json!({"sdp": "v=0"}).to_string())) + .expect("build live request"), + ) + .await + .expect("route live request"); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); +} + +#[tokio::test] +async fn json_call_is_forwarded_to_the_realtime_calls_endpoint_with_rewritten_model() { + let (status, captured) = capture_live_call( + Body::from( + json!({ + "sdp": "v=0\r\no=- 1 1 IN IP4 0.0.0.0", + "session": {"model": "gpt-realtime", "voice": "alloy"} + }) + .to_string(), + ), + "application/json", + ) + .await; + assert!(status.is_server_error() || status.is_client_error()); + let (operation, metadata) = captured.expect("captured provider endpoint request"); + let Operation::ProviderHttp(request) = operation else { + panic!("expected provider http operation"); + }; + assert_eq!(request.endpoint(), "realtime-calls"); + assert_eq!( + request.query(), + Some("intent=quicksilver&architecture=avas") + ); + let body: Value = serde_json::from_slice(request.payload().body()).expect("json body"); + assert!(body["sdp"].as_str().expect("sdp").starts_with("v=0")); + assert_eq!(body["session"]["model"], "gpt-live-1-codex"); + assert_eq!(body["session"]["voice"], "alloy"); + let header_names = request + .headers() + .iter() + .map(|header| header.name().to_owned()) + .collect::>(); + assert!(header_names.contains(&"content-type".to_owned())); + assert!(!header_names.contains(&"authorization".to_owned())); + assert!(!header_names.contains(&"x-api-key".to_owned())); + assert_eq!(metadata.endpoint, "/v1/live"); + assert!(!metadata.stream); +} + +#[tokio::test] +async fn sdp_text_is_wrapped_into_the_json_call_request() { + let (status, captured) = capture_live_call( + Body::from("v=0\r\no=- 1 1 IN IP4 0.0.0.0"), + "application/sdp", + ) + .await; + assert!(status.is_server_error() || status.is_client_error()); + let (operation, _) = captured.expect("captured provider endpoint request"); + let Operation::ProviderHttp(request) = operation else { + panic!("expected provider http operation"); + }; + let body: Value = serde_json::from_slice(request.payload().body()).expect("json body"); + assert!(body["sdp"].as_str().is_some()); + assert!(body.get("session").is_none()); +} + +#[tokio::test] +async fn multipart_call_recombines_sdp_and_session_fields() { + let boundary = "live_test_boundary"; + let multipart = format!( + "--{boundary}\r\ncontent-disposition: form-data; name=\"sdp\"\r\n\r\nv=0\r\no=- 1 1 IN IP4 0.0.0.0\r\n--{boundary}\r\ncontent-disposition: form-data; name=\"session\"\r\ncontent-type: application/json\r\n\r\n{{\"model\":\"gpt-realtime\"}}\r\n--{boundary}--\r\n" + ); + let (status, captured) = capture_live_call( + Body::from(multipart), + &format!("multipart/form-data; boundary={boundary}"), + ) + .await; + assert!(status.is_server_error() || status.is_client_error()); + let (operation, _) = captured.expect("captured provider endpoint request"); + let Operation::ProviderHttp(request) = operation else { + panic!("expected provider http operation"); + }; + let body: Value = serde_json::from_slice(request.payload().body()).expect("json body"); + assert!(body["sdp"].as_str().is_some()); + // multipart 的 session.model 同样走提取与改写链路。 + assert_eq!(body["session"]["model"], "gpt-live-1-codex"); +} + +#[tokio::test] +async fn client_identity_headers_never_reach_the_provider_operation() { + let (status, captured) = capture_live_call( + Body::from(json!({"sdp": "v=0"}).to_string()), + "application/json", + ) + .await; + assert!(status.is_server_error() || status.is_client_error()); + let (operation, _) = captured.expect("captured provider endpoint request"); + let Operation::ProviderHttp(request) = operation else { + panic!("expected provider http operation"); + }; + let header_names = request + .headers() + .iter() + .map(|header| header.name().to_ascii_lowercase()) + .collect::>(); + assert!(!header_names.contains(&"authorization".to_owned())); + assert!(!header_names.contains(&"cookie".to_owned())); + assert!(!header_names.contains(&"host".to_owned())); +} + +#[tokio::test] +async fn sideband_requires_a_websocket_upgrade() { + let response = api_router(ModelsExecution::new()) + .await + .oneshot( + Request::get("/v1/live/call_1") + .header(AUTHORIZATION, "Bearer sk_models_test") + .body(Body::empty()) + .expect("build sideband request"), + ) + .await + .expect("route sideband request"); + assert_eq!(response.status(), StatusCode::UPGRADE_REQUIRED); +} + +#[tokio::test] +async fn realtime_capability_stubs_return_501() { + for (method, path) in [ + ("POST", "/v1/realtime/sessions"), + ("POST", "/v1/realtime/client_secrets"), + ("POST", "/v1/realtime/transcription_sessions"), + ("POST", "/v1/realtime/translations"), + ("POST", "/v1/realtime/calls/call_1/accept"), + ] { + let response = api_router(ModelsExecution::new()) + .await + .oneshot( + Request::builder() + .method(method) + .uri(path) + .header(AUTHORIZATION, "Bearer sk_models_test") + .body(Body::empty()) + .expect("build realtime stub request"), + ) + .await + .expect("route realtime stub request"); + assert_eq!(response.status(), StatusCode::NOT_IMPLEMENTED, "{path}"); + let body: Value = serde_json::from_slice( + &axum::body::to_bytes(response.into_body(), 4096) + .await + .unwrap(), + ) + .unwrap(); + assert_eq!( + body["error"]["code"], "realtime_capability_not_supported", + "{path}" + ); + } +} + +#[tokio::test] +async fn direct_realtime_websocket_is_reported_unsupported() { + let response = api_router(ModelsExecution::new()) + .await + .oneshot( + Request::get("/v1/realtime?model=gpt-realtime") + .header(AUTHORIZATION, "Bearer sk_models_test") + .body(Body::empty()) + .expect("build realtime request"), + ) + .await + .expect("route realtime request"); + assert_eq!(response.status(), StatusCode::NOT_IMPLEMENTED); +} + +#[tokio::test] +async fn hangup_without_a_live_gateway_is_reported_unsupported() { + let response = api_router(ModelsExecution::new()) + .await + .oneshot( + Request::post("/v1/realtime/calls/call_unknown/hangup") + .header(AUTHORIZATION, "Bearer sk_models_test") + .header("content-type", "application/json") + .body(Body::from("{}")) + .expect("build hangup request"), + ) + .await + .expect("route hangup request"); + assert_eq!(response.status(), StatusCode::NOT_IMPLEMENTED); +} + +#[tokio::test] +async fn live_post_only_accepts_declared_methods() { + let response = api_router(ModelsExecution::new()) + .await + .oneshot( + Request::put("/v1/live") + .header(AUTHORIZATION, "Bearer sk_models_test") + .body(Body::empty()) + .expect("build live put request"), + ) + .await + .expect("route live put request"); + assert_eq!(response.status(), StatusCode::METHOD_NOT_ALLOWED); +} diff --git a/backend/crates/gateway-api/tests/openai/live/websocket.rs b/backend/crates/gateway-api/tests/openai/live/websocket.rs new file mode 100644 index 000000000..6e0f1ed65 --- /dev/null +++ b/backend/crates/gateway-api/tests/openai/live/websocket.rs @@ -0,0 +1,208 @@ +//! 通过真实 WebSocket 验证 sideband 阻塞写入、关闭与生命周期释放 + +use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, +}; +use std::time::Duration; + +use futures::{SinkExt, future::BoxFuture}; +use gateway_core::lifecycle::{ + CancellationToken, ConnectionDraining, ConnectionGuard, ConnectionLifecycle, +}; +use gateway_core::live::{ + LiveCallOutcome, LiveClose, LiveFrame, LiveGateway, LiveGatewayError, LiveHangupRequest, + LiveRelay, LiveRelayGuard, LiveRelaySink, LiveRelayStream, LiveSidebandRequest, +}; +use tokio::sync::Notify; +use tokio_tungstenite::{ + connect_async, + tungstenite::{Message, client::IntoClientRequest}, +}; + +use super::{LIVE_KEY, LiveCapture}; + +#[derive(Default)] +struct Trace { + cancellation: CancellationToken, + connections: AtomicUsize, + claims: AtomicUsize, + entered: Notify, + released: Notify, + block_close: bool, +} + +struct Guard(Arc); +impl ConnectionGuard for Guard {} +impl Drop for Guard { + fn drop(&mut self) { + self.0.connections.fetch_sub(1, Ordering::SeqCst); + self.0.released.notify_one(); + } +} +struct Claim(Arc); +impl LiveRelayGuard for Claim {} +impl Drop for Claim { + fn drop(&mut self) { + self.0.claims.fetch_sub(1, Ordering::SeqCst); + } +} + +struct Lifecycle(Arc); +impl ConnectionLifecycle for Lifecycle { + fn try_register(&self) -> Result, ConnectionDraining> { + self.0.connections.fetch_add(1, Ordering::SeqCst); + Ok(Box::new(Guard(self.0.clone()))) + } + fn cancellation(&self) -> CancellationToken { + self.0.cancellation.clone() + } + fn is_draining(&self) -> bool { + self.0.cancellation.is_cancelled() + } +} + +struct PendingStream; +impl LiveRelayStream for PendingStream { + fn next_frame(&mut self) -> BoxFuture<'_, Option> { + Box::pin(std::future::pending()) + } +} +struct PendingSink(Arc); +impl LiveRelaySink for PendingSink { + fn send_frame(&mut self, _: LiveFrame) -> BoxFuture<'_, Result<(), LiveGatewayError>> { + Box::pin(async move { + self.0.entered.notify_one(); + std::future::pending().await + }) + } + fn close(&mut self, _: Option) -> BoxFuture<'_, ()> { + Box::pin(async move { + if self.0.block_close { + self.0.entered.notify_one(); + std::future::pending().await + } + }) + } +} +struct Gateway(Arc); +impl LiveGateway for Gateway { + fn open_sideband<'a>( + &'a self, + request: LiveSidebandRequest<'a>, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { + assert!( + request.account_scope.allows( + &gateway_core::account::ProviderAccountId::new("acct_api_test").unwrap() + ) + ); + self.0.claims.fetch_add(1, Ordering::SeqCst); + let mut relay = LiveRelay::new( + None, + Box::new(PendingStream), + Box::new(PendingSink(self.0.clone())), + ); + relay.with_guard(Box::new(Claim(self.0.clone()))); + Ok(relay) + }) + } + fn hangup<'a>( + &'a self, + _: LiveHangupRequest<'a>, + ) -> BoxFuture<'a, Result> { + unreachable!("no hangup in relay lifecycle test") + } +} + +async fn blocked_relay( + trace: &Arc, +) -> ( + tokio_tungstenite::WebSocketStream>, + tokio::task::JoinHandle<()>, +) { + let admin = crate::admin::AdminTestFixture::new().await; + let execution = Arc::new(LiveCapture { + gateway: Some(Arc::new(Gateway(trace.clone()))), + ..LiveCapture::default() + }); + let router = gateway_api::initialize( + gateway_api::ApiConfig { + asset_directory: std::env::temp_dir(), + cors_allowed_origins: Vec::new(), + request_timeout_seconds: None, + request_id_header: "x-request-id".to_owned(), + }, + execution, + admin.services, + Vec::new(), + Arc::new(crate::openai::EmptyWorkerHealth), + Arc::new(Lifecycle(trace.clone())), + ) + .unwrap() + .router(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + axum::serve(listener, router).await.unwrap(); + }); + let mut request = format!("ws://{address}/v1/live/call_regression") + .into_client_request() + .unwrap(); + request.headers_mut().insert( + "authorization", + format!("Bearer {LIVE_KEY}").parse().unwrap(), + ); + let (mut socket, _) = connect_async(request).await.unwrap(); + if trace.block_close { + socket.send(Message::Close(None)).await.unwrap(); + } else { + socket.send(Message::Text("test".into())).await.unwrap(); + } + tokio::time::timeout(Duration::from_secs(2), trace.entered.notified()) + .await + .expect("relay entered blocked sink"); + (socket, server) +} + +async fn assert_released(trace: &Trace) { + tokio::time::timeout(Duration::from_secs(2), trace.released.notified()) + .await + .expect("connection released promptly"); + assert_eq!(trace.connections.load(Ordering::SeqCst), 0); + assert_eq!(trace.claims.load(Ordering::SeqCst), 0); +} + +#[tokio::test] +async fn live_shutdown_cancels_blocked_send_and_releases_both_guards() { + let trace = Arc::new(Trace::default()); + let (_socket, server) = blocked_relay(&trace).await; + trace.cancellation.cancel(); + assert_released(&trace).await; + server.abort(); +} + +#[tokio::test] +async fn live_shutdown_cancels_blocked_close_and_releases_both_guards() { + let trace = Arc::new(Trace { + block_close: true, + ..Trace::default() + }); + let (_socket, server) = blocked_relay(&trace).await; + trace.cancellation.cancel(); + assert_released(&trace).await; + server.abort(); +} + +#[tokio::test] +async fn live_close_deadline_releases_both_guards_without_shutdown() { + let trace = Arc::new(Trace { + block_close: true, + ..Trace::default() + }); + let (_socket, server) = blocked_relay(&trace).await; + tokio::time::pause(); + tokio::time::advance(Duration::from_secs(6)).await; + assert_released(&trace).await; + server.abort(); +} diff --git a/backend/crates/gateway-api/tests/openai/mod.rs b/backend/crates/gateway-api/tests/openai/mod.rs index ee559364c..4628e8bee 100644 --- a/backend/crates/gateway-api/tests/openai/mod.rs +++ b/backend/crates/gateway-api/tests/openai/mod.rs @@ -4,6 +4,7 @@ mod auth; mod endpoint; mod error; mod images; +mod live; mod middleware; mod models; mod responses; diff --git a/backend/crates/gateway-core/src/engine/execution.rs b/backend/crates/gateway-core/src/engine/execution.rs index f9f515b90..aa6e1e828 100644 --- a/backend/crates/gateway-core/src/engine/execution.rs +++ b/backend/crates/gateway-core/src/engine/execution.rs @@ -710,6 +710,12 @@ pub trait ExecutionService: Send + Sync { &self, request: StartProviderExecution, ) -> BoxFuture<'_, Result>; + + /// 返回 Provider 注册表中首个声明 Live sideband 能力的网关。 + /// `None` 表示当前组合不含语音 sideband；协议 adapter 据此回退到稳定 501。 + fn live_gateway(&self) -> Option> { + None + } } /// 只验证 Client API Key 并返回稳定 Key ID，不产生请求使用事实 @@ -2372,6 +2378,12 @@ impl ExecutionService for DefaultExecutionService { ) -> BoxFuture<'_, Result> { Box::pin(async move { self.start_provider_endpoint_inner(request).await }) } + + fn live_gateway(&self) -> Option> { + self.providers + .iter() + .find_map(|provider| provider.live_gateway()) + } } impl NestedModelExecutionPort for DefaultExecutionService { diff --git a/backend/crates/gateway-core/src/engine/provider.rs b/backend/crates/gateway-core/src/engine/provider.rs index e4602b869..a03cd28a8 100644 --- a/backend/crates/gateway-core/src/engine/provider.rs +++ b/backend/crates/gateway-core/src/engine/provider.rs @@ -1018,6 +1018,12 @@ pub trait Provider: Send + Sync { request: ProviderRequest, context: AttemptContext, ) -> Result; + + /// 返回 Provider 的 Live sideband 能力；不支持的 Provider 返回 `None`。 + /// 实现必须只暴露已钉住账号的受限拨号，不得泄露凭据或放宽账号范围。 + fn live_gateway(&self) -> Option> { + None + } } /// Provider Registry 构建错误 @@ -1135,6 +1141,11 @@ impl ProviderRegistry { pub fn is_empty(&self) -> bool { self.providers.is_empty() } + + /// 按注册顺序遍历 adapter；仅用于能力发现类查询。 + pub fn iter(&self) -> impl Iterator> { + self.providers.values() + } } impl ProviderCatalogPort for ProviderRegistry { diff --git a/backend/crates/gateway-core/src/lib.rs b/backend/crates/gateway-core/src/lib.rs index 9afb560c8..b125510f7 100644 --- a/backend/crates/gateway-core/src/lib.rs +++ b/backend/crates/gateway-core/src/lib.rs @@ -13,6 +13,7 @@ pub mod event; pub mod health; pub mod identity; pub mod lifecycle; +pub mod live; pub mod metering; pub mod middleware; pub mod operation; diff --git a/backend/crates/gateway-core/src/live.rs b/backend/crates/gateway-core/src/live.rs new file mode 100644 index 000000000..6add7a1b3 --- /dev/null +++ b/backend/crates/gateway-core/src/live.rs @@ -0,0 +1,327 @@ +//! Codex Live / Realtime 语音会话的 Core 合同。 +//! +//! 语音会话由两类请求组成：一次性 SDP 引导（走常规引擎 attempt 路径）和 +//! 通话建立后跟随 call id 的长连接 sideband 中继。sideband 不能进入 +//! 请求/响应引擎，Provider 通过本模块的 [`LiveGateway`] 暴露受限的 +//! "钉住账号拨号 + 帧中继" 能力，协议 adapter 只消费抽象帧。 + +use std::fmt; + +use bytes::Bytes; +use futures::future::BoxFuture; + +use crate::account::scope::FrozenAccountScope; +use crate::policy::ClientApiKeyId; + +/// sideband 中继在客户端与上游 WebSocket 之间搬运的帧。 +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum LiveFrame { + Text(Bytes), + Binary(Bytes), + Ping(Bytes), + Pong(Bytes), + Close(Option), +} + +/// WebSocket 关闭帧的状态码与原因。 +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LiveClose { + pub code: u16, + pub reason: String, +} + +/// sideband 的上游目标形态；决定钉住账号拨号使用的路径。 +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LiveSidebandStyle { + /// `GET /v1/live/{call_id}` → 上游 `wss://…/v1/live/{call_id}`。 + Live, + /// `GET /v1/realtime/calls/{call_id}` → 上游 `wss://…/v1/realtime/calls/{call_id}`。 + RealtimeCalls, + /// `GET /v1/realtime?call_id=…` → 上游 `wss://…/v1/realtime?intent=quicksilver&call_id=…`。 + RealtimeQuery, +} + +/// sideband 拨号请求；协议头已由协议 adapter 按允许清单过滤。 +#[derive(Debug)] +pub struct LiveSidebandRequest<'a> { + pub call_id: &'a str, + pub client_api_key_id: &'a ClientApiKeyId, + /// 本次认证冻结的账号范围与模型政策，重连也必须重新校验 + pub account_scope: &'a FrozenAccountScope, + pub style: LiveSidebandStyle, + pub protocol_headers: Vec<(String, String)>, + /// 客户端 offer 的 `Sec-WebSocket-Protocol`，原样透传给上游。 + pub subprotocols: Vec, +} + +/// 一次性账号级上游调用（hangup）请求。 +#[derive(Debug)] +pub struct LiveHangupRequest<'a> { + pub call_id: &'a str, + pub client_api_key_id: &'a ClientApiKeyId, + pub content_type: Option, + pub body: Bytes, + pub protocol_headers: Vec<(String, String)>, +} + +/// 一次性账号级上游调用的透传结果。 +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LiveCallOutcome { + pub status: u16, + pub headers: Vec<(String, String)>, + pub body: Bytes, +} + +/// sideband 中继的上游半边；帧按原始顺序搬运，不做协议解释。 +/// +/// 中继可以携带一个会话 guard（如 call 的账号认领）；`LiveRelay` 被 +/// 丢弃时 guard 一并释放，覆盖传输中断、升级失败与任务取消等全部退出路径。 +pub struct LiveRelay { + /// 上游已协商的 `Sec-WebSocket-Protocol`；协议 adapter 据此回写下游。 + pub subprotocol: Option, + receiver: Box, + sender: Box, + guard: Option>, +} + +/// 随 [`LiveRelay`] 存活的会话占用标记；丢弃时由实现方归还资源。 +pub trait LiveRelayGuard: Send {} + +impl LiveRelay { + /// 由 Provider 实现装配；Core 不感知底层 WebSocket 栈。 + #[must_use] + pub fn new( + subprotocol: Option, + receiver: Box, + sender: Box, + ) -> Self { + Self { + subprotocol, + receiver, + sender, + guard: None, + } + } + + /// 附加随中继释放的会话 guard；重复调用以最后一次为准。 + pub fn with_guard(&mut self, guard: Box) { + self.guard = Some(guard); + } + + /// 读取上游下一帧；返回 `None` 表示上游已关闭。 + pub fn next_frame(&mut self) -> BoxFuture<'_, Option> { + self.receiver.next_frame() + } + + /// 向上游写入一帧。 + pub fn send_frame(&mut self, frame: LiveFrame) -> BoxFuture<'_, Result<(), LiveGatewayError>> { + self.sender.send_frame(frame) + } + + /// 主动关闭上游半边。 + pub fn close(&mut self, close: Option) -> BoxFuture<'_, ()> { + self.sender.close(close) + } +} + +impl fmt::Debug for LiveRelay { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LiveRelay") + .field("subprotocol", &self.subprotocol) + .finish() + } +} + +/// 上游侧的帧接收端。 +pub trait LiveRelayStream: Send { + fn next_frame(&mut self) -> BoxFuture<'_, Option>; +} + +/// 上游侧的帧发送端。 +pub trait LiveRelaySink: Send { + fn send_frame(&mut self, frame: LiveFrame) -> BoxFuture<'_, Result<(), LiveGatewayError>>; + fn close(&mut self, close: Option) -> BoxFuture<'_, ()>; +} + +/// Live 会话在 Provider 内的故障分类；协议 adapter 据此投影 HTTP 状态。 +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LiveGatewayErrorKind { + /// call id 不存在或已过期。 + CallNotFound, + /// 该 call 已有 sideband 加入。 + CallBusy, + /// 调用方不是创建该 call 的 API Key，或固定账号不在当前授权范围 + OwnerMismatch, + /// call id 不满足上游标识约束。 + InvalidCallId, + /// 钉住的账号凭据不可用。 + CredentialUnavailable, + /// 上游握手或调用失败。 + UpstreamUnavailable, + /// 该 Provider 不支持请求的 live 能力。 + Unsupported, +} + +impl LiveGatewayErrorKind { + /// 错误分类的默认 HTTP 状态；上游显式状态优先于该映射。 + #[must_use] + pub const fn default_status(self) -> u16 { + match self { + Self::CallNotFound => 404, + Self::CallBusy => 409, + Self::OwnerMismatch | Self::InvalidCallId => 403, + Self::CredentialUnavailable | Self::UpstreamUnavailable => 503, + Self::Unsupported => 501, + } + } + + /// 面向 OpenAI realtime 客户端的稳定错误码。 + #[must_use] + pub const fn code(self) -> &'static str { + match self { + Self::CallNotFound => "realtime_call_not_found", + Self::CallBusy => "realtime_call_busy", + Self::OwnerMismatch => "realtime_call_scope_mismatch", + Self::InvalidCallId => "invalid_call_id", + Self::CredentialUnavailable => "codex_auth_unavailable", + Self::UpstreamUnavailable => "realtime_upstream_unavailable", + Self::Unsupported => "realtime_capability_not_supported", + } + } +} + +/// Live 会话操作失败；`status`/`body` 携带上游显式响应时优先透传。 +#[derive(Debug, Clone)] +pub struct LiveGatewayError { + kind: LiveGatewayErrorKind, + message: String, + status: Option, + body: Option, +} + +impl LiveGatewayError { + #[must_use] + pub fn new(kind: LiveGatewayErrorKind, message: impl Into) -> Self { + Self { + kind, + message: message.into(), + status: None, + body: None, + } + } + + #[must_use] + pub fn with_upstream(mut self, status: u16, body: Option) -> Self { + self.status = Some(status); + self.body = body; + self + } + + #[must_use] + pub const fn kind(&self) -> LiveGatewayErrorKind { + self.kind + } + + #[must_use] + pub fn message(&self) -> &str { + &self.message + } + + /// 投影给客户端的 HTTP 状态；上游显式状态优先。 + #[must_use] + pub const fn status(&self) -> u16 { + match self.status { + Some(status) => status, + None => self.kind.default_status(), + } + } + + #[must_use] + pub fn body(&self) -> Option<&Bytes> { + self.body.as_ref() + } +} + +impl fmt::Display for LiveGatewayError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(formatter, "{}", self.message) + } +} + +impl std::error::Error for LiveGatewayError {} + +/// Provider 暴露给协议 adapter 的 Live sideband 能力。 +/// +/// 实现方负责把 call id 钉到创建账号、复验调用方身份，并用该账号的 +/// 运行时凭据拨号；Core 与协议 adapter 不接触账号凭据。 +/// +/// call 绑定的生命周期由实现方管理：sideband 断开只释放占用，绑定保留到 +/// 会话 TTL（官方 FramelessBidi 客户端会在传输中断后重连同一 call）； +/// 确认结束（挂断成功、上游报告通话不存在）才删除绑定。 +pub trait LiveGateway: Send + Sync { + /// 为已建立的 call 建立账号级上游 sideband 中继。 + /// + /// 返回的 [`LiveRelay`] 携带本次占用的 guard；中继结束即释放占用， + /// 协议 adapter 无需另行通知会话结束。 + fn open_sideband<'a>( + &'a self, + request: LiveSidebandRequest<'a>, + ) -> BoxFuture<'a, Result>; + + /// 以钉住账号转发一次性调用（当前仅 hangup）。 + fn hangup<'a>( + &'a self, + request: LiveHangupRequest<'a>, + ) -> BoxFuture<'a, Result>; +} + +/// call id 形态校验；与上游 `Location` 的可解析范围一致。 +#[must_use] +pub fn is_valid_call_id(call_id: &str) -> bool { + !call_id.is_empty() + && call_id.len() <= 128 + && call_id + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) +} + +/// 从上游 `Location` 提取 call id；接受裸 id、路径尾段与 `?call_id=` 三种形态。 +/// +/// Location 的形态由上游合同决定（`…/live/{id}`、`…/calls/{id}`、 +/// `…/realtime?call_id={id}`、裸 id），不引入完整 URL 解析依赖。 +#[must_use] +pub fn call_id_from_location(location: &str) -> Option { + let location = location.trim(); + if is_valid_call_id(location) { + return Some(location.to_owned()); + } + if let Some(start) = location.find("call_id=") { + let rest = &location[start + "call_id=".len()..]; + let candidate = rest.split('&').next().unwrap_or(rest); + if is_valid_call_id(candidate) { + return Some(candidate.to_owned()); + } + } + let after_authority = location.split_once("//").map(|(_, rest)| rest); + let path = match after_authority { + Some(rest) => rest.split_once('/').map(|(_, path)| path).unwrap_or(""), + None => location, + }; + let segments = path + .split('?') + .next() + .unwrap_or("") + .split('/') + .filter(|segment| !segment.is_empty()) + .collect::>(); + if segments.len() < 2 { + return None; + } + let call_id = segments[segments.len() - 1]; + let previous = segments[segments.len() - 2]; + if is_valid_call_id(call_id) && (previous == "live" || previous == "calls") { + return Some(call_id.to_owned()); + } + None +} diff --git a/backend/crates/gateway-core/src/routing/snapshot.rs b/backend/crates/gateway-core/src/routing/snapshot.rs index 3db4785af..43694e33e 100644 --- a/backend/crates/gateway-core/src/routing/snapshot.rs +++ b/backend/crates/gateway-core/src/routing/snapshot.rs @@ -1176,8 +1176,13 @@ impl RuntimeSnapshot { provider: provider.as_str().to_owned(), }); } - let model_binding_valid = - matches!(operation.kind(), OperationKind::CountTokens) == upstream_model.is_some(); + // CountTokens 必须绑定目标模型；其余端点可选携带上游模型——Provider 原生 + // 端点（如 live 语音）借此让账号模型权限参与路由检查。 + let model_binding_valid = if matches!(operation.kind(), OperationKind::CountTokens) { + upstream_model.is_some() + } else { + true + }; let model_capable = upstream_model.is_none_or(|model| { match self .provider_models diff --git a/backend/crates/gateway-core/tests/live.rs b/backend/crates/gateway-core/tests/live.rs new file mode 100644 index 000000000..6a745b0fc --- /dev/null +++ b/backend/crates/gateway-core/tests/live.rs @@ -0,0 +1,45 @@ +use gateway_core::live::{call_id_from_location, is_valid_call_id}; + +#[test] +fn call_ids_accept_only_upstream_identifier_characters() { + assert!(is_valid_call_id("call_abcd-123")); + assert!(is_valid_call_id( + "RTCV_0123456789abcdefghijklmnopqrstuvwxyz" + )); + assert!(!is_valid_call_id("")); + assert!(!is_valid_call_id("bad id")); + assert!(!is_valid_call_id("slash/id")); + assert!(!is_valid_call_id(&"a".repeat(129))); +} + +#[test] +fn location_call_id_accepts_the_three_official_shapes() { + assert_eq!( + call_id_from_location("call_abcd-123"), + Some("call_abcd-123".to_owned()) + ); + assert_eq!( + call_id_from_location("https://api.openai.com/v1/live/call_1"), + Some("call_1".to_owned()) + ); + assert_eq!( + call_id_from_location("https://api.openai.com/v1/realtime/calls/call_2"), + Some("call_2".to_owned()) + ); + assert_eq!( + call_id_from_location("/backend-api/codex/realtime/calls/call_3?x=1"), + Some("call_3".to_owned()) + ); + assert_eq!( + call_id_from_location("/v1/live?call_id=call_4"), + Some("call_4".to_owned()) + ); +} + +#[test] +fn location_without_a_parseable_call_id_is_rejected() { + assert_eq!(call_id_from_location("/v1/live"), None); + assert_eq!(call_id_from_location("/v1/live?call_id=bad id"), None); + assert_eq!(call_id_from_location("/v1/live?other=call_1"), None); + assert_eq!(call_id_from_location("/v1/realtime/wrong/call_1"), None); +} diff --git a/backend/crates/gateway-core/tests/main.rs b/backend/crates/gateway-core/tests/main.rs index bc31e960d..395730830 100644 --- a/backend/crates/gateway-core/tests/main.rs +++ b/backend/crates/gateway-core/tests/main.rs @@ -8,6 +8,7 @@ mod error; mod event; mod health; mod lifecycle; +mod live; mod metering; mod middleware; mod operation; diff --git a/backend/crates/gateway-core/tests/routing/mod.rs b/backend/crates/gateway-core/tests/routing/mod.rs index a6d66860e..e0ad53344 100644 --- a/backend/crates/gateway-core/tests/routing/mod.rs +++ b/backend/crates/gateway-core/tests/routing/mod.rs @@ -591,8 +591,10 @@ fn token_count_endpoint_should_require_exact_model_capability_and_scope() { } #[test] -fn provider_http_endpoint_should_be_provider_scoped_without_a_model_binding() { - let snapshot = snapshot(); +fn provider_http_endpoint_should_carry_the_model_into_account_permissions() { + use gateway_core::account::{AccountModelAccess, AccountModelAccessMode}; + use gateway_core::error::RoutingError; + let provider = ProviderKind::new("openai").expect("provider"); let operation = Operation::ProviderHttp( ProviderHttpRequest::new( @@ -604,29 +606,52 @@ fn provider_http_endpoint_should_be_provider_scoped_without_a_model_binding() { ) .expect("provider HTTP operation"), ); + let snapshot = RuntimeSnapshot::new( + ConfigRevision::new(1).expect("revision"), + settings(), + vec![provider.clone()], + Vec::new(), + Vec::new(), + ) + .expect("snapshot") + .with_account_directory(account_directory()); + let model = UpstreamModelId::new("gpt-live-1-codex").expect("model"); let plan = snapshot .plan_provider_endpoint( &provider, - None, + Some(&model), &operation, snapshot.all_account_scope(), &RoutingContext::default(), ) - .expect("provider endpoint"); + .expect("provider endpoint with a permitted model"); assert_eq!(plan.candidates()[0].provider(), &provider); - assert_eq!(plan.candidates()[0].upstream_model(), None); - assert!( - snapshot - .plan_provider_endpoint( - &provider, - Some(&UpstreamModelId::new("gpt-5.5").expect("model")), - &operation, - snapshot.all_account_scope(), - &RoutingContext::default(), + assert_eq!(plan.candidates()[0].upstream_model(), Some(&model)); + // 账号范围整体禁止该模型时在路由层拒绝，而不是交给未授权账号服务。 + let denied_directory = Arc::new(RuntimeAccountDirectory::new(BTreeMap::from([( + ProviderAccountId::new("acct_openai").expect("account"), + RuntimeAccount::new(provider.clone(), BTreeSet::new()).with_model_access( + AccountModelAccess::new( + AccountModelAccessMode::Denylist, + vec!["gpt-live-1-codex".to_owned()], ) - .is_err(), - "raw provider HTTP cannot smuggle a model binding" - ); + .expect("model access"), + ), + )]))); + let denied_scope = Arc::new(FrozenAccountScope::new( + denied_directory, + ClientRoutingScope::all_accounts(), + )); + assert!(matches!( + snapshot.plan_provider_endpoint( + &provider, + Some(&model), + &operation, + denied_scope, + &RoutingContext::default(), + ), + Err(RoutingError::NoCapableProviderEndpoint { .. }) + )); } #[test] diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index 0ea4eefe3..6500087ec 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -98,10 +98,15 @@ pub(crate) struct SelectCodexProviderEndpointCredential<'a> { pub request_url: &'a Url, pub attempt: &'a AttemptContext, pub session_affinity: Option<&'a CodexSessionAffinity>, + /// 本端点的上游模型；提供后按账号模型权限过滤候选（如 live 语音）。 + pub upstream_model: Option<&'a str>, + /// 本端点只接受 OAuth 凭据时排除 API Key 账号，避免混合池选中后必然失败。 + pub requires_oauth: bool, } struct CredentialSelectionInput<'a> { requires_websocket: bool, + requires_oauth: bool, request_url: &'a Url, attempt: &'a AttemptContext, session_affinity_key: Option<&'a ProviderSessionAffinityKey>, @@ -297,6 +302,7 @@ impl CodexCredentialSelector { ) -> Result { let input = CredentialSelectionInput { requires_websocket: false, + requires_oauth: false, request_url: request.request_url, attempt: request.attempt, session_affinity_key: request.session_affinity_key, @@ -317,6 +323,7 @@ impl CodexCredentialSelector { ) -> Result { let input = CredentialSelectionInput { requires_websocket, + requires_oauth: false, request_url: request.request_url, attempt: request.attempt, session_affinity_key: request.session_affinity_key, @@ -370,20 +377,24 @@ impl CodexCredentialSelector { /// 为不属于 Responses 文本模型目录的 Provider 原生端点选择账号 /// /// 账号范围、健康度、配额、并发租约、cookie 与认证准备仍走同一套选择链路； - /// 原生端点没有 Responses 模型，不套用管理员配置的文本模型权限 + /// 原生端点默认没有 Responses 模型，不套用管理员配置的文本模型权限。 + /// 端点有明确上游模型（如 live 语音）时通过 `upstream_model` 让账号 + /// 模型权限参与候选过滤；`requires_oauth` 限定本端点支持的认证类型。 pub(crate) async fn select_for_provider_endpoint( &self, request: &SelectCodexProviderEndpointCredential<'_>, ) -> Result { let input = CredentialSelectionInput { requires_websocket: false, + requires_oauth: request.requires_oauth, request_url: request.request_url, attempt: request.attempt, session_affinity_key: request.session_affinity.map(CodexSessionAffinity::key), session_affinity_observation: request.session_affinity, guardian: false, }; - self.select_inner(&input, None, None).await + self.select_inner(&input, None, request.upstream_model) + .await } async fn select_inner( @@ -461,6 +472,8 @@ impl CodexCredentialSelector { .attempt .account_scope() .is_some_and(|scope| scope.allows(account.id()))) + && (!request.requires_oauth + || account.authentication_kind() == CODEX_AUTHENTICATION_KIND_OAUTH) && (diagnostic || upstream_model.is_none_or(|upstream_model| { let allowed = diff --git a/backend/crates/providers/openai/src/lib.rs b/backend/crates/providers/openai/src/lib.rs index c07972524..89f5fd644 100644 --- a/backend/crates/providers/openai/src/lib.rs +++ b/backend/crates/providers/openai/src/lib.rs @@ -179,7 +179,8 @@ pub async fn initialize( ) .map_err(OpenAiInitializeError::Provider)? .with_session_identity(session_identity) - .with_timezone(config.timezone), + .with_timezone(config.timezone) + .with_live_support(repository.clone()), ); let token_client = Arc::new( credential::token_client::openai_token_client( diff --git a/backend/crates/providers/openai/src/provider/execution.rs b/backend/crates/providers/openai/src/provider/execution.rs index 9ec43a182..93acb93bc 100644 --- a/backend/crates/providers/openai/src/provider/execution.rs +++ b/backend/crates/providers/openai/src/provider/execution.rs @@ -101,6 +101,9 @@ impl CodexProvider { request_url: &request.response_origin, attempt: &context, session_affinity: request.session_affinity.as_ref(), + // 图像/搜索端点没有对应的上游模型权限，也不限定认证类型。 + upstream_model: None, + requires_oauth: false, }) .await .map_err(map_selection_error)?; diff --git a/backend/crates/providers/openai/src/provider/live.rs b/backend/crates/providers/openai/src/provider/live.rs new file mode 100644 index 000000000..c83155bc3 --- /dev/null +++ b/backend/crates/providers/openai/src/provider/live.rs @@ -0,0 +1,750 @@ +//! Codex Live 语音会话：SDP 引导 attempt、call 会话注册表与账号级 sideband 网关。 +//! +//! 通话引导（`POST /v1/live`）复用 Provider HTTP 端点通道：Core 路由按客户端 +//! Key 的账号范围选路，Provider 把固定端点 `codex/realtime/calls` 映射到 +//! Codex backend，并把成功响应 `Location` 里的 call id 登记到进程内注册表。 +//! 后续 sideband / hangup 由 [`CodexLiveGateway`] 用钉住账号的凭据直连 +//! `api.openai.com`；音频不经过网关。 + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use bytes::Bytes; +use gateway_core::account::scope::FrozenAccountScope; +use gateway_core::engine::provider::{ProviderSelectionObservation, ProviderStream}; +use gateway_core::error::{ProviderError, ProviderErrorKind}; +use gateway_core::event::{ + GatewayEvent, ProtocolWireEvent, ProviderEvent, ProviderResponseHeader, + ProviderResponseObservation, ResponseMeta, +}; +use gateway_core::live::{ + LiveCallOutcome, LiveGateway, LiveGatewayError, LiveGatewayErrorKind, LiveHangupRequest, + LiveRelay, LiveSidebandRequest, LiveSidebandStyle, call_id_from_location, is_valid_call_id, +}; +use gateway_core::operation::{ProviderHttpMethod, ProviderHttpRequest}; +use gateway_core::policy::ClientApiKeyId; +use gateway_core::upstream::{UpstreamSendState, UpstreamTransport}; +use secrecy::ExposeSecret; +use url::Url; +use uuid::Uuid; + +use super::*; + +use crate::credential::{ + CODEX_AUTHENTICATION_KIND_OAUTH, CodexCredentialRepository, + SelectCodexProviderEndpointCredential, +}; +use crate::transport::websocket::{connect_live_sideband, into_live_relay}; +use crate::transport::{CODEX_REALTIME_CALLS_PATH, CodexRequestContext}; + +/// realtime calls 在 [`ProviderHttpRequest::endpoint`] 中的符号名； +/// Provider 将其映射到固定上游路径 [`CODEX_REALTIME_CALLS_PATH`]。 +pub(super) const LIVE_CALLS_ENDPOINT: &str = "realtime-calls"; +/// sideband / hangup 的 OpenAI 公共 API 基址；与通话引导的 chatgpt.com backend 不同。 +const OPENAI_API_WS_BASE: &str = "wss://api.openai.com/v1"; +const OPENAI_API_HTTP_BASE: &str = "https://api.openai.com/v1"; +/// call id 与创建会话的绑定寿命；过期后 sideband 无法再加入。 +const LIVE_SESSION_TTL: Duration = Duration::from_secs(3600); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum LiveClaimError { + Missing, + Busy, + OwnerMismatch, + Invalid, +} + +impl LiveClaimError { + fn kind(self) -> LiveGatewayErrorKind { + match self { + Self::Missing => LiveGatewayErrorKind::CallNotFound, + Self::Busy => LiveGatewayErrorKind::CallBusy, + Self::OwnerMismatch => LiveGatewayErrorKind::OwnerMismatch, + Self::Invalid => LiveGatewayErrorKind::InvalidCallId, + } + } + + fn message(self) -> &'static str { + match self { + Self::Missing => "codex live call not found", + Self::Busy => "codex live call already has a sideband connection", + Self::OwnerMismatch => "codex live call is outside the current API key scope", + Self::Invalid => "invalid codex live call id", + } + } +} + +struct LiveRegistryEntry { + account_id: gateway_core::account::ProviderAccountId, + upstream_model: UpstreamModelId, + client_api_key_id: ClientApiKeyId, + expires_at: Instant, + claimed: bool, +} + +/// 进程内 call id → 创建账号注册表。 +/// +/// 注册表只服务本进程的 sideband 钉住与身份复验；网关重启后未完成的语音 +/// 通话失去 sideband 能力，与上游会话自身的存活一致，不需要持久化。 +#[derive(Default)] +pub(crate) struct CodexLiveRegistry { + entries: Mutex>, +} + +impl CodexLiveRegistry { + fn sweep_expired(entries: &mut HashMap, now: Instant) { + entries.retain(|_, entry| entry.claimed || entry.expires_at > now); + } + + /// 登记成功创建的通话；call id 非法时忽略。 + pub(crate) fn register( + &self, + call_id: &str, + account_id: gateway_core::account::ProviderAccountId, + client_api_key_id: ClientApiKeyId, + upstream_model: UpstreamModelId, + ) { + if !is_valid_call_id(call_id) { + return; + } + let mut entries = self.entries.lock().expect("codex live registry poisoned"); + Self::sweep_expired(&mut entries, Instant::now()); + entries.insert( + call_id.to_owned(), + LiveRegistryEntry { + account_id, + upstream_model, + client_api_key_id, + expires_at: Instant::now() + LIVE_SESSION_TTL, + claimed: false, + }, + ); + } + + /// 认领 call：校验调用方并阻止并发 sideband；认领后暂停过期。 + /// + /// 调用方必须在认领后立即构造 [`LiveCallClaim`]（两者之间没有 await 点）， + /// guard 丢弃时归还占用，覆盖拨号取消、升级失败与中继结束等全部退出路径， + /// 不依赖显式释放调用。 + fn claim( + &self, + call_id: &str, + client_api_key_id: &ClientApiKeyId, + account_scope: &FrozenAccountScope, + ) -> Result { + if !is_valid_call_id(call_id) { + return Err(LiveClaimError::Invalid); + } + let mut entries = self.entries.lock().expect("codex live registry poisoned"); + Self::sweep_expired(&mut entries, Instant::now()); + let Some(entry) = entries.get_mut(call_id) else { + return Err(LiveClaimError::Missing); + }; + if entry.claimed { + return Err(LiveClaimError::Busy); + } + if &entry.client_api_key_id != client_api_key_id + || !account_scope.allows_model(&entry.account_id, entry.upstream_model.as_str()) + { + return Err(LiveClaimError::OwnerMismatch); + } + entry.claimed = true; + // 认领期间暂停过期；guard 释放时恢复计时。guard 覆盖全部退出路径， + // 不存在认领后无法回收的窗口。 + Ok(entry.account_id.clone()) + } + + /// 归还认领；由 [`LiveCallClaim`] 的 Drop 触发，过期计时恢复。 + fn release(&self, call_id: &str) { + let mut entries = self.entries.lock().expect("codex live registry poisoned"); + if let Some(entry) = entries.get_mut(call_id) { + entry.claimed = false; + entry.expires_at = Instant::now() + LIVE_SESSION_TTL; + } + } + + fn complete(&self, call_id: &str) { + let mut entries = self.entries.lock().expect("codex live registry poisoned"); + entries.remove(call_id); + } + + /// hangup 前的身份复验；不认领，也不因 sideband 已连接而拒绝—— + /// 挂断必须随时可用，否则通话中无法主动结束。 + fn peek_owner( + &self, + call_id: &str, + client_api_key_id: &ClientApiKeyId, + ) -> Result { + if !is_valid_call_id(call_id) { + return Err(LiveClaimError::Invalid); + } + let mut entries = self.entries.lock().expect("codex live registry poisoned"); + Self::sweep_expired(&mut entries, Instant::now()); + let Some(entry) = entries.get(call_id) else { + return Err(LiveClaimError::Missing); + }; + if &entry.client_api_key_id != client_api_key_id { + return Err(LiveClaimError::OwnerMismatch); + } + Ok(entry.account_id.clone()) + } +} + +/// 一次 sideband 认领的取消安全 guard；丢弃时归还占用并恢复过期计时。 +/// +/// guard 随 [`gateway_core::live::LiveRelay`] 存活：拨号失败、升级回调 +/// 未执行或中继结束时都会触发 Drop，条目不会永久停留在 claimed 状态。 +struct LiveCallClaim { + registry: Arc, + call_id: String, +} + +impl LiveCallClaim { + fn new(registry: Arc, call_id: String) -> Self { + Self { registry, call_id } + } +} + +impl Drop for LiveCallClaim { + fn drop(&mut self) { + self.registry.release(&self.call_id); + } +} + +impl gateway_core::live::LiveRelayGuard for LiveCallClaim {} + +impl CodexProvider { + /// 引擎 arm：把受限的 realtime calls Provider HTTP 操作发往 Codex backend。 + /// + /// `upstream_model` 是路由计划携带的实际语音模型：选号阶段按账号模型 + /// 权限过滤候选，禁止该模型的账号不会服务语音请求。 + pub(super) async fn execute_live_call( + self: Arc, + request: ProviderHttpRequest, + upstream_model: Option<&UpstreamModelId>, + context: AttemptContext, + ) -> Result { + if request.method() != ProviderHttpMethod::Post || request.endpoint() != LIVE_CALLS_ENDPOINT + { + return Err(provider_error( + ProviderErrorKind::Unsupported, + UpstreamSendState::NotSent, + )); + } + let upstream_model = upstream_model.cloned().ok_or_else(|| { + provider_error( + ProviderErrorKind::InvalidRequest, + UpstreamSendState::NotSent, + ) + })?; + let selection_started_at = Instant::now(); + let lease = self + .selector + .select_for_provider_endpoint(&SelectCodexProviderEndpointCredential { + request_url: &self.live_calls_url, + attempt: &context, + session_affinity: None, + upstream_model: Some(upstream_model.as_str()), + // realtime calls 端点绑定 ChatGPT OAuth 身份；在候选阶段就排除 + // API Key 账号，避免混合账号池选中不支持语音的账号后必然失败。 + requires_oauth: true, + }) + .await + .map_err(map_selection_error)?; + if lease.authentication().oauth().is_none() { + // 诊断等旁路仍可能到达非 OAuth 租约；此处兜底拒绝。 + return Err(provider_error( + ProviderErrorKind::Unsupported, + UpstreamSendState::NotSent, + )); + } + let account_selection_wait_ms = + u64::try_from(selection_started_at.elapsed().as_millis()).unwrap_or(u64::MAX); + let lease = Arc::new(lease); + let operation = Operation::ProviderHttp(request); + let provider_kind = ProviderKind::new(PROVIDER_NAME) + .map_err(|_| provider_error(ProviderErrorKind::Protocol, UpstreamSendState::NotSent))?; + let account_id = lease.account_id().clone(); + let provider = Arc::clone(&self); + let terminal_context = context.clone(); + context + .execute_middleware( + operation, + provider_kind, + Some(upstream_model.as_str().to_owned()), + account_id, + Box::new(move |operation, middleware_headers| { + Box::pin(async move { + provider + .execute_selected_live_call( + terminal_context, + operation, + upstream_model, + middleware_headers, + lease, + account_selection_wait_ms, + ) + .await + }) + }), + ) + .await + } + + async fn execute_selected_live_call( + self: Arc, + context: AttemptContext, + operation: Operation, + upstream_model: UpstreamModelId, + middleware_headers: Vec, + lease: Arc, + account_selection_wait_ms: u64, + ) -> Result { + let Operation::ProviderHttp(request) = operation else { + return Err(provider_error( + ProviderErrorKind::InvalidRequest, + UpstreamSendState::NotSent, + )); + }; + let allows_account_state_mutation = lease.allows_account_state_mutation(); + let provider_kind = ProviderKind::new(PROVIDER_NAME) + .map_err(|_| provider_error(ProviderErrorKind::Protocol, UpstreamSendState::NotSent))?; + let metadata = ProviderCallMetadata::new( + provider_kind, + upstream_model.clone(), + lease.account_id().clone(), + UpstreamTransport::new(HTTP_JSON_TRANSPORT).map_err(|_| { + provider_error(ProviderErrorKind::Protocol, UpstreamSendState::NotSent) + })?, + ) + .with_selection_observation(ProviderSelectionObservation::new( + account_selection_wait_ms, + lease.capacity_snapshot(), + )); + let content_type = request + .headers() + .iter() + .find(|header| header.name().eq_ignore_ascii_case("content-type")) + .and_then(|header| std::str::from_utf8(header.value()).ok()) + .map(str::to_owned); + let protocol_headers = request + .headers() + .iter() + .filter(|header| !header.name().eq_ignore_ascii_case("content-type")) + .map(|header| { + ( + header.name().to_ascii_lowercase(), + String::from_utf8_lossy(header.value()).into_owned(), + ) + }) + .collect::>(); + let events = cold_live_call_stream(ColdLiveCall { + client: self + .client_for_request(&context)? + .for_account(lease.account()) + .map_err(|_| { + provider_error(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent) + })? + .with_authentication(lease.authentication()) + .with_middleware_headers(middleware_headers), + registry: Arc::clone(&self.live_registry), + response_origin: self.live_calls_url.clone(), + endpoint_query: request.query().map(str::to_owned), + content_type, + protocol_headers, + upstream_model, + body: request.payload().body().clone(), + context, + selector: Arc::clone(&self.selector), + quota: Arc::clone(&self.quota), + lease: Arc::clone(&lease), + }); + let stream = ProviderStream::new(metadata, events, lease); + Ok(if allows_account_state_mutation { + stream.with_filtered_account_feedback( + Arc::clone(&self.account_feedback), + openai_failure_affects_account_score, + ) + } else { + stream + }) + } +} + +struct ColdLiveCall { + client: CodexBackendClient, + registry: Arc, + response_origin: Url, + endpoint_query: Option, + content_type: Option, + protocol_headers: Vec<(String, String)>, + upstream_model: UpstreamModelId, + body: Bytes, + context: AttemptContext, + selector: Arc, + quota: Arc, + lease: Arc, +} + +fn cold_live_call_stream(request: ColdLiveCall) -> EventStream { + Box::pin(async_stream::try_stream! { + let allows_account_state_mutation = request.lease.allows_account_state_mutation(); + let failure_context = OpenAiFailureContext { + client: &request.client, + selector: &request.selector, + quota: &request.quota, + response_origin: &request.response_origin, + cyber_policy_scope: None, + allows_account_state_mutation, + allows_capacity_feedback: !request.context.is_diagnostic_required_account(), + }; + let active_account = request.lease.account().clone(); + let authorization = request + .lease + .authentication() + .authorization_header() + .map_err(|_| { + provider_error(ProviderErrorKind::Unauthorized, UpstreamSendState::NotSent) + })?; + let request_id = request.context.request_id().as_str().to_owned(); + let trace = request.context.trace(); + let mut request_context = CodexRequestContext::auxiliary( + authorization.expose_secret(), + active_account.upstream_account_id(), + &request_id, + Some(request.lease.installation_id()), + ); + request_context.trace = Some(&trace); + request_context.account_selection = CodexAccountSelectionTelemetry::new( + request.lease.affinity_hit(), + request.lease.escape_reason(), + request.lease.account_switch(), + ); + + let deadline = request.context.deadline(); + if deadline.is_elapsed() { + Err(provider_error(ProviderErrorKind::Timeout, UpstreamSendState::NotSent))?; + return; + } + let cancellation = request.context.cancellation().clone(); + let attempt = tokio::select! { + biased; + _ = cancellation.cancelled() => Err(CodexHandshakeAttemptError::Cancelled), + _ = deadline.wait() => Err(CodexHandshakeAttemptError::Timeout), + response = request.client.post_live_call( + CODEX_REALTIME_CALLS_PATH, + request.endpoint_query.as_deref(), + request.content_type.as_deref(), + &request.protocol_headers, + request.body.clone(), + request_context, + ) => response.map_err(CodexHandshakeAttemptError::Client), + }; + if let Err(CodexHandshakeAttemptError::Client(error)) = &attempt { + log_client_upstream_error( + UpstreamErrorLogContext::new(&request.context, &active_account, None), + error, + ); + } + let response = match attempt.map_err(map_handshake_attempt_error) { + Ok(response) => response, + Err(mut failure) => { + if let Some(observation) = failure.observation.take() { + yield ProviderEvent::observation(observation); + } + apply_failure(&failure_context, &active_account, &failure).await; + Err(failure.error)?; + return; + } + }; + + let observation = ProviderResponseObservation::new( + UpstreamTransport::new(HTTP_JSON_TRANSPORT).map_err(|_| { + provider_error(ProviderErrorKind::Protocol, UpstreamSendState::Sent) + })?, + ) + .with_status_code(response.status) + .with_client_headers( + response + .forwarded_headers + .iter() + .map(|(name, value)| { + ProviderResponseHeader::new( + name.clone(), + Bytes::copy_from_slice(value.as_bytes()), + ) + }) + .collect(), + ); + yield ProviderEvent::observation(observation); + + if let Some(location) = response.location.as_deref() + && let Some(call_id) = call_id_from_location(location) + { + request.registry.register( + &call_id, + active_account.id().clone(), + request.context.client_api_key_ref().clone(), + request.upstream_model.clone(), + ); + } else { + tracing::warn!( + request_id = %request_id, + model = request.upstream_model.as_str(), + "codex live call response is missing a parseable Location header" + ); + } + + let response_meta = ResponseMeta::for_provider_endpoint(&request_id); + yield ProviderEvent::canonical(GatewayEvent::Started(response_meta.clone())); + let wire = ProtocolWireEvent::raw_http_body(PROVIDER_NAME, response.body).map_err(|_| { + provider_error(ProviderErrorKind::Protocol, UpstreamSendState::Sent) + })?; + yield ProviderEvent::wire(wire); + yield ProviderEvent::canonical(GatewayEvent::Completed( + response_meta.with_finish_reason(FinishReason::Stop), + )); + }) +} + +/// Core [`LiveGateway`] 的 Codex 实现：call 钉住账号 + 账号级凭据拨号。 +pub(crate) struct CodexLiveGateway { + registry: Arc, + repository: CodexCredentialRepository, + client: CodexBackendClient, +} + +impl CodexLiveGateway { + pub(crate) fn new( + registry: Arc, + repository: CodexCredentialRepository, + client: CodexBackendClient, + ) -> Self { + Self { + registry, + repository, + client, + } + } + + /// 解析钉住账号的运行时凭据；非 OAuth 账号不支持 realtime 端点。 + async fn load_pinned_credential( + &self, + account_id: &gateway_core::account::ProviderAccountId, + ) -> Result<(ProviderAccount, String), LiveGatewayError> { + let credential_unavailable = |message: &'static str| { + LiveGatewayError::new(LiveGatewayErrorKind::CredentialUnavailable, message) + }; + let account = self + .repository + .store() + .get_account(account_id) + .await + .map_err(|_| credential_unavailable("codex live pinned account is unavailable"))? + .ok_or_else(|| credential_unavailable("codex live pinned account is unavailable"))?; + if account.authentication_kind() != CODEX_AUTHENTICATION_KIND_OAUTH { + return Err(LiveGatewayError::new( + LiveGatewayErrorKind::Unsupported, + "codex live requires an OAuth codex account", + )); + } + let credential = self + .repository + .load_runtime_credential(&account) + .await + .map_err(|_| credential_unavailable("codex live pinned credential is unavailable"))?; + let authorization = credential + .authentication + .authorization_header() + .map_err(|_| credential_unavailable("codex live pinned credential is unusable"))?; + Ok((account, authorization.expose_secret().to_owned())) + } + + fn sideband_endpoint(style: LiveSidebandStyle, call_id: &str) -> String { + match style { + LiveSidebandStyle::Live => format!("{OPENAI_API_WS_BASE}/live/{call_id}"), + LiveSidebandStyle::RealtimeCalls => { + format!("{OPENAI_API_WS_BASE}/realtime/calls/{call_id}") + } + LiveSidebandStyle::RealtimeQuery => { + format!("{OPENAI_API_WS_BASE}/realtime?intent=quicksilver&call_id={call_id}") + } + } + } +} + +fn map_sideband_dial_error(error: CodexWebSocketExchangeError) -> LiveGatewayError { + match error { + CodexWebSocketExchangeError::Upstream(upstream) => { + let body = upstream + .client_response + .as_ref() + .map(|response| response.body().clone()); + LiveGatewayError::new( + LiveGatewayErrorKind::UpstreamUnavailable, + format!("codex live sideband handshake failed: {}", upstream.body), + ) + .with_upstream(upstream.status_code, body) + } + other => LiveGatewayError::new( + LiveGatewayErrorKind::UpstreamUnavailable, + format!("codex live sideband upstream unavailable: {other}"), + ), + } +} + +fn map_live_http_error(error: CodexClientError) -> LiveGatewayError { + match error { + CodexClientError::Upstream { + status: _, + client_response: Some(response), + .. + } => LiveGatewayError::new( + LiveGatewayErrorKind::UpstreamUnavailable, + format!( + "codex live upstream call failed with status {}", + response.status() + ), + ) + .with_upstream(response.status(), Some(response.body().clone())), + CodexClientError::Upstream { status, body, .. } => LiveGatewayError::new( + LiveGatewayErrorKind::UpstreamUnavailable, + format!( + "codex live upstream call failed with status {}", + status.as_u16() + ), + ) + .with_upstream(status.as_u16(), Some(Bytes::from(body))), + other => LiveGatewayError::new( + LiveGatewayErrorKind::UpstreamUnavailable, + format!("codex live upstream call failed: {other}"), + ), + } +} + +impl LiveGateway for CodexLiveGateway { + fn open_sideband<'a>( + &'a self, + request: LiveSidebandRequest<'a>, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { + let account_id = self + .registry + .claim( + request.call_id, + request.client_api_key_id, + request.account_scope, + ) + .map_err(|claim_error| { + LiveGatewayError::new(claim_error.kind(), claim_error.message()) + })?; + // 认领与 guard 构造之间没有 await 点，取消不会留下孤儿认领。 + let claim = LiveCallClaim::new(Arc::clone(&self.registry), request.call_id.to_owned()); + let (account, authorization) = self.load_pinned_credential(&account_id).await?; + if !account.enabled() { + return Err(LiveGatewayError::new( + LiveGatewayErrorKind::OwnerMismatch, + "codex live pinned account is disabled", + )); + } + let endpoint = Self::sideband_endpoint(request.style, request.call_id); + let mut headers = vec![("authorization".to_owned(), authorization)]; + if let Some(account_id) = account.upstream_account_id() { + headers.push(("chatgpt-account-id".to_owned(), account_id.to_owned())); + } + headers.extend(request.protocol_headers); + if !request.subprotocols.is_empty() { + headers.push(( + "sec-websocket-protocol".to_owned(), + request.subprotocols.join(", "), + )); + } + match connect_live_sideband(endpoint, headers, account.outbound_proxy().cloned()).await + { + Ok(sideband) => { + let mut relay = into_live_relay(sideband.stream, sideband.subprotocol); + // guard 随中继存活：传输中断只释放占用，绑定保留到会话 TTL， + // 官方 FramelessBidi 客户端会重连同一 call。 + relay.with_guard(Box::new(claim)); + Ok(relay) + } + Err(error) => { + if sideband_call_gone(&error) { + // 上游报告通话不存在（404/410，对应官方会话结束判定）， + // 立即删除绑定，避免死条目占满 TTL。 + self.registry.complete(request.call_id); + } + Err(map_sideband_dial_error(error)) + } + } + }) + } + + fn hangup<'a>( + &'a self, + request: LiveHangupRequest<'a>, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { + let account_id = self + .registry + .peek_owner(request.call_id, request.client_api_key_id) + .map_err(|claim_error| { + LiveGatewayError::new(claim_error.kind(), claim_error.message()) + })?; + let (account, authorization) = self.load_pinned_credential(&account_id).await?; + // hangup 与引导同源：必须走钉住账号的出口代理与连接池， + // 不能使用共享 client 直连发出。 + let client = self.client.for_account(&account).map_err(|_| { + LiveGatewayError::new( + LiveGatewayErrorKind::CredentialUnavailable, + "codex live hangup account client is unavailable", + ) + })?; + let request_id = Uuid::new_v4().to_string(); + let mut context = CodexRequestContext::auxiliary( + authorization.as_str(), + account.upstream_account_id(), + &request_id, + None, + ); + context.account_selection = CodexAccountSelectionTelemetry::NONE; + let url = format!( + "{OPENAI_API_HTTP_BASE}/realtime/calls/{}/hangup", + request.call_id + ); + let outcome = client + .post_live_hangup( + url, + request.content_type.as_deref(), + &request.protocol_headers, + request.body.clone(), + context, + ) + .await; + match outcome { + Ok(response) => { + if (200..300).contains(&response.status) { + self.registry.complete(request.call_id); + } + Ok(LiveCallOutcome { + status: response.status, + headers: response.forwarded_headers, + body: response.body, + }) + } + Err(error) => Err(map_live_http_error(error)), + } + }) + } +} + +/// 上游拨号失败是否等价于通话已结束；与官方 `webrtc_sideband_session_ended` +/// 一致，只把 404/410 视为会话结束信号。 +fn sideband_call_gone(error: &CodexWebSocketExchangeError) -> bool { + matches!( + error, + CodexWebSocketExchangeError::Upstream(upstream) + if matches!(upstream.status_code, 404 | 410) + ) +} diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index 0119f697d..59544ea81 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -56,9 +56,9 @@ use uuid::Uuid; use crate::credential::{ CodexAccountFailure, CodexCredentialCatalogError, CodexCredentialCatalogService, CodexCredentialLease, CodexCredentialQuotaService, CodexCredentialRefreshOutcome, - CodexCredentialRefreshService, CodexCredentialSelector, CodexCyberPolicyScope, - CodexQuotaRefreshPolicy, CodexSessionAffinity, CredentialSelectionError, RuntimeCodexCookie, - SelectCodexCredential, SelectCodexProviderEndpointCredential, + CodexCredentialRefreshService, CodexCredentialRepository, CodexCredentialSelector, + CodexCyberPolicyScope, CodexQuotaRefreshPolicy, CodexSessionAffinity, CredentialSelectionError, + RuntimeCodexCookie, SelectCodexCredential, SelectCodexProviderEndpointCredential, derive_codex_cyber_policy_session_key, derive_codex_endpoint_session_affinity, derive_codex_session_affinity, derive_previous_response_id_hash, }; @@ -93,18 +93,21 @@ use crate::transport::websocket::{ }; use crate::transport::{ CODEX_ALPHA_SEARCH_PATH, CODEX_IMAGE_EDITS_PATH, CODEX_IMAGE_GENERATIONS_PATH, - CODEX_RESPONSES_PATH, CodexAccountSelectionTelemetry, CodexBackendClient, - CodexBackendJsonResponse, CodexBackendStreamingResponse, CodexBackendTransport, - CodexClientError, CodexRateLimitUpdates, CodexRequestContext, CodexResponseMetadata, - CodexResponseMetadataUpdates, CodexTransportMetrics, CodexUpstreamDiagnostics, - CodexWebSocketPool, endpoint_url, normalize_selected_codex_downstream_body, + CODEX_REALTIME_CALLS_PATH, CODEX_RESPONSES_PATH, CodexAccountSelectionTelemetry, + CodexBackendClient, CodexBackendJsonResponse, CodexBackendStreamingResponse, + CodexBackendTransport, CodexClientError, CodexRateLimitUpdates, CodexRequestContext, + CodexResponseMetadata, CodexResponseMetadataUpdates, CodexTransportMetrics, + CodexUpstreamDiagnostics, CodexWebSocketPool, endpoint_url, + normalize_selected_codex_downstream_body, }; mod execution; mod failure; +mod live; mod observation; mod upstream_adapter; mod workers; +pub(crate) use live::{CodexLiveGateway, CodexLiveRegistry}; pub(crate) use workers::ClientReleaseServices; use execution::*; @@ -157,9 +160,12 @@ pub struct CodexProvider { image_generations_url: Url, image_edits_url: Url, search_url: Url, + live_calls_url: Url, session_identity: Option, session_transport_recovery: CodexSessionTransportRecovery, stream_max_retries: u32, + live_registry: Arc, + live_gateway: Option>, } struct PreparedGenerateRequest { @@ -258,6 +264,8 @@ impl CodexProvider { .map_err(|_| CodexProviderConfigError::InvalidBaseUrl)?; let search_url = Url::parse(&endpoint_url(&base_url, CODEX_ALPHA_SEARCH_PATH)) .map_err(|_| CodexProviderConfigError::InvalidBaseUrl)?; + let live_calls_url = Url::parse(&endpoint_url(&base_url, CODEX_REALTIME_CALLS_PATH)) + .map_err(|_| CodexProviderConfigError::InvalidBaseUrl)?; let client = CodexBackendClient::new(http, base_url, profile).with_websocket_pool(websocket_pool); Ok(Self { @@ -270,9 +278,12 @@ impl CodexProvider { image_generations_url, image_edits_url, search_url, + live_calls_url, session_identity: None, session_transport_recovery: CodexSessionTransportRecovery::default(), stream_max_retries, + live_registry: Arc::new(CodexLiveRegistry::default()), + live_gateway: None, }) } @@ -289,6 +300,18 @@ impl CodexProvider { self.session_identity = Some(identity); self } + + /// 绑定 Live 语音支持：call 注册表 + 账号级 sideband 网关。 + /// 需要凭据仓库句柄；未绑定时不暴露 sideband 与 hangup 能力。 + pub(crate) fn with_live_support(mut self, repository: CodexCredentialRepository) -> Self { + let gateway = Arc::new(CodexLiveGateway::new( + Arc::clone(&self.live_registry), + repository, + self.client.clone(), + )); + self.live_gateway = Some(gateway); + self + } } impl fmt::Debug for CodexProvider { @@ -328,6 +351,12 @@ impl Provider for CodexProvider { PROVIDER_NAME } + fn live_gateway(&self) -> Option> { + self.live_gateway + .clone() + .map(|gateway| gateway as Arc) + } + fn catalog_generation(&self) -> ProviderCatalogGeneration { self.catalog.catalog_generation() } @@ -466,6 +495,11 @@ impl Provider for CodexProvider { if let Operation::Search(search) = request.operation() { return self.execute_search(search, candidate, context).await; } + if let Operation::ProviderHttp(request) = request.operation() { + return self + .execute_live_call(request.clone(), candidate.upstream_model(), context) + .await; + } let Operation::Generate(generate) = request.operation() else { return Err(provider_error( ProviderErrorKind::Unsupported, diff --git a/backend/crates/providers/openai/src/transport/client_live.rs b/backend/crates/providers/openai/src/transport/client_live.rs new file mode 100644 index 000000000..3542770a6 --- /dev/null +++ b/backend/crates/providers/openai/src/transport/client_live.rs @@ -0,0 +1,231 @@ +//! Codex Live / Realtime 语音通话的账号级上游透明传输。 +//! +//! 语音通话 SDP 引导与 hangup 都是一次性账号级 POST； +//! 正文与响应头（`Location` 携带 call id）均不经过 serde 往返，保持上游原始字节。 + +use std::time::Instant; + +use bytes::Bytes; +use reqwest::header::HeaderMap; + +use super::{ + client::{ + CodexBackendClient, CodexBackendTransport, CodexClientError, CodexClientResult, + CodexClientVisibleUpstreamResponse, CodexRequestContext, CodexTransportMetrics, + elapsed_duration_millis, http_version_name, read_error_response_body, retry_after_seconds, + }, + diagnostics::{CodexUpstreamDiagnostics, CodexUpstreamSendPhase}, + endpoints::endpoint_url, + response_meta, +}; + +impl CodexBackendClient { + /// 向 Codex realtime calls 端点发送账号级 POST；请求与响应正文不经过 serde。 + /// + /// `extra_protocol_headers` 是客户端协议头经过允许清单过滤后的结果， + /// 依次追加在标准账号头之后，遇到同名标准头时覆盖客户端语义字段。 + pub(crate) async fn post_live_call( + &self, + endpoint_path: &'static str, + query: Option<&str>, + content_type: Option<&str>, + extra_protocol_headers: &[(String, String)], + body: Bytes, + context: CodexRequestContext<'_>, + ) -> CodexClientResult { + let mut url = endpoint_url(&self.base_url, endpoint_path); + if let Some(query) = query { + url.push('?'); + url.push_str(query); + } + self.send_live_request(url, content_type, extra_protocol_headers, body, context) + .await + } + + /// 向显式上游 URL（hangup 位于 api.openai.com）发送账号级 POST。 + pub(crate) async fn post_live_hangup( + &self, + url: String, + content_type: Option<&str>, + extra_protocol_headers: &[(String, String)], + body: Bytes, + context: CodexRequestContext<'_>, + ) -> CodexClientResult { + self.send_live_request(url, content_type, extra_protocol_headers, body, context) + .await + } + + async fn send_live_request( + &self, + url: String, + content_type: Option<&str>, + extra_protocol_headers: &[(String, String)], + body: Bytes, + context: CodexRequestContext<'_>, + ) -> CodexClientResult { + let profile = self.profile.snapshot(); + let mut headers = self.model_request_headers(&profile, context)?; + headers.insert( + reqwest::header::CONTENT_TYPE, + reqwest::header::HeaderValue::from_str(content_type.unwrap_or("application/json"))?, + ); + self.append_middleware_headers(&mut headers)?; + for (name, value) in extra_protocol_headers { + let name = reqwest::header::HeaderName::from_bytes(name.as_bytes())?; + let value = reqwest::header::HeaderValue::from_str(value)?; + headers.insert(name, value); + } + + let trace = context + .trace + .cloned() + .unwrap_or_default() + .exchange("http_json"); + trace.headers( + "upstream.request.headers", + serde_json::json!({ + "method": "POST", "endpoint": url, + }), + headers + .iter() + .map(|(name, value)| (name.as_str(), value.as_bytes())), + ); + trace.capture("upstream.request.body", &body); + + let headers_started_at = Instant::now(); + let response = self + .client + .post(url) + .headers(headers) + .body(body) + .send() + .await + .map_err(CodexClientError::HttpJson)?; + let upstream_headers_ms = elapsed_duration_millis(headers_started_at.elapsed()); + let http_version = http_version_name(response.version()).to_owned(); + let status = response.status(); + trace.headers( + "upstream.response.headers", + serde_json::json!({ + "status": status.as_u16(), "httpVersion": http_version, + "headersMs": upstream_headers_ms, + }), + response + .headers() + .iter() + .map(|(name, value)| (name.as_str(), value.as_bytes())), + ); + let diagnostics = response_meta::diagnostics(Some(status.as_u16()), response.headers()); + let set_cookie_headers = response_meta::set_cookie_headers(response.headers()); + let rate_limit_headers = response_meta::rate_limit_headers(response.headers()); + let retry_after_seconds = retry_after_seconds(response.headers(), None); + let transport_metrics = CodexTransportMetrics { + upstream_headers_ms: Some(upstream_headers_ms), + http_version: Some(http_version), + ..CodexTransportMetrics::default() + }; + + if !status.is_success() { + let content_type = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .map(|value| value.as_bytes().to_vec()); + let client_headers = response_meta::client_headers(response.headers()); + let raw_body = read_error_response_body(response).await.map_err(|source| { + CodexClientError::ErrorBodyRead { + source, + status, + diagnostics: Box::new(diagnostics.clone()), + transport: CodexBackendTransport::HttpJson, + transport_metrics: Box::new(transport_metrics.clone()), + } + })?; + trace.capture("upstream.error.body", &raw_body); + let error_body = String::from_utf8_lossy(&raw_body).into_owned(); + return Err(CodexClientError::Upstream { + status, + body: error_body, + client_response: Some(Box::new(CodexClientVisibleUpstreamResponse::new( + status, + content_type, + client_headers, + raw_body, + ))), + retry_after_seconds, + diagnostics: Box::new(diagnostics), + set_cookie_headers, + rate_limit_headers, + transport: CodexBackendTransport::HttpJson, + transport_metrics: Box::new(transport_metrics), + send_phase: CodexUpstreamSendPhase::AfterPayload, + }); + } + + let location = response + .headers() + .get(reqwest::header::LOCATION) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let content_type = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let forwarded_headers = live_forwarded_headers(response.headers()); + let body = response.bytes().await.map_err(CodexClientError::HttpJson)?; + trace.capture("upstream.response.body", &body); + Ok(CodexLiveCallResponse { + status: status.as_u16(), + body, + location, + content_type, + retry_after_seconds, + forwarded_headers, + set_cookie_headers, + rate_limit_headers, + diagnostics, + transport_metrics, + }) + } +} + +/// 上游 2xx 响应中可回传客户端的头；`Location` 是客户端获知 call id 的合同。 +fn live_forwarded_headers(headers: &HeaderMap) -> Vec<(String, String)> { + const FORWARDABLE: [&str; 5] = [ + "location", + "content-type", + "retry-after", + "x-request-id", + "openai-request-id", + ]; + headers + .iter() + .filter(|(name, _)| FORWARDABLE.contains(&name.as_str())) + .filter_map(|(name, value)| { + value + .to_str() + .ok() + .map(|value| (name.as_str().to_owned(), value.to_owned())) + }) + .collect() +} + +/// Live 通话引导与 hangup 的账号级响应。 +#[derive(Debug)] +pub struct CodexLiveCallResponse { + /// 上游 2xx 状态码。 + pub status: u16, + /// 上游原始正文（引导成功时为 SDP answer）。 + pub body: Bytes, + /// `Location` 响应头；成功引导响应携带 call id 的唯一来源。 + pub location: Option, + /// 上游 Content-Type。 + pub content_type: Option, + pub retry_after_seconds: Option, + /// 可回传客户端的响应头子集。 + pub forwarded_headers: Vec<(String, String)>, + pub set_cookie_headers: Vec, + pub rate_limit_headers: Vec<(String, String)>, + pub diagnostics: CodexUpstreamDiagnostics, + pub transport_metrics: CodexTransportMetrics, +} diff --git a/backend/crates/providers/openai/src/transport/endpoints.rs b/backend/crates/providers/openai/src/transport/endpoints.rs index a59d3c102..a307f241c 100644 --- a/backend/crates/providers/openai/src/transport/endpoints.rs +++ b/backend/crates/providers/openai/src/transport/endpoints.rs @@ -8,6 +8,8 @@ pub const CODEX_IMAGE_GENERATIONS_PATH: &str = "/codex/images/generations"; pub const CODEX_IMAGE_EDITS_PATH: &str = "/codex/images/edits"; /// `/codex/alpha/search` pub const CODEX_ALPHA_SEARCH_PATH: &str = "/codex/alpha/search"; +/// `/codex/realtime/calls`；语音通话 SDP 引导端点。 +pub const CODEX_REALTIME_CALLS_PATH: &str = "/codex/realtime/calls"; /// `/api/codex/usage` pub const CODEX_USAGE_API_PATH: &str = "/api/codex/usage"; /// `/wham/usage` diff --git a/backend/crates/providers/openai/src/transport/mod.rs b/backend/crates/providers/openai/src/transport/mod.rs index 5aaaac8f4..5aec44d08 100644 --- a/backend/crates/providers/openai/src/transport/mod.rs +++ b/backend/crates/providers/openai/src/transport/mod.rs @@ -4,6 +4,7 @@ pub mod canonical; pub mod catalog; pub mod client; mod client_json; +mod client_live; mod client_sse; pub(crate) mod connection; pub mod diagnostics; @@ -39,12 +40,13 @@ pub use self::{ CodexResponseMetadataUpdates, CodexTransportDecision, CodexTransportMetrics, build_reqwest_client, }, + client_live::CodexLiveCallResponse, diagnostics::{CodexUpstreamDiagnostics, CodexUpstreamSendPhase}, endpoints::{ CODEX_ALPHA_SEARCH_PATH, CODEX_IMAGE_EDITS_PATH, CODEX_IMAGE_GENERATIONS_PATH, - CODEX_RESPONSES_PATH, CODEX_USAGE_API_PATH, WHAM_PROFILE_STATISTICS_PATH, - WHAM_RATE_LIMIT_RESET_CREDITS_CONSUME_PATH, WHAM_RATE_LIMIT_RESET_CREDITS_PATH, - WHAM_USAGE_PATH, endpoint_url, usage_endpoint_url, + CODEX_REALTIME_CALLS_PATH, CODEX_RESPONSES_PATH, CODEX_USAGE_API_PATH, + WHAM_PROFILE_STATISTICS_PATH, WHAM_RATE_LIMIT_RESET_CREDITS_CONSUME_PATH, + WHAM_RATE_LIMIT_RESET_CREDITS_PATH, WHAM_USAGE_PATH, endpoint_url, usage_endpoint_url, }, headers::build_codex_model_headers, profile_avatar::{ diff --git a/backend/crates/providers/openai/src/transport/websocket/handshake.rs b/backend/crates/providers/openai/src/transport/websocket/handshake.rs index a010c5944..dffc5e680 100644 --- a/backend/crates/providers/openai/src/transport/websocket/handshake.rs +++ b/backend/crates/providers/openai/src/transport/websocket/handshake.rs @@ -116,6 +116,13 @@ pub fn responses_websocket_endpoint(base_url: &str) -> String { } } +/// Live sideband 复用同一条拨号路径；不计入请求连接预算。 +pub(super) async fn connect_sideband_websocket( + connection: &CodexWebSocketConnection, +) -> Result<(RawWsStream, WsResponse>>), CodexWebSocketExchangeError> { + connect_websocket(connection, false).await +} + pub(super) async fn connect_pumped_websocket( connection: &CodexWebSocketConnection, keepalive: PumpKeepalive, diff --git a/backend/crates/providers/openai/src/transport/websocket/live.rs b/backend/crates/providers/openai/src/transport/websocket/live.rs new file mode 100644 index 000000000..66708449d --- /dev/null +++ b/backend/crates/providers/openai/src/transport/websocket/live.rs @@ -0,0 +1,162 @@ +//! Codex Live sideband 的账号级上游拨号与抽象帧中继。 +//! +//! sideband 连接使用钉住账号的凭据直接拨 `api.openai.com` 的 realtime +//! WebSocket，并把上游帧映射为 Core 的 [`LiveFrame`]；协议 adapter 在 +//! 另一侧做同样映射，两侧都不解释帧内容。 + +use bytes::Bytes; +use futures::{SinkExt, StreamExt}; +use gateway_core::live::{ + LiveClose, LiveFrame, LiveGatewayError, LiveGatewayErrorKind, LiveRelay, LiveRelaySink, + LiveRelayStream, +}; +use tokio_tungstenite::tungstenite::Message; + +use super::{ + error::CodexWebSocketExchangeError, handshake::connect_sideband_websocket, + model::CodexWebSocketConnection, pump::RawWsStream, +}; + +/// sideband 上游连接；原始流加协商子协议。 +pub(crate) struct CodexLiveSideband { + pub(crate) stream: RawWsStream, + pub(crate) subprotocol: Option, +} + +/// 以给定 endpoint 与业务头拨号 sideband 上游。 +/// +/// `business_headers` 包含认证、账号与客户端协议头；标准握手字段由本层生成，子协议 +/// offer 由调用方以 `sec-websocket-protocol` 头显式携带。 +pub(crate) async fn connect_live_sideband( + endpoint: String, + mut business_headers: Vec<(String, String)>, + outbound_proxy: Option, +) -> Result { + use tungstenite::client::IntoClientRequest; + + // 业务头不包含 RFC 6455 握手字段，由底层客户端生成本次连接的随机 key + let request = endpoint.as_str().into_client_request()?; + for (name, value) in request.headers() { + if !business_headers + .iter() + .any(|(header, _)| header.eq_ignore_ascii_case(name.as_str())) + { + business_headers.push(( + name.to_string(), + String::from_utf8_lossy(value.as_bytes()).into_owned(), + )); + } + } + let connection = CodexWebSocketConnection { + endpoint, + headers: business_headers, + outbound_proxy, + connection_budget: None, + }; + let (stream, response) = connect_sideband_websocket(&connection).await?; + let subprotocol = response + .headers() + .get("sec-websocket-protocol") + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + Ok(CodexLiveSideband { + stream, + subprotocol, + }) +} + +/// 把上游原始 WebSocket 流拆装为 Core 抽象中继。 +pub(crate) fn into_live_relay(stream: RawWsStream, subprotocol: Option) -> LiveRelay { + let (sink, receiver) = stream.split(); + LiveRelay::new( + subprotocol, + Box::new(TungsteniteLiveStream { inner: receiver }), + Box::new(TungsteniteLiveSink { inner: sink }), + ) +} + +fn live_error(kind: LiveGatewayErrorKind, error: impl std::fmt::Display) -> LiveGatewayError { + LiveGatewayError::new(kind, format!("codex live sideband: {error}")) +} + +fn frame_from_message(message: Message) -> Option { + match message { + Message::Text(text) => Some(LiveFrame::Text(Bytes::copy_from_slice(text.as_bytes()))), + Message::Binary(binary) => Some(LiveFrame::Binary(binary)), + Message::Ping(payload) => Some(LiveFrame::Ping(payload)), + Message::Pong(payload) => Some(LiveFrame::Pong(payload)), + Message::Close(close) => Some(LiveFrame::Close(close.map(|frame| LiveClose { + code: u16::from(frame.code), + reason: frame.reason.as_str().to_owned(), + }))), + // 帧级响应帧由 tungstenite 自动应答，不进入中继。 + Message::Frame(_) => None, + } +} + +fn message_from_frame(frame: LiveFrame) -> Message { + match frame { + LiveFrame::Text(payload) => { + Message::Text(String::from_utf8_lossy(&payload).into_owned().into()) + } + LiveFrame::Binary(payload) => Message::Binary(payload), + LiveFrame::Ping(payload) => Message::Ping(payload), + LiveFrame::Pong(payload) => Message::Pong(payload), + LiveFrame::Close(close) => Message::Close(close.map(|close| { + tokio_tungstenite::tungstenite::protocol::CloseFrame { + code: close.code.into(), + reason: close.reason.into(), + } + })), + } +} + +struct TungsteniteLiveStream { + inner: futures::stream::SplitStream, +} + +impl LiveRelayStream for TungsteniteLiveStream { + fn next_frame(&mut self) -> futures::future::BoxFuture<'_, Option> { + Box::pin(async move { + loop { + match self.inner.next().await { + Some(Ok(message)) => { + if let Some(frame) = frame_from_message(message) { + return Some(frame); + } + } + // 上游关闭或传输失败统一折叠为流结束；关闭码投影由协议层完成。 + Some(Err(_)) | None => return None, + } + } + }) + } +} + +struct TungsteniteLiveSink { + inner: futures::stream::SplitSink, +} + +impl LiveRelaySink for TungsteniteLiveSink { + fn send_frame( + &mut self, + frame: LiveFrame, + ) -> futures::future::BoxFuture<'_, Result<(), LiveGatewayError>> { + Box::pin(async move { + self.inner + .send(message_from_frame(frame)) + .await + .map_err(|error| live_error(LiveGatewayErrorKind::UpstreamUnavailable, error)) + }) + } + + fn close(&mut self, close: Option) -> futures::future::BoxFuture<'_, ()> { + Box::pin(async move { + let _ = self + .inner + .send(message_from_frame(LiveFrame::Close(close))) + .await; + let _ = self.inner.close().await; + }) + } +} diff --git a/backend/crates/providers/openai/src/transport/websocket/mod.rs b/backend/crates/providers/openai/src/transport/websocket/mod.rs index 3471cd654..9455f4da9 100644 --- a/backend/crates/providers/openai/src/transport/websocket/mod.rs +++ b/backend/crates/providers/openai/src/transport/websocket/mod.rs @@ -6,6 +6,7 @@ mod coordinator; mod error; mod exchange; mod handshake; +mod live; mod model; mod pool; mod pump; @@ -42,5 +43,6 @@ pub(crate) use self::{ prepare_response_create_request_with_pool, }, error::WEBSOCKET_CLOSE_MESSAGE_TOO_BIG, + live::{connect_live_sideband, into_live_relay}, pool::DEFAULT_STREAM_IDLE_TIMEOUT, }; diff --git a/backend/crates/providers/openai/tests/admin.rs b/backend/crates/providers/openai/tests/admin.rs index 43772bd0a..3a718b25d 100644 --- a/backend/crates/providers/openai/tests/admin.rs +++ b/backend/crates/providers/openai/tests/admin.rs @@ -1670,6 +1670,21 @@ fn initialized_account_scope(account_id: &str) -> Arc { )) } +pub(crate) async fn initialized_test_provider( + accounts: Arc, + base_url: String, +) -> Arc { + let mut config = valid_config(); + config.config.api.base_url = base_url; + provider_openai::initialize( + config.config.clone(), + provider_ports_with(accounts, Arc::new(TestOAuthPending::default())), + ) + .await + .expect("initialized provider") + .core_provider() +} + fn provider_ports() -> ProviderStorePorts { provider_ports_with( Arc::new(MemoryAccountStore::default()), diff --git a/backend/crates/providers/openai/tests/provider/contract/mod.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs index c3414e9cd..c990dfc84 100644 --- a/backend/crates/providers/openai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/openai/tests/provider/contract/mod.rs @@ -1022,7 +1022,10 @@ fn provider_with_affinity( provider_with_affinity_and_base_url(store, session_affinity, OFFICIAL_CODEX_BASE_URL.to_owned()) } -fn provider_with_base_url(store: &Arc, base_url: String) -> Arc { +pub(super) fn provider_with_base_url( + store: &Arc, + base_url: String, +) -> Arc { provider_with_base_url_and_retry_budget( store, base_url, @@ -1167,7 +1170,7 @@ fn provider_and_quota_with_runtime_ports( (Arc::new(provider), quota, websocket_pool) } -async fn create_account(store: &Arc, id: &str) { +pub(super) async fn create_account(store: &Arc, id: &str) { create_account_with_enabled(store, id, true).await; } @@ -1474,7 +1477,7 @@ impl ExtensionSetLease for TestExtensionLease { } } -fn context_with_middleware( +pub(super) fn context_with_middleware( request_id: &str, plan: Arc, disable_fast: bool, @@ -1509,7 +1512,7 @@ fn context_with_middleware( ) } -fn context(request_id: &str, cancellation: CancellationToken) -> AttemptContext { +pub(super) fn context(request_id: &str, cancellation: CancellationToken) -> AttemptContext { context_with_fast_policy(request_id, cancellation, false) } diff --git a/backend/crates/providers/openai/tests/provider/live.rs b/backend/crates/providers/openai/tests/provider/live.rs new file mode 100644 index 000000000..65779261a --- /dev/null +++ b/backend/crates/providers/openai/tests/provider/live.rs @@ -0,0 +1,234 @@ +//! Live 引导模型事实与固定账号重连授权回归 + +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::Arc; + +use bytes::Bytes; +use futures::{StreamExt, future::BoxFuture}; +use gateway_core::account::{ + AccountModelAccess, AccountModelAccessMode, ProviderAccountId, ProviderAccountStore, +}; +use gateway_core::engine::middleware::{ + MiddlewareContext, MiddlewareError, MiddlewareNext, MiddlewarePlan, MiddlewareRequest, + MiddlewareResponse, +}; +use gateway_core::engine::provider::{Provider, ProviderRequest}; +use gateway_core::lifecycle::CancellationToken; +use gateway_core::live::{LiveGatewayErrorKind, LiveSidebandRequest, LiveSidebandStyle}; +use gateway_core::operation::{Operation, ProviderHttpMethod, ProviderHttpRequest, RawHttpPayload}; +use gateway_core::policy::ClientApiKeyId; +use gateway_core::routing::{ + ClientRoutingScope, ConfigRevision, FrozenAccountScope, ProviderKind, RoutingContext, + RuntimeAccount, RuntimeAccountDirectory, RuntimeSnapshot, UpstreamModelId, +}; +use serde_json::json; +use wiremock::{ + Mock, MockServer, ResponseTemplate, + matchers::{method, path}, +}; + +use super::contract::{context, context_with_middleware, create_account, provider_with_base_url}; +use crate::support::MemoryAccountStore; + +const MODEL: &str = "gpt-live-1-codex"; +const ACCOUNT: &str = "acct_provider_contract"; +const CALL: &str = "call_live_regression"; + +fn scope(model_access: AccountModelAccess) -> Arc { + Arc::new(FrozenAccountScope::new( + Arc::new(RuntimeAccountDirectory::new(BTreeMap::from([( + ProviderAccountId::new(ACCOUNT).unwrap(), + RuntimeAccount::new(ProviderKind::new("openai").unwrap(), BTreeSet::new()) + .with_model_access(model_access), + )]))), + ClientRoutingScope::all_accounts(), + )) +} + +fn live_request() -> ProviderRequest { + let provider = ProviderKind::new("openai").unwrap(); + let model = UpstreamModelId::new(MODEL).unwrap(); + let operation = Operation::ProviderHttp( + ProviderHttpRequest::new( + "realtime-calls", + ProviderHttpMethod::Post, + Some("intent=quicksilver&architecture=avas".into()), + Vec::new(), + RawHttpPayload::new( + "openai", + Bytes::from(json!({"sdp":"v=0", "session":{"model":MODEL}}).to_string()), + ) + .unwrap(), + ) + .unwrap(), + ); + let snapshot = RuntimeSnapshot::new( + ConfigRevision::new(1).unwrap(), + gateway_core::settings::SettingsValues::new(2, 10, "smart", BTreeMap::new(), None, None), + vec![provider.clone()], + Vec::new(), + Vec::new(), + ) + .unwrap(); + let plan = snapshot + .plan_provider_endpoint( + &provider, + Some(&model), + &operation, + scope(AccountModelAccess::default()), + &RoutingContext::default(), + ) + .unwrap(); + ProviderRequest::new(operation, plan.candidates()[0].clone()) +} + +#[derive(Debug)] +struct ModelMiddleware; +impl MiddlewarePlan for ModelMiddleware { + fn handle( + &self, + context: MiddlewareContext, + request: MiddlewareRequest, + next: MiddlewareNext, + ) -> BoxFuture<'static, Result> { + assert_eq!(context.model(), Some(MODEL)); + next.run(request) + } +} + +async fn bootstrap() -> (Arc, Arc, MockServer) { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, ACCOUNT).await; + let upstream = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/realtime/calls")) + .respond_with( + ResponseTemplate::new(201) + .insert_header("location", format!("/v1/live/{CALL}")) + .insert_header("content-type", "application/sdp") + .set_body_string("v=0\r\n"), + ) + .expect(1) + .mount(&upstream) + .await; + let provider = crate::admin::initialized_test_provider(store.clone(), upstream.uri()).await; + let request = live_request(); + let candidate = request.candidate().clone(); + let mut stream = provider + .clone() + .execute( + request, + context_with_middleware("req_live_regression", Arc::new(ModelMiddleware), false), + ) + .await + .expect("live stream"); + assert!( + stream.metadata().confirms(&candidate), + "Core must accept the frozen provider/model facts" + ); + assert!( + upstream.received_requests().await.unwrap().is_empty(), + "bootstrap must remain cold before Core accepts metadata" + ); + while let Some(event) = stream.next().await { + event.expect("bootstrap response"); + } + (provider, store, upstream) +} + +#[tokio::test] +async fn live_bootstrap_preserves_model_through_middleware_and_metadata() { + let (_provider, _store, upstream) = bootstrap().await; + upstream.verify().await; +} + +#[tokio::test] +async fn live_sideband_rejects_revoked_scope_and_model_without_claiming_call() { + let (provider, _store, _upstream) = bootstrap().await; + let gateway = provider.live_gateway().unwrap(); + let denied_model = scope( + AccountModelAccess::new(AccountModelAccessMode::Denylist, vec![MODEL.to_owned()]).unwrap(), + ); + let absent_account = Arc::new(FrozenAccountScope::new( + Arc::new(RuntimeAccountDirectory::new(BTreeMap::new())), + ClientRoutingScope::all_accounts(), + )); + let key = ClientApiKeyId::new("key_openai_contract").unwrap(); + // 同一 call 连续拒绝不能遗留 claim，后续请求仍应得到权限错误而不是 busy + for account_scope in [&absent_account, &denied_model, &absent_account] { + let error = gateway + .open_sideband(LiveSidebandRequest { + call_id: CALL, + client_api_key_id: &key, + account_scope, + style: LiveSidebandStyle::Live, + protocol_headers: Vec::new(), + subprotocols: Vec::new(), + }) + .await + .unwrap_err(); + assert_eq!(error.kind(), LiveGatewayErrorKind::OwnerMismatch); + assert_eq!(error.status(), 403); + } +} + +#[tokio::test] +async fn live_sideband_rejects_disabled_account_even_before_scope_refresh() { + let (provider, store, _upstream) = bootstrap().await; + store + .set_enabled(&ProviderAccountId::new(ACCOUNT).unwrap(), false) + .await + .unwrap(); + let gateway = provider.live_gateway().unwrap(); + let key = ClientApiKeyId::new("key_openai_contract").unwrap(); + let allowed = scope(AccountModelAccess::default()); + for _ in 0..2 { + let error = gateway + .open_sideband(LiveSidebandRequest { + call_id: CALL, + client_api_key_id: &key, + account_scope: &allowed, + style: LiveSidebandStyle::Live, + protocol_headers: Vec::new(), + subprotocols: Vec::new(), + }) + .await + .unwrap_err(); + assert_eq!(error.status(), 403); + } +} + +#[tokio::test] +async fn live_bootstrap_rejects_missing_planned_model_before_upstream_send() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, ACCOUNT).await; + let upstream = MockServer::start().await; + let provider = provider_with_base_url(&store, upstream.uri()); + let request = live_request(); + let provider_kind = ProviderKind::new("openai").unwrap(); + let snapshot = RuntimeSnapshot::new( + ConfigRevision::new(1).unwrap(), + gateway_core::settings::SettingsValues::new(2, 10, "smart", BTreeMap::new(), None, None), + vec![provider_kind.clone()], + Vec::new(), + Vec::new(), + ) + .unwrap(); + let plan = snapshot + .plan_provider_endpoint( + &provider_kind, + None, + request.operation(), + scope(AccountModelAccess::default()), + &RoutingContext::default(), + ) + .unwrap(); + let result = provider + .execute( + ProviderRequest::new(request.operation().clone(), plan.candidates()[0].clone()), + context("req_live_missing_model", CancellationToken::new()), + ) + .await; + assert!(result.is_err()); + assert!(upstream.received_requests().await.unwrap().is_empty()); +} diff --git a/backend/crates/providers/openai/tests/provider/mod.rs b/backend/crates/providers/openai/tests/provider/mod.rs index 67bf9158f..f87e8cfa8 100644 --- a/backend/crates/providers/openai/tests/provider/mod.rs +++ b/backend/crates/providers/openai/tests/provider/mod.rs @@ -2,5 +2,6 @@ mod contract; mod failure; +mod live; pub(crate) use contract::assert_local_connection_capacity_is_not_an_upstream_failure; diff --git a/docs/api.md b/docs/api.md index 1346c9762..6b5fb47d2 100644 --- a/docs/api.md +++ b/docs/api.md @@ -456,6 +456,45 @@ OpenAI 选号阶段确认本次可选账号全部额度耗尽时，HTTP 返回 ` 缺失、非法、已禁用或已删除的 Key 返回 OpenAI 风格 `401` 错误；未知查询参数返回 `400 invalid_usage_query`， 读取账本失败返回 `503 usage_unavailable`，不会用零余额掩盖故障 +### Codex Live 语音通话 + +语音通话复用数据面 Client Key 鉴权，并按该 Key 的账号分组选路。通话引导是一次性请求，走与其他 +`/v1/*` 数据面请求相同的调度、准入与请求记录；通话建立后的 sideband 与 hangup 使用引导时选中的 +账号（call id 与账号、创建 Key 绑定），音频流不经过网关。 + +| 路由 | 说明 | +| --- | --- | +| `POST /v1/live` | WebRTC SDP 引导；`POST /v1/realtime` 与 `POST /v1/realtime/calls` 为同一处理的别名，成功响应的 `Location` 会改写为 `/v1/realtime/calls/{call_id}` | +| `GET /v1/live/{call_id}` | 通话 sideband WebSocket，双向透传事件帧 | +| `GET /v1/realtime/calls/{call_id}` | 同上的 realtime 路径形态 | +| `GET /v1/realtime?call_id=…` | 同上的查询参数形态 | +| `POST /v1/realtime/calls/{call_id}/hangup` | 用钉住账号转发挂断，成功后解除绑定 | + +引导请求接受三种入口形态并统一为上游要求的 `{"sdp": …, "session"?: …}` JSON：JSON 保留原有字段、 +`application/sdp` / `text/plain` 包装为 `sdp` 字段、`multipart/form-data` 取 `sdp` 与 `session` 两个部分。 +`session.model` 或顶层 `model` 中的 `gpt-realtime` 系模型名归一为 `gpt-live-1-codex`，其余模型原样透传； +模型字段缺失时按默认语音模型参与检查。归一后的模型参与账号模型权限检查：账号范围整体禁止该模型时 +引导请求按路由错误拒绝，被禁账号不会服务语音通话。语音语义请求头按允许清单转发：`OpenAI-Alpha`、 +`X-Session-Id`、`Session-Id`、`Thread-Id`、`Originator`、`OpenAI-Safety-Identifier`、`OpenAI-Organization`、 +`OpenAI-Project`、`X-Oai-Attestation`；认证、账号身份等其余请求头一律不透传，上游账号身份由选中的账号提供。 + +引导响应透传上游状态码，并只回传 `Content-Type`、`Location`、`Retry-After`、`X-Request-Id`、 +`OpenAI-Request-Id` 五个响应头；正文按字节透传。sideband 中继不做协议解释：Text/Binary 帧双向透传， +Ping 由网关本地应答，关闭帧保留原关闭码，异常中断投影为正常关闭。同一 call id 只允许一条 sideband，重复加入返回 `409`； +非创建该通话的 Client Key 访问 sideband 或 hangup 返回 `403`；call id 未知或已过期返回 `404`。 +hangup 不受 sideband 占用限制，通话中可随时挂断；挂断与 sideband 都走钉住账号的出口代理。 +call 与账号的绑定保存在网关进程内，未连接时一小时过期；sideband 占用期间暂停过期，断开后重新计时， +客户端可重连加入同一通话。挂断成功或上游报告通话不存在时立即解除绑定。 +进程重启后未完成通话失去 sideband 与 hangup 能力 + +限制： + +- 引导与 sideband 只使用 OAuth 凭据账号；API Key 类账号不在语音调度范围。 +- 引导按当前账号范围、模型政策、额度与冷却状态选取账号；sideband 固定使用创建通话的账号，每次连接重新检查当前授权范围、模型政策与账号启用状态，失权时返回 `403`，不会切换账号 +- 已连接的 sideband 不随权限变更自动断开；仍有效的创建方 Client Key 可调用 hangup 结束通话 +- 标准实时 WebSocket（`GET /v1/realtime?model=…`）、realtime client secrets、legacy sessions、 + transcription/translation 会话与 SIP 控制返回 `501 realtime_capability_not_supported`。 + ## 4. 浏览器认证 ### 统一登录与会话