Local voice assistant on top of Canary, llama.cpp and qwentts

Pipeline en Rust de hilos y canales que une los tres motores: el micrófono
alimenta un segmentador con VAD, las intervenciones cerradas van al
reconocedor, la transcripción al modelo y cada frase que este cierra sale
hacia el sintetizador sin esperar al resto de la respuesta.

Dos reglas sostienen el diseño: ninguna etapa bloquea a la anterior —quien va
sobrado descarta trabajo en lugar de acumular retraso— y todo lo que viaja por
los canales lleva el turno al que pertenece, así que interrumpir es subir el
contador y levantar dos banderas de cancelación.

Seis crates: core (configuración, eventos, HTTP, telemetría, herramientas),
audio (cpal, VAD, anillo de reproducción), asr, llm, tts y app (supervisor de
procesos y orquestador). Los motores van como submódulos fijados a un commit,
con los cambios locales en vendor/patches.

Midiendo el pipeline aparecieron tres cuellos de botella de configuración que
valieron más que cualquier cambio de código, todos documentados en
docs/RENDIMIENTO.md:

- tts-server decodificaba el audio en bloques de 24 s, de modo que el modo
  «streaming» llegaba de una pieza: 4948 ms -> 585 ms hasta el primer audio.
- La plantilla de chat del modelo abre <think> y no lo cierra nunca, sin
  variable que lo apague: 8630 ms -> 413 ms hasta el primer token, con una
  copia de la plantilla que deja el bloque cerrado de entrada.
- Cualquier indicación de estilo junto a la guía de herramientas hace que este
  modelo de 2B deje de llamarlas y se invente el dato (8/8 aciertos con la
  guía sola, 0/8 con la persona de asistente de voz). El turno alterna ahora
  entre dos instrucciones de sistema.

La ejecución de órdenes del sistema queda implementada y apagada, tras cuatro
barreras: lista blanca sobre el ejecutable, rutas rechazadas, sin shell que
interprete metacaracteres y plazo máximo.

68 pruebas unitarias sin modelos, más seis de integración que se saltan solas
si no hay servidores y se turnan la GPU: en paralelo, los dos servidores no
caben en 4 GB y miden contención en vez de latencia.

Claude-Session: https://claude.ai/code/session_01FNxz5cSdQSscJH9H7b8uGU

1 files changed, 1767 insertions, 0 deletions

