From 08534b9b0265a07fc249d9fe68699d15ff8d2cd0 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 25 Aug 2026 09:31:43 -0400 Subject: [PATCH 01/13] fix(voice): improve Pocket playback speed quality --- src-tauri/Cargo.lock | 116 +- src-tauri/Cargo.toml | 1 - src-tauri/native/siri_tts_bridge.h | 19 + src-tauri/native/siri_tts_bridge.m | 257 +++++ src-tauri/src/commands/mod.rs | 2 +- src-tauri/src/commands/pocket_audio_player.rs | 111 ++ .../src/commands/pocket_playback_speed_dsp.rs | 309 ----- src-tauri/src/commands/pocket_voice.rs | 1007 ++--------------- 8 files changed, 498 insertions(+), 1324 deletions(-) create mode 100644 src-tauri/src/commands/pocket_audio_player.rs delete mode 100644 src-tauri/src/commands/pocket_playback_speed_dsp.rs 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..bc7b156e0 100644 --- a/src-tauri/native/siri_tts_bridge.h +++ b/src-tauri/native/siri_tts_bridge.h @@ -73,6 +73,25 @@ bool berd_siri_tts_speak( char **error_out ); +/// Opaque Pocket PCM player backed by AVAudioUnitTimePitch. Samples are +/// mono, noninterleaved float PCM. A NULL output name uses the system default. +void *berd_pocket_audio_player_create( + uint32_t sample_rate, + float rate, + const char *output_device_name, + 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_played_frames(void *player); +uint64_t berd_pocket_audio_player_pending_buffers(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..2104cc7d1 100644 --- a/src-tauri/native/siri_tts_bridge.m +++ b/src-tauri/native/siri_tts_bridge.m @@ -567,6 +567,197 @@ - (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) BOOL stopped; +- (instancetype)initWithSampleRate:(double)sampleRate + rate:(float)rate + outputDeviceName:(NSString *_Nullable)outputDeviceName + error:(NSError **)error; +- (BOOL)enqueueSamples:(const float *)samples + frameCount:(AVAudioFrameCount)frameCount + error:(NSError **)error; +- (uint64_t)playedFrames; +- (void)stop; +@end + +static AudioDeviceID BerdAudioDeviceNamed(NSString *name) { + AudioObjectPropertyAddress address = { + kAudioHardwarePropertyDevices, + kAudioObjectPropertyScopeGlobal, + kAudioObjectPropertyElementMain, + }; + UInt32 size = 0; + if (AudioObjectGetPropertyDataSize(kAudioObjectSystemObject, &address, 0, NULL, &size) != noErr) { + return kAudioObjectUnknown; + } + UInt32 count = size / sizeof(AudioDeviceID); + AudioDeviceID *devices = calloc(count, sizeof(AudioDeviceID)); + if (!devices) return kAudioObjectUnknown; + if (AudioObjectGetPropertyData(kAudioObjectSystemObject, &address, 0, NULL, &size, devices) != noErr) { + free(devices); + return kAudioObjectUnknown; + } + + AudioDeviceID match = kAudioObjectUnknown; + AudioObjectPropertyAddress nameAddress = { + kAudioObjectPropertyName, + kAudioObjectPropertyScopeGlobal, + kAudioObjectPropertyElementMain, + }; + for (UInt32 index = 0; index < count; index++) { + AudioObjectPropertyAddress streamsAddress = { + kAudioDevicePropertyStreams, + kAudioDevicePropertyScopeOutput, + kAudioObjectPropertyElementMain, + }; + UInt32 streamsSize = 0; + if (AudioObjectGetPropertyDataSize(devices[index], &streamsAddress, 0, NULL, + &streamsSize) != noErr || streamsSize == 0) { + continue; + } + CFStringRef deviceName = NULL; + UInt32 nameSize = sizeof(deviceName); + if (AudioObjectGetPropertyData(devices[index], &nameAddress, 0, NULL, &nameSize, + &deviceName) == noErr && deviceName) { + if ([(__bridge NSString *)deviceName isEqualToString:name]) match = devices[index]; + CFRelease(deviceName); + } + if (match != kAudioObjectUnknown) break; + } + free(devices); + return match; +} + +@implementation BerdPocketAudioPlayer +- (instancetype)initWithSampleRate:(double)sampleRate + rate:(float)rate + outputDeviceName:(NSString *)outputDeviceName + 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; + // Apple documents that higher overlap values reduce artifacts at the + // cost of CPU. Voice playback is mono and short-lived, so prefer quality. + _timePitch.overlap = 32.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 (outputDeviceName.length) { + AudioDeviceID device = BerdAudioDeviceNamed(outputDeviceName); + if (device == kAudioObjectUnknown) { + if (error) { + *error = BerdError(32, [NSString stringWithFormat:@"Audio output not found: %@", + outputDeviceName]); + } + return nil; + } + AudioUnit outputUnit = _engine.outputNode.audioUnit; + OSStatus status = AudioUnitSetProperty(outputUnit, + kAudioOutputUnitProperty_CurrentDevice, + kAudioUnitScope_Global, + 0, + &device, + sizeof(device)); + if (status != noErr) { + if (error) *error = BerdError(33, @"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) { + strongSelf.pendingBuffers = strongSelf.pendingBuffers > 0 + ? strongSelf.pendingBuffers - 1 + : 0; + } + }]; + if (!self.player.isPlaying) [self.player play]; + return YES; +} + +- (uint64_t)playedFrames { + AVAudioTime *renderTime = self.player.lastRenderTime; + if (!renderTime) return 0; + AVAudioTime *playerTime = [self.player playerTimeForNodeTime:renderTime]; + if (!playerTime || playerTime.sampleTime <= 0) return 0; + uint64_t frames = (uint64_t)playerTime.sampleTime; + uint64_t effectLatency = (uint64_t)llround(self.timePitch.latency * self.format.sampleRate); + return frames > effectLatency ? frames - effectLatency : 0; +} + +- (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 +1336,72 @@ bool berd_siri_tts_speak( } } +void *berd_pocket_audio_player_create( + uint32_t sampleRate, + float rate, + const char *outputDeviceNameValue, + char **errorOut +) { + @autoreleasepool { + if (errorOut) *errorOut = NULL; + NSString *outputDeviceName = outputDeviceNameValue + ? [NSString stringWithUTF8String:outputDeviceNameValue] + : nil; + NSError *error = nil; + BerdPocketAudioPlayer *player = [[BerdPocketAudioPlayer alloc] + initWithSampleRate:sampleRate + rate:rate + outputDeviceName:outputDeviceName + 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_played_frames(void *playerValue) { + if (!playerValue) return 0; + return [(__bridge BerdPocketAudioPlayer *)playerValue playedFrames]; +} + +uint64_t berd_pocket_audio_player_pending_buffers(void *playerValue) { + if (!playerValue) return 0; + BerdPocketAudioPlayer *player = (__bridge BerdPocketAudioPlayer *)playerValue; + @synchronized (player) { return player.pendingBuffers; } +} + +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..5c7b6edc6 --- /dev/null +++ b/src-tauri/src/commands/pocket_audio_player.rs @@ -0,0 +1,111 @@ +//! Safe ownership wrapper for the macOS AVAudioUnitTimePitch Pocket player. + +use std::ffi::{c_char, c_void, CStr, CString}; + +unsafe extern "C" { + fn berd_pocket_audio_player_create( + sample_rate: u32, + rate: f32, + output_device_name: *const c_char, + 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_played_frames(player: *mut c_void) -> u64; + fn berd_pocket_audio_player_pending_buffers(player: *mut c_void) -> u64; + 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, +} + +impl PocketAudioPlayer { + pub(super) fn new( + sample_rate: u32, + rate: f32, + output_device_name: Option<&str>, + ) -> Result { + let output_device_name = output_device_name + .map(CString::new) + .transpose() + .map_err(|_| "Pocket output device name cannot contain NUL bytes".to_string())?; + let mut error = std::ptr::null_mut(); + // SAFETY: The bridge copies the optional name synchronously and returns + // an owned opaque player retained until `Drop`. + let raw = unsafe { + berd_pocket_audio_player_create( + sample_rate, + rate, + output_device_name + .as_ref() + .map_or(std::ptr::null(), |name| name.as_ptr()), + &mut error, + ) + }; + if raw.is_null() { + return Err(take_error(error, "Could not start native Pocket playback")); + } + Ok(Self { raw }) + } + + 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. + unsafe { berd_pocket_audio_player_played_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 stop(&self) { + // SAFETY: `self.raw` is a live retained player and stop is idempotent. + unsafe { berd_pocket_audio_player_stop(self.raw) }; + } +} + +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 +} 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..d3de93aaf 100644 --- a/src-tauri/src/commands/pocket_voice.rs +++ b/src-tauri/src/commands/pocket_voice.rs @@ -8,8 +8,6 @@ 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}; @@ -27,19 +25,15 @@ 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}; #[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}, + native_voice::{AssistantSpeechGuard, InterruptionSensitivity, NativeVoiceState}, voice_capture::VoiceCaptureState, }; use tokio::io::AsyncWriteExt; @@ -66,192 +60,6 @@ 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")] -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 @@ -286,83 +94,6 @@ pub(crate) fn playback_latency_safety_duration(output_device: Option<&str>) -> D ) } -#[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 { @@ -548,6 +279,7 @@ impl PlaybackDeliveryLedger { } } + #[cfg(test)] 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 @@ -582,6 +314,31 @@ impl PlaybackDeliveryLedger { segments, } } + + fn snapshot_consumed_frames(&self, consumed_frames: u64) -> VoiceDeliveryProgress { + let consumed_frames = consumed_frames.saturating_sub(PLAYBACK_LATENCY_SAFETY_FRAMES); + let mut segment_start = 0_u64; + let segments = self + .segments + .iter() + .map(|(text, total_frames, synthesis_complete)| { + let played_frames = consumed_frames + .saturating_sub(segment_start) + .min(*total_frames); + segment_start = segment_start.saturating_add(*total_frames); + VoiceDeliverySegment { + text: text.clone(), + played_frames, + total_frames: *total_frames, + synthesis_complete: *synthesis_complete, + } + }) + .collect(); + VoiceDeliveryProgress { + sample_rate: berd_voice::SAMPLE_RATE, + segments, + } + } } #[cfg(target_os = "macos")] @@ -2439,10 +2196,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,53 +2203,13 @@ 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_empty_since = None::; + let playback_latency_safety_duration = playback_latency_safety_duration(output_device); let mut delivery_ledger = PlaybackDeliveryLedger::default(); let mut last_progress_emit = Instant::now(); @@ -2520,20 +2233,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_empty_since, &mut delivery_ledger, &mut last_progress_emit, false, @@ -2556,20 +2263,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_empty_since, &mut delivery_ledger, &mut last_progress_emit, true, @@ -2583,28 +2284,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 +2293,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_empty_since, &mut delivery_ledger, &mut last_progress_emit, true, @@ -2641,29 +2314,7 @@ 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() { + while !player.is_empty() { if !active.load(Ordering::SeqCst) { let delivery = pocket_delivery_snapshot(&delivery_ledger, &player); player.stop(); @@ -2674,6 +2325,8 @@ fn run_pocket_voice_stream( } std::thread::sleep(Duration::from_millis(10)); } + std::thread::sleep(playback_latency_safety_duration); + assistant_speech.take(); return Ok(PocketStreamOutcome { state: PocketStreamEventState::Completed, delivery: None, @@ -2689,6 +2342,12 @@ fn run_pocket_voice_stream( }); } Err(mpsc::RecvTimeoutError::Timeout) => { + update_pocket_assistant_speech( + &player, + &mut assistant_speech, + &mut playback_empty_since, + playback_latency_safety_duration, + ); if playback_started && last_progress_emit.elapsed() >= PLAYBACK_PROGRESS_EMIT_INTERVAL { @@ -2705,55 +2364,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,17 +2393,19 @@ 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( + player: &PocketAudioPlayer, + assistant_speech: &mut Option, + playback_empty_since: &mut Option, + playback_latency_safety_duration: Duration, ) { - 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 !player.is_empty() { + *playback_empty_since = None; + return; + } + let empty_since = playback_empty_since.get_or_insert_with(Instant::now); + if empty_since.elapsed() >= playback_latency_safety_duration { + assistant_speech.take(); } } @@ -2790,22 +2418,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 +2435,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 +2443,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_empty_since: &mut Option, delivery_ledger: &mut PlaybackDeliveryLedger, last_progress_emit: &mut Instant, flush: bool, @@ -2900,7 +2465,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 +2475,25 @@ 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.enqueue(&samples) { + callback_error = Some(error); + return false; } - let delta_len = delta.len(); - if let Err(error) = append_pocket_samples( + *playback_empty_since = None; + 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); + 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 +2512,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 +2526,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 +2536,40 @@ 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 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.enqueue(&samples) { + if let Ok(mut callback_error) = callback_error_slot.lock() { + *callback_error = Some(error); + } + return false; + } + 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,7 +2582,7 @@ fn synthesize_and_stream( player.stop(); return Ok(()); } - while !player.empty() { + while !player.is_empty() { if !active.load(Ordering::SeqCst) { player.stop(); break; @@ -3141,6 +2631,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, 0); + } + #[test] fn interruption_and_failure_capture_delivery_before_stopping_playback() { use std::cell::RefCell; @@ -3574,273 +3079,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,33 +3113,6 @@ 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")), - ) - }); - - assert_eq!( - result - .expect_err("spawn failure must propagate") - .to_string(), - "injected spawn failure" - ); - assert!(!ran.load(Ordering::SeqCst)); - assert!(begin_playback_runtime(&state, "still active").is_ok()); - } - #[test] fn parakeet_progress_is_monotonic_through_extraction_and_verification() { let mut runtime = InstallRuntime::default(); From ee015889ccb3b788ca8c1eb455184a975ada340a Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 25 Aug 2026 09:55:44 -0400 Subject: [PATCH 02/13] fix(voice): preserve native playback lifecycle --- src-tauri/native/siri_tts_bridge.m | 10 ++- src-tauri/src/commands/pocket_audio_player.rs | 23 ++++- src-tauri/src/commands/pocket_voice.rs | 86 +++---------------- 3 files changed, 40 insertions(+), 79 deletions(-) diff --git a/src-tauri/native/siri_tts_bridge.m b/src-tauri/native/siri_tts_bridge.m index 2104cc7d1..afcc5b404 100644 --- a/src-tauri/native/siri_tts_bridge.m +++ b/src-tauri/native/siri_tts_bridge.m @@ -741,7 +741,9 @@ - (uint64_t)playedFrames { AVAudioTime *playerTime = [self.player playerTimeForNodeTime:renderTime]; if (!playerTime || playerTime.sampleTime <= 0) return 0; uint64_t frames = (uint64_t)playerTime.sampleTime; - uint64_t effectLatency = (uint64_t)llround(self.timePitch.latency * self.format.sampleRate); + double rate = self.timePitch ? self.timePitch.rate : 1.0; + uint64_t effectLatency = + (uint64_t)llround(self.timePitch.latency * self.format.sampleRate * rate); return frames > effectLatency ? frames - effectLatency : 0; } @@ -1382,8 +1384,10 @@ bool berd_pocket_audio_player_enqueue( } uint64_t berd_pocket_audio_player_played_frames(void *playerValue) { - if (!playerValue) return 0; - return [(__bridge BerdPocketAudioPlayer *)playerValue playedFrames]; + @autoreleasepool { + if (!playerValue) return 0; + return [(__bridge BerdPocketAudioPlayer *)playerValue playedFrames]; + } } uint64_t berd_pocket_audio_player_pending_buffers(void *playerValue) { diff --git a/src-tauri/src/commands/pocket_audio_player.rs b/src-tauri/src/commands/pocket_audio_player.rs index 5c7b6edc6..ef3afff59 100644 --- a/src-tauri/src/commands/pocket_audio_player.rs +++ b/src-tauri/src/commands/pocket_audio_player.rs @@ -24,6 +24,7 @@ unsafe extern "C" { pub(super) struct PocketAudioPlayer { raw: *mut c_void, + delivery_safety_frames: u64, } impl PocketAudioPlayer { @@ -52,7 +53,10 @@ impl PocketAudioPlayer { if raw.is_null() { return Err(take_error(error, "Could not start native Pocket playback")); } - Ok(Self { raw }) + Ok(Self { + raw, + delivery_safety_frames: delivery_safety_frames(sample_rate, rate), + }) } pub(super) fn enqueue(&self, samples: &[f32]) -> Result<(), String> { @@ -77,6 +81,7 @@ impl PocketAudioPlayer { pub(super) fn played_frames(&self) -> u64 { // SAFETY: `self.raw` is a live retained player. unsafe { berd_pocket_audio_player_played_frames(self.raw) } + .saturating_sub(self.delivery_safety_frames) } pub(super) fn is_empty(&self) -> bool { @@ -90,6 +95,10 @@ impl PocketAudioPlayer { } } +fn delivery_safety_frames(sample_rate: u32, rate: f32) -> u64 { + (f64::from(sample_rate) * 0.1 * f64::from(rate)).ceil() as u64 +} + impl Drop for PocketAudioPlayer { fn drop(&mut self) { // SAFETY: This wrapper uniquely owns the retained bridge reference. @@ -109,3 +118,15 @@ fn take_error(error: *mut c_char, fallback: &str) -> String { unsafe { berd_siri_tts_free_string(error) }; message } + +#[cfg(test)] +mod tests { + use super::delivery_safety_frames; + + #[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); + } +} diff --git a/src-tauri/src/commands/pocket_voice.rs b/src-tauri/src/commands/pocket_voice.rs index d3de93aaf..a6c6ab4c6 100644 --- a/src-tauri/src/commands/pocket_voice.rs +++ b/src-tauri/src/commands/pocket_voice.rs @@ -60,6 +60,7 @@ 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")] fn playback_latency_safety_duration_for_transport(transport: Option) -> Duration { // CoreAudio transport FOURCC values. Bluetooth and AirPlay routes buffer @@ -93,9 +94,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(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, @@ -250,7 +248,6 @@ struct VoiceDeliveryProgress { #[derive(Debug, Default)] struct PlaybackDeliveryLedger { segments: Vec<(String, u64, bool)>, - pieces: Vec, } #[cfg(any(test, target_os = "macos"))] @@ -268,7 +265,6 @@ impl PlaybackDeliveryLedger { if !*synthesis_complete { *total = total.saturating_add(frames); } - self.pieces.push(frames); } } @@ -279,44 +275,7 @@ impl PlaybackDeliveryLedger { } } - #[cfg(test)] - 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 - .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); - let mut segment_start = 0_u64; - let segments = self - .segments - .iter() - .map(|(text, total_frames, synthesis_complete)| { - let played_frames = consumed_frames - .saturating_sub(segment_start) - .min(*total_frames); - segment_start = segment_start.saturating_add(*total_frames); - VoiceDeliverySegment { - text: text.clone(), - played_frames, - total_frames: *total_frames, - synthesis_complete: *synthesis_complete, - } - }) - .collect(); - VoiceDeliveryProgress { - sample_rate: berd_voice::SAMPLE_RATE, - segments, - } - } - fn snapshot_consumed_frames(&self, consumed_frames: u64) -> VoiceDeliveryProgress { - let consumed_frames = consumed_frames.saturating_sub(PLAYBACK_LATENCY_SAFETY_FRAMES); let mut segment_start = 0_u64; let segments = self .segments @@ -2208,12 +2167,11 @@ fn run_pocket_voice_stream( let mut first_chunk_pending = true; let mut playback_started = false; let mut assistant_speech = None::; - let mut playback_empty_since = None::; - let playback_latency_safety_duration = 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, &mut assistant_speech); if !active.load(Ordering::SeqCst) { let delivery = pocket_delivery_snapshot(&delivery_ledger, &player); player.stop(); @@ -2240,7 +2198,6 @@ fn run_pocket_voice_stream( interruption_sensitivity, suppress_capture, &mut assistant_speech, - &mut playback_empty_since, &mut delivery_ledger, &mut last_progress_emit, false, @@ -2270,7 +2227,6 @@ fn run_pocket_voice_stream( interruption_sensitivity, suppress_capture, &mut assistant_speech, - &mut playback_empty_since, &mut delivery_ledger, &mut last_progress_emit, true, @@ -2300,7 +2256,6 @@ fn run_pocket_voice_stream( interruption_sensitivity, suppress_capture, &mut assistant_speech, - &mut playback_empty_since, &mut delivery_ledger, &mut last_progress_emit, true, @@ -2325,7 +2280,6 @@ fn run_pocket_voice_stream( } std::thread::sleep(Duration::from_millis(10)); } - std::thread::sleep(playback_latency_safety_duration); assistant_speech.take(); return Ok(PocketStreamOutcome { state: PocketStreamEventState::Completed, @@ -2342,12 +2296,6 @@ fn run_pocket_voice_stream( }); } Err(mpsc::RecvTimeoutError::Timeout) => { - update_pocket_assistant_speech( - &player, - &mut assistant_speech, - &mut playback_empty_since, - playback_latency_safety_duration, - ); if playback_started && last_progress_emit.elapsed() >= PLAYBACK_PROGRESS_EMIT_INTERVAL { @@ -2396,15 +2344,8 @@ fn capture_before_stop( fn update_pocket_assistant_speech( player: &PocketAudioPlayer, assistant_speech: &mut Option, - playback_empty_since: &mut Option, - playback_latency_safety_duration: Duration, ) { - if !player.is_empty() { - *playback_empty_since = None; - return; - } - let empty_since = playback_empty_since.get_or_insert_with(Instant::now); - if empty_since.elapsed() >= playback_latency_safety_duration { + if player.is_empty() { assistant_speech.take(); } } @@ -2451,7 +2392,6 @@ fn synthesize_pocket_stream_ready( interruption_sensitivity: InterruptionSensitivity, suppress_capture: bool, assistant_speech: &mut Option, - playback_empty_since: &mut Option, delivery_ledger: &mut PlaybackDeliveryLedger, last_progress_emit: &mut Instant, flush: bool, @@ -2475,11 +2415,6 @@ fn synthesize_pocket_stream_ready( if samples.is_empty() { return true; } - if let Err(error) = player.enqueue(&samples) { - callback_error = Some(error); - return false; - } - *playback_empty_since = None; if let Err(error) = mark_pocket_playback_started( app, stream_id, @@ -2492,6 +2427,10 @@ fn synthesize_pocket_stream_ready( callback_error = Some(error); return false; } + if let Err(error) = player.enqueue(&samples) { + callback_error = Some(error); + return false; + } 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 { @@ -2613,16 +2552,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); @@ -2643,7 +2579,7 @@ mod tests { let progress = ledger.snapshot_consumed_frames(7_200); assert_eq!(progress.segments[0].played_frames, 4_800); - assert_eq!(progress.segments[1].played_frames, 0); + assert_eq!(progress.segments[1].played_frames, 2_400); } #[test] @@ -2663,7 +2599,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"), ); From 393f4195068caac149888cdac8a3d566439c1ddd Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 25 Aug 2026 10:03:04 -0400 Subject: [PATCH 03/13] fix(voice): count completed Pocket audio frames --- src-tauri/native/siri_tts_bridge.h | 4 +- src-tauri/native/siri_tts_bridge.m | 101 ++++-------------- src-tauri/src/commands/pocket_audio_player.rs | 61 +++++++---- 3 files changed, 60 insertions(+), 106 deletions(-) diff --git a/src-tauri/native/siri_tts_bridge.h b/src-tauri/native/siri_tts_bridge.h index bc7b156e0..e6f36cb27 100644 --- a/src-tauri/native/siri_tts_bridge.h +++ b/src-tauri/native/siri_tts_bridge.h @@ -78,7 +78,7 @@ bool berd_siri_tts_speak( void *berd_pocket_audio_player_create( uint32_t sample_rate, float rate, - const char *output_device_name, + uint32_t output_device_id, char **error_out ); bool berd_pocket_audio_player_enqueue( @@ -87,7 +87,7 @@ bool berd_pocket_audio_player_enqueue( uint32_t frame_count, char **error_out ); -uint64_t berd_pocket_audio_player_played_frames(void *player); +uint64_t berd_pocket_audio_player_completed_source_frames(void *player); uint64_t berd_pocket_audio_player_pending_buffers(void *player); void berd_pocket_audio_player_stop(void *player); void berd_pocket_audio_player_release(void *player); diff --git a/src-tauri/native/siri_tts_bridge.m b/src-tauri/native/siri_tts_bridge.m index afcc5b404..d445adb1a 100644 --- a/src-tauri/native/siri_tts_bridge.m +++ b/src-tauri/native/siri_tts_bridge.m @@ -573,70 +573,23 @@ @interface BerdPocketAudioPlayer : NSObject @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 stopped; - (instancetype)initWithSampleRate:(double)sampleRate rate:(float)rate - outputDeviceName:(NSString *_Nullable)outputDeviceName + outputDeviceID:(AudioDeviceID)outputDeviceID error:(NSError **)error; - (BOOL)enqueueSamples:(const float *)samples frameCount:(AVAudioFrameCount)frameCount error:(NSError **)error; -- (uint64_t)playedFrames; +- (uint64_t)completedSourceFramesSnapshot; - (void)stop; @end -static AudioDeviceID BerdAudioDeviceNamed(NSString *name) { - AudioObjectPropertyAddress address = { - kAudioHardwarePropertyDevices, - kAudioObjectPropertyScopeGlobal, - kAudioObjectPropertyElementMain, - }; - UInt32 size = 0; - if (AudioObjectGetPropertyDataSize(kAudioObjectSystemObject, &address, 0, NULL, &size) != noErr) { - return kAudioObjectUnknown; - } - UInt32 count = size / sizeof(AudioDeviceID); - AudioDeviceID *devices = calloc(count, sizeof(AudioDeviceID)); - if (!devices) return kAudioObjectUnknown; - if (AudioObjectGetPropertyData(kAudioObjectSystemObject, &address, 0, NULL, &size, devices) != noErr) { - free(devices); - return kAudioObjectUnknown; - } - - AudioDeviceID match = kAudioObjectUnknown; - AudioObjectPropertyAddress nameAddress = { - kAudioObjectPropertyName, - kAudioObjectPropertyScopeGlobal, - kAudioObjectPropertyElementMain, - }; - for (UInt32 index = 0; index < count; index++) { - AudioObjectPropertyAddress streamsAddress = { - kAudioDevicePropertyStreams, - kAudioDevicePropertyScopeOutput, - kAudioObjectPropertyElementMain, - }; - UInt32 streamsSize = 0; - if (AudioObjectGetPropertyDataSize(devices[index], &streamsAddress, 0, NULL, - &streamsSize) != noErr || streamsSize == 0) { - continue; - } - CFStringRef deviceName = NULL; - UInt32 nameSize = sizeof(deviceName); - if (AudioObjectGetPropertyData(devices[index], &nameAddress, 0, NULL, &nameSize, - &deviceName) == noErr && deviceName) { - if ([(__bridge NSString *)deviceName isEqualToString:name]) match = devices[index]; - CFRelease(deviceName); - } - if (match != kAudioObjectUnknown) break; - } - free(devices); - return match; -} - @implementation BerdPocketAudioPlayer - (instancetype)initWithSampleRate:(double)sampleRate rate:(float)rate - outputDeviceName:(NSString *)outputDeviceName + outputDeviceID:(AudioDeviceID)outputDeviceID error:(NSError **)error { self = [super init]; if (!self) return nil; @@ -673,24 +626,16 @@ - (instancetype)initWithSampleRate:(double)sampleRate [_engine connect:_player to:_engine.mainMixerNode format:_format]; } - if (outputDeviceName.length) { - AudioDeviceID device = BerdAudioDeviceNamed(outputDeviceName); - if (device == kAudioObjectUnknown) { - if (error) { - *error = BerdError(32, [NSString stringWithFormat:@"Audio output not found: %@", - outputDeviceName]); - } - return nil; - } + if (outputDeviceID != kAudioObjectUnknown) { AudioUnit outputUnit = _engine.outputNode.audioUnit; OSStatus status = AudioUnitSetProperty(outputUnit, kAudioOutputUnitProperty_CurrentDevice, kAudioUnitScope_Global, 0, - &device, - sizeof(device)); + &outputDeviceID, + sizeof(outputDeviceID)); if (status != noErr) { - if (error) *error = BerdError(33, @"Could not select the configured audio output."); + if (error) *error = BerdError(32, @"Could not select the configured audio output."); return nil; } } @@ -726,6 +671,9 @@ - (BOOL)enqueueSamples:(const float *)samples BerdPocketAudioPlayer *strongSelf = weakSelf; if (!strongSelf) return; @synchronized (strongSelf) { + if (!strongSelf.stopped) { + strongSelf.completedSourceFrames += frameCount; + } strongSelf.pendingBuffers = strongSelf.pendingBuffers > 0 ? strongSelf.pendingBuffers - 1 : 0; @@ -735,16 +683,8 @@ - (BOOL)enqueueSamples:(const float *)samples return YES; } -- (uint64_t)playedFrames { - AVAudioTime *renderTime = self.player.lastRenderTime; - if (!renderTime) return 0; - AVAudioTime *playerTime = [self.player playerTimeForNodeTime:renderTime]; - if (!playerTime || playerTime.sampleTime <= 0) return 0; - uint64_t frames = (uint64_t)playerTime.sampleTime; - double rate = self.timePitch ? self.timePitch.rate : 1.0; - uint64_t effectLatency = - (uint64_t)llround(self.timePitch.latency * self.format.sampleRate * rate); - return frames > effectLatency ? frames - effectLatency : 0; +- (uint64_t)completedSourceFramesSnapshot { + @synchronized (self) { return self.completedSourceFrames; } } - (void)stop { @@ -1341,19 +1281,16 @@ bool berd_siri_tts_speak( void *berd_pocket_audio_player_create( uint32_t sampleRate, float rate, - const char *outputDeviceNameValue, + uint32_t outputDeviceID, char **errorOut ) { @autoreleasepool { if (errorOut) *errorOut = NULL; - NSString *outputDeviceName = outputDeviceNameValue - ? [NSString stringWithUTF8String:outputDeviceNameValue] - : nil; NSError *error = nil; BerdPocketAudioPlayer *player = [[BerdPocketAudioPlayer alloc] initWithSampleRate:sampleRate rate:rate - outputDeviceName:outputDeviceName + outputDeviceID:outputDeviceID error:&error]; if (!player) { BerdSetError(errorOut, error ?: BerdError(35, @"Could not start Pocket playback.")); @@ -1383,11 +1320,9 @@ bool berd_pocket_audio_player_enqueue( } } -uint64_t berd_pocket_audio_player_played_frames(void *playerValue) { - @autoreleasepool { - if (!playerValue) return 0; - return [(__bridge BerdPocketAudioPlayer *)playerValue playedFrames]; - } +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) { diff --git a/src-tauri/src/commands/pocket_audio_player.rs b/src-tauri/src/commands/pocket_audio_player.rs index ef3afff59..fd87ceb1e 100644 --- a/src-tauri/src/commands/pocket_audio_player.rs +++ b/src-tauri/src/commands/pocket_audio_player.rs @@ -1,12 +1,12 @@ //! Safe ownership wrapper for the macOS AVAudioUnitTimePitch Pocket player. -use std::ffi::{c_char, c_void, CStr, CString}; +use std::ffi::{c_char, c_void, CStr}; unsafe extern "C" { fn berd_pocket_audio_player_create( sample_rate: u32, rate: f32, - output_device_name: *const c_char, + output_device_id: u32, error_out: *mut *mut c_char, ) -> *mut c_void; fn berd_pocket_audio_player_enqueue( @@ -15,7 +15,7 @@ unsafe extern "C" { frame_count: u32, error_out: *mut *mut c_char, ) -> bool; - fn berd_pocket_audio_player_played_frames(player: *mut c_void) -> u64; + 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_stop(player: *mut c_void); fn berd_pocket_audio_player_release(player: *mut c_void); @@ -33,22 +33,18 @@ impl PocketAudioPlayer { rate: f32, output_device_name: Option<&str>, ) -> Result { - let output_device_name = output_device_name - .map(CString::new) - .transpose() - .map_err(|_| "Pocket output device name cannot contain NUL bytes".to_string())?; + 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 copies the optional name synchronously and returns - // an owned opaque player retained until `Drop`. + // 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_name - .as_ref() - .map_or(std::ptr::null(), |name| name.as_ptr()), - &mut error, - ) + 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")); @@ -79,9 +75,12 @@ impl PocketAudioPlayer { } pub(super) fn played_frames(&self) -> u64 { - // SAFETY: `self.raw` is a live retained player. - unsafe { berd_pocket_audio_player_played_frames(self.raw) } - .saturating_sub(self.delivery_safety_frames) + // 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( + unsafe { berd_pocket_audio_player_completed_source_frames(self.raw) }, + self.delivery_safety_frames, + ) } pub(super) fn is_empty(&self) -> bool { @@ -99,6 +98,10 @@ 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) +} + impl Drop for PocketAudioPlayer { fn drop(&mut self) { // SAFETY: This wrapper uniquely owns the retained bridge reference. @@ -121,7 +124,7 @@ fn take_error(error: *mut c_char, fallback: &str) -> String { #[cfg(test)] mod tests { - use super::delivery_safety_frames; + use super::{apply_delivery_safety, delivery_safety_frames}; #[test] fn delivery_safety_tracks_playback_rate_in_source_frames() { @@ -129,4 +132,20 @@ mod tests { 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 + ); + } } From ab17b924965b2aa5259f095e539ed37512981587 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 25 Aug 2026 10:13:24 -0400 Subject: [PATCH 04/13] fix(voice): detect interrupted native output --- src-tauri/native/siri_tts_bridge.h | 3 ++- src-tauri/native/siri_tts_bridge.m | 14 ++++++++++- src-tauri/src/commands/pocket_audio_player.rs | 25 ++++++++++++++++++- src-tauri/src/commands/pocket_voice.rs | 23 +++++++++++++++-- 4 files changed, 60 insertions(+), 5 deletions(-) diff --git a/src-tauri/native/siri_tts_bridge.h b/src-tauri/native/siri_tts_bridge.h index e6f36cb27..b97cba220 100644 --- a/src-tauri/native/siri_tts_bridge.h +++ b/src-tauri/native/siri_tts_bridge.h @@ -74,7 +74,7 @@ bool berd_siri_tts_speak( ); /// Opaque Pocket PCM player backed by AVAudioUnitTimePitch. Samples are -/// mono, noninterleaved float PCM. A NULL output name uses the system default. +/// mono, noninterleaved float PCM. Device ID 0 uses the system default. void *berd_pocket_audio_player_create( uint32_t sample_rate, float rate, @@ -89,6 +89,7 @@ bool berd_pocket_audio_player_enqueue( ); 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); diff --git a/src-tauri/native/siri_tts_bridge.m b/src-tauri/native/siri_tts_bridge.m index d445adb1a..a6f28e329 100644 --- a/src-tauri/native/siri_tts_bridge.m +++ b/src-tauri/native/siri_tts_bridge.m @@ -574,6 +574,7 @@ @interface BerdPocketAudioPlayer : NSObject @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 @@ -671,8 +672,11 @@ - (BOOL)enqueueSamples:(const float *)samples BerdPocketAudioPlayer *strongSelf = weakSelf; if (!strongSelf) return; @synchronized (strongSelf) { - if (!strongSelf.stopped) { + 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 @@ -1331,6 +1335,14 @@ uint64_t berd_pocket_audio_player_pending_buffers(void *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]; diff --git a/src-tauri/src/commands/pocket_audio_player.rs b/src-tauri/src/commands/pocket_audio_player.rs index fd87ceb1e..00c1813d4 100644 --- a/src-tauri/src/commands/pocket_audio_player.rs +++ b/src-tauri/src/commands/pocket_audio_player.rs @@ -17,6 +17,7 @@ unsafe extern "C" { ) -> 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); @@ -88,6 +89,11 @@ impl PocketAudioPlayer { 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) }; @@ -102,6 +108,14 @@ fn apply_delivery_safety(completed_source_frames: u64, safety_frames: u64) -> u6 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. @@ -124,7 +138,7 @@ fn take_error(error: *mut c_char, fallback: &str) -> String { #[cfg(test)] mod tests { - use super::{apply_delivery_safety, delivery_safety_frames}; + use super::{apply_delivery_safety, delivery_safety_frames, playback_health}; #[test] fn delivery_safety_tracks_playback_rate_in_source_frames() { @@ -148,4 +162,13 @@ mod tests { 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_voice.rs b/src-tauri/src/commands/pocket_voice.rs index a6c6ab4c6..a635c81a7 100644 --- a/src-tauri/src/commands/pocket_voice.rs +++ b/src-tauri/src/commands/pocket_voice.rs @@ -2180,6 +2180,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)) => { @@ -2269,7 +2270,7 @@ fn run_pocket_voice_stream( delivery: Some(delivery), }); } - while !player.is_empty() { + loop { if !active.load(Ordering::SeqCst) { let delivery = pocket_delivery_snapshot(&delivery_ledger, &player); player.stop(); @@ -2278,6 +2279,10 @@ fn run_pocket_voice_stream( delivery: Some(delivery), }); } + player.ensure_healthy()?; + if player.is_empty() { + break; + } std::thread::sleep(Duration::from_millis(10)); } assistant_speech.take(); @@ -2415,6 +2420,10 @@ fn synthesize_pocket_stream_ready( if samples.is_empty() { return true; } + if let Err(error) = player.ensure_healthy() { + callback_error = Some(error); + return false; + } if let Err(error) = mark_pocket_playback_started( app, stream_id, @@ -2489,6 +2498,12 @@ fn synthesize_and_stream( if samples.is_empty() { return true; } + 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); @@ -2521,11 +2536,15 @@ fn synthesize_and_stream( player.stop(); return Ok(()); } - while !player.is_empty() { + loop { if !active.load(Ordering::SeqCst) { player.stop(); break; } + player.ensure_healthy()?; + if player.is_empty() { + break; + } std::thread::sleep(Duration::from_millis(10)); } Ok(()) From 3acd3d9b612438c8e9fd2f0db03f3f68d762a16f Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 25 Aug 2026 10:15:33 -0400 Subject: [PATCH 05/13] fix(voice): recheck native drain health --- src-tauri/src/commands/pocket_voice.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src-tauri/src/commands/pocket_voice.rs b/src-tauri/src/commands/pocket_voice.rs index a635c81a7..8b03a4c2f 100644 --- a/src-tauri/src/commands/pocket_voice.rs +++ b/src-tauri/src/commands/pocket_voice.rs @@ -2281,6 +2281,7 @@ fn run_pocket_voice_stream( } player.ensure_healthy()?; if player.is_empty() { + player.ensure_healthy()?; break; } std::thread::sleep(Duration::from_millis(10)); @@ -2543,6 +2544,7 @@ fn synthesize_and_stream( } player.ensure_healthy()?; if player.is_empty() { + player.ensure_healthy()?; break; } std::thread::sleep(Duration::from_millis(10)); From 07bbe24b186f36ab5d87611591e7a8865638faa8 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 25 Aug 2026 10:27:22 -0400 Subject: [PATCH 06/13] fix(voice): gate native playback imports --- src-tauri/src/commands/pocket_voice.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src-tauri/src/commands/pocket_voice.rs b/src-tauri/src/commands/pocket_voice.rs index 8b03a4c2f..e0425b681 100644 --- a/src-tauri/src/commands/pocket_voice.rs +++ b/src-tauri/src/commands/pocket_voice.rs @@ -12,7 +12,7 @@ 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}; @@ -32,8 +32,10 @@ use tauri::{AppHandle, Emitter, Manager, State}; #[cfg(target_os = "macos")] use super::pocket_audio_player::PocketAudioPlayer; +#[cfg(target_os = "macos")] +use super::native_voice::AssistantSpeechGuard; use super::{ - native_voice::{AssistantSpeechGuard, InterruptionSensitivity, NativeVoiceState}, + native_voice::{InterruptionSensitivity, NativeVoiceState}, voice_capture::VoiceCaptureState, }; use tokio::io::AsyncWriteExt; From ff2790cba00839d0f092b0f3d9b213eb5486ce04 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 25 Aug 2026 10:28:18 -0400 Subject: [PATCH 07/13] style(voice): format native playback imports --- src-tauri/src/commands/pocket_voice.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src-tauri/src/commands/pocket_voice.rs b/src-tauri/src/commands/pocket_voice.rs index e0425b681..43a972b12 100644 --- a/src-tauri/src/commands/pocket_voice.rs +++ b/src-tauri/src/commands/pocket_voice.rs @@ -30,10 +30,10 @@ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use tauri::{AppHandle, Emitter, Manager, State}; -#[cfg(target_os = "macos")] -use super::pocket_audio_player::PocketAudioPlayer; #[cfg(target_os = "macos")] use super::native_voice::AssistantSpeechGuard; +#[cfg(target_os = "macos")] +use super::pocket_audio_player::PocketAudioPlayer; use super::{ native_voice::{InterruptionSensitivity, NativeVoiceState}, voice_capture::VoiceCaptureState, From 17b726f42e72225690a0fdf6e9074d40204a79d5 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 25 Aug 2026 12:08:57 -0400 Subject: [PATCH 08/13] fix(voice): preserve Pocket output latency grace --- src-tauri/src/commands/pocket_voice.rs | 103 ++++++++++++++++++++++++- 1 file changed, 99 insertions(+), 4 deletions(-) diff --git a/src-tauri/src/commands/pocket_voice.rs b/src-tauri/src/commands/pocket_voice.rs index 43a972b12..ac9b03c79 100644 --- a/src-tauri/src/commands/pocket_voice.rs +++ b/src-tauri/src/commands/pocket_voice.rs @@ -2169,11 +2169,19 @@ fn run_pocket_voice_stream( let mut first_chunk_pending = true; let mut playback_started = false; 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, &mut assistant_speech); + update_pocket_assistant_speech( + &player, + &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(); @@ -2284,11 +2292,19 @@ fn run_pocket_voice_stream( player.ensure_healthy()?; if player.is_empty() { player.ensure_healthy()?; - break; + update_pocket_assistant_speech( + &player, + &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)); } - assistant_speech.take(); return Ok(PocketStreamOutcome { state: PocketStreamEventState::Completed, delivery: None, @@ -2352,12 +2368,37 @@ fn capture_before_stop( fn update_pocket_assistant_speech( player: &PocketAudioPlayer, assistant_speech: &mut Option, + playback_drained_at: &mut Option, + output_latency_grace: Duration, + now: Instant, ) { - if player.is_empty() { + if pocket_assistant_speech_grace_elapsed( + player.is_empty(), + 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(target_os = "macos")] #[allow(clippy::too_many_arguments)] fn mark_pocket_playback_started( @@ -3072,6 +3113,60 @@ mod tests { ); } + #[test] + 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, + )); + assert!(!pocket_assistant_speech_grace_elapsed( + false, + true, + &mut drained_at, + grace, + started + Duration::from_millis(300), + )); + assert_eq!(drained_at, None); + + assert!(!pocket_assistant_speech_grace_elapsed( + true, + true, + &mut drained_at, + grace, + started + Duration::from_millis(400), + )); + 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, + false, + &mut drained_at, + grace, + started + Duration::from_secs(1), + )); + assert_eq!(drained_at, None); + } + #[test] fn parakeet_progress_is_monotonic_through_extraction_and_verification() { let mut runtime = InstallRuntime::default(); From bdb8e35415815b2f5f0c10da2e5a0d9c6c342177 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 25 Aug 2026 12:11:25 -0400 Subject: [PATCH 09/13] fix(voice): restart grace on Pocket enqueue --- src-tauri/src/commands/pocket_voice.rs | 24 ++++++++++++++---------- 1 file changed, 14 insertions(+), 10 deletions(-) diff --git a/src-tauri/src/commands/pocket_voice.rs b/src-tauri/src/commands/pocket_voice.rs index ac9b03c79..4758247b2 100644 --- a/src-tauri/src/commands/pocket_voice.rs +++ b/src-tauri/src/commands/pocket_voice.rs @@ -2209,6 +2209,7 @@ fn run_pocket_voice_stream( interruption_sensitivity, suppress_capture, &mut assistant_speech, + &mut playback_drained_at, &mut delivery_ledger, &mut last_progress_emit, false, @@ -2238,6 +2239,7 @@ fn run_pocket_voice_stream( interruption_sensitivity, suppress_capture, &mut assistant_speech, + &mut playback_drained_at, &mut delivery_ledger, &mut last_progress_emit, true, @@ -2267,6 +2269,7 @@ fn run_pocket_voice_stream( interruption_sensitivity, suppress_capture, &mut assistant_speech, + &mut playback_drained_at, &mut delivery_ledger, &mut last_progress_emit, true, @@ -2399,6 +2402,11 @@ fn pocket_assistant_speech_grace_elapsed( now.saturating_duration_since(drained_at) >= output_latency_grace } +#[cfg(any(test, target_os = "macos"))] +fn note_pocket_audio_enqueued(playback_drained_at: &mut Option) { + *playback_drained_at = None; +} + #[cfg(target_os = "macos")] #[allow(clippy::too_many_arguments)] fn mark_pocket_playback_started( @@ -2441,6 +2449,7 @@ fn synthesize_pocket_stream_ready( interruption_sensitivity: InterruptionSensitivity, suppress_capture: bool, assistant_speech: &mut Option, + playback_drained_at: &mut Option, delivery_ledger: &mut PlaybackDeliveryLedger, last_progress_emit: &mut Instant, flush: bool, @@ -2484,6 +2493,7 @@ fn synthesize_pocket_stream_ready( callback_error = Some(error); return false; } + note_pocket_audio_enqueued(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 { @@ -3126,13 +3136,7 @@ mod tests { grace, started, )); - assert!(!pocket_assistant_speech_grace_elapsed( - false, - true, - &mut drained_at, - grace, - started + Duration::from_millis(300), - )); + note_pocket_audio_enqueued(&mut drained_at); assert_eq!(drained_at, None); assert!(!pocket_assistant_speech_grace_elapsed( @@ -3140,21 +3144,21 @@ mod tests { true, &mut drained_at, grace, - started + Duration::from_millis(400), + started + Duration::from_millis(600), )); assert!(!pocket_assistant_speech_grace_elapsed( true, true, &mut drained_at, grace, - started + Duration::from_millis(600), + started + Duration::from_millis(900), )); assert!(pocket_assistant_speech_grace_elapsed( true, true, &mut drained_at, grace, - started + Duration::from_millis(900), + started + Duration::from_millis(1_100), )); assert!(!pocket_assistant_speech_grace_elapsed( From eca09cd238741949289fbdd28f718a3826e89214 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 25 Aug 2026 12:26:42 -0400 Subject: [PATCH 10/13] fix(voice): use default time-pitch overlap --- src-tauri/native/siri_tts_bridge.m | 3 --- 1 file changed, 3 deletions(-) diff --git a/src-tauri/native/siri_tts_bridge.m b/src-tauri/native/siri_tts_bridge.m index a6f28e329..26c3632dd 100644 --- a/src-tauri/native/siri_tts_bridge.m +++ b/src-tauri/native/siri_tts_bridge.m @@ -605,9 +605,6 @@ - (instancetype)initWithSampleRate:(double)sampleRate _timePitch = [AVAudioUnitTimePitch new]; _timePitch.rate = rate; _timePitch.pitch = 0.0f; - // Apple documents that higher overlap values reduce artifacts at the - // cost of CPU. Voice playback is mono and short-lived, so prefer quality. - _timePitch.overlap = 32.0f; } _format = [[AVAudioFormat alloc] initWithCommonFormat:AVAudioPCMFormatFloat32 sampleRate:sampleRate From 5067ac79f627fe6c1c91de7c460effa674f8ce26 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 25 Aug 2026 12:31:32 -0400 Subject: [PATCH 11/13] fix(voice): bound native Pocket drain wait --- src-tauri/src/commands/pocket_audio_player.rs | 10 +- src-tauri/src/commands/pocket_voice.rs | 140 ++++++++++++++++-- 2 files changed, 137 insertions(+), 13 deletions(-) diff --git a/src-tauri/src/commands/pocket_audio_player.rs b/src-tauri/src/commands/pocket_audio_player.rs index 00c1813d4..9c067c108 100644 --- a/src-tauri/src/commands/pocket_audio_player.rs +++ b/src-tauri/src/commands/pocket_audio_player.rs @@ -78,10 +78,12 @@ impl PocketAudioPlayer { 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( - unsafe { berd_pocket_audio_player_completed_source_frames(self.raw) }, - self.delivery_safety_frames, - ) + 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 { diff --git a/src-tauri/src/commands/pocket_voice.rs b/src-tauri/src/commands/pocket_voice.rs index 4758247b2..90e0f82a1 100644 --- a/src-tauri/src/commands/pocket_voice.rs +++ b/src-tauri/src/commands/pocket_voice.rs @@ -62,6 +62,8 @@ 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(any(test, target_os = "macos"))] +const POCKET_SOURCE_COMPLETION_TIMEOUT: Duration = Duration::from_secs(2); #[cfg(target_os = "macos")] fn playback_latency_safety_duration_for_transport(transport: Option) -> Duration { @@ -277,6 +279,13 @@ impl PlaybackDeliveryLedger { } } + fn total_frames(&self) -> u64 { + self.segments + .iter() + .map(|(_, total_frames, _)| *total_frames) + .sum() + } + fn snapshot_consumed_frames(&self, consumed_frames: u64) -> VoiceDeliveryProgress { let mut segment_start = 0_u64; let segments = self @@ -2176,7 +2185,7 @@ fn run_pocket_voice_stream( let result: Result = (|| loop { update_pocket_assistant_speech( - &player, + player.is_empty(), &mut assistant_speech, &mut playback_drained_at, output_latency_grace, @@ -2283,6 +2292,13 @@ fn run_pocket_voice_stream( delivery: Some(delivery), }); } + 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); @@ -2292,11 +2308,29 @@ fn run_pocket_voice_stream( delivery: Some(delivery), }); } - player.ensure_healthy()?; - if player.is_empty() { + 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( - &player, + playback_drained, &mut assistant_speech, &mut playback_drained_at, output_latency_grace, @@ -2369,14 +2403,14 @@ fn capture_before_stop( #[cfg(target_os = "macos")] fn update_pocket_assistant_speech( - player: &PocketAudioPlayer, + playback_drained: bool, assistant_speech: &mut Option, playback_drained_at: &mut Option, output_latency_grace: Duration, now: Instant, ) { if pocket_assistant_speech_grace_elapsed( - player.is_empty(), + playback_drained, assistant_speech.is_some(), playback_drained_at, output_latency_grace, @@ -2403,10 +2437,46 @@ fn pocket_assistant_speech_grace_elapsed( } #[cfg(any(test, target_os = "macos"))] -fn note_pocket_audio_enqueued(playback_drained_at: &mut Option) { +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")] #[allow(clippy::too_many_arguments)] fn mark_pocket_playback_started( @@ -2493,7 +2563,7 @@ fn synthesize_pocket_stream_ready( callback_error = Some(error); return false; } - note_pocket_audio_enqueued(playback_drained_at); + 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 { @@ -3136,7 +3206,7 @@ mod tests { grace, started, )); - note_pocket_audio_enqueued(&mut drained_at); + reset_pocket_drain_grace(&mut drained_at); assert_eq!(drained_at, None); assert!(!pocket_assistant_speech_grace_elapsed( @@ -3171,6 +3241,58 @@ mod tests { 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!( + pocket_native_drain_status(false, timeout - Duration::from_millis(1), timeout), + PocketNativeDrainStatus::Waiting + ); + 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] fn parakeet_progress_is_monotonic_through_extraction_and_verification() { let mut runtime = InstallRuntime::default(); From 7b712eb2c270d0356489990e9319865e34618cc4 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 25 Aug 2026 13:50:49 -0400 Subject: [PATCH 12/13] chore: retrigger Pocket playback review From 01888b9ad77f79c923722ea1258f85975e350f78 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 25 Aug 2026 17:15:42 -0400 Subject: [PATCH 13/13] fix(voice): bound one-shot Pocket drain wait --- src-tauri/src/commands/pocket_voice.rs | 20 +++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/src-tauri/src/commands/pocket_voice.rs b/src-tauri/src/commands/pocket_voice.rs index 90e0f82a1..cfc652b79 100644 --- a/src-tauri/src/commands/pocket_voice.rs +++ b/src-tauri/src/commands/pocket_voice.rs @@ -2611,6 +2611,7 @@ fn synthesize_and_stream( 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_active = active.clone(); let callback_error_slot = callback_error.clone(); @@ -2634,6 +2635,7 @@ fn synthesize_and_stream( } 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() { @@ -2660,15 +2662,27 @@ fn synthesize_and_stream( player.stop(); return Ok(()); } + 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()?; - if player.is_empty() { - player.ensure_healthy()?; - break; + 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)); }