tawasuyu/brahman

1//! `nouser-nous-mock` — proveedor de embeddings determinista (sin LLM).
2//!
3//! Implementa el contrato `nouser-nous` usando los pseudo-embeddings
4//! de Phase C (`nouser_core::embed`). Sirve como:
5//!
6//! - **Mock para tests**: en `BRAHMAN_BROKER_CONTEXT=test`, el
7//!   `priority_offset` per-contexto declarado en su Card lo prioriza
8//!   sobre cualquier proveedor real.
9//! - **Bootstrap**: hasta que llegue el LLM real (Phase D futura), el
10//!   sistema funciona end-to-end con embeddings determinísticos.
11//!
12//! ## Vida del proceso
13//!
14//! 1. Sidecarea a brahman-init declarando una Card con flow output
15//!    `embed-result:json` y flow input `embed-request:json`. Su
16//!    `priority_contexts.test = { priority_offset: +1 }` lo prioriza
17//!    cuando el broker corre bajo contexto test.
18//! 2. Bind del Unix socket en `$NOUSER_NOUS_SOCKET` (default
19//!    `$XDG_RUNTIME_DIR/nouser-nous.sock`).
20//! 3. Loop: accept → read line JSON → process → write line JSON → close.
21//! 4. Cada request se loggea (info) — útil para verificar que el
22//!    consumidor está usando este proveedor.
23
24use std::collections::BTreeMap;
25use std::path::PathBuf;
26use std::time::{Instant, SystemTime, UNIX_EPOCH};
27
28use brahman_card::{
29    ulid::Ulid, Card, CardKind, ContextBias, Flow, Flows, Lifecycle, Payload, Priority,
30    Supervision, TypeRef,
31};
32use nouser_card::FileEntry;
33use nouser_core::embed;
34use nouser_nous::{
35    transport, EmbedFilePayload, EmbedRequest, EmbedResponse, EmbedTextPayload, ErrorResponse,
36    PingResponse, RequestKind, FLOW_EMBED_REQUEST, FLOW_EMBED_RESULT, FLOW_TYPE_NAME,
37};
38use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
39use tokio::net::{UnixListener, UnixStream};
40use tracing::{info, warn};
41
42/// El mock implementa el MISMO algoritmo que `nouser_core::embed`,
43/// así que reportamos el mismo `MODEL_ID` que él. De otro modo el
44/// consumer filtraría las Mónadas como "modelo distinto" y los
45/// scores quedarían vacíos.
46const MODEL_ID: &str = nouser_core::embed::MODEL_ID;
47
48#[tokio::main(flavor = "current_thread")]
49async fn main() -> std::io::Result<()> {
50    init_tracing();
51
52    // 1. Resolver socket del data-plane ANTES de armar la Card, para
53    //    declararlo en `Card.service_socket` y que los consumidores lo
54    //    descubran vía MatchEvent.
55    let sock_path = transport::provider_socket_path("mock");
56    if sock_path.exists() {
57        std::fs::remove_file(&sock_path)?;
58    }
59    if let Some(parent) = sock_path.parent() {
60        std::fs::create_dir_all(parent)?;
61    }
62    let listener = UnixListener::bind(&sock_path)?;
63    info!(socket = %sock_path.display(), "nouser-nous-mock escuchando");
64
65    // 2. Sidecar al brahman-init con la Card que declara el socket.
66    let card = build_card(sock_path.clone());
67    info!(label = %card.label, "publicando Card al brahman-init");
68    brahman_sidecar::spawn(card);
69
70    // 3. Accept loop.
71    loop {
72        let (stream, _addr) = listener.accept().await?;
73        tokio::spawn(async move {
74            if let Err(e) = handle_conn(stream).await {
75                warn!(error = %e, "conn falló");
76            }
77        });
78    }
79}
80
81fn init_tracing() {
82    tracing_subscriber::fmt()
83        .with_env_filter(
84            tracing_subscriber::EnvFilter::try_from_default_env()
85                .unwrap_or_else(|_| "info".into()),
86        )
87        .with_target(false)
88        .compact()
89        .init();
90}
91
92/// Card que el mock anuncia al brahman-init. Es kind=Ente (un proceso),
93/// con flujos JSON, bias de prioridad para contexto `test`, y el socket
94/// data-plane declarado en `service_socket` (consumidores lo reciben
95/// directo en el `MatchEvent::Available`).
96fn build_card(service_socket: std::path::PathBuf) -> Card {
97    let mut priority_contexts = BTreeMap::new();
98    priority_contexts.insert(
99        "test".into(),
100        ContextBias {
101            pin_to: None,
102            // En contexto test, este mock gana sobre cualquier real-nous.
103            priority_offset: 1,
104        },
105    );
106
107    Card {
108        schema_version: brahman_card::CARD_SCHEMA_VERSION,
109        id: Ulid::new(),
110        label: "nouser.nous_mock".into(),
111        payload: Payload::Virtual,
112        supervision: Supervision::Delegate,
113        lifecycle: Lifecycle::Daemon,
114        priority: Priority::Normal,
115        kind: CardKind::Ente,
116        service_socket: Some(service_socket),
117        flow: Flows {
118            input: vec![Flow {
119                name: FLOW_EMBED_REQUEST.into(),
120                ty: TypeRef::Primitive {
121                    name: FLOW_TYPE_NAME.into(),
122                },
123                pin_to: None,
124            }],
125            output: vec![Flow {
126                name: FLOW_EMBED_RESULT.into(),
127                ty: TypeRef::Primitive {
128                    name: FLOW_TYPE_NAME.into(),
129                },
130                pin_to: None,
131            }],
132        },
133        priority_contexts,
134        ..Default::default()
135    }
136}
137
138/// Procesa una conexión single-shot: lee una línea JSON, despacha,
139/// escribe una línea JSON, cierra.
140async fn handle_conn(stream: UnixStream) -> std::io::Result<()> {
141    let mut reader = BufReader::new(stream);
142    let mut line = String::new();
143    let n = reader.read_line(&mut line).await?;
144    if n == 0 {
145        return Ok(());
146    }
147    let req: EmbedRequest = match serde_json::from_str(&line) {
148        Ok(r) => r,
149        Err(e) => {
150            return write_error(reader.into_inner(), format!("JSON inválido: {e}")).await;
151        }
152    };
153
154    let started = Instant::now();
155    let result = match req.kind {
156        RequestKind::EmbedFile => handle_embed_file(req.payload, started),
157        RequestKind::EmbedText => handle_embed_text(req.payload, started),
158        RequestKind::Ping => handle_ping(),
159    };
160
161    let mut stream = reader.into_inner();
162    match result {
163        Ok(payload) => {
164            stream.write_all(payload.as_bytes()).await?;
165            stream.write_all(b"\n").await?;
166        }
167        Err(msg) => {
168            return write_error(stream, msg).await;
169        }
170    }
171    stream.shutdown().await?;
172    Ok(())
173}
174
175fn handle_embed_file(payload: serde_json::Value, started: Instant) -> Result<String, String> {
176    let p: EmbedFilePayload =
177        serde_json::from_value(payload).map_err(|e| format!("payload inválido: {e}"))?;
178    info!(path = %p.path, "embed_file");
179
180    let file = FileEntry {
181        id: nouser_card::FileId::from(Ulid::new()),
182        path: PathBuf::from(p.path),
183        content_hash: None,
184        size: p.size,
185        mtime_ms: p.mtime_ms,
186        extension: p.extension,
187    };
188    let v = embed::embed(&file);
189
190    let resp = EmbedResponse {
191        embedding: v.to_vec(),
192        model: MODEL_ID.into(),
193        elapsed_ms: started.elapsed().as_millis() as u64,
194    };
195    serde_json::to_string(&resp).map_err(|e| format!("encode: {e}"))
196}
197
198fn handle_embed_text(payload: serde_json::Value, started: Instant) -> Result<String, String> {
199    let p: EmbedTextPayload =
200        serde_json::from_value(payload).map_err(|e| format!("payload inválido: {e}"))?;
201    info!(text_len = p.text.len(), "embed_text");
202
203    // Mock: tratamos el texto como un "stem" sintético y rellenamos el
204    // resto del vector con ceros. No es semánticamente útil, pero respeta
205    // la forma para que el cliente no se rompa.
206    let synthetic = FileEntry {
207        id: nouser_card::FileId::from(Ulid::new()),
208        path: PathBuf::from(format!("synthetic://{}", p.text)),
209        content_hash: None,
210        size: p.text.len() as u64,
211        mtime_ms: now_ms(),
212        extension: Some("text".into()),
213    };
214    let v = embed::embed(&synthetic);
215
216    let resp = EmbedResponse {
217        embedding: v.to_vec(),
218        model: MODEL_ID.into(),
219        elapsed_ms: started.elapsed().as_millis() as u64,
220    };
221    serde_json::to_string(&resp).map_err(|e| format!("encode: {e}"))
222}
223
224fn handle_ping() -> Result<String, String> {
225    let resp = PingResponse {
226        model: MODEL_ID.into(),
227        embed_dim: embed::EMBED_DIM as u32,
228    };
229    serde_json::to_string(&resp).map_err(|e| format!("encode: {e}"))
230}
231
232async fn write_error(mut stream: UnixStream, msg: String) -> std::io::Result<()> {
233    warn!(error = %msg, "respuesta de error");
234    let resp = ErrorResponse { error: msg };
235    let json = serde_json::to_string(&resp).unwrap_or_else(|_| "{\"error\":\"encode\"}".into());
236    stream.write_all(json.as_bytes()).await?;
237    stream.write_all(b"\n").await?;
238    stream.shutdown().await?;
239    Ok(())
240}
241
242fn now_ms() -> u64 {
243    SystemTime::now()
244        .duration_since(UNIX_EPOCH)
245        .map(|d| d.as_millis() as u64)
246        .unwrap_or(0)
247}