diff --git a/src-tauri/crates/berd-call/PROTOCOL.md b/src-tauri/crates/berd-call/PROTOCOL.md index d38454cab..66809f1f9 100644 --- a/src-tauri/crates/berd-call/PROTOCOL.md +++ b/src-tauri/crates/berd-call/PROTOCOL.md @@ -301,6 +301,22 @@ The child emits: {"type":"live_event","token":u64,"text":string,"origin"?:"user"|"spokesperson"|"handoff"} ``` +On macOS, `recognition_pending.active` reports the agent speech admission hold, not +the count of unresolved native recognitions. After local speech ends the hold +expires after five seconds without a final, or, while Apple's latest result is +interim words that no final has followed, after up to ten more seconds. Noise with no +interim words keeps the five seconds. Empty finals settle their audio and resolve +their turns without emitting a `live_event`. Expiry emits +`recognition_pending.active:false` without cancelling recognition, clearing audio, +or changing the recognizer generation, so a late final is still forwarded as a +`live_event`. New speech reasserts the hold. Native finals carry no utterance +identity, so each final carries the end of its audio and resolves the ended speech +turns that audio covers, oldest first, which may include expired ones. A final for +an earlier phrase of a turn still being spoken resolves nothing, and a final with no +end resolves the oldest turn. Credit accounting is capped at 1024; at the cap attribution stays ambiguous +until the recognizer resets, and finals still forward but only the timeout +releases the hold. + For every final live-side event, the child allocates a strictly increasing token, stores it in `SessionCore`, acknowledges the runtime storage receipt, and emits `live_event`. Conventional sessions omit `origin`, which means `user`. @@ -308,6 +324,18 @@ Expert-Spokesperson sessions use `origin` to distinguish user transcripts, Spokesperson transcripts, and handoffs. User finals interrupt reserved or playing assistant output only after storage and emission. Final text is at most 64 KiB. +Conventional sessions on macOS also emit the user's words while they are still +being transcribed: + +```text +{"type":"user_interim","token":u64,"text":string} +``` + +Each `user_interim` replaces the previous one, and empty text clears it. +`token` is the token the next user final will take, so a reply composed against +interim words is still current once they finalize. The message is never stored, +never allocates a token, and never interrupts output. + Expert-Spokesperson sessions also emit two previews for hosts that show the conversation as it happens: @@ -355,7 +383,7 @@ ID and speech ID when it is ready to accept output: Before emitting `accepted`, the child installs the admitted speech's leased assistant-activity guard. `suppress_input` stops PCM admission so user input cannot interrupt the speech; `allow_barge_in` continues PCM admission with the -assistant-sensitive VAD threshold. `accepted` transfers output authority. The +same VAD threshold. `accepted` transfers output authority. The child then writes Begin and waits for `audio_begin_accepted` before delivering PCM. Readiness is bounded to two seconds; expiry emits `speech_failed` with zero output. With `allow_barge_in`, speaking or recognition pending provisionally suspends started output. If both clear without a final, the same speech and player resume. A final, targeted cancellation, host mute/reset while provisionally suspended, pause, or shutdown discards the hold and interrupts exactly once. Accepted output owns exactly one assistant-activity guard. The child removes diff --git a/src-tauri/crates/berd-call/README.md b/src-tauri/crates/berd-call/README.md index 819d49db5..6919b2122 100644 --- a/src-tauri/crates/berd-call/README.md +++ b/src-tauri/crates/berd-call/README.md @@ -334,7 +334,7 @@ updates keep the prior configuration, and private credentials, endpoints, and bundle paths never enter the snapshot. The session also projects a separately revisioned `input_during_tts` snapshot. -`allow_barge_in` keeps PCM flowing through assistant-sensitive VAD, while +`allow_barge_in` keeps PCM flowing through the same VAD threshold, while `suppress_input` drops PCM at the shared runtime for the admitted utterance. Live policy updates apply to the next admission; host mute remains an independent reason, so neither state can accidentally clear the other. @@ -432,7 +432,7 @@ CC BY 4.0, as reflected by the crate's aggregate package-license metadata. Input is paced in real time as exact 960-sample frames every 20 ms. Each clip has one second of leading silence and 6.5 seconds of trailing silence. The long tail deliberately keeps continuous recognizers supplied with capture-like PCM -through VAD settlement and the runtime's five-second live no-result bound; it +through VAD settlement and the runtime's five-second pending-utterance timeout; it is included in the reported workload. A final transcript is validated and stored in its per-utterance result before its storage receipt is acknowledged, and the next clip does not begin until authoritative speaking and @@ -462,3 +462,5 @@ Auto is the default input-during-speech preference. It mutes microphone input du ### Input activity in stdout streams `start --stream` (including `--stream live`) emits ephemeral `0\tinput_activity\t{"speaking":true,"recognizing":false}` rows when input speech or recognition state changes. Both booleans are snapshots; show activity while either is true. These rows have cursor zero, are not utterances, and must never advance a speech cursor, trigger an agent turn, or enter transcript history. They are not delivered to Codex tasks. Clear client activity on session restart, stream end, or failure. Older CLIs omit these rows. + +On macOS, conventional sessions with `--stream` also emit ephemeral `0\tuser_interim\t{"cursor":12,"text":"you should"}` rows while your words are still being transcribed. Each row replaces the previous one; empty text clears it, as do mute, input reset and session restart. `cursor` is the cursor the next `user` row will carry, so a client can show the words as a caption or let an agent start on them and keep its reply current when the final arrives. The final `user` row stays authoritative. Like `input_activity`, these rows never advance a speech cursor or enter transcript history, and older CLIs omit them. diff --git a/src-tauri/crates/berd-call/src/benchmark/stt.rs b/src-tauri/crates/berd-call/src/benchmark/stt.rs index f9cb82bf0..1f4c7de50 100644 --- a/src-tauri/crates/berd-call/src/benchmark/stt.rs +++ b/src-tauri/crates/berd-call/src/benchmark/stt.rs @@ -563,6 +563,7 @@ fn observe_event( tracker.speaking_ended = Some(now); } } + Some(VoiceInputEvent::InterimTranscript(_)) => {} Some(VoiceInputEvent::RecognitionPendingChanged(active)) => { tracker.pending = active; if active { diff --git a/src-tauri/crates/berd-call/src/host_session.rs b/src-tauri/crates/berd-call/src/host_session.rs index 136690ac5..972f4acc3 100644 --- a/src-tauri/crates/berd-call/src/host_session.rs +++ b/src-tauri/crates/berd-call/src/host_session.rs @@ -197,6 +197,13 @@ impl Transcript { } } +fn user_interim_line(cursor: u64, text: &str) -> String { + format!( + "0\tuser_interim\t{}", + json!({"cursor": cursor, "text": text}) + ) +} + fn input_activity_line(speaking: bool, recognizing: bool) -> String { format!( "0\tinput_activity\t{}", @@ -978,6 +985,7 @@ struct SessionActor { saved: Option<(std::path::PathBuf, crate::saved_settings::SavedSettings)>, input_speaking: bool, recognition_pending: bool, + interim_shown: bool, pending_polls: Vec, writer: Arc>, events: Receiver, @@ -1082,6 +1090,7 @@ impl SessionActor { saved: None, input_speaking: false, recognition_pending: false, + interim_shown: false, pending_polls: Vec::new(), writer, events, @@ -1586,11 +1595,17 @@ impl SessionActor { SessionMessage::InputMuteApplied { id, active } => { // Commit before acknowledging so a restart cannot read a stale value. self.muted.store(active, Ordering::SeqCst); + if active { + self.stream_user_interim(0, "")?; + } self.stream_input_activity()?; if let Some(response) = self.pending_settings.remove(&id) { let _ = response.send(Ok(json!({"muted":active}))); } } + SessionMessage::UserInterim { token, text } if !self.expert_spokesperson => { + self.stream_user_interim(token, &text)? + } SessionMessage::LiveEvent { token, text, @@ -1730,9 +1745,26 @@ impl SessionActor { fn reset_input_activity(&mut self) -> Result<(), String> { self.input_speaking = false; self.recognition_pending = false; + self.stream_user_interim(0, "")?; self.stream_input_activity() } + /// Ephemeral like input activity. Muted words are never shown. + fn stream_user_interim(&mut self, cursor: u64, text: &str) -> Result<(), String> { + let text = if self.muted.load(Ordering::SeqCst) { + "" + } else { + text + }; + if matches!(&*self.transcript, Transcript::Stdout { .. }) + && (self.interim_shown || !text.is_empty()) + { + self.interim_shown = !text.is_empty(); + print_flushed(&user_interim_line(cursor, text))?; + } + Ok(()) + } + fn stream_live_event( &self, token: u64, @@ -4419,6 +4451,87 @@ mod tests { ) } + #[test] + fn user_interim_stream_shows_words_and_clears_on_mute_and_reset() { + const CHILD: &str = "BERD_CALL_TEST_INTERIM_CHILD"; + if std::env::var_os(CHILD).is_none() { + let output = Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "host_session::tests::user_interim_stream_shows_words_and_clears_on_mute_and_reset", "--nocapture"]) + .env(CHILD, "1") + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8(output.stdout).unwrap(); + let rows: Vec = stdout + .lines() + .filter_map(|line| { + line.strip_prefix("0\tuser_interim\t") + .map(|text| serde_json::from_str(text).unwrap()) + }) + .collect(); + let expected: Vec<_> = [ + (3, "you sh"), + (3, "you should\tgo"), + (3, ""), + (4, "next"), + (0, ""), + (5, "back"), + (0, ""), + ] + .into_iter() + .map(|(cursor, text)| json!({"cursor": cursor, "text": text})) + .collect(); + assert_eq!(rows, expected, "{stdout}"); + return; + } + for (transcript, expert) in [ + (Transcript::Stdout { live: false }, false), + (Transcript::Silent, false), + (Transcript::Stdout { live: true }, true), + ] { + let mut child = Command::new("/bin/cat") + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .spawn() + .unwrap(); + let writer = Arc::new(Mutex::new(child.stdin.take().unwrap())); + let (_event_tx, events) = mpsc::sync_channel(1); + let (_command_tx, commands) = mpsc::sync_channel(1); + let (audio, _audio_rx) = mpsc::sync_channel(1); + let mut actor = test_actor(writer, events, commands, audio); + actor.transcript = Arc::new(transcript); + actor.expert_spokesperson = expert; + actor.muted.store(false, Ordering::SeqCst); + let interim = |token, text: &str| SessionMessage::UserInterim { + token, + text: text.into(), + }; + let mute = |active| SessionMessage::InputMuteApplied { id: 0, active }; + for event in [ + interim(3, "you sh"), + interim(3, "you should\tgo"), + interim(3, ""), + interim(4, ""), + interim(4, "next"), + mute(true), + interim(4, "muted words"), + mute(false), + interim(5, "back"), + ] { + actor.handle_event(event).unwrap(); + } + actor.reset_input_activity().unwrap(); + actor.reset_input_activity().unwrap(); + assert!(actor.transcribed.is_empty()); + child.kill().unwrap(); + child.wait().unwrap(); + } + } + #[test] fn input_activity_is_cursorless_and_preserves_both_states() { for speaking in [false, true] { diff --git a/src-tauri/crates/berd-call/src/input.rs b/src-tauri/crates/berd-call/src/input.rs index 3583e74c7..1bf58849a 100644 --- a/src-tauri/crates/berd-call/src/input.rs +++ b/src-tauri/crates/berd-call/src/input.rs @@ -42,8 +42,8 @@ const OPENAI_LIVE_RESULT_TIMEOUT: Duration = Duration::from_secs(5); const OPENAI_FINAL_WRITE_TIMEOUT: Duration = Duration::from_secs(1); const OPENAI_PRE_ROLL_FRAMES: usize = 15; const FINAL_STORAGE_TIMEOUT: Duration = Duration::from_secs(5); -#[cfg(target_os = "macos")] -const MAC_LIVE_NO_RESULT_TIMEOUT: Duration = Duration::from_secs(5); +#[cfg(any(target_os = "macos", test))] +const MAC_PENDING_UTTERANCE_TIMEOUT: Duration = Duration::from_secs(5); const FINISH_TIMEOUT: Duration = Duration::from_secs(crate::MAC_SPEECH_RECOGNITION_FINISH_TIMEOUT_SECONDS + 5 + 1); @@ -122,7 +122,11 @@ impl FinalTranscriptStorageReceipt { pub enum VoiceInputEvent { Ready, SpeakingChanged(bool), + /// Agent speech admission hold, which may expire before recognition finishes. RecognitionPendingChanged(bool), + /// Display only: words Apple is still transcribing, or empty once none are. + /// Never stored, delivered or treated as an utterance. + InterimTranscript(String), FinalTranscript { text: String, storage_receipt: FinalTranscriptStorageReceipt, @@ -727,18 +731,64 @@ fn reap_dropped_worker(worker: thread::JoinHandle<()>) { }); } +// Native finals have no utterance ID. Consume expired credits first, so an old +// final cannot release a newer speech hold. Saturation keeps attribution +// ambiguous until the recognizer is explicitly reset, without unbounded storage. +#[cfg(any(target_os = "macos", test))] +const MAX_PENDING_RECOGNITIONS: usize = 1024; + +// Apple ends a final about 0.3 s after its last word, so this only absorbs +// trailing sound the detector still counted as speech. +#[cfg(any(target_os = "macos", test))] +const FINAL_COVERAGE_SLACK_SAMPLES: u64 = 12_000; + +// How much longer than the pending-utterance timeout agents wait while Apple's +// latest result is interim words. Finals can take seven seconds or more. +#[cfg(target_os = "macos")] +const HEARD_WORDS_GRACE: Duration = Duration::from_secs(10); + +#[derive(Default)] struct PendingRecognitions { count: usize, + blocking: usize, + ambiguous: bool, + /// Where each unresolved turn's speech ended, oldest first; None while it continues. + ends: VecDeque>, + /// Where the furthest final so far ended, since one can arrive before its turn ends. + #[cfg(any(target_os = "macos", test))] + final_end: Option, + /// Apple's latest result was interim words rather than a final. Apple + /// re-reports words a final left out, and interim results end where its + /// audio does rather than at the last word, so order says more than samples. + #[cfg(target_os = "macos")] + unfinalized: bool, + /// The interim text last shown, so repeats and clears are sent once. + interim: String, } impl PendingRecognitions { fn new() -> Self { - Self { count: 0 } + Self::default() } fn begin(&mut self, events: &tokio_mpsc::Sender) -> Result<(), ()> { - self.count += 1; - if self.count == 1 { + self.begin_bounded(events, usize::MAX) + } + + fn begin_bounded( + &mut self, + events: &tokio_mpsc::Sender, + limit: usize, + ) -> Result<(), ()> { + if self.count == limit { + self.ambiguous = true; + } else { + self.count += 1; + self.ends.push_back(None); + } + let was_blocking = self.blocking > 0; + self.blocking = self.blocking.saturating_add(1).min(limit); + if !was_blocking { events .blocking_send(VoiceInputEvent::RecognitionPendingChanged(true)) .map_err(|_| ())?; @@ -746,28 +796,89 @@ impl PendingRecognitions { Ok(()) } + #[cfg(any(target_os = "macos", test))] + fn end_turn( + &mut self, + events: &tokio_mpsc::Sender, + speech_end: u64, + ) -> Result<(), ()> { + if let Some(end @ None) = self.ends.back_mut() { + *end = Some(speech_end); + } + for _ in 0..self.final_end.map_or(0, |end| self.covered(end)) { + self.resolve(events)?; + } + Ok(()) + } + + /// Ended turns, oldest first, whose speech a final ending at `end` covers. + #[cfg(any(target_os = "macos", test))] + fn covered(&self, end: u64) -> usize { + let end = end.saturating_add(FINAL_COVERAGE_SLACK_SAMPLES); + self.ends + .iter() + .take_while(|turn| turn.is_some_and(|turn| turn <= end)) + .count() + } + + /// Apple has interim words no final has followed, so a transcript is still coming. + #[cfg(target_os = "macos")] + fn words_outstanding(&self) -> bool { + self.unfinalized + } + + /// Stops waiting on interim words whose final never came, so later holds keep their timeout. + #[cfg(target_os = "macos")] + fn abandon_heard_words(&mut self) { + self.unfinalized = false; + } + + /// Interim text is best effort: a full event queue drops it rather than stall recognition. + fn show_interim(&mut self, events: &tokio_mpsc::Sender, text: &str) { + if self.interim != text { + self.interim = text.to_string(); + let _ = events.try_send(VoiceInputEvent::InterimTranscript(self.interim.clone())); + } + } + + fn result_is_current(&self) -> bool { + !self.ambiguous && self.count == self.blocking + } + fn resolve(&mut self, events: &tokio_mpsc::Sender) -> Result<(), ()> { - if self.count == 0 { + if self.count == 0 || self.ambiguous { return Ok(()); } + let current = self.result_is_current(); self.count -= 1; - if self.count == 0 { - events - .blocking_send(VoiceInputEvent::RecognitionPendingChanged(false)) - .map_err(|_| ())?; + self.ends.pop_front(); + if current { + self.blocking -= 1; + if self.blocking == 0 { + events + .blocking_send(VoiceInputEvent::RecognitionPendingChanged(false)) + .map_err(|_| ())?; + } } Ok(()) } - fn reset(&mut self, events: &tokio_mpsc::Sender) -> Result<(), ()> { - if self.count > 0 { - self.count = 0; + fn release_hold(&mut self, events: &tokio_mpsc::Sender) -> Result<(), ()> { + if self.blocking > 0 { + self.blocking = 0; events .blocking_send(VoiceInputEvent::RecognitionPendingChanged(false)) .map_err(|_| ())?; } Ok(()) } + + fn reset(&mut self, events: &tokio_mpsc::Sender) -> Result<(), ()> { + self.release_hold(events)?; + self.show_interim(events, ""); + *self = Self::new(); + Ok(()) + } } fn effective_mute_epoch( @@ -1446,15 +1557,29 @@ fn forward_mac_events( ) -> Result<(), ()> { while let Ok(event) = recognition_events.try_recv() { match event { - crate::mac_speech::MacSpeechRecognitionEvent::Final(text) => { + crate::mac_speech::MacSpeechRecognitionEvent::Final(text, end) => { let text = text.trim().to_string(); if !text.is_empty() { send_final(events, text, storage_deadline)?; } - pending.resolve(events)?; - if pending.count == 0 { + // One final can span several turns split by pauses, and one turn + // can produce several finals, so a final with an end resolves + // exactly the turns it covers. Without one it resolves the oldest. + pending.unfinalized = false; + pending.final_end = pending.final_end.max(end); + for _ in 0..end.map_or(1, |end| pending.covered(end)) { + pending.resolve(events)?; + } + if pending.blocking == 0 { *settle_deadline = None; } + if !pending.words_outstanding() { + pending.show_interim(events, ""); + } + } + crate::mac_speech::MacSpeechRecognitionEvent::Heard(_, text) => { + pending.unfinalized = true; + pending.show_interim(events, &text); } crate::mac_speech::MacSpeechRecognitionEvent::Finished => { if storage_deadline.is_none() { @@ -1488,7 +1613,7 @@ fn begin_mac_turn( events: &tokio_mpsc::Sender, settle_deadline: &mut Option, ) -> Result<(), ()> { - pending.begin(events)?; + pending.begin_bounded(events, MAX_PENDING_RECOGNITIONS)?; *settle_deadline = None; Ok(()) } @@ -1498,9 +1623,10 @@ fn end_mac_turn( pending: &PendingRecognitions, settle_deadline: &mut Option, now: Instant, + timeout: Duration, ) { - if pending.count > 0 { - *settle_deadline = Some(now + MAC_LIVE_NO_RESULT_TIMEOUT); + if pending.blocking > 0 { + *settle_deadline = Some(now + timeout); } } @@ -1510,7 +1636,10 @@ fn mac_settle_expired( settle_deadline: Option, now: Instant, ) -> bool { - pending.count > 0 && settle_deadline.is_some_and(|deadline| now >= deadline) + pending.blocking > 0 + && settle_deadline.is_some_and(|deadline| { + now >= deadline && (!pending.words_outstanding() || now >= deadline + HEARD_WORDS_GRACE) + }) } #[cfg(target_os = "macos")] @@ -1554,6 +1683,8 @@ fn mac_speech_worker( let mut pending = PendingRecognitions::new(); let mut settle_deadline = None; let mut received_audio = false; + // Samples pushed to the current recognizer, the timeline its finals report. + let mut recognized = 0_u64; if events.blocking_send(VoiceInputEvent::Ready).is_err() { return; } @@ -1571,27 +1702,14 @@ fn mac_speech_worker( return; } if mac_settle_expired(&pending, settle_deadline, Instant::now()) { - // SpeechTranscriber has no per-utterance no-result callback. End - // this recognizer generation at the bounded idle deadline so its - // suppressed late callbacks cannot resolve a later generation. - recognizer.cancel(); - let _ = pending.reset(&events); - settle_deadline = None; - input_48k.clear(); - leftover_16k.clear(); - silence_frames = 0; - vad = earshot::Detector::new(earshot::DefaultPredictor::new()); - match new_mac_recognizer(&hints) { - Ok((next, next_events)) => { - recognizer = next; - recognition_events = next_events; - received_audio = false; - } - Err(error) => { - let _ = events.blocking_send(VoiceInputEvent::Failed(error)); - return; - } + // Release admission only. Keep native recognition and all its queued + // events/audio alive so late finals still reach the conversation. + if pending.release_hold(&events).is_err() { + return; } + pending.abandon_heard_words(); + pending.show_interim(&events, ""); + settle_deadline = None; } let frame = match frames.recv_timeout(Duration::from_millis(50)) { Ok(frame) => Some(frame), @@ -1619,6 +1737,7 @@ fn mac_speech_worker( recognizer = next; recognition_events = next_events; received_audio = false; + recognized = 0; } Err(error) => { let _ = events.blocking_send(VoiceInputEvent::Failed(error)); @@ -1642,6 +1761,7 @@ fn mac_speech_worker( return; } received_audio = true; + recognized += frame.frame.samples().len() as u64; input_48k.extend_from_slice(frame.frame.samples()); while input_48k.len() >= chunk_in { @@ -1664,7 +1784,25 @@ fn mac_speech_worker( if silence_frames >= SILENCE_FLUSH_FRAMES { silence_frames = 0; in_speech = false; - end_mac_turn(&pending, &mut settle_deadline, Instant::now()); + // Pushed audio runs slightly ahead of the detector, which + // only delays coverage. + if pending + .end_turn( + &events, + recognized.saturating_sub( + (SILENCE_FLUSH_FRAMES * VAD_FRAME_SAMPLES * 3) as u64, + ), + ) + .is_err() + { + return; + } + end_mac_turn( + &pending, + &mut settle_deadline, + Instant::now(), + MAC_PENDING_UTTERANCE_TIMEOUT, + ); let _ = events.blocking_send(VoiceInputEvent::SpeakingChanged(false)); } } @@ -2737,6 +2875,7 @@ mod tests { storage_receipt.stored(); observed.push("final"); } + VoiceInputEvent::InterimTranscript(_) => {} VoiceInputEvent::Failed(error) => panic!("unexpected input failure: {error}"), } } @@ -3441,7 +3580,11 @@ mod tests { ); } let (event_tx, mut event_rx) = tokio_mpsc::channel(4); - let mut pending = PendingRecognitions { count: 2 }; + let mut pending = PendingRecognitions { + count: 2, + blocking: 2, + ..Default::default() + }; deliver_openai_turns( &mut committed, @@ -3618,7 +3761,11 @@ mod tests { &mut completed, ); let (event_tx, mut event_rx) = tokio_mpsc::channel(2); - let mut pending = PendingRecognitions { count: 1 }; + let mut pending = PendingRecognitions { + count: 1, + blocking: 1, + ..Default::default() + }; deliver_openai_turns( &mut committed, @@ -3741,12 +3888,17 @@ mod tests { recognition_tx .send(crate::mac_speech::MacSpeechRecognitionEvent::Final( text.to_string(), + None, )) .unwrap(); } let (event_tx, mut event_rx) = tokio_mpsc::channel(4); - let mut pending = PendingRecognitions { count: 2 }; - let mut settle_deadline = Some(Instant::now() + MAC_LIVE_NO_RESULT_TIMEOUT); + let mut pending = PendingRecognitions { + count: 2, + blocking: 2, + ..Default::default() + }; + let mut settle_deadline = Some(Instant::now() + MAC_PENDING_UTTERANCE_TIMEOUT); forward_mac_events( &mut recognition_rx, @@ -3769,56 +3921,304 @@ mod tests { #[cfg(target_os = "macos")] #[test] fn mac_new_speech_cancels_then_refreshes_the_settle_deadline() { - let (event_tx, _event_rx) = tokio_mpsc::channel(4); + let (event_tx, mut event_rx) = tokio_mpsc::channel(8); let mut pending = PendingRecognitions::new(); let mut deadline = None; let now = Instant::now(); begin_mac_turn(&mut pending, &event_tx, &mut deadline).unwrap(); - end_mac_turn(&pending, &mut deadline, now); + end_mac_turn(&pending, &mut deadline, now, MAC_PENDING_UTTERANCE_TIMEOUT); let first_deadline = deadline.expect("first deadline"); begin_mac_turn(&mut pending, &event_tx, &mut deadline).unwrap(); assert!(deadline.is_none()); - end_mac_turn(&pending, &mut deadline, now + Duration::from_secs(1)); + end_mac_turn( + &pending, + &mut deadline, + now + Duration::from_secs(1), + MAC_PENDING_UTTERANCE_TIMEOUT, + ); assert!(deadline.is_some_and(|deadline| deadline > first_deadline)); + assert_eq!(pending.count, 2); + assert_eq!(pending.blocking, 2); + assert!(matches!( + event_rx.try_recv(), + Ok(VoiceInputEvent::RecognitionPendingChanged(true)) + )); + assert!(event_rx.try_recv().is_err()); + let (native_tx, mut native_rx) = tokio_mpsc::unbounded_channel(); + native_tx + .send(crate::mac_speech::MacSpeechRecognitionEvent::Final( + "continued speech".into(), + None, + )) + .unwrap(); + forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); + assert_eq!(pending.blocking, 1); + assert!(deadline.is_some()); } #[cfg(target_os = "macos")] #[test] - fn mac_timeout_reset_drops_the_old_event_channel_before_new_pending() { - let (event_tx, _event_rx) = tokio_mpsc::channel(8); + fn mac_final_releases_every_turn_its_audio_covers() { + let (event_tx, mut event_rx) = tokio_mpsc::channel(16); let mut pending = PendingRecognitions::new(); let mut deadline = None; - let now = Instant::now(); + let timeout = Duration::from_secs(5); + // One sentence split by two pauses, then a turn the final does not reach. + for end in [48_000, 96_000, 200_000] { + begin_mac_turn(&mut pending, &event_tx, &mut deadline).unwrap(); + pending.end_turn(&event_tx, end).unwrap(); + end_mac_turn(&pending, &mut deadline, Instant::now(), timeout); + } + let (native_tx, mut native_rx) = tokio_mpsc::unbounded_channel(); + let final_at = + |end| crate::mac_speech::MacSpeechRecognitionEvent::Final("words".into(), end); + native_tx.send(final_at(Some(90_000))).unwrap(); + forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); + assert_eq!((pending.count, pending.blocking), (1, 1)); + assert!(deadline.is_some()); + + native_tx.send(final_at(Some(190_000))).unwrap(); + forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); + assert_eq!((pending.count, pending.blocking), (0, 0)); + assert!(deadline.is_none()); + let holds: Vec<_> = std::iter::from_fn(|| event_rx.try_recv().ok()) + .filter_map(|event| match event { + VoiceInputEvent::RecognitionPendingChanged(active) => Some(active), + _ => None, + }) + .collect(); + assert_eq!(holds, [true, false]); + } + + #[cfg(target_os = "macos")] + #[test] + fn mac_earlier_phrase_final_keeps_the_hold_for_the_rest_of_the_turn() { + let (event_tx, _event_rx) = tokio_mpsc::channel(16); + let mut pending = PendingRecognitions::new(); + let mut deadline = None; + let (native_tx, mut native_rx) = tokio_mpsc::unbounded_channel(); + let final_at = + |end| crate::mac_speech::MacSpeechRecognitionEvent::Final("words".into(), Some(end)); begin_mac_turn(&mut pending, &event_tx, &mut deadline).unwrap(); - end_mac_turn(&pending, &mut deadline, now); + // Apple finalizes the first phrase while the user is still speaking. + native_tx.send(final_at(50_000)).unwrap(); + forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); + pending.end_turn(&event_tx, 200_000).unwrap(); + end_mac_turn( + &pending, + &mut deadline, + Instant::now(), + Duration::from_secs(5), + ); + native_tx.send(final_at(120_000)).unwrap(); + forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); + assert_eq!(pending.blocking, 1); + assert!(deadline.is_some()); + + native_tx.send(final_at(195_000)).unwrap(); + forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); + assert_eq!(pending.blocking, 0); + assert!(deadline.is_none()); + } + + #[cfg(target_os = "macos")] + #[test] + fn mac_final_before_its_turn_ends_releases_the_hold_when_it_does() { + let (event_tx, _event_rx) = tokio_mpsc::channel(16); + let mut pending = PendingRecognitions::new(); + let mut deadline = None; + let (native_tx, mut native_rx) = tokio_mpsc::unbounded_channel(); + begin_mac_turn(&mut pending, &event_tx, &mut deadline).unwrap(); + // Apple finalizes the turn before the detector has heard enough silence to end it. + native_tx + .send(crate::mac_speech::MacSpeechRecognitionEvent::Final( + "words".into(), + Some(100_000), + )) + .unwrap(); + forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); + assert_eq!(pending.blocking, 1); + pending.end_turn(&event_tx, 105_000).unwrap(); + end_mac_turn( + &pending, + &mut deadline, + Instant::now(), + Duration::from_secs(5), + ); + assert_eq!((pending.count, pending.blocking), (0, 0)); + assert!(deadline.is_none()); + } + + #[test] + fn finals_cover_only_ended_turns_in_order() { + let (event_tx, _event_rx) = tokio_mpsc::channel(4); + let mut pending = PendingRecognitions::new(); + pending.begin(&event_tx).unwrap(); + pending.end_turn(&event_tx, 10_000).unwrap(); + pending.begin(&event_tx).unwrap(); + assert_eq!(pending.covered(1_000_000), 1); + pending.end_turn(&event_tx, 50_000).unwrap(); + pending.end_turn(&event_tx, 1).unwrap(); + assert_eq!(pending.covered(50_000 - FINAL_COVERAGE_SLACK_SAMPLES), 2); + assert_eq!(pending.covered(49_999 - FINAL_COVERAGE_SLACK_SAMPLES), 1); + pending.resolve(&event_tx).unwrap(); + assert_eq!(pending.covered(40_000), 1); + } + + #[cfg(target_os = "macos")] + #[test] + fn mac_hold_waits_past_its_timeout_for_words_apple_has_heard() { + let (event_tx, _event_rx) = tokio_mpsc::channel(16); + let mut pending = PendingRecognitions::new(); + let mut deadline = None; + let now = Instant::now(); + let timeout = Duration::from_secs(5); + // Two sentences: the first final arrives, the second takes longer than the timeout. + for end in [100_000, 300_000] { + begin_mac_turn(&mut pending, &event_tx, &mut deadline).unwrap(); + pending.end_turn(&event_tx, end).unwrap(); + end_mac_turn(&pending, &mut deadline, now, timeout); + } + let (native_tx, mut native_rx) = tokio_mpsc::unbounded_channel(); + use crate::mac_speech::MacSpeechRecognitionEvent::{Final, Heard}; + // Apple re-reports the second sentence's interim words after the first final. + native_tx.send(Heard(310_000, "you should".into())).unwrap(); + native_tx + .send(Final("first".into(), Some(110_000))) + .unwrap(); + native_tx.send(Heard(310_000, "you should".into())).unwrap(); + forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); + assert_eq!(pending.blocking, 1); + assert!(!mac_settle_expired(&pending, deadline, now + timeout)); assert!(mac_settle_expired( &pending, deadline, - now + MAC_LIVE_NO_RESULT_TIMEOUT + now + timeout + HEARD_WORDS_GRACE )); + pending.release_hold(&event_tx).unwrap(); + pending.abandon_heard_words(); + assert!(!pending.words_outstanding()); + // The late final still resolves its turn. + native_tx + .send(Final("second".into(), Some(310_000))) + .unwrap(); + forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); + assert_eq!(pending.count, 0); + } - let (old_tx, old_rx) = tokio_mpsc::unbounded_channel(); - drop(old_rx); - pending.reset(&event_tx).unwrap(); - deadline = None; + #[cfg(target_os = "macos")] + #[test] + fn mac_hold_keeps_its_timeout_once_a_final_follows_the_heard_words() { + let (event_tx, _event_rx) = tokio_mpsc::channel(16); + let mut pending = PendingRecognitions::new(); + let mut deadline = None; + let now = Instant::now(); + let timeout = Duration::from_secs(5); + // Words, then noise after them that Apple never transcribes. Interim + // results end where Apple's audio did, past the final's last word. + for end in [100_000, 300_000] { + begin_mac_turn(&mut pending, &event_tx, &mut deadline).unwrap(); + pending.end_turn(&event_tx, end).unwrap(); + end_mac_turn(&pending, &mut deadline, now, timeout); + } + let (native_tx, mut native_rx) = tokio_mpsc::unbounded_channel(); + use crate::mac_speech::MacSpeechRecognitionEvent::{Final, Heard}; + native_tx.send(Heard(130_000, "words".into())).unwrap(); + native_tx + .send(Final("words".into(), Some(110_000))) + .unwrap(); + forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); + assert_eq!(pending.blocking, 1); + assert!(mac_settle_expired(&pending, deadline, now + timeout)); + } + + #[cfg(target_os = "macos")] + #[test] + fn mac_hold_expiry_keeps_late_finals_without_clearing_a_new_hold() { + let (event_tx, mut event_rx) = tokio_mpsc::channel(16); + let mut pending = PendingRecognitions::new(); + let mut deadline = None; + let now = Instant::now(); begin_mac_turn(&mut pending, &event_tx, &mut deadline).unwrap(); - assert!(old_tx + end_mac_turn(&pending, &mut deadline, now, Duration::from_millis(750)); + assert!(!mac_settle_expired( + &pending, + deadline, + now + Duration::from_millis(749) + )); + assert!(mac_settle_expired( + &pending, + deadline, + now + Duration::from_millis(750) + )); + pending.release_hold(&event_tx).unwrap(); + assert_eq!(pending.count, 1); + assert_eq!(pending.blocking, 0); + begin_mac_turn(&mut pending, &event_tx, &mut deadline).unwrap(); + assert!(deadline.is_none()); + let (native_tx, mut native_rx) = tokio_mpsc::unbounded_channel(); + native_tx .send(crate::mac_speech::MacSpeechRecognitionEvent::Final( - "late old final".to_string(), + "late old final".into(), + None, )) - .is_err()); + .unwrap(); + forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); assert_eq!(pending.count, 1); - - let (new_tx, mut new_rx) = tokio_mpsc::unbounded_channel(); - new_tx + assert_eq!(pending.blocking, 1); + let mut saw_text = false; + let mut holds = Vec::new(); + while let Ok(event) = event_rx.try_recv() { + match event { + VoiceInputEvent::FinalTranscript { text, .. } => { + assert_eq!(text, "late old final"); + saw_text = true; + } + VoiceInputEvent::RecognitionPendingChanged(active) => holds.push(active), + _ => {} + } + } + assert!(saw_text); + assert_eq!(holds, [true, false, true]); + native_tx .send(crate::mac_speech::MacSpeechRecognitionEvent::Final( - "current final".to_string(), + "new final".into(), + None, )) .unwrap(); - forward_mac_events(&mut new_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); + forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); + assert_eq!(pending.blocking, 0); + } + + #[test] + fn pending_credit_saturation_stays_conservative_until_reset() { + let (tx, mut rx) = tokio_mpsc::channel(8); + let mut pending = PendingRecognitions::new(); + for _ in 0..MAX_PENDING_RECOGNITIONS + 10 { + pending + .begin_bounded(&tx, MAX_PENDING_RECOGNITIONS) + .unwrap(); + } + assert_eq!(pending.count, MAX_PENDING_RECOGNITIONS); + assert!(pending.ambiguous); + pending.release_hold(&tx).unwrap(); + pending + .begin_bounded(&tx, MAX_PENDING_RECOGNITIONS) + .unwrap(); + for _ in 0..MAX_PENDING_RECOGNITIONS + 10 { + assert!(!pending.result_is_current()); + pending.resolve(&tx).unwrap(); + } + assert_eq!(pending.blocking, 1); + pending.reset(&tx).unwrap(); assert_eq!(pending.count, 0); + assert!(!pending.ambiguous); + while rx.try_recv().is_ok() {} + pending.begin(&tx).unwrap(); + pending.resolve(&tx).unwrap(); + assert_eq!(pending.blocking, 0); } #[tokio::test] diff --git a/src-tauri/crates/berd-call/src/mac_speech.rs b/src-tauri/crates/berd-call/src/mac_speech.rs index def4ca5e5..2985c454f 100644 --- a/src-tauri/crates/berd-call/src/mac_speech.rs +++ b/src-tauri/crates/berd-call/src/mac_speech.rs @@ -23,7 +23,10 @@ pub struct MacSpeechEngineStatus { #[derive(Debug, PartialEq, Eq)] pub enum MacSpeechRecognitionEvent { - Final(String), + /// A final result and where its audio ends, in 48 kHz samples of pushed input. + Final(String, Option), + /// Interim words still being transcribed, heard through this 48 kHz sample. + Heard(u64, String), Finished, Failed(String), } @@ -43,7 +46,7 @@ unsafe extern "C" { fn berd_macos_stt_create( locale: *const c_char, hints_json: *const c_char, - event: Option, + event: Option, context: *mut c_void, error_out: *mut *mut c_char, ) -> *mut c_void; @@ -156,25 +159,40 @@ struct RecognitionContext { events: mpsc::UnboundedSender, } -fn map_recognition_event(code: i32, text: Option) -> Option { +fn map_recognition_event( + code: i32, + text: Option, + end_seconds: f64, +) -> Option { + let end = (end_seconds.is_finite() && end_seconds >= 0.0) + .then(|| (end_seconds * 48_000.0).round() as u64); match code { - 1 => Some(MacSpeechRecognitionEvent::Final(text.unwrap_or_default())), + 1 => Some(MacSpeechRecognitionEvent::Final( + text.unwrap_or_default(), + end, + )), 2 => Some(MacSpeechRecognitionEvent::Finished), 3 => Some(MacSpeechRecognitionEvent::Failed(text.unwrap_or_else( || "macOS speech recognition failed.".to_string(), ))), + 4 => end.map(|end| MacSpeechRecognitionEvent::Heard(end, text.unwrap_or_default())), _ => None, } } -unsafe extern "C" fn recognition_event(code: i32, text: *const c_char, context: *mut c_void) { +unsafe extern "C" fn recognition_event( + code: i32, + text: *const c_char, + end_seconds: f64, + context: *mut c_void, +) { if context.is_null() { return; } let context = unsafe { &*context.cast::() }; let text = (!text.is_null()).then(|| unsafe { CStr::from_ptr(text).to_string_lossy().into_owned() }); - if let Some(event) = map_recognition_event(code, text) { + if let Some(event) = map_recognition_event(code, text, end_seconds) { let _ = context.events.send(event); } } @@ -286,20 +304,32 @@ mod tests { #[test] fn native_event_codes_map_without_provider_types_escaping() { assert_eq!( - map_recognition_event(1, Some("hello".to_string())), - Some(MacSpeechRecognitionEvent::Final("hello".to_string())) + map_recognition_event(1, Some("hello".to_string()), 1.5), + Some(MacSpeechRecognitionEvent::Final( + "hello".to_string(), + Some(72_000) + )) + ); + assert_eq!( + map_recognition_event(1, Some("hello".to_string()), f64::NAN), + Some(MacSpeechRecognitionEvent::Final("hello".to_string(), None)) ); assert_eq!( - map_recognition_event(2, None), + map_recognition_event(2, None, f64::NAN), Some(MacSpeechRecognitionEvent::Finished) ); assert_eq!( - map_recognition_event(3, None), + map_recognition_event(3, None, f64::NAN), Some(MacSpeechRecognitionEvent::Failed( "macOS speech recognition failed.".to_string() )) ); - assert_eq!(map_recognition_event(99, None), None); + assert_eq!( + map_recognition_event(4, Some("hel".to_string()), 1.5), + Some(MacSpeechRecognitionEvent::Heard(72_000, "hel".to_string())) + ); + assert_eq!(map_recognition_event(4, None, f64::NAN), None); + assert_eq!(map_recognition_event(99, None, f64::NAN), None); } #[test] @@ -319,13 +349,54 @@ mod tests { ); } + #[test] + #[ignore = "runs on-device macOS speech recognition over `say` output"] + fn final_end_is_measured_on_the_pushed_audio_timeline() { + let path = std::env::temp_dir().join("berd-call-final-end.wav"); + let status = std::process::Command::new("say") + .args(["--data-format=LEF32@48000", "-o"]) + .arg(&path) + .arg("The quick brown fox jumps over the lazy dog.") + .status() + .unwrap(); + assert!(status.success()); + let wav = std::fs::read(&path).unwrap(); + let data = wav.windows(4).position(|w| w == b"data").unwrap() + 8; + let speech: Vec = wav[data..] + .chunks_exact(4) + .map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]])) + .collect(); + let lead = vec![0.0; 48_000]; + let (mut recognizer, mut events) = MacSpeechRecognizer::new().unwrap(); + for part in [&lead, &speech, &lead] { + for chunk in part.chunks(960) { + recognizer.push_48khz_mono_f32(chunk).unwrap(); + std::thread::sleep(std::time::Duration::from_millis(20)); + } + } + recognizer.finish().unwrap(); + let mut ends = Vec::new(); + while let Ok(event) = events.try_recv() { + if let MacSpeechRecognitionEvent::Final(_, end) = event { + ends.push(end.unwrap()); + } + } + let (start, stop) = (lead.len() as u64, (lead.len() + speech.len()) as u64); + let last = *ends.last().expect("a final"); + // Apple ends a final about 0.3 s after the last sound. + assert!( + last > start && last <= stop + 24_000, + "{ends:?} not within {start}..{stop}" + ); + } + #[test] fn callback_is_safe_after_the_event_receiver_is_dropped() { let (events, receiver) = mpsc::unbounded_channel(); drop(receiver); let context = Box::into_raw(Box::new(RecognitionContext { events })); let text = CString::new("ignored").unwrap(); - unsafe { recognition_event(1, text.as_ptr(), context.cast()) }; + unsafe { recognition_event(1, text.as_ptr(), f64::NAN, context.cast()) }; unsafe { drop(Box::from_raw(context)) }; } diff --git a/src-tauri/crates/berd-call/src/main.rs b/src-tauri/crates/berd-call/src/main.rs index b46fddbd1..69f6a90fc 100644 --- a/src-tauri/crates/berd-call/src/main.rs +++ b/src-tauri/crates/berd-call/src/main.rs @@ -22,7 +22,6 @@ use berd_call::input::{ VoiceInputConfig, VoiceInputControls, VoiceInputEngineConfig, VoiceInputEvent, VoiceInputFrame, VoiceInputRuntime, INPUT_FRAME_SAMPLES, }; -use berd_call::openai_realtime_protocol::accepted_handoff_tool_output; use berd_call::openai_realtime_protocol::expert_handoff_message; use berd_call::openai_realtime_protocol::expert_transcript_message; use berd_call::openai_realtime_protocol::RealtimeExpertMessage; @@ -96,6 +95,8 @@ const PCM_FRAME_BYTES: usize = INPUT_FRAME_SAMPLES * std::mem::size_of::(); const MAX_FINAL_TEXT_BYTES: usize = 64 * 1024; const MAX_SPEAK_TEXT_BYTES: usize = 16 * 1024; const MAX_HANDOFF_IDS: usize = 64; +/// Speech probability that counts as the user talking, with or without agent playback. +const VAD_THRESHOLD: f32 = 0.9; const MAX_HANDOFF_ID_BYTES: usize = 512; const MAX_HANDOFF_REASON_BYTES: usize = 4 * 1024; const INPUT_QUEUE_CAPACITY: usize = 32; @@ -2917,6 +2918,8 @@ fn run_session(config: SessionConfig, pcm_output_fd: RawFd) -> Result<(), String enum LivePlaybackInput { Samples(Vec), Finish, + #[cfg(test)] + Checkpoint(SyncSender<()>), } type LivePlaybackResult = Result<(String, u64), (String, u64, String)>; @@ -4019,7 +4022,7 @@ fn run_expert_spokesperson_session( } // The worker has stopped; its result settles this reply. Err(mpsc::TrySendError::Disconnected(_)) => {} - Err(mpsc::TrySendError::Full(LivePlaybackInput::Finish)) => { + Err(mpsc::TrySendError::Full(_)) => { unreachable!("streaming sends samples") } }, @@ -4226,14 +4229,10 @@ fn run_expert_spokesperson_session( LiveSideEvent::Handoff { call_id, .. } => call_id, _ => return Err("live handoff did not produce a handoff ID".into()), }; - // A provider that expired right behind the handoff has already - // closed the runtime. Its queued terminal event settles the handoff. - let _ = runtime.as_ref().expect("initialized runtime").send( - SpokespersonCommand::Provider(accepted_handoff_tool_output( - &call_id, - &handoff_id, - )?), - ); + runtime + .as_ref() + .expect("initialized runtime") + .acknowledge_handoff(&call_id, &handoff_id)?; } event @ (SpokespersonEvent::Expired(_) | SpokespersonEvent::SessionLost(_)) => { let (message, connection_lost) = match event { @@ -5554,6 +5553,10 @@ fn spawn_live_playback( let mut loudness = berd_call::SpokespersonLoudness::default(); while let Ok(input) = receiver.recv() { match input { + #[cfg(test)] + LivePlaybackInput::Checkpoint(reached) => { + reached.send(()).unwrap(); + } LivePlaybackInput::Samples(mut samples) => { loudness.process(&mut samples); worker_output.write(&samples)?; @@ -5734,6 +5737,9 @@ fn cancel_live_playback(active: &mut Option) { if let Some(playback) = active.as_ref() { playback.active.store(false, Ordering::SeqCst); playback.output.notify_cancel_requested(); + // The worker may be waiting for provider input instead of an output acknowledgement. + // A full queue already wakes it; its next write observes the cancelled authority. + let _ = playback.sender.try_send(LivePlaybackInput::Finish); } } @@ -5918,7 +5924,7 @@ fn flush_active_live_playback( response.pending_frames = 0; return Ok(()); } - Err(mpsc::TrySendError::Full(LivePlaybackInput::Finish)) => { + Err(mpsc::TrySendError::Full(_)) => { unreachable!("audio queue contains samples") } } @@ -5930,7 +5936,7 @@ fn flush_active_live_playback( mpsc::TrySendError::Full(LivePlaybackInput::Finish) | mpsc::TrySendError::Disconnected(_), ) => {} - Err(mpsc::TrySendError::Full(LivePlaybackInput::Samples(_))) => { + Err(mpsc::TrySendError::Full(_)) => { unreachable!("finish queue contains finish") } } @@ -6196,8 +6202,8 @@ fn acknowledge_output_ready( ) -> Result<(), String> { current.assistant_activity = input_controls.map(|controls| { controls - .begin_assistant_activity(0.65, current.input_during_tts.policy) - .expect("balanced assistant threshold is valid") + .begin_assistant_activity(VAD_THRESHOLD, current.input_during_tts.policy) + .expect("VAD threshold is valid") }); write_message( writer, @@ -7048,7 +7054,7 @@ fn stt_benchmark_target(config: &SttBackendConfig) -> Result Result Result write_message( + writer, + &SessionMessage::UserInterim { + token: *next_token, + text, + }, + )?, VoiceInputEvent::FinalTranscript { text, storage_receipt, @@ -10295,6 +10308,75 @@ mod tests { (header[3], payload) } + #[test] + fn live_cancel_wakes_a_worker_waiting_for_more_provider_audio() { + for tail_frames in [0, 1, session_audio::MAX_AUDIO_CHUNK_FRAMES - 1] { + let (child, mut host) = UnixStream::pair().unwrap(); + host.set_read_timeout(Some(Duration::from_secs(1))).unwrap(); + let transport = + Arc::new(unsafe { AudioPipeTransport::from_raw_fd(child.into_raw_fd()) }.unwrap()); + let (control, _controls) = mpsc::channel(); + let (completed, results) = mpsc::channel(); + let playback = + spawn_live_playback("response".into(), 7, None, transport, control, completed) + .unwrap(); + assert_eq!( + read_audio_record(&mut host).0, + session_audio::AUDIO_BEGIN_KIND + ); + playback + .output + .handle_ack(AudioHostAck::BeginAccepted) + .unwrap(); + playback + .sender + .send(LivePlaybackInput::Samples( + vec![0.1; session_audio::MAX_AUDIO_CHUNK_FRAMES], + )) + .unwrap(); + assert_eq!( + read_audio_record(&mut host).0, + session_audio::AUDIO_CHUNK_KIND + ); + playback + .output + .handle_ack(AudioHostAck::ChunkAccepted { sequence: 1 }) + .unwrap(); + if tail_frames > 0 { + playback + .sender + .send(LivePlaybackInput::Samples(vec![0.2; tail_frames])) + .unwrap(); + } + let (checkpoint, reached) = mpsc::sync_channel(0); + playback + .sender + .send(LivePlaybackInput::Checkpoint(checkpoint)) + .unwrap(); + reached.recv_timeout(Duration::from_secs(1)).unwrap(); + let mut active = Some(playback); + cancel_live_playback(&mut active); + assert_eq!( + read_audio_record(&mut host).0, + session_audio::AUDIO_CANCEL_KIND + ); + active + .as_ref() + .unwrap() + .output + .handle_ack(AudioHostAck::Cancelled { played_frames: 0 }) + .unwrap(); + assert_eq!( + results + .recv_timeout(Duration::from_secs(1)) + .unwrap() + .unwrap_err() + .2, + session_audio::AUDIO_CANCELLED + ); + } + } + #[test] fn live_spokesperson_playback_levels_chunks_before_the_audio_pipe() { let (child, mut host) = UnixStream::pair().unwrap(); @@ -13022,6 +13104,43 @@ mod tests { assert_eq!(core.utterances_after(0)[0].token, 1); } + #[test] + fn interim_words_carry_the_token_their_final_will_take() { + let mut core = SessionCore::default(); + let mut active = None; + let mut output = Vec::new(); + let mut next_token = 7; + handle_voice_input_event( + VoiceInputEvent::InterimTranscript("you sh".into()), + &mut core, + &mut active, + &mut next_token, + &mut output, + ) + .unwrap(); + assert_eq!(next_token, 7); + store_and_publish_voice_final( + "you should".into(), + || {}, + &mut core, + &mut active, + &mut next_token, + &mut output, + ) + .unwrap(); + let messages: Vec = String::from_utf8(output) + .unwrap() + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect(); + assert_eq!( + messages[0], + json!({"type":"user_interim","token":7,"text":"you sh"}) + ); + assert_eq!(messages[1]["type"], "live_event"); + assert_eq!(messages[1]["token"], 7); + } + #[test] fn runtime_pending_provisionally_holds_reserved_output_without_a_terminal() { let mut core = SessionCore::default(); diff --git a/src-tauri/crates/berd-call/src/openai_spokesperson.rs b/src-tauri/crates/berd-call/src/openai_spokesperson.rs index c3e285db8..4e9cbd9b0 100644 --- a/src-tauri/crates/berd-call/src/openai_spokesperson.rs +++ b/src-tauri/crates/berd-call/src/openai_spokesperson.rs @@ -1,4 +1,8 @@ use std::collections::{HashMap, HashSet, VecDeque}; +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, +}; use std::thread; use std::time::Duration; @@ -231,6 +235,13 @@ pub struct OpenAiSpokespersonRuntime { commands: mpsc::UnboundedSender, audio: mpsc::Sender, worker: Option>, + terminal_published: Arc, +} + +#[derive(Debug)] +enum RuntimeExit { + Shutdown, + Terminal, } struct PendingTruncation { @@ -322,9 +333,11 @@ impl OpenAiSpokespersonRuntime { config: OpenAiSpokespersonConfig, forward_provider_events: bool, ) -> Result<(Self, std::sync::mpsc::Receiver), String> { - let (commands, command_rx) = mpsc::unbounded_channel(); - let (audio, audio_rx) = mpsc::channel(INPUT_QUEUE_FRAMES); + let (commands, mut command_rx) = mpsc::unbounded_channel(); + let (audio, mut audio_rx) = mpsc::channel(INPUT_QUEUE_FRAMES); let (events, event_rx) = std::sync::mpsc::channel(); + let terminal_published = Arc::new(AtomicBool::new(false)); + let worker_terminal = Arc::clone(&terminal_published); let worker = thread::Builder::new() .name("berd-call-spokesperson".into()) .spawn(move || { @@ -338,15 +351,22 @@ impl OpenAiSpokespersonRuntime { return; } }; - if let Err(error) = runtime.block_on(run_inner( + let terminal = match runtime.block_on(run_inner( config, - command_rx, - audio_rx, + &mut command_rx, + &mut audio_rx, &events, forward_provider_events, )) { - let _ = events.send(SpokespersonEvent::Failed(error)); - } + Ok(RuntimeExit::Terminal) => true, + Ok(RuntimeExit::Shutdown) => false, + Err(error) => events.send(SpokespersonEvent::Failed(error)).is_ok(), + }; + // Publish the terminal before dropping the command receiver. A late handoff + // acknowledgement can then defer to that ordered event instead of masking it. + worker_terminal.store(terminal, Ordering::Release); + drop(command_rx); + drop(audio_rx); let _ = events.send(SpokespersonEvent::Closed); }) .map_err(|error| error.to_string())?; @@ -355,11 +375,23 @@ impl OpenAiSpokespersonRuntime { commands, audio, worker: Some(worker), + terminal_published, }, event_rx, )) } + /// Acknowledge a handoff unless an ordered provider terminal already ended it. + pub fn acknowledge_handoff(&self, call_id: &str, handoff_id: &str) -> Result<(), String> { + let command = SpokespersonCommand::Provider( + crate::openai_realtime_protocol::accepted_handoff_tool_output(call_id, handoff_id)?, + ); + match self.send(command) { + Err(_) if self.terminal_published.load(Ordering::Acquire) => Ok(()), + result => result, + } + } + pub fn send(&self, command: SpokespersonCommand) -> Result<(), String> { let command = match command { SpokespersonCommand::Provider(event) => provider_command(event), @@ -437,11 +469,13 @@ fn reap_spokesperson_worker( #[cfg(test)] async fn run( config: OpenAiSpokespersonConfig, - commands: mpsc::UnboundedReceiver, + mut commands: mpsc::UnboundedReceiver, events: &std::sync::mpsc::Sender, ) -> Result<(), String> { - let (_audio_tx, audio_rx) = mpsc::channel(INPUT_QUEUE_FRAMES); - run_inner(config, commands, audio_rx, events, false).await + let (_audio_tx, mut audio_rx) = mpsc::channel(INPUT_QUEUE_FRAMES); + run_inner(config, &mut commands, &mut audio_rx, events, false) + .await + .map(|_| ()) } /// The next command, or `None` once the runtime is shut down. @@ -457,11 +491,11 @@ async fn next_command( async fn run_inner( mut config: OpenAiSpokespersonConfig, - mut commands: mpsc::UnboundedReceiver, - mut audio: mpsc::Receiver, + commands: &mut mpsc::UnboundedReceiver, + audio: &mut mpsc::Receiver, events: &std::sync::mpsc::Sender, forward_provider_events: bool, -) -> Result<(), String> { +) -> Result { if let Err(existing) = rustls::crypto::aws_lc_rs::default_provider().install_default() { drop(existing); } @@ -488,7 +522,7 @@ async fn run_inner( .map_err(|_| "connect OpenAI Realtime timed out".to_string())? .map_err(|error| format!("connect OpenAI Realtime: {error}"))?, command = commands.recv() => match command { - Some(SpokespersonCommand::Shutdown) | None => return Ok(()), + Some(SpokespersonCommand::Shutdown) | None => return Ok(RuntimeExit::Shutdown), Some(_) => return Err("Spokesperson command arrived before readiness".into()), } }; @@ -528,7 +562,7 @@ async fn run_inner( && committed_input_items.is_empty() { let _ = socket.close(None).await; - return Ok(()); + return Ok(RuntimeExit::Shutdown); } let truncation_deadline = pending_truncations .values() @@ -556,7 +590,7 @@ async fn run_inner( } }, if shutdown_deadline.is_some() => { let _ = socket.close(None).await; - return Ok(()); + return Ok(RuntimeExit::Shutdown); } _ = async { match reset_deadline { @@ -600,7 +634,7 @@ async fn run_inner( return Err("Spokesperson output truncation timed out; server context is indeterminate".into()); } } - command = next_command(&mut commands, &mut audio), if shutdown_deadline.is_none() => { + command = next_command(commands, audio), if shutdown_deadline.is_none() => { match command? { Some(SpokespersonCommand::Provider(event)) => { send_json(&mut socket, event).await?; @@ -730,7 +764,7 @@ async fn run_inner( "OpenAI Realtime connection closed {detail}" )), )?; - return Ok(()); + return Ok(RuntimeExit::Terminal); } None => { send_event( @@ -739,7 +773,7 @@ async fn run_inner( "OpenAI Realtime connection ended without a close frame".into(), ), )?; - return Ok(()); + return Ok(RuntimeExit::Terminal); } Some(Ok(_)) => continue, Some(Err(error)) => { @@ -749,7 +783,7 @@ async fn run_inner( "OpenAI Realtime transport failed: {error}" )), )?; - return Ok(()); + return Ok(RuntimeExit::Terminal); } }; let Ok(value) = serde_json::from_str::(&text) else { continue }; @@ -1041,7 +1075,7 @@ async fn run_inner( || message == "Your session hit the maximum duration of 60 minutes." { send_event(events, SpokespersonEvent::Expired(message))?; - return Ok(()); + return Ok(RuntimeExit::Terminal); } else { return Err(message); } @@ -1158,6 +1192,7 @@ mod tests { commands, audio, worker: Some(worker), + terminal_published: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), }; let caller = std::thread::spawn(move || { runtime.retire_in_background(); @@ -1285,6 +1320,7 @@ mod tests { commands, audio, worker: None, + terminal_published: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), }; runtime @@ -1414,8 +1450,8 @@ mod tests { .await; let _ = socket.close(None).await; }); - let (commands, command_rx) = tokio::sync::mpsc::unbounded_channel(); - let (_audio, audio_rx) = tokio::sync::mpsc::channel(1); + let (commands, mut command_rx) = tokio::sync::mpsc::unbounded_channel(); + let (_audio, mut audio_rx) = tokio::sync::mpsc::channel(1); let (events, event_rx) = std::sync::mpsc::channel(); let ready = tokio::task::spawn_blocking(move || { let mut observed = Vec::new(); @@ -1434,8 +1470,8 @@ mod tests { Duration::from_secs(5), super::run_inner( test_config(endpoint, "test-voice", 1.0, Vec::new()), - command_rx, - audio_rx, + &mut command_rx, + &mut audio_rx, &events, true, ), @@ -2097,6 +2133,84 @@ mod tests { server.await.unwrap(); } + #[test] + fn handoff_acknowledgement_does_not_hide_unexplained_closure() { + let (commands, receiver) = tokio::sync::mpsc::unbounded_channel(); + let (audio, _audio_rx) = tokio::sync::mpsc::channel(1); + drop(receiver); + let runtime = super::OpenAiSpokespersonRuntime { + commands, + audio, + worker: None, + terminal_published: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), + }; + assert_eq!( + runtime.acknowledge_handoff("handoff-1", "handoff-external-1"), + Err("Spokesperson runtime is closed".into()), + ); + } + + #[tokio::test] + async fn handoff_acknowledgement_after_provider_terminal_preserves_its_attribution() { + for terminal in [ + json!({"type":"error","error":{"code":"session_expired","message":"expiry detail"}}), + json!({"type":"error","error":{"code":"test_failure","message":"failure detail"}}), + ] { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("ws://{}/", listener.local_addr().unwrap()); + let server = + tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut socket = accept_hdr_async(stream, require_test_authorization) + .await + .unwrap(); + let initial = receive_json(&mut socket).await; + acknowledge_initial_session(&mut socket, &initial, "test-model").await; + send_json(&mut socket, json!({ + "type":"response.function_call_arguments.done", + "response_id":"response-handoff", "call_id":"handoff-1", "name":"handoff", + "arguments":"{\"message\":\"inspect the repository\"}" + })).await; + send_json(&mut socket, terminal).await; + while socket.next().await.is_some() {} + }); + let (mut runtime, events) = OpenAiSpokespersonRuntime::spawn(test_config( + endpoint, + "test-voice", + 1.0, + Vec::new(), + )) + .unwrap(); + // Joining makes receiver closure precede acknowledgement without a timing assumption. + runtime = tokio::time::timeout( + Duration::from_secs(5), + tokio::task::spawn_blocking(move || { + runtime.worker.take().unwrap().join().unwrap(); + runtime + }), + ) + .await + .unwrap() + .unwrap(); + assert!(matches!(events.recv().unwrap(), SpokespersonEvent::Ready)); + assert!(matches!( + events.recv().unwrap(), + SpokespersonEvent::Handoff { .. } + )); + runtime + .acknowledge_handoff("handoff-1", "handoff-external-1") + .unwrap(); + match events.recv().unwrap() { + SpokespersonEvent::Expired(message) => assert_eq!(message, "expiry detail"), + SpokespersonEvent::Failed(message) => assert_eq!(message, "failure detail"), + event => panic!("original terminal was lost: {event:?}"), + } + assert!(matches!(events.recv().unwrap(), SpokespersonEvent::Closed)); + assert!(runtime.send(SpokespersonCommand::Shutdown).is_err()); + server.await.unwrap(); + } + } + #[test] fn output_truncation_timeout_is_bounded() { let now = tokio::time::Instant::now(); diff --git a/src-tauri/crates/berd-call/src/protocol.rs b/src-tauri/crates/berd-call/src/protocol.rs index cf6d578d5..7344bacb1 100644 --- a/src-tauri/crates/berd-call/src/protocol.rs +++ b/src-tauri/crates/berd-call/src/protocol.rs @@ -254,6 +254,14 @@ pub enum SessionMessage { #[serde(skip_serializing_if = "Option::is_none")] origin: Option, }, + /// The user's words still being transcribed, replaced by each update and + /// cleared with empty text. `token` is the one the next final will take, so + /// a reply composed against interim words stays current once they finalize. + /// Never an utterance itself. + UserInterim { + token: u64, + text: String, + }, /// The user's words as soon as they are transcribed, for hosts that show /// the conversation as it happens. `live_event` stays authoritative. UserTranscribed { diff --git a/src-tauri/crates/berd-call/src/session_audio.rs b/src-tauri/crates/berd-call/src/session_audio.rs index f6c33450f..8bffcb2bc 100644 --- a/src-tauri/crates/berd-call/src/session_audio.rs +++ b/src-tauri/crates/berd-call/src/session_audio.rs @@ -321,6 +321,8 @@ impl RemotePcmAudioOutput { /// Wakes the playback worker so it can serialize Cancel on the audio pipe. /// The caller owns only the cancellation flag; pipe writes remain worker-owned. pub fn notify_cancel_requested(&self) { + // Waiters check the flag under this lock, so taking it orders the wake after any check. + drop(self.state.lock().expect("remote output state")); self.changed.notify_all(); } @@ -1717,6 +1719,25 @@ mod tests { assert!(host.read(&mut byte).is_err()); } + #[test] + fn cancel_request_wakes_a_waiter_that_checked_the_flag_before_waiting() { + let (output, _host) = fixture(); + // Like the worker: the flag was checked under the lock, but the wait has not begun. + let state = output.state.lock().unwrap(); + let current = Arc::clone(&output); + let canceller = thread::spawn(move || { + current.active.store(false, Ordering::SeqCst); + current.notify_cancel_requested(); + }); + thread::sleep(Duration::from_millis(50)); + let (_state, wait) = output + .changed + .wait_timeout(state, Duration::from_secs(1)) + .unwrap(); + assert!(!wait.timed_out(), "cancellation wake was lost"); + canceller.join().unwrap(); + } + #[test] fn cancel_waits_for_quiescence_and_returns_the_settled_snapshot() { let (output, mut host) = fixture(); diff --git a/src-tauri/crates/berd-call/swift/BerdMacSpeechBridge/Sources/BerdMacSpeechBridge/BerdMacSpeechBridge.swift b/src-tauri/crates/berd-call/swift/BerdMacSpeechBridge/Sources/BerdMacSpeechBridge/BerdMacSpeechBridge.swift index e84f33489..00e270962 100644 --- a/src-tauri/crates/berd-call/swift/BerdMacSpeechBridge/Sources/BerdMacSpeechBridge/BerdMacSpeechBridge.swift +++ b/src-tauri/crates/berd-call/swift/BerdMacSpeechBridge/Sources/BerdMacSpeechBridge/BerdMacSpeechBridge.swift @@ -7,9 +7,12 @@ import Foundation private let finalEvent: Int32 = 1 private let finishedEvent: Int32 = 2 private let failedEvent: Int32 = 3 +private let heardEvent: Int32 = 4 +// The Double is where a final's or interim result's audio ends, in seconds of +// pushed input, or NaN. public typealias BerdMacSpeechEventCallback = @convention(c) ( - Int32, UnsafePointer?, UnsafeMutableRawPointer? + Int32, UnsafePointer?, Double, UnsafeMutableRawPointer? ) -> Void public typealias BerdMacSpeechProgressCallback = @convention(c) ( Double, UnsafeMutableRawPointer? @@ -130,7 +133,7 @@ private func speechTranscriber(for locale: Locale) -> SpeechTranscriber { SpeechTranscriber( locale: locale, transcriptionOptions: [], - reportingOptions: [], + reportingOptions: [.volatileResults], attributeOptions: [.audioTimeRange] ) } @@ -306,9 +309,13 @@ private final class SpeechSession: @unchecked Sendable { guard let self else { return } let text = String(result.text.characters) .trimmingCharacters(in: .whitespacesAndNewlines) - guard !text.isEmpty else { continue } - guard result.isFinal else { continue } - self.emit(finalEvent, text: text) + let end = result.range.end.isNumeric ? result.range.end.seconds : .nan + if result.isFinal { + // Empty finals still settle their audio, so the host stops waiting on it. + self.emit(finalEvent, text: text, end: end) + } else if !text.isEmpty { + self.emit(heardEvent, text: text, end: end) + } } self?.taskCompleted() } catch is CancellationError { self?.taskCompleted() } @@ -417,11 +424,11 @@ private final class SpeechSession: @unchecked Sendable { Task.detached { [analyzer] in await analyzer.cancelAndFinishNow() } } - private func emit(_ event: Int32, text: String? = nil) { + private func emit(_ event: Int32, text: String? = nil, end: Double = .nan) { lock.withLock { guard callbacksEnabled, !terminalEmitted else { return } - if let text { text.withCString { callback(event, $0, context.pointer) } } - else { callback(event, nil, context.pointer) } + if let text { text.withCString { callback(event, $0, end, context.pointer) } } + else { callback(event, nil, end, context.pointer) } } } @@ -429,7 +436,7 @@ private final class SpeechSession: @unchecked Sendable { let emitted = lock.withLock { guard callbacksEnabled, !terminalEmitted else { return false } terminalEmitted = true - error.localizedDescription.withCString { callback(failedEvent, $0, context.pointer) } + error.localizedDescription.withCString { callback(failedEvent, $0, .nan, context.pointer) } return true } if emitted { completion.signal() } @@ -444,7 +451,7 @@ private final class SpeechSession: @unchecked Sendable { completedTasks += 1 guard completedTasks == 2, callbacksEnabled, !terminalEmitted else { return false } terminalEmitted = true - callback(finishedEvent, nil, context.pointer) + callback(finishedEvent, nil, .nan, context.pointer) return true } if emitted { completion.signal() } diff --git a/src-tauri/crates/berd-call/tests/session_protocol.rs b/src-tauri/crates/berd-call/tests/session_protocol.rs index 39e4ead87..a1b0c3d73 100644 --- a/src-tauri/crates/berd-call/tests/session_protocol.rs +++ b/src-tauri/crates/berd-call/tests/session_protocol.rs @@ -31,7 +31,7 @@ struct ExpertSpokespersonTestSession { stdin: Option>>, output: mpsc::Receiver, stderr: mpsc::Receiver, - audio_host: Option>, + audio_host: Option>>, ready: Value, } @@ -192,24 +192,52 @@ impl ExpertSpokespersonTestSession { self.wait_with_stderr().0 } - fn wait_with_stderr(mut self) -> (Vec, Vec) { + fn wait_with_stderr(self) -> (Vec, Vec) { + self.wait_with_audio_policy(false) + } + + fn wait_with_audio_policy( + mut self, + allow_truncated_payload: bool, + ) -> (Vec, Vec) { self.stdin.take(); let status = self.child.0.as_mut().unwrap().wait().unwrap(); self.child.0 = None; assert!(status.success()); - self.audio_host.take().unwrap().join().unwrap(); + // Only an explicitly expected disconnect may end a partial audio frame, + // and only after the child itself has exited successfully. + check_audio_host_result( + self.audio_host.take().unwrap().join().unwrap(), + allow_truncated_payload, + ); (self.output.iter().collect(), self.stderr.iter().collect()) } } fn write_session_json(writer: &mut impl Write, value: &Value) { + try_write_session_json(writer, value).unwrap(); +} + +fn try_write_session_json(writer: &mut impl Write, value: &Value) -> std::io::Result<()> { let payload = serde_json::to_vec(value).unwrap(); - writer.write_all(b"BV").unwrap(); - writer.write_all(&[FRAME_MARKER, 1]).unwrap(); - writer - .write_all(&(payload.len() as u32).to_le_bytes()) - .unwrap(); - writer.write_all(&payload).unwrap(); + writer.write_all(b"BV")?; + writer.write_all(&[FRAME_MARKER, 1])?; + writer.write_all(&(payload.len() as u32).to_le_bytes())?; + writer.write_all(&payload) +} + +fn write_audio_replies(writer: &mut impl Write, replies: &[Value]) -> bool { + let result = replies + .iter() + .try_for_each(|reply| try_write_session_json(writer, reply)) + .and_then(|()| writer.flush()); + match result { + Ok(()) => true, + // The child may terminate with audio records still queued. The fixture + // waiter independently checks its exit status and protocol terminals. + Err(error) if error.kind() == std::io::ErrorKind::BrokenPipe => false, + Err(error) => panic!("fixture audio reply failed: {error}"), + } } fn write_session_pcm(writer: &mut impl Write, value: f32) { @@ -335,18 +363,18 @@ fn session_command() -> (Command, File, UnixStream) { fn spawn_audio_host( reader: UnixStream, stdin: Arc>, -) -> std::thread::JoinHandle<()> { +) -> std::thread::JoinHandle> { spawn_audio_host_with_played_limit(reader, stdin, None, None, None, false) } -fn spawn_audio_host_with_played_limit( +fn spawn_audio_host_with_played_limit( mut reader: UnixStream, - stdin: Arc>, + stdin: Arc>, played_frame_limit: Option, mut played_ready: Option>, mut cancellation_gate: Option, mut fail_at_limit: bool, -) -> std::thread::JoinHandle<()> { +) -> std::thread::JoinHandle> { std::thread::spawn(move || { let mut current = None::<(u64, u64, u64)>; // The speech whose player failed, and what it played: like the real host, it ignores what the session sent before hearing. @@ -355,14 +383,14 @@ fn spawn_audio_host_with_played_limit( let mut header = [0_u8; 8]; match reader.read_exact(&mut header) { Ok(()) => {} - Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => return, + Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(()), Err(error) => panic!("audio pipe read failed: {error}"), } assert_eq!(&header[..2], b"BA"); assert_eq!(header[2], FRAME_MARKER); let length = u32::from_le_bytes(header[4..8].try_into().unwrap()) as usize; let mut payload = vec![0_u8; length]; - reader.read_exact(&mut payload).unwrap(); + reader.read_exact(&mut payload)?; let message = match header[3] { 1 => { assert_eq!(payload.len(), 16); @@ -385,25 +413,20 @@ fn spawn_audio_host_with_played_limit( state.2 += frames; let reached_limit = played_frame_limit.is_some_and(|limit| state.2 >= limit); let mut writer = stdin.lock().unwrap(); - write_session_json( - &mut *writer, - &json!({"type":"audio_chunk_accepted","speech_id":speech_id,"sequence":sequence}), - ); - write_session_json( - &mut *writer, - &json!({"type":"audio_played","speech_id":speech_id,"played_frames":played_frame_limit.map_or(state.2, |limit| state.2.min(limit))}), - ); + let mut replies = vec![ + json!({"type":"audio_chunk_accepted","speech_id":speech_id,"sequence":sequence}), + json!({"type":"audio_played","speech_id":speech_id,"played_frames":played_frame_limit.map_or(state.2, |limit| state.2.min(limit))}), + ]; if reached_limit && fail_at_limit { fail_at_limit = false; let played_frames = played_frame_limit.unwrap(); failed = Some((speech_id, played_frames)); current = None; - write_session_json( - &mut *writer, - &json!({"type":"audio_failed","speech_id":speech_id,"played_frames":played_frames,"message":"PCM audio output stopped unexpectedly"}), - ); + replies.push(json!({"type":"audio_failed","speech_id":speech_id,"played_frames":played_frames,"message":"PCM audio output stopped unexpectedly"})); + } + if !write_audio_replies(&mut *writer, &replies) { + return Ok(()); } - writer.flush().unwrap(); if reached_limit { if let Some(ready) = played_ready.take() { let _ = ready.send(()); @@ -423,7 +446,7 @@ fn spawn_audio_host_with_played_limit( if let Some(gate) = cancellation_gate.take_if(|gate| gate.at_drain) { let _ = gate.observed.send(()); if gate.release.recv().is_err() { - return; + return Ok(()); } continue; } @@ -435,11 +458,14 @@ fn spawn_audio_host_with_played_limit( let speech_id = u64::from_le_bytes(payload.try_into().unwrap()); if let Some((_, played_frames)) = failed.filter(|(id, _)| *id == speech_id) { let mut writer = stdin.lock().unwrap(); - write_session_json( + if !write_audio_replies( &mut *writer, - &json!({"type":"audio_cancelled","speech_id":speech_id,"played_frames":played_frames}), - ); - writer.flush().unwrap(); + &[ + json!({"type":"audio_cancelled","speech_id":speech_id,"played_frames":played_frames}), + ], + ) { + return Ok(()); + } continue; } let played_frames = current @@ -457,10 +483,116 @@ fn spawn_audio_host_with_played_limit( kind => panic!("unknown audio record kind {kind}"), }; let mut writer = stdin.lock().unwrap(); - write_session_json(&mut *writer, &message); - writer.flush().unwrap(); + if !write_audio_replies(&mut *writer, &[message]) { + return Ok(()); + } + } + }) +} + +fn check_audio_host_result(result: std::io::Result<()>, allow_truncated_payload: bool) { + if let Err(error) = result { + assert!( + allow_truncated_payload && error.kind() == std::io::ErrorKind::UnexpectedEof, + "fixture audio read failed: {error}" + ); + } +} + +#[test] +fn audio_fixture_payload_errors_require_explicit_shutdown_allowance() { + for allow in [false, true] { + check_audio_host_result(Ok(()), allow); + for kind in [ + std::io::ErrorKind::UnexpectedEof, + std::io::ErrorKind::PermissionDenied, + std::io::ErrorKind::ConnectionReset, + ] { + let accepted = + std::panic::catch_unwind(|| check_audio_host_result(Err(kind.into()), allow)) + .is_ok(); + assert_eq!(accepted, allow && kind == std::io::ErrorKind::UnexpectedEof); } + } +} + +#[test] +fn audio_fixture_truncated_payload_does_not_panic_before_shutdown_is_checked() { + let (input, _peer) = UnixStream::pair().unwrap(); + let (mut audio, reader) = UnixStream::pair().unwrap(); + audio.write_all(b"BA").unwrap(); + audio.write_all(&[FRAME_MARKER, 1]).unwrap(); + audio.write_all(&16_u32.to_le_bytes()).unwrap(); + // The child exits halfway through a valid begin payload. + audio.write_all(&1_u64.to_le_bytes()).unwrap(); + drop(audio); + let host = spawn_audio_host_with_played_limit( + reader, + Arc::new(Mutex::new(input)), + None, + None, + None, + false, + ); + let result = host + .join() + .expect("payload EOF panicked before child status was checked"); + assert_eq!( + result.as_ref().unwrap_err().kind(), + std::io::ErrorKind::UnexpectedEof + ); + check_audio_host_result(result, true); +} + +#[test] +fn audio_fixture_reply_errors_remain_visible_except_for_closed_input() { + struct FailingWriter(std::io::ErrorKind); + impl Write for FailingWriter { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Err(self.0.into()) + } + } + let reply = json!({"type":"audio_cancelled","speech_id":1,"played_frames":0}); + assert!(!write_audio_replies( + &mut FailingWriter(std::io::ErrorKind::BrokenPipe), + std::slice::from_ref(&reply) + )); + assert!(std::panic::catch_unwind(|| { + write_audio_replies( + &mut FailingWriter(std::io::ErrorKind::PermissionDenied), + &[reply], + ); }) + .is_err()); +} + +#[test] +fn audio_fixture_can_finish_a_queued_reply_after_child_input_closes() { + let (input, closed) = UnixStream::pair().unwrap(); + drop(closed); + let (mut audio, reader) = UnixStream::pair().unwrap(); + // Queue a valid begin record after the child has closed its input. No sleeps + // or scheduler assumptions: the fixture's first acknowledgement must fail. + audio.write_all(b"BA").unwrap(); + audio.write_all(&[FRAME_MARKER, 1]).unwrap(); + audio.write_all(&16_u32.to_le_bytes()).unwrap(); + audio.write_all(&1_u64.to_le_bytes()).unwrap(); + audio.write_all(&24_000_u64.to_le_bytes()).unwrap(); + drop(audio); + let host = spawn_audio_host_with_played_limit( + reader, + Arc::new(Mutex::new(input)), + None, + None, + None, + false, + ); + host.join() + .expect("late fixture reply panicked after child shutdown") + .unwrap(); } #[test] @@ -946,7 +1078,7 @@ fn active_spokesperson_disconnect_is_specific_terminal_and_does_not_replay() { fatal["message"], "Spokesperson connection was lost during an active turn" ); - let (remaining, stderr) = session.wait_with_stderr(); + let (remaining, stderr) = session.wait_with_audio_policy(true); assert!(remaining .iter() .all(|message| message["type"] != "live_event")); @@ -3264,7 +3396,7 @@ fn siri_remote_output_supports_consecutive_turns_and_cancellation() { let mut child_process = child.0.take().unwrap(); assert!(child_process.wait().unwrap().success()); reader.join().unwrap(); - audio_host.join().unwrap(); + audio_host.join().unwrap().unwrap(); } #[test] @@ -3325,7 +3457,7 @@ fn speech_whose_player_stops_is_interrupted_where_it_was_heard_and_the_call_goes send(json!({"type":"shutdown"})); drop(stdin); assert!(child.0.take().unwrap().wait().unwrap().success()); - audio_host.join().unwrap(); + audio_host.join().unwrap().unwrap(); } /// Accepts one OpenAI transcription session that never hears anything. diff --git a/src-tauri/src/commands/native_voice.rs b/src-tauri/src/commands/native_voice.rs index 9e5c3c1f2..e9e19f7bb 100644 --- a/src-tauri/src/commands/native_voice.rs +++ b/src-tauri/src/commands/native_voice.rs @@ -1781,6 +1781,8 @@ pub async fn start_native_voice_conversation( // The runtime owns recognition-pending sequencing. Berd's // renderer does not project that state yet. } + // Berd's renderer does not show interim words yet. + berd_call::input::VoiceInputEvent::InterimTranscript(_) => {} berd_call::input::VoiceInputEvent::FinalTranscript { text, storage_receipt,