From 9d71ecf60627a3950f1335a7118ae7fe273c11b8 Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Sun, 12 Apr 2026 12:48:01 -0700 Subject: [PATCH] Trim realtime text mirror plumbing --- codex-rs/core/src/codex.rs | 13 +++---------- codex-rs/core/src/realtime_conversation.rs | 16 ++++++---------- 2 files changed, 9 insertions(+), 20 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 6bd8ccf5b4..f5be00103b 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_without_realtime_text_mirror( + let _ = handlers::user_input_or_turn_without_realtime_text_mirror( self, self.next_internal_sub_id(), Op::UserInput { @@ -4960,7 +4960,8 @@ mod handlers { } pub async fn user_input_or_turn(sess: &Arc, sub_id: String, op: Op) { - let accepted_items = user_input_or_turn_inner(sess, sub_id, op).await; + let accepted_items = + user_input_or_turn_without_realtime_text_mirror(sess, sub_id, op).await; if let Some(items) = accepted_items { mirror_user_text_to_realtime(sess, &items).await; } @@ -4970,14 +4971,6 @@ mod handlers { sess: &Arc, sub_id: String, op: Op, - ) { - let _ = user_input_or_turn_inner(sess, sub_id, op).await; - } - - async fn user_input_or_turn_inner( - sess: &Arc, - sub_id: String, - op: Op, ) -> Option> { let (items, updates, responsesapi_client_metadata) = match op { Op::UserTurn { diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 40a8ba498f..8200ba4908 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -403,10 +403,6 @@ impl RealtimeConversationManager { } pub(crate) async fn text_in(&self, text: String) -> CodexResult<()> { - self.queue_text_input(text).await - } - - async fn queue_text_input(&self, text: String) -> CodexResult<()> { let sender = { let guard = self.state.lock().await; guard.as_ref().map(|state| state.user_text_tx.clone()) @@ -908,7 +904,7 @@ pub(crate) async fn handle_text( sub_id: String, params: ConversationTextParams, ) { - debug!(text = %params.text, "[realtime-text] sending realtime conversation text input"); + debug!(text = %params.text, "[realtime-text] appending realtime conversation text input"); if let Err(err) = sess.conversation.text_in(params.text).await { error!("failed to append realtime text: {err}"); if sess.conversation.running_state().await.is_some() { @@ -943,9 +939,9 @@ fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> { loop { let result = tokio::select! { - // Text that should be sent into realtime. + // Text typed by the user that should be sent into realtime. user_text = user_text_rx.recv() => { - handle_realtime_text_input( + handle_user_text_input( user_text, &writer, &events_tx, @@ -990,12 +986,12 @@ fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> { }) } -async fn handle_realtime_text_input( - input: Result, +async fn handle_user_text_input( + text: Result, writer: &RealtimeWebsocketWriter, events_tx: &Sender, ) -> anyhow::Result<()> { - let text = input.context("text input channel closed")?; + let text = text.context("user text input channel closed")?; if let Err(err) = writer.send_conversation_item_create(text).await { let mapped_error = map_api_error(err);