diff --git a/desktop/src-tauri/src/huddle/playout.rs b/desktop/src-tauri/src/huddle/playout.rs index 66c80f5d6c6..68c416a9766 100644 --- a/desktop/src-tauri/src/huddle/playout.rs +++ b/desktop/src-tauri/src/huddle/playout.rs @@ -183,23 +183,6 @@ fn same_occupancy( && index_to_epoch.get(&peer_idx) == Some(&epoch) } -fn mix_remote_stt_samples(mix: &mut Vec, samples: &[f32]) { - if mix.len() < samples.len() { - mix.resize(samples.len(), 0.0); - } - for (mixed, sample) in mix.iter_mut().zip(samples) { - *mixed = (*mixed + *sample).clamp(-1.0, 1.0); - } -} - -fn f32_samples_to_le_bytes(samples: &[f32]) -> Vec { - let mut bytes = Vec::with_capacity(std::mem::size_of_val(samples)); - for sample in samples { - bytes.extend_from_slice(&sample.to_le_bytes()); - } - bytes -} - /// One remote peer's slot: jitter buffer + dedicated rodio Player. /// /// Per-frame seq/timestamp come from the v2 wire header (sender-authored). @@ -280,7 +263,6 @@ pub(crate) async fn run_playout_recv_loop( tts_active: Arc, tts_cancel: Arc, local_tts_publishers: super::tts::LocalTtsPublishers, - remote_stt_pipeline: Arc>>>, agent_pubkeys: Arc>>, human_floor: HumanFloor, ) { @@ -338,7 +320,6 @@ pub(crate) async fn run_playout_recv_loop( // per idle peer into rodio forever. `is_active` is a 500 ms // grace past the last received packet, far longer than typical // DTX comfort-noise cadence. - let mut remote_stt_mix = Vec::new(); for (peer_idx, slot) in peers.iter_mut() { if !slot.is_active() { // Still drain the frame to keep NetEq's internal clock @@ -360,17 +341,10 @@ pub(crate) async fn run_playout_recv_loop( ); slot.player.skip_one(); } - if !is_locally_synthesized_peer(*peer_idx, &local_tts_publishers) { - let remote_agent = { - let agents = agent_pubkeys - .lock() - .unwrap_or_else(|error| error.into_inner()); - is_agent_peer(*peer_idx, &index_to_pubkey, &agents) - }; - if !remote_agent { - mix_remote_stt_samples(&mut remote_stt_mix, &samples); - } - } + // Remote peers are played out, never transcribed: + // this device signs transcripts with the local + // user's key, so another participant's speech has + // no honest path into that pipeline. slot.player.append(SamplesBuffer::new(channels, rate, samples)); } Err(e) => { @@ -380,18 +354,6 @@ pub(crate) async fn run_playout_recv_loop( } } } - if !remote_stt_mix.is_empty() { - let pipeline = remote_stt_pipeline - .lock() - .unwrap_or_else(|error| error.into_inner()) - .as_ref() - .and_then(std::sync::Weak::upgrade); - if let Some(pipeline) = pipeline { - let _ = pipeline.push_remote_audio(f32_samples_to_le_bytes( - &remote_stt_mix, - )); - } - } } _ = speaker_tick.tick() => { release_expired_remote_floors( @@ -776,8 +738,10 @@ mod tests { assert!(!is_locally_synthesized_peer(9, &local_publishers)); } + /// Agent audio plays out, but must never acquire the human floor — + /// otherwise one agent's speech would suppress another agent's response. #[test] - fn remote_agent_identity_is_excluded_from_human_stt() { + fn remote_agent_identity_is_excluded_from_the_human_floor() { let peers = std::collections::HashMap::from([(3, "human".to_owned()), (4, "AGENT".to_owned())]); let agents = vec!["agent".to_owned()]; @@ -812,16 +776,4 @@ mod tests { "frame for an unoccupied index is dropped" ); } - - #[test] - fn remote_human_stt_mix_sums_and_clamps_concurrent_speakers() { - let mut mix = Vec::new(); - mix_remote_stt_samples(&mut mix, &[0.4, -0.7, 0.2]); - mix_remote_stt_samples(&mut mix, &[0.8, -0.6, -0.1]); - - assert_eq!(mix, vec![1.0, -1.0, 0.1]); - let bytes = f32_samples_to_le_bytes(&mix); - assert_eq!(bytes.len(), std::mem::size_of_val(mix.as_slice())); - assert_eq!(f32::from_le_bytes(bytes[0..4].try_into().unwrap()), 1.0); - } } diff --git a/desktop/src-tauri/src/huddle/relay_api.rs b/desktop/src-tauri/src/huddle/relay_api.rs index 190397aa054..247bc9eb2f6 100644 --- a/desktop/src-tauri/src/huddle/relay_api.rs +++ b/desktop/src-tauri/src/huddle/relay_api.rs @@ -187,20 +187,12 @@ pub(crate) async fn connect_audio_relay( let keys = state.keys.lock().map_err(|e| e.to_string())?.clone(); // TTS interrupt flags — recv task cancels TTS when remote humans speak. - let ( - tts_cancel, - tts_active, - local_tts_publishers, - remote_stt_pipeline, - agent_pubkeys, - human_floor, - ) = { + let (tts_cancel, tts_active, local_tts_publishers, agent_pubkeys, human_floor) = { let hs = state.huddle()?; ( Arc::clone(&hs.tts_cancel), Arc::clone(&hs.tts_active), Arc::clone(&hs.local_tts_publishers), - Arc::clone(&hs.remote_stt_pipeline), Arc::clone(&hs.agent_pubkeys), hs.human_floor.clone(), ) @@ -233,7 +225,6 @@ pub(crate) async fn connect_audio_relay( tts_cancel, tts_active, local_tts_publishers, - remote_stt_pipeline, agent_pubkeys, human_floor, output_device_name, @@ -458,7 +449,6 @@ struct AudioRelayPipelineArgs { tts_cancel: Arc, tts_active: Arc, local_tts_publishers: super::tts::LocalTtsPublishers, - remote_stt_pipeline: Arc>>>, agent_pubkeys: Arc>>, human_floor: super::human_floor::HumanFloor, output_device_name: Option, @@ -475,7 +465,6 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String tts_cancel, tts_active, local_tts_publishers, - remote_stt_pipeline, agent_pubkeys, human_floor, output_device_name, @@ -588,7 +577,6 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String tts_active, tts_cancel, local_tts_publishers, - remote_stt_pipeline, agent_pubkeys, human_floor, )); diff --git a/desktop/src-tauri/src/huddle/state.rs b/desktop/src-tauri/src/huddle/state.rs index c7aff1bf7e2..ac0668553b9 100644 --- a/desktop/src-tauri/src/huddle/state.rs +++ b/desktop/src-tauri/src/huddle/state.rs @@ -7,7 +7,7 @@ use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; use std::sync::{ atomic::{AtomicBool, AtomicU64, Ordering}, - Arc, Mutex, Weak, + Arc, Mutex, }; use super::agent_voice::AgentVoiceSettings; @@ -79,12 +79,6 @@ pub struct HuddleState { /// Active STT pipeline — not serialized, not cloned. #[serde(skip)] pub stt_pipeline: Option>, - /// Weak STT handle shared with the audio receive loop so remote human - /// speech can reach transcription even when the pipeline hot-starts after - /// the Huddle audio socket was connected. The state-owned strong handle - /// above remains the sole owner and teardown clears both atomically. - #[serde(skip)] - pub remote_stt_pipeline: Arc>>>, /// Active TTS pipeline — not serialized, not cloned. #[serde(skip)] pub tts_pipeline: Option>, @@ -198,7 +192,6 @@ impl Clone for HuddleState { agent_pubkeys: Arc::new(Mutex::new(agent_pubkeys_snapshot)), agent_voice_settings: self.agent_voice_settings.clone(), stt_pipeline: None, // Never clone the pipeline handle. - remote_stt_pipeline: Arc::new(Mutex::new(None)), tts_pipeline: None, // Never clone the pipeline handle. local_tts_publishers: Arc::clone(&self.local_tts_publishers), is_creator: self.is_creator, @@ -235,7 +228,6 @@ impl Default for HuddleState { agent_pubkeys: Arc::new(Mutex::new(Vec::new())), agent_voice_settings: BTreeMap::new(), stt_pipeline: None, - remote_stt_pipeline: Arc::new(Mutex::new(None)), tts_pipeline: None, local_tts_publishers: tts::LocalTtsPublishers::default(), is_creator: false, @@ -258,19 +250,15 @@ impl Default for HuddleState { } impl HuddleState { + /// Install the STT pipeline. The state holds the only handle: the audio + /// receive loop must not reach it, because this pipeline's transcripts are + /// signed with the local user's key and it therefore accepts this device's + /// microphone alone. pub(crate) fn set_stt_pipeline(&mut self, pipeline: Arc) { - *self - .remote_stt_pipeline - .lock() - .unwrap_or_else(|error| error.into_inner()) = Some(Arc::downgrade(&pipeline)); self.stt_pipeline = Some(pipeline); } pub(crate) fn take_stt_pipeline(&mut self) -> Option> { - self.remote_stt_pipeline - .lock() - .unwrap_or_else(|error| error.into_inner()) - .take(); self.stt_pipeline.take() } diff --git a/desktop/src-tauri/src/huddle/stt.rs b/desktop/src-tauri/src/huddle/stt.rs index c27bf38b649..9accc3449d9 100644 --- a/desktop/src-tauri/src/huddle/stt.rs +++ b/desktop/src-tauri/src/huddle/stt.rs @@ -17,6 +17,12 @@ //! //! The worker runs on a dedicated `std::thread` (not async) because //! sherpa-onnx is CPU-bound and not Send-safe across await points. +//! +//! **Attribution invariant.** Every transcript this pipeline emits is signed +//! with the local user's key, so only this device's own microphone may enter +//! it. Audio from any other participant — decoded remote peers included — must +//! never reach `push_audio`: transcribing it here would publish one person's +//! speech as another's, on every listening desktop independently. use std::{ collections::VecDeque, @@ -53,26 +59,15 @@ const MAX_SPEECH_SAMPLES: usize = 16_000 * 30; /// task without holding a Mutex across await points. #[derive(Debug)] pub struct SttPipeline { - /// Send raw PCM bytes (f32 LE, 48 kHz mono) into the pipeline. - audio_tx: SyncSender, + /// Send raw PCM bytes (f32 LE, 48 kHz mono) from the local microphone + /// into the pipeline. See the attribution invariant in the module docs. + audio_tx: SyncSender>, /// Signals the worker thread to stop. shutdown: Arc, /// Worker thread handle — taken on drop to join cleanly. thread: Option>, } -#[derive(Debug)] -struct SttAudioInput { - pcm_bytes: Vec, - origin: SttAudioOrigin, -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -enum SttAudioOrigin { - Local, - RemoteHuman, -} - impl SttPipeline { /// Spawn the pipeline thread. /// @@ -102,7 +97,7 @@ impl SttPipeline { human_floor: HumanFloor, output_device: Option, ) -> Result<(Self, tokio_mpsc::Receiver), String> { - let (audio_tx, audio_rx) = mpsc::sync_channel::(AUDIO_QUEUE_DEPTH); + let (audio_tx, audio_rx) = mpsc::sync_channel::>(AUDIO_QUEUE_DEPTH); let (text_tx, text_rx) = tokio_mpsc::channel::(64); let shutdown = Arc::new(AtomicBool::new(false)); @@ -144,23 +139,15 @@ impl SttPipeline { self.thread.as_ref().is_none_or(|h| h.is_finished()) } - /// Feed raw PCM bytes into the pipeline. + /// Feed raw PCM bytes from **this device's microphone** into the pipeline. + /// + /// The only audio entry point, deliberately. Its output is signed with the + /// local user's key, so admitting another participant's audio here would + /// attribute their speech to this user — see the module docs. /// /// Non-blocking. Drops audio silently if the pipeline can't keep up — /// better to lose frames than to stall the UI thread. pub fn push_audio(&self, pcm_bytes: Vec) -> Result<(), String> { - self.push_audio_from(pcm_bytes, SttAudioOrigin::Local) - } - - /// Feed decoded remote-human PCM into transcription. Unlike the desktop - /// microphone path, this is not gated by the desktop PTT or mute state: the - /// remote participant already made their transmission choice on their own - /// device before the relay delivered these samples. - pub fn push_remote_audio(&self, pcm_bytes: Vec) -> Result<(), String> { - self.push_audio_from(pcm_bytes, SttAudioOrigin::RemoteHuman) - } - - fn push_audio_from(&self, pcm_bytes: Vec, origin: SttAudioOrigin) -> Result<(), String> { // Reject non-4-byte-aligned input — would silently truncate in bytes_to_f32. if !pcm_bytes.len().is_multiple_of(4) { return Err(format!( @@ -169,7 +156,7 @@ impl SttPipeline { )); } // Drop audio if the pipeline can't keep up — better than blocking the UI. - let _ = self.audio_tx.try_send(SttAudioInput { pcm_bytes, origin }); + let _ = self.audio_tx.try_send(pcm_bytes); Ok(()) } } @@ -405,11 +392,11 @@ impl SttStreamState { #[derive(Debug)] enum SttLoopInput { Tick, - Batch(Vec), + Batch(Vec>), } fn run_stt_receive_loop( - audio_rx: Receiver, + audio_rx: Receiver>, shutdown: &AtomicBool, human_floor: HumanFloor, mut process: impl FnMut(SttLoopInput, &mut local_barge_in::LocalBargeIn), @@ -443,7 +430,7 @@ fn run_stt_receive_loop( #[allow(clippy::too_many_arguments)] fn stt_worker( model_dir: PathBuf, - audio_rx: Receiver, + audio_rx: Receiver>, text_tx: tokio_mpsc::Sender, shutdown: Arc, ptt_active: Option>, @@ -488,9 +475,7 @@ fn stt_worker( } }; - // ── 2. Independent local and remote processing state ───────────────────── - // Separate resampler/VAD state prevents simultaneous desktop and remote - // speech from being serialized into one artificial utterance. + // ── 2. Local microphone processing state ───────────────────────────────── let mut local_stream = match SttStreamState::new() { Ok(stream) => stream, Err(error) => { @@ -498,13 +483,6 @@ fn stt_worker( return; } }; - let mut remote_stream = match SttStreamState::new() { - Ok(stream) => stream, - Err(error) => { - eprintln!("buzz-desktop: {error}"); - return; - } - }; let speculative_enabled = stt_speculative_decode(); let mut transmit_was_active = ptt_active .as_ref() @@ -546,28 +524,18 @@ fn stt_worker( } } SttLoopInput::Batch(batch) => { - for input in batch { - let (stream, ptt_gate, manual_gate, track_local_floor) = match input.origin { - SttAudioOrigin::Local => ( - &mut local_stream, - ptt_active.as_ref(), - manual_mic_unmuted.as_ref(), - true, - ), - SttAudioOrigin::RemoteHuman => (&mut remote_stream, None, None, false), - }; + for pcm_bytes in batch { process_stt_input( - stream, - &input.pcm_bytes, + &mut local_stream, + &pcm_bytes, speculative_enabled, &recognizer, &text_tx, - ptt_gate, - manual_gate, + ptt_active.as_ref(), + manual_mic_unmuted.as_ref(), &human_floor, local_barge_in_state, output_device.as_deref(), - track_local_floor, ); } } @@ -591,7 +559,6 @@ fn process_stt_input( human_floor: &HumanFloor, local_barge_in_state: &mut local_barge_in::LocalBargeIn, output_device: Option<&str>, - track_local_floor: bool, ) { stream .input_buf_48k @@ -614,7 +581,6 @@ fn process_stt_input( human_floor, local_barge_in_state, output_device, - track_local_floor, ); } } @@ -671,7 +637,6 @@ fn process_16k_samples( human_floor: &HumanFloor, local_barge_in_state: &mut local_barge_in::LocalBargeIn, output_device: Option<&str>, - track_local_floor: bool, ) { let (speculative_enabled, speculative) = speculative; leftover.extend_from_slice(samples); @@ -692,20 +657,17 @@ fn process_16k_samples( endpoint.process_frame(frame, prob, accepts_audio, flush_allowed, flush_frames); // Open-mic VAD semantics also apply when a PTT-mode user manually // opens the mic. A held shortcut keeps its explicit key-down cancel. - let local_barge_in = track_local_floor - && local_barge_in::enabled(ptt_active.is_some(), manually_open, ptt_held); - if track_local_floor { - if local_barge_in { - local_barge_in_state.observe( - prob, - action == VadFrameAction::ConfirmedOnset, - human_floor, - output_device, - VAD_ONSET_THRESHOLD, - ); - } else { - local_barge_in_state.release(human_floor); - } + let local_barge_in = local_barge_in::enabled(ptt_active.is_some(), manually_open, ptt_held); + if local_barge_in { + local_barge_in_state.observe( + prob, + action == VadFrameAction::ConfirmedOnset, + human_floor, + output_device, + VAD_ONSET_THRESHOLD, + ); + } else { + local_barge_in_state.release(human_floor); } match action { diff --git a/desktop/src-tauri/src/huddle/stt_tests.rs b/desktop/src-tauri/src/huddle/stt_tests.rs index d7425970ced..3fba9adafa6 100644 --- a/desktop/src-tauri/src/huddle/stt_tests.rs +++ b/desktop/src-tauri/src/huddle/stt_tests.rs @@ -1,9 +1,9 @@ use std::sync::{atomic::AtomicBool, mpsc, Arc, Barrier}; use super::{ - has_enough_voiced_audio, run_stt_receive_loop, vad_flush_allowed, HumanFloor, SttAudioInput, - SttAudioOrigin, SttLoopInput, VadEndpoint, VadFrameAction, MIN_VOICED_FRAMES, - SILENCE_FLUSH_FRAMES, VAD_FRAME_SAMPLES, VAD_ONSET_FRAMES, VAD_PRE_ROLL_FRAMES, + has_enough_voiced_audio, run_stt_receive_loop, vad_flush_allowed, HumanFloor, SttLoopInput, + VadEndpoint, VadFrameAction, MIN_VOICED_FRAMES, SILENCE_FLUSH_FRAMES, VAD_FRAME_SAMPLES, + VAD_ONSET_FRAMES, VAD_PRE_ROLL_FRAMES, }; #[derive(Clone, Copy)] @@ -34,12 +34,7 @@ fn assert_worker_exit_releases_floor(exit: WorkerExit) { ); }); - audio_tx - .send(SttAudioInput { - pcm_bytes: Vec::new(), - origin: SttAudioOrigin::Local, - }) - .expect("worker receiver is open"); + audio_tx.send(Vec::new()).expect("worker receiver is open"); acquired.wait(); assert!(human_floor.is_blocked()); match exit {