berd-call: stream the user's words while they are transcribed

src-tauri/crates/berd-call/PROTOCOL.md

@@ -312,6 +312,18 @@ Expert-Spokesperson sessions use `origin` to distinguish user transcripts,
 Spokesperson transcripts, and handoffs. User finals interrupt reserved or playing
 assistant output only after storage and emission. Final text is at most 64 KiB.
 
+Conventional sessions on macOS also emit the user's words while they are still
+being transcribed:
+
+```text
+{"type":"user_interim","token":u64,"text":string}
+```
+
+Each `user_interim` replaces the previous one, and empty text clears it.
+`token` is the token the next user final will take, so a reply composed against
+interim words is still current once they finalize. The message is never stored,
+never allocates a token, and never interrupts output.
+
 Expert-Spokesperson sessions also emit two previews for hosts that show the
 conversation as it happens:
 

src-tauri/crates/berd-call/README.md

@@ -496,4 +496,6 @@ Auto is the default input-during-speech preference. It mutes microphone input du
 
 `start --stream` (including `--stream live`) 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.
+
 Session protocol 8 defines `recognition_pending.active` as the agent speech admission hold, not the count of unresolved native recognitions. The hold may expire while recognition remains active. `DetectionSettings.pending_utterance_timeout_ms` is saved with device profiles and defaults to 5000 for older profiles. In-flight deadlines retain their original duration. Native finals have no utterance identity, so expired credits are consumed before newer holds and excluded from detector tuning. At 1024 unresolved credits attribution becomes conservatively ambiguous until an explicit recognizer reset; finals still forward, but holds resolve only by timeout during that period.

src-tauri/crates/berd-call/src/benchmark/stt.rs

@@ -569,6 +569,7 @@ fn observe_event(
                 tracker.speaking_ended = Some(now);
             }
         }
