jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-11 20:01:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(desktop): coordinate TTS playback (#6341)

Browse Source 
## Summary

- introduce a `PlaybackCoordinator` as the exclusive owner of the rodio
`Player`
- serialize append, cancellation, synthesis-flight, and drain-release
state
- cancel without blocking the coordinator by replacing the player queue
- preserve first-append/onset bookkeeping across cancellation
- increase the existing playback lead-in from 20 ms (480 samples) to a
derived 22 ms (528 samples), covering rodio 0.22.2's 512-source-sample
add-time bootstrap span

This is the behavior-preserving coordinator prerequisite for the
follow-up floor-authorization policy. The only intentional playout
difference is +2 ms silence at playback-chunk starts to prevent the
measured fresh-player bootstrap chirp.

## Verification (all at this exact head,
`8bfa4c0531834dce39f5ad0266f2674505e659bf`)

- `cargo clippy --manifest-path desktop/src-tauri/Cargo.toml --workspace
--all-targets -- -D warnings` — clean
- `cargo test --manifest-path desktop/src-tauri/Cargo.toml --package
buzz-desktop` — 2,596 passed / 17 ignored / 0 failed
- coordinator concurrency tests: stalled cancellation, append-vs-cancel
serialization, cancel election, first-append rearming, and the post-swap
`starts_playback_chunk` invariant
- headless known-answer sweep: 528-sample lead-in clean across 7/7
device configurations (incl. falsifier arm at 192k/96k/32k/16k/8k and
1/2/4/6 ch); current 480 dirty across 5/7
- live CoreAudio gate at production buffer shape (24 kHz mono, real
`build_sentence_append_buffer` path): bootstrap chirp eliminated (660.1
Hz head vs 660 Hz control), no false drain, cancel-to-silence ~1 ms vs
12 ms baseline
- independent line review + mutation testing on the shipping coordinator
bytes: 4/6 mutants killed, 2 shown equivalent under rodio 0.22.2
`Player::drop` semantics

## Playout cost

- +2 ms lead-in at each playback-chunk start versus main
- first post-cancel utterance receives 22 ms total intentional silence
(not stacked cushions)

## Provenance

Authored and reviewed by Buzz agents (Wren: implementation; Dawn: line
review, known-answer control, mutation testing; Max: live CoreAudio
gates; Eva: coordination and final verification), with review records on
the Buzz relay. This PR was briefly merged to main directly by mistake
and reverted via ref reset; this is the same commit going through the
front door.

---------

Signed-off-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz>
Signed-off-by: Dawn <c6237ef84fa537c78dcee78efd2d4e59f728859c7f194da42ac51ededfa0be05@buzz.block.builderlab.xyz>
Co-authored-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz>
Co-authored-by: Dawn <c6237ef84fa537c78dcee78efd2d4e59f728859c7f194da42ac51ededfa0be05@buzz.block.builderlab.xyz>
This commit is contained in:
Tyler
2026-08-19 21:26:03 -04:00
committed by GitHub
parent 9c2f05346f
commit 09718fbb60
8 changed files with 896 additions and 313 deletions
Download Patch File Download Diff File Expand all files Collapse all files

240
desktop/src-tauri/src/huddle/tts.rs
View File

