rtx-csm: verify barge-in end-to-end + slow-mock LLM for testing

Phase 6c.3c verification: extended converse_client with
--barge-in-after-ms flag that injects a 200 ms audio frame N ms after
the assistant starts speaking, then watches for the
{"event":"barge_in"} server response.

Verified end-to-end on Metal:
  Input: 10.43s LibriSpeech FLAC
  STT transcript: matched correctly
  Mock LLM streamed 4-sentence response with 50ms/token delays
  Client injected barge-in 100 ms into TTS streaming
  Server log: "barge-in detected (4800 samples carried over)"
  Client log: "[server] barge_in event received -- TTS cancelled"
  WAV file: 1.60s of TTS captured before cutoff

Mock LLM upgraded to multi-sentence with tokio::time::sleep(50ms)
between chunks — exercises the streaming pipeline long enough for
barge-in tests to fire mid-response.

The full Rust Unmute conversational stack is now feature-verified:
voice-in, voice-out, interruptible, VAD-driven, authed, metrics-
instrumented. Strategic Phase 6 deliverable shipped end-to-end.

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>

crates/models/rtx-csm/examples/converse_client.rs

@@ -35,6 +35,11 @@ struct Cli {
     /// RTX_AUTH_TOKEN env var if not provided.
     #[arg(long)]
     auth_token: Option<String>,
+    /// If set, after receiving N ms of assistant audio, inject a chunk
+    /// of new audio (the same input WAV's first 200 ms by default) to
+    /// simulate barge-in. Useful for testing 6c.3c.
+    #[arg(long)]
+    barge_in_after_ms: Option<u64>,
 }
 
 #[tokio::main]
@@ -86,6 +91,17 @@ async fn main() -> Result<()> {
     let mut response_pcm: Vec<f32> = Vec::new();
     let t = std::time::Instant::now();
     let mut first_audio_ms: Option<u128> = None;
+    let mut barge_in_fired = false;
+    // Pre-encode 200 ms of the input as the barge-in payload (it's just
+    // the first ~4800 samples, encoded as i16 LE bytes).
+    let barge_payload: Vec<u8> = samples
+        .iter()
+        .take((PCM_RATE as f32 * 0.2) as usize)
+        .flat_map(|s| {
+            let v = (s.clamp(-1.0, 1.0) * i16::MAX as f32) as i16;
+            v.to_le_bytes().to_vec()
+        })
+        .collect();
 
     while let Some(msg) = ws.next().await {
         let msg = msg?;
@@ -98,6 +114,21 @@ async fn main() -> Result<()> {
                     let s = i16::from_le_bytes([c[0], c[1]]);
                     response_pcm.push(s as f32 / i16::MAX as f32);
                 }
+                // Optional: inject barge-in once we've heard N ms of audio.
+                if let Some(after_ms) = cli.barge_in_after_ms {
+                    if !barge_in_fired
+                        && first_audio_ms
+                            .is_some_and(|t0| t.elapsed().as_millis() - t0 > after_ms as u128)
+                    {
+                        println!(
+                            "[barge-in] injecting {} bytes of audio after {} ms",
+                            barge_payload.len(),
+                            after_ms
+                        );
+                        ws.send(Message::Binary(barge_payload.clone().into())).await?;
+                        barge_in_fired = true;
+                    }
+                }
             }
             Message::Text(t) => {
                 let parsed: serde_json::Value =
@@ -119,6 +150,20 @@ async fn main() -> Result<()> {
                         println!("assistant: {assistant:?}");
                         break;
                     }
+                    "barge_in" => {
+                        println!("[server] barge_in event received — TTS cancelled");
+                        // For this test we just exit; a real client would
+                        // continue streaming the new turn's audio + EOT.
+                        if cli.barge_in_after_ms.is_some() {
+                            println!(
+                                "[barge-in test] success — server correctly detected barge-in"
+                            );
+                            break;
+                        }
+                    }
+                    "vad_eot" => {
+                        println!("[server] vad_eot event received — auto end-of-turn");
+                    }
                     "error" => {
                         let msg = parsed
                             .get("msg")

crates/models/rtx-csm/examples/converse_server.rs

@@ -45,7 +45,7 @@ use axum::{
     Router,
 };
 use clap::Parser;
-use futures_util::stream;
+use futures_util::{stream, StreamExt};
 use rtx_csm::{
     converse::{Converse, ConverseOptions, FlushPolicy},
     error::Result as CsmResult,
@@ -113,9 +113,10 @@ struct Cli {
     vad_consecutive: u32,
 }
 
-/// Echo-style mock LLM. Reads the most recent user message from the
+/// Multi-sentence mock LLM. Reads the most recent user message and
-/// chat history and returns a fixed acknowledgment + repeats the heard
+/// returns a fixed multi-sentence acknowledgment with a small delay
-/// transcript. Useful for end-to-end testing without an API key.
+/// between sentence boundaries so the streaming pipeline runs long
+/// enough to be barge-in-testable.
 struct MockLlm;
 
 #[async_trait]
@@ -136,13 +137,22 @@ impl LlmClient for MockLlm {
                 }
             })
             .unwrap_or_default();
-        let response = format!("I heard you say: {}.", user_text);
+        let response = format!(
-        // Emit it as a small number of chunks to exercise streaming.
+            "I heard you. You said: {}. That is interesting. Tell me more about it.",
-        let chunks: Vec<CsmResult<String>> = response
+            user_text
+        );
+        // Stream tokens one whitespace-split chunk at a time, with a small
+        // sleep between chunks so the pipeline takes a few seconds end to
+        // end (lets barge-in tests fire mid-stream).
+        let chunks: Vec<String> = response
             .split_inclusive(' ')
-            .map(|s| Ok(s.to_string()))
+            .map(|s| s.to_string())
             .collect();
-        Ok(Box::pin(stream::iter(chunks)))
+        let s = stream::iter(chunks).then(|s| async move {
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+            Ok::<_, rtx_csm::CsmError>(s)
+        });
+        Ok(Box::pin(s))
     }
 }