diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs index 58ade3b75f..17ed27f439 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -42,6 +42,7 @@ use std::collections::VecDeque; use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; +use std::time::Duration; use tokio::sync::Mutex; use tokio::sync::mpsc; use tokio::sync::oneshot; @@ -908,15 +909,28 @@ impl RealtimeWebsocketClient { default_headers: HeaderMap, transcript_state: RealtimeTranscriptState, ) -> Result { - self.connect_sideband( - config, - call_id, - extra_headers, - default_headers, - RealtimeSessionInitialization::ExistingCall, - transcript_state, - ) - .await + // Instant Connect returns a call ID before ICE activates the call. Its sideband + // can briefly return 404 while the client connects media in parallel. + let mut activation_delays = [100, 300, 800].into_iter(); + loop { + let result = self + .connect_sideband( + config.clone(), + call_id, + extra_headers.clone(), + default_headers.clone(), + RealtimeSessionInitialization::ExistingCall, + transcript_state.clone(), + ) + .await; + if matches!(&result, Err(ApiError::Api { status, .. }) if *status == StatusCode::NOT_FOUND) + && let Some(delay_ms) = activation_delays.next() + { + sleep(Duration::from_millis(delay_ms)).await; + continue; + } + return result; + } } async fn connect_sideband( diff --git a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs index bd6f463852..37a028eef2 100644 --- a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs +++ b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs @@ -3,6 +3,7 @@ use std::collections::HashMap; use std::future::Future; use std::time::Duration; +use codex_api::ApiError; use codex_api::Provider; use codex_api::RealtimeAudioFrame; use codex_api::RealtimeEvent; @@ -23,11 +24,15 @@ use codex_protocol::protocol::RealtimeVoice; use futures::SinkExt; use futures::StreamExt; use http::HeaderMap; +use http::StatusCode; use serde_json::Value; use serde_json::json; use tokio::net::TcpListener; 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; type RealtimeWsStream = tokio_tungstenite::WebSocketStream; @@ -287,6 +292,86 @@ async fn realtime_ws_connect_webrtc_sideband_retries_join_until_server_is_availa server.await.expect("server task"); } +#[tokio::test] +async fn realtime_ws_existing_call_retries_activation_404_with_a_bounded_budget() { + for (status, failures, activates) in [ + (StatusCode::NOT_FOUND, 3, true), + (StatusCode::NOT_FOUND, 4, false), + (StatusCode::GONE, 1, false), + ] { + tokio::time::timeout(Duration::from_secs(5), async { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("address"); + let server = tokio::spawn(async move { + for _ in 0..failures { + let (stream, _) = listener.accept().await.expect("accept handshake"); + accept_hdr_async(stream, move |request: &Request, _: Response| { + assert!( + request + .uri() + .query() + .expect("query") + .contains("call_id=rtc_prepared") + ); + Err(http::Response::builder() + .status(status) + .body(None) + .expect("error response")) + }) + .await + .expect_err("call is not active"); + } + if activates { + let (stream, _) = listener.accept().await.expect("accept activated call"); + let mut ws = accept_async(stream).await.expect("successful handshake"); + // Existing-call attachment must not overwrite the prepared session. + assert!(matches!( + ws.next().await.expect("close").expect("frame"), + Message::Close(_) + )); + } + }); + let mut provider = test_provider(format!("http://{addr}")); + provider.retry.max_attempts = 0; + let client = + test_client(provider).with_webrtc_sideband_base_url(format!("http://{addr}")); + let result = client + .connect_existing_call_sideband( + RealtimeSessionConfig { + instructions: "unused existing-call prompt".to_string(), + initial_items: Vec::new(), + delegation_ack_filler: None, + model: None, + session_id: None, + event_parser: RealtimeEventParser::RealtimeV2, + session_mode: RealtimeSessionMode::Conversational, + output_modality: RealtimeOutputModality::Audio, + voice: RealtimeVoice::Marin, + }, + "rtc_prepared", + HeaderMap::new(), + HeaderMap::new(), + RealtimeTranscriptState::default(), + ) + .await; + if activates { + result + .expect("connect after activation") + .close() + .await + .expect("close"); + } else { + assert!( + matches!(result, Err(ApiError::Api { status: actual, .. }) if actual == status) + ); + } + server.await.expect("server task"); + }) + .await + .expect("bounded sideband handshake attempts"); + } +} + #[tokio::test] async fn realtime_ws_e2e_send_while_next_event_waits() { let (addr, server) = spawn_realtime_ws_server(|mut ws: RealtimeWsStream| async move {