use crate::auth::SharedAuthProvider;use crate::endpoint::realtime_websocket::RealtimeSessionConfig;use crate::endpoint::realtime_websocket::session_update_session_json;use crate::endpoint::session::EndpointSession;use crate::error::ApiError;use crate::provider::Provider;use bytes::Bytes;use codex_client::HttpTransport;use codex_client::Request;use codex_client::RequestBody;use codex_client::RequestTelemetry;use codex_protocol::protocol::RealtimeConversationArchitecture;use http::HeaderMap;use http::HeaderValue;use http::Method;use http::header::CONTENT_TYPE;use http::header::LOCATION;use serde::Serialize;use serde_json::Value;use serde_json::to_string;use serde_json::to_value;use std::sync::Arc;use tracing::instrument;use tracing::trace;const MULTIPART_BOUNDARY: &str = "codex-realtime-call-boundary";const MULTIPART_CONTENT_TYPE: &str = "multipart/form-data; boundary=codex-realtime-call-boundary";pub struct RealtimeCallClient<T: HttpTransport> { session: EndpointSession<T>,}/// Answer from creating a WebRTC Realtime call.////// `sdp` configures the peer connection. `call_id` is parsed from the response `Location` header/// and is later used by the server-side sideband WebSocket to join this exact call.#[derive(Debug, Clone, PartialEq, Eq)]pub struct RealtimeCallResponse { pub sdp: String, pub call_id: String,}#[derive(Serialize)]struct BackendRealtimeCallRequest<'a> { sdp: &'a str, session: &'a Value,}impl<T: HttpTransport> RealtimeCallClient<T> { pub fn new(transport: T, provider: Provider, auth: SharedAuthProvider) -> Self { Self { session: EndpointSession::new(transport, provider, auth), } } pub fn with_telemetry(self, request: Option<Arc<dyn RequestTelemetry>>) -> Self { Self { session: self.session.with_request_telemetry(request), } } fn path() -> &'static str { "realtime/calls" } fn uses_backend_request_shape(&self) -> bool { self.session.provider().base_url.contains("/backend-api") } #[instrument( name = "realtime_call.create", level = "info", skip_all, fields( http.method = "POST", api.path = "realtime/calls" ) )] pub async fn create(&self, sdp: String) -> Result<RealtimeCallResponse, ApiError> { self.create_with_headers(sdp, HeaderMap::new()).await } pub async fn create_with_session( &self, sdp: String, session_config: RealtimeSessionConfig, ) -> Result<RealtimeCallResponse, ApiError> { self.create_with_session_and_headers(sdp, session_config, HeaderMap::new()) .await } pub async fn create_with_headers( &self, sdp: String, extra_headers: HeaderMap, ) -> Result<RealtimeCallResponse, ApiError> { let resp = self .session .execute_with( Method::POST, Self::path(), extra_headers, /*body*/ None, |req| { req.headers .insert(CONTENT_TYPE, HeaderValue::from_static("application/sdp")); req.body = Some(RequestBody::Raw(Bytes::from(sdp.clone()))); }, ) .await?; let sdp = decode_sdp_response(resp.body.as_ref())?; let call_id = decode_call_id_from_location(&resp.headers)?; Ok(RealtimeCallResponse { sdp, call_id }) } pub async fn create_with_session_and_headers( &self, sdp: String, session_config: RealtimeSessionConfig, extra_headers: HeaderMap, ) -> Result<RealtimeCallResponse, ApiError> { self.create_with_session_architecture_and_headers( sdp, session_config, RealtimeConversationArchitecture::RealtimeApi, extra_headers, ) .await } pub async fn create_with_session_architecture_and_headers( &self, sdp: String, session_config: RealtimeSessionConfig, architecture: RealtimeConversationArchitecture, extra_headers: HeaderMap, ) -> Result<RealtimeCallResponse, ApiError> { trace!(target: "codex_api::realtime_websocket::wire", "realtime call request SDP: {sdp}"); // WebRTC can begin inference as soon as the peer connection comes up, so the initial // session payload is sent with call creation. The sideband WebSocket still sends its normal // session.update after it joins. let mut session = realtime_session_json(session_config)?; if let Some(session) = session.as_object_mut() { session.remove("id"); } // TODO(aibrahim): Align the SIWC route with the API multipart shape and remove this branch. if self.uses_backend_request_shape() { let body = to_value(BackendRealtimeCallRequest { sdp: &sdp, session: &session, }) .map_err(|err| ApiError::Stream(format!("failed to encode realtime call: {err}")))?; let resp = self .session .execute_with( Method::POST, Self::path(), extra_headers, Some(body), |req| configure_realtime_call_request(req, architecture), ) .await?; let sdp = decode_sdp_response(resp.body.as_ref())?; let call_id = decode_call_id_from_location(&resp.headers)?; return Ok(RealtimeCallResponse { sdp, call_id }); } let session = to_string(&session).map_err(|err| ApiError::InvalidRequest { message: err.to_string(), })?; let mut body = Vec::new(); body.extend_from_slice(format!("--{MULTIPART_BOUNDARY}\r\n").as_bytes()); body.extend_from_slice(b"Content-Disposition: form-data; name=\"sdp\"\r\n"); body.extend_from_slice(b"Content-Type: application/sdp\r\n\r\n"); body.extend_from_slice(sdp.as_bytes()); body.extend_from_slice(b"\r\n"); body.extend_from_slice(format!("--{MULTIPART_BOUNDARY}\r\n").as_bytes()); body.extend_from_slice(b"Content-Disposition: form-data; name=\"session\"\r\n"); body.extend_from_slice(b"Content-Type: application/json\r\n\r\n"); body.extend_from_slice(session.as_bytes()); body.extend_from_slice(b"\r\n"); body.extend_from_slice(format!("--{MULTIPART_BOUNDARY}--\r\n").as_bytes()); let resp = self .session .execute_with( Method::POST, Self::path(), extra_headers, /*body*/ None, |req| { configure_realtime_call_request(req, architecture); req.headers.insert( CONTENT_TYPE, HeaderValue::from_static(MULTIPART_CONTENT_TYPE), ); req.body = Some(RequestBody::Raw(Bytes::from(body.clone()))); }, ) .await?; let sdp = decode_sdp_response(resp.body.as_ref())?; let call_id = decode_call_id_from_location(&resp.headers)?; Ok(RealtimeCallResponse { sdp, call_id }) }}fn configure_realtime_call_request( request: &mut Request, architecture: RealtimeConversationArchitecture,) { match architecture { RealtimeConversationArchitecture::RealtimeApi => {} RealtimeConversationArchitecture::Avas => { append_query_pair(&mut request.url, "intent", "quicksilver"); append_query_pair(&mut request.url, "architecture", "avas"); } }}fn append_query_pair(url: &mut String, key: &str, value: &str) { if url.contains('?') { url.push('&'); } else { url.push('?'); } url.push_str(key); url.push('='); url.push_str(value);}fn realtime_session_json(session_config: RealtimeSessionConfig) -> Result<Value, ApiError> { session_update_session_json(session_config) .map_err(|err| ApiError::Stream(format!("failed to encode realtime call session: {err}")))}fn decode_sdp_response(body: &[u8]) -> Result<String, ApiError> { String::from_utf8(body.to_vec()).map_err(|err| { ApiError::Stream(format!( "failed to decode realtime call SDP response: {err}" )) })}fn decode_call_id_from_location(headers: &HeaderMap) -> Result<String, ApiError> { let location = headers .get(LOCATION) .ok_or_else(|| ApiError::Stream("realtime call response missing Location".to_string()))? .to_str() .map_err(|err| ApiError::Stream(format!("invalid realtime call Location: {err}")))?; trace!("realtime call Location: {location}"); location .split('?') .next() .unwrap_or(location) .rsplit('/') .find(|segment| is_realtime_call_id_segment(segment)) .map(str::to_string) .ok_or_else(|| { ApiError::Stream(format!( "realtime call Location does not contain a call id: {location}" )) })}fn is_realtime_call_id_segment(segment: &str) -> bool { if segment.starts_with("rtc_") && segment.len() > "rtc_".len() { return true; } if segment.len() != 36 { return false; } segment.char_indices().all(|(index, ch)| match index { 8 | 13 | 18 | 23 => ch == '-', _ => ch.is_ascii_hexdigit(), })}#[cfg(test)]mod tests { use super::*; use crate::auth::AuthProvider; use crate::endpoint::realtime_websocket::RealtimeEventParser; use crate::endpoint::realtime_websocket::RealtimeOutputModality; use crate::endpoint::realtime_websocket::RealtimeSessionMode; use crate::provider::RetryConfig; use codex_client::Request; use codex_client::Response; use codex_client::StreamResponse; use codex_client::TransportError; use codex_protocol::protocol::RealtimeVoice; use http::StatusCode; use pretty_assertions::assert_eq; use std::sync::Mutex; use std::time::Duration; #[derive(Clone)] struct CapturingTransport { last_request: Arc<Mutex<Option<Request>>>, response_headers: HeaderMap, } impl CapturingTransport { fn new() -> Self { Self::with_location("/v1/realtime/calls/rtc_test") } fn with_location(location: &str) -> Self { let mut response_headers = HeaderMap::new(); response_headers.insert(LOCATION, HeaderValue::from_str(location).unwrap()); Self { last_request: Arc::new(Mutex::new(None)), response_headers, } } fn without_location() -> Self { Self { last_request: Arc::new(Mutex::new(None)), response_headers: HeaderMap::new(), } } } impl HttpTransport for CapturingTransport { async fn execute(&self, req: Request) -> Result<Response, TransportError> { *self.last_request.lock().unwrap() = Some(req); Ok(Response { status: StatusCode::OK, headers: self.response_headers.clone(), body: Bytes::from_static(b"v=0\r\n"), }) } async fn stream(&self, _req: Request) -> Result<StreamResponse, TransportError> { Err(TransportError::Build("stream should not run".to_string())) } } #[derive(Clone, Default)] struct DummyAuth; impl AuthProvider for DummyAuth { fn add_auth_headers(&self, headers: &mut HeaderMap) { headers.insert( http::header::AUTHORIZATION, HeaderValue::from_static("Bearer test-token"), ); } } fn provider(base_url: &str) -> Provider { Provider { name: "test".to_string(), base_url: base_url.to_string(), query_params: None, headers: HeaderMap::new(), retry: RetryConfig { max_attempts: 1, base_delay: Duration::from_millis(1), retry_429: false, retry_5xx: true, retry_transport: true, }, stream_idle_timeout: Duration::from_secs(1), } } fn realtime_session_config(session_id: &str) -> RealtimeSessionConfig { RealtimeSessionConfig { instructions: "hi".to_string(), model: Some("gpt-realtime".to_string()), session_id: Some(session_id.to_string()), event_parser: RealtimeEventParser::RealtimeV2, session_mode: RealtimeSessionMode::Conversational, output_modality: RealtimeOutputModality::Audio, voice: RealtimeVoice::Marin, } } #[tokio::test] async fn sends_sdp_offer_as_raw_body() { let transport = CapturingTransport::new(); let client = RealtimeCallClient::new( transport.clone(), provider("https://api.openai.com/v1"), Arc::new(DummyAuth), ); let response = client .create("v=offer\r\n".to_string()) .await .expect("request should succeed"); assert_eq!( response, RealtimeCallResponse { sdp: "v=0\r\n".to_string(), call_id: "rtc_test".to_string(), } ); let request = transport.last_request.lock().unwrap().clone().unwrap(); assert_eq!(request.method, Method::POST); assert_eq!(request.url, "https://api.openai.com/v1/realtime/calls"); assert_eq!( request.headers.get(CONTENT_TYPE).unwrap(), HeaderValue::from_static("application/sdp") ); assert_eq!( request .headers .get(http::header::AUTHORIZATION) .and_then(|value| value.to_str().ok()), Some("Bearer test-token") ); assert_eq!( request.body, Some(RequestBody::Raw(Bytes::from_static(b"v=offer\r\n"))) ); } #[tokio::test] async fn extracts_call_id_from_forwarded_backend_location() { let transport = CapturingTransport::with_location("/v1/realtime/calls/calls/rtc_backend_test"); let client = RealtimeCallClient::new( transport.clone(), provider("https://chatgpt.com/backend-api/codex"), Arc::new(DummyAuth), ); let response = client .create("v=offer\r\n".to_string()) .await .expect("request should succeed"); assert_eq!( response, RealtimeCallResponse { sdp: "v=0\r\n".to_string(), call_id: "rtc_backend_test".to_string(), } ); let request = transport.last_request.lock().unwrap().clone().unwrap(); assert_eq!(request.method, Method::POST); assert_eq!( request.url, "https://chatgpt.com/backend-api/codex/realtime/calls" ); assert_eq!( request.body, Some(RequestBody::Raw(Bytes::from_static(b"v=offer\r\n"))) ); } #[tokio::test] async fn sends_api_session_call_as_multipart_body() { let transport = CapturingTransport::new(); let client = RealtimeCallClient::new( transport.clone(), provider("https://api.openai.com/v1"), Arc::new(DummyAuth), ); let response = client .create_with_session( "v=offer\r\n".to_string(), realtime_session_config("sess-api"), ) .await .expect("request should succeed"); assert_eq!( response, RealtimeCallResponse { sdp: "v=0\r\n".to_string(), call_id: "rtc_test".to_string(), } ); let request = transport.last_request.lock().unwrap().clone().unwrap(); assert_eq!(request.method, Method::POST); assert_eq!(request.url, "https://api.openai.com/v1/realtime/calls"); assert_eq!( request.headers.get(CONTENT_TYPE).unwrap(), HeaderValue::from_static(MULTIPART_CONTENT_TYPE) ); let Some(RequestBody::Raw(body)) = request.body else { panic!("multipart body should be raw"); }; let body = std::str::from_utf8(&body).expect("multipart body should be utf-8"); let mut session = realtime_session_json(realtime_session_config("sess-api")) .expect("session should encode"); session .as_object_mut() .expect("session should be an object") .remove("id"); let session = to_string(&session).expect("session should serialize"); assert_eq!( body, format!( "--codex-realtime-call-boundary\r\n\ Content-Disposition: form-data; name=\"sdp\"\r\n\ Content-Type: application/sdp\r\n\ \r\n\ v=offer\r\n\ \r\n\ --codex-realtime-call-boundary\r\n\ Content-Disposition: form-data; name=\"session\"\r\n\ Content-Type: application/json\r\n\ \r\n\ {session}\r\n\ --codex-realtime-call-boundary--\r\n" ) ); } #[tokio::test] async fn sends_avas_session_call_query_params() { let transport = CapturingTransport::new(); let client = RealtimeCallClient::new( transport.clone(), provider("https://api.openai.com/v1"), Arc::new(DummyAuth), ); let response = client .create_with_session_architecture_and_headers( "v=offer\r\n".to_string(), realtime_session_config("sess-api"), RealtimeConversationArchitecture::Avas, HeaderMap::new(), ) .await .expect("request should succeed"); assert_eq!( response, RealtimeCallResponse { sdp: "v=0\r\n".to_string(), call_id: "rtc_test".to_string(), } ); let request = transport.last_request.lock().unwrap().clone().unwrap(); assert_eq!(request.method, Method::POST); assert_eq!( request.url, "https://api.openai.com/v1/realtime/calls?intent=quicksilver&architecture=avas" ); } #[tokio::test] async fn sends_backend_session_call_as_json_body() { let transport = CapturingTransport::new(); let client = RealtimeCallClient::new( transport.clone(), provider("https://chatgpt.com/backend-api/codex"), Arc::new(DummyAuth), ); let response = client .create_with_session( "v=offer\r\n".to_string(), realtime_session_config("sess-backend"), ) .await .expect("request should succeed"); assert_eq!( response, RealtimeCallResponse { sdp: "v=0\r\n".to_string(), call_id: "rtc_test".to_string(), } ); let request = transport.last_request.lock().unwrap().clone().unwrap(); assert_eq!(request.method, Method::POST); assert_eq!( request.url, "https://chatgpt.com/backend-api/codex/realtime/calls" ); let mut expected_session = realtime_session_json(realtime_session_config("sess-backend")) .expect("session should encode"); expected_session .as_object_mut() .expect("session should be an object") .remove("id"); assert_eq!( request.body, Some(RequestBody::Json( to_value(BackendRealtimeCallRequest { sdp: "v=offer\r\n", session: &expected_session, }) .expect("request should encode") )) ); } #[tokio::test] async fn errors_when_location_is_missing() { let transport = CapturingTransport::without_location(); let client = RealtimeCallClient::new( transport, provider("https://api.openai.com/v1"), Arc::new(DummyAuth), ); let err = client .create("v=offer\r\n".to_string()) .await .expect_err("request should require Location"); assert_eq!( err.to_string(), "stream error: realtime call response missing Location" ); } #[test] fn rejects_location_without_call_id() { let mut headers = HeaderMap::new(); headers.insert(LOCATION, HeaderValue::from_static("/v1/realtime/calls")); let err = decode_call_id_from_location(&headers) .expect_err("Location without rtc_ segment should fail"); assert_eq!( err.to_string(), "stream error: realtime call Location does not contain a call id: /v1/realtime/calls" ); } #[test] fn accepts_uuid_call_id_from_location() { let mut headers = HeaderMap::new(); headers.insert( LOCATION, HeaderValue::from_static("/v1/realtime/calls/019eb97d-8e9a-7ff3-94b0-ea019babd5d7"), ); let call_id = decode_call_id_from_location(&headers).expect("UUID call id should parse"); assert_eq!(call_id, "019eb97d-8e9a-7ff3-94b0-ea019babd5d7"); }}