diff --git a/codex-rs/tui/src/chatwidget/realtime.rs b/codex-rs/tui/src/chatwidget/realtime.rs index e2e8afde17..ccda3290f4 100644 --- a/codex-rs/tui/src/chatwidget/realtime.rs +++ b/codex-rs/tui/src/chatwidget/realtime.rs @@ -1,6 +1,7 @@ //! TUI orchestration for an app-server-signaled, locally owned WebRTC voice session. //! Completed captions and both speakers' partials stay bounded across widget replacement. //! Interleaved speakers retain separate displays so settled caption text never reanimates. +//! Speech recovery suppresses stale queued answers while preserving unspoken text fallbacks. mod recording_controls; mod transcript_replay; @@ -1012,10 +1013,15 @@ impl ChatWidget { pub(crate) fn take_undelivered_realtime_speech_for_replay( &mut self, ) -> Vec<(ThreadId, String, ThreadItem)> { + let input_generation = self.realtime_conversation.input_generation; self.realtime_conversation .pending_speech .drain(..) - .filter(|delivery| !delivery.captioned) + .filter(|delivery| { + !delivery.captioned + && (delivery.state == PendingSpeechState::AwaitingTurn + || delivery.input_generation == input_generation) + }) .map(|delivery| (delivery.thread_id, delivery.turn_id, delivery.item)) .collect() } @@ -1144,10 +1150,14 @@ impl ChatWidget { } fn restore_realtime_speech(&mut self, delivery: PendingRealtimeSpeech) { - if delivery.captioned { - return; - } - if self.thread_id() != Some(delivery.thread_id) { + // Do not append old queued speech under a newer question. An answer that + // never reached speech still needs its text fallback: a delayed user + // transcript can advance the generation after its delegation starts. + if delivery.captioned + || (delivery.state != PendingSpeechState::AwaitingTurn + && delivery.input_generation != self.realtime_conversation.input_generation) + || self.thread_id() != Some(delivery.thread_id) + { return; } self.forget_realtime_turn_origin(&delivery.turn_id); @@ -1185,13 +1195,7 @@ impl ChatWidget { } self.forget_realtime_turn_origin(turn_id); for delivery in waiting { - if self.thread_id() == Some(delivery.thread_id) { - self.handle_thread_item( - delivery.item, - delivery.turn_id, - super::ThreadItemRenderSource::Live, - ); - } + self.restore_realtime_speech(delivery); } } @@ -1842,6 +1846,8 @@ impl ChatWidget { .is_some(); self.realtime_conversation = RealtimeConversationUiState { attempt_id: self.realtime_conversation.attempt_id, + // Keep parked speech tied to its input until it is taken for replay. + input_generation: self.realtime_conversation.input_generation, pending_speech: std::mem::take(&mut self.realtime_conversation.pending_speech), pending_history_cells, accepted_transcripts, diff --git a/codex-rs/tui/src/chatwidget/realtime_tests/snapshots/codex_tui__chatwidget__realtime__tests__speech_recovery__paraphrased_voice_recovery.snap b/codex-rs/tui/src/chatwidget/realtime_tests/snapshots/codex_tui__chatwidget__realtime__tests__speech_recovery__paraphrased_voice_recovery.snap new file mode 100644 index 0000000000..defab35db5 --- /dev/null +++ b/codex-rs/tui/src/chatwidget/realtime_tests/snapshots/codex_tui__chatwidget__realtime__tests__speech_recovery__paraphrased_voice_recovery.snap @@ -0,0 +1,10 @@ +--- +source: tui/src/chatwidget/realtime_tests/speech_recovery.rs +expression: rendered +--- + +› Question 16 + +• Spoken paraphrase 16 + +› Next question diff --git a/codex-rs/tui/src/chatwidget/realtime_tests/speech_recovery.rs b/codex-rs/tui/src/chatwidget/realtime_tests/speech_recovery.rs index 7b695149d9..d3c8c012f8 100644 --- a/codex-rs/tui/src/chatwidget/realtime_tests/speech_recovery.rs +++ b/codex-rs/tui/src/chatwidget/realtime_tests/speech_recovery.rs @@ -3,6 +3,158 @@ use super::*; use pretty_assertions::assert_eq; +fn history_text(events: &mut tokio::sync::mpsc::UnboundedReceiver) -> String { + std::iter::from_fn(|| events.try_recv().ok()) + .filter_map(|event| match event { + AppEvent::InsertHistoryCell(cell) if !cell.as_any().is::() => { + Some(cell.display_lines(/*width*/ 80)) + } + _ => None, + }) + .flatten() + .map(|line| line.to_string()) + .collect::>() + .join("\n") +} + +#[tokio::test] +async fn paraphrased_voice_answers_do_not_reappear_on_overflow_or_stop() { + let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await; + let thread_id = activate_voice(&mut chat); + let mut last_exchange = String::new(); + for index in 0..=super::super::MAX_PENDING_SPEECH_DELIVERIES { + let turn_id = format!("turn-{index}"); + let question = format!("Question {index}"); + chat.on_realtime_transcript_delta("user".into(), question.clone()); + chat.on_realtime_transcript_done("user".into(), question.clone()); + start_item( + &mut chat, + thread_id, + &turn_id, + user_item(&format!( + "{question}" + )), + ); + let answer = agent_item( + &format!("answer-{index}"), + &format!("Backend answer {index}"), + Some(MessagePhase::FinalAnswer), + ); + start_item(&mut chat, thread_id, &turn_id, answer.clone()); + complete_item(&mut chat, thread_id, &turn_id, answer.clone()); + finish_turn( + &mut chat, + thread_id, + &turn_id, + vec![answer], + TurnStatus::Completed, + ); + let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = ops.try_recv().unwrap() + else { + panic!("completed voice turn should queue speech"); + }; + chat.accept_realtime_speech(delivery_id); + let spoken = format!("Spoken paraphrase {index}"); + chat.on_realtime_transcript_delta("assistant".into(), spoken.clone()); + chat.on_realtime_transcript_done("assistant".into(), spoken); + commit_realtime_history_events(&mut chat, &mut events); + last_exchange = history_text(&mut events); + assert!( + !last_exchange.contains("Backend answer"), + "old backend answer appeared during turn {index}" + ); + } + chat.on_realtime_transcript_done("user".into(), "Next question".into()); + chat.stop_realtime_conversation(); + commit_realtime_history_events(&mut chat, &mut events); + let rendered = format!("{last_exchange}\n{}", history_text(&mut events)); + insta::assert_snapshot!("paraphrased_voice_recovery", rendered); +} + +#[tokio::test] +async fn failed_speech_recovers_only_for_the_current_input() { + for superseded in [false, true] { + let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await; + let thread_id = activate_voice(&mut chat); + let turn_id = "voice-turn"; + start_item( + &mut chat, + thread_id, + turn_id, + user_item("question"), + ); + let answer = agent_item( + "answer", + "Undelivered answer", + Some(MessagePhase::FinalAnswer), + ); + start_item(&mut chat, thread_id, turn_id, answer.clone()); + complete_item(&mut chat, thread_id, turn_id, answer.clone()); + finish_turn( + &mut chat, + thread_id, + turn_id, + vec![answer], + TurnStatus::Completed, + ); + let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = ops.try_recv().unwrap() + else { + panic!("completed voice turn should queue speech"); + }; + if superseded { + chat.note_realtime_typed_input("New task"); + } + while events.try_recv().is_ok() {} + chat.restore_undelivered_realtime_speech(delivery_id); + chat.restore_undelivered_realtime_speech(delivery_id); + assert_eq!( + history_text(&mut events), + if superseded { + "" + } else { + "• Undelivered answer" + } + ); + } +} + +#[tokio::test] +async fn delayed_voice_transcript_preserves_unspoken_text_fallback() { + for completed_before_transcript in [false, true] { + let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await; + let thread_id = activate_voice(&mut chat); + let turn_id = "voice-turn"; + start_item( + &mut chat, + thread_id, + turn_id, + user_item("question"), + ); + let answer = agent_item("answer", "Current answer", Some(MessagePhase::FinalAnswer)); + start_item(&mut chat, thread_id, turn_id, answer.clone()); + if completed_before_transcript { + complete_item(&mut chat, thread_id, turn_id, answer.clone()); + } + chat.on_realtime_transcript_done("user".into(), "question".into()); + commit_realtime_history_events(&mut chat, &mut events); + while events.try_recv().is_ok() {} + if !completed_before_transcript { + complete_item(&mut chat, thread_id, turn_id, answer.clone()); + } + finish_turn( + &mut chat, + thread_id, + turn_id, + vec![answer], + TurnStatus::Completed, + ); + insta::allow_duplicates! { + insta::assert_snapshot!(history_text(&mut events), @"• Current answer"); + } + assert!(ops.try_recv().is_err()); + } +} + #[tokio::test] async fn queued_voice_answers_return_to_history_once_if_voice_closes_before_delivery() { let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await; @@ -70,7 +222,7 @@ async fn queued_voice_answers_return_to_history_once_if_voice_closes_before_deli } #[tokio::test] -async fn hidden_and_queued_voice_answers_share_a_lossless_sixteen_item_cap() { +async fn pending_voice_answers_stay_bounded_without_restoring_old_inputs() { let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await; let thread_id = activate_voice(&mut chat); let mut oldest_delivery_id = None; @@ -79,6 +231,7 @@ async fn hidden_and_queued_voice_answers_share_a_lossless_sixteen_item_cap() { chat.park_voice(); } let turn_id = format!("turn-{index}"); + chat.on_realtime_transcript_done("user".into(), format!("Question {index}")); start_item( &mut chat, thread_id, @@ -92,19 +245,18 @@ async fn hidden_and_queued_voice_answers_share_a_lossless_sixteen_item_cap() { ); start_item(&mut chat, thread_id, &turn_id, answer.clone()); complete_item(&mut chat, thread_id, &turn_id, answer.clone()); + finish_turn( + &mut chat, + thread_id, + &turn_id, + vec![answer], + TurnStatus::Completed, + ); + let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = ops.try_recv().unwrap() + else { + panic!("completed voice turn should queue speech"); + }; if index == 0 { - finish_turn( - &mut chat, - thread_id, - &turn_id, - vec![answer], - TurnStatus::Completed, - ); - let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = - ops.try_recv().unwrap() - else { - panic!("first voice answer should queue speech"); - }; oldest_delivery_id = Some(delivery_id); } } @@ -124,13 +276,22 @@ async fn hidden_and_queued_voice_answers_share_a_lossless_sixteen_item_cap() { restored += 1; } } - assert_eq!(restored, 1); + assert_eq!(restored, 0); assert!(ops.try_recv().is_err()); chat.stop_realtime_conversation(); chat.reset_realtime_conversation(); + let last = 2 * super::super::MAX_PENDING_SPEECH_DELIVERIES; assert_eq!( - chat.take_undelivered_realtime_speech_for_replay().len(), - super::super::MAX_PENDING_SPEECH_DELIVERIES + chat.take_undelivered_realtime_speech_for_replay(), + vec![( + thread_id, + format!("turn-{last}"), + agent_item( + &format!("answer-{last}"), + &format!("Answer number {last}"), + Some(MessagePhase::FinalAnswer), + ), + )] ); }