diff --git a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs index 4ae9187ea..975819dc7 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -1225,14 +1225,14 @@ async fn webrtc_v1_start_posts_offer_returns_sdp_and_joins_sideband() -> Result< "v=offer\r\n", v1_session_create_json(), )?; + + let session_update = harness.sideband_outbound_request(/*request_index*/ 0).await; + assert_v1_session_update(&session_update)?; assert_eq!( harness.realtime_server.single_handshake().uri(), "/v1/realtime?intent=quicksilver&call_id=rtc_e2e" ); - let session_update = harness.sideband_outbound_request(/*request_index*/ 0).await; - assert_v1_session_update(&session_update)?; - let closed = timeout( Duration::from_millis(100), harness diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index eff209b62..249b3ae15 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -196,6 +196,12 @@ struct RealtimeInputTask { event_parser: RealtimeEventParser, } +struct RealtimeInputChannels { + user_text_rx: Receiver, + handoff_output_rx: Receiver, + audio_rx: Receiver, +} + impl RealtimeHandoffState { fn new(output_tx: Sender, session_kind: RealtimeSessionKind) -> Self { Self { @@ -212,7 +218,6 @@ struct ConversationState { audio_tx: Sender, user_text_tx: Sender, session_kind: RealtimeSessionKind, - writer: RealtimeWebsocketWriter, handoff: RealtimeHandoffState, input_task: JoinHandle<()>, fanout_task: Option>, @@ -284,39 +289,6 @@ impl RealtimeConversationManager { RealtimeEventParser::RealtimeV2 => RealtimeSessionKind::V2, }; - let client = RealtimeWebsocketClient::new(api_provider); - let (connection, sdp) = if let Some(sdp) = sdp { - let call = model_client - .create_realtime_call_with_headers( - sdp, - session_config.clone(), - extra_headers.unwrap_or_default(), - ) - .await?; - let connection = client - .connect_webrtc_sideband( - session_config, - &call.call_id, - call.sideband_headers, - default_headers(), - ) - .await - .map_err(map_api_error)?; - (connection, Some(call.sdp)) - } else { - let connection = client - .connect( - session_config, - extra_headers.unwrap_or_default(), - default_headers(), - ) - .await - .map_err(map_api_error)?; - (connection, None) - }; - - let writer = connection.writer(); - let events = connection.events(); let (audio_tx, audio_rx) = async_channel::bounded::(AUDIO_IN_QUEUE_CAPACITY); let (user_text_tx, user_text_rx) = @@ -328,24 +300,62 @@ impl RealtimeConversationManager { let realtime_active = Arc::new(AtomicBool::new(true)); let handoff = RealtimeHandoffState::new(handoff_output_tx, session_kind); - let task = spawn_realtime_input_task(RealtimeInputTask { - writer: writer.clone(), - events, + let input_channels = RealtimeInputChannels { user_text_rx, handoff_output_rx, audio_rx, - events_tx, - handoff_state: handoff.clone(), - session_kind, - event_parser, - }); + }; + + let client = RealtimeWebsocketClient::new(api_provider); + let (task, sdp) = if let Some(sdp) = sdp { + let call = model_client + .create_realtime_call_with_headers( + sdp, + session_config.clone(), + extra_headers.unwrap_or_default(), + ) + .await?; + let task = spawn_webrtc_sideband_input_task(RealtimeWebrtcSidebandInputTask { + client, + session_config, + call_id: call.call_id, + sideband_headers: call.sideband_headers, + input_channels, + events_tx, + handoff_state: handoff.clone(), + session_kind, + event_parser, + realtime_active: Arc::clone(&realtime_active), + }); + (task, Some(call.sdp)) + } else { + let connection = client + .connect( + session_config, + extra_headers.unwrap_or_default(), + default_headers(), + ) + .await + .map_err(map_api_error)?; + let task = spawn_realtime_input_task(RealtimeInputTask { + writer: connection.writer(), + events: connection.events(), + user_text_rx: input_channels.user_text_rx, + handoff_output_rx: input_channels.handoff_output_rx, + audio_rx: input_channels.audio_rx, + events_tx, + handoff_state: handoff.clone(), + session_kind, + event_parser, + }); + (task, None) + }; let mut guard = self.state.lock().await; *guard = Some(ConversationState { audio_tx, user_text_tx, session_kind, - writer, handoff, input_task: task, fanout_task: None, @@ -1004,6 +1014,83 @@ pub(crate) async fn handle_close(sess: &Arc, sub_id: String) { } fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> { + tokio::spawn(run_realtime_input_task(input)) +} + +struct RealtimeWebrtcSidebandInputTask { + client: RealtimeWebsocketClient, + session_config: RealtimeSessionConfig, + call_id: String, + sideband_headers: HeaderMap, + input_channels: RealtimeInputChannels, + events_tx: Sender, + handoff_state: RealtimeHandoffState, + session_kind: RealtimeSessionKind, + event_parser: RealtimeEventParser, + realtime_active: Arc, +} + +fn spawn_webrtc_sideband_input_task(input: RealtimeWebrtcSidebandInputTask) -> JoinHandle<()> { + let RealtimeWebrtcSidebandInputTask { + client, + session_config, + call_id, + sideband_headers, + input_channels, + events_tx, + handoff_state, + session_kind, + event_parser, + realtime_active, + } = input; + + tokio::spawn(async move { + if !realtime_active.load(Ordering::Relaxed) { + return; + } + + let connection = match client + .connect_webrtc_sideband( + session_config, + &call_id, + sideband_headers, + default_headers(), + ) + .await + { + Ok(connection) => connection, + Err(err) => { + if realtime_active.load(Ordering::Relaxed) { + let mapped_error = map_api_error(err); + warn!("failed to connect realtime sideband: {mapped_error}"); + let _ = events_tx + .send(RealtimeEvent::Error(mapped_error.to_string())) + .await; + } + return; + } + }; + + if !realtime_active.load(Ordering::Relaxed) { + return; + } + + run_realtime_input_task(RealtimeInputTask { + writer: connection.writer(), + events: connection.events(), + user_text_rx: input_channels.user_text_rx, + handoff_output_rx: input_channels.handoff_output_rx, + audio_rx: input_channels.audio_rx, + events_tx, + handoff_state, + session_kind, + event_parser, + }) + .await; + }) +} + +async fn run_realtime_input_task(input: RealtimeInputTask) { let RealtimeInputTask { writer, events, @@ -1016,57 +1103,55 @@ fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> { event_parser, } = input; - tokio::spawn(async move { - let mut output_audio_state: Option = None; - let mut response_create_queue = RealtimeResponseCreateQueue::default(); + let mut output_audio_state: Option = None; + let mut response_create_queue = RealtimeResponseCreateQueue::default(); - loop { - let result = tokio::select! { - // Text typed by the user that should be sent into realtime. - user_text = user_text_rx.recv() => { - handle_user_text_input( - user_text, - &writer, - &events_tx, - ) - .await - } - // Background agent progress or final output that should be sent back to realtime. - background_agent_output = handoff_output_rx.recv() => { - handle_handoff_output( - background_agent_output, - &writer, - &events_tx, - &handoff_state, - event_parser, - &mut response_create_queue, - ) - .await - } - // Events received from the realtime server. - realtime_event = events.next_event() => { - handle_realtime_server_event( - realtime_event, - &writer, - &events_tx, - &handoff_state, - session_kind, - &mut output_audio_state, - &mut response_create_queue, - ) + loop { + let result = tokio::select! { + // Text typed by the user that should be sent into realtime. + user_text = user_text_rx.recv() => { + handle_user_text_input( + user_text, + &writer, + &events_tx, + ) .await - } - // Audio frames captured from the user microphone. - user_audio_frame = audio_rx.recv() => { - handle_user_audio_input(user_audio_frame, &writer, &events_tx) - .await - } - }; - if result.is_err() { - break; } + // Background agent progress or final output that should be sent back to realtime. + background_agent_output = handoff_output_rx.recv() => { + handle_handoff_output( + background_agent_output, + &writer, + &events_tx, + &handoff_state, + event_parser, + &mut response_create_queue, + ) + .await + } + // Events received from the realtime server. + realtime_event = events.next_event() => { + handle_realtime_server_event( + realtime_event, + &writer, + &events_tx, + &handoff_state, + session_kind, + &mut output_audio_state, + &mut response_create_queue, + ) + .await + } + // Audio frames captured from the user microphone. + user_audio_frame = audio_rx.recv() => { + handle_user_audio_input(user_audio_frame, &writer, &events_tx) + .await + } + }; + if result.is_err() { + break; } - }) + } } async fn handle_user_text_input( diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 96aa979f9..ff273a77c 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -120,6 +120,21 @@ fn websocket_request_instructions( .map(str::to_owned) } +async fn wait_for_websocket_request( + server: &core_test_support::responses::WebSocketTestServer, + connection_index: usize, + request_index: usize, +) -> Result { + timeout( + Duration::from_secs(2), + server.wait_for_request(connection_index, request_index), + ) + .await + .with_context(|| { + format!("timed out waiting for websocket request {connection_index}/{request_index}") + }) +} + fn expected_realtime_backend_prompt() -> String { REALTIME_BACKEND_PROMPT .trim_end() @@ -456,6 +471,7 @@ async fn conversation_webrtc_start_posts_generated_session() -> Result<()> { skip_if_no_network!(Ok(())); let server = start_mock_server().await; + let sideband_accept_delay = Duration::from_millis(1000); let capture = RealtimeCallRequestCapture::new(); Mock::given(method("POST")) .and(path_regex(".*/realtime/calls$")) @@ -468,12 +484,15 @@ async fn conversation_webrtc_start_posts_generated_session() -> Result<()> { .mount(&server) .await; let realtime_server = start_websocket_server_with_headers(vec![WebSocketConnectionConfig { - requests: vec![vec![json!({ - "type": "session.updated", - "session": { "id": "sess_webrtc", "instructions": "backend prompt" } - })]], + requests: vec![ + vec![json!({ + "type": "session.updated", + "session": { "id": "sess_webrtc", "instructions": "backend prompt" } + })], + vec![], + ], response_headers: Vec::new(), - accept_delay: None, + accept_delay: Some(sideband_accept_delay), close_after_requests: false, }]) .await; @@ -510,6 +529,16 @@ async fn conversation_webrtc_start_posts_generated_session() -> Result<()> { .await .unwrap_or_else(|err: ErrorEvent| panic!("conversation call create failed: {err:?}")); assert_eq!(created.sdp, "v=answer\r\n"); + assert!( + realtime_server.handshakes().is_empty(), + "SDP should be emitted before the delayed sideband websocket joins" + ); + + test.codex + .submit(Op::RealtimeConversationText(ConversationTextParams { + text: "queued before sideband".to_string(), + })) + .await?; let session_updated = wait_for_event_match(&test.codex, |msg| match msg { EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { @@ -566,9 +595,12 @@ async fn conversation_webrtc_start_posts_generated_session() -> Result<()> { // Phase 3: the server joins that same call over the direct sideband WebSocket, sends the // ordinary session.update, and keeps the conversation alive until the client closes it. - let session_update = realtime_server - .wait_for_request(/*connection_index*/ 0, /*request_index*/ 0) - .await; + let session_update = wait_for_websocket_request( + &realtime_server, + /*connection_index*/ 0, + /*request_index*/ 0, + ) + .await?; assert_eq!( session_update.body_json()["type"].as_str(), Some("session.update") @@ -578,6 +610,16 @@ async fn conversation_webrtc_start_posts_generated_session() -> Result<()> { .context("session.update should include instructions")? .contains("startup context") ); + let queued_text = wait_for_websocket_request( + &realtime_server, + /*connection_index*/ 0, + /*request_index*/ 1, + ) + .await?; + assert_eq!( + websocket_request_text(&queued_text).as_deref(), + Some("queued before sideband") + ); let handshake = realtime_server.single_handshake(); assert_eq!( handshake.uri(), @@ -603,6 +645,176 @@ async fn conversation_webrtc_start_posts_generated_session() -> Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn conversation_webrtc_close_while_sideband_connecting_drops_pending_join() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + Mock::given(method("POST")) + .and(path_regex(".*/realtime/calls$")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("Location", "/v1/realtime/calls/calls/rtc_close_pending") + .set_body_string("v=answer\r\n"), + ) + .mount(&server) + .await; + let realtime_server = start_websocket_server_with_headers(vec![WebSocketConnectionConfig { + requests: vec![vec![]], + response_headers: Vec::new(), + accept_delay: Some(Duration::from_millis(500)), + close_after_requests: false, + }]) + .await; + + let realtime_ws_base_url = realtime_server.uri().to_string(); + let mut builder = test_codex().with_config(move |config| { + 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.realtime.version = RealtimeWsVersion::V1; + }); + let test = builder.build(&server).await?; + + test.codex + .submit(Op::RealtimeConversationStart(ConversationStartParams { + output_modality: RealtimeOutputModality::Audio, + prompt: Some(Some("backend prompt".to_string())), + realtime_session_id: None, + transport: Some(ConversationStartTransport::Webrtc { + sdp: "v=offer\r\n".to_string(), + }), + voice: None, + })) + .await?; + + let sdp = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationSdp(created) => Some(created.sdp.clone()), + _ => None, + }) + .await; + assert_eq!(sdp, "v=answer\r\n"); + assert!( + realtime_server.handshakes().is_empty(), + "sideband websocket should still be pending when SDP is emitted" + ); + + test.codex.submit(Op::RealtimeConversationClose).await?; + let closed = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationClosed(closed) => Some(closed.clone()), + _ => None, + }) + .await; + assert_eq!(closed.reason.as_deref(), Some("requested")); + + let stale_event = timeout(Duration::from_millis(700), async { + wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::Error(message), + }) => Some(format!("stale realtime error: {message}")), + EventMsg::RealtimeConversationClosed(closed) => { + Some(format!("stale close event: {:?}", closed.reason)) + } + _ => None, + }) + .await + }) + .await; + assert!( + stale_event.is_err(), + "pending sideband task leaked after close: {:?}", + stale_event.ok() + ); + assert!( + realtime_server.handshakes().is_empty(), + "pending sideband task should abort before websocket handshake completes" + ); + + realtime_server.shutdown().await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn conversation_webrtc_sideband_connect_failure_closes_with_error() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + Mock::given(method("POST")) + .and(path_regex(".*/realtime/calls$")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("Location", "/v1/realtime/calls/calls/rtc_sideband_failure") + .set_body_string("v=answer\r\n"), + ) + .mount(&server) + .await; + let mut builder = test_codex().with_config(|config| { + 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("http://127.0.0.1:1".to_string()); + config.realtime.version = RealtimeWsVersion::V1; + }); + let test = builder.build(&server).await?; + + test.codex + .submit(Op::RealtimeConversationStart(ConversationStartParams { + output_modality: RealtimeOutputModality::Audio, + prompt: Some(Some("backend prompt".to_string())), + realtime_session_id: None, + transport: Some(ConversationStartTransport::Webrtc { + sdp: "v=offer\r\n".to_string(), + }), + voice: None, + })) + .await?; + + let started = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationStarted(started) => Some(started.clone()), + _ => None, + }) + .await; + assert!(started.realtime_session_id.is_some()); + + let sdp = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationSdp(created) => Some(created.sdp.clone()), + _ => None, + }) + .await; + assert_eq!(sdp, "v=answer\r\n"); + + let err = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::Error(message), + }) => Some(message.clone()), + _ => None, + }) + .await; + assert!(!err.is_empty()); + + let closed = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationClosed(closed) => Some(closed.clone()), + _ => None, + }) + .await; + assert_eq!(closed.reason.as_deref(), Some("error")); + + test.codex + .submit(Op::RealtimeConversationText(ConversationTextParams { + text: "after sideband failure".to_string(), + })) + .await?; + let err = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::Error(err) => Some(err.clone()), + _ => None, + }) + .await; + assert_eq!(err.message, "conversation is not running"); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() -> Result<()> { if std::env::var_os(REALTIME_CONVERSATION_TEST_SUBPROCESS_ENV_VAR).is_none() {