diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index 5d5ee50ce..11ef77893 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -51,7 +51,6 @@ dependencies = [ "sha2", "sherpa-onnx", "sqlx", - "ssstretch", "swift-rs", "sysinfo", "tar", @@ -991,17 +990,6 @@ dependencies = [ "cc", ] -[[package]] -name = "codespan-reporting" -version = "0.13.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "af491d569909a7e4dee0ad7db7f5341fef5c614d5b8ec8cf765732aba3cff681" -dependencies = [ - "serde", - "termcolor", - "unicode-width 0.2.2", -] - [[package]] name = "colorchoice" version = "1.0.5" @@ -1331,68 +1319,6 @@ version = "0.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52560adf09603e58c9a7ee1fe1dcb95a16927b17c127f0ac02d6e768a0e25bc1" -[[package]] -name = "cxx" -version = "1.0.199" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "824894a4a85dca76d4c95c2b9098c036f5a29f627b30c12780774f6654e60974" -dependencies = [ - "cc", - "cxx-build", - "cxxbridge-cmd", - "cxxbridge-flags", - "cxxbridge-macro", - "foldhash 0.2.0", - "link-cplusplus", -] - -[[package]] -name = "cxx-build" -version = "1.0.199" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f1ae0b651ea5b0000b19513aef5a03f194d7e3486f2d9258b658da8677fe9036" -dependencies = [ - "cc", - "codespan-reporting", - "indexmap 2.13.1", - "proc-macro2", - "quote", - "scratch", - "syn 3.0.3", -] - -[[package]] -name = "cxxbridge-cmd" -version = "1.0.199" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb05f91d3fb8435d9bab6ac5ce6ac1868be774325fb7fb2a91be39393b21388e" -dependencies = [ - "clap", - "codespan-reporting", - "indexmap 2.13.1", - "proc-macro2", - "quote", - "syn 3.0.3", -] - -[[package]] -name = "cxxbridge-flags" -version = "1.0.199" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bf293202e0e3e98495785745389e8d0755b217e66f19194a5c695c25e03282ef" - -[[package]] -name = "cxxbridge-macro" -version = "1.0.199" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca001d746947c7249ed9d332a10f7a59daedbafeb0ec68c5c18a7db7a93f6ccc" -dependencies = [ - "indexmap 2.13.1", - "proc-macro2", - "quote", - "syn 3.0.3", -] - [[package]] name = "darling" version = "0.20.11" @@ -3346,15 +3272,6 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2e126dda6f34391ab7b444f9922055facc83c07a910da3eb16f1e4d9c45dc777" -[[package]] -name = "link-cplusplus" -version = "1.0.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f78c730aaa7d0b9336a299029ea49f9ee53b0ed06e9202e8cb7db9bae7b8c82" -dependencies = [ - "cc", -] - [[package]] name = "linux-raw-sys" version = "0.12.1" @@ -3526,7 +3443,7 @@ checksum = "5f98efec8807c63c752b5bd61f862c165c115b0a35685bdcfd9238c7aeb592b7" dependencies = [ "cfg-if", "miette-derive", - "unicode-width 0.1.14", + "unicode-width", ] [[package]] @@ -5483,12 +5400,6 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" -[[package]] -name = "scratch" -version = "1.0.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d68f2ec51b097e4c1a75b681a8bec621909b5e91f15bb7b840c4f2f7b01148b2" - [[package]] name = "security-framework" version = "3.7.0" @@ -6196,16 +6107,6 @@ dependencies = [ "url", ] -[[package]] -name = "ssstretch" -version = "0.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b4ee31a0a494b76c8d3047aac804c5f4eb4b6e96c75414e3043b2212301bb8c6" -dependencies = [ - "cxx", - "cxx-build", -] - [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -6958,15 +6859,6 @@ dependencies = [ "new_debug_unreachable", ] -[[package]] -name = "termcolor" -version = "1.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "06794f8f6c5c898b3275aebefa6b8a1cb24cd2c6c79397ab15774837a0bc5755" -dependencies = [ - "winapi-util", -] - [[package]] name = "terminal_size" version = "0.4.4" @@ -7557,12 +7449,6 @@ version = "0.1.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af" -[[package]] -name = "unicode-width" -version = "0.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b4ac048d71ede7ee76d585517add45da530660ef4390e49b098733c6e897f254" - [[package]] name = "unicode_categories" version = "0.1.1" diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 505641514..c73cb6ff3 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -71,7 +71,6 @@ tar = "0.4" # 1.13.5 and reintroduces the CI failure; re-pin it with # `cargo update -p sherpa-onnx-sys --precise 1.12.40`. sherpa-onnx = "~1.12.40" -ssstretch = "0.1.0" semver = "1" serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/src-tauri/native/siri_tts_bridge.h b/src-tauri/native/siri_tts_bridge.h index 9302a0ffb..b97cba220 100644 --- a/src-tauri/native/siri_tts_bridge.h +++ b/src-tauri/native/siri_tts_bridge.h @@ -73,6 +73,26 @@ bool berd_siri_tts_speak( char **error_out ); +/// Opaque Pocket PCM player backed by AVAudioUnitTimePitch. Samples are +/// mono, noninterleaved float PCM. Device ID 0 uses the system default. +void *berd_pocket_audio_player_create( + uint32_t sample_rate, + float rate, + uint32_t output_device_id, + char **error_out +); +bool berd_pocket_audio_player_enqueue( + void *player, + const float *samples, + uint32_t frame_count, + char **error_out +); +uint64_t berd_pocket_audio_player_completed_source_frames(void *player); +uint64_t berd_pocket_audio_player_pending_buffers(void *player); +bool berd_pocket_audio_player_failed(void *player); +void berd_pocket_audio_player_stop(void *player); +void berd_pocket_audio_player_release(void *player); + /// Frees strings returned by this bridge. void berd_siri_tts_free_string(char *value); diff --git a/src-tauri/native/siri_tts_bridge.m b/src-tauri/native/siri_tts_bridge.m index f60035274..26c3632dd 100644 --- a/src-tauri/native/siri_tts_bridge.m +++ b/src-tauri/native/siri_tts_bridge.m @@ -567,6 +567,140 @@ - (void)cancel { } @end +@interface BerdPocketAudioPlayer : NSObject +@property(nonatomic, strong) AVAudioEngine *engine; +@property(nonatomic, strong) AVAudioPlayerNode *player; +@property(nonatomic, strong) AVAudioUnitTimePitch *timePitch; +@property(nonatomic, strong) AVAudioFormat *format; +@property(nonatomic, assign) uint64_t pendingBuffers; +@property(nonatomic, assign) uint64_t completedSourceFrames; +@property(nonatomic, assign) BOOL playbackFailed; +@property(nonatomic, assign) BOOL stopped; +- (instancetype)initWithSampleRate:(double)sampleRate + rate:(float)rate + outputDeviceID:(AudioDeviceID)outputDeviceID + error:(NSError **)error; +- (BOOL)enqueueSamples:(const float *)samples + frameCount:(AVAudioFrameCount)frameCount + error:(NSError **)error; +- (uint64_t)completedSourceFramesSnapshot; +- (void)stop; +@end + +@implementation BerdPocketAudioPlayer +- (instancetype)initWithSampleRate:(double)sampleRate + rate:(float)rate + outputDeviceID:(AudioDeviceID)outputDeviceID + error:(NSError **)error { + self = [super init]; + if (!self) return nil; + if (!(sampleRate > 0) || !isfinite(rate) || rate < 0.75f || rate > 2.0f) { + if (error) *error = BerdError(30, @"Pocket playback speed or sample rate is invalid."); + return nil; + } + + _engine = [AVAudioEngine new]; + _player = [AVAudioPlayerNode new]; + if (fabsf(rate - 1.0f) > 0.0001f) { + _timePitch = [AVAudioUnitTimePitch new]; + _timePitch.rate = rate; + _timePitch.pitch = 0.0f; + } + _format = [[AVAudioFormat alloc] initWithCommonFormat:AVAudioPCMFormatFloat32 + sampleRate:sampleRate + channels:1 + interleaved:NO]; + if (!_format) { + if (error) *error = BerdError(31, @"Could not create the Pocket PCM format."); + return nil; + } + + [_engine attachNode:_player]; + if (_timePitch) { + [_engine attachNode:_timePitch]; + [_engine connect:_player to:_timePitch format:_format]; + [_engine connect:_timePitch to:_engine.mainMixerNode format:_format]; + } else { + [_engine connect:_player to:_engine.mainMixerNode format:_format]; + } + + if (outputDeviceID != kAudioObjectUnknown) { + AudioUnit outputUnit = _engine.outputNode.audioUnit; + OSStatus status = AudioUnitSetProperty(outputUnit, + kAudioOutputUnitProperty_CurrentDevice, + kAudioUnitScope_Global, + 0, + &outputDeviceID, + sizeof(outputDeviceID)); + if (status != noErr) { + if (error) *error = BerdError(32, @"Could not select the configured audio output."); + return nil; + } + } + + [_engine prepare]; + if (![_engine startAndReturnError:error]) return nil; + return self; +} + +- (BOOL)enqueueSamples:(const float *)samples + frameCount:(AVAudioFrameCount)frameCount + error:(NSError **)error { + @synchronized (self) { + if (self.stopped) { + if (error) *error = BerdError(NSUserCancelledError, @"Pocket playback stopped."); + return NO; + } + } + if (!samples || frameCount == 0) return YES; + AVAudioPCMBuffer *buffer = [[AVAudioPCMBuffer alloc] + initWithPCMFormat:self.format frameCapacity:frameCount]; + if (!buffer || !buffer.floatChannelData[0]) { + if (error) *error = BerdError(34, @"Could not allocate a Pocket playback buffer."); + return NO; + } + buffer.frameLength = frameCount; + memcpy(buffer.floatChannelData[0], samples, sizeof(float) * frameCount); + @synchronized (self) { self.pendingBuffers += 1; } + __weak typeof(self) weakSelf = self; + [self.player scheduleBuffer:buffer + completionCallbackType:AVAudioPlayerNodeCompletionDataPlayedBack + completionHandler:^(__unused AVAudioPlayerNodeCompletionCallbackType type) { + BerdPocketAudioPlayer *strongSelf = weakSelf; + if (!strongSelf) return; + @synchronized (strongSelf) { + BOOL playing = strongSelf.engine.isRunning && strongSelf.player.isPlaying; + if (!strongSelf.stopped && playing) { + strongSelf.completedSourceFrames += frameCount; + } else if (!strongSelf.stopped) { + strongSelf.playbackFailed = YES; + } + strongSelf.pendingBuffers = strongSelf.pendingBuffers > 0 + ? strongSelf.pendingBuffers - 1 + : 0; + } + }]; + if (!self.player.isPlaying) [self.player play]; + return YES; +} + +- (uint64_t)completedSourceFramesSnapshot { + @synchronized (self) { return self.completedSourceFrames; } +} + +- (void)stop { + @synchronized (self) { + if (self.stopped) return; + self.stopped = YES; + self.pendingBuffers = 0; + } + [self.player stop]; + [self.engine stop]; +} + +- (void)dealloc { [self stop]; } +@end + static void BerdDownloadedVoices( NSString *language, NSString *voiceName, @@ -1145,6 +1279,77 @@ bool berd_siri_tts_speak( } } +void *berd_pocket_audio_player_create( + uint32_t sampleRate, + float rate, + uint32_t outputDeviceID, + char **errorOut +) { + @autoreleasepool { + if (errorOut) *errorOut = NULL; + NSError *error = nil; + BerdPocketAudioPlayer *player = [[BerdPocketAudioPlayer alloc] + initWithSampleRate:sampleRate + rate:rate + outputDeviceID:outputDeviceID + error:&error]; + if (!player) { + BerdSetError(errorOut, error ?: BerdError(35, @"Could not start Pocket playback.")); + return NULL; + } + return (__bridge_retained void *)player; + } +} + +bool berd_pocket_audio_player_enqueue( + void *playerValue, + const float *samples, + uint32_t frameCount, + char **errorOut +) { + @autoreleasepool { + if (errorOut) *errorOut = NULL; + if (!playerValue) { + BerdSetError(errorOut, BerdError(36, @"Pocket playback is unavailable.")); + return false; + } + NSError *error = nil; + BOOL enqueued = [(__bridge BerdPocketAudioPlayer *)playerValue + enqueueSamples:samples frameCount:frameCount error:&error]; + if (!enqueued) BerdSetError(errorOut, error ?: BerdError(37, @"Could not queue Pocket audio.")); + return enqueued; + } +} + +uint64_t berd_pocket_audio_player_completed_source_frames(void *playerValue) { + if (!playerValue) return 0; + return [(__bridge BerdPocketAudioPlayer *)playerValue completedSourceFramesSnapshot]; +} + +uint64_t berd_pocket_audio_player_pending_buffers(void *playerValue) { + if (!playerValue) return 0; + BerdPocketAudioPlayer *player = (__bridge BerdPocketAudioPlayer *)playerValue; + @synchronized (player) { return player.pendingBuffers; } +} + +bool berd_pocket_audio_player_failed(void *playerValue) { + if (!playerValue) return true; + BerdPocketAudioPlayer *player = (__bridge BerdPocketAudioPlayer *)playerValue; + @synchronized (player) { + return player.playbackFailed || (!player.stopped && !player.engine.isRunning); + } +} + +void berd_pocket_audio_player_stop(void *playerValue) { + if (!playerValue) return; + [(__bridge BerdPocketAudioPlayer *)playerValue stop]; +} + +void berd_pocket_audio_player_release(void *playerValue) { + if (!playerValue) return; + CFBridgingRelease(playerValue); +} + void berd_siri_tts_free_string(char *value) { free(value); } diff --git a/src-tauri/src/commands/mod.rs b/src-tauri/src/commands/mod.rs index 55584379c..07e46b266 100644 --- a/src-tauri/src/commands/mod.rs +++ b/src-tauri/src/commands/mod.rs @@ -39,7 +39,7 @@ pub mod notifications; pub mod openai_realtime; pub mod path_resolver; #[cfg(target_os = "macos")] -mod pocket_playback_speed_dsp; +mod pocket_audio_player; pub mod pocket_voice; pub mod project_icons; pub mod pull_requests; diff --git a/src-tauri/src/commands/pocket_audio_player.rs b/src-tauri/src/commands/pocket_audio_player.rs new file mode 100644 index 000000000..9c067c108 --- /dev/null +++ b/src-tauri/src/commands/pocket_audio_player.rs @@ -0,0 +1,176 @@ +//! Safe ownership wrapper for the macOS AVAudioUnitTimePitch Pocket player. + +use std::ffi::{c_char, c_void, CStr}; + +unsafe extern "C" { + fn berd_pocket_audio_player_create( + sample_rate: u32, + rate: f32, + output_device_id: u32, + error_out: *mut *mut c_char, + ) -> *mut c_void; + fn berd_pocket_audio_player_enqueue( + player: *mut c_void, + samples: *const f32, + frame_count: u32, + error_out: *mut *mut c_char, + ) -> bool; + fn berd_pocket_audio_player_completed_source_frames(player: *mut c_void) -> u64; + fn berd_pocket_audio_player_pending_buffers(player: *mut c_void) -> u64; + fn berd_pocket_audio_player_failed(player: *mut c_void) -> bool; + fn berd_pocket_audio_player_stop(player: *mut c_void); + fn berd_pocket_audio_player_release(player: *mut c_void); + fn berd_siri_tts_free_string(value: *mut c_char); +} + +pub(super) struct PocketAudioPlayer { + raw: *mut c_void, + delivery_safety_frames: u64, +} + +impl PocketAudioPlayer { + pub(super) fn new( + sample_rate: u32, + rate: f32, + output_device_name: Option<&str>, + ) -> Result { + let output_device_id = output_device_name + .map(|name| { + coreaudio::audio_unit::macos_helpers::get_device_id_from_name(name, false) + .ok_or_else(|| format!("audio output not found: {name}")) + }) + .transpose()? + .unwrap_or(0); + let mut error = std::ptr::null_mut(); + // SAFETY: The bridge receives a resolved CoreAudio device ID and + // returns an owned opaque player retained until `Drop`. + let raw = unsafe { + berd_pocket_audio_player_create(sample_rate, rate, output_device_id, &mut error) + }; + if raw.is_null() { + return Err(take_error(error, "Could not start native Pocket playback")); + } + Ok(Self { + raw, + delivery_safety_frames: delivery_safety_frames(sample_rate, rate), + }) + } + + pub(super) fn enqueue(&self, samples: &[f32]) -> Result<(), String> { + if samples.is_empty() { + return Ok(()); + } + let frame_count = u32::try_from(samples.len()) + .map_err(|_| "Pocket audio chunk is too large to queue".to_string())?; + let mut error = std::ptr::null_mut(); + // SAFETY: The bridge copies `frame_count` samples before returning and + // `self.raw` remains retained for this wrapper's lifetime. + let enqueued = unsafe { + berd_pocket_audio_player_enqueue(self.raw, samples.as_ptr(), frame_count, &mut error) + }; + if enqueued { + Ok(()) + } else { + Err(take_error(error, "Could not queue native Pocket audio")) + } + } + + pub(super) fn played_frames(&self) -> u64 { + // SAFETY: `self.raw` is a live retained player. The bridge counts only + // source buffers confirmed played back, so idle queue gaps add nothing. + apply_delivery_safety(self.completed_source_frames(), self.delivery_safety_frames) + } + + pub(super) fn completed_source_frames(&self) -> u64 { + // SAFETY: `self.raw` is a live retained player. + unsafe { berd_pocket_audio_player_completed_source_frames(self.raw) } + } + + pub(super) fn is_empty(&self) -> bool { + // SAFETY: `self.raw` is a live retained player. + unsafe { berd_pocket_audio_player_pending_buffers(self.raw) == 0 } + } + + pub(super) fn ensure_healthy(&self) -> Result<(), String> { + // SAFETY: `self.raw` is a live retained player. + playback_health(unsafe { berd_pocket_audio_player_failed(self.raw) }) + } + + pub(super) fn stop(&self) { + // SAFETY: `self.raw` is a live retained player and stop is idempotent. + unsafe { berd_pocket_audio_player_stop(self.raw) }; + } +} + +fn delivery_safety_frames(sample_rate: u32, rate: f32) -> u64 { + (f64::from(sample_rate) * 0.1 * f64::from(rate)).ceil() as u64 +} + +fn apply_delivery_safety(completed_source_frames: u64, safety_frames: u64) -> u64 { + completed_source_frames.saturating_sub(safety_frames) +} + +fn playback_health(failed: bool) -> Result<(), String> { + if failed { + Err("Pocket audio output stopped unexpectedly".to_string()) + } else { + Ok(()) + } +} + +impl Drop for PocketAudioPlayer { + fn drop(&mut self) { + // SAFETY: This wrapper uniquely owns the retained bridge reference. + unsafe { berd_pocket_audio_player_release(self.raw) }; + } +} + +fn take_error(error: *mut c_char, fallback: &str) -> String { + if error.is_null() { + return fallback.to_string(); + } + // SAFETY: Bridge errors are NUL-terminated malloc strings and are released + // through the paired bridge function after copying. + let message = unsafe { CStr::from_ptr(error) } + .to_string_lossy() + .into_owned(); + unsafe { berd_siri_tts_free_string(error) }; + message +} + +#[cfg(test)] +mod tests { + use super::{apply_delivery_safety, delivery_safety_frames, playback_health}; + + #[test] + fn delivery_safety_tracks_playback_rate_in_source_frames() { + assert_eq!(delivery_safety_frames(24_000, 0.75), 1_800); + assert_eq!(delivery_safety_frames(24_000, 1.0), 2_400); + assert_eq!(delivery_safety_frames(24_000, 2.0), 4_800); + } + + #[test] + fn silent_queue_gaps_do_not_advance_delivery() { + let safety = delivery_safety_frames(24_000, 1.0); + let first_buffer_completed = 4_800; + assert_eq!(apply_delivery_safety(first_buffer_completed, safety), 2_400); + + let after_silent_gap = first_buffer_completed; + assert_eq!(apply_delivery_safety(after_silent_gap, safety), 2_400); + + let second_buffer_completed = 9_600; + assert_eq!( + apply_delivery_safety(second_buffer_completed, safety), + 7_200 + ); + } + + #[test] + fn unexpected_output_stops_fail_playback() { + assert!(playback_health(false).is_ok()); + assert_eq!( + playback_health(true).expect_err("unexpected stop must fail"), + "Pocket audio output stopped unexpectedly" + ); + } +} diff --git a/src-tauri/src/commands/pocket_playback_speed_dsp.rs b/src-tauri/src/commands/pocket_playback_speed_dsp.rs deleted file mode 100644 index 25cfb90af..000000000 --- a/src-tauri/src/commands/pocket_playback_speed_dsp.rs +++ /dev/null @@ -1,309 +0,0 @@ -//! Stateful pitch-preserving processing for streamed Pocket TTS PCM. -//! -//! Pocket emits cumulative PCM after each decoder block. A single stretcher -//! must span those blocks so its analysis window and latency are paid once per -//! synthesis attempt instead of once per callback. - -const MIN_PLAYBACK_SPEED: f32 = 0.75; -const MAX_PLAYBACK_SPEED: f32 = 2.0; -const DEFAULT_PLAYBACK_SPEED: f32 = 1.0; -const UNITY_EPSILON: f32 = 0.000_1; - -pub(super) struct StreamingSpeedProcessor { - speed: f32, - sample_rate: u32, - stretch: Option, - input_latency: usize, - output_latency: usize, - reset_pre_roll: usize, - trim_remaining: usize, - total_input: usize, - requested_output: usize, - emitted_output: usize, -} - -impl StreamingSpeedProcessor { - pub(super) fn new(speed: f32, sample_rate: u32) -> Result { - validate_speed(speed)?; - if (speed - DEFAULT_PLAYBACK_SPEED).abs() <= UNITY_EPSILON { - return Ok(Self { - speed, - sample_rate, - stretch: None, - input_latency: 0, - output_latency: 0, - reset_pre_roll: 0, - trim_remaining: 0, - total_input: 0, - requested_output: 0, - emitted_output: 0, - }); - } - - let mut stretch = ssstretch::Stretch::new(); - stretch.preset_default(1, sample_rate as f32); - let input_latency = stretch.input_latency().max(0) as usize; - let output_latency = stretch.output_latency().max(0) as usize; - let reset_pre_roll = (input_latency as f64 / speed as f64).ceil() as usize; - - Ok(Self { - speed, - sample_rate, - stretch: Some(stretch), - input_latency, - output_latency, - reset_pre_roll, - trim_remaining: reset_pre_roll.saturating_add(output_latency), - total_input: 0, - requested_output: 0, - emitted_output: 0, - }) - } - - pub(super) fn process(&mut self, input: &[f32]) -> Result, String> { - if input.is_empty() { - return Ok(Vec::new()); - } - self.total_input = self - .total_input - .checked_add(input.len()) - .ok_or_else(|| "Pocket streamed audio is too large".to_string())?; - if self.stretch.is_none() { - self.requested_output = self.total_input; - self.emitted_output = self.total_input; - return Ok(input.to_vec()); - } - - let target_output = stretched_len(self.total_input, self.speed); - let output_samples = target_output.saturating_sub(self.requested_output); - self.requested_output = target_output; - let inputs = [input.to_vec()]; - let mut output = [Vec::with_capacity(output_samples)]; - self.stretch - .as_mut() - .expect("non-unity processor has a stretcher") - .process_vec( - &inputs, - i32_len(input.len())?, - &mut output, - i32_len(output_samples)?, - ); - Ok(self.trim_and_count(output[0].as_slice())) - } - - pub(super) fn expected_output_frames(&self) -> usize { - stretched_len(self.total_input, self.speed) - } - - pub(super) fn finish(&mut self) -> Result, String> { - if self.stretch.is_none() { - return Ok(Vec::new()); - } - - let latency_input = [vec![0.0; self.input_latency]]; - let mut latency_output = [Vec::with_capacity(self.reset_pre_roll)]; - self.stretch - .as_mut() - .expect("non-unity processor has a stretcher") - .process_vec( - &latency_input, - i32_len(self.input_latency)?, - &mut latency_output, - i32_len(self.reset_pre_roll)?, - ); - let mut output = self.trim_and_count(latency_output[0].as_slice()); - - let mut flushed = [Vec::with_capacity(self.output_latency)]; - self.stretch - .as_mut() - .expect("non-unity processor has a stretcher") - .flush_vec(&mut flushed, i32_len(self.output_latency)?); - output.extend(self.trim_and_count(flushed[0].as_slice())); - - let expected = stretched_len(self.total_input, self.speed); - if self.trim_remaining != 0 || self.emitted_output != expected { - return Err(format!( - "time stretcher emitted {} samples with {} latency samples remaining, expected {expected}", - self.emitted_output, self.trim_remaining - )); - } - Ok(output) - } - - pub(super) fn drain_and_reset(&mut self) -> Result, String> { - let output = self.finish()?; - *self = Self::new(self.speed, self.sample_rate)?; - Ok(output) - } - - fn trim_and_count(&mut self, samples: &[f32]) -> Vec { - let trim = self.trim_remaining.min(samples.len()); - self.trim_remaining -= trim; - let output = samples[trim..].to_vec(); - self.emitted_output = self.emitted_output.saturating_add(output.len()); - output - } -} - -fn stretched_len(input_len: usize, speed: f32) -> usize { - (input_len as f64 / speed as f64).round() as usize -} - -fn validate_speed(speed: f32) -> Result<(), String> { - if speed.is_finite() && (MIN_PLAYBACK_SPEED..=MAX_PLAYBACK_SPEED).contains(&speed) { - Ok(()) - } else { - Err(format!( - "Pocket playback speed must be between {MIN_PLAYBACK_SPEED} and {MAX_PLAYBACK_SPEED}" - )) - } -} - -fn i32_len(length: usize) -> Result { - i32::try_from(length).map_err(|_| "audio chunk is too large to process".to_string()) -} - -#[cfg(test)] -mod tests { - use super::*; - - const SAMPLE_RATE: u32 = 24_000; - - #[test] - fn unity_streaming_is_a_bit_exact_bypass() { - let first = vec![0.0, 0.125]; - let second = vec![-0.5, 1.0]; - let mut processor = StreamingSpeedProcessor::new(1.0, SAMPLE_RATE).expect("processor"); - - assert_eq!(processor.process(&first).expect("first"), first); - assert_eq!(processor.process(&second).expect("second"), second); - assert!(processor.finish().expect("finish").is_empty()); - } - - #[test] - fn streaming_blocks_preserve_length_pitch_order_and_continuity() { - let input: Vec = (0..48_000) - .map(|sample| { - let frequency = if sample < 16_000 { - 220.0 - } else if sample < 32_000 { - 440.0 - } else { - 660.0 - }; - (2.0 * std::f32::consts::PI * frequency * sample as f32 / SAMPLE_RATE as f32).sin() - }) - .collect(); - let mut processor = StreamingSpeedProcessor::new(1.25, SAMPLE_RATE).expect("processor"); - let mut output = Vec::new(); - let mut callback_boundaries = Vec::new(); - for block in input.chunks(1_920) { - let processed = processor.process(block).expect("stream block"); - if !processed.is_empty() && !output.is_empty() { - callback_boundaries.push(output.len()); - } - output.extend(processed); - } - output.extend(processor.finish().expect("finish")); - - assert_eq!(output.len(), 38_400); - assert_frequency(&output[2_000..10_000], 220.0, 8.0); - assert_frequency(&output[14_000..23_000], 440.0, 10.0); - assert_frequency(&output[27_000..36_000], 660.0, 12.0); - for boundary in callback_boundaries { - let jump = (output[boundary] - output[boundary - 1]).abs(); - assert!(jump < 0.5, "callback boundary jump was {jump}"); - } - } - - #[test] - fn two_x_streaming_preserves_pitch_and_tail() { - let frequency = 220.0_f32; - let input: Vec = (0..SAMPLE_RATE * 2) - .map(|sample| { - (2.0 * std::f32::consts::PI * frequency * sample as f32 / SAMPLE_RATE as f32).sin() - }) - .collect(); - let mut processor = StreamingSpeedProcessor::new(2.0, SAMPLE_RATE).expect("processor"); - let mut output = Vec::new(); - for block in input.chunks(1_920) { - output.extend(processor.process(block).expect("stream block")); - } - output.extend(processor.finish().expect("finish")); - - assert_eq!(output.len(), SAMPLE_RATE as usize); - assert!( - root_mean_square(&output[..480]) > 0.2, - "initial latency was not removed" - ); - assert!( - root_mean_square(&output[output.len() - 480..]) > 0.2, - "speech tail was truncated" - ); - assert_frequency(&output[2_000..], frequency, 4.0); - } - - #[test] - fn two_x_boundary_drain_emits_complete_stretched_length() { - let input: Vec = (0..SAMPLE_RATE) - .map(|sample| { - (2.0 * std::f32::consts::PI * 220.0 * sample as f32 / SAMPLE_RATE as f32).sin() - }) - .collect(); - let mut processor = StreamingSpeedProcessor::new(2.0, SAMPLE_RATE).expect("processor"); - let mut output = Vec::new(); - for block in input.chunks(1_920) { - output.extend(processor.process(block).expect("stream block")); - } - - let tail = processor.drain_and_reset().expect("boundary drain"); - assert!(!tail.is_empty(), "boundary drain did not emit a tail"); - output.extend(tail); - - assert_eq!(output.len(), stretched_len(input.len(), 2.0)); - } - - #[test] - fn two_x_processing_continues_after_boundary_reset() { - let first = vec![0.25; SAMPLE_RATE as usize]; - let second: Vec = (0..SAMPLE_RATE) - .map(|sample| { - (2.0 * std::f32::consts::PI * 440.0 * sample as f32 / SAMPLE_RATE as f32).sin() - }) - .collect(); - let mut processor = StreamingSpeedProcessor::new(2.0, SAMPLE_RATE).expect("processor"); - for block in first.chunks(1_920) { - processor.process(block).expect("first stream block"); - } - processor.drain_and_reset().expect("boundary drain"); - - let mut output = Vec::new(); - for block in second.chunks(1_920) { - output.extend(processor.process(block).expect("second stream block")); - } - output.extend(processor.finish().expect("finish second segment")); - - assert_eq!(output.len(), stretched_len(second.len(), 2.0)); - assert_frequency(&output[2_000..], 440.0, 8.0); - } - - fn assert_frequency(samples: &[f32], expected: f32, tolerance: f32) { - let measured = zero_crossing_frequency(samples); - assert!( - (measured - expected).abs() <= tolerance, - "expected {expected} Hz, measured {measured} Hz" - ); - } - - fn zero_crossing_frequency(samples: &[f32]) -> f32 { - let crossings = samples - .windows(2) - .filter(|pair| pair[0] <= 0.0 && pair[1] > 0.0) - .count(); - crossings as f32 * SAMPLE_RATE as f32 / samples.len() as f32 - } - - fn root_mean_square(samples: &[f32]) -> f32 { - (samples.iter().map(|sample| sample * sample).sum::() / samples.len() as f32).sqrt() - } -} diff --git a/src-tauri/src/commands/pocket_voice.rs b/src-tauri/src/commands/pocket_voice.rs index b32b07341..cfc652b79 100644 --- a/src-tauri/src/commands/pocket_voice.rs +++ b/src-tauri/src/commands/pocket_voice.rs @@ -8,13 +8,11 @@ use std::io::Write; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, Ordering}; #[cfg(target_os = "macos")] -use std::sync::atomic::{AtomicU64, AtomicUsize}; -#[cfg(target_os = "macos")] use std::sync::mpsc; use std::sync::{Arc, Mutex, OnceLock}; use std::time::{Duration, Instant, SystemTime}; -#[cfg(any(test, target_os = "macos"))] +#[cfg(target_os = "macos")] use berd_voice::SAMPLE_RATE; #[cfg(target_os = "macos")] use berd_voice::{load_text_to_speech, load_voice_style, PocketTts, VoiceStyle}; @@ -27,9 +25,7 @@ use objc2_core_audio::{ AudioObjectGetPropertyDataSize, AudioObjectID, AudioObjectPropertyAddress, }; #[cfg(target_os = "macos")] -use rodio::buffer::SamplesBuffer; -#[cfg(target_os = "macos")] -use rodio::{ChannelCount, DeviceTrait, Player, SampleRate, Source}; +use rodio::DeviceTrait; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use tauri::{AppHandle, Emitter, Manager, State}; @@ -37,7 +33,7 @@ use tauri::{AppHandle, Emitter, Manager, State}; #[cfg(target_os = "macos")] use super::native_voice::AssistantSpeechGuard; #[cfg(target_os = "macos")] -use super::pocket_playback_speed_dsp::StreamingSpeedProcessor; +use super::pocket_audio_player::PocketAudioPlayer; use super::{ native_voice::{InterruptionSensitivity, NativeVoiceState}, voice_capture::VoiceCaptureState, @@ -66,192 +62,9 @@ const BLUETOOTH_PLAYBACK_LATENCY_SAFETY_DURATION: Duration = Duration::from_mill const AIRPLAY_PLAYBACK_LATENCY_SAFETY_DURATION: Duration = Duration::from_secs(2); #[cfg(target_os = "macos")] const UNKNOWN_PLAYBACK_LATENCY_SAFETY_DURATION: Duration = Duration::from_secs(2); -#[cfg(target_os = "macos")] +#[cfg(any(test, target_os = "macos"))] const POCKET_SOURCE_COMPLETION_TIMEOUT: Duration = Duration::from_secs(2); -#[cfg(target_os = "macos")] -struct PocketPlaybackSource { - input: I, - generation: u64, - pending_audio_sources: Arc, - completed_generation: Arc, - completion_sender: mpsc::SyncSender, - completion_sent: bool, -} - -#[cfg(target_os = "macos")] -enum PocketPlaybackEvent { - SourceFinished, - ShutdownImmediately, - ShutdownAfterPlayback, - #[cfg(test)] - Probe(mpsc::Sender<()>), -} - -#[cfg(target_os = "macos")] -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -enum PocketPlaybackMonitorOutcome { - Completed, - SourceCompletionTimedOut, -} - -#[cfg(target_os = "macos")] -fn apply_pocket_playback_monitor_outcome( - result: Result, - outcome: PocketPlaybackMonitorOutcome, -) -> Result { - if outcome == PocketPlaybackMonitorOutcome::SourceCompletionTimedOut { - log::warn!( - "Pocket playback source completion bookkeeping timed out after the player drained" - ); - } - result -} - -#[cfg(target_os = "macos")] -fn run_pocket_playback_monitor( - playback_completion_receiver: mpsc::Receiver, - assistant_speech: Arc>>, - completed_generation: Arc, - source_completion_timeout: Duration, - playback_latency_safety_duration: Duration, -) -> PocketPlaybackMonitorOutcome { - let mut shutdown_after_playback = false; - while let Ok(event) = playback_completion_receiver.recv() { - match event { - event @ (PocketPlaybackEvent::SourceFinished - | PocketPlaybackEvent::ShutdownAfterPlayback) => { - if matches!(event, PocketPlaybackEvent::ShutdownAfterPlayback) { - let target_generation = assistant_speech - .lock() - .expect("assistant speech") - .as_ref() - .map(|(generation, _)| *generation); - let Some(target_generation) = target_generation else { - return PocketPlaybackMonitorOutcome::Completed; - }; - let source_completion_deadline = Instant::now() + source_completion_timeout; - while completed_generation.load(Ordering::Acquire) != target_generation { - let source_wait_remaining = - source_completion_deadline.saturating_duration_since(Instant::now()); - if source_wait_remaining.is_zero() { - assistant_speech.lock().expect("assistant speech").take(); - return PocketPlaybackMonitorOutcome::SourceCompletionTimedOut; - } - match playback_completion_receiver - .recv_timeout(source_wait_remaining.min(Duration::from_millis(10))) - { - Ok(PocketPlaybackEvent::SourceFinished) - | Ok(PocketPlaybackEvent::ShutdownAfterPlayback) - | Err(mpsc::RecvTimeoutError::Timeout) => {} - #[cfg(test)] - Ok(PocketPlaybackEvent::Probe(sender)) => { - let _ = sender.send(()); - } - Ok(PocketPlaybackEvent::ShutdownImmediately) - | Err(mpsc::RecvTimeoutError::Disconnected) => { - return PocketPlaybackMonitorOutcome::Completed; - } - } - } - } - shutdown_after_playback |= - matches!(event, PocketPlaybackEvent::ShutdownAfterPlayback); - let mut generation = completed_generation.load(Ordering::Acquire); - let mut grace_started = Instant::now(); - loop { - let grace_remaining = - playback_latency_safety_duration.saturating_sub(grace_started.elapsed()); - match playback_completion_receiver.recv_timeout(grace_remaining) { - Ok(PocketPlaybackEvent::SourceFinished) => { - generation = completed_generation.load(Ordering::Acquire); - grace_started = Instant::now(); - } - #[cfg(test)] - Ok(PocketPlaybackEvent::Probe(sender)) => { - let _ = sender.send(()); - } - Ok(PocketPlaybackEvent::ShutdownImmediately) - | Err(mpsc::RecvTimeoutError::Disconnected) => { - return PocketPlaybackMonitorOutcome::Completed; - } - Ok(PocketPlaybackEvent::ShutdownAfterPlayback) => { - shutdown_after_playback = true; - } - Err(mpsc::RecvTimeoutError::Timeout) => { - let latest_generation = completed_generation.load(Ordering::Acquire); - if latest_generation != generation { - generation = latest_generation; - grace_started = Instant::now(); - continue; - } - release_completed_pocket_assistant_speech( - generation, - &assistant_speech, - ); - let latest_generation = completed_generation.load(Ordering::Acquire); - if latest_generation != generation { - generation = latest_generation; - grace_started = Instant::now(); - continue; - } - if shutdown_after_playback { - return PocketPlaybackMonitorOutcome::Completed; - } - break; - } - } - } - } - #[cfg(test)] - PocketPlaybackEvent::Probe(sender) => { - let _ = sender.send(()); - } - PocketPlaybackEvent::ShutdownImmediately => break, - } - } - PocketPlaybackMonitorOutcome::Completed -} - -#[cfg(target_os = "macos")] -fn spawn_pocket_playback_monitor_with( - task: impl FnOnce() -> PocketPlaybackMonitorOutcome + Send + 'static, - spawn: F, -) -> std::io::Result> -where - F: FnOnce( - Box PocketPlaybackMonitorOutcome + Send>, - ) -> std::io::Result>, -{ - spawn(Box::new(task)) -} - -#[cfg(target_os = "macos")] -fn spawn_pocket_playback_monitor( - playback_completion_receiver: mpsc::Receiver, - assistant_speech: Arc>>, - completed_generation: Arc, - source_completion_timeout: Duration, - playback_latency_safety_duration: Duration, -) -> std::io::Result> { - spawn_pocket_playback_monitor_with( - move || { - run_pocket_playback_monitor( - playback_completion_receiver, - assistant_speech, - completed_generation, - source_completion_timeout, - playback_latency_safety_duration, - ) - }, - |task| { - std::thread::Builder::new() - .name("pocket-playback-monitor".to_string()) - .spawn(task) - }, - ) -} - #[cfg(target_os = "macos")] fn playback_latency_safety_duration_for_transport(transport: Option) -> Duration { // CoreAudio transport FOURCC values. Bluetooth and AirPlay routes buffer @@ -285,86 +98,6 @@ pub(crate) fn playback_latency_safety_duration(output_device: Option<&str>) -> D device_id.and_then(|id| get_device_transport_type(id).ok()), ) } - -#[cfg(target_os = "macos")] -impl PocketPlaybackSource { - fn new( - input: I, - generation: u64, - pending_audio_sources: Arc, - completed_generation: Arc, - completion_sender: mpsc::SyncSender, - ) -> Self { - Self { - input, - generation, - pending_audio_sources, - completed_generation, - completion_sender, - completion_sent: false, - } - } - - fn finish(&mut self) { - if std::mem::replace(&mut self.completion_sent, true) { - return; - } - let previous = self.pending_audio_sources.fetch_sub(1, Ordering::AcqRel); - debug_assert!(previous > 0, "Pocket playback source count underflow"); - if previous == 1 { - self.completed_generation - .store(self.generation, Ordering::Release); - let _ = self - .completion_sender - .try_send(PocketPlaybackEvent::SourceFinished); - } - } -} - -#[cfg(target_os = "macos")] -impl Iterator for PocketPlaybackSource { - type Item = I::Item; - - fn next(&mut self) -> Option { - let next = self.input.next(); - if next.is_none() { - self.finish(); - } - next - } - - fn size_hint(&self) -> (usize, Option) { - self.input.size_hint() - } -} - -#[cfg(target_os = "macos")] -impl Drop for PocketPlaybackSource { - fn drop(&mut self) { - self.finish(); - } -} - -#[cfg(target_os = "macos")] -impl Source for PocketPlaybackSource { - fn current_span_len(&self) -> Option { - self.input.current_span_len() - } - - fn channels(&self) -> ChannelCount { - self.input.channels() - } - - fn sample_rate(&self) -> SampleRate { - self.input.sample_rate() - } - - fn total_duration(&self) -> Option { - self.input.total_duration() - } -} -#[cfg(any(test, target_os = "macos"))] -const PLAYBACK_LATENCY_SAFETY_FRAMES: u64 = SAMPLE_RATE as u64 / 10; const PARAKEET_ARCHIVE: Artifact = Artifact { filename: "parakeet.tar.bz2", size: 104_337_827, @@ -519,7 +252,6 @@ struct VoiceDeliveryProgress { #[derive(Debug, Default)] struct PlaybackDeliveryLedger { segments: Vec<(String, u64, bool)>, - pieces: Vec, } #[cfg(any(test, target_os = "macos"))] @@ -537,7 +269,6 @@ impl PlaybackDeliveryLedger { if !*synthesis_complete { *total = total.saturating_add(frames); } - self.pieces.push(frames); } } @@ -548,18 +279,14 @@ impl PlaybackDeliveryLedger { } } - fn snapshot(&self, queued_pieces: usize, current_piece_frames: u64) -> VoiceDeliveryProgress { - let completed_pieces = self.pieces.len().saturating_sub(queued_pieces); - let completed_frames = self - .pieces + fn total_frames(&self) -> u64 { + self.segments .iter() - .take(completed_pieces) - .copied() - .sum::(); - let current_total = self.pieces.get(completed_pieces).copied().unwrap_or(0); - let consumed_frames = completed_frames - .saturating_add(current_piece_frames.min(current_total)) - .saturating_sub(PLAYBACK_LATENCY_SAFETY_FRAMES); + .map(|(_, total_frames, _)| *total_frames) + .sum() + } + + fn snapshot_consumed_frames(&self, consumed_frames: u64) -> VoiceDeliveryProgress { let mut segment_start = 0_u64; let segments = self .segments @@ -2439,10 +2166,6 @@ fn run_pocket_voice_stream( interruption_sensitivity: InterruptionSensitivity, suppress_capture: bool, ) -> Result { - use std::num::NonZero; - - use rodio::cpal::traits::HostTrait; - let version = base.join(CACHE_VERSION); let engine = load_text_to_speech( version @@ -2450,57 +2173,24 @@ fn run_pocket_voice_stream( .ok_or_else(|| "Pocket model path is not valid UTF-8".to_string())?, )?; let style = load_voice_style(&version.join("voices").join(voice.filename))?; - let sink = if let Some(name) = output_device { - let host = rodio::cpal::default_host(); - let mut matching = None; - for device in host - .output_devices() - .map_err(|error| format!("enumerate audio outputs: {error}"))? - { - if device - .description() - .ok() - .is_some_and(|description| description.name() == name) - { - matching = Some(device); - break; - } - } - let device = matching.ok_or_else(|| format!("audio output not found: {name}"))?; - rodio::DeviceSinkBuilder::from_device(device) - .map_err(|error| format!("configure audio output {name}: {error}"))? - .open_stream() - .map_err(|error| format!("open audio output {name}: {error}"))? - } else { - rodio::DeviceSinkBuilder::open_default_sink() - .map_err(|error| format!("open default audio output: {error}"))? - }; - let channels = - NonZero::new(1_u16).ok_or_else(|| "Pocket channel count invariant failed".to_string())?; - let rate = NonZero::new(SAMPLE_RATE) - .ok_or_else(|| "Pocket sample rate invariant failed".to_string())?; - let player = Player::connect_new(sink.mixer()); - let mut speed_processor = StreamingSpeedProcessor::new(speed, SAMPLE_RATE)?; + let player = PocketAudioPlayer::new(SAMPLE_RATE, speed, output_device)?; let mut pending = String::new(); let mut first_chunk_pending = true; let mut playback_started = false; - let assistant_speech = Arc::new(Mutex::new(None::<(u64, AssistantSpeechGuard)>)); - let pending_audio_sources = Arc::new(AtomicUsize::new(0)); - let completed_generation = Arc::new(AtomicU64::new(0)); - let (playback_completion_sender, playback_completion_receiver) = mpsc::sync_channel(1); - let mut playback_generation = 0_u64; - let playback_monitor = spawn_pocket_playback_monitor( - playback_completion_receiver, - Arc::clone(&assistant_speech), - Arc::clone(&completed_generation), - POCKET_SOURCE_COMPLETION_TIMEOUT, - playback_latency_safety_duration(output_device), - ) - .map_err(|error| format!("start Pocket playback monitor: {error}"))?; + let mut assistant_speech = None::; + let mut playback_drained_at = None; + let output_latency_grace = playback_latency_safety_duration(output_device); let mut delivery_ledger = PlaybackDeliveryLedger::default(); let mut last_progress_emit = Instant::now(); let result: Result = (|| loop { + update_pocket_assistant_speech( + player.is_empty(), + &mut assistant_speech, + &mut playback_drained_at, + output_latency_grace, + Instant::now(), + ); if !active.load(Ordering::SeqCst) { let delivery = pocket_delivery_snapshot(&delivery_ledger, &player); player.stop(); @@ -2509,6 +2199,7 @@ fn run_pocket_voice_stream( delivery: Some(delivery), }); } + player.ensure_healthy()?; let command = receiver.recv_timeout(Duration::from_millis(20)); match command { Ok(PocketStreamCommand::Append(text)) => { @@ -2520,20 +2211,14 @@ fn run_pocket_voice_stream( &style, &active, &player, - channels, - rate, - &mut speed_processor, &mut pending, &mut first_chunk_pending, &mut playback_started, &native_voice, interruption_sensitivity, suppress_capture, - &assistant_speech, - &pending_audio_sources, - &completed_generation, - &playback_completion_sender, - &mut playback_generation, + &mut assistant_speech, + &mut playback_drained_at, &mut delivery_ledger, &mut last_progress_emit, false, @@ -2556,20 +2241,14 @@ fn run_pocket_voice_stream( &style, &active, &player, - channels, - rate, - &mut speed_processor, &mut pending, &mut first_chunk_pending, &mut playback_started, &native_voice, interruption_sensitivity, suppress_capture, - &assistant_speech, - &pending_audio_sources, - &completed_generation, - &playback_completion_sender, - &mut playback_generation, + &mut assistant_speech, + &mut playback_drained_at, &mut delivery_ledger, &mut last_progress_emit, true, @@ -2583,28 +2262,6 @@ fn run_pocket_voice_stream( delivery: Some(delivery), }); } - let tail = speed_processor.drain_and_reset()?; - if !tail.is_empty() { - let tail_len = tail.len(); - append_pocket_samples( - app, - stream_id, - &player, - channels, - rate, - tail, - &native_voice, - interruption_sensitivity, - suppress_capture, - &mut playback_started, - &assistant_speech, - &pending_audio_sources, - &completed_generation, - &playback_completion_sender, - &mut playback_generation, - )?; - delivery_ledger.append_frames(tail_len); - } } Ok(PocketStreamCommand::Finish) => { if !synthesize_pocket_stream_ready( @@ -2614,20 +2271,14 @@ fn run_pocket_voice_stream( &style, &active, &player, - channels, - rate, - &mut speed_processor, &mut pending, &mut first_chunk_pending, &mut playback_started, &native_voice, interruption_sensitivity, suppress_capture, - &assistant_speech, - &pending_audio_sources, - &completed_generation, - &playback_completion_sender, - &mut playback_generation, + &mut assistant_speech, + &mut playback_drained_at, &mut delivery_ledger, &mut last_progress_emit, true, @@ -2641,29 +2292,14 @@ fn run_pocket_voice_stream( delivery: Some(delivery), }); } - let tail = speed_processor.finish()?; - if !tail.is_empty() { - let tail_len = tail.len(); - append_pocket_samples( - app, - stream_id, - &player, - channels, - rate, - tail, - &native_voice, - interruption_sensitivity, - suppress_capture, - &mut playback_started, - &assistant_speech, - &pending_audio_sources, - &completed_generation, - &playback_completion_sender, - &mut playback_generation, - )?; - delivery_ledger.append_frames(tail_len); - } - while !player.empty() { + let drain_timeout = pocket_native_drain_timeout( + delivery_ledger.total_frames(), + player.completed_source_frames(), + speed, + ); + let drain_started = Instant::now(); + let mut completion_timed_out = false; + loop { if !active.load(Ordering::SeqCst) { let delivery = pocket_delivery_snapshot(&delivery_ledger, &player); player.stop(); @@ -2672,6 +2308,38 @@ fn run_pocket_voice_stream( delivery: Some(delivery), }); } + if !completion_timed_out { + player.ensure_healthy()?; + match pocket_native_drain_status( + player.is_empty(), + drain_started.elapsed(), + drain_timeout, + ) { + PocketNativeDrainStatus::Waiting => {} + PocketNativeDrainStatus::Drained => { + player.ensure_healthy()?; + } + PocketNativeDrainStatus::TimedOut => { + log::warn!("Pocket native buffer completion bookkeeping timed out"); + player.stop(); + reset_pocket_drain_grace(&mut playback_drained_at); + completion_timed_out = true; + } + } + } + let playback_drained = completion_timed_out || player.is_empty(); + if playback_drained { + update_pocket_assistant_speech( + playback_drained, + &mut assistant_speech, + &mut playback_drained_at, + output_latency_grace, + Instant::now(), + ); + if assistant_speech.is_none() { + break; + } + } std::thread::sleep(Duration::from_millis(10)); } return Ok(PocketStreamOutcome { @@ -2705,55 +2373,22 @@ fn run_pocket_voice_stream( } })(); - let wait_for_output_latency = matches!( - &result, - Ok(PocketStreamOutcome { - state: PocketStreamEventState::Completed, - .. - }) - ); - let result = result.map_err(|error| { + assistant_speech.take(); + result.map_err(|error| { let delivery = delivery_with_played_audio(capture_before_stop( || pocket_delivery_snapshot(&delivery_ledger, &player), || player.stop(), )); PocketStreamFailure { error, delivery } - }); - - let shutdown = if wait_for_output_latency { - PocketPlaybackEvent::ShutdownAfterPlayback - } else { - PocketPlaybackEvent::ShutdownImmediately - }; - let _ = playback_completion_sender.send(shutdown); - let playback_monitor_outcome = playback_monitor.join(); - if let Ok(mut assistant_speech) = assistant_speech.lock() { - assistant_speech.take(); - } - match playback_monitor_outcome { - Ok(outcome) => apply_pocket_playback_monitor_outcome(result, outcome), - Err(_) => Err(PocketStreamFailure { - error: "Pocket playback monitor panicked".to_string(), - delivery: delivery_with_played_audio(pocket_delivery_snapshot( - &delivery_ledger, - &player, - )), - }), - } + }) } #[cfg(target_os = "macos")] fn pocket_delivery_snapshot( ledger: &PlaybackDeliveryLedger, - player: &Player, + player: &PocketAudioPlayer, ) -> VoiceDeliveryProgress { - // Read the queue depth first. If the player advances to the next source - // before get_pos(), pairing the newer (smaller) position with the older - // (larger) queue depth can only undercount delivery. - let queued_pieces = player.len(); - let current_piece_frames = - (player.get_pos().as_secs_f64() * f64::from(SAMPLE_RATE)).round() as u64; - ledger.snapshot(queued_pieces, current_piece_frames) + ledger.snapshot_consumed_frames(player.played_frames()) } #[cfg(any(test, target_os = "macos"))] @@ -2767,18 +2402,79 @@ fn capture_before_stop( } #[cfg(target_os = "macos")] -fn release_completed_pocket_assistant_speech( - completed_generation: u64, - assistant_speech: &Mutex>, +fn update_pocket_assistant_speech( + playback_drained: bool, + assistant_speech: &mut Option, + playback_drained_at: &mut Option, + output_latency_grace: Duration, + now: Instant, ) { - if let Ok(mut assistant_speech) = assistant_speech.lock() { - if assistant_speech - .as_ref() - .is_some_and(|(generation, _)| *generation == completed_generation) - { - assistant_speech.take(); - } + if pocket_assistant_speech_grace_elapsed( + playback_drained, + assistant_speech.is_some(), + playback_drained_at, + output_latency_grace, + now, + ) { + assistant_speech.take(); + } +} + +#[cfg(any(test, target_os = "macos"))] +fn pocket_assistant_speech_grace_elapsed( + playback_drained: bool, + guard_active: bool, + playback_drained_at: &mut Option, + output_latency_grace: Duration, + now: Instant, +) -> bool { + if !guard_active || !playback_drained { + *playback_drained_at = None; + return false; } + let drained_at = *playback_drained_at.get_or_insert(now); + now.saturating_duration_since(drained_at) >= output_latency_grace +} + +#[cfg(any(test, target_os = "macos"))] +fn reset_pocket_drain_grace(playback_drained_at: &mut Option) { + *playback_drained_at = None; +} + +#[cfg(any(test, target_os = "macos"))] +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum PocketNativeDrainStatus { + Waiting, + Drained, + TimedOut, +} + +#[cfg(any(test, target_os = "macos"))] +fn pocket_native_drain_status( + playback_drained: bool, + elapsed: Duration, + timeout: Duration, +) -> PocketNativeDrainStatus { + if playback_drained { + PocketNativeDrainStatus::Drained + } else if elapsed >= timeout { + PocketNativeDrainStatus::TimedOut + } else { + PocketNativeDrainStatus::Waiting + } +} + +#[cfg(any(test, target_os = "macos"))] +fn pocket_native_drain_timeout( + total_source_frames: u64, + completed_source_frames: u64, + rate: f32, +) -> Duration { + let remaining_source_frames = total_source_frames.saturating_sub(completed_source_frames); + let remaining_playback_seconds = + remaining_source_frames as f64 / f64::from(berd_voice::SAMPLE_RATE) / f64::from(rate); + Duration::from_secs_f64(remaining_playback_seconds) + .saturating_add(POCKET_SOURCE_COMPLETION_TIMEOUT) } #[cfg(target_os = "macos")] @@ -2790,22 +2486,12 @@ fn mark_pocket_playback_started( interruption_sensitivity: InterruptionSensitivity, suppress_capture: bool, playback_started: &mut bool, - generation: u64, - assistant_speech: &Mutex>, + assistant_speech: &mut Option, ) -> Result<(), String> { - let mut assistant_speech = assistant_speech - .lock() - .map_err(|_| "Pocket assistant speech guard lock was poisoned".to_string())?; - if assistant_speech - .as_ref() - .is_some_and(|(active_generation, _)| *active_generation == generation) - { - return Ok(()); + if assistant_speech.is_none() { + *assistant_speech = + Some(native_voice.begin_assistant_speech(interruption_sensitivity, suppress_capture)); } - *assistant_speech = Some(( - generation, - native_voice.begin_assistant_speech(interruption_sensitivity, suppress_capture), - )); if !*playback_started { *playback_started = true; emit_pocket_stream_event(app, stream_id, PocketStreamEventState::Started, None, None); @@ -2817,53 +2503,6 @@ fn mark_pocket_playback_started( Ok(()) } -#[cfg(target_os = "macos")] -#[allow(clippy::too_many_arguments)] -fn append_pocket_samples( - app: &AppHandle, - stream_id: &str, - player: &Player, - channels: std::num::NonZero, - rate: std::num::NonZero, - samples: Vec, - native_voice: &NativeVoiceState, - interruption_sensitivity: InterruptionSensitivity, - suppress_capture: bool, - playback_started: &mut bool, - assistant_speech: &Mutex>, - pending_audio_sources: &Arc, - completed_generation: &Arc, - playback_completion_sender: &mpsc::SyncSender, - playback_generation: &mut u64, -) -> Result<(), String> { - let previous_sources = pending_audio_sources.fetch_add(1, Ordering::AcqRel); - if previous_sources == 0 { - *playback_generation = playback_generation.wrapping_add(1).max(1); - } - let generation = *playback_generation; - if let Err(error) = mark_pocket_playback_started( - app, - stream_id, - native_voice, - interruption_sensitivity, - suppress_capture, - playback_started, - generation, - assistant_speech, - ) { - pending_audio_sources.fetch_sub(1, Ordering::AcqRel); - return Err(error); - } - player.append(PocketPlaybackSource::new( - SamplesBuffer::new(channels, rate, samples), - generation, - Arc::clone(pending_audio_sources), - Arc::clone(completed_generation), - playback_completion_sender.clone(), - )); - Ok(()) -} - #[cfg(target_os = "macos")] #[allow(clippy::too_many_arguments)] fn synthesize_pocket_stream_ready( @@ -2872,21 +2511,15 @@ fn synthesize_pocket_stream_ready( engine: &PocketTts, style: &VoiceStyle, active: &Arc, - player: &Player, - channels: std::num::NonZero, - rate: std::num::NonZero, - speed_processor: &mut StreamingSpeedProcessor, + player: &PocketAudioPlayer, pending: &mut String, first_chunk_pending: &mut bool, playback_started: &mut bool, native_voice: &NativeVoiceState, interruption_sensitivity: InterruptionSensitivity, suppress_capture: bool, - assistant_speech: &Mutex>, - pending_audio_sources: &Arc, - completed_generation: &Arc, - playback_completion_sender: &mpsc::SyncSender, - playback_generation: &mut u64, + assistant_speech: &mut Option, + playback_drained_at: &mut Option, delivery_ledger: &mut PlaybackDeliveryLedger, last_progress_emit: &mut Instant, flush: bool, @@ -2900,7 +2533,7 @@ fn synthesize_pocket_stream_ready( } let text = text.trim().to_string(); delivery_ledger.begin_segment(text.clone()); - let output_start = speed_processor.expected_output_frames(); + let mut segment_frames = 0_u64; let mut callback_error = None; let completed = engine.synth_chunk_streaming(&text, style, STREAMING_EMIT_FRAMES, &mut |samples| { @@ -2910,38 +2543,29 @@ fn synthesize_pocket_stream_ready( if samples.is_empty() { return true; } - let delta = match speed_processor.process(&samples) { - Ok(processed) => processed, - Err(error) => { - callback_error = Some(error); - return false; - } - }; - if delta.is_empty() { - return true; + if let Err(error) = player.ensure_healthy() { + callback_error = Some(error); + return false; } - let delta_len = delta.len(); - if let Err(error) = append_pocket_samples( + if let Err(error) = mark_pocket_playback_started( app, stream_id, - player, - channels, - rate, - delta, native_voice, interruption_sensitivity, suppress_capture, playback_started, assistant_speech, - pending_audio_sources, - completed_generation, - playback_completion_sender, - playback_generation, ) { callback_error = Some(error); return false; } - delivery_ledger.append_frames(delta_len); + if let Err(error) = player.enqueue(&samples) { + callback_error = Some(error); + return false; + } + reset_pocket_drain_grace(playback_drained_at); + segment_frames = segment_frames.saturating_add(samples.len() as u64); + delivery_ledger.append_frames(samples.len()); if last_progress_emit.elapsed() >= PLAYBACK_PROGRESS_EMIT_INTERVAL { emit_pocket_stream_event( app, @@ -2960,10 +2584,7 @@ fn synthesize_pocket_stream_ready( if !completed { return Ok(false); } - let final_total_frames = speed_processor - .expected_output_frames() - .saturating_sub(output_start) as u64; - delivery_ledger.complete_segment(final_total_frames); + delivery_ledger.complete_segment(segment_frames); } Ok(true) } @@ -2977,14 +2598,9 @@ fn synthesize_and_stream( active: Arc, speed: f32, ) -> Result<(), String> { - use std::cell::RefCell; - use std::num::NonZero; - use std::rc::Rc; use std::sync::Mutex; use std::time::Duration; - use rodio::cpal::traits::HostTrait; - let version = base.join(CACHE_VERSION); let engine = load_text_to_speech( version @@ -2992,94 +2608,48 @@ fn synthesize_and_stream( .ok_or_else(|| "Pocket model path is not valid UTF-8".to_string())?, )?; let style = load_voice_style(&version.join("voices").join(voice.filename))?; - let sink = if let Some(name) = output_device { - let host = rodio::cpal::default_host(); - let mut matching = None; - for device in host - .output_devices() - .map_err(|error| format!("enumerate audio outputs: {error}"))? - { - if device - .description() - .ok() - .is_some_and(|description| description.name() == name) - { - matching = Some(device); - break; - } - } - let device = matching.ok_or_else(|| format!("audio output not found: {name}"))?; - rodio::DeviceSinkBuilder::from_device(device) - .map_err(|error| format!("configure audio output {name}: {error}"))? - .open_stream() - .map_err(|error| format!("open audio output {name}: {error}"))? - } else { - rodio::DeviceSinkBuilder::open_default_sink() - .map_err(|error| format!("open default audio output: {error}"))? - }; - let channels = - NonZero::new(1_u16).ok_or_else(|| "Pocket channel count invariant failed".to_string())?; - let rate = NonZero::new(SAMPLE_RATE) - .ok_or_else(|| "Pocket sample rate invariant failed".to_string())?; - let player = Arc::new(Player::connect_new(sink.mixer())); - let speed_processor = Rc::new(RefCell::new(StreamingSpeedProcessor::new( - speed, - SAMPLE_RATE, - )?)); + let player = PocketAudioPlayer::new(SAMPLE_RATE, speed, output_device)?; let callback_error = Arc::new(Mutex::new(None::)); let playback_started = Arc::new(AtomicBool::new(false)); + let mut total_source_frames = 0_u64; - let callback_player = player.clone(); let callback_active = active.clone(); - let callback_speed_processor = speed_processor.clone(); let callback_error_slot = callback_error.clone(); let callback_started = playback_started.clone(); - let mut on_audio = move |samples: Vec| { + let mut on_audio = |samples: Vec| { if !callback_active.load(Ordering::SeqCst) { return false; } if samples.is_empty() { return true; } - let delta = match callback_speed_processor.borrow_mut().process(&samples) { - Ok(processed) => processed, - Err(error) => { + if let Err(error) = player.ensure_healthy() { + if let Ok(mut callback_error) = callback_error_slot.lock() { + *callback_error = Some(error); + } + return false; + } + if let Err(error) = player.enqueue(&samples) { + if let Ok(mut callback_error) = callback_error_slot.lock() { + *callback_error = Some(error); + } + return false; + } + total_source_frames = total_source_frames.saturating_add(samples.len() as u64); + if !callback_started.swap(true, Ordering::SeqCst) { + println!("VOICE_CONVERSATION_PLAYBACK_STARTED"); + if let Err(error) = std::io::stdout().flush() { if let Ok(mut callback_error) = callback_error_slot.lock() { - *callback_error = Some(error); + *callback_error = Some(format!("signal Pocket playback start: {error}")); } return false; } - }; - if !delta.is_empty() { - callback_player.append(SamplesBuffer::new(channels, rate, delta)); - if !callback_started.swap(true, Ordering::SeqCst) { - println!("VOICE_CONVERSATION_PLAYBACK_STARTED"); - if let Err(error) = std::io::stdout().flush() { - if let Ok(mut callback_error) = callback_error_slot.lock() { - *callback_error = Some(format!("signal Pocket playback start: {error}")); - } - return false; - } - } } true }; let completed = engine.synth_chunk_streaming(text, &style, STREAMING_EMIT_FRAMES, &mut on_audio)?; - if completed { - let tail = speed_processor.borrow_mut().finish()?; - if !tail.is_empty() { - player.append(SamplesBuffer::new(channels, rate, tail)); - if !playback_started.swap(true, Ordering::SeqCst) { - println!("VOICE_CONVERSATION_PLAYBACK_STARTED"); - std::io::stdout() - .flush() - .map_err(|error| format!("signal Pocket playback start: {error}"))?; - } - } - } - if let Some(error) = callback_error .lock() .map_err(|_| "Pocket callback error lock was poisoned".to_string())? @@ -3092,11 +2662,28 @@ fn synthesize_and_stream( player.stop(); return Ok(()); } - while !player.empty() { + let drain_timeout = + pocket_native_drain_timeout(total_source_frames, player.completed_source_frames(), speed); + let drain_started = Instant::now(); + loop { if !active.load(Ordering::SeqCst) { player.stop(); break; } + player.ensure_healthy()?; + match pocket_native_drain_status(player.is_empty(), drain_started.elapsed(), drain_timeout) + { + PocketNativeDrainStatus::Waiting => {} + PocketNativeDrainStatus::Drained => { + player.ensure_healthy()?; + break; + } + PocketNativeDrainStatus::TimedOut => { + log::warn!("Pocket one-shot native buffer completion bookkeeping timed out"); + player.stop(); + break; + } + } std::thread::sleep(Duration::from_millis(10)); } Ok(()) @@ -3123,16 +2710,13 @@ mod tests { let mut ledger = PlaybackDeliveryLedger::default(); ledger.begin_segment("First sentence.".to_string()); ledger.append_frames(4_800); - assert!(!ledger.snapshot(1, 0).segments[0].synthesis_complete); + assert!(!ledger.snapshot_consumed_frames(0).segments[0].synthesis_complete); ledger.complete_segment(4_800); ledger.begin_segment("Second sentence.".to_string()); ledger.append_frames(4_800); ledger.complete_segment(4_800); - // One source has completed and the next is 50 ms in. The 100 ms - // output-latency allowance leaves 3,600 safely delivered frames in - // the first segment and none in the second. - let progress = ledger.snapshot(1, 1_200); + let progress = ledger.snapshot_consumed_frames(3_600); assert_eq!(progress.segments[0].played_frames, 3_600); assert_eq!(progress.segments[0].total_frames, 4_800); assert!(progress.segments[0].synthesis_complete); @@ -3141,6 +2725,21 @@ mod tests { assert!(progress.segments[1].synthesis_complete); } + #[test] + fn playback_ledger_maps_native_consumed_frames_across_segments() { + let mut ledger = PlaybackDeliveryLedger::default(); + ledger.begin_segment("First sentence.".to_string()); + ledger.append_frames(4_800); + ledger.complete_segment(4_800); + ledger.begin_segment("Second sentence.".to_string()); + ledger.append_frames(4_800); + ledger.complete_segment(4_800); + + let progress = ledger.snapshot_consumed_frames(7_200); + assert_eq!(progress.segments[0].played_frames, 4_800); + assert_eq!(progress.segments[1].played_frames, 2_400); + } + #[test] fn interruption_and_failure_capture_delivery_before_stopping_playback() { use std::cell::RefCell; @@ -3158,7 +2757,7 @@ mod tests { let delivery = capture_before_stop( || { calls.borrow_mut().push("snapshot"); - ledger.snapshot(2, 1_200) + ledger.snapshot_consumed_frames(3_600) }, || calls.borrow_mut().push("stop"), ); @@ -3574,273 +3173,6 @@ mod tests { )); } - #[cfg(target_os = "macos")] - #[test] - fn stale_playback_completion_does_not_release_a_new_burst() { - let native_voice = NativeVoiceState::default(); - let assistant_speech = Mutex::new(Some(( - 2, - native_voice.begin_assistant_speech(InterruptionSensitivity::More, false), - ))); - - release_completed_pocket_assistant_speech(1, &assistant_speech); - assert!(assistant_speech.lock().expect("assistant speech").is_some()); - - release_completed_pocket_assistant_speech(2, &assistant_speech); - assert!(assistant_speech.lock().expect("assistant speech").is_none()); - } - - #[cfg(target_os = "macos")] - #[test] - fn dropped_pocket_source_finishes_pending_playback_once() { - let pending = Arc::new(AtomicUsize::new(1)); - let completed_generation = Arc::new(AtomicU64::new(0)); - let (sender, receiver) = mpsc::sync_channel(2); - let source = PocketPlaybackSource::new( - SamplesBuffer::new( - std::num::NonZeroU16::new(1).expect("channels"), - std::num::NonZeroU32::new(SAMPLE_RATE).expect("sample rate"), - vec![0.0], - ), - 7, - Arc::clone(&pending), - Arc::clone(&completed_generation), - sender, - ); - - drop(source); - - assert_eq!(pending.load(Ordering::Acquire), 0); - assert_eq!(completed_generation.load(Ordering::Acquire), 7); - assert!(matches!( - receiver.try_recv(), - Ok(PocketPlaybackEvent::SourceFinished) - )); - assert_eq!(receiver.try_iter().count(), 0); - } - - #[cfg(target_os = "macos")] - #[test] - fn exhausted_pocket_source_does_not_finish_again_when_dropped() { - let pending = Arc::new(AtomicUsize::new(1)); - let completed_generation = Arc::new(AtomicU64::new(0)); - let (sender, receiver) = mpsc::sync_channel(2); - let mut source = PocketPlaybackSource::new( - SamplesBuffer::new( - std::num::NonZeroU16::new(1).expect("channels"), - std::num::NonZeroU32::new(SAMPLE_RATE).expect("sample rate"), - vec![0.0], - ), - 9, - Arc::clone(&pending), - Arc::clone(&completed_generation), - sender, - ); - - assert!(source.next().is_some()); - assert!(source.next().is_none()); - drop(source); - - assert_eq!(pending.load(Ordering::Acquire), 0); - assert_eq!(completed_generation.load(Ordering::Acquire), 9); - assert!(matches!( - receiver.try_recv(), - Ok(PocketPlaybackEvent::SourceFinished) - )); - assert_eq!(receiver.try_iter().count(), 0); - } - - #[cfg(target_os = "macos")] - #[test] - fn playback_monitor_rechecks_generation_after_a_lost_completion_wakeup() { - let native_voice = NativeVoiceState::default(); - let assistant_speech = Arc::new(Mutex::new(Some(( - 3, - native_voice.begin_assistant_speech(InterruptionSensitivity::More, false), - )))); - let completed_generation = Arc::new(AtomicU64::new(1)); - let (sender, receiver) = mpsc::sync_channel(1); - let monitor = spawn_pocket_playback_monitor( - receiver, - Arc::clone(&assistant_speech), - Arc::clone(&completed_generation), - Duration::from_secs(1), - Duration::from_millis(10), - ) - .expect("spawn playback monitor"); - - sender - .send(PocketPlaybackEvent::SourceFinished) - .expect("send initial completion"); - let (probe_sender, probe_receiver) = mpsc::channel(); - sender - .send(PocketPlaybackEvent::Probe(probe_sender)) - .expect("send monitor probe"); - probe_receiver - .recv_timeout(Duration::from_secs(1)) - .expect("monitor entered completion grace"); - - // A full capacity-one channel can drop this generation's wakeup after - // the atomic state advances. The monitor must still observe the latest - // generation when its current grace expires. - completed_generation.store(3, Ordering::Release); - - let deadline = Instant::now() + Duration::from_secs(1); - while assistant_speech.lock().expect("assistant speech").is_some() { - assert!(Instant::now() < deadline, "latest guard was not released"); - std::thread::yield_now(); - } - - sender - .send(PocketPlaybackEvent::ShutdownImmediately) - .expect("stop playback monitor"); - monitor.join().expect("join playback monitor"); - } - - #[cfg(target_os = "macos")] - #[test] - fn playback_monitor_preserves_grace_when_shutdown_precedes_source_completion() { - let native_voice = NativeVoiceState::default(); - let assistant_speech = Arc::new(Mutex::new(Some(( - 7, - native_voice.begin_assistant_speech(InterruptionSensitivity::More, false), - )))); - let completed_generation = Arc::new(AtomicU64::new(0)); - let (sender, receiver) = mpsc::sync_channel(4); - let monitor = spawn_pocket_playback_monitor( - receiver, - Arc::clone(&assistant_speech), - Arc::clone(&completed_generation), - Duration::from_secs(1), - Duration::from_millis(20), - ) - .expect("spawn playback monitor"); - - sender - .send(PocketPlaybackEvent::ShutdownAfterPlayback) - .expect("send graceful shutdown"); - let (first_probe_sender, first_probe_receiver) = mpsc::channel(); - sender - .send(PocketPlaybackEvent::Probe(first_probe_sender)) - .expect("probe shutdown grace"); - first_probe_receiver - .recv_timeout(Duration::from_secs(1)) - .expect("monitor waits for source completion"); - - std::thread::sleep(Duration::from_millis(40)); - let (late_probe_sender, late_probe_receiver) = mpsc::channel(); - sender - .send(PocketPlaybackEvent::Probe(late_probe_sender)) - .expect("probe after an elapsed route interval"); - late_probe_receiver - .recv_timeout(Duration::from_secs(1)) - .expect("monitor still waits after an elapsed route interval"); - assert!(assistant_speech.lock().expect("assistant speech").is_some()); - - completed_generation.store(7, Ordering::Release); - sender - .send(PocketPlaybackEvent::SourceFinished) - .expect("send delayed source completion"); - let (second_probe_sender, second_probe_receiver) = mpsc::channel(); - sender - .send(PocketPlaybackEvent::Probe(second_probe_sender)) - .expect("probe restarted source grace"); - second_probe_receiver - .recv_timeout(Duration::from_secs(1)) - .expect("monitor observes delayed source completion"); - assert!(assistant_speech.lock().expect("assistant speech").is_some()); - - monitor.join().expect("join playback monitor"); - assert!(assistant_speech.lock().expect("assistant speech").is_none()); - } - - #[cfg(target_os = "macos")] - #[test] - fn playback_monitor_preserves_grace_when_source_completion_precedes_shutdown() { - let native_voice = NativeVoiceState::default(); - let assistant_speech = Arc::new(Mutex::new(Some(( - 9, - native_voice.begin_assistant_speech(InterruptionSensitivity::More, false), - )))); - let completed_generation = Arc::new(AtomicU64::new(9)); - let (sender, receiver) = mpsc::sync_channel(3); - let monitor = spawn_pocket_playback_monitor( - receiver, - Arc::clone(&assistant_speech), - Arc::clone(&completed_generation), - Duration::from_secs(1), - Duration::from_millis(20), - ) - .expect("spawn playback monitor"); - - sender - .send(PocketPlaybackEvent::SourceFinished) - .expect("send source completion"); - let (probe_sender, probe_receiver) = mpsc::channel(); - sender - .send(PocketPlaybackEvent::Probe(probe_sender)) - .expect("probe source grace"); - probe_receiver - .recv_timeout(Duration::from_secs(1)) - .expect("monitor entered source grace"); - assert!(assistant_speech.lock().expect("assistant speech").is_some()); - - sender - .send(PocketPlaybackEvent::ShutdownAfterPlayback) - .expect("send graceful shutdown"); - monitor.join().expect("join playback monitor"); - assert!(assistant_speech.lock().expect("assistant speech").is_none()); - } - - #[cfg(target_os = "macos")] - #[test] - fn playback_monitor_times_out_a_missing_source_completion_and_releases_ownership() { - let state = PocketVoiceState::default(); - let active = begin_playback_runtime(&state, "already active").expect("first playback"); - let native_voice = NativeVoiceState::default(); - let assistant_speech = Arc::new(Mutex::new(Some(( - 11, - native_voice.begin_assistant_speech(InterruptionSensitivity::More, true), - )))); - let completed_generation = Arc::new(AtomicU64::new(0)); - let (sender, receiver) = mpsc::sync_channel(1); - let monitor = spawn_pocket_playback_monitor( - receiver, - Arc::clone(&assistant_speech), - completed_generation, - Duration::from_millis(20), - Duration::from_millis(10), - ) - .expect("spawn playback monitor"); - - let outcome = run_with_playback_cleanup(&state.playback, &active, || { - sender - .send(PocketPlaybackEvent::ShutdownAfterPlayback) - .expect("send graceful shutdown"); - monitor.join().expect("join playback monitor") - }); - - assert_eq!( - outcome, - PocketPlaybackMonitorOutcome::SourceCompletionTimedOut - ); - assert!(matches!( - apply_pocket_playback_monitor_outcome( - Ok(PocketStreamOutcome { - state: PocketStreamEventState::Completed, - delivery: None, - }), - outcome, - ), - Ok(PocketStreamOutcome { - state: PocketStreamEventState::Completed, - .. - }) - )); - assert!(assistant_speech.lock().expect("assistant speech").is_none()); - assert!(begin_playback_runtime(&state, "still active").is_ok()); - } - #[cfg(target_os = "macos")] #[test] fn playback_drain_grace_covers_high_latency_transports() { @@ -3875,31 +3207,104 @@ mod tests { ); } - #[cfg(target_os = "macos")] #[test] - fn pocket_monitor_spawn_failure_releases_playback_for_a_new_stream() { - let state = PocketVoiceState::default(); - let active = begin_playback_runtime(&state, "already active").expect("first playback"); - let ran = Arc::new(AtomicBool::new(false)); - let task_ran = Arc::clone(&ran); - let result = run_with_playback_cleanup(&state.playback, &active, || { - spawn_pocket_playback_monitor_with( - move || { - task_ran.store(true, Ordering::SeqCst); - PocketPlaybackMonitorOutcome::Completed - }, - |_task| Err(std::io::Error::other("injected spawn failure")), - ) - }); + fn playback_drain_grace_restarts_when_a_new_burst_is_enqueued() { + let started = Instant::now(); + let grace = Duration::from_millis(500); + let mut drained_at = None; + + assert!(!pocket_assistant_speech_grace_elapsed( + true, + true, + &mut drained_at, + grace, + started, + )); + reset_pocket_drain_grace(&mut drained_at); + assert_eq!(drained_at, None); + + assert!(!pocket_assistant_speech_grace_elapsed( + true, + true, + &mut drained_at, + grace, + started + Duration::from_millis(600), + )); + assert!(!pocket_assistant_speech_grace_elapsed( + true, + true, + &mut drained_at, + grace, + started + Duration::from_millis(900), + )); + assert!(pocket_assistant_speech_grace_elapsed( + true, + true, + &mut drained_at, + grace, + started + Duration::from_millis(1_100), + )); + + assert!(!pocket_assistant_speech_grace_elapsed( + true, + false, + &mut drained_at, + grace, + started + Duration::from_secs(1), + )); + assert_eq!(drained_at, None); + } + #[test] + fn native_drain_times_out_after_expected_remaining_audio() { + let timeout = pocket_native_drain_timeout(72_000, 24_000, 2.0); + assert_eq!(timeout, Duration::from_secs(3)); assert_eq!( - result - .expect_err("spawn failure must propagate") - .to_string(), - "injected spawn failure" + pocket_native_drain_status(false, timeout - Duration::from_millis(1), timeout), + PocketNativeDrainStatus::Waiting ); - assert!(!ran.load(Ordering::SeqCst)); - assert!(begin_playback_runtime(&state, "still active").is_ok()); + assert_eq!( + pocket_native_drain_status(false, timeout, timeout), + PocketNativeDrainStatus::TimedOut + ); + assert_eq!( + pocket_native_drain_status(true, timeout, timeout), + PocketNativeDrainStatus::Drained + ); + } + + #[cfg(target_os = "macos")] + #[test] + fn native_drain_timeout_releases_guard_after_route_grace() { + let native_voice = NativeVoiceState::default(); + let mut assistant_speech = + Some(native_voice.begin_assistant_speech(InterruptionSensitivity::More, false)); + let mut drained_at = Some(Instant::now()); + let timed_out_at = Instant::now(); + let route_grace = Duration::from_millis(500); + + reset_pocket_drain_grace(&mut drained_at); + update_pocket_assistant_speech( + true, + &mut assistant_speech, + &mut drained_at, + route_grace, + timed_out_at, + ); + assert!(assistant_speech.is_some()); + + update_pocket_assistant_speech( + true, + &mut assistant_speech, + &mut drained_at, + route_grace, + timed_out_at + route_grace, + ); + assert!(assistant_speech.is_none()); + + assistant_speech = + Some(native_voice.begin_assistant_speech(InterruptionSensitivity::More, false)); + assert!(assistant_speech.is_some()); } #[test]