diff --git a/src-tauri/crates/berd-call/PROTOCOL.md b/src-tauri/crates/berd-call/PROTOCOL.md index f1b35e7ca..8a5d6f7a4 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. @@ -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} +{"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} @@ -376,12 +377,14 @@ 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. A resumed speech does not emit another 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. 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/README.md b/src-tauri/crates/berd-call/README.md index 54d326b65..b381855f8 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`. `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`. diff --git a/src-tauri/crates/berd-call/src/host_session.rs b/src-tauri/crates/berd-call/src/host_session.rs index 805697456..7c356919d 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 @@ -1311,6 +1328,9 @@ impl SessionActor { } } } + SessionMessage::SpeechUsed { tts: settings, .. } => { + crate::telemetry::speech_used(&settings); + } SessionMessage::TtsSettingsResult { id, outcome, @@ -1387,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 })?; } @@ -3059,6 +3079,78 @@ 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 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") diff --git a/src-tauri/crates/berd-call/src/main.rs b/src-tauri/crates/berd-call/src/main.rs index 06f811c17..a72a01194 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; @@ -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, @@ -4548,6 +4412,32 @@ fn publish_speech_started( 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::SpeechUsed { + id: current.prepare_id, + speech_id: current.speech_id, + tts: current.tts.snapshot().settings.clone(), + }, + )?; + current.usage_reported = true; + } + Ok(()) +} + fn finish_playback( core: &mut SessionCore, active: &mut Option, @@ -4556,6 +4446,7 @@ fn finish_playback( writer: &mut impl Write, ) -> Result<(), String> { 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) @@ -4728,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, @@ -4737,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" @@ -4756,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, )?; } } @@ -5183,7 +5153,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; @@ -5813,9 +5783,117 @@ 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), (true, 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); + 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, + &mut active, + &mut messages_out, + ) + .unwrap(); + let rows = messages(&messages_out); + 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.iter() + .filter(|row| row["type"] == "speech_started") + .count(), + usize::from(already_started) + ); + assert_eq!(rows.last().unwrap()["type"], "speech_interrupted"); } } + #[test] + fn speech_start_is_reported_once_even_after_resume() { + let mut core = SessionCore::default(); + let current = active_playback(&mut core); + 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, + }] + ); + } + 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(); @@ -7962,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(); @@ -8003,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(); @@ -8155,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(); @@ -8247,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(); @@ -8433,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 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 }) + .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(); @@ -8448,6 +8621,7 @@ mod tests { finish_shutdown_playback( &receiver, + &mpsc::channel().1, &mut core, &mut active, &mut output, @@ -8473,6 +8647,7 @@ mod tests { let mut output = Vec::new(); finish_shutdown_playback( &receiver, + &mpsc::channel().1, &mut core, &mut active, &mut output, @@ -8492,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 87e4afdfc..279366baf 100644 --- a/src-tauri/crates/berd-call/src/protocol.rs +++ b/src-tauri/crates/berd-call/src/protocol.rs @@ -242,6 +242,11 @@ pub enum SessionMessage { id: u64, speech_id: u64, }, + SpeechUsed { + id: u64, + speech_id: u64, + tts: TtsSettings, + }, SpeechCompleted { id: u64, speech_id: u64, @@ -284,7 +289,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 +306,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..720b04ddf 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 used(&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_used(settings: &TtsSettings) { + if let Ok(mut call) = CALL.lock() { + if let Some(call) = call.as_mut() { + call.voices.used(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.used(&openai("alloy")); + requests.used(&openai("alloy")); + requests.used(&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.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, + }); + 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..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,7 +792,13 @@ 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(); + 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 .args([ @@ -777,6 +812,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()) @@ -788,20 +824,28 @@ 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 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("output_ready_result" | "speech_started") => {} + Some("speech_used") => used.lock().unwrap().push(message["tts"].clone()), + Some("speech_started") => {} + Some("output_ready_result") => {} _ => break message, } }; @@ -814,14 +858,105 @@ 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 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 + .recv_timeout(Duration::from_secs(5)) + .unwrap()["voice"], + expected + ); + } send(json!({"type":"shutdown"})); drop(stdin); assert!(child.0.take().unwrap().wait().unwrap().success()); 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);