Support caller-owned sideband channels for realtime attachments (#58953)

## What changed

Add `RealtimeSidebandTransport` and `Op::RealtimeConversationAttachStream` so callers can attach Core to an existing realtime call through bounded JSON channels without opening or reconnecting a socket.

Read events independently of outbound backpressure, enforce a 1 MiB message limit, and drain accepted input on detach or channel closure before flushing the transcript. Pending writes are never replayed.

## Testing

Add transport and Core integration tests covering delegation without a callback socket, stalled output, attachment replacement and detach, channel closure, oversized messages, and preservation of queued transcripts in history.

GitOrigin-RevId: 06b7d1dfe2388054438d72d925e510a959dfe106

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<WsCommand>,
-    pump_task: tokio::task::JoinHandle<()>,
+    transport_task: tokio::task::JoinHandle<()>,
+    sideband_input_stop: Option<tokio::sync::watch::Sender<bool>>,
     terminal_policy_error: tokio::sync::watch::Receiver<Option<NetworkPolicyDenied>>,
 }
 
@@ -90,7 +92,7 @@ impl WsStream {
         let (tx_message, rx_message) = async_channel::unbounded::<Result<Message, WsError>>();
 
         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();
     }
 }
 

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::<WsCommand>(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;

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<String>,
+    mpsc::Receiver<String>,
+) {
+    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());
+}

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<codex_protocol::protocol::RealtimeSidebandTransport>,
     api_provider: ApiProvider,
     http_client_factory: HttpClientFactory,
     realtime_sideband_base_url: Option<String>,
@@ -662,6 +664,7 @@ impl RealtimeConversationManager {
     ) -> CodexResult<RealtimeStartOutput> {
         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<codex_protocol::protocol::RealtimeSidebandTransport>,
     api_provider: ApiProvider,
     http_client_factory: HttpClientFactory,
     realtime_sideband_base_url: Option<String>,
@@ -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<Session>,
+    sub_id: String,
+    params: ConversationStartParams,
+    transport: codex_protocol::protocol::RealtimeSidebandTransport,
+) -> CodexResult<()> {
+    if !matches!(
+        &params.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<Session>, expected: Option<&str>, stop_sub_id: &str) {
     let current = {
         let mut state = sess.conversation.state.lock().await;

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;
+    })
+}

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,

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<bool>,
     messages: mpsc::UnboundedReceiver<Value>,

codex-rs/protocol/src/protocol.rs

@@ -222,6 +222,15 @@ pub struct ConversationStartParams {
     pub voice: Option<RealtimeVoice>,
 }
 
+/// 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<String>,
+    pub outgoing: tokio::sync::mpsc::Sender<String>,
+}
+
 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<CodexResult<()>>,
     },
 
+    /// Attach an existing call using caller-provided I/O without opening a socket.
+    RealtimeConversationAttachStream {
+        params: ConversationStartParams,
+        transport: RealtimeSidebandTransport,
+        reply: oneshot::Sender<CodexResult<()>>,
+    },
+
     /// 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",