tawasuyu/brahman

1//! El cliente: consume un daemon presentándose como un `Provider`.
2//!
3//! Un [`DaemonClient`] implementa `verbo_core::Provider`, así que
4//! cualquier consumidor (`fana-semantic`, `badu`, `chasqui`) lo usa sin
5//! saber que el modelo vive en otro proceso. Cada llamada es un
6//! round-trip independiente: sin estado de conexión que reparar.
7
8use std::path::{Path, PathBuf};
9
10use async_trait::async_trait;
11use tokio::net::UnixStream;
12use verbo_core::{EmbedError, EmbeddingVector, ModelId, Provider};
13
14use crate::wire::{read_frame, write_frame, Request, Response};
15
16/// Cliente de un [`crate::Daemon`]. Se comporta como un `Provider`
17/// local — los consumidores no notan que el modelo es remoto.
18pub struct DaemonClient {
19    path: PathBuf,
20    model: ModelId,
21}
22
23impl DaemonClient {
24    /// Conecta a un daemon y hace el handshake del modelo. El `ModelId`
25    /// queda cacheado: marca los vectores y nunca cambia en vida del
26    /// daemon.
27    pub async fn connect(path: impl AsRef<Path>) -> Result<Self, EmbedError> {
28        let path = path.as_ref().to_path_buf();
29        let model = match round_trip(&path, &Request::ModelId).await? {
30            Response::ModelId(m) => m,
31            other => return Err(unexpected(other)),
32        };
33        Ok(Self { path, model })
34    }
35}
36
37/// Mapea una respuesta fuera de contrato a un `EmbedError`.
38fn unexpected(r: Response) -> EmbedError {
39    match r {
40        Response::Error(e) => EmbedError::Backend(e),
41        _ => EmbedError::Backend("respuesta del daemon verbo inesperada".into()),
42    }
43}
44
45/// Un round-trip completo: conecta, manda el request, lee la respuesta.
46async fn round_trip(path: &Path, req: &Request) -> Result<Response, EmbedError> {
47    let mut stream = UnixStream::connect(path)
48        .await
49        .map_err(|e| EmbedError::Backend(format!("conexión al daemon verbo: {e}")))?;
50    write_frame(&mut stream, req)
51        .await
52        .map_err(|e| EmbedError::Backend(format!("envío al daemon verbo: {e}")))?;
53    match read_frame::<_, Response>(&mut stream).await {
54        Ok(Some(resp)) => Ok(resp),
55        Ok(None) => Err(EmbedError::Backend(
56            "el daemon verbo cerró la conexión sin responder".into(),
57        )),
58        Err(e) => Err(EmbedError::Backend(format!("lectura del daemon verbo: {e}"))),
59    }
60}
61
62#[async_trait]
63impl Provider for DaemonClient {
64    fn model_id(&self) -> &ModelId {
65        &self.model
66    }
67
68    async fn embed(&self, text: &str) -> Result<EmbeddingVector, EmbedError> {
69        match round_trip(&self.path, &Request::Embed(text.to_string())).await? {
70            Response::Embed(v) => Ok(v),
71            other => Err(unexpected(other)),
72        }
73    }
74
75    async fn embed_batch(&self, texts: &[String]) -> Result<Vec<EmbeddingVector>, EmbedError> {
76        match round_trip(&self.path, &Request::EmbedBatch(texts.to_vec())).await? {
77            Response::EmbedBatch(v) => Ok(v),
78            other => Err(unexpected(other)),
79        }
80    }
81}