diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index e38fa9b83e4..2efacce2b19 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -958,14 +958,19 @@ async fn resolve_new_session_channel_context( /// On error from `session_new_full()`, returns the `AcpError` — caller handles /// error reporting. Model-switch failures are logged and gracefully ignored /// (the agent proceeds with its default model). +struct NewSessionChannelContext<'a> { + huddle_instructions: Option<&'a str>, + canvas: Option<&'a str>, + name: Option<&'a str>, + id: Option, + channel_type: Option<&'a str>, +} + async fn create_session_and_apply_model( agent: &mut OwnedAgent, ctx: &PromptContext, agent_core: Option<&str>, - agent_canvas: Option<&str>, - channel_name: Option<&str>, - channel_id: Option, - channel_type: Option<&str>, + channel: NewSessionChannelContext<'_>, ) -> Result { // Build base_prompt + system_prompt + agent core + canvas metadata into a // single prompt. Standard protocol-v2 agents receive it in `session/new`; @@ -975,24 +980,27 @@ async fn create_session_and_apply_model( // `[Channel Canvas]` header; both are appended with a blank-line separator. let is_goose = agent.agent_name == "goose"; let combined_system_prompt = with_canvas( - with_core( - with_team( - framed_system_prompt(&ctx.cwd, ctx.base_prompt, ctx.system_prompt.as_deref()), - ctx.team_instructions.as_deref(), + with_huddle_instructions( + with_core( + with_team( + framed_system_prompt(&ctx.cwd, ctx.base_prompt, ctx.system_prompt.as_deref()), + ctx.team_instructions.as_deref(), + ), + agent_core, ), - agent_core, + channel.huddle_instructions, ), - agent_canvas, + channel.canvas, ); let session_title = ctx .session_title .as_deref() - .map(|agent_name| compose_session_title(agent_name, channel_name)); + .map(|agent_name| compose_session_title(agent_name, channel.name)); let mcp_servers = mcp_servers_with_git_origin( &ctx.mcp_servers, - channel_id, - channel_type, + channel.id, + channel.channel_type, ctx.session_title.as_deref(), ); @@ -1394,6 +1402,21 @@ fn with_core(framed: Option, core: Option<&str>) -> Option { } } +/// Append owner-signed huddle instructions to this channel session's system prompt. +fn with_huddle_instructions(prompt: Option, instructions: Option<&str>) -> Option { + let instructions = instructions + .map(str::trim) + .filter(|value| !value.is_empty()); + match (prompt, instructions) { + (Some(prompt), Some(instructions)) => { + Some(format!("{prompt}\n\n[Huddle Instructions]\n{instructions}")) + } + (None, Some(instructions)) => Some(format!("[Huddle Instructions]\n{instructions}")), + (Some(prompt), None) => Some(prompt), + (None, None) => None, + } +} + /// Append the `[Channel Canvas]` metadata section onto the accumulated system prompt. /// /// The canvas section already carries its `[Channel Canvas]` header (from @@ -1616,6 +1639,7 @@ pub async fn run_prompt_task( // prevents a stale revision A surviving a failed create and being re-used by // the next attempt after the canvas was cleared. let mut pending_canvas: Option<(Uuid, String)> = None; + let mut huddle_instructions: Option = None; // Channel name for the session title, from the same single resolve the // canvas DM check uses — see `resolve_new_session_channel_context`. let mut title_channel: Option = None; @@ -1628,6 +1652,10 @@ pub async fn run_prompt_task( resolve_new_session_channel_context(&ctx.channel_info, *cid).await; title_channel = resolved_channel; origin_channel_type = resolved_channel_type; + if let Some(owner) = ctx.agent_owner_pubkey.as_ref() { + huddle_instructions = + fetch_huddle_instructions(*cid, owner, &ctx.rest_client).await; + } // A confirmed DM never receives a canvas section; an undeterminable // channel type fails closed as a DM for the same reason. if needs_canvas && !is_dm { @@ -1670,10 +1698,13 @@ pub async fn run_prompt_task( &mut agent, &ctx, agent_core.as_deref(), - agent_canvas.as_deref(), - title_channel.as_deref(), - Some(*cid), - origin_channel_type.as_deref(), + NewSessionChannelContext { + huddle_instructions: huddle_instructions.as_deref(), + canvas: agent_canvas.as_deref(), + name: title_channel.as_deref(), + id: Some(*cid), + channel_type: origin_channel_type.as_deref(), + }, ) .await { @@ -1728,8 +1759,19 @@ pub async fn run_prompt_task( if let Some(sid) = &agent.state.heartbeat_session { (sid.clone(), false) } else { - match create_session_and_apply_model(&mut agent, &ctx, None, None, None, None, None) - .await + match create_session_and_apply_model( + &mut agent, + &ctx, + None, + NewSessionChannelContext { + huddle_instructions: None, + canvas: None, + name: None, + id: None, + channel_type: None, + }, + ) + .await { Ok(sid) => { tracing::info!( @@ -1798,6 +1840,7 @@ pub async fn run_prompt_task( system_prompt: ctx.system_prompt.as_deref(), team_instructions: ctx.team_instructions.as_deref(), agent_core: agent_core.as_deref(), + huddle_instructions: huddle_instructions.as_deref(), agent_canvas: agent_canvas.as_deref(), }; // Delivery state is committed only after ACP confirms success. Existing @@ -2056,6 +2099,7 @@ pub async fn run_prompt_task( b, &crate::queue::FormatPromptArgs { agent_core: standing.agent_core, + huddle_instructions: standing.huddle_instructions, channel_info: channel_info.as_ref(), conversation_context: conversation_context.as_ref(), conversation_context_had_delivered_events, @@ -2638,6 +2682,67 @@ pub(crate) async fn fetch_channel_info( .await } +/// Fetch owner-signed huddle instructions for a new channel session. +/// +/// The event is promoted into the system role, so accepting any channel member's +/// event would be a privilege escalation. Only the configured agent owner's +/// valid signature is accepted; absence or failure simply yields no section. +async fn fetch_huddle_instructions( + channel_id: Uuid, + owner: &nostr::PublicKey, + rest: &RestClient, +) -> Option { + use nostr::{Alphabet, SingleLetterTag}; + + let h_tag = SingleLetterTag::lowercase(Alphabet::H); + let filter = nostr::Filter::new() + .kind(nostr::Kind::Custom( + buzz_core::kind::KIND_HUDDLE_GUIDELINES as u16, + )) + .author(*owner) + .custom_tags(h_tag, [channel_id.to_string()]) + .limit(1); + let json = match timeout( + CONTEXT_FETCH_TIMEOUT, + rest.query(std::slice::from_ref(&filter)), + ) + .await + { + Ok(Ok(json)) => json, + Ok(Err(error)) => { + tracing::warn!(channel = %channel_id, "huddle instructions query failed: {error}"); + return None; + } + Err(_) => { + tracing::warn!(channel = %channel_id, "huddle instructions query timed out"); + return None; + } + }; + huddle_instructions_from_query_response(json.as_array()?, channel_id, owner) +} + +fn huddle_instructions_from_query_response( + events: &[serde_json::Value], + channel_id: Uuid, + owner: &nostr::PublicKey, +) -> Option { + let raw = events.first()?; + let event = serde_json::from_value::(raw.clone()).ok()?; + event.verify().ok()?; + let channel_id = channel_id.to_string(); + if event.pubkey != *owner + || event.kind.as_u16() as u32 != buzz_core::kind::KIND_HUDDLE_GUIDELINES + || !event + .tags + .iter() + .any(|tag| tag.kind().to_string() == "h" && tag.content() == Some(channel_id.as_str())) + { + return None; + } + let content = event.content.trim(); + (!content.is_empty()).then(|| content.to_owned()) +} + /// Fetch the latest canvas event for `channel_id` and return a rendered /// `[Channel Canvas]` metadata section, or `None` if absent/blank/error. /// @@ -4451,6 +4556,7 @@ mod tests { system_prompt: Some("you are Eva"), team_instructions: Some("ship small"), agent_core: Some("[Agent Memory — core]\nremember this"), + huddle_instructions: Some("reply immediately"), agent_canvas: Some("[Channel Canvas]\ncanvas content"), } } @@ -4466,6 +4572,7 @@ mod tests { "[System]", "[Team Instructions]", "[Agent Memory — core]", + "[Huddle Instructions]", "[Channel Canvas]", "do the thing", ] @@ -7496,6 +7603,59 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" } } + // ── huddle instructions ───────────────────────────────────────────────── + + #[test] + fn huddle_instructions_append_as_system_section() { + assert_eq!( + with_huddle_instructions(Some("base".into()), Some(" reply now ")).as_deref(), + Some("base\n\n[Huddle Instructions]\nreply now") + ); + } + + #[test] + fn huddle_instructions_require_owner_signature_and_channel() { + let owner = Keys::generate(); + let stranger = Keys::generate(); + let channel = Uuid::parse_str("00f1ccaf-1506-4dd7-9a0e-fa67e9e486ae").unwrap(); + let event = |keys: &Keys, channel_id: Uuid| { + let channel_id = channel_id.to_string(); + let h_tag = Tag::parse(["h", channel_id.as_str()]).unwrap(); + serde_json::to_value( + EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_HUDDLE_GUIDELINES as u16), + "reply immediately", + ) + .tags([h_tag]) + .sign_with_keys(keys) + .unwrap(), + ) + .unwrap() + }; + + assert_eq!( + huddle_instructions_from_query_response( + &[event(&owner, channel)], + channel, + &owner.public_key(), + ) + .as_deref(), + Some("reply immediately") + ); + assert!(huddle_instructions_from_query_response( + &[event(&stranger, channel)], + channel, + &owner.public_key(), + ) + .is_none()); + assert!(huddle_instructions_from_query_response( + &[event(&owner, Uuid::new_v4())], + channel, + &owner.public_key(), + ) + .is_none()); + } + // ── render_canvas_section ──────────────────────────────────────────────── #[test] diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index dabee13afd5..b0f0fa248e3 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -1448,6 +1448,8 @@ fn format_conversation_context( #[derive(Default)] pub struct FormatPromptArgs<'a> { pub agent_core: Option<&'a str>, + /// Owner-signed instructions for an active huddle channel. + pub huddle_instructions: Option<&'a str>, pub channel_info: Option<&'a PromptChannelInfo>, pub conversation_context: Option<&'a ConversationContext>, /// True when delivery-delta filtering removed at least one event that this @@ -1496,13 +1498,14 @@ pub(crate) struct StandingContext<'a> { pub system_prompt: Option<&'a str>, pub team_instructions: Option<&'a str>, pub agent_core: Option<&'a str>, + pub huddle_instructions: Option<&'a str>, pub agent_canvas: Option<&'a str>, } impl StandingContext<'_> { /// Render the sections in the order legacy agents have always seen them. pub(crate) fn sections(&self) -> Vec { - let mut sections = Vec::with_capacity(5); + let mut sections = Vec::with_capacity(6); if let Some(bp) = self.base_prompt { sections.push(base_section(bp)); } @@ -1519,6 +1522,13 @@ impl StandingContext<'_> { if let Some(core) = self.agent_core { sections.push(core.to_string()); } + if let Some(instructions) = self + .huddle_instructions + .map(str::trim) + .filter(|value| !value.is_empty()) + { + sections.push(format!("[Huddle Instructions]\n{instructions}")); + } if let Some(canvas) = self.agent_canvas { sections.push(canvas.to_string()); } @@ -1587,6 +1597,7 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec usize { + std::env::var("BUZZ_TTS_THREADS") + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|&n| n >= 1) + .unwrap_or(TTS_NUM_THREADS) +} + /// Loaded reference voice samples and their original sample rate. #[derive(Debug, Clone)] pub struct VoiceStyle { @@ -83,13 +93,13 @@ pub fn load_text_to_speech(model_dir: &str) -> Result { } } Ok(PocketTts { - inner: Mutex::new(AprilPocketTts::load(&dir, TTS_NUM_THREADS)?), + inner: Mutex::new(AprilPocketTts::load(&dir, tts_num_threads())?), }) } impl PocketTts { - /// Split text into synthesis units that satisfy the bundle's exact - /// 50-token input limit. + /// Split text into model-safe synthesis units that satisfy the bundle's + /// exact 50-token input limit, packing sentences whenever they fit. pub fn split_text_into_chunks(&self, text: &str) -> Result, String> { let Some(prepared) = prepare_april_prompt(text) else { return Ok(Vec::new()); @@ -100,6 +110,23 @@ impl PocketTts { .split_prompt(&prepared) } + /// Split text into ordered playback units, keeping the first sentence + /// separate so it reaches synthesis before the remainder is packed. + /// + /// Units are contiguous substrings of the prepared model prompt and may + /// retain boundary whitespace. Concatenating them with `chunks.concat()` + /// reconstructs that prompt exactly, and each unit's prepared token count + /// is at most 50. + pub fn split_text_for_playback(&self, text: &str) -> Result, String> { + let Some(prepared) = prepare_april_prompt(text) else { + return Ok(Vec::new()); + }; + self.inner + .lock() + .map_err(|_| "Pocket TTS engine lock poisoned".to_string())? + .split_playback_prompt(&prepared) + } + /// Synthesize text with the supplied reference voice. /// /// Pocket detects language from text and this model uses one synthesis @@ -127,6 +154,36 @@ impl PocketTts { } Ok(samples) } + + /// EXPERIMENTAL (latency): streaming synthesis. Invokes `on_audio` with + /// PCM deltas as soon as roughly `emit_frames` Flow LM frames (80 ms of + /// audio each) have been generated and decoded. Concatenated deltas equal + /// one `synth_chunk` result. The callback runs on the caller thread and + /// returns `false` to cancel; the function then returns Ok(false). + pub fn synth_chunk_streaming( + &self, + text: &str, + style: &VoiceStyle, + emit_frames: usize, + on_audio: &mut dyn FnMut(Vec) -> bool, + ) -> Result { + let Some(prepared) = prepare_april_prompt(text) else { + return Ok(true); + }; + let mut engine = self + .inner + .lock() + .map_err(|_| "Pocket TTS engine lock poisoned".to_string())?; + let chunks = engine.split_prompt(&prepared)?; + for chunk in chunks { + let prepared = prepare_april_prompt(&chunk) + .ok_or_else(|| "Pocket TTS prompt chunk became empty".to_string())?; + if !engine.synth_chunk_streaming(&prepared, style, emit_frames, on_audio)? { + return Ok(false); + } + } + Ok(true) + } } #[cfg(test)] @@ -148,6 +205,92 @@ mod tests { .any(|artifact| artifact.filename == "flow_lm_main.onnx")); } + /// Which splitter each production function delegates to, across the whole + /// file rather than one hand-picked window. + /// + /// A wrong delegation can reinstate either shipped defect in one token: + /// removing first-sentence priority from playback, or re-isolating sentence + /// one inside units that already fit. Asserting the whole map means a new + /// delegation must be declared here to compile green. + fn splitter_delegations(source: &str) -> Vec<(String, Vec)> { + let production = source + .split_once("\n#[cfg(test)]") + .map_or(source, |(production, _)| production); + // Scan code only. Prose cannot call a splitter, but it can contain + // ` fn `, which would end a body early and hide a call after it, and it + // can name a splitter, which would report a call the code never makes. + let production: String = production + .lines() + .map(|line| line.split_once("//").map_or(line, |(code, _)| code)) + .collect::>() + .join("\n"); + let mut out = Vec::new(); + let mut rest = production.as_str(); + while let Some((_, after)) = rest.split_once(" fn ") { + let (name, body) = after + .split_once('(') + .expect("a function signature has an argument list"); + // End at this function's own closing brace, not at the next ` fn `: + // a body provably stops where its braces balance, so no later + // function's calls are attributed here and none of this one's are + // dropped. + let inner = body.split_once('{').map_or("", |(_, inner)| inner); + let mut depth = 1usize; + let body = inner + .char_indices() + .find(|&(_, ch)| { + depth = match ch { + '{' => depth + 1, + '}' => depth - 1, + _ => depth, + }; + depth == 0 + }) + .map_or(inner, |(end, _)| &inner[..end]); + let mut calls = Vec::new(); + // Check the isolating spelling first: ".split_prompt(" is a + // substring of neither, but a naive contains() on the shorter name + // would also match the longer one. + for _ in 0..body.matches(".split_playback_prompt(").count() { + calls.push("split_playback_prompt".to_string()); + } + let plain = body.matches(".split_prompt(").count(); + for _ in 0..plain { + calls.push("split_prompt".to_string()); + } + if !calls.is_empty() { + out.push((name.trim().to_string(), calls)); + } + rest = after; + } + out + } + + #[test] + fn every_production_splitter_delegation_is_declared() { + let source = include_str!("pocket.rs"); + let actual = splitter_delegations(source); + let expected: Vec<(String, Vec)> = vec![ + // Model units: pack sentences, never isolate. + ("split_text_into_chunks".into(), vec!["split_prompt".into()]), + // Playback units: isolate sentence one for time-to-first-audio. + ( + "split_text_for_playback".into(), + vec!["split_playback_prompt".into()], + ), + // Synthesis receives an already-packed unit: re-isolating here + // re-adds the per-sentence seam this PR removes. + ("synth_chunk".into(), vec!["split_prompt".into()]), + ("synth_chunk_streaming".into(), vec!["split_prompt".into()]), + ]; + assert_eq!( + actual, expected, + "a production function changed which splitter it calls (or a new \ + one appeared); isolating outside split_text_for_playback delays \ + first audio, packing inside it removes the guarantee" + ); + } + #[test] #[ignore = "requires BUZZ_POCKET_TEST_MODEL_DIR"] fn production_api_emits_non_silent_april_int8_pcm() { diff --git a/crates/buzz-voice/src/pocket_april.rs b/crates/buzz-voice/src/pocket_april.rs index 43826df5c99..9ace5001daa 100644 --- a/crates/buzz-voice/src/pocket_april.rs +++ b/crates/buzz-voice/src/pocket_april.rs @@ -36,6 +36,13 @@ const DECODER_CHUNK_FRAMES: usize = 12; const TOKENS_PER_SECOND_ESTIMATE: f32 = 3.0; const GENERATION_SECONDS_PADDING: f32 = 2.0; +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum TextBoundary { + Sentence, + Clause, + Word, +} + #[derive(Debug, Deserialize)] struct Bundle { schema_version: u32, @@ -89,13 +96,129 @@ struct StateValue { value: DynValue, } -struct CachedVoice { - samples_ptr: usize, +/// Stable identity for a reference voice: a content hash of the sample +/// buffer plus its length and rate. Buffer addresses are NOT part of the +/// key — voice switching clones and drops sample buffers, so the allocator +/// can hand a different voice the same address, and an address-based key +/// would then restore the previous voice's cached state. +#[derive(PartialEq, Eq, Clone, Copy, Debug)] +struct VoiceKey { + content_hash: u64, samples_len: usize, sample_rate: i32, +} + +fn voice_key(style: &VoiceStyle) -> VoiceKey { + use std::hash::Hasher; + let mut hasher = std::hash::DefaultHasher::new(); + for sample in &style.samples { + hasher.write_u32(sample.to_bits()); + } + VoiceKey { + content_hash: hasher.finish(), + samples_len: style.samples.len(), + sample_rate: style.sample_rate, + } +} + +struct CachedVoice { + key: VoiceKey, embeddings: Vec, } +/// EXPERIMENTAL (latency): a dtype-tagged copy of one recurrent state tensor, +/// used to snapshot the Flow LM state right after voice conditioning so +/// subsequent chunks skip the ~160 ms `condition_voice` pass entirely. +enum SnapshotTensor { + F32(Vec, Vec), + I64(Vec, Vec), + Bool(Vec, Vec), +} + +struct CachedConditioning { + key: VoiceKey, + state: Vec<(StateSpec, SnapshotTensor)>, +} + +fn snapshot_state(state: &[StateValue]) -> Result, String> { + state + .iter() + .map(|value| { + let tensor = match value.spec.dtype { + StateDtype::Float32 => { + let (shape, data) = value + .value + .try_extract_tensor::() + .map_err(ort_error("snapshot f32 state"))?; + SnapshotTensor::F32(shape.to_vec(), data.to_vec()) + } + StateDtype::Int64 => { + let (shape, data) = value + .value + .try_extract_tensor::() + .map_err(ort_error("snapshot i64 state"))?; + SnapshotTensor::I64(shape.to_vec(), data.to_vec()) + } + StateDtype::Bool => { + let (shape, data) = value + .value + .try_extract_tensor::() + .map_err(ort_error("snapshot bool state"))?; + SnapshotTensor::Bool(shape.to_vec(), data.to_vec()) + } + }; + Ok((value.spec.clone(), tensor)) + }) + .collect() +} + +fn restore_state(snapshot: &[(StateSpec, SnapshotTensor)]) -> Result, String> { + snapshot + .iter() + .map(|(spec, tensor)| { + let value = match tensor { + SnapshotTensor::F32(shape, data) => { + if data.is_empty() { + Tensor::::new(&ort::memory::Allocator::default(), shape.clone()) + .map_err(ort_error("restore empty f32 state"))? + .into_dyn() + } else { + Tensor::from_array((shape.clone(), data.clone().into_boxed_slice())) + .map_err(ort_error("restore f32 state"))? + .into_dyn() + } + } + SnapshotTensor::I64(shape, data) => { + if data.is_empty() { + Tensor::::new(&ort::memory::Allocator::default(), shape.clone()) + .map_err(ort_error("restore empty i64 state"))? + .into_dyn() + } else { + Tensor::from_array((shape.clone(), data.clone().into_boxed_slice())) + .map_err(ort_error("restore i64 state"))? + .into_dyn() + } + } + SnapshotTensor::Bool(shape, data) => { + if data.is_empty() { + Tensor::::new(&ort::memory::Allocator::default(), shape.clone()) + .map_err(ort_error("restore empty bool state"))? + .into_dyn() + } else { + Tensor::from_array((shape.clone(), data.clone().into_boxed_slice())) + .map_err(ort_error("restore bool state"))? + .into_dyn() + } + } + }; + Ok(StateValue { + spec: spec.clone(), + value, + }) + }) + .collect() +} + pub(crate) struct AprilPocketTts { bundle: Bundle, tokenizer: Tokenizer, @@ -106,6 +229,10 @@ pub(crate) struct AprilPocketTts { flow: Session, mimi_decoder: Session, cached_voice: Option, + /// EXPERIMENTAL (latency): post-`condition_voice` Flow LM state, cached + /// per reference voice. Restoring it replaces the ~160 ms conditioning + /// pass on every chunk after the first for a given voice. + cached_conditioning: Option, } #[derive(Debug, Clone, PartialEq)] @@ -239,6 +366,7 @@ impl AprilPocketTts { tokenizer, bos_embedding, cached_voice: None, + cached_conditioning: None, }) } @@ -246,62 +374,23 @@ impl AprilPocketTts { &self, prepared: &AprilPreparedPrompt, ) -> Result, String> { - if self.token_count(&prepared.text)? <= self.bundle.max_token_per_chunk { + if self.prepared_token_count(&prepared.text)? <= self.bundle.max_token_per_chunk { return Ok(vec![prepared.text.clone()]); } + split_model_at_natural_boundaries(&prepared.text, self.bundle.max_token_per_chunk, |text| { + self.prepared_token_count(text) + }) + } - let mut chunks = Vec::new(); - let mut current = String::new(); - for word in prepared.text.split_whitespace() { - let candidate = if current.is_empty() { - word.to_string() - } else { - format!("{current} {word}") - }; - if self.prepared_token_count(&candidate)? <= self.bundle.max_token_per_chunk { - current = candidate; - continue; - } - if !current.is_empty() { - chunks.push(std::mem::take(&mut current)); - } - - if self.prepared_token_count(word)? <= self.bundle.max_token_per_chunk { - current = word.to_string(); - continue; - } - - let mut fragment = String::new(); - for ch in word.chars() { - let candidate = format!("{fragment}{ch}"); - if !fragment.is_empty() - && self.prepared_token_count(&candidate)? > self.bundle.max_token_per_chunk - { - chunks.push(std::mem::take(&mut fragment)); - } - fragment.push(ch); - } - current = fragment; - } - if !current.is_empty() { - chunks.push(current); - } - - chunks - .into_iter() - .map(|text| { - let chunk = prepare_april_prompt(&text) - .ok_or_else(|| "Pocket TTS prompt chunk became empty".to_string())?; - let token_count = self.token_count(&chunk.text)?; - if token_count > self.bundle.max_token_per_chunk { - return Err(format!( - "Pocket TTS prompt chunk has {token_count} tokens; maximum is {}", - self.bundle.max_token_per_chunk - )); - } - Ok(chunk.text) - }) - .collect() + pub(crate) fn split_playback_prompt( + &self, + prepared: &AprilPreparedPrompt, + ) -> Result, String> { + split_playback_at_natural_boundaries( + &prepared.text, + self.bundle.max_token_per_chunk, + |text| self.prepared_token_count(text), + ) } pub(crate) fn synth_chunk( @@ -309,8 +398,11 @@ impl AprilPocketTts { prepared: &AprilPreparedPrompt, style: &VoiceStyle, ) -> Result, String> { - let voice_embeddings = self.voice_embeddings(style)?; - let mut flow_state = self.condition_voice(&voice_embeddings)?; + // EXPERIMENTAL (latency bench): phase timing, enabled by BUZZ_TTS_PHASE_LOG=1. + let phase_log = std::env::var("BUZZ_TTS_PHASE_LOG").is_ok_and(|v| v == "1"); + let t0 = std::time::Instant::now(); + let mut flow_state = self.conditioned_flow_state(style)?; + let t_condition = t0.elapsed(); let token_ids = self .tokenizer .encode(prepared.text.as_str(), false) @@ -334,10 +426,244 @@ impl AprilPocketTts { let token_count = token_ids.len(); let text_embeddings = self.text_embeddings(token_ids)?; self.run_flow_main_prefix(&text_embeddings, &mut flow_state)?; + let t_prefix = t0.elapsed(); let max_frames = estimate_max_frames(token_count, self.bundle.frame_rate); let latents = self.generate_latents(max_frames, prepared.frames_after_eos, &mut flow_state)?; - self.decode_latents(&latents) + let t_generate = t0.elapsed(); + let audio = self.decode_latents(&latents)?; + if phase_log { + eprintln!( + "tts-phase: condition={:.0}ms prefix={:.0}ms generate={:.0}ms decode={:.0}ms frames={} audio_s={:.2}", + t_condition.as_secs_f64() * 1e3, + (t_prefix - t_condition).as_secs_f64() * 1e3, + (t_generate - t_prefix).as_secs_f64() * 1e3, + (t0.elapsed() - t_generate).as_secs_f64() * 1e3, + latents.len() / self.bundle.latent_dim, + audio.len() as f64 / self.bundle.sample_rate as f64, + ); + } + Ok(audio) + } + + /// EXPERIMENTAL (latency): return a fresh Flow LM state conditioned on + /// the reference voice, restoring a cached snapshot when the same voice + /// samples were conditioned before. Keyed by voice content, like + /// `cached_voice` — never by buffer address. + fn conditioned_flow_state(&mut self, style: &VoiceStyle) -> Result, String> { + let key = voice_key(style); + if let Some(cached) = &self.cached_conditioning { + if cached.key == key { + return restore_state(&cached.state); + } + } + let voice_embeddings = self.voice_embeddings(style)?; + let state = self.condition_voice(&voice_embeddings)?; + self.cached_conditioning = Some(CachedConditioning { + key, + state: snapshot_state(&state)?, + }); + Ok(state) + } + + /// EXPERIMENTAL (latency): streaming synthesis — interleaves the Flow LM + /// frame loop with incremental stateful Mimi decoding, invoking + /// `on_audio` with each decoded delta as soon as ~`emit_frames` latent + /// frames exist (80 ms of audio per frame). The Mimi decoder carries its + /// recurrent state across deltas, so the concatenated deltas are the same + /// audio `synth_chunk` would return. Returns Ok(false) when the callback + /// requested cancellation. + pub(crate) fn synth_chunk_streaming( + &mut self, + prepared: &AprilPreparedPrompt, + style: &VoiceStyle, + emit_frames: usize, + on_audio: &mut dyn FnMut(Vec) -> bool, + ) -> Result { + let mut flow_state = self.conditioned_flow_state(style)?; + let token_ids = self + .tokenizer + .encode(prepared.text.as_str(), false) + .map_err(|err| format!("tokenize Pocket TTS prompt: {err}"))? + .get_ids() + .iter() + .copied() + .map(i64::from) + .collect::>(); + if token_ids.is_empty() { + return Ok(true); + } + if token_ids.len() > self.bundle.max_token_per_chunk { + return Err(format!( + "Pocket TTS prompt has {} tokens; split_text_into_chunks maximum is {}", + token_ids.len(), + self.bundle.max_token_per_chunk + )); + } + + let token_count = token_ids.len(); + let text_embeddings = self.text_embeddings(token_ids)?; + self.run_flow_main_prefix(&text_embeddings, &mut flow_state)?; + let max_frames = estimate_max_frames(token_count, self.bundle.frame_rate); + let emit_frames = emit_frames.max(1); + + let mut mimi_state = initialize_state(&self.bundle.mimi_state_manifest)?; + let mut pending: Vec = Vec::with_capacity(emit_frames * self.bundle.latent_dim); + let mut current = vec![f32::NAN; self.bundle.latent_dim]; + let mut eos_step = None; + let mut rng = rand::rng(); + + for step in 0..max_frames { + let sequence = Tensor::from_array(( + vec![1_i64, 1, self.bundle.latent_dim as i64], + current.clone().into_boxed_slice(), + )) + .map_err(ort_error("create latent input"))?; + let text_embeddings = Tensor::::new( + &ort::memory::Allocator::default(), + [1_i64, 0, self.bundle.conditioning_dim as i64], + ) + .map_err(ort_error("create empty text input"))?; + let mut inputs = vec![ + (Cow::Borrowed("sequence"), SessionInputValue::from(sequence)), + ( + Cow::Borrowed("text_embeddings"), + SessionInputValue::from(text_embeddings), + ), + ]; + append_state_inputs(&mut inputs, &flow_state); + // Scoped: `outputs` borrows `self.flow_main`; it must drop before + // `decode_frames` takes `&mut self` below. + let (conditioning, eos_logit) = { + let mut outputs = self + .flow_main + .run(inputs) + .map_err(ort_error("run Pocket TTS Flow LM"))?; + let conditioning = outputs[0] + .try_extract_tensor::() + .map_err(ort_error("extract Flow LM conditioning"))? + .1 + .to_vec(); + let eos_logit = outputs[1] + .try_extract_tensor::() + .map_err(ort_error("extract Flow LM EOS logit"))? + .1 + .first() + .copied() + .ok_or_else(|| "Flow LM returned empty EOS logit".to_string())?; + replace_state_from_outputs(&mut flow_state, &mut outputs)?; + (conditioning, eos_logit) + }; + + if eos_logit > EOS_LOGIT_THRESHOLD && eos_step.is_none() { + eos_step = Some(step); + } + if eos_step.is_some_and(|eos| step >= eos + prepared.frames_after_eos) { + break; + } + + let mut noise = + normal_noise(&mut rng, self.bundle.latent_dim, DEFAULT_TEMPERATURE.sqrt()); + let conditioning = Tensor::from_array(( + vec![1_i64, self.bundle.conditioning_dim as i64], + conditioning.into_boxed_slice(), + )) + .map_err(ort_error("create flow conditioning"))?; + let s = Tensor::from_array((vec![1_i64, 1], vec![0.0_f32].into_boxed_slice())) + .map_err(ort_error("create flow start tensor"))?; + let t = Tensor::from_array((vec![1_i64, 1], vec![1.0_f32].into_boxed_slice())) + .map_err(ort_error("create flow end tensor"))?; + let x = Tensor::from_array(( + vec![1_i64, self.bundle.latent_dim as i64], + noise.clone().into_boxed_slice(), + )) + .map_err(ort_error("create flow noise tensor"))?; + let outputs = self + .flow + .run(ort::inputs![ + "c" => conditioning, + "s" => s, + "t" => t, + "x" => x, + ]) + .map_err(ort_error("run Pocket TTS flow"))?; + let flow = outputs[0] + .try_extract_tensor::() + .map_err(ort_error("extract Pocket TTS flow"))? + .1; + if flow.len() != noise.len() { + return Err(format!( + "flow returned {} values; expected {}", + flow.len(), + noise.len() + )); + } + for (sample, delta) in noise.iter_mut().zip(flow) { + *sample += *delta; + } + drop(outputs); + current.clone_from(&noise); + pending.extend_from_slice(&noise); + + if pending.len() >= emit_frames * self.bundle.latent_dim { + let audio = self.decode_frames(&pending, &mut mimi_state)?; + pending.clear(); + if !audio.is_empty() && !on_audio(audio) { + return Ok(false); + } + } + } + if !pending.is_empty() { + let audio = self.decode_frames(&pending, &mut mimi_state)?; + if !audio.is_empty() && !on_audio(audio) { + return Ok(false); + } + } + Ok(true) + } + + /// EXPERIMENTAL (latency): decode a batch of latent frames with a + /// caller-held Mimi state, so successive calls continue one stream. + fn decode_frames( + &mut self, + latents: &[f32], + state: &mut [StateValue], + ) -> Result, String> { + if latents.is_empty() { + return Ok(Vec::new()); + } + if !latents.len().is_multiple_of(self.bundle.latent_dim) { + return Err(format!( + "latent buffer has {} values, not divisible by {}", + latents.len(), + self.bundle.latent_dim + )); + } + let frame_count = latents.len() / self.bundle.latent_dim; + let mut audio = Vec::new(); + for start in (0..frame_count).step_by(DECODER_CHUNK_FRAMES) { + let end = (start + DECODER_CHUNK_FRAMES).min(frame_count); + let values = + latents[start * self.bundle.latent_dim..end * self.bundle.latent_dim].to_vec(); + let latent = Tensor::from_array(( + vec![1_i64, (end - start) as i64, self.bundle.latent_dim as i64], + values.into_boxed_slice(), + )) + .map_err(ort_error("create Mimi latent tensor"))?; + let mut inputs = vec![(Cow::Borrowed("latent"), SessionInputValue::from(latent))]; + append_state_inputs(&mut inputs, state); + let mut outputs = self + .mimi_decoder + .run(inputs) + .map_err(ort_error("run Mimi decoder"))?; + let samples = outputs[0] + .try_extract_tensor::() + .map_err(ort_error("extract Mimi audio"))? + .1; + audio.extend_from_slice(samples); + replace_state_from_outputs(state, &mut outputs)?; + } + Ok(audio) } fn prepared_token_count(&self, text: &str) -> Result { @@ -356,13 +682,9 @@ impl AprilPocketTts { } fn voice_embeddings(&mut self, style: &VoiceStyle) -> Result, String> { - let key = ( - style.samples.as_ptr() as usize, - style.samples.len(), - style.sample_rate, - ); + let key = voice_key(style); if let Some(cached) = &self.cached_voice { - if (cached.samples_ptr, cached.samples_len, cached.sample_rate) == key { + if cached.key == key { return Ok(cached.embeddings.clone()); } } @@ -403,9 +725,7 @@ impl AprilPocketTts { embeddings.extend_from_slice(&self.bos_embedding); embeddings.extend_from_slice(encoded); self.cached_voice = Some(CachedVoice { - samples_ptr: key.0, - samples_len: key.1, - sample_rate: key.2, + key, embeddings: embeddings.clone(), }); Ok(embeddings) @@ -638,6 +958,180 @@ impl AprilPocketTts { } } +fn split_model_at_natural_boundaries( + text: &str, + max_tokens: usize, + token_count: F, +) -> Result, String> +where + F: FnMut(&str) -> Result, +{ + split_at_natural_boundaries(text, max_tokens, false, token_count) +} + +fn split_playback_at_natural_boundaries( + text: &str, + max_tokens: usize, + token_count: F, +) -> Result, String> +where + F: FnMut(&str) -> Result, +{ + split_at_natural_boundaries(text, max_tokens, true, token_count) +} + +fn split_at_natural_boundaries( + text: &str, + max_tokens: usize, + isolate_first_sentence: bool, + mut token_count: F, +) -> Result, String> +where + F: FnMut(&str) -> Result, +{ + if text.is_empty() { + return Ok(Vec::new()); + } + + let mut chunks = Vec::new(); + let mut start = 0; + while start < text.len() { + while text[start..] + .chars() + .next() + .is_some_and(char::is_whitespace) + { + start += text[start..] + .chars() + .next() + .expect("checked above") + .len_utf8(); + } + if start == text.len() { + break; + } + + let mut first_sentence_end = None; + let mut sentence_end = None; + let mut clause_end = None; + let mut word_end = None; + for (offset, ch) in text[start..].char_indices() { + let end = start + offset + ch.len_utf8(); + let at_word_end = + end == text.len() || text[end..].chars().next().is_some_and(char::is_whitespace); + let at_clause_end = matches!(ch, '—' | '–') + && !text[end..] + .chars() + .next() + .is_some_and(is_closing_punctuation); + if !at_word_end && !at_clause_end { + continue; + } + // Prepared token counts are monotonic in prefix length, so once a + // candidate overflows the limit no longer candidate can fit. Stop + // scanning instead of tokenizing every remaining boundary: that + // kept this loop superlinear in prompt length, and the cost landed + // before the first chunk reached synthesis. + if token_count(&text[start..end])? > max_tokens { + break; + } + + word_end = Some(end); + match natural_boundary(&text[start..end], end == text.len()) { + TextBoundary::Sentence => { + first_sentence_end.get_or_insert(end); + sentence_end = Some(end); + } + TextBoundary::Clause => clause_end = Some(end), + TextBoundary::Word => {} + } + } + + let preferred_end = if isolate_first_sentence && chunks.is_empty() { + first_sentence_end.or(clause_end).or(word_end) + } else { + sentence_end.or(clause_end).or(word_end) + }; + let end = if let Some(end) = preferred_end { + end + } else { + // A single word can itself exceed the model limit. Preserve a + // scalar boundary as the final safety case without losing UTF-8. + let mut scalar_end = None; + for (offset, ch) in text[start..].char_indices() { + if ch.is_whitespace() { + break; + } + let end = start + offset + ch.len_utf8(); + if token_count(&text[start..end])? <= max_tokens { + scalar_end = Some(end); + } + } + scalar_end.ok_or_else(|| { + format!( + "Pocket TTS prompt cannot fit one character within the {max_tokens}-token limit" + ) + })? + }; + + let mut next_start = end; + while text[next_start..] + .chars() + .next() + .is_some_and(char::is_whitespace) + { + next_start += text[next_start..] + .chars() + .next() + .expect("checked above") + .len_utf8(); + } + chunks.push(text[start..next_start].to_string()); + start = next_start; + } + + debug_assert_eq!(chunks.concat(), text); + Ok(chunks) +} + +fn natural_boundary(candidate: &str, is_end_of_text: bool) -> TextBoundary { + if is_end_of_text { + return TextBoundary::Sentence; + } + + let mut chars = candidate.chars().rev(); + let mut last = chars.next(); + while last.is_some_and(is_closing_punctuation) { + last = chars.next(); + } + match last { + Some('.' | '!' | '?') if !looks_like_abbreviation(candidate) => TextBoundary::Sentence, + Some(',' | ';' | ':' | '—' | '–') => TextBoundary::Clause, + _ => TextBoundary::Word, + } +} + +fn is_closing_punctuation(ch: char) -> bool { + matches!(ch, '"' | '\'' | '”' | '’' | ')' | ']' | '}') +} + +fn looks_like_abbreviation(candidate: &str) -> bool { + const ABBREVIATIONS: &[&str] = &[ + "Dr.", "Mr.", "Mrs.", "Ms.", "Prof.", "Sr.", "Jr.", "St.", "Ave.", "Rd.", "Blvd.", "Dept.", + "Inc.", "Ltd.", "Co.", "Corp.", "etc.", "vs.", "i.e.", "e.g.", "Ph.D.", + ]; + + let candidate = candidate.trim_end_matches(is_closing_punctuation); + let last_word = candidate + .rsplit_once(char::is_whitespace) + .map_or(candidate, |(_, word)| word); + ABBREVIATIONS.contains(&last_word) + || (last_word.ends_with('.') + && last_word[..last_word.len() - 1] + .chars() + .all(|ch| ch.is_ascii_digit())) +} + fn load_session(path: PathBuf, num_threads: usize) -> Result { if !path.is_file() { return Err(format!("missing Pocket TTS file: {}", path.display())); @@ -861,6 +1355,215 @@ mod tests { assert_eq!(shape_len(&[2, 1, 8, 1000, 64]).expect("shape"), 1_024_000); } + /// The two engine splitters must keep OPPOSITE isolation polarity. + /// + /// The guards in `pocket.rs` pin which engine method each public API calls, + /// but they cannot see what the method itself does: pointing + /// `split_playback_prompt` at the model wrapper leaves every call site's + /// source text untouched while first-sentence isolation silently stops + /// happening, so the first playback unit becomes the whole utterance and + /// first audio waits on generating all of it. + #[test] + fn engine_splitters_keep_opposite_isolation_polarity() { + let source = include_str!("pocket_april.rs"); + let production = source + .split_once("\n#[cfg(test)]") + .map_or(source, |(production, _)| production); + + // A method's own code, and nothing else. Ending at the method's own + // closing brace keeps the NEXT method's doc comment out, and stripping + // `//` to end of line keeps prose out: neither can call a splitter, so + // scanning either reports drift in a method that has not changed. + let method_code = |name: &str| -> String { + let (_, body) = production + .split_once(name) + .unwrap_or_else(|| panic!("{name} exists")); + let (body, _) = body + .split_once("\n }\n") + .unwrap_or_else(|| panic!("{name} has a closing brace")); + body.lines() + .map(|line| line.split_once("//").map_or(line, |(code, _)| code)) + .collect::>() + .join("\n") + }; + let model = method_code("fn split_prompt"); + let model = model.as_str(); + let playback = method_code("fn split_playback_prompt"); + let playback = playback.as_str(); + + assert_eq!( + ( + model.matches("split_model_at_natural_boundaries(").count(), + model + .matches("split_playback_at_natural_boundaries(") + .count(), + ), + (1, 0), + "split_prompt must pack sentences: isolating here peels sentence \ + one off every already-packed unit" + ); + assert_eq!( + ( + playback + .matches("split_playback_at_natural_boundaries(") + .count(), + playback + .matches("split_model_at_natural_boundaries(") + .count(), + ), + (1, 0), + "split_playback_prompt must isolate sentence one: packing here \ + makes the first playback unit the whole utterance and delays \ + first audio by the full generation" + ); + + // Calling the isolating splitter is necessary but not sufficient: a + // short circuit before the call can return the whole utterance as one + // unit while leaving the delegated splitter unchanged. Playback must + // delegate unconditionally so sentence one remains the first unit. + for control_flow in ["if ", "match ", "else", "return"] { + assert!( + !playback.contains(control_flow), + "split_playback_prompt must delegate unconditionally, found \ + `{control_flow}`: a branch before the split can return the \ + whole utterance as the first playback unit, delaying first \ + audio by the full generation" + ); + } + } + + fn whitespace_token_count(text: &str) -> Result { + Ok(text.split_whitespace().count()) + } + + #[test] + fn playback_split_keeps_first_sentence_separate_then_packs_the_remainder() { + let text = "One two. Three four. Five six."; + let chunks = split_playback_at_natural_boundaries(text, 4, whitespace_token_count).unwrap(); + assert_eq!(chunks, ["One two. ", "Three four. Five six."]); + assert_eq!(chunks.concat(), text); + } + + #[test] + fn model_split_packs_multiple_sentences_within_limit() { + let text = "One two. Three four. Five six."; + let chunks = split_model_at_natural_boundaries(text, 4, whitespace_token_count).unwrap(); + assert_eq!(chunks, ["One two. Three four. ", "Five six."]); + assert_eq!(chunks.concat(), text); + } + + #[test] + fn playback_then_model_split_does_not_isolate_later_sentences_again() { + let text = "Alpha one. Beta two. Gamma three."; + let playback = + split_playback_at_natural_boundaries(text, 50, whitespace_token_count).unwrap(); + assert_eq!(playback, ["Alpha one. ", "Beta two. Gamma three."]); + + let model: Vec<_> = playback + .iter() + .flat_map(|chunk| { + split_model_at_natural_boundaries(chunk.trim(), 50, whitespace_token_count).unwrap() + }) + .collect(); + assert_eq!(model, ["Alpha one.", "Beta two. Gamma three."]); + } + + #[test] + fn natural_split_prefers_preceding_sentence_boundary() { + let text = "One two. Three four five six."; + let chunks = split_at_natural_boundaries(text, 5, true, whitespace_token_count).unwrap(); + assert_eq!(chunks, ["One two. ", "Three four five six."]); + assert_eq!(chunks.concat(), text); + } + + #[test] + fn oversized_sentence_uses_clause_then_word_fallback() { + let clause_text = "One two three, four five six seven."; + let clause_chunks = + split_at_natural_boundaries(clause_text, 5, true, whitespace_token_count).unwrap(); + assert_eq!(clause_chunks, ["One two three, ", "four five six seven."]); + assert_eq!(clause_chunks.concat(), clause_text); + + let word_text = "One two three four five six."; + let word_chunks = + split_at_natural_boundaries(word_text, 4, true, whitespace_token_count).unwrap(); + assert_eq!(word_chunks, ["One two three four ", "five six."]); + assert_eq!(word_chunks.concat(), word_text); + } + + #[test] + fn natural_split_preserves_unicode_punctuation_and_abbreviations() { + let text = "“Café naïve?” Maybe—yes, definitely; 東京 speaks."; + let chunks = split_at_natural_boundaries(text, 3, true, whitespace_token_count).unwrap(); + assert_eq!( + chunks, + ["“Café naïve?” ", "Maybe—yes, definitely; ", "東京 speaks."] + ); + assert_eq!(chunks.concat(), text); + + let abbreviation = "Dr. Smith waits. Then leaves."; + let chunks = + split_at_natural_boundaries(abbreviation, 3, true, whitespace_token_count).unwrap(); + assert_eq!(chunks, ["Dr. Smith waits. ", "Then leaves."]); + assert_eq!(chunks.concat(), abbreviation); + + let unspaced_clause = "alpha beta—gamma delta"; + let chunks = + split_at_natural_boundaries(unspaced_clause, 2, true, whitespace_token_count).unwrap(); + assert_eq!(chunks, ["alpha beta—", "gamma delta"]); + assert_eq!(chunks.concat(), unspaced_clause); + } + + #[test] + fn natural_split_does_not_treat_numeric_punctuation_as_unspaced_clauses() { + let text = "Meet at 12:30 with 1,000 guests onward."; + let chunks = split_at_natural_boundaries(text, 3, true, whitespace_token_count).unwrap(); + assert_eq!(chunks, ["Meet at 12:30 ", "with 1,000 guests ", "onward."]); + assert_eq!(chunks.concat(), text); + } + + #[test] + fn oversized_word_uses_utf8_scalar_boundary_without_loss() { + let text = "éééé"; + let chunks = + split_at_natural_boundaries(text, 3, true, |chunk| Ok(chunk.chars().count())).unwrap(); + assert_eq!(chunks, ["ééé", "é"]); + assert_eq!(chunks.concat(), text); + } + + #[test] + fn natural_split_stops_counting_tokens_past_the_limit() { + // Each boundary scan must stop at the first overflowing candidate + // rather than tokenizing every remaining boundary. Scanning to + // end-of-text makes tokenizer input grow superlinearly in prompt + // length, and that cost is paid before the first chunk reaches + // synthesis, taxing time-to-first-audio on long prompts. + let sentence = "The relay finished its migration and the channel list refreshed. "; + let tokenized_bytes = |repeats: usize| -> usize { + let text = sentence.repeat(repeats).trim_end().to_string(); + let total = std::cell::Cell::new(0_usize); + let chunks = split_at_natural_boundaries(&text, 50, true, |chunk| { + total.set(total.get() + chunk.len()); + whitespace_token_count(chunk) + }) + .expect("split repeated sentences"); + assert_eq!(chunks.concat(), text); + assert!(chunks.len() > 1); + total.get() + }; + + // Doubling the prompt must not multiply tokenizer work superlinearly. + // Bounded scans grow ~2x here; scanning to end-of-text grows ~5.5x. + let single = tokenized_bytes(12); + let double = tokenized_bytes(24); + assert!( + double < single * 3, + "doubling the prompt grew tokenizer input from {single} to {double} bytes \ + ({:.1}x); bounded scans stay near 2x", + double as f64 / single as f64, + ); + } + #[test] fn normal_noise_has_requested_length() { let mut rng = rand::rng(); @@ -874,6 +1577,234 @@ mod tests { assert_eq!(estimate_max_frames(300, 12.5), 1_275); } + /// Regression (review finding): the voice caches must key on CONTENT. + /// Voice switching clones and drops sample buffers, so a new voice with + /// the same length and rate can land at a recycled address — an + /// address-based key would then restore the previous voice's state and + /// speak with the wrong voice. + #[test] + fn voice_key_is_content_based_not_address_based() { + let style_a = VoiceStyle { + samples: vec![0.1, -0.2, 0.3, -0.4], + sample_rate: 24_000, + }; + // Same length, same rate, different content — MUST key differently, + // regardless of what address the allocator hands out. + let style_b = VoiceStyle { + samples: vec![0.4, -0.3, 0.2, -0.1], + sample_rate: 24_000, + }; + assert_ne!(voice_key(&style_a), voice_key(&style_b)); + + // Same content in a fresh allocation — MUST key identically, so the + // cache still hits across clones of the same voice. + let style_a_clone = VoiceStyle { + samples: style_a.samples.clone(), + sample_rate: style_a.sample_rate, + }; + assert_ne!( + style_a.samples.as_ptr(), + style_a_clone.samples.as_ptr(), + "clone must be a distinct allocation for this test to mean anything" + ); + assert_eq!(voice_key(&style_a), voice_key(&style_a_clone)); + + // Same content at a different rate is a different voice identity. + let style_a_resampled = VoiceStyle { + samples: style_a.samples.clone(), + sample_rate: 16_000, + }; + assert_ne!(voice_key(&style_a), voice_key(&style_a_resampled)); + } + + #[test] + #[ignore = "requires BUZZ_POCKET_TEST_MODEL_DIR"] + fn switching_between_equal_length_voices_reconditions_the_flow_state() { + let dir = std::env::var("BUZZ_POCKET_TEST_MODEL_DIR") + .expect("set BUZZ_POCKET_TEST_MODEL_DIR to the verified April bundle"); + let style_a = + crate::pocket::load_voice_style(&Path::new(&dir).join("reference_sample.wav")) + .expect("load reference voice"); + // Voice B: same length, same rate, different content (reversed + // samples) — the exact shape an address-recycling collision takes. + let style_b = VoiceStyle { + samples: style_a.samples.iter().rev().copied().collect(), + sample_rate: style_a.sample_rate, + }; + assert_eq!(style_a.samples.len(), style_b.samples.len()); + + // Engine 1: condition A (primes both caches), then switch to B. + let mut engine = AprilPocketTts::load(Path::new(&dir), 1).expect("load April bundle"); + let state_a = snapshot_state( + &engine + .conditioned_flow_state(&style_a) + .expect("condition A"), + ) + .expect("snapshot A"); + let state_b_after_switch = snapshot_state( + &engine + .conditioned_flow_state(&style_b) + .expect("condition B"), + ) + .expect("snapshot B after switch"); + // Warm hit on the SAME voice: the cached restore must reproduce the + // original conditioning bit-for-bit (cache warm == cache cold). + let state_b_warm_hit = snapshot_state( + &engine + .conditioned_flow_state(&style_b) + .expect("condition B warm"), + ) + .expect("snapshot B warm hit"); + + // Engine 2: fresh process conditions B with no cache in play. + let mut fresh = AprilPocketTts::load(Path::new(&dir), 1).expect("load April bundle"); + let state_b_fresh = snapshot_state( + &fresh + .conditioned_flow_state(&style_b) + .expect("condition B fresh"), + ) + .expect("snapshot B fresh"); + + // The switched state must equal a from-scratch conditioning of B and + // must NOT be A's cached state. + assert!( + snapshots_equal(&state_b_after_switch, &state_b_fresh), + "switching voices must recondition, not replay the cache" + ); + assert!( + !snapshots_equal(&state_b_after_switch, &state_a), + "equal-length distinct voices must produce distinct conditioning" + ); + // And the warm cache hit must be indistinguishable from recomputing. + assert!( + snapshots_equal(&state_b_warm_hit, &state_b_fresh), + "a warm conditioning-cache hit must equal a cold recompute" + ); + } + + fn snapshots_equal( + a: &[(StateSpec, SnapshotTensor)], + b: &[(StateSpec, SnapshotTensor)], + ) -> bool { + // f32 compares bitwise: state tensors legitimately contain NaN fill, + // and NaN != NaN under float equality would make identical states + // compare unequal. + a.len() == b.len() + && a.iter().zip(b).all(|((_, ta), (_, tb))| match (ta, tb) { + (SnapshotTensor::F32(sa, da), SnapshotTensor::F32(sb, db)) => { + sa == sb + && da.len() == db.len() + && da.iter().zip(db).all(|(x, y)| x.to_bits() == y.to_bits()) + } + (SnapshotTensor::I64(sa, da), SnapshotTensor::I64(sb, db)) => sa == sb && da == db, + (SnapshotTensor::Bool(sa, da), SnapshotTensor::Bool(sb, db)) => { + sa == sb && da == db + } + _ => false, + }) + } + + #[test] + #[ignore = "requires BUZZ_POCKET_TEST_MODEL_DIR"] + fn incremental_stateful_decode_matches_batch_decode() { + let dir = std::env::var("BUZZ_POCKET_TEST_MODEL_DIR") + .expect("set BUZZ_POCKET_TEST_MODEL_DIR to the verified April bundle"); + let mut engine = AprilPocketTts::load(Path::new(&dir), 1).expect("load April bundle"); + let style = crate::pocket::load_voice_style(&Path::new(&dir).join("reference_sample.wav")) + .expect("load reference voice"); + + // Generate one real latent sequence (the RNG makes repeat synths + // differ, so both decode paths must consume the SAME latents). + let prepared = + prepare_april_prompt("The relay deploy finished and every check passed cleanly.") + .expect("prepare prompt"); + let mut flow_state = engine + .conditioned_flow_state(&style) + .expect("condition voice"); + let token_ids = engine + .tokenizer + .encode(prepared.text.as_str(), false) + .expect("tokenize") + .get_ids() + .iter() + .copied() + .map(i64::from) + .collect::>(); + let token_count = token_ids.len(); + let text_embeddings = engine.text_embeddings(token_ids).expect("text embeddings"); + engine + .run_flow_main_prefix(&text_embeddings, &mut flow_state) + .expect("prefix"); + let max_frames = estimate_max_frames(token_count, engine.bundle.frame_rate); + let latents = engine + .generate_latents(max_frames, prepared.frames_after_eos, &mut flow_state) + .expect("generate latents"); + let frame_count = latents.len() / engine.bundle.latent_dim; + assert!( + frame_count > DECODER_CHUNK_FRAMES, + "need a multi-chunk case" + ); + + // Batch: the production decode (fresh state, 12-frame steps). + let batch = engine.decode_latents(&latents).expect("batch decode"); + + // Incremental chunkings: 12-frame deltas through one carried Mimi + // state must be bit-exact (the production batch path itself steps by + // DECODER_CHUNK_FRAMES=12 through one state). Sub-12 chunkings are + // measured for the record but are NOT exact — the decoder has + // intra-chunk lookahead — so streaming must emit at >= 12 frames. + for delta_frames in [6usize, 4, 2, 1] { + let mut state = + initialize_state(&engine.bundle.mimi_state_manifest).expect("mimi state"); + let mut streamed = Vec::new(); + for chunk in latents.chunks(delta_frames * engine.bundle.latent_dim) { + streamed.extend( + engine + .decode_frames(chunk, &mut state) + .expect("delta decode"), + ); + } + assert_eq!(batch.len(), streamed.len(), "sample count must match"); + let max_diff = batch + .iter() + .zip(&streamed) + .map(|(a, b)| (a - b).abs()) + .fold(0.0f32, f32::max); + let rms_batch = (batch.iter().map(|s| s * s).sum::() / batch.len() as f32).sqrt(); + let rms_err = (batch + .iter() + .zip(&streamed) + .map(|(a, b)| (a - b) * (a - b)) + .sum::() + / batch.len() as f32) + .sqrt(); + eprintln!( + "delta_frames={delta_frames}: max|diff|={max_diff:.6} rms_err={rms_err:.6} snr_db={:.1}", + 20.0 * (rms_batch / rms_err.max(1e-12)).log10() + ); + } + let mut state = initialize_state(&engine.bundle.mimi_state_manifest).expect("mimi state"); + let mut streamed = Vec::new(); + for chunk in latents.chunks(DECODER_CHUNK_FRAMES * engine.bundle.latent_dim) { + streamed.extend( + engine + .decode_frames(chunk, &mut state) + .expect("delta decode"), + ); + } + + assert_eq!(batch.len(), streamed.len(), "sample count must match"); + let max_diff = batch + .iter() + .zip(&streamed) + .map(|(a, b)| (a - b).abs()) + .fold(0.0f32, f32::max); + assert!( + max_diff <= 1.0e-4, + "incremental decode diverged from batch decode: max |diff| = {max_diff}" + ); + } + #[test] #[ignore = "requires BUZZ_POCKET_TEST_MODEL_DIR"] fn tokenizer_matches_sentencepiece_reference_including_unknown_words() { @@ -910,8 +1841,10 @@ mod tests { assert!(chunks.len() > 1); assert!(chunks.iter().all(|chunk| { - engine.token_count(chunk).expect("tokenize chunk") <= engine.bundle.max_token_per_chunk + engine.prepared_token_count(chunk).expect("tokenize chunk") + <= engine.bundle.max_token_per_chunk })); + assert_eq!(chunks.concat(), prepared.text); } #[test] @@ -925,16 +1858,20 @@ mod tests { let chunks = engine.split_prompt(&prepared).expect("split long sentence"); let token_counts: Vec<_> = chunks .iter() - .map(|chunk| engine.token_count(chunk).expect("count tokens")) + .map(|chunk| engine.prepared_token_count(chunk).expect("count tokens")) .collect(); - assert_eq!( - chunks, - [ - "And sometimes, when I am certain the reader is rested, I will engage him with a sentence of considerable length, a sentence that burns with energy and builds with all the.", - "Impetus of a crescendo, the roll of the drums, the crash of the cymbals–sounds that say listen to this, it is important.", - ] - ); - assert_eq!(token_counts, [48, 44]); + assert!(token_counts + .iter() + .all(|&count| count <= engine.bundle.max_token_per_chunk)); + assert_eq!(chunks.concat(), prepared.text); + assert!(chunks.len() > 1); + assert!(chunks[..chunks.len() - 1].iter().all(|chunk| { + chunk + .trim_end() + .chars() + .last() + .is_some_and(|ch| ['.', '!', '?', ',', ';', ':', '—', '–'].contains(&ch)) + })); } } diff --git a/desktop/src-tauri/src/huddle/agent_tts_routing.rs b/desktop/src-tauri/src/huddle/agent_tts_routing.rs index 2ee3ec0d41a..87a56c0dbbc 100644 --- a/desktop/src-tauri/src/huddle/agent_tts_routing.rs +++ b/desktop/src-tauri/src/huddle/agent_tts_routing.rs @@ -25,8 +25,9 @@ pub(super) fn classify_agent_tts_runtime( } /// Maximum text length accepted for TTS synthesis. -/// ~2000 chars is 1–2 minutes of speech. Longer messages are truncated. -pub(super) const MAX_TTS_TEXT_LEN: usize = 2000; +/// This high safety cap keeps unexpectedly large events bounded while allowing +/// normal long-form huddle replies to play in full. +pub(super) const MAX_TTS_TEXT_LEN: usize = 8_096; pub(super) fn normalize_agent_tts_text(text: String) -> String { if text.chars().count() > MAX_TTS_TEXT_LEN { diff --git a/desktop/src-tauri/src/huddle/agent_tts_routing_tests.rs b/desktop/src-tauri/src/huddle/agent_tts_routing_tests.rs index cb550d7005b..c9ebabe6b62 100644 --- a/desktop/src-tauri/src/huddle/agent_tts_routing_tests.rs +++ b/desktop/src-tauri/src/huddle/agent_tts_routing_tests.rs @@ -47,6 +47,7 @@ fn disabled_is_the_only_intentional_runtime_no_op() { #[test] fn assistant_text_truncation_is_unicode_safe_before_voice_routing() { + assert_eq!(MAX_TTS_TEXT_LEN, 8_096); let input = "🦀".repeat(MAX_TTS_TEXT_LEN + 1); let output = normalize_agent_tts_text(input); assert_eq!( diff --git a/desktop/src-tauri/src/huddle/agents.rs b/desktop/src-tauri/src/huddle/agents.rs index 41a348d8889..a64f540f5eb 100644 --- a/desktop/src-tauri/src/huddle/agents.rs +++ b/desktop/src-tauri/src/huddle/agents.rs @@ -29,40 +29,21 @@ use super::{pipeline::start_auto_enabled_transcription, HuddlePhase}; // ── Constants ───────────────────────────────────────────────────────────────── -/// Voice-mode guidelines posted as kind:48106 (huddle guidelines) to the -/// ephemeral channel at huddle start. Agents see them via EOSE replay. -/// Instructs agents on voice-mode etiquette: TTS constraints, brevity, -/// self-selection, and sentence-at-a-time delivery. +/// Voice-mode instructions posted as kind:48106 to the ephemeral channel at +/// huddle start. Agents load this event into the channel session system prompt. /// -/// Why sentence-at-a-time: the desktop speaks each agent message as it -/// arrives (queued, in order), so an agent that sends its first sentence -/// immediately — then the rest as separate messages — cuts time-to-first- -/// audio from "full reply generated" to "first sentence generated". This is -/// the prompt-level equivalent of token streaming, with no harness changes. -/// -/// Build voice-mode guidelines with the parent channel ID so agents know -/// where "the main channel" is. +/// Keep this deliberately short: the invariant that matters is that a directly +/// addressed user interrupts every other activity and receives an immediate +/// spoken response. pub fn voice_mode_guidelines(parent_channel_id: &str) -> String { format!( "\ You are in a live voice huddle attached to channel {parent_channel_id}. -Your text is read aloud via TTS, message by message, in the order sent. - -Latency matters most: reply IMMEDIATELY — do not compose your full reply -before sending anything. The moment your first sentence is formed, send it -as its own `buzz messages send` tool call: it is what breaks the silence. -Then send each following sentence the same way — one sentence per separate -`buzz messages send` call. Never hold a finished sentence back to bundle it -with the next one. - -- If not addressed or relevant: do nothing. Do not respond. -- Keep the whole reply short — a few sentences at most. Start with the answer, no preamble. -- No markdown, code blocks, lists, or structured data — say it naturally. -- To share code or detailed data: say \"I'll post that in the main channel\" and do so. -- When you need a tool, say one short sentence first (e.g. \"Let me check.\"), then run it, then summarize the key finding verbally. -- If a new human message arrives mid-reply, you were interrupted: drop your unsent sentences and respond to the new message instead. -- In multi-agent huddles, identify yourself only when needed. -- Use your Buzz tools proactively when asked." +Your messages are read aloud in the order sent. +Reply immediately whenever a user addresses you, no matter what else is happening. +Send your first sentence as soon as it is formed, then send each following sentence separately. +Speak plainly and briefly without markdown; post code or long detail to the attached channel instead. +If you are not addressed, stay silent." ) } @@ -317,7 +298,15 @@ fn contains_member(members: &[(String, Option)], pubkey: &str) -> bool { #[cfg(test)] mod tests { - use super::contains_member; + use super::{contains_member, voice_mode_guidelines}; + + #[test] + fn voice_mode_guidelines_are_short_and_pin_immediate_reply() { + let guidelines = voice_mode_guidelines("parent-channel"); + assert_eq!(guidelines.lines().count(), 6); + assert!(guidelines.contains("Reply immediately whenever a user addresses you")); + assert!(guidelines.contains("parent-channel")); + } #[test] fn existing_parent_membership_is_preserved_regardless_of_role() { diff --git a/desktop/src-tauri/src/huddle/commands.rs b/desktop/src-tauri/src/huddle/commands.rs index 993d8e54eba..e4f25a93fcb 100644 --- a/desktop/src-tauri/src/huddle/commands.rs +++ b/desktop/src-tauri/src/huddle/commands.rs @@ -7,7 +7,9 @@ use uuid::Uuid; use crate::{app_state::AppState, events, relay::submit_event}; -use super::{relay_api::validate_pubkey_hex, HuddlePhase}; +use super::pipeline::start_auto_enabled_transcription; +use super::relay_api::MAX_HUDDLE_AGENTS; +use super::{agents, relay_api::validate_pubkey_hex, HuddlePhase}; /// Update the clickable microphone control independently from the PTT shortcut. #[tauri::command] @@ -130,3 +132,85 @@ pub async fn remove_agent_from_huddle( Ok(()) } + +/// Add an agent to the active huddle. +/// +/// Steps: +/// 1. Validates the huddle is in the Connected or Active phase. +/// 2. Adds the agent to both the ephemeral and parent channels (kind:9000). +/// 3. Only appends the agent pubkey to `agent_pubkeys` if the ephemeral add +/// succeeded — failed adds (policy rejection) are NOT p-tagged. +/// +/// Returns a structured `AgentAddResult` so the frontend can surface +/// parent-channel errors without treating them as hard failures. +/// +/// The running ACP process for this agent auto-subscribes when it receives +/// the kind:9000 membership notification — no separate process spawn needed. +#[tauri::command] +pub async fn add_agent_to_huddle( + agent_pubkey: String, + state: State<'_, AppState>, +) -> Result { + validate_pubkey_hex(&agent_pubkey)?; + + let (eph_id, parent_id, huddle_generation) = { + let hs = state.huddle()?; + if !matches!(hs.phase, HuddlePhase::Connected | HuddlePhase::Active) { + return Err("no active huddle".to_string()); + } + + // Enforce agent cap on incremental adds too. + let current_agent_count = hs + .agent_pubkeys + .lock() + .unwrap_or_else(|e| e.into_inner()) + .len(); + if current_agent_count >= MAX_HUDDLE_AGENTS { + return Err(format!( + "agent limit reached: {} (max {})", + current_agent_count, MAX_HUDDLE_AGENTS + )); + } + + let eph = hs + .ephemeral_channel_id + .clone() + .ok_or("no ephemeral channel")?; + let parent = hs.parent_channel_id.clone().ok_or("no parent channel")?; + (eph, parent, hs.huddle_generation) + }; + + let eph_uuid = Uuid::parse_str(&eph_id).map_err(|e| e.to_string())?; + let parent_uuid = Uuid::parse_str(&parent_id).map_err(|e| e.to_string())?; + + // Returns Err only if the ephemeral add fails — parent failure is in the result. + let result = agents::add_agent_to_huddle(eph_uuid, parent_uuid, &agent_pubkey, &state).await?; + + // Ephemeral add succeeded — register it only if this is still the huddle + // that initiated the relay operation. + let transcription_auto_enabled = { + let mut hs = state.huddle()?; + if !hs.is_current_huddle(&eph_id, huddle_generation) { + return Ok(result); + } + let mut pubkeys = hs.agent_pubkeys.lock().unwrap_or_else(|e| e.into_inner()); + if !pubkeys.contains(&agent_pubkey) { + pubkeys.push(agent_pubkey.clone()); + } + drop(pubkeys); + if !hs.participants.contains(&agent_pubkey) { + hs.participants.push(agent_pubkey.clone()); + } + hs.maybe_auto_enable_transcription_for_agents() + }; + + // No guidelines re-post needed — the agent sees the original kind:48106 + // guidelines via EOSE replay when it subscribes to the ephemeral channel. + if transcription_auto_enabled { + start_auto_enabled_transcription(&state, &eph_id).await; + } else { + state.emit_huddle_state_changed(); + } + + Ok(result) +} diff --git a/desktop/src-tauri/src/huddle/latency_bench.rs b/desktop/src-tauri/src/huddle/latency_bench.rs new file mode 100644 index 00000000000..710854b5337 --- /dev/null +++ b/desktop/src-tauri/src/huddle/latency_bench.rs @@ -0,0 +1,324 @@ +//! Ad-hoc baseline latency bench for the STT -> fake LLM -> TTS pipeline. +//! +//! Drives the REAL production machinery: +//! - `SttPipeline::new` (rubato 48k->16k, earshot VAD, 300 ms silence flush, +//! Parakeet TDT-CTC 110M int8 via sherpa-onnx, 1 thread) +//! - `TtsPipeline::new_with_voice` (warmup synth, chunker, synth_chunk, +//! rodio persistent Player, 20 ms lead-in) +//! +//! with a fake LLM in place of the relay/agent leg. +//! +//! Audio is fed in real-time 100 ms batches (mirroring the AudioWorklet +//! cadence) so VAD endpointing behaves exactly like production. +//! +//! Timestamps captured per turn: +//! t_speech_end last voiced sample delivered to push_audio (wall clock, +//! derived from the WAV's last voiced sample + feed pacing) +//! t_transcript text_rx yields the transcript +//! t_speak fake-LLM reply handed to TtsPipeline::speak +//! t_first_audio tts_active rising edge = first player.append accepted +//! +//! Run: +//! BUZZ_BENCH_WAV=<48k f32 mono wav> cargo test --release -p buzz-desktop \ +//! --lib huddle::latency_bench -- --ignored --nocapture + +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, +}; +use std::time::{Duration, Instant}; + +use super::stt::SttPipeline; +use super::tts::TtsPipeline; + +/// Read a mono 32-bit-float WAV (as produced by `afconvert -d LEF32@48000`). +/// Minimal parser: walks RIFF chunks, asserts fmt = IEEE float mono 48 kHz. +fn read_wav_f32_48k(path: &str) -> Vec { + let bytes = std::fs::read(path).expect("read wav"); + assert_eq!(&bytes[0..4], b"RIFF"); + assert_eq!(&bytes[8..12], b"WAVE"); + let mut pos = 12usize; + let mut fmt_ok = false; + let mut data: Option<(usize, usize)> = None; + while pos + 8 <= bytes.len() { + let id = &bytes[pos..pos + 4]; + let len = u32::from_le_bytes(bytes[pos + 4..pos + 8].try_into().unwrap()) as usize; + let body = pos + 8; + match id { + b"fmt " => { + let format = u16::from_le_bytes(bytes[body..body + 2].try_into().unwrap()); + let channels = u16::from_le_bytes(bytes[body + 2..body + 4].try_into().unwrap()); + let rate = u32::from_le_bytes(bytes[body + 4..body + 8].try_into().unwrap()); + let bits = u16::from_le_bytes(bytes[body + 14..body + 16].try_into().unwrap()); + assert_eq!(format, 3, "expected IEEE float wav"); + assert_eq!(channels, 1, "expected mono"); + assert_eq!(rate, 48_000, "expected 48 kHz"); + assert_eq!(bits, 32); + fmt_ok = true; + } + b"data" => data = Some((body, len)), + _ => {} + } + pos = body + len + (len & 1); + } + assert!(fmt_ok, "fmt chunk missing"); + let (off, len) = data.expect("data chunk missing"); + bytes[off..off + len] + .chunks_exact(4) + .map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]])) + .collect() +} + +/// Index (in samples) one past the last sample whose |amplitude| exceeds the +/// threshold — "when the user stopped speaking" on the feed timeline. +fn last_voiced_sample(samples: &[f32], threshold: f32) -> usize { + samples + .iter() + .rposition(|s| s.abs() > threshold) + .map(|i| i + 1) + .unwrap_or(0) +} + +/// Poll a tokio mpsc receiver from sync context for up to `timeout`. +/// 1 ms poll keeps timestamp error negligible against ~100 ms scales. +fn tokio_recv_with_timeout( + rx: &mut tokio::sync::mpsc::Receiver, + timeout: Duration, +) -> Option { + let deadline = Instant::now() + timeout; + loop { + if let Ok(t) = rx.try_recv() { + return Some(t); + } + if Instant::now() >= deadline { + return None; + } + std::thread::sleep(Duration::from_millis(1)); + } +} + +struct TurnResult { + label: &'static str, + transcript: String, + stt_ms: f64, + llm_ms: f64, + tts_ms: f64, + e2e_ms: f64, +} + +#[test] +#[ignore = "ad-hoc latency baseline; needs models in ~/.buzz/models and an audio output device"] +fn baseline_stt_fake_llm_tts_first_audio() { + let home = dirs::home_dir().expect("home"); + let stt_dir = home.join(".buzz/models/parakeet-tdt-ctc-110m-en"); + let tts_dir = home.join(".buzz/models/pocket-tts"); + assert!( + stt_dir.join("model.int8.onnx").exists(), + "parakeet model missing" + ); + assert!(tts_dir.join("bundle.json").exists(), "pocket model missing"); + + let wav_path = std::env::var("BUZZ_BENCH_WAV").expect("set BUZZ_BENCH_WAV"); + let samples_48k = read_wav_f32_48k(&wav_path); + let speech_end_sample = last_voiced_sample(&samples_48k, 0.015); + let audio_dur_s = samples_48k.len() as f64 / 48_000.0; + let speech_end_s = speech_end_sample as f64 / 48_000.0; + eprintln!( + "bench: utterance {wav_path}: {audio_dur_s:.2} s total, speech ends at {speech_end_s:.2} s" + ); + + let llm_delay_ms: u64 = std::env::var("BUZZ_BENCH_LLM_MS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(0); + + // ── Bring up the real pipelines, exactly as maybe_start_* do ──────────── + let tts_active = Arc::new(AtomicBool::new(false)); + let tts_cancel = Arc::new(AtomicBool::new(false)); + + let t = Instant::now(); + let tts = TtsPipeline::new_with_voice( + tts_dir, + Arc::clone(&tts_active), + Arc::clone(&tts_cancel), + "eve", + None, // default output device + None, // no Tauri app handle + ) + .expect("tts pipeline"); + eprintln!( + "bench: TTS pipeline ready (engine load + warmup + audio prime) in {:.0} ms", + t.elapsed().as_secs_f64() * 1e3 + ); + + let t = Instant::now(); + let (stt, mut text_rx) = SttPipeline::new(stt_dir, None, None).expect("stt pipeline"); + // Recognizer loads inside the worker thread; give it time, then verify + // liveness via a first throwaway feed below. + std::thread::sleep(Duration::from_secs(2)); + assert!(!stt.is_finished(), "stt worker died during init"); + eprintln!( + "bench: STT pipeline spawned ({:.0} ms incl. settle sleep)", + t.elapsed().as_secs_f64() * 1e3 + ); + + // Fake LLM replies: short / medium / long, cycled across turns. + let replies: [(&'static str, &'static str); 3] = [ + ("reply_short", "Let me check."), + ("reply_medium", "Got it. The relay deploy finished about two minutes ago and all checks passed."), + ("reply_long", "Here's where things stand. The relay deploy finished cleanly and every health check is green. Two pods restarted during rollout, which is expected, and message latency is back to normal."), + ]; + let turns: usize = std::env::var("BUZZ_BENCH_TURNS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(6); + + // 100 ms batches at 48 kHz, matching the AudioWorklet push cadence. + const BATCH: usize = 4_800; + let mut results: Vec = Vec::new(); + + let stt = Arc::new(stt); + for turn in 0..turns { + let (label, reply) = replies[turn % replies.len()]; + + // Feed the utterance in real time from a separate thread (the + // AudioWorklet role), then trailing silence so the 300 ms VAD flush + // fires. The main thread meanwhile timestamps transcript arrival — + // recv must NOT be serialized behind the silence feed, or the + // measurement floor becomes the feed loop instead of the STT path. + let feeder_stt = Arc::clone(&stt); + let feeder_samples = samples_48k.clone(); + let feed_start = Instant::now(); + let feeder = std::thread::spawn(move || { + let mut cursor = 0usize; + while cursor < feeder_samples.len() { + let end = (cursor + BATCH).min(feeder_samples.len()); + let bytes: Vec = feeder_samples[cursor..end] + .iter() + .flat_map(|s| s.to_le_bytes()) + .collect(); + feeder_stt.push_audio(bytes).expect("push"); + cursor = end; + // Pace to real time. + let target = feed_start + Duration::from_millis((cursor / 48) as u64); + let now = Instant::now(); + if target > now { + std::thread::sleep(target - now); + } + } + // Trailing silence: 1 s guarantees the 300 ms flush window closes. + let silence = vec![0u8; BATCH * 4]; + for _ in 0..10 { + feeder_stt + .push_audio(silence.clone()) + .expect("push silence"); + std::thread::sleep(Duration::from_millis(100)); + } + }); + let t_speech_end = feed_start + Duration::from_secs_f64(speech_end_s); + + // Transcript arrival. An utterance with an intra-sentence pause can + // VAD-split into multiple segments; keep the LAST one delivered so the + // turn aligns with the true end of speech. The extra "is another + // segment coming?" wait below is a HARNESS artifact (prod forwards + // every segment immediately) and is excluded from all timings. + let mut transcript = text_rx + .blocking_recv() + .expect("stt channel closed before transcript"); + let mut t_transcript = Instant::now(); + let mut segments = 1usize; + loop { + match text_rx.try_recv() { + Ok(t) => { + transcript = t; + t_transcript = Instant::now(); + segments += 1; + } + Err(_) => { + if feeder.is_finished() { + // Feed done (incl. 1 s trailing silence): any final + // segment has already flushed and decoded. One short + // grace poll covers a decode still in flight. + match tokio_recv_with_timeout(&mut text_rx, Duration::from_millis(500)) { + Some(t) => { + transcript = t; + t_transcript = Instant::now(); + segments += 1; + } + None => break, + } + } else { + // Feeder still delivering audio — a later segment may + // arrive any time until the feed (plus flush window) + // completes. Keep waiting; do NOT break early or the + // tail segment leaks into the next turn. + if let Some(t) = + tokio_recv_with_timeout(&mut text_rx, Duration::from_millis(100)) + { + transcript = t; + t_transcript = Instant::now(); + segments += 1; + } + } + } + } + } + feeder.join().expect("feeder"); + + // Fake LLM. Applied AFTER the harness-only segment wait; llm_ms is the + // configured delay, so the harness wait never leaks into any timing. + if llm_delay_ms > 0 { + std::thread::sleep(Duration::from_millis(llm_delay_ms)); + } + let t_speak = Instant::now(); + tts.speak(reply.to_string()).expect("speak"); + + // First audio: tts_active rising edge == first accepted player append. + let deadline = Instant::now() + Duration::from_secs(30); + while !tts_active.load(Ordering::Acquire) { + assert!(Instant::now() < deadline, "no first audio within 30 s"); + std::thread::sleep(Duration::from_micros(500)); + } + let t_first_audio = Instant::now(); + + let stt_ms = (t_transcript - t_speech_end).as_secs_f64() * 1e3; + // llm_ms is exactly the configured fake-LLM delay; tts is measured + // from speak() to first accepted append. e2e composes the three real + // legs so the harness-only segment wait (between t_transcript and the + // fake-LLM sleep) never inflates the pipeline number. + let llm_ms = llm_delay_ms as f64; + let tts_ms = (t_first_audio - t_speak).as_secs_f64() * 1e3; + let e2e_ms = stt_ms + llm_ms + tts_ms; + eprintln!( + "bench turn {turn} [{label}]: stt={stt_ms:.0}ms llm={llm_ms:.0}ms tts_first_audio={tts_ms:.0}ms e2e={e2e_ms:.0}ms segments={segments} transcript={transcript:?}" + ); + results.push(TurnResult { + label, + transcript, + stt_ms, + llm_ms, + tts_ms, + e2e_ms, + }); + + // Wait for playback to drain + prod cooldown before the next turn. + while tts_active.load(Ordering::Acquire) { + std::thread::sleep(Duration::from_millis(20)); + } + std::thread::sleep(Duration::from_millis(500)); + } + + // Summary JSON for the write-up. + println!("["); + for (i, r) in results.iter().enumerate() { + let comma = if i + 1 < results.len() { "," } else { "" }; + println!( + " {{\"turn\":{i},\"label\":\"{}\",\"stt_ms\":{:.1},\"llm_ms\":{:.1},\"tts_first_audio_ms\":{:.1},\"e2e_ms\":{:.1},\"transcript\":{:?}}}{comma}", + r.label, r.stt_ms, r.llm_ms, r.tts_ms, r.e2e_ms, r.transcript + ); + } + println!("]"); + + stt.shutdown(); + tts.shutdown(); +} diff --git a/desktop/src-tauri/src/huddle/mod.rs b/desktop/src-tauri/src/huddle/mod.rs index fcf29d688b9..1feb2073b09 100644 --- a/desktop/src-tauri/src/huddle/mod.rs +++ b/desktop/src-tauri/src/huddle/mod.rs @@ -29,6 +29,8 @@ pub mod agents; pub mod audio_output; mod commands; pub mod jitter; +#[cfg(test)] +mod latency_bench; pub mod models; pub mod pipeline; pub mod playout; @@ -69,7 +71,8 @@ pub(super) fn drain_until_shutdown( // ── Re-exports ──────────────────────────────────────────────────────────────── pub use commands::{ - interrupt_huddle_speech, remove_agent_from_huddle, set_huddle_manual_mic_unmuted, + add_agent_to_huddle, interrupt_huddle_speech, remove_agent_from_huddle, + set_huddle_manual_mic_unmuted, }; pub use state::{HuddleJoinInfo, HuddlePhase, HuddleState, VoiceInputMode}; pub use transcription::{set_huddle_transcription_enabled, start_stt_pipeline}; @@ -91,7 +94,7 @@ use agent_tts_routing::{ pub use pipeline::check_pipeline_hotstart; use pipeline::{ await_inflight_tts_start, maybe_start_stt_pipeline, maybe_start_tts_pipeline, - post_connect_setup, start_auto_enabled_transcription, PostConnectOutcome, + post_connect_setup, PostConnectOutcome, }; use relay_api::{ count_human_members, fetch_channel_members, parse_channel_uuid, validate_pubkey_hex, @@ -915,85 +918,3 @@ pub async fn speak_agent_message( eprintln!("buzz-desktop: tts stage=queue status=failed reason=closed route_id={route_id}") }) } - -/// Add an agent to the active huddle. -/// -/// Steps: -/// 1. Validates the huddle is in the Connected or Active phase. -/// 2. Adds the agent to both the ephemeral and parent channels (kind:9000). -/// 3. Only appends the agent pubkey to `agent_pubkeys` if the ephemeral add -/// succeeded — failed adds (policy rejection) are NOT p-tagged. -/// -/// Returns a structured `AgentAddResult` so the frontend can surface -/// parent-channel errors without treating them as hard failures. -/// -/// The running ACP process for this agent auto-subscribes when it receives -/// the kind:9000 membership notification — no separate process spawn needed. -#[tauri::command] -pub async fn add_agent_to_huddle( - agent_pubkey: String, - state: State<'_, AppState>, -) -> Result { - validate_pubkey_hex(&agent_pubkey)?; - - let (eph_id, parent_id, huddle_generation) = { - let hs = state.huddle()?; - if !matches!(hs.phase, HuddlePhase::Connected | HuddlePhase::Active) { - return Err("no active huddle".to_string()); - } - - // Enforce agent cap on incremental adds too. - let current_agent_count = hs - .agent_pubkeys - .lock() - .unwrap_or_else(|e| e.into_inner()) - .len(); - if current_agent_count >= MAX_HUDDLE_AGENTS { - return Err(format!( - "agent limit reached: {} (max {})", - current_agent_count, MAX_HUDDLE_AGENTS - )); - } - - let eph = hs - .ephemeral_channel_id - .clone() - .ok_or("no ephemeral channel")?; - let parent = hs.parent_channel_id.clone().ok_or("no parent channel")?; - (eph, parent, hs.huddle_generation) - }; - - let eph_uuid = Uuid::parse_str(&eph_id).map_err(|e| e.to_string())?; - let parent_uuid = Uuid::parse_str(&parent_id).map_err(|e| e.to_string())?; - - // Returns Err only if the ephemeral add fails — parent failure is in the result. - let result = agents::add_agent_to_huddle(eph_uuid, parent_uuid, &agent_pubkey, &state).await?; - - // Ephemeral add succeeded — register it only if this is still the huddle - // that initiated the relay operation. - let transcription_auto_enabled = { - let mut hs = state.huddle()?; - if !hs.is_current_huddle(&eph_id, huddle_generation) { - return Ok(result); - } - let mut pubkeys = hs.agent_pubkeys.lock().unwrap_or_else(|e| e.into_inner()); - if !pubkeys.contains(&agent_pubkey) { - pubkeys.push(agent_pubkey.clone()); - } - drop(pubkeys); - if !hs.participants.contains(&agent_pubkey) { - hs.participants.push(agent_pubkey.clone()); - } - hs.maybe_auto_enable_transcription_for_agents() - }; - - // No guidelines re-post needed — the agent sees the original kind:48106 - // guidelines via EOSE replay when it subscribes to the ephemeral channel. - if transcription_auto_enabled { - start_auto_enabled_transcription(&state, &eph_id).await; - } else { - state.emit_huddle_state_changed(); - } - - Ok(result) -} diff --git a/desktop/src-tauri/src/huddle/pipeline.rs b/desktop/src-tauri/src/huddle/pipeline.rs index b05b6b7fe47..afa7aed8e05 100644 --- a/desktop/src-tauri/src/huddle/pipeline.rs +++ b/desktop/src-tauri/src/huddle/pipeline.rs @@ -277,10 +277,6 @@ pub(crate) async fn post_connect_setup( /// /// Returns `Ok(true)` if the pipeline was started, `Ok(false)` if models are /// not ready (voice-only mode), or `Err` on a real failure. -/// -/// Creates the shared `tts_active` flag and passes it to the STT pipeline -/// for barge-in / echo gating. The same flag is later passed to the TTS -/// pipeline so it can signal when audio is playing. pub(crate) async fn maybe_start_stt_pipeline( state: &AppState, ephemeral_channel_id: &str, @@ -309,7 +305,6 @@ pub(crate) async fn maybe_start_stt_pipeline( // Take the old pipeline OUT of the lock before dropping — Drop joins // the worker thread (~200ms) and must not block under the mutex. let ( - tts_active, agent_pubkeys_arc, session_gen, expected_generation, @@ -345,7 +340,6 @@ pub(crate) async fn maybe_start_stt_pipeline( None }; ( - Arc::clone(&hs.tts_active), Arc::clone(&hs.agent_pubkeys), Arc::clone(&hs.session_generation), hs.session_generation.load(Ordering::Acquire), @@ -359,12 +353,7 @@ pub(crate) async fn maybe_start_stt_pipeline( drop(old_stt); let constructed = tokio::task::spawn_blocking(move || { - stt::SttPipeline::new( - model_dir, - tts_active, - ptt_active_for_stt, - manual_mic_unmuted_for_stt, - ) + stt::SttPipeline::new(model_dir, ptt_active_for_stt, manual_mic_unmuted_for_stt) }) .await; let (pipeline, text_rx) = match constructed { diff --git a/desktop/src-tauri/src/huddle/preprocessing.rs b/desktop/src-tauri/src/huddle/preprocessing.rs index ce85e3145e3..8eeddc2bea0 100644 --- a/desktop/src-tauri/src/huddle/preprocessing.rs +++ b/desktop/src-tauri/src/huddle/preprocessing.rs @@ -12,87 +12,6 @@ //! → numbers → words → "forty two" //! → collapse whitespace → clean string //! ``` -//! -//! Also provides `split_sentences` — the single sentence-boundary splitter used -//! by both the TTS batching pipeline and the Supertonic text chunker. - -use regex::Regex; -use std::sync::LazyLock; - -// ── Sentence splitting ──────────────────────────────────────────────────────── - -/// Regex: a sentence-ending punctuation mark followed by whitespace. -static RE_SENTENCE_BOUNDARY: LazyLock = LazyLock::new(|| Regex::new(r"([.!?])\s+").unwrap()); - -/// Common abbreviations that end with a period but are NOT sentence boundaries. -const ABBREVIATIONS: &[&str] = &[ - "Dr.", "Mr.", "Mrs.", "Ms.", "Prof.", "Sr.", "Jr.", "St.", "Ave.", "Rd.", "Blvd.", "Dept.", - "Inc.", "Ltd.", "Co.", "Corp.", "etc.", "vs.", "i.e.", "e.g.", "Ph.D.", -]; - -/// Split text into sentence-sized chunks. -/// -/// Combines regex-based boundary detection with: -/// - Abbreviation awareness (`Dr.`, `Mr.`, etc. don't split) -/// - Digit-before-period check (avoids splitting `1.` `2.` numbered lists) -/// - `\n` and `—` treated as sentence breaks -/// -/// Returns non-empty, trimmed strings. -pub fn split_sentences(text: &str) -> Vec { - // First, split on newlines and em-dashes to get coarse segments. - let coarse: Vec<&str> = text.split(['\n', '—']).collect(); - - let mut sentences = Vec::new(); - - for segment in coarse { - let segment = segment.trim(); - if segment.is_empty() { - continue; - } - // Within each segment, split on sentence-ending punctuation. - let matches: Vec<_> = RE_SENTENCE_BOUNDARY.find_iter(segment).collect(); - if matches.is_empty() { - sentences.push(segment.to_string()); - continue; - } - - let mut last_end = 0usize; - for m in &matches { - let before = &segment[last_end..m.start()]; - let punc_char = &segment[m.start()..m.start() + 1]; - - // Skip if this looks like an abbreviation. - let combined = format!("{}{}", before.trim(), punc_char); - let is_abbrev = ABBREVIATIONS.iter().any(|a| combined.ends_with(a)); - - // Skip if the character before the period is a digit (numbered list). - let is_digit_period = punc_char == "." - && !before.is_empty() - && before.ends_with(|c: char| c.is_ascii_digit()); - - if !is_abbrev && !is_digit_period { - let piece = segment[last_end..m.end()].trim(); - if !piece.is_empty() { - sentences.push(piece.to_string()); - } - last_end = m.end(); - } - } - - if last_end < segment.len() { - let tail = segment[last_end..].trim(); - if !tail.is_empty() { - sentences.push(tail.to_string()); - } - } - } - - if sentences.is_empty() { - vec![text.to_string()] - } else { - sentences - } -} // ── Public API ──────────────────────────────────────────────────────────────── @@ -602,49 +521,6 @@ mod tests { assert_eq!(out, "hello world"); } - #[test] - fn split_sentences_basic() { - let result = split_sentences("Hello world. How are you? I'm fine!"); - assert_eq!(result, vec!["Hello world.", "How are you?", "I'm fine!"]); - } - - #[test] - fn split_sentences_newline_break() { - let result = split_sentences("First line.\nSecond line."); - assert_eq!(result, vec!["First line.", "Second line."]); - } - - #[test] - fn split_sentences_em_dash_break() { - let result = split_sentences("Start here—then continue."); - assert_eq!(result, vec!["Start here", "then continue."]); - } - - #[test] - fn split_sentences_abbreviations() { - let result = split_sentences("Dr. Smith went home. He was tired."); - assert_eq!(result, vec!["Dr. Smith went home.", "He was tired."]); - } - - #[test] - fn split_sentences_numbered_list() { - let result = split_sentences("1. First item. 2. Second item."); - // "1." and "2." should NOT cause a split (digit before period). - assert_eq!(result, vec!["1. First item.", "2. Second item."]); - } - - #[test] - fn split_sentences_single() { - let result = split_sentences("Just one sentence"); - assert_eq!(result, vec!["Just one sentence"]); - } - - #[test] - fn split_sentences_empty() { - let result = split_sentences(""); - assert_eq!(result, vec![""]); - } - #[test] fn filters_trivial_responses() { assert_eq!(preprocess_for_tts("."), ""); diff --git a/desktop/src-tauri/src/huddle/state.rs b/desktop/src-tauri/src/huddle/state.rs index 7acf5fe633b..c615ff19c2e 100644 --- a/desktop/src-tauri/src/huddle/state.rs +++ b/desktop/src-tauri/src/huddle/state.rs @@ -137,6 +137,8 @@ pub struct HuddleState { pub ptt_active: Arc, /// True while the clickable microphone control is manually unmuted. /// In PTT mode, either this flag or `ptt_active` opens the STT gate. + /// Defaults to muted so push-to-talk actually gates the microphone + /// until the user explicitly opens it. #[serde(skip)] pub manual_mic_unmuted: Arc, } @@ -226,7 +228,7 @@ impl Default for HuddleState { session_generation: Arc::new(AtomicU64::new(0)), voice_input_mode: VoiceInputMode::default(), ptt_active: Arc::new(AtomicBool::new(false)), - manual_mic_unmuted: Arc::new(AtomicBool::new(true)), + manual_mic_unmuted: Arc::new(AtomicBool::new(false)), } } } @@ -339,10 +341,10 @@ mod tests { } #[test] - fn defaults_to_push_to_talk_with_an_open_microphone() { + fn defaults_to_push_to_talk_with_a_muted_microphone() { let state = HuddleState::default(); assert_eq!(state.voice_input_mode, super::VoiceInputMode::PushToTalk); - assert!(state.manual_mic_unmuted.load(Ordering::Acquire)); + assert!(!state.manual_mic_unmuted.load(Ordering::Acquire)); } #[test] diff --git a/desktop/src-tauri/src/huddle/stt.rs b/desktop/src-tauri/src/huddle/stt.rs index 70a80886402..19a28b150b3 100644 --- a/desktop/src-tauri/src/huddle/stt.rs +++ b/desktop/src-tauri/src/huddle/stt.rs @@ -61,15 +61,11 @@ pub struct SttPipeline { impl SttPipeline { /// Spawn the pipeline thread. /// - /// `tts_active` is a shared flag set by the TTS pipeline while audio is - /// playing. The STT worker uses it to: - /// - discard accumulated speech so local playback cannot feed back into STT - /// - apply a cooldown after TTS stops before re-enabling STT - /// - /// Open-mic VAD cannot distinguish a nearby human from the app's own native - /// TTS playback because it has no acoustic echo reference. Local mic frames - /// therefore never cancel TTS. Push-to-talk and remote participant speech - /// remain explicit, reliable barge-in paths. + /// Mic input is transcribed even while agent TTS is playing: the huddle UI + /// already tells users to wear headphones, so speaker bleed is accepted in + /// exchange for never dropping human speech that overlaps agent audio. + /// Local mic frames still never cancel TTS — push-to-talk and remote + /// participant speech remain the explicit barge-in paths. /// /// `ptt_active` and `manual_mic_unmuted` are present when the PTT shortcut /// is enabled. The pipeline accepts speech while either input path is open; @@ -86,7 +82,6 @@ impl SttPipeline { /// thread on every `recv_timeout` call). pub fn new( model_dir: PathBuf, - tts_active: Arc, ptt_active: Option>, manual_mic_unmuted: Option>, ) -> Result<(Self, tokio_mpsc::Receiver), String> { @@ -105,7 +100,6 @@ impl SttPipeline { audio_rx, text_tx, shutdown_worker, - tts_active, ptt_active_worker, manual_mic_unmuted_worker, ) @@ -166,6 +160,11 @@ impl Drop for SttPipeline { /// How many 16 kHz samples of silence before we flush to STT. /// 300 ms × 16 000 Hz / 256 samples-per-frame ≈ 19 frames. /// Previous value (28 frames / 450 ms) felt sluggish in conversation. +/// +/// This window is a turn-taking quality knob, not a latency lever: an earlier +/// env override (`BUZZ_STT_FLUSH_MS`) let it be lowered to 150 ms, which split +/// natural mid-sentence pauses into separate messages and confused the +/// listening agents. Reverted — the window is fixed at the production value. const SILENCE_FLUSH_FRAMES: usize = 19; /// earshot requires exactly 256 samples per frame at 16 kHz. @@ -183,12 +182,6 @@ const MIN_VOICED_FRAMES: usize = 12; /// How long the worker waits on the audio channel before checking the shutdown flag. const RECV_TIMEOUT: Duration = Duration::from_millis(50); -/// 150 ms cooldown after TTS stops before STT re-enables. -/// Prevents the tail of TTS audio from being transcribed as speech. -/// This remains shorter than the previous 200 ms gate that ate the first word, -/// but is long enough for speaker/AEC tail audio to leave the microphone path. -const TTS_COOLDOWN: Duration = Duration::from_millis(150); - /// Number of ONNX Runtime intra-op threads used by the offline recognizer. /// /// Held at 1 (conservative) until we have a local A/B on real huddle audio. @@ -200,12 +193,31 @@ const TTS_COOLDOWN: Duration = Duration::from_millis(150); /// shows it's safe on the minimum-spec target. const STT_NUM_THREADS: i32 = 1; +/// EXPERIMENTAL (latency bench): override recognizer intra-op threads via +/// `BUZZ_STT_THREADS`. Default preserves the production single thread. +fn stt_num_threads() -> i32 { + std::env::var("BUZZ_STT_THREADS") + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|&n| n >= 1) + .unwrap_or(STT_NUM_THREADS) +} + +/// EXPERIMENTAL (latency bench): `BUZZ_STT_SPECULATIVE=1` starts the Parakeet +/// decode at the FIRST silent VAD frame instead of after the full flush +/// window, overlapping the ~150-250 ms decode with the silence wait. If +/// speech resumes, the speculative result is discarded. When silence holds +/// to the flush threshold the transcript is emitted immediately, so the STT +/// leg collapses to ~max(flush window, decode time). +fn stt_speculative_decode() -> bool { + std::env::var("BUZZ_STT_SPECULATIVE").is_ok_and(|v| v == "1") +} + fn stt_worker( model_dir: PathBuf, audio_rx: Receiver>, text_tx: tokio_mpsc::Sender, shutdown: Arc, - tts_active: Arc, ptt_active: Option>, manual_mic_unmuted: Option>, ) { @@ -248,7 +260,7 @@ fn stt_worker( let mut cfg = OfflineRecognizerConfig::default(); cfg.model_config.nemo_ctc.model = Some(model_path.to_string_lossy().into_owned()); cfg.model_config.tokens = Some(tokens_path.to_string_lossy().into_owned()); - cfg.model_config.num_threads = STT_NUM_THREADS; + cfg.model_config.num_threads = stt_num_threads(); // Explicit — defaults are not part of the API contract, and noisy debug // logging in release builds would be expensive on every VAD chunk. cfg.model_config.debug = false; @@ -275,11 +287,14 @@ fn stt_worker( let mut in_speech = false; // Number of frames earshot classified as voiced in the current segment. let mut voiced_frames = 0; - // Timestamp when TTS last stopped — used for the playback-tail cooldown. - let mut tts_stopped_at: Option = None; + // Silence flush window (frames) — fixed at the production value. + let flush_frames = SILENCE_FLUSH_FRAMES; + // EXPERIMENTAL: speculative decode result + the voiced-frame count it was + // computed at. Valid only while no new voiced frame has arrived since. + let speculative_enabled = stt_speculative_decode(); + let mut speculative: Option<(String, usize)> = None; // ── 5. Main loop ────────────────────────────────────────────────────────── - let mut tts_was_active = false; let mut transmit_was_active = ptt_active .as_ref() .is_some_and(|ptt| ptt.load(Ordering::Acquire)) @@ -292,14 +307,6 @@ fn stt_worker( break; } - // Track TTS transitions to set the cooldown timer. - let tts_now = tts_active.load(Ordering::Acquire); - if tts_was_active && !tts_now { - // TTS just stopped — record the timestamp for the cooldown window. - tts_stopped_at = Some(std::time::Instant::now()); - } - tts_was_active = tts_now; - // Track the combined manual/PTT transmission edge. When both paths // close, the worklet stops sending frames, so flush here rather than // waiting for silence that will never arrive. @@ -348,10 +355,10 @@ fn stt_worker( &mut silence_frames, &mut in_speech, &mut voiced_frames, + flush_frames, + (speculative_enabled, &mut speculative), &recognizer, &text_tx, - &tts_active, - &mut tts_stopped_at, ptt_active.as_ref(), manual_mic_unmuted.as_ref(), ); @@ -391,14 +398,16 @@ fn resample_chunk(resampler: &mut rubato::Fft, chunk_48k: &[f32]) -> Vec), recognizer: &sherpa_onnx::OfflineRecognizer, text_tx: &tokio_mpsc::Sender, - tts_active: &Arc, - tts_stopped_at: &mut Option, ptt_active: Option<&Arc>, manual_mic_unmuted: Option<&Arc>, ) { + let (speculative_enabled, speculative) = speculative; leftover.extend_from_slice(samples); while leftover.len() >= VAD_FRAME_SAMPLES { @@ -424,54 +434,26 @@ fn process_16k_samples( let is_speech = prob > VAD_THRESHOLD; let manually_open = manual_mic_unmuted.is_some_and(|manual| manual.load(Ordering::Acquire)); + let ptt_held = ptt_active.is_some_and(|ptt| ptt.load(Ordering::Acquire)); // Shortcut-enabled mode accepts input from either the held shortcut or // a manually open microphone. - let is_speech = if let Some(ptt) = ptt_active { - is_speech && (ptt.load(Ordering::Acquire) || manually_open) + let is_speech = if ptt_active.is_some() { + is_speech && (ptt_held || manually_open) } else { is_speech }; - - let tts_playing = tts_active.load(Ordering::Acquire); - - // While TTS is playing, discard local mic input. The native TTS output - // is not available as an echo-cancellation reference to this worker, so - // VAD cannot reliably tell speaker feedback from a human interruption. - // Push-to-talk and remote participant audio provide the intentional - // cancellation paths instead. - if tts_playing { - *in_speech = false; - speech_buf.clear(); - *silence_frames = 0; - *voiced_frames = 0; - continue; - } - - // TTS not playing — check cooldown window. - if let Some(stopped) = *tts_stopped_at { - if stopped.elapsed() < TTS_COOLDOWN { - // Still in cooldown — discard but keep tracking speech state. - if !is_speech { - *in_speech = false; - } - speech_buf.clear(); - *silence_frames = 0; - *voiced_frames = 0; - continue; - } else { - // Cooldown expired — clear the timer and reset all segment state. - *tts_stopped_at = None; - *in_speech = false; - *silence_frames = 0; - *voiced_frames = 0; - } - } + // A held shortcut means "I am not done talking": silence never ends + // the utterance while it is held. VAD pause flushing applies in pure + // VAD mode, or with a manually open mic once the shortcut is up. + let vad_flush_allowed = vad_flush_allowed(ptt_active.is_some(), manually_open, ptt_held); if is_speech { *silence_frames = 0; *in_speech = true; *voiced_frames += 1; speech_buf.extend_from_slice(&frame); + // New voiced audio invalidates any speculative decode. + speculative.take(); // OOM guard: flush and reset if the buffer exceeds 30 s of audio. if speech_buf.len() >= MAX_SPEECH_SAMPLES { @@ -486,11 +468,29 @@ fn process_16k_samples( speech_buf.extend_from_slice(&frame); *silence_frames += 1; - // A manually open microphone behaves like normal VAD. A - // shortcut-only transmission stays grouped until key release. - if (ptt_active.is_none() || manually_open) && *silence_frames >= SILENCE_FLUSH_FRAMES { - // End of utterance — transcribe. - flush_to_stt(speech_buf, *voiced_frames, recognizer, text_tx); + // EXPERIMENTAL: kick the Parakeet decode at the first silent + // frame so it overlaps the flush window. speech_buf keeps + // accumulating silence afterwards, but trailing silence does not + // change the transcript; any resumed speech invalidates the + // speculative result above. + if speculative_enabled + && speculative.is_none() + && vad_flush_allowed + && has_enough_voiced_audio(*voiced_frames) + { + speculative.replace((decode_speech(recognizer, speech_buf), *voiced_frames)); + } + + // A manually open microphone behaves like normal VAD. A held + // shortcut keeps the utterance grouped until key release. + if vad_flush_allowed && *silence_frames >= flush_frames { + // End of utterance — transcribe (or emit the speculative decode). + match speculative.take() { + Some((text, decoded_at)) if decoded_at == *voiced_frames => { + send_transcript(text, text_tx); + } + _ => flush_to_stt(speech_buf, *voiced_frames, recognizer, text_tx), + } speech_buf.clear(); *silence_frames = 0; *in_speech = false; @@ -514,16 +514,22 @@ fn flush_to_stt( if speech_buf.is_empty() || !has_enough_voiced_audio(voiced_frames) { return; } + send_transcript(decode_speech(recognizer, speech_buf), text_tx); +} +/// Run the Parakeet decode on a speech buffer and return the trimmed text. +fn decode_speech(recognizer: &sherpa_onnx::OfflineRecognizer, speech_buf: &[f32]) -> String { let stream = recognizer.create_stream(); stream.accept_waveform(16_000, speech_buf); recognizer.decode(&stream); - let text = stream + stream .get_result() .map(|r| r.text.trim().to_string()) - .unwrap_or_default(); + .unwrap_or_default() +} +fn send_transcript(text: String, text_tx: &tokio_mpsc::Sender) { if !text.is_empty() { if let Err(e) = text_tx.blocking_send(text) { eprintln!("buzz-desktop: STT text channel closed: {e}"); @@ -535,6 +541,17 @@ fn has_enough_voiced_audio(voiced_frames: usize) -> bool { voiced_frames >= MIN_VOICED_FRAMES } +/// Whether a silence run may end the current utterance and flush it to STT. +/// +/// Pure VAD mode (no shortcut configured) always allows pause flushing. When +/// the push-to-talk shortcut is configured, a held shortcut is an explicit +/// "I am not done talking" signal, so silence never flushes while it is held +/// — even if the microphone is also manually open. A manually open mic with +/// the shortcut up behaves like normal VAD. +fn vad_flush_allowed(ptt_mode: bool, manually_open: bool, ptt_held: bool) -> bool { + !ptt_mode || (manually_open && !ptt_held) +} + /// Convert raw bytes (f32 LE) to f32 samples. /// Caller should ensure `bytes.len() % 4 == 0`; extra bytes are silently truncated. /// @@ -553,7 +570,7 @@ use super::drain_until_shutdown; #[cfg(test)] mod tests { - use super::{has_enough_voiced_audio, MIN_VOICED_FRAMES}; + use super::{has_enough_voiced_audio, vad_flush_allowed, MIN_VOICED_FRAMES}; #[test] fn short_vad_blips_do_not_reach_the_recognizer() { @@ -561,4 +578,19 @@ mod tests { assert!(!has_enough_voiced_audio(MIN_VOICED_FRAMES - 1)); assert!(has_enough_voiced_audio(MIN_VOICED_FRAMES)); } + + #[test] + fn held_push_to_talk_never_silence_flushes() { + // Pure VAD mode: silence always ends the utterance. + assert!(vad_flush_allowed(false, false, false)); + // Shortcut configured, nothing transmitting: nothing to flush anyway, + // but the pause path stays closed. + assert!(!vad_flush_allowed(true, false, false)); + // Shortcut held: "I am not done talking" — never flush on silence, + // regardless of the manual mic state. + assert!(!vad_flush_allowed(true, false, true)); + assert!(!vad_flush_allowed(true, true, true)); + // Manually open mic with the shortcut up: normal VAD behavior. + assert!(vad_flush_allowed(true, true, false)); + } } diff --git a/desktop/src-tauri/src/huddle/tts.rs b/desktop/src-tauri/src/huddle/tts.rs index 6a56f85444c..aca2339a3c4 100644 --- a/desktop/src-tauri/src/huddle/tts.rs +++ b/desktop/src-tauri/src/huddle/tts.rs @@ -7,9 +7,9 @@ //! → bounded sync_channel (TEXT_QUEUE_DEPTH = 8) //! → tts_worker thread (owns 1 Pocket TTS engine + 1 persistent Player) //! 1. Preprocess text -//! 2. Split into sentences -//! 3. Synthesize each sentence individually → f32 PCM -//! 4. Clamp to full scale + fade out each sentence +//! 2. Split into tokenizer-safe natural units, prioritizing sentence one +//! 3. Synthesize each unit → f32 PCM +//! 4. Clamp to full scale + fade out each unit //! 5. Append each buffer to the persistent rodio Player (gapless) //! 6. While audio is draining, keep pulling queued text items and //! synthesizing ahead — playback of item N overlaps synthesis of @@ -41,7 +41,7 @@ use std::{ sync::{ atomic::{AtomicBool, AtomicU64, Ordering}, mpsc::{self, SyncSender}, - Arc, Mutex, MutexGuard, PoisonError, + Arc, Mutex, }, thread, time::{Duration, Instant}, @@ -50,7 +50,7 @@ use std::{ use super::pocket::{ load_text_to_speech, load_voice_style, DEFAULT_VOICE, SAMPLE_RATE, VOICE_FILE_EXT, }; -use super::preprocessing::{preprocess_for_tts, split_sentences}; +use super::preprocessing::preprocess_for_tts; #[path = "tts_voice_transition.rs"] mod voice_transition; @@ -69,6 +69,9 @@ mod pipeline_controls; #[path = "tts_speaker_cancellation.rs"] mod speaker_cancellation; use speaker_cancellation::*; +#[path = "tts_streaming.rs"] +mod streaming; +use streaming::*; // ── Constants ───────────────────────────────────────────────────────────────── @@ -99,38 +102,11 @@ const SYNTH_STEPS: usize = 1; /// the leading waveform is important. const FADE_OUT_SAMPLES: usize = (SAMPLE_RATE as f64 * 0.008) as usize; -/// Length of the zero-sample cushion prepended before each synthesized -/// sentence chunk, so the OS audio device / rodio mixer has a fully-quiet -/// ramp-up window before the real onset hits. -/// -/// This used to be applied only before the first sentence of a whole response. -/// That still left later sentence chunks vulnerable to first-syllable clipping -/// when their first phoneme was soft (notably `I'm` / `I've`) and rodio crossed -/// from an explicit silence buffer straight into non-zero speech. 20 ms ≈ 480 -/// samples is enough to cover a CoreAudio buffer turnover without being audible -/// as latency. At sentence boundaries this lead-in is budgeted out of the -/// existing inter-sentence pause, so it does not lengthen multi-sentence gaps. +/// Length of the zero-sample cushion prepended when playback is idle, so the +/// OS audio device / rodio mixer has a fully-quiet ramp-up window before the +/// real onset hits. Continuously queued chunks receive no synthetic padding. const SENTENCE_LEAD_IN_SAMPLES: usize = (SAMPLE_RATE as f64 * 0.020) as usize; -/// Approximate character budget for one synthesis chunk. -/// -/// Upstream pocket-tts groups sentences into chunks of up to -/// `MAX_TOKEN_PER_CHUNK = 50` tokenizer tokens (`default_parameters.py`) — -/// typically multi-sentence chunks — because every `generate()` call is an -/// independent generation with a cold FlowLM start, and each chunk boundary -/// is an exposed prosody seam (kyutai-labs/pocket-tts #151; the Kyutai team -/// names chunk stitching as the reliability lever). Our previous -/// sentence-per-call path created ~2–4× more seams than upstream. -/// -/// This character budget performs only coarse sentence packing. The April -/// engine applies its SentencePiece tokenizer afterward and refines every -/// result at the bundle's exact 50-token boundary. -const MAX_CHUNK_CHARS: usize = 200; - -/// Silence inserted between sentences by the TTS pipeline (seconds). -/// Injected as a silent buffer between each synthesized sentence chunk. -const INTER_SENTENCE_SILENCE: f32 = 0.1; - type WorkerControlState = ( Arc, Arc, @@ -450,7 +426,9 @@ fn tts_worker( // `tts_active` lifecycle: set on the first append while idle, cleared // whenever the player has fully drained — either in the idle timeout // arm or on item receipt before synthesis begins. - let silence_buf_len = (INTER_SENTENCE_SILENCE * SAMPLE_RATE as f32) as usize; + // EXPERIMENTAL (latency bench): `Some(emit_frames)` = stream PCM deltas + // out of Pocket as they are generated (see tts_streaming.rs). + let tts_streaming = streaming_emit_frames(); // `first_append` = "no audio queued since the player last went idle". // Flipped by `build_sentence_append_buffer` on the first real append; the // idle branch below uses it to decide when to drop `tts_active` and to @@ -705,17 +683,20 @@ fn tts_worker( continue; } - // Split into sentences, then group into synthesis chunks: the first - // sentence stays alone (fast time-to-first-audio), the rest pack - // greedily up to MAX_CHUNK_CHARS. Playback of each model unit overlaps - // synthesis of the next one. The Pocket engine applies its exact - // 50-token split; keeping those units within one playback chunk avoids - // adding fades and pauses at token-only boundaries. - let sentences: Vec = split_sentences(&text) - .into_iter() - .filter(|s| !s.trim().is_empty()) - .collect(); - let chunks = group_sentences_into_chunks(&sentences, MAX_CHUNK_CHARS); + // Let Pocket's tokenizer-aware splitter isolate the first sentence for + // minimum time-to-first-audio, then pack later sentences into the + // largest natural units within the model's exact 50-token limit. Once + // each unit is appended, generation of the next proceeds while rodio + // plays the already-queued audio. + let chunks = match engine.split_text_for_playback(&text) { + Ok(chunks) => chunks, + Err(_) => { + eprintln!( + "buzz-desktop: tts stage=synthesis status=failed reason=chunking route_id={route_id}" + ); + continue; + } + }; if chunks.is_empty() { eprintln!( "buzz-desktop: tts stage=synthesis status=empty reason=no_chunks route_id={route_id}" @@ -747,6 +728,41 @@ fn tts_worker( continue; } + // EXPERIMENTAL (latency bench): streaming synthesis path — see + // tts_streaming.rs for the mechanics and exactness constraints. + if let Some(emit_frames) = tts_streaming { + let outcome = synthesize_streaming( + &engine, + text, + &style, + emit_frames, + (&cancel, &voice_cancel, &shutdown), + StreamingPlayback { + player: &player, + first_append: &mut first_append, + route_id, + }, + &mut |prepared| { + if !append_audio( + prepared, + route_id, + speaker_pubkey.as_deref(), + speaker_generation, + ) { + return false; + } + appended_audio = true; + last_route_id = route_id; + true + }, + ); + if let Some(outcome) = outcome { + synthesis_outcome = outcome; + break 'playback_chunks; + } + continue; + } + let model_chunks = match engine.split_text_into_chunks(text) { Ok(model_chunks) => model_chunks, Err(_) => { @@ -811,7 +827,6 @@ fn tts_worker( samples, chunk_index, &mut first_append, - silence_buf_len, player.empty(), ) { if !append_audio( @@ -842,9 +857,7 @@ fn tts_worker( } } } - if let Some(prepared) = - playback_audio.finish(&mut first_append, silence_buf_len, player.empty()) - { + if let Some(prepared) = playback_audio.finish(&mut first_append, player.empty()) { if !append_audio( prepared, route_id, @@ -882,90 +895,6 @@ fn tts_worker( tts_active.store(false, Ordering::Release); } -// ── Helpers ─────────────────────────────────────────────────────────────────── - -/// Check for cancel or shutdown. Returns `true` if the caller should break/continue. -/// On cancel: drains the text queue and clears the cancel flag. -/// -/// `player` pairs the Player with the `player_ops` mutex shared with the -/// barge-in monitor thread; the cancel/shutdown clear runs under that lock so -/// it is serialized with the monitor's stale-branch re-check (see the monitor -/// block in `tts_worker`). -fn handle_cancel_or_shutdown( - cancel_signals: CancelSignals<'_>, - shutdown: &AtomicBool, - tts_active: &AtomicBool, - text_state: CancelTextState<'_>, - voice_change_ack: &VoiceChangeAck, - active_route_id: Option, - player: Option<(&rodio::Player, &Mutex<()>)>, -) -> bool { - let (cancel, voice_cancel) = cancel_signals; - let (text_rx, deferred_text, current_text) = text_state; - if shutdown.load(Ordering::Acquire) { - eprintln!( - "buzz-desktop: tts stage=cancellation reason=shutdown route_id={}", - active_route_id.unwrap_or(0) - ); - if let Some((p, ops)) = player { - let _ops = lock_player_ops(ops); - p.clear(); - } - tts_active.store(false, Ordering::Release); - return true; - } - if cancel.load(Ordering::Acquire) || voice_cancel.load(Ordering::Acquire) { - // Serialize with begin_voice_change so the generation boundary and - // cancel consumption are observed as one transition. - let pending_voice_change = voice_change_ack - .lock() - .unwrap_or_else(|error| error.into_inner()); - // Consume at the serialization point. A later barge-in remains true - // for the next pass instead of being overwritten after queue cleanup. - let barge_in = cancel.swap(false, Ordering::AcqRel); - voice_cancel.store(false, Ordering::Release); - eprintln!( - "buzz-desktop: tts stage=cancellation reason={} route_id={}", - if barge_in { "barge_in" } else { "voice_switch" }, - active_route_id.unwrap_or(0) - ); - let preserve_generation = (!barge_in) - .then(|| { - pending_voice_change - .as_ref() - .map(|pending| pending.generation) - }) - .flatten(); - retain_cancelled_text(deferred_text, current_text, text_rx, preserve_generation); - if let Some((p, ops)) = player { - let _ops = lock_player_ops(ops); - // `Player::clear()` removes queued sources AND pauses the player - // (rodio 0.22 `clear()` ends with `self.pause()`). With one - // persistent Player for the worker's lifetime, the un-pause is - // mandatory: without `play()`, every append after a barge-in - // would queue silently forever. - p.clear(); - p.play(); - // Consume the flag under the lock: once released with - // `cancel == false`, the monitor's stale branch no-ops instead - // of clearing the fresh post-cancel utterance. - } - tts_active.store(false, Ordering::Release); - return true; - } - false -} - -/// Acquire the `player_ops` lock, recovering from poison. -/// -/// The data under the mutex is `()` — it only serializes Player mutations — -/// so a panicked holder leaves nothing inconsistent to observe and recovery -/// is always safe. Without this, a worker panic would wedge the monitor (or -/// vice versa) on `unwrap()`. -fn lock_player_ops(ops: &Mutex<()>) -> MutexGuard<'_, ()> { - ops.lock().unwrap_or_else(PoisonError::into_inner) -} - // ── Tests ───────────────────────────────────────────────────────────────────── #[cfg(test)] diff --git a/desktop/src-tauri/src/huddle/tts_audio.rs b/desktop/src-tauri/src/huddle/tts_audio.rs index 58300b7497e..80bf0c4661c 100644 --- a/desktop/src-tauri/src/huddle/tts_audio.rs +++ b/desktop/src-tauri/src/huddle/tts_audio.rs @@ -10,15 +10,11 @@ pub(super) struct PreparedModelAudio { /// on the first and last unit that actually produced audio. pub(super) struct PlaybackChunkAudio { pending: Option<(Vec, usize)>, - appended: bool, } impl PlaybackChunkAudio { pub(super) fn new() -> Self { - Self { - pending: None, - appended: false, - } + Self { pending: None } } pub(super) fn push( @@ -26,36 +22,26 @@ impl PlaybackChunkAudio { samples: Vec, chunk_index: usize, first_append: &mut bool, - silence_buf_len: usize, playback_idle: bool, ) -> Option { if samples.is_empty() { return None; } let previous = self.pending.replace((samples, chunk_index))?; - let prepared = prepare_model_audio( - previous, - first_append, - silence_buf_len, - !self.appended || playback_idle, - false, - ); - self.appended = true; + let prepared = prepare_model_audio(previous, first_append, playback_idle, false); Some(prepared) } pub(super) fn finish( &mut self, first_append: &mut bool, - silence_buf_len: usize, playback_idle: bool, ) -> Option { let pending = self.pending.take()?; Some(prepare_model_audio( pending, first_append, - silence_buf_len, - !self.appended || playback_idle, + playback_idle, true, )) } @@ -64,7 +50,6 @@ impl PlaybackChunkAudio { fn prepare_model_audio( (samples, chunk_index): (Vec, usize), first_append: &mut bool, - silence_buf_len: usize, starts_playback_chunk: bool, ends_playback_chunk: bool, ) -> PreparedModelAudio { @@ -74,13 +59,7 @@ fn prepare_model_audio( apply_fade_out(&mut audio); } PreparedModelAudio { - buffer: build_sentence_append_buffer( - first_append, - audio, - silence_buf_len, - starts_playback_chunk, - ends_playback_chunk, - ), + buffer: build_sentence_append_buffer(first_append, audio, starts_playback_chunk), sample_count, chunk_index, } @@ -103,9 +82,7 @@ pub(super) fn apply_fade_out(samples: &mut [f32]) { pub(super) fn build_sentence_append_buffer( first_append: &mut bool, audio: Vec, - silence_buf_len: usize, starts_playback_chunk: bool, - ends_playback_chunk: bool, ) -> Vec { if *first_append { *first_append = false; @@ -116,117 +93,73 @@ pub(super) fn build_sentence_append_buffer( } else { 0 }; - let trailing_silence_len = if ends_playback_chunk { - silence_buf_len.saturating_sub(SENTENCE_LEAD_IN_SAMPLES) - } else { - 0 - }; - let mut buffer = Vec::with_capacity(lead_in_len + audio.len() + trailing_silence_len); + let mut buffer = Vec::with_capacity(lead_in_len + audio.len()); buffer.extend(std::iter::repeat_n(0.0_f32, lead_in_len)); buffer.extend(audio); - buffer.extend(std::iter::repeat_n(0.0_f32, trailing_silence_len)); buffer } -pub(super) fn group_sentences_into_chunks(sentences: &[String], max_chars: usize) -> Vec { - let mut chunks: Vec = Vec::new(); - for (index, sentence) in sentences.iter().enumerate() { - let sentence = sentence.trim(); - if sentence.is_empty() { - continue; - } - if index == 0 || chunks.is_empty() { - chunks.push(sentence.to_string()); - continue; - } - let can_merge = chunks.len() > 1 - && chunks - .last() - .is_some_and(|chunk| chunk.len() + 1 + sentence.len() <= max_chars); - if can_merge { - if let Some(last) = chunks.last_mut() { - last.push(' '); - last.push_str(sentence); - } - } else { - chunks.push(sentence.to_string()); - } - } - chunks -} - #[cfg(test)] mod tests { use super::*; #[test] - fn multi_unit_audio_decorates_only_outer_playback_boundaries() { + fn model_units_are_queued_contiguously_without_injected_silence() { let mut chunk = PlaybackChunkAudio::new(); let mut first_append = true; - let silence = SENTENCE_LEAD_IN_SAMPLES + 100; assert!(chunk - .push(vec![0.4; 16], 0, &mut first_append, silence, false) + .push(vec![0.4; 16], 0, &mut first_append, false) .is_none()); let first = chunk - .push(vec![0.5; 16], 1, &mut first_append, silence, false) + .push(vec![0.5; 16], 1, &mut first_append, false) .expect("first ready model unit"); - assert_eq!(first.buffer.len(), SENTENCE_LEAD_IN_SAMPLES + 16); - assert!(first.buffer[..SENTENCE_LEAD_IN_SAMPLES] - .iter() - .all(|sample| *sample == 0.0)); - assert_eq!(first.buffer[SENTENCE_LEAD_IN_SAMPLES], 0.4); + assert_eq!(first.buffer, vec![0.4; 16]); let last = chunk - .finish(&mut first_append, silence, false) + .finish(&mut first_append, false) .expect("last ready model unit"); - assert_eq!(last.buffer.len(), 16 + 100); - assert_eq!(last.buffer.last(), Some(&0.0)); + assert_eq!(last.buffer.len(), 16); + assert_eq!(last.sample_count, 16); } #[test] - fn empty_edge_units_do_not_steal_lead_in_or_trailing_boundary() { + fn empty_edge_units_do_not_steal_audio_boundaries() { let mut chunk = PlaybackChunkAudio::new(); let mut first_append = true; - let silence = SENTENCE_LEAD_IN_SAMPLES + 100; assert!(chunk - .push(Vec::new(), 0, &mut first_append, silence, false) + .push(Vec::new(), 0, &mut first_append, false) .is_none()); assert!(chunk - .push(vec![0.5; 16], 1, &mut first_append, silence, false) + .push(vec![0.5; 16], 1, &mut first_append, false) .is_none()); assert!(chunk - .push(Vec::new(), 2, &mut first_append, silence, false) + .push(Vec::new(), 2, &mut first_append, false) .is_none()); let only = chunk - .finish(&mut first_append, silence, false) + .finish(&mut first_append, false) .expect("only audible model unit"); - assert_eq!(only.buffer.len(), SENTENCE_LEAD_IN_SAMPLES + 16 + 100); - assert!(only.buffer[..SENTENCE_LEAD_IN_SAMPLES] - .iter() - .all(|sample| *sample == 0.0)); - assert_eq!(only.buffer.last(), Some(&0.0)); + assert_eq!(only.buffer.len(), 16); } #[test] fn playback_underrun_rearms_the_onset_cushion() { let mut chunk = PlaybackChunkAudio::new(); let mut first_append = true; - let silence = SENTENCE_LEAD_IN_SAMPLES + 100; assert!(chunk - .push(vec![0.4; 16], 0, &mut first_append, silence, false) + .push(vec![0.4; 16], 0, &mut first_append, false) .is_none()); let first = chunk - .push(vec![0.5; 16], 1, &mut first_append, silence, false) - .expect("first model unit"); - assert_eq!(first.buffer.len(), SENTENCE_LEAD_IN_SAMPLES + 16); + .push(vec![0.5; 16], 1, &mut first_append, false) + .expect("first ready model unit"); + assert_eq!(first.buffer.len(), 16); let after_underrun = chunk - .push(vec![0.6; 16], 2, &mut first_append, silence, true) - .expect("model unit after underrun"); + .push(vec![0.6; 16], 2, &mut first_append, true) + .expect("second ready model unit"); assert_eq!(after_underrun.buffer.len(), SENTENCE_LEAD_IN_SAMPLES + 16); assert!(after_underrun.buffer[..SENTENCE_LEAD_IN_SAMPLES] .iter() diff --git a/desktop/src-tauri/src/huddle/tts_streaming.rs b/desktop/src-tauri/src/huddle/tts_streaming.rs new file mode 100644 index 00000000000..2bb401c43f5 --- /dev/null +++ b/desktop/src-tauri/src/huddle/tts_streaming.rs @@ -0,0 +1,104 @@ +//! EXPERIMENTAL (latency bench): streaming synthesis path for the TTS worker. +//! +//! `BUZZ_TTS_STREAMING=1` streams PCM deltas out of Pocket as they are +//! generated instead of waiting for the full first-chunk synthesis. +//! `BUZZ_TTS_EMIT_FRAMES` tunes the delta size in Flow LM frames (80 ms of +//! audio each). Default 12 = the Mimi decoder's native chunk, which keeps +//! streamed audio bit-identical to the batch path; smaller deltas are faster +//! to first audio but diverge (~23 dB SNR vs batch — decoder intra-chunk +//! lookahead). + +use super::*; + +use crate::huddle::pocket::{PocketTts, VoiceStyle}; + +/// Read the streaming env overrides once per worker: `Some(emit_frames)` +/// when `BUZZ_TTS_STREAMING=1`, `None` for the production batch path. +pub(super) fn streaming_emit_frames() -> Option { + std::env::var("BUZZ_TTS_STREAMING") + .is_ok_and(|v| v == "1") + .then(|| { + std::env::var("BUZZ_TTS_EMIT_FRAMES") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(12) + }) +} + +/// Playback context threaded through one streamed chunk. +pub(super) struct StreamingPlayback<'a> { + pub(super) player: &'a rodio::Player, + pub(super) first_append: &'a mut bool, + pub(super) route_id: u64, +} + +/// Synthesize one text chunk through `synth_chunk_streaming`, appending PCM +/// deltas to the player as they are generated so first audio lands after +/// ~`emit_frames` of generation instead of after the whole first-chunk +/// synthesis. Delta boundary decoration reuses `PlaybackChunkAudio`: lead-in +/// on the first delta, fade-out only on the final one. +/// +/// `signals` = (cancel, voice_cancel, shutdown); `append_audio` returns +/// `false` to abort (its own cancellation checks and logging apply). Returns +/// `None` on success or `Some(outcome)` — the worker's `synthesis_outcome` +/// label — when the chunk was cancelled or failed. +pub(super) fn synthesize_streaming( + engine: &PocketTts, + text: &str, + style: &VoiceStyle, + emit_frames: usize, + signals: (&AtomicBool, &AtomicBool, &AtomicBool), + playback: StreamingPlayback<'_>, + append_audio: &mut dyn FnMut(PreparedModelAudio) -> bool, +) -> Option<&'static str> { + let (cancel, voice_cancel, shutdown) = signals; + let StreamingPlayback { + player, + first_append, + route_id, + } = playback; + let mut playback_audio = PlaybackChunkAudio::new(); + let mut delta_index = 0usize; + let stream_result = engine.synth_chunk_streaming(text, style, emit_frames, &mut |samples| { + if cancel.load(Ordering::Acquire) + || voice_cancel.load(Ordering::Acquire) + || shutdown.load(Ordering::Acquire) + { + return false; + } + let chunk_index = delta_index; + delta_index += 1; + if let Some(prepared) = + playback_audio.push(samples, chunk_index, first_append, player.empty()) + { + if !append_audio(prepared) { + return false; + } + } + true + }); + match stream_result { + Ok(true) => { + if let Some(prepared) = playback_audio.finish(first_append, player.empty()) { + if !append_audio(prepared) { + *first_append = true; + return Some("cancelled"); + } + } + None + } + Ok(false) => { + eprintln!( + "buzz-desktop: tts stage=synthesis status=cancelled reason=stream_callback route_id={route_id}" + ); + *first_append = true; + Some("cancelled") + } + Err(_) => { + eprintln!( + "buzz-desktop: tts stage=synthesis status=failed reason=inference route_id={route_id}" + ); + Some("failed") + } + } +} diff --git a/desktop/src-tauri/src/huddle/tts_tests.rs b/desktop/src-tauri/src/huddle/tts_tests.rs index 1dee4de90cc..50e4d17ced5 100644 --- a/desktop/src-tauri/src/huddle/tts_tests.rs +++ b/desktop/src-tauri/src/huddle/tts_tests.rs @@ -785,98 +785,63 @@ fn apply_fade_out_single_sample() { // ── build_sentence_append_buffer tests ─────────────────────────────────── -/// REGRESSION: every chunk needs an onset cushion; synthesized chunks -/// can start with speech energy within the first millisecond. -#[test] -fn lead_in_pad_is_present_for_every_sentence_chunk() { - const SENTENCE_AUDIO_LEN: usize = 1000; - const SILENCE_BUF_LEN: usize = 2400; // 100 ms at 24 kHz, like production - const N_SENTENCES: usize = 5; - - let mut first = true; - - for _ in 0..N_SENTENCES { - let buf = build_sentence_append_buffer( - &mut first, - vec![0.5_f32; SENTENCE_AUDIO_LEN], - SILENCE_BUF_LEN, - true, - true, - ); - - assert_eq!(buf.len(), SENTENCE_AUDIO_LEN + SILENCE_BUF_LEN); - assert!( - buf[..SENTENCE_LEAD_IN_SAMPLES].iter().all(|&s| s == 0.0), - "lead-in pad must be pure silence" - ); - assert!( - buf[SENTENCE_LEAD_IN_SAMPLES..SENTENCE_LEAD_IN_SAMPLES + SENTENCE_AUDIO_LEN] - .iter() - .all(|&s| s == 0.5), - "sentence audio must immediately follow the lead-in" - ); - assert!( - buf[SENTENCE_LEAD_IN_SAMPLES + SENTENCE_AUDIO_LEN..] - .iter() - .all(|&s| s == 0.0), - "trailing gap must be pure silence" - ); - } - - assert!(!first, "first_append flag must be cleared after first call"); -} - -/// `first_append` still flips on the first call for `tts_active` gating. +/// `first_append` still flips on the first append for `tts_active` gating. #[test] fn build_sentence_append_buffer_flips_first_append() { let mut first = true; - let _ = build_sentence_append_buffer(&mut first, vec![0.5; 100], 2400, true, true); + let buf = build_sentence_append_buffer(&mut first, vec![0.5; 100], false); + assert_eq!(buf, vec![0.5; 100]); assert!(!first, "first call must flip the flag"); - - // Subsequent call: still has a per-sentence lead-in, flag stays false. - let buf = build_sentence_append_buffer(&mut first, vec![0.5; 100], 2400, true, true); - assert!(buf[..SENTENCE_LEAD_IN_SAMPLES].iter().all(|&s| s == 0.0)); - assert!(!first); } -/// Leading silence is exactly the lead-in; no pre-audio gap is double-counted. +/// Playback chunks are contiguous: Pocket's generated pause is not extended +/// with a fixed inter-sentence silence budget. #[test] -fn first_sentence_leading_silence_is_exactly_lead_in() { +fn sentence_append_buffer_does_not_inject_silence() { let mut first = true; - let buf = build_sentence_append_buffer(&mut first, vec![0.5; 100], 2400, true, true); - assert!(buf[..SENTENCE_LEAD_IN_SAMPLES].iter().all(|&s| s == 0.0)); - assert_eq!(buf[SENTENCE_LEAD_IN_SAMPLES], 0.5); + let first_buf = build_sentence_append_buffer(&mut first, vec![0.5; 100], false); + let second_buf = build_sentence_append_buffer(&mut first, vec![0.25; 100], false); + + assert_eq!(first_buf, vec![0.5; 100]); + assert_eq!(second_buf, vec![0.25; 100]); } -/// Tail silence plus the next lead-in preserves the 100 ms sentence gap. +/// If generation falls behind playback, retain the onset cushion that protects +/// the first phoneme while the output path wakes back up. #[test] -fn sentence_gap_budget_is_preserved() { +fn idle_playback_gets_an_onset_cushion() { let mut first = true; - let silence_buf_len = 2400; - let first_buf = - build_sentence_append_buffer(&mut first, vec![0.5; 100], silence_buf_len, true, true); - let second_buf = - build_sentence_append_buffer(&mut first, vec![0.5; 100], silence_buf_len, true, true); + let buf = build_sentence_append_buffer(&mut first, vec![0.5; 100], true); - let first_tail = &first_buf[SENTENCE_LEAD_IN_SAMPLES + 100..]; - let second_lead = &second_buf[..SENTENCE_LEAD_IN_SAMPLES]; - assert_eq!(first_tail.len(), silence_buf_len - SENTENCE_LEAD_IN_SAMPLES); - assert_eq!(second_lead.len(), SENTENCE_LEAD_IN_SAMPLES); - assert_eq!(first_tail.len() + second_lead.len(), silence_buf_len); + assert_eq!(buf.len(), SENTENCE_LEAD_IN_SAMPLES + 100); + assert!(buf[..SENTENCE_LEAD_IN_SAMPLES].iter().all(|&s| s == 0.0)); + assert_eq!(buf[SENTENCE_LEAD_IN_SAMPLES], 0.5); } -/// Regression guard: one contiguous rodio source per synthesized sentence. #[test] -fn sentence_append_buffer_is_one_contiguous_source() { - let mut first = true; - let buf = build_sentence_append_buffer(&mut first, vec![0.5; 100], 2400, true, true); +fn tts_worker_uses_distinct_playback_and_model_splitters() { + let source = include_str!("tts.rs"); + let playback_calls = source.matches("engine.split_text_for_playback(").count(); + let model_calls = source.matches("engine.split_text_into_chunks(").count(); - assert_eq!(buf.len(), 2400 + 100); - assert!(buf[..SENTENCE_LEAD_IN_SAMPLES].iter().all(|&s| s == 0.0)); + assert_eq!( + (playback_calls, model_calls), + (1, 1), + "the worker must isolate sentence one only in the outer playback split" + ); + + // Counts alone are order-blind: swapping the two call sites keeps them at + // (1, 1) while the outer split stops isolating sentence one, which delays + // first audio by a whole generation. Pin the ORDER too. + let playback_at = source + .find("engine.split_text_for_playback(") + .expect("outer playback split exists"); + let model_at = source + .find("engine.split_text_into_chunks(") + .expect("inner model split exists"); assert!( - buf[SENTENCE_LEAD_IN_SAMPLES..SENTENCE_LEAD_IN_SAMPLES + 100] - .iter() - .all(|&s| s == 0.5) + playback_at < model_at, + "the playback split must be the OUTER pass; swapping the two delays first audio" ); } @@ -907,79 +872,3 @@ fn clamp_to_full_scale_empty_buffer() { let out = clamp_to_full_scale(Vec::new()); assert!(out.is_empty()); } - -// ── group_sentences_into_chunks tests ───────────────────────────────────── - -fn s(v: &[&str]) -> Vec { - v.iter().map(|x| x.to_string()).collect() -} - -/// The first sentence always stands alone — it bounds time-to-first-audio. -/// Even when the whole message would fit in one chunk, sentence one must -/// not wait on synthesis of the rest. -#[test] -fn chunk_grouping_first_sentence_is_always_alone() { - let chunks = group_sentences_into_chunks(&s(&["Hi there.", "Short.", "Tiny."]), 200); - assert_eq!(chunks[0], "Hi there."); - assert_eq!(chunks.len(), 2); - assert_eq!(chunks[1], "Short. Tiny."); -} - -/// Sentences after the first pack greedily up to the char budget, then -/// spill into a new chunk. Fewer generate() calls = fewer prosody seams. -#[test] -fn chunk_grouping_packs_up_to_budget_then_spills() { - let a = "A".repeat(50) + "."; - let b = "B".repeat(50) + "."; - let c = "C".repeat(50) + "."; - let d = "D".repeat(50) + "."; - // Budget of 110: b+c fits (51+1+51 = 103), adding d (103+1+51) does not. - let chunks = group_sentences_into_chunks(&s(&[&a, &b, &c, &d]), 110); - assert_eq!(chunks.len(), 3, "chunks: {chunks:?}"); - assert_eq!(chunks[0], a); - assert_eq!(chunks[1], format!("{b} {c}")); - assert_eq!(chunks[2], d); -} - -/// A single sentence longer than the coarse budget is passed through here; -/// the loaded April engine subsequently enforces its exact 50-token limit. -#[test] -fn chunk_grouping_oversized_sentence_passes_through() { - let long = "word ".repeat(60).trim_end().to_string() + "."; - assert!(long.len() > 200); - let chunks = group_sentences_into_chunks(&s(&["First.", &long]), 200); - assert_eq!(chunks, vec!["First.".to_string(), long]); -} - -/// Single-sentence messages — the common huddle case, since agents are -/// prompted to send one sentence per message — are unaffected by grouping. -#[test] -fn chunk_grouping_single_sentence_unchanged() { - let chunks = group_sentences_into_chunks(&s(&["Just one sentence here."]), 200); - assert_eq!(chunks, vec!["Just one sentence here.".to_string()]); -} - -/// Empty and whitespace-only entries are dropped, and never produce -/// empty chunks (which would synthesize as garbage). -#[test] -fn chunk_grouping_skips_blank_sentences() { - let chunks = group_sentences_into_chunks(&s(&["", " ", "Real sentence.", " ", "Two."]), 200); - assert_eq!(chunks[0], "Real sentence."); - assert_eq!(chunks.len(), 2); - assert_eq!(chunks[1], "Two."); -} - -/// Empty input produces no chunks (the worker loop then synthesizes nothing). -#[test] -fn chunk_grouping_empty_input() { - assert!(group_sentences_into_chunks(&[], 200).is_empty()); -} - -/// Chunks joined with a single space preserve each sentence's terminal -/// punctuation — the model sees natural multi-sentence prose, matching the -/// shape upstream's ~50-token chunker produces. -#[test] -fn chunk_grouping_preserves_punctuation_at_joins() { - let chunks = group_sentences_into_chunks(&s(&["Lead.", "Really?", "Yes!", "Good."]), 200); - assert_eq!(chunks[1], "Really? Yes! Good."); -} diff --git a/desktop/src-tauri/src/huddle/tts_tests/token_split.rs b/desktop/src-tauri/src/huddle/tts_tests/token_split.rs index b9249c9afc4..404f8a8153f 100644 --- a/desktop/src-tauri/src/huddle/tts_tests/token_split.rs +++ b/desktop/src-tauri/src/huddle/tts_tests/token_split.rs @@ -2,7 +2,7 @@ use super::*; /// The onset cushion covers 20 ms at the production sample rate. #[test] -fn sentence_lead_in_is_sane() { +fn chunk_lead_in_is_sane() { assert_eq!(SENTENCE_LEAD_IN_SAMPLES, 480, "20 ms × 24 kHz"); } @@ -11,14 +11,11 @@ fn sentence_lead_in_is_sane() { #[test] fn token_split_units_do_not_add_sentence_boundary_padding() { let mut first = true; - let silence_buf_len = 2400; - let first_unit = - build_sentence_append_buffer(&mut first, vec![0.5; 100], silence_buf_len, true, false); - let last_unit = - build_sentence_append_buffer(&mut first, vec![0.25; 100], silence_buf_len, false, true); + let first_unit = build_sentence_append_buffer(&mut first, vec![0.5; 100], false); + let last_unit = build_sentence_append_buffer(&mut first, vec![0.25; 100], false); - assert_eq!(first_unit.len(), SENTENCE_LEAD_IN_SAMPLES + 100); + assert_eq!(first_unit.len(), 100); assert_eq!(first_unit.last(), Some(&0.5)); assert_eq!(last_unit.first(), Some(&0.25)); - assert_eq!(first_unit.len() + last_unit.len(), 200 + silence_buf_len); + assert_eq!(first_unit.len() + last_unit.len(), 200); } diff --git a/desktop/src-tauri/src/huddle/tts_voice_transition.rs b/desktop/src-tauri/src/huddle/tts_voice_transition.rs index a60d3506ffa..99b165bfe81 100644 --- a/desktop/src-tauri/src/huddle/tts_voice_transition.rs +++ b/desktop/src-tauri/src/huddle/tts_voice_transition.rs @@ -5,7 +5,7 @@ use std::{ sync::{ atomic::{AtomicBool, AtomicU64, Ordering}, mpsc::{self, SyncSender}, - Arc, Mutex, + Arc, Mutex, MutexGuard, PoisonError, }, }; @@ -64,7 +64,7 @@ impl PlaybackProbe { } pub(super) fn set_synthesis_in_flight(&self, in_flight: bool) { - let _ops = super::lock_player_ops(&self.player_ops); + let _ops = lock_player_ops(&self.player_ops); self.synthesis_in_flight.store(in_flight, Ordering::Release); } @@ -202,7 +202,7 @@ pub(super) fn request_active_speaker_cancel( let Some(player) = playback_probe.player() else { return false; }; - let _ops = super::lock_player_ops(&playback_probe.player_ops); + let _ops = lock_player_ops(&playback_probe.player_ops); let playback_live = !player.empty() || playback_probe.synthesis_in_flight.load(Ordering::Acquire); request_active_speaker_cancel_while_locked( @@ -472,6 +472,88 @@ fn log_cancelled_route(route_id: u64, reason: &str) { eprintln!("buzz-desktop: tts stage=queue status=dropped reason={reason} route_id={route_id}"); } +/// Check for cancel or shutdown. Returns `true` if the caller should break/continue. +/// On cancel: drains the text queue and clears the cancel flag. +/// +/// `player` pairs the Player with the `player_ops` mutex shared with the +/// barge-in monitor thread; the cancel/shutdown clear runs under that lock so +/// it is serialized with the monitor's stale-branch re-check (see the monitor +/// block in `tts_worker`). +pub(super) fn handle_cancel_or_shutdown( + cancel_signals: CancelSignals<'_>, + shutdown: &AtomicBool, + tts_active: &AtomicBool, + text_state: CancelTextState<'_>, + voice_change_ack: &VoiceChangeAck, + active_route_id: Option, + player: Option<(&rodio::Player, &Mutex<()>)>, +) -> bool { + let (cancel, voice_cancel) = cancel_signals; + let (text_rx, deferred_text, current_text) = text_state; + if shutdown.load(Ordering::Acquire) { + eprintln!( + "buzz-desktop: tts stage=cancellation reason=shutdown route_id={}", + active_route_id.unwrap_or(0) + ); + if let Some((p, ops)) = player { + let _ops = lock_player_ops(ops); + p.clear(); + } + tts_active.store(false, Ordering::Release); + return true; + } + if cancel.load(Ordering::Acquire) || voice_cancel.load(Ordering::Acquire) { + // Serialize with begin_voice_change so the generation boundary and + // cancel consumption are observed as one transition. + let pending_voice_change = voice_change_ack + .lock() + .unwrap_or_else(|error| error.into_inner()); + // Consume at the serialization point. A later barge-in remains true + // for the next pass instead of being overwritten after queue cleanup. + let barge_in = cancel.swap(false, Ordering::AcqRel); + voice_cancel.store(false, Ordering::Release); + eprintln!( + "buzz-desktop: tts stage=cancellation reason={} route_id={}", + if barge_in { "barge_in" } else { "voice_switch" }, + active_route_id.unwrap_or(0) + ); + let preserve_generation = (!barge_in) + .then(|| { + pending_voice_change + .as_ref() + .map(|pending| pending.generation) + }) + .flatten(); + retain_cancelled_text(deferred_text, current_text, text_rx, preserve_generation); + if let Some((p, ops)) = player { + let _ops = lock_player_ops(ops); + // `Player::clear()` removes queued sources AND pauses the player + // (rodio 0.22 `clear()` ends with `self.pause()`). With one + // persistent Player for the worker's lifetime, the un-pause is + // mandatory: without `play()`, every append after a barge-in + // would queue silently forever. + p.clear(); + p.play(); + // Consume the flag under the lock: once released with + // `cancel == false`, the monitor's stale branch no-ops instead + // of clearing the fresh post-cancel utterance. + } + tts_active.store(false, Ordering::Release); + return true; + } + false +} + +/// Acquire the `player_ops` lock, recovering from poison. +/// +/// The data under the mutex is `()` — it only serializes Player mutations — +/// so a panicked holder leaves nothing inconsistent to observe and recovery +/// is always safe. Without this, a worker panic would wedge the monitor (or +/// vice versa) on `unwrap()`. +pub(super) fn lock_player_ops(ops: &Mutex<()>) -> MutexGuard<'_, ()> { + ops.lock().unwrap_or_else(PoisonError::into_inner) +} + #[cfg(test)] mod speaker_generation_tests { use super::*; diff --git a/desktop/src/app/AppHuddleShell.tsx b/desktop/src/app/AppHuddleShell.tsx index 68f9eacd6fd..3e32697e9ec 100644 --- a/desktop/src/app/AppHuddleShell.tsx +++ b/desktop/src/app/AppHuddleShell.tsx @@ -1,7 +1,8 @@ -import type * as React from "react"; +import * as React from "react"; import { AppHuddleBar } from "@/app/AppHuddleBar"; import * as BuzzTheme from "@/app/BuzzThemeSurfaces"; -import { HuddleProvider } from "@/features/huddle"; +import { HuddleProvider, useHuddle } from "@/features/huddle"; +import { HUDDLE_SHORTCUT_EVENT } from "@/shared/lib/keyboard-shortcuts"; import { RemindMeLaterProvider } from "@/features/reminders/ui/RemindMeLaterProvider"; import { cn } from "@/shared/lib/cn"; @@ -19,6 +20,28 @@ type AppHuddleShellProps = { onVisibilityChange: (visible: boolean) => void; }; +type HuddleShortcutHandlerProps = { + children: React.ReactNode; +}; + +function HuddleShortcutHandler({ children }: HuddleShortcutHandlerProps) { + const { activeEphemeralChannelId, leaveHuddle } = useHuddle(); + + React.useEffect(() => { + if (!activeEphemeralChannelId) return; + + function handleHuddleShortcut() { + void leaveHuddle(); + } + + window.addEventListener(HUDDLE_SHORTCUT_EVENT, handleHuddleShortcut); + return () => + window.removeEventListener(HUDDLE_SHORTCUT_EVENT, handleHuddleShortcut); + }, [activeEphemeralChannelId, leaveHuddle]); + + return children; +} + export function AppHuddleShell({ children, currentPubkey, @@ -42,42 +65,44 @@ export function AppHuddleShell({ onShowHuddleInMainApp={isRoom ? undefined : onShowHuddleInMainApp} onViewHuddleChannel={isRoom ? undefined : onViewHuddleChannel} > - -
+ +