From 159bda93c6b985ed8197227ded98018a96c2bf13 Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Wed, 4 Mar 2026 17:09:40 -0800 Subject: [PATCH] realtime: fix output format rate and smooth playback buffer --- .../endpoint/realtime_websocket/methods.rs | 5 + .../endpoint/realtime_websocket/protocol.rs | 1 + codex-rs/tui/src/voice.rs | 99 ++++++++++++++++--- 3 files changed, 90 insertions(+), 15 deletions(-) diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs index 7fd0c2b979..7d4f760a7a 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -347,6 +347,7 @@ impl RealtimeWebsocketWriter { output: SessionAudioOutput { format: SessionAudioOutputFormat { kind: "audio/pcm".to_string(), + rate: 24_000, }, voice: "marin".to_string(), }, @@ -902,6 +903,10 @@ mod tests { first_json["session"]["audio"]["output"]["format"]["type"], Value::String("audio/pcm".to_string()) ); + assert_eq!( + first_json["session"]["audio"]["output"]["format"]["rate"], + Value::from(24_000) + ); assert_eq!( first_json["session"]["audio"]["output"]["voice"], Value::String("marin".to_string()) diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs index 9a62163ff3..ca88c5d91f 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs @@ -67,6 +67,7 @@ pub(super) struct SessionAudioOutput { pub(super) struct SessionAudioOutputFormat { #[serde(rename = "type")] pub(super) kind: String, + pub(super) rate: u32, } #[derive(Debug, Clone, Serialize)] diff --git a/codex-rs/tui/src/voice.rs b/codex-rs/tui/src/voice.rs index 227d27c88f..569f6ab4e8 100644 --- a/codex-rs/tui/src/voice.rs +++ b/codex-rs/tui/src/voice.rs @@ -484,7 +484,7 @@ fn convert_u16_to_i16_and_peak(input: &[u16], out: &mut Vec) -> u16 { pub(crate) struct RealtimeAudioPlayer { _stream: cpal::Stream, - queue: Arc>>, + queue: Arc>, output_sample_rate: u32, output_channels: u16, } @@ -495,8 +495,14 @@ impl RealtimeAudioPlayer { crate::audio_device::select_configured_output_device_and_config(config)?; let output_sample_rate = config.sample_rate().0; let output_channels = config.channels(); - let queue = Arc::new(Mutex::new(VecDeque::new())); - let stream = build_output_stream(&device, &config, Arc::clone(&queue))?; + let prebuffer_samples = output_prebuffer_samples(output_sample_rate, output_channels); + let queue = Arc::new(Mutex::new(OutputAudioQueue::default())); + let stream = build_output_stream( + &device, + &config, + Arc::clone(&queue), + prebuffer_samples, + )?; stream .play() .map_err(|e| format!("failed to start output stream: {e}"))?; @@ -537,13 +543,14 @@ impl RealtimeAudioPlayer { .lock() .map_err(|_| "failed to lock output audio queue".to_string())?; // TODO(aibrahim): Cap or trim this queue if we observe producer bursts outrunning playback. - guard.extend(converted); + guard.samples.extend(converted); Ok(()) } pub(crate) fn clear(&self) { if let Ok(mut guard) = self.queue.lock() { - guard.clear(); + guard.samples.clear(); + guard.primed = false; } } } @@ -551,14 +558,17 @@ impl RealtimeAudioPlayer { fn build_output_stream( device: &cpal::Device, config: &cpal::SupportedStreamConfig, - queue: Arc>>, + queue: Arc>, + prebuffer_samples: usize, ) -> Result { let config_any: cpal::StreamConfig = config.clone().into(); match config.sample_format() { cpal::SampleFormat::F32 => device .build_output_stream( &config_any, - move |output: &mut [f32], _| fill_output_f32(output, &queue), + move |output: &mut [f32], _| { + fill_output_f32(output, &queue, prebuffer_samples) + }, move |err| error!("audio output error: {err}"), None, ) @@ -566,7 +576,9 @@ fn build_output_stream( cpal::SampleFormat::I16 => device .build_output_stream( &config_any, - move |output: &mut [i16], _| fill_output_i16(output, &queue), + move |output: &mut [i16], _| { + fill_output_i16(output, &queue, prebuffer_samples) + }, move |err| error!("audio output error: {err}"), None, ) @@ -574,7 +586,9 @@ fn build_output_stream( cpal::SampleFormat::U16 => device .build_output_stream( &config_any, - move |output: &mut [u16], _| fill_output_u16(output, &queue), + move |output: &mut [u16], _| { + fill_output_u16(output, &queue, prebuffer_samples) + }, move |err| error!("audio output error: {err}"), None, ) @@ -583,20 +597,67 @@ fn build_output_stream( } } -fn fill_output_i16(output: &mut [i16], queue: &Arc>>) { +#[derive(Default)] +struct OutputAudioQueue { + samples: VecDeque, + primed: bool, +} + +fn output_prebuffer_samples(sample_rate: u32, channels: u16) -> usize { + let samples_per_second = (sample_rate as usize).saturating_mul(channels as usize); + // 120ms jitter buffer smooths websocket burstiness without adding too much latency. + ((samples_per_second as u64) * 120 / 1_000) as usize +} + +fn should_output_silence( + queue: &mut OutputAudioQueue, + min_buffer_samples: usize, +) -> bool { + if !queue.primed { + if queue.samples.len() < min_buffer_samples { + return true; + } + queue.primed = true; + } + + if queue.samples.is_empty() { + queue.primed = false; + return true; + } + + false +} + +fn fill_output_i16( + output: &mut [i16], + queue: &Arc>, + prebuffer_samples: usize, +) { if let Ok(mut guard) = queue.lock() { + if should_output_silence(&mut guard, prebuffer_samples) { + output.fill(0); + return; + } for sample in output { - *sample = guard.pop_front().unwrap_or(0); + *sample = guard.samples.pop_front().unwrap_or(0); } return; } output.fill(0); } -fn fill_output_f32(output: &mut [f32], queue: &Arc>>) { +fn fill_output_f32( + output: &mut [f32], + queue: &Arc>, + prebuffer_samples: usize, +) { if let Ok(mut guard) = queue.lock() { + if should_output_silence(&mut guard, prebuffer_samples) { + output.fill(0.0); + return; + } for sample in output { - let v = guard.pop_front().unwrap_or(0); + let v = guard.samples.pop_front().unwrap_or(0); *sample = (v as f32) / (i16::MAX as f32); } return; @@ -604,10 +665,18 @@ fn fill_output_f32(output: &mut [f32], queue: &Arc>>) { output.fill(0.0); } -fn fill_output_u16(output: &mut [u16], queue: &Arc>>) { +fn fill_output_u16( + output: &mut [u16], + queue: &Arc>, + prebuffer_samples: usize, +) { if let Ok(mut guard) = queue.lock() { + if should_output_silence(&mut guard, prebuffer_samples) { + output.fill(32768); + return; + } for sample in output { - let v = guard.pop_front().unwrap_or(0); + let v = guard.samples.pop_front().unwrap_or(0); *sample = (v as i32 + 32768).clamp(0, u16::MAX as i32) as u16; } return;