diff --git a/src-tauri/crates/berd-call/README.md b/src-tauri/crates/berd-call/README.md index 7f01b1039..819d49db5 100644 --- a/src-tauri/crates/berd-call/README.md +++ b/src-tauri/crates/berd-call/README.md @@ -143,7 +143,7 @@ device; the existing Berd Siri player and the CLI use the same decoder. in [PROTOCOL.md](PROTOCOL.md). The host supplies persisted status-sound settings and semantic `working` / `waiting` updates through that protocol; the runtime owns the five-second cadence, speech suppression, and macOS Pop/Purr playback. -The default mode is `working`; `working-and-waiting` also plays idle cues, and `off` disables both. Status sounds default to a gain of `0.8`. Cues play immediately on a status transition and repeat every five seconds. Conversation audio cancels the current cue and restarts the full cadence when it ends; missed cues are never replayed. Repeated identical status updates preserve the cadence. The host determines when work begins; the chained client signals working on its first live tool call and stays working until the run ends. +The default mode is `working`; `working-and-waiting` also plays idle cues, and `off` disables both. Status sounds default to a gain of `0.8`. Cues play immediately on a status transition and repeat every five seconds. Conversation audio lets an already-playing cue finish and skips scheduled cues without shifting the cadence. A status transition during conversation audio is announced when that audio ends; skipped periodic cues are never replayed. Repeated identical status updates preserve the cadence. Status cues follow the current `--input-during-tts` policy, including the selected output route. Cue suppression pauses incoming PCM without discarding active speech or pending recognition. The host determines when work begins; the chained client signals working on its first live tool call and stays working until the run ends. Status-cue playback is single-flight: a cadence tick never queues behind a cue that is still draining. Hosts claim overlapping user and assistant audio through shared activity guards, and the runtime resumes cues only after the final guard diff --git a/src-tauri/crates/berd-call/src/host_session.rs b/src-tauri/crates/berd-call/src/host_session.rs index 166c44ede..cfa4e0d47 100644 --- a/src-tauri/crates/berd-call/src/host_session.rs +++ b/src-tauri/crates/berd-call/src/host_session.rs @@ -5159,6 +5159,7 @@ mod tests { }), input_policy: Some(InputDuringTtsPolicy::SuppressInput), status_sounds: None, + ..Default::default() } } diff --git a/src-tauri/crates/berd-call/src/input.rs b/src-tauri/crates/berd-call/src/input.rs index e0b4c51dd..3583e74c7 100644 --- a/src-tauri/crates/berd-call/src/input.rs +++ b/src-tauri/crates/berd-call/src/input.rs @@ -165,6 +165,7 @@ struct ControlState { transition: Mutex, muted: AtomicBool, mute_epoch: AtomicU64, + cue_suppressions: AtomicU64, assistant_speaking: AtomicBool, assistant_vad_threshold: AtomicU32, } @@ -294,6 +295,7 @@ impl Default for VoiceInputControls { transition: Mutex::new(ControlTransitionState::default()), muted: AtomicBool::new(false), mute_epoch: AtomicU64::new(0), + cue_suppressions: AtomicU64::new(0), assistant_speaking: AtomicBool::new(false), assistant_vad_threshold: AtomicU32::new(0.5_f32.to_bits()), }), @@ -302,6 +304,31 @@ impl Default for VoiceInputControls { } impl VoiceInputControls { + /// Pauses new captured frames without invalidating buffered speech or recognition. + pub fn begin_cue_activity( + &self, + policy: InputDuringTtsPolicy, + output_device: Option<&str>, + ) -> Result { + let suppress = policy.resolve(output_device) == InputDuringTtsPolicy::SuppressInput; + if suppress { + self.shared + .cue_suppressions + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { + count.checked_add(1) + }) + .map_err(|_| "Too many overlapping cue activity guards".to_string())?; + } + Ok(CueInputActivityGuard { + controls: self.clone(), + suppress, + }) + } + + pub fn is_input_suppressed(&self) -> bool { + self.is_muted() || self.shared.cue_suppressions.load(Ordering::Acquire) > 0 + } + pub fn is_muted(&self) -> bool { self.shared.muted.load(Ordering::Acquire) } @@ -455,6 +482,23 @@ impl VoiceInputControls { } } +#[must_use = "cue input suppression ends when the guard is dropped"] +pub struct CueInputActivityGuard { + controls: VoiceInputControls, + suppress: bool, +} + +impl Drop for CueInputActivityGuard { + fn drop(&mut self) { + if self.suppress { + self.controls + .shared + .cue_suppressions + .fetch_sub(1, Ordering::AcqRel); + } + } +} + #[must_use = "assistant activity ends when the guard is dropped"] pub struct AssistantActivityGuard { controls: VoiceInputControls, @@ -590,11 +634,11 @@ impl VoiceInputRuntime { if self.shutdown.load(Ordering::Acquire) { return Err("Voice input recognition is no longer running.".to_string()); } - if self.controls.is_muted() { + if self.controls.is_input_suppressed() { return Ok(()); } let mute_epoch = self.controls.mute_epoch(); - if self.controls.is_muted() || mute_epoch != self.controls.mute_epoch() { + if self.controls.is_input_suppressed() || mute_epoch != self.controls.mute_epoch() { return Ok(()); } match self.frame_tx.try_send(QueuedFrame { frame, mute_epoch }) { @@ -2166,6 +2210,45 @@ fn openai_worker( mod tests { use super::*; + #[test] + fn cue_auto_policy_uses_selected_speaker_route() { + let controls = VoiceInputControls::default(); + let guard = controls + .begin_cue_activity(InputDuringTtsPolicy::Auto, Some("Test Speakers")) + .unwrap(); + assert!(controls.is_input_suppressed()); + assert_eq!(controls.mute_epoch(), 0); + drop(guard); + assert!(!controls.is_input_suppressed()); + } + + #[test] + fn cue_suppression_preserves_recognition_epoch_and_composes_with_host_mute() { + let controls = VoiceInputControls::default(); + let epoch = controls.mute_epoch(); + let first = controls + .begin_cue_activity(InputDuringTtsPolicy::SuppressInput, None) + .unwrap(); + let second = controls + .begin_cue_activity(InputDuringTtsPolicy::SuppressInput, None) + .unwrap(); + assert!(controls.is_input_suppressed()); + assert!(!controls.is_muted()); + assert_eq!(controls.mute_epoch(), epoch); + drop(first); + assert!(controls.is_input_suppressed()); + drop(second); + assert!(!controls.is_input_suppressed()); + assert_eq!(controls.mute_epoch(), epoch); + let allow = controls + .begin_cue_activity(InputDuringTtsPolicy::AllowBargeIn, None) + .unwrap(); + assert!(!controls.is_input_suppressed()); + controls.set_host_muted(true); + drop(allow); + assert!(controls.is_input_suppressed()); + } + #[test] fn default_output_is_classified_by_its_name() { for name in ["USB Speakers", "Bluetooth SPEAKER", "Altavoces externos"] { @@ -3104,6 +3187,34 @@ mod tests { assert_ne!(queued.mute_epoch, controls.mute_epoch()); } + #[test] + fn cue_pause_drops_only_new_pcm_and_keeps_recognition_epoch() { + let (frame_tx, frame_rx) = mpsc::sync_channel(INPUT_QUEUE_FRAMES); + let controls = VoiceInputControls::default(); + let runtime = VoiceInputRuntime { + frame_tx, + controls: controls.clone(), + shutdown: Arc::new(AtomicBool::new(false)), + discard_on_shutdown: Arc::new(AtomicBool::new(false)), + shutdown_mute_epoch: Arc::new(AtomicU64::new(0)), + worker: None, + }; + runtime.try_push_frame(silence_frame()).unwrap(); + let before = frame_rx.try_recv().unwrap(); + let cue = controls + .begin_cue_activity(InputDuringTtsPolicy::SuppressInput, None) + .unwrap(); + runtime.try_push_frame(silence_frame()).unwrap(); + assert!(frame_rx.try_recv().is_err()); + assert_eq!(before.mute_epoch, controls.mute_epoch()); + drop(cue); + runtime.try_push_frame(silence_frame()).unwrap(); + let after = frame_rx.try_recv().unwrap(); + assert_eq!(before.mute_epoch, after.mute_epoch); + runtime.signal_shutdown(false); + assert!(!runtime.discard_on_shutdown.load(Ordering::Acquire)); + } + #[test] fn input_during_tts_policy_controls_frame_admission() { let (frame_tx, frame_rx) = mpsc::sync_channel(INPUT_QUEUE_FRAMES); diff --git a/src-tauri/crates/berd-call/src/main.rs b/src-tauri/crates/berd-call/src/main.rs index 7d4872a73..dc6137581 100644 --- a/src-tauri/crates/berd-call/src/main.rs +++ b/src-tauri/crates/berd-call/src/main.rs @@ -50,7 +50,7 @@ use berd_call::spokesperson_voice_update::{ validate_voice_update_settings, VoiceBarrierAction, VoiceUpdateAction, VoiceUpdatePurpose, VoiceUpdateQueue, VoiceUpdateRequest, VoiceUpdateTransaction, }; -use berd_call::StatusSoundRuntime; +use berd_call::ManagedStatusSoundRuntime; use berd_call::{ estimated_spoken_through_utf8, local_assets::{ @@ -2220,7 +2220,7 @@ fn run_session(config: SessionConfig, pcm_output_fd: RawFd) -> Result<(), String let mut held: Option = None; let mut active: Option = None; input_progress.enter("status_assets"); - let mut status_sound_runtime = StatusSoundRuntime::default(); + let status_sound_runtime = ManagedStatusSoundRuntime::spawn(None)?; loop { input_progress.enter("voice_events"); @@ -2295,7 +2295,8 @@ fn run_session(config: SessionConfig, pcm_output_fd: RawFd) -> Result<(), String } let conversation_active = standard_session_status_cues_suppressed(&core, active.is_some()); input_progress.enter("status_sounds"); - let status_sound_result = status_sound_runtime.poll(conversation_active); + let status_sound_result = + status_sound_runtime.queue_conversation_active(conversation_active); if let Err(message) = status_sound_result { eprintln!("status sound playback disabled: {message}"); } @@ -2463,11 +2464,13 @@ fn run_session(config: SessionConfig, pcm_output_fd: RawFd) -> Result<(), String return Ok(()); } input_progress.enter("hello_ready"); + status_sound_runtime.set_input_controls(runtime.controls())?; input_controls = Some(runtime.controls()); input_runtime = Some(runtime); input_events = Some(events); initialized = true; - status_sound_runtime.set_output_device(status_sound_output_device); + status_sound_runtime.set_output_device(status_sound_output_device)?; + status_sound_runtime.set_input_policy(input_during_tts)?; let input_policy = InputDuringTtsSlot::new(input_during_tts); let session = VoiceSessionSnapshot { tts: slot.snapshot()?, @@ -2500,7 +2503,7 @@ fn run_session(config: SessionConfig, pcm_output_fd: RawFd) -> Result<(), String status, settings, }) => { - status_sound_runtime.update(status, settings); + status_sound_runtime.update(status, settings)?; write_message( &mut writer, &SessionMessage::ConversationStatusApplied { @@ -2586,6 +2589,7 @@ fn run_session(config: SessionConfig, pcm_output_fd: RawFd) -> Result<(), String Ok(snapshot) => (InputDuringTtsOutcome::Applied, snapshot), Err(snapshot) => (InputDuringTtsOutcome::Rejected, snapshot), }; + status_sound_runtime.set_input_policy(snapshot.policy)?; write_message( &mut writer, &SessionMessage::InputDuringTtsResult { @@ -3613,7 +3617,11 @@ fn run_expert_spokesperson_session( let mut queued_tts_settings = VoiceUpdateQueue::::default(); let mut pending_voice_update: Option = None; input_progress.enter("status_assets"); - let mut status_sound_runtime = StatusSoundRuntime::default(); + let cue_input_controls = VoiceInputControls::default(); + let status_sound_runtime = ManagedStatusSoundRuntime::spawn_with_input_controls( + None, + Some(cue_input_controls.clone()), + )?; loop { input_progress.enter("maintenance"); @@ -4619,7 +4627,8 @@ fn run_expert_spokesperson_session( let conversation_active = expert_session_status_cues_suppressed(&turn_gate, active.is_some()); input_progress.enter("status_sounds"); - let status_sound_result = status_sound_runtime.poll(conversation_active); + let status_sound_result = + status_sound_runtime.queue_conversation_active(conversation_active); if let Err(message) = status_sound_result { eprintln!("status sound playback disabled: {message}"); } @@ -4695,7 +4704,7 @@ fn run_expert_spokesperson_session( } Input::Pcm(frame) => { if spokesperson_pcm_allowed( - input_muted, + input_muted || cue_input_controls.is_input_suppressed(), active.is_some(), input_during_tts_slot .as_ref() @@ -4776,6 +4785,7 @@ fn run_expert_spokesperson_session( rate: spokesperson_config.speed(), }, }; + status_sound_runtime.set_input_policy(input_during_tts)?; let input_policy = InputDuringTtsSlot::new(input_during_tts); let snapshot = VoiceSessionSnapshot { tts: match &reply_tts { @@ -4813,7 +4823,7 @@ fn run_expert_spokesperson_session( session_tts = Some(tts); input_during_tts_slot = Some(input_policy); initialized = true; - status_sound_runtime.set_output_device(status_sound_output_device); + status_sound_runtime.set_output_device(status_sound_output_device)?; turn_gate.lifecycle.session_started(Instant::now()); write_message( &mut writer, @@ -5237,7 +5247,7 @@ fn run_expert_spokesperson_session( status, settings, }) => { - status_sound_runtime.update(status, settings); + status_sound_runtime.update(status, settings)?; write_message( &mut writer, &SessionMessage::ConversationStatusApplied { @@ -5305,6 +5315,13 @@ fn run_expert_spokesperson_session( .expect("hello initialized input policy"), &mut writer, )?; + status_sound_runtime.set_input_policy( + input_during_tts_slot + .as_ref() + .expect("initialized input policy") + .snapshot()? + .policy, + )?; } Input::Request(SessionRequest::SetTtsSettings { id, @@ -13460,6 +13477,7 @@ mod tests { }), input_policy: Some(berd_call::input::InputDuringTtsPolicy::SuppressInput), status_sounds: None, + ..Default::default() }, ) .unwrap(); diff --git a/src-tauri/crates/berd-call/src/saved_settings.rs b/src-tauri/crates/berd-call/src/saved_settings.rs index a5bb2253c..80c87ce2b 100644 --- a/src-tauri/crates/berd-call/src/saved_settings.rs +++ b/src-tauri/crates/berd-call/src/saved_settings.rs @@ -15,6 +15,9 @@ pub(crate) struct SavedSettings { // Left out until chosen, so a file without one still opens in older berd-call. #[serde(default, skip_serializing_if = "Option::is_none")] pub status_sounds: Option, + // Read preferences written by the optional debug-sound build without saving its setting. + #[serde(default, rename = "vad_debug_sounds", skip_serializing)] + pub _legacy_debug_sounds: Option, } pub(crate) fn path() -> Result { @@ -235,6 +238,38 @@ pub(crate) fn with_spokesperson(arguments: &[String], on: bool, direct: bool) -> mod tests { use super::*; + #[test] + fn optional_preferences_are_ignored_without_losing_other_settings() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("settings.json"); + let settings = SavedSettings { + arguments: vec!["--rate".into(), "1.5".into()], + tts: Some(TtsSettings::Siri { + voice: "Aaron".into(), + language: "en-GB".into(), + rate: 1.5, + }), + input_policy: Some(InputDuringTtsPolicy::SuppressInput), + status_sounds: Some(StatusSoundMode::WorkingAndWaiting), + ..Default::default() + }; + let expected = serde_json::to_value(&settings).unwrap(); + let mut legacy = expected.clone(); + legacy["vad_debug_sounds"] = serde_json::json!(true); + std::fs::write(&path, serde_json::to_vec(&legacy).unwrap()).unwrap(); + let restored = load(&path).unwrap(); + assert_eq!(serde_json::to_value(&restored).unwrap(), expected); + save(&path, &restored).unwrap(); + let written: serde_json::Value = + serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap(); + assert!(written.get("vad_debug_sounds").is_none()); + assert_eq!(written, expected); + assert_eq!( + serde_json::to_value(load(&path).unwrap()).unwrap(), + expected + ); + } + #[test] fn auto_and_explicit_input_preferences_survive_reopen() { let directory = tempfile::tempdir().unwrap(); diff --git a/src-tauri/crates/berd-call/src/status_sounds.rs b/src-tauri/crates/berd-call/src/status_sounds.rs index 67e8a462c..b588f38a7 100644 --- a/src-tauri/crates/berd-call/src/status_sounds.rs +++ b/src-tauri/crates/berd-call/src/status_sounds.rs @@ -99,9 +99,10 @@ impl StatusSoundStateMachine { pub struct StatusSoundRuntime { machine: StatusSoundStateMachine, next_tick: Option, + pending_status_cue: bool, player: StatusSoundPlayer, + input_policy: crate::input::InputDuringTtsPolicy, output_device: Option, - conversation_active: bool, input_controls: Option, } @@ -110,9 +111,10 @@ impl Default for StatusSoundRuntime { Self { machine: StatusSoundStateMachine::default(), next_tick: None, + pending_status_cue: false, player: StatusSoundPlayer::new(), + input_policy: crate::input::InputDuringTtsPolicy::Auto, output_device: None, - conversation_active: false, input_controls: None, } } @@ -131,6 +133,10 @@ impl StatusSoundRuntime { } pub fn update(&mut self, status: ConversationStatus, settings: StatusSoundSettings) { + self.pending_status_cue |= self + .machine + .current + .is_some_and(|(previous, _)| previous != status); let changed = self.machine.update(status, settings); if changed { self.player.stop(); @@ -143,29 +149,27 @@ impl StatusSoundRuntime { } fn update_conversation_activity(&mut self, conversation_active: bool) { - if conversation_active { - self.stop(); - } else if self.conversation_active { + if !conversation_active && self.pending_status_cue { self.next_tick = Some(Instant::now()); + self.pending_status_cue = false; } - self.conversation_active = conversation_active; } pub fn poll(&mut self, conversation_active: bool) -> Result { self.update_conversation_activity(conversation_active); - if conversation_active { - self.player.reap(); - return Ok(false); - } self.player.reap(); let now = Instant::now(); if self.next_tick.is_some_and(|deadline| now >= deadline) { - self.next_tick = Some(now + STATUS_SOUND_INTERVAL); + let deadline = self.next_tick.unwrap(); + let elapsed = now.duration_since(deadline).as_nanos(); + let ticks = elapsed / STATUS_SOUND_INTERVAL.as_nanos() + 1; + self.next_tick = Some(deadline + STATUS_SOUND_INTERVAL.mul_f64(ticks as f64)); if let Some(cue) = self.machine.tick(conversation_active) { self.player.play( cue, self.output_device.as_deref(), self.input_controls.as_ref(), + self.input_policy, )?; } } @@ -180,6 +184,8 @@ enum StatusSoundCommand { applied: Option>>, }, OutputDevice(Option), + InputPolicy(crate::input::InputDuringTtsPolicy), + InputControls(crate::input::VoiceInputControls), Shutdown, } @@ -248,6 +254,14 @@ impl ManagedStatusSoundRuntime { runtime.set_output_device(device); None } + Ok(StatusSoundCommand::InputPolicy(policy)) => { + runtime.input_policy = policy; + None + } + Ok(StatusSoundCommand::InputControls(controls)) => { + runtime.set_input_controls(Some(controls)); + None + } Ok(StatusSoundCommand::Shutdown) | Err(RecvTimeoutError::Disconnected) => { runtime.stop(); break; @@ -273,6 +287,20 @@ impl ManagedStatusSoundRuntime { }) } + pub fn set_input_policy( + &self, + policy: crate::input::InputDuringTtsPolicy, + ) -> Result<(), String> { + self.send(StatusSoundCommand::InputPolicy(policy)) + } + + pub fn set_input_controls( + &self, + controls: crate::input::VoiceInputControls, + ) -> Result<(), String> { + self.send(StatusSoundCommand::InputControls(controls)) + } + pub fn update( &self, status: ConversationStatus, @@ -282,7 +310,17 @@ impl ManagedStatusSoundRuntime { self.send(StatusSoundCommand::Update(status, settings)) } + /// Queue activity without waiting for native audio operations. PCM-owning loops + /// must use this path so a slow output device cannot stop microphone draining. + pub fn queue_conversation_active(&self, active: bool) -> Result<(), String> { + self.update_explicit_activity(active, false) + } + pub fn set_conversation_active(&self, active: bool) -> Result<(), String> { + self.update_explicit_activity(active, true) + } + + fn update_explicit_activity(&self, active: bool, wait: bool) -> Result<(), String> { let mut activity = self .inner .activity @@ -293,7 +331,12 @@ impl ManagedStatusSoundRuntime { activity.explicit_active = active; let current = activity.is_active(); if previous != current { - if let Err(error) = self.apply_conversation_active(current) { + let result = if wait { + self.apply_conversation_active(current) + } else { + self.send_conversation_active(current, None) + }; + if let Err(error) = result { activity.explicit_active = previous_explicit; let _ = self.send_conversation_active(previous, None); return Err(error); @@ -428,6 +471,7 @@ impl StatusSoundPlayer { _cue: StatusSoundCue, _output_device: Option<&str>, _input_controls: Option<&crate::input::VoiceInputControls>, + _input_policy: crate::input::InputDuringTtsPolicy, ) -> Result<(), String> { Err("status sound playback is only available on macOS".into()) } @@ -452,7 +496,7 @@ struct ActiveStatusSound { player: Box, drain_deadline: Instant, output_tail_deadline: Option, - _input_activity: Option, + _input_activity: Option, } #[cfg(target_os = "macos")] @@ -477,6 +521,7 @@ impl StatusSoundPlayer { cue: StatusSoundCue, output_device: Option<&str>, input_controls: Option<&crate::input::VoiceInputControls>, + input_policy: crate::input::InputDuringTtsPolicy, ) -> Result<(), String> { self.reap(); if self.active.is_some() { @@ -488,27 +533,58 @@ impl StatusSoundPlayer { } .as_ref() .map_err(Clone::clone)?; + Self::play_asset( + &mut self.active, + asset, + cue.volume, + output_device, + input_controls, + input_policy, + ) + } + + fn play_asset( + active: &mut Option, + asset: &StatusSoundAsset, + volume: f32, + output_device: Option<&str>, + input_controls: Option<&crate::input::VoiceInputControls>, + input_policy: crate::input::InputDuringTtsPolicy, + ) -> Result<(), String> { let player = crate::macos_audio_output::PocketAudioPlayer::new( asset.sample_rate, 1.0, output_device, )?; + Self::enqueue_asset( + active, + Box::new(player), + asset, + volume, + input_controls, + input_policy.resolve(output_device), + ) + } + + fn enqueue_asset( + active: &mut Option, + player: Box, + asset: &StatusSoundAsset, + volume: f32, + input_controls: Option<&crate::input::VoiceInputControls>, + input_policy: crate::input::InputDuringTtsPolicy, + ) -> Result<(), String> { let samples = asset .samples .iter() - .map(|sample| sample * cue.volume) + .map(|sample| sample * volume) .collect::>(); let input_activity = input_controls - .map(|controls| { - controls.begin_assistant_activity( - 0.65, - crate::input::InputDuringTtsPolicy::SuppressInput, - ) - }) + .map(|controls| controls.begin_cue_activity(input_policy, None)) .transpose()?; - player.enqueue(&samples)?; - self.active = Some(ActiveStatusSound { - player: Box::new(player), + player.write(&samples)?; + *active = Some(ActiveStatusSound { + player, drain_deadline: Instant::now() + Duration::from_secs_f64(samples.len() as f64 / f64::from(asset.sample_rate)) + STATUS_SOUND_DRAIN_GRACE, @@ -570,6 +646,61 @@ fn load_system_sound(name: &str) -> Result { mod tests { use super::*; + #[cfg(target_os = "macos")] + #[test] + fn native_cue_enqueue_obeys_policy_and_releases_guard_on_failure() { + struct Output { + fail: bool, + } + impl crate::PcmAudioOutput for Output { + fn write(&self, _: &[f32]) -> Result<(), String> { + if self.fail { + Err("enqueue failed".into()) + } else { + Ok(()) + } + } + fn cancel(&self) {} + fn is_drained(&self) -> bool { + false + } + fn check_health(&self) -> Result<(), String> { + Ok(()) + } + fn played_frames(&self) -> u64 { + 0 + } + } + use crate::input::{InputDuringTtsPolicy, VoiceInputControls}; + for policy in [ + InputDuringTtsPolicy::AllowBargeIn, + InputDuringTtsPolicy::SuppressInput, + ] { + for fail in [false, true] { + let controls = VoiceInputControls::default(); + let mut active = None; + let result = StatusSoundPlayer::enqueue_asset( + &mut active, + Box::new(Output { fail }), + &StatusSoundAsset { + sample_rate: 48000, + samples: vec![0.01; 960], + }, + 0.08, + Some(&controls), + policy, + ); + assert_eq!(result.is_err(), fail); + assert_eq!( + controls.is_input_suppressed(), + !fail && policy == InputDuringTtsPolicy::SuppressInput + ); + drop(active); + assert!(!controls.is_input_suppressed()); + } + } + } + fn settings(mode: StatusSoundMode) -> StatusSoundSettings { StatusSoundSettings { mode, @@ -582,6 +713,38 @@ mod tests { assert_eq!(actual, Some(expected)); } + #[test] + fn session_activity_does_not_wait_for_a_stalled_audio_worker() { + let (commands, receiver) = mpsc::channel(); + let runtime = ManagedStatusSoundRuntime { + inner: Arc::new(StatusSoundWorker { + commands, + worker: Mutex::new(None), + activity: Mutex::new(StatusSoundActivityState::default()), + }), + }; + let (done, result) = mpsc::channel(); + let caller = thread::spawn(move || { + let outcome = runtime.queue_conversation_active(true); + done.send(outcome).unwrap(); + }); + // The audio worker deliberately never services the command. + let outcome = result.recv_timeout(Duration::from_millis(100)); + // Release the old blocking implementation before asserting the result. + if let Ok(StatusSoundCommand::ConversationActive { + applied: Some(applied), + .. + }) = receiver.try_recv() + { + let _ = applied.send(Ok(())); + } + caller.join().unwrap(); + assert!( + outcome.is_ok(), + "microphone processing waited for the audio worker" + ); + } + #[test] fn managed_worker_finish_joins_and_is_idempotent_across_clones() { let runtime = ManagedStatusSoundRuntime::spawn(None).unwrap(); @@ -698,6 +861,7 @@ mod tests { let explicit = StatusSoundSettings { mode: StatusSoundMode::Working, volume: 0.25, + ..Default::default() }; assert_eq!(explicit.validate().unwrap().volume, 0.25); } @@ -708,6 +872,7 @@ mod tests { assert!(StatusSoundSettings { mode: StatusSoundMode::Working, volume, + ..Default::default() } .validate() .is_err()); @@ -740,27 +905,50 @@ mod tests { } #[test] - fn working_cue_is_due_immediately_after_speech() { - let mut runtime = StatusSoundRuntime::default(); - runtime.update( - ConversationStatus::Working, - settings(StatusSoundMode::Working), - ); - runtime.update_conversation_activity(true); - runtime.update_conversation_activity(false); - assert!(runtime.next_tick.unwrap() <= Instant::now()); + fn speech_preserves_the_existing_cue_schedule() { + for status in [ConversationStatus::Working, ConversationStatus::Waiting] { + let mut runtime = StatusSoundRuntime::default(); + runtime.update(status, settings(StatusSoundMode::WorkingAndWaiting)); + let deadline = Instant::now() + STATUS_SOUND_INTERVAL; + runtime.next_tick = Some(deadline); + runtime.poll(true).unwrap(); + runtime.update_conversation_activity(false); + assert_eq!(runtime.next_tick, Some(deadline)); + } + } + + #[test] + fn status_change_during_speech_is_announced_when_speech_finishes() { + for (before, after) in [ + (ConversationStatus::Working, ConversationStatus::Waiting), + (ConversationStatus::Waiting, ConversationStatus::Working), + ] { + let mut runtime = StatusSoundRuntime::default(); + let settings = settings(StatusSoundMode::WorkingAndWaiting); + runtime.update(before, settings); + runtime.poll(true).unwrap(); + runtime.update(after, settings); + runtime.poll(true).unwrap(); + runtime.update_conversation_activity(false); + assert!(runtime.next_tick.unwrap() <= Instant::now()); + } } #[test] - fn waiting_cue_is_due_immediately_after_speech() { + fn cues_due_during_speech_are_skipped_without_catching_up() { let mut runtime = StatusSoundRuntime::default(); runtime.update( - ConversationStatus::Waiting, - settings(StatusSoundMode::WorkingAndWaiting), + ConversationStatus::Working, + settings(StatusSoundMode::Working), ); - runtime.update_conversation_activity(true); - runtime.update_conversation_activity(false); - assert!(runtime.next_tick.unwrap() <= Instant::now()); + let deadline = Instant::now() - Duration::from_secs(1); + runtime.next_tick = Some(deadline); + assert!(!runtime.poll(true).unwrap()); + let next = runtime.next_tick.unwrap(); + assert!(next > Instant::now()); + assert_eq!(next, deadline + STATUS_SOUND_INTERVAL); + assert!(!runtime.poll(false).unwrap()); + assert_eq!(runtime.next_tick, Some(next)); } #[test] @@ -781,6 +969,37 @@ mod tests { runtime.finish().unwrap(); } + #[cfg(target_os = "macos")] + #[test] + fn speech_does_not_cancel_a_sample_already_playing() { + struct PendingOutput; + impl crate::PcmAudioOutput for PendingOutput { + fn write(&self, _: &[f32]) -> Result<(), String> { + Ok(()) + } + fn cancel(&self) {} + fn is_drained(&self) -> bool { + false + } + fn check_health(&self) -> Result<(), String> { + Ok(()) + } + fn played_frames(&self) -> u64 { + 0 + } + } + let mut runtime = StatusSoundRuntime::default(); + runtime.player.active = Some(ActiveStatusSound { + player: Box::new(PendingOutput), + drain_deadline: Instant::now() + Duration::from_secs(10), + output_tail_deadline: None, + _input_activity: None, + }); + assert!(runtime.poll(true).unwrap()); + assert!(runtime.player.active.is_some()); + runtime.stop(); + } + #[cfg(target_os = "macos")] #[test] fn failed_or_expired_pending_cue_releases_microphone_suppression() { @@ -796,7 +1015,7 @@ mod tests { } fn cancel(&self) { assert!( - self.controls.is_muted(), + self.controls.is_input_suppressed(), "cancel before releasing input suppression" ); self.cancelled.store(true, Ordering::SeqCst); @@ -835,21 +1054,21 @@ mod tests { output_tail_deadline: None, _input_activity: Some( controls - .begin_assistant_activity( - 0.65, + .begin_cue_activity( crate::input::InputDuringTtsPolicy::SuppressInput, + None, ) .unwrap(), ), }), }; - assert!(controls.is_muted()); + assert!(controls.is_input_suppressed()); player.reap(); assert_eq!(player.is_active(), !failed && !expired); - assert_eq!(controls.is_muted(), !failed && !expired); + assert_eq!(controls.is_input_suppressed(), !failed && !expired); assert_eq!(cancelled.load(Ordering::SeqCst), failed || expired); player.stop(); - assert!(!controls.is_muted()); + assert!(!controls.is_input_suppressed()); } } @@ -868,13 +1087,14 @@ mod tests { }, None, Some(&controls), + crate::input::InputDuringTtsPolicy::SuppressInput, ) .unwrap(); } assert!(player.active.is_some()); - assert!(controls.is_muted()); + assert!(controls.is_input_suppressed()); player.stop(); - assert!(!controls.is_muted()); + assert!(!controls.is_input_suppressed()); } #[cfg(target_os = "macos")] @@ -891,15 +1111,16 @@ mod tests { }, None, Some(&controls), + crate::input::InputDuringTtsPolicy::SuppressInput, ) .unwrap(); let deadline = Instant::now() + Duration::from_secs(5); while player.is_active() { - assert!(controls.is_muted()); + assert!(controls.is_input_suppressed()); assert!(Instant::now() < deadline, "status cue did not drain"); player.reap(); thread::sleep(Duration::from_millis(10)); } - assert!(!controls.is_muted()); + assert!(!controls.is_input_suppressed()); } }