feat(berd-call): count speech requests by voice in call telemetry

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.

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`.

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")