From 7f56b2f538994f661aadf3584d91bf198b2c8e58 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Thu, 5 Nov 2026 09:48:44 -0500 Subject: [PATCH 1/8] feat(berd-call): count speech requests by voice in call telemetry Signed-off-by: John Tennant Co-authored-by: Darcy <756b3e94a13d7979c09d32291e4fd4da46848bdaa53340021ac24fb843e28d57@buzz.block.builderlab.xyz> --- src-tauri/crates/berd-call/PROTOCOL.md | 8 +- src-tauri/crates/berd-call/README.md | 4 + .../crates/berd-call/src/host_session.rs | 6 + src-tauri/crates/berd-call/src/main.rs | 31 ++- src-tauri/crates/berd-call/src/protocol.rs | 6 +- .../crates/berd-call/src/session_framing.rs | 2 +- src-tauri/crates/berd-call/src/telemetry.rs | 184 +++++++++++++++++- .../berd-call/tests/session_protocol.rs | 23 ++- 8 files changed, 244 insertions(+), 20 deletions(-) diff --git a/src-tauri/crates/berd-call/PROTOCOL.md b/src-tauri/crates/berd-call/PROTOCOL.md index f1b35e7ca..3bd0b19ad 100644 --- a/src-tauri/crates/berd-call/PROTOCOL.md +++ b/src-tauri/crates/berd-call/PROTOCOL.md @@ -58,12 +58,12 @@ initial policy; a host-specific `auto` mode must be resolved before the request. {"type":"hello","id":1,"input_during_tts":"allow_barge_in","status_sound_output_device":"MacBook Pro Speakers"} ``` -The response uses `protocol:6` as the exact session message-set version. The +The response uses `protocol:7` as the exact session message-set version. The parent must reject a version it does not support; the JSON version is independent of the fixed binary framing marker described below: ```json -{"type":"ready","id":1,"protocol":6,"session":{"tts":{"revision":1,"backend":"siri","voice":"Aaron","language":"en-US","rate":1.0},"input_during_tts":{"revision":1,"policy":"allow_barge_in"}}} +{"type":"ready","id":1,"protocol":7,"session":{"tts":{"revision":1,"backend":"siri","voice":"Aaron","language":"en-US","rate":1.0},"input_during_tts":{"revision":1,"policy":"allow_barge_in"}}} ``` The `session.tts` object is the authoritative, sanitized TTS configuration. @@ -358,7 +358,7 @@ cancellation and shutdown terminals. {"type":"not_admitted","id":u64,"reason":"paused"|"in_progress"|"cancelled"|"empty_text"|"invalid_voice"} {"type":"state","id":u64,"confirmed_token":u64,"utterances_after":[{"token":u64,"text":string}]} {"type":"cancel_result","id":u64,"outcome":"cancelled"|"stale","speech_id":u64|null} -{"type":"speech_started","id":u64,"speech_id":u64} +{"type":"speech_started","id":u64,"speech_id":u64,"tts":TtsSettings} {"type":"speech_completed","id":u64,"speech_id":u64} {"type":"speech_interrupted","id":u64,"speech_id":u64,"spoken_through_utf8":u64} {"type":"speech_failed","id":u64,"speech_id":u64,"message":string} @@ -376,7 +376,7 @@ names the targeted speech. The result is emitted before the runtime requests the host's quiescent `audio_cancelled` barrier. Every speech event carries the originating prepare ID. `speech_started` appears only after the first PCM Chunk is accepted by the host, and exactly one terminal -message follows every admission. +message follows every admission. When available, its `tts` object contains the sanitized settings held by that speech's synthesis lease, including any per-request voice override. This has the same backend-specific fields as `session.tts` without a revision. It does not change the session default; a resumed speech does not emit another start. On `shutdown`, all complete earlier frames are processed in order. The parent keeps stdin open while the child cancels and drains output. The child then diff --git a/src-tauri/crates/berd-call/README.md b/src-tauri/crates/berd-call/README.md index 54d326b65..269a40bba 100644 --- a/src-tauri/crates/berd-call/README.md +++ b/src-tauri/crates/berd-call/README.md @@ -448,3 +448,7 @@ Auto is the default input-during-speech preference. It mutes microphone input du `start --stream` emits ephemeral `0\tinput_activity\t{"speaking":true,"recognizing":false}` rows when input speech or recognition state changes. Both booleans are snapshots; show activity while either is true. These rows have cursor zero, are not utterances, and must never advance a speech cursor, trigger an agent turn, or enter transcript history. They are not delivered to Codex tasks. Clear client activity on session restart, stream end, or failure. Older CLIs omit these rows. On macOS, conventional sessions with `--stream` also emit ephemeral `0\tuser_interim\t{"cursor":12,"text":"you should"}` rows while your words are still being transcribed. Each row replaces the previous one; empty text clears it, as do mute, input reset and session restart. `cursor` is the cursor the next `user` row will carry, so a client can show the words as a caption or let an agent start on them and keep its reply current when the final arrives. The final `user` row stays authoritative. Like `input_activity`, these rows never advance a speech cursor or enter transcript history, and older CLIs omit them. + +### Call voice telemetry + +With telemetry enabled, standalone calls emit one start and one end event using `berd-call-otlp-logs-v2`. The existing `tts_voice` remains the call's starting default. The ended event also carries `tts_voice_requests`, an OTLP key/value list of catalog display names and numeric request counts. A request counts once when playback begins, including interrupted speech; queued or rejected speech that never plays does not count, and resuming a request does not count it again. Per-request overrides and later settings use the voice held by that speech's synthesis lease. Unknown voices are grouped as `custom`; no agent names, text or per-speech events are exported. Deploy gateway v2 support before releasing this client. diff --git a/src-tauri/crates/berd-call/src/host_session.rs b/src-tauri/crates/berd-call/src/host_session.rs index 805697456..c03698b29 100644 --- a/src-tauri/crates/berd-call/src/host_session.rs +++ b/src-tauri/crates/berd-call/src/host_session.rs @@ -1311,6 +1311,12 @@ impl SessionActor { } } } + SessionMessage::SpeechStarted { + tts: Some(settings), + .. + } => { + crate::telemetry::speech_started(&settings); + } SessionMessage::TtsSettingsResult { id, outcome, diff --git a/src-tauri/crates/berd-call/src/main.rs b/src-tauri/crates/berd-call/src/main.rs index 06f811c17..46159a4a4 100644 --- a/src-tauri/crates/berd-call/src/main.rs +++ b/src-tauri/crates/berd-call/src/main.rs @@ -4543,7 +4543,11 @@ fn publish_speech_started( ) -> Result<(), String> { if core.mark_started(speech_id) { let id = active.map_or(0, |current| current.prepare_id); - write_message(writer, &SessionMessage::SpeechStarted { id, speech_id })?; + let tts = active.map(|current| current.tts.snapshot().settings.clone()); + write_message( + writer, + &SessionMessage::SpeechStarted { id, speech_id, tts }, + )?; } Ok(()) } @@ -5816,6 +5820,31 @@ mod tests { } } + #[test] + fn speech_start_reports_its_lease_once_even_after_resume() { + let mut core = SessionCore::default(); + let current = active_playback(&mut core); + let settings = current.tts.snapshot().settings.clone(); + let mut messages = Vec::new(); + publish_speech_started(current.speech_id, &mut core, Some(¤t), &mut messages) + .unwrap(); + publish_speech_started(current.speech_id, &mut core, Some(¤t), &mut messages) + .unwrap(); + let records: Vec = String::from_utf8(messages) + .unwrap() + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect(); + assert_eq!( + records, + vec![SessionMessage::SpeechStarted { + id: current.prepare_id, + speech_id: current.speech_id, + tts: Some(settings), + }] + ); + } + fn read_audio_record(reader: &mut impl Read) -> (u8, Vec) { let mut header = [0; session_audio::AUDIO_FRAME_HEADER_BYTES]; reader.read_exact(&mut header).unwrap(); diff --git a/src-tauri/crates/berd-call/src/protocol.rs b/src-tauri/crates/berd-call/src/protocol.rs index 87e4afdfc..f62131e9a 100644 --- a/src-tauri/crates/berd-call/src/protocol.rs +++ b/src-tauri/crates/berd-call/src/protocol.rs @@ -241,6 +241,8 @@ pub enum SessionMessage { SpeechStarted { id: u64, speech_id: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + tts: Option, }, SpeechCompleted { id: u64, @@ -284,7 +286,7 @@ mod tests { assert_eq!( serde_json::to_string(&SessionMessage::Ready { id: 4, - protocol: 6, + protocol: 7, session: VoiceSessionSnapshot { tts: TtsConfigurationSnapshot { revision: 1, @@ -301,7 +303,7 @@ mod tests { }, }) .unwrap(), - r#"{"type":"ready","id":4,"protocol":6,"session":{"tts":{"revision":1,"backend":"openai","model":"gpt-4o-mini-tts","voice":"marin","rate":1.0},"input_during_tts":{"revision":1,"policy":"allow_barge_in"}}}"# + r#"{"type":"ready","id":4,"protocol":7,"session":{"tts":{"revision":1,"backend":"openai","model":"gpt-4o-mini-tts","voice":"marin","rate":1.0},"input_during_tts":{"revision":1,"policy":"allow_barge_in"}}}"# ); assert_eq!( serde_json::from_str::( diff --git a/src-tauri/crates/berd-call/src/session_framing.rs b/src-tauri/crates/berd-call/src/session_framing.rs index 8fedbf5d8..94fa6aa49 100644 --- a/src-tauri/crates/berd-call/src/session_framing.rs +++ b/src-tauri/crates/berd-call/src/session_framing.rs @@ -1,4 +1,4 @@ -pub(crate) const SESSION_PROTOCOL_VERSION: u32 = 6; +pub(crate) const SESSION_PROTOCOL_VERSION: u32 = 7; pub(crate) const FRAME_MAGIC: [u8; 2] = *b"BV"; pub(crate) const FRAME_MARKER: u8 = 3; pub(crate) const JSON_FRAME_KIND: u8 = 1; diff --git a/src-tauri/crates/berd-call/src/telemetry.rs b/src-tauri/crates/berd-call/src/telemetry.rs index 6013d3faa..5751858e5 100644 --- a/src-tauri/crates/berd-call/src/telemetry.rs +++ b/src-tauri/crates/berd-call/src/telemetry.rs @@ -7,6 +7,7 @@ //! only cost to a call is the bounded flush when it ends. #![cfg_attr(not(target_os = "macos"), allow(dead_code))] +use std::collections::BTreeMap; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{mpsc, Mutex}; @@ -17,7 +18,7 @@ use berd_call::telemetry_store as store; use berd_call::TtsSettings; use serde_json::{json, Value}; -const SCHEMA_VERSION: &str = "berd-call-otlp-logs-v1"; +const SCHEMA_VERSION: &str = "berd-call-otlp-logs-v2"; const ENFORCED: bool = cfg!(feature = "telemetry-enforced"); const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); @@ -149,8 +150,69 @@ impl Context { } } +/// Only catalog display names leave the process. Capture the catalog once per +/// Siri session startup so speech-start handling never waits for enumeration. +#[derive(Default)] +struct VoiceRequests { + siri: Vec, + counts: BTreeMap, +} + +impl VoiceRequests { + fn started(&mut self, settings: &TtsSettings) { + let name = match settings { + TtsSettings::Siri { + voice, language, .. + } => self + .siri + .iter() + .find(|known| { + berd_call::siri::SiriVoiceIdentity::new(voice, language) + .is_ok_and(|identity| known.matches(&identity)) + }) + .map(|known| known.display_name.as_str()), + TtsSettings::Pocket { voice, .. } => berd_call::pocket_assets::voices() + .iter() + .find(|known| known.id == voice) + .map(|known| known.name), + TtsSettings::OpenAi { voice, .. } => berd_call::openai::OPENAI_VOICE_IDS + .contains(&voice.as_str()) + .then_some(voice.as_str()), + } + .filter(|name| valid_voice_name(name)) + .unwrap_or("custom"); + // Bound the wire map even if a native catalog grows unexpectedly. + let name = if self.counts.len() >= 511 && !self.counts.contains_key(name) { + "custom" + } else { + name + }; + let count = self.counts.entry(name.to_string()).or_default(); + *count = count.saturating_add(1).min(i64::MAX as u64); + } + + fn attribute(&self) -> Value { + let values: Vec<_> = self + .counts + .iter() + .map(|(name, count)| json!({"key": name, "value": {"intValue": count.to_string()}})) + .collect(); + json!({"key": "tts_voice_requests", "value": {"kvlistValue": {"values": values}}}) + } +} + +fn valid_voice_name(name: &str) -> bool { + !name.is_empty() + && name.chars().count() <= 64 + && name.chars().next().is_some_and(char::is_alphanumeric) + && name + .chars() + .all(|ch| ch.is_alphanumeric() || " ._()'-".contains(ch)) +} + struct Call { context: Context, + voices: VoiceRequests, started: Instant, records: mpsc::Sender, sent: mpsc::Receiver<()>, @@ -169,6 +231,16 @@ pub(crate) fn assistant_response() { ASSISTANT_RESPONSES.fetch_add(1, Ordering::Relaxed); } +/// Session playback reports the settings held by its actual synthesis lease. +/// No per-speech record is exported; the call accumulates counts until end. +pub(crate) fn speech_started(settings: &TtsSettings) { + if let Ok(mut call) = CALL.lock() { + if let Some(call) = call.as_mut() { + call.voices.started(settings); + } + } +} + pub(crate) fn user_stopped() { USER_STOPPED.store(true, Ordering::Relaxed); } @@ -179,13 +251,32 @@ pub(crate) fn start(session_arguments: &[String], tts: &TtsSettings) { return; }; let Ok(mut call) = CALL.lock() else { return }; - if call.is_some() || !consented(&dir) { + if !consented(&dir) { return; } - let Some(context) = Context::new(session_arguments, tts, || { + // Backend changes restart the session within the same call. Refresh the + // Siri catalog there, while retaining all counts and the starting default. + let siri = || { berd_call::siri::load_voice_catalog(None) - .map(|catalog| catalog.voices.into_iter().map(|voice| voice.name).collect()) + .map(|catalog| catalog.voices) .unwrap_or_default() + }; + if let Some(call) = call.as_mut() { + if matches!(tts, TtsSettings::Siri { .. }) && call.voices.siri.is_empty() { + call.voices.siri = siri(); + } + return; + } + let voices = VoiceRequests { + siri: if matches!(tts, TtsSettings::Siri { .. }) { + siri() + } else { + Vec::new() + }, + ..VoiceRequests::default() + }; + let Some(context) = Context::new(session_arguments, tts, || { + voices.siri.iter().map(|voice| voice.name.clone()).collect() }) else { return; }; @@ -218,6 +309,7 @@ pub(crate) fn start(session_arguments: &[String], tts: &TtsSettings) { )); *call = Some(Call { context, + voices, started: Instant::now(), records, sent, @@ -236,8 +328,8 @@ pub(crate) fn end(failed: bool) { } else { "clean-shutdown" }; - let _ = call.records.send(record( - "berd_voice_conversation_ended", + let _ = call.records.send(ended_record( + &call.voices, SystemTime::now(), &ended_attributes( &call.context, @@ -251,6 +343,19 @@ pub(crate) fn end(failed: bool) { let _ = call.sent.recv_timeout(FLUSH_TIMEOUT); } +fn ended_record( + voices: &VoiceRequests, + at: SystemTime, + event_attributes: &[(&'static str, String)], +) -> Value { + let mut record = record("berd_voice_conversation_ended", at, event_attributes); + record["attributes"] + .as_array_mut() + .unwrap() + .push(voices.attribute()); + record +} + fn ended_attributes( context: &Context, duration: Duration, @@ -472,11 +577,20 @@ mod tests { &context.attributes(), ), ); + let mut requests = VoiceRequests::default(); + let openai = |voice: &str| TtsSettings::OpenAi { + model: "test".into(), + voice: voice.into(), + rate: 1.0, + }; + requests.started(&openai("alloy")); + requests.started(&openai("alloy")); + requests.started(&openai("marin")); let ended = payload( production(), ID, - record( - "berd_voice_conversation_ended", + ended_record( + &requests, at(1_762_265_100), &ended_attributes(&context, Duration::from_millis(300_250), 4, 3, "user"), ), @@ -535,6 +649,10 @@ mod tests { {"key": "user_utterance_count", "value": {"stringValue": "4"}}, {"key": "assistant_response_count", "value": {"stringValue": "3"}}, {"key": "end_reason", "value": {"stringValue": "user"}}, + {"key": "tts_voice_requests", "value": {"kvlistValue": {"values": [ + {"key": "alloy", "value": {"intValue": "2"}}, + {"key": "marin", "value": {"intValue": "1"}}, + ]}}}, ]) ) ); @@ -690,7 +808,7 @@ mod tests { .collect::>() }; assert_eq!(header("authorization"), ["Bearer token-1"]); - assert_eq!(header("x-berd-schema-version"), ["berd-call-otlp-logs-v1"]); + assert_eq!(header("x-berd-schema-version"), ["berd-call-otlp-logs-v2"]); assert_eq!(header("content-type"), ["application/json"]); let resource = &requests[1].body["resourceLogs"][0]["resource"]["attributes"]; assert_eq!( @@ -714,6 +832,54 @@ mod tests { assert_eq!(requests[0].line, "POST /v1/bootstrap HTTP/1.1"); } + #[test] + fn requests_use_catalog_display_names_and_reset_for_each_call() { + let mut requests = VoiceRequests { + siri: vec![berd_call::siri::SiriVoice { + name: "en-US-F".into(), + display_name: "Zack".into(), + language: "en-US".into(), + size_bytes: 0, + installed: true, + }], + ..VoiceRequests::default() + }; + let siri = |language: &str| TtsSettings::Siri { + voice: "en-US-F".into(), + language: language.into(), + rate: 1.0, + }; + requests.started(&siri("en-US")); + requests.started(&siri("en-US")); + requests.started(&siri("en_US")); + requests.started(&pocket(1.0)); + requests.started(&siri("en-AU")); // Same code in a different locale is not this identity. + requests.started(&TtsSettings::OpenAi { + model: "test".into(), + voice: "private override".into(), + rate: 1.0, + }); + assert_eq!( + requests.counts, + BTreeMap::from([("Zack".into(), 3), ("Anna".into(), 1), ("custom".into(), 2),]) + ); + let ended = ended_record(&requests, UNIX_EPOCH, &[]); + assert_eq!( + ended["attributes"], + json!([{ + "key": "tts_voice_requests", "value": {"kvlistValue": {"values": [ + {"key": "Anna", "value": {"intValue": "1"}}, + {"key": "Zack", "value": {"intValue": "3"}}, + {"key": "custom", "value": {"intValue": "2"}}, + ]}} + }]) + ); + assert!(VoiceRequests::default().counts.is_empty()); + assert!(!serde_json::to_string(&ended) + .unwrap() + .contains("private override")); + } + #[test] fn ended_attributes_append_duration_counts_and_reason() { let context = context(&["session"], &pocket(9.0)).unwrap(); diff --git a/src-tauri/crates/berd-call/tests/session_protocol.rs b/src-tauri/crates/berd-call/tests/session_protocol.rs index c6c54dd2f..3985f92b3 100644 --- a/src-tauri/crates/berd-call/tests/session_protocol.rs +++ b/src-tauri/crates/berd-call/tests/session_protocol.rs @@ -763,7 +763,7 @@ fn siri_remote_output_supports_consecutive_turns_and_cancellation() { #[test] fn speech_whose_player_stops_is_interrupted_where_it_was_heard_and_the_call_goes_on() { - let (speech_url, _speech_requests) = spawn_speech_server(); + let (speech_url, speech_requests) = spawn_speech_server(); let (mut command, pcm_guard, host) = session_command(); let child = command .args([ @@ -777,6 +777,7 @@ fn speech_whose_player_stops_is_interrupted_where_it_was_heard_and_the_call_goes &spawn_transcription_server(), ]) .env("OPENAI_API_KEY", "test-key") + .env("OPENAI_TTS_VOICE", "alloy") .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::null()) @@ -795,13 +796,15 @@ fn speech_whose_player_stops_is_interrupted_where_it_was_heard_and_the_call_goes write_session_json(&mut *stdin, &message); stdin.flush().unwrap(); }; + let started = Mutex::new(Vec::new()); let terminal = |id: u64| loop { let message = output.recv_timeout(Duration::from_secs(5)).unwrap(); match message["type"].as_str() { Some("admitted") => send(json!({ "type":"output_ready","id":id,"speech_id":message["speech_id"] })), - Some("output_ready_result" | "speech_started") => {} + Some("speech_started") => started.lock().unwrap().push(message["tts"].clone()), + Some("output_ready_result") => {} _ => break message, } }; @@ -814,8 +817,22 @@ fn speech_whose_player_stops_is_interrupted_where_it_was_heard_and_the_call_goes assert_eq!(interrupted["type"], "speech_interrupted", "{interrupted}"); let heard = interrupted["spoken_through_utf8"].as_u64().unwrap(); assert!(heard < text.len() as u64, "{interrupted}"); - send(json!({"type":"prepare_speak","id":3,"acknowledgement":null,"text":"Still here."})); + send( + json!({"type":"prepare_speak","id":3,"acknowledgement":null,"text":"Still here.","voice":"marin"}), + ); assert_eq!(terminal(3)["type"], "speech_completed"); + let started = started.into_inner().unwrap(); + assert_eq!(started.len(), 2); + assert_eq!(started[0]["voice"], "alloy"); + assert_eq!(started[1]["voice"], "marin"); + for expected in ["alloy", "marin"] { + assert_eq!( + speech_requests + .recv_timeout(Duration::from_secs(5)) + .unwrap()["voice"], + expected + ); + } send(json!({"type":"shutdown"})); drop(stdin); assert!(child.0.take().unwrap().wait().unwrap().success()); From 20595cdcb665b28c030d449391b6006f18761ac5 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Thu, 5 Nov 2026 10:25:14 -0500 Subject: [PATCH 2/8] fix(berd-call): count playback hidden by cancellation Signed-off-by: John Tennant Co-authored-by: Darcy <756b3e94a13d7979c09d32291e4fd4da46848bdaa53340021ac24fb843e28d57@buzz.block.builderlab.xyz> --- src-tauri/crates/berd-call/src/main.rs | 113 ++++++++++++++++++++- src-tauri/crates/berd-call/src/protocol.rs | 5 + 2 files changed, 114 insertions(+), 4 deletions(-) diff --git a/src-tauri/crates/berd-call/src/main.rs b/src-tauri/crates/berd-call/src/main.rs index 46159a4a4..ea64030e8 100644 --- a/src-tauri/crates/berd-call/src/main.rs +++ b/src-tauri/crates/berd-call/src/main.rs @@ -33,9 +33,9 @@ use berd_call::{ LocalAssetLockError, LocalAssetRoots, LocalInstallError, LocalInstallErrorKind, LocalInstallPhase, LocalInstallProgress, }, - ConfiguredTtsSlot, DeliveryProgress, TtsBackend, TtsConfiguration, TtsConfigurationLease, - TtsConfigurationRejection, TtsConfigurationRejectionKind, TtsSettings, WavSynthesisErrorKind, - DEFAULT_VAD_THRESHOLD, + ConfiguredTtsSlot, DeliveryProgress, PcmAudioOutput, TtsBackend, TtsConfiguration, + TtsConfigurationLease, TtsConfigurationRejection, TtsConfigurationRejectionKind, TtsSettings, + WavSynthesisErrorKind, DEFAULT_VAD_THRESHOLD, }; use serde::Serialize; @@ -4559,6 +4559,28 @@ fn finish_playback( message: SessionMessage, writer: &mut impl Write, ) -> Result<(), String> { + if let Some(current) = active + .as_ref() + .filter(|current| current.speech_id == speech_id) + { + // Cancellation can suppress SpeechStarted after audio has already played. + // Marking it here deduplicates usage without publishing a late start. + if current + .output + .as_ref() + .is_some_and(|output| output.played_frames() > 0) + && core.mark_started(speech_id) + { + write_message( + writer, + &SessionMessage::SpeechUsed { + id: current.prepare_id, + speech_id, + tts: current.tts.snapshot().settings.clone(), + }, + )?; + } + } if core.finish(speech_id) { if active .as_ref() @@ -5187,7 +5209,7 @@ fn synthesize_to_output_with_finish( #[cfg(test)] mod tests { use super::*; - use berd_call::{PcmAudioOutput, TtsOutcome, TtsPcmSpec}; + use berd_call::{TtsOutcome, TtsPcmSpec}; use serde_json::{json, Value}; use std::io::{Cursor, Read, Write}; use std::os::fd::IntoRawFd; @@ -5820,6 +5842,89 @@ mod tests { } } + #[test] + fn cancelled_playback_reports_used_voice_without_a_late_start() { + for (already_started, played_frames) in [(false, 64), (true, 64), (false, 0)] { + let mut core = SessionCore::default(); + let mut current = active_playback(&mut core); + let speech_id = current.speech_id; + let settings = current.tts.snapshot().settings.clone(); + let (child, mut host) = UnixStream::pair().unwrap(); + let transport = + unsafe { AudioPipeTransport::from_raw_fd(child.into_raw_fd()) }.unwrap(); + let (control_sender, _control_receiver) = mpsc::channel(); + let authority = current.active.as_ref().unwrap().clone(); + let output = Arc::new( + RemotePcmAudioOutput::new( + speech_id, + LongRemoteTts.pcm_spec(), + Arc::new(transport), + Arc::clone(&authority), + control_sender, + ) + .unwrap(), + ); + current.output = Some(Arc::clone(&output)); + let worker_output = Arc::clone(&output); + let worker = thread::spawn(move || { + worker_output.start()?; + worker_output.write(&[0.1; 4096]) + }); + assert_eq!( + read_audio_record(&mut host).0, + session_audio::AUDIO_BEGIN_KIND + ); + output.handle_ack(AudioHostAck::BeginAccepted).unwrap(); + assert_eq!( + read_audio_record(&mut host).0, + session_audio::AUDIO_CHUNK_KIND + ); + let mut messages_out = Vec::new(); + if already_started { + publish_speech_started(speech_id, &mut core, Some(¤t), &mut messages_out) + .unwrap(); + } + authority.store(false, Ordering::SeqCst); + output.notify_cancel_requested(); + assert_eq!( + read_audio_record(&mut host).0, + session_audio::AUDIO_CANCEL_KIND + ); + assert!(!output + .handle_ack(AudioHostAck::ChunkAccepted { sequence: 1 }) + .unwrap()); + output + .handle_ack(AudioHostAck::Cancelled { played_frames }) + .unwrap(); + assert!(worker.join().unwrap().is_err()); + let mut active = Some(current); + handle_playback_event( + PlaybackEvent::Interrupted(speech_id, 0), + &mut core, + &mut active, + &mut messages_out, + ) + .unwrap(); + let rows = messages(&messages_out); + if !already_started && played_frames == 0 { + assert_eq!(rows.len(), 1); + assert_eq!(rows[0]["type"], "speech_interrupted"); + continue; + } + assert_eq!(rows.len(), 2); + assert_eq!( + rows[0]["type"], + if already_started { + "speech_started" + } else { + "speech_used" + } + ); + assert_eq!(rows[0]["tts"], serde_json::to_value(settings).unwrap()); + assert_eq!(rows[1]["type"], "speech_interrupted"); + } + } + #[test] fn speech_start_reports_its_lease_once_even_after_resume() { let mut core = SessionCore::default(); diff --git a/src-tauri/crates/berd-call/src/protocol.rs b/src-tauri/crates/berd-call/src/protocol.rs index f62131e9a..8852e1fff 100644 --- a/src-tauri/crates/berd-call/src/protocol.rs +++ b/src-tauri/crates/berd-call/src/protocol.rs @@ -244,6 +244,11 @@ pub enum SessionMessage { #[serde(default, skip_serializing_if = "Option::is_none")] tts: Option, }, + SpeechUsed { + id: u64, + speech_id: u64, + tts: TtsSettings, + }, SpeechCompleted { id: u64, speech_id: u64, From 22b986ad690413beb8818fe5ee494d419e1703b3 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Thu, 5 Nov 2026 10:25:35 -0500 Subject: [PATCH 3/8] fix(berd-call): collect final playback usage before shutdown Signed-off-by: John Tennant Co-authored-by: Darcy <756b3e94a13d7979c09d32291e4fd4da46848bdaa53340021ac24fb843e28d57@buzz.block.builderlab.xyz> --- src-tauri/crates/berd-call/PROTOCOL.md | 5 +- .../crates/berd-call/src/host_session.rs | 57 ++++++++++++++++++- 2 files changed, 59 insertions(+), 3 deletions(-) diff --git a/src-tauri/crates/berd-call/PROTOCOL.md b/src-tauri/crates/berd-call/PROTOCOL.md index 3bd0b19ad..4593e6467 100644 --- a/src-tauri/crates/berd-call/PROTOCOL.md +++ b/src-tauri/crates/berd-call/PROTOCOL.md @@ -359,6 +359,7 @@ cancellation and shutdown terminals. {"type":"state","id":u64,"confirmed_token":u64,"utterances_after":[{"token":u64,"text":string}]} {"type":"cancel_result","id":u64,"outcome":"cancelled"|"stale","speech_id":u64|null} {"type":"speech_started","id":u64,"speech_id":u64,"tts":TtsSettings} +{"type":"speech_used","id":u64,"speech_id":u64,"tts":TtsSettings} {"type":"speech_completed","id":u64,"speech_id":u64} {"type":"speech_interrupted","id":u64,"speech_id":u64,"spoken_through_utf8":u64} {"type":"speech_failed","id":u64,"speech_id":u64,"message":string} @@ -378,10 +379,12 @@ Every speech event carries the originating prepare ID. `speech_started` appears only after the first PCM Chunk is accepted by the host, and exactly one terminal message follows every admission. When available, its `tts` object contains the sanitized settings held by that speech's synthesis lease, including any per-request voice override. This has the same backend-specific fields as `session.tts` without a revision. It does not change the session default; a resumed speech does not emit another start. +`speech_used` reports confirmed playback when cancellation suppressed `speech_started`. It precedes the terminal event and carries the synthesis lease settings. A speech emits either its start or this fallback usage event, so hosts can count used voices without announcing a late playback start. + On `shutdown`, all complete earlier frames are processed in order. The parent keeps stdin open while the child cancels and drains output. The child then finishes the input runtime while continuing to drain events and storage -receipts, flushes, and exits. EOF, malformed framing, fatal +receipts, flushes, and exits. The parent drains stdout through EOF before finalizing call telemetry. EOF, malformed framing, fatal input failure, or process death cancels both authorities without transparent restart. A fatal error is flushed exactly once and followed by no protocol output. diff --git a/src-tauri/crates/berd-call/src/host_session.rs b/src-tauri/crates/berd-call/src/host_session.rs index c03698b29..90c035cbb 100644 --- a/src-tauri/crates/berd-call/src/host_session.rs +++ b/src-tauri/crates/berd-call/src/host_session.rs @@ -296,7 +296,7 @@ fn run_session( thread::sleep(Duration::from_millis(2)); } drop(capture); - child.wait_for_exit()?; + actor.finish_child(&mut child)?; let restart = actor.restart.take(); actor.finish_pending(if restart.is_some() { "voice session restarted" @@ -989,6 +989,23 @@ impl SessionActor { } } + fn finish_child(&mut self, child: &mut SessionProcess) -> Result<(), String> { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + match self + .events + .recv_timeout(deadline.saturating_duration_since(Instant::now())) + { + Ok(event) => self.handle_event(event)?, + Err(mpsc::RecvTimeoutError::Disconnected) => break, + Err(mpsc::RecvTimeoutError::Timeout) => { + return Err("voice session event stream did not close within 5 seconds".into()); + } + } + } + child.wait_for_exit() + } + fn poll(&mut self) -> Result<(), String> { if !self.stopping { if let Some(muted) = self @@ -1314,7 +1331,8 @@ impl SessionActor { SessionMessage::SpeechStarted { tts: Some(settings), .. - } => { + } + | SessionMessage::SpeechUsed { tts: settings, .. } => { crate::telemetry::speech_started(&settings); } SessionMessage::TtsSettingsResult { @@ -3065,6 +3083,41 @@ mod tests { assert!(child.wait().unwrap().success()); } + #[test] + fn stopping_drains_events_delivered_after_the_final_poll() { + let mut process = Command::new("/bin/sh") + .args([ + "-c", + "read -r signal; echo '{\"type\":\"input_speaking\",\"active\":true}'", + ]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .spawn() + .unwrap(); + let writer = Arc::new(Mutex::new(process.stdin.take().unwrap())); + let (sender, events) = mpsc::sync_channel(1); + spawn_stdout_reader(process.stdout.take().unwrap(), sender).unwrap(); + let (_commands, commands) = mpsc::sync_channel(1); + let (audio, _audio) = mpsc::sync_channel(1); + let mut actor = test_actor(writer, events, commands, audio); + actor.stopping = true; + actor.poll().unwrap(); + assert!(!actor.input_speaking); + writeln!(actor.writer.lock().unwrap(), "finish").unwrap(); + let mut child = SessionProcess { + child: process, + stdin: None, + stdout: None, + audio: None, + reaped: false, + }; + actor.finish_child(&mut child).unwrap(); + assert!( + actor.input_speaking, + "final stdout events must reach the actor before telemetry ends" + ); + } + #[test] fn status_sounds_reported_while_stopping_wait_for_the_next_session() { let mut child = Command::new("/usr/bin/true") From f7099d92a7d2da25f011dde77fa8d99ff2804b6f Mon Sep 17 00:00:00 2001 From: John Tennant Date: Thu, 5 Nov 2026 10:43:43 -0500 Subject: [PATCH 4/8] fix(berd-call): count confirmed playback through shutdown Signed-off-by: John Tennant Co-authored-by: Darcy <756b3e94a13d7979c09d32291e4fd4da46848bdaa53340021ac24fb843e28d57@buzz.block.builderlab.xyz> --- src-tauri/crates/berd-call/PROTOCOL.md | 8 +- .../crates/berd-call/src/host_session.rs | 8 +- src-tauri/crates/berd-call/src/main.rs | 462 ++++++++++-------- src-tauri/crates/berd-call/src/protocol.rs | 2 - src-tauri/crates/berd-call/src/telemetry.rs | 24 +- .../berd-call/tests/session_protocol.rs | 150 +++++- 6 files changed, 404 insertions(+), 250 deletions(-) diff --git a/src-tauri/crates/berd-call/PROTOCOL.md b/src-tauri/crates/berd-call/PROTOCOL.md index 4593e6467..8a5d6f7a4 100644 --- a/src-tauri/crates/berd-call/PROTOCOL.md +++ b/src-tauri/crates/berd-call/PROTOCOL.md @@ -358,7 +358,7 @@ cancellation and shutdown terminals. {"type":"not_admitted","id":u64,"reason":"paused"|"in_progress"|"cancelled"|"empty_text"|"invalid_voice"} {"type":"state","id":u64,"confirmed_token":u64,"utterances_after":[{"token":u64,"text":string}]} {"type":"cancel_result","id":u64,"outcome":"cancelled"|"stale","speech_id":u64|null} -{"type":"speech_started","id":u64,"speech_id":u64,"tts":TtsSettings} +{"type":"speech_started","id":u64,"speech_id":u64} {"type":"speech_used","id":u64,"speech_id":u64,"tts":TtsSettings} {"type":"speech_completed","id":u64,"speech_id":u64} {"type":"speech_interrupted","id":u64,"speech_id":u64,"spoken_through_utf8":u64} @@ -377,12 +377,12 @@ names the targeted speech. The result is emitted before the runtime requests the host's quiescent `audio_cancelled` barrier. Every speech event carries the originating prepare ID. `speech_started` appears only after the first PCM Chunk is accepted by the host, and exactly one terminal -message follows every admission. When available, its `tts` object contains the sanitized settings held by that speech's synthesis lease, including any per-request voice override. This has the same backend-specific fields as `session.tts` without a revision. It does not change the session default; a resumed speech does not emit another start. +message follows every admission. A resumed speech does not emit another start. -`speech_used` reports confirmed playback when cancellation suppressed `speech_started`. It precedes the terminal event and carries the synthesis lease settings. A speech emits either its start or this fallback usage event, so hosts can count used voices without announcing a late playback start. +`speech_used` appears once when the host confirms positive played frames. Its `tts` object contains the sanitized settings held by that speech's synthesis lease, including per-request voice overrides, with the same backend-specific fields as `session.tts` without a revision. It does not change the session default. Accepting a chunk without playing any frames does not emit usage, and resuming the same speech does not emit it twice. On `shutdown`, all complete earlier frames are processed in order. The parent -keeps stdin open while the child cancels and drains output. The child then +keeps stdin open while the child cancels and drains output. The child continues handling audio acknowledgements while waiting for playback to settle. The child then finishes the input runtime while continuing to drain events and storage receipts, flushes, and exits. The parent drains stdout through EOF before finalizing call telemetry. EOF, malformed framing, fatal input failure, or process death cancels both authorities without transparent diff --git a/src-tauri/crates/berd-call/src/host_session.rs b/src-tauri/crates/berd-call/src/host_session.rs index 90c035cbb..2d8b84c01 100644 --- a/src-tauri/crates/berd-call/src/host_session.rs +++ b/src-tauri/crates/berd-call/src/host_session.rs @@ -1328,12 +1328,8 @@ impl SessionActor { } } } - SessionMessage::SpeechStarted { - tts: Some(settings), - .. - } - | SessionMessage::SpeechUsed { tts: settings, .. } => { - crate::telemetry::speech_started(&settings); + SessionMessage::SpeechUsed { tts: settings, .. } => { + crate::telemetry::speech_used(&settings); } SessionMessage::TtsSettingsResult { id, diff --git a/src-tauri/crates/berd-call/src/main.rs b/src-tauri/crates/berd-call/src/main.rs index ea64030e8..2871ab2d5 100644 --- a/src-tauri/crates/berd-call/src/main.rs +++ b/src-tauri/crates/berd-call/src/main.rs @@ -139,6 +139,7 @@ struct ActivePlayback { input_during_tts: InputDuringTtsSnapshot, tts: TtsConfigurationLease, suspension_requested: bool, + usage_reported: bool, } #[derive(Clone, Debug, PartialEq)] @@ -2115,6 +2116,7 @@ fn run_session(config: SessionConfig, pcm_output_fd: RawFd) -> Result<(), String let status_sound_runtime = ManagedStatusSoundRuntime::spawn(None)?; loop { + publish_speech_usage(active.as_mut(), &mut writer)?; input_progress.enter("voice_events"); if let Some(events) = input_events.as_mut() { while let Ok(event) = events.try_recv() { @@ -2282,6 +2284,7 @@ fn run_session(config: SessionConfig, pcm_output_fd: RawFd) -> Result<(), String interrupt_active(&mut core, &mut active, &mut writer)?; finish_shutdown_playback( &playback_rx, + &control_rx, &mut core, &mut active, &mut writer, @@ -2571,51 +2574,28 @@ fn run_session(config: SessionConfig, pcm_output_fd: RawFd) -> Result<(), String )?; } } - Input::Request(SessionRequest::AudioBeginAccepted { speech_id }) => { - if let Err(message) = - handle_audio_ack(speech_id, AudioHostAck::BeginAccepted, active.as_ref()) - { - write_protocol_fatal( - &mut writer, - "invalid host audio acknowledgement", - &message, - )?; - abort_active(&active); - return Ok(()); - } - } - Input::Request(SessionRequest::AudioBeginFailed { - speech_id, - played_frames, - message, - }) => { - eprintln!("host audio begin failed: {message}"); - if let Err(message) = handle_audio_ack( - speech_id, - AudioHostAck::BeginFailed { - played_frames, - message, - }, - active.as_ref(), - ) { - write_protocol_fatal( - &mut writer, - "invalid host audio acknowledgement", - &message, - )?; - abort_active(&active); - return Ok(()); + Input::Request( + request @ (SessionRequest::AudioBeginAccepted { .. } + | SessionRequest::AudioBeginFailed { .. } + | SessionRequest::AudioChunkAccepted { .. } + | SessionRequest::AudioPlayed { .. } + | SessionRequest::AudioSuspended { .. } + | SessionRequest::AudioResumed { .. } + | SessionRequest::AudioDrained { .. } + | SessionRequest::AudioFailed { .. } + | SessionRequest::AudioCancelled { .. }), + ) => { + match &request { + SessionRequest::AudioBeginFailed { message, .. } => { + eprintln!("host audio begin failed: {message}") + } + SessionRequest::AudioFailed { message, .. } => { + eprintln!("host audio output failed: {message}") + } + _ => {} } - } - Input::Request(SessionRequest::AudioChunkAccepted { - speech_id, - sequence, - }) => { - match handle_audio_ack( - speech_id, - AudioHostAck::ChunkAccepted { sequence }, - active.as_ref(), - ) { + let (speech_id, ack) = audio_ack(request).expect("audio request matched"); + match handle_audio_ack(speech_id, ack, active.as_ref()) { Ok(true) => { publish_speech_started(speech_id, &mut core, active.as_ref(), &mut writer)? } @@ -2631,123 +2611,6 @@ fn run_session(config: SessionConfig, pcm_output_fd: RawFd) -> Result<(), String } } } - Input::Request(SessionRequest::AudioPlayed { - speech_id, - played_frames, - }) => { - if let Err(message) = handle_audio_ack( - speech_id, - AudioHostAck::Played { played_frames }, - active.as_ref(), - ) { - write_protocol_fatal( - &mut writer, - "invalid host audio acknowledgement", - &message, - )?; - abort_active(&active); - return Ok(()); - } - } - Input::Request(SessionRequest::AudioSuspended { - speech_id, - played_frames, - }) => { - if let Err(message) = handle_audio_ack( - speech_id, - AudioHostAck::Suspended { played_frames }, - active.as_ref(), - ) { - write_protocol_fatal( - &mut writer, - "invalid host audio acknowledgement", - &message, - )?; - abort_active(&active); - return Ok(()); - } - } - Input::Request(SessionRequest::AudioResumed { - speech_id, - played_frames, - }) => { - if let Err(message) = handle_audio_ack( - speech_id, - AudioHostAck::Resumed { played_frames }, - active.as_ref(), - ) { - write_protocol_fatal( - &mut writer, - "invalid host audio acknowledgement", - &message, - )?; - abort_active(&active); - return Ok(()); - } - } - Input::Request(SessionRequest::AudioDrained { - speech_id, - sequence, - played_frames, - }) => { - if let Err(message) = handle_audio_ack( - speech_id, - AudioHostAck::Drained { - sequence, - played_frames, - }, - active.as_ref(), - ) { - write_protocol_fatal( - &mut writer, - "invalid host audio acknowledgement", - &message, - )?; - abort_active(&active); - return Ok(()); - } - } - Input::Request(SessionRequest::AudioFailed { - speech_id, - played_frames, - message, - }) => { - eprintln!("host audio output failed: {message}"); - if let Err(message) = handle_audio_ack( - speech_id, - AudioHostAck::Failed { - played_frames, - message, - }, - active.as_ref(), - ) { - write_protocol_fatal( - &mut writer, - "invalid host audio acknowledgement", - &message, - )?; - abort_active(&active); - return Ok(()); - } - } - Input::Request(SessionRequest::AudioCancelled { - speech_id, - played_frames, - }) => { - if let Err(message) = handle_audio_ack( - speech_id, - AudioHostAck::Cancelled { played_frames }, - active.as_ref(), - ) { - write_protocol_fatal( - &mut writer, - "invalid host audio acknowledgement", - &message, - )?; - abort_active(&active); - return Ok(()); - } - } Input::Request(SessionRequest::QueryState { id, after }) => { write_state(&mut writer, id, after, &core)? } @@ -4282,6 +4145,7 @@ fn process_prepare( input_during_tts, tts, suspension_requested: false, + usage_reported: false, }); write_message( writer, @@ -4543,11 +4407,33 @@ fn publish_speech_started( ) -> Result<(), String> { if core.mark_started(speech_id) { let id = active.map_or(0, |current| current.prepare_id); - let tts = active.map(|current| current.tts.snapshot().settings.clone()); + write_message(writer, &SessionMessage::SpeechStarted { id, speech_id })?; + } + Ok(()) +} + +fn publish_speech_usage( + current: Option<&mut ActivePlayback>, + writer: &mut impl Write, +) -> Result<(), String> { + let Some(current) = current else { + return Ok(()); + }; + if !current.usage_reported + && current + .output + .as_ref() + .is_some_and(|output| output.played_frames() > 0) + { write_message( writer, - &SessionMessage::SpeechStarted { id, speech_id, tts }, + &SessionMessage::SpeechUsed { + id: current.prepare_id, + speech_id: current.speech_id, + tts: current.tts.snapshot().settings.clone(), + }, )?; + current.usage_reported = true; } Ok(()) } @@ -4559,29 +4445,8 @@ fn finish_playback( message: SessionMessage, writer: &mut impl Write, ) -> Result<(), String> { - if let Some(current) = active - .as_ref() - .filter(|current| current.speech_id == speech_id) - { - // Cancellation can suppress SpeechStarted after audio has already played. - // Marking it here deduplicates usage without publishing a late start. - if current - .output - .as_ref() - .is_some_and(|output| output.played_frames() > 0) - && core.mark_started(speech_id) - { - write_message( - writer, - &SessionMessage::SpeechUsed { - id: current.prepare_id, - speech_id, - tts: current.tts.snapshot().settings.clone(), - }, - )?; - } - } if core.finish(speech_id) { + publish_speech_usage(active.as_mut(), writer)?; if active .as_ref() .is_some_and(|current| current.speech_id == speech_id) @@ -4754,8 +4619,71 @@ fn abort_active(active: &Option) { } } +fn audio_ack(request: SessionRequest) -> Option<(u64, AudioHostAck)> { + use SessionRequest::*; + let (speech_id, ack) = match request { + AudioBeginAccepted { speech_id } => (speech_id, AudioHostAck::BeginAccepted), + AudioBeginFailed { + speech_id, + played_frames, + message, + } => ( + speech_id, + AudioHostAck::BeginFailed { + played_frames, + message, + }, + ), + AudioChunkAccepted { + speech_id, + sequence, + } => (speech_id, AudioHostAck::ChunkAccepted { sequence }), + AudioPlayed { + speech_id, + played_frames, + } => (speech_id, AudioHostAck::Played { played_frames }), + AudioSuspended { + speech_id, + played_frames, + } => (speech_id, AudioHostAck::Suspended { played_frames }), + AudioResumed { + speech_id, + played_frames, + } => (speech_id, AudioHostAck::Resumed { played_frames }), + AudioDrained { + speech_id, + sequence, + played_frames, + } => ( + speech_id, + AudioHostAck::Drained { + sequence, + played_frames, + }, + ), + AudioFailed { + speech_id, + played_frames, + message, + } => ( + speech_id, + AudioHostAck::Failed { + played_frames, + message, + }, + ), + AudioCancelled { + speech_id, + played_frames, + } => (speech_id, AudioHostAck::Cancelled { played_frames }), + _ => return None, + }; + Some((speech_id, ack)) +} + fn finish_shutdown_playback( playback_rx: &Receiver, + control_rx: &Receiver, core: &mut SessionCore, active: &mut Option, writer: &mut impl Write, @@ -4763,17 +4691,30 @@ fn finish_shutdown_playback( ) -> Result<(), String> { let deadline = Instant::now() + timeout; while active.is_some() { + while let Ok(control) = control_rx.try_recv() { + if let Input::Request(request) = control.input { + if let Some((speech_id, ack)) = audio_ack(request) { + handle_audio_ack(speech_id, ack, active.as_ref())?; + } + } + if Instant::now() >= deadline { + break; + } + } + publish_speech_usage(active.as_mut(), writer)?; let remaining = deadline.saturating_duration_since(Instant::now()); let event = if remaining.is_zero() { Err(mpsc::RecvTimeoutError::Timeout) } else { - playback_rx.recv_timeout(remaining) + playback_rx.recv_timeout(remaining.min(Duration::from_millis(2))) }; match event { Ok(event) => handle_playback_event(event, core, active, writer)?, + Err(mpsc::RecvTimeoutError::Timeout) if Instant::now() < deadline => continue, Err(error) => { - let current = active.take().expect("active playback exists"); - core.finish(current.speech_id); + let current = active.as_ref().expect("active playback exists"); + let id = current.prepare_id; + let speech_id = current.speech_id; let message = match error { mpsc::RecvTimeoutError::Timeout => { "playback cancellation timed out during shutdown" @@ -4782,13 +4723,16 @@ fn finish_shutdown_playback( "playback worker disconnected during shutdown" } }; - write_message( - writer, - &SessionMessage::SpeechFailed { - id: current.prepare_id, - speech_id: current.speech_id, + finish_playback( + core, + active, + speech_id, + SessionMessage::SpeechFailed { + id, + speech_id, message: message.into(), }, + writer, )?; } } @@ -5839,12 +5783,13 @@ mod tests { input_during_tts: test_input_policy(), tts: test_tts_lease(), suspension_requested: false, + usage_reported: false, } } #[test] fn cancelled_playback_reports_used_voice_without_a_late_start() { - for (already_started, played_frames) in [(false, 64), (true, 64), (false, 0)] { + for (already_started, played_frames) in [(false, 64), (true, 64), (false, 0), (true, 0)] { let mut core = SessionCore::default(); let mut current = active_playback(&mut core); let speech_id = current.speech_id; @@ -5898,6 +5843,8 @@ mod tests { .unwrap(); assert!(worker.join().unwrap().is_err()); let mut active = Some(current); + publish_speech_usage(active.as_mut(), &mut messages_out).unwrap(); + publish_speech_usage(active.as_mut(), &mut messages_out).unwrap(); handle_playback_event( PlaybackEvent::Interrupted(speech_id, 0), &mut core, @@ -5906,30 +5853,28 @@ mod tests { ) .unwrap(); let rows = messages(&messages_out); - if !already_started && played_frames == 0 { - assert_eq!(rows.len(), 1); - assert_eq!(rows[0]["type"], "speech_interrupted"); - continue; + let usage: Vec<_> = rows + .iter() + .filter(|row| row["type"] == "speech_used") + .collect(); + assert_eq!(usage.len(), usize::from(played_frames > 0)); + if let Some(usage) = usage.first() { + assert_eq!(usage["tts"], serde_json::to_value(settings).unwrap()); } - assert_eq!(rows.len(), 2); assert_eq!( - rows[0]["type"], - if already_started { - "speech_started" - } else { - "speech_used" - } + rows.iter() + .filter(|row| row["type"] == "speech_started") + .count(), + usize::from(already_started) ); - assert_eq!(rows[0]["tts"], serde_json::to_value(settings).unwrap()); - assert_eq!(rows[1]["type"], "speech_interrupted"); + assert_eq!(rows.last().unwrap()["type"], "speech_interrupted"); } } #[test] - fn speech_start_reports_its_lease_once_even_after_resume() { + fn speech_start_is_reported_once_even_after_resume() { let mut core = SessionCore::default(); let current = active_playback(&mut core); - let settings = current.tts.snapshot().settings.clone(); let mut messages = Vec::new(); publish_speech_started(current.speech_id, &mut core, Some(¤t), &mut messages) .unwrap(); @@ -5945,7 +5890,6 @@ mod tests { vec![SessionMessage::SpeechStarted { id: current.prepare_id, speech_id: current.speech_id, - tts: Some(settings), }] ); } @@ -8096,6 +8040,7 @@ mod tests { input_during_tts: test_input_policy(), tts: test_tts_lease(), suspension_requested: false, + usage_reported: false, }); let mut held = None; let mut output = Vec::new(); @@ -8137,6 +8082,7 @@ mod tests { input_during_tts: test_input_policy(), tts: test_tts_lease(), suspension_requested: false, + usage_reported: false, }); let mut output = Vec::new(); @@ -8289,6 +8235,7 @@ mod tests { input_during_tts: test_input_policy(), tts: test_tts_lease(), suspension_requested: false, + usage_reported: false, }); let mut next_token = 1; let mut output = Vec::new(); @@ -8381,6 +8328,7 @@ mod tests { input_during_tts: test_input_policy(), tts: test_tts_lease(), suspension_requested: false, + usage_reported: false, }); let mut next_token = 1; let mut output = Vec::new(); @@ -8567,6 +8515,97 @@ mod tests { assert!(output.cancelled.load(Ordering::SeqCst)); } + #[test] + fn shutdown_settles_queued_audio_acknowledgements_before_counting_usage() { + let mut core = SessionCore::default(); + let mut current = active_playback(&mut core); + let speech_id = current.speech_id; + let (child, mut host) = UnixStream::pair().unwrap(); + let transport = unsafe { AudioPipeTransport::from_raw_fd(child.into_raw_fd()) }.unwrap(); + let (controls, _controls) = mpsc::channel(); + let authority = current.active.as_ref().unwrap().clone(); + let audio = Arc::new( + RemotePcmAudioOutput::new( + speech_id, + LongRemoteTts.pcm_spec(), + Arc::new(transport), + authority, + controls, + ) + .unwrap(), + ); + current.output = Some(Arc::clone(&audio)); + let (playback_tx, playback_rx) = mpsc::channel(); + let output = Arc::clone(&audio); + let worker = thread::spawn(move || { + output.start().unwrap(); + assert!(output.write(&[0.1; 4096]).is_err()); + playback_tx + .send(PlaybackEvent::Interrupted(speech_id, 0)) + .unwrap(); + }); + assert_eq!( + read_audio_record(&mut host).0, + session_audio::AUDIO_BEGIN_KIND + ); + audio.handle_ack(AudioHostAck::BeginAccepted).unwrap(); + assert_eq!( + read_audio_record(&mut host).0, + session_audio::AUDIO_CHUNK_KIND + ); + let mut active = Some(current); + let mut messages_out = Vec::new(); + interrupt_active(&mut core, &mut active, &mut messages_out).unwrap(); + assert_eq!( + read_audio_record(&mut host).0, + session_audio::AUDIO_CANCEL_KIND + ); + let (sender, receiver) = mpsc::channel(); + for request in [ + SessionRequest::AudioChunkAccepted { + speech_id, + sequence: 1, + }, + SessionRequest::AudioCancelled { + speech_id, + played_frames: 64, + }, + ] { + sender + .send(OrderedControl { + after_pcm: 999, + input: Input::Request(request), + }) + .unwrap(); + } + finish_shutdown_playback( + &playback_rx, + &receiver, + &mut core, + &mut active, + &mut messages_out, + Duration::from_millis(100), + ) + .unwrap(); + // Release the worker even on the unfixed path, so a failing test leaves no blocked thread. + if audio.played_frames() == 0 { + audio + .handle_ack(AudioHostAck::ChunkAccepted { sequence: 1 }) + .unwrap(); + audio + .handle_ack(AudioHostAck::Cancelled { played_frames: 64 }) + .unwrap(); + } + worker.join().unwrap(); + let rows = messages(&messages_out); + assert_eq!( + rows.iter() + .map(|row| row["type"].as_str().unwrap()) + .collect::>(), + ["speech_used", "speech_interrupted"] + ); + } + #[test] fn shutdown_drains_started_before_interrupted_terminal() { let mut core = SessionCore::default(); @@ -8582,6 +8621,7 @@ mod tests { finish_shutdown_playback( &receiver, + &mpsc::channel().1, &mut core, &mut active, &mut output, @@ -8607,6 +8647,7 @@ mod tests { let mut output = Vec::new(); finish_shutdown_playback( &receiver, + &mpsc::channel().1, &mut core, &mut active, &mut output, @@ -8626,6 +8667,7 @@ mod tests { let mut output = Vec::new(); finish_shutdown_playback( &receiver, + &mpsc::channel().1, &mut core, &mut active, &mut output, diff --git a/src-tauri/crates/berd-call/src/protocol.rs b/src-tauri/crates/berd-call/src/protocol.rs index 8852e1fff..279366baf 100644 --- a/src-tauri/crates/berd-call/src/protocol.rs +++ b/src-tauri/crates/berd-call/src/protocol.rs @@ -241,8 +241,6 @@ pub enum SessionMessage { SpeechStarted { id: u64, speech_id: u64, - #[serde(default, skip_serializing_if = "Option::is_none")] - tts: Option, }, SpeechUsed { id: u64, diff --git a/src-tauri/crates/berd-call/src/telemetry.rs b/src-tauri/crates/berd-call/src/telemetry.rs index 5751858e5..720b04ddf 100644 --- a/src-tauri/crates/berd-call/src/telemetry.rs +++ b/src-tauri/crates/berd-call/src/telemetry.rs @@ -159,7 +159,7 @@ struct VoiceRequests { } impl VoiceRequests { - fn started(&mut self, settings: &TtsSettings) { + fn used(&mut self, settings: &TtsSettings) { let name = match settings { TtsSettings::Siri { voice, language, .. @@ -233,10 +233,10 @@ pub(crate) fn assistant_response() { /// Session playback reports the settings held by its actual synthesis lease. /// No per-speech record is exported; the call accumulates counts until end. -pub(crate) fn speech_started(settings: &TtsSettings) { +pub(crate) fn speech_used(settings: &TtsSettings) { if let Ok(mut call) = CALL.lock() { if let Some(call) = call.as_mut() { - call.voices.started(settings); + call.voices.used(settings); } } } @@ -583,9 +583,9 @@ mod tests { voice: voice.into(), rate: 1.0, }; - requests.started(&openai("alloy")); - requests.started(&openai("alloy")); - requests.started(&openai("marin")); + requests.used(&openai("alloy")); + requests.used(&openai("alloy")); + requests.used(&openai("marin")); let ended = payload( production(), ID, @@ -849,12 +849,12 @@ mod tests { language: language.into(), rate: 1.0, }; - requests.started(&siri("en-US")); - requests.started(&siri("en-US")); - requests.started(&siri("en_US")); - requests.started(&pocket(1.0)); - requests.started(&siri("en-AU")); // Same code in a different locale is not this identity. - requests.started(&TtsSettings::OpenAi { + requests.used(&siri("en-US")); + requests.used(&siri("en-US")); + requests.used(&siri("en_US")); + requests.used(&pocket(1.0)); + requests.used(&siri("en-AU")); // Same code in a different locale is not this identity. + requests.used(&TtsSettings::OpenAi { model: "test".into(), voice: "private override".into(), rate: 1.0, diff --git a/src-tauri/crates/berd-call/tests/session_protocol.rs b/src-tauri/crates/berd-call/tests/session_protocol.rs index 3985f92b3..44582b88a 100644 --- a/src-tauri/crates/berd-call/tests/session_protocol.rs +++ b/src-tauri/crates/berd-call/tests/session_protocol.rs @@ -118,7 +118,7 @@ fn spawn_audio_host( reader: UnixStream, stdin: Arc>, ) -> std::thread::JoinHandle> { - spawn_audio_host_with_played_limit(reader, stdin, None, false) + spawn_audio_host_with_played_limit(reader, stdin, None, false, false) } fn spawn_audio_host_with_played_limit( @@ -126,6 +126,7 @@ fn spawn_audio_host_with_played_limit( stdin: Arc>, played_frame_limit: Option, mut fail_at_limit: bool, + mut shutdown_before_ack: bool, ) -> std::thread::JoinHandle> { std::thread::spawn(move || { let mut current = None::<(u64, u64, u64)>; @@ -169,6 +170,10 @@ fn spawn_audio_host_with_played_limit( json!({"type":"audio_chunk_accepted","speech_id":speech_id,"sequence":sequence}), json!({"type":"audio_played","speech_id":speech_id,"played_frames":played_frame_limit.map_or(state.2, |limit| state.2.min(limit))}), ]; + if shutdown_before_ack { + shutdown_before_ack = false; + replies.insert(0, json!({"type":"shutdown"})); + } if reached_limit && fail_at_limit { fail_at_limit = false; let played_frames = played_frame_limit.unwrap(); @@ -262,7 +267,8 @@ fn audio_fixture_truncated_payload_does_not_panic_before_shutdown_is_checked() { // The child exits halfway through a valid begin payload. audio.write_all(&1_u64.to_le_bytes()).unwrap(); drop(audio); - let host = spawn_audio_host_with_played_limit(reader, Arc::new(Mutex::new(input)), None, false); + let host = + spawn_audio_host_with_played_limit(reader, Arc::new(Mutex::new(input)), None, false, false); let result = host .join() .expect("payload EOF panicked before child status was checked"); @@ -311,7 +317,8 @@ fn audio_fixture_can_finish_a_queued_reply_after_child_input_closes() { audio.write_all(&1_u64.to_le_bytes()).unwrap(); audio.write_all(&24_000_u64.to_le_bytes()).unwrap(); drop(audio); - let host = spawn_audio_host_with_played_limit(reader, Arc::new(Mutex::new(input)), None, false); + let host = + spawn_audio_host_with_played_limit(reader, Arc::new(Mutex::new(input)), None, false, false); host.join() .expect("late fixture reply panicked after child shutdown") .unwrap(); @@ -688,12 +695,20 @@ fn siri_remote_output_supports_consecutive_turns_and_cancellation() { write_session_json(&mut *stdin, &message); stdin.flush().unwrap(); }; + let usage = Mutex::new(Vec::new()); let receive = || -> Value { - let line = receiver - .recv_timeout(Duration::from_secs(10)) - .expect("child did not emit a bounded response") - .unwrap(); - serde_json::from_str(&line).unwrap() + loop { + let line = receiver + .recv_timeout(Duration::from_secs(10)) + .expect("child did not emit a bounded response") + .unwrap(); + let message: Value = serde_json::from_str(&line).unwrap(); + if message["type"] == "speech_used" { + usage.lock().unwrap().push(message); + } else { + return message; + } + } }; send(json!({"type":"hello","id":1,"input_during_tts":"allow_barge_in"})); @@ -753,6 +768,20 @@ fn siri_remote_output_supports_consecutive_turns_and_cancellation() { send(json!({"type":"cancel","id":5})); assert_eq!(receive()["type"], "cancel_result"); assert_eq!(receive()["type"], "speech_interrupted"); + let usage = usage.into_inner().unwrap(); + assert_eq!(usage[0]["id"], 2); + assert_eq!(usage[1]["id"], 3); + assert!(usage + .iter() + .all(|event| [2, 3, 4].contains(&event["id"].as_u64().unwrap()))); + assert_eq!( + usage + .iter() + .map(|event| event["speech_id"].as_u64().unwrap()) + .collect::>() + .len(), + usage.len() + ); send(json!({"type":"shutdown"})); drop(stdin); let mut child_process = child.0.take().unwrap(); @@ -763,6 +792,12 @@ fn siri_remote_output_supports_consecutive_turns_and_cancellation() { #[test] fn speech_whose_player_stops_is_interrupted_where_it_was_heard_and_the_call_goes_on() { + for played_frames in [0, 1_200] { + assert_player_stop_usage(played_frames); + } +} + +fn assert_player_stop_usage(played_frames: u64) { let (speech_url, speech_requests) = spawn_speech_server(); let (mut command, pcm_guard, host) = session_command(); let child = command @@ -789,21 +824,27 @@ fn speech_whose_player_stops_is_interrupted_where_it_was_heard_and_the_call_goes let stdin = Arc::new(Mutex::new(process.stdin.take().unwrap())); let output = spawn_session_message_reader(process.stdout.take().unwrap()); // The host's player stops partway through the first speech, as when the speakers change. - let audio_host = - spawn_audio_host_with_played_limit(host, Arc::clone(&stdin), Some(1_200), true); + let audio_host = spawn_audio_host_with_played_limit( + host, + Arc::clone(&stdin), + Some(played_frames), + true, + false, + ); let send = |message: Value| { let mut stdin = stdin.lock().unwrap(); write_session_json(&mut *stdin, &message); stdin.flush().unwrap(); }; - let started = Mutex::new(Vec::new()); + let used = Mutex::new(Vec::new()); let terminal = |id: u64| loop { let message = output.recv_timeout(Duration::from_secs(5)).unwrap(); match message["type"].as_str() { Some("admitted") => send(json!({ "type":"output_ready","id":id,"speech_id":message["speech_id"] })), - Some("speech_started") => started.lock().unwrap().push(message["tts"].clone()), + Some("speech_used") => used.lock().unwrap().push(message["tts"].clone()), + Some("speech_started") => {} Some("output_ready_result") => {} _ => break message, } @@ -821,10 +862,12 @@ fn speech_whose_player_stops_is_interrupted_where_it_was_heard_and_the_call_goes json!({"type":"prepare_speak","id":3,"acknowledgement":null,"text":"Still here.","voice":"marin"}), ); assert_eq!(terminal(3)["type"], "speech_completed"); - let started = started.into_inner().unwrap(); - assert_eq!(started.len(), 2); - assert_eq!(started[0]["voice"], "alloy"); - assert_eq!(started[1]["voice"], "marin"); + let used = used.into_inner().unwrap(); + assert_eq!(used.len(), if played_frames > 0 { 2 } else { 1 }); + if played_frames > 0 { + assert_eq!(used[0]["voice"], "alloy"); + } + assert_eq!(used.last().unwrap()["voice"], "marin"); for expected in ["alloy", "marin"] { assert_eq!( speech_requests @@ -839,6 +882,81 @@ fn speech_whose_player_stops_is_interrupted_where_it_was_heard_and_the_call_goes audio_host.join().unwrap().unwrap(); } +#[test] +fn shutdown_before_first_chunk_ack_reports_only_confirmed_usage() { + for played_frames in [0, 64] { + let (speech_url, _speech_requests) = spawn_speech_server(); + let (mut command, pcm_guard, host) = session_command(); + let child = command + .args([ + "--tts-backend", + "openai", + "--tts-url", + &speech_url, + "--stt-backend", + "openai", + "--stt-url", + &spawn_transcription_server(), + ]) + .env("OPENAI_API_KEY", "test-key") + .env("OPENAI_TTS_VOICE", "alloy") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + drop(pcm_guard); + let mut child = ChildGuard(Some(child)); + let process = child.0.as_mut().unwrap(); + let stdin = Arc::new(Mutex::new(process.stdin.take().unwrap())); + let output = spawn_session_message_reader(process.stdout.take().unwrap()); + let host = spawn_audio_host_with_played_limit( + host, + Arc::clone(&stdin), + Some(played_frames), + false, + true, + ); + let send = |message: Value| { + let mut stdin = stdin.lock().unwrap(); + write_session_json(&mut *stdin, &message); + stdin.flush().unwrap(); + }; + send(json!({"type":"hello","id":1,"input_during_tts":"allow_barge_in"})); + assert_eq!( + output.recv_timeout(Duration::from_secs(5)).unwrap()["type"], + "ready" + ); + send( + json!({"type":"prepare_speak","id":2,"acknowledgement":null,"text":"A reply stopped before its first acknowledgement."}), + ); + let mut usage = Vec::new(); + loop { + let message = output.recv_timeout(Duration::from_secs(5)).unwrap(); + match message["type"].as_str() { + Some("admitted") => { + send(json!({"type":"output_ready","id":2,"speech_id":message["speech_id"]})) + } + Some("speech_used") => usage.push(message["tts"].clone()), + Some("output_ready_result") => {} + Some("speech_interrupted") => break, + Some("speech_failed") + if message["message"] == "remote PCM output was cancelled" => + { + break + } + _ => panic!("unexpected shutdown event: {message}"), + } + } + assert_eq!(usage.len(), usize::from(played_frames > 0)); + if let Some(settings) = usage.first() { + assert_eq!(settings["voice"], "alloy"); + } + assert!(child.0.take().unwrap().wait().unwrap().success()); + host.join().unwrap().unwrap(); + } +} + /// Accepts one OpenAI transcription session that never hears anything. fn spawn_transcription_server() -> String { let (endpoint_tx, endpoint_rx) = mpsc::sync_channel(1); From 80e163473b81838eeeba924c08fb627879aff11c Mon Sep 17 00:00:00 2001 From: John Tennant Date: Thu, 5 Nov 2026 10:43:44 -0500 Subject: [PATCH 5/8] docs(berd-call): state voice telemetry gateway compatibility Signed-off-by: John Tennant Co-authored-by: Darcy <756b3e94a13d7979c09d32291e4fd4da46848bdaa53340021ac24fb843e28d57@buzz.block.builderlab.xyz> --- src-tauri/crates/berd-call/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src-tauri/crates/berd-call/README.md b/src-tauri/crates/berd-call/README.md index 269a40bba..99b659571 100644 --- a/src-tauri/crates/berd-call/README.md +++ b/src-tauri/crates/berd-call/README.md @@ -451,4 +451,4 @@ On macOS, conventional sessions with `--stream` also emit ephemeral `0\tuser_int ### Call voice telemetry -With telemetry enabled, standalone calls emit one start and one end event using `berd-call-otlp-logs-v2`. The existing `tts_voice` remains the call's starting default. The ended event also carries `tts_voice_requests`, an OTLP key/value list of catalog display names and numeric request counts. A request counts once when playback begins, including interrupted speech; queued or rejected speech that never plays does not count, and resuming a request does not count it again. Per-request overrides and later settings use the voice held by that speech's synthesis lease. Unknown voices are grouped as `custom`; no agent names, text or per-speech events are exported. Deploy gateway v2 support before releasing this client. +With telemetry enabled, standalone calls emit one start and one end event using `berd-call-otlp-logs-v2`. The existing `tts_voice` remains the call's starting default. The ended event also carries `tts_voice_requests`, an OTLP key/value list of catalog display names and numeric request counts. A request counts once when the host confirms positive played frames, including interrupted speech; queued or rejected speech that never plays does not count, and resuming a request does not count it again. Per-request overrides and later settings use the voice held by that speech's synthesis lease. Unknown voices are grouped as `custom`; no agent names, text or per-speech events are exported. Telemetry uploads require a gateway that accepts `berd-call-otlp-logs-v2`. From a75a7552099eddd4151edd7a32bda108570fd969 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Thu, 5 Nov 2026 10:58:29 -0500 Subject: [PATCH 6/8] docs(berd-call): explain shutdown regression cleanup Signed-off-by: John Tennant Co-authored-by: Darcy <756b3e94a13d7979c09d32291e4fd4da46848bdaa53340021ac24fb843e28d57@buzz.block.builderlab.xyz> --- src-tauri/crates/berd-call/src/main.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src-tauri/crates/berd-call/src/main.rs b/src-tauri/crates/berd-call/src/main.rs index 2871ab2d5..a72a01194 100644 --- a/src-tauri/crates/berd-call/src/main.rs +++ b/src-tauri/crates/berd-call/src/main.rs @@ -8587,7 +8587,7 @@ mod tests { Duration::from_millis(100), ) .unwrap(); - // Release the worker even on the unfixed path, so a failing test leaves no blocked thread. + // Release the worker if shutdown leaves acknowledgements unprocessed, so a failing test cannot leave a blocked thread. if audio.played_frames() == 0 { audio .handle_ack(AudioHostAck::ChunkAccepted { sequence: 1 }) From ae208a3b74e4cb0ecd20aa3a97fcc2b33ae19ea4 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Thu, 5 Nov 2026 11:13:19 -0500 Subject: [PATCH 7/8] fix(berd-call): skip admission handshakes while stopping Co-authored-by: Darcy <756b3e94a13d7979c09d32291e4fd4da46848bdaa53340021ac24fb843e28d57@buzz.block.builderlab.xyz> Signed-off-by: John Tennant --- .../crates/berd-call/src/host_session.rs | 39 ++++++++++++++++++- 1 file changed, 38 insertions(+), 1 deletion(-) diff --git a/src-tauri/crates/berd-call/src/host_session.rs b/src-tauri/crates/berd-call/src/host_session.rs index 2d8b84c01..7c356919d 100644 --- a/src-tauri/crates/berd-call/src/host_session.rs +++ b/src-tauri/crates/berd-call/src/host_session.rs @@ -1407,7 +1407,7 @@ impl SessionActor { ))); } SessionMessage::Admitted { id, speech_id, .. } - if self.pending_prepare_id() == Some(id) => + if !self.stopping && self.pending_prepare_id() == Some(id) => { send_request(&self.writer, &SessionRequest::OutputReady { id, speech_id })?; } @@ -3114,6 +3114,43 @@ mod tests { ); } + #[test] + fn stopping_drains_admission_without_writing_to_the_exited_child() { + let mut process = Command::new("/usr/bin/true") + .stdin(Stdio::piped()) + .spawn() + .unwrap(); + let writer = Arc::new(Mutex::new(process.stdin.take().unwrap())); + assert!(process.wait().unwrap().success()); + let (sender, events) = mpsc::sync_channel(1); + sender + .send(SessionMessage::Admitted { + id: 2, + speech_id: 3, + confirmed_token: 0, + }) + .unwrap(); + drop(sender); + let (_commands, commands) = mpsc::sync_channel(1); + let (audio, _audio) = mpsc::sync_channel(1); + let mut actor = test_actor(writer, events, commands, audio); + actor.pending_speak = Some(PendingSpeak { + prepare_id: 2, + text: "test".into(), + non_blocking: false, + response: None, + }); + actor.stopping = true; + let mut child = SessionProcess { + child: process, + stdin: None, + stdout: None, + audio: None, + reaped: true, + }; + actor.finish_child(&mut child).unwrap(); + } + #[test] fn status_sounds_reported_while_stopping_wait_for_the_next_session() { let mut child = Command::new("/usr/bin/true") From c670b5d80f85d42e08bd217873f36d06c6cebf4b Mon Sep 17 00:00:00 2001 From: John Tennant Date: Thu, 5 Nov 2026 12:16:01 -0500 Subject: [PATCH 8/8] docs(berd-call): describe starting voice telemetry directly Co-authored-by: Darcy <756b3e94a13d7979c09d32291e4fd4da46848bdaa53340021ac24fb843e28d57@buzz.block.builderlab.xyz> Signed-off-by: John Tennant --- src-tauri/crates/berd-call/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src-tauri/crates/berd-call/README.md b/src-tauri/crates/berd-call/README.md index 99b659571..b381855f8 100644 --- a/src-tauri/crates/berd-call/README.md +++ b/src-tauri/crates/berd-call/README.md @@ -451,4 +451,4 @@ On macOS, conventional sessions with `--stream` also emit ephemeral `0\tuser_int ### Call voice telemetry -With telemetry enabled, standalone calls emit one start and one end event using `berd-call-otlp-logs-v2`. The existing `tts_voice` remains the call's starting default. The ended event also carries `tts_voice_requests`, an OTLP key/value list of catalog display names and numeric request counts. A request counts once when the host confirms positive played frames, including interrupted speech; queued or rejected speech that never plays does not count, and resuming a request does not count it again. Per-request overrides and later settings use the voice held by that speech's synthesis lease. Unknown voices are grouped as `custom`; no agent names, text or per-speech events are exported. Telemetry uploads require a gateway that accepts `berd-call-otlp-logs-v2`. +With telemetry enabled, standalone calls emit one start and one end event using `berd-call-otlp-logs-v2`. `tts_voice` records the call's starting default. The ended event also carries `tts_voice_requests`, an OTLP key/value list of catalog display names and numeric request counts. A request counts once when the host confirms positive played frames, including interrupted speech; queued or rejected speech that never plays does not count, and resuming a request does not count it again. Per-request overrides and later settings use the voice held by that speech's synthesis lease. Unknown voices are grouped as `custom`; no agent names, text or per-speech events are exported. Telemetry uploads require a gateway that accepts `berd-call-otlp-logs-v2`.