From e7e6267ab3e084e2c3d9647acbd83952e4e687ad Mon Sep 17 00:00:00 2001 From: kmeelu-oai Date: Mon, 4 May 2026 15:28:14 -0700 Subject: [PATCH] Make realtime sideband startup async (#20715) ## Summary Moves the WebRTC realtime sideband websocket join out of the voice start critical path. Call creation still posts the SDP offer and session config synchronously so the client gets the SDP answer, but the sideband websocket now connects in the input task async and doesn't block conversation state installation. This lets the normal realtime input channels buffer text, handoff output, and audio while the WebRTC sideband websocket is connecting. If the sideband join fails while the conversation is still active, the task sends a RealtimeEvent::Error through the existing events_tx / fanout path. To rephrase this: * No longer blocked on sideband: the client can receive the SDP answer earlier, set up the WebRTC peer connection, and let the media leg progress while the sideband websocket joins. * Still blocked on sideband: queued text, handoff output, and sideband server events cannot flow until connect_webrtc_sideband(...).await finishes and then run_realtime_input_task(...) starts ## Validation - `env CODEX_SKIP_VENDORED_BWRAP=1 cargo test --manifest-path codex-rs/Cargo.toml -p codex-core --test all conversation_webrtc_start_posts_generated_session` `CODEX_SKIP_VENDORED_BWRAP=1` is needed in this local environment because `libcap.pc` is not installed for the vendored bubblewrap build. ## Testing I tested this locally by running `cargo run -p codex-cli --bin codex -- --enable realtime_conversation` and invoking `/realtime`. Then, we get logs emitted in `~/.codex/log/codex-tui.log`. ### Before the Change Logging commit (https://github.com/openai/codex/commit/c0299e6edf1222fa0c43c1796e4811976c26fecd) ``` 2026-05-04T16:06:09.251956Z INFO session_loop{thread_id=019df3b9-e3d8-7271-b13a-b880119aa4c2}:submission_dispatch{otel.name="op.dispatch.realtime_conversation_start" submission.id="019df3bd-65df-7ee2-8125-1d6701fe39d2" codex.op="realtime_conversation_start"}: codex_core::realtime_conversation: starting realtime conversation 2026-05-04T16:06:09.251980Z INFO session_loop{thread_id=019df3b9-e3d8-7271-b13a-b880119aa4c2}:submission_dispatch{otel.name="op.dispatch.realtime_conversation_start" submission.id="019df3bd-65df-7ee2-8125-1d6701fe39d2" codex.op="realtime_conversation_start"}: codex_core::realtime_conversation: creating realtime call transport="webrtc" 2026-05-04T16:06:10.365722Z INFO session_loop{thread_id=019df3b9-e3d8-7271-b13a-b880119aa4c2}:submission_dispatch{otel.name="op.dispatch.realtime_conversation_start" submission.id="019df3bd-65df-7ee2-8125-1d6701fe39d2" codex.op="realtime_conversation_start"}: codex_core::realtime_conversation: realtime call created; sdp answer ready transport="webrtc" call_id=rtc_u0_Dbq65nhak5eLjQZ73yhAy elapsed_ms=1113 total_elapsed_ms=1113 2026-05-04T16:06:10.365843Z INFO session_loop{thread_id=019df3b9-e3d8-7271-b13a-b880119aa4c2}:submission_dispatch{otel.name="op.dispatch.realtime_conversation_start" submission.id="019df3bd-65df-7ee2-8125-1d6701fe39d2" codex.op="realtime_conversation_start"}: codex_core::realtime_conversation: connecting realtime sideband websocket call_id=rtc_u0_Dbq65nhak5eLjQZ73yhAy 2026-05-04T16:06:10.784528Z INFO session_loop{thread_id=019df3b9-e3d8-7271-b13a-b880119aa4c2}:submission_dispatch{otel.name="op.dispatch.realtime_conversation_start" submission.id="019df3bd-65df-7ee2-8125-1d6701fe39d2" codex.op="realtime_conversation_start"}: codex_core::realtime_conversation: connected realtime sideband websocket call_id=rtc_u0_Dbq65nhak5eLjQZ73yhAy elapsed_ms=418 total_elapsed_ms=1532 2026-05-04T16:06:10.784665Z INFO session_loop{thread_id=019df3b9-e3d8-7271-b13a-b880119aa4c2}:submission_dispatch{otel.name="op.dispatch.realtime_conversation_start" submission.id="019df3bd-65df-7ee2-8125-1d6701fe39d2" codex.op="realtime_conversation_start"}: codex_core::realtime_conversation: realtime conversation started ``` ### After the Change Logging commit (https://github.com/openai/codex/commit/c8b00ac21adf4f8dd1fe3a81403a2bb6183fe13b) ``` 2026-05-04T15:41:24.080363Z INFO ... codex_core::realtime_conversation: starting realtime conversation 2026-05-04T15:41:24.080434Z INFO ... codex_core::realtime_conversation: creating realtime call transport="webrtc" 2026-05-04T15:41:25.106906Z INFO ... codex_core::realtime_conversation: realtime call created; sdp answer ready transport="webrtc" call_id=rtc_u0_Dbpi8nhak5eLjQZ73yhAy elapsed_ms=1026 total_elapsed_ms=1026 2026-05-04T15:41:25.107067Z INFO ... codex_core::realtime_conversation: spawned realtime sideband connection task transport="webrtc" total_elapsed_ms=1026 2026-05-04T15:41:25.107160Z INFO ... codex_core::realtime_conversation: realtime conversation started 2026-05-04T15:41:25.107185Z INFO codex_core::realtime_conversation: connecting realtime sideband websocket call_id=rtc_u0_Dbpi8nhak5eLjQZ73yhAy 2026-05-04T15:41:25.107352Z INFO ... codex_core::realtime_conversation: sent realtime sdp answer to client 2026-05-04T15:41:26.076685Z INFO codex_core::realtime_conversation: connected realtime sideband websocket call_id=rtc_u0_Dbpi8nhak5eLjQZ73yhAy elapsed_ms=969 total_elapsed_ms=1996 2026-05-04T15:41:26.573893Z INFO codex_core::realtime_conversation: realtime session updated realtime_session_id=sess_u0_Dbpi8nhak5eLjQZ73yhAy 2026-05-04T15:41:26.573970Z INFO codex_core::realtime_conversation: received realtime conversation event event=SessionUpdated { ... } ``` ### Conclusion Here we see that we saved about a half a second in conversation startup (1532ms -> 969ms). This also checks out with my sanity tests; I was seeing at most a second of saving. --------- Co-authored-by: Codex --- .../tests/suite/v2/realtime_conversation.rs | 6 +- codex-rs/core/src/realtime_conversation.rs | 265 ++++++++++++------ .../core/tests/suite/realtime_conversation.rs | 228 ++++++++++++++- 3 files changed, 398 insertions(+), 101 deletions(-) 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() {