voice_loop: Phase 1 barge-in (duck mic + cancel/truncate)

src/voice_loop/audio.rs

@@ -130,24 +130,28 @@ pub(super) fn start_output_stream(
     Ok((stream, sample_rate))
 }
 
+// Returns the number of source (24 kHz) PCM samples decoded from `delta`, so
+// the caller can track how much assistant audio has been enqueued for
+// conversation.item.truncate accounting on barge-in.
 pub(super) fn enqueue_audio_delta(
     delta: &str,
     queue: Arc<Mutex<PlaybackBuffer>>,
     output_rate: u32,
-) -> Result<(), String> {
+) -> Result<usize, String> {
     let bytes = base64::engine::general_purpose::STANDARD
         .decode(delta)
         .map_err(|e| format!("invalid audio delta base64: {e}"))?;
     let mut pcm = Vec::with_capacity(bytes.len() / 2);
     for chunk in bytes.chunks_exact(2) {
         pcm.push(i16::from_le_bytes([chunk[0], chunk[1]]));
     }
+    let source_samples = pcm.len();
     let resampled = resample_i16(&pcm, 24_000, output_rate);
     let mut guard = queue.lock().map_err(|_| "playback queue poisoned")?;
     guard
         .samples
         .extend(resampled.into_iter().map(|s| s as f32 / i16::MAX as f32));
-    Ok(())
+    Ok(source_samples)
 }
 
 pub(super) fn playback_depth_ms(
@@ -161,6 +165,26 @@ pub(super) fn playback_depth_ms(
     Some(guard.samples.len() * 1_000 / output_rate as usize)
 }
 
+// Drop any queued playback audio. Used on barge-in so we stop voicing the
+// (now-cancelled) assistant turn the moment the user starts speaking.
+pub(super) fn clear_playback(queue: &Arc<Mutex<PlaybackBuffer>>) {
+    if let Ok(mut guard) = queue.lock() {
+        guard.samples.clear();
+        guard.started = false;
+    }
+}
+
+// Attenuate mic samples by `gain` (0.0..=1.0). Used to duck the mic while the
+// assistant is speaking so speaker leakage doesn't trip the server-side VAD,
+// while leaving real user speech loud enough to barge in.
+pub(super) fn duck_samples(samples: &[i16], gain: f32) -> Vec<i16> {
+    let gain = gain.clamp(0.0, 1.0);
+    samples
+        .iter()
+        .map(|s| (*s as f32 * gain).round().clamp(i16::MIN as f32, i16::MAX as f32) as i16)
+        .collect()
+}
+
 pub(super) fn i16_to_le_bytes(samples: &[i16]) -> Vec<u8> {
     let mut bytes = Vec::with_capacity(samples.len() * 2);
     for sample in samples {
@@ -282,3 +306,55 @@ impl FromF32 for u16 {
         ((value.clamp(-1.0, 1.0) * 0.5 + 0.5) * u16::MAX as f32) as u16
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn duck_samples_zero_gain_silences() {
+        let input = vec![1000_i16, -2000, 3000, -4000];
+        let out = duck_samples(&input, 0.0);
+        assert!(out.iter().all(|s| *s == 0));
+    }
+
+    #[test]
+    fn duck_samples_unity_gain_passthrough() {
+        let input = vec![1000_i16, -2000, 3000, -4000];
+        let out = duck_samples(&input, 1.0);
+        assert_eq!(out, input);
+    }
+
+    #[test]
+    fn duck_samples_half_gain_attenuates() {
+        let input = vec![10_000_i16, -10_000];
+        let out = duck_samples(&input, 0.5);
+        // Allow ±1 for f32 rounding.
+        assert!((out[0] - 5000).abs() <= 1, "got {}", out[0]);
+        assert!((out[1] + 5000).abs() <= 1, "got {}", out[1]);
+    }
+
+    #[test]
+    fn duck_samples_clamps_gain_above_one() {
+        // Negative gain or gain >1.0 must not panic or saturate weirdly.
+        let input = vec![10_000_i16];
+        let out_high = duck_samples(&input, 5.0);
+        assert_eq!(out_high, input, "gain >1.0 should clamp to 1.0");
+        let out_neg = duck_samples(&input, -1.0);
+        assert_eq!(out_neg, vec![0], "negative gain should clamp to 0.0");
+    }
+
+    #[test]
+    fn clear_playback_resets_buffer_and_start_flag() {
+        let buf = Arc::new(Mutex::new(PlaybackBuffer::new()));
+        {
+            let mut guard = buf.lock().unwrap();
+            guard.samples.extend([0.1_f32, 0.2, 0.3]);
+            guard.started = true;
+        }
+        clear_playback(&buf);
+        let guard = buf.lock().unwrap();
+        assert!(guard.samples.is_empty());
+        assert!(!guard.started);
+    }
+}

src/voice_loop/mod.rs

@@ -6,8 +6,8 @@ use crate::orchestrator::{
     OpenAiSummarizer, OrchestratorBridge, OrchestratorJobManager, OrchestratorProvider,
 };
 use audio::{
-    AudioChunk, PlaybackBuffer, enqueue_audio_delta, i16_to_le_bytes, playback_depth_ms,
-    resample_i16, start_input_stream, start_output_stream,
+    AudioChunk, PlaybackBuffer, clear_playback, duck_samples, enqueue_audio_delta, i16_to_le_bytes,
+    playback_depth_ms, resample_i16, start_input_stream, start_output_stream,
 };
 use settings::ResolvedVoiceSettings;
 use base64::Engine;
@@ -96,6 +96,16 @@ struct VoiceLoop<'a> {
 async fn run_voice_loop(mut state: VoiceLoop<'_>) -> Result<(), String> {
     let mut response_active = false;
     let mut deferred_response_creates = VecDeque::<serde_json::Value>::new();
+    // Barge-in bookkeeping. `current_item_id` is the id of the assistant
+    // message item that's currently being voiced; `enqueued_assistant_ms`
+    // is the total assistant audio (in wall-clock ms) we've handed to the
+    // local playback buffer since the response started. Both reset on
+    // response.done. On input_audio_buffer.speech_started while a response
+    // is active we use them to emit conversation.item.truncate with an
+    // accurate audio_end_ms.
+    let mut current_item_id: Option<String> = None;
+    let mut enqueued_assistant_ms: u64 = 0;
+    let mic_ducking_gain = state.voice_settings.mic_ducking_gain;
 
     let url = format!("wss://api.openai.com/v1/realtime?model={}", state.model);
     let mut request = url
@@ -127,14 +137,32 @@ async fn run_voice_loop(mut state: VoiceLoop<'_>) -> Result<(), String> {
     loop {
         tokio::select! {
             Some(chunk) = state.mic_rx.recv() => {
-                if response_active || playback_depth_ms(&state.playback, state.output_rate).unwrap_or(0) > 50 {
-                    continue;
-                }
                 let pcm24 = resample_i16(&chunk.samples, chunk.sample_rate, 24_000);
                 if pcm24.is_empty() {
                     continue;
                 }
-                let bytes = i16_to_le_bytes(&pcm24);
+                // While the assistant is speaking (or its audio is still
+                // draining out of the local playback buffer), duck the mic
+                // so speaker echo stays below the server VAD threshold, but
+                // keep streaming so a deliberate user utterance can still
+                // trigger input_audio_buffer.speech_started for barge-in.
+                let speaking = response_active
+                    || playback_depth_ms(&state.playback, state.output_rate).unwrap_or(0) > 50;
+                let pcm_to_send = if speaking {
+                    if mic_ducking_gain <= 0.0 {
+                        // Legacy behavior: fully gate the mic during
+                        // assistant speech (opt-in, disables barge-in).
+                        continue;
+                    }
+                    if mic_ducking_gain >= 1.0 {
+                        pcm24
+                    } else {
+                        duck_samples(&pcm24, mic_ducking_gain)
+                    }
+                } else {
+                    pcm24
+                };
+                let bytes = i16_to_le_bytes(&pcm_to_send);
                 let audio = base64::engine::general_purpose::STANDARD.encode(bytes);
                 let event = json!({
                     "type": "input_audio_buffer.append",
@@ -170,6 +198,8 @@ async fn run_voice_loop(mut state: VoiceLoop<'_>) -> Result<(), String> {
                     &mut state.orchestrator_bridge,
                     &state.orchestrator_jobs,
                     &mut response_active,
+                    &mut current_item_id,
+                    &mut enqueued_assistant_ms,
                     Arc::clone(&state.playback),
                     state.output_rate,
                 )
@@ -257,36 +287,89 @@ async fn handle_realtime_event(
     orchestrator_bridge: &mut OrchestratorBridge,
     orchestrator_jobs: &OrchestratorJobManager,
     response_active: &mut bool,
+    current_item_id: &mut Option<String>,
+    enqueued_assistant_ms: &mut u64,
     playback: Arc<Mutex<PlaybackBuffer>>,
     output_rate: u32,
 ) -> Result<Vec<serde_json::Value>, String> {
     let event_type = value.get("type").and_then(|v| v.as_str()).unwrap_or("");
+    let mut emitted: Vec<serde_json::Value> = Vec::new();
     match event_type {
         "session.created" | "session.updated" => {
             eprintln!("realtime event: {event_type}");
         }
         "response.created" => {
             *response_active = true;
+            *current_item_id = None;
+            *enqueued_assistant_ms = 0;
             eprintln!("realtime event: response.created");
         }
+        "response.output_item.added" => {
+            // Capture the assistant message item id so we can truncate it
+            // accurately on barge-in. Only the first audio-bearing item per
+            // response is tracked; subsequent items in the same response
+            // overwrite (the most recently active item is what would be
+            // mid-flight when the user speaks).
+            if let Some(id) = value
+                .get("item")
+                .and_then(|item| item.get("id"))
+                .and_then(|id| id.as_str())
+            {
+                *current_item_id = Some(id.to_string());
+            }
+        }
         "response.done" => {
             *response_active = false;
+            *current_item_id = None;
+            *enqueued_assistant_ms = 0;
             eprintln!(
                 "realtime event: response.done playback_depth_ms={}",
                 playback_depth_ms(&playback, output_rate).unwrap_or(0)
             );
         }
         "response.output_audio.delta" | "response.audio.delta" => {
             if let Some(delta) = value.get("delta").and_then(|v| v.as_str()) {
-                enqueue_audio_delta(delta, playback, output_rate)?;
+                let samples_24k = enqueue_audio_delta(delta, Arc::clone(&playback), output_rate)?;
+                // 24 kHz mono: 24 samples == 1 ms.
+                let delta_ms = (samples_24k / 24) as u64;
+                *enqueued_assistant_ms = enqueued_assistant_ms.saturating_add(delta_ms);
+            }
+        }
+        "input_audio_buffer.speech_started" => {
+            // Server-side VAD detected user speech. If the assistant is
+            // mid-response, cancel it, mark the played duration on the
+            // assistant item, and drop any audio still in the local buffer
+            // so the user hears their barge-in immediately.
+            if *response_active {
+                let remaining_ms = playback_depth_ms(&playback, output_rate).unwrap_or(0) as u64;
+                let played_ms = enqueued_assistant_ms.saturating_sub(remaining_ms);
+                eprintln!(
+                    "realtime barge-in: speech_started enqueued_ms={} remaining_ms={} played_ms={}",
+                    enqueued_assistant_ms, remaining_ms, played_ms
+                );
+                emitted.push(json!({ "type": "response.cancel" }));
+                if let Some(item_id) = current_item_id.take() {
+                    emitted.push(json!({
+                        "type": "conversation.item.truncate",
+                        "item_id": item_id,
+                        "content_index": 0,
+                        "audio_end_ms": played_ms,
+                    }));
+                }
+                clear_playback(&playback);
+                // Keep response_active=true until the server confirms the
+                // cancellation with response.done; that's our source of
+                // truth for "the assistant turn is over".
             }
         }
         "error" => {
             eprintln!("Realtime error: {}", value);
         }
         _ => {}
     }
-    orchestrator_bridge
+    let mut bridge_events = orchestrator_bridge
         .handle_realtime_event(&value, orchestrator_jobs)
-        .await
+        .await?;
+    emitted.append(&mut bridge_events);
+    Ok(emitted)
 }

src/voice_loop/session.rs

@@ -102,6 +102,7 @@ mod tests {
         let settings = ResolvedVoiceSettings {
             voice: "marin".to_string(),
             instructions: "base".to_string(),
+            mic_ducking_gain: 0.25,
         };
         let config = session_update_json_for("test-model", &settings);
         let tools = config["session"]["tools"]
@@ -132,6 +133,7 @@ mod tests {
         let settings = ResolvedVoiceSettings {
             voice: "cedar".to_string(),
             instructions: "custom instructions".to_string(),
+            mic_ducking_gain: 0.25,
         };
         let config = session_update_json_for("test-model", &settings);
         assert_eq!(