From c9438f6891d37449075612902b1bf7b508a9b5e9 Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Sat, 11 Apr 2026 23:28:13 -0700 Subject: [PATCH] Mirror user text into realtime --- codex-rs/core/src/codex.rs | 45 +++++++- .../core/tests/suite/realtime_conversation.rs | 107 ++++++++++++++++++ ..._turn_is_sent_to_realtime_when_active.snap | 9 ++ codex-rs/tui/src/chatwidget.rs | 10 -- codex-rs/tui/src/chatwidget/realtime.rs | 26 ----- 5 files changed, 159 insertions(+), 38 deletions(-) create mode 100644 codex-rs/core/tests/suite/snapshots/all__suite__realtime_conversation__conversation_user_text_turn_is_sent_to_realtime_when_active.snap diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index f44a79528f..9100850ff2 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -2265,7 +2265,7 @@ impl Session { } pub(crate) async fn route_realtime_text_input(self: &Arc, text: String) { - handlers::user_input_or_turn( + handlers::user_input_or_turn_with_realtime_text_mirror( self, self.next_internal_sub_id(), Op::UserInput { @@ -2276,6 +2276,7 @@ impl Session { final_output_json_schema: None, responsesapi_client_metadata: None, }, + handlers::RealtimeTextMirror::Disabled, ) .await; } @@ -4910,6 +4911,7 @@ mod handlers { use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::Settings; use codex_protocol::dynamic_tools::DynamicToolResponse; + use codex_protocol::items::UserMessageItem; use codex_protocol::mcp::RequestId as ProtocolRequestId; use codex_protocol::user_input::UserInput; use codex_rmcp_client::ElicitationAction; @@ -4917,6 +4919,7 @@ mod handlers { use serde_json::Value; use std::path::PathBuf; use std::sync::Arc; + use tracing::debug; use tracing::info; use tracing::warn; @@ -4957,7 +4960,22 @@ mod handlers { } } + pub(super) enum RealtimeTextMirror { + Enabled, + Disabled, + } + pub async fn user_input_or_turn(sess: &Arc, sub_id: String, op: Op) { + user_input_or_turn_with_realtime_text_mirror(sess, sub_id, op, RealtimeTextMirror::Enabled) + .await; + } + + pub(super) async fn user_input_or_turn_with_realtime_text_mirror( + sess: &Arc, + sub_id: String, + op: Op, + realtime_text_mirror: RealtimeTextMirror, + ) { let (items, updates, responsesapi_client_metadata) = match op { Op::UserTurn { cwd, @@ -5016,6 +5034,13 @@ mod handlers { ), _ => unreachable!(), }; + let realtime_text = match realtime_text_mirror { + RealtimeTextMirror::Enabled => { + let text = UserMessageItem::new(&items).message(); + (!text.is_empty()).then_some(text) + } + RealtimeTextMirror::Disabled => None, + }; let Ok(current_context) = sess.new_turn_with_sub_id(sub_id.clone(), updates).await else { // new_turn_with_sub_id already emits the error event. @@ -5031,7 +5056,10 @@ mod handlers { ) .await { - Ok(_) => current_context.session_telemetry.user_prompt(&items), + Ok(_) => { + current_context.session_telemetry.user_prompt(&items); + mirror_user_text_to_realtime(sess, realtime_text).await; + } Err(SteerInputError::NoActiveTurn(items)) => { if let Some(responsesapi_client_metadata) = responsesapi_client_metadata { current_context @@ -5047,6 +5075,7 @@ mod handlers { crate::tasks::RegularTask::new(), ) .await; + mirror_user_text_to_realtime(sess, realtime_text).await; } Err(err) => { sess.send_event_raw(Event { @@ -5058,6 +5087,18 @@ mod handlers { } } + async fn mirror_user_text_to_realtime(sess: &Arc, realtime_text: Option) { + let Some(text) = realtime_text else { + return; + }; + if sess.conversation.running_state().await.is_none() { + return; + } + if let Err(err) = sess.conversation.text_in(text).await { + debug!("failed to mirror user text to realtime conversation: {err}"); + } + } + /// Records an inter-agent assistant envelope, then lets the shared pending-work scheduler /// decide whether an idle session should start a regular turn. pub async fn inter_agent_communication( diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index a8f7f10fe6..6261590058 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -1626,6 +1626,113 @@ async fn conversation_startup_context_is_truncated_and_sent_once_per_start() -> Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn conversation_user_text_turn_is_sent_to_realtime_when_active() -> Result<()> { + skip_if_no_network!(Ok(())); + + let api_server = start_mock_server().await; + let response_mock = responses::mount_sse_once( + &api_server, + responses::sse(vec![ + responses::ev_response_created("resp_user_text"), + responses::ev_assistant_message("msg_user_text", "ack"), + responses::ev_completed("resp_user_text"), + ]), + ) + .await; + + let realtime_server = start_websocket_server(vec![vec![ + vec![json!({ + "type": "session.updated", + "session": { "id": "sess_user_text", "instructions": "backend prompt" } + })], + vec![], + ]]) + .await; + + let mut builder = test_codex().with_config({ + let realtime_base_url = realtime_server.uri().to_string(); + move |config| { + config.experimental_realtime_ws_base_url = Some(realtime_base_url); + config.realtime.version = RealtimeWsVersion::V1; + } + }); + let test = builder.build(&api_server).await?; + + test.codex + .submit(Op::RealtimeConversationStart(ConversationStartParams { + prompt: Some(Some("backend prompt".to_string())), + session_id: None, + transport: None, + voice: None, + })) + .await?; + + let session_updated = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::SessionUpdated { session_id, .. }, + }) => Some(session_id.clone()), + _ => None, + }) + .await; + assert_eq!(session_updated, "sess_user_text"); + + let user_text = "typed follow-up for realtime"; + test.codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: user_text.to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + }) + .await?; + + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + let realtime_text_request = wait_for_matching_websocket_request( + &realtime_server, + "normal user turn text mirrored to realtime", + |request| websocket_request_text(request).as_deref() == Some(user_text), + ) + .await; + let model_user_texts = response_mock.single_request().message_input_texts("user"); + assert_eq!( + ( + model_user_texts.iter().any(|text| text == user_text), + websocket_request_text(&realtime_text_request), + ), + (true, Some(user_text.to_string())), + ); + + let realtime_request_body = realtime_text_request.body_json(); + let content = &realtime_request_body["item"]["content"][0]; + let snapshot = format!( + "type: {}\nitem.type: {}\nitem.role: {}\ncontent[0].type: {}\ncontent[0].text: {}", + realtime_request_body["type"].as_str().unwrap_or_default(), + realtime_request_body["item"]["type"] + .as_str() + .unwrap_or_default(), + realtime_request_body["item"]["role"] + .as_str() + .unwrap_or_default(), + content["type"].as_str().unwrap_or_default(), + content["text"].as_str().unwrap_or_default(), + ); + insta::assert_snapshot!( + "conversation_user_text_turn_is_sent_to_realtime_when_active", + snapshot + ); + + realtime_server.shutdown().await; + api_server.shutdown().await; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__realtime_conversation__conversation_user_text_turn_is_sent_to_realtime_when_active.snap b/codex-rs/core/tests/suite/snapshots/all__suite__realtime_conversation__conversation_user_text_turn_is_sent_to_realtime_when_active.snap new file mode 100644 index 0000000000..070f7ce09b --- /dev/null +++ b/codex-rs/core/tests/suite/snapshots/all__suite__realtime_conversation__conversation_user_text_turn_is_sent_to_realtime_when_active.snap @@ -0,0 +1,9 @@ +--- +source: core/tests/suite/realtime_conversation.rs +expression: snapshot +--- +type: conversation.item.create +item.type: message +item.role: user +content[0].type: text +content[0].text: typed follow-up for realtime diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index fe83404523..7f47edbb99 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -5050,11 +5050,6 @@ impl ChatWidget { { return; } - let Some(user_message) = - self.maybe_defer_user_message_for_realtime(user_message) - else { - return; - }; let should_submit_now = self.is_session_configured() && !self.is_plan_streaming_in_tui(); if should_submit_now { @@ -5085,11 +5080,6 @@ impl ChatWidget { .bottom_pane .take_recent_submission_mention_bindings(), }; - let Some(user_message) = - self.maybe_defer_user_message_for_realtime(user_message) - else { - return; - }; self.queue_user_message(user_message); } InputResult::Command(cmd) => { diff --git a/codex-rs/tui/src/chatwidget/realtime.rs b/codex-rs/tui/src/chatwidget/realtime.rs index f45ae5b9aa..03a59c224c 100644 --- a/codex-rs/tui/src/chatwidget/realtime.rs +++ b/codex-rs/tui/src/chatwidget/realtime.rs @@ -29,7 +29,6 @@ pub(super) struct RealtimeConversationUiState { pub(super) phase: RealtimeConversationPhase, requested_close: bool, session_id: Option, - warned_audio_only_submission: bool, transport: RealtimeConversationUiTransport, #[cfg(not(target_os = "linux"))] pub(super) meter_placeholder_id: Option, @@ -191,28 +190,6 @@ impl ChatWidget { self.last_rendered_user_message_event.as_ref() != Some(&key) } - pub(super) fn maybe_defer_user_message_for_realtime( - &mut self, - user_message: UserMessage, - ) -> Option { - if !self.realtime_conversation.is_live() { - return Some(user_message); - } - - self.restore_user_message_to_composer(user_message); - if !self.realtime_conversation.warned_audio_only_submission { - self.realtime_conversation.warned_audio_only_submission = true; - self.add_info_message( - "Realtime voice mode is audio-only. Use /realtime to stop.".to_string(), - /*hint*/ None, - ); - } else { - self.request_redraw(); - } - - None - } - fn realtime_footer_hint_items() -> Vec<(String, String)> { vec![("/realtime".to_string(), "stop live voice".to_string())] } @@ -238,7 +215,6 @@ impl ChatWidget { self.realtime_conversation.phase = RealtimeConversationPhase::Starting; self.realtime_conversation.requested_close = false; self.realtime_conversation.session_id = None; - self.realtime_conversation.warned_audio_only_submission = false; self.set_footer_hint_override(Some(Self::realtime_footer_hint_items())); match self.config.realtime.transport { RealtimeTransport::Websocket => { @@ -297,7 +273,6 @@ impl ChatWidget { self.realtime_conversation.phase = RealtimeConversationPhase::Inactive; self.realtime_conversation.requested_close = false; self.realtime_conversation.session_id = None; - self.realtime_conversation.warned_audio_only_submission = false; self.realtime_conversation.transport = RealtimeConversationUiTransport::Websocket; } @@ -320,7 +295,6 @@ impl ChatWidget { return; } self.realtime_conversation.session_id = ev.session_id; - self.realtime_conversation.warned_audio_only_submission = false; self.set_footer_hint_override(Some(Self::realtime_footer_hint_items())); if self.realtime_conversation_uses_webrtc() { self.realtime_conversation.phase = RealtimeConversationPhase::Starting;