diff --git a/vendor/patches/canary-rs.patch b/vendor/patches/canary-rs.patch
new file mode 100644
index 0000000..c579ee0
--- /dev/null
+++ b/vendor/patches/canary-rs.patch
@@ -0,0 +1,1767 @@
+diff --git a/Cargo.toml b/Cargo.toml
+index 575a037..65469d9 100644
+--- a/Cargo.toml
++++ b/Cargo.toml
+@@ -41,6 +41,10 @@ path = "examples/inspect_model.rs"
+ name = "decoder_smoke"
+ path = "examples/decoder_smoke.rs"
+ 
++[[example]]
++name = "bench_live"
++path = "examples/bench_live.rs"
++
+ [features]
+ default = ["cpu", "ort-defaults"]
+ cpu = []
+diff --git a/README.md b/README.md
+index ea711f1..34ecfc5 100644
+--- a/README.md
++++ b/README.md
+@@ -41,6 +41,21 @@ let mut stream = model.stream("en", "en", stream_cfg)?;
+ // stream.push_samples(&audio_chunk, sample_rate, channels)?;
+ ```
+ 
++### Streaming
++
++`CanaryStream` decodes a rolling window of audio, and its defaults are tuned for live input where
++decoding is slower than real time:
++
++- `tail_silence_duration` (0.4 s) pads the end of every window. Canary was trained on complete
++  utterances and very often emits `<|endoftext|>` immediately — returning *no text at all* — for
++  audio that stops mid-word, which is what a live window always looks like.
++- `max_windows_per_push` (1) decodes only the newest window and drops the backlog, so latency
++  stays bounded by one decode instead of growing without limit. Set it to `0` when replaying
++  recorded audio, where every window matters and wall-clock does not.
++- `min_commit_duration` (2.0 s) keeps the text of very short windows out of `delta_text`. The
++  first fractions of a second of an utterance decode poorly, and a committed transcript cannot be
++  taken back. Their text is still available in `StreamChunk::result` for display.
++
+ **Note**: When using `canary-180m-flash` don't enable the `use_itn` option in `SessionConfig`, as this model doesn't seem to be trained with inverse text normalization and enabling it causes empty output.
+ 
+ ## Features
+@@ -55,13 +70,54 @@ See the `examples` directory for more usage examples, including live microphone
+ 
+ ```bash
+ # Defaults to CPU as execution provider
+-cargo run --example live
++cargo run --release --example live -- en en
+ ```
+ 
++The `live` example runs a threaded pipeline — audio callback, capture/VAD, decode worker(s) and
++rendering each on their own thread, connected by channels — so a slow decode never blocks capture
++and stale windows are dropped rather than queued. It shows the stabilized transcript plus a dimmed,
++still-changing tail, and re-decodes each finished utterance for a clean final line. Tuning knobs:
++
++| Variable | Default | Meaning |
++| --- | --- | --- |
++| `CANARY_LIVE_WINDOW` | `6.0` | Rolling window in seconds; longer means more context, fewer updates |
++| `CANARY_LIVE_STEP` | `0.4` | How far the window advances between decodes |
++| `CANARY_LIVE_STABILITY` | `2` | Windows that must agree before a word is committed |
++| `CANARY_LIVE_TAIL_SILENCE` | `0.4` | Silence appended to each decoded window |
++| `CANARY_LIVE_SILENCE_HOLD` | `0.8` | Silence that ends an utterance |
++| `CANARY_LIVE_MIN_UTTERANCE` | `0.3` | Shortest utterance worth a final decode |
++| `CANARY_LIVE_MAX_UTTERANCE` | `20.0` | Longest utterance before one is forced |
++| `CANARY_LIVE_FINAL_MODEL` | `0` | Load a second model instance so final decodes run in parallel |
++| `CANARY_LIVE_STATS` | `0` | Print decode time, lag and dropped windows to stderr |
++| `CANARY_LIVE_WIDTH` | `100` | Characters of the live line to keep on screen |
++
++`CANARY_INTER_THREADS` / `CANARY_INTRA_THREADS` set the ONNX Runtime thread pools for every
++example.
++
+ You can provide an environment variable called `CANARY_EXECUTION_PROVIDER` with the matching feature flag to select an execution provider. For example, to use CUDA:
+ 
+ ```bash
+-CANARY_EXECUTION_PROVIDER=cuda CANARY_CUDA_DEVICE_ID=1 cargo run --example transcribe --features cuda
++CANARY_EXECUTION_PROVIDER=cuda CANARY_CUDA_DEVICE_ID=0 cargo run --example transcribe --features cuda
++```
++
++By default, a requested execution provider that cannot be loaded is a hard error rather than a
++silent fall back to CPU — ONNX Runtime's own behaviour, which otherwise looks like a working GPU
++session running an order of magnitude slower than expected. The CUDA provider needs cuDNN 9
++alongside the CUDA runtime; without it you will see
++
++```
++Failed to load library libonnxruntime_providers_cuda.so with error: libcudnn.so.9: cannot open shared object file
++```
++
++Set `CANARY_REQUIRE_EP=0` (or `ExecutionConfig::with_require_execution_provider(false)`) to opt
++back into the silent fallback.
++
++Note that the published Canary ONNX exports are int8-quantized, and the CUDA provider has limited
++coverage of int8 operators — on a modest GPU it can be no faster than a multi-core CPU. Measure
++with the `bench_live` example before assuming the GPU wins:
++
++```bash
++cargo run --release --example bench_live -- audio.wav en
+ ```
+ 
+ Or to use CoreML with Neural Engine + low precision:
+diff --git a/examples/live.rs b/examples/live.rs
+index 46d3ad8..a60cfdf 100644
+--- a/examples/live.rs
++++ b/examples/live.rs
+@@ -1,248 +1,505 @@
+ #[path = "shared/utils.rs"]
+ mod utils;
+ 
+-use canary_rs::{Canary, StreamConfig};
+-use cpal::Sample;
++use canary_rs::{Canary, CanarySession, StreamConfig};
+ use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
++use cpal::{Sample, SampleFormat, SupportedStreamConfig};
+ use std::io::Write;
+ use std::sync::atomic::{AtomicBool, Ordering};
+-use std::sync::{Arc, Mutex, mpsc};
+-use std::time::Duration;
+-
+-// Live streaming example:
+-// - Captures microphone audio via cpal.
+-// - Runs a rolling window decode for low-latency updates.
+-// - Tracks a simple noise floor to avoid decoding silence.
+-// - On detected utterance end, re-decodes the full utterance for a final result.
++use std::sync::mpsc::{self, Receiver, Sender};
++use std::sync::Arc;
++use std::thread::{self, JoinHandle};
++use std::time::{Duration, Instant};
++
++// Live streaming transcription.
++//
++// Decoding a window is slower than recording it — on a laptop CPU a 6 s window costs roughly
++// 800 ms while it only advances 0.4 s of audio — so anything that decodes every window in order
++// falls behind the speaker without bound. The pipeline below is built around that fact: each
++// stage runs on its own thread, connected by channels, and the decode stage is allowed to throw
++// work away.
++//
++//   cpal callback ─samples─> capture thread ─jobs─> decode worker(s) ─updates─> render (main)
++//        realtime            resample + VAD          Canary sessions           terminal
++//
++// * The capture thread never blocks the audio callback: it owns resampling, the noise-floor VAD
++//   and the utterance buffer.
++// * A decode worker drains its whole queue before running the model, keeps only the newest
++//   rolling window and drops the backlog, so latency is bounded by one decode instead of growing.
++// * The end of an utterance re-decodes the whole thing for a clean final line. That decode is
++//   long, so it can optionally run on its own thread against a second model instance
++//   (`CANARY_LIVE_FINAL_MODEL=1`) rather than stalling the partials.
++
++/// Sample rate the model works at; the input stream is opened here directly when the device
++/// allows it, which removes resampling from the hot path entirely.
++const MODEL_SAMPLE_RATE: u32 = 16_000;
++
++/// How much audio the VAD looks at per decision.
++const VAD_FRAME_SECONDS: f32 = 0.1;
++/// Audio kept before speech is detected, so an utterance does not start clipped.
++const PREROLL_SECONDS: f32 = 0.3;
++
++const MIN_SILENCE_THRESHOLD: f32 = 0.0008;
++const MAX_SILENCE_THRESHOLD: f32 = 0.02;
++
+ fn main() -> Result<(), Box<dyn std::error::Error>> {
+-    // Use en -> en or the first and second command line arguments as source and target language codes
+-    let source_lang = match std::env::args().nth(1) {
+-        Some(lang) => lang,
+-        None => "en".to_string(),
+-    };
+-    let target_lang = match std::env::args().nth(2) {
+-        Some(lang) => lang,
+-        None => source_lang.clone(),
+-    };
++    let opts = Options::from_env();
+     println!(
+         "Source language: {}, Target language: {}",
+-        source_lang, target_lang
++        opts.source_lang, opts.target_lang
+     );
+ 
+     println!("Loading Canary model...");
+-
+     let config = utils::execution_config_from_env();
+     let model_dir =
+         std::env::var("CANARY_MODEL_DIR").unwrap_or_else(|_| "canary-180m-flash".to_string());
+-    let model = Canary::from_pretrained(&model_dir, Some(config))?;
+-    let stream_cfg = StreamConfig::new()
+-        .with_window_duration(8.0)
+-        .with_step_duration(0.5)
+-        .with_emit_partial(true)
+-        .with_pad_partial(false)
+-        .with_stability_window(3);
+-    let mut stream_state =
+-        model.stream(source_lang.clone(), target_lang.clone(), stream_cfg.clone())?;
+-    let mut full_session = model.session();
+-
++    let model = Canary::from_pretrained(&model_dir, Some(config.clone()))?;
++
++    // A second instance means finals and partials decode in parallel: sessions cloned from one
++    // `Canary` share its ONNX sessions behind a mutex, so they would serialize instead.
++    let final_model = if opts.dedicated_final_model {
++        println!("Loading a second model instance for final decodes...");
++        Some(Canary::from_pretrained(&model_dir, Some(config))?)
++    } else {
++        None
++    };
+     println!("Model loaded successfully!");
+-
+-    let host = cpal::default_host();
+-    let device = host
+-        .default_input_device()
+-        .ok_or("No input device available")?;
+-    let supported_config = device.default_input_config()?;
+-    let sample_rate = supported_config.sample_rate() as usize;
+-    let channels = supported_config.channels() as usize;
+-    let cpal_config: cpal::StreamConfig = supported_config.clone().into();
+-
+-    // Feed chunks at roughly the streaming step size to avoid multi-second output bursts.
+-    let chunk_seconds = stream_cfg.step_duration.max(0.05);
+-    let chunk_samples = ((sample_rate as f32 * chunk_seconds).round() as usize).max(1) * channels;
+-    let min_silence_threshold = 0.0008_f32;
+-    let max_silence_threshold = 0.02_f32;
+-    let silence_hold_seconds = 0.8_f32;
+-    let silence_hold_samples =
+-        ((sample_rate as f32 * silence_hold_seconds).round() as usize).max(1) * channels;
+-    let min_utterance_seconds = 0.3_f32;
+-    let min_utterance_samples =
+-        ((sample_rate as f32 * min_utterance_seconds).round() as usize).max(1) * channels;
+-
+-    let buffer = Arc::new(Mutex::new(Vec::<f32>::with_capacity(chunk_samples * 2)));
+-    let (tx, rx) = mpsc::channel::<Vec<f32>>();
+-
+-    let err_fn = |err| eprintln!("stream error: {}", err);
+-
+-    let buffer_clone = Arc::clone(&buffer);
+-    let tx_clone = tx.clone();
+-    let stream = match supported_config.sample_format() {
+-        cpal::SampleFormat::F32 => device.build_input_stream(
+-            &cpal_config,
+-            move |data: &[f32], _| {
+-                push_input_data(data, chunk_samples, &buffer_clone, &tx_clone);
+-            },
+-            err_fn,
+-            None,
+-        )?,
+-        cpal::SampleFormat::I16 => device.build_input_stream(
+-            &cpal_config,
+-            move |data: &[i16], _| {
+-                push_input_data(data, chunk_samples, &buffer_clone, &tx_clone);
+-            },
+-            err_fn,
+-            None,
+-        )?,
+-        cpal::SampleFormat::U16 => device.build_input_stream(
+-            &cpal_config,
+-            move |data: &[u16], _| {
+-                push_input_data(data, chunk_samples, &buffer_clone, &tx_clone);
+-            },
+-            err_fn,
+-            None,
+-        )?,
+-        sample_format => {
+-            return Err(format!("Unsupported sample format: {:?}", sample_format).into());
++    let (job_tx, job_rx) = mpsc::channel::<Job>();
++    let (update_tx, update_rx) = mpsc::channel::<Update>();
++
++    let mut workers: Vec<JoinHandle<()>> = Vec::new();
++    let final_tx = match final_model {
++        Some(final_model) => {
++            let (final_tx, final_rx) = mpsc::channel::<Job>();
++            let opts = opts.clone();
++            let updates = update_tx.clone();
++            workers.push(thread::spawn(move || {
++                run_final_worker(final_model, final_rx, updates, &opts)
++            }));
++            Some(final_tx)
+         }
++        None => None,
++    };
++    {
++        let opts = opts.clone();
++        let updates = update_tx.clone();
++        workers.push(thread::spawn(move || {
++            run_partial_worker(model, job_rx, updates, &opts)
++        }));
++    }
++    drop(update_tx);
++
++    let (capture_tx, capture_rx) = mpsc::channel::<Capture>();
++    let input = InputStream::open(capture_tx)?;
++    eprintln!(
++        "Input: {} Hz, {} ch ({:?}){}",
++        input.config.sample_rate(),
++        input.config.channels(),
++        input.config.sample_format(),
++        if input.needs_conversion() {
++            " — converting to 16 kHz mono"
++        } else {
++            ""
++        }
++    );
++
++    let capture = {
++        let opts = opts.clone();
++        let format = input.format();
++        thread::spawn(move || run_capture(capture_rx, job_tx, final_tx, format, &opts))
+     };
+ 
+     println!("Listening... press Enter to stop.");
+-    stream.play()?;
++    input.stream.play()?;
+ 
+     let stop = Arc::new(AtomicBool::new(false));
+-    let stop_clone = Arc::clone(&stop);
+-    std::thread::spawn(move || {
+-        let mut line = String::new();
+-        let _ = std::io::stdin().read_line(&mut line);
+-        stop_clone.store(true, Ordering::SeqCst);
+-    });
+-
+-    let mut last_len = 0usize;
+-    let mut final_line = String::new();
+-    let mut silence_samples = 0usize;
+-    let mut noise_floor = 0.0_f32;
+-    let mut utterance_audio: Vec<f32> = Vec::new();
+-    let mut in_utterance = false;
+-    while !stop.load(Ordering::SeqCst) {
+-        match rx.recv_timeout(Duration::from_millis(100)) {
+-            Ok(chunk) => {
+-                let chunk_rms = rms(&chunk);
+-                if noise_floor == 0.0 {
+-                    noise_floor = chunk_rms;
+-                } else if chunk_rms < noise_floor * 1.5 {
+-                    noise_floor = noise_floor * 0.95 + chunk_rms * 0.05;
+-                }
+-                let silence_threshold =
+-                    (noise_floor * 3.0).clamp(min_silence_threshold, max_silence_threshold);
+-
+-                if chunk_rms < silence_threshold {
+-                    if in_utterance {
+-                        silence_samples = silence_samples.saturating_add(chunk.len());
+-                        utterance_audio.extend_from_slice(&chunk);
+-                        if silence_samples >= silence_hold_samples {
+-                            if utterance_audio.len() >= min_utterance_samples {
+-                                let final_result = full_session.transcribe_samples(
+-                                    &utterance_audio,
+-                                    sample_rate,
+-                                    channels,
+-                                    &source_lang,
+-                                    &target_lang,
+-                                )?;
+-                                let final_text =
+-                                    normalize_punctuation_spacing(final_result.text.trim());
+-                                if !final_text.is_empty() {
+-                                    if last_len > 0 {
+-                                        print!("\r{:<width$}", "", width = last_len);
+-                                    }
+-                                    print!("\r{}", final_text);
+-                                    println!();
+-                                    let _ = std::io::stdout().flush();
+-                                    last_len = 0;
+-                                }
+-                            } else if last_len > 0 {
+-                                println!();
+-                                last_len = 0;
+-                            }
+-
+-                            stream_state.reset();
+-                            final_line.clear();
+-                            utterance_audio.clear();
+-                            silence_samples = 0;
+-                            in_utterance = false;
+-                        }
+-                    }
+-                    continue;
+-                }
+-                silence_samples = 0;
+-                in_utterance = true;
+-                utterance_audio.extend_from_slice(&chunk);
+-
+-                let results = stream_state.push_samples(&chunk, sample_rate, channels)?;
+-                for result in results {
+-                    let delta = result.delta_text.trim();
+-                    if delta.is_empty() {
+-                        continue;
+-                    }
++    {
++        let stop = Arc::clone(&stop);
++        thread::spawn(move || {
++            let mut line = String::new();
++            let _ = std::io::stdin().read_line(&mut line);
++            stop.store(true, Ordering::SeqCst);
++        });
++    }
+ 
+-                    let delta = normalize_punctuation_spacing(delta);
+-                    let delta = strip_duplicate_leading_punct(&final_line, &delta);
+-                    if delta.is_empty() {
+-                        continue;
+-                    }
++    let mut render = Renderer::new(&opts);
++    let mut stream = Some(input.stream);
++    loop {
++        match update_rx.recv_timeout(Duration::from_millis(100)) {
++            Ok(update) => render.apply(update),
++            Err(mpsc::RecvTimeoutError::Timeout) => {}
++            Err(mpsc::RecvTimeoutError::Disconnected) => break,
++        }
++        // Dropping the stream ends the capture thread, which ends the workers, which closes
++        // `update_rx` — but only after the in-flight decodes have been rendered.
++        if stop.load(Ordering::SeqCst) {
++            stream = None;
++        }
++    }
++    drop(stream);
++    render.finish();
+ 
+-                    append_with_space(&mut final_line, &delta);
++    let _ = capture.join();
++    for worker in workers {
++        let _ = worker.join();
++    }
++    Ok(())
++}
+ 
+-                    let mut line_break = false;
+-                    if ends_with_sentence_punct(&final_line) {
+-                        line_break = true;
+-                    }
++/// Runtime knobs, all overridable from the environment so the pipeline can be tuned without a
++/// rebuild.
++#[derive(Clone)]
++struct Options {
++    source_lang: String,
++    target_lang: String,
++    /// Rolling window length; longer means more context but fewer updates per second.
++    window: f32,
++    /// How far the window advances between decodes.
++    step: f32,
++    /// Windows that must agree before a word is committed to the stable transcript.
++    stability: usize,
++    /// Silence appended to each decoded window; see `StreamConfig::tail_silence_duration`.
++    tail_silence: f32,
++    /// Silence that ends an utterance.
++    silence_hold: f32,
++    /// Shortest utterance worth a final decode.
++    min_utterance: f32,
++    /// Longest utterance before one is forced, bounding final-decode cost and memory.
++    max_utterance: f32,
++    dedicated_final_model: bool,
++    stats: bool,
++    width: usize,
++}
+ 
+-                    if !final_line.is_empty() && has_alnum(&final_line) {
+-                        last_len = last_len.max(final_line.len());
+-                        print!("\r{:<width$}", final_line, width = last_len);
+-                        let _ = std::io::stdout().flush();
+-                    }
++impl Options {
++    fn from_env() -> Self {
++        let source_lang = std::env::args().nth(1).unwrap_or_else(|| "en".to_string());
++        let target_lang = std::env::args().nth(2).unwrap_or_else(|| source_lang.clone());
++        Self {
++            source_lang,
++            target_lang,
++            window: env_f32("CANARY_LIVE_WINDOW").unwrap_or(6.0),
++            step: env_f32("CANARY_LIVE_STEP").unwrap_or(0.4),
++            stability: env_usize("CANARY_LIVE_STABILITY").unwrap_or(2),
++            tail_silence: env_f32("CANARY_LIVE_TAIL_SILENCE")
++                .unwrap_or(StreamConfig::default().tail_silence_duration),
++            silence_hold: env_f32("CANARY_LIVE_SILENCE_HOLD").unwrap_or(0.8),
++            min_utterance: env_f32("CANARY_LIVE_MIN_UTTERANCE").unwrap_or(0.3),
++            max_utterance: env_f32("CANARY_LIVE_MAX_UTTERANCE").unwrap_or(20.0),
++            dedicated_final_model: env_bool("CANARY_LIVE_FINAL_MODEL").unwrap_or(false),
++            stats: env_bool("CANARY_LIVE_STATS").unwrap_or(false),
++            width: env_usize("CANARY_LIVE_WIDTH")
++                .or_else(|| env_usize("COLUMNS"))
++                .unwrap_or(100),
++        }
++    }
+ 
+-                    if line_break {
+-                        println!();
+-                        last_len = 0;
+-                        final_line.clear();
+-                    }
+-                }
+-            }
+-            Err(mpsc::RecvTimeoutError::Timeout) => {}
+-            Err(mpsc::RecvTimeoutError::Disconnected) => break,
++    fn stream_config(&self) -> StreamConfig {
++        StreamConfig::new()
++            .with_window_duration(self.window)
++            .with_step_duration(self.step)
++            .with_emit_partial(true)
++            .with_pad_partial(false)
++            .with_stability_window(self.stability)
++            .with_tail_silence_duration(self.tail_silence)
++            // The whole point of the pipeline: never work through a backlog of stale windows.
++            .with_max_windows_per_push(1)
++    }
++}
++
++fn env_f32(key: &str) -> Option<f32> {
++    std::env::var(key).ok()?.trim().parse().ok()
++}
++
++fn env_usize(key: &str) -> Option<usize> {
++    std::env::var(key).ok()?.trim().parse().ok()
++}
++
++fn env_bool(key: &str) -> Option<bool> {
++    match std::env::var(key).ok()?.trim().to_lowercase().as_str() {
++        "1" | "true" | "yes" => Some(true),
++        "0" | "false" | "no" => Some(false),
++        _ => None,
++    }
++}
++
++/// One buffer straight from the audio callback, stamped so the pipeline can report how far
++/// behind the speaker it is running.
++struct Capture {
++    samples: Vec<f32>,
++    at: Instant,
++}
++
++/// Work handed to a decode thread. `seq` identifies the utterance so that a final arriving late
++/// is not mistaken for the transcript of the utterance already in progress.
++enum Job {
++    /// Newly captured audio for the rolling partial window, at 16 kHz mono.
++    Audio {
++        seq: u64,
++        samples: Vec<f32>,
++        at: Instant,
++    },
++    /// A finished utterance, to be decoded in full.
++    Final { seq: u64, audio: Vec<f32> },
++    /// The utterance ended with nothing worth decoding.
++    Reset { seq: u64 },
++}
++
++enum Update {
++    /// Stable transcript so far plus the still-changing tail of the newest window.
++    Partial {
++        seq: u64,
++        committed: String,
++        volatile: String,
++        stats: Stats,
++    },
++    /// Whole-utterance transcript; replaces everything shown for `seq`.
++    Final {
++        seq: u64,
++        text: String,
++        stats: Stats,
++    },
++    /// The utterance produced nothing.
++    Discard { seq: u64 },
++    Error(String),
++}
++
++#[derive(Clone, Copy)]
++struct Stats {
++    decode: Duration,
++    /// Age of the newest sample in the decoded audio when the result was ready.
++    lag: Duration,
++    /// Steps of audio thrown away as stale before this decode.
++    dropped: usize,
++}
++
++// ---------------------------------------------------------------------------------------------
++// Capture
++// ---------------------------------------------------------------------------------------------
++
++/// The input stream plus the format it actually opened with.
++struct InputStream {
++    stream: cpal::Stream,
++    config: SupportedStreamConfig,
++}
++
++impl InputStream {
++    /// Opens the default input device, preferring 16 kHz mono so no conversion is needed.
++    fn open(tx: Sender<Capture>) -> Result<Self, Box<dyn std::error::Error>> {
++        let host = cpal::default_host();
++        let device = host
++            .default_input_device()
++            .ok_or("No input device available")?;
++
++        let config = Self::preferred_config(&device)?;
++        let stream_config: cpal::StreamConfig = config.clone().into();
++        let err_fn = |err| eprintln!("stream error: {}", err);
++
++        // The callback runs on a realtime thread: convert to f32, hand it over, return.
++        macro_rules! build {
++            ($sample:ty) => {
++                device.build_input_stream(
++                    &stream_config,
++                    move |data: &[$sample], _: &_| {
++                        let samples = data.iter().map(|s| f32::from_sample(*s)).collect();
++                        let _ = tx.send(Capture {
++                            samples,
++                            at: Instant::now(),
++                        });
++                    },
++                    err_fn,
++                    None,
++                )?
++            };
+         }
++
++        let stream = match config.sample_format() {
++            SampleFormat::F32 => build!(f32),
++            SampleFormat::I16 => build!(i16),
++            SampleFormat::U16 => build!(u16),
++            other => return Err(format!("Unsupported sample format: {:?}", other).into()),
++        };
++
++        Ok(Self { stream, config })
+     }
+ 
+-    Ok(())
++    fn preferred_config(
++        device: &cpal::Device,
++    ) -> Result<SupportedStreamConfig, Box<dyn std::error::Error>> {
++        let native = device
++            .supported_input_configs()?
++            .filter(|range| range.channels() == 1)
++            .filter(|range| {
++                range.min_sample_rate() <= MODEL_SAMPLE_RATE
++                    && MODEL_SAMPLE_RATE <= range.max_sample_rate()
++            })
++            .find(|range| range.sample_format() == SampleFormat::F32)
++            .map(|range| range.with_sample_rate(MODEL_SAMPLE_RATE));
++
++        match native {
++            Some(config) => Ok(config),
++            None => Ok(device.default_input_config()?),
++        }
++    }
++
++    fn needs_conversion(&self) -> bool {
++        self.config.sample_rate() != MODEL_SAMPLE_RATE || self.config.channels() != 1
++    }
++
++    fn format(&self) -> InputFormat {
++        InputFormat {
++            sample_rate: self.config.sample_rate() as usize,
++            channels: self.config.channels() as usize,
++        }
++    }
++}
++
++#[derive(Clone, Copy)]
++struct InputFormat {
++    sample_rate: usize,
++    channels: usize,
+ }
+ 
+-fn push_input_data<T: Sample>(
+-    input: &[T],
+-    chunk_samples: usize,
+-    buffer: &Arc<Mutex<Vec<f32>>>,
+-    sender: &mpsc::Sender<Vec<f32>>,
+-) where
+-    f32: cpal::FromSample<T>,
+-{
+-    let mut buffer = match buffer.lock() {
+-        Ok(guard) => guard,
+-        Err(poisoned) => poisoned.into_inner(),
++/// Turns captured audio into decode jobs: converts to 16 kHz mono, tracks the noise floor, and
++/// cuts the stream into utterances.
++fn run_capture(
++    rx: Receiver<Capture>,
++    job_tx: Sender<Job>,
++    final_tx: Option<Sender<Job>>,
++    format: InputFormat,
++    opts: &Options,
++) {
++    let rate = MODEL_SAMPLE_RATE as f32;
++    let frame_samples = (rate * VAD_FRAME_SECONDS).max(1.0) as usize;
++    let preroll_samples = (rate * PREROLL_SECONDS) as usize;
++    let silence_hold_samples = (rate * opts.silence_hold) as usize;
++    let min_utterance_samples = (rate * opts.min_utterance) as usize;
++    let max_utterance_samples = (rate * opts.max_utterance) as usize;
++
++    let mut frame: Vec<f32> = Vec::with_capacity(frame_samples * 2);
++    let mut preroll: std::collections::VecDeque<f32> = std::collections::VecDeque::new();
++    let mut utterance: Vec<f32> = Vec::new();
++    let mut noise_floor = 0.0f32;
++    let mut silence_run = 0usize;
++    let mut speaking = false;
++    let mut seq = 0u64;
++
++    // Sending a boundary is the same work whether or not a dedicated final worker exists.
++    let send_boundary = |job: Job| -> bool {
++        match (&final_tx, &job) {
++            (Some(final_tx), Job::Final { seq, .. }) => {
++                job_tx.send(Job::Reset { seq: *seq }).is_ok() && final_tx.send(job).is_ok()
++            }
++            _ => job_tx.send(job).is_ok(),
++        }
+     };
+ 
+-    buffer.extend(input.iter().map(|sample| f32::from_sample(*sample)));
++    while let Ok(capture) = rx.recv() {
++        let mono = to_model_audio(capture.samples, format);
++        frame.extend_from_slice(&mono);
++
++        while frame.len() >= frame_samples {
++            let chunk: Vec<f32> = frame.drain(..frame_samples).collect();
++            let level = rms(&chunk);
+ 
+-    while buffer.len() >= chunk_samples {
+-        let chunk: Vec<f32> = buffer.drain(..chunk_samples).collect();
+-        let _ = sender.send(chunk);
++            // Track the quietest recent level as the noise floor, and only adapt downwards so a
++            // steady voice cannot drag the threshold up over itself.
++            if noise_floor == 0.0 {
++                noise_floor = level;
++            } else if level < noise_floor * 1.5 {
++                noise_floor = noise_floor * 0.95 + level * 0.05;
++            }
++            let threshold = (noise_floor * 3.0).clamp(MIN_SILENCE_THRESHOLD, MAX_SILENCE_THRESHOLD);
++
++            if level < threshold && !speaking {
++                preroll.extend(chunk.iter().copied());
++                while preroll.len() > preroll_samples {
++                    preroll.pop_front();
++                }
++                continue;
++            }
++
++            if !speaking {
++                speaking = true;
++                seq += 1;
++                utterance.clear();
++                utterance.extend(preroll.drain(..));
++            }
++
++            if level < threshold {
++                silence_run += chunk.len();
++            } else {
++                silence_run = 0;
++            }
++            utterance.extend_from_slice(&chunk);
++
++            let ended = silence_run >= silence_hold_samples;
++            let too_long = utterance.len() >= max_utterance_samples;
++            if !ended && !too_long {
++                let job = Job::Audio {
++                    seq,
++                    samples: chunk,
++                    at: capture.at,
++                };
++                if job_tx.send(job).is_err() {
++                    return;
++                }
++                continue;
++            }
++
++            let long_enough = utterance.len() >= min_utterance_samples;
++            let boundary = if long_enough {
++                Job::Final {
++                    seq,
++                    audio: std::mem::take(&mut utterance),
++                }
++            } else {
++                Job::Reset { seq }
++            };
++            if !send_boundary(boundary) {
++                return;
++            }
++
++            utterance.clear();
++            preroll.clear();
++            silence_run = 0;
++            // A forced cut lands mid-sentence, so keep listening as if speech never stopped.
++            speaking = too_long && !ended;
++            if speaking {
++                seq += 1;
++            }
++        }
+     }
+ }
+ 
+-fn ends_with_sentence_punct(text: &str) -> bool {
+-    text.chars()
+-        .rev()
+-        .find(|ch| !ch.is_whitespace())
+-        .map_or(false, |ch| matches!(ch, '.' | '!' | '?'))
++/// Converts a captured buffer to the 16 kHz mono the model expects.
++fn to_model_audio(samples: Vec<f32>, format: InputFormat) -> Vec<f32> {
++    let mono = if format.channels > 1 {
++        samples
++            .chunks(format.channels)
++            .map(|frame| frame.iter().sum::<f32>() / format.channels as f32)
++            .collect()
++    } else {
++        samples
++    };
++
++    if format.sample_rate == MODEL_SAMPLE_RATE as usize {
++        return mono;
++    }
++    // Cheap linear resampling is enough here: the input is already band-limited by the device
++    // and the alternative costs more than the decode it feeds.
++    let ratio = MODEL_SAMPLE_RATE as f64 / format.sample_rate as f64;
++    let out_len = (mono.len() as f64 * ratio) as usize;
++    (0..out_len)
++        .map(|i| {
++            let pos = i as f64 / ratio;
++            let idx = pos as usize;
++            let frac = (pos - idx as f64) as f32;
++            let a = mono.get(idx).copied().unwrap_or(0.0);
++            let b = mono.get(idx + 1).copied().unwrap_or(a);
++            a + (b - a) * frac
++        })
++        .collect()
+ }
+ 
+ fn rms(samples: &[f32]) -> f32 {
+@@ -253,19 +510,398 @@ fn rms(samples: &[f32]) -> f32 {
+     (sum / samples.len() as f32).sqrt()
+ }
+ 
+-fn has_alnum(text: &str) -> bool {
+-    text.chars().any(|ch| ch.is_alphanumeric())
++// ---------------------------------------------------------------------------------------------
++// Decoding
++// ---------------------------------------------------------------------------------------------
++
++/// Runs the rolling-window decode, and the final decode too unless a dedicated worker took it.
++fn run_partial_worker(model: Canary, rx: Receiver<Job>, tx: Sender<Update>, opts: &Options) {
++    let mut stream = match model.stream(
++        opts.source_lang.clone(),
++        opts.target_lang.clone(),
++        opts.stream_config(),
++    ) {
++        Ok(stream) => stream,
++        Err(err) => {
++            let _ = tx.send(Update::Error(format!("stream setup failed: {}", err)));
++            return;
++        }
++    };
++    let mut session = model.session();
++    let mut committed = String::new();
++
++    for batch in Batches::new(rx) {
++        // Audio queued behind an utterance boundary belongs to a transcript that is about to be
++        // decoded in full, so decoding a window of it would only spend time to be overwritten.
++        let mut pending: Vec<f32> = Vec::new();
++        let mut pending_seq = 0u64;
++        let mut newest = Instant::now();
++        let mut dropped = 0usize;
++
++        for job in batch {
++            match job {
++                Job::Audio { seq, samples, at } => {
++                    pending_seq = seq;
++                    pending.extend_from_slice(&samples);
++                    newest = at;
++                }
++                Job::Reset { seq } => {
++                    dropped += steps_in(&pending, opts);
++                    pending.clear();
++                    stream.reset();
++                    committed.clear();
++                    let _ = tx.send(Update::Discard { seq });
++                }
++                Job::Final { seq, audio } => {
++                    dropped += steps_in(&pending, opts);
++                    pending.clear();
++                    stream.reset();
++                    committed.clear();
++                    if decode_final(&mut session, &audio, seq, &tx, opts).is_err() {
++                        return;
++                    }
++                }
++            }
++        }
++
++        if pending.is_empty() {
++            continue;
++        }
++        dropped += steps_in(&pending, opts).saturating_sub(1);<