openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-29 19:08:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Detach realtime sideband observers without ending the session (#57817)

Browse Source 
## 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
This commit is contained in:
Jiahao
2026-10-28 20:10:15 +00:00
committed by copyberry
parent c92700d299
commit ef186f812a
8 changed files with 253 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/src/protocol/common.rs
+5 -2
View File
@@ -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)) => {
codex-rs/app-server/src/message_processor.rs
+2 -1
View File
@@ -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, .. } => {
codex-rs/app-server/src/request_processors/turn_processor.rs
+5 -3
View File
@@ -271,8 +271,9 @@ impl TurnRequestProcessor {
&self,
request_id: &ConnectionRequestId,
params: ThreadRealtimeStartParams,
sideband_base_url: Option<String>,
) -> Result<Option<ClientResponsePayload>, 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<String>,
) -> Result<Option<ThreadRealtimeStartResponse>, JSONRPCErrorError> {
let attaches_existing_call = matches!(
&params.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,
}
}
}),
codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs
+23 -5
View File
@@ -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<Result<(), WsError>>,
},
Close {
frame: Option<CloseFrame>,
tx_result: oneshot::Sender<Result<(), WsError>>,
},
}
@@ -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<CloseFrame>) -> 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 {
codex-rs/codex-api/tests/realtime_websocket_e2e.rs
+105
View File
@@ -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<tokio::net::TcpStream>;
@@ -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::<Value>(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 {
codex-rs/core/src/realtime_conversation.rs
+3
View File
@@ -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<Box<RealtimePendingOutbound>>,
@@ -2154,6 +2156,7 @@ async fn run_realtime_input_task(
let mut output_audio_state: Option<OutputAudioState> = 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,
codex-rs/core/src/realtime_conversation/sideband.rs
+34 -3
View File
@@ -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,
codex-rs/core/tests/suite/realtime_conversation.rs
+76
View File
@@ -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(()));