Keep voice conversations running across TUI thread navigation (#47381)

## Why

Switching threads stopped the active voice conversation. Users should be able to view another task while continuing the call on its original thread.

## What changed

- Retain the voice owner's live widget and subscription in the background, routing voice controls, speech, and connection callbacks to that owner.
- Restore the live call and transcript when returning to its thread, without duplicating captions or letting background UI events affect the visible task.
- Preserve undelivered speech and defer background errors until the owning thread is shown.
- Stop background voice on shutdown, disconnect, or removal of the owning session.

## Testing

Extend regression and end-to-end coverage for thread switching during delegated speech, voice control routing, transcript restoration, caption deduplication, and cleanup on disconnect or archive/delete.

GitOrigin-RevId: ce010dce6fa8cd6fa44949a131f3f16e8cb80c4d
This commit is contained in:
Eric Traut
2026-09-22 22:45:08 +00:00
committed by copyberry
parent 72ec34ee9a
commit a837a656ff
35 changed files with 870 additions and 140 deletions
+5
View File
@@ -641,6 +641,8 @@ pub(crate) struct App {
pending_realtime_transcript_replay:
HashMap<ThreadId, VecDeque<crate::chatwidget::RealtimeTranscriptRecord>>,
realtime_replay_order: VecDeque<ThreadId>,
background_voice: Option<Box<ChatWidget>>,
background_voice_error: Option<(ThreadId, String)>,
temporary_structured_requests: HashMap<ThreadId, mpsc::UnboundedSender<ServerNotification>>,
/// Track title generation across thread switches and deduplicate automatic requests.
pending_thread_titles: HashMap<(ThreadId, ThreadTitleDestination), CancellationToken>,
@@ -1100,6 +1102,9 @@ impl App {
}
// Allow widgets to process any pending timers before rendering.
let had_active_modal = self.chat_widget.has_active_modal();
if let Some(owner) = self.background_voice.as_mut() {
owner.refresh_realtime_microphone_level();
}
self.chat_widget.pre_draw_tick();
self.refresh_agents_overview_usage(app_server, tui.frame_requester());
let rendered_area = self.render_chat_widget_frame(tui, screen_size)?;
+12 -2
View File
@@ -634,10 +634,19 @@ impl App {
match startup_draft.as_deref_mut() {
Some(draft) => {
draft
.run_until(tui, self.shutdown_current_thread(app_server))
.run_until(
tui,
self.detach_current_thread_for_navigation(
app_server,
Some(root_thread_id),
),
)
.await?
}
None => self.shutdown_current_thread(app_server).await,
None => {
self.detach_current_thread_for_navigation(app_server, Some(root_thread_id))
.await
}
}
}
// Explicit choices carry across cold resumes and new sessions.
@@ -727,6 +736,7 @@ impl App {
for thread_id in previous_thread_ids {
if previous_running_thread_ids.is_empty()
&& thread_id != root_thread_id
&& self.voice_owner_thread_id() != Some(thread_id)
&& Some(thread_id) != previous_displayed_thread_id
&& !self.agents_overview.blank_sessions.contains_key(&thread_id)
&& let Err(error) = StartupDraftPump::run_with_optional_draft(
@@ -178,6 +178,8 @@ impl App {
let events = tui.event_stream();
let operation = async {
let result = async {
self.stop_voice_for_removed_thread(app_server, thread_id)
.await?;
if let Some(primary) = primary_thread_id
&& primary != thread_id
{
@@ -276,7 +278,9 @@ impl App {
self.pending_thread_switch_resets += 1;
self.app_event_tx
.send(AppEvent::ResetTranscriptForThreadSwitch);
self.reset_thread_event_state();
self.detach_current_thread_for_navigation(app_server, /*destination*/ None)
.await;
self.reset_thread_event_state().await;
let init = self.chatwidget_init_for_forked_or_resumed_thread(
tui,
self.config.clone(),
@@ -698,7 +698,9 @@ async fn lifecycle_removes_background_and_current_tasks_without_losing_the_dashb
// Removal must not depend on the overview having the primary or its ancestors cached.
app.agents_overview.threads.remove(&primary);
}
crate::chatwidget::activate_voice_for_thread(&mut app.chat_widget, primary);
Box::pin(app.handle_event(&mut tui, &mut app_server, confirmed)).await?;
assert_eq!(app.voice_owner_thread_id(), None);
assert_eq!(
(
app.primary_thread_id,
+21 -10
View File
@@ -405,7 +405,13 @@ impl App {
}
}
}
if self.primary_thread_id.is_none() && !self.pending_startup_thread_start {
let background_voice = self.background_voice.as_ref().is_some_and(|owner| {
owner.thread_id() == Some(thread_id) && owner.realtime_conversation_is_running()
});
if self.primary_thread_id.is_none()
&& !self.pending_startup_thread_start
&& !background_voice
{
return;
}
if self.primary_thread_id.is_some()
@@ -430,9 +436,7 @@ impl App {
{
return;
}
let result = if self.primary_thread_id == Some(thread_id)
|| self.primary_thread_id.is_none()
{
let result = if self.primary_thread_id.is_none() && !background_voice {
self.enqueue_primary_thread_notification(notification).await
} else {
self.enqueue_thread_notification(thread_id, notification)
@@ -602,6 +606,11 @@ impl App {
}
let thread_id = server_request_thread_id(&request);
let background_voice = self.background_voice.as_ref().is_some_and(|owner| {
owner.realtime_conversation_is_running()
&& owner.thread_id().is_some()
&& owner.thread_id() == thread_id
});
if thread_id.is_some_and(|thread_id| self.abandoned_side_threads.contains(&thread_id)) {
if let Err(err) = self
.reject_app_server_request(
@@ -617,6 +626,7 @@ impl App {
}
if thread_id.is_some()
&& self.primary_thread_id.is_none()
&& !background_voice
&& self.pending_startup_thread_start
{
self.pending_primary_events
@@ -640,6 +650,7 @@ impl App {
if let Some(thread_id) = thread_id
&& self.primary_thread_id != Some(thread_id)
&& !unsupported_request
&& !background_voice
&& let Some(requests) = self.agents_overview.dispatched_requests.get_mut(&thread_id)
{
requests.push(request);
@@ -647,6 +658,7 @@ impl App {
}
if thread_id.is_some()
&& self.primary_thread_id.is_none()
&& !background_voice
&& !self.pending_startup_thread_start
&& !unsupported_request
{
@@ -727,12 +739,11 @@ impl App {
return;
};
let result =
if self.primary_thread_id == Some(thread_id) || self.primary_thread_id.is_none() {
self.enqueue_primary_thread_request(request).await
} else {
self.enqueue_thread_request(thread_id, request).await
};
let result = if self.primary_thread_id.is_none() && !background_voice {
self.enqueue_primary_thread_request(request).await
} else {
self.enqueue_thread_request(thread_id, request).await
};
if let Err(err) = result {
tracing::warn!("failed to enqueue app-server request: {err}");
}
+1 -1
View File
@@ -388,7 +388,7 @@ async fn running_turn_dismisses_announcement_across_thread_resets() {
/*replay_kind*/ None,
);
assert_eq!(app.composer_hint(/*width*/ 80), None);
app.reset_thread_event_state();
app.reset_thread_event_state().await;
let tip = app.composer_tips.select(
/*turn_count*/ 0,
|| panic!("an already running turn must end the startup announcement"),
+77 -28
View File
@@ -528,7 +528,7 @@ impl App {
} else {
None
};
self.shutdown_current_thread(app_server).await;
self.detach_current_thread_for_navigation(app_server, Some(forked.session.thread_id)).await;
match self
.replace_chat_widget_with_app_server_thread(
tui,
@@ -1006,21 +1006,48 @@ impl App {
self.chat_widget.pre_draw_tick();
self.render_chat_widget_frame(tui, screen_size)?;
}
let parked_voice = match &op {
AppCommand::RealtimeConversationStart { thread_id, .. }
| AppCommand::RealtimeConversationStop { thread_id }
| AppCommand::RealtimeConversationSpeech { thread_id, .. } => self
.background_voice
.as_ref()
.is_some_and(|owner| owner.thread_id() == Some(*thread_id)),
_ => false,
};
let visible_thread = self.active_thread_id;
if parked_voice
&& let Some(owner) = self.background_voice.as_mut()
{
std::mem::swap(&mut self.chat_widget, owner);
self.active_thread_id = self.chat_widget.thread_id();
}
self.chat_widget.prepare_local_op_submission(&op);
if let Err(err) = self.submit_active_thread_op(app_server, op).await {
if let Some(delivery_id) = realtime_speech_delivery_id {
self.chat_widget
.restore_undelivered_realtime_speech(delivery_id);
}
if self.recover_transport_error(&err)
{
let result = self.submit_active_thread_op(app_server, op).await;
if result.is_err()
&& let Some(delivery_id) = realtime_speech_delivery_id
{
self.chat_widget.restore_undelivered_realtime_speech(delivery_id);
}
if parked_voice
&& let Some(owner) = self.background_voice.as_mut()
{
std::mem::swap(&mut self.chat_widget, owner);
self.active_thread_id = visible_thread;
}
if let Err(err) = result {
if self.recover_transport_error(&err) {
return Ok(AppRunControl::Continue);
}
let chat_widget = match self.background_voice.as_deref_mut() {
Some(owner) if parked_voice => owner,
_ => &mut self.chat_widget,
};
let unsupported_permissions = err
.downcast_ref::<UnsupportedLegacyPermissionProfile>()
.is_some();
if unsupported_permissions {
self.chat_widget
chat_widget
.set_queue_autosend_suppressed(/*suppressed*/ true);
}
let handled = is_user_turn
@@ -1029,19 +1056,18 @@ impl App {
Some(TypedRequestError::Server { method, .. })
if method == "turn/start"
) || unsupported_permissions)
&& self
.chat_widget
&& chat_widget
.handle_turn_start_rejection(format!("Failed to start turn: {err:#}"));
if is_realtime_conversation {
let message = format!("Voice conversation failed: {err:#}");
if is_realtime_stop {
if self.chat_widget.thread_id() == realtime_stop_thread_id {
self.chat_widget.record_realtime_failure();
self.chat_widget.reset_realtime_conversation();
self.chat_widget.add_error_message(message);
if chat_widget.thread_id() == realtime_stop_thread_id {
chat_widget.record_realtime_failure();
chat_widget.reset_realtime_conversation();
chat_widget.add_realtime_error(message);
}
} else {
self.chat_widget.on_realtime_error(message);
chat_widget.on_realtime_error(message);
}
tracing::error!(error = ?err, "realtime conversation request failed");
} else if handled {
@@ -1894,8 +1920,15 @@ impl App {
}
return Ok(control);
}
AppEvent::BackgroundVoiceError { thread_id, message } => {
if self.chat_widget.thread_id() == Some(thread_id) {
self.chat_widget.add_error_message(message);
} else {
self.background_voice_error = Some((thread_id, message));
}
}
AppEvent::VoiceControl { thread_id, control } => {
if thread_id == self.chat_widget.thread_id() {
if thread_id == self.chat_widget.thread_id() || self.voice_owner_thread_id().is_some() {
self.control_voice(control);
}
}
@@ -1904,9 +1937,8 @@ impl App {
attempt_id,
result,
} => {
if self.chat_widget.thread_id() == Some(thread_id) {
self.chat_widget
.on_realtime_webrtc_offer_created(thread_id, attempt_id, result);
if let Some(owner) = self.voice_widget_for_thread(thread_id) {
owner.on_realtime_webrtc_offer_created(thread_id, attempt_id, result);
} else if let Ok(offer) = result {
offer.handle.close();
}
@@ -1916,9 +1948,8 @@ impl App {
attempt_id,
result,
} => {
if self.chat_widget.thread_id() == Some(thread_id) {
self.chat_widget
.on_realtime_webrtc_connected(attempt_id, result);
if let Some(owner) = self.voice_widget_for_thread(thread_id) {
owner.on_realtime_webrtc_connected(attempt_id, result);
}
}
AppEvent::StopRealtimeConversation { thread_id } => {
@@ -3311,7 +3342,9 @@ impl App {
// its shutdown completion does not trigger agent failover.
self.pending_shutdown_exit_thread_id =
self.active_thread_id.or(self.chat_widget.thread_id());
if self.pending_shutdown_exit_thread_id.is_some() {
if self.pending_shutdown_exit_thread_id.is_some()
|| self.voice_owner_thread_id().is_some()
{
// This is a UI escape-hatch budget, not a protocol
// deadline. A healthy local thread/unsubscribe round trip
// should finish comfortably inside two seconds, while a
@@ -3367,7 +3400,13 @@ impl App {
}
}
Ok(match app_server.thread_archive(thread_id).await {
let result = async {
self.stop_voice_for_removed_thread(app_server, thread_id)
.await?;
app_server.thread_archive(thread_id).await
}
.await;
Ok(match result {
Ok(()) if matches!(self.app_server_target, AppServerTarget::Embedded) => {
AppRunControl::Exit(ExitReason::Archived(thread_id))
}
@@ -3384,7 +3423,9 @@ impl App {
self.pending_thread_switch_resets += 1;
self.app_event_tx
.send(AppEvent::ResetTranscriptForThreadSwitch);
self.reset_thread_event_state();
self.detach_current_thread_for_navigation(app_server, /*destination*/ None)
.await;
self.reset_thread_event_state().await;
let init = self.chatwidget_init_for_forked_or_resumed_thread(
tui,
self.config.clone(),
@@ -3427,7 +3468,13 @@ impl App {
}
}
Ok(match app_server.thread_delete(thread_id).await {
let result = async {
self.stop_voice_for_removed_thread(app_server, thread_id)
.await?;
app_server.thread_delete(thread_id).await
}
.await;
Ok(match result {
Ok(()) if matches!(self.app_server_target, AppServerTarget::Embedded) => {
AppRunControl::Exit(ExitReason::ThreadRemoved)
}
@@ -3444,7 +3491,9 @@ impl App {
self.pending_thread_switch_resets += 1;
self.app_event_tx
.send(AppEvent::ResetTranscriptForThreadSwitch);
self.reset_thread_event_state();
self.detach_current_thread_for_navigation(app_server, /*destination*/ None)
.await;
self.reset_thread_event_state().await;
let init = self.chatwidget_init_for_forked_or_resumed_thread(
tui,
self.config.clone(),
+2 -1
View File
@@ -207,7 +207,8 @@ impl App {
if self.backtrack.overlay_preview_active && self.chat_widget.no_modal_or_popup_active() {
return KeymapContextSet::browsing().with_voice_toggle(&self.keymap);
}
let voice_available = self.chat_widget.realtime_microphone_shortcut_available();
let voice_available = self.chat_widget.realtime_microphone_shortcut_available()
|| self.voice_owner_thread_id().is_some();
let contexts = self.chat_widget.keymap_contexts();
let contexts = if self.chat_widget.no_modal_or_popup_active() {
let contexts = contexts
+12
View File
@@ -294,6 +294,18 @@ impl App {
let Some(thread_id) = self.chat_widget.thread_id() else {
return HashMap::new();
};
if let Some(owner) = self.background_voice.as_ref()
&& owner.thread_id() == Some(thread_id)
{
for text in self.chat_widget.prepare_background_voice_replay(owner) {
let key = ("assistant".to_string(), normalized_replay_text(text));
if let Some(count) = replayed_voice_texts.get_mut(&key) {
*count = count.saturating_sub(/*rhs*/ 1);
}
}
self.pending_realtime_transcript_replay
.insert(thread_id, owner.realtime_transcript_cells_for_replay());
}
self.realtime_replay_order
.retain(|saved| *saved != thread_id);
let mut retained_assistant_captions = HashMap::<String, usize>::new();
+4
View File
@@ -185,6 +185,10 @@ impl App {
{
self.chat_widget.restore_user_message_to_composer(message);
}
if let Some(owner) = self.background_voice.as_mut() {
owner.reset_realtime_conversation();
}
self.retire_background_voice();
self.reconnect.offline = true;
// Cached blank sessions are usable only while this connection owns a subscription.
self.agents_overview.blank_sessions.clear();
+2 -1
View File
@@ -201,7 +201,8 @@ impl App {
};
let retry_thread_id = started.session.thread_id;
self.shutdown_current_thread(app_server).await;
self.detach_current_thread_for_navigation(app_server, Some(retry_thread_id))
.await;
if let Err(err) = self
.replace_chat_widget_with_app_server_thread(
tui,
+49 -25
View File
@@ -485,7 +485,9 @@ impl App {
/// This helper copies every known nickname/role from `AgentNavigationState` into the
/// replacement widget so that replayed collab items render agent names immediately.
pub(super) fn replace_chat_widget(&mut self, mut chat_widget: ChatWidget) {
self.retain_realtime_replay_state_before_replace();
if !self.chat_widget.realtime_conversation_is_running() {
self.retain_realtime_replay_state_before_replace();
}
chat_widget.cyber_policy_notice = self.chat_widget.cyber_policy_notice.clone();
self.commit_animation = None;
// Transfer the last-written terminal title to the replacement widget
@@ -520,7 +522,11 @@ impl App {
}
chat_widget.restore_kill_buffer_snapshot(self.chat_widget.take_kill_buffer_snapshot());
crate::markdown_render::preferences::init(chat_widget.local_settings.tui.rendering);
self.chat_widget = chat_widget;
let mut previous = std::mem::replace(&mut self.chat_widget, chat_widget);
if previous.realtime_conversation_is_running() {
previous.park_voice();
self.background_voice = Some(Box::new(previous));
}
self.sync_active_agent_label();
}
@@ -690,10 +696,6 @@ impl App {
self.recap.seed_from_progress(recap_progress, now);
self.schedule_recap_check(thread_id, now);
// The old widget owns the local helper and its undelivered answers.
// Transfer replay state before stopping its backend voice session.
self.retain_realtime_replay_state_before_replace();
self.stop_realtime_conversation(app_server).await;
self.render_thread_snapshot(tui, app_server, thread_id, snapshot, !is_replay_only)?;
if is_replay_only
&& self
@@ -794,9 +796,28 @@ impl App {
Ok(())
}
pub(super) fn reset_thread_event_state(&mut self) {
self.abort_all_thread_event_listeners();
self.thread_event_channels.clear();
pub(super) async fn reset_thread_event_state(&mut self) {
let voice_owner = self.voice_owner_thread_id();
if voice_owner.is_some() {
for (thread_id, channel) in &self.thread_event_channels {
if Some(*thread_id) != voice_owner {
for request in channel.store.lock().await.pending_replay_requests() {
self.pending_app_server_requests
.resolve_notification(&thread_id.to_string(), request.id());
}
}
}
}
self.thread_event_listener_tasks.retain(|id, task| {
if Some(*id) == voice_owner {
true
} else {
task.abort();
false
}
});
self.thread_event_channels
.retain(|id, _| Some(*id) == voice_owner);
self.pending_realtime_speech_replay.clear();
self.pending_realtime_transcript_replay.clear();
self.realtime_replay_order.clear();
@@ -810,7 +831,9 @@ impl App {
self.last_subagent_backfill_attempt = None;
self.primary_session_configured = None;
self.pending_primary_events.clear();
self.pending_app_server_requests.clear();
if voice_owner.is_none() {
self.pending_app_server_requests.clear();
}
self.pending_startup_thread_start = false;
self.pending_server_version_notice = None;
self.chat_widget.set_pending_thread_approvals(Vec::new());
@@ -975,14 +998,11 @@ impl App {
.await
{
Ok(mut started) => {
self.shutdown_current_thread(app_server).await;
let tracked_thread_ids: Vec<ThreadId> =
self.thread_event_channels.keys().copied().collect();
for thread_id in tracked_thread_ids {
if let Err(err) = app_server.thread_unsubscribe(thread_id).await {
tracing::warn!("failed to unsubscribe tracked thread {thread_id}: {err}");
}
}
self.detach_current_thread_for_navigation(
app_server,
Some(started.session.thread_id),
)
.await;
self.local_settings = self.local_settings.reloaded(&config);
self.refresh_server_version_overview_notice(CODEX_CLI_VERSION);
self.config = config;
@@ -1056,11 +1076,19 @@ impl App {
// Initial messages are for freshly attached primary threads only. Thread switches and
// resume/fork flows pass `None` so they cannot replay old history and then auto-submit a new
// user turn by accident.
if let Some(mut receiver) = self.active_thread_rx.take() {
while let Ok(event) = receiver.try_recv() {
self.handle_thread_event_now_recovering_file_changes(event)
.await;
}
self.active_thread_rx = Some(receiver);
}
self.store_active_thread_receiver().await;
self.reset_for_thread_switch(tui)?;
self.pending_thread_switch_resets += 1;
self.app_event_tx
.send(AppEvent::ResetTranscriptForThreadSwitch);
self.reset_thread_event_state();
self.reset_thread_event_state().await;
let init = self.chatwidget_init_for_forked_or_resumed_thread(
tui,
self.config.clone(),
@@ -1307,12 +1335,8 @@ impl App {
&& self.chat_widget.thread_id() == Some(resumed_thread_id))
.then(|| self.chat_widget.capture_thread_input_state())
.flatten();
if self.chat_widget.thread_id() == Some(resumed_thread_id) {
// The successful resume has already subscribed this connection to the same thread.
self.shutdown_side_threads(app_server).await;
} else {
self.shutdown_current_thread(app_server).await;
}
self.detach_current_thread_for_navigation(app_server, Some(resumed_thread_id))
.await;
self.local_settings = local_settings;
self.refresh_server_version_overview_notice(CODEX_CLI_VERSION);
self.config = resume_config;
+2
View File
@@ -805,6 +805,8 @@ See the Codex keymap documentation for supported actions and examples."
pending_realtime_speech_replay: HashMap::new(),
pending_realtime_transcript_replay: HashMap::new(),
realtime_replay_order: VecDeque::new(),
background_voice: None,
background_voice_error: None,
temporary_structured_requests: HashMap::new(),
pending_thread_titles: HashMap::new(),
thread_event_listener_tasks: HashMap::new(),
+2
View File
@@ -86,6 +86,8 @@ pub(crate) async fn make_test_app() -> App {
pending_realtime_speech_replay: HashMap::new(),
pending_realtime_transcript_replay: HashMap::new(),
realtime_replay_order: VecDeque::new(),
background_voice: None,
background_voice_error: None,
temporary_structured_requests: HashMap::new(),
pending_thread_titles: HashMap::new(),
thread_event_listener_tasks: HashMap::new(),
+30 -1
View File
@@ -614,6 +614,31 @@ async fn enqueue_primary_thread_session_replays_buffered_approval_after_attach()
};
*request_id = AppServerRequestId::Integer(2);
params.thread_id = background_thread_id.to_string();
app.pending_app_server_requests
.note_server_request(&background_request);
app.enqueue_thread_request(background_thread_id, background_request.clone())
.await?;
crate::chatwidget::activate_voice_for_thread(&mut app.chat_widget, thread_id);
let mut tui = crate::tui::test_support::make_test_tui()?;
app.replace_chat_widget_with_app_server_thread(
&mut tui,
AppServerStartedThread {
session: test_thread_session(ThreadId::new(), app.config.cwd.to_path_buf()),
turns: Vec::new(),
blocks_direct_input: false,
task_tools_available: false,
},
session_lifecycle::ThreadAttachPresentation::SessionLineage,
/*initial_user_message*/ None,
)
.await?;
Box::pin(app.select_agent_thread(&mut tui, &mut app_server, thread_id)).await?;
assert!(
!app.pending_app_server_requests
.contains_server_request(&background_request)
);
app.pending_app_server_requests
.note_server_request(&background_request);
app.enqueue_thread_request(background_thread_id, background_request)
@@ -844,7 +869,7 @@ async fn reset_thread_event_state_aborts_listener_tasks() {
.await
.expect("listener task should report it started");
app.reset_thread_event_state();
app.reset_thread_event_state().await;
assert_eq!(app.thread_event_listener_tasks.is_empty(), true);
assert!(app.pending_server_profiles.is_empty());
@@ -6019,6 +6044,8 @@ async fn make_test_app() -> App {
pending_realtime_speech_replay: HashMap::new(),
pending_realtime_transcript_replay: HashMap::new(),
realtime_replay_order: VecDeque::new(),
background_voice: None,
background_voice_error: None,
temporary_structured_requests: HashMap::new(),
pending_thread_titles: HashMap::new(),
thread_event_listener_tasks: HashMap::new(),
@@ -6123,6 +6150,8 @@ pub(super) async fn make_test_app_with_channels() -> (
pending_realtime_speech_replay: HashMap::new(),
pending_realtime_transcript_replay: HashMap::new(),
realtime_replay_order: VecDeque::new(),
background_voice: None,
background_voice_error: None,
temporary_structured_requests: HashMap::new(),
pending_thread_titles: HashMap::new(),
thread_event_listener_tasks: HashMap::new(),
+19 -1
View File
@@ -264,7 +264,24 @@ async fn remote_disconnect_retires_voice_before_reconnecting() -> Result<()> {
thread_id, /*attempt_id*/ 0, /*input_generation*/ 0
));
assert!(app.begin_reconnect());
let (replacement, _, _, _) = make_chatwidget_manual_with_sender().await;
app.replace_chat_widget(replacement);
app.active_thread_id = None;
let (mut session, _, proxy) =
super::session_lifecycle_requests::start_recording_remote_app_server(&app.config).await?;
proxy.abort();
assert!(proxy.await.unwrap_err().is_cancelled());
let mut tui = crate::tui::test_support::make_test_tui()?;
app.handle_event(
&mut tui,
&mut session,
AppEvent::CodexOp(Op::RealtimeConversationStop { thread_id }),
)
.await?;
assert_eq!(
(app.voice_owner_thread_id(), app.active_thread_id),
(None, None)
);
assert_eq!(
(
app.reconnect.offline,
@@ -278,6 +295,7 @@ async fn remote_disconnect_retires_voice_before_reconnecting() -> Result<()> {
assert!(!app.chat_widget.is_current_realtime_attempt(
thread_id, /*attempt_id*/ 0, /*input_generation*/ 0
));
session.shutdown().await?;
Ok(())
}
@@ -140,6 +140,19 @@ async fn delegated_core_events_keep_private_output_hidden_and_deliver_final_spee
.thread_realtime_append_speech(thread_id, "fixture prompt".to_string())
.await?;
// Keep the real Core delegation running while another session is visible.
let other = app_server.start_thread(&app.config).await?;
let other_id = other.session.thread_id;
app.replace_chat_widget_with_app_server_thread(
&mut tui,
other,
super::super::session_lifecycle::ThreadAttachPresentation::Fresh,
/*initial_user_message*/ None,
)
.await?;
assert_eq!(app.voice_owner_thread_id(), Some(thread_id));
assert_eq!(app.chat_widget.thread_id(), Some(other_id));
let mut rendered = Vec::new();
let speech = timeout(Duration::from_secs(15), async {
loop {
@@ -158,7 +171,7 @@ async fn delegated_core_events_keep_private_output_hidden_and_deliver_final_spee
}
app.handle_event(&mut tui, &mut app_server, event).await?;
}
op = ops.recv() => {
op = ops.recv(), if !ops.is_closed() => {
let op = op.expect("TUI command stream should stay open");
if matches!(op, Op::RealtimeConversationSpeech { .. }) {
break Ok::<_, color_eyre::Report>(op);
@@ -191,6 +204,11 @@ async fn delegated_core_events_keep_private_output_hidden_and_deliver_final_spee
"channel": "speakable"
})
);
assert_eq!(app.chat_widget.thread_id(), Some(other_id));
assert_eq!(app.voice_owner_thread_id(), Some(thread_id));
app.select_agent_thread(&mut tui, &mut app_server, thread_id)
.await?;
assert!(app.chat_widget.realtime_conversation_is_running());
let rendered = rendered.join("\n");
assert!(!rendered.contains("PRIVATE REASONING"), "{rendered}");
assert!(!rendered.contains("PRIVATE COMMENTARY"), "{rendered}");
@@ -211,6 +229,31 @@ async fn delegated_core_events_keep_private_output_hidden_and_deliver_final_spee
"a final answer must be sent only once"
);
// Replaying a still-live answer must not duplicate the caption that arrives later.
app.chat_widget.handle_server_notification(
ServerNotification::ThreadRealtimeTranscriptDone(
codex_app_server_protocol::ThreadRealtimeTranscriptDoneNotification {
thread_id: thread_id.to_string(),
role: "assistant".into(),
text: "[ANALYSIS] is the marker you asked about.".into(),
},
),
/*replay_kind*/ None,
);
while let Ok(event) = app_events.try_recv() {
app.handle_event(&mut tui, &mut app_server, event).await?;
}
assert_eq!(
app.transcript_cells
.iter()
.flat_map(|cell| cell.display_lines(/*width*/ 100))
.filter(|line| line
.to_string()
.contains("[ANALYSIS] is the marker you asked about."))
.count(),
1,
);
// A typed request on the same live thread must follow the normal visible path.
app.chat_widget
.restore_user_message_to_composer(UserMessage::from("typed follow-up"));
@@ -219,9 +262,6 @@ async fn delegated_core_events_keep_private_output_hidden_and_deliver_final_spee
crossterm::event::KeyCode::Enter,
crossterm::event::KeyModifiers::NONE,
));
let typed_op = next_user_turn_op(&mut ops);
app.handle_event(&mut tui, &mut app_server, AppEvent::CodexOp(typed_op))
.await?;
let mut typed_rendered = Vec::new();
timeout(Duration::from_secs(10), async {
loop {
@@ -243,7 +283,7 @@ async fn delegated_core_events_keep_private_output_hidden_and_deliver_final_spee
}
app.handle_event(&mut tui, &mut app_server, event).await?;
}
op = ops.recv() => {
op = ops.recv(), if !ops.is_closed() => {
let op = op.expect("TUI command stream should stay open");
assert!(!matches!(op, Op::RealtimeConversationSpeech { .. }), "typed answer must not be spoken");
app.handle_event(&mut tui, &mut app_server, AppEvent::CodexOp(op)).await?;
+104 -20
View File
@@ -27,15 +27,20 @@ fn normalize_voice_snapshot_directory(rendered: &str, cwd: &Path) -> String {
"{placeholder}{}",
" ".repeat(cwd.len().saturating_sub(placeholder.len()))
);
rendered.replace(&cwd, &padded_placeholder)
rendered
.replace(&cwd, &padded_placeholder)
.lines()
.map(str::trim_end)
.collect::<Vec<_>>()
.join("\n")
}
#[test]
fn voice_snapshot_directory_keeps_header_width_for_windows_paths() {
let rendered = "│ directory: C:\\tmp\\project │";
let rendered = "│ directory: C:\\tmp\\project │\n gpt-test · C:\\tmp\\project";
assert_eq!(
normalize_voice_snapshot_directory(rendered, Path::new("C:\\tmp\\project")),
"│ directory: /tmp/project │"
"│ directory: /tmp/project │\n gpt-test · /tmp/project"
);
}
@@ -200,6 +205,9 @@ async fn shutdown_stops_backend_voice_once_through_app_server() -> Result<()> {
.handle_thread_session_quiet(test_thread_session(thread_id, app.config.cwd.to_path_buf()));
crate::chatwidget::activate_voice_for_thread(&mut app.chat_widget, thread_id);
let (other, _, _, _) = make_chatwidget_manual_with_sender().await;
app.replace_chat_widget(other);
assert_eq!(app.voice_owner_thread_id(), Some(thread_id));
app.shutdown_current_thread(&mut app_server).await;
app.shutdown_current_thread(&mut app_server).await;
@@ -221,6 +229,7 @@ async fn stalled_voice_stop_leaves_time_for_shutdown_unsubscribe() -> Result<()>
)
.await?;
let thread_id = ThreadId::new();
app.ensure_thread_channel(thread_id);
app.active_thread_id = Some(thread_id);
app.chat_widget
.handle_thread_session_quiet(test_thread_session(thread_id, app.config.cwd.to_path_buf()));
@@ -246,8 +255,8 @@ async fn stalled_voice_stop_leaves_time_for_shutdown_unsubscribe() -> Result<()>
}
#[tokio::test]
async fn switching_agent_threads_stops_backend_voice_once_through_app_server() -> Result<()> {
let (mut app, _events, mut ops) = make_test_app_with_channels().await;
async fn switching_agent_threads_preserves_backend_voice_and_routes_pending_speech() -> Result<()> {
let (mut app, mut events, mut ops) = make_test_app_with_channels().await;
let (mut app_server, requests, proxy) = start_recording_realtime_speech_app_server(
&app.config,
RealtimeRequestBehavior::AcceptSpeech,
@@ -339,19 +348,60 @@ async fn switching_agent_threads_stops_backend_voice_once_through_app_server() -
Box::pin(app.select_agent_thread(&mut tui, &mut app_server, target)).await?;
assert_eq!(app.chat_widget.thread_id(), Some(target));
assert_eq!(
app.pending_realtime_speech_replay[&source],
vec![(turn_id.into(), answer)]
);
assert_eq!(app.pending_realtime_transcript_replay[&source].len(), 1);
assert_eq!(
app.pending_realtime_transcript_replay[&source][0].text,
"queued before switch"
);
assert_eq!(app.voice_owner_thread_id(), Some(source));
assert!(recorded_params(&requests, "thread/realtime/stop").is_empty());
Box::pin(app.handle_event(&mut tui, &mut app_server, AppEvent::CodexOp(speech))).await?;
assert_eq!(app.active_thread_id, Some(target));
assert_eq!(app.chat_widget.thread_id(), Some(target));
// The newer user caption interrupted this queued answer, even across navigation.
assert!(recorded_params(&requests, "thread/realtime/appendSpeech").is_empty());
Box::pin(app.handle_event(
&mut tui,
&mut app_server,
AppEvent::VoiceControl {
thread_id: Some(source),
control: crate::app_event::VoiceControl::Mute,
},
))
.await?;
// A queued command still reaches its owner after navigation.
while let Ok(event) = events.try_recv() {
Box::pin(app.handle_event(&mut tui, &mut app_server, event)).await?;
}
Box::pin(app.handle_tui_event(
&mut tui,
&mut app_server,
tui::TuiEvent::Key(KeyCode::F(8).into()),
))
.await?;
while let Ok(event) = events.try_recv() {
Box::pin(app.handle_event(&mut tui, &mut app_server, event)).await?;
}
assert_eq!(app.voice_owner_thread_id(), None);
let stop = ops.try_recv()?;
Box::pin(app.handle_event(&mut tui, &mut app_server, AppEvent::CodexOp(stop))).await?;
assert_eq!(
recorded_params(&requests, "thread/realtime/stop"),
vec![serde_json::json!({"threadId": source.to_string()})]
);
// Resuming the ended owner must keep its newly attached subscription.
app.detach_current_thread_for_navigation(&mut app_server, Some(source))
.await;
assert_eq!(
recorded_params(&requests, "thread/unsubscribe"),
vec![serde_json::json!({"threadId": target.to_string()})]
);
app.detach_current_thread_for_navigation(&mut app_server, Some(ThreadId::new()))
.await;
assert!(
recorded_params(&requests, "thread/unsubscribe")
.contains(&serde_json::json!({"threadId": source.to_string()}))
);
Box::pin(app.select_agent_thread(&mut tui, &mut app_server, source)).await?;
assert!(std::iter::from_fn(|| events.try_recv().ok()).any(|event| {
matches!(event, AppEvent::InsertHistoryCell(cell) if cell.display_lines(/*width*/ 80)
.iter().any(|line| line.to_string().contains("Start voice mode before muting")))
}));
app_server.shutdown().await?;
proxy.await??;
Ok(())
@@ -382,7 +432,7 @@ async fn switching_threads_keeps_the_source_voice_partial_only_on_reattach() {
app.replace_chat_widget(side_widget);
app.chat_widget
.handle_thread_session_quiet(test_thread_session(side, app.config.cwd.to_path_buf()));
assert_eq!(app.pending_realtime_transcript_replay[&source].len(), 1);
let other_ui = render_bottom_popup(&app.chat_widget, /*width*/ 80);
while let Ok(event) = side_events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
assert!(
@@ -428,7 +478,13 @@ async fn switching_threads_keeps_the_source_voice_partial_only_on_reattach() {
assert_eq!(rendered.matches("spoken partial completed").count(), 1);
insta::assert_snapshot!(
"voice_partial_completed_after_thread_switch",
normalize_voice_snapshot_directory(&rendered, &app.config.cwd)
normalize_voice_snapshot_directory(
&format!(
"Other task:\n{other_ui}\nRestored owner:\n{}\n{rendered}",
render_bottom_popup(&app.chat_widget, /*width*/ 80)
),
&app.config.cwd,
)
);
while let Ok(event) = side_events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
@@ -455,7 +511,7 @@ async fn queued_voice_caption_after_switch_returns_once_to_its_source_thread() {
let (other_widget, _, _, _) = make_chatwidget_manual_with_sender().await;
app.active_thread_id = Some(ThreadId::new());
app.replace_chat_widget(other_widget);
assert!(app.pending_realtime_transcript_replay.contains_key(&source));
assert_eq!(app.voice_owner_thread_id(), Some(source));
let (source_widget, _, _, _) = make_chatwidget_manual_with_sender().await;
app.active_thread_id = Some(source);
@@ -486,6 +542,18 @@ async fn queued_voice_caption_after_switch_returns_once_to_its_source_thread() {
.await
.unwrap();
app.enqueue_thread_notification(
source,
ServerNotification::ThreadRealtimeClosed(
codex_app_server_protocol::ThreadRealtimeClosedNotification {
thread_id: source.to_string(),
reason: Some("requested".into()),
},
),
)
.await
.unwrap();
for notification in [
ServerNotification::ThreadRealtimeTranscriptDelta(
codex_app_server_protocol::ThreadRealtimeTranscriptDeltaNotification {
@@ -537,7 +605,6 @@ async fn queued_voice_caption_after_switch_returns_once_to_its_source_thread() {
)
.await
.unwrap();
assert_eq!(app.pending_realtime_transcript_replay[&source].len(), 2);
while let Ok(event) = other_events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
assert!(
@@ -740,7 +807,7 @@ async fn inactive_voice_replay_is_bounded_and_discarded_with_its_thread() {
.contains_key(&ids[16])
);
assert!(!app.realtime_replay_order.contains(&ids[16]));
app.reset_thread_event_state();
app.reset_thread_event_state().await;
assert!(app.pending_realtime_transcript_replay.is_empty());
assert!(app.pending_realtime_speech_replay.is_empty());
assert!(app.realtime_replay_order.is_empty());
@@ -978,7 +1045,6 @@ async fn completed_voice_caption_survives_repeated_thread_replacement() {
app.replace_chat_widget(side_widget);
app.chat_widget
.handle_thread_session_quiet(test_thread_session(side, app.config.cwd.to_path_buf()));
assert_eq!(app.pending_realtime_transcript_replay[&source].len(), 1);
while let Ok(event) = side_events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
assert!(
@@ -1262,6 +1328,17 @@ async fn switching_threads_retains_undelivered_voice_answer_after_replay_evictio
app.replace_chat_widget(side_widget);
app.chat_widget
.handle_thread_session_quiet(test_thread_session(side, app.config.cwd.to_path_buf()));
app.stop_realtime_conversation(&mut app_server).await;
app.enqueue_thread_notification(
original,
ServerNotification::ThreadRealtimeClosed(
codex_app_server_protocol::ThreadRealtimeClosedNotification {
thread_id: original.to_string(),
reason: Some("requested".into()),
},
),
)
.await?;
assert_eq!(app.pending_realtime_speech_replay[&original].len(), 1);
// Reconnect discards the old AppEvent receiver before installing a new widget.
let (replacement_tx, _replacement_rx) = tokio::sync::mpsc::unbounded_channel();
@@ -1374,6 +1451,13 @@ async fn switching_threads_retains_undelivered_voice_answer_after_replay_evictio
let (side_widget, _, _, _) = make_chatwidget_manual_with_sender().await;
app.active_thread_id = Some(side);
app.replace_chat_widget(side_widget);
app.app_server_target = AppServerTarget::LocalDaemon {
allow_embedded_fallback: true,
endpoint: crate::RemoteAppServerEndpoint::UnixSocket {
socket_path: test_path_buf("/tmp/unused.sock").abs(),
},
};
assert!(app.begin_reconnect());
assert_eq!(app.pending_realtime_speech_replay[&original].len(), 1);
let (original_widget, _, mut early_events, _) = make_chatwidget_manual_with_sender().await;
app.active_thread_id = Some(original);
@@ -3,6 +3,7 @@
use super::*;
use crate::app::tests::session_lifecycle_requests::recorded_params;
use crate::app::tests::session_lifecycle_requests::start_recording_remote_app_server;
use pretty_assertions::assert_eq;
#[tokio::test]
async fn rejected_voice_start_does_not_leave_local_capture_active() -> Result<()> {
@@ -99,6 +100,118 @@ async fn rejected_voice_stop_clears_the_owned_session() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn background_voice_closes_with_its_owner_not_the_visible_thread() -> Result<()> {
for owner_closed in [false, true] {
let (mut app, _events, mut ops) = make_test_app_with_channels().await;
let owner = ThreadId::new();
let visible = ThreadId::new();
crate::chatwidget::activate_voice_for_thread(&mut app.chat_widget, owner);
let (replacement, _, _, _) =
crate::chatwidget::tests::make_chatwidget_manual_with_sender().await;
app.replace_chat_widget(replacement);
app.chat_widget
.handle_thread_session_quiet(test_thread_session(
visible,
app.config.cwd.to_path_buf(),
));
app.active_thread_id = Some(visible);
let closed = if owner_closed { owner } else { visible };
app.enqueue_thread_notification(
closed,
ServerNotification::ThreadClosed(codex_app_server_protocol::ThreadClosedNotification {
thread_id: closed.to_string(),
}),
)
.await?;
assert_eq!(
app.voice_owner_thread_id(),
(!owner_closed).then_some(owner)
);
assert!(ops.try_recv().is_err());
}
Ok(())
}
#[tokio::test]
async fn lifecycle_actions_stop_parked_voice_before_removing_owner() -> Result<()> {
use crate::app_event::AgentsOverviewAction;
for (action, slash_command) in [
(AgentsOverviewAction::Archive, false),
(AgentsOverviewAction::Delete, false),
(AgentsOverviewAction::Archive, true),
(AgentsOverviewAction::Delete, true),
] {
let (mut app, _, _) = make_test_app_with_channels().await;
let (mut server, requests, proxy) = start_recording_remote_app_server(&app.config).await?;
let started = server.start_thread(&app.config).await?;
let root = started.session.thread_id;
let owner = if slash_command {
ThreadId::from_string(
&app_test_support::create_fake_parented_rollout_with_source(
&app.config.codex_home,
"2026-09-22T12-01-00",
"2026-09-22T12:01:00Z",
"Voice child",
Some(&app.config.model_provider_id),
/*git_info*/ None,
codex_protocol::protocol::SessionSource::SubAgent(
codex_protocol::protocol::SubAgentSource::ThreadSpawn {
parent_thread_id: root,
depth: 1,
agent_path: None,
agent_nickname: None,
agent_role: None,
},
),
codex_protocol::SessionId::from(root),
root,
)
.expect("materialize child session"),
)?
} else {
root
};
crate::chatwidget::activate_voice_for_thread(&mut app.chat_widget, owner);
let (replacement, _, _, _) =
crate::chatwidget::tests::make_chatwidget_manual_with_sender().await;
app.replace_chat_widget(replacement);
let mut tui = crate::tui::test_support::make_test_tui()?;
tui.pause_events();
if slash_command {
Box::pin(server.resume_thread(
&app.local_settings,
app.config.clone(),
owner,
crate::app_server_session::ResumeModelSettings::PreserveExistingThread,
))
.await?;
app.active_thread_id = Some(root);
app.app_server_target = AppServerTarget::Remote {
endpoint: crate::resolve_remote_addr("ws://127.0.0.1:9")?,
};
match action {
AgentsOverviewAction::Archive => {
Box::pin(app.archive_current_thread(&mut tui, &mut server)).await?;
}
AgentsOverviewAction::Delete => {
Box::pin(app.delete_current_thread(&mut tui, &mut server)).await?;
}
}
} else {
Box::pin(app.run_agents_overview_action(&mut tui, &mut server, owner, action)).await?;
}
assert_eq!(app.voice_owner_thread_id(), None);
assert_eq!(
recorded_params(&requests, "thread/realtime/stop"),
vec![serde_json::json!({"threadId": owner.to_string()})]
);
server.shutdown().await?;
proxy.await??;
}
Ok(())
}
#[tokio::test]
async fn voice_mute_shortcut_reaches_the_active_widget() -> Result<()> {
let (mut app, mut events, _) = make_test_app_with_channels().await;
@@ -716,6 +716,11 @@ goals = true
let primary_thread_id = ThreadId::new();
app.primary_thread_id = Some(primary_thread_id);
let voice_owner = ThreadId::new();
let (mut owner, _, _, _) = crate::chatwidget::tests::make_chatwidget_manual_with_sender().await;
crate::chatwidget::activate_voice_for_thread(&mut owner, voice_owner);
owner.park_voice();
app.background_voice = Some(Box::new(owner));
Box::pin(app.retry_safety_buffered_turn(
&mut tui,
&mut app_server,
@@ -790,6 +795,7 @@ goals = true
},
))
.await;
assert_eq!(app.voice_owner_thread_id(), Some(voice_owner));
if scenario == SafetyRetryScenario::UnsupportedPermissions {
assert_eq!(app.active_thread_id, Some(source_thread_id));
@@ -1,8 +1,18 @@
---
source: tui/src/app/tests/realtime_requests.rs
assertion_line: 302
expression: "normalize_voice_snapshot_directory(&rendered, &app.config.cwd)"
expression: "normalize_voice_snapshot_directory(&format!(\"Other task:\\n{other_ui}\\nRestored owner:\\n{}\\n{rendered}\",\nrender_bottom_popup(&app.chat_widget, 80)), &app.config.cwd,)"
---
Other task:
› Ask Codex to do anything
gpt-test default · /tmp/project
Restored owner:
voice ● listening ctrl+x mute /voice stop
mic ▁▁▁▁▁▁ codex ▁▁▁▁▁▁
› Ask Codex to do anything
gpt-test default · /tmp/project
╭────────────────────────────────────────╮
│ >_ OpenAI Codex (v0.0.0) │
│ │
+19 -10
View File
@@ -18,7 +18,13 @@ const REALTIME_STOP_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 1);
impl App {
pub(super) async fn stop_realtime_conversation(&mut self, app_server: &mut AppServerSession) {
let Some(thread_id) = self.chat_widget.reset_realtime_conversation() else {
let thread_id = self
.background_voice
.as_mut()
.and_then(|owner| owner.reset_realtime_conversation())
.or_else(|| self.chat_widget.reset_realtime_conversation());
self.retire_background_voice();
let Some(thread_id) = thread_id else {
return;
};
match tokio::time::timeout(
@@ -39,14 +45,8 @@ impl App {
pub(super) async fn shutdown_current_thread(&mut self, app_server: &mut AppServerSession) {
self.stop_realtime_conversation(app_server).await;
self.shutdown_side_threads(app_server).await;
if let Some(thread_id) = self.chat_widget.thread_id() {
if let Err(err) = app_server.thread_unsubscribe(thread_id).await {
tracing::warn!("failed to unsubscribe thread {thread_id}: {err}");
}
self.abort_thread_event_listener(thread_id);
self.pending_server_profiles.remove(&thread_id);
}
self.detach_current_thread_for_navigation(app_server, /*destination*/ None)
.await;
}
pub(super) async fn shutdown_side_threads(&mut self, app_server: &mut AppServerSession) {
@@ -1144,6 +1144,7 @@ impl App {
thread_id: ThreadId,
notification: ServerNotification,
) -> Result<()> {
self.deliver_background_voice_notification(thread_id, &notification);
if self.abandoned_side_threads.contains(&thread_id) {
return Ok(());
}
@@ -1261,7 +1262,13 @@ impl App {
guard.push_notification_ref(&notification);
Some(notification)
} else {
self.retain_inactive_realtime_transcript(thread_id, &notification);
if self
.background_voice
.as_ref()
.is_none_or(|owner| owner.thread_id() != Some(thread_id))
{
self.retain_inactive_realtime_transcript(thread_id, &notification);
}
guard.push_notification(notification);
None
};
@@ -1601,6 +1608,7 @@ impl App {
&replayed_final_items,
retained_assistant_captions,
);
self.restore_voice_owner_after_replay();
let pending = std::mem::take(&mut self.pending_primary_events);
for pending_event in pending {
match pending_event {
@@ -1910,6 +1918,7 @@ impl App {
&replayed_final_items,
retained_assistant_captions,
);
self.restore_voice_owner_after_replay();
self.chat_widget
.set_queue_autosend_suppressed(/*suppressed*/ false);
self.chat_widget
+139 -9
View File
@@ -1,17 +1,147 @@
//! Dispatches voice controls through the app before task navigation can change their target.
//! Keeps the single voice owner's live widget and subscription across navigation.
//! Parked widgets process their own events; only voice and account updates leave their sender.
use super::*;
use crate::app_event::VoiceControl;
impl App {
pub(super) fn control_voice(&mut self, control: crate::app_event::VoiceControl) {
use crate::app_event::VoiceControl;
match control {
VoiceControl::Toggle if !self.reconnect.offline => {
self.chat_widget.toggle_realtime_conversation()
pub(super) fn voice_owner_thread_id(&self) -> Option<ThreadId> {
self.background_voice
.as_deref()
.filter(|owner| owner.realtime_conversation_is_running())
.or_else(|| {
self.chat_widget
.realtime_conversation_is_running()
.then_some(&self.chat_widget)
})
.and_then(ChatWidget::thread_id)
}
pub(super) async fn stop_voice_for_removed_thread(
&mut self,
app_server: &mut AppServerSession,
thread_id: ThreadId,
) -> Result<()> {
if let Some(owner) = self.voice_owner_thread_id()
&& (owner == thread_id
|| app_server
.thread_read(owner, /*include_turns*/ false)
.await?
.session_id
== thread_id.to_string())
{
self.stop_realtime_conversation(app_server).await;
}
Ok(())
}
pub(super) fn voice_widget_for_thread(
&mut self,
thread_id: ThreadId,
) -> Option<&mut ChatWidget> {
if let Some(owner) = self.background_voice.as_deref_mut()
&& owner.thread_id() == Some(thread_id)
{
return Some(owner);
}
(self.chat_widget.thread_id() == Some(thread_id)).then_some(&mut self.chat_widget)
}
pub(super) fn control_voice(&mut self, control: VoiceControl) {
if let Some(thread_id) = self.voice_owner_thread_id()
&& let Some(owner) = self.voice_widget_for_thread(thread_id)
{
match control {
VoiceControl::Toggle => owner.toggle_realtime_conversation(),
VoiceControl::Stop => owner.stop_realtime_conversation(),
VoiceControl::Mute => owner.toggle_realtime_microphone(),
}
VoiceControl::Toggle => {}
VoiceControl::Stop => self.chat_widget.stop_realtime_conversation(),
VoiceControl::Mute => self.chat_widget.toggle_realtime_microphone(),
} else if matches!(control, VoiceControl::Toggle) && !self.reconnect.offline {
// Preserve the ended owner's transcript before allowing another call.
self.retire_background_voice();
self.chat_widget.toggle_realtime_conversation();
} else if matches!(control, VoiceControl::Mute) {
self.chat_widget.toggle_realtime_microphone();
}
}
pub(super) fn retire_background_voice(&mut self) {
if let Some(mut owner) = self.background_voice.take() {
std::mem::swap(&mut self.chat_widget, &mut owner);
self.retain_realtime_replay_state_before_replace();
std::mem::swap(&mut self.chat_widget, &mut owner);
}
}
pub(super) fn restore_voice_owner_after_replay(&mut self) {
if let Some(mut owner) = self
.background_voice
.take_if(|owner| owner.thread_id() == self.chat_widget.thread_id())
{
self.chat_widget.resume_background_voice(&mut owner);
}
if let Some((_, message)) = self
.background_voice_error
.take_if(|(thread_id, _)| Some(*thread_id) == self.chat_widget.thread_id())
{
self.chat_widget.add_error_message(message);
}
}
pub(super) fn deliver_background_voice_notification(
&mut self,
thread_id: ThreadId,
notification: &ServerNotification,
) {
if matches!(
notification,
ServerNotification::ThreadClosed(_)
| ServerNotification::ThreadArchived(_)
| ServerNotification::ThreadDeleted(_)
) && self.voice_owner_thread_id() == Some(thread_id)
&& let Some(owner) = self.voice_widget_for_thread(thread_id)
{
// The owning session is gone; no close acknowledgment can be required.
owner.reset_realtime_conversation();
}
if let Some(owner) = self.background_voice.as_mut()
&& owner.thread_id() == Some(thread_id)
{
owner.handle_server_notification(notification.clone(), /*replay_kind*/ None);
}
if self
.background_voice
.as_ref()
.is_some_and(|owner| !owner.may_receive_realtime_transcripts())
{
self.retire_background_voice();
}
}
pub(super) async fn detach_current_thread_for_navigation(
&mut self,
app_server: &mut AppServerSession,
destination: Option<ThreadId>,
) {
self.shutdown_side_threads(app_server).await;
let thread_ids: Vec<_> = self.thread_event_channels.keys().copied().collect();
for thread_id in thread_ids {
if self.voice_owner_thread_id() == Some(thread_id)
|| destination == Some(thread_id)
|| self
.agents_overview
.dispatched_requests
.contains_key(&thread_id)
|| self.agents_overview.blank_sessions.contains_key(&thread_id)
{
continue;
}
if let Err(err) = app_server.thread_unsubscribe(thread_id).await {
tracing::warn!("failed to unsubscribe thread {thread_id}: {err}");
}
self.abort_thread_event_listener(thread_id);
self.pending_server_profiles.remove(&thread_id);
}
}
}
+4
View File
@@ -1094,6 +1094,10 @@ pub(crate) enum AppEvent {
thread_id: Option<ThreadId>,
control: VoiceControl,
},
BackgroundVoiceError {
thread_id: ThreadId,
message: String,
},
/// Finish buffering initial resume replay after all replay events have been queued.
EndInitialHistoryReplayBuffer,
+27 -1
View File
@@ -4,6 +4,9 @@
//! without duplicating event construction or session logging behavior.
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use crate::app_command::AppCommand;
use crate::app_command::UserVerificationResponse;
@@ -23,16 +26,39 @@ use crate::session_log;
#[derive(Clone, Debug)]
pub(crate) struct AppEventSender {
pub app_event_tx: UnboundedSender<AppEvent>,
pub(crate) voice_only: Arc<AtomicBool>,
}
impl AppEventSender {
pub(crate) fn new(app_event_tx: UnboundedSender<AppEvent>) -> Self {
Self { app_event_tx }
Self {
app_event_tx,
voice_only: Arc::default(),
}
}
/// Send an event to the app event channel. If it fails, we swallow the
/// error and log it.
pub(crate) fn send(&self, event: AppEvent) {
// A parked voice owner keeps processing its session, but its UI and queued
// typed work must not act on the visible session. Thread-scoped metadata still applies.
if self.voice_only.load(Ordering::Relaxed)
&& !matches!(
&event,
AppEvent::CodexOp(
AppCommand::RealtimeConversationStart { .. }
| AppCommand::RealtimeConversationStop { .. }
| AppCommand::RealtimeConversationSpeech { .. }
) | AppEvent::RealtimeWebrtcOfferCreated { .. }
| AppEvent::RealtimeWebrtcConnected { .. }
| AppEvent::StopRealtimeConversation { .. }
| AppEvent::BackgroundVoiceError { .. }
| AppEvent::SyncThreadGitBranch { .. }
| AppEvent::RefreshRateLimits { .. }
)
{
return;
}
// Record inbound events for high-fidelity session replay.
// Avoid double-logging Ops; those are logged at the point of submission.
if !matches!(event, AppEvent::CodexOp(_)) {
@@ -33,6 +33,8 @@ impl ChatWidget {
terminal_title_invalid_items_warned,
session_telemetry,
} = common;
// Keep asynchronous widget callbacks scoped separately from the app and other widgets.
let app_event_tx = AppEventSender::new(app_event_tx.app_event_tx);
let model = model.filter(|m| !m.trim().is_empty());
let mut config = config;
config.model = model.clone();
+46 -14
View File
@@ -35,6 +35,9 @@ use std::sync::atomic::Ordering;
use std::time::Duration;
use std::time::Instant;
#[path = "realtime_navigation.rs"]
mod navigation;
#[cfg(test)]
#[path = "realtime_tests.rs"]
pub(crate) mod tests;
@@ -58,6 +61,7 @@ const INTERRUPTION_ACKNOWLEDGMENT: Duration = Duration::from_millis(400);
static NEXT_REALTIME_ATTEMPT_ID: AtomicU64 = AtomicU64::new(1);
static NEXT_REALTIME_SPEECH_DELIVERY_ID: AtomicU64 = AtomicU64::new(1);
#[derive(Clone)]
struct PendingRealtimeSpeech {
state: PendingSpeechState,
captioned: bool,
@@ -67,6 +71,7 @@ struct PendingRealtimeSpeech {
item: ThreadItem,
}
#[derive(Clone)]
pub(crate) struct RealtimeTranscriptRecord {
pub(crate) role: String,
pub(crate) text: String,
@@ -803,7 +808,14 @@ impl ChatWidget {
if questions.is_some() {
return false;
}
if from_replay || !self.is_realtime_delegated_agent_item(turn_id, item_id) {
if from_replay {
return self
.realtime_conversation
.pending_speech
.iter()
.any(|pending| pending.turn_id == turn_id && pending.item.id() == item_id);
}
if !self.is_realtime_delegated_agent_item(turn_id, item_id) {
return false;
}
let Some(RealtimeAgentItemOrigin::Delegated {
@@ -1011,10 +1023,23 @@ impl ChatWidget {
pub(crate) fn take_realtime_transcript_cells_for_replay(
&mut self,
) -> VecDeque<RealtimeTranscriptRecord> {
let mut records = std::mem::take(&mut self.realtime_conversation.accepted_transcripts);
let records = self.realtime_transcript_cells_for_replay();
self.realtime_conversation.accepted_transcripts.clear();
self.realtime_conversation.pending_history_cells.clear();
self.realtime_conversation.interleaved_transcript = None;
self.realtime_conversation.interleaved_transcript_cell = None;
self.realtime_conversation.transcript_role = None;
self.realtime_conversation.transcript.clear();
self.realtime_conversation.live_transcript_cell = None;
records
}
pub(crate) fn realtime_transcript_cells_for_replay(
&self,
) -> VecDeque<RealtimeTranscriptRecord> {
let mut records = self.realtime_conversation.accepted_transcripts.clone();
// Accepted records include cells already flushed into the old widget as
// well as those still deferred behind another assistant stream.
self.realtime_conversation.pending_history_cells.clear();
let mut retain_partial = |role: String, text: String| {
if text.trim().is_empty() {
return;
@@ -1037,14 +1062,11 @@ impl ChatWidget {
before_turn_id: None,
});
};
if let Some((role, text)) = self.realtime_conversation.interleaved_transcript.take() {
self.realtime_conversation.interleaved_transcript_cell = None;
if let Some((role, text)) = self.realtime_conversation.interleaved_transcript.clone() {
retain_partial(role, text);
}
if let Some(role) = self.realtime_conversation.transcript_role.take() {
let text = std::mem::take(&mut self.realtime_conversation.transcript);
retain_partial(role, text);
self.realtime_conversation.live_transcript_cell = None;
if let Some(role) = self.realtime_conversation.transcript_role.clone() {
retain_partial(role, self.realtime_conversation.transcript.clone());
}
records
}
@@ -1107,6 +1129,9 @@ impl ChatWidget {
}
pub(crate) fn restore_undelivered_realtime_speech(&mut self, delivery_id: u64) {
if self.app_event_tx.voice_only.load(Ordering::Relaxed) {
return;
}
if let Some(index) = self
.realtime_conversation
.pending_speech
@@ -1134,6 +1159,10 @@ impl ChatWidget {
}
fn restore_all_undelivered_realtime_speech(&mut self) {
// Preserve parked recovery without undoing overflow eviction.
if self.app_event_tx.voice_only.load(Ordering::Relaxed) {
return;
}
while let Some(delivery) = self.realtime_conversation.pending_speech.pop_front() {
self.restore_realtime_speech(delivery);
}
@@ -1638,7 +1667,7 @@ impl ChatWidget {
self.record_realtime_failure();
}
self.stop_realtime_conversation();
self.add_error_message(message);
self.add_realtime_error(message);
}
pub(super) fn on_realtime_conversation_closed(&mut self, reason: Option<String>) {
@@ -1695,10 +1724,12 @@ impl ChatWidget {
&& reason != "error"
&& !(failed && reason == "requested")
{
self.add_info_message(
format!("Voice conversation ended: {reason}"),
/*hint*/ None,
);
let message = format!("Voice conversation ended: {reason}");
if self.app_event_tx.voice_only.load(Ordering::Relaxed) && reason != "requested" {
self.add_realtime_error(message);
} else {
self.add_info_message(message, /*hint*/ None);
}
}
self.request_redraw();
}
@@ -1807,6 +1838,7 @@ impl ChatWidget {
.is_some();
self.realtime_conversation = RealtimeConversationUiState {
attempt_id: self.realtime_conversation.attempt_id,
pending_speech: std::mem::take(&mut self.realtime_conversation.pending_speech),
pending_history_cells,
accepted_transcripts,
delegated_reasoning_turns,
@@ -12,7 +12,7 @@ impl ChatWidget {
return;
}
} else if self.realtime_conversation.phase != RealtimeConversationPhase::Starting {
self.add_error_message("Start voice mode before muting the microphone.".to_string());
self.add_realtime_error("Start voice mode before muting the microphone.".to_string());
return;
}
@@ -126,7 +126,7 @@ impl ChatWidget {
}
}
pub(in crate::chatwidget) fn refresh_realtime_microphone_level(&mut self) {
pub(crate) fn refresh_realtime_microphone_level(&mut self) {
if !matches!(
self.realtime_conversation.phase,
RealtimeConversationPhase::Starting | RealtimeConversationPhase::Active
@@ -494,6 +494,13 @@ async fn voice_terminal_title_tracks_capture_activity_and_user_settings() {
check(&mut chat, "project");
chat.thread_id = Some(thread_id);
check(&mut chat, "● project");
chat.park_voice();
let (mut restored, _, _, _) = make_chatwidget_manual_with_sender().await;
restored.thread_id = Some(thread_id);
restored.local_settings.tui.animations = false;
restored.resume_background_voice(&mut chat);
assert_eq!(restored.last_terminal_title.as_deref(), Some("● project"));
chat = restored;
chat.reset_realtime_conversation();
assert_eq!(chat.last_terminal_title.as_deref(), Some("project"));
activate_voice(&mut chat);
@@ -0,0 +1,60 @@
//! Transfers live voice ownership into a reconstructed view without replaying the call.
use super::*;
impl ChatWidget {
pub(crate) fn add_realtime_error(&mut self, message: String) {
if self.app_event_tx.voice_only.load(Ordering::Relaxed)
&& let Some(thread_id) = self.thread_id()
{
self.app_event_tx.send(AppEvent::BackgroundVoiceError {
thread_id,
message: message.chars().take(/*n*/ 4096).collect(),
});
} else {
self.add_error_message(message);
}
}
pub(crate) fn park_voice(&mut self) {
self.app_event_tx.voice_only.store(true, Ordering::Relaxed);
self.set_queue_autosend_suppressed(/*suppressed*/ true);
self.stop_rate_limit_poller();
}
pub(crate) fn prepare_background_voice_replay(
&mut self,
owner: &Self,
) -> impl Iterator<Item = &str> {
self.realtime_conversation.pending_speech = owner
.realtime_conversation
.pending_speech
.iter()
.filter(|pending| !pending.captioned)
.cloned()
.collect();
self.realtime_conversation
.pending_speech
.iter()
.filter_map(|pending| match &pending.item {
codex_app_server_protocol::ThreadItem::AgentMessage { text, .. } => {
Some(text.as_str())
}
_ => None,
})
}
pub(crate) fn resume_background_voice(&mut self, owner: &mut ChatWidget) {
let replay = std::mem::replace(
&mut self.realtime_conversation,
std::mem::take(&mut owner.realtime_conversation),
);
// The reconstructed transcript has already restored historical captions.
// Keep its pending insertions, while preserving the live call's partials,
// delivery acknowledgments, and input generations.
self.realtime_conversation.pending_history_cells = replay.pending_history_cells;
self.update_realtime_footer();
self.refresh_terminal_title();
self.refresh_thread_usage_after_turn();
}
}
@@ -84,17 +84,36 @@ async fn voice_cannot_start_on_a_parent_owned_thread() {
#[tokio::test]
async fn stopping_while_the_offer_is_pending_resets_the_session() {
let (mut chat, _sender, _events, mut ops) = make_chatwidget_manual_with_sender().await;
let (mut chat, sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
let (abort, _registration) = AbortHandle::new_pair();
let observed_abort = abort.clone();
chat.realtime_conversation.phase = RealtimeConversationPhase::Starting;
chat.realtime_conversation.startup_abort = Some(abort);
chat.stop_realtime_conversation();
let (mut restored, _, _, mut ops) = make_chatwidget_manual_with_sender().await;
let callback = chat.app_event_tx.clone();
chat.park_voice();
callback.send(AppEvent::ResetTranscriptForThreadSwitch);
assert!(events.try_recv().is_err());
callback.send(AppEvent::RefreshRateLimits {
origin: crate::app_event::RateLimitRefreshOrigin::Recovery,
});
assert!(matches!(
events.try_recv(),
Ok(AppEvent::RefreshRateLimits {
origin: crate::app_event::RateLimitRefreshOrigin::Recovery
})
));
sender.send(AppEvent::ResetTranscriptForThreadSwitch);
assert!(events.try_recv().is_ok());
restored.resume_background_voice(&mut chat);
drop(chat);
assert!(!observed_abort.is_aborted());
restored.stop_realtime_conversation();
assert_eq!(
(
chat.realtime_conversation.phase,
restored.realtime_conversation.phase,
ops.try_recv().ok(),
observed_abort.is_aborted(),
),
@@ -74,7 +74,10 @@ async fn hidden_and_queued_voice_answers_share_a_lossless_sixteen_item_cap() {
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;
for index in 0..=super::super::MAX_PENDING_SPEECH_DELIVERIES {
for index in 0..=2 * super::super::MAX_PENDING_SPEECH_DELIVERIES {
if index == super::super::MAX_PENDING_SPEECH_DELIVERIES + 1 {
chat.park_voice();
}
let turn_id = format!("turn-{index}");
start_item(
&mut chat,
@@ -123,6 +126,12 @@ async fn hidden_and_queued_voice_answers_share_a_lossless_sixteen_item_cap() {
}
assert_eq!(restored, 1);
assert!(ops.try_recv().is_err());
chat.stop_realtime_conversation();
chat.reset_realtime_conversation();
assert_eq!(
chat.take_undelivered_realtime_speech_for_replay().len(),
super::super::MAX_PENDING_SPEECH_DELIVERIES
);
}
#[tokio::test]
@@ -250,6 +250,9 @@ impl ChatWidget {
/// Animated titles record their next refresh for the foreground loop, independently
/// of full TUI redraws.
fn refresh_terminal_title_from_selections(&mut self, selections: &StatusSurfaceSelections) {
if self.app_event_tx.voice_only.load(Ordering::Relaxed) {
return;
}
self.last_terminal_title_requires_action =
self.terminal_title_shows_action_required_with_selections(selections);
let now = Instant::now();
+1 -1
View File
@@ -374,7 +374,7 @@ pub(crate) async fn make_chatwidget_manual_with_sender() -> (
tokio::sync::mpsc::UnboundedReceiver<Op>,
) {
let (widget, rx, op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
let app_event_tx = widget.app_event_tx.clone();
let app_event_tx = AppEventSender::new(widget.app_event_tx.app_event_tx.clone());
(widget, app_event_tx, rx, op_rx)
}
@@ -732,7 +732,9 @@ async fn unchanged_thread_usage_has_bounded_settlement_refreshes() {
})),
));
chat.refresh_thread_usage_after_turn();
let (mut owner, _, _, _) = make_chatwidget_manual_with_sender().await;
owner.park_voice();
chat.resume_background_voice(&mut owner);
for attempt in 0..=THREAD_USAGE_SETTLEMENT_DELAYS.len() {
let request_id = match rx.try_recv() {
Ok(AppEvent::RefreshThreadUsage { request_id, .. }) => request_id,