diff --git a/src-tauri/crates/berd-call/PROTOCOL.md b/src-tauri/crates/berd-call/PROTOCOL.md index e1b7b7e62..1c8922022 100644 --- a/src-tauri/crates/berd-call/PROTOCOL.md +++ b/src-tauri/crates/berd-call/PROTOCOL.md @@ -281,6 +281,21 @@ 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. +Expert-Spokesperson sessions also emit two previews for hosts that show the +conversation as it happens: + +```text +{"type":"user_transcribed","token":u64,"text":string} +{"type":"spokesperson_generated","text":string} +``` + +`user_transcribed` follows a user final's `live_event` with its token and plain +text. `spokesperson_generated` carries a reply's whole text once the provider +finishes generating it, before playback, and only after no user speech is still +being transcribed. A reply already spoken or cut off by then gets none. The +Spokesperson transcript recorded after playback stays authoritative. Neither +message changes Expert delivery. + ## Confirmation and admission `acknowledgement` is a request-local causal cutoff. `null` uses the stored diff --git a/src-tauri/crates/berd-call/README.md b/src-tauri/crates/berd-call/README.md index 6eb5cd441..752a84bc1 100644 --- a/src-tauri/crates/berd-call/README.md +++ b/src-tauri/crates/berd-call/README.md @@ -64,7 +64,16 @@ berd-call stop server listens only on `127.0.0.1:5222` by default; all four commands accept `--port` when more than one local call must be addressed. `--stream` writes one TSV row per delivered transcript event after the `cursor`, `role`, and `text` -header. The standalone host intentionally uses only the default devices. It +header. In Expert-Spokesperson mode the Expert, like a GPT-Live delegate, hears +the conversation only when the Spokesperson hands off: the stream prints +nothing until then, and then every row since the last handoff, ending with the +`handoff` row. `--stream live` instead shows the conversation as it happens: a +`user` row as soon as the words are transcribed, and a `spokesperson_generated` +row with cursor `0` as soon as a reply is generated, before it is spoken. The +`spokesperson` or `spokesperson_interrupted` row after playback carries the +reply's cursor and what was heard. A conventional call has no handoffs, so it +always streams each row. A `--codex` task, like `--stream`, wakes only at +handoffs. The standalone host intentionally uses only the default devices. It suppresses recognition during playback when the default output identifies itself as built-in speakers, preventing the call from transcribing its own speech; headphone-like outputs retain barge-in. diff --git a/src-tauri/crates/berd-call/src/cli_help.rs b/src-tauri/crates/berd-call/src/cli_help.rs index 157e0a26d..cc5448b85 100644 --- a/src-tauri/crates/berd-call/src/cli_help.rs +++ b/src-tauri/crates/berd-call/src/cli_help.rs @@ -33,14 +33,17 @@ Run `berd-call help ` for command-specific usage."#; const START_HELP: &str = r#"Run a local voice call backed by the shared session runtime. Usage: - berd-call start [--port PORT] [--stream | --codex] [--non-blocking] + berd-call start [--port PORT] [--stream [handoffs|live] | --codex] [--non-blocking] [--no-menu-bar] --voice NAME --language BCP47 [session options] The foreground process owns the default microphone and audio output. Session options are the same as `berd-call session`, except the PCM descriptor is owned -internally. `--stream` prints cursor, role, and text as TSV. `--codex` delivers -the same records to the calling Codex task (requires CODEX_THREAD_ID). A +internally. `--stream` prints cursor, role, and text as TSV. In +Expert-Spokesperson mode it prints nothing until the Spokesperson hands off, +then everything since the last handoff; `--stream live` prints each row as it +happens. `--codex` delivers the handoff batches to the calling Codex task +(requires CODEX_THREAD_ID). A menu-bar item controls speech rate, input muting, and ending the call; pass `--no-menu-bar` to omit it. diff --git a/src-tauri/crates/berd-call/src/host_session.rs b/src-tauri/crates/berd-call/src/host_session.rs index d827ae2fc..e28968fcf 100644 --- a/src-tauri/crates/berd-call/src/host_session.rs +++ b/src-tauri/crates/berd-call/src/host_session.rs @@ -1,6 +1,6 @@ #![cfg(target_os = "macos")] -use std::collections::VecDeque; +use std::collections::{HashSet, VecDeque}; use std::fs::File; use std::io::{BufRead, BufReader, Read, Write}; use std::os::fd::{FromRawFd, RawFd}; @@ -145,7 +145,7 @@ pub(crate) fn run(options: StartOptions) -> Result<(), String> { /// Where live transcript records go: nowhere, stdout TSV, or a Codex task. pub(crate) enum Transcript { Silent, - Stdout, + Stdout { live: bool }, Codex(CodexRelay), } @@ -153,7 +153,7 @@ impl Transcript { fn for_options(options: &StartOptions) -> Result { Ok(match options.transcript { TranscriptDestination::None => Self::Silent, - TranscriptDestination::Stdout => Self::Stdout, + TranscriptDestination::Stdout { live } => Self::Stdout { live }, TranscriptDestination::Codex => { let target = CodexTarget::from_environment()?; let executable = std::env::current_exe().map_err(|error| { @@ -174,7 +174,7 @@ impl Transcript { fn start(&self) -> Result<(), String> { match self { - Self::Stdout => print_flushed("cursor\trole\ttext"), + Self::Stdout { .. } => print_flushed("cursor\trole\ttext"), Self::Codex(_) => { println!("Berd Call is delivering transcripts to this Codex task."); Ok(()) @@ -192,7 +192,7 @@ impl Transcript { ) -> Result<(), String> { match self { Self::Silent => Ok(()), - Self::Stdout => print_flushed(&format!( + Self::Stdout { .. } => print_flushed(&format!( "{}\t{role}\t{}", cursor.unwrap_or_default(), stream_text(text) @@ -213,7 +213,7 @@ impl Transcript { fn finish(&self, text: &str) { match self { Self::Silent => {} - Self::Stdout => { + Self::Stdout { .. } => { let _ = self.record(None, "lifecycle", None, text); } Self::Codex(relay) => relay.finish(Some(CodexRecord { @@ -771,6 +771,10 @@ struct SessionActor { session: Arc>, transcript: Arc, expert_spokesperson: bool, + // User rows already streamed when transcribed, left out of the Expert batch. + transcribed: HashSet, + // Rows a handoffs stream or Codex task gets only when the Spokesperson hands off. + held: Vec, stopping: bool, restart: Option, muted: Arc, @@ -831,6 +835,8 @@ impl SessionActor { session, transcript, expert_spokesperson, + transcribed: HashSet::new(), + held: Vec::new(), stopping: false, restart: None, muted: Arc::new(AtomicBool::new(false)), @@ -1191,8 +1197,27 @@ impl SessionActor { text, origin, } if !self.expert_spokesperson => self.stream_live_event(token, origin, &text)?, - SessionMessage::ExpertDelivery { events, .. } if self.expert_spokesperson => { - self.stream_expert_delivery(&events)? + // Only a live stream shows the conversation as it happens. + SessionMessage::UserTranscribed { token, text } + if self.expert_spokesperson + && matches!(*self.transcript, Transcript::Stdout { live: true }) => + { + self.transcript.record(Some(token), "user", None, &text)?; + self.transcribed.insert(token); + } + SessionMessage::SpokespersonGenerated { text } + if self.expert_spokesperson + && matches!(*self.transcript, Transcript::Stdout { live: true }) => + { + self.transcript + .record(None, "spokesperson_generated", None, &text)? + } + SessionMessage::ExpertDelivery { + events, + handoff_ids, + .. + } if self.expert_spokesperson => { + self.stream_expert_delivery(events, !handoff_ids.is_empty())? } SessionMessage::Pending { id, utterances } if self.pending_prepare_id() == Some(id) => { self.respond_speak(Ok(json!({"spoke":false,"utterances":utterances}))); @@ -1306,8 +1331,21 @@ impl SessionActor { self.transcript.record(Some(token), role, None, text) } - fn stream_expert_delivery(&self, events: &[RealtimeExpertDeliveryEvent]) -> Result<(), String> { - for event in events { + fn stream_expert_delivery( + &mut self, + events: Vec, + handoff: bool, + ) -> Result<(), String> { + self.held.extend(events); + if !handoff + && matches!( + *self.transcript, + Transcript::Stdout { live: false } | Transcript::Codex(_) + ) + { + return Ok(()); + } + for event in unstreamed(&std::mem::take(&mut self.held), &mut self.transcribed) { self.transcript.record( Some(event.cursor), expert_delivery_role_name(event.role), @@ -2537,6 +2575,16 @@ fn not_admitted_reason_name(reason: NotAdmittedReason) -> &'static str { } } +/// The batch's events not already streamed as they happened. +fn unstreamed<'a>( + events: &'a [RealtimeExpertDeliveryEvent], + transcribed: &'a mut HashSet, +) -> impl Iterator { + events.iter().filter(move |event| { + !(event.role == RealtimeExpertDeliveryRole::User && transcribed.remove(&event.cursor)) + }) +} + fn expert_delivery_role_name(role: RealtimeExpertDeliveryRole) -> &'static str { match role { RealtimeExpertDeliveryRole::User => "user", @@ -2551,6 +2599,64 @@ fn expert_delivery_role_name(role: RealtimeExpertDeliveryRole) -> &'static str { mod tests { use super::*; + #[test] + fn expert_batches_leave_out_user_rows_already_streamed() { + let event = |cursor, role| RealtimeExpertDeliveryEvent { + cursor, + role, + text: format!("event {cursor}"), + handoff_id: None, + }; + let events = [ + event(1, RealtimeExpertDeliveryRole::User), + event(2, RealtimeExpertDeliveryRole::User), + event(3, RealtimeExpertDeliveryRole::Spokesperson), + ]; + let mut transcribed = HashSet::from([1, 3]); + let cursors: Vec<_> = unstreamed(&events, &mut transcribed) + .map(|event| event.cursor) + .collect(); + assert_eq!(cursors, [2, 3]); + assert_eq!(transcribed, HashSet::from([3])); + } + + #[test] + fn handoffs_stream_holds_rows_until_a_handoff() { + 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); + let event = |cursor, role| RealtimeExpertDeliveryEvent { + cursor, + role, + text: format!("event {cursor}"), + handoff_id: None, + }; + actor + .stream_expert_delivery( + vec![ + event(1, RealtimeExpertDeliveryRole::User), + event(2, RealtimeExpertDeliveryRole::Spokesperson), + ], + false, + ) + .unwrap(); + assert_eq!(actor.held.len(), 2); + actor + .stream_expert_delivery(vec![event(3, RealtimeExpertDeliveryRole::Handoff)], true) + .unwrap(); + assert!(actor.held.is_empty()); + drop(actor); + let _ = child.kill(); + let _ = child.wait(); + } + #[test] fn microphone_teardown_reports_a_blocked_forwarder() { let (_release, blocked) = mpsc::sync_channel::<()>(1); @@ -3205,7 +3311,7 @@ mod tests { commands, audio_commands, session, - Arc::new(Transcript::Stdout), + Arc::new(Transcript::Stdout { live: false }), false, ) } diff --git a/src-tauri/crates/berd-call/src/main.rs b/src-tauri/crates/berd-call/src/main.rs index 7e9d0cb1a..7346405ec 100644 --- a/src-tauri/crates/berd-call/src/main.rs +++ b/src-tauri/crates/berd-call/src/main.rs @@ -732,7 +732,11 @@ fn request_bundled_app_update(app_path: &Path) { #[derive(Clone, Copy, Debug, PartialEq)] enum TranscriptDestination { None, - Stdout, + /// `live` streams an Expert-Spokesperson call as it happens; otherwise + /// its rows wait for the next handoff. + Stdout { + live: bool, + }, Codex, } @@ -777,12 +781,17 @@ fn parse_start_args(args: &[String]) -> Result { if transcript != TranscriptDestination::None { return Err("choose one of --stream or --codex for transcript delivery".into()); } - transcript = if flag == "--stream" { - TranscriptDestination::Stdout - } else { + index += 1; + transcript = if flag == "--codex" { TranscriptDestination::Codex + } else { + let mode = args.get(index).map(String::as_str); + let live = mode == Some("live"); + if matches!(mode, Some("live" | "handoffs")) { + index += 1; + } + TranscriptDestination::Stdout { live } }; - index += 1; } "--no-menu-bar" if menu_bar => { menu_bar = false; @@ -3153,6 +3162,8 @@ fn run_expert_spokesperson_session( let mut waiting_responses = VecDeque::::new(); let mut active: Option = None; let mut turn_gate = ExpertTurnGate::new(spokesperson_renew_after()); + // Generated replies waiting to be shown until the words they answer are. + let mut generated = Vec::::new(); let mut session_tts: Option = None; let mut input_during_tts_slot: Option = None; let mut input_muted = false; @@ -3363,6 +3374,12 @@ fn run_expert_spokesperson_session( } SpokespersonEvent::UserTurnDiscarded { item_id } => { turn_gate.discard_user_turn(&item_id); + show_generated_replies( + &mut generated, + &responses, + &turn_gate, + &mut writer, + )?; } SpokespersonEvent::UserFinal { item_id, text } => { turn_gate.resolve_user_final(&item_id); @@ -3372,7 +3389,20 @@ fn run_expert_spokesperson_session( LiveSideEvent::UserTranscript { text: text.clone() }, &mut writer, )?; + write_message( + &mut writer, + &SessionMessage::UserTranscribed { + token: emitted_live_token, + text: text.clone(), + }, + )?; core.record_user_turn(text); + show_generated_replies( + &mut generated, + &responses, + &turn_gate, + &mut writer, + )?; } SpokespersonEvent::ResponseStarted { response_id } => { turn_gate.response_started(&response_id, responses.len())?; @@ -3431,7 +3461,19 @@ fn run_expert_spokesperson_session( response.playback_complete = true; } } + if status == SpokespersonResponseStatus::Completed + && response.delivery.has_transcript() + && !response.playback_complete + { + generated.push(response_id.clone()); + } } + show_generated_replies( + &mut generated, + &responses, + &turn_gate, + &mut writer, + )?; publish_live_response_if_complete( &response_id, &mut responses, @@ -5148,6 +5190,34 @@ fn publish_live_response_if_complete( Ok(()) } +/// Shows each generated reply before it is spoken, once no user words are +/// still being transcribed, so a live transcript reads in order. A reply +/// already spoken or cut off is left to its authoritative transcript. +fn show_generated_replies( + generated: &mut Vec, + responses: &HashMap, + turn_gate: &ExpertTurnGate, + writer: &mut impl Write, +) -> Result<(), String> { + if turn_gate.lifecycle.transcribing() { + return Ok(()); + } + for response_id in generated.drain(..) { + if let Some(response) = responses + .get(&response_id) + .filter(|response| !response.playback_complete && !response.interrupted) + { + write_message( + writer, + &SessionMessage::SpokespersonGenerated { + text: response.delivery.transcript(), + }, + )?; + } + } + Ok(()) +} + fn live_truncation_pending(responses: &HashMap) -> bool { responses .values() @@ -8063,6 +8133,40 @@ mod tests { assert_eq!(emitted[1]["type"], "expert_delivery"); } + #[test] + fn generated_replies_wait_for_the_words_they_answer_and_are_shown_once() { + let reply = |text: &str| { + let mut response = LiveResponse::new(None, None); + response + .delivery + .replace_transcript("assistant", 0, 0, text.into()) + .unwrap(); + response + }; + let mut spoken = reply("Already spoken."); + spoken.playback_complete = true; + let responses = HashMap::from([ + ("response-a".into(), reply("Twelve.")), + ("response-b".into(), spoken), + ]); + let mut turn_gate = ExpertTurnGate::new(Duration::from_secs(60)); + turn_gate.begin_user_speaking("user-1".into()); + turn_gate.finish_user_speaking(); + let mut generated = vec!["response-a".to_string(), "response-b".to_string()]; + let mut output = Vec::new(); + + show_generated_replies(&mut generated, &responses, &turn_gate, &mut output).unwrap(); + assert!(output.is_empty()); + + turn_gate.resolve_user_final("user-1"); + show_generated_replies(&mut generated, &responses, &turn_gate, &mut output).unwrap(); + show_generated_replies(&mut generated, &responses, &turn_gate, &mut output).unwrap(); + let emitted = messages(&output); + assert_eq!(emitted.len(), 1); + assert_eq!(emitted[0]["type"], "spokesperson_generated"); + assert_eq!(emitted[0]["text"], "Twelve."); + } + #[test] fn late_audio_after_interruption_requires_zero_ms_truncation_before_release() { let mut response = LiveResponse::new(None, None); @@ -11355,7 +11459,10 @@ mod tests { ])) .unwrap(); assert_eq!(parsed.port, 5300); - assert_eq!(parsed.transcript, TranscriptDestination::Stdout); + assert_eq!( + parsed.transcript, + TranscriptDestination::Stdout { live: false } + ); assert!(!parsed.expert_spokesperson); assert_eq!( parsed.session_arguments, @@ -11385,6 +11492,18 @@ mod tests { let parsed = start(&["--codex", "--non-blocking"]).unwrap(); assert_eq!(parsed.transcript, TranscriptDestination::Codex); assert!(parsed.non_blocking); + for (flags, live) in [ + (&["--stream"][..], false), + (&["--stream", "handoffs"], false), + (&["--stream", "live", "--non-blocking"], true), + ] { + let parsed = start(flags).unwrap(); + assert_eq!(parsed.transcript, TranscriptDestination::Stdout { live }); + assert_eq!( + parsed.session_arguments[1..], + ["--voice", "Aaron", "--language", "en-US"] + ); + } assert!(start(&["--codex", "--stream"]).is_err()); assert!(start(&["--codex", "--codex"]).is_err()); assert!(start(&["--non-blocking"]).is_err()); diff --git a/src-tauri/crates/berd-call/src/protocol.rs b/src-tauri/crates/berd-call/src/protocol.rs index 9767ee5c2..32eca4e91 100644 --- a/src-tauri/crates/berd-call/src/protocol.rs +++ b/src-tauri/crates/berd-call/src/protocol.rs @@ -235,6 +235,18 @@ pub enum SessionMessage { #[serde(skip_serializing_if = "Option::is_none")] origin: Option, }, + /// The user's words as soon as they are transcribed, for hosts that show + /// the conversation as it happens. `live_event` stays authoritative. + UserTranscribed { + token: u64, + text: String, + }, + /// The Spokesperson's whole reply once it is generated, before it is + /// spoken. The transcript recorded after playback stays authoritative and + /// says how much was heard when the user cut it off. + SpokespersonGenerated { + text: String, + }, /// A crate-selected batch that is ready to wake the external Expert. /// `live_event` remains the authoritative transcript projection; this /// message alone controls when the host schedules an Expert turn. diff --git a/src-tauri/crates/berd-call/src/realtime_host_lifecycle.rs b/src-tauri/crates/berd-call/src/realtime_host_lifecycle.rs index f34009e76..ac70ea306 100644 --- a/src-tauri/crates/berd-call/src/realtime_host_lifecycle.rs +++ b/src-tauri/crates/berd-call/src/realtime_host_lifecycle.rs @@ -90,6 +90,11 @@ impl RealtimeHostLifecycle { self.activity.user_speaking } + /// Whether user speech has started that is not yet transcribed or discarded. + pub fn transcribing(&self) -> bool { + !self.activity.pending_user_items.is_empty() + } + pub fn is_busy(&self, work: RealtimeHostWork) -> bool { self.activity.is_busy(work) }