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

Merge PR 1: coordinate TTS playback

Browse Source 
Co-authored-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz>
Signed-off-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz>
This commit is contained in:
Wren
2026-08-19 14:21:12 -04:00
parent bbd20fae75 8bfa4c0531
commit 203735f15a
6 changed files with 446 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files

236
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,74 +440,72 @@ 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
},
);
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 +518,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 +529,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 +562,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 +571,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 +588,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 +612,7 @@ fn tts_worker(
);
continue;
}
if !player.empty()
if !playback.empty()
&& queued_text
.speaker_pubkey
.as_deref()
@@ -639,18 +644,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 +717,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 +738,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 +790,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 +815,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(|first_append, empty| {
playback_audio.push(samples, chunk_index, first_append, empty)
}) {
if !append_audio(
prepared,
route_id,
speaker_pubkey.as_deref(),
speaker_generation,
) {
first_append = true;
synthesis_outcome = "cancelled";
break 'playback_chunks;
}
@@ -857,14 +850,15 @@ fn tts_worker(
}
}
}
if let Some(prepared) = playback_audio.finish(&mut first_append, player.empty()) {
if let Some(prepared) = playback
.prepare_audio(|first_append, empty| playback_audio.finish(first_append, empty))
{
if !append_audio(
prepared,
route_id,
speaker_pubkey.as_deref(),
speaker_generation,
) {
first_append = true;
synthesis_outcome = "cancelled";
break 'playback_chunks;
}
@@ -884,8 +878,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();

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

@@ -0,0 +1,262 @@
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,
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)
}
pub(super) fn append_if<S>(&self, source: S, authorize: impl FnOnce(bool) -> bool) -> 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;
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()
}
pub(super) fn prepare_audio<R>(&self, prepare: impl FnOnce(&mut bool, bool) -> R) -> R {
let mut state = self.lock();
let empty = state.player.empty();
prepare(&mut state.first_append, 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. The old player is dropped
/// after releasing the coordinator, so rodio's teardown cannot extend the
/// critical section. Concurrent cancel observers elect exactly one
/// replacement because replacement resets both liveness signals.
pub(super) fn cancel_if_live(&self, authorize: impl FnOnce() -> bool) -> 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);
std::mem::replace(&mut state.player, Player::connect_new(&self.mixer))
};
drop(old_player);
true
}
}
#[cfg(test)]
mod tests {
use std::{
num::NonZero,
sync::{
atomic::{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 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(|first_append, starts_playback_chunk| {
assert!(*first_append, "replacement must rearm the first append");
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")));
}
}

50
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,
@@ -20,28 +19,24 @@ pub(super) fn spawn_tts_monitor(state: TtsMonitorState) -> std::io::Result<threa
let mut last_activity_pubkey: Option<String> = None;
let mut next_activity_tick = Instant::now();
while !state.stop.load(Ordering::Acquire) {
if state.cancel.load(Ordering::Acquire)
|| state.voice_cancel.load(Ordering::Acquire)
if (state.cancel.load(Ordering::Acquire)
|| state.voice_cancel.load(Ordering::Acquire))
&& state.playback.cancel_if_live(|| {
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
.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 +89,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,10 +99,7 @@ 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();
if playback.cancel_if_live(|| take_cancelled_active_speaker(&cancelled, active_speaker)) {
tts_active.store(false, Ordering::Release);
}
}
@@ -133,7 +124,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,11 +136,8 @@ 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();
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;
}

21
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| {
@@ -68,9 +63,9 @@ 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())
{
if let Some(prepared) = playback.prepare_audio(|first_append, empty| {
playback_audio.push(samples, chunk_index, first_append, empty)
}) {
if !append_audio(prepared) {
return false;
}
@@ -79,9 +74,10 @@ 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(|first_append, empty| playback_audio.finish(first_append, empty))
{
if !append_audio(prepared) {
*first_append = true;
return Some("cancelled");
}
}
@@ -91,7 +87,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(_) => {

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

@@ -1,9 +1,11 @@
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

118
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,9 +472,8 @@ 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();
if let Some(playback) = playback {
playback.cancel_if_live(|| true);
}
tts_active.store(false, Ordering::Release);
return true;
@@ -525,18 +501,11 @@ 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.
if let Some(playback) = playback {
// Consume the flag at the coordinator serialization point: once
// released with `cancel == false`, a stale monitor observation
// cannot replace fresh post-cancel playback.
playback.cancel_if_live(|| true);
}
tts_active.store(false, Ordering::Release);
return true;
@@ -544,16 +513,6 @@ pub(super) fn handle_cancel_or_shutdown(
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)
}
#[cfg(test)]
mod speaker_generation_tests {
use super::*;
@@ -562,16 +521,15 @@ 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
}