rtx-csm: Phase 6c.1 — text -> LLM -> sentence buffer -> CSM TTS

Composable orchestrator stitching the LLM-output side of the
conversational stack:

  prompt + history -> LlmClient.generate_stream -> sentence buffer
    -> per-sentence Generator.generate -> post-process -> watermark
    -> Vec<Utterance> stream of (text, audio, latency)

CSM is sentence-level (best prosody on full sentences), so the buffer
flushes when terminal punctuation appears anywhere in the buffer
(Punctuation policy: . ! ? \n) or at any clause boundary
(Eager policy: + , ; :).

src/converse.rs:
- Converse<L: LlmClient> orchestrator
- Utterance { text, audio, tts_latency_ms } per sentence
- FlushPolicy { Punctuation, Eager }
- find_first_boundary scans the whole buffer (not just the last char)
  so "Sentence one. Word two" emits "Sentence one." immediately rather
  than waiting for the next terminal mark
- 3 unit tests for boundary detection + policy modes

examples/converse.rs:
- --mock mode: hardcoded 20-token "sleepy turtle" stream, no API key
- live mode: any OpenAI-compatible endpoint via OpenAiCompatibleClient
- writes the concatenated audio to a single WAV

Verified mock end-to-end on Metal: 2 utterances emitted as expected
(sentence 1 hits max_audio_ms cap at 6s; sentence 2 EOTs naturally at
4.88s), total 10.88s of audio in 31.5s wall-clock.

Phase 6c.1 ships the half-duplex (text-in -> voice-out) pipeline. Full
duplex (audio-in -> voice-out) is 6c.2, blocked on 6a's STT word
emission landing.

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

crates/models/rtx-csm/Cargo.toml

@@ -193,3 +193,7 @@ path = "examples/stt_demo.rs"
 [[example]]
 name = "llm_chat"
 path = "examples/llm_chat.rs"
+
+[[example]]
+name = "converse"
+path = "examples/converse.rs"

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

