External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(desktop): serialize Pocket voice transitions

Browse Source 
Signed-off-by: John Tennant <jtennant@squareup.com>
This commit is contained in:
John Tennant
2026-07-29 12:20:48 -04:00
committed by John Tennant
parent 5e70797c69
commit 68d9c37fc0
5 changed files with 940 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
desktop/src-tauri/src/app_state.rs
+10 -4
View File
@@ -30,8 +30,9 @@ pub struct AppState {
/// Workspace-provided relay URL override. Set by `apply_workspace` on app
/// init and takes priority over env vars and compile-time defaults.
pub relay_url_override: Mutex<Option<String>>,
/// Set during setup when managed agents are eligible for launch restore;
/// consumed after workspace identity/relay install to avoid the fallback.
/// Set during backend setup when managed agents are eligible for launch
/// restore. `apply_workspace` consumes it after installing the workspace
/// relay and identity, so agents never start against the fallback relay.
pub managed_agent_restore_pending: AtomicBool,
/// Whether desktop may repair managed-agent kind:0 profiles from its local
/// records. Disabled by the agent-managed profiles experiment so an agent's
@@ -39,8 +40,8 @@ pub struct AppState {
pub managed_agent_profile_reconcile_enabled: AtomicBool,
/// Shared shutdown signal checked by launch-time agent restoration.
pub shutdown_started: AtomicBool,
/// Serializes managed-runtime transitions that change the protected PID set:
/// spawn/register, adoption, stop, shutdown, and sweep snapshots.
/// Serializes every managed-runtime transition that changes the protected
/// PID set: spawn/register, adoption, stop, shutdown, and sweep snapshots.
/// Never perform network I/O while holding this lock.
pub managed_agent_runtime_transition: Mutex<()>,
pub managed_agents_store_lock: Mutex<()>,
@@ -50,6 +51,11 @@ pub struct AppState {
pub tts_settings: Mutex<crate::huddle::tts_settings::TtsSettings>,
pub tts_settings_load_error: Mutex<Option<String>>,
pub tts_settings_transition: tokio::sync::Mutex<()>,
/// Tauri app handle — stored after setup so huddle commands can emit
/// `huddle-state-changed` events without needing the handle threaded
/// through every call site.
///
/// Set once during `setup()` in `lib.rs`; never cleared.
pub app_handle: Mutex<Option<AppHandle>>,
/// Selected audio output device name. `None` = system default.
/// Used by `connect_audio_relay` and TTS pipeline when opening sinks.
desktop/src-tauri/src/huddle/pipeline.rs
+135 -16
View File
@@ -16,7 +16,7 @@ use crate::events;
use super::models;
use super::relay_api::{self, fetch_channel_members, parse_channel_uuid};
use super::state::{HuddlePhase, VoiceInputMode};
use super::state::{HuddlePhase, HuddleState, VoiceInputMode};
use super::stt;
use super::tts;
@@ -216,7 +216,7 @@ pub(crate) async fn maybe_start_tts_pipeline(state: &AppState) -> Result<bool, S
.lock()
.unwrap_or_else(|e| e.into_inner())
.clone();
let voice = state
let initial_voice = state
.tts_settings
.lock()
.map_err(|error| format!("text-to-speech settings lock poisoned: {error}"))
@@ -224,7 +224,13 @@ pub(crate) async fn maybe_start_tts_pipeline(state: &AppState) -> Result<bool, S
super::tts_settings::pocket_voice_name(&settings.voice_preferences).to_string()
})?;
let constructed = tokio::task::spawn_blocking(move || {
tts::TtsPipeline::new_with_voice(model_dir, tts_active, tts_cancel, &voice, output_device)
tts::TtsPipeline::new_with_voice(
model_dir,
tts_active,
tts_cancel,
&initial_voice,
output_device,
)
})
.await;
let pipeline = match constructed {
@@ -241,20 +247,38 @@ pub(crate) async fn maybe_start_tts_pipeline(state: &AppState) -> Result<bool, S
}
};
{
let mut hs = state.huddle()?;
hs.tts_starting.store(false, Ordering::Release);
// Phase check: huddle may have been torn down during construction.
if !matches!(hs.phase, HuddlePhase::Connected | HuddlePhase::Active) {
return Ok(false);
}
// Final check: another path may have created a pipeline while we were constructing.
if hs.tts_pipeline.is_some() {
return Ok(false);
}
hs.tts_pipeline = Some(pipeline);
}
finalize_tts_pipeline_start(state, move |voice, huddle| {
pipeline.select_voice_before_publish(voice);
huddle.tts_pipeline = Some(pipeline);
})
}
/// Publish a constructed TTS pipeline against the latest settings.
///
/// Construction happens outside locks and can overlap a voice change or OFF
/// transition. Holding the huddle lock while re-reading settings gives either
/// transition a safe ordering: it updates the installed pipeline afterward,
/// or this finalizer observes the new setting before publishing.
fn finalize_tts_pipeline_start(
state: &AppState,
publish: impl FnOnce(&str, &mut HuddleState),
) -> Result<bool, String> {
let mut huddle = state.huddle()?;
huddle.tts_starting.store(false, Ordering::Release);
if !huddle.tts_enabled
|| !matches!(huddle.phase, HuddlePhase::Connected | HuddlePhase::Active)
|| huddle.tts_pipeline.is_some()
{
return Ok(false);
}
let voice = state
.tts_settings
.lock()
.map_err(|error| format!("text-to-speech settings lock poisoned: {error}"))
.map(|settings| {
super::tts_settings::pocket_voice_name(&settings.voice_preferences).to_string()
})?;
publish(&voice, &mut huddle);
Ok(true)
}
@@ -364,3 +388,98 @@ pub(crate) fn spawn_transcription_task(
}
});
}
#[cfg(test)]
mod tts_start_race_tests {
use std::sync::{atomic::Ordering, Arc, Barrier, Mutex};
use crate::app_state::build_app_state;
use super::{finalize_tts_pipeline_start, HuddlePhase};
#[test]
fn construction_reconciles_a_voice_selected_while_starting() {
let state = Arc::new(build_app_state());
{
let mut huddle = state.huddle().expect("huddle state");
huddle.phase = HuddlePhase::Active;
huddle.tts_enabled = true;
huddle.tts_starting.store(true, Ordering::Release);
}
let constructed = Arc::new(Barrier::new(2));
let publish = Arc::new(Barrier::new(2));
let selected_voice = Arc::new(Mutex::new(None));
let worker_state = Arc::clone(&state);
let worker_constructed = Arc::clone(&constructed);
let worker_publish = Arc::clone(&publish);
let worker_voice = Arc::clone(&selected_voice);
let worker = std::thread::spawn(move || {
worker_constructed.wait();
worker_publish.wait();
finalize_tts_pipeline_start(&worker_state, |voice, _| {
*worker_voice.lock().expect("selected voice") = Some(voice.to_string());
})
});
constructed.wait();
assert!(state
.huddle()
.expect("huddle state")
.tts_starting
.load(Ordering::Acquire));
state
.tts_settings
.lock()
.expect("text-to-speech settings")
.voice_preferences = vec!["pocket:marius".to_string()];
publish.wait();
assert!(worker.join().expect("starter thread").expect("finalize"));
assert_eq!(
*selected_voice.lock().expect("selected voice"),
Some("marius".to_string())
);
}
#[test]
fn construction_is_discarded_when_disabled_while_starting() {
let state = Arc::new(build_app_state());
{
let mut huddle = state.huddle().expect("huddle state");
huddle.phase = HuddlePhase::Active;
huddle.tts_enabled = true;
huddle.tts_starting.store(true, Ordering::Release);
}
let constructed = Arc::new(Barrier::new(2));
let publish = Arc::new(Barrier::new(2));
let did_publish = Arc::new(Mutex::new(false));
let worker_state = Arc::clone(&state);
let worker_constructed = Arc::clone(&constructed);
let worker_publish = Arc::clone(&publish);
let worker_did_publish = Arc::clone(&did_publish);
let worker = std::thread::spawn(move || {
worker_constructed.wait();
worker_publish.wait();
finalize_tts_pipeline_start(&worker_state, |_, _| {
*worker_did_publish.lock().expect("publish flag") = true;
})
});
constructed.wait();
{
let mut huddle = state.huddle().expect("huddle state");
huddle.tts_enabled = false;
}
publish.wait();
assert!(!worker.join().expect("starter thread").expect("finalize"));
assert!(!*did_publish.lock().expect("publish flag"));
assert!(!state
.huddle()
.expect("huddle state")
.tts_starting
.load(Ordering::Acquire));
}
}
desktop/src-tauri/src/huddle/tts.rs
+348 -91
View File
@@ -35,10 +35,11 @@
//! can gate microphone input while the agent is speaking.
use std::{
collections::VecDeque,
num::NonZero,
path::PathBuf,
sync::{
atomic::{AtomicBool, Ordering},
atomic::{AtomicBool, AtomicU64, Ordering},
mpsc::{self, SyncSender},
Arc, Mutex, MutexGuard, PoisonError,
},
@@ -46,9 +47,50 @@ use std::{
time::Duration,
};
use super::pocket::{load_text_to_speech, load_voice_style, SAMPLE_RATE, VOICE_FILE_EXT};
use super::pocket::{
load_text_to_speech, load_voice_style, VoiceStyle, SAMPLE_RATE, VOICE_FILE_EXT,
};
use super::preprocessing::{preprocess_for_tts, split_sentences};
#[derive(Debug)]
struct PendingVoiceChange {
generation: u64,
acknowledged: tokio::sync::oneshot::Sender<()>,
}
type VoiceChangeAck = Arc<Mutex<Option<PendingVoiceChange>>>;
type WorkerVoiceState = (Arc<Mutex<String>>, Arc<AtomicU64>, VoiceChangeAck);
type WorkerCancelSignals = (Arc<AtomicBool>, Arc<AtomicBool>);
type CancelTextState<'a> = (
&'a mpsc::Receiver<QueuedText>,
&'a mut VecDeque<QueuedText>,
&'a mut Option<QueuedText>,
);
type CancelSignals<'a> = (&'a AtomicBool, &'a AtomicBool);
#[derive(Debug)]
struct QueuedText {
generation: u64,
text: String,
}
#[derive(Clone, Debug)]
pub(crate) struct TtsTextSender {
text_tx: SyncSender<QueuedText>,
generation: u64,
}
impl TtsTextSender {
pub(crate) fn send(&self, text: String) -> Result<(), String> {
self.text_tx
.send(QueuedText {
generation: self.generation,
text,
})
.map_err(|error| error.to_string())
}
}
// ── Constants ─────────────────────────────────────────────────────────────────
/// Maximum number of queued text items.
@@ -117,7 +159,7 @@ const INTER_SENTENCE_SILENCE: f32 = 0.1;
#[derive(Debug)]
pub struct TtsPipeline {
/// Send preprocessed text into the pipeline.
text_tx: SyncSender<String>,
text_tx: SyncSender<QueuedText>,
/// `true` while the agent is speaking. Shared with the STT pipeline for gating.
#[allow(dead_code)]
pub tts_active: Arc<AtomicBool>,
@@ -127,9 +169,16 @@ pub struct TtsPipeline {
/// Kept alive here so the Arc isn't dropped — the worker holds a clone.
#[allow(dead_code)]
cancel: Arc<AtomicBool>,
/// Internal cancellation used only for voice changes. Kept separate so a
/// concurrent human barge-in always clears every queued message.
voice_cancel: Arc<AtomicBool>,
/// Selected manifest voice. The worker reloads only the lightweight style
/// when this changes; the warmed Pocket engine and audio player stay alive.
voice: Arc<Mutex<String>>,
/// Tags messages so a voice change drops only pre-change queue entries.
voice_generation: Arc<AtomicU64>,
/// Completed after the worker drains pre-change text and installs the new style.
voice_change_ack: VoiceChangeAck,
/// Worker thread handle — taken on drop to join cleanly.
thread: Option<thread::JoinHandle<()>>,
}
@@ -146,15 +195,21 @@ impl TtsPipeline {
voice: &str,
output_device: Option<String>,
) -> Result<Self, String> {
let (text_tx, text_rx) = mpsc::sync_channel::<String>(TEXT_QUEUE_DEPTH);
let (text_tx, text_rx) = mpsc::sync_channel::<QueuedText>(TEXT_QUEUE_DEPTH);
let shutdown = Arc::new(AtomicBool::new(false));
// cancel is passed in from HuddleState.tts_cancel — shared with STT for barge-in.
let shutdown_worker = Arc::clone(&shutdown);
let cancel_worker = Arc::clone(&cancel);
let voice_cancel = Arc::new(AtomicBool::new(false));
let worker_voice_cancel = Arc::clone(&voice_cancel);
let tts_active_worker = Arc::clone(&tts_active);
let voice = Arc::new(Mutex::new(voice.to_string()));
let voice_worker = Arc::clone(&voice);
let voice_generation = Arc::new(AtomicU64::new(1));
let worker_voice_generation = Arc::clone(&voice_generation);
let voice_change_ack = Arc::new(Mutex::new(None));
let worker_voice_change_ack = Arc::clone(&voice_change_ack);
let model_dir_worker = model_dir.clone();
let handle = thread::Builder::new()
@@ -162,11 +217,15 @@ impl TtsPipeline {
.spawn(move || {
tts_worker(
model_dir_worker,
voice_worker,
(
voice_worker,
worker_voice_generation,
worker_voice_change_ack,
),
text_rx,
tts_active_worker,
shutdown_worker,
cancel_worker,
(cancel_worker, worker_voice_cancel),
output_device,
)
})
@@ -177,7 +236,10 @@ impl TtsPipeline {
tts_active,
shutdown,
cancel,
voice_cancel,
voice,
voice_generation,
voice_change_ack,
thread: Some(handle),
})
}
@@ -187,17 +249,25 @@ impl TtsPipeline {
/// Non-blocking. Returns `Err` if the queue is full (bounded at
/// `TEXT_QUEUE_DEPTH`) — caller may log and discard.
pub fn speak(&self, text: String) -> Result<(), String> {
self.text_tx.try_send(text).map_err(|e| {
eprintln!("buzz-desktop: TTS queue saturated, dropping message: {e}");
format!("TTS queue full, dropping: {e}")
})
self.text_tx
.try_send(QueuedText {
generation: self.voice_generation.load(Ordering::Acquire),
text,
})
.map_err(|e| {
eprintln!("buzz-desktop: TTS queue saturated, dropping message: {e}");
format!("TTS queue full, dropping: {e}")
})
}
/// Clone the bounded queue sender so callers can apply backpressure without
/// holding the huddle mutex. Disabling TTS drops the receiver and unblocks
/// any waiting sender while the shared cancellation flag stops playback.
pub(crate) fn text_sender(&self) -> SyncSender<String> {
self.text_tx.clone()
pub(crate) fn text_sender(&self) -> TtsTextSender {
TtsTextSender {
text_tx: self.text_tx.clone(),
generation: self.voice_generation.load(Ordering::Acquire),
}
}
/// Select a bundled Pocket voice for subsequent speech.
@@ -205,9 +275,23 @@ impl TtsPipeline {
/// Current playback and queued text are cancelled immediately so content
/// cannot continue in the old voice. The worker keeps its warmed inference
/// engine and reloads only the reference style before the next utterance.
pub fn select_voice(&self, voice: &str) {
pub fn select_voice(&self, voice: &str) -> Option<tokio::sync::oneshot::Receiver<()>> {
begin_voice_change(
&self.voice,
&self.voice_generation,
&self.voice_cancel,
&self.voice_change_ack,
voice,
)
}
/// Reconcile the voice of a pipeline that has not been published yet.
///
/// No caller can enqueue text before publication, so raising the shared
/// cancellation flag here would create a race that could discard the first
/// message queued immediately after installation.
pub(crate) fn select_voice_before_publish(&self, voice: &str) {
*self.voice.lock().unwrap_or_else(|error| error.into_inner()) = voice.to_string();
self.cancel.store(true, Ordering::Release);
}
/// Signal the worker thread to stop.
@@ -237,13 +321,15 @@ impl Drop for TtsPipeline {
fn tts_worker(
model_dir: PathBuf,
selected_voice: Arc<Mutex<String>>,
text_rx: mpsc::Receiver<String>,
voice_state: WorkerVoiceState,
text_rx: mpsc::Receiver<QueuedText>,
tts_active: Arc<AtomicBool>,
shutdown: Arc<AtomicBool>,
cancel: Arc<AtomicBool>,
cancel_signals: WorkerCancelSignals,
output_device: Option<String>,
) {
let (selected_voice, voice_generation, voice_change_ack) = voice_state;
let (cancel, voice_cancel) = cancel_signals;
// ── 1. Initialise TTS engine ──────────────────────────────────────────────
let model_dir_str = model_dir.to_string_lossy().to_string();
@@ -254,7 +340,12 @@ fn tts_worker(
"buzz-desktop: TTS engine init failed (model_dir={}): {e}. TTS disabled.",
model_dir.display()
);
drain_until_shutdown(text_rx, &shutdown);
drain_tts_until_shutdown(
text_rx,
&shutdown,
(&cancel, &voice_cancel),
&voice_change_ack,
);
return;
}
};
@@ -271,7 +362,12 @@ fn tts_worker(
eprintln!(
"buzz-desktop: TTS voice style load failed ({voice_name}): {e}. TTS disabled."
);
drain_until_shutdown(text_rx, &shutdown);
drain_tts_until_shutdown(
text_rx,
&shutdown,
(&cancel, &voice_cancel),
&voice_change_ack,
);
return;
}
};
@@ -304,7 +400,12 @@ fn tts_worker(
Ok(h) => h,
Err(e) => {
eprintln!("buzz-desktop: TTS audio output failed: {e}. TTS disabled.");
drain_until_shutdown(text_rx, &shutdown);
drain_tts_until_shutdown(
text_rx,
&shutdown,
(&cancel, &voice_cancel),
&voice_change_ack,
);
return;
}
};
@@ -374,6 +475,7 @@ fn tts_worker(
let monitor = {
let player = Arc::clone(&player);
let cancel = Arc::clone(&cancel);
let voice_cancel = Arc::clone(&voice_cancel);
let tts_active = Arc::clone(&tts_active);
let stop = Arc::clone(&monitor_stop);
let player_ops = Arc::clone(&player_ops);
@@ -381,12 +483,12 @@ fn tts_worker(
.name("tts-barge-in-monitor".into())
.spawn(move || {
while !stop.load(Ordering::Acquire) {
if cancel.load(Ordering::Acquire) {
if cancel.load(Ordering::Acquire) || voice_cancel.load(Ordering::Acquire) {
let _ops = lock_player_ops(&player_ops);
// Re-check under the lock: the worker may have
// consumed this cancel (and appended fresh audio)
// between the load above and the lock acquisition.
if cancel.load(Ordering::Acquire) {
if cancel.load(Ordering::Acquire) || voice_cancel.load(Ordering::Acquire) {
// clear() pauses the persistent player; play()
// un-pauses (see handle_cancel_or_shutdown).
// Idempotent — safe to repeat every tick until
@@ -420,13 +522,16 @@ fn tts_worker(
// idle branch below uses it to decide when to drop `tts_active` and to
// arm a fresh lead-in cushion for the next utterance.
let mut first_append = true;
let mut deferred_text = VecDeque::new();
loop {
let mut no_current_text = None;
if handle_cancel_or_shutdown(
&cancel,
(&cancel, &voice_cancel),
&shutdown,
&tts_active,
&text_rx,
(&text_rx, &mut deferred_text, &mut no_current_text),
&voice_change_ack,
Some((&player, &player_ops)),
) {
if shutdown.load(Ordering::Acquire) {
@@ -441,65 +546,39 @@ fn tts_worker(
// Voice changes cancel the old utterance/queue and are observed here,
// before receiving subsequent text. A bad bundled asset falls back to
// Mary without discarding the already-warmed Pocket engine.
let requested_voice = selected_voice
.lock()
.unwrap_or_else(|error| error.into_inner())
.clone();
if requested_voice != voice_name {
let requested_path = model_dir.join(format!("{requested_voice}.{VOICE_FILE_EXT}"));
match load_voice_style(&requested_path) {
Ok(requested_style) => {
style = requested_style;
voice_name = requested_voice;
}
Err(error) => {
use super::pocket::DEFAULT_VOICE;
eprintln!(
"buzz-desktop: Pocket voice {requested_voice} is unavailable ({error}); falling back to Mary"
);
let fallback_path = model_dir.join(format!("{DEFAULT_VOICE}.{VOICE_FILE_EXT}"));
match load_voice_style(&fallback_path) {
Ok(fallback_style) => {
style = fallback_style;
voice_name = DEFAULT_VOICE.to_string();
*selected_voice
.lock()
.unwrap_or_else(|lock_error| lock_error.into_inner()) =
DEFAULT_VOICE.to_string();
}
Err(fallback_error) => {
eprintln!(
"buzz-desktop: Mary voice fallback is unavailable: {fallback_error}"
);
continue;
}
}
}
}
let voice_ready =
reconcile_selected_voice(&model_dir, &selected_voice, &mut voice_name, &mut style);
acknowledge_voice_change(&voice_change_ack, &voice_cancel);
if !voice_ready {
continue;
}
let raw_text = match text_rx.recv_timeout(RECV_TIMEOUT) {
Ok(t) => t,
Err(mpsc::RecvTimeoutError::Timeout) => {
// Nothing queued. If playback has also finished, the agent
// has gone quiet — release the mic gate and reset the
// lead-in so the next utterance gets a fresh cushion.
if player.empty() && !first_append {
tts_active.store(false, Ordering::Release);
first_append = true;
let mut queued_text = Some(match deferred_text.pop_front() {
Some(text) => text,
None => match text_rx.recv_timeout(RECV_TIMEOUT) {
Ok(text) => text,
Err(mpsc::RecvTimeoutError::Timeout) => {
// Nothing queued. If playback has also finished, the agent
// has gone quiet — release the mic gate and reset the
// lead-in so the next utterance gets a fresh cushion.
if player.empty() && !first_append {
tts_active.store(false, Ordering::Release);
first_append = true;
}
continue;
}
continue;
}
Err(mpsc::RecvTimeoutError::Disconnected) => break,
};
Err(mpsc::RecvTimeoutError::Disconnected) => break,
},
});
// Check cancel again after unblocking — a cancel may have arrived
// while we were waiting.
if handle_cancel_or_shutdown(
&cancel,
(&cancel, &voice_cancel),
&shutdown,
&tts_active,
&text_rx,
(&text_rx, &mut deferred_text, &mut queued_text),
&voice_change_ack,
Some((&player, &player_ops)),
) {
if shutdown.load(Ordering::Acquire) {
@@ -508,6 +587,21 @@ fn tts_worker(
first_append = true;
continue;
}
let Some(queued_text) = queued_text else {
continue;
};
if queued_text.generation < voice_generation.load(Ordering::Acquire) {
continue;
}
let raw_text = queued_text.text;
// The selected voice can change while this worker is blocked in
// recv_timeout. Reconcile again after receipt so the first message
// queued after an unpublished pipeline is installed cannot use the
// voice captured when construction began.
if !reconcile_selected_voice(&model_dir, &selected_voice, &mut voice_name, &mut style) {
continue;
}
// If playback already drained while we were waiting for this item,
// the agent is silent — release the mic gate BEFORE preprocessing/
@@ -541,11 +635,13 @@ fn tts_worker(
let chunks = group_sentences_into_chunks(&sentences, MAX_CHUNK_CHARS);
'playback_chunks: for chunk in &chunks {
let mut no_current_text = None;
if handle_cancel_or_shutdown(
&cancel,
(&cancel, &voice_cancel),
&shutdown,
&tts_active,
&text_rx,
(&text_rx, &mut deferred_text, &mut no_current_text),
&voice_change_ack,
Some((&player, &player_ops)),
) {
first_append = true;
@@ -566,11 +662,13 @@ fn tts_worker(
};
let model_chunk_count = model_chunks.len();
for (model_chunk_index, model_chunk) in model_chunks.iter().enumerate() {
let mut no_current_text = None;
if handle_cancel_or_shutdown(
&cancel,
(&cancel, &voice_cancel),
&shutdown,
&tts_active,
&text_rx,
(&text_rx, &mut deferred_text, &mut no_current_text),
&voice_change_ack,
Some((&player, &player_ops)),
) {
first_append = true;
@@ -608,11 +706,11 @@ fn tts_worker(
// does the full consume (drain queue, reset lead-in) on
// the next iteration.
let _ops = lock_player_ops(&player_ops);
if cancel.load(Ordering::Acquire) {
if cancel.load(Ordering::Acquire) || voice_cancel.load(Ordering::Acquire) {
// Nothing appended; the loop-top consume re-arms
// `first_append` (the flag is still set — the worker
// is its only consumer).
break;
break 'playback_chunks;
}
player.append(SamplesBuffer::new(channels, rate, buf));
// NOTE: tts_active is set AFTER player.append(), not
@@ -643,11 +741,133 @@ fn tts_worker(
let _ = handle.join();
}
finish_voice_change_ack(&voice_change_ack);
tts_active.store(false, Ordering::Release);
}
// ── Helpers ───────────────────────────────────────────────────────────────────
fn begin_voice_change(
selected_voice: &Mutex<String>,
voice_generation: &AtomicU64,
voice_cancel: &AtomicBool,
voice_change_ack: &VoiceChangeAck,
voice: &str,
) -> Option<tokio::sync::oneshot::Receiver<()>> {
let mut pending_ack = voice_change_ack
.lock()
.unwrap_or_else(|error| error.into_inner());
let mut selected = selected_voice
.lock()
.unwrap_or_else(|error| error.into_inner());
if selected.as_str() == voice {
return None;
}
let (sender, receiver) = tokio::sync::oneshot::channel();
voice_cancel.store(true, Ordering::Release);
let generation = voice_generation.fetch_add(1, Ordering::AcqRel) + 1;
if let Some(superseded) = pending_ack.replace(PendingVoiceChange {
generation,
acknowledged: sender,
}) {
let _ = superseded.acknowledged.send(());
}
*selected = voice.to_string();
Some(receiver)
}
fn acknowledge_voice_change(voice_change_ack: &VoiceChangeAck, voice_cancel: &AtomicBool) {
let mut pending_ack = voice_change_ack
.lock()
.unwrap_or_else(|error| error.into_inner());
if voice_cancel.load(Ordering::Acquire) {
return;
}
if let Some(pending) = pending_ack.take() {
let _ = pending.acknowledged.send(());
}
}
fn finish_voice_change_ack(voice_change_ack: &VoiceChangeAck) {
if let Some(pending) = voice_change_ack
.lock()
.unwrap_or_else(|error| error.into_inner())
.take()
{
let _ = pending.acknowledged.send(());
}
}
fn drain_tts_until_shutdown(
text_rx: mpsc::Receiver<QueuedText>,
shutdown: &AtomicBool,
cancel_signals: CancelSignals<'_>,
voice_change_ack: &VoiceChangeAck,
) {
let (cancel, voice_cancel) = cancel_signals;
loop {
if cancel.swap(false, Ordering::AcqRel) | voice_cancel.swap(false, Ordering::AcqRel) {
while text_rx.try_recv().is_ok() {}
}
acknowledge_voice_change(voice_change_ack, voice_cancel);
if shutdown.load(Ordering::Acquire) {
break;
}
match text_rx.recv_timeout(RECV_TIMEOUT) {
Ok(_) | Err(mpsc::RecvTimeoutError::Timeout) => {}
Err(mpsc::RecvTimeoutError::Disconnected) => break,
}
}
finish_voice_change_ack(voice_change_ack);
}
fn reconcile_selected_voice(
model_dir: &std::path::Path,
selected_voice: &Mutex<String>,
voice_name: &mut String,
style: &mut VoiceStyle,
) -> bool {
let requested_voice = selected_voice
.lock()
.unwrap_or_else(|error| error.into_inner())
.clone();
if requested_voice == *voice_name {
return true;
}
let requested_path = model_dir.join(format!("{requested_voice}.{VOICE_FILE_EXT}"));
match load_voice_style(&requested_path) {
Ok(requested_style) => {
*style = requested_style;
*voice_name = requested_voice;
true
}
Err(error) => {
use super::pocket::DEFAULT_VOICE;
eprintln!(
"buzz-desktop: Pocket voice {requested_voice} is unavailable ({error}); falling back to Mary"
);
let fallback_path = model_dir.join(format!("{DEFAULT_VOICE}.{VOICE_FILE_EXT}"));
match load_voice_style(&fallback_path) {
Ok(fallback_style) => {
*style = fallback_style;
*voice_name = DEFAULT_VOICE.to_string();
*selected_voice
.lock()
.unwrap_or_else(|lock_error| lock_error.into_inner()) =
DEFAULT_VOICE.to_string();
true
}
Err(fallback_error) => {
eprintln!("buzz-desktop: Mary voice fallback is unavailable: {fallback_error}");
false
}
}
}
}
}
/// Check for cancel or shutdown. Returns `true` if the caller should break/continue.
/// On cancel: drains the text queue and clears the cancel flag.
///
@@ -656,12 +876,15 @@ fn tts_worker(
/// it is serialized with the monitor's stale-branch re-check (see the monitor
/// block in `tts_worker`).
fn handle_cancel_or_shutdown(
cancel: &AtomicBool,
cancel_signals: CancelSignals<'_>,
shutdown: &AtomicBool,
tts_active: &AtomicBool,
text_rx: &mpsc::Receiver<String>,
text_state: CancelTextState<'_>,
voice_change_ack: &VoiceChangeAck,
player: Option<(&rodio::Player, &Mutex<()>)>,
) -> bool {
let (cancel, voice_cancel) = cancel_signals;
let (text_rx, deferred_text, current_text) = text_state;
if shutdown.load(Ordering::Acquire) {
if let Some((p, ops)) = player {
let _ops = lock_player_ops(ops);
@@ -670,7 +893,24 @@ fn handle_cancel_or_shutdown(
tts_active.store(false, Ordering::Release);
return true;
}
if cancel.load(Ordering::Acquire) {
if cancel.load(Ordering::Acquire) || voice_cancel.load(Ordering::Acquire) {
// Serialize with begin_voice_change so the generation boundary and
// cancel consumption are observed as one transition.
let pending_voice_change = voice_change_ack
.lock()
.unwrap_or_else(|error| error.into_inner());
// Consume at the serialization point. A later barge-in remains true
// for the next pass instead of being overwritten after queue cleanup.
let barge_in = cancel.swap(false, Ordering::AcqRel);
voice_cancel.store(false, Ordering::Release);
let preserve_generation = (!barge_in)
.then(|| {
pending_voice_change
.as_ref()
.map(|pending| pending.generation)
})
.flatten();
retain_cancelled_text(deferred_text, current_text, text_rx, preserve_generation);
if let Some((p, ops)) = player {
let _ops = lock_player_ops(ops);
// `Player::clear()` removes queued sources AND pauses the player
@@ -683,11 +923,6 @@ fn handle_cancel_or_shutdown(
// Consume the flag under the lock: once released with
// `cancel == false`, the monitor's stale branch no-ops instead
// of clearing the fresh post-cancel utterance.
while text_rx.try_recv().is_ok() {}
cancel.store(false, Ordering::Release);
} else {
while text_rx.try_recv().is_ok() {}
cancel.store(false, Ordering::Release);
}
tts_active.store(false, Ordering::Release);
return true;
@@ -695,6 +930,31 @@ fn handle_cancel_or_shutdown(
false
}
fn retain_cancelled_text(
deferred_text: &mut VecDeque<QueuedText>,
current_text: &mut Option<QueuedText>,
text_rx: &mpsc::Receiver<QueuedText>,
preserve_generation: Option<u64>,
) {
if let Some(generation) = preserve_generation {
deferred_text.retain(|text| text.generation >= generation);
if let Some(text) = current_text.take() {
if text.generation >= generation {
deferred_text.push_front(text);
}
}
while let Ok(text) = text_rx.try_recv() {
if text.generation >= generation {
deferred_text.push_back(text);
}
}
} else {
deferred_text.clear();
current_text.take();
while text_rx.try_recv().is_ok() {}
}
}
/// Acquire the `player_ops` lock, recovering from poison.
///
/// The data under the mutex is `()` — it only serializes Player mutations —
@@ -834,9 +1094,6 @@ fn group_sentences_into_chunks(sentences: &[String], max_chars: usize) -> Vec<St
chunks
}
// drain_until_shutdown lives in super (huddle/mod.rs) — shared with stt.rs.
use super::drain_until_shutdown;
// ── Tests ─────────────────────────────────────────────────────────────────────
#[cfg(test)]
desktop/src-tauri/src/huddle/tts_settings.rs
+136 -17
View File
@@ -6,21 +6,31 @@
//! embedded in installation-global settings or future agent identity without a
//! schema change. Availability is intentionally client-local.
use std::path::{Path, PathBuf};
use std::{
path::{Path, PathBuf},
sync::Arc,
time::Duration,
};
use serde::{Deserialize, Serialize};
use tauri::{AppHandle, Manager, State};
use crate::{app_state::AppState, managed_agents::storage::atomic_write_json_restricted};
use super::{models, pocket::DEFAULT_VOICE, HuddlePhase};
use super::{models, pocket::DEFAULT_VOICE, HuddlePhase, HuddleState};
const SETTINGS_FILE: &str = "tts-settings.json";
const CURRENT_VERSION: u32 = 1;
const VOICE_CHANGE_ACK_TIMEOUT: Duration = Duration::from_secs(5);
pub const POCKET_BACKEND_ID: &str = "pocket";
pub const MARY_VOICE_KEY: &str = "pocket:mary";
pub const MARIUS_VOICE_KEY: &str = "pocket:marius";
type VoiceChangeWait = (
Arc<super::tts::TtsPipeline>,
tokio::sync::oneshot::Receiver<()>,
);
const VOICE_AVAILABILITY_BUNDLED: &str = "bundled";
const VOICE_AVAILABILITY_INSTALLED: &str = "installed";
@@ -338,11 +348,28 @@ fn commit_effective_off(state: &AppState) -> Result<(), String> {
Ok(())
}
fn enable_tts_runtime(huddle: &mut HuddleState, voice: &str) -> Option<VoiceChangeWait> {
huddle.tts_enabled = true;
// OFF removes the pipeline. Clear a prior cancellation only when enabling
// a fresh pipeline; an idempotent ON write must not erase a voice
// transition that the existing worker still needs to drain.
if huddle.tts_pipeline.is_none() {
huddle
.tts_cancel
.store(false, std::sync::atomic::Ordering::Release);
}
huddle.tts_pipeline.as_ref().and_then(|pipeline| {
pipeline
.select_voice(voice)
.map(|acknowledged| (Arc::clone(pipeline), acknowledged))
})
}
async fn apply_tts_settings(
settings: TtsSettings,
app: &AppHandle,
state: &AppState,
) -> Result<TtsSettings, String> {
) -> Result<Option<VoiceChangeWait>, String> {
if settings.version != CURRENT_VERSION {
return Err(format!(
"Unsupported text-to-speech settings version: {}",
@@ -366,18 +393,18 @@ async fn apply_tts_settings(
.map_err(|error| format!("text-to-speech settings lock poisoned: {error}"))? =
settings.clone();
let mut voice_change_wait = None;
if settings.agent_text_to_speech {
let active = {
let (active, voice_change_ack) = {
let mut huddle = state.huddle()?;
huddle.tts_enabled = true;
huddle
.tts_cancel
.store(false, std::sync::atomic::Ordering::Release);
if let Some(pipeline) = huddle.tts_pipeline.as_ref() {
pipeline.select_voice(&pocket_voice_name(&settings.voice_preferences));
}
matches!(huddle.phase, HuddlePhase::Connected | HuddlePhase::Active)
let voice_change_ack =
enable_tts_runtime(&mut huddle, &pocket_voice_name(&settings.voice_preferences));
(
matches!(huddle.phase, HuddlePhase::Connected | HuddlePhase::Active),
voice_change_ack,
)
};
voice_change_wait = voice_change_ack;
if active {
if let Err(error) = super::pipeline::maybe_start_tts_pipeline(state).await {
eprintln!("buzz-desktop: could not hot-start text to speech: {error}");
@@ -385,7 +412,50 @@ async fn apply_tts_settings(
}
state.emit_huddle_state_changed();
}
Ok(settings)
Ok(voice_change_wait)
}
fn current_settings(state: &AppState) -> Result<TtsSettings, String> {
state
.tts_settings
.lock()
.map_err(|error| format!("text-to-speech settings lock poisoned: {error}"))
.map(|settings| settings.clone())
}
async fn finish_voice_change(voice_change: Option<VoiceChangeWait>) -> Result<(), String> {
let Some((pipeline, acknowledged)) = voice_change else {
return Ok(());
};
wait_for_voice_change_ack(acknowledged, VOICE_CHANGE_ACK_TIMEOUT, || {
pipeline.is_finished()
})
.await
}
async fn wait_for_voice_change_ack(
mut acknowledged: tokio::sync::oneshot::Receiver<()>,
timeout: Duration,
mut worker_is_finished: impl FnMut() -> bool,
) -> Result<(), String> {
let deadline = tokio::time::sleep(timeout);
tokio::pin!(deadline);
loop {
tokio::select! {
_ = &mut acknowledged => return Ok(()),
_ = &mut deadline => {
return Err(
"Pocket TTS is still finishing the previous voice. Turn Agent text to speech off and try again."
.to_string(),
);
}
_ = tokio::time::sleep(Duration::from_millis(25)) => {
if worker_is_finished() {
return Ok(());
}
}
}
}
}
/// Compatibility command for the huddle speaker button. It updates the same
@@ -396,14 +466,17 @@ pub async fn set_tts_enabled(
app: AppHandle,
state: State<'_, AppState>,
) -> Result<TtsSettings, String> {
let _transition = state.tts_settings_transition.lock().await;
let transition = state.tts_settings_transition.lock().await;
let mut settings = state
.tts_settings
.lock()
.map_err(|error| format!("text-to-speech settings lock poisoned: {error}"))?
.clone();
settings.agent_text_to_speech = enabled;
apply_tts_settings(settings, &app, &state).await
let voice_change = apply_tts_settings(settings, &app, &state).await?;
drop(transition);
finish_voice_change(voice_change).await?;
current_settings(&state)
}
fn settings_with_pocket_voice(
@@ -436,14 +509,17 @@ pub async fn set_pocket_voice(
app: AppHandle,
state: State<'_, AppState>,
) -> Result<TtsSettings, String> {
let _transition = state.tts_settings_transition.lock().await;
let transition = state.tts_settings_transition.lock().await;
let settings = state
.tts_settings
.lock()
.map_err(|error| format!("text-to-speech settings lock poisoned: {error}"))?
.clone();
let settings = settings_with_pocket_voice(settings, &voice_key)?;
apply_tts_settings(settings, &app, &state).await
let voice_change = apply_tts_settings(settings, &app, &state).await?;
drop(transition);
finish_voice_change(voice_change).await?;
current_settings(&state)
}
#[tauri::command]
@@ -499,6 +575,49 @@ pub async fn preview_pocket_voice(
mod tests {
use super::*;
#[tokio::test]
async fn stalled_voice_change_returns_an_actionable_error() {
let (_keep_pending, acknowledged) = tokio::sync::oneshot::channel();
let error = wait_for_voice_change_ack(acknowledged, Duration::from_millis(1), || false)
.await
.expect_err("stalled worker should time out");
assert!(error.contains("Turn Agent text to speech off"));
}
#[test]
fn idempotent_enable_preserves_an_existing_pipeline_cancel() {
let state = crate::app_state::build_app_state();
let model_dir = tempfile::tempdir().expect("temp model dir");
let cancel = state.huddle().expect("huddle state").tts_cancel.clone();
let pipeline = Arc::new(
super::super::tts::TtsPipeline::new_with_voice(
model_dir.path().to_path_buf(),
Arc::new(std::sync::atomic::AtomicBool::new(false)),
Arc::clone(&cancel),
"reference_sample",
None,
)
.expect("pipeline"),
);
pipeline.shutdown();
for _ in 0..100 {
if pipeline.is_finished() {
break;
}
std::thread::sleep(Duration::from_millis(1));
}
assert!(pipeline.is_finished(), "test pipeline should stop");
cancel.store(true, std::sync::atomic::Ordering::Release);
let mut huddle = state.huddle().expect("huddle state");
huddle.tts_pipeline = Some(pipeline);
assert!(enable_tts_runtime(&mut huddle, "reference_sample").is_none());
assert!(cancel.load(std::sync::atomic::Ordering::Acquire));
}
#[test]
fn defaults_are_backwards_compatible_and_use_mary() {
assert_eq!(
desktop/src-tauri/src/huddle/tts_voice_selection_tests.rs
+311 -3
View File
@@ -3,7 +3,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
#[test]
fn selecting_a_voice_immediately_raises_cancel_and_retains_the_engine_handle() {
fn selecting_a_voice_raises_only_the_internal_cancel_and_retains_the_engine_handle() {
let model_dir = tempfile::tempdir().expect("temp model dir");
let active = Arc::new(AtomicBool::new(false));
let cancel = Arc::new(AtomicBool::new(false));
@@ -16,9 +16,10 @@ fn selecting_a_voice_immediately_raises_cancel_and_retains_the_engine_handle() {
)
.expect("pipeline handle");
pipeline.select_voice("marius");
let _acknowledged = pipeline.select_voice("marius");
assert!(cancel.load(Ordering::Acquire));
assert!(!cancel.load(Ordering::Acquire));
assert!(pipeline.voice_cancel.load(Ordering::Acquire));
assert_eq!(
pipeline
.voice
@@ -28,3 +29,310 @@ fn selecting_a_voice_immediately_raises_cancel_and_retains_the_engine_handle() {
"marius"
);
}
#[test]
fn reconciling_an_unpublished_pipeline_does_not_cancel_its_first_message() {
let model_dir = tempfile::tempdir().expect("temp model dir");
let active = Arc::new(AtomicBool::new(false));
let cancel = Arc::new(AtomicBool::new(false));
let pipeline = TtsPipeline::new_with_voice(
model_dir.path().to_path_buf(),
active,
Arc::clone(&cancel),
"reference_sample",
None,
)
.expect("pipeline handle");
pipeline.select_voice_before_publish("marius");
assert!(!cancel.load(Ordering::Acquire));
assert_eq!(
pipeline
.voice
.lock()
.unwrap_or_else(|error| error.into_inner())
.as_str(),
"marius"
);
}
#[test]
fn received_text_reconciles_a_voice_changed_while_the_worker_was_waiting() {
let model_dir = tempfile::tempdir().expect("temp model dir");
let bundled_voice =
std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("resources/pocket-voices/marius.wav");
std::fs::copy(
&bundled_voice,
model_dir.path().join("reference_sample.wav"),
)
.expect("Mary test voice");
std::fs::copy(&bundled_voice, model_dir.path().join("marius.wav")).expect("Marius test voice");
let selected_voice = Arc::new(std::sync::Mutex::new("reference_sample".to_string()));
let mut style =
load_voice_style(&model_dir.path().join("reference_sample.wav")).expect("initial style");
let waiting = Arc::new(std::sync::Barrier::new(2));
let (text_tx, text_rx) = std::sync::mpsc::channel();
let worker_voice = Arc::clone(&selected_voice);
let worker_waiting = Arc::clone(&waiting);
let worker_model_dir = model_dir.path().to_path_buf();
let worker = std::thread::spawn(move || {
let mut voice_name = "reference_sample".to_string();
worker_waiting.wait();
let text = text_rx.recv().expect("first queued text");
assert!(reconcile_selected_voice(
&worker_model_dir,
&worker_voice,
&mut voice_name,
&mut style,
));
(text, voice_name)
});
waiting.wait();
*selected_voice.lock().expect("selected voice") = "marius".to_string();
text_tx
.send("first message".to_string())
.expect("queue first message");
assert_eq!(
worker.join().expect("worker"),
("first message".to_string(), "marius".to_string())
);
}
#[test]
fn an_in_hand_post_change_message_survives_cancellation() {
let selected_voice = Arc::new(std::sync::Mutex::new("reference_sample".to_string()));
let voice_generation = AtomicU64::new(1);
let barge_in = AtomicBool::new(false);
let voice_cancel = Arc::new(AtomicBool::new(false));
let voice_change_ack = Arc::new(std::sync::Mutex::new(None));
let (text_tx, text_rx) = std::sync::mpsc::sync_channel(1);
let mut acknowledged = begin_voice_change(
&selected_voice,
&voice_generation,
&voice_cancel,
&voice_change_ack,
"marius",
)
.expect("voice changed");
assert!(voice_cancel.load(Ordering::Acquire));
assert!(matches!(
acknowledged.try_recv(),
Err(tokio::sync::oneshot::error::TryRecvError::Empty)
));
acknowledge_voice_change(&voice_change_ack, &voice_cancel);
assert!(matches!(
acknowledged.try_recv(),
Err(tokio::sync::oneshot::error::TryRecvError::Empty)
));
text_tx
.send(QueuedText {
generation: voice_generation.load(Ordering::Acquire),
text: "new message".to_string(),
})
.expect("new message");
let mut current_text = Some(text_rx.recv().expect("in-hand new message"));
let shutdown = AtomicBool::new(false);
let active = AtomicBool::new(true);
let mut deferred_text = VecDeque::from([
QueuedText {
generation: 1,
text: "old message".to_string(),
},
QueuedText {
generation: voice_generation.load(Ordering::Acquire),
text: "later new message".to_string(),
},
]);
assert!(handle_cancel_or_shutdown(
(&barge_in, &voice_cancel),
&shutdown,
&active,
(&text_rx, &mut deferred_text, &mut current_text),
&voice_change_ack,
None,
));
acknowledge_voice_change(&voice_change_ack, &voice_cancel);
acknowledged.blocking_recv().expect("voice change ack");
assert_eq!(
deferred_text
.pop_front()
.expect("preserved post-change message")
.text,
"new message"
);
assert_eq!(
deferred_text
.pop_front()
.expect("later post-change message")
.text,
"later new message"
);
assert!(text_rx.try_recv().is_err());
}
#[test]
fn superseding_voice_change_removes_earlier_deferred_messages() {
let selected_voice = std::sync::Mutex::new("reference_sample".to_string());
let voice_generation = AtomicU64::new(1);
let barge_in = AtomicBool::new(false);
let voice_cancel = AtomicBool::new(false);
let voice_change_ack = Arc::new(std::sync::Mutex::new(None));
let (_text_tx, text_rx) = std::sync::mpsc::channel();
let shutdown = AtomicBool::new(false);
let active = AtomicBool::new(true);
let mut deferred_text = VecDeque::new();
let mut current_text = None;
let first = begin_voice_change(
&selected_voice,
&voice_generation,
&voice_cancel,
&voice_change_ack,
"marius",
)
.expect("first voice change");
deferred_text.push_back(QueuedText {
generation: voice_generation.load(Ordering::Acquire),
text: "message for Marius".to_string(),
});
assert!(handle_cancel_or_shutdown(
(&barge_in, &voice_cancel),
&shutdown,
&active,
(&text_rx, &mut deferred_text, &mut current_text),
&voice_change_ack,
None,
));
acknowledge_voice_change(&voice_change_ack, &voice_cancel);
first.blocking_recv().expect("first acknowledgement");
let _second = begin_voice_change(
&selected_voice,
&voice_generation,
&voice_cancel,
&voice_change_ack,
"reference_sample",
)
.expect("second voice change");
assert!(handle_cancel_or_shutdown(
(&barge_in, &voice_cancel),
&shutdown,
&active,
(&text_rx, &mut deferred_text, &mut current_text),
&voice_change_ack,
None,
));
assert!(deferred_text.is_empty());
}
#[test]
fn barge_in_clears_deferred_voice_change_messages() {
let barge_in = AtomicBool::new(true);
let voice_cancel = AtomicBool::new(false);
let shutdown = AtomicBool::new(false);
let active = AtomicBool::new(true);
let voice_change_ack = Arc::new(std::sync::Mutex::new(None));
let (_text_tx, text_rx) = std::sync::mpsc::channel();
let mut deferred_text = VecDeque::from([QueuedText {
generation: 2,
text: "deferred message".to_string(),
}]);
let mut current_text = None;
assert!(handle_cancel_or_shutdown(
(&barge_in, &voice_cancel),
&shutdown,
&active,
(&text_rx, &mut deferred_text, &mut current_text),
&voice_change_ack,
None,
));
assert!(deferred_text.is_empty());
}
#[test]
fn barge_in_during_a_voice_change_clears_post_change_messages() {
let selected_voice = std::sync::Mutex::new("reference_sample".to_string());
let voice_generation = AtomicU64::new(1);
let barge_in = AtomicBool::new(false);
let voice_cancel = AtomicBool::new(false);
let voice_change_ack = Arc::new(std::sync::Mutex::new(None));
let (_text_tx, text_rx) = std::sync::mpsc::channel();
let shutdown = AtomicBool::new(false);
let active = AtomicBool::new(true);
let mut deferred_text = VecDeque::new();
let mut current_text = None;
let _acknowledged = begin_voice_change(
&selected_voice,
&voice_generation,
&voice_cancel,
&voice_change_ack,
"marius",
)
.expect("voice change");
deferred_text.push_back(QueuedText {
generation: voice_generation.load(Ordering::Acquire),
text: "post-change message".to_string(),
});
barge_in.store(true, Ordering::Release);
assert!(handle_cancel_or_shutdown(
(&barge_in, &voice_cancel),
&shutdown,
&active,
(&text_rx, &mut deferred_text, &mut current_text),
&voice_change_ack,
None,
));
assert!(deferred_text.is_empty());
}
#[test]
fn a_sender_captured_before_voice_change_is_stale_even_if_it_sends_after_drain() {
let selected_voice = std::sync::Mutex::new("reference_sample".to_string());
let voice_generation = Arc::new(AtomicU64::new(1));
let barge_in = AtomicBool::new(false);
let voice_cancel = AtomicBool::new(false);
let voice_change_ack = Arc::new(std::sync::Mutex::new(None));
let (text_tx, text_rx) = std::sync::mpsc::sync_channel(1);
let old_sender = TtsTextSender {
text_tx,
generation: voice_generation.load(Ordering::Acquire),
};
let shutdown = AtomicBool::new(false);
let active = AtomicBool::new(true);
let mut deferred_text = VecDeque::new();
let mut current_text = None;
let _acknowledged = begin_voice_change(
&selected_voice,
&voice_generation,
&voice_cancel,
&voice_change_ack,
"marius",
)
.expect("voice change");
assert!(handle_cancel_or_shutdown(
(&barge_in, &voice_cancel),
&shutdown,
&active,
(&text_rx, &mut deferred_text, &mut current_text),
&voice_change_ack,
None,
));
old_sender
.send("late old message".to_string())
.expect("late send");
let late = text_rx.recv().expect("late queued text");
assert!(late.generation < voice_generation.load(Ordering::Acquire));
}