Текстовый файл из учебного комплекта, не свидетельство установленного runtime. Вернуться к учебнику · Исходный файл
1 //! TUI orchestration for an app-server-signaled, locally owned WebRTC voice session. 2 //! Completed captions and both speakers' partials stay bounded across widget replacement. 3 //! Interleaved speakers retain separate displays so settled caption text never reanimates. 4 5 mod recording_controls; 6 mod transcript_replay; 7 8 use super::ChatWidget; 9 use super::HistoryCell; 10 use super::PARENT_OWNED_INPUT_MESSAGE; 11 use super::realtime_split_flap::SplitFlapTranscriptCell; 12 use super::realtime_split_flap::VoiceAmplitudeHistory; 13 use crate::app_command::AppCommand; 14 use crate::app_event::AppEvent; 15 use crate::bottom_pane::VoiceStripPhase; 16 use crate::bottom_pane::VoiceStripState; 17 use crate::history_cell; 18 use crate::key_hint::KeyBindingListExt; 19 use crate::motion::MotionMode; 20 use codex_app_server_protocol::ThreadItem; 21 use codex_app_server_protocol::UserInput; 22 use codex_features::Feature; 23 use codex_protocol::ThreadId; 24 use codex_protocol::models::MessagePhase; 25 use codex_realtime_webrtc::RealtimeWebrtcSession; 26 use codex_realtime_webrtc::RealtimeWebrtcSessionHandle; 27 use codex_realtime_webrtc::StartedRealtimeWebrtcSession; 28 use crossterm::event::KeyEvent; 29 use crossterm::event::KeyEventKind; 30 use futures::future::AbortHandle; 31 use std::collections::HashMap; 32 use std::collections::VecDeque; 33 use std::collections::hash_map::DefaultHasher; 34 use std::hash::Hash; 35 use std::hash::Hasher; 36 use std::sync::atomic::AtomicU64; 37 use std::sync::atomic::Ordering; 38 use std::time::Duration; 39 use std::time::Instant; 40 41 #[cfg(test)] 42 #[path = "realtime_tests.rs"] 43 pub(crate) mod tests; 44 45 pub(crate) const MAX_TRANSCRIPT_BYTES: usize = 1024; 46 const MAX_PENDING_TRANSCRIPT_CELLS: usize = 32; 47 pub(crate) const MAX_REPLAY_TRANSCRIPT_CELLS: usize = MAX_PENDING_TRANSCRIPT_CELLS + 2; 48 const MAX_PENDING_SPEECH_DELIVERIES: usize = 16; 49 const MAX_RETAINED_DELEGATED_REASONING_TURNS: usize = 16; 50 const MAX_PENDING_SPEECH_ITEM_BYTES: usize = 64 * 1024; 51 const MAX_PENDING_SPEECH_TURN_ID_BYTES: usize = 512; 52 // Leave room for Core's optional backend prefix within its 1,000-token speech limit. 53 const MAX_SPEAKABLE_FINAL_TOKENS: usize = 990; 54 const AUDIO_METER_SEGMENTS: usize = 5; 55 const AUDIO_METER_NOISE_FLOOR: u16 = 512; 56 const AUDIO_METER_FULL_SCALE: u16 = 8192; 57 const MAX_REALTIME_AUDIO_METER_FRAMES: usize = 12; 58 const MICROPHONE_METER_INTERVAL: Duration = Duration::from_millis(100); 59 const SPEAKER_ACTIVITY_HOLD: Duration = Duration::from_millis(500); 60 const INTERRUPTION_ACKNOWLEDGMENT: Duration = Duration::from_millis(400); 61 static NEXT_REALTIME_ATTEMPT_ID: AtomicU64 = AtomicU64::new(1); 62 static NEXT_REALTIME_SPEECH_DELIVERY_ID: AtomicU64 = AtomicU64::new(1); 63 64 struct PendingRealtimeSpeech { 65 state: PendingSpeechState, 66 captioned: bool, 67 input_generation: u64, 68 thread_id: ThreadId, 69 turn_id: String, 70 item: ThreadItem, 71 } 72 73 pub(crate) struct RealtimeTranscriptRecord { 74 pub(crate) role: String, 75 pub(crate) text: String, 76 pub(crate) complete: bool, 77 pub(crate) before_turn_id: Option<String>, 78 } 79 80 #[derive(Clone, Copy, PartialEq, Eq)] 81 enum PendingSpeechState { 82 AwaitingTurn, 83 Queued(u64), 84 Accepted, 85 } 86 87 fn can_retain_realtime_speech(turn_id: &str, item: &ThreadItem) -> bool { 88 turn_id.len() <= MAX_PENDING_SPEECH_TURN_ID_BYTES 89 && serde_json::to_vec(item) 90 .is_ok_and(|encoded| encoded.len() <= MAX_PENDING_SPEECH_ITEM_BYTES) 91 } 92 93 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] 94 pub(super) enum RealtimeConversationPhase { 95 #[default] 96 Inactive, 97 Starting, 98 Active, 99 Stopping, 100 } 101 102 #[derive(Clone, Copy, Debug)] 103 pub(super) enum RealtimeAgentItemOrigin { 104 Typed, 105 Delegated { 106 may_speak: bool, 107 completed: bool, 108 suppressed_nonfinal: bool, 109 input_generation: u64, 110 }, 111 } 112 113 #[derive(Clone, Copy, Debug)] 114 enum RealtimeTurnOrigin { 115 Typed { 116 input_generation: u64, 117 }, 118 Delegated { 119 may_speak: bool, 120 input_generation: u64, 121 }, 122 } 123 124 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] 125 enum StartupRetry { 126 #[default] 127 Available, 128 WaitingForStop, 129 Used, 130 } 131 132 #[derive(Default)] 133 pub(super) struct RealtimeConversationUiState { 134 startup_retry: StartupRetry, 135 pub(super) phase: RealtimeConversationPhase, 136 recover_late_transcripts: bool, 137 attempt_id: u64, 138 thread_id: Option<ThreadId>, 139 pub(super) handle: Option<RealtimeWebrtcSessionHandle>, 140 startup_abort: Option<AbortHandle>, 141 backend_started: bool, 142 webrtc_connected: bool, 143 active_since: Option<Instant>, 144 failure_recorded: bool, 145 microphone_muted: bool, 146 microphone_level: usize, 147 speaker_level: usize, 148 microphone_intensity: u8, 149 speaker_intensity: u8, 150 microphone_history: VoiceAmplitudeHistory, 151 speaker_history: VoiceAmplitudeHistory, 152 audio_meter_history: VecDeque<(u8, u8)>, 153 next_audio_meter_sample_at: Option<Instant>, 154 speaker_active_until: Option<Instant>, 155 interruption_acknowledged_until: Option<Instant>, 156 speaker_suppression_generation: Option<u64>, 157 transcript_role: Option<String>, 158 transcript: String, 159 // The other speaker's bounded partial while duplex deltas interleave. 160 interleaved_transcript: Option<(String, String)>, 161 interleaved_transcript_cell: Option<Box<dyn HistoryCell>>, 162 transcript_input_generation: Option<u64>, 163 assistant_transcript_generation: Option<u64>, 164 assistant_caption_started_after_speech_queue: bool, 165 pub(super) live_transcript_cell: Option<Box<dyn HistoryCell>>, 166 pub(super) pending_history_cells: VecDeque<Box<dyn HistoryCell>>, 167 accepted_transcripts: VecDeque<RealtimeTranscriptRecord>, 168 replay_transcripts: Option<VecDeque<RealtimeTranscriptRecord>>, 169 latest_input_was_voice: bool, 170 input_generation: u64, 171 latest_voice_input_fingerprint: Option<(usize, u64)>, 172 pending_typed_input: Option<String>, 173 turn_origins: HashMap<String, RealtimeTurnOrigin>, 174 delegated_reasoning_turns: VecDeque<String>, 175 pub(super) agent_items: HashMap<(String, String), RealtimeAgentItemOrigin>, 176 pending_speech: VecDeque<PendingRealtimeSpeech>, 177 } 178 179 impl RealtimeConversationUiState { 180 pub(super) fn live_transcript_cells(&self) -> impl Iterator<Item = &Box<dyn HistoryCell>> { 181 // Keep the user above the reply even when their last packets interleave. 182 let cells = if self.transcript_role.as_deref() == Some("user") { 183 [ 184 &self.live_transcript_cell, 185 &self.interleaved_transcript_cell, 186 ] 187 } else { 188 [ 189 &self.interleaved_transcript_cell, 190 &self.live_transcript_cell, 191 ] 192 }; 193 cells.into_iter().filter_map(Option::as_ref) 194 } 195 } 196 197 pub(crate) fn realtime_delegation_input(items: &[UserInput]) -> Option<&str> { 198 let [ 199 UserInput::Text { 200 text, 201 text_elements, 202 }, 203 ] = items 204 else { 205 return None; 206 }; 207 if !text_elements.is_empty() { 208 return None; 209 } 210 let body = text 211 .trim() 212 .strip_prefix("<realtime_delegation>") 213 .and_then(|body| body.strip_suffix("</realtime_delegation>"))?; 214 let (_, input) = body.split_once("<input>")?; 215 input.split_once("</input>").map(|(input, _)| input) 216 } 217 218 pub(crate) fn is_private_realtime_agent_item(item: &ThreadItem) -> bool { 219 if matches!(item, ThreadItem::Reasoning { .. }) { 220 return true; 221 } 222 let ThreadItem::AgentMessage { text, phase, .. } = item else { 223 return false; 224 }; 225 let text = text.trim_start(); 226 matches!(phase, Some(MessagePhase::Commentary)) 227 || (!matches!(phase, Some(MessagePhase::FinalAnswer)) 228 && (text.starts_with("[ANALYSIS]") || text.starts_with("[COMMENTARY]"))) 229 } 230 231 pub(crate) fn realtime_delegation_display_text(input: &str) -> String { 232 input 233 .replace("<", "<") 234 .replace(">", ">") 235 .replace("&", "&") 236 } 237 238 fn realtime_input_fingerprint(input: &str) -> (usize, u64) { 239 let input = input.trim(); 240 let mut hasher = DefaultHasher::new(); 241 input.hash(&mut hasher); 242 (input.len(), hasher.finish()) 243 } 244 245 impl ChatWidget { 246 pub(crate) fn realtime_conversation_is_running(&self) -> bool { 247 self.realtime_conversation.phase != RealtimeConversationPhase::Inactive 248 } 249 250 pub(crate) fn may_receive_realtime_transcripts(&self) -> bool { 251 self.realtime_conversation.phase != RealtimeConversationPhase::Inactive 252 || self.realtime_conversation.recover_late_transcripts 253 } 254 255 pub(crate) fn toggle_realtime_conversation(&mut self) { 256 if self.realtime_conversation.phase == RealtimeConversationPhase::Stopping { 257 self.realtime_conversation.startup_retry = StartupRetry::Used; 258 self.add_info_message( 259 "Voice conversation is still stopping.".to_string(), 260 /*hint*/ None, 261 ); 262 return; 263 } 264 265 if self.realtime_conversation.phase != RealtimeConversationPhase::Inactive { 266 self.stop_realtime_conversation(); 267 return; 268 } 269 270 if self.has_misalignment_policy_violation() { 271 self.show_misalignment_policy_precaution(); 272 return; 273 } 274 275 if self.side_conversation_active() { 276 self.add_error_message( 277 "Voice mode is unavailable in side conversations. Return to the main thread first." 278 .to_string(), 279 ); 280 return; 281 } 282 283 if self.blocks_direct_input { 284 self.add_error_message(PARENT_OWNED_INPUT_MESSAGE.to_string()); 285 return; 286 } 287 288 if !self.config.features.enabled(Feature::RealtimeConversation) { 289 self.add_error_message("Voice conversations are not enabled.".to_string()); 290 return; 291 } 292 293 if !RealtimeWebrtcSession::is_supported() { 294 self.add_error_message( 295 "Voice requires macOS, an MSVC-based Windows build, or a glibc-based Linux build." 296 .to_string(), 297 ); 298 return; 299 } 300 301 let Some(thread_id) = self.thread_id() else { 302 self.add_error_message("Start a conversation before using voice mode.".to_string()); 303 return; 304 }; 305 306 self.session_telemetry 307 .counter("codex.voice.session.start", /*inc*/ 1, &[]); 308 self.start_realtime_conversation(thread_id); 309 } 310 311 fn start_realtime_conversation(&mut self, thread_id: ThreadId) { 312 self.realtime_conversation.recover_late_transcripts = false; 313 self.realtime_conversation.attempt_id = 314 NEXT_REALTIME_ATTEMPT_ID.fetch_add(1, Ordering::Relaxed); 315 self.realtime_conversation.thread_id = Some(thread_id); 316 let attempt_id = self.realtime_conversation.attempt_id; 317 let (startup_abort, abort_registration) = AbortHandle::new_pair(); 318 self.realtime_conversation.startup_abort = Some(startup_abort); 319 self.realtime_conversation.phase = RealtimeConversationPhase::Starting; 320 self.update_realtime_footer(); 321 let app_event_tx = self.app_event_tx.clone(); 322 std::thread::spawn(move || { 323 let result = 324 RealtimeWebrtcSession::start(abort_registration).map_err(|error| error.to_string()); 325 app_event_tx.send(AppEvent::RealtimeWebrtcOfferCreated { 326 thread_id, 327 attempt_id, 328 result, 329 }); 330 }); 331 self.request_redraw(); 332 } 333 334 pub(super) fn stop_realtime_conversation(&mut self) { 335 self.realtime_conversation.startup_retry = StartupRetry::Used; 336 if matches!( 337 self.realtime_conversation.phase, 338 RealtimeConversationPhase::Inactive | RealtimeConversationPhase::Stopping 339 ) { 340 return; 341 } 342 self.finish_realtime_session_metrics(); 343 self.restore_all_undelivered_realtime_speech(); 344 345 // Stopping may prevent final transcript events from arriving. 346 self.finish_realtime_partial_transcripts(); 347 348 if let Some(handle) = self.realtime_conversation.handle.take() { 349 handle.close(); 350 self.realtime_conversation.phase = RealtimeConversationPhase::Stopping; 351 self.refresh_terminal_title(); 352 self.bottom_pane.set_voice_strip(/*state*/ None); 353 let Some(thread_id) = self.realtime_conversation.thread_id else { 354 self.reset_realtime_conversation(); 355 return; 356 }; 357 if !self.submit_op(AppCommand::RealtimeConversationStop { thread_id }) { 358 self.record_realtime_failure(); 359 self.reset_realtime_conversation(); 360 } 361 } else { 362 if let Some(thread_id) = self.reset_realtime_conversation() { 363 self.submit_op(AppCommand::RealtimeConversationStop { thread_id }); 364 } 365 } 366 self.request_redraw(); 367 } 368 369 pub(crate) fn on_realtime_webrtc_offer_created( 370 &mut self, 371 thread_id: ThreadId, 372 attempt_id: u64, 373 result: Result<StartedRealtimeWebrtcSession, String>, 374 ) { 375 if self.realtime_conversation.phase != RealtimeConversationPhase::Starting 376 || self.realtime_conversation.attempt_id != attempt_id 377 { 378 if let Ok(offer) = result { 379 offer.handle.close(); 380 } 381 return; 382 } 383 384 let offer = match result { 385 Ok(offer) => offer, 386 Err(error) => { 387 self.on_realtime_error(format!("Failed to start voice mode: {error}")); 388 return; 389 } 390 }; 391 self.realtime_conversation.startup_abort = None; 392 if self.realtime_conversation.microphone_muted 393 && let Err(error) = offer.handle.set_microphone_muted(/*muted*/ true) 394 { 395 offer.handle.close(); 396 self.on_realtime_error(format!("Failed to restore microphone mute: {error}")); 397 return; 398 } 399 self.realtime_conversation.handle = Some(offer.handle); 400 self.update_realtime_footer(); 401 self.refresh_terminal_title(); 402 self.frame_requester 403 .schedule_frame_in(MICROPHONE_METER_INTERVAL); 404 if !self.submit_op(AppCommand::RealtimeConversationStart { 405 thread_id, 406 offer_sdp: offer.offer_sdp.into(), 407 }) { 408 self.record_realtime_failure(); 409 self.reset_realtime_conversation(); 410 } 411 } 412 413 pub(super) fn on_realtime_conversation_sdp(&mut self, answer_sdp: String) { 414 if self.realtime_conversation.phase != RealtimeConversationPhase::Starting { 415 return; 416 } 417 let Some(handle) = self.realtime_conversation.handle.clone() else { 418 return; 419 }; 420 let Some(thread_id) = self.thread_id() else { 421 return; 422 }; 423 let attempt_id = self.realtime_conversation.attempt_id; 424 let app_event_tx = self.app_event_tx.clone(); 425 std::thread::spawn(move || { 426 let result = handle.apply_answer_sdp(answer_sdp); 427 app_event_tx.send(AppEvent::RealtimeWebrtcConnected { 428 thread_id, 429 attempt_id, 430 result, 431 }); 432 }); 433 } 434 435 pub(super) fn on_realtime_conversation_started(&mut self) { 436 if self.realtime_conversation.phase != RealtimeConversationPhase::Starting { 437 return; 438 } 439 self.realtime_conversation.backend_started = true; 440 self.maybe_activate_realtime_conversation(); 441 } 442 443 pub(crate) fn on_realtime_webrtc_connected( 444 &mut self, 445 attempt_id: u64, 446 result: Result<(), codex_realtime_webrtc::ConnectionError>, 447 ) { 448 if self.realtime_conversation.phase != RealtimeConversationPhase::Starting 449 || self.realtime_conversation.attempt_id != attempt_id 450 { 451 return; 452 } 453 if let Err(error) = result { 454 if error == codex_realtime_webrtc::ConnectionError::NegotiationTimedOut 455 && self.realtime_conversation.startup_retry == StartupRetry::Available 456 && let Some(thread_id) = self.realtime_conversation.thread_id 457 { 458 if let Some(handle) = self.realtime_conversation.handle.take() { 459 handle.close(); 460 } 461 self.realtime_conversation.phase = RealtimeConversationPhase::Stopping; 462 self.realtime_conversation.startup_retry = StartupRetry::WaitingForStop; 463 self.refresh_terminal_title(); 464 if !self.submit_op(AppCommand::RealtimeConversationStop { thread_id }) { 465 self.record_realtime_failure(); 466 self.reset_realtime_conversation(); 467 } else { 468 self.add_info_message( 469 "Voice connection timed out. Retrying once after cleanup.".into(), 470 /*hint*/ None, 471 ); 472 } 473 return; 474 } 475 self.on_realtime_error(format!("Failed to connect voice mode: {error}")); 476 return; 477 } 478 479 self.realtime_conversation.webrtc_connected = true; 480 self.maybe_activate_realtime_conversation(); 481 } 482 483 fn maybe_activate_realtime_conversation(&mut self) { 484 if self.realtime_conversation.phase != RealtimeConversationPhase::Starting 485 || !self.realtime_conversation.backend_started 486 || !self.realtime_conversation.webrtc_connected 487 { 488 return; 489 } 490 self.realtime_conversation.phase = RealtimeConversationPhase::Active; 491 self.realtime_conversation.active_since = Some(Instant::now()); 492 self.session_telemetry 493 .counter("codex.voice.session.connected", /*inc*/ 1, &[]); 494 let running_delegation = 495 self.turn_lifecycle 496 .last_turn_id 497 .as_deref() 498 .is_some_and(|turn_id| { 499 matches!( 500 self.realtime_conversation.turn_origins.get(turn_id), 501 Some(RealtimeTurnOrigin::Delegated { 502 may_speak: true, 503 .. 504 }) 505 ) 506 }); 507 self.realtime_conversation.latest_input_was_voice = 508 !self.turn_lifecycle.agent_turn_running || running_delegation; 509 if self.turn_lifecycle.agent_turn_running 510 && let Some(turn_id) = self.turn_lifecycle.last_turn_id.clone() 511 { 512 self.realtime_conversation 513 .turn_origins 514 .entry(turn_id) 515 .or_insert(RealtimeTurnOrigin::Typed { 516 input_generation: self.realtime_conversation.input_generation, 517 }); 518 } 519 self.update_realtime_footer(); 520 self.frame_requester 521 .schedule_frame_in(MICROPHONE_METER_INTERVAL); 522 self.add_info_message( 523 "Voice conversation started.".to_string(), 524 Some("Use /voice mute to mute or /voice to stop.".to_string()), 525 ); 526 } 527 528 pub(crate) fn is_current_realtime_attempt( 529 &self, 530 thread_id: ThreadId, 531 attempt_id: u64, 532 input_generation: u64, 533 ) -> bool { 534 self.realtime_conversation.phase == RealtimeConversationPhase::Active 535 && self.realtime_conversation.thread_id == Some(thread_id) 536 && self.realtime_conversation.attempt_id == attempt_id 537 && self.realtime_conversation.input_generation == input_generation 538 && self.realtime_conversation.latest_input_was_voice 539 } 540 541 pub(super) fn note_realtime_typed_input(&mut self, text: &str) { 542 self.invalidate_realtime_voice_input(); 543 if self.realtime_conversation.phase == RealtimeConversationPhase::Active { 544 self.realtime_conversation.pending_typed_input = Some(text.to_string()); 545 } 546 } 547 548 fn invalidate_realtime_voice_input(&mut self) { 549 if self.realtime_conversation.phase != RealtimeConversationPhase::Active { 550 return; 551 } 552 self.realtime_conversation.latest_input_was_voice = false; 553 self.realtime_conversation.input_generation = self 554 .realtime_conversation 555 .input_generation 556 .wrapping_add(/*rhs*/ 1); 557 // Typing supersedes the old voice answer. Suppress the helper before 558 // another queued audio frame can play, even when the mic is muted. 559 self.realtime_conversation.speaker_suppression_generation = 560 Some(self.realtime_conversation.input_generation); 561 if let Some(handle) = self.realtime_conversation.handle.as_ref() { 562 handle.set_speaker_suppressed(/*suppressed*/ true); 563 } 564 self.realtime_conversation.speaker_level = 0; 565 self.realtime_conversation.speaker_active_until = None; 566 self.update_realtime_footer(); 567 for origin in self.realtime_conversation.turn_origins.values_mut() { 568 if let RealtimeTurnOrigin::Delegated { may_speak, .. } = origin { 569 *may_speak = false; 570 } 571 } 572 for origin in self.realtime_conversation.agent_items.values_mut() { 573 if let RealtimeAgentItemOrigin::Delegated { may_speak, .. } = origin { 574 *may_speak = false; 575 } 576 } 577 } 578 579 pub(super) fn note_realtime_user_item_started(&mut self, turn_id: &str, items: &[UserInput]) { 580 if !matches!( 581 self.realtime_conversation.phase, 582 RealtimeConversationPhase::Starting | RealtimeConversationPhase::Active 583 ) { 584 if realtime_delegation_input(items).is_some() { 585 self.remember_realtime_delegated_reasoning_turn(turn_id); 586 } 587 return; 588 } 589 let matches_local_typed_input = self 590 .realtime_conversation 591 .pending_typed_input 592 .as_ref() 593 .is_some_and(|pending| { 594 items 595 .iter() 596 .any(|item| matches!(item, UserInput::Text { text, .. } if text == pending)) 597 }); 598 if matches_local_typed_input { 599 self.realtime_conversation.pending_typed_input = None; 600 } 601 if let Some(input) = realtime_delegation_input(items) 602 && !matches_local_typed_input 603 { 604 let transcript_tail_flush = items.iter().any(|item| { 605 matches!(item, UserInput::Text { text, .. } 606 if text.contains("<source>transcript_tail_flush</source>")) 607 }); 608 let stale_voice_input = !self.realtime_conversation.latest_input_was_voice 609 && self 610 .realtime_conversation 611 .latest_voice_input_fingerprint 612 .is_some_and(|latest| { 613 latest 614 == realtime_input_fingerprint(&realtime_delegation_display_text(input)) 615 }); 616 let (typed_turn_generation, delegated_turn_generation) = 617 match self.realtime_conversation.turn_origins.get(turn_id) { 618 Some(RealtimeTurnOrigin::Typed { input_generation }) => { 619 (Some(*input_generation), None) 620 } 621 Some(RealtimeTurnOrigin::Delegated { 622 input_generation, .. 623 }) => (None, Some(*input_generation)), 624 None => (None, None), 625 }; 626 let voice_supersedes_typed_turn = typed_turn_generation.is_some_and(|generation| { 627 generation != self.realtime_conversation.input_generation 628 || !self.realtime_conversation.latest_input_was_voice 629 }) && !transcript_tail_flush 630 && !stale_voice_input; 631 let typed_turn = typed_turn_generation.is_some() && !voice_supersedes_typed_turn; 632 let may_speak = !transcript_tail_flush && !stale_voice_input && !typed_turn; 633 if may_speak { 634 let replaces_current_delegation = self 635 .realtime_conversation 636 .turn_origins 637 .values() 638 .any(|origin| { 639 matches!( 640 origin, 641 RealtimeTurnOrigin::Delegated { 642 may_speak: true, 643 input_generation, 644 } if *input_generation == self.realtime_conversation.input_generation 645 ) 646 }); 647 if !self.realtime_conversation.latest_input_was_voice || replaces_current_delegation 648 { 649 self.realtime_conversation.input_generation = self 650 .realtime_conversation 651 .input_generation 652 .wrapping_add(/*rhs*/ 1); 653 if self 654 .realtime_conversation 655 .speaker_suppression_generation 656 .is_some() 657 { 658 self.realtime_conversation.speaker_suppression_generation = 659 Some(self.realtime_conversation.input_generation); 660 } 661 } 662 self.realtime_conversation.latest_input_was_voice = true; 663 } 664 let delegated_origin = RealtimeTurnOrigin::Delegated { 665 may_speak, 666 input_generation: self.realtime_conversation.input_generation, 667 }; 668 let supersedes_previous_delegation = may_speak 669 && delegated_turn_generation.is_some_and(|generation| { 670 generation != self.realtime_conversation.input_generation 671 }); 672 if voice_supersedes_typed_turn || supersedes_previous_delegation { 673 self.realtime_conversation 674 .turn_origins 675 .insert(turn_id.to_string(), delegated_origin); 676 } else { 677 self.realtime_conversation 678 .turn_origins 679 .entry(turn_id.to_string()) 680 .or_insert(delegated_origin); 681 } 682 if matches!( 683 self.realtime_conversation.turn_origins.get(turn_id), 684 Some(RealtimeTurnOrigin::Delegated { .. }) 685 ) { 686 self.remember_realtime_delegated_reasoning_turn(turn_id); 687 } 688 return; 689 } 690 691 self.invalidate_realtime_voice_input(); 692 self.realtime_conversation.turn_origins.insert( 693 turn_id.to_string(), 694 RealtimeTurnOrigin::Typed { 695 input_generation: self.realtime_conversation.input_generation, 696 }, 697 ); 698 self.realtime_conversation 699 .delegated_reasoning_turns 700 .retain(|saved| saved != turn_id); 701 } 702 703 pub(crate) fn remember_realtime_delegated_reasoning_turn(&mut self, turn_id: &str) { 704 if turn_id.len() > 512 705 || self 706 .realtime_conversation 707 .delegated_reasoning_turns 708 .iter() 709 .any(|saved| saved == turn_id) 710 { 711 return; 712 } 713 self.realtime_conversation 714 .delegated_reasoning_turns 715 .push_back(turn_id.to_string()); 716 if self.realtime_conversation.delegated_reasoning_turns.len() 717 > MAX_RETAINED_DELEGATED_REASONING_TURNS 718 { 719 self.realtime_conversation 720 .delegated_reasoning_turns 721 .pop_front(); 722 } 723 } 724 725 pub(super) fn is_realtime_delegated_reasoning_turn(&self, turn_id: &str) -> bool { 726 self.realtime_conversation 727 .delegated_reasoning_turns 728 .iter() 729 .any(|saved| saved == turn_id) 730 } 731 732 pub(super) fn is_realtime_delegated_reasoning_item( 733 &self, 734 turn_id: &str, 735 item_id: &str, 736 ) -> bool { 737 self.is_realtime_delegated_reasoning_turn(turn_id) 738 && !matches!( 739 self.realtime_conversation 740 .agent_items 741 .get(&(turn_id.to_string(), item_id.to_string())), 742 Some(RealtimeAgentItemOrigin::Typed) 743 ) 744 } 745 746 pub(super) fn should_hide_realtime_delegation(&self, turn_id: &str) -> bool { 747 matches!( 748 self.realtime_conversation.phase, 749 RealtimeConversationPhase::Starting | RealtimeConversationPhase::Active 750 ) && matches!( 751 self.realtime_conversation.turn_origins.get(turn_id), 752 Some(RealtimeTurnOrigin::Delegated { .. }) 753 ) 754 } 755 756 pub(super) fn is_realtime_delegated_agent_item( 757 &mut self, 758 turn_id: &str, 759 item_id: &str, 760 ) -> bool { 761 if !matches!( 762 self.realtime_conversation.phase, 763 RealtimeConversationPhase::Starting | RealtimeConversationPhase::Active 764 ) { 765 return false; 766 } 767 let default_origin = match self.realtime_conversation.turn_origins.get(turn_id) { 768 Some(RealtimeTurnOrigin::Delegated { 769 may_speak, 770 input_generation, 771 }) => RealtimeAgentItemOrigin::Delegated { 772 may_speak: *may_speak, 773 completed: false, 774 suppressed_nonfinal: false, 775 input_generation: *input_generation, 776 }, 777 Some(RealtimeTurnOrigin::Typed { .. }) | None => RealtimeAgentItemOrigin::Typed, 778 }; 779 matches!( 780 self.realtime_conversation 781 .agent_items 782 .entry((turn_id.to_string(), item_id.to_string())) 783 .or_insert(default_origin), 784 RealtimeAgentItemOrigin::Delegated { .. } 785 ) 786 } 787 788 pub(super) fn complete_realtime_delegated_agent_item( 789 &mut self, 790 turn_id: &str, 791 item: &ThreadItem, 792 from_replay: bool, 793 ) -> bool { 794 let ThreadItem::AgentMessage { 795 id: item_id, 796 text, 797 phase, 798 questions, 799 .. 800 } = item 801 else { 802 return false; 803 }; 804 // The ordinary completion path opens the structured question editor. 805 // A question-bearing answer must remain visible instead of being speech-only. 806 if questions.is_some() { 807 return false; 808 } 809 if from_replay || !self.is_realtime_delegated_agent_item(turn_id, item_id) { 810 return false; 811 } 812 let Some(RealtimeAgentItemOrigin::Delegated { 813 may_speak, 814 completed, 815 suppressed_nonfinal, 816 .. 817 }) = self 818 .realtime_conversation 819 .agent_items 820 .get_mut(&(turn_id.to_string(), item_id.to_string())) 821 else { 822 return false; 823 }; 824 if *completed { 825 return true; 826 } 827 // The delegated agent's commentary and reasoning are private to the 828 // voice handoff. Never put them in the fallback queue, even when the 829 // item is too large to retain. 830 let trimmed = text.trim(); 831 let explicitly_final = matches!(phase, Some(MessagePhase::FinalAnswer)); 832 if is_private_realtime_agent_item(item) 833 || (!explicitly_final && trimmed == "[FINAL]") 834 || trimmed.is_empty() 835 { 836 *completed = true; 837 *suppressed_nonfinal = true; 838 return true; 839 } 840 // Typing ends the old item's speech ownership, but it does not make 841 // private voice commentary into typed output. Only a visible answer 842 // returns to the ordinary completion path. 843 if !*may_speak { 844 return false; 845 } 846 // A final item that cannot be retained for delivery recovery stays in 847 // ordinary history instead of becoming a voice-only answer. 848 if !can_retain_realtime_speech(turn_id, item) { 849 return false; 850 } 851 let Some(thread_id) = self.realtime_conversation.thread_id else { 852 return false; 853 }; 854 let input_generation = self.realtime_conversation.input_generation; 855 *completed = true; 856 self.transcript.last_completed_agent_message = 857 Some((turn_id.to_string(), item_id.to_string())); 858 if self.realtime_conversation.pending_speech.len() >= MAX_PENDING_SPEECH_DELIVERIES 859 && let Some(oldest) = self.realtime_conversation.pending_speech.pop_front() 860 { 861 self.restore_realtime_speech(oldest); 862 } 863 self.realtime_conversation 864 .pending_speech 865 .push_back(PendingRealtimeSpeech { 866 state: PendingSpeechState::AwaitingTurn, 867 captioned: false, 868 input_generation, 869 thread_id, 870 turn_id: turn_id.to_string(), 871 item: item.clone(), 872 }); 873 874 true 875 } 876 877 pub(super) fn speak_completed_realtime_delegation(&mut self, turn_id: &str, item: &ThreadItem) { 878 let ThreadItem::AgentMessage { 879 id: item_id, text, .. 880 } = item 881 else { 882 return; 883 }; 884 let Some(RealtimeAgentItemOrigin::Delegated { 885 may_speak: true, 886 completed: true, 887 suppressed_nonfinal, 888 input_generation, 889 }) = self 890 .realtime_conversation 891 .agent_items 892 .get(&(turn_id.to_string(), item_id.to_string())) 893 else { 894 return; 895 }; 896 let input_generation = *input_generation; 897 let suppressed_nonfinal = *suppressed_nonfinal; 898 if input_generation != self.realtime_conversation.input_generation { 899 return; 900 } 901 let trimmed = text.trim(); 902 if is_private_realtime_agent_item(item) 903 || trimmed.is_empty() 904 || !self.realtime_conversation.latest_input_was_voice 905 { 906 return; 907 } 908 let Some(thread_id) = self.realtime_conversation.thread_id else { 909 return; 910 }; 911 if self.thread_id() != Some(thread_id) { 912 return; 913 } 914 let text = trimmed 915 .strip_prefix("[FINAL]") 916 .map(str::trim_start) 917 .unwrap_or(trimmed) 918 .to_string(); 919 if text.is_empty() { 920 return; 921 } 922 let Some(RealtimeTurnOrigin::Delegated { 923 may_speak, 924 input_generation: turn_input_generation, 925 }) = self.realtime_conversation.turn_origins.get_mut(turn_id) 926 else { 927 return; 928 }; 929 if !*may_speak || *turn_input_generation != input_generation { 930 return; 931 } 932 *may_speak = false; 933 if !can_retain_realtime_speech(turn_id, item) 934 || codex_utils_string::approx_token_count(&text) > MAX_SPEAKABLE_FINAL_TOKENS 935 { 936 self.remove_waiting_realtime_speech(turn_id, item_id); 937 self.finish_realtime_turn(turn_id); 938 self.handle_thread_item( 939 item.clone(), 940 turn_id.to_string(), 941 super::ThreadItemRenderSource::Live, 942 ); 943 return; 944 } 945 let delivery_id = NEXT_REALTIME_SPEECH_DELIVERY_ID.fetch_add(1, Ordering::Relaxed); 946 let index = self 947 .realtime_conversation 948 .pending_speech 949 .iter_mut() 950 .position(|pending| { 951 pending.turn_id == turn_id 952 && matches!(&pending.item, ThreadItem::AgentMessage { id, .. } if id == item_id) 953 && pending.state == PendingSpeechState::AwaitingTurn 954 }); 955 let index = match index { 956 Some(index) => index, 957 None if suppressed_nonfinal => { 958 if self.realtime_conversation.pending_speech.len() >= MAX_PENDING_SPEECH_DELIVERIES 959 && let Some(oldest) = self.realtime_conversation.pending_speech.pop_front() 960 { 961 self.restore_realtime_speech(oldest); 962 } 963 self.realtime_conversation 964 .pending_speech 965 .push_back(PendingRealtimeSpeech { 966 state: PendingSpeechState::AwaitingTurn, 967 captioned: false, 968 input_generation, 969 thread_id, 970 turn_id: turn_id.to_string(), 971 item: item.clone(), 972 }); 973 self.realtime_conversation.pending_speech.len() - 1 974 } 975 None => return, 976 }; 977 let pending = &mut self.realtime_conversation.pending_speech[index]; 978 pending.item = item.clone(); 979 pending.state = PendingSpeechState::Queued(delivery_id); 980 if !self.submit_op(AppCommand::RealtimeConversationSpeech { 981 thread_id, 982 attempt_id: self.realtime_conversation.attempt_id, 983 input_generation, 984 delivery_id, 985 text: text.into(), 986 }) { 987 self.restore_undelivered_realtime_speech(delivery_id); 988 self.on_realtime_error("Failed to deliver the voice response.".to_string()); 989 } else { 990 self.realtime_conversation.pending_speech.retain(|pending| { 991 pending.turn_id != turn_id || pending.state != PendingSpeechState::AwaitingTurn 992 }); 993 } 994 } 995 996 pub(crate) fn has_pending_realtime_speech(&self, delivery_id: u64) -> bool { 997 self.realtime_conversation 998 .pending_speech 999 .iter() 1000 .any(|delivery| delivery.state == PendingSpeechState::Queued(delivery_id)) 1001 } 1002 1003 pub(crate) fn take_undelivered_realtime_speech_for_replay( 1004 &mut self, 1005 ) -> Vec<(ThreadId, String, ThreadItem)> { 1006 self.realtime_conversation 1007 .pending_speech 1008 .drain(..) 1009 .filter(|delivery| !delivery.captioned) 1010 .map(|delivery| (delivery.thread_id, delivery.turn_id, delivery.item)) 1011 .collect() 1012 } 1013 1014 pub(crate) fn take_realtime_transcript_cells_for_replay( 1015 &mut self, 1016 ) -> VecDeque<RealtimeTranscriptRecord> { 1017 let mut records = std::mem::take(&mut self.realtime_conversation.accepted_transcripts); 1018 // Accepted records include cells already flushed into the old widget as 1019 // well as those still deferred behind another assistant stream. 1020 self.realtime_conversation.pending_history_cells.clear(); 1021 let mut retain_partial = |role: String, text: String| { 1022 if text.trim().is_empty() { 1023 return; 1024 } 1025 if let Some(partial) = records 1026 .iter_mut() 1027 .rev() 1028 .find(|record| record.role == role && !record.complete) 1029 { 1030 partial.text = text; 1031 return; 1032 } 1033 if records.len() >= MAX_REPLAY_TRANSCRIPT_CELLS { 1034 records.pop_front(); 1035 } 1036 records.push_back(RealtimeTranscriptRecord { 1037 role, 1038 text, 1039 complete: false, 1040 before_turn_id: None, 1041 }); 1042 }; 1043 if let Some((role, text)) = self.realtime_conversation.interleaved_transcript.take() { 1044 self.realtime_conversation.interleaved_transcript_cell = None; 1045 retain_partial(role, text); 1046 } 1047 if let Some(role) = self.realtime_conversation.transcript_role.take() { 1048 let text = std::mem::take(&mut self.realtime_conversation.transcript); 1049 retain_partial(role, text); 1050 self.realtime_conversation.live_transcript_cell = None; 1051 } 1052 records 1053 } 1054 1055 pub(crate) fn restore_realtime_transcript_cells( 1056 &mut self, 1057 records: VecDeque<RealtimeTranscriptRecord>, 1058 ) { 1059 self.realtime_conversation.recover_late_transcripts = true; 1060 for record in records { 1061 if self.realtime_conversation.accepted_transcripts.len() >= MAX_REPLAY_TRANSCRIPT_CELLS 1062 { 1063 self.realtime_conversation.accepted_transcripts.pop_front(); 1064 } 1065 if record.complete { 1066 if self.realtime_conversation.pending_history_cells.len() 1067 >= MAX_REPLAY_TRANSCRIPT_CELLS 1068 { 1069 self.realtime_conversation.pending_history_cells.pop_front(); 1070 } 1071 let cell = self.realtime_transcript_history_cell(&record.role, &record.text); 1072 self.realtime_conversation 1073 .pending_history_cells 1074 .push_back(cell); 1075 self.bump_active_cell_revision(); 1076 } else { 1077 // Keep an unfinished caption editable until its late completion or close. 1078 self.on_realtime_transcript_delta(record.role.clone(), record.text.clone()); 1079 } 1080 self.realtime_conversation 1081 .accepted_transcripts 1082 .push_back(record); 1083 } 1084 self.flush_realtime_transcript_history(); 1085 } 1086 1087 pub(crate) fn render_undelivered_realtime_speech( 1088 &mut self, 1089 thread_id: ThreadId, 1090 turn_id: String, 1091 item: ThreadItem, 1092 ) { 1093 if self.thread_id() == Some(thread_id) { 1094 self.handle_thread_item(item, turn_id, super::ThreadItemRenderSource::Live); 1095 } 1096 } 1097 1098 pub(crate) fn accept_realtime_speech(&mut self, delivery_id: u64) { 1099 if let Some(delivery) = self 1100 .realtime_conversation 1101 .pending_speech 1102 .iter_mut() 1103 .find(|delivery| delivery.state == PendingSpeechState::Queued(delivery_id)) 1104 { 1105 delivery.state = PendingSpeechState::Accepted; 1106 } 1107 self.realtime_conversation 1108 .pending_speech 1109 .retain(|delivery| !delivery.captioned); 1110 } 1111 1112 pub(crate) fn restore_undelivered_realtime_speech(&mut self, delivery_id: u64) { 1113 if let Some(index) = self 1114 .realtime_conversation 1115 .pending_speech 1116 .iter() 1117 .position(|delivery| delivery.state == PendingSpeechState::Queued(delivery_id)) 1118 && let Some(delivery) = self.realtime_conversation.pending_speech.remove(index) 1119 { 1120 self.restore_realtime_speech(delivery); 1121 } 1122 } 1123 1124 fn restore_realtime_speech(&mut self, delivery: PendingRealtimeSpeech) { 1125 if delivery.captioned { 1126 return; 1127 } 1128 if self.thread_id() != Some(delivery.thread_id) { 1129 return; 1130 } 1131 self.forget_realtime_turn_origin(&delivery.turn_id); 1132 self.handle_thread_item( 1133 delivery.item, 1134 delivery.turn_id, 1135 super::ThreadItemRenderSource::Live, 1136 ); 1137 } 1138 1139 fn restore_all_undelivered_realtime_speech(&mut self) { 1140 while let Some(delivery) = self.realtime_conversation.pending_speech.pop_front() { 1141 self.restore_realtime_speech(delivery); 1142 } 1143 } 1144 1145 pub(super) fn finish_realtime_turn(&mut self, turn_id: &str) { 1146 let mut waiting = Vec::new(); 1147 let mut index = 0; 1148 while index < self.realtime_conversation.pending_speech.len() { 1149 if self.realtime_conversation.pending_speech[index].turn_id == turn_id 1150 && self.realtime_conversation.pending_speech[index].state 1151 == PendingSpeechState::AwaitingTurn 1152 { 1153 if let Some(delivery) = self.realtime_conversation.pending_speech.remove(index) { 1154 waiting.push(delivery); 1155 } 1156 } else { 1157 index += 1; 1158 } 1159 } 1160 self.forget_realtime_turn_origin(turn_id); 1161 for delivery in waiting { 1162 if self.thread_id() == Some(delivery.thread_id) { 1163 self.handle_thread_item( 1164 delivery.item, 1165 delivery.turn_id, 1166 super::ThreadItemRenderSource::Live, 1167 ); 1168 } 1169 } 1170 } 1171 1172 fn remove_waiting_realtime_speech(&mut self, turn_id: &str, item_id: &str) { 1173 self.realtime_conversation.pending_speech.retain(|pending| { 1174 pending.turn_id != turn_id 1175 || pending.state != PendingSpeechState::AwaitingTurn 1176 || !matches!(&pending.item, ThreadItem::AgentMessage { id, .. } if id == item_id) 1177 }); 1178 } 1179 1180 fn forget_realtime_turn_origin(&mut self, turn_id: &str) { 1181 self.realtime_conversation.turn_origins.remove(turn_id); 1182 self.realtime_conversation 1183 .delegated_reasoning_turns 1184 .retain(|saved| saved != turn_id); 1185 self.realtime_conversation 1186 .agent_items 1187 .retain(|(item_turn_id, _), _| item_turn_id != turn_id); 1188 } 1189 1190 pub(super) fn on_realtime_transcript_delta(&mut self, role: String, delta: String) { 1191 if self.realtime_conversation.phase == RealtimeConversationPhase::Inactive 1192 && !self.realtime_conversation.recover_late_transcripts 1193 { 1194 return; 1195 } 1196 let active = matches!( 1197 self.realtime_conversation.phase, 1198 RealtimeConversationPhase::Starting | RealtimeConversationPhase::Active 1199 ); 1200 // Duplex transcript chunks can interleave. Display role changes are not input boundaries. 1201 if active 1202 && role == "user" 1203 && self 1204 .realtime_conversation 1205 .transcript_input_generation 1206 .is_none() 1207 { 1208 let interrupted = self.realtime_conversation.phase == RealtimeConversationPhase::Active 1209 && !self.realtime_conversation.microphone_muted 1210 && !delta.trim().is_empty() 1211 && (self.realtime_conversation.speaker_level > 0 1212 || self 1213 .realtime_conversation 1214 .speaker_active_until 1215 .is_some_and(|deadline| deadline > Instant::now())); 1216 self.realtime_conversation.input_generation = self 1217 .realtime_conversation 1218 .input_generation 1219 .wrapping_add(/*rhs*/ 1); 1220 self.realtime_conversation.transcript_input_generation = 1221 Some(self.realtime_conversation.input_generation); 1222 if interrupted && self.local_settings.tui.animations { 1223 self.realtime_conversation.interruption_acknowledged_until = 1224 Some(Instant::now() + INTERRUPTION_ACKNOWLEDGMENT); 1225 } 1226 self.suppress_active_realtime_speaker(); 1227 } 1228 if active 1229 && role == "assistant" 1230 && !delta.trim().is_empty() 1231 && self 1232 .realtime_conversation 1233 .assistant_transcript_generation 1234 .is_none() 1235 { 1236 self.realtime_conversation.assistant_transcript_generation = 1237 Some(self.realtime_conversation.input_generation); 1238 self.realtime_conversation 1239 .assistant_caption_started_after_speech_queue = self 1240 .realtime_conversation 1241 .pending_speech 1242 .iter() 1243 .any(|delivery| { 1244 delivery.input_generation == self.realtime_conversation.input_generation 1245 && delivery.state != PendingSpeechState::AwaitingTurn 1246 }); 1247 } 1248 if active { 1249 self.resume_realtime_speaker_for(&role, &delta); 1250 } 1251 if active 1252 && role == "user" 1253 && self.realtime_conversation.transcript_input_generation 1254 == Some(self.realtime_conversation.input_generation) 1255 { 1256 self.realtime_conversation.latest_input_was_voice = true; 1257 } 1258 if self.realtime_conversation.transcript_role.as_deref() != Some(role.as_str()) { 1259 let previous = self.realtime_conversation.transcript_role.take(); 1260 let previous_text = std::mem::take(&mut self.realtime_conversation.transcript); 1261 let saved = self.realtime_conversation.interleaved_transcript.take(); 1262 let saved_cell = self 1263 .realtime_conversation 1264 .interleaved_transcript_cell 1265 .take(); 1266 let previous_cell = self.realtime_conversation.live_transcript_cell.take(); 1267 self.realtime_conversation.transcript = match saved { 1268 Some((saved_role, saved_text)) if saved_role == role => { 1269 self.realtime_conversation.live_transcript_cell = saved_cell; 1270 saved_text 1271 } 1272 _ => String::new(), 1273 }; 1274 if let Some(previous_role) = previous { 1275 self.realtime_conversation.interleaved_transcript = 1276 Some((previous_role, previous_text)); 1277 self.realtime_conversation.interleaved_transcript_cell = previous_cell; 1278 } 1279 self.realtime_conversation.transcript_role = Some(role); 1280 } 1281 self.realtime_conversation.transcript.push_str(&delta); 1282 let mut discarded_prefix_bytes = 0; 1283 if self.realtime_conversation.transcript.len() > MAX_TRANSCRIPT_BYTES { 1284 let mut start = self.realtime_conversation.transcript.len() - MAX_TRANSCRIPT_BYTES; 1285 while !self 1286 .realtime_conversation 1287 .transcript 1288 .is_char_boundary(start) 1289 { 1290 start += 1; 1291 } 1292 self.realtime_conversation.transcript.drain(..start); 1293 discarded_prefix_bytes = start; 1294 } 1295 let role = self 1296 .realtime_conversation 1297 .transcript_role 1298 .as_deref() 1299 .unwrap_or(""); 1300 // Keep recognition and interruption state current without echoing each partial utterance. 1301 // The final event commits the ordinary user history cell once. 1302 if role == "user" && !self.local_settings.tui.animations { 1303 if self 1304 .realtime_conversation 1305 .live_transcript_cell 1306 .take() 1307 .is_some() 1308 { 1309 self.bump_active_cell_revision(); 1310 self.request_redraw(); 1311 } 1312 return; 1313 } 1314 let previous = self 1315 .realtime_conversation 1316 .live_transcript_cell 1317 .as_deref() 1318 .and_then(|cell| cell.as_any().downcast_ref::<SplitFlapTranscriptCell>()); 1319 let live_cell = SplitFlapTranscriptCell::new( 1320 |text| self.realtime_transcript_history_cell(role, text), 1321 role, 1322 &self.realtime_conversation.transcript, 1323 previous, 1324 discarded_prefix_bytes, 1325 MotionMode::from_animations_enabled(self.local_settings.tui.animations), 1326 self.frame_requester.clone(), 1327 ); 1328 self.realtime_conversation.live_transcript_cell = Some(Box::new(live_cell)); 1329 self.bump_active_cell_revision(); 1330 self.request_redraw(); 1331 } 1332 1333 pub(super) fn on_realtime_transcript_done(&mut self, role: String, mut text: String) { 1334 if matches!( 1335 self.realtime_conversation.phase, 1336 RealtimeConversationPhase::Inactive | RealtimeConversationPhase::Stopping 1337 ) { 1338 if self.realtime_conversation.phase == RealtimeConversationPhase::Inactive 1339 && !self.realtime_conversation.recover_late_transcripts 1340 { 1341 return; 1342 } 1343 if self.realtime_conversation.transcript_role.as_deref() == Some(role.as_str()) { 1344 self.realtime_conversation.transcript_role = None; 1345 self.realtime_conversation.transcript.clear(); 1346 self.realtime_conversation.live_transcript_cell = None; 1347 self.bump_active_cell_revision(); 1348 } else if self 1349 .realtime_conversation 1350 .interleaved_transcript 1351 .as_ref() 1352 .is_some_and(|(saved_role, _)| saved_role == &role) 1353 { 1354 self.realtime_conversation.interleaved_transcript = None; 1355 self.realtime_conversation.interleaved_transcript_cell = None; 1356 self.bump_active_cell_revision(); 1357 } 1358 if text.trim().is_empty() { 1359 if let Some(index) = self 1360 .realtime_conversation 1361 .accepted_transcripts 1362 .iter() 1363 .rposition(|record| record.role == role && !record.complete) 1364 { 1365 self.realtime_conversation 1366 .accepted_transcripts 1367 .remove(index); 1368 } 1369 return; 1370 } 1371 if text.len() > MAX_TRANSCRIPT_BYTES { 1372 let mut end = MAX_TRANSCRIPT_BYTES; 1373 while !text.is_char_boundary(end) { 1374 end -= 1; 1375 } 1376 text.truncate(end); 1377 } 1378 let display_text = if let Some(partial) = self 1379 .realtime_conversation 1380 .accepted_transcripts 1381 .iter_mut() 1382 .rev() 1383 .find(|record| record.role == role && !record.complete) 1384 { 1385 partial.text = text.clone(); 1386 partial.complete = true; 1387 text.clone() 1388 } else { 1389 // A separate final can share a prefix with an earlier caption. 1390 // Show it in full rather than guessing it extends that caption. 1391 if self.realtime_conversation.accepted_transcripts.len() 1392 >= MAX_REPLAY_TRANSCRIPT_CELLS 1393 { 1394 self.realtime_conversation.accepted_transcripts.pop_front(); 1395 } 1396 self.realtime_conversation.accepted_transcripts.push_back( 1397 RealtimeTranscriptRecord { 1398 role: role.clone(), 1399 text: text.clone(), 1400 complete: true, 1401 before_turn_id: None, 1402 }, 1403 ); 1404 text.clone() 1405 }; 1406 if display_text.is_empty() { 1407 return; 1408 } 1409 if self.realtime_conversation.pending_history_cells.len() >= MAX_REPLAY_TRANSCRIPT_CELLS 1410 { 1411 self.realtime_conversation.pending_history_cells.pop_front(); 1412 } 1413 let cell = self.realtime_transcript_history_cell(&role, &display_text); 1414 self.realtime_conversation 1415 .pending_history_cells 1416 .push_back(cell); 1417 self.bump_active_cell_revision(); 1418 self.flush_realtime_transcript_history(); 1419 return; 1420 } 1421 if !matches!( 1422 self.realtime_conversation.phase, 1423 RealtimeConversationPhase::Starting | RealtimeConversationPhase::Active 1424 ) { 1425 return; 1426 } 1427 let has_text = !text.trim().is_empty(); 1428 if role == "assistant" && has_text { 1429 // appendSpeech's RPC acknowledgement only means the request was 1430 // queued. A subsequent assistant caption is the closest signal 1431 // available here that its answer reached the voice session. The 1432 // protocol does not identify which speech request produced it. 1433 // A done-only caption has no generation owner, so it cannot retire 1434 // a newer answer's fallback. 1435 if self 1436 .realtime_conversation 1437 .assistant_caption_started_after_speech_queue 1438 && let Some(caption_generation) = 1439 self.realtime_conversation.assistant_transcript_generation 1440 && let Some(delivery) = 1441 self.realtime_conversation 1442 .pending_speech 1443 .iter_mut() 1444 .find(|delivery| { 1445 !delivery.captioned 1446 && delivery.input_generation == caption_generation 1447 && delivery.state != PendingSpeechState::AwaitingTurn 1448 && matches!(&delivery.item, ThreadItem::AgentMessage { text: answer, .. } 1449 if answer.trim().strip_prefix("[FINAL]").unwrap_or(answer.trim()) 1450 .split_whitespace().eq(text.split_whitespace())) 1451 }) 1452 { 1453 delivery.captioned = true; 1454 } 1455 self.realtime_conversation 1456 .pending_speech 1457 .retain(|delivery| { 1458 !delivery.captioned || matches!(delivery.state, PendingSpeechState::Queued(_)) 1459 }); 1460 } 1461 if has_text 1462 && self 1463 .realtime_conversation 1464 .transcript_input_generation 1465 .is_none() 1466 { 1467 self.resume_realtime_speaker_for(&role, &text); 1468 } 1469 // Check ownership before retiring the old output: its final chunk must not unmute it. 1470 if role == "assistant" { 1471 self.realtime_conversation.assistant_transcript_generation = None; 1472 self.realtime_conversation 1473 .assistant_caption_started_after_speech_queue = false; 1474 } 1475 let voice_input_fingerprint = (role == "user").then(|| realtime_input_fingerprint(&text)); 1476 if role == "user" { 1477 match self 1478 .realtime_conversation 1479 .transcript_input_generation 1480 .take() 1481 { 1482 Some(generation) => { 1483 if generation == self.realtime_conversation.input_generation { 1484 self.realtime_conversation.latest_input_was_voice = has_text; 1485 if !has_text { 1486 self.release_realtime_speaker(); 1487 } 1488 } 1489 } 1490 None => { 1491 let old_superseded_transcript = 1492 !self.realtime_conversation.latest_input_was_voice 1493 && self 1494 .realtime_conversation 1495 .latest_voice_input_fingerprint 1496 .is_some_and(|latest| Some(latest) == voice_input_fingerprint); 1497 if old_superseded_transcript || !has_text { 1498 return; 1499 } 1500 self.realtime_conversation.input_generation = self 1501 .realtime_conversation 1502 .input_generation 1503 .wrapping_add(/*rhs*/ 1); 1504 self.realtime_conversation.latest_input_was_voice = true; 1505 self.suppress_active_realtime_speaker(); 1506 } 1507 } 1508 } 1509 if self.realtime_conversation.transcript_role.as_deref() == Some(role.as_str()) { 1510 self.realtime_conversation.transcript.clear(); 1511 self.realtime_conversation.transcript_role = None; 1512 self.realtime_conversation.live_transcript_cell = None; 1513 self.bump_active_cell_revision(); 1514 } else if self 1515 .realtime_conversation 1516 .interleaved_transcript 1517 .as_ref() 1518 .is_some_and(|(saved_role, _)| saved_role == &role) 1519 { 1520 self.realtime_conversation.interleaved_transcript = None; 1521 self.realtime_conversation.interleaved_transcript_cell = None; 1522 self.bump_active_cell_revision(); 1523 } 1524 if text.len() > MAX_TRANSCRIPT_BYTES { 1525 let mut end = MAX_TRANSCRIPT_BYTES; 1526 while !text.is_char_boundary(end) { 1527 end -= 1; 1528 } 1529 text.truncate(end); 1530 } 1531 if role == "user" { 1532 self.realtime_conversation.latest_voice_input_fingerprint = voice_input_fingerprint; 1533 } 1534 if !text.trim().is_empty() { 1535 while self.realtime_conversation.accepted_transcripts.len() 1536 >= MAX_PENDING_TRANSCRIPT_CELLS 1537 { 1538 self.realtime_conversation.accepted_transcripts.pop_front(); 1539 } 1540 self.realtime_conversation 1541 .accepted_transcripts 1542 .push_back(RealtimeTranscriptRecord { 1543 role: role.clone(), 1544 text: text.clone(), 1545 complete: true, 1546 before_turn_id: None, 1547 }); 1548 while self.realtime_conversation.pending_history_cells.len() 1549 >= MAX_PENDING_TRANSCRIPT_CELLS 1550 { 1551 self.realtime_conversation.pending_history_cells.pop_front(); 1552 tracing::warn!("dropping the oldest deferred voice transcript"); 1553 } 1554 let cell = self.realtime_transcript_history_cell(&role, &text); 1555 self.realtime_conversation 1556 .pending_history_cells 1557 .push_back(cell); 1558 self.bump_active_cell_revision(); 1559 self.flush_realtime_transcript_history(); 1560 } 1561 } 1562 1563 fn finish_realtime_partial_transcripts(&mut self) { 1564 if let Some((role, text)) = self.realtime_conversation.interleaved_transcript.take() { 1565 self.realtime_conversation.interleaved_transcript_cell = None; 1566 self.on_realtime_transcript_done(role, text); 1567 } 1568 if let Some(role) = self.realtime_conversation.transcript_role.clone() { 1569 let text = std::mem::take(&mut self.realtime_conversation.transcript); 1570 self.on_realtime_transcript_done(role, text); 1571 } 1572 } 1573 1574 fn realtime_transcript_history_cell(&self, role: &str, text: &str) -> Box<dyn HistoryCell> { 1575 if role == "user" { 1576 Box::new(history_cell::new_spoken_user_prompt(text.to_string())) 1577 } else { 1578 Box::new(history_cell::AgentMarkdownCell::new_spoken( 1579 text.to_string(), 1580 &self.config.cwd, 1581 )) 1582 } 1583 } 1584 1585 pub(super) fn flush_realtime_transcript_history(&mut self) { 1586 if self.stream_controller.is_some() 1587 || self.plan_stream_controller.is_some() 1588 || self.pending_stream_consolidations > 0 1589 { 1590 return; 1591 } 1592 if !self.realtime_conversation.pending_history_cells.is_empty() { 1593 self.app_event_tx 1594 .send(AppEvent::CommitRealtimeTranscriptHistory); 1595 self.request_redraw(); 1596 } 1597 } 1598 1599 pub(crate) fn take_realtime_transcript_history(&mut self) -> Vec<Box<dyn HistoryCell>> { 1600 if self.stream_controller.is_some() 1601 || self.plan_stream_controller.is_some() 1602 || self.pending_stream_consolidations > 0 1603 { 1604 return Vec::new(); 1605 } 1606 let mut cells = Vec::new(); 1607 while let Some(cell) = self.realtime_conversation.pending_history_cells.pop_front() { 1608 if let Some(active) = self.take_history_insertion_prefix(cell.as_ref()) { 1609 cells.push(active); 1610 } 1611 cells.push(cell); 1612 } 1613 if !cells.is_empty() { 1614 self.bump_active_cell_revision(); 1615 } 1616 cells 1617 } 1618 1619 pub(crate) fn record_realtime_failure(&mut self) { 1620 if self.realtime_conversation.phase == RealtimeConversationPhase::Inactive 1621 || self.realtime_conversation.failure_recorded 1622 { 1623 return; 1624 } 1625 self.realtime_conversation.failure_recorded = true; 1626 self.session_telemetry 1627 .counter("codex.voice.session.failure", /*inc*/ 1, &[]); 1628 } 1629 1630 pub(super) fn realtime_retry_cleanup_pending(&self) -> bool { 1631 self.realtime_conversation.startup_retry == StartupRetry::WaitingForStop 1632 } 1633 1634 pub(crate) fn on_realtime_error(&mut self, message: String) { 1635 if self.realtime_conversation.phase == RealtimeConversationPhase::Inactive { 1636 return; 1637 } 1638 if self.realtime_conversation.phase != RealtimeConversationPhase::Stopping 1639 || self.realtime_retry_cleanup_pending() 1640 { 1641 self.record_realtime_failure(); 1642 } 1643 self.stop_realtime_conversation(); 1644 self.add_error_message(message); 1645 } 1646 1647 pub(super) fn on_realtime_conversation_closed(&mut self, reason: Option<String>) { 1648 if self.realtime_conversation.phase == RealtimeConversationPhase::Inactive { 1649 if self.realtime_conversation.recover_late_transcripts { 1650 self.finish_realtime_partial_transcripts(); 1651 self.realtime_conversation.recover_late_transcripts = false; 1652 } 1653 return; 1654 } 1655 let retry_after_stop = 1656 self.realtime_conversation.startup_retry == StartupRetry::WaitingForStop; 1657 let retry_after_early_close = self.realtime_conversation.phase 1658 == RealtimeConversationPhase::Starting 1659 && self.realtime_conversation.startup_retry == StartupRetry::Available 1660 && reason.as_deref() == Some("transport_closed"); 1661 if self.realtime_conversation.phase == RealtimeConversationPhase::Stopping 1662 && reason.as_deref() != Some("requested") 1663 { 1664 return; 1665 } 1666 // A transport close may arrive without final transcript events. Recover 1667 // answers before flushing partial captions, which do not prove delivery. 1668 self.restore_all_undelivered_realtime_speech(); 1669 self.finish_realtime_partial_transcripts(); 1670 let muted = self.realtime_conversation.microphone_muted; 1671 let thread_id = self.realtime_conversation.thread_id; 1672 let retry_thread_id = if retry_after_stop || retry_after_early_close { 1673 thread_id.filter(|id| Some(*id) == self.thread_id()) 1674 } else { 1675 None 1676 }; 1677 if retry_thread_id.is_none() 1678 && (reason.is_none() || matches!(reason.as_deref(), Some("transport_closed" | "error"))) 1679 { 1680 self.record_realtime_failure(); 1681 } 1682 let failed = self.realtime_conversation.failure_recorded; 1683 self.reset_realtime_conversation(); 1684 if let Some(thread_id) = retry_thread_id { 1685 // The old backend is closed. A late peer result belongs to its attempt ID. 1686 self.realtime_conversation.startup_retry = StartupRetry::Used; 1687 self.realtime_conversation.microphone_muted = muted; 1688 if retry_after_early_close { 1689 self.add_info_message( 1690 "Voice connection closed during startup. Retrying once.".into(), 1691 /*hint*/ None, 1692 ); 1693 } 1694 self.start_realtime_conversation(thread_id); 1695 return; 1696 } 1697 if let Some(reason) = reason 1698 && reason != "error" 1699 && !(failed && reason == "requested") 1700 { 1701 self.add_info_message( 1702 format!("Voice conversation ended: {reason}"), 1703 /*hint*/ None, 1704 ); 1705 } 1706 self.request_redraw(); 1707 } 1708 1709 fn finish_realtime_session_metrics(&mut self) { 1710 if let Some(active_since) = self.realtime_conversation.active_since.take() { 1711 self.session_telemetry 1712 .counter("codex.voice.session.ended", /*inc*/ 1, &[]); 1713 self.session_telemetry.record_duration( 1714 "codex.voice.session.duration", 1715 active_since.elapsed(), 1716 &[], 1717 ); 1718 } 1719 } 1720 1721 pub(crate) fn reset_realtime_conversation(&mut self) -> Option<ThreadId> { 1722 self.finish_realtime_session_metrics(); 1723 let should_refresh_terminal_title = self.realtime_conversation.phase 1724 != RealtimeConversationPhase::Inactive 1725 && self.last_terminal_title.is_some(); 1726 self.restore_all_undelivered_realtime_speech(); 1727 // A delegated item is hidden until the turn completes so it can be spoken. 1728 // If voice ends first, let the later turn completion render it normally. 1729 if self 1730 .transcript 1731 .last_completed_agent_message 1732 .as_ref() 1733 .is_some_and(|completed_item| { 1734 matches!( 1735 self.realtime_conversation.agent_items.get(completed_item), 1736 Some(RealtimeAgentItemOrigin::Delegated { 1737 completed: true, 1738 .. 1739 }) 1740 ) 1741 }) 1742 { 1743 self.transcript.last_completed_agent_message = None; 1744 } 1745 // The helper may already have exited while app-server still owns the 1746 // voice session. An acknowledged backend start still needs a stop RPC. 1747 let backend_thread_id = if self.realtime_conversation.handle.is_some() 1748 || self.realtime_conversation.backend_started 1749 { 1750 self.realtime_conversation.thread_id 1751 } else { 1752 None 1753 }; 1754 if let Some(abort) = self.realtime_conversation.startup_abort.take() { 1755 abort.abort(); 1756 } 1757 if let Some(handle) = self.realtime_conversation.handle.take() { 1758 handle.close(); 1759 } 1760 // Direct resets (for example, switching to a resumed thread) do not 1761 // pass through the normal stop/close transcript flush. 1762 let partials = [ 1763 self.realtime_conversation.interleaved_transcript.take(), 1764 self.realtime_conversation 1765 .transcript_role 1766 .take() 1767 .map(|role| { 1768 ( 1769 role, 1770 std::mem::take(&mut self.realtime_conversation.transcript), 1771 ) 1772 }), 1773 ]; 1774 for (role, text) in partials.into_iter().flatten() { 1775 if text.trim().is_empty() { 1776 continue; 1777 } 1778 if self.realtime_conversation.accepted_transcripts.len() >= MAX_REPLAY_TRANSCRIPT_CELLS 1779 { 1780 self.realtime_conversation.accepted_transcripts.pop_front(); 1781 } 1782 if self.realtime_conversation.pending_history_cells.len() >= MAX_REPLAY_TRANSCRIPT_CELLS 1783 { 1784 self.realtime_conversation.pending_history_cells.pop_front(); 1785 } 1786 let cell = self.realtime_transcript_history_cell(&role, &text); 1787 self.realtime_conversation 1788 .pending_history_cells 1789 .push_back(cell); 1790 self.bump_active_cell_revision(); 1791 self.realtime_conversation 1792 .accepted_transcripts 1793 .push_back(RealtimeTranscriptRecord { 1794 role, 1795 text, 1796 complete: true, 1797 before_turn_id: None, 1798 }); 1799 } 1800 let pending_history_cells = 1801 std::mem::take(&mut self.realtime_conversation.pending_history_cells); 1802 let accepted_transcripts = 1803 std::mem::take(&mut self.realtime_conversation.accepted_transcripts); 1804 let delegated_reasoning_turns = 1805 std::mem::take(&mut self.realtime_conversation.delegated_reasoning_turns); 1806 let had_live_transcript = self 1807 .realtime_conversation 1808 .live_transcript_cells() 1809 .next() 1810 .is_some(); 1811 self.realtime_conversation = RealtimeConversationUiState { 1812 attempt_id: self.realtime_conversation.attempt_id, 1813 pending_history_cells, 1814 accepted_transcripts, 1815 delegated_reasoning_turns, 1816 ..RealtimeConversationUiState::default() 1817 }; 1818 if had_live_transcript { 1819 self.bump_active_cell_revision(); 1820 } 1821 self.bottom_pane.set_voice_strip(/*state*/ None); 1822 self.flush_realtime_transcript_history(); 1823 if should_refresh_terminal_title { 1824 self.refresh_terminal_title(); 1825 } 1826 backend_thread_id 1827 } 1828 }