Skip to content

fix(desktop): transcribe only the local microphone in Huddles #7043

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hr-o wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(desktop): transcribe only the local microphone in Huddles #7043

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 7 additions & 55 deletions desktop/src-tauri/src/huddle/playout.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,23 +183,6 @@ fn same_occupancy(
&& index_to_epoch.get(&peer_idx) == Some(&epoch)
}

fn mix_remote_stt_samples(mix: &mut Vec<f32>, 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<u8> {
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).
Expand Down Expand Up @@ -280,7 +263,6 @@ pub(crate) async fn run_playout_recv_loop(
tts_active: Arc<AtomicBool>,
tts_cancel: Arc<AtomicBool>,
local_tts_publishers: super::tts::LocalTtsPublishers,
remote_stt_pipeline: Arc<std::sync::Mutex<Option<std::sync::Weak<super::stt::SttPipeline>>>>,
agent_pubkeys: Arc<std::sync::Mutex<Vec<String>>>,
human_floor: HumanFloor,
) {
Expand Down Expand Up @@ -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
Expand All @@ -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) => {
Expand All @@ -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(
Expand Down Expand Up @@ -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()];
Expand Down Expand Up @@ -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);
}
}
14 changes: 1 addition & 13 deletions desktop/src-tauri/src/huddle/relay_api.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
)
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -458,7 +449,6 @@ struct AudioRelayPipelineArgs {
tts_cancel: Arc<AtomicBool>,
tts_active: Arc<AtomicBool>,
local_tts_publishers: super::tts::LocalTtsPublishers,
remote_stt_pipeline: Arc<std::sync::Mutex<Option<std::sync::Weak<super::stt::SttPipeline>>>>,
agent_pubkeys: Arc<std::sync::Mutex<Vec<String>>>,
human_floor: super::human_floor::HumanFloor,
output_device_name: Option<String>,
Expand All @@ -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,
Expand Down Expand Up @@ -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,
));
Expand Down
22 changes: 5 additions & 17 deletions desktop/src-tauri/src/huddle/state.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -79,12 +79,6 @@ pub struct HuddleState {
/// Active STT pipeline — not serialized, not cloned.
#[serde(skip)]
pub stt_pipeline: Option<Arc<stt::SttPipeline>>,
/// 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<Mutex<Option<Weak<stt::SttPipeline>>>>,
/// Active TTS pipeline — not serialized, not cloned.
#[serde(skip)]
pub tts_pipeline: Option<Arc<tts::TtsPipeline>>,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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<stt::SttPipeline>) {
*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<Arc<stt::SttPipeline>> {
self.remote_stt_pipeline
.lock()
.unwrap_or_else(|error| error.into_inner())
.take();
self.stt_pipeline.take()
}

Expand Down
Loading