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, 351 insertions, 0 deletions

diff --git a/crates/asist-core/src/http.rs b/crates/asist-core/src/http.rs
new file mode 100644
index 0000000..32b6680
--- /dev/null
+++ b/crates/asist-core/src/http.rs
@@ -0,0 +1,351 @@
+//! Cliente HTTP/1.1 mínimo para hablar con los servidores locales.
+//!
+//! Los tres motores (llama-server, tts-server) escuchan en localhost sobre HTTP
+//! plano, así que no hace falta TLS ni un runtime asíncrono. A cambio de las
+//! ~300 líneas de aquí se gana lo único que ninguna librería genérica da
+//! cómodo y que este pipeline necesita de verdad: **abortar una respuesta a
+//! media descarga**. Cuando el usuario interrumpe al asistente hay que dejar de
+//! leer audio ya generado y cerrar el socket en ese mismo instante; con un
+//! cliente que sólo expone `read_to_end` eso no se puede hacer.
+
+use std::io::{BufRead, BufReader, Read, Write};
+use std::net::TcpStream;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use crate::error::{Error, Result};
+
+/// Bandera compartida que aborta una lectura en curso.
+///
+/// Se comprueba entre bloques, así que el corte ocurre como muy tarde un
+/// `read` después de activarla —del orden de milisegundos con los tamaños de
+/// bloque que usan los servidores.
+#[derive(Clone, Default)]
+pub struct Cancel(Arc<AtomicBool>);
+
+impl Cancel {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn cancel(&self) {
+        self.0.store(true, Ordering::SeqCst);
+    }
+
+    pub fn is_cancelled(&self) -> bool {
+        self.0.load(Ordering::SeqCst)
+    }
+
+    pub fn reset(&self) {
+        self.0.store(false, Ordering::SeqCst);
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct HttpClient {
+    /// `host:puerto` del servidor local.
+    pub authority: String,
+    pub connect_timeout: Duration,
+    /// Tiempo máximo sin recibir un solo byte. No es el tiempo total: una
+    /// síntesis larga puede tardar minutos siempre que siga fluyendo.
+    pub read_timeout: Duration,
+}
+
+impl HttpClient {
+    pub fn new(authority: impl Into<String>) -> Self {
+        Self {
+            authority: authority.into(),
+            connect_timeout: Duration::from_secs(5),
+            read_timeout: Duration::from_secs(120),
+        }
+    }
+
+    pub fn with_read_timeout(mut self, timeout: Duration) -> Self {
+        self.read_timeout = timeout;
+        self
+    }
+
+    fn connect(&self, path: &str) -> Result<TcpStream> {
+        let addr = resolve(&self.authority, path)?;
+        let stream = TcpStream::connect_timeout(&addr, self.connect_timeout).map_err(|e| {
+            Error::Transport {
+                url: format!("http://{}{}", self.authority, path),
+                source: e,
+            }
+        })?;
+        stream.set_read_timeout(Some(self.read_timeout))?;
+        // Los cuerpos son pequeños y la latencia manda: nada de Nagle.
+        let _ = stream.set_nodelay(true);
+        Ok(stream)
+    }
+
+    fn send(&self, method: &str, path: &str, body: Option<&[u8]>) -> Result<Response> {
+        let mut stream = self.connect(path)?;
+        let mut head = format!(
+            "{method} {path} HTTP/1.1\r\nHost: {}\r\nConnection: close\r\nAccept: */*\r\n",
+            self.authority
+        );
+        if let Some(body) = body {
+            head.push_str("Content-Type: application/json\r\n");
+            head.push_str(&format!("Content-Length: {}\r\n", body.len()));
+        }
+        head.push_str("\r\n");
+
+        let url = format!("http://{}{}", self.authority, path);
+        let io = |e: std::io::Error| Error::Transport {
+            url: url.clone(),
+            source: e,
+        };
+        stream.write_all(head.as_bytes()).map_err(io)?;
+        if let Some(body) = body {
+            stream.write_all(body).map_err(io)?;
+        }
+        stream.flush().map_err(io)?;
+
+        Response::read_head(BufReader::new(stream), url)
+    }
+
+    /// GET que devuelve el cuerpo completo ya decodificado.
+    pub fn get(&self, path: &str) -> Result<Vec<u8>> {
+        self.send("GET", path, None)?.into_body()
+    }
+
+    /// POST de JSON que devuelve el cuerpo completo.
+    pub fn post_json(&self, path: &str, body: &serde_json::Value) -> Result<Vec<u8>> {
+        let payload = serde_json::to_vec(body)?;
+        self.send("POST", path, Some(&payload))?.into_body()
+    }
+
+    /// POST de JSON cuya respuesta se consume a trozos según llega.
+    ///
+    /// `on_chunk` recibe cada bloque en cuanto está disponible y devuelve
+    /// `false` para cortar; junto a `cancel` son las dos vías por las que el
+    /// barge-in detiene una síntesis a medias.
+    pub fn post_json_streaming(
+        &self,
+        path: &str,
+        body: &serde_json::Value,
+        cancel: &Cancel,
+        mut on_chunk: impl FnMut(&[u8]) -> bool,
+    ) -> Result<()> {
+        let payload = serde_json::to_vec(body)?;
+        let mut response = self.send("POST", path, Some(&payload))?;
+        response.error_for_status()?;
+        response.stream(cancel, &mut on_chunk)
+    }
+
+    /// POST de JSON que entrega la respuesta línea a línea (SSE de llama-server).
+    pub fn post_json_lines(
+        &self,
+        path: &str,
+        body: &serde_json::Value,
+        cancel: &Cancel,
+        mut on_line: impl FnMut(&str) -> bool,
+    ) -> Result<()> {
+        let mut pending = Vec::new();
+        self.post_json_streaming(path, body, cancel, |chunk| {
+            pending.extend_from_slice(chunk);
+            while let Some(nl) = pending.iter().position(|b| *b == b'\n') {
+                let line: Vec<u8> = pending.drain(..=nl).collect();
+                let line = String::from_utf8_lossy(&line);
+                if !on_line(line.trim_end_matches(['\r', '\n'])) {
+                    return false;
+                }
+            }
+            true
+        })
+    }
+
+    /// Sondea `/health` y devuelve `true` si el servidor está listo.
+    pub fn healthy(&self, path: &str) -> bool {
+        self.get(path).is_ok()
+    }
+}
+
+fn resolve(authority: &str, path: &str) -> Result<std::net::SocketAddr> {
+    use std::net::ToSocketAddrs;
+    authority
+        .to_socket_addrs()
+        .map_err(|e| Error::Transport {
+            url: format!("http://{authority}{path}"),
+            source: e,
+        })?
+        .next()
+        .ok_or_else(|| Error::Config(format!("no se pudo resolver «{authority}»")))
+}
+
+/// Codificación del cuerpo anunciada por el servidor.
+enum Body {
+    Chunked,
+    Length(usize),
+    /// Sin longitud ni chunked: el cuerpo acaba cuando cierra el socket.
+    UntilClose,
+}
+
+pub struct Response {
+    pub status: u16,
+    url: String,
+    reader: BufReader<TcpStream>,
+    body: Body,
+}
+
+impl Response {
+    fn read_head(mut reader: BufReader<TcpStream>, url: String) -> Result<Self> {
+        let io = |e: std::io::Error| Error::Transport {
+            url: url.clone(),
+            source: e,
+        };
+
+        let mut line = String::new();
+        reader.read_line(&mut line).map_err(io)?;
+        let status = line
+            .split_whitespace()
+            .nth(1)
+            .and_then(|s| s.parse::<u16>().ok())
+            .ok_or_else(|| Error::Config(format!("respuesta HTTP ilegible de {url}: {line:?}")))?;
+
+        let mut body = Body::UntilClose;
+        loop {
+            let mut header = String::new();
+            if reader.read_line(&mut header).map_err(io)? == 0 {
+                break;
+            }
+            let header = header.trim_end();
+            if header.is_empty() {
+                break;
+            }
+            let Some((name, value)) = header.split_once(':') else {
+                continue;
+            };
+            let (name, value) = (name.trim().to_ascii_lowercase(), value.trim());
+            match name.as_str() {
+                "transfer-encoding" if value.eq_ignore_ascii_case("chunked") => {
+                    body = Body::Chunked;
+                }
+                "content-length" => {
+                    if let Ok(n) = value.parse() {
+                        body = Body::Length(n);
+                    }
+                }
+                _ => {}
+            }
+        }
+
+        Ok(Self {
+            status,
+            url,
+            reader,
+            body,
+        })
+    }
+
+    fn error_for_status(&mut self) -> Result<()> {
+        if (200..300).contains(&self.status) {
+            return Ok(());
+        }
+        let status = self.status;
+        let url = self.url.clone();
+        // Leer el cuerpo del error es lo que convierte un «500» en un mensaje útil.
+        let body = self.drain().unwrap_or_default();
+        Err(Error::Http {
+            status,
+            url,
+            body: String::from_utf8_lossy(&body).chars().take(500).collect(),
+        })
+    }
+
+    fn into_body(mut self) -> Result<Vec<u8>> {
+        self.error_for_status()?;
+        self.drain()
+    }
+
+    fn drain(&mut self) -> Result<Vec<u8>> {
+        let mut out = Vec::new();
+        let cancel = Cancel::new();
+        self.stream(&cancel, &mut |chunk| {
+            out.extend_from_slice(chunk);
+            true
+        })?;
+        Ok(out)
+    }
+
+    fn stream(&mut self, cancel: &Cancel, on_chunk: &mut dyn FnMut(&[u8]) -> bool) -> Result<()> {
+        let url = self.url.clone();
+        let io = |e: std::io::Error| Error::Transport {
+            url: url.clone(),
+            source: e,
+        };
+        let mut buf = vec![0u8; 16 * 1024];
+
+        match self.body {
+            Body::Chunked => loop {
+                if cancel.is_cancelled() {
+                    return Err(Error::Cancelled);
+                }
+                let mut size_line = String::new();
+                if self.reader.read_line(&mut size_line).map_err(io)? == 0 {
+                    return Ok(());
+                }
+                let size_line = size_line.trim();
+                if size_line.is_empty() {
+                    continue;
+                }
+                // Las extensiones de chunk van tras un ';' y aquí no interesan.
+                let size =
+                    usize::from_str_radix(size_line.split(';').next().unwrap_or("0").trim(), 16)
+                        .map_err(|_| {
+                            Error::Config(format!("tamaño de chunk ilegible: {size_line:?}"))
+                        })?;
+                if size == 0 {
+                    return Ok(());
+                }
+                let mut left = size;
+                while left > 0 {
+                    if cancel.is_cancelled() {
+                        return Err(Error::Cancelled);
+                    }
+                    let want = left.min(buf.len());
+                    self.reader.read_exact(&mut buf[..want]).map_err(io)?;
+                    if !on_chunk(&buf[..want]) {
+                        return Err(Error::Cancelled);
+                    }
+                    left -= want;
+                }
+                // CRLF de cierre del chunk.
+                let mut crlf = [0u8; 2];
+                self.reader.read_exact(&mut crlf).map_err(io)?;
+            },
+            Body::Length(total) => {
+                let mut left = total;
+                while left > 0 {
+                    if cancel.is_cancelled() {
+                        return Err(Error::Cancelled);
+                    }
+                    let want = left.min(buf.len());
+                    let n = self.reader.read(&mut buf[..want]).map_err(io)?;
+                    if n == 0 {
+                        return Ok(());
+                    }
+                    if !on_chunk(&buf[..n]) {
+                        return Err(Error::Cancelled);
+                    }
+                    left -= n;
+                }
+                Ok(())
+            }
+            Body::UntilClose => loop {
+                if cancel.is_cancelled() {
+                    return Err(Error::Cancelled);
+                }
+                let n = self.reader.read(&mut buf).map_err(io)?;
+                if n == 0 {
+                    return Ok(());
+                }
+                if !on_chunk(&buf[..n]) {
+                    return Err(Error::Cancelled);
+                }
+            },
+        }
+    }
+}