sergio/takana

1//! `AgentClient`: cliente síncrono del bus de agente. Ver `docs/07-agent-bus.md` §3.
2//!
3//! Modelo: una conexión = un thread del que llama. El reader thread interno demultiplexa
4//! eventos por tipo: respuestas correlativas a un comando se devuelven al caller via
5//! `recv_*_event_blocking`; eventos asíncronos (`Modified`, `Crashed`) se acumulan en una
6//! cola que el caller drena con `drain_async()`.
7//!
8//! No hay correlation-id en el protocolo todavía (el SDD 07 no lo exige), así que la
9//! correlación se hace por **tipo**: tras un `Compile` esperamos `BuildReady|BuildFailed`,
10//! cualquier otra cosa que no sea `Modified`/`Crashed` se reporta como protocol error.
11
12use std::collections::VecDeque;
13use std::io::{BufRead, BufReader, Write};
14use std::os::unix::net::UnixStream;
15use std::path::Path;
16use std::sync::{Arc, Condvar, Mutex};
17use std::thread;
18use std::time::{Duration, Instant};
19
20use takana_core::proto::{Cap, Command, Event, Peer, RecipeInline};
21
22#[derive(Debug, thiserror::Error)]
23pub enum ClientError {
24    #[error("io: {0}")]
25    Io(#[from] std::io::Error),
26    #[error("json: {0}")]
27    Json(#[from] serde_json::Error),
28    #[error("protocolo: {0}")]
29    Protocol(String),
30    #[error("timeout esperando {0}")]
31    Timeout(&'static str),
32    #[error("el bus devolvió error code={code} msg={msg}")]
33    Bus { code: String, msg: String },
34    #[error("la conexión se cerró")]
35    Closed,
36}
37
38pub type Result<T> = std::result::Result<T, ClientError>;
39
40#[derive(Debug, Clone)]
41pub struct Welcome {
42    pub ver: u32,
43    pub caps: Vec<Cap>,
44    pub peer: Peer,
45}
46
47/// Resultado de `compile`: el hash del artefacto sellado más el veredicto de la evidencia (H1c).
48/// `verdict` es `None` cuando la receta no declaró evidencia.
49#[derive(Debug, Clone)]
50pub struct CompileOutcome {
51    pub artifact: String,
52    pub verdict: Option<takana_core::proto::EvidenceVerdict>,
53}
54
55/// Estado compartido entre el reader thread y el caller.
56struct Inbox {
57    /// Eventos asíncronos (Modified/Crashed) en orden de llegada.
58    async_events: VecDeque<Event>,
59    /// Próximo evento "respuesta" al último comando. Sólo cabe uno a la vez en este
60    /// cliente síncrono (el caller envía + espera antes del siguiente).
61    pending: Option<Event>,
62    closed: bool,
63}
64
65#[derive(Clone)]
66struct Shared {
67    inbox: Arc<Mutex<Inbox>>,
68    cv: Arc<Condvar>,
69}
70
71pub struct AgentClient {
72    writer: UnixStream,
73    shared: Shared,
74    pub welcome: Welcome,
75}
76
77impl AgentClient {
78    /// Conecta al socket, hace handshake `Hello` y arranca el reader thread. Devuelve un
79    /// cliente listo para enviar comandos.
80    pub fn connect(sock_path: &Path) -> Result<Self> {
81        Self::connect_with_client_name(sock_path, "hammer-agent")
82    }
83
84    pub fn connect_with_client_name(sock_path: &Path, client: &str) -> Result<Self> {
85        let stream = UnixStream::connect(sock_path)?;
86        let reader_stream = stream.try_clone()?;
87        let writer = stream;
88
89        let shared = Shared {
90            inbox: Arc::new(Mutex::new(Inbox {
91                async_events: VecDeque::new(),
92                pending: None,
93                closed: false,
94            })),
95            cv: Arc::new(Condvar::new()),
96        };
97        // Reader thread: drena el socket línea a línea hasta EOF.
98        let s_for_reader = shared.clone();
99        thread::Builder::new()
100            .name("agent-client-reader".into())
101            .spawn(move || reader_loop(reader_stream, s_for_reader))
102            .ok();
103
104        let mut me = Self {
105            writer,
106            shared,
107            welcome: Welcome {
108                ver: 0,
109                caps: vec![],
110                peer: Peer { uid: 0, gid: 0, pid: 0 },
111            },
112        };
113        me.send(&Command::Hello {
114            ver: takana_core::proto::PROTOCOL_VERSION,
115            client: client.to_string(),
116        })?;
117        let ev = me.recv_pending(Duration::from_secs(5))?;
118        match ev {
119            Event::Welcome { ver, caps, peer } => {
120                me.welcome = Welcome { ver, caps, peer };
121            }
122            Event::Error { code, msg } => return Err(ClientError::Bus { code, msg }),
123            other => {
124                return Err(ClientError::Protocol(format!(
125                    "esperaba Welcome, llegó {other:?}"
126                )));
127            }
128        }
129        Ok(me)
130    }
131
132    /// Bloquea hasta `BuildReady` o `BuildFailed`. `timeout` es el wall-clock máximo —
133    /// el lab puede tardar varios minutos, así que el default debería ser generoso.
134    pub fn compile(&mut self, recipe: RecipeInline, timeout: Duration) -> Result<CompileOutcome> {
135        self.send(&Command::Compile { recipe })?;
136        match self.recv_pending(timeout)? {
137            Event::BuildReady { artifact, verdict, .. } => Ok(CompileOutcome { artifact, verdict }),
138            Event::BuildFailed { reason, .. } => Err(ClientError::Bus {
139                code: "build_failed".into(),
140                msg: reason,
141            }),
142            Event::Error { code, msg } => Err(ClientError::Bus { code, msg }),
143            other => Err(ClientError::Protocol(format!(
144                "esperaba BuildReady|BuildFailed, llegó {other:?}"
145            ))),
146        }
147    }
148
149    pub fn inject(
150        &mut self,
151        artifact: &str,
152        target: &str,
153        overlay: Option<&str>,
154        timeout: Duration,
155    ) -> Result<usize> {
156        self.send(&Command::Inject {
157            artifact: artifact.to_string(),
158            target: target.to_string(),
159            overlay: overlay.map(String::from),
160        })?;
161        match self.recv_pending(timeout)? {
162            Event::Injected { files, .. } => Ok(files),
163            Event::Error { code, msg } => Err(ClientError::Bus { code, msg }),
164            other => Err(ClientError::Protocol(format!("esperaba Injected, llegó {other:?}"))),
165        }
166    }
167
168    pub fn query_file(&mut self, path: &str, timeout: Duration) -> Result<serde_json::Value> {
169        self.send(&Command::Query {
170            what: "file".into(),
171            path: Some(path.to_string()),
172            name: None,
173            expr: None,
174        })?;
175        match self.recv_pending(timeout)? {
176            Event::QueryResult { value, .. } => Ok(value),
177            Event::Error { code, msg } => Err(ClientError::Bus { code, msg }),
178            other => Err(ClientError::Protocol(format!(
179                "esperaba QueryResult, llegó {other:?}"
180            ))),
181        }
182    }
183
184    /// Evalúa una expresión del mini-lenguaje (`bin:grep`, `pin:musl`, …) en el daemon.
185    /// El JSON devuelto sigue el schema de `takana_core::query::eval`.
186    pub fn query_expr(&mut self, expr: &str, timeout: Duration) -> Result<serde_json::Value> {
187        self.send(&Command::Query {
188            what: "expr".into(),
189            path: None,
190            name: None,
191            expr: Some(expr.to_string()),
192        })?;
193        match self.recv_pending(timeout)? {
194            Event::QueryResult { value, .. } => Ok(value),
195            Event::Error { code, msg } => Err(ClientError::Bus { code, msg }),
196            other => Err(ClientError::Protocol(format!(
197                "esperaba QueryResult, llegó {other:?}"
198            ))),
199        }
200    }
201
202    pub fn init(&mut self, cmd: &str, timeout: Duration) -> Result<()> {
203        self.send(&Command::Init { cmd: cmd.to_string() })?;
204        match self.recv_pending(timeout)? {
205            Event::InitAck { .. } => Ok(()),
206            Event::Error { code, msg } => Err(ClientError::Bus { code, msg }),
207            other => Err(ClientError::Protocol(format!("esperaba InitAck, llegó {other:?}"))),
208        }
209    }
210
211    /// Vacía la cola de eventos asíncronos. No bloquea.
212    pub fn drain_async(&self) -> Vec<Event> {
213        let Ok(mut g) = self.shared.inbox.lock() else { return vec![] };
214        std::mem::take(&mut g.async_events).into_iter().collect()
215    }
216
217    /// Espera el próximo evento asíncrono (típicamente para verificar que un `Modified`
218    /// específico haya llegado tras una operación).
219    pub fn next_async(&self, timeout: Duration) -> Result<Event> {
220        let g = self.shared.inbox.lock().unwrap();
221        let (mut g, _) = self
222            .shared
223            .cv
224            .wait_timeout_while(g, timeout, |i| {
225                !i.closed && i.async_events.is_empty()
226            })
227            .unwrap();
228        if let Some(ev) = g.async_events.pop_front() {
229            Ok(ev)
230        } else if g.closed {
231            Err(ClientError::Closed)
232        } else {
233            Err(ClientError::Timeout("async event"))
234        }
235    }
236
237    fn send(&mut self, cmd: &Command) -> Result<()> {
238        let line = serde_json::to_string(cmd)?;
239        self.writer.write_all(line.as_bytes())?;
240        self.writer.write_all(b"\n")?;
241        self.writer.flush()?;
242        Ok(())
243    }
244
245    fn recv_pending(&self, timeout: Duration) -> Result<Event> {
246        let deadline = Instant::now() + timeout;
247        let g = self.shared.inbox.lock().unwrap();
248        let (mut g, _) = self
249            .shared
250            .cv
251            .wait_timeout_while(g, timeout, |i| !i.closed && i.pending.is_none())
252            .unwrap();
253        if let Some(ev) = g.pending.take() {
254            return Ok(ev);
255        }
256        if g.closed {
257            return Err(ClientError::Closed);
258        }
259        // Si el wait expiró exactamente en el momento de un evento, vuelvo a comprobar.
260        if Instant::now() >= deadline {
261            return Err(ClientError::Timeout("response"));
262        }
263        Err(ClientError::Timeout("response"))
264    }
265}
266
267fn reader_loop(stream: UnixStream, shared: Shared) {
268    let reader = BufReader::new(stream);
269    for line in reader.lines() {
270        let line = match line {
271            Ok(l) => l,
272            Err(e) => {
273                tracing::warn!(error = %e, "agent-client: read error, cerrando");
274                break;
275            }
276        };
277        if line.trim().is_empty() {
278            continue;
279        }
280        let ev: Event = match serde_json::from_str(&line) {
281            Ok(e) => e,
282            Err(e) => {
283                tracing::warn!(error = %e, line = %line, "agent-client: línea no parseable");
284                continue;
285            }
286        };
287        let is_async = matches!(
288            &ev,
289            Event::Modified { .. } | Event::Crashed { .. }
290        );
291        let mut g = match shared.inbox.lock() {
292            Ok(g) => g,
293            Err(_) => break,
294        };
295        if is_async {
296            g.async_events.push_back(ev);
297        } else {
298            // Si ya hay un pending sin recoger, eso es un bug del caller (envió dos
299            // comandos antes de leer la primera respuesta). Lo logueamos y reemplazamos.
300            if g.pending.is_some() {
301                tracing::warn!(
302                    "agent-client: pending ya tenía respuesta; el caller no la recogió"
303                );
304            }
305            g.pending = Some(ev);
306        }
307        shared.cv.notify_all();
308    }
309    if let Ok(mut g) = shared.inbox.lock() {
310        g.closed = true;
311        shared.cv.notify_all();
312    }
313}