diff --git a/codex-rs/codex-api/src/endpoint/mod.rs b/codex-rs/codex-api/src/endpoint/mod.rs index 5d01a15fe3..ba8ed96e6c 100644 --- a/codex-rs/codex-api/src/endpoint/mod.rs +++ b/codex-rs/codex-api/src/endpoint/mod.rs @@ -20,6 +20,7 @@ pub use realtime_websocket::RealtimeEventParser; pub use realtime_websocket::RealtimeOutputModality; pub use realtime_websocket::RealtimeSessionConfig; pub use realtime_websocket::RealtimeSessionMode; +pub use realtime_websocket::RealtimeTranscriptState; pub use realtime_websocket::RealtimeWebsocketClient; pub use realtime_websocket::RealtimeWebsocketConnection; pub use realtime_websocket::RealtimeWebsocketEvents; 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 ebd79efd57..4a7bd31b7c 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -31,6 +31,7 @@ use futures::SinkExt; use futures::StreamExt; use http::HeaderMap; use http::HeaderValue; +use http::StatusCode; use std::collections::HashMap; use std::collections::VecDeque; use std::sync::Arc; @@ -52,10 +53,13 @@ use tracing::info; use tracing::trace; use tracing::warn; use tungstenite::protocol::WebSocketConfig; +use tungstenite::protocol::frame::coding::CloseCode; use url::Url; const REALTIME_WIRE_LOG_TARGET: &str = "codex_api::realtime_websocket::wire"; const OPENAI_REALTIME_API_BASE_URL: &str = "https://api.openai.com/v1"; +const MAX_ACTIVE_TRANSCRIPT_BYTES: usize = 8 * 1024; +const TRUNCATED_TRANSCRIPT_PREFIX: &str = "…"; struct WsStream { tx_command: mpsc::Sender, @@ -221,15 +225,19 @@ pub struct RealtimeWebsocketWriter { pub struct RealtimeWebsocketEvents { rx_message: async_channel::Receiver>, pending_events: Arc>>, - active_transcript: Arc>, + transcript_state: RealtimeTranscriptState, event_parser: RealtimeEventParser, is_closed: Arc, } +#[derive(Clone, Default)] +pub struct RealtimeTranscriptState { + active: Arc>, +} + #[derive(Default)] struct ActiveTranscriptState { entries: Vec, - last_handoff_entry_count: usize, new_input_entry: bool, new_output_entry: bool, } @@ -277,6 +285,7 @@ impl RealtimeWebsocketConnection { stream: WsStream, rx_message: async_channel::Receiver>, event_parser: RealtimeEventParser, + transcript_state: RealtimeTranscriptState, ) -> Self { let stream = Arc::new(stream); let is_closed = Arc::new(AtomicBool::new(false)); @@ -290,7 +299,7 @@ impl RealtimeWebsocketConnection { events: RealtimeWebsocketEvents { rx_message, pending_events: Arc::new(Mutex::new(VecDeque::new())), - active_transcript: Arc::new(Mutex::new(ActiveTranscriptState::default())), + transcript_state, event_parser, is_closed, }, @@ -486,11 +495,12 @@ impl RealtimeWebsocketWriter { } impl RealtimeWebsocketEvents { + pub fn transcript_state(&self) -> RealtimeTranscriptState { + self.transcript_state.clone() + } + pub async fn take_transcript_tail(&self) -> Vec { - let mut active_transcript = self.active_transcript.lock().await; - let tail = active_transcript.entries[active_transcript.last_handoff_entry_count..].to_vec(); - active_transcript.last_handoff_entry_count = active_transcript.entries.len(); - tail + self.transcript_state.take_transcript_tail().await } pub async fn next_event(&self) -> Result, ApiError> { @@ -514,6 +524,12 @@ impl RealtimeWebsocketEvents { } Err(_) => { self.is_closed.store(true, Ordering::SeqCst); + if self.event_parser == RealtimeEventParser::FramelessBidi { + error!("realtime websocket event stream ended unexpectedly"); + return Err(ApiError::Stream( + "realtime websocket event stream ended unexpectedly".to_string(), + )); + } info!("realtime websocket event stream ended"); return Ok(None); } @@ -536,7 +552,20 @@ impl RealtimeWebsocketEvents { frame.as_ref().map(|frame| frame.code), frame.as_ref().map(|frame| frame.reason.as_str()) ); - return Ok(None); + if self.event_parser != RealtimeEventParser::FramelessBidi { + return Ok(None); + } + return match frame { + Some(frame) if frame.code == CloseCode::Normal => Ok(None), + Some(frame) => Err(ApiError::Stream(format!( + "realtime websocket closed unexpectedly: code={:?} reason={:?}", + frame.code, frame.reason + ))), + None => Err(ApiError::Stream( + "realtime websocket closed unexpectedly without a status code" + .to_string(), + )), + }; } Message::Binary(_) => { return Ok(Some(RealtimeEvent::Error( @@ -567,7 +596,7 @@ impl RealtimeWebsocketEvents { } async fn update_active_transcript(&self, event: &mut RealtimeEvent) { - let mut active_transcript = self.active_transcript.lock().await; + let mut active_transcript = self.transcript_state.active.lock().await; match event { RealtimeEvent::InputAudioSpeechStarted(_) => { active_transcript.new_input_entry = true; @@ -609,10 +638,7 @@ impl RealtimeWebsocketEvents { } RealtimeEvent::HandoffRequested(handoff) => { append_handoff_input(&mut active_transcript.entries, &handoff.input_transcript); - handoff.active_transcript = active_transcript.entries - [active_transcript.last_handoff_entry_count..] - .to_vec(); - active_transcript.last_handoff_entry_count = active_transcript.entries.len(); + handoff.active_transcript = std::mem::take(&mut active_transcript.entries); active_transcript.new_input_entry = true; active_transcript.new_output_entry = true; } @@ -628,9 +654,48 @@ impl RealtimeWebsocketEvents { | RealtimeEvent::ConversationItemAdded(_) | RealtimeEvent::Error(_) => {} } + truncate_active_transcript(&mut active_transcript.entries); } } +impl RealtimeTranscriptState { + pub async fn take_transcript_tail(&self) -> Vec { + std::mem::take(&mut self.active.lock().await.entries) + } +} + +fn truncate_active_transcript(entries: &mut Vec) { + let mut total_bytes = transcript_entries_bytes(entries); + while total_bytes > MAX_ACTIVE_TRANSCRIPT_BYTES && entries.len() > 1 { + total_bytes = total_bytes.saturating_sub(transcript_entry_bytes(&entries[0])); + entries.remove(0); + } + let Some(entry) = entries.first_mut() else { + return; + }; + let entry_overhead = entry.role.len() + 3; + let max_text_bytes = MAX_ACTIVE_TRANSCRIPT_BYTES.saturating_sub(entry_overhead); + if entry.text.len() <= max_text_bytes { + return; + } + let mut start = entry + .text + .len() + .saturating_sub(max_text_bytes.saturating_sub(TRUNCATED_TRANSCRIPT_PREFIX.len())); + while !entry.text.is_char_boundary(start) { + start += 1; + } + entry.text = format!("{TRUNCATED_TRANSCRIPT_PREFIX}{}", &entry.text[start..]); +} + +fn transcript_entries_bytes(entries: &[RealtimeTranscriptEntry]) -> usize { + entries.iter().map(transcript_entry_bytes).sum() +} + +fn transcript_entry_bytes(entry: &RealtimeTranscriptEntry) -> usize { + entry.role.len() + entry.text.len() + 3 +} + fn append_transcript_delta( entries: &mut Vec, role: &str, @@ -735,6 +800,7 @@ impl RealtimeWebsocketClient { extra_headers, default_headers, /*initialize_session*/ true, + RealtimeTranscriptState::default(), ) .await } @@ -745,6 +811,7 @@ impl RealtimeWebsocketClient { call_id: &str, extra_headers: HeaderMap, default_headers: HeaderMap, + transcript_state: RealtimeTranscriptState, ) -> Result { // The WebRTC call already exists; this loop only retries joining its sideband control // socket. Once joined, the returned connection is the same reader/writer state that the @@ -756,10 +823,12 @@ impl RealtimeWebsocketClient { call_id, extra_headers.clone(), default_headers.clone(), + transcript_state.clone(), ) .await; match result { Ok(connection) => return Ok(connection), + Err(err) if webrtc_sideband_session_ended(&err) => return Err(err), Err(err) if attempt < self.provider.retry.max_attempts => { let delay = backoff(self.provider.retry.base_delay, attempt + 1); warn!( @@ -785,6 +854,7 @@ impl RealtimeWebsocketClient { call_id: &str, extra_headers: HeaderMap, default_headers: HeaderMap, + transcript_state: RealtimeTranscriptState, ) -> Result { // Keep the parser/session query shaping from standalone realtime while replacing the model // query with a call_id join onto an existing WebRTC session. @@ -795,6 +865,7 @@ impl RealtimeWebsocketClient { extra_headers, default_headers, /*initialize_session*/ false, + transcript_state, ) .await } @@ -821,6 +892,7 @@ impl RealtimeWebsocketClient { extra_headers: HeaderMap, default_headers: HeaderMap, initialize_session: bool, + transcript_state: RealtimeTranscriptState, ) -> Result { ensure_rustls_crypto_provider(); @@ -848,7 +920,7 @@ impl RealtimeWebsocketClient { connector, ) .await - .map_err(|err| ApiError::Stream(format!("failed to connect realtime websocket: {err}")))?; + .map_err(map_realtime_websocket_connect_error)?; info!( ws_url = %ws_url, status = %response.status(), @@ -856,7 +928,12 @@ impl RealtimeWebsocketClient { ); let (stream, rx_message) = WsStream::new(stream); - let connection = RealtimeWebsocketConnection::new(stream, rx_message, config.event_parser); + let connection = RealtimeWebsocketConnection::new( + stream, + rx_message, + config.event_parser, + transcript_state, + ); if initialize_session || config.event_parser != RealtimeEventParser::FramelessBidi { debug!( session_id = config.session_id.as_deref().unwrap_or(""), @@ -881,6 +958,24 @@ impl RealtimeWebsocketClient { } } +fn webrtc_sideband_session_ended(err: &ApiError) -> bool { + matches!( + err, + ApiError::Api { status, .. } + if matches!(*status, StatusCode::NOT_FOUND | StatusCode::GONE) + ) +} + +fn map_realtime_websocket_connect_error(err: WsError) -> ApiError { + match err { + WsError::Http(response) => ApiError::Api { + status: response.status(), + message: "realtime websocket handshake failed".to_string(), + }, + err => ApiError::Stream(format!("failed to connect realtime websocket: {err}")), + } +} + fn merge_request_headers( provider_headers: &HeaderMap, extra_headers: HeaderMap, @@ -1047,6 +1142,7 @@ mod tests { use codex_protocol::protocol::RealtimeTranscriptDone; use codex_protocol::protocol::RealtimeVoice; use http::HeaderValue; + use http::StatusCode; use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; @@ -1055,6 +1151,7 @@ mod tests { use tokio::net::TcpListener; use tokio_tungstenite::accept_async; use tokio_tungstenite::tungstenite::Message; + use tungstenite::protocol::CloseFrame; #[test] fn parse_session_updated_event() { @@ -1152,7 +1249,7 @@ mod tests { let events = RealtimeWebsocketEvents { rx_message, pending_events: Arc::new(Mutex::new(VecDeque::new())), - active_transcript: Arc::new(Mutex::new(ActiveTranscriptState::default())), + transcript_state: RealtimeTranscriptState::default(), event_parser: RealtimeEventParser::V1, is_closed: Arc::new(AtomicBool::new(false)), }; @@ -1197,6 +1294,156 @@ mod tests { assert_eq!(events.take_transcript_tail().await, vec![]); } + #[tokio::test] + async fn transcript_state_survives_reconnect_and_reconciles_done() { + let transcript_state = RealtimeTranscriptState::default(); + let (_first_tx, first_rx) = async_channel::unbounded(); + let first_events = RealtimeWebsocketEvents { + rx_message: first_rx, + pending_events: Arc::new(Mutex::new(VecDeque::new())), + transcript_state: transcript_state.clone(), + event_parser: RealtimeEventParser::FramelessBidi, + is_closed: Arc::new(AtomicBool::new(false)), + }; + let mut partial = RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta { + delta: "hello wor".to_string(), + }); + first_events.update_active_transcript(&mut partial).await; + + let (_second_tx, second_rx) = async_channel::unbounded(); + let second_events = RealtimeWebsocketEvents { + rx_message: second_rx, + pending_events: Arc::new(Mutex::new(VecDeque::new())), + transcript_state, + event_parser: RealtimeEventParser::FramelessBidi, + is_closed: Arc::new(AtomicBool::new(false)), + }; + let mut done = RealtimeEvent::InputTranscriptDone(RealtimeTranscriptDone { + text: "hello world".to_string(), + }); + second_events.update_active_transcript(&mut done).await; + + assert_eq!( + second_events.take_transcript_tail().await, + vec![RealtimeTranscriptEntry { + role: "user".to_string(), + text: "hello world".to_string(), + }] + ); + assert_eq!(second_events.take_transcript_tail().await, vec![]); + } + + #[tokio::test] + async fn active_transcript_retains_a_bounded_suffix() { + let (_tx_message, rx_message) = async_channel::unbounded(); + let events = RealtimeWebsocketEvents { + rx_message, + pending_events: Arc::new(Mutex::new(VecDeque::new())), + transcript_state: RealtimeTranscriptState::default(), + event_parser: RealtimeEventParser::FramelessBidi, + is_closed: Arc::new(AtomicBool::new(false)), + }; + let mut oversized = RealtimeEvent::InputTranscriptDelta(RealtimeTranscriptDelta { + delta: format!("old{}new", "x".repeat(MAX_ACTIVE_TRANSCRIPT_BYTES)), + }); + events.update_active_transcript(&mut oversized).await; + + let tail = events.take_transcript_tail().await; + assert!(transcript_entries_bytes(&tail) <= MAX_ACTIVE_TRANSCRIPT_BYTES); + assert!(tail[0].text.starts_with(TRUNCATED_TRANSCRIPT_PREFIX)); + assert!(tail[0].text.ends_with("new")); + } + + #[test] + fn terminal_sideband_handshake_statuses_are_not_retryable() { + for status in [StatusCode::NOT_FOUND, StatusCode::GONE] { + assert!(webrtc_sideband_session_ended(&ApiError::Api { + status, + message: "session ended".to_string(), + })); + } + assert!(!webrtc_sideband_session_ended(&ApiError::Api { + status: StatusCode::INTERNAL_SERVER_ERROR, + message: "retry".to_string(), + })); + } + + enum TestRealtimeTermination { + Close(Option), + EventChannelEnd, + } + + async fn realtime_termination_is_transport_error( + event_parser: RealtimeEventParser, + termination: TestRealtimeTermination, + ) -> bool { + let (tx_message, rx_message) = async_channel::unbounded(); + let events = RealtimeWebsocketEvents { + rx_message, + pending_events: Arc::new(Mutex::new(VecDeque::new())), + transcript_state: RealtimeTranscriptState::default(), + event_parser, + is_closed: Arc::new(AtomicBool::new(false)), + }; + match termination { + TestRealtimeTermination::Close(close_code) => { + let frame = close_code.map(|code| CloseFrame { + code, + reason: "session ended".into(), + }); + tx_message + .send(Ok(Message::Close(frame))) + .await + .expect("send close frame"); + } + TestRealtimeTermination::EventChannelEnd => drop(tx_message), + } + events.next_event().await.is_err() + } + + #[tokio::test] + async fn websocket_termination_classification_preserves_protocol_behavior() { + use CloseCode::Away; + use CloseCode::Normal; + use RealtimeEventParser::FramelessBidi; + use RealtimeEventParser::RealtimeV2; + use RealtimeEventParser::V1; + use TestRealtimeTermination::Close; + use TestRealtimeTermination::EventChannelEnd; + + for (event_parser, termination, expect_transport_error) in [ + (FramelessBidi, Close(Some(Normal)), false), + (FramelessBidi, Close(Some(Away)), true), + (FramelessBidi, Close(None), true), + (FramelessBidi, EventChannelEnd, true), + (V1, Close(Some(Away)), false), + (V1, EventChannelEnd, false), + (RealtimeV2, Close(Some(Away)), false), + ] { + assert_eq!( + realtime_termination_is_transport_error(event_parser, termination).await, + expect_transport_error, + "unexpected termination classification for {event_parser:?}" + ); + } + } + + #[test] + fn websocket_handshake_error_preserves_http_status() { + let response = http::Response::builder() + .status(StatusCode::GONE) + .body(None) + .expect("build response"); + + assert!(matches!( + map_realtime_websocket_connect_error(WsError::Http(Box::new(response))), + ApiError::Api { + status: StatusCode::GONE, + .. + } + )); + } + #[test] fn parse_input_transcript_delta_event() { let payload = json!({ diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs index 6bb4808fae..e954e489a5 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs @@ -9,6 +9,7 @@ mod protocol_frameless_bidi; mod protocol_v1; mod protocol_v2; +pub use methods::RealtimeTranscriptState; pub use methods::RealtimeWebsocketClient; pub use methods::RealtimeWebsocketConnection; pub use methods::RealtimeWebsocketEvents; diff --git a/codex-rs/codex-api/src/lib.rs b/codex-rs/codex-api/src/lib.rs index 03055b75bb..83af9f0dc0 100644 --- a/codex-rs/codex-api/src/lib.rs +++ b/codex-rs/codex-api/src/lib.rs @@ -59,6 +59,7 @@ pub use crate::endpoint::RealtimeEventParser; pub use crate::endpoint::RealtimeOutputModality; pub use crate::endpoint::RealtimeSessionConfig; pub use crate::endpoint::RealtimeSessionMode; +pub use crate::endpoint::RealtimeTranscriptState; pub use crate::endpoint::RealtimeWebsocketClient; pub use crate::endpoint::RealtimeWebsocketConnection; pub use crate::endpoint::RealtimeWebsocketEvents; diff --git a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs index 1fd1c59878..696d0f5052 100644 --- a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs +++ b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs @@ -10,6 +10,7 @@ use codex_api::RealtimeEventParser; use codex_api::RealtimeOutputModality; use codex_api::RealtimeSessionConfig; use codex_api::RealtimeSessionMode; +use codex_api::RealtimeTranscriptState; use codex_api::RealtimeWebsocketClient; use codex_api::RetryConfig; use codex_protocol::protocol::RealtimeHandoffRequested; @@ -263,6 +264,7 @@ async fn realtime_ws_connect_webrtc_sideband_retries_join_until_server_is_availa "rtc_test", HeaderMap::new(), HeaderMap::new(), + RealtimeTranscriptState::default(), ) .await .expect("connect on retry"); diff --git a/codex-rs/core/src/context/realtime_delegation.rs b/codex-rs/core/src/context/realtime_delegation.rs index 98274df87c..ba3a92ce87 100644 --- a/codex-rs/core/src/context/realtime_delegation.rs +++ b/codex-rs/core/src/context/realtime_delegation.rs @@ -1,5 +1,8 @@ use super::ContextualUserFragment; +const MAX_REALTIME_DELEGATION_FIELD_BYTES: usize = 4 * 1024; +const TRUNCATION_MARKER: &str = "…"; + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum RealtimeDelegationSource { Handoff, @@ -41,7 +44,7 @@ impl ContextualUserFragment for RealtimeDelegation<'_> { } fn body(&self) -> String { - let input = escape_xml_text(self.input); + let input = escape_xml_text_bounded(self.input, Retain::Start); let source = match self.source { RealtimeDelegationSource::Handoff => "", RealtimeDelegationSource::TranscriptTailFlush => { @@ -49,7 +52,7 @@ impl ContextualUserFragment for RealtimeDelegation<'_> { } }; if let Some(transcript_delta) = self.transcript_delta.filter(|text| !text.is_empty()) { - let transcript_delta = escape_xml_text(transcript_delta); + let transcript_delta = escape_xml_text_bounded(transcript_delta, Retain::End); return format!( "\n{source} {input}\n {transcript_delta}\n" ); @@ -59,6 +62,36 @@ impl ContextualUserFragment for RealtimeDelegation<'_> { } } +#[derive(Clone, Copy)] +enum Retain { + Start, + End, +} + +fn escape_xml_text_bounded(input: &str, retain: Retain) -> String { + let escaped = escape_xml_text(input); + if escaped.len() <= MAX_REALTIME_DELEGATION_FIELD_BYTES { + return escaped; + } + let retained_bytes = MAX_REALTIME_DELEGATION_FIELD_BYTES - TRUNCATION_MARKER.len(); + match retain { + Retain::Start => { + let mut end = retained_bytes; + while !escaped.is_char_boundary(end) { + end -= 1; + } + format!("{}{TRUNCATION_MARKER}", &escaped[..end]) + } + Retain::End => { + let mut start = escaped.len() - retained_bytes; + while !escaped.is_char_boundary(start) { + start += 1; + } + format!("{TRUNCATION_MARKER}{}", &escaped[start..]) + } + } +} + fn escape_xml_text(input: &str) -> String { input .replace('&', "&") diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 325bd34394..f9c96bea3b 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -80,9 +80,11 @@ use tracing::info; use tracing::warn; mod bem; +mod sideband; use self::bem::ChannelParser as BemChannelParser; use self::bem::message_phase as bem_message_phase; +use self::sideband::spawn_webrtc_sideband_input_task; const AUDIO_IN_QUEUE_CAPACITY: usize = 256; const TEXT_IN_QUEUE_CAPACITY: usize = 64; @@ -317,7 +319,7 @@ fn take_last_bytes_at_char_boundary(text: &str, max_bytes: usize) -> &str { &text[start..] } -#[derive(Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq)] enum RealtimeOutbound { StandaloneHandoff { text: String, @@ -366,14 +368,13 @@ impl RealtimeResponseCreateQueue { async fn request_create( &mut self, writer: &RealtimeWebsocketWriter, - events_tx: &Sender, reason: &str, ) -> anyhow::Result<()> { if self.active_default_response { self.pending_create = true; return Ok(()); } - self.send_create_now(writer, events_tx, reason).await + self.send_create_now(writer, reason).await } fn mark_started(&mut self) { @@ -383,7 +384,6 @@ impl RealtimeResponseCreateQueue { async fn mark_finished( &mut self, writer: &RealtimeWebsocketWriter, - events_tx: &Sender, reason: &str, ) -> anyhow::Result<()> { self.active_default_response = false; @@ -391,27 +391,24 @@ impl RealtimeResponseCreateQueue { return Ok(()); } self.pending_create = false; - self.send_create_now(writer, events_tx, reason).await + self.send_create_now(writer, reason).await } async fn send_create_now( &mut self, writer: &RealtimeWebsocketWriter, - events_tx: &Sender, reason: &str, ) -> anyhow::Result<()> { if let Err(err) = writer.send_response_create().await { - let mapped_error = map_api_error(err); - let error_message = mapped_error.to_string(); - if error_message.starts_with(REALTIME_ACTIVE_RESPONSE_ERROR_PREFIX) { + if matches!(&err, ApiError::Stream(message) if message.starts_with(REALTIME_ACTIVE_RESPONSE_ERROR_PREFIX)) + { warn!("realtime response.create raced an active response; deferring"); self.active_default_response = true; self.pending_create = true; return Ok(()); } - warn!("failed to send {reason} response.create: {mapped_error}"); - let _ = events_tx.send(RealtimeEvent::Error(error_message)).await; - return Err(mapped_error.into()); + warn!("failed to send {reason} response.create: {err}"); + return Err(err.into()); } self.active_default_response = true; Ok(()) @@ -428,11 +425,14 @@ struct RealtimeInputTask { handoff_state: RealtimeHandoffState, session_kind: RealtimeSessionKind, event_parser: RealtimeEventParser, - flush_transcript_tail_on_session_end: bool, - transcript_tail_tx: Sender, stop_token: CancellationToken, } +struct RealtimeTranscriptTailFlush { + enabled: bool, + tx: Sender, +} + struct RealtimeInputChannels { text_rx: Receiver, handoff_output_rx: Receiver, @@ -594,6 +594,10 @@ impl RealtimeConversationManager { Some(base_url) => client.with_webrtc_sideband_base_url(base_url), None => client, }; + let transcript_tail_flush = RealtimeTranscriptTailFlush { + enabled: flush_transcript_tail_on_session_end, + tx: transcript_tail_tx, + }; let (task, sdp) = if let Some(sdp) = sdp { let call = model_client .create_realtime_call_with_headers( @@ -614,8 +618,7 @@ impl RealtimeConversationManager { session_kind, event_parser, realtime_active: Arc::clone(&realtime_active), - flush_transcript_tail_on_session_end, - transcript_tail_tx, + transcript_tail_flush, stop_token: stop_token.clone(), }); (task, Some(call.sdp)) @@ -628,20 +631,21 @@ impl RealtimeConversationManager { ) .await .map_err(map_api_error)?; - let task = spawn_realtime_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, - flush_transcript_tail_on_session_end, - transcript_tail_tx, - stop_token: stop_token.clone(), - }); + let task = spawn_realtime_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) }; @@ -1718,8 +1722,34 @@ pub(crate) async fn handle_close(sess: &Arc, sub_id: String) { end_realtime_conversation(sess, sub_id, RealtimeConversationEnd::Requested).await; } -fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> { - tokio::spawn(run_realtime_input_task(input)) +fn spawn_realtime_input_task( + input: RealtimeInputTask, + transcript_tail_flush: RealtimeTranscriptTailFlush, +) -> JoinHandle<()> { + tokio::spawn(async move { + let events_tx = input.events_tx.clone(); + let transcript_state = input.events.transcript_state(); + let exit = run_realtime_input_task(input, /*pending_outbound*/ None).await; + if let RealtimeInputTaskExit::TransportLost { err, .. } = exit { + 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; + }) +} + +enum RealtimeInputTaskExit { + Terminal, + TransportLost { + err: ApiError, + pending_outbound: Option>, + }, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +enum RealtimePendingOutbound { + Text(ConversationTextParams), + Handoff(RealtimeOutbound), } struct RealtimeWebrtcSidebandInputTask { @@ -1733,78 +1763,14 @@ struct RealtimeWebrtcSidebandInputTask { session_kind: RealtimeSessionKind, event_parser: RealtimeEventParser, realtime_active: Arc, - flush_transcript_tail_on_session_end: bool, - transcript_tail_tx: Sender, + transcript_tail_flush: RealtimeTranscriptTailFlush, stop_token: CancellationToken, } -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, - flush_transcript_tail_on_session_end, - transcript_tail_tx, - stop_token, - } = input; - - tokio::spawn(async move { - if !realtime_active.load(Ordering::Relaxed) { - return; - } - - let connection = match tokio::select! { - connection = client.connect_webrtc_sideband( - session_config, - &call_id, - sideband_headers, - default_headers(), - ) => connection, - _ = stop_token.cancelled() => return, - } { - 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(), - text_rx: input_channels.text_rx, - handoff_output_rx: input_channels.handoff_output_rx, - audio_rx: input_channels.audio_rx, - events_tx, - handoff_state, - session_kind, - event_parser, - flush_transcript_tail_on_session_end, - transcript_tail_tx, - stop_token, - }) - .await; - }) -} - -async fn run_realtime_input_task(input: RealtimeInputTask) { +async fn run_realtime_input_task( + input: RealtimeInputTask, + pending_outbound: Option>, +) -> RealtimeInputTaskExit { let RealtimeInputTask { writer, events, @@ -1815,67 +1781,162 @@ async fn run_realtime_input_task(input: RealtimeInputTask) { handoff_state, session_kind, event_parser, - flush_transcript_tail_on_session_end, - transcript_tail_tx, stop_token, } = input; let mut output_audio_state: Option = None; let mut response_create_queue = RealtimeResponseCreateQueue::default(); + if let Some(pending_outbound) = pending_outbound + && let Err(err) = send_realtime_pending_outbound( + &pending_outbound, + &writer, + &handoff_state, + event_parser, + &mut response_create_queue, + ) + .await + { + return classify_realtime_input_error_with_pending(err, Some(pending_outbound)); + } + loop { let result = tokio::select! { - _ = stop_token.cancelled() => break, + _ = stop_token.cancelled() => Err(RealtimeInputTaskExit::Terminal), + realtime_event = events.next_event() => { + match realtime_event { + Ok(Some(event)) => { + handle_realtime_server_event( + event, + &writer, + &events_tx, + &handoff_state, + session_kind, + &mut output_audio_state, + &mut response_create_queue, + ) + .await + .map_err(classify_realtime_input_error) + } + Ok(None) => Err(RealtimeInputTaskExit::Terminal), + Err(err) => Err(RealtimeInputTaskExit::TransportLost { + err, + pending_outbound: None, + }), + } + } // Text input that should be sent into realtime. text = text_rx.recv() => { + let pending_outbound = text + .as_ref() + .ok() + .cloned() + .map(RealtimePendingOutbound::Text) + .map(Box::new); handle_text_input( text, &writer, - &events_tx, ) - .await + .await + .map_err(|err| { + classify_realtime_input_error_with_pending(err, pending_outbound) + }) } // Background agent progress or final output that should be sent back to realtime. background_agent_output = handoff_output_rx.recv() => { + let pending_outbound = background_agent_output + .as_ref() + .ok() + .cloned() + .map(RealtimePendingOutbound::Handoff) + .map(Box::new); 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 + .map_err(|err| { + classify_realtime_input_error_with_pending(err, pending_outbound) + }) } // Audio frames captured from the user microphone. user_audio_frame = audio_rx.recv() => { - handle_user_audio_input(user_audio_frame, &writer, &events_tx) + handle_user_audio_input(user_audio_frame, &writer) .await + .map_err(classify_realtime_input_error) } }; - if result.is_err() { - break; + if let Err(exit) = result { + break exit; } } +} - if flush_transcript_tail_on_session_end - && let Some(transcript_delta) = - realtime_transcript_delta(&events.take_transcript_tail().await) +fn classify_realtime_input_error(err: anyhow::Error) -> RealtimeInputTaskExit { + classify_realtime_input_error_with_pending(err, /*pending_outbound*/ None) +} + +fn classify_realtime_input_error_with_pending( + err: anyhow::Error, + pending_outbound: Option>, +) -> RealtimeInputTaskExit { + match err.downcast::() { + Ok(err) => RealtimeInputTaskExit::TransportLost { + err, + pending_outbound, + }, + Err(err) => { + warn!("realtime input task stopped: {err}"); + RealtimeInputTaskExit::Terminal + } + } +} + +async fn send_realtime_pending_outbound( + pending_outbound: &RealtimePendingOutbound, + writer: &RealtimeWebsocketWriter, + handoff_state: &RealtimeHandoffState, + event_parser: RealtimeEventParser, + response_create_queue: &mut RealtimeResponseCreateQueue, +) -> anyhow::Result<()> { + match pending_outbound { + RealtimePendingOutbound::Text(params) => writer + .send_conversation_item_create(params.text.clone(), params.role) + .await + .map_err(anyhow::Error::from), + RealtimePendingOutbound::Handoff(output) => { + handle_handoff_output( + Ok(output.clone()), + writer, + handoff_state, + event_parser, + response_create_queue, + ) + .await + } + } +} + +async fn report_realtime_transport_loss(events_tx: &Sender, err: ApiError) { + let mapped_error = map_api_error(err); + error!("realtime stream closed: {mapped_error}"); + let _ = events_tx + .send(RealtimeEvent::Error(mapped_error.to_string())) + .await; +} + +async fn flush_realtime_transcript_tail( + transcript_tail_flush: &RealtimeTranscriptTailFlush, + transcript_tail: &[RealtimeTranscriptEntry], +) { + if transcript_tail_flush.enabled + && let Some(transcript_delta) = realtime_transcript_delta(transcript_tail) { - let _ = transcript_tail_tx + let _ = transcript_tail_flush + .tx .send(wrap_realtime_delegation_input( REALTIME_SESSION_ENDED_HANDOFF_INSTRUCTION, Some(&transcript_delta), @@ -1888,21 +1949,12 @@ async fn run_realtime_input_task(input: RealtimeInputTask) { async fn handle_text_input( params: Result, writer: &RealtimeWebsocketWriter, - events_tx: &Sender, ) -> anyhow::Result<()> { let params = params.context("text input channel closed")?; - - if let Err(err) = writer + writer .send_conversation_item_create(params.text, params.role) .await - { - let mapped_error = map_api_error(err); - warn!("failed to send input text: {mapped_error}"); - let _ = events_tx - .send(RealtimeEvent::Error(mapped_error.to_string())) - .await; - return Err(mapped_error.into()); - } + .map_err(anyhow::Error::from)?; Ok(()) } @@ -1968,7 +2020,6 @@ fn v3_output_writer( async fn handle_handoff_output( handoff_output: Result, writer: &RealtimeWebsocketWriter, - events_tx: &Sender, handoff_state: &RealtimeHandoffState, event_parser: RealtimeEventParser, response_create_queue: &mut RealtimeResponseCreateQueue, @@ -2092,7 +2143,7 @@ async fn handle_handoff_output( Err(err) } else { return response_create_queue - .request_create(writer, events_tx, "standalone handoff") + .request_create(writer, "standalone handoff") .await; } } @@ -2104,7 +2155,7 @@ async fn handle_handoff_output( Err(err) } else { return response_create_queue - .request_create(writer, events_tx, "standalone handoff") + .request_create(writer, "standalone handoff") .await; } } @@ -2145,7 +2196,7 @@ async fn handle_handoff_output( Err(err) } else { return response_create_queue - .request_create(writer, events_tx, "handoff") + .request_create(writer, "handoff") .await; } } @@ -2162,18 +2213,14 @@ async fn handle_handoff_output( }, }; if let Err(err) = result { - let mapped_error = map_api_error(err); - warn!("failed to send handoff output: {mapped_error}"); - let _ = events_tx - .send(RealtimeEvent::Error(mapped_error.to_string())) - .await; - return Err(mapped_error.into()); + warn!("failed to send handoff output: {err}"); + return Err(err.into()); } Ok(()) } async fn handle_realtime_server_event( - event: Result, ApiError>, + event: RealtimeEvent, writer: &RealtimeWebsocketWriter, events_tx: &Sender, handoff_state: &RealtimeHandoffState, @@ -2181,23 +2228,6 @@ async fn handle_realtime_server_event( output_audio_state: &mut Option, response_create_queue: &mut RealtimeResponseCreateQueue, ) -> anyhow::Result<()> { - let event = match event { - Ok(Some(event)) => event, - Ok(None) => anyhow::bail!("realtime event stream ended"), - Err(err) => { - let mapped_error = map_api_error(err); - if events_tx - .send(RealtimeEvent::Error(mapped_error.to_string())) - .await - .is_err() - { - return Err(mapped_error.into()); - } - error!("realtime stream closed: {mapped_error}"); - return Err(mapped_error.into()); - } - }; - let should_stop = match &event { RealtimeEvent::AudioOut(frame) => { match session_kind { @@ -2217,7 +2247,8 @@ async fn handle_realtime_server_event( .item_id .as_deref() .is_none_or(|item_id| item_id == output_audio_state.item_id) - && let Err(err) = writer + { + writer .send_payload( json!({ "type": "conversation.item.truncate", @@ -2228,9 +2259,7 @@ async fn handle_realtime_server_event( .to_string(), ) .await - { - let mapped_error = map_api_error(err); - warn!("failed to truncate realtime audio: {mapped_error}"); + .map_err(anyhow::Error::from)?; } } } @@ -2249,7 +2278,7 @@ async fn handle_realtime_server_event( RealtimeSessionKind::V1 => {} RealtimeSessionKind::V2 => { response_create_queue - .mark_finished(writer, events_tx, "deferred") + .mark_finished(writer, "deferred") .await?; } } @@ -2261,7 +2290,7 @@ async fn handle_realtime_server_event( RealtimeSessionKind::V1 => {} RealtimeSessionKind::V2 => { response_create_queue - .mark_finished(writer, events_tx, "deferred") + .mark_finished(writer, "deferred") .await?; } } @@ -2287,17 +2316,11 @@ async fn handle_realtime_server_event( ) .await { - let mapped_error = map_api_error(err); - warn!( - "failed to send handoff steering acknowledgement: {mapped_error}" - ); - let _ = events_tx - .send(RealtimeEvent::Error(mapped_error.to_string())) - .await; - return Err(mapped_error.into()); + warn!("failed to send handoff steering acknowledgement: {err}"); + return Err(err.into()); } response_create_queue - .request_create(writer, events_tx, "handoff steering") + .request_create(writer, "handoff steering") .await?; } None => { @@ -2319,12 +2342,8 @@ async fn handle_realtime_server_event( .send_conversation_function_call_output(noop.call_id.clone(), String::new()) .await { - let mapped_error = map_api_error(err); - warn!("failed to send realtime noop function output: {mapped_error}"); - let _ = events_tx - .send(RealtimeEvent::Error(mapped_error.to_string())) - .await; - return Err(mapped_error.into()); + warn!("failed to send realtime noop function output: {err}"); + return Err(err.into()); } } } @@ -2359,18 +2378,12 @@ async fn handle_realtime_server_event( async fn handle_user_audio_input( frame: Result, writer: &RealtimeWebsocketWriter, - events_tx: &Sender, ) -> anyhow::Result<()> { let frame = frame.context("user audio input channel closed")?; - - if let Err(err) = writer.send_audio_frame(frame).await { - let mapped_error = map_api_error(err); - error!("failed to send input audio: {mapped_error}"); - let _ = events_tx - .send(RealtimeEvent::Error(mapped_error.to_string())) - .await; - return Err(mapped_error.into()); - } + writer + .send_audio_frame(frame) + .await + .map_err(anyhow::Error::from)?; Ok(()) } diff --git a/codex-rs/core/src/realtime_conversation/sideband.rs b/codex-rs/core/src/realtime_conversation/sideband.rs new file mode 100644 index 0000000000..7ce1456301 --- /dev/null +++ b/codex-rs/core/src/realtime_conversation/sideband.rs @@ -0,0 +1,162 @@ +use super::RealtimeInputTask; +use super::RealtimeInputTaskExit; +use super::RealtimeWebrtcSidebandInputTask; +use super::flush_realtime_transcript_tail; +use super::report_realtime_transport_loss; +use super::run_realtime_input_task; +use codex_api::ApiError; +use codex_api::RealtimeEvent; +use codex_api::RealtimeEventParser; +use codex_api::RealtimeTranscriptState; +use codex_api::map_api_error; +use codex_login::default_client::default_headers; +use http::StatusCode; +use std::sync::atomic::Ordering; +use std::time::Duration; +use tokio::task::JoinHandle; +use tokio::time::Instant; +use tokio::time::sleep; +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); + +pub(super) 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, + transcript_tail_flush, + stop_token, + } = input; + + tokio::spawn(async move { + let mut reconnecting = false; + let mut rapid_disconnects = 0_u32; + let mut pending_outbound = None; + let transcript_state = RealtimeTranscriptState::default(); + while realtime_active.load(Ordering::Relaxed) { + let connection = match tokio::select! { + biased; + _ = stop_token.cancelled() => break, + connection = client.connect_webrtc_sideband( + session_config.clone(), + &call_id, + sideband_headers.clone(), + default_headers(), + transcript_state.clone(), + ) => connection, + } { + Ok(connection) => connection, + Err(err) => { + if !realtime_active.load(Ordering::Relaxed) || stop_token.is_cancelled() { + break; + } + if reconnecting && webrtc_sideband_session_ended(&err) { + info!( + call_id, + "realtime sideband session ended while reconnecting" + ); + break; + } + let mapped_error = map_api_error(err); + warn!( + reconnecting, + "failed to connect realtime sideband: {mapped_error}" + ); + let _ = events_tx + .send(RealtimeEvent::Error(mapped_error.to_string())) + .await; + break; + } + }; + + if !realtime_active.load(Ordering::Relaxed) || stop_token.is_cancelled() { + break; + } + + let connected_at = Instant::now(); + let exit = run_realtime_input_task( + RealtimeInputTask { + writer: connection.writer(), + events: connection.events(), + text_rx: input_channels.text_rx.clone(), + handoff_output_rx: input_channels.handoff_output_rx.clone(), + audio_rx: input_channels.audio_rx.clone(), + events_tx: events_tx.clone(), + handoff_state: handoff_state.clone(), + session_kind, + event_parser, + stop_token: stop_token.clone(), + }, + pending_outbound.take(), + ) + .await; + + let RealtimeInputTaskExit::TransportLost { + err, + pending_outbound: failed_outbound, + } = exit + else { + break; + }; + pending_outbound = failed_outbound; + if !realtime_active.load(Ordering::Relaxed) || stop_token.is_cancelled() { + break; + } + if event_parser != RealtimeEventParser::FramelessBidi { + report_realtime_transport_loss(&events_tx, err).await; + break; + } + if connected_at.elapsed() >= STABLE_CONNECTION_DURATION { + rapid_disconnects = 0; + } + rapid_disconnects = rapid_disconnects.saturating_add(1); + let delay = reconnect_delay(rapid_disconnects); + warn!( + call_id, + delay_ms = delay.as_millis(), + "live sideband transport lost; reconnecting: {err}" + ); + reconnecting = true; + tokio::select! { + biased; + _ = stop_token.cancelled() => break, + _ = sleep(delay) => {} + } + } + + let transcript_tail = transcript_state.take_transcript_tail().await; + flush_realtime_transcript_tail(&transcript_tail_flush, &transcript_tail).await; + }) +} + +fn webrtc_sideband_session_ended(err: &ApiError) -> bool { + matches!( + err, + ApiError::Api { status, .. } + if matches!(*status, StatusCode::NOT_FOUND | StatusCode::GONE) + ) +} + +fn reconnect_delay(rapid_disconnects: u32) -> Duration { + let multiplier = 2_u32.saturating_pow(rapid_disconnects.saturating_sub(1)); + RECONNECT_BASE_DELAY + .saturating_mul(multiplier) + .min(RECONNECT_MAX_DELAY) +} + +#[cfg(test)] +#[path = "sideband_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/realtime_conversation/sideband_tests.rs b/codex-rs/core/src/realtime_conversation/sideband_tests.rs new file mode 100644 index 0000000000..47f60fd912 --- /dev/null +++ b/codex-rs/core/src/realtime_conversation/sideband_tests.rs @@ -0,0 +1,36 @@ +use super::*; +use pretty_assertions::assert_eq; + +#[test] +fn reconnect_delay_backs_off_and_caps() { + assert_eq!( + reconnect_delay(/*rapid_disconnects*/ 1), + Duration::from_millis(200) + ); + assert_eq!( + reconnect_delay(/*rapid_disconnects*/ 2), + Duration::from_millis(400) + ); + assert_eq!( + reconnect_delay(/*rapid_disconnects*/ 3), + Duration::from_millis(800) + ); + assert_eq!( + reconnect_delay(/*rapid_disconnects*/ 10), + RECONNECT_MAX_DELAY + ); +} + +#[test] +fn recognizes_terminal_sideband_statuses() { + for status in [StatusCode::NOT_FOUND, StatusCode::GONE] { + assert!(webrtc_sideband_session_ended(&ApiError::Api { + status, + message: "session ended".to_string(), + })); + } + assert!(!webrtc_sideband_session_ended(&ApiError::Api { + status: StatusCode::INTERNAL_SERVER_ERROR, + message: "retry".to_string(), + })); +} diff --git a/codex-rs/core/src/realtime_conversation_tests.rs b/codex-rs/core/src/realtime_conversation_tests.rs index 53b5faae15..70075592d1 100644 --- a/codex-rs/core/src/realtime_conversation_tests.rs +++ b/codex-rs/core/src/realtime_conversation_tests.rs @@ -1,17 +1,25 @@ use super::AGENT_FINAL_MESSAGE_PREFIX; use super::HANDOFF_STREAM_TRUNCATION_MARKER; use super::RealtimeHandoffState; +use super::RealtimeInputTaskExit; +use super::RealtimeOutbound; +use super::RealtimePendingOutbound; use super::RealtimeSessionKind; use super::RealtimeStreamedItem; +use super::classify_realtime_input_error; +use super::classify_realtime_input_error_with_pending; use super::realtime_delegation_from_handoff; use super::realtime_request_headers; use super::realtime_text_from_handoff_request; use super::wrap_realtime_delegation_input; use crate::context::RealtimeDelegationSource; use async_channel::bounded; +use codex_api::ApiError; use codex_api::RealtimeEventParser; use codex_protocol::models::MessagePhase; use codex_protocol::protocol::CodexResponseHandoffMode; +use codex_protocol::protocol::ConversationTextParams; +use codex_protocol::protocol::ConversationTextRole; use codex_protocol::protocol::RealtimeHandoffRequested; use codex_protocol::protocol::RealtimeTranscriptEntry; use pretty_assertions::assert_eq; @@ -147,6 +155,55 @@ fn wraps_realtime_delegation_input_with_xml_escaping_without_transcript() { ); } +#[test] +fn bounds_realtime_delegation_fields_and_keeps_latest_transcript() { + let input = format!("start{}input-end", "x".repeat(8 * 1024)); + let transcript = format!("transcript-start{}latest", "y".repeat(8 * 1024)); + let rendered = wrap_realtime_delegation_input( + &input, + Some(&transcript), + RealtimeDelegationSource::Handoff, + ); + + assert!(rendered.len() < 9 * 1024); + assert!(rendered.contains("start")); + assert!(!rendered.contains("input-end")); + assert!(!rendered.contains("transcript-start")); + assert!(rendered.contains("latest")); +} + +#[test] +fn classifies_outbound_api_failures_as_transport_loss() { + for pending_outbound in [ + RealtimePendingOutbound::Text(ConversationTextParams { + text: "retry me".to_string(), + role: ConversationTextRole::User, + }), + RealtimePendingOutbound::Handoff(RealtimeOutbound::StandaloneHandoff { + text: "retry this handoff".to_string(), + phase: Some(MessagePhase::FinalAnswer), + }), + ] { + let exit = classify_realtime_input_error_with_pending( + ApiError::Stream("failed to send realtime request".to_string()).into(), + Some(Box::new(pending_outbound.clone())), + ); + let RealtimeInputTaskExit::TransportLost { + err: ApiError::Stream(_), + pending_outbound: Some(actual_pending_outbound), + } = exit + else { + panic!("outbound API failure should preserve pending output for reconnect"); + }; + assert_eq!(*actual_pending_outbound, pending_outbound); + } + + assert!(matches!( + classify_realtime_input_error(anyhow::anyhow!("input channel closed")), + RealtimeInputTaskExit::Terminal + )); +} + #[tokio::test] async fn clears_active_handoff_explicitly() { let (tx, _rx) = bounded(1); diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index c6b5634485..fa5ee2fb66 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -24,8 +24,10 @@ use codex_protocol::protocol::RealtimeAudioFrame; use codex_protocol::protocol::RealtimeConversationRealtimeEvent; use codex_protocol::protocol::RealtimeConversationVersion; use codex_protocol::protocol::RealtimeEvent; +use codex_protocol::protocol::RealtimeHandoffRequested; use codex_protocol::protocol::RealtimeNoopRequested; use codex_protocol::protocol::RealtimeOutputModality; +use codex_protocol::protocol::RealtimeTranscriptEntry; use codex_protocol::protocol::RealtimeVoice; use codex_protocol::protocol::SessionSource; use codex_protocol::user_input::UserInput; @@ -41,6 +43,8 @@ use core_test_support::test_codex::TestCodex; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; use core_test_support::wait_for_event_match; +use futures::SinkExt; +use futures::StreamExt; use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; @@ -51,14 +55,18 @@ use std::sync::Arc; use std::sync::Mutex; use std::time::Duration; use tokio::io::AsyncReadExt; +use tokio::io::AsyncWriteExt; use tokio::sync::oneshot; use tokio::time::timeout; +use tokio_tungstenite::accept_async; +use tokio_tungstenite::tungstenite::Message; use uuid::Uuid; use wiremock::Match; use wiremock::Mock; use wiremock::Request as WiremockRequest; use wiremock::ResponseTemplate; use wiremock::matchers::method; +use wiremock::matchers::path; use wiremock::matchers::path_regex; const STARTUP_CONTEXT_HEADER: &str = "Startup context from Codex."; @@ -815,6 +823,276 @@ async fn conversation_webrtc_start_posts_generated_session() -> Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn conversation_webrtc_live_reconnects_sideband_after_unclean_disconnect() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + Mock::given(method("POST")) + .and(path("/v1/live")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("Location", "/v1/live/rtc_reconnect") + .set_body_string("v=answer\r\n"), + ) + .mount(&server) + .await; + let _response_mock = responses::mount_sse_once( + &server, + responses::sse(vec![ + responses::ev_response_created("resp-reconnect"), + responses::ev_assistant_message("msg-reconnect", "ok"), + responses::ev_completed("resp-reconnect"), + ]), + ) + .await; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let realtime_ws_base_url = format!("ws://{}", listener.local_addr()?); + let (inbound_flood_started_tx, inbound_flood_started_rx) = oneshot::channel(); + let sideband_server = tokio::spawn(async move { + let (stream, _) = listener.accept().await?; + let mut websocket = accept_async(stream).await?; + websocket + .send(Message::Text( + json!({ + "type": "input_transcript.added", + "item": { "text": "hello wor" } + }) + .to_string() + .into(), + )) + .await?; + let (mut websocket_tx, mut websocket_rx) = websocket.split(); + inbound_flood_started_tx + .send(()) + .map_err(|_| anyhow::anyhow!("realtime client dropped before inbound flood"))?; + let inbound_flood = tokio::spawn(async move { + let mut event_count = 0_u32; + loop { + websocket_tx + .send(Message::Text( + json!({ + "type": "session.updated", + "session": { "id": "sess_reconnect_flood" } + }) + .to_string() + .into(), + )) + .await?; + event_count += 1; + if event_count.is_multiple_of(64) { + tokio::task::yield_now().await; + } + } + #[allow(unreachable_code)] + Ok::<(), anyhow::Error>(()) + }); + timeout(Duration::from_secs(5), async { + loop { + let message = websocket_rx + .next() + .await + .context("sideband closed before outbound text")??; + let Message::Text(payload) = message else { + continue; + }; + let payload: Value = serde_json::from_str(&payload)?; + let saw_outbound_text = payload["type"] == "session.context.append" + && payload["content"] + .as_array() + .into_iter() + .flatten() + .filter_map(|content| content["text"].as_str()) + .any(|text| text.contains("outbound during inbound flood")); + if saw_outbound_text { + break Ok::<(), anyhow::Error>(()); + } + } + }) + .await + .context("outbound text was starved by sustained inbound traffic")??; + inbound_flood.abort(); + drop(websocket_rx); + + let (stream, _) = listener.accept().await?; + let mut websocket = accept_async(stream).await?; + websocket + .send(Message::Text( + json!({ + "type": "turn.done", + "turn": { "role": "user", "transcript": "hello world" } + }) + .to_string() + .into(), + )) + .await?; + websocket + .send(Message::Text( + json!({ + "type": "output_transcript.added", + "item": { "text": "after reconnect" } + }) + .to_string() + .into(), + )) + .await?; + websocket + .send(Message::Text( + json!({ + "type": "delegation.created", + "item": { + "id": "handoff_reconnect", + "type": "delegation", + "target": "client", + "content": [{ "type": "input_text", "text": "hello world" }] + } + }) + .to_string() + .into(), + )) + .await?; + drop(websocket); + + let (mut stream, _) = listener.accept().await?; + let mut request = vec![0_u8; 4096]; + let request_len = stream.read(&mut request).await?; + if !request[..request_len].starts_with(b"GET ") { + anyhow::bail!("expected reconnect websocket handshake"); + } + stream + .write_all(b"HTTP/1.1 410 Gone\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .await?; + if timeout(Duration::from_secs(1), listener.accept()) + .await + .is_ok() + { + anyhow::bail!("terminal sideband response was retried"); + } + Ok::<(), anyhow::Error>(()) + }); + + 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::V3; + }); + let test = builder.build(&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, + codex_response_handoff_channel_prefixes: None, + model: None, + output_modality: RealtimeOutputModality::Audio, + include_startup_context: true, + initial_items: Vec::new(), + realtime_start_instructions: None, + realtime_end_instructions: None, + prompt: Some(Some("backend prompt".to_string())), + realtime_session_id: None, + transport: Some(ConversationStartTransport::Webrtc { + sdp: "v=offer\r\n".to_string(), + }), + version: Some(RealtimeConversationVersion::V3), + voice: None, + })) + .await?; + timeout(Duration::from_secs(5), inbound_flood_started_rx) + .await + .context("timed out waiting for sustained inbound traffic")? + .context("sideband server stopped before sustained inbound traffic")?; + test.codex + .submit(Op::RealtimeConversationText(ConversationTextParams { + text: "outbound during inbound flood".to_string(), + role: ConversationTextRole::User, + })) + .await?; + + let handoff = timeout(Duration::from_secs(10), async { + let mut saw_post_reconnect_transcript = false; + loop { + match test.codex.next_event().await?.msg { + EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::OutputTranscriptDelta(delta), + }) if delta.delta == "after reconnect" => { + saw_post_reconnect_transcript = true; + } + EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::HandoffRequested(handoff), + }) => { + if !saw_post_reconnect_transcript { + anyhow::bail!("handoff arrived before the post-reconnect transcript"); + } + break Ok::<_, anyhow::Error>(handoff); + } + EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::Error(message), + }) => anyhow::bail!("reconnect emitted a realtime error: {message}"), + EventMsg::RealtimeConversationClosed(closed) => { + anyhow::bail!("reconnect closed the conversation early: {closed:?}") + } + EventMsg::Error(err) => { + anyhow::bail!("reconnect emitted an app error: {}", err.message) + } + _ => {} + } + } + }) + .await + .context("timed out waiting for the reconnected sideband")??; + assert_eq!( + handoff, + RealtimeHandoffRequested { + handoff_id: "handoff_reconnect".to_string(), + item_id: "handoff_reconnect".to_string(), + input_transcript: "hello world".to_string(), + active_transcript: vec![ + RealtimeTranscriptEntry { + role: "user".to_string(), + text: "hello world".to_string(), + }, + RealtimeTranscriptEntry { + role: "assistant".to_string(), + text: "after reconnect".to_string(), + } + ], + } + ); + + let closed = timeout(Duration::from_secs(10), async { + loop { + match test.codex.next_event().await?.msg { + EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::Error(message), + }) => anyhow::bail!("terminal reconnect emitted a realtime error: {message}"), + EventMsg::RealtimeConversationClosed(closed) => { + break Ok::<_, anyhow::Error>(closed); + } + EventMsg::Error(err) => { + anyhow::bail!("terminal reconnect emitted an app error: {}", err.message) + } + _ => {} + } + } + }) + .await + .context("timed out waiting for the terminal reconnect response")??; + assert_eq!(closed.reason.as_deref(), Some("transport_closed")); + timeout(Duration::from_secs(10), sideband_server) + .await + .context("timed out waiting for sideband server")???; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn conversation_webrtc_start_uses_avas_query() -> Result<()> { skip_if_no_network!(Ok(()));