Allow per-call sideband endpoints for existing realtime calls (#41923)

## What changed

- Add an in-process `sideband_base_url` override to `ConversationStartTransport::ExistingCall`, while keeping app-server requests on the configured or default endpoint.
- Prefer the per-call endpoint when attaching and reconnecting a sideband, falling back to `experimental_realtime_ws_base_url` when no override is supplied.
- Preserve runtime authentication headers on the selected endpoint without adding bearer authorization.

## Testing

- Cover configured and per-call endpoint selection, authentication headers, call-specific handshake paths, and reconnect behavior.

GitOrigin-RevId: 16c439fe0a9ff8bcdda4ec615bb9983e30935052

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;

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(())
 }
 

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(())
+}