@@ -0,0 +1,196 @@
+//! End-to-end LLM → CSM TTS pipeline demo.
+//!
+//! Streams an LLM response token-by-token, flushes each completed
+//! sentence to CSM, writes the concatenated audio to a single WAV.
+//!
+//! Two modes:
+//!   --mock          : use a hardcoded mock LLM (no API key needed; useful
+//!                     for local end-to-end testing)
+//!   --base/--model  : use a real OpenAI-compatible endpoint
+//!
+//! Usage (mock):
+//! ```
+//! cargo run -p rtx-csm --release --features metal --example converse -- \
+//!     --mock --speaker 0 --out /tmp/converse_out.wav
+//! ```
+//!
+//! Usage (live LLM):
+//! ```
+//! export OPENAI_API_KEY=sk-...
+//! cargo run -p rtx-csm --release --features metal --example converse -- \
+//!     --base "https://api.openai.com/v1" --model gpt-4o-mini \
+//!     --prompt "Tell me a two-sentence story about a sleepy turtle." \
+//!     --out /tmp/converse_out.wav
+//! ```
+
+use anyhow::{Context, Result};
+use async_trait::async_trait;
+use clap::Parser;
+use futures_util::{stream, Stream};
+use rtx_csm::{
+    audio_io,
+    converse::{Converse, ConverseOptions, FlushPolicy, Utterance},
+    llm_client::{ChatMessage, GenConfig, LlmClient, OpenAiCompatibleClient, TokenStream},
+    Generator,
+};
+use rtx_csm::error::Result as CsmResult;
+use std::path::PathBuf;
+use std::pin::Pin;
+
+#[derive(Debug, Parser)]
+#[command(name = "converse")]
+struct Cli {
+    /// Use the hardcoded mock LLM instead of a real API.
+    #[arg(long)]
+    mock: bool,
+    /// LLM base URL (e.g. https://api.openai.com/v1).
+    #[arg(long, default_value = "https://api.openai.com/v1")]
+    base: String,
+    #[arg(long, default_value = "gpt-4o-mini")]
+    model: String,
+    #[arg(long)]
+    api_key: Option<String>,
+    #[arg(long, default_value = "Tell me a two-sentence story about a sleepy turtle.")]
+    prompt: String,
+    #[arg(long, default_value = "You are a concise storyteller.")]
+    system: String,
+    #[arg(long, default_value_t = 0)]
+    speaker: u32,
+    #[arg(long, default_value_t = 0.7)]
+    temperature: f32,
+    #[arg(long, default_value_t = 256)]
+    max_tokens: u32,
+    /// Eager flush (every comma) for lower first-audio latency.
+    #[arg(long)]
+    eager: bool,
+    /// Output WAV path (concatenated full assistant response).
+    #[arg(long)]
+    out: PathBuf,
+    #[arg(long)]
+    cpu: bool,
+}
+
+/// Mock LLM: emits a fixed response chunked into small token-like pieces
+/// so the streaming pipeline is exercised without a network call.
+struct MockLlm {
+    chunks: Vec<&'static str>,
+}
+
+impl MockLlm {
+    fn new() -> Self {
+        Self {
+            chunks: vec![
+                "Once ", "upon ", "a ", "time ", "there ", "was ", "a ", "very ", "sleepy ",
+                "turtle. ", "She ", "yawned ", "loudly ", "and ", "fell ", "asleep ", "in ",
+                "the ", "warm ", "sun.",
+            ],
+        }
+    }
+}
+
+#[async_trait]
+impl LlmClient for MockLlm {
+    async fn generate_stream(
+        &self,
+        _messages: Vec<ChatMessage>,
+        _config: GenConfig,
+    ) -> CsmResult<TokenStream> {
+        let chunks: Vec<CsmResult<String>> =
+            self.chunks.iter().map(|s| Ok(s.to_string())).collect();
+        let s: Pin<Box<dyn Stream<Item = CsmResult<String>> + Send>> =
+            Box::pin(stream::iter(chunks));
+        Ok(s)
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::fmt().init();
+    let cli = Cli::parse();
+
+    let device = if cli.cpu {
+        candle_core::Device::Cpu
+    } else {
+        Generator::default_device()?
+    };
+    println!("device: {device:?}");
+    let mut generator = Generator::load_csm_1b(&device)?;
+    println!(
+        "loaded CSM-1B (sr={} Hz)",
+        generator.config.sample_rate
+    );
+
+    let messages = vec![
+        ChatMessage::system(&cli.system),
+        ChatMessage::user(&cli.prompt),
+    ];
+    let gen_cfg = GenConfig {
+        max_tokens: Some(cli.max_tokens),
+        temperature: cli.temperature,
+        ..GenConfig::default()
+    };
+    let opts = ConverseOptions {
+        speaker: cli.speaker,
+        flush: if cli.eager { FlushPolicy::Eager } else { FlushPolicy::Punctuation },
+        ..ConverseOptions::default()
+    };
+
+    let mut all: Vec<Utterance> = Vec::new();
+    let t = std::time::Instant::now();
+    let full_text = if cli.mock {
+        let llm = MockLlm::new();
+        let mut conv = Converse::new(&llm, &mut generator);
+        conv.run(messages, gen_cfg, opts, |u| {
+            println!(
+                "  utt ({} samples, tts {}ms): {:?}",
+                u.audio.len(),
+                u.tts_latency_ms,
+                u.text
+            );
+            all.push(u.clone());
+            Ok(())
+        })
+        .await?
+    } else {
+        let api_key = cli
+            .api_key
+            .or_else(|| std::env::var("OPENAI_API_KEY").ok())
+            .ok_or_else(|| anyhow::anyhow!("set --api-key or OPENAI_API_KEY (or use --mock)"))?;
+        let llm = OpenAiCompatibleClient::new(&cli.base, api_key, &cli.model);
+        let mut conv = Converse::new(&llm, &mut generator);
+        conv.run(messages, gen_cfg, opts, |u| {
+            println!(
+                "  utt ({} samples, tts {}ms): {:?}",
+                u.audio.len(),
+                u.tts_latency_ms,
+                u.text
+            );
+            all.push(u.clone());
+            Ok(())
+        })
+        .await?
+    };
+
+    println!(
+        "\n== full assistant response ({} chars, {} sentences) ==",
+        full_text.len(),
+        all.len()
+    );
+    println!("{full_text}");
+
+    // Concatenate and write.
+    let mut concat: Vec<f32> = Vec::new();
+    for u in &all {
+        concat.extend_from_slice(&u.audio);
+    }
+    audio_io::write_wav_24k_mono(&cli.out, &concat).context("write wav")?;
+    println!(
+        "\nwrote {} ({} samples = {:.2}s @ 24 kHz, total {:.2}s elapsed)",
+        cli.out.display(),
+        concat.len(),
+        concat.len() as f32 / 24000.0,
+        t.elapsed().as_secs_f32()
+    );
+
+    Ok(())
+}

crates/models/rtx-csm/src/converse.rs

@@ -0,0 +1,254 @@
+//! Composable orchestrator for the LLM → TTS half of the conversational
+//! stack. Takes a prompt + chat history, streams LLM tokens, buffers them
+//! into sentences, and dispatches each completed sentence to CSM TTS as
+//! soon as it's ready. PCM is emitted as the model produces it.
+//!
+//! ## Why sentence buffering?
+//!
+//! CSM is a sentence-level TTS — its prosody is best when fed a complete
+//! sentence at a time, not token-by-token. The LLM streams tokens; we
+//! accumulate until a terminal punctuation mark (`.`, `!`, `?`) or newline,
+//! then flush the sentence to TTS. This trades a small chunk of buffering
+//! latency (typically <1s for short sentences) for natural prosody.
+//!
+//! ## Pipeline
+//!
+//! ```text
+//!   prompt + history
+//!     │
+//!     ▼
+//!   LlmClient.generate_stream → token chunks (k1, k2, ...)
+//!     │
+//!     ▼ buffer until terminal punctuation
+//!   Sentence "Hello."
+//!     │
+//!     ▼
+//!   Generator.generate(text) → 24 kHz PCM (Vec<f32>)
+//!     │
+//!     ▼ on_audio callback fires once per sentence
+//!   caller (e.g. WebSocket sink, file writer)
+//! ```
+//!
+//! When the next implementation lands a streaming TTS variant that takes
+//! token-level input directly (no sentence boundary needed), the
+//! orchestrator can be swapped to that path. Today's CSM is the bottleneck.
+//!
+//! ## Half-duplex; full duplex (6c.2) is deferred
+//!
+//! This orchestrator handles the LLM → TTS direction only. To get full
+//! voice-in/voice-out, pair this with the Phase 6a STT once its word
+//! emission is fixed: `audio_in → STT → user_text → Converse → audio_out`.
+
+use crate::audio_io;
+use crate::error::{CsmError, Result};
+use crate::generator::Generator;
+use crate::llm_client::{ChatMessage, GenConfig, LlmClient};
+use crate::post::PostProcess;
+use crate::GenerateOptions;
+use futures_util::StreamExt;
+use std::path::Path;
+
+/// One unit emitted by [`Converse::run`] each time a sentence completes.
+#[derive(Debug, Clone)]
+pub struct Utterance {
+    pub text: String,
+    /// 24 kHz mono PCM, post-processed and (if a watermarker is installed
+    /// on the Generator) watermarked.
+    pub audio: Vec<f32>,
+    /// Wall-clock time-to-first-audio for THIS sentence, in milliseconds.
+    /// (Time from sentence-buffer flush to PCM produced.)
+    pub tts_latency_ms: u128,
+}
+
+/// How aggressively to flush sentences. `Punctuation` waits for `. ! ?`
+/// or newline; `Eager` flushes more often for lower latency at the cost
+/// of less natural prosody.
+#[derive(Debug, Clone, Copy)]
+pub enum FlushPolicy {
+    Punctuation,
+    /// Flush at every comma, semicolon, or punctuation mark — useful for
+    /// long-running monologues where you want low first-audio latency.
+    Eager,
+}
+
+#[derive(Debug, Clone)]
+pub struct ConverseOptions {
+    pub generate: GenerateOptions,
+    pub speaker: u32,
+    pub flush: FlushPolicy,
+    /// Skip sentences shorter than this (after trim). Avoids dispatching
+    /// tiny "k." or single-character fragments to TTS.
+    pub min_sentence_chars: usize,
+}
+
+impl Default for ConverseOptions {
+    fn default() -> Self {
+        Self {
+            generate: GenerateOptions {
+                max_audio_ms: 6_000,
+                ..GenerateOptions::default()
+            },
+            speaker: 0,
+            flush: FlushPolicy::Punctuation,
+            min_sentence_chars: 2,
+        }
+    }
+}
+
+/// Locate the byte index AFTER the first sentence-boundary character in
+/// the buffer per policy. Returns `Some(end)` so the caller can do
+/// `buf.drain(..end)` to extract the completed sentence (including the
+/// punctuation mark and any trailing whitespace up to the boundary).
+/// Returns `None` if no boundary is present yet.
+fn find_first_boundary(buf: &str, policy: FlushPolicy) -> Option<usize> {
+    let is_boundary = |c: char| match policy {
+        FlushPolicy::Punctuation => matches!(c, '.' | '!' | '?' | '\n'),
+        FlushPolicy::Eager => matches!(c, '.' | '!' | '?' | '\n' | ',' | ';' | ':'),
+    };
+    for (i, c) in buf.char_indices() {
+        if is_boundary(c) {
+            return Some(i + c.len_utf8());
+        }
+    }
+    None
+}
+
+/// Buffer ends in a "flushable" boundary (used by the trailing-text
+/// fallback at end-of-stream). For the live streaming path we use
+/// [`find_first_boundary`] to handle "turtle. She" style mid-buffer cases.
+fn should_flush(buf: &str, policy: FlushPolicy) -> bool {
+    find_first_boundary(buf, policy).is_some()
+}
+
+pub struct Converse<'a, L: LlmClient> {
+    llm: &'a L,
+    generator: &'a mut Generator,
+    post: PostProcess,
+}
+
+impl<'a, L: LlmClient> Converse<'a, L> {
+    pub fn new(llm: &'a L, generator: &'a mut Generator) -> Self {
+        Self {
+            llm,
+            generator,
+            post: PostProcess::default(),
+        }
+    }
+
+    /// Override post-processing (HPF + declick + LUFS). Pass
+    /// [`PostProcess::disabled`] to skip.
+    pub fn with_post(mut self, post: PostProcess) -> Self {
+        self.post = post;
+        self
+    }
+
+    /// Run the full pipeline: stream LLM tokens, flush sentences to TTS,
+    /// invoke `on_utterance` for each completed (text, audio) pair as
+    /// they're produced. Returns the full assistant message text once
+    /// the LLM stream ends.
+    pub async fn run<F>(
+        &mut self,
+        messages: Vec<ChatMessage>,
+        gen_cfg: GenConfig,
+        opts: ConverseOptions,
+        mut on_utterance: F,
+    ) -> Result<String>
+    where
+        F: FnMut(&Utterance) -> Result<()>,
+    {
+        let mut stream = self.llm.generate_stream(messages, gen_cfg).await?;
+        let mut buf = String::new();
+        let mut full = String::new();
+
+        while let Some(chunk) = stream.next().await {
+            let chunk = chunk?;
+            full.push_str(&chunk);
+            buf.push_str(&chunk);
+            // Flush as many complete sentences as the buffer contains.
+            while let Some(end) = find_first_boundary(&buf, opts.flush) {
+                let sentence: String = buf.drain(..end).collect();
+                let trimmed = sentence.trim();
+                if trimmed.chars().count() < opts.min_sentence_chars {
+                    continue;
+                }
+                let utt = self.synthesize(trimmed, &opts)?;
+                on_utterance(&utt)?;
+            }
+        }
+
+        // Trailing text without terminal punctuation — flush as one
+        // final sentence so callers don't lose the tail.
+        let trailing = buf.trim().to_string();
+        if trailing.chars().count() >= opts.min_sentence_chars {
+            let utt = self.synthesize(&trailing, &opts)?;
+            on_utterance(&utt)?;
+        }
+        Ok(full)
+    }
+
+    fn synthesize(&mut self, sentence: &str, opts: &ConverseOptions) -> Result<Utterance> {
+        let t = std::time::Instant::now();
+        let mut pcm = self
+            .generator
+            .generate(sentence, opts.speaker, &[], opts.generate)
+            .map_err(|e| CsmError::Config(format!("converse generate: {e}")))?;
+        self.post
+            .apply(&mut pcm, self.generator.config.sample_rate)
+            .map_err(|e| CsmError::Config(format!("converse post: {e}")))?;
+        if let Some(wm) = self.generator.watermarker.as_ref() {
+            pcm = wm.embed(&pcm)?;
+        }
+        Ok(Utterance {
+            text: sentence.to_string(),
+            audio: pcm,
+            tts_latency_ms: t.elapsed().as_millis(),
+        })
+    }
+}
+
+/// Convenience: write all utterances concatenated into a single 24 kHz WAV.
+pub fn write_concatenated_wav<I>(utterances: I, out: &Path) -> Result<()>
+where
+    I: IntoIterator<Item = Utterance>,
+{
+    let mut all: Vec<f32> = Vec::new();
+    for u in utterances {
+        all.extend(u.audio);
+    }
+    audio_io::write_wav_24k_mono(out, &all)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn flush_punctuation_modes() {
+        assert!(should_flush("Hello.", FlushPolicy::Punctuation));
+        assert!(should_flush("Hello. World", FlushPolicy::Punctuation));
+        assert!(should_flush("Wait!", FlushPolicy::Punctuation));
+        assert!(should_flush("What?", FlushPolicy::Punctuation));
+        assert!(should_flush("Line\n", FlushPolicy::Punctuation));
+        assert!(!should_flush("Hello", FlushPolicy::Punctuation));
+        assert!(!should_flush("Hello, world", FlushPolicy::Punctuation));
+        assert!(should_flush("Hello,", FlushPolicy::Eager));
+        assert!(should_flush("Hello, world", FlushPolicy::Eager));
+        assert!(should_flush("then;", FlushPolicy::Eager));
+        assert!(!should_flush("Hello world", FlushPolicy::Eager));
+    }
+
+    #[test]
+    fn flush_empty_buffer() {
+        assert!(!should_flush("", FlushPolicy::Punctuation));
+        assert!(!should_flush("", FlushPolicy::Eager));
+    }
+
+    #[test]
+    fn find_boundary_returns_index_after_punctuation() {
+        // "Hello. World" → boundary at index 6 (after the .).
+        assert_eq!(find_first_boundary("Hello. World", FlushPolicy::Punctuation), Some(6));
+        assert_eq!(find_first_boundary("Hello", FlushPolicy::Punctuation), None);
+        // Eager: comma at index 5, returns 6.
+        assert_eq!(find_first_boundary("Hello, world", FlushPolicy::Eager), Some(6));
+    }
+}

crates/models/rtx-csm/src/lib.rs

@@ -8,6 +8,7 @@ pub mod audio_io;
 pub mod audioseal;
 pub mod audioseal_convert;
 pub mod config;
+pub mod converse;
 pub mod csm_fork;
 pub mod csm_quantized;
 pub mod error;