diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index 84047d2d2f..129ad3b5a8 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -1272,7 +1272,10 @@ impl TurnRequestProcessor { ConversationStartTransport::Webrtc { sdp } } ThreadRealtimeStartTransport::ExistingCall { call_id } => { - ConversationStartTransport::ExistingCall { call_id } + ConversationStartTransport::ExistingCall { + call_id, + sideband_base_url: None, + } } }), version: params.version, diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index f60d27dcee..bec3a05985 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -1191,7 +1191,16 @@ async fn prepare_realtime_start( .clone() .unwrap_or(ConversationStartTransport::Websocket); let mut api_provider = provider.to_api_provider(Some(AuthMode::ApiKey))?; - let realtime_sideband_base_url = config.experimental_realtime_ws_base_url.clone(); + let realtime_sideband_base_url = match &transport { + ConversationStartTransport::ExistingCall { + sideband_base_url, .. + } => sideband_base_url + .clone() + .or_else(|| config.experimental_realtime_ws_base_url.clone()), + ConversationStartTransport::Websocket | ConversationStartTransport::Webrtc { .. } => { + config.experimental_realtime_ws_base_url.clone() + } + }; if let Some(realtime_ws_base_url) = &realtime_sideband_base_url { api_provider.base_url = realtime_ws_base_url.clone(); } @@ -1525,7 +1534,7 @@ async fn handle_start_inner( let (sdp, existing_call_id) = match transport { ConversationStartTransport::Websocket => (None, None), ConversationStartTransport::Webrtc { sdp } => (Some(sdp), None), - ConversationStartTransport::ExistingCall { call_id } => (None, Some(call_id)), + ConversationStartTransport::ExistingCall { call_id, .. } => (None, Some(call_id)), }; let mode_instructions = RealtimeModeInstructions { start: realtime_start_instructions, diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 4d434aed55..5b6961228b 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -126,6 +126,7 @@ mod prompt_debug_tests; mod quota_exceeded; mod realtime_conversation; mod realtime_initial_items; +mod realtime_sideband_endpoint; mod remote_env; mod remote_models; mod request_compression; diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index c93e25425a..de16d730ed 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -64,8 +64,10 @@ use tokio::io::AsyncReadExt; use tokio::io::AsyncWriteExt; use tokio::sync::oneshot; use tokio::time::timeout; -use tokio_tungstenite::accept_async; +use tokio_tungstenite::accept_hdr_async; use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::tungstenite::handshake::server::Request; +use tokio_tungstenite::tungstenite::handshake::server::Response; use uuid::Uuid; use wiremock::Match; use wiremock::Mock; @@ -514,7 +516,7 @@ async fn conversation_start_defaults_to_v2_and_gpt_realtime_1_5() -> Result<()> } #[test_matrix( - [ConversationStartTransport::Websocket, ConversationStartTransport::ExistingCall { call_id: "rtc_existing".to_string() }], + [ConversationStartTransport::Websocket, ConversationStartTransport::ExistingCall { call_id: "rtc_existing".to_string(), sideband_base_url: None }], [None, Some(ThreadSource::User)] )] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -1034,22 +1036,29 @@ async fn conversation_webrtc_start_posts_generated_session() -> Result<()> { Ok(()) } +#[derive(Clone, Copy)] +enum SidebandReconnectTransport { + Webrtc, + ExistingCallConfiguredEndpoint, + ExistingCallPerCallEndpoint, +} + +#[test_case(SidebandReconnectTransport::Webrtc; "core-created webrtc")] #[test_case( - ConversationStartTransport::Webrtc { sdp: "v=offer\r\n".to_string() }; - "core-created webrtc" + SidebandReconnectTransport::ExistingCallConfiguredEndpoint; + "client-created existing call" )] #[test_case( - ConversationStartTransport::ExistingCall { call_id: "rtc_reconnect".to_string() }; - "client-created existing call" + SidebandReconnectTransport::ExistingCallPerCallEndpoint; + "client-created existing call with per-call endpoint" )] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn conversation_webrtc_live_reconnects_sideband_after_unclean_disconnect( - transport: ConversationStartTransport, + transport: SidebandReconnectTransport, ) -> Result<()> { skip_if_no_network!(Ok(())); - let attaches_existing_call = - matches!(&transport, ConversationStartTransport::ExistingCall { .. }); + let attaches_existing_call = !matches!(transport, SidebandReconnectTransport::Webrtc); let server = start_mock_server().await; Mock::given(method("POST")) .and(path("/v1/live")) @@ -1072,10 +1081,41 @@ async fn conversation_webrtc_live_reconnects_sideband_after_unclean_disconnect( let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; let realtime_ws_base_url = format!("ws://{}", listener.local_addr()?); + let unused_endpoint = start_websocket_server(vec![vec![]]).await; + let configured_base_url = match transport { + SidebandReconnectTransport::Webrtc + | SidebandReconnectTransport::ExistingCallConfiguredEndpoint => { + realtime_ws_base_url.clone() + } + SidebandReconnectTransport::ExistingCallPerCallEndpoint => { + unused_endpoint.uri().to_string() + } + }; + let transport = match transport { + SidebandReconnectTransport::Webrtc => ConversationStartTransport::Webrtc { + sdp: "v=offer\r\n".to_string(), + }, + SidebandReconnectTransport::ExistingCallConfiguredEndpoint => { + ConversationStartTransport::ExistingCall { + call_id: "rtc_reconnect".to_string(), + sideband_base_url: None, + } + } + SidebandReconnectTransport::ExistingCallPerCallEndpoint => { + ConversationStartTransport::ExistingCall { + call_id: "rtc_reconnect".to_string(), + sideband_base_url: Some(realtime_ws_base_url), + } + } + }; let (inbound_flood_started_tx, inbound_flood_started_rx) = oneshot::channel(); let sideband_server = tokio::spawn(async move { + let validate_handshake = |request: &Request, response: Response| { + assert_eq!(request.uri().to_string(), "/v1/live/rtc_reconnect"); + Ok(response) + }; let (stream, _) = listener.accept().await?; - let mut websocket = accept_async(stream).await?; + let mut websocket = accept_hdr_async(stream, validate_handshake).await?; websocket .send(Message::Text( json!({ @@ -1139,7 +1179,7 @@ async fn conversation_webrtc_live_reconnects_sideband_after_unclean_disconnect( drop(websocket_rx); let (stream, _) = listener.accept().await?; - let mut websocket = accept_async(stream).await?; + let mut websocket = accept_hdr_async(stream, validate_handshake).await?; websocket .send(Message::Text( json!({ @@ -1180,8 +1220,8 @@ async fn conversation_webrtc_live_reconnects_sideband_after_unclean_disconnect( let (mut stream, _) = listener.accept().await?; let mut request = vec![0_u8; 4096]; let request_len = stream.read(&mut request).await?; - if !request[..request_len].starts_with(b"GET ") { - anyhow::bail!("expected reconnect websocket handshake"); + if !request[..request_len].starts_with(b"GET /v1/live/rtc_reconnect HTTP/1.1\r\n") { + anyhow::bail!("expected reconnect websocket handshake for the same call"); } stream .write_all(b"HTTP/1.1 410 Gone\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") @@ -1199,7 +1239,7 @@ async fn conversation_webrtc_live_reconnects_sideband_after_unclean_disconnect( config.experimental_realtime_ws_backend_prompt = Some("backend prompt".to_string()); config.experimental_realtime_ws_model = Some("realtime-test-model".to_string()); config.experimental_realtime_ws_startup_context = Some(String::new()); - config.experimental_realtime_ws_base_url = Some(realtime_ws_base_url); + config.experimental_realtime_ws_base_url = Some(configured_base_url); config.realtime.version = RealtimeWsVersion::V3; }); let test = builder.build_with_auto_env(&server).await?; @@ -1315,6 +1355,9 @@ async fn conversation_webrtc_live_reconnects_sideband_after_unclean_disconnect( timeout(Duration::from_secs(10), sideband_server) .await .context("timed out waiting for sideband server")???; + assert!(unused_endpoint.handshakes().is_empty()); + test.codex.shutdown_and_wait().await?; + unused_endpoint.shutdown().await; Ok(()) } diff --git a/codex-rs/core/tests/suite/realtime_sideband_endpoint.rs b/codex-rs/core/tests/suite/realtime_sideband_endpoint.rs new file mode 100644 index 0000000000..b94be5d07f --- /dev/null +++ b/codex-rs/core/tests/suite/realtime_sideband_endpoint.rs @@ -0,0 +1,125 @@ +//! Verifies that existing-call sidebands honor trusted per-call endpoint overrides, +//! fall back to the configured endpoint, and preserve runtime authentication without +//! adding bearer credentials. + +use anyhow::Result; +use codex_login::AuthHeaders; +use codex_login::CodexAuth; +use codex_protocol::protocol::CodexResponseHandoffMode; +use codex_protocol::protocol::ConversationStartParams; +use codex_protocol::protocol::ConversationStartTransport; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::Op; +use codex_protocol::protocol::RealtimeConversationVersion; +use codex_protocol::protocol::RealtimeOutputModality; +use core_test_support::responses::start_mock_server; +use core_test_support::responses::start_websocket_server; +use core_test_support::skip_if_no_network; +use core_test_support::test_codex::test_codex; +use core_test_support::wait_for_event; +use core_test_support::wait_for_event_match; +use http::HeaderMap; +use http::HeaderValue; +use pretty_assertions::assert_eq; +use serde_json::json; +use test_case::test_case; + +#[derive(Clone, Copy)] +enum EndpointSelection { + Configured, + PerCall, +} + +#[test_case(EndpointSelection::Configured; "configured endpoint")] +#[test_case(EndpointSelection::PerCall; "per-call endpoint")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn existing_call_uses_selected_endpoint_and_runtime_auth( + selection: EndpointSelection, +) -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let events = vec![vec![vec![json!({ + "type": "session.started", + "session": { "id": "rtc_existing" } + })]]]; + let configured_endpoint = start_websocket_server(events.clone()).await; + let call_endpoint = start_websocket_server(events).await; + let configured_base_url = configured_endpoint.uri().to_string(); + let mut headers = HeaderMap::new(); + headers.insert( + "x-runtime-auth", + HeaderValue::from_static("runtime-credential"), + ); + let mut builder = test_codex() + .with_auth(CodexAuth::Headers(AuthHeaders::new(headers))) + .with_config(move |config| { + config.experimental_realtime_ws_base_url = Some(configured_base_url); + }); + let test = builder.build_with_auto_env(&server).await?; + + test.codex + .submit(Op::RealtimeConversationStart(ConversationStartParams { + client_managed_handoffs: false, + delegation_ack_filler: None, + flush_transcript_tail_on_session_end: false, + codex_responses_as_items: false, + codex_response_item_prefix: None, + codex_response_handoff_mode: CodexResponseHandoffMode::Thinking, + codex_response_handoff_channel_prefixes: None, + model: None, + output_modality: RealtimeOutputModality::Audio, + include_startup_context: false, + initial_items: Vec::new(), + realtime_start_instructions: None, + realtime_end_instructions: None, + prompt: None, + realtime_session_id: None, + transport: Some(ConversationStartTransport::ExistingCall { + call_id: "rtc_existing".to_string(), + sideband_base_url: match selection { + EndpointSelection::Configured => None, + EndpointSelection::PerCall => Some(call_endpoint.uri().to_string()), + }, + }), + version: Some(RealtimeConversationVersion::V3), + voice: None, + })) + .await?; + + wait_for_event_match(&test.codex, |event| match event { + EventMsg::RealtimeConversationStarted(_) => Some(Ok(())), + EventMsg::Error(error) => Some(Err(anyhow::anyhow!("{error:?}"))), + _ => None, + }) + .await?; + + let (selected, unused) = match selection { + EndpointSelection::Configured => (&configured_endpoint, &call_endpoint), + EndpointSelection::PerCall => (&call_endpoint, &configured_endpoint), + }; + let handshake = selected.single_handshake(); + assert_eq!( + ( + handshake.uri(), + handshake.header("x-runtime-auth"), + handshake.header("authorization"), + ), + ( + "/v1/live/rtc_existing", + Some("runtime-credential".to_string()), + None, + ), + ); + assert!(unused.handshakes().is_empty()); + + test.codex.submit(Op::RealtimeConversationClose).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::RealtimeConversationClosed(_)) + }) + .await; + test.codex.shutdown_and_wait().await?; + configured_endpoint.shutdown().await; + call_endpoint.shutdown().await; + Ok(()) +} diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index db862f866a..06ba9705f7 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -254,8 +254,16 @@ pub struct ConversationStartParams { #[derive(Debug, Clone, PartialEq)] pub enum ConversationStartTransport { Websocket, - Webrtc { sdp: String }, - ExistingCall { call_id: String }, + Webrtc { + sdp: String, + }, + ExistingCall { + call_id: String, + /// Endpoint selected by the embedding runtime for this call's sideband. + /// This is an in-process override, not a client-supplied API parameter. + /// `None` uses the configured endpoint or the default public API. + sideband_base_url: Option, + }, } #[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]