Retry realtime sideband attachment while an existing call activates (#51460)

## Why

Instant Connect can return a call ID before ICE activates the call, causing sideband attachment to briefly fail with HTTP `404` while the client connects media.

## What changed

Retry `404` responses from `connect_existing_call_sideband` up to three times, waiting 100, 300, and 800 milliseconds. Return the error once this activation retry budget is exhausted, and preserve the existing handling of other errors.

## Testing

Add WebSocket coverage for successful attachment after three `404` responses, failure after four `404` responses, and no retry on HTTP `410`. Verify that successful attachment does not overwrite the prepared session configuration.

GitOrigin-RevId: 5d18ff7eeeb9abd56388808828382f7350c7ae57
This commit is contained in:
bullman-oai
2026-10-06 19:13:04 +00:00
committed by copyberry
parent 414d165b47
commit 44984d2081
2 changed files with 108 additions and 9 deletions
@@ -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<RealtimeWebsocketConnection, ApiError> {
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(
@@ -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<tokio::net::TcpStream>;
@@ -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 {