openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-31 19:28:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Switch realtime TUI to WebRTC transport

Browse Source
This commit is contained in:
Justin Uberti
2026-03-20 16:08:45 -07:00
parent ddda43919f
commit fc01075f74
20 changed files with 1183 additions and 3489 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/realtime_conversation.rs
+39 -62
View File
@@ -8,14 +8,12 @@ use crate::error::Result as CodexResult;
use crate::realtime_context::build_realtime_startup_context;
use async_channel::Receiver;
use async_channel::Sender;
use async_channel::TrySendError;
use codex_api::Provider as ApiProvider;
use codex_api::RealtimeAudioFrame;
use codex_api::RealtimeEvent;
use codex_api::RealtimeSessionConfig;
use codex_api::RealtimeWebsocketClient;
use codex_api::endpoint::realtime_websocket::RealtimeWebsocketEvents;
use codex_api::endpoint::realtime_websocket::RealtimeWebsocketWriter;
use codex_api::RealtimeWebrtcClient;
use codex_api::endpoint::realtime_webrtc::RealtimeWebrtcEvents;
use codex_api::endpoint::realtime_webrtc::RealtimeWebrtcWriter;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::ConversationAudioParams;
use codex_protocol::protocol::ConversationAudioTruncateParams;
@@ -24,6 +22,7 @@ use codex_protocol::protocol::ConversationTextParams;
use codex_protocol::protocol::ErrorEvent;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RealtimeAudioFrame;
use codex_protocol::protocol::RealtimeConversationClosedEvent;
use codex_protocol::protocol::RealtimeConversationRealtimeEvent;
use codex_protocol::protocol::RealtimeConversationStartedEvent;
@@ -42,7 +41,6 @@ use tracing::error;
use tracing::info;
use tracing::warn;
const AUDIO_IN_QUEUE_CAPACITY: usize = 256;
const USER_TEXT_IN_QUEUE_CAPACITY: usize = 64;
const HANDOFF_OUT_QUEUE_CAPACITY: usize = 64;
const OUTPUT_EVENTS_QUEUE_CAPACITY: usize = 256;
@@ -132,9 +130,8 @@ impl RealtimeHandoffState {
#[allow(dead_code)]
struct ConversationState {
audio_tx: Sender<RealtimeAudioFrame>,
user_text_tx: Sender<String>,
writer: RealtimeWebsocketWriter,
writer: RealtimeWebrtcWriter,
handoff: RealtimeHandoffState,
task: JoinHandle<()>,
realtime_active: Arc<AtomicBool>,
@@ -178,7 +175,7 @@ impl RealtimeConversationManager {
model,
session_id,
};
let client = RealtimeWebsocketClient::new(api_provider);
let client = RealtimeWebrtcClient::new(api_provider);
let connection = client
.connect(
session_config,
@@ -190,8 +187,6 @@ impl RealtimeConversationManager {
let writer = connection.writer();
let events = connection.events();
let (audio_tx, audio_rx) =
async_channel::bounded::<RealtimeAudioFrame>(AUDIO_IN_QUEUE_CAPACITY);
let (user_text_tx, user_text_rx) =
async_channel::bounded::<String>(USER_TEXT_IN_QUEUE_CAPACITY);
let (handoff_output_tx, handoff_output_rx) =
@@ -206,14 +201,12 @@ impl RealtimeConversationManager {
events,
user_text_rx,
handoff_output_rx,
audio_rx,
events_tx,
handoff.clone(),
);
let mut guard = self.state.lock().await;
*guard = Some(ConversationState {
audio_tx,
user_text_tx,
writer,
handoff,
@@ -224,27 +217,10 @@ impl RealtimeConversationManager {
}
pub(crate) async fn audio_in(&self, frame: RealtimeAudioFrame) -> CodexResult<()> {
let sender = {
let guard = self.state.lock().await;
guard.as_ref().map(|state| state.audio_tx.clone())
};
let Some(sender) = sender else {
return Err(CodexErr::InvalidRequest(
"conversation is not running".to_string(),
));
};
match sender.try_send(frame) {
Ok(()) => Ok(()),
Err(TrySendError::Full(_)) => {
warn!("dropping input audio frame due to full queue");
Ok(())
}
Err(TrySendError::Closed(_)) => Err(CodexErr::InvalidRequest(
"conversation is not running".to_string(),
)),
}
let _ = frame;
Err(CodexErr::InvalidRequest(
"realtime audio input is handled by WebRTC native audio, not PCM append".to_string(),
))
}
pub(crate) async fn text_in(&self, text: String) -> CodexResult<()> {
@@ -270,25 +246,17 @@ impl RealtimeConversationManager {
&self,
params: ConversationAudioTruncateParams,
) -> CodexResult<()> {
let writer = {
let is_running = {
let guard = self.state.lock().await;
guard.as_ref().map(|state| state.writer.clone())
guard.is_some()
};
let Some(writer) = writer else {
if !is_running {
return Err(CodexErr::InvalidRequest(
"conversation is not running".to_string(),
));
};
}
writer
.send_conversation_item_truncate(
params.item_id,
params.content_index,
params.audio_end_ms,
)
.await
.map_err(map_api_error)?;
let _ = params;
Ok(())
}
@@ -608,11 +576,10 @@ pub(crate) async fn handle_close(sess: &Arc<Session>, sub_id: String) {
}
fn spawn_realtime_input_task(
writer: RealtimeWebsocketWriter,
events: RealtimeWebsocketEvents,
writer: RealtimeWebrtcWriter,
events: RealtimeWebrtcEvents,
user_text_rx: Receiver<String>,
handoff_output_rx: Receiver<HandoffOutput>,
audio_rx: Receiver<RealtimeAudioFrame>,
events_tx: Sender<RealtimeEvent>,
handoff_state: RealtimeHandoffState,
) -> JoinHandle<()> {
@@ -806,18 +773,6 @@ fn spawn_realtime_input_task(
}
}
}
frame = audio_rx.recv() => {
match frame {
Ok(frame) => {
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}");
break;
}
}
Err(_) => break,
}
}
}
}
})
@@ -843,9 +798,11 @@ async fn send_conversation_error(
mod tests {
use super::HandoffOutput;
use super::INTERRUPT_TOOL_OUTPUT_CANCELLED;
use super::RealtimeConversationManager;
use super::RealtimeHandoffState;
use super::realtime_text_from_handoff_request;
use async_channel::bounded;
use codex_protocol::protocol::RealtimeAudioFrame;
use codex_protocol::protocol::RealtimeHandoffRequested;
use codex_protocol::protocol::RealtimeTranscriptEntry;
use pretty_assertions::assert_eq;
@@ -901,6 +858,26 @@ mod tests {
assert_eq!(realtime_text_from_handoff_request(&handoff), None);
}
#[tokio::test]
async fn rejects_pcm_audio_append_in_webrtc_mode() {
let manager = RealtimeConversationManager::new();
let err = manager
.audio_in(RealtimeAudioFrame {
data: "AQID".to_string(),
sample_rate: 24000,
num_channels: 1,
samples_per_channel: Some(480),
})
.await
.expect_err("PCM append should be unsupported");
assert_eq!(
err.to_string(),
"realtime audio input is handled by WebRTC native audio, not PCM append"
);
}
#[tokio::test]
async fn clears_active_handoff_explicitly() {
let (tx, _rx) = bounded(1);
codex-rs/core/tests/suite/realtime_conversation.rs
-66
View File
@@ -6,7 +6,6 @@ use codex_core::auth::OPENAI_API_KEY_ENV_VAR;
use codex_protocol::ThreadId;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::ConversationAudioParams;
use codex_protocol::protocol::ConversationAudioTruncateParams;
use codex_protocol::protocol::ConversationStartParams;
use codex_protocol::protocol::ConversationTextParams;
use codex_protocol::protocol::ErrorEvent;
@@ -274,71 +273,6 @@ async fn conversation_start_audio_text_close_round_trip() -> Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conversation_audio_truncate_sends_truncate_event() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_websocket_server(vec![
vec![],
vec![
vec![json!({
"type": "session.updated",
"session": { "id": "sess_truncate", "instructions": "backend prompt" }
})],
vec![],
],
])
.await;
let mut builder = test_codex();
let test = builder.build_with_websocket_server(&server).await?;
assert!(server.wait_for_handshakes(1, Duration::from_secs(2)).await);
test.codex
.submit(Op::RealtimeConversationStart(ConversationStartParams {
prompt: "backend prompt".to_string(),
session_id: None,
}))
.await?;
wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::SessionUpdated { session_id, .. },
}) if session_id == "sess_truncate" => Some(()),
_ => None,
})
.await;
test.codex
.submit(Op::RealtimeConversationAudioTruncate(
ConversationAudioTruncateParams {
item_id: "item_audio_1".to_string(),
content_index: 0,
audio_end_ms: 320,
},
))
.await?;
let truncate_request = wait_for_matching_websocket_request(
&server,
"realtime conversation.item.truncate request",
|request| request.body_json()["type"].as_str() == Some("conversation.item.truncate"),
)
.await;
assert_eq!(
truncate_request.body_json(),
json!({
"type": "conversation.item.truncate",
"item_id": "item_audio_1",
"content_index": 0,
"audio_end_ms": 320
})
);
server.shutdown().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial(openai_api_key_env)]
async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() -> Result<()> {