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 e7a8bf1bd4..9faa75738f 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -63,11 +63,13 @@ use tungstenite::protocol::frame::coding::CloseCode; use url::Url; const REALTIME_WIRE_LOG_TARGET: &str = "codex_api::realtime_websocket::wire"; +mod sideband; const OPENAI_REALTIME_API_BASE_URL: &str = "https://api.openai.com/v1"; struct WsStream { tx_command: mpsc::Sender, - pump_task: tokio::task::JoinHandle<()>, + transport_task: tokio::task::JoinHandle<()>, + sideband_input_stop: Option>, terminal_policy_error: tokio::sync::watch::Receiver>, } @@ -90,7 +92,7 @@ impl WsStream { let (tx_message, rx_message) = async_channel::unbounded::>(); let (terminal_policy_tx, terminal_policy_error) = tokio::sync::watch::channel(None); - let pump_task = tokio::spawn(async move { + let transport_task = tokio::spawn(async move { let mut inner = inner; let report_write_policy_error = |error: &WsError| { if let Some(denied) = codex_websocket_client::network_policy_denial(error) { @@ -191,7 +193,8 @@ impl WsStream { ( Self { tx_command, - pump_task, + transport_task, + sideband_input_stop: None, terminal_policy_error, }, rx_message, @@ -238,7 +241,7 @@ impl WsStream { impl Drop for WsStream { fn drop(&mut self) { - self.pump_task.abort(); + self.transport_task.abort(); } } diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods/sideband.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods/sideband.rs new file mode 100644 index 0000000000..12c448aaef --- /dev/null +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods/sideband.rs @@ -0,0 +1,158 @@ +//! Adapts caller-owned, bounded JSON channels to the existing realtime parser and writer. +//! +//! ```text +//! Caller -- incoming JSON --> read loop -- Message::Text --> Core event parser +//! Caller <-- outgoing JSON -- write loop <-- WsCommand ----- Core writer +//! ``` +//! +//! Each channel item carries one JSON message of up to 1 MiB. The loops run +//! independently so Core can still receive events while a command waits for +//! space in the caller's queue. Shutdown seals incoming messages and drains accepted +//! input before closing both directions, including a frame awaiting parser capacity. + +use super::*; +use codex_protocol::protocol::RealtimeSidebandTransport; + +const MAX_MESSAGE_BYTES: usize = 1024 * 1024; + +impl RealtimeWebsocketConnection { + /// Uses an already authenticated, caller-owned stream without opening or reconnecting a socket. + pub fn from_sideband( + transport: RealtimeSidebandTransport, + event_parser: RealtimeEventParser, + ) -> Self { + let RealtimeSidebandTransport { + mut incoming, + outgoing, + } = transport; + + let (tx_command, mut rx_command) = mpsc::channel::(32); + let (tx_message, rx_message) = async_channel::bounded(32); + let (_, terminal_policy_error) = tokio::sync::watch::channel(None); + + let (input_stop, mut input_stopped) = tokio::sync::watch::channel(false); + let transport_input_stop = input_stop.clone(); + let transport_task = tokio::spawn(async move { + let read = async { + let mut is_draining = false; + loop { + let text = if is_draining { + // A reserved but unsent permit must not hold shutdown open. + // try_recv() will drain messages already queued and exit when emptied. + incoming.try_recv().ok() + } else { + tokio::select! { + biased; + // Core requested shutdown, or the transport's write loop ended. + _ = input_stopped.changed() => { + incoming.close(); + is_draining = true; + continue; + } + // The caller dropped its receiver for Core commands. + // Detect this even when the writer is idle, then drain accepted input. + _ = outgoing.closed() => { + incoming.close(); + is_draining = true; + continue; + } + text = incoming.recv() => text, + } + }; + let Some(text) = text else { + break; + }; + if text.len() > MAX_MESSAGE_BYTES { + break; + } + if tx_message + .send(Ok(Message::Text(text.into()))) + .await + .is_err() + { + break; + } + } + }; + + let write = async { + while let Some(command) = rx_command.recv().await { + // These commands are still called WsCommand for legacy reasons. + match command { + WsCommand::Send { + message: Message::Text(text), + tx_result, + } => { + let result = if text.len() > MAX_MESSAGE_BYTES { + Err(WsError::Io(std::io::Error::other( + "sideband message too large", + ))) + } else { + outgoing + .send(text.to_string()) + .await + .map_err(|_| WsError::ConnectionClosed) + }; + let failed = result.is_err(); + // Tell Core's writer whether the caller's channel accepted the command. + let _ = tx_result.send(result); + if failed { + break; + } + } + WsCommand::Close { + frame: _, + tx_result, + } => { + let _ = tx_result.send(Ok(())); + break; + } + WsCommand::Send { tx_result, .. } => { + let _ = tx_result.send(Err(WsError::Io(std::io::Error::other( + "sideband requires JSON text", + )))); + break; + } + } + } + }; + + tokio::pin!(read); + tokio::select! { + _ = &mut read => {}, + _ = write => { + transport_input_stop.send_replace(true); + // Preserve the read future: it may already own a frame that + // has not yet fit into the parser queue. + read.await; + }, + } + }); + + Self::new( + WsStream { + tx_command, + transport_task, + sideband_input_stop: Some(input_stop), + terminal_policy_error, + }, + rx_message, + event_parser, + RealtimeTranscriptState::default(), + ) + } +} + +impl RealtimeWebsocketWriter { + /// Seal caller-owned input so its accepted events can drain without waiting for producers. + /// Socket-backed connections are unaffected. + pub fn close_sideband_input(&self) { + if let Some(input_stop) = &self.stream.sideband_input_stop { + input_stop.send_replace(true); + } + } +} + +#[cfg(test)] +#[path = "sideband_tests.rs"] +mod tests; diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods/sideband_tests.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods/sideband_tests.rs new file mode 100644 index 0000000000..04021ee942 --- /dev/null +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods/sideband_tests.rs @@ -0,0 +1,210 @@ +#![cfg(test)] + +use super::*; +use pretty_assertions::assert_eq; +use std::time::Duration; +use tokio::time::timeout; + +fn connection() -> ( + RealtimeWebsocketConnection, + mpsc::Sender, + mpsc::Receiver, +) { + let (events, incoming) = mpsc::channel(1); + let (outgoing, commands) = mpsc::channel(1); + ( + RealtimeWebsocketConnection::from_sideband( + RealtimeSidebandTransport { incoming, outgoing }, + RealtimeEventParser::FramelessBidi, + ), + events, + commands, + ) +} + +/// A blocked command receiver must not prevent receiving a realtime event on the other half. +#[tokio::test] +async fn stalled_output_does_not_block_input() { + let (connection, events, commands) = connection(); + let writer = connection.writer(); + writer.send_payload("first".into()).await.unwrap(); + let blocked = tokio::spawn(async move { writer.send_payload("second".into()).await }); + events + .send(r#"{"type":"session.started","session":{"id":"call"}}"#.into()) + .await + .unwrap(); + assert!( + timeout(Duration::from_secs(1), connection.next_event()) + .await + .unwrap() + .unwrap() + .is_some() + ); + assert!(!blocked.is_finished()); + drop(commands); + assert!( + timeout(Duration::from_secs(1), blocked) + .await + .unwrap() + .unwrap() + .is_err() + ); + assert!( + timeout(Duration::from_secs(1), connection.next_event()) + .await + .unwrap() + .is_err() + ); +} + +/// Ending the input stream closes the output too and never attempts reconnection. +#[tokio::test] +async fn input_half_close_ends_attachment() { + let (connection, events, mut commands) = connection(); + drop(events); + assert!( + timeout(Duration::from_secs(1), connection.next_event()) + .await + .unwrap() + .is_err() + ); + assert_eq!( + timeout(Duration::from_secs(1), commands.recv()) + .await + .unwrap(), + None + ); + assert!( + connection + .writer() + .send_payload("late".into()) + .await + .is_err() + ); +} + +/// A caller dropping its response half releases an idle transport without waiting for another command. +#[tokio::test] +async fn response_half_close_ends_idle_attachment() { + let (_connection, events, commands) = connection(); + drop(commands); + timeout(Duration::from_secs(1), events.closed()) + .await + .unwrap(); +} + +/// Oversized payloads fail the transport before entering the parser or caller's outgoing queue. +#[tokio::test] +async fn oversized_payloads_close_both_directions() { + let (input_connection, events, mut commands) = connection(); + events + .send("x".repeat(MAX_MESSAGE_BYTES + 1)) + .await + .unwrap(); + assert!( + timeout(Duration::from_secs(1), input_connection.next_event()) + .await + .unwrap() + .is_err() + ); + assert_eq!( + timeout(Duration::from_secs(1), commands.recv()) + .await + .unwrap(), + None + ); + let (output_connection, _events, mut commands) = connection(); + assert!( + output_connection + .writer() + .send_payload("x".repeat(MAX_MESSAGE_BYTES + 1)) + .await + .is_err() + ); + assert_eq!( + timeout(Duration::from_secs(1), commands.recv()) + .await + .unwrap(), + None + ); +} + +/// A final transcript accepted before peer hangup must reach both the parser and transcript tail. +#[tokio::test(flavor = "current_thread")] +async fn queued_transcript_survives_peer_hangup() { + for _ in 0..64 { + let (connection, events, commands) = connection(); + events + .try_send(r#"{"type":"input_transcript.added","item":{"text":"last words"}}"#.into()) + .unwrap(); + drop(events); + drop(commands); + assert_eq!( + timeout(Duration::from_secs(1), connection.next_event()) + .await + .unwrap() + .unwrap(), + Some( + codex_protocol::protocol::RealtimeEvent::InputTranscriptDelta( + codex_protocol::protocol::RealtimeTranscriptDelta { + delta: "last words".into() + } + ) + ) + ); + assert_eq!( + connection.events().take_transcript_tail().await, + vec![codex_protocol::protocol::RealtimeTranscriptEntry { + role: "user".into(), + text: "last words".into() + }] + ); + } +} + +/// Explicit shutdown preserves a frame blocked on parser capacity and does not wait for an unsent permit. +#[tokio::test] +async fn input_stop_drains_backpressure_without_waiting_for_reserved_permit() { + let (events, incoming) = mpsc::channel(64); + let (outgoing, _commands) = mpsc::channel(1); + let connection = RealtimeWebsocketConnection::from_sideband( + RealtimeSidebandTransport { incoming, outgoing }, + RealtimeEventParser::FramelessBidi, + ); + let _permit = events.reserve().await.unwrap(); + for _ in 0..63 { + events + .try_send(r#"{"type":"input_transcript.added","item":{"text":"tail"}}"#.into()) + .unwrap(); + } + timeout(Duration::from_secs(1), async { + while events.capacity() < 33 { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + connection.writer().close_sideband_input(); + for _ in 0..63 { + assert_eq!( + timeout(Duration::from_secs(1), connection.next_event()) + .await + .unwrap() + .unwrap(), + Some( + codex_protocol::protocol::RealtimeEvent::InputTranscriptDelta( + codex_protocol::protocol::RealtimeTranscriptDelta { + delta: "tail".into() + } + ) + ) + ); + } + assert!( + timeout(Duration::from_secs(1), connection.next_event()) + .await + .unwrap() + .is_err() + ); + assert!(events.is_closed()); +} diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index ffa74f075d..6f26c1b228 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -88,6 +88,7 @@ use tracing::info; use tracing::warn; mod bem; +mod caller_sideband; mod existing_call; mod sideband; mod transcript_tail; @@ -545,6 +546,7 @@ struct ConversationState { } struct RealtimeStart { + sideband: Option, api_provider: ApiProvider, http_client_factory: HttpClientFactory, realtime_sideband_base_url: Option, @@ -662,6 +664,7 @@ impl RealtimeConversationManager { ) -> CodexResult { let session_id = start.session_config.session_id.clone(); let RealtimeStart { + sideband, api_provider, http_client_factory, realtime_sideband_base_url, @@ -729,7 +732,29 @@ impl RealtimeConversationManager { event_parser, tx: transcript_tail_tx, }; - let (task, sdp) = if let Some(sdp) = sdp { + // Use the sideband transport passed from outside. + let (task, sdp) = if let Some(transport) = sideband { + // Existing naming uses websocket, actual link is defined by the transport. + let connection = + codex_api::RealtimeWebsocketConnection::from_sideband(transport, event_parser); + + let task = caller_sideband::spawn_input_task( + RealtimeInputTask { + writer: connection.writer(), + events: connection.events(), + text_rx: input_channels.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, + stop_token: stop_token.clone(), + }, + transcript_tail_flush, + ); + (task, None) + } else if let Some(sdp) = sdp { let call = model_client .create_realtime_call_with_headers( sdp, @@ -1340,6 +1365,7 @@ pub(crate) async fn handle_start( } struct PreparedRealtimeConversationStart { + sideband: Option, api_provider: ApiProvider, http_client_factory: HttpClientFactory, realtime_sideband_base_url: Option, @@ -1500,6 +1526,7 @@ async fn prepare_realtime_start( extra_headers.insert(X_CODEX_TURN_METADATA_HEADER, metadata); } Ok(PreparedRealtimeConversationStart { + sideband: None, api_provider, http_client_factory, realtime_sideband_base_url, @@ -1725,6 +1752,7 @@ async fn handle_start_inner( prepared_start: PreparedRealtimeConversationStart, ) -> CodexResult<()> { let PreparedRealtimeConversationStart { + sideband, api_provider, http_client_factory, realtime_sideband_base_url, @@ -1756,6 +1784,7 @@ async fn handle_start_inner( }; let event_parser = session_config.event_parser; let start = RealtimeStart { + sideband, api_provider, http_client_factory, realtime_sideband_base_url, @@ -2090,6 +2119,28 @@ pub(crate) async fn handle_attach( handle_start_inner(sess, &sub_id, prepared).await } +/// Attaches Core to an existing call using the caller’s channels to receive +/// events and send commands. +pub(crate) async fn handle_attach_stream( + sess: &Arc, + sub_id: String, + params: ConversationStartParams, + transport: codex_protocol::protocol::RealtimeSidebandTransport, +) -> CodexResult<()> { + if !matches!( + ¶ms.transport, + Some(ConversationStartTransport::ExistingCall { .. }) + ) { + return Err(CodexErr::InvalidRequest( + "sideband stream requires an existing call".into(), + )); + } + + let mut prepared = prepare_realtime_start(sess, params).await?; + prepared.sideband = Some(transport); + handle_start_inner(sess, &sub_id, prepared).await +} + pub(crate) async fn handle_detach(sess: &Arc, expected: Option<&str>, stop_sub_id: &str) { let current = { let mut state = sess.conversation.state.lock().await; diff --git a/codex-rs/core/src/realtime_conversation/caller_sideband.rs b/codex-rs/core/src/realtime_conversation/caller_sideband.rs new file mode 100644 index 0000000000..956b037b0b --- /dev/null +++ b/codex-rs/core/src/realtime_conversation/caller_sideband.rs @@ -0,0 +1,114 @@ +//! Drives caller-owned existing calls without coupling inbound events to outbound backpressure. +//! Detach stops writes and drains accepted inbound events before flushing the transcript. +//! Pending writes are never replayed. +//! Existing calls support v1/v3 only, so inbound handling does not issue v2 response-control writes. + +use super::RealtimeInputTask; +use super::RealtimeInputTaskExit; +use super::RealtimeResponseCreateQueue; +use super::RealtimeSessionKind; +use super::RealtimeTranscriptTailFlush; +use super::classify_realtime_input_error; +use super::flush_realtime_transcript_tail; +use super::handle_handoff_output; +use super::handle_realtime_server_event; +use super::handle_text_input; +use super::handle_user_audio_input; +use super::realtime_transcript_delta; +use super::report_realtime_transport_loss; +use crate::context::RealtimeDelegation; +use tokio::task::JoinHandle; + +pub(super) fn spawn_input_task( + input: RealtimeInputTask, + transcript_tail_flush: RealtimeTranscriptTailFlush, +) -> JoinHandle<()> { + tokio::spawn(async move { + let RealtimeInputTask { + writer, + events, + text_rx, + handoff_output_rx, + audio_rx, + events_tx, + handoff_state, + session_kind, + event_parser, + stop_token, + } = input; + debug_assert!(matches!(session_kind, RealtimeSessionKind::V1)); + let transcript_state = events.transcript_state(); + + let read = async { + let mut output_audio_state = None; + let mut response_create_queue = RealtimeResponseCreateQueue::default(); + loop { + let Some(event) = events.next_event().await? else { + return anyhow::Ok(()); + }; + let completed = events + .take_completed_transcript(&event, RealtimeDelegation::FIELD_BYTES) + .await; + handle_realtime_server_event( + (event, realtime_transcript_delta(&completed)), + &writer, + &events_tx, + &handoff_state, + session_kind, + &mut output_audio_state, + &mut response_create_queue, + ) + .await?; + } + }; + let write = async { + let mut response_create_queue = RealtimeResponseCreateQueue::default(); + loop { + tokio::select! { + text = text_rx.recv() => handle_text_input(text, &writer).await?, + output = handoff_output_rx.recv() => { + handle_handoff_output( + output, + &writer, + &handoff_state, + event_parser, + &mut response_create_queue, + ).await?; + } + audio = audio_rx.recv() => handle_user_audio_input(audio, &writer).await?, + } + } + }; + + // Stop accepting input before draining it, and keep the read future alive + // even when it already owns an event awaiting delivery to the fanout task. + let result = { + tokio::pin!(read); + tokio::select! { + biased; + _ = stop_token.cancelled() => { + writer.close_sideband_input(); + let _ = read.await; + Ok(()) + }, + result = &mut read => result, + result = write => { + writer.close_sideband_input(); + let _ = read.await; + result + }, + } + }; + // Dropping the last writer aborts the transport pump without waiting for + // space in the host's outgoing queue (including for a close command). + drop(writer); + if let Err(err) = result + && let RealtimeInputTaskExit::TransportLost { err, .. } = + classify_realtime_input_error(err) + { + report_realtime_transport_loss(&events_tx, err).await; + } + let transcript_tail = transcript_state.take_transcript_tail().await; + flush_realtime_transcript_tail(&transcript_tail_flush, &transcript_tail).await; + }) +} diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index f4f6dcceb5..b139eb074c 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -546,6 +546,18 @@ pub(super) async fn submission_loop( let _ = reply.send(result); false } + Op::RealtimeConversationAttachStream { + params, + transport, + reply, + } => { + let result = crate::realtime_conversation::handle_attach_stream( + &sess, sub.id, params, transport, + ) + .await; + let _ = reply.send(result); + false + } Op::RealtimeConversationDetach { realtime_session_id, reply, diff --git a/codex-rs/core/tests/suite/realtime_attachment.rs b/codex-rs/core/tests/suite/realtime_attachment.rs index ff7b3738d3..38e63ac3fa 100644 --- a/codex-rs/core/tests/suite/realtime_attachment.rs +++ b/codex-rs/core/tests/suite/realtime_attachment.rs @@ -1,6 +1,7 @@ //! Exercises attachment races through Core submissions and real sideband sockets. use anyhow::Result; +use codex_api::RealtimeEvent; use codex_protocol::protocol::ConversationStartParams; use codex_protocol::protocol::ConversationStartTransport; use codex_protocol::protocol::ConversationTextParams; @@ -9,11 +10,13 @@ use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::Op; use codex_protocol::protocol::RealtimeConversationVersion; use codex_protocol::protocol::RealtimeOutputModality; +use codex_protocol::protocol::RealtimeSidebandTransport; use codex_protocol::protocol::ThreadHistoryMode; use core_test_support::responses::ResponsesRequest; use core_test_support::responses::start_mock_server; use core_test_support::test_codex::TestCodex; use core_test_support::test_codex::test_codex; +use core_test_support::wait_for_event; use futures::SinkExt; use futures::StreamExt; use pretty_assertions::assert_eq; @@ -32,6 +35,202 @@ struct Sideband { task: tokio::task::JoinHandle<()>, } +/// A caller-owned stream drives a real Core delegation and returns context without any callback socket. +#[tokio::test] +async fn initiated_sideband_delegates_without_callback() -> Result<()> { + let api = start_mock_server().await; + let response = core_test_support::responses::mount_sse_once( + &api, + core_test_support::responses::sse(vec![ + core_test_support::responses::ev_response_created("stream-handoff"), + core_test_support::responses::ev_assistant_message( + "answer", + "Stream delegation completed", + ), + core_test_support::responses::ev_completed("stream-handoff"), + ]), + ) + .await; + let test = test_codex() + .with_history_mode(ThreadHistoryMode::Paginated) + .with_model_info_override("gpt-5.5", |model| model.use_responses_lite = false) + .with_config(|config| config.base_instructions = Some(String::new())) + .build_with_auto_env(&api) + .await?; + // Any regression to the existing-call WebSocket path is both observed and fails attachment. + let callback = TcpListener::bind("127.0.0.1:0").await?; + let callback_url = format!("ws://{}", callback.local_addr()?); + let (events, incoming) = mpsc::channel(32); + let (outgoing, mut commands) = mpsc::channel(32); + let (reply, ready) = oneshot::channel(); + test.codex + .submit(Op::RealtimeConversationAttachStream { + params: params("stream", &callback_url), + transport: RealtimeSidebandTransport { incoming, outgoing }, + reply, + }) + .await?; + timeout(Duration::from_secs(10), ready).await???; + events + .send( + json!({ + "type": "delegation.created", + "offset_ms": 0, + "item": { + "id": "stream-handoff", + "type": "delegation", + "target": "client", + "content": [{ + "type": "input_text", + "text": "complete the stream delegation" + }] + } + }) + .to_string(), + ) + .await?; + timeout(Duration::from_secs(20), async { + while let Some(command) = commands.recv().await { + let command: Value = serde_json::from_str(&command)?; + if command["type"] == "delegation.context.append" + && command.to_string().contains("Stream delegation completed") + { + assert_eq!(command["delegation_item_id"], "stream-handoff"); + return anyhow::Ok(()); + } + } + anyhow::bail!("Core closed before returning delegation context") + }) + .await??; + assert_eq!(response.requests().len(), 1); + assert!( + timeout(Duration::from_millis(100), callback.accept()) + .await + .is_err() + ); + drop(events); + timeout(Duration::from_secs(10), async { + while commands.recv().await.is_some() {} + }) + .await?; + test.codex.shutdown_and_wait().await?; + Ok(()) +} + +/// Inbound events, replacement, and detach stay live even when the host never drains output. +#[tokio::test] +async fn initiated_sideband_backpressure_keeps_events_and_lifecycle_live() -> Result<()> { + let api = start_mock_server().await; + let test = test_codex().build_with_auto_env(&api).await?; + let mut transports = Vec::new(); + for session in ["original", "replacement"] { + let (events, incoming) = mpsc::channel(/*buffer*/ 1); + let (outgoing, commands) = mpsc::channel(/*buffer*/ 1); + // Occupy the host's queue before attaching, and retain its receiver until after detach. + outgoing.send("occupied".to_owned()).await?; + let (reply, ready) = oneshot::channel(); + test.codex + .submit(Op::RealtimeConversationAttachStream { + params: params(session, "ws://unused.invalid"), + transport: RealtimeSidebandTransport { incoming, outgoing }, + reply, + }) + .await?; + timeout(Duration::from_secs(/*secs*/ 10), ready).await???; + + // Core's text queue holds 64 items. Accepting 65 submissions before the + // barrier below proves that the writer has dequeued text into a blocked send. + for _ in 0..65 { + test.codex + .submit(Op::RealtimeConversationText(ConversationTextParams { + text: "blocked output".to_owned(), + role: ConversationTextRole::User, + })) + .await?; + } + // A mismatched detach is an acknowledged no-op on the submission loop. + detach(&test, "not-selected").await?; + events + .send(json!({"type":"session.started", "session":{"id":session}}).to_string()) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::RealtimeConversationRealtime(event) + if matches!(&event.payload, RealtimeEvent::SessionUpdated { realtime_session_id, .. } + if realtime_session_id == session)) + }) + .await; + transports.push((events, commands)); + } + detach(&test, "replacement").await?; + for (events, mut commands) in transports { + timeout(Duration::from_secs(/*secs*/ 10), events.closed()).await?; + assert_eq!(commands.recv().await, Some("occupied".to_owned())); + assert_eq!(commands.recv().await, None); + } + test.codex.shutdown_and_wait().await?; + Ok(()) +} + +/// Detach is a history barrier for accepted final speech, including when both caller halves disappear. +#[tokio::test(flavor = "current_thread")] +async fn initiated_sideband_detach_preserves_queued_transcript_in_history() -> Result<()> { + let api = start_mock_server().await; + let test = test_codex().build_with_auto_env(&api).await?; + test.codex.ensure_rollout_materialized().await; + for index in 0..8 { + let session = format!("hangup-{index}"); + let completed_turn = index % 2 == 1; + let text = if completed_turn { + format!( + "{}final words before hangup {index}", + "completed words ".repeat(800) + ) + } else { + format!("final words before hangup {index}") + }; + let (events, incoming) = mpsc::channel(32); + let (outgoing, commands) = mpsc::channel(32); + let (reply, ready) = oneshot::channel(); + let mut start_params = params(&session, "ws://unused.invalid"); + // Completed turns must persist even when the optional unfinished-tail flush is off. + start_params.flush_transcript_tail_on_session_end = !completed_turn; + test.codex + .submit(Op::RealtimeConversationAttachStream { + params: start_params, + transport: RealtimeSidebandTransport { incoming, outgoing }, + reply, + }) + .await?; + timeout(Duration::from_secs(10), ready).await???; + let event = if completed_turn { + json!({"type": "turn.done", "turn": {"role": "user", "transcript": text}}) + } else { + json!({"type": "input_transcript.added", "item": {"text": text}}) + }; + events.try_send(event.to_string())?; + drop(events); + drop(commands); + detach(&test, &session).await?; + let history = test.codex.load_history(/*include_archived*/ false).await?; + let tails = history.items.iter().filter(|item| { + matches!(item, + codex_history::RolloutItem::ResponseItem(codex_history::ResponseItemEnvelope { + item: codex_protocol::models::ResponseItem::Message { role, content, .. }, .. + }) if role == "user" && content.iter().any(|item| matches!(item, + codex_protocol::models::ContentItem::InputText { text: recorded } + if recorded.contains(&text) + )) + ) + }).count(); + assert_eq!( + tails, 1, + "detach must persist the accepted final transcript exactly once" + ); + } + test.codex.shutdown_and_wait().await?; + Ok(()) +} + struct Connection { accept: oneshot::Sender, messages: mpsc::UnboundedReceiver, diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 4050afcc71..3ac83794bc 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -222,6 +222,15 @@ pub struct ConversationStartParams { pub voice: Option, } +/// A caller-owned realtime sideband. Dropping either direction seals input and ends +/// the attachment after accepted messages drain. The caller supplies bounded channels +/// and one complete JSON message per item. +#[derive(Debug)] +pub struct RealtimeSidebandTransport { + pub incoming: tokio::sync::mpsc::Receiver, + pub outgoing: tokio::sync::mpsc::Sender, +} + impl fmt::Debug for ConversationStartParams { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { // Text, instructions, and transport endpoints can contain credentials. @@ -720,6 +729,13 @@ pub enum Op { reply: oneshot::Sender>, }, + /// Attach an existing call using caller-provided I/O without opening a socket. + RealtimeConversationAttachStream { + params: ConversationStartParams, + transport: RealtimeSidebandTransport, + reply: oneshot::Sender>, + }, + /// Send audio input to the running realtime conversation stream. #[debug("RealtimeConversationAudio {{ item_id: {:?}, sample_rate: {}, num_channels: {}, samples_per_channel: {:?} }}", _0.frame.item_id, _0.frame.sample_rate, _0.frame.num_channels, _0.frame.samples_per_channel)] RealtimeConversationAudio(ConversationAudioParams), @@ -1094,6 +1110,7 @@ impl Op { Self::CleanBackgroundTerminals => "clean_background_terminals", Self::RealtimeConversationStart(_) => "realtime_conversation_start", Self::RealtimeConversationAttach { .. } => "realtime_conversation_attach", + Self::RealtimeConversationAttachStream { .. } => "realtime_conversation_attach_stream", Self::RealtimeConversationAudio(_) => "realtime_conversation_audio", Self::RealtimeConversationText(_) => "realtime_conversation_text", Self::RealtimeConversationSpeech(_) => "realtime_conversation_speech",