openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reconnect WebRTC Realtime sideband transports (#39257)

Browse Source 
## What changed

- Reconnect frameless bidirectional WebRTC sideband sockets after unexpected transport loss with capped exponential backoff.
- Preserve pending text or handoff output and bounded transcript state across reconnects.
- Treat `404 Not Found` and `410 Gone` handshake responses as terminal session completion, and preserve HTTP status information from failed WebSocket handshakes.
- Bound Realtime delegation input and transcript fields while retaining the most relevant portion of each.

## Testing

- Add unit coverage for reconnect backoff, terminal status handling, transport error classification, transcript continuity and truncation, and pending outbound messages.
- Add an end-to-end WebRTC conversation test covering an unclean disconnect, reconnect, transcript reconciliation, outbound progress during sustained inbound traffic, and terminal session completion.

GitOrigin-RevId: 544fd7e542c3434d0bb42cfbcc7143604419a037
This commit is contained in:
Peter Bakkum
2026-08-18 18:41:40 +00:00
committed by copyberry
parent fa9a05f2d2
commit ecb8013dfa
11 changed files with 1046 additions and 215 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/codex-api/src/endpoint/mod.rs
+1
View File
@@ -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;
codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs
+263 -16
View File
@@ -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<WsCommand>,
@@ -221,15 +225,19 @@ pub struct RealtimeWebsocketWriter {
pub struct RealtimeWebsocketEvents {
rx_message: async_channel::Receiver<Result<Message, WsError>>,
pending_events: Arc<Mutex<VecDeque<RealtimeEvent>>>,
active_transcript: Arc<Mutex<ActiveTranscriptState>>,
transcript_state: RealtimeTranscriptState,
event_parser: RealtimeEventParser,
is_closed: Arc<AtomicBool>,
}
#[derive(Clone, Default)]
pub struct RealtimeTranscriptState {
active: Arc<Mutex<ActiveTranscriptState>>,
}
#[derive(Default)]
struct ActiveTranscriptState {
entries: Vec<RealtimeTranscriptEntry>,
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<Result<Message, WsError>>,
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<RealtimeTranscriptEntry> {
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<Option<RealtimeEvent>, 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<RealtimeTranscriptEntry> {
std::mem::take(&mut self.active.lock().await.entries)
}
}
fn truncate_active_transcript(entries: &mut Vec<RealtimeTranscriptEntry>) {
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<RealtimeTranscriptEntry>,
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<RealtimeWebsocketConnection, ApiError> {
// 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<RealtimeWebsocketConnection, ApiError> {
// 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<RealtimeWebsocketConnection, ApiError> {
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("<none>"),
@@ -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<CloseCode>),
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!({
codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs
+1
View File
@@ -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;
codex-rs/codex-api/src/lib.rs
+1
View File
@@ -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;
codex-rs/codex-api/tests/realtime_websocket_e2e.rs
+2
View File
@@ -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");
codex-rs/core/src/context/realtime_delegation.rs
+35 -2
View File
@@ -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>{input}</input>\n <transcript_delta>{transcript_delta}</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('&', "&amp;")
codex-rs/core/src/realtime_conversation.rs
+210 -197
View File
@@ -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<RealtimeEvent>,
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<RealtimeEvent>,
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<RealtimeEvent>,
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<String>,
stop_token: CancellationToken,
}
struct RealtimeTranscriptTailFlush {
enabled: bool,
tx: Sender<String>,
}
struct RealtimeInputChannels {
text_rx: Receiver<ConversationTextParams>,
handoff_output_rx: Receiver<RealtimeOutbound>,
@@ -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<Session>, 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<Box<RealtimePendingOutbound>>,
},
}
#[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<AtomicBool>,
flush_transcript_tail_on_session_end: bool,
transcript_tail_tx: Sender<String>,
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<Box<RealtimePendingOutbound>>,
) -> 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<OutputAudioState> = 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<Box<RealtimePendingOutbound>>,
) -> RealtimeInputTaskExit {
match err.downcast::<ApiError>() {
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<RealtimeEvent>, 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<ConversationTextParams, RecvError>,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
) -> 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<RealtimeOutbound, RecvError>,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
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<Option<RealtimeEvent>, ApiError>,
event: RealtimeEvent,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
handoff_state: &RealtimeHandoffState,
@@ -2181,23 +2228,6 @@ async fn handle_realtime_server_event(
output_audio_state: &mut Option<OutputAudioState>,
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<RealtimeAudioFrame, RecvError>,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
) -> 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(())
}
codex-rs/core/src/realtime_conversation/sideband.rs
+162
View File
@@ -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;
codex-rs/core/src/realtime_conversation/sideband_tests.rs
+36
View File
@@ -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(),
}));
}
codex-rs/core/src/realtime_conversation_tests.rs
+57
View File
@@ -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("<input>start"));
assert!(!rendered.contains("input-end"));
assert!(!rendered.contains("transcript-start"));
assert!(rendered.contains("latest</transcript_delta>"));
}
#[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);
codex-rs/core/tests/suite/realtime_conversation.rs
+278
View File
@@ -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(()));