Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 11 additions & 9 deletions src-tauri/crates/berd-call/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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::{
Expand Down Expand Up @@ -2220,7 +2220,7 @@ fn run_session(config: SessionConfig, pcm_output_fd: RawFd) -> Result<(), String
let mut held: Option<HeldPrepare> = None;
let mut active: Option<ActivePlayback> = 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");
Expand Down Expand Up @@ -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}");
}
Expand Down Expand Up @@ -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()?,
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -3613,7 +3614,7 @@ fn run_expert_spokesperson_session(
let mut queued_tts_settings = VoiceUpdateQueue::<PendingSpokespersonSettingsUpdate>::default();
let mut pending_voice_update: Option<VoiceUpdateTransaction> = 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");
Expand Down Expand Up @@ -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}");
}
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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 {
Expand Down
49 changes: 48 additions & 1 deletion src-tauri/crates/berd-call/src/status_sounds.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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);
Expand Down Expand Up @@ -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();
Expand Down
Loading