From 7337171bf82a8c9eb252e8bd71f561c518ededbd Mon Sep 17 00:00:00 2001 From: Zack <04f47b290b155b3b8b782b11501cdb15f1e21e82dfceefd01ff0ed405459d927@buzz.block.builderlab.xyz> Date: Thu, 22 Oct 2026 23:32:31 -0400 Subject: [PATCH 01/13] fix(berd-call): wake idle live playback on cancellation Cancellation revoked output authority and notified output waiters, but a live playback worker could already be waiting for its next provider message. Queue Finish as well so the existing cancellation path can serialize Cancel without another provider event. A full input queue already gives the worker work that observes the revoked authority. The regression synchronizes after one chunk and checks cancellation with no tail, one buffered sample and a nearly full partial chunk. Signed-off-by: Zack <04f47b290b155b3b8b782b11501cdb15f1e21e82dfceefd01ff0ed405459d927@buzz.block.builderlab.xyz> (cherry picked from commit 80b97e4c2be90e6db13656d6d2a31aaca462199f) --- src-tauri/crates/berd-call/src/main.rs | 84 +++++++++++++++++++++++++- 1 file changed, 81 insertions(+), 3 deletions(-) diff --git a/src-tauri/crates/berd-call/src/main.rs b/src-tauri/crates/berd-call/src/main.rs index b46fddbd1..5c2379631 100644 --- a/src-tauri/crates/berd-call/src/main.rs +++ b/src-tauri/crates/berd-call/src/main.rs @@ -2917,6 +2917,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 +4021,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") } }, @@ -5554,6 +5556,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 +5740,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 +5927,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 +5939,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") } } @@ -10295,6 +10304,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(); From b47ba813c84c3b52e60960b6d9041dddb1c6145b Mon Sep 17 00:00:00 2001 From: Henry Date: Thu, 22 Oct 2026 15:42:26 -0400 Subject: [PATCH 02/13] test(berd-call): cover a cancel wake racing the worker's flag check The playback worker checks the cancellation flag under the state lock and then waits on the condvar. A cancel request that stores the flag and notifies between those two steps is lost, so the worker sleeps until its operation timeout. This reproduces the CI timeout in autonomous_spokesperson_cancel_is_correlated_before_audio_cancellation_and_then_stale. Signed-off-by: Henry Co-Authored-By: Claude Opus 5.5 (cherry picked from commit f6097e517a72726dd91a09d1428b02f640a9a8ff) --- .../crates/berd-call/src/session_audio.rs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/src-tauri/crates/berd-call/src/session_audio.rs b/src-tauri/crates/berd-call/src/session_audio.rs index f6c33450f..ef16ce9af 100644 --- a/src-tauri/crates/berd-call/src/session_audio.rs +++ b/src-tauri/crates/berd-call/src/session_audio.rs @@ -1717,6 +1717,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(); From f12ef5f685b3b227b5377d2a8ad9691788b17434 Mon Sep 17 00:00:00 2001 From: Henry Date: Thu, 22 Oct 2026 15:46:32 -0400 Subject: [PATCH 03/13] fix(berd-call): order cancel wakes after the worker's flag check notify_cancel_requested stored the flag and notified without the state lock, so a worker that had checked the flag but not yet waited missed the wake and slept until its operation timeout. Taking the lock before notifying orders the wake after any such check. Signed-off-by: Henry Co-Authored-By: Claude Opus 5.5 (cherry picked from commit 14943c91e2c8a39ead0a4575b87f5ba2f30c1682) --- src-tauri/crates/berd-call/src/session_audio.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src-tauri/crates/berd-call/src/session_audio.rs b/src-tauri/crates/berd-call/src/session_audio.rs index ef16ce9af..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(); } From 8259c7d9adf632cd4f5233582a53622e0b3e90b2 Mon Sep 17 00:00:00 2001 From: Zack <04f47b290b155b3b8b782b11501cdb15f1e21e82dfceefd01ff0ed405459d927@buzz.block.builderlab.xyz> Date: Thu, 22 Oct 2026 13:26:58 -0400 Subject: [PATCH 04/13] fix(berd-call): preserve ordered handoff terminal failures Signed-off-by: Zack <04f47b290b155b3b8b782b11501cdb15f1e21e82dfceefd01ff0ed405459d927@buzz.block.builderlab.xyz> (cherry picked from commit 6b5afdf000b6a776d2253948789df6a7b4d407df) --- src-tauri/crates/berd-call/src/main.rs | 13 +- .../berd-call/src/openai_spokesperson.rs | 164 +++++++++++++++--- 2 files changed, 143 insertions(+), 34 deletions(-) diff --git a/src-tauri/crates/berd-call/src/main.rs b/src-tauri/crates/berd-call/src/main.rs index 5c2379631..ed16c99f6 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; @@ -4228,14 +4227,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 { 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(); From 1c6f70880a161481d8b1468c428c96bda5234c6c Mon Sep 17 00:00:00 2001 From: Zack <04f47b290b155b3b8b782b11501cdb15f1e21e82dfceefd01ff0ed405459d927@buzz.block.builderlab.xyz> Date: Thu, 22 Oct 2026 13:41:00 -0400 Subject: [PATCH 05/13] test(berd-call): tolerate audio replies after child shutdown Signed-off-by: Zack <04f47b290b155b3b8b782b11501cdb15f1e21e82dfceefd01ff0ed405459d927@buzz.block.builderlab.xyz> (cherry picked from commit 46fdf603bdc8ff64ee0712f2c178b397a58544de) --- .../berd-call/tests/session_protocol.rs | 121 ++++++++++++++---- 1 file changed, 94 insertions(+), 27 deletions(-) diff --git a/src-tauri/crates/berd-call/tests/session_protocol.rs b/src-tauri/crates/berd-call/tests/session_protocol.rs index 39e4ead87..2b090e6fd 100644 --- a/src-tauri/crates/berd-call/tests/session_protocol.rs +++ b/src-tauri/crates/berd-call/tests/session_protocol.rs @@ -203,13 +203,29 @@ impl ExpertSpokespersonTestSession { } 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) { @@ -339,9 +355,9 @@ fn spawn_audio_host( 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, @@ -385,25 +401,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; } - writer.flush().unwrap(); if reached_limit { if let Some(ready) = played_ready.take() { let _ = ready.send(()); @@ -435,11 +446,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; + } continue; } let played_frames = current @@ -457,12 +471,65 @@ 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; + } } }) } +#[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, + ); + assert!( + host.join().is_ok(), + "late fixture reply panicked after child shutdown" + ); +} + #[test] fn session_rejects_a_read_only_pcm_descriptor_before_hello() { let mut descriptors = [-1; 2]; From f5278816a016a2ba17b38374905c6fe96ba4858b Mon Sep 17 00:00:00 2001 From: Zack <04f47b290b155b3b8b782b11501cdb15f1e21e82dfceefd01ff0ed405459d927@buzz.block.builderlab.xyz> Date: Thu, 22 Oct 2026 14:35:34 -0400 Subject: [PATCH 06/13] test(berd-call): validate partial audio EOF after intentional disconnect Signed-off-by: Zack <04f47b290b155b3b8b782b11501cdb15f1e21e82dfceefd01ff0ed405459d927@buzz.block.builderlab.xyz> (cherry picked from commit 71214694e4d42b81c57f7a8e1ad235380bd85465) --- .../berd-call/tests/session_protocol.rs | 101 ++++++++++++++---- 1 file changed, 83 insertions(+), 18 deletions(-) diff --git a/src-tauri/crates/berd-call/tests/session_protocol.rs b/src-tauri/crates/berd-call/tests/session_protocol.rs index 2b090e6fd..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,12 +192,24 @@ 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()) } } @@ -351,7 +363,7 @@ 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) } @@ -362,7 +374,7 @@ fn spawn_audio_host_with_played_limit( 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. @@ -371,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); @@ -413,7 +425,7 @@ fn spawn_audio_host_with_played_limit( 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; + return Ok(()); } if reached_limit { if let Some(ready) = played_ready.take() { @@ -434,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; } @@ -452,7 +464,7 @@ fn spawn_audio_host_with_played_limit( json!({"type":"audio_cancelled","speech_id":speech_id,"played_frames":played_frames}), ], ) { - return; + return Ok(()); } continue; } @@ -472,12 +484,66 @@ fn spawn_audio_host_with_played_limit( }; let mut writer = stdin.lock().unwrap(); if !write_audio_replies(&mut *writer, &[message]) { - return; + 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); @@ -524,10 +590,9 @@ fn audio_fixture_can_finish_a_queued_reply_after_child_input_closes() { None, false, ); - assert!( - host.join().is_ok(), - "late fixture reply panicked after child shutdown" - ); + host.join() + .expect("late fixture reply panicked after child shutdown") + .unwrap(); } #[test] @@ -1013,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")); @@ -3331,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] @@ -3392,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. From bd1d4f5b03b755d0cf7a68c2923ab91230e7932b Mon Sep 17 00:00:00 2001 From: Aaron Date: Fri, 23 Oct 2026 09:36:23 -0400 Subject: [PATCH 07/13] fix(berd-call): expire speech holds without cancelling recognition The idle timeout after local speech ended used to cancel the native recognizer and start a new generation, discarding Apple's queued audio and any final still on its way. A slow recognition therefore lost its transcript, and a final the recognizer had already produced could not release a newer hold. Expiry now releases only the agent speech admission hold. Recognition, its queued audio and the recognizer generation continue, so a late final is still forwarded. Native finals have no utterance identity, so hold credits are tracked separately from unresolved recognitions: an expired credit is consumed before a newer blocking one, and a final for an old utterance cannot release a newer hold. Credit accounting is capped at 1024 and stays ambiguous until the recognizer resets. The pending-utterance timeout is the existing fixed five seconds. Ported from 2eda9918c in #426 without the detection settings, CLI flag and protocol field it added there. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Aaron --- src-tauri/crates/berd-call/PROTOCOL.md | 11 ++ src-tauri/crates/berd-call/README.md | 2 +- src-tauri/crates/berd-call/src/input.rs | 236 +++++++++++++++++------- 3 files changed, 185 insertions(+), 64 deletions(-) diff --git a/src-tauri/crates/berd-call/PROTOCOL.md b/src-tauri/crates/berd-call/PROTOCOL.md index d38454cab..7d7242e20 100644 --- a/src-tauri/crates/berd-call/PROTOCOL.md +++ b/src-tauri/crates/berd-call/PROTOCOL.md @@ -301,6 +301,17 @@ 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. 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 a final first resolves the oldest hold credit, which may be an expired +one. 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`. diff --git a/src-tauri/crates/berd-call/README.md b/src-tauri/crates/berd-call/README.md index 819d49db5..41313d086 100644 --- a/src-tauri/crates/berd-call/README.md +++ b/src-tauri/crates/berd-call/README.md @@ -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 diff --git a/src-tauri/crates/berd-call/src/input.rs b/src-tauri/crates/berd-call/src/input.rs index 3583e74c7..1f407dd45 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,6 +122,7 @@ impl FinalTranscriptStorageReceipt { pub enum VoiceInputEvent { Ready, SpeakingChanged(bool), + /// Agent speech admission hold, which may expire before recognition finishes. RecognitionPendingChanged(bool), FinalTranscript { text: String, @@ -727,18 +728,41 @@ 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; + +#[derive(Default)] struct PendingRecognitions { count: usize, + blocking: usize, + ambiguous: bool, } 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; + } + 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 +770,42 @@ impl PendingRecognitions { Ok(()) } + 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(|_| ())?; + 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 = Self::new(); + Ok(()) + } } fn effective_mute_epoch( @@ -1452,7 +1490,7 @@ fn forward_mac_events( send_final(events, text, storage_deadline)?; } pending.resolve(events)?; - if pending.count == 0 { + if pending.blocking == 0 { *settle_deadline = None; } } @@ -1488,7 +1526,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 +1536,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 +1549,7 @@ 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) } #[cfg(target_os = "macos")] @@ -1571,27 +1610,12 @@ 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; } + settle_deadline = None; } let frame = match frames.recv_timeout(Duration::from_millis(50)) { Ok(frame) => Some(frame), @@ -1664,7 +1688,12 @@ 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()); + end_mac_turn( + &pending, + &mut settle_deadline, + Instant::now(), + MAC_PENDING_UTTERANCE_TIMEOUT, + ); let _ = events.blocking_send(VoiceInputEvent::SpeakingChanged(false)); } } @@ -3441,7 +3470,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 +3651,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, @@ -3745,8 +3782,12 @@ mod tests { .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 +3810,125 @@ 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(), + )) + .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_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(); - end_mac_turn(&pending, &mut deadline, now); + 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 + MAC_LIVE_NO_RESULT_TIMEOUT + now + Duration::from_millis(750) )); - - let (old_tx, old_rx) = tokio_mpsc::unbounded_channel(); - drop(old_rx); - pending.reset(&event_tx).unwrap(); - deadline = None; + 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!(old_tx + 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(), )) - .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(), )) .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] From 22523930b81ed6ebb2823908cc9acf11ca0a9cbd Mon Sep 17 00:00:00 2001 From: Aaron Date: Fri, 23 Oct 2026 09:37:40 -0400 Subject: [PATCH 08/13] fix(berd-call): release the speech hold when a final covers its turns Each VAD turn adds a pending recognition credit, but SpeechTranscriber emits one final for a sentence that pauses split into several turns. Resolving one credit per final left the rest outstanding, so in John's recent calls 55 of 57 holds ran the full pending-utterance timeout even though Apple had finished 0 to 4 seconds after he stopped. Pass each final's audio end through the native bridge and resolve exactly the ended turns whose speech it covers. A final for an earlier phrase of a still-pending turn resolves nothing, so the rest of the sentence keeps the hold. Finals without an end keep the old behaviour. Ported from 43efb0e00 in #426 without its capture hooks. A turn's end is the pushed audio minus the host's silence flush window, since the host declares speech over after that much silence. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Aaron --- src-tauri/crates/berd-call/PROTOCOL.md | 6 +- src-tauri/crates/berd-call/src/input.rs | 131 +++++++++++++++++- src-tauri/crates/berd-call/src/mac_speech.rs | 86 ++++++++++-- .../BerdMacSpeechBridge.swift | 16 ++- 4 files changed, 216 insertions(+), 23 deletions(-) diff --git a/src-tauri/crates/berd-call/PROTOCOL.md b/src-tauri/crates/berd-call/PROTOCOL.md index 7d7242e20..b14f65e97 100644 --- a/src-tauri/crates/berd-call/PROTOCOL.md +++ b/src-tauri/crates/berd-call/PROTOCOL.md @@ -307,8 +307,10 @@ expires after five seconds without a final. 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 a final first resolves the oldest hold credit, which may be an expired -one. Credit accounting is capped at 1024; at the cap attribution stays ambiguous +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. diff --git a/src-tauri/crates/berd-call/src/input.rs b/src-tauri/crates/berd-call/src/input.rs index 1f407dd45..fec02153a 100644 --- a/src-tauri/crates/berd-call/src/input.rs +++ b/src-tauri/crates/berd-call/src/input.rs @@ -734,11 +734,18 @@ fn reap_dropped_worker(worker: thread::JoinHandle<()>) { #[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; + #[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>, } impl PendingRecognitions { @@ -759,6 +766,7 @@ impl PendingRecognitions { 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); @@ -770,6 +778,23 @@ impl PendingRecognitions { Ok(()) } + #[cfg(any(target_os = "macos", test))] + fn end_turn(&mut self, speech_end: u64) { + if let Some(end @ None) = self.ends.back_mut() { + *end = Some(speech_end); + } + } + + /// 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() + } + fn result_is_current(&self) -> bool { !self.ambiguous && self.count == self.blocking } @@ -780,6 +805,7 @@ impl PendingRecognitions { } let current = self.result_is_current(); self.count -= 1; + self.ends.pop_front(); if current { self.blocking -= 1; if self.blocking == 0 { @@ -1484,12 +1510,17 @@ 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)?; + // 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. + for _ in 0..end.map_or(1, |end| pending.covered(end)) { + pending.resolve(events)?; + } if pending.blocking == 0 { *settle_deadline = None; } @@ -1593,6 +1624,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; } @@ -1643,6 +1676,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)); @@ -1666,6 +1700,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 { @@ -1688,6 +1723,13 @@ fn mac_speech_worker( if silence_frames >= SILENCE_FLUSH_FRAMES { silence_frames = 0; in_speech = false; + // Pushed audio runs slightly ahead of the detector, which + // only delays coverage. + pending.end_turn( + recognized.saturating_sub( + (SILENCE_FLUSH_FRAMES * VAD_FRAME_SAMPLES * 3) as u64, + ), + ); end_mac_turn( &pending, &mut settle_deadline, @@ -3778,6 +3820,7 @@ mod tests { recognition_tx .send(crate::mac_speech::MacSpeechRecognitionEvent::Final( text.to_string(), + None, )) .unwrap(); } @@ -3839,6 +3882,7 @@ mod tests { 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(); @@ -3846,6 +3890,87 @@ mod tests { assert!(deadline.is_some()); } + #[cfg(target_os = "macos")] + #[test] + 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 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(end); + 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(); + // 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(200_000); + 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()); + } + + #[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(10_000); + pending.begin(&event_tx).unwrap(); + assert_eq!(pending.covered(1_000_000), 1); + pending.end_turn(50_000); + pending.end_turn(1); + 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_expiry_keeps_late_finals_without_clearing_a_new_hold() { @@ -3874,6 +3999,7 @@ mod tests { native_tx .send(crate::mac_speech::MacSpeechRecognitionEvent::Final( "late old final".into(), + None, )) .unwrap(); forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); @@ -3896,6 +4022,7 @@ mod tests { native_tx .send(crate::mac_speech::MacSpeechRecognitionEvent::Final( "new final".into(), + None, )) .unwrap(); forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap(); diff --git a/src-tauri/crates/berd-call/src/mac_speech.rs b/src-tauri/crates/berd-call/src/mac_speech.rs index def4ca5e5..93b7b3d4c 100644 --- a/src-tauri/crates/berd-call/src/mac_speech.rs +++ b/src-tauri/crates/berd-call/src/mac_speech.rs @@ -23,7 +23,8 @@ 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), Finished, Failed(String), } @@ -43,7 +44,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,9 +157,17 @@ 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 { match code { - 1 => Some(MacSpeechRecognitionEvent::Final(text.unwrap_or_default())), + 1 => Some(MacSpeechRecognitionEvent::Final( + text.unwrap_or_default(), + (end_seconds.is_finite() && end_seconds >= 0.0) + .then(|| (end_seconds * 48_000.0).round() as u64), + )), 2 => Some(MacSpeechRecognitionEvent::Finished), 3 => Some(MacSpeechRecognitionEvent::Failed(text.unwrap_or_else( || "macOS speech recognition failed.".to_string(), @@ -167,14 +176,19 @@ fn map_recognition_event(code: i32, text: Option) -> Option() }; 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 +300,27 @@ 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(99, None, f64::NAN), None); } #[test] @@ -319,13 +340,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/swift/BerdMacSpeechBridge/Sources/BerdMacSpeechBridge/BerdMacSpeechBridge.swift b/src-tauri/crates/berd-call/swift/BerdMacSpeechBridge/Sources/BerdMacSpeechBridge/BerdMacSpeechBridge.swift index e84f33489..02c0dc8e7 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 @@ -8,8 +8,9 @@ private let finalEvent: Int32 = 1 private let finishedEvent: Int32 = 2 private let failedEvent: Int32 = 3 +// The Double is where a final'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? @@ -308,7 +309,8 @@ private final class SpeechSession: @unchecked Sendable { .trimmingCharacters(in: .whitespacesAndNewlines) guard !text.isEmpty else { continue } guard result.isFinal else { continue } - self.emit(finalEvent, text: text) + let end = result.range.end + self.emit(finalEvent, text: text, end: end.isNumeric ? end.seconds : .nan) } self?.taskCompleted() } catch is CancellationError { self?.taskCompleted() } @@ -417,11 +419,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 +431,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 +446,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() } From ec237b0cac7f542ffe1e4f8adc126f6614d83d4b Mon Sep 17 00:00:00 2001 From: Aaron Date: Fri, 23 Oct 2026 09:38:41 -0400 Subject: [PATCH 09/13] fix(berd-call): keep the speech hold while Apple still has words to finalize The hold lifted at the pending-utterance timeout even when Apple was still transcribing. On a live call a second sentence took about seven seconds to finalize, so an agent's reply to the first sentence started playing and was then cut off by the late transcript. The bridge now reports Apple's interim results. While Apple's latest result is interim words, the hold lasts up to ten seconds beyond the timeout, until a final follows them; noise with no interim words keeps the timeout. Apple re-reports words a final leaves out as a fresh interim result, and interim results end where its audio does rather than at the last word, so the hold follows result order, not sample positions. Empty finals now reach the host so they settle their audio. Ported from 09ee43fd9, e173655ac and 4d2f03488 in #426 as their final combined behavior. The ten-second grace is a fixed constant, and the capture-only `heard` and `settled` records are left out with the capture infrastructure. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Aaron --- src-tauri/crates/berd-call/PROTOCOL.md | 5 +- src-tauri/crates/berd-call/src/input.rs | 97 ++++++++++++++++++- src-tauri/crates/berd-call/src/mac_speech.rs | 13 ++- .../BerdMacSpeechBridge.swift | 17 ++-- 4 files changed, 122 insertions(+), 10 deletions(-) diff --git a/src-tauri/crates/berd-call/PROTOCOL.md b/src-tauri/crates/berd-call/PROTOCOL.md index b14f65e97..6ed3ae561 100644 --- a/src-tauri/crates/berd-call/PROTOCOL.md +++ b/src-tauri/crates/berd-call/PROTOCOL.md @@ -303,7 +303,10 @@ The child emits: 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. Expiry emits +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 diff --git a/src-tauri/crates/berd-call/src/input.rs b/src-tauri/crates/berd-call/src/input.rs index fec02153a..db3d1fb56 100644 --- a/src-tauri/crates/berd-call/src/input.rs +++ b/src-tauri/crates/berd-call/src/input.rs @@ -739,6 +739,11 @@ const MAX_PENDING_RECOGNITIONS: usize = 1024; #[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, @@ -746,6 +751,11 @@ struct PendingRecognitions { ambiguous: bool, /// Where each unresolved turn's speech ended, oldest first; None while it continues. ends: VecDeque>, + /// 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, } impl PendingRecognitions { @@ -795,6 +805,18 @@ impl PendingRecognitions { .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; + } + fn result_is_current(&self) -> bool { !self.ambiguous && self.count == self.blocking } @@ -1518,6 +1540,7 @@ fn forward_mac_events( // 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; for _ in 0..end.map_or(1, |end| pending.covered(end)) { pending.resolve(events)?; } @@ -1525,6 +1548,7 @@ fn forward_mac_events( *settle_deadline = None; } } + crate::mac_speech::MacSpeechRecognitionEvent::Heard(_) => pending.unfinalized = true, crate::mac_speech::MacSpeechRecognitionEvent::Finished => { if storage_deadline.is_none() { pending.reset(events)?; @@ -1580,7 +1604,10 @@ fn mac_settle_expired( settle_deadline: Option, now: Instant, ) -> bool { - pending.blocking > 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")] @@ -1648,6 +1675,7 @@ fn mac_speech_worker( if pending.release_hold(&events).is_err() { return; } + pending.abandon_heard_words(); settle_deadline = None; } let frame = match frames.recv_timeout(Duration::from_millis(50)) { @@ -3971,6 +3999,73 @@ mod tests { 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(end); + 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)).unwrap(); + native_tx + .send(Final("first".into(), Some(110_000))) + .unwrap(); + native_tx.send(Heard(310_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)); + assert!(mac_settle_expired( + &pending, + deadline, + 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); + } + + #[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(end); + 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)).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() { diff --git a/src-tauri/crates/berd-call/src/mac_speech.rs b/src-tauri/crates/berd-call/src/mac_speech.rs index 93b7b3d4c..8f9255e24 100644 --- a/src-tauri/crates/berd-call/src/mac_speech.rs +++ b/src-tauri/crates/berd-call/src/mac_speech.rs @@ -25,6 +25,8 @@ pub struct MacSpeechEngineStatus { pub enum MacSpeechRecognitionEvent { /// A final result and where its audio ends, in 48 kHz samples of pushed input. Final(String, Option), + /// Words still being transcribed, heard through this 48 kHz sample. + Heard(u64), Finished, Failed(String), } @@ -162,16 +164,18 @@ fn map_recognition_event( 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(), - (end_seconds.is_finite() && end_seconds >= 0.0) - .then(|| (end_seconds * 48_000.0).round() as u64), + end, )), 2 => Some(MacSpeechRecognitionEvent::Finished), 3 => Some(MacSpeechRecognitionEvent::Failed(text.unwrap_or_else( || "macOS speech recognition failed.".to_string(), ))), + 4 => end.map(MacSpeechRecognitionEvent::Heard), _ => None, } } @@ -320,6 +324,11 @@ mod tests { "macOS speech recognition failed.".to_string() )) ); + assert_eq!( + map_recognition_event(4, None, 1.5), + Some(MacSpeechRecognitionEvent::Heard(72_000)) + ); + assert_eq!(map_recognition_event(4, None, f64::NAN), None); assert_eq!(map_recognition_event(99, None, f64::NAN), None); } 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 02c0dc8e7..13f2e5c8d 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,8 +7,10 @@ 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 audio ends, in seconds of pushed input, or NaN. +// 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?, Double, UnsafeMutableRawPointer? ) -> Void @@ -131,7 +133,7 @@ private func speechTranscriber(for locale: Locale) -> SpeechTranscriber { SpeechTranscriber( locale: locale, transcriptionOptions: [], - reportingOptions: [], + reportingOptions: [.volatileResults], attributeOptions: [.audioTimeRange] ) } @@ -307,10 +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 } - let end = result.range.end - self.emit(finalEvent, text: text, end: end.isNumeric ? end.seconds : .nan) + 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, end: end) + } } self?.taskCompleted() } catch is CancellationError { self?.taskCompleted() } From 5f53d28b7c9791899c17429c5f6391e340b69f49 Mon Sep 17 00:00:00 2001 From: Aaron Date: Fri, 23 Oct 2026 09:40:06 -0400 Subject: [PATCH 10/13] feat(berd-call): stream the user's words while they are transcribed Hosts could only show what the user said once Apple finalized it, often seconds after they stopped talking. Conventional sessions on macOS now emit `user_interim` with Apple's interim text, replaced by each update and cleared when its final arrives, on mute, input reset or restart. `--stream` carries it as an ephemeral `0\tuser_interim` row whose JSON gives the text and the cursor the next `user` row will take. A client can show the words as a caption, or let an agent start on them and keep its reply current when the final lands. The row never advances a cursor or enters transcript history, and Codex tasks never see it. Ported from 7b28c3ecd in #428, re-applied on the speech-hold commits without the capture trace calls that branch carried. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Aaron --- src-tauri/crates/berd-call/PROTOCOL.md | 12 ++ src-tauri/crates/berd-call/README.md | 2 + .../crates/berd-call/src/benchmark/stt.rs | 1 + .../crates/berd-call/src/host_session.rs | 113 ++++++++++++++++++ src-tauri/crates/berd-call/src/input.rs | 30 ++++- src-tauri/crates/berd-call/src/mac_speech.rs | 10 +- src-tauri/crates/berd-call/src/main.rs | 44 +++++++ src-tauri/crates/berd-call/src/protocol.rs | 8 ++ .../BerdMacSpeechBridge.swift | 2 +- 9 files changed, 212 insertions(+), 10 deletions(-) diff --git a/src-tauri/crates/berd-call/PROTOCOL.md b/src-tauri/crates/berd-call/PROTOCOL.md index 6ed3ae561..078dad8fa 100644 --- a/src-tauri/crates/berd-call/PROTOCOL.md +++ b/src-tauri/crates/berd-call/PROTOCOL.md @@ -324,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: diff --git a/src-tauri/crates/berd-call/README.md b/src-tauri/crates/berd-call/README.md index 41313d086..d1397d3d5 100644 --- a/src-tauri/crates/berd-call/README.md +++ b/src-tauri/crates/berd-call/README.md @@ -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 db3d1fb56..2f9b7fa6f 100644 --- a/src-tauri/crates/berd-call/src/input.rs +++ b/src-tauri/crates/berd-call/src/input.rs @@ -124,6 +124,9 @@ pub enum VoiceInputEvent { 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, @@ -756,6 +759,8 @@ struct PendingRecognitions { /// 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 { @@ -817,6 +822,14 @@ impl PendingRecognitions { 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 } @@ -851,6 +864,7 @@ impl PendingRecognitions { fn reset(&mut self, events: &tokio_mpsc::Sender) -> Result<(), ()> { self.release_hold(events)?; + self.show_interim(events, ""); *self = Self::new(); Ok(()) } @@ -1547,8 +1561,14 @@ fn forward_mac_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::Heard(_) => pending.unfinalized = true, crate::mac_speech::MacSpeechRecognitionEvent::Finished => { if storage_deadline.is_none() { pending.reset(events)?; @@ -1676,6 +1696,7 @@ fn mac_speech_worker( return; } pending.abandon_heard_words(); + pending.show_interim(&events, ""); settle_deadline = None; } let frame = match frames.recv_timeout(Duration::from_millis(50)) { @@ -2836,6 +2857,7 @@ mod tests { storage_receipt.stored(); observed.push("final"); } + VoiceInputEvent::InterimTranscript(_) => {} VoiceInputEvent::Failed(error) => panic!("unexpected input failure: {error}"), } } @@ -4016,11 +4038,11 @@ mod tests { 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)).unwrap(); + 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)).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)); @@ -4057,7 +4079,7 @@ mod tests { } let (native_tx, mut native_rx) = tokio_mpsc::unbounded_channel(); use crate::mac_speech::MacSpeechRecognitionEvent::{Final, Heard}; - native_tx.send(Heard(130_000)).unwrap(); + native_tx.send(Heard(130_000, "words".into())).unwrap(); native_tx .send(Final("words".into(), Some(110_000))) .unwrap(); diff --git a/src-tauri/crates/berd-call/src/mac_speech.rs b/src-tauri/crates/berd-call/src/mac_speech.rs index 8f9255e24..2985c454f 100644 --- a/src-tauri/crates/berd-call/src/mac_speech.rs +++ b/src-tauri/crates/berd-call/src/mac_speech.rs @@ -25,8 +25,8 @@ pub struct MacSpeechEngineStatus { pub enum MacSpeechRecognitionEvent { /// A final result and where its audio ends, in 48 kHz samples of pushed input. Final(String, Option), - /// Words still being transcribed, heard through this 48 kHz sample. - Heard(u64), + /// Interim words still being transcribed, heard through this 48 kHz sample. + Heard(u64, String), Finished, Failed(String), } @@ -175,7 +175,7 @@ fn map_recognition_event( 3 => Some(MacSpeechRecognitionEvent::Failed(text.unwrap_or_else( || "macOS speech recognition failed.".to_string(), ))), - 4 => end.map(MacSpeechRecognitionEvent::Heard), + 4 => end.map(|end| MacSpeechRecognitionEvent::Heard(end, text.unwrap_or_default())), _ => None, } } @@ -325,8 +325,8 @@ mod tests { )) ); assert_eq!( - map_recognition_event(4, None, 1.5), - Some(MacSpeechRecognitionEvent::Heard(72_000)) + 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); diff --git a/src-tauri/crates/berd-call/src/main.rs b/src-tauri/crates/berd-call/src/main.rs index ed16c99f6..6f508a945 100644 --- a/src-tauri/crates/berd-call/src/main.rs +++ b/src-tauri/crates/berd-call/src/main.rs @@ -7857,6 +7857,13 @@ fn handle_voice_input_event( )?; update_provisional_suspension(core, active)?; } + VoiceInputEvent::InterimTranscript(text) => write_message( + writer, + &SessionMessage::UserInterim { + token: *next_token, + text, + }, + )?, VoiceInputEvent::FinalTranscript { text, storage_receipt, @@ -13095,6 +13102,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/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/swift/BerdMacSpeechBridge/Sources/BerdMacSpeechBridge/BerdMacSpeechBridge.swift b/src-tauri/crates/berd-call/swift/BerdMacSpeechBridge/Sources/BerdMacSpeechBridge/BerdMacSpeechBridge.swift index 13f2e5c8d..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 @@ -314,7 +314,7 @@ private final class SpeechSession: @unchecked Sendable { // 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, end: end) + self.emit(heardEvent, text: text, end: end) } } self?.taskCompleted() From 4dda673bf63ca42b7ca0c4a926cb4857c4db91c3 Mon Sep 17 00:00:00 2001 From: Henry Date: Fri, 23 Oct 2026 07:42:20 -0400 Subject: [PATCH 11/13] fix(berd): ignore interim words in native voice conversations Berd's native voice handler matches every voice input event, so the new interim transcript event broke its build. Berd doesn't show interim words yet. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Henry (cherry picked from commit e347e12a6b8bee0ce2b4204cf1e06dadd2f59229) --- src-tauri/src/commands/native_voice.rs | 2 ++ 1 file changed, 2 insertions(+) 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, From e2b9c156aa182c4ba105fcbc8ad0137096e006e3 Mon Sep 17 00:00:00 2001 From: Aaron Date: Fri, 23 Oct 2026 10:05:42 -0400 Subject: [PATCH 12/13] fix(berd-call): use one VAD threshold for listening and interruption The call detected the user's speech at 0.5 while listening and 0.65 during agent playback. One threshold is simpler to reason about and to tune, so both now use 0.9, the value John converged on in live calls. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Aaron --- src-tauri/crates/berd-call/PROTOCOL.md | 2 +- src-tauri/crates/berd-call/README.md | 2 +- src-tauri/crates/berd-call/src/main.rs | 14 ++++++++------ 3 files changed, 10 insertions(+), 8 deletions(-) diff --git a/src-tauri/crates/berd-call/PROTOCOL.md b/src-tauri/crates/berd-call/PROTOCOL.md index 078dad8fa..66809f1f9 100644 --- a/src-tauri/crates/berd-call/PROTOCOL.md +++ b/src-tauri/crates/berd-call/PROTOCOL.md @@ -383,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 d1397d3d5..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. diff --git a/src-tauri/crates/berd-call/src/main.rs b/src-tauri/crates/berd-call/src/main.rs index 6f508a945..69f6a90fc 100644 --- a/src-tauri/crates/berd-call/src/main.rs +++ b/src-tauri/crates/berd-call/src/main.rs @@ -95,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; @@ -6200,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, @@ -7052,7 +7054,7 @@ fn stt_benchmark_target(config: &SttBackendConfig) -> Result Result Result Date: Fri, 23 Oct 2026 11:32:20 -0400 Subject: [PATCH 13/13] fix(berd-call): release a turn whose final arrived before it ended Apple can finalize a complete utterance during the 1.2-second silence window, before the detector records the turn's end. That final could not cover a turn still in progress, and its coverage was then lost, so the hold waited the full five-second timeout. Remember the furthest final and resolve the turns it covers when a turn ends. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Henry --- src-tauri/crates/berd-call/src/input.rs | 72 ++++++++++++++++++++----- 1 file changed, 59 insertions(+), 13 deletions(-) diff --git a/src-tauri/crates/berd-call/src/input.rs b/src-tauri/crates/berd-call/src/input.rs index 2f9b7fa6f..1bf58849a 100644 --- a/src-tauri/crates/berd-call/src/input.rs +++ b/src-tauri/crates/berd-call/src/input.rs @@ -754,6 +754,9 @@ struct PendingRecognitions { 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. @@ -794,10 +797,18 @@ impl PendingRecognitions { } #[cfg(any(target_os = "macos", test))] - fn end_turn(&mut self, speech_end: u64) { + 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. @@ -1555,6 +1566,7 @@ fn forward_mac_events( // 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)?; } @@ -1774,11 +1786,17 @@ fn mac_speech_worker( in_speech = false; // Pushed audio runs slightly ahead of the detector, which // only delays coverage. - pending.end_turn( - recognized.saturating_sub( - (SILENCE_FLUSH_FRAMES * VAD_FRAME_SAMPLES * 3) as u64, - ), - ); + 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, @@ -3950,7 +3968,7 @@ mod tests { // 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(end); + 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(); @@ -3987,7 +4005,7 @@ mod tests { // 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(200_000); + pending.end_turn(&event_tx, 200_000).unwrap(); end_mac_turn( &pending, &mut deadline, @@ -4005,16 +4023,44 @@ mod tests { 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(10_000); + pending.end_turn(&event_tx, 10_000).unwrap(); pending.begin(&event_tx).unwrap(); assert_eq!(pending.covered(1_000_000), 1); - pending.end_turn(50_000); - pending.end_turn(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(); @@ -4032,7 +4078,7 @@ mod tests { // 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(end); + 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(); @@ -4074,7 +4120,7 @@ mod tests { // 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(end); + 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();