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