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 <noreply@openai.com>
This commit is contained in:
2han9wen71an
2026-10-05 23:48:55 +08:00
committed by GitHub
co-authored by Codex
parent 78ba7e412e
commit 22acbd1d10
37 changed files with 3563 additions and 40 deletions
+18
View File
@@ -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"
+1 -1
View File
@@ -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"] }
@@ -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
{
@@ -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<ApiState>,
connect_info: Option<Extension<ConnectInfo<SocketAddr>>>,
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<ApiState>,
axum::extract::Path(call_id): axum::extract::Path<String>,
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<Value>)>), Box<Response>> {
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::<Value>(&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<ProviderEvent>) -> Option<Bytes> {
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()
}
@@ -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<crate::ApiState> {
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<ProviderHttpHeader> {
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<String> {
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<ProviderHttpHeader>,
) -> Result<ProviderHttpRequest, Box<Response>> {
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<String> {
let payload = serde_json::from_slice::<Value>(body).ok()?;
let extract = |value: Option<&Value>| -> Option<String> {
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>) -> 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::<serde_json::Map<String, Value>>(&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<Value>,
) -> Result<Bytes, Box<Response>> {
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<Value>)>,
) -> Result<(Bytes, Option<String>, Option<String>), Box<Response>> {
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))
}
}
}
@@ -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<ApiState>,
axum::extract::Path(call_id): axum::extract::Path<String>,
uri: Uri,
headers: HeaderMap,
upgrade: Result<WebSocketUpgrade, WebSocketUpgradeRejection>,
) -> 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<ApiState>,
uri: Uri,
headers: HeaderMap,
upgrade: Result<WebSocketUpgrade, WebSocketUpgradeRejection>,
) -> 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<WebSocketUpgrade, WebSocketUpgradeRejection>,
) -> 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<dyn ConnectionGuard>,
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<dyn ConnectionGuard>,
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;
}
@@ -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;
@@ -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<ApiState> {
// 官方 OpenAI 模型详情合同使用 path ID;它不属于 Admin API 约束
.route(MODEL_DETAIL_PATH, get(model_detail))
.merge(usage::router())
.merge(live::router())
// OpenAI 数据面正文属于客户端/上游协议;代理不能用私有大小上限提前拒绝
// 上游本可接受的未来 payload
.layer(DefaultBodyLimit::disable())
@@ -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<UpstreamModelId>,
client_ip: Option<IpAddr>,
user_agent: Option<String>,
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<Arc<dyn gateway_core::live::LiveGateway>> {
self.execution.live_gateway()
}
pub(crate) fn try_register_connection(
&self,
) -> Result<Box<dyn ConnectionGuard>, ConnectionDraining> {
@@ -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",
@@ -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<Mutex<Option<StartProviderExecution>>>,
gateway: Option<Arc<dyn gateway_core::live::LiveGateway>>,
}
impl LiveCapture {
fn new() -> Arc<Self> {
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<Arc<dyn gateway_core::live::LiveGateway>> {
self.gateway.clone()
}
fn authenticate(
&self,
plaintext: &str,
) -> Result<AuthenticatedClient, ClientAuthenticationError> {
if plaintext == LIVE_KEY {
Ok(authenticated_client(LIVE_KEY))
} else {
Err(ClientAuthenticationError::InvalidKey)
}
}
fn public_models(&self, _: &AuthenticatedClient) -> Vec<gateway_core::routing::PublicModelId> {
Vec::new()
}
fn contains_public_model(
&self,
_: &AuthenticatedClient,
_: &gateway_core::routing::PublicModelId,
) -> bool {
false
}
fn start(&self, _: StartExecution) -> BoxFuture<'_, Result<StartedExecution, GatewayError>> {
Box::pin(async {
Err(GatewayError::new(
GatewayErrorKind::Internal,
"live test must not start a response",
))
})
}
fn start_provider_endpoint(
&self,
request: StartProviderExecution,
) -> BoxFuture<'_, Result<StartedExecution, GatewayError>> {
*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<dyn ExecutionService>)
.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::<Vec<_>>();
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::<Vec<_>>();
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);
}
@@ -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<Trace>);
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<Trace>);
impl LiveRelayGuard for Claim {}
impl Drop for Claim {
fn drop(&mut self) {
self.0.claims.fetch_sub(1, Ordering::SeqCst);
}
}
struct Lifecycle(Arc<Trace>);
impl ConnectionLifecycle for Lifecycle {
fn try_register(&self) -> Result<Box<dyn ConnectionGuard>, 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<LiveFrame>> {
Box::pin(std::future::pending())
}
}
struct PendingSink(Arc<Trace>);
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<LiveClose>) -> BoxFuture<'_, ()> {
Box::pin(async move {
if self.0.block_close {
self.0.entered.notify_one();
std::future::pending().await
}
})
}
}
struct Gateway(Arc<Trace>);
impl LiveGateway for Gateway {
fn open_sideband<'a>(
&'a self,
request: LiveSidebandRequest<'a>,
) -> BoxFuture<'a, Result<LiveRelay, LiveGatewayError>> {
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<LiveCallOutcome, LiveGatewayError>> {
unreachable!("no hangup in relay lifecycle test")
}
}
async fn blocked_relay(
trace: &Arc<Trace>,
) -> (
tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>,
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();
}
@@ -4,6 +4,7 @@ mod auth;
mod endpoint;
mod error;
mod images;
mod live;
mod middleware;
mod models;
mod responses;
@@ -710,6 +710,12 @@ pub trait ExecutionService: Send + Sync {
&self,
request: StartProviderExecution,
) -> BoxFuture<'_, Result<StartedExecution, GatewayError>>;
/// 返回 Provider 注册表中首个声明 Live sideband 能力的网关。
/// `None` 表示当前组合不含语音 sideband;协议 adapter 据此回退到稳定 501。
fn live_gateway(&self) -> Option<Arc<dyn crate::live::LiveGateway>> {
None
}
}
/// 只验证 Client API Key 并返回稳定 Key ID,不产生请求使用事实
@@ -2372,6 +2378,12 @@ impl ExecutionService for DefaultExecutionService {
) -> BoxFuture<'_, Result<StartedExecution, GatewayError>> {
Box::pin(async move { self.start_provider_endpoint_inner(request).await })
}
fn live_gateway(&self) -> Option<Arc<dyn crate::live::LiveGateway>> {
self.providers
.iter()
.find_map(|provider| provider.live_gateway())
}
}
impl NestedModelExecutionPort for DefaultExecutionService {
@@ -1018,6 +1018,12 @@ pub trait Provider: Send + Sync {
request: ProviderRequest,
context: AttemptContext,
) -> Result<ProviderStream, ProviderError>;
/// 返回 Provider 的 Live sideband 能力;不支持的 Provider 返回 `None`。
/// 实现必须只暴露已钉住账号的受限拨号,不得泄露凭据或放宽账号范围。
fn live_gateway(&self) -> Option<Arc<dyn crate::live::LiveGateway>> {
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<Item = &Arc<dyn Provider>> {
self.providers.values()
}
}
impl ProviderCatalogPort for ProviderRegistry {
+1
View File
@@ -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;
+327
View File
@@ -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<LiveClose>),
}
/// 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<String>,
}
/// 一次性账号级上游调用(hangup)请求。
#[derive(Debug)]
pub struct LiveHangupRequest<'a> {
pub call_id: &'a str,
pub client_api_key_id: &'a ClientApiKeyId,
pub content_type: Option<String>,
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<String>,
receiver: Box<dyn LiveRelayStream>,
sender: Box<dyn LiveRelaySink>,
guard: Option<Box<dyn LiveRelayGuard>>,
}
/// 随 [`LiveRelay`] 存活的会话占用标记;丢弃时由实现方归还资源。
pub trait LiveRelayGuard: Send {}
impl LiveRelay {
/// 由 Provider 实现装配;Core 不感知底层 WebSocket 栈。
#[must_use]
pub fn new(
subprotocol: Option<String>,
receiver: Box<dyn LiveRelayStream>,
sender: Box<dyn LiveRelaySink>,
) -> Self {
Self {
subprotocol,
receiver,
sender,
guard: None,
}
}
/// 附加随中继释放的会话 guard;重复调用以最后一次为准。
pub fn with_guard(&mut self, guard: Box<dyn LiveRelayGuard>) {
self.guard = Some(guard);
}
/// 读取上游下一帧;返回 `None` 表示上游已关闭。
pub fn next_frame(&mut self) -> BoxFuture<'_, Option<LiveFrame>> {
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<LiveClose>) -> 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<LiveFrame>>;
}
/// 上游侧的帧发送端。
pub trait LiveRelaySink: Send {
fn send_frame(&mut self, frame: LiveFrame) -> BoxFuture<'_, Result<(), LiveGatewayError>>;
fn close(&mut self, close: Option<LiveClose>) -> 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<u16>,
body: Option<Bytes>,
}
impl LiveGatewayError {
#[must_use]
pub fn new(kind: LiveGatewayErrorKind, message: impl Into<String>) -> Self {
Self {
kind,
message: message.into(),
status: None,
body: None,
}
}
#[must_use]
pub fn with_upstream(mut self, status: u16, body: Option<Bytes>) -> 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<LiveRelay, LiveGatewayError>>;
/// 以钉住账号转发一次性调用(当前仅 hangup)。
fn hangup<'a>(
&'a self,
request: LiveHangupRequest<'a>,
) -> BoxFuture<'a, Result<LiveCallOutcome, LiveGatewayError>>;
}
/// 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<String> {
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::<Vec<_>>();
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
}
@@ -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
+45
View File
@@ -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);
}
@@ -8,6 +8,7 @@ mod error;
mod event;
mod health;
mod lifecycle;
mod live;
mod metering;
mod middleware;
mod operation;
@@ -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]
@@ -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<CodexCredentialLease, CredentialSelectionError> {
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<CodexCredentialLease, CredentialSelectionError> {
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<CodexCredentialLease, CredentialSelectionError> {
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 =
+2 -1
View File
@@ -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(
@@ -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)?;
@@ -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<HashMap<String, LiveRegistryEntry>>,
}
impl CodexLiveRegistry {
fn sweep_expired(entries: &mut HashMap<String, LiveRegistryEntry>, 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<gateway_core::account::ProviderAccountId, LiveClaimError> {
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<gateway_core::account::ProviderAccountId, LiveClaimError> {
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<CodexLiveRegistry>,
call_id: String,
}
impl LiveCallClaim {
fn new(registry: Arc<CodexLiveRegistry>, 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<Self>,
request: ProviderHttpRequest,
upstream_model: Option<&UpstreamModelId>,
context: AttemptContext,
) -> Result<ProviderStream, ProviderError> {
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<Self>,
context: AttemptContext,
operation: Operation,
upstream_model: UpstreamModelId,
middleware_headers: Vec<MiddlewareHeader>,
lease: Arc<CodexCredentialLease>,
account_selection_wait_ms: u64,
) -> Result<ProviderStream, ProviderError> {
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::<Vec<_>>();
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<CodexLiveRegistry>,
response_origin: Url,
endpoint_query: Option<String>,
content_type: Option<String>,
protocol_headers: Vec<(String, String)>,
upstream_model: UpstreamModelId,
body: Bytes,
context: AttemptContext,
selector: Arc<CodexCredentialSelector>,
quota: Arc<CodexCredentialQuotaService>,
lease: Arc<CodexCredentialLease>,
}
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<CodexLiveRegistry>,
repository: CodexCredentialRepository,
client: CodexBackendClient,
}
impl CodexLiveGateway {
pub(crate) fn new(
registry: Arc<CodexLiveRegistry>,
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<LiveRelay, LiveGatewayError>> {
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<LiveCallOutcome, LiveGatewayError>> {
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)
)
}
@@ -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<CodexSessionIdentity>,
session_transport_recovery: CodexSessionTransportRecovery,
stream_max_retries: u32,
live_registry: Arc<CodexLiveRegistry>,
live_gateway: Option<Arc<CodexLiveGateway>>,
}
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<Arc<dyn gateway_core::live::LiveGateway>> {
self.live_gateway
.clone()
.map(|gateway| gateway as Arc<dyn gateway_core::live::LiveGateway>)
}
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,
@@ -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<CodexLiveCallResponse> {
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<CodexLiveCallResponse> {
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<CodexLiveCallResponse> {
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<String>,
/// 上游 Content-Type。
pub content_type: Option<String>,
pub retry_after_seconds: Option<u64>,
/// 可回传客户端的响应头子集。
pub forwarded_headers: Vec<(String, String)>,
pub set_cookie_headers: Vec<String>,
pub rate_limit_headers: Vec<(String, String)>,
pub diagnostics: CodexUpstreamDiagnostics,
pub transport_metrics: CodexTransportMetrics,
}
@@ -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`
@@ -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::{
@@ -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<Option<Vec<u8>>>), CodexWebSocketExchangeError> {
connect_websocket(connection, false).await
}
pub(super) async fn connect_pumped_websocket(
connection: &CodexWebSocketConnection,
keepalive: PumpKeepalive,
@@ -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<String>,
}
/// 以给定 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<gateway_core::account::OutboundProxy>,
) -> Result<CodexLiveSideband, CodexWebSocketExchangeError> {
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<String>) -> 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<LiveFrame> {
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<RawWsStream>,
}
impl LiveRelayStream for TungsteniteLiveStream {
fn next_frame(&mut self) -> futures::future::BoxFuture<'_, Option<LiveFrame>> {
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<RawWsStream, Message>,
}
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<LiveClose>) -> futures::future::BoxFuture<'_, ()> {
Box::pin(async move {
let _ = self
.inner
.send(message_from_frame(LiveFrame::Close(close)))
.await;
let _ = self.inner.close().await;
})
}
}
@@ -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,
};
@@ -1670,6 +1670,21 @@ fn initialized_account_scope(account_id: &str) -> Arc<FrozenAccountScope> {
))
}
pub(crate) async fn initialized_test_provider(
accounts: Arc<MemoryAccountStore>,
base_url: String,
) -> Arc<dyn gateway_core::engine::provider::Provider> {
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()),
@@ -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<MemoryAccountStore>, base_url: String) -> Arc<CodexProvider> {
pub(super) fn provider_with_base_url(
store: &Arc<MemoryAccountStore>,
base_url: String,
) -> Arc<CodexProvider> {
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<MemoryAccountStore>, id: &str) {
pub(super) async fn create_account(store: &Arc<MemoryAccountStore>, 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<dyn MiddlewarePlan>,
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)
}
@@ -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<FrozenAccountScope> {
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<MiddlewareResponse, MiddlewareError>> {
assert_eq!(context.model(), Some(MODEL));
next.run(request)
}
}
async fn bootstrap() -> (Arc<dyn Provider>, Arc<MemoryAccountStore>, 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());
}
@@ -2,5 +2,6 @@
mod contract;
mod failure;
mod live;
pub(crate) use contract::assert_local_connection_capacity_is_not_an_upstream_failure;
+39
View File
@@ -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. 浏览器认证
### 统一登录与会话