diff --git a/src-tauri/crates/berd-call/src/main.rs b/src-tauri/crates/berd-call/src/main.rs index 7d4872a73..ed94a06ae 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}"); } @@ -2467,7 +2468,7 @@ fn run_session(config: SessionConfig, pcm_output_fd: RawFd) -> Result<(), String 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)?; let input_policy = InputDuringTtsSlot::new(input_during_tts); let session = VoiceSessionSnapshot { tts: slot.snapshot()?, @@ -2500,7 +2501,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 { @@ -3613,7 +3614,7 @@ 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 status_sound_runtime = ManagedStatusSoundRuntime::spawn(None)?; loop { input_progress.enter("maintenance"); @@ -4619,7 +4620,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}"); } @@ -4813,7 +4815,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 +5239,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 { diff --git a/src-tauri/crates/berd-call/src/status_sounds.rs b/src-tauri/crates/berd-call/src/status_sounds.rs index 85a824636..f8b9c3a16 100644 --- a/src-tauri/crates/berd-call/src/status_sounds.rs +++ b/src-tauri/crates/berd-call/src/status_sounds.rs @@ -278,7 +278,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 @@ -289,7 +299,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); @@ -578,6 +593,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();