@@ -18,10 +18,9 @@
//! → cancel flag: a 10 ms barge-in monitor thread silences the player and
//! releases tts_active on the flag's rising edge (~15 ms flag-to-silence,
//! even mid-sentence while the worker is blocked in synth_chunk); the
//! worker then consumes the flag — drain queue + clear + play (un-pause).
//! Monitor clears and worker player mutations are serialized through the
//! `player_ops` mutex, with the flag re-checked under the lock — see the
//! monitor block in `tts_worker` for the race this closes.
//! worker then consumes the flag and drains stale text. Every Player
//! operation is serialized by `PlaybackCoordinator`; cancellation swaps in
//! a fresh queue and drops the old Player after releasing the coordinator.
//! ```
//!
//! Lookahead pipelining spans *items*, not just sentences within one item:
@@ -55,6 +54,9 @@ use super::preprocessing::preprocess_for_tts;
#[path = "tts_voice_transition.rs"]
mod voice_transition;
use voice_transition::*;
#[path = "tts_playback.rs"]
mod playback;
use playback::*;
#[path = "tts_startup.rs"]
mod startup;
use startup::await_worker_startup;
@@ -102,10 +104,26 @@ const SYNTH_STEPS: usize = 1;
/// the leading waveform is important.
const FADE_OUT_SAMPLES: usize = (SAMPLE_RATE as f64 * 0.008) as usize;
/// Length of the zero-sample cushion prepended when playback is idle, so the
/// OS audio device / rodio mixer has a fully-quiet ramp-up window before the
/// real onset hits. Continuously queued chunks receive no synthetic padding.
const SENTENCE_LEAD_IN_SAMPLES: usize = (SAMPLE_RATE as f64 * 0.020) as usize;
/// rodio 0.22.2 bootstraps `UniformSourceIterator` when a source is added to
/// the mixer (`conversions/uniform.rs:49-66`). Its empty queue's 512-sample
/// span (`queue.rs::SourcesQueueInput::new`) can therefore retain placeholder
/// format metadata until the next span. The lead-in covers that whole span,
/// rounded up to the next millisecond, while preserving the product's existing
/// 20 ms quiet ramp-up. Continuously queued chunks receive no synthetic padding.
const SAMPLES_PER_MS: usize = SAMPLE_RATE as usize / 1_000;
const PRODUCT_RAMP_UP_MS: usize = 20;
const PRODUCT_RAMP_UP_SAMPLES: usize = PRODUCT_RAMP_UP_MS * SAMPLES_PER_MS;
const RODIO_ADD_BOOTSTRAP_SPAN_SAMPLES: usize = 512;
const RODIO_ADD_BOOTSTRAP_CUSHION_MS: usize =
RODIO_ADD_BOOTSTRAP_SPAN_SAMPLES.div_ceil(SAMPLES_PER_MS);
const SENTENCE_LEAD_IN_SAMPLES: usize = {
let bootstrap_cushion = RODIO_ADD_BOOTSTRAP_CUSHION_MS * SAMPLES_PER_MS;
if PRODUCT_RAMP_UP_SAMPLES > bootstrap_cushion {
PRODUCT_RAMP_UP_SAMPLES
} else {
bootstrap_cushion
}
};
type WorkerControlState = (
Arc<AtomicBool>,
@@ -332,7 +350,6 @@ fn tts_worker(
// ── 3. Initialise rodio output device ─────────────────────────────────────
use rodio::buffer::SamplesBuffer;
use rodio::Player;
let sink_handle = match super::audio_output::open_output_sink_by_name(output_device.as_deref())
{
@@ -360,28 +377,24 @@ fn tts_worker(
}
};
// Single persistent Player for the lifetime of the worker — all sentence
// buffers from all text items append here, and rodio plays them gaplessly.
// Persistence is what enables cross-item pipelining: the worker never
// waits for one item to drain before synthesizing the next.
//
// Shared (Arc) with the barge-in monitor thread below, which needs to
// silence it while this thread is blocked inside `synth_chunk`.
let player = Arc::new(Player::connect_new(sink_handle.mixer()));
playback_probe.install(Arc::clone(&player));
// One coordinator owns the current Player and every operation on it.
// Cancellation replaces its queue while the output stream and mixer stay
// alive, preserving cross-item pipelining without exposing Player handles.
let playback = Arc::new(PlaybackCoordinator::new(sink_handle.mixer()));
playback_probe.install(Arc::clone(&playback));
// Prime the audio output stream with a short silent buffer.
// On macOS, CoreAudio initializes the output device lazily on first use.
// Without this, the first real append races against device startup and
// player.empty() returns true before audio has started draining — causing
// playback.empty() returns true before audio has started draining — causing
// the first TTS message to be truncated after a few words.
{
let silence = vec![0.0f32; SAMPLE_RATE as usize / 10]; // 100ms of silence
player.append(SamplesBuffer::new(channels, rate, silence));
playback.append_untracked(SamplesBuffer::new(channels, rate, silence));
// Wait for the silent buffer to drain — this ensures the output stream
// is fully initialized before the first real utterance.
let deadline = std::time::Instant::now() + AUDIO_PRIME_TIMEOUT;
while !player.empty() {
while !playback.empty() {
if std::time::Instant::now() >= deadline {
eprintln!("buzz-desktop: tts stage=startup status=failed reason=output_prime");
let _ = startup_tx.send(Err(
@@ -397,16 +410,14 @@ fn tts_worker(
}
eprintln!("buzz-desktop: tts stage=startup status=ready");
let player_ops = Arc::clone(&playback_probe.player_ops);
let activity_frames = Arc::new(Mutex::new(VecDeque::<TtsSpeakerActivityFrame>::new()));
let monitor_stop = Arc::new(AtomicBool::new(false));
let monitor = spawn_tts_monitor(TtsMonitorState {
player: Arc::clone(&player),
playback: Arc::clone(&playback),
cancel: Arc::clone(&cancel),
voice_cancel: Arc::clone(&voice_cancel),
tts_active: Arc::clone(&tts_active),
stop: Arc::clone(&monitor_stop),
player_ops: Arc::clone(&player_ops),
activity_frames: Arc::clone(&activity_frames),
active_speaker: Arc::clone(&active_speaker),
speaker_cancel: Arc::clone(&speaker_cancel),
@@ -429,75 +440,77 @@ fn tts_worker(
// EXPERIMENTAL (latency bench): `Some(emit_frames)` = stream PCM deltas
// out of Pocket as they are generated (see tts_streaming.rs).
let tts_streaming = streaming_emit_frames();
// `first_append` = "no audio queued since the player last went idle".
// Flipped by `build_sentence_append_buffer` on the first real append; the
// 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 last_route_id = 0;
let mut deferred_text = VecDeque::new();
let append_audio = |prepared: PreparedModelAudio,
route_id: u64,
speaker_pubkey: Option<&str>,
speaker_generation: u64| {
let _ops = lock_player_ops(&player_ops);
if cancel.load(Ordering::Acquire)
|| voice_cancel.load(Ordering::Acquire)
|| shutdown.load(Ordering::Acquire)
{
let reason = if shutdown.load(Ordering::Acquire) {
"shutdown"
} else if cancel.load(Ordering::Acquire) {
"barge_in"
} else {
"voice_switch"
};
eprintln!(
"buzz-desktop: tts stage=synthesis status=cancelled reason={reason} route_id={route_id}"
);
return false;
}
let speaker_is_current = speaker_pubkey.is_none_or(|pubkey| {
current_speaker_generation(&speaker_generations, pubkey) == speaker_generation
let sample_count = prepared.sample_count;
let chunk_index = prepared.chunk_index;
let activity = speaker_pubkey.map(|pubkey| {
build_tts_speaker_activity_frames(&prepared.buffer, pubkey, SAMPLE_RATE as usize)
});
if !speaker_is_current {
eprintln!(
"buzz-desktop: tts stage=synthesis status=cancelled reason=speaker_removed route_id={route_id}"
);
let accepted = playback.append_if(
SamplesBuffer::new(channels, rate, prepared.buffer),
|player_empty| {
if cancel.load(Ordering::Acquire)
|| voice_cancel.load(Ordering::Acquire)
|| shutdown.load(Ordering::Acquire)
{
let reason = if shutdown.load(Ordering::Acquire) {
"shutdown"
} else if cancel.load(Ordering::Acquire) {
"barge_in"
} else {
"voice_switch"
};
eprintln!(
"buzz-desktop: tts stage=synthesis status=cancelled reason={reason} route_id={route_id}"
);
return false;
}
if speaker_pubkey.is_some_and(|pubkey| {
current_speaker_generation(&speaker_generations, pubkey) != speaker_generation
}) {
eprintln!(
"buzz-desktop: tts stage=synthesis status=cancelled reason=speaker_removed route_id={route_id}"
);
return false;
}
if let Some(pubkey) = speaker_pubkey {
let mut active = active_speaker
.lock()
.unwrap_or_else(|error| error.into_inner());
if player_empty {
active.take();
}
if active
.as_deref()
.is_some_and(|current| !current.eq_ignore_ascii_case(pubkey))
{
return false;
}
active.get_or_insert_with(|| pubkey.to_ascii_lowercase());
activity_frames
.lock()
.unwrap_or_else(|error| error.into_inner())
.extend(activity.unwrap_or_default());
}
true
},
// Publish activity under the coordinator: an append and the mic
// gate it implies are one transition, so a cancellation that
// replaces this player cannot have its release overwritten by a
// `true` landing after the fact.
|| tts_active.store(true, Ordering::Release),
);
if !accepted {
return false;
}
if let Some(pubkey) = speaker_pubkey {
let mut active = active_speaker
.lock()
.unwrap_or_else(|error| error.into_inner());
if player.empty() {
active.take();
}
if active
.as_deref()
.is_some_and(|current| !current.eq_ignore_ascii_case(pubkey))
{
return false;
}
active.get_or_insert_with(|| pubkey.to_ascii_lowercase());
}
if let Some(pubkey) = speaker_pubkey {
activity_frames
.lock()
.unwrap_or_else(|error| error.into_inner())
.extend(build_tts_speaker_activity_frames(
&prepared.buffer,
pubkey,
SAMPLE_RATE as usize,
));
}
player.append(SamplesBuffer::new(channels, rate, prepared.buffer));
eprintln!(
"buzz-desktop: tts stage=player status=append_accepted route_id={route_id} chunk_index={} sample_count={}",
prepared.chunk_index, prepared.sample_count
"buzz-desktop: tts stage=player status=append_accepted route_id={route_id} chunk_index={chunk_index} sample_count={sample_count}"
);
// Set this only after append so STT remains open during synthesis.
tts_active.store(true, Ordering::Release);
true
};
@@ -509,9 +522,8 @@ fn tts_worker(
&speaker_generations,
&tts_active,
(&text_rx, &mut deferred_text, &mut no_current_text),
Some((&player, &player_ops)),
Some(&playback),
) {
first_append = true;
continue;
}
if handle_cancel_or_shutdown(
@@ -521,14 +533,13 @@ fn tts_worker(
(&text_rx, &mut deferred_text, &mut no_current_text),
&voice_change_ack,
None,
Some((&player, &player_ops)),
Some(&playback),
) {
if shutdown.load(Ordering::Acquire) {
break;
}
// Cancel consumed: queued audio cleared, queue drained. The next
// append starts a new utterance and needs its own lead-in cushion.
first_append = true;
continue;
}
@@ -555,7 +566,7 @@ fn tts_worker(
// 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 {
playback.release_if_drained(|| {
tts_active.store(false, Ordering::Release);
active_speaker
.lock()
@@ -564,8 +575,7 @@ fn tts_worker(
eprintln!(
"buzz-desktop: tts stage=player status=drained route_id={last_route_id}"
);
first_append = true;
}
});
continue;
}
Err(mpsc::RecvTimeoutError::Disconnected) => break,
@@ -582,12 +592,11 @@ fn tts_worker(
(&text_rx, &mut deferred_text, &mut queued_text),
&voice_change_ack,
pending_route_id,
Some((&player, &player_ops)),
Some(&playback),
) {
if shutdown.load(Ordering::Acquire) {
break;
}
first_append = true;
continue;
}
let Some(queued_text) = queued_text else {
@@ -607,7 +616,7 @@ fn tts_worker(
);
continue;
}
if !player.empty()
if !playback.empty()
&& queued_text
.speaker_pubkey
.as_deref()
@@ -639,18 +648,14 @@ fn tts_worker(
// release stale ownership before doing any potentially slow voice or
// synthesis work. Serialize the drain decision with Stop and append so
// those paths observe one coherent utterance boundary.
{
let _ops = lock_player_ops(&player_ops);
if player.empty() && !first_append {
tts_active.store(false, Ordering::Release);
active_speaker
.lock()
.unwrap_or_else(|error| error.into_inner())
.take();
eprintln!("buzz-desktop: tts stage=player status=drained route_id={last_route_id}");
first_append = true;
}
}
playback.release_if_drained(|| {
tts_active.store(false, Ordering::Release);
active_speaker
.lock()
.unwrap_or_else(|error| error.into_inner())
.take();
eprintln!("buzz-desktop: tts stage=player status=drained route_id={last_route_id}");
});
// From this point until the item finishes, an empty player can mean a
// voice-preparation or synthesis gap rather than a drained utterance.
@@ -716,9 +721,8 @@ fn tts_worker(
(&text_rx, &mut deferred_text, &mut no_current_text),
&voice_change_ack,
Some(route_id),
Some((&player, &player_ops)),
Some(&playback),
) {
first_append = true;
synthesis_outcome = "cancelled";
break;
}
@@ -738,8 +742,7 @@ fn tts_worker(
emit_frames,
(&cancel, &voice_cancel, &shutdown),
StreamingPlayback {
player: &player,
first_append: &mut first_append,
playback: &playback,
route_id,
},
&mut |prepared| {
@@ -791,9 +794,8 @@ fn tts_worker(
(&text_rx, &mut deferred_text, &mut no_current_text),
&voice_change_ack,
Some(route_id),
Some((&player, &player_ops)),
Some(&playback),
) {
first_append = true;
synthesis_outcome = "cancelled";
break 'playback_chunks;
}
@@ -817,25 +819,20 @@ fn tts_worker(
// synthesis that completed after cancellation so stale audio
// never reaches the player, while keeping buzz-voice's
// extracted April engine API unchanged.
first_append = true;
synthesis_outcome = "cancelled";
break 'playback_chunks;
}
match synthesis {
Ok(samples) if !samples.is_empty() => {
if let Some(prepared) = playback_audio.push(
samples,
chunk_index,
&mut first_append,
player.empty(),
) {
if let Some(prepared) = playback
.prepare_audio(|empty| playback_audio.push(samples, chunk_index, empty))
{
if !append_audio(
prepared,
route_id,
speaker_pubkey.as_deref(),
speaker_generation,
) {
first_append = true;
synthesis_outcome = "cancelled";
break 'playback_chunks;
}
@@ -857,14 +854,13 @@ fn tts_worker(
}
}
}
if let Some(prepared) = playback_audio.finish(&mut first_append, player.empty()) {
if let Some(prepared) = playback.prepare_audio(|empty| playback_audio.finish(empty)) {
if !append_audio(
prepared,
route_id,
speaker_pubkey.as_deref(),
speaker_generation,
) {
first_append = true;
synthesis_outcome = "cancelled";
break 'playback_chunks;
}
@@ -884,8 +880,8 @@ fn tts_worker(
}
}
// Stop the barge-in monitor before exiting — it holds a Player clone,
// and an orphaned monitor would keep ticking against a dead pipeline.
// Stop the barge-in monitor before exiting so an orphaned monitor cannot
// keep ticking against a dead pipeline.
monitor_stop.store(true, Ordering::Release);
if let Ok(handle) = monitor {
let _ = handle.join();

61
desktop/src-tauri/src/huddle/tts_audio.rs
View File

@@ -21,35 +21,24 @@ impl PlaybackChunkAudio {
&mut self,
samples: Vec<f32>,
chunk_index: usize,
first_append: &mut bool,
playback_idle: bool,
) -> Option<PreparedModelAudio> {
if samples.is_empty() {
return None;
}
let previous = self.pending.replace((samples, chunk_index))?;
let prepared = prepare_model_audio(previous, first_append, playback_idle, false);
let prepared = prepare_model_audio(previous, playback_idle, false);
Some(prepared)
}
pub(super) fn finish(
&mut self,
first_append: &mut bool,
playback_idle: bool,
) -> Option<PreparedModelAudio> {
pub(super) fn finish(&mut self, playback_idle: bool) -> Option<PreparedModelAudio> {
let pending = self.pending.take()?;
Some(prepare_model_audio(
pending,
first_append,
playback_idle,
true,
))
Some(prepare_model_audio(pending, playback_idle, true))
}
}
fn prepare_model_audio(
(samples, chunk_index): (Vec<f32>, usize),
first_append: &mut bool,
starts_playback_chunk: bool,
ends_playback_chunk: bool,
) -> PreparedModelAudio {
@@ -59,7 +48,7 @@ fn prepare_model_audio(
apply_fade_out(&mut audio);
}
PreparedModelAudio {
buffer: build_sentence_append_buffer(first_append, audio, starts_playback_chunk),
buffer: build_sentence_append_buffer(audio, starts_playback_chunk),
sample_count,
chunk_index,
}
@@ -80,14 +69,9 @@ pub(super) fn apply_fade_out(samples: &mut [f32]) {
}
pub(super) fn build_sentence_append_buffer(
first_append: &mut bool,
audio: Vec<f32>,
starts_playback_chunk: bool,
) -> Vec<f32> {
if *first_append {
*first_append = false;
}
let lead_in_len = if starts_playback_chunk {
SENTENCE_LEAD_IN_SAMPLES
} else {
@@ -106,19 +90,14 @@ mod tests {
#[test]
fn model_units_are_queued_contiguously_without_injected_silence() {
let mut chunk = PlaybackChunkAudio::new();
let mut first_append = true;
assert!(chunk
.push(vec![0.4; 16], 0, &mut first_append, false)
.is_none());
assert!(chunk.push(vec![0.4; 16], 0, false).is_none());
let first = chunk
.push(vec![0.5; 16], 1, &mut first_append, false)
.push(vec![0.5; 16], 1, false)
.expect("first ready model unit");
assert_eq!(first.buffer, vec![0.4; 16]);
let last = chunk
.finish(&mut first_append, false)
.expect("last ready model unit");
let last = chunk.finish(false).expect("last ready model unit");
assert_eq!(last.buffer.len(), 16);
assert_eq!(last.sample_count, 16);
}
@@ -126,39 +105,27 @@ mod tests {
#[test]
fn empty_edge_units_do_not_steal_audio_boundaries() {
let mut chunk = PlaybackChunkAudio::new();
let mut first_append = true;
assert!(chunk
.push(Vec::new(), 0, &mut first_append, false)
.is_none());
assert!(chunk
.push(vec![0.5; 16], 1, &mut first_append, false)
.is_none());
assert!(chunk
.push(Vec::new(), 2, &mut first_append, false)
.is_none());
assert!(chunk.push(Vec::new(), 0, false).is_none());
assert!(chunk.push(vec![0.5; 16], 1, false).is_none());
assert!(chunk.push(Vec::new(), 2, false).is_none());
let only = chunk
.finish(&mut first_append, false)
.expect("only audible model unit");
let only = chunk.finish(false).expect("only audible model unit");
assert_eq!(only.buffer.len(), 16);
}
#[test]
fn playback_underrun_rearms_the_onset_cushion() {
let mut chunk = PlaybackChunkAudio::new();
let mut first_append = true;
assert!(chunk
.push(vec![0.4; 16], 0, &mut first_append, false)
.is_none());
assert!(chunk.push(vec![0.4; 16], 0, false).is_none());
let first = chunk
.push(vec![0.5; 16], 1, &mut first_append, false)
.push(vec![0.5; 16], 1, false)
.expect("first ready model unit");
assert_eq!(first.buffer.len(), 16);
let after_underrun = chunk
.push(vec![0.6; 16], 2, &mut first_append, true)
.push(vec![0.6; 16], 2, true)
.expect("second ready model unit");
assert_eq!(after_underrun.buffer.len(), SENTENCE_LEAD_IN_SAMPLES + 16);
assert!(after_underrun.buffer[..SENTENCE_LEAD_IN_SAMPLES]

485
desktop/src-tauri/src/huddle/tts_playback.rs Normal file
View File

@@ -0,0 +1,485 @@
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use rodio::{mixer::Mixer, Player, Source};
/// Serializes every operation on the TTS player and owns the utterance-boundary
/// bookkeeping that must change atomically when playback is replaced.
///
/// Poison recovery is sound because `PlaybackState` has no partially-valid
/// representation: `Player` replacement is a single assignment, booleans are
/// independently valid at either value, and no mutable reference to the state
/// leaves the locked operation that created it.
pub(super) struct PlaybackCoordinator {
mixer: Mixer,
state: Mutex<PlaybackState>,
}
struct PlaybackState {
player: Player,
/// `true` while no append has been committed since the last utterance
/// boundary. Only `append_if` clears it, so it records appends that were
/// actually queued — never one the authorization refused.
first_append: bool,
synthesis_in_flight: bool,
synthesis_generation: u64,
}
pub(super) struct SynthesisFlightGuard {
playback: Arc<PlaybackCoordinator>,
generation: u64,
}
impl Drop for SynthesisFlightGuard {
fn drop(&mut self) {
let mut state = self.playback.lock();
if state.synthesis_generation == self.generation {
state.synthesis_in_flight = false;
}
}
}
impl PlaybackCoordinator {
pub(super) fn new(mixer: &Mixer) -> Self {
Self {
mixer: mixer.clone(),
state: Mutex::new(PlaybackState {
player: Player::connect_new(mixer),
first_append: true,
synthesis_in_flight: false,
synthesis_generation: 0,
}),
}
}
fn lock(&self) -> MutexGuard<'_, PlaybackState> {
self.state.lock().unwrap_or_else(PoisonError::into_inner)
}
/// Queue `source` when `authorize` accepts, then publish the append with
/// `commit` before releasing the coordinator. `commit` runs under the lock
/// so an append and the activity state it implies are one transition: a
/// concurrent cancellation either replaces the queue before this append is
/// authorized, or observes the committed state after it — never lands its
/// own release between the two and gets overwritten.
pub(super) fn append_if<S>(
&self,
source: S,
authorize: impl FnOnce(bool) -> bool,
commit: impl FnOnce(),
) -> bool
where
S: Source<Item = f32> + Send + 'static,
{
let mut state = self.lock();
if !authorize(state.player.empty()) {
return false;
}
state.player.append(source);
state.first_append = false;
commit();
true
}
pub(super) fn append_untracked<S>(&self, source: S)
where
S: Source<Item = f32> + Send + 'static,
{
self.lock().player.append(source);
}
pub(super) fn empty(&self) -> bool {
self.lock().player.empty()
}
/// Observe playback emptiness under the coordinator so the onset decision
/// for the audio being built is serialized with append and cancellation.
pub(super) fn prepare_audio<R>(&self, prepare: impl FnOnce(bool) -> R) -> R {
let state = self.lock();
let empty = state.player.empty();
prepare(empty)
}
pub(super) fn release_if_drained(&self, release: impl FnOnce()) -> bool {
let mut state = self.lock();
if !state.player.empty() || state.first_append {
return false;
}
release();
state.first_append = true;
true
}
pub(super) fn begin_synthesis(self: &Arc<Self>) -> SynthesisFlightGuard {
let generation = {
let mut state = self.lock();
state.synthesis_generation = state.synthesis_generation.wrapping_add(1);
state.synthesis_in_flight = true;
state.synthesis_generation
};
SynthesisFlightGuard {
playback: Arc::clone(self),
generation,
}
}
pub(super) fn with_playback_live<R>(&self, observe: impl FnOnce(bool) -> R) -> R {
let state = self.lock();
observe(!state.player.empty() || state.synthesis_in_flight)
}
/// Replace live playback with a fresh queue, publishing the replacement
/// with `commit` before releasing the coordinator. The old player is
/// dropped after releasing, so rodio's teardown cannot extend the critical
/// section. Concurrent cancel observers elect exactly one replacement
/// because replacement resets both liveness signals.
///
/// `commit` is the mirror of `append_if`'s: a replacement and the activity
/// state it implies are one transition, so an append that wins the lock
/// handoff after this cancellation cannot have its own publication
/// overwritten by a `false` landing late.
pub(super) fn cancel_if_live(
&self,
authorize: impl FnOnce() -> bool,
commit: impl FnOnce(),
) -> bool {
let old_player = {
let mut state = self.lock();
if (state.player.empty() && !state.synthesis_in_flight) || !authorize() {
return false;
}
state.first_append = true;
state.synthesis_in_flight = false;
state.synthesis_generation = state.synthesis_generation.wrapping_add(1);
let old_player = std::mem::replace(&mut state.player, Player::connect_new(&self.mixer));
commit();
old_player
};
drop(old_player);
true
}
}
#[cfg(test)]
mod tests {
use std::{
num::NonZero,
sync::{
atomic::{AtomicBool, AtomicUsize, Ordering},
Arc, Barrier,
},
thread,
time::{Duration, Instant},
};
use rodio::buffer::SamplesBuffer;
use super::*;
fn coordinator() -> (Arc<PlaybackCoordinator>, rodio::mixer::MixerSource) {
let channels = NonZero::new(1).expect("nonzero channels");
let rate = NonZero::new(24_000).expect("nonzero rate");
let (mixer, source) = rodio::mixer::mixer(channels, rate);
(Arc::new(PlaybackCoordinator::new(&mixer)), source)
}
fn append_second(playback: &PlaybackCoordinator) {
playback.append_if(
SamplesBuffer::new(
NonZero::new(1).expect("nonzero channels"),
NonZero::new(24_000).expect("nonzero rate"),
vec![0.25; 24_000],
),
|_| true,
|| {},
);
}
#[test]
fn cancel_replaces_playback_without_waiting_for_the_mixer() {
let (playback, _unpulled_source) = coordinator();
append_second(&playback);
let started = Instant::now();
assert!(playback.cancel_if_live(|| true, || {}));
assert!(started.elapsed() < Duration::from_millis(50));
assert!(playback.empty());
}
#[test]
fn concurrent_cancel_observers_elect_exactly_one_replacement() {
let (playback, _unpulled_source) = coordinator();
append_second(&playback);
let barrier = Arc::new(Barrier::new(3));
let replacements = Arc::new(AtomicUsize::new(0));
let mut threads = Vec::new();
for _ in 0..2 {
let playback = Arc::clone(&playback);
let barrier = Arc::clone(&barrier);
let replacements = Arc::clone(&replacements);
threads.push(thread::spawn(move || {
barrier.wait();
if playback.cancel_if_live(|| true, || {}) {
replacements.fetch_add(1, Ordering::Relaxed);
}
}));
}
barrier.wait();
for thread in threads {
thread.join().expect("cancel observer");
}
assert_eq!(replacements.load(Ordering::Relaxed), 1);
}
#[test]
fn append_and_cancel_are_one_serialized_public_operation() {
let (playback, _unpulled_source) = coordinator();
append_second(&playback);
let append_authorized = Arc::new(Barrier::new(2));
let release_append = Arc::new(Barrier::new(2));
let append_thread = {
let playback = Arc::clone(&playback);
let append_authorized = Arc::clone(&append_authorized);
let release_append = Arc::clone(&release_append);
thread::spawn(move || {
playback.append_if(
SamplesBuffer::new(
NonZero::new(1).expect("nonzero channels"),
NonZero::new(24_000).expect("nonzero rate"),
vec![0.5; 24_000],
),
|_| {
append_authorized.wait();
release_append.wait();
true
},
|| {},
)
})
};
append_authorized.wait();
let cancel_thread = {
let playback = Arc::clone(&playback);
thread::spawn(move || playback.cancel_if_live(|| true, || {}))
};
release_append.wait();
assert!(append_thread.join().expect("append"));
assert!(cancel_thread.join().expect("cancel"));
assert!(
playback.empty(),
"cancel must replace the queue after append"
);
}
#[test]
fn an_accepted_append_publishes_its_activity_inside_the_append_transition() {
let (playback, _unpulled_source) = coordinator();
let committed = Arc::new(AtomicBool::new(false));
let appended = playback.append_if(
SamplesBuffer::new(
NonZero::new(1).expect("nonzero channels"),
NonZero::new(24_000).expect("nonzero rate"),
vec![0.25; 24_000],
),
|_| true,
|| {
// The coordinator is still held, so no cancellation can land a
// release between queueing this audio and publishing it.
assert!(
playback.state.try_lock().is_err(),
"commit must run inside the append's critical section"
);
committed.store(true, Ordering::Release);
},
);
assert!(appended);
assert!(
committed.load(Ordering::Acquire),
"an accepted append must publish"
);
assert!(
playback.state.try_lock().is_ok(),
"the coordinator is released once the append returns"
);
}
#[test]
fn a_refused_append_publishes_nothing_and_leaves_the_onset_armed() {
let (playback, _unpulled_source) = coordinator();
// The worker builds its buffer under the coordinator, then the append
// is refused — cancelled, or owned by another speaker.
playback.prepare_audio(|starts_playback_chunk| {
assert!(starts_playback_chunk, "a fresh coordinator is idle");
});
let appended = playback.append_if(
SamplesBuffer::new(
NonZero::new(1).expect("nonzero channels"),
NonZero::new(24_000).expect("nonzero rate"),
vec![0.25; 24_000],
),
|_| false,
|| panic!("a refused append must publish nothing"),
);
assert!(!appended);
// Nothing was queued, so this is not a drained utterance: releasing
// here would drop the mic gate and log a drain for audio that never
// played, and cost the next append its onset cushion.
assert!(
!playback.release_if_drained(|| panic!("a refused append is not a drain")),
"a refused append must not present as a drained utterance"
);
}
#[test]
fn an_appended_utterance_still_releases_exactly_once_when_it_drains() {
let (playback, mut source) = coordinator();
append_second(&playback);
assert!(
!playback.release_if_drained(|| panic!("queued audio is not drained")),
"queued audio must not release"
);
while !playback.empty() {
assert!(
source.next().is_some(),
"the mixer source outlives the queue"
);
}
let releases = Arc::new(AtomicUsize::new(0));
for _ in 0..2 {
let releases = Arc::clone(&releases);
playback.release_if_drained(move || {
releases.fetch_add(1, Ordering::Relaxed);
});
}
assert_eq!(
releases.load(Ordering::Relaxed),
1,
"a drained utterance releases once and rearms the onset"
);
}
/// The reverse direction of the same barrier: a cancellation must publish
/// its release *inside* the replacement. The cancelling thread is
/// otherwise past its replacement and about to release the mic gate, while
/// an append that wins the coordinator handoff has already published its
/// own `true` — a `false` landing outside the replacement would ungate the
/// mic for audio that is actually playing, and VAD would hear our own TTS
/// and barge in on it.
#[test]
fn a_replacement_publishes_its_release_inside_the_cancel_transition() {
let (playback, _unpulled_source) = coordinator();
append_second(&playback);
let committed = Arc::new(AtomicBool::new(false));
let replaced = playback.cancel_if_live(
|| true,
|| {
// Still held: no append can commit its own activity between
// this replacement and the release it implies.
assert!(
playback.state.try_lock().is_err(),
"commit must run inside the cancellation's critical section"
);
committed.store(true, Ordering::Release);
},
);
assert!(replaced, "live playback must be replaced");
assert!(
committed.load(Ordering::Acquire),
"a replacement must publish"
);
assert!(
playback.state.try_lock().is_ok(),
"the coordinator is released once the cancellation returns"
);
}
/// A cancellation that replaces nothing publishes nothing: the caller
/// still owns releasing the gate, and a replacement that never happened
/// must not present as one.
#[test]
fn a_cancellation_with_nothing_live_publishes_nothing() {
let (playback, _unpulled_source) = coordinator();
assert!(!playback.cancel_if_live(|| true, || panic!("nothing was replaced")));
assert!(!playback.cancel_if_live(|| false, || panic!("cancellation was refused")));
}
/// Both publication directions under real contention: whichever
/// transition takes the coordinator last decides, and the activity flag
/// must describe the player that survived.
#[test]
fn a_cancellation_and_an_append_never_disagree_about_the_mic_gate() {
for _ in 0..256 {
let (playback, _unpulled_source) = coordinator();
let tts_active = Arc::new(AtomicBool::new(true));
append_second(&playback);
let barrier = Arc::new(Barrier::new(2));
let canceller = {
let playback = Arc::clone(&playback);
let tts_active = Arc::clone(&tts_active);
let barrier = Arc::clone(&barrier);
thread::spawn(move || {
barrier.wait();
playback.cancel_if_live(|| true, || tts_active.store(false, Ordering::Release))
})
};
let appender = {
let playback = Arc::clone(&playback);
let tts_active = Arc::clone(&tts_active);
let barrier = Arc::clone(&barrier);
thread::spawn(move || {
barrier.wait();
playback.append_if(
SamplesBuffer::new(
NonZero::new(1).expect("nonzero channels"),
NonZero::new(24_000).expect("nonzero rate"),
vec![0.5; 24_000],
),
|_| true,
|| tts_active.store(true, Ordering::Release),
)
})
};
let replaced = canceller.join().expect("canceller");
let appended = appender.join().expect("appender");
assert!(replaced, "live playback must be replaced");
assert!(appended, "the append is authorized either way");
assert_eq!(
tts_active.load(Ordering::Acquire),
!playback.empty(),
"the mic gate must agree with the player that survived"
);
}
}
#[test]
fn cancellation_rearms_first_append_and_releases_activity_once() {
let (playback, _unpulled_source) = coordinator();
append_second(&playback);
assert!(playback.cancel_if_live(|| true, || {}));
assert!(!playback.release_if_drained(|| panic!("fresh replacement is not a drain")));
playback.prepare_audio(|starts_playback_chunk| {
assert!(
starts_playback_chunk,
"the first append after replacement must carry the onset cushion"
);
});
append_second(&playback);
assert!(!playback.release_if_drained(|| panic!("queued audio is not drained")));
}
}

212
desktop/src-tauri/src/huddle/tts_speaker_cancellation.rs
View File

@@ -1,12 +1,11 @@
use super::*;
pub(super) struct TtsMonitorState {
pub(super) player: Arc<rodio::Player>,
pub(super) playback: Arc<PlaybackCoordinator>,
pub(super) cancel: Arc<AtomicBool>,
pub(super) voice_cancel: Arc<AtomicBool>,
pub(super) tts_active: Arc<AtomicBool>,
pub(super) stop: Arc<AtomicBool>,
pub(super) player_ops: Arc<Mutex<()>>,
pub(super) activity_frames: Arc<Mutex<VecDeque<TtsSpeakerActivityFrame>>>,
pub(super) active_speaker: ActiveSpeaker,
pub(super) speaker_cancel: SpeakerCancellation,
@@ -23,25 +22,28 @@ pub(super) fn spawn_tts_monitor(state: TtsMonitorState) -> std::io::Result<threa
if state.cancel.load(Ordering::Acquire)
|| state.voice_cancel.load(Ordering::Acquire)
{
let _ops = lock_player_ops(&state.player_ops);
if state.cancel.load(Ordering::Acquire)
|| state.voice_cancel.load(Ordering::Acquire)
{
state.player.clear();
state.player.play();
state
.active_speaker
.lock()
.unwrap_or_else(|error| error.into_inner())
.take();
state.tts_active.store(false, Ordering::Release);
}
state.playback.cancel_if_live(
|| {
state.cancel.load(Ordering::Acquire)
|| state.voice_cancel.load(Ordering::Acquire)
},
// Publish the release inside the replacement so an
// append winning the lock handoff cannot have its own
// activity publication overwritten by this `false`.
|| {
state
.active_speaker
.lock()
.unwrap_or_else(|error| error.into_inner())
.take();
state.tts_active.store(false, Ordering::Release);
},
);
}
silence_cancelled_speaker(
&state.speaker_cancel,
&state.active_speaker,
&state.player,
&state.player_ops,
&state.playback,
&state.tts_active,
);
if let Some(ref app) = state.activity_app {
@@ -94,8 +96,7 @@ pub(super) fn spawn_tts_monitor(state: TtsMonitorState) -> std::io::Result<threa
pub(super) fn silence_cancelled_speaker(
cancellation: &SpeakerCancellation,
active_speaker: &ActiveSpeaker,
player: &rodio::Player,
player_ops: &Mutex<()>,
playback: &PlaybackCoordinator,
tts_active: &AtomicBool,
) {
let Some(cancelled) = cancellation
@@ -105,12 +106,10 @@ pub(super) fn silence_cancelled_speaker(
else {
return;
};
let _ops = lock_player_ops(player_ops);
if take_cancelled_active_speaker(&cancelled, active_speaker) {
player.clear();
player.play();
tts_active.store(false, Ordering::Release);
}
playback.cancel_if_live(
|| take_cancelled_active_speaker(&cancelled, active_speaker),
|| tts_active.store(false, Ordering::Release),
);
}
fn take_cancelled_active_speaker(cancelled: &str, active_speaker: &ActiveSpeaker) -> bool {
@@ -133,7 +132,7 @@ pub(super) fn consume_speaker_cancel(
generations: &SpeakerGenerations,
tts_active: &AtomicBool,
text_state: CancelTextState<'_>,
player: Option<(&rodio::Player, &Mutex<()>)>,
playback: Option<&PlaybackCoordinator>,
) -> bool {
let Some(cancelled) = cancellation
.lock()
@@ -145,12 +144,11 @@ pub(super) fn consume_speaker_cancel(
let (text_rx, deferred_text, current_text) = text_state;
retain_current_speaker_text(generations, deferred_text, current_text, text_rx);
let mut cleared_player = false;
if let Some((player, player_ops)) = player {
let _ops = lock_player_ops(player_ops);
if take_cancelled_active_speaker(&cancelled, active_speaker) {
player.clear();
player.play();
tts_active.store(false, Ordering::Release);
if let Some(playback) = playback {
if playback.cancel_if_live(
|| take_cancelled_active_speaker(&cancelled, active_speaker),
|| tts_active.store(false, Ordering::Release),
) {
cleared_player = true;
}
}
@@ -164,6 +162,31 @@ pub(super) fn consume_speaker_cancel(
mod tests {
use super::*;
use std::sync::Barrier;
use rodio::buffer::SamplesBuffer;
/// Headless coordinator: a real mixer with its source held unpulled, so
/// the queue never drains and no output device is opened.
fn coordinator() -> (Arc<PlaybackCoordinator>, rodio::mixer::MixerSource) {
let channels = std::num::NonZero::new(1).expect("nonzero channels");
let rate = std::num::NonZero::new(24_000).expect("nonzero rate");
let (mixer, source) = rodio::mixer::mixer(channels, rate);
(Arc::new(PlaybackCoordinator::new(&mixer)), source)
}
/// Append as `speaker` the way the worker does: activity is published
/// inside the append transition.
fn speak(playback: &PlaybackCoordinator, tts_active: &AtomicBool) {
let channels = std::num::NonZero::new(1).expect("nonzero channels");
let rate = std::num::NonZero::new(24_000).expect("nonzero rate");
assert!(playback.append_if(
SamplesBuffer::new(channels, rate, vec![0.25; 24_000]),
|_| true,
|| tts_active.store(true, Ordering::Release),
));
}
#[test]
fn stale_targeted_cancel_does_not_release_the_next_speaker() {
let active_speaker = Arc::new(Mutex::new(Some("bob".to_string())));
@@ -174,4 +197,129 @@ mod tests {
Some("bob")
);
}
/// The wedge Mari found: the monitor silences the cancelled speaker while
/// the worker is mid-append. The monitor takes `active_speaker`, so the
/// worker's later `consume_speaker_cancel` fails authorization and never
/// clears `tts_active` — if the worker's `true` could land after the
/// monitor's `false`, mic gating stays active with nothing playing.
#[test]
fn a_targeted_cancel_racing_an_append_leaves_the_mic_gate_released() {
for _ in 0..64 {
let (playback, _unpulled_source) = coordinator();
let tts_active = Arc::new(AtomicBool::new(false));
let active_speaker: ActiveSpeaker = Arc::new(Mutex::new(None));
let speaker_cancel: SpeakerCancellation = Arc::new(Mutex::new(None));
speak(&playback, &tts_active);
active_speaker
.lock()
.expect("active speaker")
.replace("alice".to_string());
speaker_cancel
.lock()
.expect("speaker cancel")
.replace("alice".to_string());
let barrier = Arc::new(Barrier::new(2));
let monitor = {
let playback = Arc::clone(&playback);
let tts_active = Arc::clone(&tts_active);
let active_speaker = Arc::clone(&active_speaker);
let speaker_cancel = Arc::clone(&speaker_cancel);
let barrier = Arc::clone(&barrier);
thread::spawn(move || {
barrier.wait();
silence_cancelled_speaker(
&speaker_cancel,
&active_speaker,
&playback,
&tts_active,
);
})
};
let worker = {
let playback = Arc::clone(&playback);
let tts_active = Arc::clone(&tts_active);
let barrier = Arc::clone(&barrier);
thread::spawn(move || {
barrier.wait();
// The worker appends the next chunk of the utterance the
// monitor is cancelling.
playback.append_if(
SamplesBuffer::new(
std::num::NonZero::new(1).expect("nonzero channels"),
std::num::NonZero::new(24_000).expect("nonzero rate"),
vec![0.25; 24_000],
),
|_| true,
|| tts_active.store(true, Ordering::Release),
)
})
};
let appended = worker.join().expect("worker");
monitor.join().expect("monitor");
// Whichever order the two took the coordinator, the surviving
// activity flag must describe the surviving player.
assert_eq!(
tts_active.load(Ordering::Acquire),
!playback.empty(),
"the mic gate must agree with the player that survived \
(appended={appended})"
);
}
}
/// The worker arm of the same race: the monitor already took the speaker,
/// so `consume_speaker_cancel` is not authorized to clear anything. It
/// must not report a clear it did not perform, and it must not disturb the
/// activity flag the monitor already published.
#[test]
fn consuming_a_cancel_the_monitor_already_handled_preserves_the_released_gate() {
let (playback, _unpulled_source) = coordinator();
let tts_active = Arc::new(AtomicBool::new(false));
let active_speaker: ActiveSpeaker = Arc::new(Mutex::new(None));
let speaker_cancel: SpeakerCancellation = Arc::new(Mutex::new(None));
let generations: SpeakerGenerations = Arc::new(Mutex::new(HashMap::new()));
let (_text_tx, text_rx) = mpsc::channel::<QueuedText>();
let mut deferred_text = VecDeque::new();
let mut current_text = None;
speak(&playback, &tts_active);
active_speaker
.lock()
.expect("active speaker")
.replace("alice".to_string());
speaker_cancel
.lock()
.expect("speaker cancel")
.replace("alice".to_string());
silence_cancelled_speaker(&speaker_cancel, &active_speaker, &playback, &tts_active);
assert!(
!tts_active.load(Ordering::Acquire),
"the monitor releases the mic gate it cancelled"
);
let cleared = consume_speaker_cancel(
&speaker_cancel,
&active_speaker,
&generations,
&tts_active,
(&text_rx, &mut deferred_text, &mut current_text),
Some(&playback),
);
assert!(
!cleared,
"the worker must not claim a clear the monitor already performed"
);
assert!(
!tts_active.load(Ordering::Acquire),
"the released mic gate must survive the worker's pass"
);
assert!(playback.empty(), "the cancelled utterance stays silenced");
}
}

15
desktop/src-tauri/src/huddle/tts_streaming.rs
View File

@@ -27,8 +27,7 @@ pub(super) fn streaming_emit_frames() -> Option<usize> {
/// Playback context threaded through one streamed chunk.
pub(super) struct StreamingPlayback<'a> {
pub(super) player: &'a rodio::Player,
pub(super) first_append: &'a mut bool,
pub(super) playback: &'a PlaybackCoordinator,
pub(super) route_id: u64,
}
@@ -52,11 +51,7 @@ pub(super) fn synthesize_streaming(
append_audio: &mut dyn FnMut(PreparedModelAudio) -> bool,
) -> Option<&'static str> {
let (cancel, voice_cancel, shutdown) = signals;
let StreamingPlayback {
player,
first_append,
route_id,
} = playback;
let StreamingPlayback { playback, route_id } = playback;
let mut playback_audio = PlaybackChunkAudio::new();
let mut delta_index = 0usize;
let stream_result = engine.synth_chunk_streaming(text, style, emit_frames, &mut |samples| {
@@ -69,7 +64,7 @@ pub(super) fn synthesize_streaming(
let chunk_index = delta_index;
delta_index += 1;
if let Some(prepared) =
playback_audio.push(samples, chunk_index, first_append, player.empty())
playback.prepare_audio(|empty| playback_audio.push(samples, chunk_index, empty))
{
if !append_audio(prepared) {
return false;
@@ -79,9 +74,8 @@ pub(super) fn synthesize_streaming(
});
match stream_result {
Ok(true) => {
if let Some(prepared) = playback_audio.finish(first_append, player.empty()) {
if let Some(prepared) = playback.prepare_audio(|empty| playback_audio.finish(empty)) {
if !append_audio(prepared) {
*first_append = true;
return Some("cancelled");
}
}
@@ -91,7 +85,6 @@ pub(super) fn synthesize_streaming(
eprintln!(
"buzz-desktop: tts stage=synthesis status=cancelled reason=stream_callback route_id={route_id}"
);
*first_append = true;
Some("cancelled")
}
Err(_) => {

17
desktop/src-tauri/src/huddle/tts_tests.rs
View File

@@ -785,22 +785,12 @@ fn apply_fade_out_single_sample() {
// ── build_sentence_append_buffer tests ───────────────────────────────────
/// `first_append` still flips on the first append for `tts_active` gating.
#[test]
fn build_sentence_append_buffer_flips_first_append() {
let mut first = true;
let buf = build_sentence_append_buffer(&mut first, vec![0.5; 100], false);
assert_eq!(buf, vec![0.5; 100]);
assert!(!first, "first call must flip the flag");
}
/// Playback chunks are contiguous: Pocket's generated pause is not extended
/// with a fixed inter-sentence silence budget.
#[test]
fn sentence_append_buffer_does_not_inject_silence() {
let mut first = true;
let first_buf = build_sentence_append_buffer(&mut first, vec![0.5; 100], false);
let second_buf = build_sentence_append_buffer(&mut first, vec![0.25; 100], false);
let first_buf = build_sentence_append_buffer(vec![0.5; 100], false);
let second_buf = build_sentence_append_buffer(vec![0.25; 100], false);
assert_eq!(first_buf, vec![0.5; 100]);
assert_eq!(second_buf, vec![0.25; 100]);
@@ -810,8 +800,7 @@ fn sentence_append_buffer_does_not_inject_silence() {
/// the first phoneme while the output path wakes back up.
#[test]
fn idle_playback_gets_an_onset_cushion() {
let mut first = true;
let buf = build_sentence_append_buffer(&mut first, vec![0.5; 100], true);
let buf = build_sentence_append_buffer(vec![0.5; 100], true);
assert_eq!(buf.len(), SENTENCE_LEAD_IN_SAMPLES + 100);
assert!(buf[..SENTENCE_LEAD_IN_SAMPLES].iter().all(|&s| s == 0.0));

11
desktop/src-tauri/src/huddle/tts_tests/token_split.rs
View File

@@ -1,18 +1,19 @@
use super::*;
/// The onset cushion covers 20 ms at the production sample rate.
/// The onset cushion rounds rodio's 512-sample bootstrap span up to 22 ms at
/// the 24 kHz production sample rate (528 samples).
#[test]
fn chunk_lead_in_is_sane() {
assert_eq!(SENTENCE_LEAD_IN_SAMPLES, 480, "20 ms × 24 kHz");
assert_eq!(RODIO_ADD_BOOTSTRAP_SPAN_SAMPLES, 512);
assert_eq!(SENTENCE_LEAD_IN_SAMPLES, 528, "22 ms × 24 kHz");
}
/// Model-token splits remain contiguous: only the playback chunk as a whole
/// receives its onset cushion and trailing sentence gap.
#[test]
fn token_split_units_do_not_add_sentence_boundary_padding() {
let mut first = true;
let first_unit = build_sentence_append_buffer(&mut first, vec![0.5; 100], false);
let last_unit = build_sentence_append_buffer(&mut first, vec![0.25; 100], false);
let first_unit = build_sentence_append_buffer(vec![0.5; 100], false);
let last_unit = build_sentence_append_buffer(vec![0.25; 100], false);
assert_eq!(first_unit.len(), 100);
assert_eq!(first_unit.last(), Some(&0.5));

168
desktop/src-tauri/src/huddle/tts_voice_transition.rs
View File

@@ -5,10 +5,12 @@ use std::{
sync::{
atomic::{AtomicBool, AtomicU64, Ordering},
mpsc::{self, SyncSender},
Arc, Mutex, MutexGuard, PoisonError,
Arc, Mutex,
},
};
use super::{PlaybackCoordinator, SynthesisFlightGuard};
use crate::huddle::pocket::{load_voice_style, VoiceStyle, DEFAULT_VOICE, VOICE_FILE_EXT};
#[derive(Debug)]
@@ -32,51 +34,29 @@ pub(super) type CancelSignals<'a> = (&'a AtomicBool, &'a AtomicBool);
#[derive(Clone)]
pub(super) struct PlaybackProbe {
player: Arc<Mutex<Option<Arc<rodio::Player>>>>,
pub(super) player_ops: Arc<Mutex<()>>,
synthesis_in_flight: Arc<AtomicBool>,
}
pub(super) struct SynthesisFlightGuard {
playback_probe: PlaybackProbe,
}
impl Drop for SynthesisFlightGuard {
fn drop(&mut self) {
self.playback_probe.set_synthesis_in_flight(false);
}
playback: Arc<Mutex<Option<Arc<PlaybackCoordinator>>>>,
}
impl PlaybackProbe {
pub(super) fn new() -> Self {
Self {
player: Arc::new(Mutex::new(None)),
player_ops: Arc::new(Mutex::new(())),
synthesis_in_flight: Arc::new(AtomicBool::new(false)),
playback: Arc::new(Mutex::new(None)),
}
}
pub(super) fn install(&self, player: Arc<rodio::Player>) {
self.player
pub(super) fn install(&self, playback: Arc<PlaybackCoordinator>) {
self.playback
.lock()
.unwrap_or_else(|error| error.into_inner())
.replace(player);
.replace(playback);
}
pub(super) fn set_synthesis_in_flight(&self, in_flight: bool) {
let _ops = lock_player_ops(&self.player_ops);
self.synthesis_in_flight.store(in_flight, Ordering::Release);
pub(super) fn begin_synthesis(&self) -> Option<SynthesisFlightGuard> {
self.playback().map(|playback| playback.begin_synthesis())
}
pub(super) fn begin_synthesis(&self) -> SynthesisFlightGuard {
self.set_synthesis_in_flight(true);
SynthesisFlightGuard {
playback_probe: self.clone(),
}
}
fn player(&self) -> Option<Arc<rodio::Player>> {
self.player
pub(super) fn playback(&self) -> Option<Arc<PlaybackCoordinator>> {
self.playback
.lock()
.unwrap_or_else(|error| error.into_inner())
.clone()
@@ -199,19 +179,18 @@ pub(super) fn request_active_speaker_cancel(
playback_probe: &PlaybackProbe,
expected_speaker_pubkey: &str,
) -> bool {
let Some(player) = playback_probe.player() else {
let Some(playback) = playback_probe.playback() else {
return false;
};
let _ops = lock_player_ops(&playback_probe.player_ops);
let playback_live =
!player.empty() || playback_probe.synthesis_in_flight.load(Ordering::Acquire);
request_active_speaker_cancel_while_locked(
generations,
active_speaker,
cancellation,
playback_live,
expected_speaker_pubkey,
)
playback.with_playback_live(|playback_live| {
request_active_speaker_cancel_while_locked(
generations,
active_speaker,
cancellation,
playback_live,
expected_speaker_pubkey,
)
})
}
fn request_active_speaker_cancel_while_locked(
@@ -475,10 +454,8 @@ fn log_cancelled_route(route_id: u64, reason: &str) {
/// Check for cancel or shutdown. Returns `true` if the caller should break/continue.
/// On cancel: drains the text queue and clears the cancel flag.
///
/// `player` pairs the Player with the `player_ops` mutex shared with the
/// barge-in monitor thread; the cancel/shutdown clear runs under that lock so
/// it is serialized with the monitor's stale-branch re-check (see the monitor
/// block in `tts_worker`).
/// `playback` is the coordinator shared with the barge-in monitor; replacing
/// playback is serialized with append and with the monitor's stale observation.
pub(super) fn handle_cancel_or_shutdown(
cancel_signals: CancelSignals<'_>,
shutdown: &AtomicBool,
@@ -486,7 +463,7 @@ pub(super) fn handle_cancel_or_shutdown(
text_state: CancelTextState<'_>,
voice_change_ack: &VoiceChangeAck,
active_route_id: Option<u64>,
player: Option<(&rodio::Player, &Mutex<()>)>,
playback: Option<&PlaybackCoordinator>,
) -> bool {
let (cancel, voice_cancel) = cancel_signals;
let (text_rx, deferred_text, current_text) = text_state;
@@ -495,11 +472,7 @@ pub(super) fn handle_cancel_or_shutdown(
"buzz-desktop: tts stage=cancellation reason=shutdown route_id={}",
active_route_id.unwrap_or(0)
);
if let Some((p, ops)) = player {
let _ops = lock_player_ops(ops);
p.clear();
}
tts_active.store(false, Ordering::Release);
release_playback(playback, tts_active);
return true;
}
if cancel.load(Ordering::Acquire) || voice_cancel.load(Ordering::Acquire) {
@@ -525,33 +498,27 @@ pub(super) fn handle_cancel_or_shutdown(
})
.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
// (rodio 0.22 `clear()` ends with `self.pause()`). With one
// persistent Player for the worker's lifetime, the un-pause is
// mandatory: without `play()`, every append after a barge-in
// would queue silently forever.
p.clear();
p.play();
// 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.
}
tts_active.store(false, Ordering::Release);
// Consume the flag at the coordinator serialization point: once
// released with `cancel == false`, a stale monitor observation cannot
// replace fresh post-cancel playback.
release_playback(playback, tts_active);
return true;
}
false
}
/// Acquire the `player_ops` lock, recovering from poison.
///
/// The data under the mutex is `()` — it only serializes Player mutations —
/// so a panicked holder leaves nothing inconsistent to observe and recovery
/// is always safe. Without this, a worker panic would wedge the monitor (or
/// vice versa) on `unwrap()`.
pub(super) fn lock_player_ops(ops: &Mutex<()>) -> MutexGuard<'_, ()> {
ops.lock().unwrap_or_else(PoisonError::into_inner)
/// Silence playback and release the mic gate as one transition. When a player
/// is live the release is published inside the replacement, so an append that
/// wins the coordinator handoff cannot have its own activity publication
/// overwritten by this `false`. With nothing live there is no transition to
/// join and the gate is released directly.
fn release_playback(playback: Option<&PlaybackCoordinator>, tts_active: &AtomicBool) {
let released = playback.is_some_and(|playback| {
playback.cancel_if_live(|| true, || tts_active.store(false, Ordering::Release))
});
if !released {
tts_active.store(false, Ordering::Release);
}
}
#[cfg(test)]
@@ -562,16 +529,16 @@ mod speaker_generation_tests {
let channels = std::num::NonZero::new(1).expect("non-zero channels");
let sample_rate = std::num::NonZero::new(24_000).expect("non-zero sample rate");
let (mixer, _mixer_source) = rodio::mixer::mixer(channels, sample_rate);
let player = Arc::new(rodio::Player::connect_new(&mixer));
let playback = Arc::new(PlaybackCoordinator::new(&mixer));
if playback_live {
player.append(rodio::buffer::SamplesBuffer::new(
channels,
sample_rate,
vec![0.0; 24_000],
));
playback.append_if(
rodio::buffer::SamplesBuffer::new(channels, sample_rate, vec![0.0; 24_000]),
|_| true,
|| {},
);
}
let probe = PlaybackProbe::new();
probe.install(player);
probe.install(playback);
probe
}
@@ -586,6 +553,43 @@ mod speaker_generation_tests {
}
}
/// A cancellation releases the mic gate whether or not there was audio to
/// silence. With a live player the release rides inside the replacement;
/// with nothing live there is no transition to join, and skipping the
/// release would strand the gate open with the worker already past the
/// utterance.
#[test]
fn cancellation_releases_the_mic_gate_with_or_without_live_playback() {
for playback_live in [false, true] {
let probe = playback_probe(playback_live);
let playback = probe.playback().expect("installed coordinator");
let cancel = AtomicBool::new(true);
let voice_cancel = AtomicBool::new(false);
let shutdown = AtomicBool::new(false);
let tts_active = AtomicBool::new(true);
let voice_change_ack = Arc::new(Mutex::new(None));
let (_text_tx, text_rx) = mpsc::channel();
let mut deferred_text = VecDeque::new();
let mut current_text = None;
assert!(handle_cancel_or_shutdown(
(&cancel, &voice_cancel),
&shutdown,
&tts_active,
(&text_rx, &mut deferred_text, &mut current_text),
&voice_change_ack,
None,
Some(&playback),
));
assert!(
!tts_active.load(Ordering::Acquire),
"cancellation must release the mic gate (playback_live={playback_live})"
);
assert!(playback.empty(), "cancellation silences any queued audio");
}
}
#[test]
fn removing_a_speaker_invalidates_only_that_speakers_queued_text() {
let generations = Arc::new(Mutex::new(HashMap::new()));