diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index 6e35ac8bb..7ce8c3c52 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -9,9 +9,9 @@ dependencies = [ "anyhow", "audioadapter-buffers", "base64 0.22.1", + "berd-voice", "block2", "builderbot-auth", - "berd-voice", "bytes", "bzip2 0.6.1", "cc", @@ -34,6 +34,7 @@ dependencies = [ "nucleo-matcher", "objc2", "objc2-app-kit", + "objc2-avf-audio", "objc2-foundation", "objc2-user-notifications", "percent-encoding", @@ -49,6 +50,7 @@ dependencies = [ "sherpa-onnx", "sqlx", "ssstretch", + "swift-rs", "sysinfo", "tar", "tauri", @@ -580,6 +582,20 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a8241f3ebb85c056b509d4327ad0358fbbba6ffb340bf388f26350aeda225b1" +[[package]] +name = "berd-voice" +version = "0.1.0" +dependencies = [ + "ort", + "ort-sys", + "rand 0.10.2", + "sentencepiece-model", + "serde", + "serde_json", + "sherpa-onnx", + "tokenizers", +] + [[package]] name = "berdctl" version = "0.6.2" @@ -713,20 +729,6 @@ version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" -[[package]] -name = "berd-voice" -version = "0.1.0" -dependencies = [ - "ort", - "ort-sys", - "rand 0.10.2", - "sentencepiece-model", - "serde", - "serde_json", - "sherpa-onnx", - "tokenizers", -] - [[package]] name = "bytemuck" version = "1.25.2" @@ -3932,6 +3934,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6948501a91121d6399b79abaa33a8aa4ea7857fe019f341b8c23ad6e81b79b08" dependencies = [ "bitflags 2.13.1", + "block2", "libc", "objc2", "objc2-core-audio", @@ -3946,7 +3949,11 @@ version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13a380031deed8e99db00065c45937da434ca987c034e13b87e4441f9e4090be" dependencies = [ + "bitflags 2.13.1", + "block2", "objc2", + "objc2-audio-toolbox", + "objc2-core-audio-types", "objc2-foundation", ] diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index ccc523bca..4c57c6e2d 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -21,6 +21,9 @@ exclude = ["plugins/app-test-driver"] tauri-build = { version = "2", features = [] } cc = "1" +[target.'cfg(target_os = "macos")'.build-dependencies] +swift-rs = { version = "1.0.7", features = ["build"] } + [dependencies] anyhow = "1" base64 = "0.22" @@ -109,6 +112,7 @@ windows-sys = { version = "0.59", features = [ block2 = "0.6" objc2 = "0.6" objc2-app-kit = { version = "0.3.2", features = ["NSApplication", "NSImage", "NSMenu", "NSMenuItem", "NSAlert", "NSButton", "NSControl", "NSCell", "NSResponder", "NSView"] } +objc2-avf-audio = { version = "0.3.2", features = ["AVAudioApplication", "block2"] } objc2-foundation = { version = "0.3.2", features = ["NSDictionary", "NSError", "NSFileManager", "NSObject", "NSProcessInfo", "NSString", "NSURL"] } objc2-user-notifications = "0.3.2" rodio = { version = "0.22", default-features = false, features = ["playback"] } diff --git a/src-tauri/build.rs b/src-tauri/build.rs index 69ecf7ee2..121ff7745 100644 --- a/src-tauri/build.rs +++ b/src-tauri/build.rs @@ -18,7 +18,11 @@ fn main() { for framework in ["Foundation", "AVFoundation", "AudioToolbox", "CoreAudio"] { println!("cargo:rustc-link-lib=framework={framework}"); } + if std::env::var("CARGO_CFG_TARGET_OS").as_deref() == Ok("macos") { + swift_rs::SwiftLinker::new("14.0") + .with_package("BerdAirPodsBridge", "swift/BerdAirPodsBridge") + .link(); + } } - tauri_build::build() } diff --git a/src-tauri/src/commands/mod.rs b/src-tauri/src/commands/mod.rs index 465a05e20..485664fe5 100644 --- a/src-tauri/src/commands/mod.rs +++ b/src-tauri/src/commands/mod.rs @@ -31,6 +31,7 @@ pub mod local_mcp_inventory; pub mod message_queues; pub mod migration; pub mod model_setup; +mod native_input_mute; pub mod native_voice; pub mod notifications; #[cfg(feature = "block-voice-dictation")] diff --git a/src-tauri/src/commands/native_input_mute.rs b/src-tauri/src/commands/native_input_mute.rs new file mode 100644 index 000000000..f6ee2f323 --- /dev/null +++ b/src-tauri/src/commands/native_input_mute.rs @@ -0,0 +1,170 @@ +use std::sync::{ + atomic::{AtomicBool, AtomicU64, Ordering}, + Arc, +}; + +pub fn start(input_muted: &Arc, mute_epoch: &Arc, on_change: F) -> bool +where + F: Fn(bool) + Send + Sync + 'static, +{ + clear(input_muted); + + #[cfg(target_os = "macos")] + return match macos::install(Arc::clone(input_muted), Arc::clone(mute_epoch), on_change) { + Ok(()) => true, + Err(error) => { + log::info!("AirPods input mute listener is unavailable: {error}"); + false + } + }; + + #[cfg(not(target_os = "macos"))] + { + let _ = (mute_epoch, on_change); + false + } +} + +pub fn stop(input_muted: &Arc) { + clear(input_muted); + + #[cfg(target_os = "macos")] + if let Err(error) = macos::uninstall() { + log::info!("Could not stop the AirPods input mute listener: {error}"); + } +} + +pub fn set_muted( + input_muted: &AtomicBool, + mute_epoch: &AtomicU64, + muted: bool, +) -> Result<(), String> { + #[cfg(target_os = "macos")] + { + macos::set_muted(muted)?; + apply_change(input_muted, mute_epoch, muted, &|_| {}); + Ok(()) + } + + #[cfg(not(target_os = "macos"))] + { + let _ = (input_muted, mute_epoch, muted); + Err("native microphone mute is only available on macOS".to_string()) + } +} + +fn clear(input_muted: &AtomicBool) { + input_muted.store(false, Ordering::Release); +} + +#[cfg(any(target_os = "macos", test))] +fn apply_change( + input_muted: &AtomicBool, + mute_epoch: &AtomicU64, + muted: bool, + on_change: &dyn Fn(bool), +) { + if input_muted.swap(muted, Ordering::AcqRel) != muted { + if muted { + mute_epoch.fetch_add(1, Ordering::AcqRel); + } + on_change(muted); + } +} + +#[cfg(target_os = "macos")] +mod macos { + use super::*; + use block2::RcBlock; + use objc2::runtime::Bool; + use objc2_avf_audio::AVAudioApplication; + + extern "C" { + fn berd_airpods_capture_start() -> bool; + fn berd_airpods_capture_stop(); + } + + pub fn install( + input_muted: Arc, + mute_epoch: Arc, + on_change: F, + ) -> Result<(), String> + where + F: Fn(bool) + Send + Sync + 'static, + { + // SAFETY: Berd's minimum macOS version is 14.0, where + // AVAudioApplication and these selectors are public API. + let application = unsafe { AVAudioApplication::sharedInstance() }; + let handler = RcBlock::new(move |muted: Bool| { + let muted = muted.as_bool(); + apply_change(&input_muted, &mute_epoch, muted, &|muted| { + log::info!("AirPods input mute changed muted={muted}"); + on_change(muted); + }); + Bool::YES + }); + // SAFETY: The block has the generated AVFAudio signature. The API + // copies and retains it until a later registration or cancellation. + if let Err(error) = + unsafe { application.setInputMuteStateChangeHandler_error(Some(&handler)) } + { + return Err(error.localizedDescription().to_string()); + } + if let Err(error) = unsafe { application.setInputMuted_error(false) } { + let _ = unsafe { application.setInputMuteStateChangeHandler_error(None) }; + return Err(error.localizedDescription().to_string()); + } + // SAFETY: The Swift bridge owns one process-global AVAudioEngine and + // exposes a C-compatible lifecycle API. + if !unsafe { berd_airpods_capture_start() } { + let _ = unsafe { application.setInputMuteStateChangeHandler_error(None) }; + return Err("macOS microphone capture could not start".to_string()); + } + log::info!("AirPods input mute listener started"); + Ok(()) + } + + pub fn uninstall() -> Result<(), String> { + // SAFETY: Berd targets macOS 14+, and nil is the documented way to + // cancel the process-wide handler at the end of a call lifecycle. + let application = unsafe { AVAudioApplication::sharedInstance() }; + // Do not leave another Berd microphone feature inheriting the voice + // conversation's last input-mute state after its handler is gone. + let reset_result = unsafe { application.setInputMuted_error(false) } + .map_err(|error| error.localizedDescription().to_string()); + let handler_result = unsafe { application.setInputMuteStateChangeHandler_error(None) } + .map_err(|error| error.localizedDescription().to_string()); + unsafe { berd_airpods_capture_stop() }; + reset_result.and(handler_result) + } + + pub fn set_muted(muted: bool) -> Result<(), String> { + let application = unsafe { AVAudioApplication::sharedInstance() }; + unsafe { application.setInputMuted_error(muted) } + .map_err(|error| error.localizedDescription().to_string()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn lifecycle_boundary_clears_mute() { + let input_muted = Arc::new(AtomicBool::new(true)); + clear(&input_muted); + assert!(!input_muted.load(Ordering::Acquire)); + } + + #[test] + fn mute_edge_advances_epoch_even_after_immediate_unmute() { + let input_muted = AtomicBool::new(false); + let mute_epoch = AtomicU64::new(0); + + apply_change(&input_muted, &mute_epoch, true, &|_| {}); + apply_change(&input_muted, &mute_epoch, false, &|_| {}); + + assert!(!input_muted.load(Ordering::Acquire)); + assert_eq!(mute_epoch.load(Ordering::Acquire), 1); + } +} diff --git a/src-tauri/src/commands/native_voice.rs b/src-tauri/src/commands/native_voice.rs index eaff54ede..ffaa4a376 100644 --- a/src-tauri/src/commands/native_voice.rs +++ b/src-tauri/src/commands/native_voice.rs @@ -4,7 +4,7 @@ use std::{ collections::VecDeque, path::PathBuf, sync::{ - atomic::{AtomicBool, AtomicUsize, Ordering}, + atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}, mpsc::{self, Receiver, SyncSender, TrySendError}, Arc, Mutex, }, @@ -16,7 +16,9 @@ use serde::Serialize; use tauri::{AppHandle, Emitter, Manager, State, WebviewWindow}; use tokio::sync::mpsc as tokio_mpsc; -use super::{pocket_voice::parakeet_model_dir, voice_capture::VoiceCaptureState}; +use super::{ + native_input_mute, pocket_voice::parakeet_model_dir, voice_capture::VoiceCaptureState, +}; const EVENT_NAME: &str = "voice-conversation:event"; const MAX_AUDIO_BATCH_BYTES: usize = 100 * 1024; @@ -47,6 +49,8 @@ pub struct NativeVoiceStatus { session_id: Option, owner_window_label: Option, revision: u64, + native_microphone_mute_control: bool, + native_microphone_muted: bool, } #[derive(Clone, Debug, Serialize)] @@ -79,6 +83,7 @@ enum NativeVoiceEvent { owner_window_label: String, line: String, revision: u64, + native_microphone_mute_control: bool, }, User { session_id: String, @@ -93,6 +98,11 @@ enum NativeVoiceEvent { activity: &'static str, revision: u64, }, + InputMute { + session_id: String, + muted: bool, + revision: u64, + }, CleanShutdown { session_id: String, revision: u64, @@ -112,6 +122,7 @@ struct Runtime { revision: u64, owner: Option, pipeline: Option, + native_microphone_mute_control: bool, } #[derive(Clone)] @@ -124,6 +135,8 @@ pub struct NativeVoiceState { runtime: Arc>, pending: Arc>>, capture_suppressions: Arc, + input_muted: Arc, + input_mute_epoch: Arc, } #[must_use = "capture suppression ends when the guard is dropped"] @@ -169,27 +182,56 @@ enum SttMessage { } struct SttPipeline { - audio_tx: SyncSender>, + audio_tx: SyncSender, audio_seen: AtomicBool, shutdown: Arc, + discard_on_shutdown: Arc, + input_muted: Arc, + input_mute_epoch: Arc, thread: Option>, } +struct AudioBatch { + bytes: Vec, + mute_epoch: u64, +} + impl SttPipeline { - fn new(model_dir: PathBuf) -> Result<(Self, tokio_mpsc::Receiver), String> { + fn new( + model_dir: PathBuf, + input_muted: Arc, + input_mute_epoch: Arc, + ) -> Result<(Self, tokio_mpsc::Receiver), String> { let (audio_tx, audio_rx) = mpsc::sync_channel(AUDIO_QUEUE_DEPTH); let (event_tx, event_rx) = tokio_mpsc::channel(64); let shutdown = Arc::new(AtomicBool::new(false)); + let discard_on_shutdown = Arc::new(AtomicBool::new(false)); let worker_shutdown = Arc::clone(&shutdown); + let worker_discard_on_shutdown = Arc::clone(&discard_on_shutdown); + let worker_input_muted = Arc::clone(&input_muted); + let worker_input_mute_epoch = Arc::clone(&input_mute_epoch); let thread = thread::Builder::new() .name("berd-native-stt".into()) - .spawn(move || stt_worker(model_dir, audio_rx, event_tx, worker_shutdown)) + .spawn(move || { + stt_worker( + model_dir, + audio_rx, + event_tx, + worker_shutdown, + worker_discard_on_shutdown, + worker_input_muted, + worker_input_mute_epoch, + ) + }) .map_err(|error| format!("start native transcription: {error}"))?; Ok(( Self { audio_tx, audio_seen: AtomicBool::new(false), shutdown, + discard_on_shutdown, + input_muted, + input_mute_epoch, thread: Some(thread), }, event_rx, @@ -206,13 +248,22 @@ impl SttPipeline { if !bytes.len().is_multiple_of(4) { return Err("audio batch must contain complete f32 samples".to_string()); } + if self.input_muted.load(Ordering::Acquire) { + return Ok(()); + } + let mute_epoch = self.input_mute_epoch.load(Ordering::Acquire); + if self.input_muted.load(Ordering::Acquire) + || mute_epoch != self.input_mute_epoch.load(Ordering::Acquire) + { + return Ok(()); + } if !self.audio_seen.swap(true, Ordering::AcqRel) { log::info!( "Native Parakeet received its first audio batch ({} bytes)", bytes.len() ); } - match self.audio_tx.try_send(bytes) { + match self.audio_tx.try_send(AudioBatch { bytes, mute_epoch }) { Ok(()) => Ok(()), Err(TrySendError::Full(_)) => Err( "Native voice audio overrun: transcription could not keep up with microphone input." @@ -225,9 +276,20 @@ impl SttPipeline { } fn begin_shutdown(&mut self) -> Option> { - self.shutdown.store(true, Ordering::Release); + self.signal_shutdown(); self.thread.take() } + + fn signal_shutdown(&self) { + self.latch_muted_shutdown(); + self.shutdown.store(true, Ordering::Release); + } + + fn latch_muted_shutdown(&self) { + if self.input_muted.load(Ordering::Acquire) { + self.discard_on_shutdown.store(true, Ordering::Release); + } + } } impl Drop for SttPipeline { @@ -272,6 +334,9 @@ fn status(app: &AppHandle, state: &NativeVoiceState) -> NativeVoiceStatus { .as_ref() .map(|owner| owner.window_label.clone()), revision: runtime.revision, + native_microphone_mute_control: runtime.native_microphone_mute_control, + native_microphone_muted: runtime.session_id.is_some() + && state.input_muted.load(Ordering::Acquire), } } @@ -389,7 +454,11 @@ pub async fn start_native_voice_conversation( return Err(error); } }; - let (pipeline, mut events) = match SttPipeline::new(model_dir) { + let (pipeline, mut events) = match SttPipeline::new( + model_dir, + Arc::clone(&state.input_muted), + Arc::clone(&state.input_mute_epoch), + ) { Ok(result) => result, Err(error) => { if microphone_claimed { @@ -398,7 +467,7 @@ pub async fn start_native_voice_conversation( return Err(error); } }; - let (revision, lifecycle_id) = { + let (revision, lifecycle_id, runtime_mute_control) = { let mut runtime = state .runtime .lock() @@ -416,9 +485,24 @@ pub async fn start_native_voice_conversation( window_label: window_label.clone(), }); runtime.pipeline = Some(pipeline); + let runtime_revision = runtime.revision; + let mute_window = webview_window.clone(); + let mute_session_id = session_id.clone(); + runtime.native_microphone_mute_control = + native_input_mute::start(&state.input_muted, &state.input_mute_epoch, move |muted| { + let _ = mute_window.emit( + EVENT_NAME, + NativeVoiceEvent::InputMute { + session_id: mute_session_id.clone(), + muted, + revision: runtime_revision, + }, + ); + }); ( runtime.revision, runtime.lifecycle_id.clone().unwrap_or_default(), + runtime.native_microphone_mute_control, ) }; let _ = webview_window.emit( @@ -428,6 +512,7 @@ pub async fn start_native_voice_conversation( owner_window_label: window_label.clone(), line: "Native Parakeet voice conversation is on".to_string(), revision, + native_microphone_mute_control: runtime_mute_control, }, ); @@ -435,6 +520,7 @@ pub async fn start_native_voice_conversation( let event_window = webview_window.clone(); let runtime = Arc::clone(&state.runtime); let pending = Arc::clone(&state.pending); + let input_muted = Arc::clone(&state.input_muted); tauri::async_runtime::spawn(async move { while let Some(event) = events.recv().await { let active = runtime.lock().ok().is_some_and(|current| { @@ -507,6 +593,8 @@ pub async fn start_native_voice_conversation( { break; } + native_input_mute::stop(&input_muted); + current.native_microphone_mute_control = false; current.session_id = None; current.lifecycle_id = None; current.owner = None; @@ -573,6 +661,8 @@ pub async fn stop_native_voice_conversation( .lock() .map_err(|_| "native voice state lock was poisoned".to_string())?; if runtime.revision == revision && runtime.session_id == session_id { + native_input_mute::stop(&state.input_muted); + runtime.native_microphone_mute_control = false; runtime.session_id = None; runtime.lifecycle_id = None; runtime.owner = None; @@ -630,6 +720,8 @@ impl NativeVoiceState { .lock() .map_err(|_| "native voice state lock was poisoned".to_string())?; if runtime.revision == revision && runtime.session_id == session_id { + native_input_mute::stop(&self.input_muted); + runtime.native_microphone_mute_control = false; runtime.session_id = None; runtime.lifecycle_id = None; runtime.owner = None; @@ -664,11 +756,13 @@ impl NativeVoiceState { { return false; } - ( - runtime.session_id.clone(), - runtime.revision, - runtime.pipeline.take(), - ) + let pipeline = runtime.pipeline.take(); + if let Some(pipeline) = pipeline.as_ref() { + pipeline.signal_shutdown(); + } + native_input_mute::stop(&self.input_muted); + runtime.native_microphone_mute_control = false; + (runtime.session_id.clone(), runtime.revision, pipeline) }; if pipeline.is_none() { if let Ok(mut runtime) = self.runtime.lock() { @@ -701,6 +795,11 @@ impl NativeVoiceState { let Ok(mut runtime) = self.runtime.lock() else { return; }; + if let Some(pipeline) = runtime.pipeline.as_ref() { + pipeline.latch_muted_shutdown(); + } + native_input_mute::stop(&self.input_muted); + runtime.native_microphone_mute_control = false; ( runtime.session_id.clone(), runtime.revision, @@ -719,6 +818,39 @@ impl NativeVoiceState { } } +#[tauri::command] +pub fn set_native_voice_input_muted( + state: State<'_, NativeVoiceState>, + webview_window: WebviewWindow, + session_id: String, + revision: u64, + muted: bool, +) -> Result<(), String> { + let runtime = state + .runtime + .lock() + .map_err(|_| "native voice state lock was poisoned".to_string())?; + if !owns_native_mute_control(&runtime, webview_window.label(), &session_id, revision) { + return Err("Native microphone mute is unavailable for this conversation.".to_string()); + } + native_input_mute::set_muted(&state.input_muted, &state.input_mute_epoch, muted) +} + +fn owns_native_mute_control( + runtime: &Runtime, + window_label: &str, + session_id: &str, + revision: u64, +) -> bool { + runtime.native_microphone_mute_control + && runtime.session_id.as_deref() == Some(session_id) + && runtime.revision == revision + && runtime + .owner + .as_ref() + .is_some_and(|owner| owner.window_label == window_label) +} + #[tauri::command] pub fn push_native_voice_audio( request: tauri::ipc::Request<'_>, @@ -769,9 +901,12 @@ fn enqueue_pending_transcript( fn stt_worker( model_dir: PathBuf, - audio_rx: Receiver>, + audio_rx: Receiver, event_tx: tokio_mpsc::Sender, shutdown: Arc, + discard_on_shutdown: Arc, + input_muted: Arc, + input_mute_epoch: Arc, ) { use rubato::{Fft, FixedSync, Resampler}; use sherpa_onnx::{OfflineRecognizer, OfflineRecognizerConfig}; @@ -810,14 +945,42 @@ fn stt_worker( let mut speech = Vec::new(); let mut silence_frames = 0_usize; let mut in_speech = false; + let mut observed_mute_epoch = input_mute_epoch.load(Ordering::Acquire); while !shutdown.load(Ordering::Acquire) { - let bytes = match audio_rx.recv_timeout(Duration::from_millis(50)) { - Ok(bytes) => bytes, - Err(mpsc::RecvTimeoutError::Timeout) => continue, + let batch = match audio_rx.recv_timeout(Duration::from_millis(50)) { + Ok(batch) => Some(batch), + Err(mpsc::RecvTimeoutError::Timeout) => None, Err(mpsc::RecvTimeoutError::Disconnected) => break, }; + let shutting_down = shutdown.load(Ordering::Acquire); + if shutting_down && (discard_on_shutdown.load(Ordering::Acquire) || batch.is_none()) { + break; + } + let current_mute_epoch = input_mute_epoch.load(Ordering::Acquire); + if current_mute_epoch != observed_mute_epoch { + observed_mute_epoch = current_mute_epoch; + if clear_buffered_audio( + &mut input_48k, + &mut leftover_16k, + &mut speech, + &mut silence_frames, + &mut in_speech, + ) { + let _ = event_tx.blocking_send(SttMessage::Speaking(false)); + } + } + if !shutting_down && input_muted.load(Ordering::Acquire) { + continue; + } + let Some(batch) = batch else { + continue; + }; + if batch.mute_epoch != observed_mute_epoch { + continue; + } input_48k.extend( - bytes + batch + .bytes .chunks_exact(4) .map(|sample| f32::from_le_bytes([sample[0], sample[1], sample[2], sample[3]])), ); @@ -839,7 +1002,14 @@ fn stt_worker( silence_frames = 0; speech.extend_from_slice(&frame); if speech.len() >= MAX_SPEECH_SAMPLES { - flush_speech(&speech, &recognizer, &event_tx, None); + flush_speech( + &speech, + &recognizer, + &event_tx, + None, + &input_mute_epoch, + observed_mute_epoch, + ); speech.clear(); in_speech = false; let _ = event_tx.blocking_send(SttMessage::Speaking(false)); @@ -848,7 +1018,14 @@ fn stt_worker( speech.extend_from_slice(&frame); silence_frames += 1; if silence_frames >= SILENCE_FLUSH_FRAMES { - flush_speech(&speech, &recognizer, &event_tx, None); + flush_speech( + &speech, + &recognizer, + &event_tx, + None, + &input_mute_epoch, + observed_mute_epoch, + ); speech.clear(); silence_frames = 0; in_speech = false; @@ -858,13 +1035,34 @@ fn stt_worker( } } } - if !speech.is_empty() { + if !speech.is_empty() && !discard_on_shutdown.load(Ordering::Acquire) { let (delivered_tx, delivered_rx) = mpsc::sync_channel(0); - flush_speech(&speech, &recognizer, &event_tx, Some(delivered_tx)); + flush_speech( + &speech, + &recognizer, + &event_tx, + Some(delivered_tx), + &input_mute_epoch, + observed_mute_epoch, + ); let _ = delivered_rx.recv_timeout(Duration::from_secs(5)); } } +fn clear_buffered_audio( + input_48k: &mut Vec, + leftover_16k: &mut Vec, + speech: &mut Vec, + silence_frames: &mut usize, + in_speech: &mut bool, +) -> bool { + input_48k.clear(); + leftover_16k.clear(); + speech.clear(); + *silence_frames = 0; + std::mem::take(in_speech) +} + fn resample(resampler: &mut rubato::Fft, samples: &[f32]) -> Vec { use audioadapter_buffers::direct::InterleavedSlice; use rubato::Resampler; @@ -882,6 +1080,8 @@ fn flush_speech( recognizer: &sherpa_onnx::OfflineRecognizer, event_tx: &tokio_mpsc::Sender, delivered: Option>, + input_mute_epoch: &AtomicU64, + expected_mute_epoch: u64, ) { if speech.is_empty() { return; @@ -893,6 +1093,12 @@ fn flush_speech( .get_result() .map(|result| result.text.trim().to_string()) .unwrap_or_default(); + if input_mute_epoch.load(Ordering::Acquire) != expected_mute_epoch { + if let Some(delivered) = delivered { + let _ = delivered.send(()); + } + return; + } log::info!( "Native Parakeet finalized {} samples into {} text characters", speech.len(), @@ -919,6 +1125,24 @@ fn deliver_recognition_result( mod tests { use super::*; + #[test] + fn native_mute_control_is_bound_to_window_session_and_revision() { + let runtime = Runtime { + session_id: Some("session-1".to_string()), + revision: 4, + owner: Some(RuntimeOwner { + window_label: "main".to_string(), + }), + native_microphone_mute_control: true, + ..Runtime::default() + }; + + assert!(owns_native_mute_control(&runtime, "main", "session-1", 4)); + assert!(!owns_native_mute_control(&runtime, "other", "session-1", 4)); + assert!(!owns_native_mute_control(&runtime, "main", "session-2", 4)); + assert!(!owns_native_mute_control(&runtime, "main", "session-1", 5)); + } + #[test] fn speaker_playback_blocks_vad_ingestion_until_all_guards_finish() { let state = NativeVoiceState::default(); @@ -972,6 +1196,9 @@ mod tests { let pipeline = SttPipeline { audio_tx: sender, shutdown: Arc::new(AtomicBool::new(false)), + discard_on_shutdown: Arc::new(AtomicBool::new(false)), + input_muted: Arc::new(AtomicBool::new(false)), + input_mute_epoch: Arc::new(AtomicU64::new(0)), audio_seen: AtomicBool::new(false), thread: None, }; @@ -985,6 +1212,9 @@ mod tests { let pipeline = SttPipeline { audio_tx: sender, shutdown: Arc::new(AtomicBool::new(false)), + discard_on_shutdown: Arc::new(AtomicBool::new(false)), + input_muted: Arc::new(AtomicBool::new(false)), + input_mute_epoch: Arc::new(AtomicU64::new(0)), audio_seen: AtomicBool::new(false), thread: None, }; @@ -996,6 +1226,119 @@ mod tests { .contains("overrun")); } + #[test] + fn input_mute_discards_audio_and_unmute_resumes_queueing() { + let (sender, receiver) = mpsc::sync_channel(1); + let input_muted = Arc::new(AtomicBool::new(true)); + let input_mute_epoch = Arc::new(AtomicU64::new(1)); + let pipeline = SttPipeline { + audio_tx: sender, + shutdown: Arc::new(AtomicBool::new(false)), + discard_on_shutdown: Arc::new(AtomicBool::new(false)), + input_muted: Arc::clone(&input_muted), + input_mute_epoch: Arc::clone(&input_mute_epoch), + audio_seen: AtomicBool::new(false), + thread: None, + }; + + pipeline + .push(vec![0; 4]) + .expect("muted microphone input is accepted and discarded"); + assert!(receiver.try_recv().is_err()); + + input_muted.store(false, Ordering::Release); + pipeline.push(vec![0; 4]).expect("unmuted audio is queued"); + assert_eq!( + receiver.try_recv().expect("unmuted audio").bytes, + vec![0; 4] + ); + } + + #[test] + fn queued_audio_retains_epoch_across_fast_mute_unmute() { + let (sender, receiver) = mpsc::sync_channel(1); + let input_mute_epoch = Arc::new(AtomicU64::new(0)); + let pipeline = SttPipeline { + audio_tx: sender, + shutdown: Arc::new(AtomicBool::new(false)), + discard_on_shutdown: Arc::new(AtomicBool::new(false)), + input_muted: Arc::new(AtomicBool::new(false)), + input_mute_epoch: Arc::clone(&input_mute_epoch), + audio_seen: AtomicBool::new(false), + thread: None, + }; + + pipeline.push(vec![0; 4]).expect("audio queues before mute"); + input_mute_epoch.fetch_add(1, Ordering::AcqRel); + + let batch = receiver.try_recv().expect("queued audio"); + assert_ne!(batch.mute_epoch, input_mute_epoch.load(Ordering::Acquire)); + } + + #[test] + fn input_mute_clears_partial_utterance_even_without_a_new_batch() { + let mut input_48k = vec![0.1]; + let mut leftover_16k = vec![0.2]; + let mut speech = vec![0.3]; + let mut silence_frames = 4; + let mut in_speech = true; + + assert!(clear_buffered_audio( + &mut input_48k, + &mut leftover_16k, + &mut speech, + &mut silence_frames, + &mut in_speech, + )); + assert!(input_48k.is_empty()); + assert!(leftover_16k.is_empty()); + assert!(speech.is_empty()); + assert_eq!(silence_frames, 0); + assert!(!in_speech); + } + + #[test] + fn muted_shutdown_keeps_final_utterance_discarded_after_handler_reset() { + let (sender, _receiver) = mpsc::sync_channel(1); + let input_muted = Arc::new(AtomicBool::new(true)); + let discard_on_shutdown = Arc::new(AtomicBool::new(false)); + let mut pipeline = SttPipeline { + audio_tx: sender, + shutdown: Arc::new(AtomicBool::new(false)), + discard_on_shutdown: Arc::clone(&discard_on_shutdown), + input_muted: Arc::clone(&input_muted), + input_mute_epoch: Arc::new(AtomicU64::new(1)), + audio_seen: AtomicBool::new(false), + thread: None, + }; + + pipeline.begin_shutdown(); + input_muted.store(false, Ordering::Release); + + assert!(discard_on_shutdown.load(Ordering::Acquire)); + } + + #[test] + fn unmuted_shutdown_keeps_final_utterance_after_later_mute_event() { + let (sender, _receiver) = mpsc::sync_channel(1); + let input_muted = Arc::new(AtomicBool::new(false)); + let discard_on_shutdown = Arc::new(AtomicBool::new(false)); + let mut pipeline = SttPipeline { + audio_tx: sender, + shutdown: Arc::new(AtomicBool::new(false)), + discard_on_shutdown: Arc::clone(&discard_on_shutdown), + input_muted: Arc::clone(&input_muted), + input_mute_epoch: Arc::new(AtomicU64::new(0)), + audio_seen: AtomicBool::new(false), + thread: None, + }; + + pipeline.begin_shutdown(); + input_muted.store(true, Ordering::Release); + + assert!(!discard_on_shutdown.load(Ordering::Acquire)); + } + #[test] fn only_owning_window_can_inject_audio() { let state = NativeVoiceState::default(); @@ -1008,6 +1351,9 @@ mod tests { runtime.pipeline = Some(SttPipeline { audio_tx: sender, shutdown: Arc::new(AtomicBool::new(false)), + discard_on_shutdown: Arc::new(AtomicBool::new(false)), + input_muted: Arc::new(AtomicBool::new(false)), + input_mute_epoch: Arc::new(AtomicU64::new(0)), audio_seen: AtomicBool::new(false), thread: None, }); @@ -1016,13 +1362,17 @@ mod tests { assert!(push_audio_for_window(&state, "other-window", vec![0; 4]).is_err()); assert!(receiver.try_recv().is_err()); push_audio_for_window(&state, "owner-window", vec![0; 4]).expect("owner can send audio"); - assert_eq!(receiver.try_recv().expect("owner audio queued"), vec![0; 4]); + assert_eq!( + receiver.try_recv().expect("owner audio queued").bytes, + vec![0; 4] + ); } #[tokio::test] async fn window_destroy_schedules_blocked_worker_join_off_callback() { let state = NativeVoiceState::default(); let (sender, _receiver) = mpsc::sync_channel(1); + let shutdown = Arc::new(AtomicBool::new(false)); let worker = thread::spawn(|| thread::sleep(Duration::from_millis(250))); { let mut runtime = state.runtime.lock().expect("lock native runtime"); @@ -1032,7 +1382,10 @@ mod tests { }); runtime.pipeline = Some(SttPipeline { audio_tx: sender, - shutdown: Arc::new(AtomicBool::new(false)), + shutdown: Arc::clone(&shutdown), + discard_on_shutdown: Arc::new(AtomicBool::new(false)), + input_muted: Arc::new(AtomicBool::new(false)), + input_mute_epoch: Arc::new(AtomicU64::new(0)), audio_seen: AtomicBool::new(false), thread: Some(worker), }); @@ -1041,6 +1394,7 @@ mod tests { let started = std::time::Instant::now(); assert!(state.stop_for_window_destroyed("owner-window")); assert!(started.elapsed() < Duration::from_millis(50)); + assert!(shutdown.load(Ordering::Acquire)); tokio::time::sleep(Duration::from_millis(300)).await; assert!(state .runtime diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index f90b7b950..8672de987 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -646,6 +646,7 @@ pub fn run() { commands::native_voice::start_native_voice_conversation, commands::native_voice::stop_native_voice_conversation, commands::native_voice::push_native_voice_audio, + commands::native_voice::set_native_voice_input_muted, commands::voice_capture::register_voice_renderer_instance, commands::window_session::get_session_window_support, commands::window_session::open_session_window, diff --git a/src-tauri/swift/BerdAirPodsBridge/Package.swift b/src-tauri/swift/BerdAirPodsBridge/Package.swift new file mode 100644 index 000000000..5e216c9a1 --- /dev/null +++ b/src-tauri/swift/BerdAirPodsBridge/Package.swift @@ -0,0 +1,25 @@ +// swift-tools-version: 5.9 +import PackageDescription + +let package = Package( + name: "BerdAirPodsBridge", + platforms: [.macOS(.v14)], + products: [ + .library( + name: "BerdAirPodsBridge", + type: .static, + targets: ["BerdAirPodsBridge"] + ) + ], + targets: [ + .target( + name: "BerdObjCExceptionCatch", + publicHeadersPath: "include" + ), + .target( + name: "BerdAirPodsBridge", + dependencies: ["BerdObjCExceptionCatch"], + linkerSettings: [.linkedFramework("AVFAudio")] + ) + ] +) diff --git a/src-tauri/swift/BerdAirPodsBridge/Sources/BerdAirPodsBridge/BerdAirPodsBridge.swift b/src-tauri/swift/BerdAirPodsBridge/Sources/BerdAirPodsBridge/BerdAirPodsBridge.swift new file mode 100644 index 000000000..e20d632f2 --- /dev/null +++ b/src-tauri/swift/BerdAirPodsBridge/Sources/BerdAirPodsBridge/BerdAirPodsBridge.swift @@ -0,0 +1,158 @@ +import AVFAudio +import BerdObjCExceptionCatch +import Foundation + +private let bridgeQueue = DispatchQueue(label: "com.berd.airpods-capture") + +@available(macOS 14.0, *) +private final class AirPodsCapture: @unchecked Sendable { + private var engine: AVAudioEngine? + private var inputNode: AVAudioInputNode? + private var configurationObserver: NSObjectProtocol? + private var inputMuteObserver: NSObjectProtocol? + private var restart: DispatchWorkItem? + private var generation: UInt64 = 0 + private var stopped = false + + init() throws { + try startEngine() + } + + func stop() { + guard !stopped else { return } + stopped = true + generation &+= 1 + restart?.cancel() + restart = nil + tearDownEngine() + } + + private func startEngine() throws { + let engine = AVAudioEngine() + var caughtInputNode: AVAudioInputNode? + var inputNodeError: NSError? + BerdTryObjCBlock({ caughtInputNode = engine.inputNode }, &inputNodeError) + guard inputNodeError == nil, let inputNode = caughtInputNode else { + throw CaptureError.objectiveC(inputNodeError?.localizedDescription ?? "no input node") + } + + let format = inputNode.outputFormat(forBus: 0) + guard format.sampleRate > 0, format.channelCount > 0 else { + throw CaptureError.noInputFormat + } + var tapError: NSError? + BerdTryObjCBlock({ + inputNode.installTap(onBus: 0, bufferSize: 4096, format: format) { _, _ in } + }, &tapError) + guard tapError == nil else { + throw CaptureError.objectiveC(tapError?.localizedDescription ?? "install tap failed") + } + + do { + engine.prepare() + try engine.start() + } catch { + removeTap(from: inputNode) + throw error + } + configurationObserver = NotificationCenter.default.addObserver( + forName: .AVAudioEngineConfigurationChange, + object: engine, + queue: nil + ) { [weak self] _ in + self?.scheduleRestart() + } + // Observing this notification and reading the state opts this capture + // session into AirPods mute delivery. Rust's handler remains the event path. + inputMuteObserver = NotificationCenter.default.addObserver( + forName: AVAudioApplication.inputMuteStateChangeNotification, + object: nil, + queue: nil + ) { _ in + _ = AVAudioApplication.shared.isInputMuted + } + self.engine = engine + self.inputNode = inputNode + } + + private func tearDownEngine() { + if let configurationObserver { + NotificationCenter.default.removeObserver(configurationObserver) + self.configurationObserver = nil + } + if let inputMuteObserver { + NotificationCenter.default.removeObserver(inputMuteObserver) + self.inputMuteObserver = nil + } + if let inputNode { removeTap(from: inputNode) } + engine?.stop() + inputNode = nil + engine = nil + } + + private func removeTap(from inputNode: AVAudioInputNode) { + var error: NSError? + BerdTryObjCBlock({ inputNode.removeTap(onBus: 0) }, &error) + } + + private func scheduleRestart() { + bridgeQueue.async { [weak self] in + guard let self, !stopped else { return } + generation &+= 1 + let expectedGeneration = generation + restart?.cancel() + let work = DispatchWorkItem { [weak self] in + self?.restartEngine(expectedGeneration) + } + restart = work + bridgeQueue.asyncAfter(deadline: .now() + 0.15, execute: work) + } + } + + private func restartEngine(_ expectedGeneration: UInt64) { + guard !stopped, generation == expectedGeneration else { return } + restart = nil + tearDownEngine() + do { + try startEngine() + } catch { + let work = DispatchWorkItem { [weak self] in + self?.restartEngine(expectedGeneration) + } + restart = work + bridgeQueue.asyncAfter(deadline: .now() + 1, execute: work) + } + } + + private enum CaptureError: Error { + case noInputFormat + case objectiveC(String) + } +} + +@available(macOS 14.0, *) +private var activeCapture: AirPodsCapture? + +@_cdecl("berd_airpods_capture_start") +public func berdAirPodsCaptureStart() -> Bool { + guard #available(macOS 14.0, *) else { return false } + return bridgeQueue.sync { + do { + activeCapture?.stop() + activeCapture = try AirPodsCapture() + return true + } catch { + activeCapture = nil + return false + } + } +} + +@_cdecl("berd_airpods_capture_stop") +public func berdAirPodsCaptureStop() { + guard #available(macOS 14.0, *) else { return } + bridgeQueue.sync { + activeCapture?.stop() + activeCapture = nil + } +} diff --git a/src-tauri/swift/BerdAirPodsBridge/Sources/BerdObjCExceptionCatch/BerdObjCExceptionCatch.m b/src-tauri/swift/BerdAirPodsBridge/Sources/BerdObjCExceptionCatch/BerdObjCExceptionCatch.m new file mode 100644 index 000000000..350cd0658 --- /dev/null +++ b/src-tauri/swift/BerdAirPodsBridge/Sources/BerdObjCExceptionCatch/BerdObjCExceptionCatch.m @@ -0,0 +1,19 @@ +#import "BerdObjCExceptionCatch.h" + +// Adapted from voice-conversation-cli's VCTryObjCBlock. +BOOL BerdTryObjCBlock(void (NS_NOESCAPE ^_Nonnull block)(void), + NSError *_Nullable *_Nullable error) { + @try { + block(); + return YES; + } @catch (NSException *exception) { + if (error) { + *error = [NSError errorWithDomain:@"com.berd.objc-exception" + code:-1 + userInfo:@{ + NSLocalizedDescriptionKey: exception.reason ?: exception.name, + }]; + } + return NO; + } +} diff --git a/src-tauri/swift/BerdAirPodsBridge/Sources/BerdObjCExceptionCatch/include/BerdObjCExceptionCatch.h b/src-tauri/swift/BerdAirPodsBridge/Sources/BerdObjCExceptionCatch/include/BerdObjCExceptionCatch.h new file mode 100644 index 000000000..f23c87a37 --- /dev/null +++ b/src-tauri/swift/BerdAirPodsBridge/Sources/BerdObjCExceptionCatch/include/BerdObjCExceptionCatch.h @@ -0,0 +1,4 @@ +#import + +BOOL BerdTryObjCBlock(void (NS_NOESCAPE ^_Nonnull block)(void), + NSError *_Nullable *_Nullable error); diff --git a/src/features/voice-conversation/api/voiceConversation.test.ts b/src/features/voice-conversation/api/voiceConversation.test.ts index 4bd905ec8..15c4fac47 100644 --- a/src/features/voice-conversation/api/voiceConversation.test.ts +++ b/src/features/voice-conversation/api/voiceConversation.test.ts @@ -23,8 +23,12 @@ vi.mock("../lib/nativeMicrophone", () => ({ import { acknowledgeVoiceConversationTranscript, + applyVoiceConversationMicrophoneMuteEvent, + applyVoiceConversationTerminalEvent, drainVoiceConversationTranscripts, + getVoiceConversationMicrophoneMuted, getVoiceConversationStatus, + hydrateVoiceConversationMicrophone, listenToVoiceConversation, reconcileVoiceConversationMicrophone, setVoiceConversationMicrophoneMuted, @@ -33,6 +37,16 @@ import { stopVoiceConversation, } from "./voiceConversation"; +function deferred() { + let resolve!: (value: T) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((promiseResolve, promiseReject) => { + resolve = promiseResolve; + reject = promiseReject; + }); + return { promise, reject, resolve }; +} + describe("voice conversation API", () => { beforeEach(() => { stopActiveMicrophoneForTest(); @@ -147,6 +161,25 @@ describe("voice conversation API", () => { expect(mocks.stopMicrophone).toHaveBeenCalledOnce(); }); + it("hydrates browser capture from the authoritative native mute state", async () => { + const status = { + available: true, + unavailableReason: null, + lifecycle: "running", + sessionId: "session-1", + ownerWindowLabel: "main", + revision: 3, + nativeMicrophoneMuteControl: true, + nativeMicrophoneMuted: true, + } as const; + + await hydrateVoiceConversationMicrophone(status); + + expect(getVoiceConversationMicrophoneMuted()).toBe(true); + expect(mocks.startMicrophone).toHaveBeenCalledOnce(); + expect(mocks.setMicrophoneMuted).toHaveBeenLastCalledWith(true); + }); + it("does not attach browser capture in a non-owning window", async () => { const status = { available: true, @@ -179,15 +212,171 @@ describe("voice conversation API", () => { expect(mocks.startMicrophone).toHaveBeenCalledOnce(); expect(mocks.stopMicrophone).not.toHaveBeenCalled(); - expect(mocks.setMicrophoneMuted.mock.calls).toEqual([ - [false], - [true], - [true], - [false], - ]); + expect(mocks.setMicrophoneMuted).toHaveBeenCalledWith(true); + expect(mocks.setMicrophoneMuted).toHaveBeenLastCalledWith(false); expect(mocks.invoke).not.toHaveBeenCalled(); }); + it("routes UI mute through macOS while keeping browser capture in sync", async () => { + const status = { + available: true, + unavailableReason: null, + lifecycle: "running", + sessionId: "session-1", + ownerWindowLabel: "main", + revision: 3, + nativeMicrophoneMuteControl: true, + } as const; + mocks.invoke.mockResolvedValue(undefined); + + await reconcileVoiceConversationMicrophone(status); + await setVoiceConversationMicrophoneMuted(true, status); + + expect(mocks.invoke).toHaveBeenCalledWith("set_native_voice_input_muted", { + sessionId: "session-1", + revision: 3, + muted: true, + }); + expect(mocks.setMicrophoneMuted).toHaveBeenLastCalledWith(true); + }); + + it("serializes opposite native mute intents so the latest command wins", async () => { + const status = { + available: true, + unavailableReason: null, + lifecycle: "running", + sessionId: "session-1", + ownerWindowLabel: "main", + revision: 3, + nativeMicrophoneMuteControl: true, + } as const; + const first = deferred(); + const second = deferred(); + mocks.invoke + .mockReturnValueOnce(first.promise) + .mockReturnValueOnce(second.promise); + + await reconcileVoiceConversationMicrophone(status); + const mute = setVoiceConversationMicrophoneMuted(true, status); + await vi.waitFor(() => expect(mocks.invoke).toHaveBeenCalledOnce()); + expect(mocks.invoke).toHaveBeenLastCalledWith( + "set_native_voice_input_muted", + expect.objectContaining({ muted: true }), + ); + const unmute = setVoiceConversationMicrophoneMuted(false, status); + expect(mocks.setMicrophoneMuted).toHaveBeenLastCalledWith(false); + first.resolve(); + await vi.waitFor(() => expect(mocks.invoke).toHaveBeenCalledTimes(2)); + expect(mocks.setMicrophoneMuted).toHaveBeenLastCalledWith(false); + expect(mocks.invoke).toHaveBeenLastCalledWith( + "set_native_voice_input_muted", + expect.objectContaining({ muted: false }), + ); + second.resolve(); + await Promise.all([mute, unmute]); + + expect(mocks.setMicrophoneMuted).toHaveBeenLastCalledWith(false); + }); + + it("rolls browser capture back to the last applied mute after a failure", async () => { + const status = { + available: true, + unavailableReason: null, + lifecycle: "running", + sessionId: "session-1", + ownerWindowLabel: "main", + revision: 3, + nativeMicrophoneMuteControl: true, + } as const; + mocks.invoke + .mockResolvedValueOnce(undefined) + .mockRejectedValueOnce(new Error("mute unavailable")); + + await reconcileVoiceConversationMicrophone(status); + await setVoiceConversationMicrophoneMuted(true, status); + await expect( + setVoiceConversationMicrophoneMuted(false, status), + ).rejects.toThrow("mute unavailable"); + + expect(mocks.setMicrophoneMuted).toHaveBeenLastCalledWith(true); + }); + + it("does not let a stale UI completion replace a newer stem state", async () => { + const status = { + available: true, + unavailableReason: null, + lifecycle: "running", + sessionId: "session-1", + ownerWindowLabel: "main", + revision: 3, + nativeMicrophoneMuteControl: true, + } as const; + const pending = deferred(); + mocks.invoke + .mockReturnValueOnce(pending.promise) + .mockRejectedValueOnce(new Error("mute unavailable")); + + await reconcileVoiceConversationMicrophone(status); + const staleMute = setVoiceConversationMicrophoneMuted(true, status); + await vi.waitFor(() => expect(mocks.invoke).toHaveBeenCalledOnce()); + applyVoiceConversationMicrophoneMuteEvent(false); + pending.resolve(); + await staleMute; + + await expect( + setVoiceConversationMicrophoneMuted(true, status), + ).rejects.toThrow("mute unavailable"); + expect(mocks.setMicrophoneMuted).toHaveBeenLastCalledWith(false); + }); + + it("does not run queued mute work after the conversation stops", async () => { + const running = { + available: true, + unavailableReason: null, + lifecycle: "running", + sessionId: "session-1", + ownerWindowLabel: "main", + revision: 3, + nativeMicrophoneMuteControl: true, + } as const; + const stopped = { + ...running, + lifecycle: "stopped" as const, + sessionId: null, + ownerWindowLabel: null, + revision: 4, + }; + const pending = deferred(); + mocks.invoke.mockImplementation((command: string) => { + if (command === "set_native_voice_input_muted") return pending.promise; + if (command === "stop_native_voice_conversation") { + return Promise.resolve(stopped); + } + return Promise.resolve(undefined); + }); + + await reconcileVoiceConversationMicrophone(running); + const muting = setVoiceConversationMicrophoneMuted(true, running); + await vi.waitFor(() => + expect(mocks.invoke).toHaveBeenCalledWith( + "set_native_voice_input_muted", + expect.any(Object), + ), + ); + const queuedUnmute = setVoiceConversationMicrophoneMuted(false, running); + await stopVoiceConversation(); + pending.resolve(); + await Promise.all([muting, queuedUnmute]); + + expect( + mocks.invoke.mock.calls.filter( + ([command]) => command === "set_native_voice_input_muted", + ), + ).toHaveLength(1); + expect(mocks.startMicrophone).toHaveBeenCalledOnce(); + expect(mocks.stopMicrophone).toHaveBeenCalledOnce(); + }); + it("restores the previous mute state when initial capture fails", async () => { const status = { available: true, @@ -266,7 +455,7 @@ describe("voice conversation API", () => { }); }); - it("stops browser capture when native voice shuts down elsewhere", async () => { + it("stops browser capture for a validated terminal event", async () => { mocks.invoke.mockResolvedValue({ available: true, unavailableReason: null, @@ -276,18 +465,7 @@ describe("voice conversation API", () => { revision: 3, }); await startVoiceConversation("session-1"); - mocks.listen.mockImplementation(async (_name, handler) => { - handler({ - payload: { - type: "cleanShutdown", - sessionId: "session-1", - revision: 4, - }, - }); - return vi.fn(); - }); - - await listenToVoiceConversation(vi.fn()); + applyVoiceConversationTerminalEvent(); expect(mocks.stopMicrophone).toHaveBeenCalledOnce(); }); diff --git a/src/features/voice-conversation/api/voiceConversation.ts b/src/features/voice-conversation/api/voiceConversation.ts index e213f5486..f8ca73e9e 100644 --- a/src/features/voice-conversation/api/voiceConversation.ts +++ b/src/features/voice-conversation/api/voiceConversation.ts @@ -12,6 +12,15 @@ let microphoneGeneration = 0; let microphoneStart: { generation: number; promise: Promise } | null = null; let microphoneMuted = false; +let appliedMicrophoneMuted = false; +let microphoneMuteIntent = 0; +let microphoneMuteQueue: Promise = Promise.resolve(); + +function resetMicrophoneMuteState(): void { + microphoneMuteIntent += 1; + microphoneMuted = false; + appliedMicrophoneMuted = false; +} function stopActiveMicrophone(): void { microphoneGeneration += 1; @@ -59,26 +68,73 @@ export async function reconcileVoiceConversationMicrophone( } } +export async function hydrateVoiceConversationMicrophone( + status: VoiceConversationStatus, +): Promise { + if (status.lifecycle === "running") { + applyVoiceConversationMicrophoneMuteEvent( + status.nativeMicrophoneMuted ?? false, + ); + } + await reconcileVoiceConversationMicrophone(status); +} + export async function setVoiceConversationMicrophoneMuted( muted: boolean, status: VoiceConversationStatus, ): Promise { - const previous = microphoneMuted; + const intent = ++microphoneMuteIntent; microphoneMuted = muted; - try { + activeMicrophone?.setMuted(muted); + const operation = microphoneMuteQueue.then(async () => { + if (intent !== microphoneMuteIntent) return; await reconcileVoiceConversationMicrophone(status); + if (intent !== microphoneMuteIntent) return; + activeMicrophone?.setMuted(microphoneMuted); + if (status.nativeMicrophoneMuteControl) { + await invoke("set_native_voice_input_muted", { + sessionId: status.sessionId, + revision: status.revision, + muted, + }); + } + if (intent === microphoneMuteIntent) { + appliedMicrophoneMuted = muted; + } + }); + microphoneMuteQueue = operation.catch(() => undefined); + try { + await operation; } catch (error) { - microphoneMuted = previous; - activeMicrophone?.setMuted(previous); + if (intent === microphoneMuteIntent) { + microphoneMuted = appliedMicrophoneMuted; + activeMicrophone?.setMuted(appliedMicrophoneMuted); + } throw error; } } +export function applyVoiceConversationMicrophoneMuteEvent(muted: boolean) { + microphoneMuteIntent += 1; + microphoneMuted = muted; + appliedMicrophoneMuted = muted; + activeMicrophone?.setMuted(muted); +} + +export function getVoiceConversationMicrophoneMuted(): boolean { + return microphoneMuted; +} + +export function applyVoiceConversationTerminalEvent(): void { + resetMicrophoneMuteState(); + stopActiveMicrophone(); +} + export function stopActiveMicrophoneForTest(): void { if (!import.meta.env.DEV) { throw new Error("Native microphone test controls are development-only."); } - microphoneMuted = false; + resetMicrophoneMuteState(); stopActiveMicrophone(); } @@ -109,6 +165,10 @@ export interface VoiceConversationStatus { ownerWindowLabel: string | null; /** Monotonic native lifecycle revision used to reject stale responses/events. */ revision: number; + /** macOS owns an input session capable of receiving headset mute controls. */ + nativeMicrophoneMuteControl?: boolean; + /** Authoritative native input-mute state for renderer recovery. */ + nativeMicrophoneMuted?: boolean; } export type VoiceConversationEvent = @@ -118,6 +178,7 @@ export type VoiceConversationEvent = ownerWindowLabel: string; line: string; revision: number; + nativeMicrophoneMuteControl: boolean; } | { type: "user"; @@ -138,6 +199,12 @@ export type VoiceConversationEvent = | "assistant-idle"; revision: number; } + | { + type: "inputMute"; + sessionId: string; + muted: boolean; + revision: number; + } | { type: "cleanShutdown"; sessionId: string; @@ -208,7 +275,7 @@ export function rejectVoiceConversationTranscript( export async function startVoiceConversation( sessionId: string, ): Promise { - microphoneMuted = false; + resetMicrophoneMuteState(); const { rendererId, rendererEpoch } = await getRendererInstance(); const status = await invoke( "start_native_voice_conversation", @@ -231,7 +298,7 @@ export async function startVoiceConversation( } export async function stopVoiceConversation(): Promise { - microphoneMuted = false; + resetMicrophoneMuteState(); stopActiveMicrophone(); const { rendererId, rendererEpoch } = await getRendererInstance(); return invoke("stop_native_voice_conversation", { @@ -244,13 +311,6 @@ export function listenToVoiceConversation( onEvent: (event: VoiceConversationEvent) => void, ): Promise { return listen(VOICE_CONVERSATION_EVENT, (event) => { - if ( - event.payload.type === "cleanShutdown" || - (event.payload.type === "error" && event.payload.terminal) - ) { - microphoneMuted = false; - stopActiveMicrophone(); - } onEvent(event.payload); }); } diff --git a/src/features/voice-conversation/hooks/useVoiceConversationController.ts b/src/features/voice-conversation/hooks/useVoiceConversationController.ts index df364dd87..cec0a633d 100644 --- a/src/features/voice-conversation/hooks/useVoiceConversationController.ts +++ b/src/features/voice-conversation/hooks/useVoiceConversationController.ts @@ -315,6 +315,7 @@ function ensureVoiceEventDeliveryInitialized() { return; } if (event.type === "activity") return; + if (event.type === "inputMute") return; if (event.type !== "user" || !event.text.trim()) return; if ( hasDeliveredVoiceTranscript( @@ -744,11 +745,13 @@ export function useVoiceConversationController({ const toggleMicrophoneMute = useCallback(async () => { if (status.lifecycle !== "running") return; try { - await setMicrophoneMuted(!microphoneMuted); + await setMicrophoneMuted( + !useVoiceConversationStore.getState().microphoneMuted, + ); } catch (muteError) { addErrorNotification(status.sessionId, errorText(muteError)); } - }, [microphoneMuted, setMicrophoneMuted, status.lifecycle, status.sessionId]); + }, [setMicrophoneMuted, status.lifecycle, status.sessionId]); return useMemo( () => ({ diff --git a/src/features/voice-conversation/stores/voiceConversationStore.test.ts b/src/features/voice-conversation/stores/voiceConversationStore.test.ts index af1510a15..ce6691b04 100644 --- a/src/features/voice-conversation/stores/voiceConversationStore.test.ts +++ b/src/features/voice-conversation/stores/voiceConversationStore.test.ts @@ -6,9 +6,13 @@ import type { } from "../api/voiceConversation"; const mocks = vi.hoisted(() => ({ + applyMicrophoneMuteEvent: vi.fn(), + applyTerminalEvent: vi.fn(), acknowledge: vi.fn(), drain: vi.fn(), + getMicrophoneMuted: vi.fn(), getStatus: vi.fn(), + hydrateMicrophone: vi.fn(), listen: vi.fn(), reconcileMicrophone: vi.fn(), reject: vi.fn(), @@ -18,9 +22,13 @@ const mocks = vi.hoisted(() => ({ })); vi.mock("../api/voiceConversation", () => ({ + applyVoiceConversationMicrophoneMuteEvent: mocks.applyMicrophoneMuteEvent, + applyVoiceConversationTerminalEvent: mocks.applyTerminalEvent, acknowledgeVoiceConversationTranscript: mocks.acknowledge, drainVoiceConversationTranscripts: mocks.drain, + getVoiceConversationMicrophoneMuted: mocks.getMicrophoneMuted, getVoiceConversationStatus: mocks.getStatus, + hydrateVoiceConversationMicrophone: mocks.hydrateMicrophone, listenToVoiceConversation: mocks.listen, reconcileVoiceConversationMicrophone: mocks.reconcileMicrophone, rejectVoiceConversationTranscript: mocks.reject, @@ -46,10 +54,12 @@ function status( function deferred() { let resolve!: (value: T) => void; - const promise = new Promise((resolver) => { - resolve = resolver; + let reject!: (reason?: unknown) => void; + const promise = new Promise((promiseResolve, promiseReject) => { + resolve = promiseResolve; + reject = promiseReject; }); - return { promise, resolve }; + return { promise, reject, resolve }; } describe("voice conversation store lifecycle ordering", () => { @@ -58,7 +68,10 @@ describe("voice conversation store lifecycle ordering", () => { beforeEach(() => { vi.resetModules(); mocks.acknowledge.mockReset().mockResolvedValue(undefined); + mocks.applyMicrophoneMuteEvent.mockReset(); + mocks.applyTerminalEvent.mockReset(); mocks.drain.mockReset().mockResolvedValue([]); + mocks.getMicrophoneMuted.mockReset().mockReturnValue(false); mocks.getStatus.mockReset().mockResolvedValue(status("stopped", 0)); mocks.start.mockReset(); mocks.stop.mockReset(); @@ -66,6 +79,7 @@ describe("voice conversation store lifecycle ordering", () => { emit = callback; return vi.fn(); }); + mocks.hydrateMicrophone.mockReset().mockResolvedValue(undefined); mocks.reconcileMicrophone.mockReset().mockResolvedValue(undefined); mocks.reject .mockReset() @@ -122,7 +136,167 @@ describe("voice conversation store lifecycle ordering", () => { await loadStore(); + expect(mocks.hydrateMicrophone).toHaveBeenCalledWith(running); + }); + + it("hydrates an already-muted native lifecycle", async () => { + const running = { + ...status("running", 2, "session-1"), + nativeMicrophoneMuteControl: true, + nativeMicrophoneMuted: true, + }; + mocks.getStatus.mockResolvedValue(running); + + const store = await loadStore(); + + expect(mocks.hydrateMicrophone).toHaveBeenCalledWith(running); + expect(store.getState().microphoneMuted).toBe(true); + }); + + it("refreshes mute state for the same running lifecycle", async () => { + const unmuted = { + ...status("running", 2, "session-1"), + nativeMicrophoneMuteControl: true, + nativeMicrophoneMuted: false, + }; + const muted = { ...unmuted, nativeMicrophoneMuted: true }; + mocks.getStatus.mockResolvedValueOnce(unmuted).mockResolvedValueOnce(muted); + const store = await loadStore(); + + await store.getState().init(); + + expect(mocks.hydrateMicrophone).toHaveBeenLastCalledWith(muted); + expect(store.getState().microphoneMuted).toBe(true); + }); + + it("hydrates mute state when startup arrives before status", async () => { + const response = deferred(); + mocks.getStatus.mockReturnValue(response.promise); + const { useVoiceConversationStore } = await import( + "./voiceConversationStore" + ); + const initializing = useVoiceConversationStore.getState().init(); + await vi.waitFor(() => expect(mocks.getStatus).toHaveBeenCalledOnce()); + emit({ + type: "startup", + sessionId: "session-1", + ownerWindowLabel: "main", + line: "type\tid\ttext", + revision: 2, + nativeMicrophoneMuteControl: true, + }); + const muted = { + ...status("running", 2, "session-1"), + nativeMicrophoneMuteControl: true, + nativeMicrophoneMuted: true, + }; + response.resolve(muted); + await initializing; + + expect(mocks.hydrateMicrophone).toHaveBeenCalledWith(muted); + expect(useVoiceConversationStore.getState().microphoneMuted).toBe(true); + }); + + it("preserves a mute event that arrives while hydration is pending", async () => { + const hydration = deferred(); + const muted = { + ...status("running", 2, "session-1"), + nativeMicrophoneMuteControl: true, + nativeMicrophoneMuted: true, + }; + mocks.getStatus.mockResolvedValue(muted); + mocks.hydrateMicrophone.mockReturnValue(hydration.promise); + const { useVoiceConversationStore } = await import( + "./voiceConversationStore" + ); + + const initializing = useVoiceConversationStore.getState().init(); + await vi.waitFor(() => + expect(mocks.hydrateMicrophone).toHaveBeenCalledWith(muted), + ); + emit({ + type: "inputMute", + sessionId: "session-1", + muted: false, + revision: 2, + }); + hydration.resolve(); + await initializing; + + expect(useVoiceConversationStore.getState().microphoneMuted).toBe(false); + }); + + it("does not hydrate over a pending microphone mute request", async () => { + const muteRequest = deferred(); + const recoveryStatus = deferred(); + const running = { + ...status("running", 2, "session-1"), + nativeMicrophoneMuteControl: true, + nativeMicrophoneMuted: false, + }; + mocks.getStatus.mockResolvedValue(running); + mocks.setMicrophoneMuted.mockReturnValue(muteRequest.promise); + const { useVoiceConversationStore } = await import( + "./voiceConversationStore" + ); + await useVoiceConversationStore.getState().init(); + mocks.hydrateMicrophone.mockClear(); + mocks.reconcileMicrophone.mockClear(); + mocks.getStatus.mockReturnValueOnce(recoveryStatus.promise); + + const muting = useVoiceConversationStore + .getState() + .setMicrophoneMuted(true); + await vi.waitFor(() => + expect(mocks.setMicrophoneMuted).toHaveBeenCalledWith(true, running), + ); + const recovering = useVoiceConversationStore.getState().init(); + await vi.waitFor(() => expect(mocks.getStatus).toHaveBeenCalledTimes(2)); + + muteRequest.resolve(); + await muting; + recoveryStatus.resolve(running); + await recovering; + + expect(mocks.hydrateMicrophone).not.toHaveBeenCalled(); expect(mocks.reconcileMicrophone).toHaveBeenCalledWith(running); + expect(useVoiceConversationStore.getState().microphoneMuted).toBe(true); + }); + + it("does not let stale status overwrite a newer mute event", async () => { + const response = deferred(); + const current = { + ...status("running", 3, "session-1"), + nativeMicrophoneMuteControl: true, + nativeMicrophoneMuted: false, + }; + mocks.getStatus + .mockReturnValueOnce(response.promise) + .mockResolvedValueOnce(current); + const { useVoiceConversationStore } = await import( + "./voiceConversationStore" + ); + + const initializing = useVoiceConversationStore.getState().init(); + await vi.waitFor(() => expect(mocks.getStatus).toHaveBeenCalledOnce()); + emit({ + type: "inputMute", + sessionId: "session-1", + muted: false, + revision: 3, + }); + response.resolve({ + ...status("running", 2, "session-1"), + nativeMicrophoneMuteControl: true, + nativeMicrophoneMuted: true, + }); + await initializing; + + expect(mocks.hydrateMicrophone).not.toHaveBeenCalled(); + expect(mocks.reconcileMicrophone).toHaveBeenCalledWith(current); + expect(mocks.getStatus).toHaveBeenCalledTimes(2); + expect(useVoiceConversationStore.getState().microphoneMuted).toBe(false); + expect(useVoiceConversationStore.getState().status.revision).toBe(3); }); it("refreshes availability when installation changes without a lifecycle revision", async () => { @@ -217,6 +391,23 @@ describe("voice conversation store lifecycle ordering", () => { uiState: "off", error: null, }); + expect(mocks.applyTerminalEvent).toHaveBeenCalledOnce(); + }); + + it("does not tear down capture for a stale terminal event", async () => { + const store = await loadStore(); + store.setState({ + status: status("running", 5, "session-2"), + uiState: "listening", + }); + + emit({ type: "cleanShutdown", sessionId: "session-1", revision: 4 }); + + expect(mocks.applyTerminalEvent).not.toHaveBeenCalled(); + expect(store.getState()).toMatchObject({ + status: status("running", 5, "session-2"), + uiState: "listening", + }); }); it("returns to off after a no-op stop with an unchanged revision", async () => { @@ -253,7 +444,7 @@ describe("voice conversation store lifecycle ordering", () => { await expect(first).resolves.toEqual(status("stopped", 2)); }); - it("does not let a stale start response regress a startup event", async () => { + it("preserves native mute control when startup wins the response race", async () => { const store = await loadStore(); const response = deferred(); mocks.start.mockReturnValue(response.promise); @@ -265,6 +456,61 @@ describe("voice conversation store lifecycle ordering", () => { ownerWindowLabel: "main", line: "type\tid\ttext", revision: 2, + nativeMicrophoneMuteControl: true, + }); + response.resolve({ + ...status("running", 2, "session-1"), + nativeMicrophoneMuteControl: true, + }); + await starting; + + expect(store.getState()).toMatchObject({ + status: status("running", 2, "session-1"), + uiState: "listening", + error: null, + }); + expect(store.getState().status.nativeMicrophoneMuteControl).toBe(true); + }); + + it("preserves a stem mute observed immediately before startup", async () => { + const store = await loadStore(); + + emit({ + type: "inputMute", + sessionId: "session-1", + muted: true, + revision: 2, + }); + emit({ + type: "startup", + sessionId: "session-1", + ownerWindowLabel: "main", + line: "type\tid\ttext", + revision: 2, + nativeMicrophoneMuteControl: true, + }); + + expect(mocks.applyMicrophoneMuteEvent).toHaveBeenCalledWith(true); + expect(store.getState()).toMatchObject({ + status: status("running", 2, "session-1"), + uiState: "listening", + microphoneMuted: true, + }); + }); + + it("ignores an older start response after a newer startup event", async () => { + const store = await loadStore(); + const response = deferred(); + mocks.start.mockReturnValue(response.promise); + + const starting = store.getState().start("session-1"); + emit({ + type: "startup", + sessionId: "session-1", + ownerWindowLabel: "main", + line: "type\tid\ttext", + revision: 2, + nativeMicrophoneMuteControl: true, }); response.resolve(status("starting", 1, "session-1")); await starting; @@ -274,6 +520,7 @@ describe("voice conversation store lifecycle ordering", () => { uiState: "listening", error: null, }); + expect(store.getState().status.nativeMicrophoneMuteControl).toBe(true); }); it("reconciles status after a failed stop", async () => { @@ -367,6 +614,109 @@ describe("voice conversation store lifecycle ordering", () => { }); }); + it("keeps the latest UI mute intent when requests settle out of order", async () => { + const store = await loadStore(); + store.setState({ + status: status("running", 2, "session-1"), + uiState: "listening", + }); + const mute = deferred(); + const unmute = deferred(); + mocks.setMicrophoneMuted + .mockReturnValueOnce(mute.promise) + .mockReturnValueOnce(unmute.promise); + + const muting = store.getState().setMicrophoneMuted(true); + const unmuting = store.getState().setMicrophoneMuted(false); + expect(store.getState().microphoneMuted).toBe(false); + + unmute.resolve(); + await unmuting; + mute.resolve(); + await muting; + + expect(store.getState().microphoneMuted).toBe(false); + }); + + it("rolls the latest failed intent back to the last successful mute", async () => { + const store = await loadStore(); + store.setState({ + status: status("running", 2, "session-1"), + uiState: "listening", + }); + const mute = deferred(); + const unmute = deferred(); + mocks.setMicrophoneMuted + .mockReturnValueOnce(mute.promise) + .mockReturnValueOnce(unmute.promise); + + const muting = store.getState().setMicrophoneMuted(true); + const unmuting = store.getState().setMicrophoneMuted(false); + mute.resolve(); + await muting; + mocks.getMicrophoneMuted.mockReturnValue(true); + unmute.reject(new Error("mute unavailable")); + await expect(unmuting).rejects.toThrow("mute unavailable"); + + expect(store.getState()).toMatchObject({ + microphoneMuted: true, + uiState: "error", + error: "mute unavailable", + }); + }); + + it("keeps a newer stem state after a stale UI request settles", async () => { + const store = await loadStore(); + store.setState({ + status: status("running", 2, "session-1"), + uiState: "listening", + }); + const pending = deferred(); + mocks.setMicrophoneMuted.mockReturnValueOnce(pending.promise); + + const staleMute = store.getState().setMicrophoneMuted(true); + emit({ + type: "inputMute", + sessionId: "session-1", + muted: false, + revision: 3, + }); + pending.resolve(); + await staleMute; + mocks.getMicrophoneMuted.mockReturnValue(false); + mocks.setMicrophoneMuted.mockRejectedValueOnce( + new Error("mute unavailable"), + ); + + await expect(store.getState().setMicrophoneMuted(true)).rejects.toThrow( + "mute unavailable", + ); + expect(store.getState().microphoneMuted).toBe(false); + }); + + it("applies a current stem mute event to capture and UI", async () => { + const store = await loadStore(); + store.setState({ + status: status("running", 2, "session-1"), + uiState: "user-speaking", + userSpeaking: true, + }); + + emit({ + type: "inputMute", + sessionId: "session-1", + muted: true, + revision: 3, + }); + + expect(mocks.applyMicrophoneMuteEvent).toHaveBeenCalledWith(true); + expect(store.getState()).toMatchObject({ + microphoneMuted: true, + userSpeaking: false, + uiState: "listening", + }); + }); + it("ignores stale user-speaking activity while the microphone is muted", async () => { const store = await loadStore(); store.setState({ @@ -394,8 +744,14 @@ describe("voice conversation store lifecycle ordering", () => { store.setState({ status: status("running", 2, "session-1"), uiState: "listening", - microphoneMuted: true, }); + emit({ + type: "inputMute", + sessionId: "session-1", + muted: true, + revision: 3, + }); + mocks.getMicrophoneMuted.mockReturnValue(true); mocks.setMicrophoneMuted.mockRejectedValueOnce( new Error("microphone unavailable"), ); diff --git a/src/features/voice-conversation/stores/voiceConversationStore.ts b/src/features/voice-conversation/stores/voiceConversationStore.ts index efff65a0c..b4b48459b 100644 --- a/src/features/voice-conversation/stores/voiceConversationStore.ts +++ b/src/features/voice-conversation/stores/voiceConversationStore.ts @@ -1,9 +1,13 @@ import { create } from "zustand"; import { + applyVoiceConversationTerminalEvent, + applyVoiceConversationMicrophoneMuteEvent, acknowledgeVoiceConversationTranscript, drainVoiceConversationTranscripts, + getVoiceConversationMicrophoneMuted, getVoiceConversationStatus, + hydrateVoiceConversationMicrophone, listenToVoiceConversation, reconcileVoiceConversationMicrophone, rejectVoiceConversationTranscript, @@ -56,6 +60,9 @@ interface VoiceConversationStore { let initialized = false; let stopInFlight: Promise | null = null; +let microphoneMuteIntent = 0; +let microphoneMuteStateVersion = 0; +let pendingMicrophoneMuteRequests = 0; const eventSubscribers = new Set< (event: VoiceConversationEvent) => void | Promise >(); @@ -158,6 +165,31 @@ function shouldApplyResponseRevision( return revision > current.revision; } +function isSameRunningLifecycle( + current: VoiceConversationStatus, + next: VoiceConversationStatus, +) { + return ( + current.lifecycle === "running" && + next.lifecycle === "running" && + current.sessionId === next.sessionId && + current.revision === next.revision && + (current.ownerWindowLabel === null || + current.ownerWindowLabel === next.ownerWindowLabel) + ); +} + +async function getRecoveryStatus( + currentStatus: () => VoiceConversationStatus, +): Promise { + let status = await getVoiceConversationStatus(); + const current = currentStatus(); + if (current.lifecycle === "running" && status.revision < current.revision) { + status = await getVoiceConversationStatus(); + } + return status; +} + export const useVoiceConversationStore = create( (set, get) => ({ status: VOICE_CONVERSATION_OFF_STATUS, @@ -181,8 +213,33 @@ export const useVoiceConversationStore = create( init: async () => { if (initialized) { try { - const status = await getVoiceConversationStatus(); - await reconcileVoiceConversationMicrophone(status); + const muteStateVersion = microphoneMuteStateVersion; + const muteRequestWasPending = pendingMicrophoneMuteRequests > 0; + const status = await getRecoveryStatus(() => get().status); + const currentStatus = get().status; + const shouldAdopt = shouldApplyResponseRevision( + currentStatus, + status.revision, + ); + const sameRunningLifecycle = isSameRunningLifecycle( + currentStatus, + status, + ); + const shouldReconcile = shouldAdopt || sameRunningLifecycle; + const shouldHydrate = + (shouldAdopt || sameRunningLifecycle) && + !muteRequestWasPending && + pendingMicrophoneMuteRequests === 0 && + muteStateVersion === microphoneMuteStateVersion; + if (shouldHydrate) { + await hydrateVoiceConversationMicrophone(status); + } else if (shouldReconcile) { + await reconcileVoiceConversationMicrophone(status); + } + const applyHydratedMute = + shouldHydrate && + pendingMicrophoneMuteRequests === 0 && + muteStateVersion === microphoneMuteStateVersion; set((state) => { if ( shouldApplyResponseRevision(state.status, status.revision) || @@ -196,6 +253,11 @@ export const useVoiceConversationStore = create( state.uiState === "error" ? state.uiState : uiStateForStatus(status), + microphoneMuted: applyHydratedMute + ? status.lifecycle === "running" + ? (status.nativeMicrophoneMuted ?? false) + : false + : state.microphoneMuted, hydrated: true, }; } @@ -206,6 +268,11 @@ export const useVoiceConversationStore = create( available: status.available, unavailableReason: status.unavailableReason, }, + microphoneMuted: applyHydratedMute + ? status.lifecycle === "running" + ? (status.nativeMicrophoneMuted ?? false) + : false + : state.microphoneMuted, hydrated: true, }; } @@ -229,6 +296,21 @@ export const useVoiceConversationStore = create( await listenToVoiceConversation((event) => { if (!shouldApplyEventRevision(get().status, event.revision)) return; + if (event.type === "inputMute") { + microphoneMuteIntent += 1; + microphoneMuteStateVersion += 1; + applyVoiceConversationMicrophoneMuteEvent(event.muted); + } else if (event.type === "startup") { + microphoneMuteIntent += 1; + } else if ( + event.type === "cleanShutdown" || + (event.type === "error" && event.terminal) + ) { + microphoneMuteIntent += 1; + microphoneMuteStateVersion += 1; + applyVoiceConversationTerminalEvent(); + } + set((state) => { switch (event.type) { case "startup": @@ -236,13 +318,14 @@ export const useVoiceConversationStore = create( ...state, status: { ...state.status, - lifecycle: "running", + lifecycle: "running" as const, sessionId: event.sessionId, ownerWindowLabel: event.ownerWindowLabel, revision: event.revision, + nativeMicrophoneMuteControl: + event.nativeMicrophoneMuteControl, }, uiState: "listening", - microphoneMuted: false, error: null, }; case "user": @@ -250,7 +333,7 @@ export const useVoiceConversationStore = create( ...state, status: { ...state.status, - lifecycle: "running", + lifecycle: "running" as const, sessionId: event.sessionId, revision: event.revision, }, @@ -287,6 +370,25 @@ export const useVoiceConversationStore = create( uiState: activityUiState(nextState), }; } + case "inputMute": { + const nextState = { + ...state, + status: { + ...state.status, + lifecycle: "running" as const, + sessionId: event.sessionId, + revision: event.revision, + }, + microphoneMuted: event.muted, + userSpeaking: event.muted ? false : state.userSpeaking, + }; + return { + status: nextState.status, + microphoneMuted: event.muted, + userSpeaking: nextState.userSpeaking, + uiState: activityUiState(nextState), + }; + } case "cleanShutdown": return { ...state, @@ -296,6 +398,7 @@ export const useVoiceConversationStore = create( sessionId: null, ownerWindowLabel: null, revision: event.revision, + nativeMicrophoneMuteControl: false, }, uiState: "off", userSpeaking: false, @@ -314,6 +417,7 @@ export const useVoiceConversationStore = create( sessionId: null, ownerWindowLabel: null, revision: event.revision, + nativeMicrophoneMuteControl: false, } : { ...state.status, @@ -357,23 +461,65 @@ export const useVoiceConversationStore = create( } try { - const status = await getVoiceConversationStatus(); - await reconcileVoiceConversationMicrophone(status); - set((state) => - shouldApplyResponseRevision(state.status, status.revision) || - (!state.hydrated && - state.status.revision === 0 && - state.uiState === "off") - ? { - status, - uiState: - state.uiState === "error" - ? state.uiState - : uiStateForStatus(status), - hydrated: true, - } - : { hydrated: true }, + const muteStateVersion = microphoneMuteStateVersion; + const muteRequestWasPending = pendingMicrophoneMuteRequests > 0; + const status = await getRecoveryStatus(() => get().status); + const currentStatus = get().status; + const shouldAdopt = + shouldApplyResponseRevision(currentStatus, status.revision) || + (!get().hydrated && + get().status.revision === 0 && + get().uiState === "off"); + const sameRunningLifecycle = isSameRunningLifecycle( + currentStatus, + status, ); + const shouldReconcile = shouldAdopt || sameRunningLifecycle; + const shouldHydrate = + (shouldAdopt || sameRunningLifecycle) && + !muteRequestWasPending && + pendingMicrophoneMuteRequests === 0 && + muteStateVersion === microphoneMuteStateVersion; + if (shouldHydrate) { + await hydrateVoiceConversationMicrophone(status); + } else if (shouldReconcile) { + await reconcileVoiceConversationMicrophone(status); + } + const applyHydratedMute = + shouldHydrate && + pendingMicrophoneMuteRequests === 0 && + muteStateVersion === microphoneMuteStateVersion; + set((state) => { + if ( + shouldApplyResponseRevision(state.status, status.revision) || + (!state.hydrated && + state.status.revision === 0 && + state.uiState === "off") + ) { + return { + status, + uiState: + state.uiState === "error" + ? state.uiState + : uiStateForStatus(status), + microphoneMuted: applyHydratedMute + ? status.lifecycle === "running" + ? (status.nativeMicrophoneMuted ?? false) + : false + : state.microphoneMuted, + hydrated: true, + }; + } + if (isSameRunningLifecycle(state.status, status)) { + return { + microphoneMuted: applyHydratedMute + ? (status.nativeMicrophoneMuted ?? false) + : state.microphoneMuted, + hydrated: true, + }; + } + return { hydrated: true }; + }); } catch (error) { set({ error: error instanceof Error ? error.message : String(error), @@ -384,6 +530,8 @@ export const useVoiceConversationStore = create( }, start: async (sessionId) => { + microphoneMuteIntent += 1; + microphoneMuteStateVersion += 1; set({ uiState: "starting", microphoneMuted: false, error: null }); try { const status = await startVoiceConversation(sessionId); @@ -418,6 +566,8 @@ export const useVoiceConversationStore = create( stop: () => { if (stopInFlight) return stopInFlight; + microphoneMuteIntent += 1; + microphoneMuteStateVersion += 1; set({ uiState: "stopping", microphoneMuted: false, @@ -493,18 +643,38 @@ export const useVoiceConversationStore = create( setMicrophoneMuted: async (microphoneMuted) => { const current = get(); if (current.status.lifecycle !== "running") return; + const intent = ++microphoneMuteIntent; + microphoneMuteStateVersion += 1; + pendingMicrophoneMuteRequests += 1; + set((state) => { + const nextState = { + ...state, + microphoneMuted, + userSpeaking: microphoneMuted ? false : state.userSpeaking, + }; + return { + microphoneMuted, + userSpeaking: nextState.userSpeaking, + uiState: activityUiState(nextState), + }; + }); try { await setVoiceConversationMicrophoneMuted( microphoneMuted, current.status, ); } catch (error) { + if (intent !== microphoneMuteIntent) return; set({ + microphoneMuted: getVoiceConversationMicrophoneMuted(), uiState: "error", error: error instanceof Error ? error.message : String(error), }); throw error; + } finally { + pendingMicrophoneMuteRequests -= 1; } + if (intent !== microphoneMuteIntent) return; set((state) => { if ( state.status.lifecycle !== "running" ||