+        Some(VoiceInputEvent::InterimTranscript(_)) => {}
         Some(VoiceInputEvent::RecognitionPendingChanged(active)) => {
             tracker.pending = active;
             if active {

src-tauri/crates/berd-call/src/host_session.rs

@@ -201,6 +201,13 @@ impl Transcript {
     }
 }
 
+fn user_interim_line(cursor: u64, text: &str) -> String {
+    format!(
+        "0\tuser_interim\t{}",
+        json!({"cursor": cursor, "text": text})
+    )
+}
+
 fn input_activity_line(speaking: bool, recognizing: bool) -> String {
     format!(
         "0\tinput_activity\t{}",
@@ -1007,6 +1014,7 @@ struct SessionActor {
     saved: Option<(std::path::PathBuf, crate::saved_settings::SavedSettings)>,
     input_speaking: bool,
     recognition_pending: bool,
+    interim_shown: bool,
     pending_polls: Vec<PendingInputPoll>,
     writer: Arc<Mutex<ChildStdin>>,
     events: Receiver<SessionMessage>,
@@ -1178,6 +1186,7 @@ impl SessionActor {
             saved: None,
             input_speaking: false,
             recognition_pending: false,
+            interim_shown: false,
             pending_polls: Vec::new(),
             writer,
             events,
@@ -1779,11 +1788,17 @@ impl SessionActor {
             SessionMessage::InputMuteApplied { id, active } => {
                 // Commit before acknowledging so a restart cannot read a stale value.
                 self.muted.store(active, Ordering::SeqCst);
+                if active {
+                    self.stream_user_interim(0, "")?;
+                }
                 self.stream_input_activity()?;
                 if let Some(response) = self.pending_settings.remove(&id) {
                     let _ = response.send(Ok(json!({"muted":active})));
                 }
             }
+            SessionMessage::UserInterim { token, text } if !self.expert_spokesperson => {
+                self.stream_user_interim(token, &text)?
+            }
             SessionMessage::LiveEvent {
                 token,
                 text,
@@ -1937,9 +1952,26 @@ impl SessionActor {
     fn reset_input_activity(&mut self) -> Result<(), String> {
         self.input_speaking = false;
         self.recognition_pending = false;
+        self.stream_user_interim(0, "")?;
         self.stream_input_activity()
     }
 
+    /// Ephemeral like input activity. Muted words are never shown.
+    fn stream_user_interim(&mut self, cursor: u64, text: &str) -> Result<(), String> {
+        let text = if self.muted.load(Ordering::SeqCst) {
+            ""
+        } else {
+            text
+        };
+        if matches!(&*self.transcript, Transcript::Stdout { .. })
+            && (self.interim_shown || !text.is_empty())
+        {
+            self.interim_shown = !text.is_empty();
+            print_flushed(&user_interim_line(cursor, text))?;
+        }
+        Ok(())
+    }
+
     fn stream_live_event(
         &self,
         token: u64,
@@ -4971,6 +5003,87 @@ mod tests {
         }
     }
 
+    #[test]
+    fn user_interim_stream_shows_words_and_clears_on_mute_and_reset() {
+        const CHILD: &str = "BERD_CALL_TEST_INTERIM_CHILD";
+        if std::env::var_os(CHILD).is_none() {
+            let output = Command::new(std::env::current_exe().unwrap())
+                .args(["--exact", "host_session::tests::user_interim_stream_shows_words_and_clears_on_mute_and_reset", "--nocapture"])
+                .env(CHILD, "1")
+                .output()
+                .unwrap();
+            assert!(
+                output.status.success(),
+                "{}",
+                String::from_utf8_lossy(&output.stderr)
+            );
+            let stdout = String::from_utf8(output.stdout).unwrap();
+            let rows: Vec<Value> = stdout
+                .lines()
+                .filter_map(|line| {
+                    line.strip_prefix("0\tuser_interim\t")
+                        .map(|text| serde_json::from_str(text).unwrap())
+                })
+                .collect();
+            let expected: Vec<_> = [
+                (3, "you sh"),
+                (3, "you should\tgo"),
+                (3, ""),
+                (4, "next"),
+                (0, ""),
+                (5, "back"),
+                (0, ""),
+            ]
+            .into_iter()
+            .map(|(cursor, text)| json!({"cursor": cursor, "text": text}))
+            .collect();
+            assert_eq!(rows, expected, "{stdout}");
+            return;
+        }
+        for (transcript, expert) in [
+            (Transcript::Stdout { live: false }, false),
+            (Transcript::Silent, false),
+            (Transcript::Stdout { live: true }, true),
+        ] {
+            let mut child = Command::new("/bin/cat")
+                .stdin(Stdio::piped())
+                .stdout(Stdio::null())
+                .spawn()
+                .unwrap();
+            let writer = Arc::new(Mutex::new(child.stdin.take().unwrap()));
+            let (_event_tx, events) = mpsc::sync_channel(1);
+            let (_command_tx, commands) = mpsc::sync_channel(1);
+            let (audio, _audio_rx) = mpsc::sync_channel(1);
+            let mut actor = test_actor(writer, events, commands, audio);
+            actor.transcript = Arc::new(transcript);
+            actor.expert_spokesperson = expert;
+            actor.muted.store(false, Ordering::SeqCst);
+            let interim = |token, text: &str| SessionMessage::UserInterim {
+                token,
+                text: text.into(),
+            };
+            let mute = |active| SessionMessage::InputMuteApplied { id: 0, active };
+            for event in [
+                interim(3, "you sh"),
+                interim(3, "you should\tgo"),
+                interim(3, ""),
+                interim(4, ""),
+                interim(4, "next"),
+                mute(true),
+                interim(4, "muted words"),
+                mute(false),
+                interim(5, "back"),
+            ] {
+                actor.handle_event(event).unwrap();
+            }
+            actor.reset_input_activity().unwrap();
+            actor.reset_input_activity().unwrap();
+            assert!(actor.transcribed.is_empty());
+            child.kill().unwrap();
+            child.wait().unwrap();
+        }
+    }
+
     #[test]
     fn device_switch_restores_the_saved_pair_and_ignores_old_learning_before_ack() {
         let setup = |name: &str| berd_call::macos_audio_route::AudioSetup {

src-tauri/crates/berd-call/src/input.rs

@@ -156,6 +156,9 @@ pub enum VoiceInputEvent {
     SpeakingReset,
     /// Agent speech admission hold, which may expire before recognition finishes.
     RecognitionPendingChanged(bool),
+    /// Display only: words Apple is still transcribing, or empty once none are.
+    /// Never stored, delivered or treated as an utterance.
+    InterimTranscript(String),
     FinalTranscript {
         text: String,
         storage_receipt: FinalTranscriptStorageReceipt,
@@ -934,6 +937,8 @@ struct PendingRecognitions {
     /// audio does rather than at the last word, so order says more than samples.
     #[cfg(any(target_os = "macos", test))]
     unfinalized: bool,
+    /// The interim text last shown, so repeats and clears are sent once.
+    interim: String,
 }
 
 impl PendingRecognitions {
@@ -1007,6 +1012,14 @@ impl PendingRecognitions {
             .any(|&at| first_word < at && at + FINAL_COVERAGE_SLACK_SAMPLES <= end)
     }
 
+    /// Interim text is best effort: a full event queue drops it rather than stall recognition.
+    fn show_interim(&mut self, events: &tokio_mpsc::Sender<VoiceInputEvent>, text: &str) {
+        if self.interim != text {
+            self.interim = text.to_string();
+            let _ = events.try_send(VoiceInputEvent::InterimTranscript(self.interim.clone()));
+        }
+    }
+
     fn result_is_current(&self) -> bool {
         !self.ambiguous && self.count == self.blocking
     }
@@ -1058,6 +1071,7 @@ impl PendingRecognitions {
 
     fn reset(&mut self, events: &tokio_mpsc::Sender<VoiceInputEvent>) -> Result<(), ()> {
         self.release_hold(events)?;
+        self.show_interim(events, "");
         *self = Self::new();
         Ok(())
     }
@@ -1803,12 +1817,16 @@ fn forward_mac_events_with_detection(
                 if pending.blocking == 0 {
                     *settle_deadline = None;
                 }
+                if !pending.words_outstanding() {
+                    pending.show_interim(events, "");
+                }
             }
             crate::mac_speech::MacSpeechRecognitionEvent::FinalStart(start) => {
                 pending.first_word = Some(start);
             }
-            crate::mac_speech::MacSpeechRecognitionEvent::Heard(end) => {
+            crate::mac_speech::MacSpeechRecognitionEvent::Heard(end, text) => {
                 pending.unfinalized = true;
+                pending.show_interim(events, &text);
                 // Apple repeats interim results, so only new audio is traced.
                 if end > pending.heard {
                     pending.heard = end;
@@ -2064,6 +2082,7 @@ fn mac_speech_worker(
             if pending.expire(&events).is_err() {
                 return;
             }
+            pending.show_interim(&events, "");
             settle_deadline = None;
         }
         let frame = match frames.recv_timeout(Duration::from_millis(50)) {
@@ -3374,6 +3393,7 @@ mod tests {
                     storage_receipt.stored();
                     observed.push("final");
                 }
+                VoiceInputEvent::InterimTranscript(_) => {}
                 VoiceInputEvent::Failed(error) => panic!("unexpected input failure: {error}"),
             }
         }
@@ -4657,11 +4677,11 @@ mod tests {
         let (native_tx, mut native_rx) = tokio_mpsc::unbounded_channel();
         use crate::mac_speech::MacSpeechRecognitionEvent::{Final, Heard};
         // Apple re-reports the second sentence's interim words after the first final.
-        native_tx.send(Heard(310_000)).unwrap();
+        native_tx.send(Heard(310_000, "you should".into())).unwrap();
         native_tx
             .send(Final("first".into(), Some(110_000)))
             .unwrap();
-        native_tx.send(Heard(310_000)).unwrap();
+        native_tx.send(Heard(310_000, "you should".into())).unwrap();
         forward_mac_events(&mut native_rx, &event_tx, &mut pending, &mut deadline, None).unwrap();
         assert_eq!(pending.blocking, 1);
         let words_wait =
@@ -4705,7 +4725,7 @@ mod tests {
         }
         let (native_tx, mut native_rx) = tokio_mpsc::unbounded_channel();
         use crate::mac_speech::MacSpeechRecognitionEvent::{Final, Heard};
-        native_tx.send(Heard(130_000)).unwrap();
+        native_tx.send(Heard(130_000, "words".into())).unwrap();
         native_tx
             .send(Final("words".into(), Some(110_000)))
             .unwrap();
@@ -4902,7 +4922,7 @@ mod tests {
         use crate::mac_speech::MacSpeechRecognitionEvent::Heard;
         let mut learning = Learning::all();
         learning.turn(48_000, 5);
-        learning.hear([Heard(50_000)]);
+        learning.hear([Heard(50_000, "words".into())]);
         learning.pending.expire(&learning.events).unwrap();
         learning.deadline = None;
         learning.hear([final_of("late", 50_000)]);
@@ -4913,7 +4933,7 @@ mod tests {
             assert_eq!(learning.settings().interim_words_wait_ms, 10500);
             learning.turn(48_000 * (n + 2), -1);
             learning.hear([
-                Heard(48_000 * (n + 2) + 100),
+                Heard(48_000 * (n + 2) + 100, "words".into()),
                 final_of("prompt", 48_000 * (n + 2)),
             ]);
         }

src-tauri/crates/berd-call/src/mac_speech.rs

@@ -27,8 +27,8 @@ pub enum MacSpeechRecognitionEvent {
     Final(String, Option<u64>),
     /// Sent just before a final: its first word starts at this 48 kHz sample.
     FinalStart(u64),
-    /// Words still being transcribed, heard through this 48 kHz sample.
-    Heard(u64),
+    /// Interim words still being transcribed, heard through this 48 kHz sample.
+    Heard(u64, String),
     /// Apple finalized everything before this 48 kHz sample.
     Settled(u64),
     Finished,
@@ -179,7 +179,7 @@ fn map_recognition_event(
         3 => Some(MacSpeechRecognitionEvent::Failed(text.unwrap_or_else(
             || "macOS speech recognition failed.".to_string(),
         ))),
-        4 => end.map(MacSpeechRecognitionEvent::Heard),
+        4 => end.map(|end| MacSpeechRecognitionEvent::Heard(end, text.unwrap_or_default())),
         5 => end.map(MacSpeechRecognitionEvent::Settled),
         6 => end.map(MacSpeechRecognitionEvent::FinalStart),
         _ => None,
@@ -331,8 +331,8 @@ mod tests {
             ))
         );
         assert_eq!(
-            map_recognition_event(4, None, 1.5),
-            Some(MacSpeechRecognitionEvent::Heard(72_000))
+            map_recognition_event(4, Some("hel".to_string()), 1.5),
+            Some(MacSpeechRecognitionEvent::Heard(72_000, "hel".to_string()))
         );
         assert_eq!(map_recognition_event(4, None, f64::NAN), None);
         assert_eq!(