From ef186f812ae447283776a2968fe1f8083e0d8ce7 Mon Sep 17 00:00:00 2001 From: Jiahao Date: Wed, 28 Oct 2026 18:31:52 +0000 Subject: [PATCH] Detach realtime sideband observers without ending the session (#57817) ## Why Stopping a sideband observer should close its WebSocket while preserving the frontend-owned realtime session. The existing frameless session close also sends `session.close`, so observer shutdown needs a transport-only close. ## What changed - Add `close_transport()` to send a normal WebSocket close without `session.close`. - Close the sideband transport on cancellation or when the call becomes inactive, including an existing connection awaiting adoption by the input task. Bound the close attempt to five seconds. - Thread an optional sideband base URL through realtime start processing and allow nested thread ID fields in request serialization scopes. ## Testing Add WebSocket and core conversation regression tests covering observer detachment for both realtime parsers and conversation versions. Verify that explicit frameless session close still sends `session.close`. GitOrigin-RevId: 874a8980b0fcd45058e976b00c0d968f5b4b2404 --- .../src/protocol/common.rs | 7 +- codex-rs/app-server/src/message_processor.rs | 3 +- .../src/request_processors/turn_processor.rs | 8 +- .../endpoint/realtime_websocket/methods.rs | 28 ++++- .../codex-api/tests/realtime_websocket_e2e.rs | 105 ++++++++++++++++++ codex-rs/core/src/realtime_conversation.rs | 3 + .../src/realtime_conversation/sideband.rs | 37 +++++- .../core/tests/suite/realtime_conversation.rs | 76 +++++++++++++ 8 files changed, 253 insertions(+), 14 deletions(-) diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 00ac5b6a4c..2936c01855 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -148,9 +148,12 @@ macro_rules! serialization_scope_expr { ($actual_params:ident, global_shared_read($key:literal)) => { Some(ClientRequestSerializationScope::GlobalSharedRead($key)) }; - ($actual_params:ident, thread_id($params:ident . $field:ident)) => { + // Match one or more dot-separated fields, allowing both `params.thread_id` + // and nested paths like `params.start.thread_id`. The resolved ID selects + // the per-thread request queue, preserving ordering with other thread operations. + ($actual_params:ident, thread_id($params:ident . $($field:ident).+)) => { Some(ClientRequestSerializationScope::Thread { - thread_id: $actual_params.$field.clone(), + thread_id: $actual_params.$($field).+.clone(), }) }; ($actual_params:ident, optional_thread_id($params:ident . $field:ident)) => { diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 93e0fbdaeb..417165a635 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -1723,8 +1723,9 @@ impl MessageProcessor { .await } ClientRequest::ThreadRealtimeStart { params, .. } => { + let sideband_base_url = None; self.turn_processor - .thread_realtime_start(&request_id, params) + .thread_realtime_start(&request_id, params, sideband_base_url) .await } ClientRequest::ThreadRealtimeAppendAudio { params, .. } => { 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 4ece6caa05..e5be25d12d 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -271,8 +271,9 @@ impl TurnRequestProcessor { &self, request_id: &ConnectionRequestId, params: ThreadRealtimeStartParams, + sideband_base_url: Option, ) -> Result, JSONRPCErrorError> { - self.thread_realtime_start_inner(request_id, params) + self.thread_realtime_start_inner(request_id, params, sideband_base_url) .await .map(|response| response.map(Into::into)) } @@ -1198,6 +1199,7 @@ impl TurnRequestProcessor { &self, request_id: &ConnectionRequestId, params: ThreadRealtimeStartParams, + sideband_base_url: Option, ) -> Result, JSONRPCErrorError> { let attaches_existing_call = matches!( ¶ms.transport, @@ -1276,10 +1278,10 @@ impl TurnRequestProcessor { ThreadRealtimeStartTransport::Webrtc { sdp } => { ConversationStartTransport::Webrtc { sdp } } - ThreadRealtimeStartTransport::ExistingCall { call_id } => { + ThreadRealtimeStartTransport::ExistingCall { call_id, .. } => { ConversationStartTransport::ExistingCall { call_id, - sideband_base_url: None, + sideband_base_url, } } }), 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 29f5d59250..2ce624c85a 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -55,6 +55,7 @@ use tracing::error; use tracing::info; use tracing::trace; use tracing::warn; +use tungstenite::protocol::CloseFrame; use tungstenite::protocol::WebSocketConfig; use tungstenite::protocol::frame::coding::CloseCode; use url::Url; @@ -76,6 +77,7 @@ enum WsCommand { tx_result: oneshot::Sender>, }, Close { + frame: Option, tx_result: oneshot::Sender>, }, } @@ -116,9 +118,9 @@ impl WsStream { break; } } - WsCommand::Close { tx_result } => { + WsCommand::Close { frame, tx_result } => { info!("realtime websocket sending close"); - let result = inner.send(Message::Close(None)).await; + let result = inner.send(Message::Close(frame)).await; if let Err(err) = &result { error!("realtime websocket close failed: {err}"); report_write_policy_error(err); @@ -226,8 +228,10 @@ impl WsStream { .await } - async fn close(&self) -> Result<(), WsError> { - self.request(|tx_result| WsCommand::Close { tx_result }) + /// Ask the socket pump to send a close frame to the connected peer and + /// return the write result. + async fn close(&self, frame: Option) -> Result<(), WsError> { + self.request(|tx_result| WsCommand::Close { frame, tx_result }) .await } } @@ -493,7 +497,7 @@ impl RealtimeWebsocketWriter { ))); } } - if let Err(err) = self.stream.close().await + if let Err(err) = self.stream.close(/*frame*/ None).await && !matches!(err, WsError::ConnectionClosed | WsError::AlreadyClosed) { return Err(ApiError::Stream(format!( @@ -503,6 +507,20 @@ impl RealtimeWebsocketWriter { Ok(()) } + /// Detach an observer without ending the frontend-owned realtime session. + pub async fn close_transport(&self) -> Result<(), ApiError> { + if self.is_closed.swap(true, Ordering::SeqCst) { + return Ok(()); + } + self.stream + .close(Some(CloseFrame { + code: CloseCode::Normal, + reason: "".into(), + })) + .await + .map_err(|err| ApiError::Stream(format!("failed to close websocket transport: {err}"))) + } + async fn send_json(&self, message: &RealtimeOutboundMessage) -> Result<(), ApiError> { match message { RealtimeOutboundMessage::DelegationContextAppend { diff --git a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs index 37a028eef2..d42d4ec7d9 100644 --- a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs +++ b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs @@ -25,6 +25,7 @@ use futures::SinkExt; use futures::StreamExt; use http::HeaderMap; use http::StatusCode; +use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; use tokio::net::TcpListener; @@ -33,6 +34,8 @@ 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 tokio_tungstenite::tungstenite::protocol::CloseFrame; +use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode; type RealtimeWsStream = tokio_tungstenite::WebSocketStream; @@ -372,6 +375,108 @@ async fn realtime_ws_existing_call_retries_activation_404_with_a_bounded_budget( } } +#[tokio::test] +async fn realtime_ws_existing_call_transport_close_preserves_session() { + for event_parser in [ + RealtimeEventParser::RealtimeV2, + RealtimeEventParser::FramelessBidi, + ] { + let (addr, server) = spawn_realtime_ws_server(|mut ws| async move { + assert_eq!( + ws.next().await.expect("close frame").expect("valid close"), + Message::Close(Some(CloseFrame { + code: CloseCode::Normal, + reason: "".into(), + })), + "observer detach must not send an application session.close" + ); + }) + .await; + let client = test_client(test_provider(format!("http://{addr}"))) + .with_webrtc_sideband_base_url(format!("http://{addr}")); + let connection = client + .connect_existing_call_sideband( + RealtimeSessionConfig { + instructions: String::new(), + initial_items: Vec::new(), + delegation_ack_filler: None, + model: None, + session_id: None, + event_parser, + session_mode: RealtimeSessionMode::Conversational, + output_modality: RealtimeOutputModality::Audio, + voice: RealtimeVoice::Marin, + }, + "rtc_frontend_owned", + HeaderMap::new(), + HeaderMap::new(), + RealtimeTranscriptState::default(), + ) + .await + .expect("attach observer"); + connection + .writer() + .close_transport() + .await + .expect("detach transport"); + connection.close().await.expect("already detached"); + tokio::time::timeout(Duration::from_secs(5), server) + .await + .expect("close was flushed before dropping connection") + .expect("server task"); + } +} + +#[tokio::test] +async fn realtime_ws_frameless_session_close_still_ends_session() { + let (addr, server) = spawn_realtime_ws_server(|mut ws| async move { + let session_close = ws + .next() + .await + .expect("session close") + .expect("valid frame"); + assert_eq!( + serde_json::from_str::(session_close.to_text().expect("text")).expect("json"), + json!({"type": "session.close"}) + ); + assert_eq!( + ws.next() + .await + .expect("websocket close") + .expect("valid close"), + Message::Close(None) + ); + }) + .await; + let client = test_client(test_provider(format!("http://{addr}"))) + .with_webrtc_sideband_base_url(format!("http://{addr}")); + let connection = client + .connect_existing_call_sideband( + RealtimeSessionConfig { + instructions: String::new(), + initial_items: Vec::new(), + delegation_ack_filler: None, + model: None, + session_id: None, + event_parser: RealtimeEventParser::FramelessBidi, + session_mode: RealtimeSessionMode::Conversational, + output_modality: RealtimeOutputModality::Audio, + voice: RealtimeVoice::Marin, + }, + "rtc_session_owner", + HeaderMap::new(), + HeaderMap::new(), + RealtimeTranscriptState::default(), + ) + .await + .expect("connect session"); + connection.close().await.expect("close session"); + tokio::time::timeout(Duration::from_secs(5), server) + .await + .expect("close flushed") + .expect("server task"); +} + #[tokio::test] async fn realtime_ws_e2e_send_while_next_event_waits() { let (addr, server) = spawn_realtime_ws_server(|mut ws: RealtimeWsStream| async move { diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index d44a19b5de..b689988cfd 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -818,6 +818,7 @@ impl RealtimeConversationManager { { let mut guard = self.state.lock().await; if let Some(state) = guard.conversation.as_mut() + // Only attach the task to the conversation currently stored in the manager. && Arc::ptr_eq(&state.realtime_active, realtime_active) { state.fanout_task = fanout_task.take(); @@ -2134,6 +2135,7 @@ enum RealtimeSidebandSessionInitialization { }, } +// The main loop for realtime inputs, including user audio/text and model commentary/results. async fn run_realtime_input_task( input: RealtimeInputTask, pending_outbound: Option>, @@ -2154,6 +2156,7 @@ async fn run_realtime_input_task( let mut output_audio_state: Option = None; let mut response_create_queue = RealtimeResponseCreateQueue::default(); + // Send any previously unsent messages before entering the main event loop. if let Some(pending_outbound) = pending_outbound && let Err(err) = send_realtime_pending_outbound( &pending_outbound, diff --git a/codex-rs/core/src/realtime_conversation/sideband.rs b/codex-rs/core/src/realtime_conversation/sideband.rs index d0b835550b..e378592c5a 100644 --- a/codex-rs/core/src/realtime_conversation/sideband.rs +++ b/codex-rs/core/src/realtime_conversation/sideband.rs @@ -9,6 +9,7 @@ use codex_api::ApiError; use codex_api::RealtimeEvent; use codex_api::RealtimeEventParser; use codex_api::RealtimeTranscriptState; +use codex_api::RealtimeWebsocketConnection; use codex_api::map_api_error; use codex_login::default_client::default_headers; use http::StatusCode; @@ -17,12 +18,14 @@ use std::time::Duration; use tokio::task::JoinHandle; use tokio::time::Instant; use tokio::time::sleep; +use tokio::time::timeout; use tracing::info; use tracing::warn; const RECONNECT_BASE_DELAY: Duration = Duration::from_millis(200); const RECONNECT_MAX_DELAY: Duration = Duration::from_secs(5); const STABLE_CONNECTION_DURATION: Duration = Duration::from_secs(30); +const TRANSPORT_CLOSE_TIMEOUT: Duration = Duration::from_secs(5); pub(super) fn spawn_webrtc_sideband_input_task( input: RealtimeWebrtcSidebandInputTask, @@ -110,11 +113,14 @@ pub(super) fn spawn_webrtc_sideband_input_task( } }; + // Stop when the call is inactive or cancellation has been requested. if !realtime_active.load(Ordering::Relaxed) || stop_token.is_cancelled() { + close_sideband_transport(&connection).await; break; } let connected_at = Instant::now(); + // Run the realtime input loop. let exit = run_realtime_input_task( RealtimeInputTask { writer: connection.writer(), @@ -132,6 +138,12 @@ pub(super) fn spawn_webrtc_sideband_input_task( ) .await; + // Cancellation returns Terminal, so close the socket before checking for TransportLost. + if !realtime_active.load(Ordering::Relaxed) || stop_token.is_cancelled() { + close_sideband_transport(&connection).await; + break; + } + let RealtimeInputTaskExit::TransportLost { err, pending_outbound: failed_outbound, @@ -140,9 +152,7 @@ pub(super) fn spawn_webrtc_sideband_input_task( break; }; pending_outbound = failed_outbound; - if !realtime_active.load(Ordering::Relaxed) || stop_token.is_cancelled() { - break; - } + if event_parser != RealtimeEventParser::FramelessBidi || matches!( &err, @@ -170,11 +180,32 @@ pub(super) fn spawn_webrtc_sideband_input_task( } } + if let RealtimeSidebandSessionInitialization::ExistingCall { initial_connection } = + &mut session_initialization + && let Some(connection) = initial_connection.take() + { + // Cancellation may win before the task takes its already connected observer. + close_sideband_transport(&connection).await; + } let transcript_tail = transcript_state.take_transcript_tail().await; flush_realtime_transcript_tail(&transcript_tail_flush, &transcript_tail).await; }) } +async fn close_sideband_transport(connection: &RealtimeWebsocketConnection) { + // The session belongs to the frontend; writer.close() would also send SessionClose. + match timeout( + TRANSPORT_CLOSE_TIMEOUT, + connection.writer().close_transport(), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(_)) => warn!("failed to close realtime sideband transport"), + Err(_) => warn!("timed out closing realtime sideband transport"), + } +} + fn webrtc_sideband_session_ended(err: &ApiError) -> bool { matches!( err, diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 3d0dc16401..883842746c 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -76,6 +76,8 @@ 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 tokio_tungstenite::tungstenite::protocol::CloseFrame; +use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode; use uuid::Uuid; use wiremock::Match; use wiremock::Mock; @@ -1893,6 +1895,80 @@ async fn conversation_webrtc_start_uses_configured_call_base_url_for_avas() -> R Ok(()) } +#[test_case(RealtimeConversationVersion::V1; "preview")] +#[test_case(RealtimeConversationVersion::V3; "frameless")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn conversation_existing_call_stop_closes_only_observer_transport( + version: RealtimeConversationVersion, +) -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let realtime_ws_base_url = format!("ws://{}", listener.local_addr()?); + let observer = tokio::spawn(async move { + let (stream, _) = listener.accept().await?; + let mut websocket = tokio_tungstenite::accept_async(stream).await?; + // The first frame must be the transport close, never session initialization or SessionClose. + let frame = websocket.next().await.context("observer frame")??; + Ok::<_, anyhow::Error>(frame) + }); + let mut builder = test_codex().with_config(move |config| { + config.experimental_realtime_ws_base_url = Some(realtime_ws_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: + codex_protocol::protocol::CodexResponseHandoffMode::Thinking, + backend_reasoning_status: false, + 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, + voice_session_id: None, + transport: Some(ConversationStartTransport::ExistingCall { + call_id: "rtc_frontend_owned".to_string(), + sideband_base_url: None, + }), + version: Some(version), + voice: None, + })) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::RealtimeConversationStarted(_)) + }) + .await; + test.codex.submit(Op::RealtimeConversationClose).await?; + let closed = wait_for_event_match(&test.codex, |event| match event { + EventMsg::RealtimeConversationClosed(closed) => Some(closed.clone()), + _ => None, + }) + .await; + assert_eq!(closed.reason.as_deref(), Some("requested")); + assert_eq!( + timeout(Duration::from_secs(5), observer) + .await + .context("observer close timed out")???, + Message::Close(Some(CloseFrame { + code: CloseCode::Normal, + reason: "".into() + })) + ); + test.codex.shutdown_and_wait().await?; + 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(()));