Expand TUI regression coverage for realtime voice conversations (#43645)

## What changed

- Add app-to-RPC tests for voice startup, shutdown, thread switching, speech delivery, and rejected or stale speech, using a recording app server without audio hardware.
- Cover session retries, typed and voice handoffs, reasoning visibility, and recovery of undelivered answers in focused chat widget test modules.
- Add transcript and buffered replay tests for caption reconciliation, partial transcript preservation, and voice output ownership across thread changes, with rendering snapshots.

GitOrigin-RevId: 8a76523907289f94f257fe97386db7d2e7f33ca5
This commit is contained in:
Benjamin Carlsson
2026-09-08 02:09:29 +00:00
committed by copyberry
parent e7637306bc
commit 6fee98cc85
32 changed files with 5396 additions and 88 deletions
+2
View File
@@ -44,6 +44,8 @@ mod plugin_catalog;
mod rate_limits;
#[path = "tests/realtime_handoff_e2e.rs"]
mod realtime_handoff_e2e;
#[path = "tests/realtime_requests.rs"]
mod realtime_requests;
#[path = "tests/realtime_start.rs"]
mod realtime_start;
#[path = "tests/recap_generation_tests.rs"]
@@ -71,6 +71,360 @@ fn buffered_replay_keeps_unfinished_and_unrelated_deltas_in_order() {
assert_eq!(store.snapshot().events.len(), 6);
}
#[test]
fn evicted_voice_delegation_marker_still_suppresses_private_replay() {
let mut store = ThreadEventStore::new(/*capacity*/ 2);
let voice_request = ThreadItem::UserMessage {
id: "user".into(),
client_id: None,
content: vec![codex_app_server_protocol::UserInput::Text {
text: "<realtime_delegation><input>question</input></realtime_delegation>".into(),
text_elements: Vec::new(),
}],
};
store.push_notification(ServerNotification::ItemStarted(
codex_app_server_protocol::ItemStartedNotification {
thread_id: "thread".into(),
turn_id: "voice-turn".into(),
started_at_ms: 0,
item: voice_request.clone(),
},
));
store.push_notification(delta("thread", "voice-turn", "reasoning"));
let private = ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id: "thread".into(),
turn_id: "voice-turn".into(),
completed_at_ms: 0,
item: ThreadItem::AgentMessage {
id: "reasoning".into(),
text: "[ANALYSIS] private".into(),
phase: None,
memory_citation: None,
delivery: None,
questions: None,
},
});
store.push_notification(private.clone());
assert!(store.snapshot().events.is_empty());
store.push_notification(delta("thread", "voice-turn", "unfinished"));
assert!(store.snapshot().events.is_empty());
assert_eq!(store.snapshot().delegated_turns, vec!["voice-turn"]);
for notification in [
ServerNotification::ReasoningSummaryPartAdded(
codex_app_server_protocol::ReasoningSummaryPartAddedNotification {
thread_id: "thread".into(),
turn_id: "voice-turn".into(),
item_id: "reasoning".into(),
summary_index: 0,
},
),
ServerNotification::ReasoningSummaryTextDelta(
codex_app_server_protocol::ReasoningSummaryTextDeltaNotification {
thread_id: "thread".into(),
turn_id: "voice-turn".into(),
item_id: "reasoning".into(),
delta: "private summary".into(),
summary_index: 0,
},
),
ServerNotification::ReasoningTextDelta(
codex_app_server_protocol::ReasoningTextDeltaNotification {
thread_id: "thread".into(),
turn_id: "voice-turn".into(),
item_id: "reasoning".into(),
delta: "private raw reasoning".into(),
content_index: 0,
},
),
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id: "thread".into(),
turn_id: "voice-turn".into(),
completed_at_ms: 0,
item: ThreadItem::Reasoning {
id: "reasoning".into(),
summary: vec!["private summary".into()],
content: vec!["private raw reasoning".into()],
},
}),
] {
store.push_notification(notification);
assert!(store.snapshot().events.is_empty());
}
let mut resumed = ThreadEventStore::new(/*capacity*/ 1);
resumed.set_turns(vec![Turn {
id: "voice-turn".into(),
items: vec![voice_request],
items_view: codex_app_server_protocol::TurnItemsView::Summary,
status: TurnStatus::InProgress,
error: None,
started_at: None,
completed_at: None,
duration_ms: None,
}]);
resumed.push_notification(delta("thread", "voice-turn", "unfinished"));
let mut newer = resumed.turns[0].clone();
newer.id = "newer-turn".into();
resumed.set_turns(vec![resumed.turns[0].clone(), newer]);
assert!(resumed.snapshot().events.is_empty());
resumed.push_notification(delta("thread", "newer-turn", "unfinished"));
assert!(resumed.snapshot().events.is_empty());
let mut ordinary = ThreadEventStore::new(/*capacity*/ 2);
let ServerNotification::ItemCompleted(mut normal) = private else {
unreachable!()
};
normal.turn_id = "normal-turn".into();
ordinary.push_notification(ServerNotification::ItemCompleted(normal));
assert_eq!(ordinary.snapshot().events.len(), 1);
}
#[test]
fn saved_voice_turn_suppresses_reasoning_without_hiding_typed_reasoning() {
let reasoning = ThreadItem::Reasoning {
id: "reasoning".into(),
summary: vec!["private summary".into()],
content: vec!["private raw reasoning".into()],
};
let voice_request = ThreadItem::UserMessage {
id: "voice-request".into(),
client_id: None,
content: vec![codex_app_server_protocol::UserInput::Text {
text: "<realtime_delegation><input>question</input></realtime_delegation>".into(),
text_elements: Vec::new(),
}],
};
let turn = |id: &str, items| Turn {
id: id.into(),
items,
items_view: codex_app_server_protocol::TurnItemsView::Full,
status: TurnStatus::Completed,
error: None,
started_at: None,
completed_at: None,
duration_ms: None,
};
let mut store = ThreadEventStore::new(/*capacity*/ 4);
store.set_turns(vec![
turn("voice", vec![voice_request.clone(), reasoning.clone()]),
turn("typed", vec![reasoning.clone()]),
]);
assert_eq!(
store.snapshot().turns,
vec![
turn("voice", vec![voice_request]),
turn("typed", vec![reasoning])
]
);
}
#[test]
fn snapshot_keeps_typed_output_before_voice_handoff_in_the_same_turn() {
let commentary = |id: &str, text: &str| ThreadItem::AgentMessage {
id: id.into(),
text: text.into(),
phase: Some(codex_protocol::models::MessagePhase::Commentary),
questions: None,
memory_citation: None,
delivery: None,
};
let marker = ThreadItem::UserMessage {
id: "voice".into(),
client_id: None,
content: vec![codex_app_server_protocol::UserInput::Text {
text: "<realtime_delegation><input>question</input></realtime_delegation>".into(),
text_elements: Vec::new(),
}],
};
let typed = commentary("typed", "Earlier typed commentary");
let private = commentary("private", "Later voice-private commentary");
let mut store = ThreadEventStore::new(/*capacity*/ 8);
store.set_turns(vec![Turn {
id: "shared".into(),
items: vec![typed.clone(), marker.clone(), private.clone()],
items_view: codex_app_server_protocol::TurnItemsView::Full,
status: TurnStatus::InProgress,
error: None,
started_at: None,
completed_at: None,
duration_ms: None,
}]);
for item in [typed.clone(), marker.clone(), private] {
store.push_notification(ServerNotification::ItemCompleted(
ItemCompletedNotification {
thread_id: "thread".into(),
turn_id: "shared".into(),
completed_at_ms: 0,
item,
},
));
}
let snapshot = store.snapshot();
assert_eq!(snapshot.turns[0].items, vec![typed.clone(), marker]);
let completed = snapshot
.events
.into_iter()
.filter_map(|event| match event {
ThreadBufferedEvent::Notification(notification) => match *notification {
ServerNotification::ItemCompleted(item) => Some(item.item),
_ => None,
},
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(completed, vec![typed, snapshot.turns[0].items[1].clone()]);
}
#[test]
fn snapshot_keeps_typed_item_completed_after_voice_handoff() {
let typed = ThreadItem::AgentMessage {
id: "typed".into(),
text: "Typed commentary completed late".into(),
phase: Some(codex_protocol::models::MessagePhase::Commentary),
questions: None,
memory_citation: None,
delivery: None,
};
let private = ThreadItem::AgentMessage {
id: "private".into(),
text: "Voice-private commentary".into(),
phase: Some(codex_protocol::models::MessagePhase::Commentary),
questions: None,
memory_citation: None,
delivery: None,
};
let marker = ThreadItem::UserMessage {
id: "voice".into(),
client_id: None,
content: vec![codex_app_server_protocol::UserInput::Text {
text: "<realtime_delegation><input>question</input></realtime_delegation>".into(),
text_elements: Vec::new(),
}],
};
let mut store = ThreadEventStore::new(/*capacity*/ 8);
for item in [typed.clone(), marker, private.clone()] {
store.push_notification(ServerNotification::ItemStarted(
codex_app_server_protocol::ItemStartedNotification {
thread_id: "thread".into(),
turn_id: "shared".into(),
started_at_ms: 0,
item,
},
));
}
for item in [typed.clone(), private] {
store.push_notification(ServerNotification::ItemCompleted(
ItemCompletedNotification {
thread_id: "thread".into(),
turn_id: "shared".into(),
completed_at_ms: 0,
item,
},
));
}
let completed = store
.snapshot()
.events
.into_iter()
.filter_map(|event| match event {
ThreadBufferedEvent::Notification(notification) => match *notification {
ServerNotification::ItemCompleted(n) => Some(n.item),
_ => None,
},
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(completed, vec![typed]);
}
#[tokio::test]
async fn evicted_voice_marker_survives_widget_snapshot_for_late_reasoning() {
let thread_id = ThreadId::new();
let mut store = ThreadEventStore::new(/*capacity*/ 1);
store.push_notification(ServerNotification::ItemStarted(
codex_app_server_protocol::ItemStartedNotification {
thread_id: thread_id.to_string(),
turn_id: "voice-turn".into(),
started_at_ms: 0,
item: ThreadItem::UserMessage {
id: "user".into(),
client_id: None,
content: vec![codex_app_server_protocol::UserInput::Text {
text: "<realtime_delegation><input>question</input></realtime_delegation>"
.into(),
text_elements: Vec::new(),
}],
},
},
));
store.push_notification(delta(&thread_id.to_string(), "voice-turn", "private"));
let snapshot = store.snapshot();
assert!(snapshot.events.is_empty());
assert!(snapshot.turns.is_empty());
assert_eq!(snapshot.delegated_turns, vec!["voice-turn"]);
let mut app = make_test_app().await;
let (widget, _sender, mut events, _ops) =
crate::chatwidget::tests::make_chatwidget_manual_with_sender().await;
app.active_thread_id = Some(thread_id);
app.replace_chat_widget(widget);
app.replay_thread_snapshot(snapshot, /*resume_restored_queue*/ false);
app.chat_widget.handle_server_notification(
ServerNotification::ReasoningSummaryTextDelta(
codex_app_server_protocol::ReasoningSummaryTextDeltaNotification {
thread_id: thread_id.to_string(),
turn_id: "voice-turn".into(),
item_id: "late-reasoning".into(),
delta: "private after switch".into(),
summary_index: 0,
},
),
/*replay_kind*/ None,
);
app.chat_widget.handle_server_notification(
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id: thread_id.to_string(),
turn_id: "voice-turn".into(),
completed_at_ms: 0,
item: ThreadItem::Reasoning {
id: "late-reasoning".into(),
summary: vec!["private after switch".into()],
content: Vec::new(),
},
}),
/*replay_kind*/ None,
);
app.chat_widget.handle_server_notification(
delta(&thread_id.to_string(), "voice-turn", "private-commentary"),
/*replay_kind*/ None,
);
app.chat_widget.handle_server_notification(
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id: thread_id.to_string(),
turn_id: "voice-turn".into(),
completed_at_ms: 0,
item: ThreadItem::AgentMessage {
id: "private-commentary".into(),
text: "[COMMENTARY] private after switch".into(),
phase: None,
memory_citation: None,
delivery: None,
questions: None,
},
}),
/*replay_kind*/ None,
);
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
let rendered = cell
.transcript_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<String>();
assert!(!rendered.contains("private"), "{rendered}");
}
}
}
#[tokio::test]
async fn buffered_replay_renders_completed_text_without_streaming_again() {
let (mut app, mut events, _ops) = make_test_app_with_channels().await;
File diff suppressed because it is too large Load Diff
@@ -197,11 +197,56 @@ pub(super) async fn start_recording_remote_app_server(
/// Proxies a real app server while optionally rejecting modern pagination like an older server.
pub(super) async fn start_recording_app_server_with_history(
config: &Config,
history_capabilities: HistoryCapabilities,
blocked_thread_list: Option<(ThreadId, oneshot::Sender<()>, oneshot::Receiver<()>)>,
failed_thread_name: Option<&'static str>,
thread_params_mode: crate::app_server_session::ThreadParamsMode,
loader_overrides: LoaderOverrides,
) -> Result<RecordingAppServer> {
start_recording_app_server_with_realtime_speech(
config,
history_capabilities,
blocked_thread_list,
failed_thread_name,
thread_params_mode,
RealtimeRequestBehavior::Forward,
loader_overrides,
)
.await
}
#[derive(Clone, Copy, PartialEq, Eq)]
pub(super) enum RealtimeRequestBehavior {
Forward,
AcceptStart,
AcceptSpeech,
AcceptSpeechAndStallStop,
}
pub(super) async fn start_recording_realtime_speech_app_server(
config: &Config,
realtime_behavior: RealtimeRequestBehavior,
) -> Result<RecordingAppServer> {
start_recording_app_server_with_realtime_speech(
config,
HistoryCapabilities::Current,
/*blocked_thread_list*/ None,
/*failed_thread_name*/ None,
crate::app_server_session::ThreadParamsMode::Embedded,
realtime_behavior,
LoaderOverrides::default(),
)
.await
}
async fn start_recording_app_server_with_realtime_speech(
config: &Config,
history_capabilities: HistoryCapabilities,
mut blocked_thread_list: Option<(ThreadId, oneshot::Sender<()>, oneshot::Receiver<()>)>,
failed_thread_name: Option<&'static str>,
thread_params_mode: crate::app_server_session::ThreadParamsMode,
realtime_behavior: RealtimeRequestBehavior,
loader_overrides: LoaderOverrides,
) -> Result<RecordingAppServer> {
let state_db =
@@ -256,6 +301,11 @@ pub(super) async fn start_recording_app_server_with_history(
.lock()
.expect("request recorder lock")
.push(request.clone());
if realtime_behavior == RealtimeRequestBehavior::AcceptSpeechAndStallStop
&& request.method == "thread/realtime/stop"
{
continue;
}
let request_id = request.id.clone();
let params = request.params.as_ref();
let requires_pagination = match request.method.as_str() {
@@ -286,7 +336,20 @@ pub(super) async fn start_recording_app_server_with_history(
.is_some_and(|tools| {
tools.iter().any(|tool| tool["type"] == "namespace")
});
let response = if let HistoryCapabilities::ConfigReadUnsupported(code) =
let fake_realtime_response = (realtime_behavior
== RealtimeRequestBehavior::AcceptStart
&& request.method == "thread/realtime/start")
|| (matches!(
realtime_behavior,
RealtimeRequestBehavior::AcceptSpeech
| RealtimeRequestBehavior::AcceptSpeechAndStallStop
) && request.method == "thread/realtime/appendSpeech");
let response = if fake_realtime_response {
JSONRPCMessage::Response(JSONRPCResponse {
id: request_id,
result: serde_json::json!({}),
})
} else if let HistoryCapabilities::ConfigReadUnsupported(code) =
history_capabilities
&& request.method == "config/read"
{
@@ -0,0 +1,7 @@
---
source: tui/src/app/tests/realtime_requests.rs
assertion_line: 1083
expression: "rendered.join(\"\\n\").replace(&thread_id.to_string(), \"<thread-id>\")"
---
• An answer to keep.
■ Voice conversation failed: thread/realtime/appendSpeech failed in TUI: thread/realtime/appendSpeech failed: thread not found: <thread-id> (code -32600)
@@ -0,0 +1,23 @@
---
source: tui/src/app/tests/realtime_requests.rs
assertion_line: 645
expression: "normalize_voice_snapshot_directory(&rendered, &app.config.cwd)"
---
╭────────────────────────────────────────╮
│ >_ OpenAI Codex (v0.0.0) │
│ │
│ model: gpt-test /model to change │
│ directory: /tmp/project │
╰────────────────────────────────────────╯
To get started, describe a task or try one of these commands:
/init - create an AGENTS.md file with instructions for Codex
/status - show current session configuration
/permissions - choose what Codex is allowed to do
/model - choose what model and reasoning effort to use
/review - review any changes and find issues
› buffered question
• buffered answer
@@ -0,0 +1,20 @@
---
source: tui/src/app/tests/realtime_requests.rs
assertion_line: 182
expression: "rendered.replace(&app.config.cwd.display().to_string(), \"/tmp/project\")"
---
╭────────────────────────────────────────╮
│ >_ OpenAI Codex (v0.0.0) │
│ │
│ model: gpt-test /model to change │
│ directory: /tmp/project │
╰────────────────────────────────────────╯
To get started, describe a task or try one of these commands:
/init - create an AGENTS.md file with instructions for Codex
/status - show current session configuration
/permissions - choose what Codex is allowed to do
/model - choose what model and reasoning effort to use
/review - review any changes and find issues
• spoken complete
@@ -0,0 +1,21 @@
---
source: tui/src/app/tests/realtime_requests.rs
assertion_line: 302
expression: "normalize_voice_snapshot_directory(&rendered, &app.config.cwd)"
---
╭────────────────────────────────────────╮
│ >_ OpenAI Codex (v0.0.0) │
│ │
│ model: gpt-test /model to change │
│ directory: /tmp/project │
╰────────────────────────────────────────╯
To get started, describe a task or try one of these commands:
/init - create an AGENTS.md file with instructions for Codex
/status - show current session configuration
/permissions - choose what Codex is allowed to do
/model - choose what model and reasoning effort to use
/review - review any changes and find issues
› spoken partial completed
@@ -0,0 +1,33 @@
---
source: tui/src/app/tests/realtime_requests.rs
assertion_line: 561
expression: "normalize_voice_snapshot_directory(&rendered, &app.config.cwd)"
---
╭────────────────────────────────────────╮
│ >_ OpenAI Codex (v0.0.0) │
│ │
│ model: gpt-test /model to change │
│ directory: /tmp/project │
╰────────────────────────────────────────╯
To get started, describe a task or try one of these commands:
/init - create an AGENTS.md file with instructions for Codex
/status - show current session configuration
/permissions - choose what Codex is allowed to do
/model - choose what model and reasoning effort to use
/review - review any changes and find issues
› show <code> & tests
• spoken answer
› typed words
› show <code> & tests
› typed words
• different caption
+118 -86
View File
@@ -1,11 +1,37 @@
//! Focused visual coverage and shared setup for voice behavior tests.
//! Regression coverage for thread-owned voice sessions and transcript/handoff safety.
//! Synthetic events preserve typed turns and reject stale session generations.
use super::RealtimeConversationPhase;
use crate::app_command::AppCommand;
use crate::app_event::AppEvent;
use crate::chatwidget::ChatWidget;
use crate::chatwidget::ReplayKind;
use crate::chatwidget::tests::make_chatwidget_manual_with_sender;
use crate::chatwidget::tests::render_bottom_popup;
use codex_app_server_protocol::AgentMessageDeltaNotification;
use codex_app_server_protocol::ItemCompletedNotification;
use codex_app_server_protocol::ItemStartedNotification;
use codex_app_server_protocol::ReasoningSummaryPartAddedNotification;
use codex_app_server_protocol::ReasoningSummaryTextDeltaNotification;
use codex_app_server_protocol::ReasoningTextDeltaNotification;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::Turn;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnItemsView;
use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::UserInput;
use codex_protocol::ThreadId;
use codex_protocol::items::AsyncUserInputQuestion;
use codex_protocol::models::MessagePhase;
use futures::future::AbortHandle;
use std::collections::VecDeque;
fn activate_voice(chat: &mut ChatWidget) -> ThreadId {
let thread_id = ThreadId::new();
activate_voice_for_thread(chat, thread_id);
thread_id
}
pub(crate) fn activate_voice_for_thread(chat: &mut ChatWidget, thread_id: ThreadId) {
chat.thread_id = Some(thread_id);
@@ -15,93 +41,99 @@ pub(crate) fn activate_voice_for_thread(chat: &mut ChatWidget, thread_id: Thread
chat.realtime_conversation.latest_input_was_voice = true;
}
#[tokio::test]
async fn enabling_voice_on_an_open_thread_snapshots_the_new_thread_notice() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
chat.set_feature_enabled(
codex_features::Feature::RealtimeConversation,
/*enabled*/ true,
fn user_item(text: &str) -> ThreadItem {
ThreadItem::UserMessage {
id: format!("user-{text}"),
client_id: None,
content: vec![UserInput::Text {
text: text.to_string(),
text_elements: Vec::new(),
}],
}
}
fn agent_item(item_id: &str, text: &str, phase: Option<MessagePhase>) -> ThreadItem {
ThreadItem::AgentMessage {
id: item_id.to_string(),
text: text.to_string(),
phase,
questions: None,
memory_citation: None,
delivery: None,
}
}
fn start_item(chat: &mut ChatWidget, thread_id: ThreadId, turn_id: &str, item: ThreadItem) {
chat.handle_server_notification(
ServerNotification::ItemStarted(ItemStartedNotification {
thread_id: thread_id.to_string(),
turn_id: turn_id.to_string(),
item,
started_at_ms: 0,
}),
/*replay_kind*/ None,
);
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
}
fn complete_item(chat: &mut ChatWidget, thread_id: ThreadId, turn_id: &str, item: ThreadItem) {
chat.handle_server_notification(
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id: thread_id.to_string(),
turn_id: turn_id.to_string(),
item,
completed_at_ms: 0,
}),
/*replay_kind*/ None,
);
}
fn finish_turn(
chat: &mut ChatWidget,
thread_id: ThreadId,
turn_id: &str,
items: Vec<ThreadItem>,
status: TurnStatus,
) {
chat.handle_server_notification(
ServerNotification::TurnCompleted(TurnCompletedNotification {
thread_id: thread_id.to_string(),
turn: Turn {
id: turn_id.to_string(),
items,
items_view: TurnItemsView::Summary,
status,
error: None,
started_at: None,
completed_at: None,
duration_ms: None,
},
}),
/*replay_kind*/ None,
);
}
fn without_completion_metadata(rendered: &str) -> String {
rendered
.lines()
.filter_map(|line| {
let stable = line
.split_once(" done ")
.map_or(line, |(before, _)| before);
(!stable.is_empty()).then_some(stable)
})
.collect::<Vec<_>>()
.join("\n");
insta::assert_snapshot!(rendered, @"• Voice conversations will be available in new threads.");
.join("\n")
}
#[tokio::test]
async fn voice_footer_renders_the_main_conversation_states() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
let mut states = Vec::new();
for (label, phase, muted, level, speaker_level, role, transcript) in [
(
"connecting",
RealtimeConversationPhase::Starting,
false,
0,
0,
None,
"",
),
(
"listening",
RealtimeConversationPhase::Active,
false,
0,
0,
None,
"",
),
(
"speaking",
RealtimeConversationPhase::Active,
false,
4,
5,
None,
"",
),
(
"muted",
RealtimeConversationPhase::Active,
true,
4,
4,
None,
"",
),
(
"transcript",
RealtimeConversationPhase::Active,
false,
2,
2,
Some("assistant"),
"Hello there",
),
] {
chat.realtime_conversation.phase = phase;
chat.realtime_conversation.microphone_muted = muted;
chat.realtime_conversation.microphone_level = level;
chat.realtime_conversation.speaker_level = speaker_level;
chat.realtime_conversation.transcript_role = role.map(str::to_string);
chat.realtime_conversation.transcript = transcript.to_string();
chat.update_realtime_footer();
states.push(format!(
"{label}:\n{}",
render_bottom_popup(&chat, /*width*/ 80)
));
}
insta::assert_snapshot!(states.join("\n\n"));
}
#[path = "realtime_tests/caption_replay.rs"]
mod caption_replay;
#[path = "realtime_tests/handoff_privacy.rs"]
mod handoff_privacy;
#[path = "realtime_tests/handoffs.rs"]
mod handoffs;
#[path = "realtime_tests/lifecycle.rs"]
mod lifecycle;
#[path = "realtime_tests/speech_recovery.rs"]
mod speech_recovery;
#[path = "realtime_tests/transcripts.rs"]
mod transcripts;
@@ -0,0 +1,392 @@
//! Replay and captions reconcile once across thread and session changes.
use super::*;
use pretty_assertions::assert_eq;
#[tokio::test]
async fn replay_preserves_typed_updates_before_voice_steers_the_turn() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
chat.thread_id = Some(ThreadId::new());
chat.replay_thread_turns(
vec![Turn {
id: "typed-then-voice".into(),
items: vec![
user_item("Typed question"),
agent_item(
"typed-update",
"Checking the typed request",
Some(MessagePhase::Commentary),
),
ThreadItem::Reasoning {
id: "typed-reasoning".into(),
summary: vec!["Typed reasoning summary".into()],
content: Vec::new(),
},
user_item(
"<realtime_delegation><input>spoken correction</input></realtime_delegation>",
),
agent_item(
"private-update",
"Private voice commentary",
Some(MessagePhase::Commentary),
),
ThreadItem::Reasoning {
id: "private-reasoning".into(),
summary: vec!["Private voice reasoning".into()],
content: Vec::new(),
},
],
items_view: TurnItemsView::Full,
status: TurnStatus::Completed,
error: None,
started_at: None,
completed_at: None,
duration_ms: None,
}],
ReplayKind::ThreadSnapshot,
);
chat.flush_answer_stream_with_separator();
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.transcript_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
assert!(
rendered.contains("Checking the typed request"),
"{rendered}"
);
assert!(rendered.contains("Typed reasoning summary"), "{rendered}");
assert!(!rendered.contains("Private voice"), "{rendered}");
insta::assert_snapshot!("typed_turn_replay_before_voice_steering", rendered);
}
#[tokio::test]
async fn in_progress_voice_replay_restores_the_late_reasoning_guard() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
let thread_id = ThreadId::new();
chat.thread_id = Some(thread_id);
let voice_request =
user_item("<realtime_delegation><input>question</input></realtime_delegation>");
chat.replay_thread_turns(
vec![Turn {
id: "saved-voice-turn".into(),
items: vec![voice_request.clone()],
items_view: TurnItemsView::Full,
status: TurnStatus::InProgress,
error: None,
started_at: None,
completed_at: None,
duration_ms: None,
}],
ReplayKind::ThreadSnapshot,
);
chat.handle_server_notification(
ServerNotification::ItemStarted(ItemStartedNotification {
thread_id: thread_id.to_string(),
turn_id: "buffered-voice-turn".into(),
item: voice_request,
started_at_ms: 0,
}),
Some(ReplayKind::ThreadSnapshot),
);
while events.try_recv().is_ok() {}
for turn_id in ["saved-voice-turn", "buffered-voice-turn"] {
chat.handle_server_notification(
ServerNotification::ReasoningSummaryTextDelta(ReasoningSummaryTextDeltaNotification {
thread_id: thread_id.to_string(),
turn_id: turn_id.into(),
item_id: "late-reasoning".into(),
delta: "private after switch".into(),
summary_index: 0,
}),
/*replay_kind*/ None,
);
complete_item(
&mut chat,
thread_id,
turn_id,
ThreadItem::Reasoning {
id: "late-reasoning".into(),
summary: vec!["private after switch".into()],
content: Vec::new(),
},
);
assert!(chat.is_realtime_delegated_reasoning_turn(turn_id));
}
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
let rendered = cell
.transcript_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<String>();
assert!(!rendered.contains("private"), "{rendered}");
}
}
}
#[tokio::test]
async fn accepted_voice_answer_with_only_an_old_caption_returns_to_history_on_close() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "accepted-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>question</input></realtime_delegation>"),
);
let answer = agent_item("answer", "Accepted answer", Some(MessagePhase::FinalAnswer));
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = ops.try_recv().unwrap() else {
panic!("completed voice turn should queue speech");
};
while events.try_recv().is_ok() {}
chat.accept_realtime_speech(delivery_id);
assert!(!chat.has_pending_realtime_speech(delivery_id));
chat.realtime_conversation.assistant_transcript_generation =
Some(chat.realtime_conversation.input_generation.wrapping_sub(1));
chat.on_realtime_transcript_done("assistant".into(), "Earlier speech".into());
assert_eq!(chat.realtime_conversation.pending_speech.len(), 1);
chat.on_realtime_conversation_closed(Some("transport_closed".into()));
chat.restore_undelivered_realtime_speech(delivery_id);
let mut answers = 0;
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
let rendered = cell
.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<String>();
answers += usize::from(rendered.contains("Accepted answer"));
}
}
assert_eq!(answers, 1);
}
#[tokio::test]
async fn unrelated_caption_started_before_speech_queue_keeps_answer_fallback() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "answer-after-old-caption";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>question</input></realtime_delegation>"),
);
chat.on_realtime_transcript_delta("assistant".into(), "Unrelated ".into());
let answer = agent_item("answer", "New answer", Some(MessagePhase::FinalAnswer));
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = ops.try_recv().unwrap() else {
panic!("completed voice turn should queue speech");
};
while events.try_recv().is_ok() {}
chat.accept_realtime_speech(delivery_id);
chat.on_realtime_transcript_done("assistant".into(), "Unrelated caption".into());
assert_eq!(chat.realtime_conversation.pending_speech.len(), 1);
chat.on_realtime_conversation_closed(Some("transport_closed".into()));
let restored = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.filter(|text| text.contains("New answer"))
.count();
assert_eq!(restored, 1);
}
#[tokio::test]
async fn done_only_caption_cannot_retire_a_newer_voice_answer() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "done-only-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>question</input></realtime_delegation>"),
);
let answer = agent_item("answer", "New answer", Some(MessagePhase::FinalAnswer));
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = ops.try_recv().unwrap() else {
panic!("completed voice turn should queue speech");
};
while events.try_recv().is_ok() {}
chat.accept_realtime_speech(delivery_id);
assert!(
chat.realtime_conversation
.assistant_transcript_generation
.is_none()
);
chat.on_realtime_transcript_done("assistant".into(), "Old done-only caption".into());
assert_eq!(chat.realtime_conversation.pending_speech.len(), 1);
chat.on_realtime_conversation_closed(Some("transport_closed".into()));
let restored = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.filter(|text| text.contains("New answer"))
.count();
assert_eq!(restored, 1);
}
#[tokio::test]
async fn captioned_voice_answer_does_not_duplicate_on_close() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "captioned-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>question</input></realtime_delegation>"),
);
let answer = agent_item(
"answer",
"Captioned answer",
Some(MessagePhase::FinalAnswer),
);
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = ops.try_recv().unwrap() else {
panic!("completed voice turn should queue speech");
};
while events.try_recv().is_ok() {}
// A delta establishes which input generation owns the completed caption.
chat.on_realtime_transcript_delta("assistant".into(), "Captioned ".into());
chat.on_realtime_transcript_done("assistant".into(), "Captioned answer".into());
chat.accept_realtime_speech(delivery_id);
chat.reset_realtime_conversation();
chat.restore_undelivered_realtime_speech(delivery_id);
let mut rendered = Vec::new();
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
rendered.extend(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string()),
);
}
}
insta::assert_snapshot!("captioned_voice_answer_on_close", rendered.join("\n"));
}
#[tokio::test]
async fn restored_partial_caption_accepts_late_completion_without_duplicate_history() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
chat.restore_realtime_transcript_cells(VecDeque::from([
super::super::RealtimeTranscriptRecord {
role: "user".into(),
text: "last ".into(),
complete: false,
},
]));
let live = chat
.realtime_conversation
.live_transcript_cell
.as_ref()
.unwrap();
insta::assert_snapshot!(
"restored_voice_partial_live",
live.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n")
);
chat.on_realtime_transcript_delta("user".into(), "words".into());
let retained = chat.take_realtime_transcript_cells_for_replay();
assert_eq!(retained.len(), 1);
assert_eq!(retained[0].text, "last words");
chat.restore_realtime_transcript_cells(retained);
chat.on_realtime_transcript_done("user".into(), "last words".into());
assert!(chat.realtime_conversation.live_transcript_cell.is_none());
assert_eq!(chat.realtime_conversation.accepted_transcripts.len(), 1);
assert_eq!(
chat.realtime_conversation.accepted_transcripts[0].text,
"last words"
);
assert!(chat.realtime_conversation.accepted_transcripts[0].complete);
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<String>(),
),
_ => None,
})
.collect::<String>();
assert_eq!(rendered.matches("last words").count(), 1);
}
#[tokio::test]
async fn empty_late_completion_discards_the_restored_partial() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
chat.restore_realtime_transcript_cells(VecDeque::from([
super::super::RealtimeTranscriptRecord {
role: "user".into(),
text: "unfinished".into(),
complete: false,
},
]));
chat.on_realtime_transcript_done("user".into(), String::new());
assert!(chat.realtime_conversation.live_transcript_cell.is_none());
assert!(chat.realtime_conversation.accepted_transcripts.is_empty());
assert!(events.try_recv().is_err());
}
@@ -0,0 +1,481 @@
//! Typed and spoken steering preserve ownership and private reasoning boundaries.
use super::*;
use pretty_assertions::assert_eq;
fn complete_stale_handoff(chat: &mut ChatWidget, thread_id: ThreadId, question: &str) {
let turn_id = "stale-turn";
start_item(
chat,
thread_id,
turn_id,
user_item(&format!(
"<realtime_delegation><input>{question}</input></realtime_delegation>"
)),
);
let answer = agent_item(
"stale-answer",
"stale answer",
Some(MessagePhase::FinalAnswer),
);
start_item(chat, thread_id, turn_id, answer.clone());
complete_item(chat, thread_id, turn_id, answer.clone());
finish_turn(
chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
}
#[tokio::test]
async fn typed_turn_remains_a_normal_text_response_while_voice_is_active() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "typed-turn";
chat.turn_lifecycle.last_turn_id = Some(turn_id.to_string());
let typed = user_item("pick a random number");
start_item(&mut chat, thread_id, turn_id, typed.clone());
complete_item(&mut chat, thread_id, turn_id, typed);
let answer = agent_item("typed-answer", "73", Some(MessagePhase::FinalAnswer));
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
assert!(matches!(
chat.pending_notification.as_ref(),
Some(crate::chatwidget::Notification::AgentTurnComplete { response }) if response == "73"
));
let mut rendered = String::new();
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
for line in cell.transcript_lines(/*width*/ 80) {
rendered.push_str(&line.to_string());
}
}
}
assert!(rendered.contains("pick a random number"));
assert!(rendered.contains("73"));
assert!(ops.try_recv().is_err());
}
#[tokio::test]
async fn typed_steering_restores_normal_output_for_an_existing_voice_item() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "shared-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>earlier voice</input></realtime_delegation>"),
);
let answer = agent_item(
"shared-answer",
"typed answer",
Some(MessagePhase::FinalAnswer),
);
start_item(&mut chat, thread_id, turn_id, answer.clone());
let private = agent_item(
"private-commentary",
"private voice commentary",
Some(MessagePhase::Commentary),
);
start_item(&mut chat, thread_id, turn_id, private.clone());
start_item(
&mut chat,
thread_id,
turn_id,
user_item("new typed question"),
);
complete_item(&mut chat, thread_id, turn_id, private);
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
let mut rendered = String::new();
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
for line in cell.display_lines(/*width*/ 80) {
rendered.push_str(&line.to_string());
}
}
}
assert!(rendered.contains("typed answer"), "{rendered}");
assert!(!rendered.contains("private voice commentary"), "{rendered}");
insta::assert_snapshot!(
"typed_steering_keeps_voice_commentary_private",
without_completion_metadata(&rendered)
);
assert!(ops.try_recv().is_err());
}
#[tokio::test]
async fn newer_voice_steering_an_existing_typed_turn_is_spoken_once() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "shared-typed-turn";
start_item(&mut chat, thread_id, turn_id, user_item("typed task"));
let earlier_typed = agent_item("typed-item", "existing typed work", /*phase*/ None);
start_item(&mut chat, thread_id, turn_id, earlier_typed);
chat.on_realtime_transcript_delta("user".to_string(), "spoken follow-up".to_string());
chat.on_realtime_transcript_done("user".to_string(), "spoken follow-up".to_string());
let delegation =
user_item("<realtime_delegation><input>spoken follow-up</input></realtime_delegation>");
start_item(&mut chat, thread_id, turn_id, delegation.clone());
complete_item(&mut chat, thread_id, turn_id, delegation);
let answer = agent_item(
"voice-item",
"answer to the spoken follow-up",
/*phase*/ None,
);
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
assert!(matches!(
ops.try_recv(),
Ok(AppCommand::RealtimeConversationSpeech { text, .. })
if text.as_str() == "answer to the spoken follow-up"
));
assert!(ops.try_recv().is_err());
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
assert!(
cell.display_lines(/*width*/ 80)
.iter()
.all(|line| !line.to_string().contains("realtime_delegation"))
);
}
}
}
#[tokio::test]
async fn voice_handoff_preserves_started_typed_reasoning_but_hides_new_reasoning() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "mixed-reasoning-turn";
start_item(&mut chat, thread_id, turn_id, user_item("typed task"));
start_item(
&mut chat,
thread_id,
turn_id,
ThreadItem::Reasoning {
id: "typed-reasoning".into(),
summary: Vec::new(),
content: Vec::new(),
},
);
chat.handle_server_notification(
ServerNotification::ReasoningSummaryTextDelta(ReasoningSummaryTextDeltaNotification {
thread_id: thread_id.to_string(),
turn_id: turn_id.into(),
item_id: "typed-reasoning".into(),
delta: "Typed beginning ".into(),
summary_index: 0,
}),
/*replay_kind*/ None,
);
let delegation =
user_item("<realtime_delegation><input>spoken follow-up</input></realtime_delegation>");
start_item(&mut chat, thread_id, turn_id, delegation);
chat.handle_server_notification(
ServerNotification::ReasoningSummaryTextDelta(ReasoningSummaryTextDeltaNotification {
thread_id: thread_id.to_string(),
turn_id: turn_id.into(),
item_id: "typed-reasoning".into(),
delta: "and typed tail".into(),
summary_index: 0,
}),
/*replay_kind*/ None,
);
complete_item(
&mut chat,
thread_id,
turn_id,
ThreadItem::Reasoning {
id: "typed-reasoning".into(),
summary: vec!["Typed beginning and typed tail".into()],
content: Vec::new(),
},
);
start_item(
&mut chat,
thread_id,
turn_id,
ThreadItem::Reasoning {
id: "private-reasoning".into(),
summary: Vec::new(),
content: Vec::new(),
},
);
chat.handle_server_notification(
ServerNotification::ReasoningSummaryTextDelta(ReasoningSummaryTextDeltaNotification {
thread_id: thread_id.to_string(),
turn_id: turn_id.into(),
item_id: "private-reasoning".into(),
delta: "private after handoff".into(),
summary_index: 0,
}),
/*replay_kind*/ None,
);
complete_item(
&mut chat,
thread_id,
turn_id,
ThreadItem::Reasoning {
id: "private-reasoning".into(),
summary: vec!["private after handoff".into()],
content: Vec::new(),
},
);
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.transcript_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
assert!(
rendered.contains("Typed beginning and typed tail"),
"{rendered}"
);
assert!(!rendered.contains("private after handoff"), "{rendered}");
insta::assert_snapshot!("typed_reasoning_through_voice_handoff", rendered);
}
#[tokio::test]
async fn voice_delegation_without_transcript_can_steer_an_existing_typed_turn() {
let (mut chat, _sender, _events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "shared-turn-without-transcript";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("earlier typed task"),
);
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>new spoken request</input></realtime_delegation>"),
);
let answer = agent_item(
"voice-answer",
"new spoken answer",
Some(MessagePhase::FinalAnswer),
);
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
assert!(matches!(
ops.try_recv(),
Ok(AppCommand::RealtimeConversationSpeech { text, .. }) if text.as_str() == "new spoken answer"
));
assert!(ops.try_recv().is_err());
}
#[tokio::test]
async fn typed_agent_item_completes_visibly_after_voice_handoff() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "typed-then-voice";
start_item(&mut chat, thread_id, turn_id, user_item("typed request"));
let typed = agent_item(
"typed-commentary",
"Visible typed commentary",
Some(MessagePhase::Commentary),
);
start_item(&mut chat, thread_id, turn_id, typed.clone());
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>voice request</input></realtime_delegation>"),
);
complete_item(&mut chat, thread_id, turn_id, typed);
let private = agent_item(
"voice-commentary",
"Hidden voice commentary",
Some(MessagePhase::Commentary),
);
start_item(&mut chat, thread_id, turn_id, private.clone());
complete_item(&mut chat, thread_id, turn_id, private);
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
assert!(rendered.contains("Visible typed commentary"), "{rendered}");
assert!(!rendered.contains("Hidden voice commentary"), "{rendered}");
}
#[tokio::test]
async fn stale_voice_handoff_after_newer_typed_input_is_never_spoken() {
let (mut chat, _sender, _events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
chat.on_realtime_transcript_done("user".to_string(), "How are you doing".to_string());
chat.note_realtime_typed_input("what's the best pizza in new york?");
complete_stale_handoff(&mut chat, thread_id, "How are you doing");
assert!(ops.try_recv().is_err());
}
#[tokio::test]
async fn typed_submission_suppresses_old_speaker_audio() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
chat.note_realtime_typed_input("typed correction");
let generation = chat.realtime_conversation.input_generation;
assert_eq!(
chat.realtime_conversation.speaker_suppression_generation,
Some(generation)
);
start_item(
&mut chat,
thread_id,
"typed-turn",
user_item("typed correction"),
);
assert_eq!(
chat.realtime_conversation.speaker_suppression_generation,
Some(chat.realtime_conversation.input_generation)
);
}
#[tokio::test]
async fn late_voice_transcript_after_typed_input_cannot_revive_a_stale_handoff() {
let (mut chat, _sender, _events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
chat.on_realtime_transcript_delta("user".to_string(), "How are ".to_string());
chat.note_realtime_typed_input("what's the best pizza in new york?");
chat.on_realtime_transcript_delta("user".to_string(), "you doing".to_string());
chat.on_realtime_transcript_done("user".to_string(), "How are you doing".to_string());
complete_stale_handoff(&mut chat, thread_id, "How are you doing");
assert!(!chat.realtime_conversation.latest_input_was_voice);
assert!(ops.try_recv().is_err());
}
#[tokio::test]
async fn duplicate_delta_less_transcript_after_typed_input_cannot_revive_old_voice() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
chat.on_realtime_transcript_done("user".to_string(), "old question".to_string());
while events.try_recv().is_ok() {}
chat.note_realtime_typed_input("newer typed question");
let typed_generation = chat.realtime_conversation.input_generation;
chat.on_realtime_transcript_done("user".to_string(), "old question".to_string());
assert_eq!(
chat.realtime_conversation.input_generation,
typed_generation
);
assert!(!chat.realtime_conversation.latest_input_was_voice);
assert!(events.try_recv().is_err());
complete_stale_handoff(&mut chat, thread_id, "old question");
assert!(ops.try_recv().is_err());
}
#[tokio::test]
async fn stale_voice_handoff_longer_than_the_display_limit_cannot_be_revived() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let old_question = format!("{} final words", "é".repeat(600));
chat.on_realtime_transcript_done("user".to_string(), old_question.clone());
while events.try_recv().is_ok() {}
chat.note_realtime_typed_input("a newer typed question");
let typed_generation = chat.realtime_conversation.input_generation;
chat.on_realtime_transcript_done("user".to_string(), old_question.clone());
assert_eq!(
chat.realtime_conversation.input_generation,
typed_generation
);
assert!(!chat.realtime_conversation.latest_input_was_voice);
assert!(events.try_recv().is_err());
complete_stale_handoff(&mut chat, thread_id, &old_question);
assert!(ops.try_recv().is_err());
}
#[tokio::test]
async fn new_voice_handoff_without_transcript_still_speaks_after_a_typed_turn() {
let (mut chat, _sender, _events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
chat.note_realtime_typed_input("a previous typed question");
let turn_id = "new-voice-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>fresh spoken request</input></realtime_delegation>"),
);
let answer = agent_item(
"answer",
"The voice answer",
Some(MessagePhase::FinalAnswer),
);
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
assert!(matches!(
ops.try_recv(),
Ok(AppCommand::RealtimeConversationSpeech { text, .. }) if text.as_str() == "The voice answer"
));
assert!(ops.try_recv().is_err());
}
@@ -0,0 +1,424 @@
//! Voice delegation routes completed answers without exposing private protocol text.
use super::*;
use pretty_assertions::assert_eq;
#[tokio::test]
async fn closing_voice_before_turn_completion_restores_the_delegated_answer_once() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "voice-turn-closing";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>the answer?</input></realtime_delegation>"),
);
let answer = agent_item(
"final-answer",
"The answer is 42.",
Some(MessagePhase::FinalAnswer),
);
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
assert!(events.try_recv().is_err());
chat.on_realtime_conversation_closed(/*reason*/ None);
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
let mut visible_answers = 0;
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
visible_answers += cell
.display_lines(/*width*/ 80)
.iter()
.any(|line| line.to_string().contains("The answer is 42."))
as usize;
}
}
assert_eq!(visible_answers, 1);
assert!(ops.try_recv().is_err());
}
#[tokio::test]
async fn delegated_voice_work_speaks_only_the_completed_final_answer_once() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "voice-turn";
chat.turn_lifecycle.last_turn_id = Some(turn_id.to_string());
let delegation =
user_item("<realtime_delegation><input>best pizza?</input></realtime_delegation>");
start_item(&mut chat, thread_id, turn_id, delegation.clone());
complete_item(&mut chat, thread_id, turn_id, delegation);
let commentary = agent_item(
"commentary",
"[COMMENTARY] checking current guides",
Some(MessagePhase::Commentary),
);
start_item(&mut chat, thread_id, turn_id, commentary.clone());
complete_item(&mut chat, thread_id, turn_id, commentary);
let first_final = agent_item("first-final", "Earlier guess", /*phase*/ None);
start_item(&mut chat, thread_id, turn_id, first_final.clone());
complete_item(&mut chat, thread_id, turn_id, first_final);
let final_answer = agent_item(
"last-final",
"[FINAL] Lucali in Brooklyn",
/*phase*/ None,
);
start_item(&mut chat, thread_id, turn_id, final_answer.clone());
chat.handle_server_notification(
ServerNotification::AgentMessageDelta(AgentMessageDeltaNotification {
thread_id: thread_id.to_string(),
turn_id: turn_id.to_string(),
item_id: "last-final".to_string(),
delta: "Lucali".to_string(),
}),
/*replay_kind*/ None,
);
complete_item(&mut chat, thread_id, turn_id, final_answer.clone());
assert!(ops.try_recv().is_err());
assert!(events.try_recv().is_err());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![final_answer],
TurnStatus::Completed,
);
assert!(
chat.pending_notification.is_none(),
"voice-delegated turns should not trigger desktop completion notifications"
);
assert!(matches!(
ops.try_recv(),
Ok(AppCommand::RealtimeConversationSpeech { thread_id: spoken_thread_id, text, .. })
if spoken_thread_id == thread_id && text.as_str() == "Lucali in Brooklyn"
));
assert!(ops.try_recv().is_err());
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
assert!(
cell.display_lines(/*width*/ 80)
.iter()
.all(|line| !line.to_string().contains("Earlier guess"))
);
}
}
}
#[tokio::test]
async fn delegation_started_before_peer_connection_keeps_its_voice_origin() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
chat.realtime_conversation.phase = RealtimeConversationPhase::Starting;
chat.realtime_conversation.backend_started = true;
chat.realtime_conversation.webrtc_connected = false;
let turn_id = "startup-delegation";
chat.turn_lifecycle.agent_turn_running = true;
chat.turn_lifecycle.last_turn_id = Some(turn_id.to_string());
let delegation =
user_item("<realtime_delegation><input>spoken question</input></realtime_delegation>");
start_item(&mut chat, thread_id, turn_id, delegation.clone());
complete_item(&mut chat, thread_id, turn_id, delegation);
chat.on_realtime_webrtc_connected(chat.realtime_conversation.attempt_id, Ok(()));
let final_answer = agent_item(
"startup-final",
"[FINAL] Spoken answer",
/*phase*/ None,
);
start_item(&mut chat, thread_id, turn_id, final_answer.clone());
complete_item(&mut chat, thread_id, turn_id, final_answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![final_answer],
TurnStatus::Completed,
);
assert!(matches!(
ops.try_recv(),
Ok(AppCommand::RealtimeConversationSpeech { text, .. }) if text.as_str() == "Spoken answer"
));
assert!(ops.try_recv().is_err());
let mut rendered_history = Vec::new();
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
let rendered = cell
.transcript_lines(/*width*/ 80)
.iter()
.map(ToString::to_string)
.collect::<String>();
assert!(!rendered.contains("Spoken answer"));
assert!(!rendered.contains("<realtime_delegation>"));
rendered_history.push(rendered);
}
}
insta::assert_snapshot!(
"voice_delegation_during_connection",
without_completion_metadata(&rendered_history.join("\n"))
);
}
#[tokio::test]
async fn delegated_answer_with_async_question_opens_the_editor_instead_of_speech() {
let (mut chat, _sender, _events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "question-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>choose</input></realtime_delegation>"),
);
let answer = ThreadItem::AgentMessage {
id: "question-answer".into(),
text: "Which option?".into(),
phase: Some(MessagePhase::FinalAnswer),
questions: Some(vec![AsyncUserInputQuestion {
title: "Which option?".into(),
options: None,
}]),
memory_citation: None,
delivery: None,
};
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
assert_eq!(
chat.bottom_pane
.questions
.as_ref()
.map(|editor| editor.unanswered_count()),
Some(1)
);
assert!(
ops.try_recv().is_err(),
"question-bearing answer stays in the TUI"
);
}
#[tokio::test]
async fn typed_turn_running_during_peer_connection_stays_typed() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
chat.realtime_conversation.phase = RealtimeConversationPhase::Starting;
chat.realtime_conversation.backend_started = true;
chat.realtime_conversation.webrtc_connected = false;
chat.turn_lifecycle.agent_turn_running = true;
chat.turn_lifecycle.last_turn_id = Some("typed-startup".to_string());
chat.on_realtime_webrtc_connected(chat.realtime_conversation.attempt_id, Ok(()));
assert!(!chat.realtime_conversation.latest_input_was_voice);
assert!(matches!(
chat.realtime_conversation.turn_origins.get("typed-startup"),
Some(super::super::RealtimeTurnOrigin::Typed { .. })
));
assert_eq!(chat.thread_id(), Some(thread_id));
}
#[tokio::test]
async fn delegated_item_that_becomes_final_at_turn_completion_is_recoverable() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "changed-final-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>question</input></realtime_delegation>"),
);
let analysis = agent_item(
"same-id",
"[ANALYSIS] secret",
Some(MessagePhase::Commentary),
);
start_item(&mut chat, thread_id, turn_id, analysis.clone());
complete_item(&mut chat, thread_id, turn_id, analysis);
let final_answer = agent_item("same-id", "Public answer", Some(MessagePhase::FinalAnswer));
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![final_answer],
TurnStatus::Completed,
);
let AppCommand::RealtimeConversationSpeech {
delivery_id, text, ..
} = ops.try_recv().unwrap()
else {
panic!("changed final should be delivered");
};
assert_eq!(text.as_str(), "Public answer");
chat.restore_undelivered_realtime_speech(delivery_id);
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
for line in cell.display_lines(/*width*/ 80) {
assert!(!line.to_string().contains("secret"));
}
}
}
}
#[tokio::test]
async fn explicit_final_answer_can_explain_private_channel_markers() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "marker-final-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>explain markers</input></realtime_delegation>"),
);
let answer = agent_item(
"marker-final",
"[COMMENTARY] is a label in the documentation.",
Some(MessagePhase::FinalAnswer),
);
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
let AppCommand::RealtimeConversationSpeech {
delivery_id, text, ..
} = ops.try_recv().expect("final answer should be delivered")
else {
panic!("expected speech delivery");
};
assert_eq!(
text.as_str(),
"[COMMENTARY] is a label in the documentation."
);
assert!(ops.try_recv().is_err());
chat.restore_undelivered_realtime_speech(delivery_id);
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.transcript_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
insta::assert_snapshot!(
"explicit_final_answer_with_channel_marker",
without_completion_metadata(&rendered)
);
}
#[tokio::test]
async fn newer_spoken_input_prevents_an_older_voice_turn_from_speaking() {
let (mut chat, _sender, _events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
chat.on_realtime_transcript_delta("user".to_string(), "first question".to_string());
chat.on_realtime_transcript_done("user".to_string(), "first question".to_string());
let old_turn_id = "old-voice-turn";
start_item(
&mut chat,
thread_id,
old_turn_id,
user_item("<realtime_delegation><input>first question</input></realtime_delegation>"),
);
let old_answer = agent_item(
"old-answer",
"answer to the earlier question",
Some(MessagePhase::FinalAnswer),
);
start_item(&mut chat, thread_id, old_turn_id, old_answer.clone());
complete_item(&mut chat, thread_id, old_turn_id, old_answer.clone());
chat.on_realtime_transcript_delta("user".to_string(), "newer question".to_string());
chat.on_realtime_transcript_done("user".to_string(), "newer question".to_string());
finish_turn(
&mut chat,
thread_id,
old_turn_id,
vec![old_answer],
TurnStatus::Completed,
);
assert!(ops.try_recv().is_err());
}
#[tokio::test]
async fn newer_voice_delegation_on_the_same_turn_speaks_only_its_own_answer() {
let (mut chat, _sender, _events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
chat.on_realtime_transcript_done("user".to_string(), "first".to_string());
let turn_id = "shared-voice-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>first question</input></realtime_delegation>"),
);
let old_answer = agent_item(
"old-answer",
"answer to the earlier question",
Some(MessagePhase::FinalAnswer),
);
start_item(&mut chat, thread_id, turn_id, old_answer.clone());
complete_item(&mut chat, thread_id, turn_id, old_answer);
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>newer question</input></realtime_delegation>"),
);
let latest_answer = agent_item(
"latest-answer",
"answer to the newer question",
Some(MessagePhase::FinalAnswer),
);
start_item(&mut chat, thread_id, turn_id, latest_answer.clone());
complete_item(&mut chat, thread_id, turn_id, latest_answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![latest_answer],
TurnStatus::Completed,
);
assert!(matches!(
ops.try_recv(),
Ok(AppCommand::RealtimeConversationSpeech { text, .. })
if text.as_str() == "answer to the newer question"
));
let owner = chat.realtime_conversation.speaker_suppression_generation;
assert_eq!(owner, Some(chat.realtime_conversation.input_generation));
chat.on_realtime_transcript_delta("assistant".to_string(), "answer".to_string());
let owner = chat.realtime_conversation.speaker_suppression_generation;
assert!(owner.is_none());
assert!(ops.try_recv().is_err());
}
@@ -0,0 +1,581 @@
//! Voice startup, shutdown, and retry maintain one active owned session.
use super::*;
use pretty_assertions::assert_eq;
#[tokio::test]
async fn enabling_voice_on_an_open_thread_snapshots_the_new_thread_notice() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
chat.set_feature_enabled(
codex_features::Feature::RealtimeConversation,
/*enabled*/ true,
);
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
insta::assert_snapshot!(rendered, @"• Voice conversations will be available in new threads.");
}
#[tokio::test]
async fn voice_cannot_start_in_a_side_conversation() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
chat.set_side_conversation_active(/*active*/ true);
chat.toggle_realtime_conversation();
let Ok(AppEvent::InsertHistoryCell(cell)) = events.try_recv() else {
panic!("voice should report that side conversations are unsupported");
};
assert!(
cell.display_lines(/*width*/ 80)
.iter()
.any(|line| line.to_string().contains("side conversations"))
);
assert!(ops.try_recv().is_err());
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Inactive
);
}
#[tokio::test]
async fn mute_during_startup_is_saved_before_the_offer_handle_exists() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
chat.realtime_conversation.phase = RealtimeConversationPhase::Starting;
chat.realtime_conversation.thread_id = Some(ThreadId::new());
chat.toggle_realtime_microphone();
assert!(chat.realtime_conversation.microphone_muted);
assert!(events.try_recv().is_err());
chat.toggle_realtime_microphone();
assert!(!chat.realtime_conversation.microphone_muted);
assert!(events.try_recv().is_err());
}
#[tokio::test]
async fn voice_cannot_start_on_a_parent_owned_thread() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
chat.blocks_direct_input = true;
chat.toggle_realtime_conversation();
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Inactive
);
assert!(matches!(
events.try_recv(),
Ok(AppEvent::InsertHistoryCell(_))
));
}
#[tokio::test]
async fn voice_footer_renders_the_main_conversation_states() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
let mut states = Vec::new();
for (label, phase, muted, level, speaker_level, role, transcript) in [
(
"connecting",
RealtimeConversationPhase::Starting,
false,
0,
0,
None,
"",
),
(
"listening",
RealtimeConversationPhase::Active,
false,
0,
0,
None,
"",
),
(
"speaking",
RealtimeConversationPhase::Active,
false,
4,
5,
None,
"",
),
(
"muted",
RealtimeConversationPhase::Active,
true,
4,
4,
None,
"",
),
(
"transcript",
RealtimeConversationPhase::Active,
false,
2,
2,
Some("assistant"),
"Hello there",
),
] {
chat.realtime_conversation.phase = phase;
chat.realtime_conversation.microphone_muted = muted;
chat.realtime_conversation.microphone_level = level;
chat.realtime_conversation.speaker_level = speaker_level;
chat.realtime_conversation.transcript_role = role.map(str::to_string);
chat.realtime_conversation.transcript = transcript.to_string();
chat.update_realtime_footer();
states.push(format!(
"{label}:\n{}",
render_bottom_popup(&chat, /*width*/ 80)
));
}
insta::assert_snapshot!(states.join("\n\n"));
}
#[tokio::test]
async fn narrow_voice_footer_keeps_the_stop_control_before_meters() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
chat.realtime_conversation.speaker_active_until =
Some(std::time::Instant::now() + super::super::SPEAKER_ACTIVITY_HOLD);
chat.update_realtime_footer();
let footer = render_bottom_popup(&chat, /*width*/ 45);
assert!(footer.contains("voice ● speaking"));
assert!(footer.contains("/voice stop"));
chat.realtime_conversation.speaker_active_until = None;
chat.realtime_conversation.speaker_level = 1;
chat.on_realtime_transcript_delta("user".to_string(), "stop".to_string());
assert_eq!(chat.realtime_conversation.speaker_level, 0);
assert_eq!(chat.realtime_conversation.speaker_active_until, None);
chat.realtime_conversation.speaker_active_until =
Some(std::time::Instant::now() - super::super::SPEAKER_ACTIVITY_HOLD);
chat.update_realtime_footer();
assert!(render_bottom_popup(&chat, /*width*/ 45).contains("voice ● listening"));
for (peak, expected) in [
(0, 0),
(1, 0),
(511, 0),
(512, 0),
(513, 1),
(4096, 3),
(6144, 4),
(6656, 4),
(6657, 5),
(8192, 5),
(32768, 5),
(u16::MAX, 5),
] {
assert_eq!(super::super::audio_meter_level(peak), expected);
}
assert_eq!(super::super::audio_meter_bar(/*level*/ 0), "░░░░░");
assert_eq!(super::super::audio_meter_bar(/*level*/ 5), "█████");
}
#[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 (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();
assert_eq!(
(
chat.realtime_conversation.phase,
ops.try_recv().ok(),
observed_abort.is_aborted(),
),
(RealtimeConversationPhase::Inactive, None, true)
);
}
#[tokio::test]
async fn resetting_pending_voice_aborts_without_reporting_a_backend_session() {
let (mut chat, _sender, _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.thread_id = Some(ThreadId::new());
chat.realtime_conversation.startup_abort = Some(abort);
assert_eq!(
(
chat.reset_realtime_conversation(),
chat.realtime_conversation.phase,
observed_abort.is_aborted(),
),
(None, RealtimeConversationPhase::Inactive, true)
);
}
#[tokio::test]
async fn audio_failure_cancels_pending_voice_and_reports_the_device_error() {
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.on_realtime_error("speaker stream failed: device disconnected".to_string());
let Ok(AppEvent::InsertHistoryCell(cell)) = events.try_recv() else {
panic!("voice should report the speaker failure");
};
assert!(
cell.display_lines(/*width*/ 80)
.iter()
.any(|line| line.to_string().contains("device disconnected"))
);
assert_eq!(
(
chat.realtime_conversation.phase,
observed_abort.is_aborted()
),
(RealtimeConversationPhase::Inactive, true)
);
}
#[tokio::test]
async fn voice_waits_for_the_previous_session_to_close_before_restarting() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
chat.realtime_conversation.phase = RealtimeConversationPhase::Stopping;
chat.toggle_realtime_conversation();
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Stopping
);
chat.on_realtime_conversation_closed(/*reason*/ Some("requested".into()));
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Inactive
);
}
#[tokio::test]
async fn canceled_offer_is_ignored_after_a_new_start_attempt() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
chat.realtime_conversation.phase = RealtimeConversationPhase::Starting;
chat.realtime_conversation.attempt_id = 2;
chat.on_realtime_webrtc_offer_created(
ThreadId::new(),
/*attempt_id*/ 1,
Err("canceled offer failed".to_string()),
);
assert_eq!(
(
chat.realtime_conversation.phase,
chat.realtime_conversation.attempt_id,
events.try_recv().is_err()
),
(RealtimeConversationPhase::Starting, 2, true)
);
}
#[tokio::test]
async fn voice_becomes_active_only_after_backend_and_current_peer_are_ready() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
chat.realtime_conversation.phase = RealtimeConversationPhase::Starting;
chat.realtime_conversation.attempt_id = 2;
chat.on_realtime_webrtc_connected(/*attempt_id*/ 1, Ok(()));
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Starting
);
chat.on_realtime_webrtc_connected(/*attempt_id*/ 2, Ok(()));
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Starting
);
chat.on_realtime_conversation_started();
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Active
);
let Ok(AppEvent::InsertHistoryCell(cell)) = events.try_recv() else {
panic!("voice start should insert its history banner");
};
insta::assert_snapshot!(
"voice_start_banner",
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n")
);
}
#[tokio::test]
async fn normal_voice_close_renders_the_ended_message() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
while events.try_recv().is_ok() {}
chat.on_realtime_conversation_closed(Some("transport_closed".into()));
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
insta::assert_snapshot!("voice_ended_message", rendered);
}
#[tokio::test]
async fn voice_waits_for_peer_when_backend_starts_first() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
chat.realtime_conversation.phase = RealtimeConversationPhase::Starting;
chat.realtime_conversation.attempt_id = 2;
chat.on_realtime_conversation_started();
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Starting
);
chat.on_realtime_webrtc_connected(/*attempt_id*/ 2, Ok(()));
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Active
);
}
#[tokio::test]
async fn startup_retry_waits_for_closed_then_uses_a_fresh_attempt_and_preserves_mute() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
chat.realtime_conversation.phase = RealtimeConversationPhase::Starting;
chat.realtime_conversation.attempt_id = 0;
chat.realtime_conversation.microphone_muted = true;
chat.on_realtime_webrtc_connected(
/*attempt_id*/ 0,
Err(codex_realtime_webrtc::ConnectionError::NegotiationTimedOut),
);
assert!(
matches!(ops.try_recv().unwrap(), AppCommand::RealtimeConversationStop { thread_id: id } if id == thread_id)
);
assert_eq!(
(
chat.realtime_conversation.phase,
chat.realtime_conversation.startup_retry
),
(
RealtimeConversationPhase::Stopping,
super::super::StartupRetry::WaitingForStop
)
);
assert!(ops.try_recv().is_err());
let mut rendered = Vec::new();
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
rendered.extend(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string()),
);
}
}
insta::assert_snapshot!("voice_startup_retry", rendered.join("\n"));
chat.on_realtime_conversation_closed(Some("transport_closed".into()));
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Stopping
);
assert_eq!(chat.realtime_conversation.attempt_id, 0);
chat.on_realtime_conversation_closed(Some("requested".into()));
let new_attempt = chat.realtime_conversation.attempt_id;
assert_ne!(new_attempt, 0);
assert_eq!(
(
chat.realtime_conversation.phase,
chat.realtime_conversation.startup_retry,
chat.realtime_conversation.microphone_muted
),
(
RealtimeConversationPhase::Starting,
super::super::StartupRetry::Used,
true
)
);
// A completion from the first attempt cannot finish or fail the retry.
chat.on_realtime_webrtc_connected(
/*attempt_id*/ 0,
Err(codex_realtime_webrtc::ConnectionError::NegotiationTimedOut),
);
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Starting
);
chat.on_realtime_webrtc_connected(new_attempt, Ok(()));
chat.on_realtime_conversation_started();
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Active
);
chat.reset_realtime_conversation();
}
#[tokio::test]
async fn startup_retry_is_cancelled_while_waiting_for_backend_close() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
chat.realtime_conversation.phase = RealtimeConversationPhase::Starting;
chat.on_realtime_webrtc_connected(
/*attempt_id*/ 0,
Err(codex_realtime_webrtc::ConnectionError::NegotiationTimedOut),
);
chat.toggle_realtime_conversation();
chat.on_realtime_conversation_closed(Some("transport_closed".into()));
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Stopping
);
chat.on_realtime_conversation_closed(Some("requested".into()));
assert_eq!(
(
chat.realtime_conversation.phase,
chat.realtime_conversation.attempt_id
),
(RealtimeConversationPhase::Inactive, 0)
);
}
#[tokio::test]
async fn startup_transport_close_before_peer_timeout_retries_once_and_ignores_old_result() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
chat.realtime_conversation.phase = RealtimeConversationPhase::Starting;
chat.realtime_conversation.attempt_id = 0;
// The close can also overtake the backend's started notification.
chat.realtime_conversation.backend_started = false;
chat.realtime_conversation.microphone_muted = true;
chat.on_realtime_conversation_closed(Some("transport_closed".into()));
let retry_attempt = chat.realtime_conversation.attempt_id;
assert_ne!(retry_attempt, 0);
assert_eq!(
(
chat.realtime_conversation.phase,
chat.realtime_conversation.startup_retry,
chat.realtime_conversation.microphone_muted,
),
(
RealtimeConversationPhase::Starting,
super::super::StartupRetry::Used,
true,
)
);
assert!(
ops.try_recv().is_err(),
"closed backend needs no extra stop"
);
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
insta::assert_snapshot!("voice_startup_transport_close_retry", rendered);
chat.on_realtime_webrtc_connected(
/*attempt_id*/ 0,
Err(codex_realtime_webrtc::ConnectionError::NegotiationTimedOut),
);
assert_eq!(
(
chat.realtime_conversation.phase,
chat.realtime_conversation.attempt_id
),
(RealtimeConversationPhase::Starting, retry_attempt)
);
chat.on_realtime_conversation_started();
chat.on_realtime_webrtc_connected(retry_attempt, Ok(()));
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Active
);
chat.on_realtime_conversation_closed(Some("transport_closed".into()));
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Inactive
);
assert!(ops.try_recv().is_err(), "retry budget must be used");
}
#[tokio::test]
async fn startup_retry_never_retries_twice_or_retries_other_errors_or_active_sessions() {
for (phase, budget, error) in [
(
RealtimeConversationPhase::Starting,
super::super::StartupRetry::Used,
codex_realtime_webrtc::ConnectionError::NegotiationTimedOut,
),
(
RealtimeConversationPhase::Starting,
super::super::StartupRetry::Available,
codex_realtime_webrtc::ConnectionError::Failed,
),
(
RealtimeConversationPhase::Active,
super::super::StartupRetry::Available,
codex_realtime_webrtc::ConnectionError::NegotiationTimedOut,
),
] {
let (mut chat, _sender, _events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
chat.realtime_conversation.phase = phase;
chat.realtime_conversation.startup_retry = budget;
chat.on_realtime_webrtc_connected(/*attempt_id*/ 0, Err(error));
assert_ne!(
chat.realtime_conversation.startup_retry,
super::super::StartupRetry::WaitingForStop
);
if phase == RealtimeConversationPhase::Starting {
assert!(matches!(
ops.try_recv(),
Ok(AppCommand::RealtimeConversationStop { thread_id: stopped }) if stopped == thread_id
));
}
assert!(ops.try_recv().is_err(), "no retry may be queued");
}
}
@@ -0,0 +1,6 @@
---
source: tui/src/chatwidget/realtime_tests/caption_replay.rs
assertion_line: 1675
expression: "rendered.join(\"\\n\")"
---
• Captioned answer
@@ -0,0 +1,7 @@
---
source: tui/src/chatwidget/realtime_tests/caption_replay.rs
assertion_line: 2528
expression: "live.display_lines(80).into_iter().map(|line|\nline.to_string()).collect::<Vec<_>>().join(\"\\n\")"
---
› last
@@ -0,0 +1,12 @@
---
source: tui/src/chatwidget/realtime_tests/caption_replay.rs
assertion_line: 1586
expression: rendered
---
› Typed question
• Checking the typed request
• Typed reasoning summary
› spoken correction
@@ -0,0 +1,6 @@
---
source: tui/src/chatwidget/realtime_tests/handoff_privacy.rs
assertion_line: 1188
expression: rendered
---
• Typed beginning and typed tail
@@ -0,0 +1,6 @@
---
source: tui/src/chatwidget/realtime_tests/handoff_privacy.rs
assertion_line: 1046
expression: rendered
---
• typed answer
@@ -0,0 +1,6 @@
---
source: tui/src/chatwidget/realtime_tests/handoffs.rs
assertion_line: 918
expression: rendered
---
• [COMMENTARY] is a label in the documentation.
@@ -0,0 +1,6 @@
---
source: tui/src/chatwidget/realtime_tests/handoffs.rs
assertion_line: 514
expression: "rendered_history.join(\"\\n\")"
---
• Voice conversation started. Use /voice mute to mute or /voice to stop.
@@ -0,0 +1,6 @@
---
source: tui/src/chatwidget/realtime_tests/lifecycle.rs
assertion_line: 2510
expression: rendered
---
• Voice conversation ended: transport_closed
@@ -1,5 +1,5 @@
---
source: tui/src/chatwidget/realtime_tests.rs
source: tui/src/chatwidget/realtime_tests/lifecycle.rs
assertion_line: 176
expression: "states.join(\"\\n\\n\")"
---
@@ -0,0 +1,6 @@
---
source: tui/src/chatwidget/realtime_tests/lifecycle.rs
assertion_line: 1372
expression: "cell.display_lines(80).into_iter().map(|line|\nline.to_string()).collect::<Vec<_>>().join(\"\\n\")"
---
• Voice conversation started. Use /voice mute to mute or /voice to stop.
@@ -0,0 +1,6 @@
---
source: tui/src/chatwidget/realtime_tests/lifecycle.rs
assertion_line: 1425
expression: "rendered.join(\"\\n\")"
---
• Voice connection timed out. Retrying once after cleanup.
@@ -0,0 +1,6 @@
---
source: tui/src/chatwidget/realtime_tests/lifecycle.rs
assertion_line: 2670
expression: rendered
---
• Voice connection closed during startup. Retrying once.
@@ -0,0 +1,8 @@
---
source: tui/src/chatwidget/realtime_tests/transcripts.rs
assertion_line: 218
expression: "rendered.join(\"\\n\")"
---
• Hello there
› Can you hear me?
@@ -0,0 +1,9 @@
---
source: tui/src/chatwidget/realtime_tests/transcripts.rs
assertion_line: 397
expression: rendered
---
› Correction
• First answer
@@ -0,0 +1,9 @@
---
source: tui/src/chatwidget/realtime_tests/transcripts.rs
assertion_line: 446
expression: rendered
---
› Correction
• First answer
@@ -0,0 +1,8 @@
---
source: tui/src/chatwidget/realtime_tests/transcripts.rs
assertion_line: 500
expression: rendered
---
• First
• First and last
• Voice conversation ended: requested
@@ -0,0 +1,604 @@
//! Undelivered speech returns to visible history without leaks or clipping.
use super::*;
use pretty_assertions::assert_eq;
#[tokio::test]
async fn queued_voice_answers_return_to_history_once_if_voice_closes_before_delivery() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let mut deliveries = Vec::new();
for (turn_id, item_id, text) in [
("first-turn", "first-answer", "First answer"),
("second-turn", "second-answer", "Second answer"),
] {
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>question</input></realtime_delegation>"),
);
let answer = agent_item(item_id, text, Some(MessagePhase::FinalAnswer));
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = ops.try_recv().unwrap()
else {
panic!("completed voice turn should queue speech");
};
deliveries.push(delivery_id);
}
assert_eq!(chat.realtime_conversation.pending_speech.len(), 2);
while events.try_recv().is_ok() {}
chat.stop_realtime_conversation();
chat.reset_realtime_conversation();
for delivery_id in deliveries {
chat.restore_undelivered_realtime_speech(delivery_id);
}
let mut rendered = Vec::new();
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
rendered.push(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<String>(),
);
}
}
assert_eq!(
rendered
.iter()
.filter(|line| line.contains("First answer"))
.count(),
1
);
assert_eq!(
rendered
.iter()
.filter(|line| line.contains("Second answer"))
.count(),
1
);
}
#[tokio::test]
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 {
let turn_id = format!("turn-{index}");
start_item(
&mut chat,
thread_id,
&turn_id,
user_item("<realtime_delegation><input>question</input></realtime_delegation>"),
);
let answer = agent_item(
&format!("answer-{index}"),
&format!("Answer number {index}"),
Some(MessagePhase::FinalAnswer),
);
start_item(&mut chat, thread_id, &turn_id, answer.clone());
complete_item(&mut chat, thread_id, &turn_id, answer.clone());
if index == 0 {
finish_turn(
&mut chat,
thread_id,
&turn_id,
vec![answer],
TurnStatus::Completed,
);
let AppCommand::RealtimeConversationSpeech { delivery_id, .. } =
ops.try_recv().unwrap()
else {
panic!("first voice answer should queue speech");
};
oldest_delivery_id = Some(delivery_id);
}
}
assert_eq!(
chat.realtime_conversation.pending_speech.len(),
super::super::MAX_PENDING_SPEECH_DELIVERIES
);
assert!(!chat.has_pending_realtime_speech(oldest_delivery_id.unwrap()));
let mut restored = 0;
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event
&& cell
.display_lines(/*width*/ 80)
.iter()
.any(|line| line.to_string().contains("Answer number 0"))
{
restored += 1;
}
}
assert_eq!(restored, 1);
assert!(ops.try_recv().is_err());
}
#[tokio::test]
async fn delegated_analysis_never_enters_history_on_completion_overflow_or_stop() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "private-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>question</input></realtime_delegation>"),
);
let private_items = [
agent_item(
"analysis",
"[ANALYSIS] private thought",
/*phase*/ None,
),
agent_item(
"commentary",
"private commentary",
Some(MessagePhase::Commentary),
),
agent_item(
"oversized-analysis",
&format!(
"[ANALYSIS] {}",
"secret".repeat(super::super::MAX_PENDING_SPEECH_ITEM_BYTES)
),
/*phase*/ None,
),
];
for item in &private_items {
start_item(&mut chat, thread_id, turn_id, item.clone());
complete_item(&mut chat, thread_id, turn_id, item.clone());
}
assert!(chat.realtime_conversation.pending_speech.is_empty());
finish_turn(
&mut chat,
thread_id,
turn_id,
private_items.to_vec(),
TurnStatus::Completed,
);
for index in 0..=super::super::MAX_PENDING_SPEECH_DELIVERIES {
let turn_id = format!("answer-turn-{index}");
start_item(
&mut chat,
thread_id,
&turn_id,
user_item("<realtime_delegation><input>question</input></realtime_delegation>"),
);
let answer = agent_item(
&format!("answer-{index}"),
&format!("Public answer {index}"),
Some(MessagePhase::FinalAnswer),
);
start_item(&mut chat, thread_id, &turn_id, answer.clone());
complete_item(&mut chat, thread_id, &turn_id, answer);
}
chat.stop_realtime_conversation();
chat.reset_realtime_conversation();
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
let lines = cell.display_lines(/*width*/ 80);
assert!(lines.iter().all(|line| {
!line.to_string().contains("private") && !line.to_string().contains("secret")
}));
}
}
assert!(matches!(
ops.try_recv(),
Ok(AppCommand::RealtimeConversationStop { .. })
));
assert!(ops.try_recv().is_err(), "no private speech may be queued");
}
#[tokio::test]
async fn delegated_reasoning_never_enters_live_history() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "private-reasoning-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>question</input></realtime_delegation>"),
);
chat.config.show_raw_agent_reasoning = true;
for notification in [
ServerNotification::ReasoningSummaryPartAdded(ReasoningSummaryPartAddedNotification {
thread_id: thread_id.to_string(),
turn_id: turn_id.into(),
item_id: "reasoning".into(),
summary_index: 0,
}),
ServerNotification::ReasoningSummaryTextDelta(ReasoningSummaryTextDeltaNotification {
thread_id: thread_id.to_string(),
turn_id: turn_id.into(),
item_id: "reasoning".into(),
delta: "private summary".into(),
summary_index: 0,
}),
ServerNotification::ReasoningTextDelta(ReasoningTextDeltaNotification {
thread_id: thread_id.to_string(),
turn_id: turn_id.into(),
item_id: "reasoning".into(),
delta: "private raw reasoning".into(),
content_index: 0,
}),
] {
chat.handle_server_notification(notification, /*replay_kind*/ None);
}
complete_item(
&mut chat,
thread_id,
turn_id,
ThreadItem::Reasoning {
id: "reasoning".into(),
summary: vec!["private summary".into()],
content: vec!["private raw reasoning".into()],
},
);
chat.reset_realtime_conversation();
start_item(
&mut chat,
thread_id,
"late-start-turn",
user_item("<realtime_delegation><input>delayed input</input></realtime_delegation>"),
);
assert!(chat.is_realtime_delegated_reasoning_turn("late-start-turn"));
chat.handle_server_notification(
ServerNotification::ReasoningSummaryTextDelta(ReasoningSummaryTextDeltaNotification {
thread_id: thread_id.to_string(),
turn_id: turn_id.into(),
item_id: "late-reasoning".into(),
delta: "private late summary".into(),
summary_index: 0,
}),
/*replay_kind*/ None,
);
complete_item(
&mut chat,
thread_id,
turn_id,
ThreadItem::Reasoning {
id: "late-reasoning".into(),
summary: vec!["private late summary".into()],
content: Vec::new(),
},
);
chat.handle_server_notification(
ServerNotification::ReasoningSummaryTextDelta(ReasoningSummaryTextDeltaNotification {
thread_id: thread_id.to_string(),
turn_id: "late-start-turn".into(),
item_id: "late-start-reasoning".into(),
delta: "private delayed start".into(),
summary_index: 0,
}),
/*replay_kind*/ None,
);
complete_item(
&mut chat,
thread_id,
"late-start-turn",
ThreadItem::Reasoning {
id: "late-start-reasoning".into(),
summary: vec!["private delayed start".into()],
content: Vec::new(),
},
);
chat.handle_server_notification(
ServerNotification::AgentMessageDelta(AgentMessageDeltaNotification {
thread_id: thread_id.to_string(),
turn_id: turn_id.into(),
item_id: "late-commentary".into(),
delta: "private commentary stream".into(),
}),
/*replay_kind*/ None,
);
complete_item(
&mut chat,
thread_id,
turn_id,
agent_item(
"late-commentary",
"private commentary item",
Some(MessagePhase::Commentary),
),
);
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![agent_item(
"private-none",
"[ANALYSIS] private fallback",
/*phase*/ None,
)],
TurnStatus::Completed,
);
chat.flush_answer_stream_with_separator();
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
let rendered = cell
.transcript_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n");
assert!(!rendered.contains("private"), "{rendered}");
}
}
assert!(!chat.is_realtime_delegated_reasoning_turn(turn_id));
chat.handle_server_notification(
ServerNotification::ReasoningSummaryTextDelta(ReasoningSummaryTextDeltaNotification {
thread_id: thread_id.to_string(),
turn_id: "typed-turn".into(),
item_id: "typed-reasoning".into(),
delta: "ordinary summary".into(),
summary_index: 0,
}),
/*replay_kind*/ None,
);
complete_item(
&mut chat,
thread_id,
"typed-turn",
ThreadItem::Reasoning {
id: "typed-reasoning".into(),
summary: vec!["ordinary summary".into()],
content: Vec::new(),
},
);
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.transcript_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<String>();
assert!(rendered.contains("ordinary summary"), "{rendered}");
}
#[tokio::test]
async fn answer_exceeding_speech_budget_is_shown_in_full_instead() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "text-only-final-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>question</input></realtime_delegation>"),
);
while events.try_recv().is_ok() {}
let text = format!("Start. {} End.", "word ".repeat(/*n*/ 800));
let answer = agent_item("text-only-answer", &text, Some(MessagePhase::FinalAnswer));
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
assert!(
ops.try_recv().is_err(),
"an over-budget answer must not be clipped for speech"
);
assert!(chat.realtime_conversation.pending_speech.is_empty());
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.transcript_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
assert_eq!(
without_completion_metadata(&rendered)
.split_whitespace()
.collect::<Vec<_>>()
.join(" "),
format!("• {text}")
.split_whitespace()
.collect::<Vec<_>>()
.join(" ")
);
}
#[tokio::test]
async fn oversized_delegated_answer_remains_in_normal_history() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "large-final-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>question</input></realtime_delegation>"),
);
let answer = agent_item(
"large-answer",
&"word ".repeat(super::super::MAX_PENDING_SPEECH_ITEM_BYTES / 5),
Some(MessagePhase::FinalAnswer),
);
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
assert!(chat.realtime_conversation.pending_speech.is_empty());
assert!(ops.try_recv().is_err());
assert!(events.try_recv().is_ok());
assert_eq!(
chat.transcript.last_completed_agent_message,
Some((turn_id.to_string(), "large-answer".to_string()))
);
}
#[tokio::test]
async fn final_item_that_grows_after_completion_is_not_queued_without_recovery() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "expanded-final-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>question</input></realtime_delegation>"),
);
let short = agent_item("expanded-answer", "short", Some(MessagePhase::FinalAnswer));
start_item(&mut chat, thread_id, turn_id, short.clone());
complete_item(&mut chat, thread_id, turn_id, short);
let expanded_text = "word ".repeat(super::super::MAX_PENDING_SPEECH_ITEM_BYTES / 5);
let expanded = agent_item(
"expanded-answer",
&expanded_text,
Some(MessagePhase::FinalAnswer),
);
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![expanded],
TurnStatus::Completed,
);
assert!(chat.realtime_conversation.pending_speech.is_empty());
assert!(ops.try_recv().is_err());
let recovered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(cell.raw_lines()),
_ => None,
})
.flatten()
.map(|line| line.to_string())
.filter(|text| text == "short" || text.starts_with("word "))
.collect::<Vec<_>>();
assert_eq!(recovered, vec![expanded_text.trim_end().to_string()]);
}
#[tokio::test]
async fn interrupted_voice_delegation_never_speaks_a_completed_item() {
let (mut chat, _sender, _events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let turn_id = "interrupted-turn";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>stop me</input></realtime_delegation>"),
);
let answer = agent_item("answer", "Never say this", Some(MessagePhase::FinalAnswer));
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Interrupted,
);
assert!(ops.try_recv().is_err());
}
#[tokio::test]
async fn replay_projects_the_delegated_request_without_internal_xml() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
chat.replay_thread_item(
user_item(
"<realtime_delegation><input>show &lt;code&gt; &amp; tests</input></realtime_delegation>",
),
"old-turn".to_string(),
ReplayKind::ResumeInitialMessages,
);
let Ok(AppEvent::InsertHistoryCell(cell)) = events.try_recv() else {
panic!("a resumed voice delegation should preserve the original user request");
};
let rendered = cell
.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n");
assert!(rendered.contains("show <code> & tests"));
assert!(!rendered.contains("realtime_delegation"));
assert!(ops.try_recv().is_err());
}
#[tokio::test]
async fn locally_typed_delegation_shaped_text_keeps_its_normal_answer() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
let typed_text = "<realtime_delegation><input>explain this xml</input></realtime_delegation>";
chat.note_realtime_typed_input(typed_text);
let turn_id = "typed-xml-turn";
let prompt = user_item(typed_text);
start_item(&mut chat, thread_id, turn_id, prompt.clone());
complete_item(&mut chat, thread_id, turn_id, prompt);
let answer = agent_item("xml-answer", "It is an XML element.", /*phase*/ None);
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
let mut saw_answer = false;
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
saw_answer |= cell
.display_lines(/*width*/ 80)
.iter()
.any(|line| line.to_string().contains("It is an XML element."));
}
}
assert!(saw_answer);
assert!(ops.try_recv().is_err());
}
#[tokio::test]
async fn new_user_input_invalidates_already_queued_voice_speech() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
chat.realtime_conversation.attempt_id = 4;
let generation = chat.realtime_conversation.input_generation;
assert!(chat.is_current_realtime_attempt(thread_id, /*attempt_id*/ 4, generation));
chat.note_realtime_typed_input("newer typed message");
assert!(!chat.is_current_realtime_attempt(thread_id, /*attempt_id*/ 4, generation));
chat.on_realtime_transcript_done("user".to_string(), "newer spoken message".to_string());
assert!(!chat.is_current_realtime_attempt(thread_id, /*attempt_id*/ 4, generation));
}
@@ -0,0 +1,696 @@
//! Partial and completed captions stay ordered and bounded through voice shutdown.
use super::*;
use pretty_assertions::assert_eq;
fn seed_interleaved_partials(chat: &mut ChatWidget) {
chat.on_realtime_transcript_delta("assistant".into(), "First ".into());
chat.on_realtime_transcript_delta("user".into(), "Correction".into());
chat.on_realtime_transcript_delta("assistant".into(), "answer".into());
}
#[tokio::test]
async fn interleaved_transcripts_keep_interrupted_speech_suppressed() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
chat.on_realtime_transcript_delta("assistant".into(), "The long answer".into());
chat.on_realtime_transcript_delta("user".into(), "Actually, ".into());
let generation = chat.realtime_conversation.input_generation;
chat.on_realtime_transcript_delta("assistant".into(), " keeps going".into());
assert_eq!(
chat.realtime_conversation.speaker_suppression_generation,
Some(generation)
);
chat.on_realtime_transcript_done("user".into(), "Actually, make it short".into());
chat.on_realtime_transcript_delta("assistant".into(), " and going".into());
chat.on_realtime_transcript_done("assistant".into(), "The long answer keeps going".into());
assert_eq!(
chat.realtime_conversation.speaker_suppression_generation,
Some(generation)
);
chat.on_realtime_transcript_delta("assistant".into(), "Short answer".into());
assert!(
chat.realtime_conversation
.speaker_suppression_generation
.is_none()
);
}
#[tokio::test]
async fn delayed_user_transcript_suppresses_old_audio_after_microphone_mute() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
chat.on_realtime_transcript_delta("assistant".into(), "Old answer".into());
chat.realtime_conversation.microphone_muted = true;
chat.on_realtime_transcript_delta("user".into(), "Actually".into());
assert_eq!(
(
chat.realtime_conversation.microphone_muted,
chat.realtime_conversation.speaker_suppression_generation,
),
(true, Some(chat.realtime_conversation.input_generation))
);
}
#[tokio::test]
async fn interleaved_transcripts_do_not_discard_the_interrupting_request() {
let (mut chat, _sender, _events, mut ops) = make_chatwidget_manual_with_sender().await;
let thread_id = activate_voice(&mut chat);
chat.on_realtime_transcript_delta("assistant".into(), "The long answer".into());
chat.on_realtime_transcript_delta("user".into(), "Actually, ".into());
let generation = chat.realtime_conversation.input_generation;
let turn_id = "interruption";
start_item(
&mut chat,
thread_id,
turn_id,
user_item("<realtime_delegation><input>make it short</input></realtime_delegation>"),
);
chat.on_realtime_transcript_delta("assistant".into(), " keeps going".into());
chat.on_realtime_transcript_delta("user".into(), "make it short".into());
chat.on_realtime_transcript_done("assistant".into(), "The long answer keeps going".into());
chat.on_realtime_transcript_done("user".into(), "Actually, make it short".into());
let answer = agent_item(
"short-answer",
"Short answer",
Some(MessagePhase::FinalAnswer),
);
start_item(&mut chat, thread_id, turn_id, answer.clone());
complete_item(&mut chat, thread_id, turn_id, answer.clone());
finish_turn(
&mut chat,
thread_id,
turn_id,
vec![answer],
TurnStatus::Completed,
);
assert!(matches!(
ops.try_recv(),
Ok(AppCommand::RealtimeConversationSpeech { input_generation, text, .. })
if input_generation == generation && text.as_str() == "Short answer"
));
assert!(ops.try_recv().is_err());
}
#[tokio::test]
async fn completed_user_transcript_keeps_the_old_assistant_turn_suppressed() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
chat.on_realtime_transcript_delta("assistant".into(), "Old answer".into());
chat.on_realtime_transcript_done("user".into(), "Stop, I have a correction".into());
let generation = chat.realtime_conversation.input_generation;
chat.on_realtime_transcript_done("assistant".into(), "Old answer".into());
assert_eq!(
chat.realtime_conversation.speaker_suppression_generation,
Some(generation)
);
chat.on_realtime_transcript_delta("assistant".into(), "Go".into());
chat.on_realtime_transcript_done("assistant".into(), "Go ahead".into());
assert!(
chat.realtime_conversation
.speaker_suppression_generation
.is_none()
);
}
#[tokio::test]
async fn final_only_assistant_transcript_cannot_release_an_interruption() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
chat.on_realtime_transcript_delta("user".into(), "Stop".into());
let generation = chat.realtime_conversation.input_generation;
chat.on_realtime_transcript_done("user".into(), "Stop".into());
chat.on_realtime_transcript_done("assistant".into(), "Late old answer".into());
assert_eq!(
chat.realtime_conversation.speaker_suppression_generation,
Some(generation)
);
}
#[tokio::test]
async fn transcript_deltas_track_the_current_speaker() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
chat.realtime_conversation.phase = RealtimeConversationPhase::Active;
let owner = |chat: &ChatWidget| chat.realtime_conversation.speaker_suppression_generation;
chat.on_realtime_transcript_delta("user".to_string(), "Hello ".to_string());
chat.on_realtime_transcript_delta("user".to_string(), "there".to_string());
assert_eq!(
(
chat.realtime_conversation.transcript_role.as_deref(),
chat.realtime_conversation.transcript.as_str()
),
(Some("user"), "Hello there")
);
chat.on_realtime_transcript_done("assistant".to_string(), "Old".to_string());
assert_eq!(
chat.realtime_conversation.speaker_suppression_generation,
Some(chat.realtime_conversation.input_generation)
);
chat.on_realtime_transcript_done("user".to_string(), "Hello there".to_string());
chat.on_realtime_transcript_delta("assistant".to_string(), "Hi".to_string());
assert!(owner(&chat).is_none());
chat.on_realtime_transcript_done("assistant".to_string(), "Hi".to_string());
chat.on_realtime_transcript_delta("user".to_string(), "Again".to_string());
chat.on_realtime_transcript_done("user".to_string(), "Again".to_string());
assert!(owner(&chat).is_some());
chat.on_realtime_transcript_delta("assistant".to_string(), "".to_string());
assert!(owner(&chat).is_some());
chat.on_realtime_transcript_delta("assistant".to_string(), "Hi".to_string());
assert!(owner(&chat).is_none());
assert_eq!(
(
chat.realtime_conversation.transcript_role.as_deref(),
chat.realtime_conversation.transcript.as_str()
),
(Some("assistant"), "Hi")
);
chat.on_realtime_transcript_done("user".to_string(), "Again".to_string());
let generation = chat.realtime_conversation.input_generation;
assert_eq!(owner(&chat), Some(generation));
chat.on_realtime_transcript_done("user".to_string(), " ".to_string());
assert_eq!(owner(&chat), Some(generation));
chat.on_realtime_transcript_done("assistant".to_string(), "".to_string());
assert_eq!(owner(&chat), Some(generation));
chat.on_realtime_transcript_done("assistant".to_string(), "Hi".to_string());
assert_eq!(owner(&chat), Some(generation));
chat.on_realtime_transcript_delta("user".to_string(), "noise".to_string());
chat.on_realtime_transcript_done("user".to_string(), "".to_string());
assert!(owner(&chat).is_none());
}
#[tokio::test]
async fn voice_transcripts_stream_in_the_conversation_instead_of_the_footer() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
chat.config.animations = false;
activate_voice(&mut chat);
chat.update_realtime_footer();
chat.on_realtime_transcript_delta("user".to_string(), "pick a ".to_string());
chat.on_realtime_transcript_delta("user".to_string(), "number".to_string());
let live = chat
.active_cell_transcript_lines(/*width*/ 80)
.unwrap_or_default()
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n");
assert!(live.contains("pick a number"));
let footer = render_bottom_popup(&chat, /*width*/ 80);
assert!(footer.contains("voice ● listening"));
let status_line = footer
.lines()
.find(|line| line.contains("voice ● listening"))
.unwrap_or_default();
assert!(!status_line.contains("pick a number"));
assert!(events.try_recv().is_err());
chat.on_realtime_transcript_done("user".to_string(), "pick a number".to_string());
assert!(chat.active_cell_transcript_lines(/*width*/ 80).is_none());
let Ok(AppEvent::InsertHistoryCell(cell)) = events.try_recv() else {
panic!("the completed voice transcript should become one normal user turn");
};
assert!(
cell.display_lines(/*width*/ 80)
.iter()
.any(|line| line.to_string().contains("pick a number"))
);
assert!(events.try_recv().is_err());
}
#[tokio::test]
async fn unexpected_voice_close_preserves_partial_transcripts_once() {
for (role, text) in [
("user", "Part of my request"),
("assistant", "Part of the answer"),
] {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
chat.on_realtime_transcript_delta(role.to_string(), text.to_string());
chat.on_realtime_conversation_closed(/*reason*/ None);
chat.on_realtime_transcript_done(role.to_string(), text.to_string());
let mut appearances = 0;
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
appearances +=
cell.display_lines(/*width*/ 80)
.iter()
.any(|line| line.to_string().contains(text)) as usize;
}
}
assert_eq!(appearances, 1, "{role} transcript should be preserved once");
assert!(chat.active_cell_transcript_lines(/*width*/ 80).is_none());
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Inactive
);
}
}
#[tokio::test]
async fn interleaved_partial_transcripts_survive_voice_close() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
seed_interleaved_partials(&mut chat);
chat.on_realtime_conversation_closed(/*reason*/ None);
chat.on_realtime_transcript_done("user".into(), "Correction".into());
chat.on_realtime_transcript_done("assistant".into(), "First answer".into());
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
assert_eq!(rendered.matches("Correction").count(), 1);
assert_eq!(rendered.matches("First answer").count(), 1);
insta::assert_snapshot!("interleaved_partials_on_close", rendered);
}
#[tokio::test]
async fn interleaved_partials_transfer_once_on_thread_switch() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
seed_interleaved_partials(&mut chat);
let cells = chat.take_realtime_transcript_cells_for_replay();
let rendered = cells
.iter()
.map(|record| record.text.as_str())
.collect::<String>();
assert_eq!(cells.len(), 2);
assert_eq!(rendered.matches("Correction").count(), 1);
assert_eq!(rendered.matches("First answer").count(), 1);
assert!(chat.take_realtime_transcript_cells_for_replay().is_empty());
}
#[tokio::test]
async fn intentional_stop_preserves_both_interleaved_partials_once() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
seed_interleaved_partials(&mut chat);
chat.stop_realtime_conversation();
chat.on_realtime_conversation_closed(/*reason*/ None);
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
assert_eq!(rendered.matches("Correction").count(), 1);
assert_eq!(rendered.matches("First answer").count(), 1);
insta::assert_snapshot!("interleaved_partials_on_stop", rendered);
}
#[tokio::test]
async fn stopping_voice_keeps_the_complete_late_final_caption() {
let (mut chat, _sender, mut events, mut ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
chat.on_realtime_transcript_delta("assistant".into(), "First ".into());
chat.finish_realtime_partial_transcripts();
chat.realtime_conversation.phase = RealtimeConversationPhase::Stopping;
chat.on_realtime_transcript_delta("assistant".into(), "and last".into());
chat.on_realtime_transcript_done("assistant".into(), "First and last".into());
chat.on_realtime_conversation_closed(Some("requested".into()));
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
assert_eq!(rendered.matches("First").count(), 2);
assert_eq!(rendered.matches("and last").count(), 1);
assert!(ops.try_recv().is_err());
insta::assert_snapshot!("voice_stop_late_transcript", rendered);
}
#[tokio::test]
async fn separate_late_finals_with_a_shared_prefix_keep_both_full_captions() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
chat.realtime_conversation.phase = RealtimeConversationPhase::Stopping;
chat.on_realtime_transcript_done("assistant".into(), "Hello".into());
chat.on_realtime_transcript_done("assistant".into(), "Hello again".into());
let captions = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(captions, vec!["• Hello", "• Hello again"]);
}
#[tokio::test]
async fn direct_reset_preserves_both_partial_speakers_for_replay() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
chat.on_realtime_transcript_delta("assistant".into(), "First ".into());
chat.on_realtime_transcript_delta("user".into(), "Correction".into());
chat.on_realtime_transcript_delta("assistant".into(), "answer".into());
chat.reset_realtime_conversation();
let records = chat.take_realtime_transcript_cells_for_replay();
assert_eq!(
records
.into_iter()
.map(|record| (record.role, record.text))
.collect::<Vec<_>>(),
vec![
("user".to_string(), "Correction".to_string()),
("assistant".to_string(), "First answer".to_string())
]
);
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<String>(),
),
_ => None,
})
.collect::<String>();
assert_eq!(rendered.matches("Correction").count(), 1);
assert_eq!(rendered.matches("First answer").count(), 1);
}
#[tokio::test]
async fn stopping_voice_preserves_the_live_transcript_once() {
for command in ["/voice", "/voice stop"] {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
chat.config.animations = true;
activate_voice(&mut chat);
chat.on_realtime_transcript_done("user".to_string(), "Earlier question".to_string());
chat.on_realtime_transcript_done("assistant".to_string(), "Earlier answer".to_string());
chat.on_realtime_transcript_delta("assistant".to_string(), "Answer in ".to_string());
chat.on_realtime_transcript_delta("assistant".to_string(), "progress".to_string());
if command == "/voice" {
chat.handle_slash_command_dispatch(crate::slash_command::SlashCommand::Voice);
} else {
chat.handle_slash_command_with_args_dispatch(
crate::slash_command::SlashCommand::Voice,
"stop".to_string(),
Vec::new(),
);
}
chat.on_realtime_transcript_done("assistant".to_string(), "Answer in progress".to_string());
chat.on_realtime_conversation_closed(/*reason*/ None);
chat.stop_realtime_conversation();
assert_eq!(
chat.realtime_conversation.phase,
RealtimeConversationPhase::Inactive
);
assert!(chat.active_cell_transcript_key().is_none());
let mut rendered = Vec::new();
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
assert!(cell.transcript_animation_tick().is_none());
rendered.extend(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string()),
);
}
}
insta::allow_duplicates! {
insta::assert_snapshot!(rendered.join("\n"), @r"
› Earlier question
• Earlier answer
• Answer in progress
");
}
}
}
#[tokio::test]
async fn transcript_completion_waits_for_normal_agent_stream_consolidation() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
chat.on_agent_message_delta("normal typed output".to_string());
chat.on_realtime_transcript_done("assistant".to_string(), "voice output".to_string());
chat.on_realtime_transcript_delta(
"assistant".to_string(),
"unfinished voice output".to_string(),
);
chat.stop_realtime_conversation();
assert_eq!(chat.realtime_conversation.pending_history_cells.len(), 2);
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
assert!(
cell.display_lines(/*width*/ 80)
.iter()
.all(|line| !line.to_string().contains("voice output"))
);
}
}
chat.finalize_completed_assistant_message(Some("normal typed output"));
chat.note_stream_consolidation_completed();
chat.flush_realtime_transcript_history();
let mut rendered = Vec::new();
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
rendered.extend(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string()),
);
}
}
assert_eq!(
rendered,
[
"• normal typed output",
"• voice output",
"• unfinished voice output"
]
);
}
#[tokio::test]
async fn transcript_handoff_moves_deferred_repeats_and_partial_once() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
chat.on_agent_message_delta("ordinary stream".to_string());
for _ in 0..super::super::MAX_PENDING_TRANSCRIPT_CELLS {
chat.on_realtime_transcript_done("assistant".to_string(), "same words".to_string());
}
chat.on_realtime_transcript_delta("assistant".to_string(), "partial words".to_string());
assert_eq!(
chat.realtime_conversation.pending_history_cells.len(),
super::super::MAX_PENDING_TRANSCRIPT_CELLS
);
let cells = chat.take_realtime_transcript_cells_for_replay();
assert_eq!(cells.len(), super::super::MAX_PENDING_TRANSCRIPT_CELLS + 1);
assert!(chat.take_realtime_transcript_cells_for_replay().is_empty());
chat.restore_realtime_transcript_cells(cells);
chat.finalize_completed_assistant_message(Some("ordinary stream"));
chat.note_stream_consolidation_completed();
chat.flush_realtime_transcript_history();
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.transcript_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<String>(),
),
_ => None,
})
.collect::<String>();
assert_eq!(
rendered.matches("same words").count(),
super::super::MAX_PENDING_TRANSCRIPT_CELLS
);
assert_eq!(rendered.matches("partial words").count(), 0);
let live = chat
.realtime_conversation
.live_transcript_cell
.as_ref()
.unwrap();
assert!(
live.display_lines(/*width*/ 80)
.iter()
.any(|line| line.to_string().contains("partial words"))
);
}
#[tokio::test]
async fn dropped_deferred_captions_do_not_return_on_thread_replacement() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
activate_voice(&mut chat);
chat.on_agent_message_delta("ordinary stream".to_string());
for index in 0..(super::super::MAX_PENDING_TRANSCRIPT_CELLS + 2) {
chat.on_realtime_transcript_done("assistant".into(), format!("caption {index}"));
}
assert_eq!(
chat.realtime_conversation.pending_history_cells.len(),
super::super::MAX_PENDING_TRANSCRIPT_CELLS
);
let records = chat.take_realtime_transcript_cells_for_replay();
assert_eq!(records.len(), super::super::MAX_PENDING_TRANSCRIPT_CELLS);
assert_eq!(
records.front().map(|record| record.text.as_str()),
Some("caption 2")
);
assert_eq!(
records.back().map(|record| record.text.as_str()),
Some("caption 33")
);
}
#[tokio::test]
async fn transcripts_are_preserved_while_the_peer_is_connecting() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
chat.realtime_conversation.phase = RealtimeConversationPhase::Starting;
chat.realtime_conversation.backend_started = true;
chat.on_realtime_transcript_delta("assistant".to_string(), "Hello".to_string());
assert_eq!(
(
chat.realtime_conversation.transcript_role.as_deref(),
chat.realtime_conversation.transcript.as_str(),
),
(Some("assistant"), "Hello")
);
chat.on_realtime_transcript_done("assistant".to_string(), "Hello there".to_string());
let Ok(AppEvent::InsertHistoryCell(cell)) = events.try_recv() else {
panic!("voice should preserve transcripts received before the peer connects");
};
assert!(
cell.display_lines(/*width*/ 80)
.iter()
.any(|line| line.to_string().contains("Hello there"))
);
assert_eq!(
(
chat.realtime_conversation.phase,
chat.realtime_conversation.transcript_role.as_deref(),
chat.realtime_conversation.transcript.as_str(),
),
(RealtimeConversationPhase::Starting, None, "")
);
}
#[tokio::test]
async fn transcript_deltas_keep_a_bounded_utf8_suffix() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
chat.realtime_conversation.phase = RealtimeConversationPhase::Active;
chat.on_realtime_transcript_delta("user".to_string(), "é".repeat(600));
chat.on_realtime_transcript_delta("user".to_string(), "done".to_string());
assert_eq!(
chat.realtime_conversation.transcript,
format!("{}done", "é".repeat(510))
);
}
#[tokio::test]
async fn completed_transcript_is_added_to_history_and_clears_the_caption() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
chat.realtime_conversation.phase = RealtimeConversationPhase::Active;
chat.on_realtime_transcript_delta("assistant".to_string(), "Hello".to_string());
chat.on_realtime_transcript_done("assistant".to_string(), "Hello there".to_string());
chat.on_realtime_transcript_done("user".to_string(), "Can you hear me?".to_string());
let mut rendered = Vec::new();
while let Ok(event) = events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
rendered.extend(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string()),
);
}
}
assert_eq!(
(
chat.realtime_conversation.transcript_role.as_deref(),
chat.realtime_conversation.transcript.as_str(),
),
(None, "")
);
insta::assert_snapshot!(rendered.join("\n"));
}
#[tokio::test]
async fn completed_transcript_preserves_the_other_speakers_caption() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
chat.realtime_conversation.phase = RealtimeConversationPhase::Active;
chat.on_realtime_transcript_delta("assistant".to_string(), "Still speaking".to_string());
chat.on_realtime_transcript_done("user".to_string(), "Interruption".to_string());
assert_eq!(
(
chat.realtime_conversation.transcript_role.as_deref(),
chat.realtime_conversation.transcript.as_str(),
),
(Some("assistant"), "Still speaking")
);
}