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        })?;
166        match self.recv_pending(timeout)? {
167            Event::QueryResult { value, .. } => Ok(value),
168            Event::Error { code, msg } => Err(ClientError::Bus { code, msg }),
169            other => Err(ClientError::Protocol(format!(
170                "esperaba QueryResult, llegó {other:?}"
171            ))),
172        }
173    }
174
175    pub fn init(&mut self, cmd: &str, timeout: Duration) -> Result<()> {
176        self.send(&Command::Init { cmd: cmd.to_string() })?;
177        match self.recv_pending(timeout)? {
178            Event::InitAck { .. } => Ok(()),
179            Event::Error { code, msg } => Err(ClientError::Bus { code, msg }),
180            other => Err(ClientError::Protocol(format!("esperaba InitAck, llegó {other:?}"))),
181        }
182    }
183
184    /// Vacía la cola de eventos asíncronos. No bloquea.
185    pub fn drain_async(&self) -> Vec<Event> {
186        let Ok(mut g) = self.shared.inbox.lock() else { return vec![] };
187        std::mem::take(&mut g.async_events).into_iter().collect()
188    }
189
190    /// Espera el próximo evento asíncrono (típicamente para verificar que un `Modified`
191    /// específico haya llegado tras una operación).
192    pub fn next_async(&self, timeout: Duration) -> Result<Event> {
193        let g = self.shared.inbox.lock().unwrap();
194        let (mut g, _) = self
195            .shared
196            .cv
197            .wait_timeout_while(g, timeout, |i| {
198                !i.closed && i.async_events.is_empty()
199            })
200            .unwrap();
201        if let Some(ev) = g.async_events.pop_front() {
202            Ok(ev)
203        } else if g.closed {
204            Err(ClientError::Closed)
205        } else {
206            Err(ClientError::Timeout("async event"))
207        }
208    }
209
210    fn send(&mut self, cmd: &Command) -> Result<()> {
211        let line = serde_json::to_string(cmd)?;
212        self.writer.write_all(line.as_bytes())?;
213        self.writer.write_all(b"\n")?;
214        self.writer.flush()?;
215        Ok(())
216    }
217
218    fn recv_pending(&self, timeout: Duration) -> Result<Event> {
219        let deadline = Instant::now() + timeout;
220        let g = self.shared.inbox.lock().unwrap();
221        let (mut g, _) = self
222            .shared
223            .cv
224            .wait_timeout_while(g, timeout, |i| !i.closed && i.pending.is_none())
225            .unwrap();
226        if let Some(ev) = g.pending.take() {
227            return Ok(ev);
228        }
229        if g.closed {
230            return Err(ClientError::Closed);
231        }
232        // Si el wait expiró exactamente en el momento de un evento, vuelvo a comprobar.
233        if Instant::now() >= deadline {
234            return Err(ClientError::Timeout("response"));
235        }
236        Err(ClientError::Timeout("response"))
237    }
238}
239
240fn reader_loop(stream: UnixStream, shared: Shared) {
241    let reader = BufReader::new(stream);
242    for line in reader.lines() {
243        let line = match line {
244            Ok(l) => l,
245            Err(e) => {
246                tracing::warn!(error = %e, "agent-client: read error, cerrando");
247                break;
248            }
249        };
250        if line.trim().is_empty() {
251            continue;
252        }
253        let ev: Event = match serde_json::from_str(&line) {
254            Ok(e) => e,
255            Err(e) => {
256                tracing::warn!(error = %e, line = %line, "agent-client: línea no parseable");
257                continue;
258            }
259        };
260        let is_async = matches!(
261            &ev,
262            Event::Modified { .. } | Event::Crashed { .. }
263        );
264        let mut g = match shared.inbox.lock() {
265            Ok(g) => g,
266            Err(_) => break,
267        };
268        if is_async {
269            g.async_events.push_back(ev);
270        } else {
271            // Si ya hay un pending sin recoger, eso es un bug del caller (envió dos
272            // comandos antes de leer la primera respuesta). Lo logueamos y reemplazamos.
273            if g.pending.is_some() {
274                tracing::warn!(
275                    "agent-client: pending ya tenía respuesta; el caller no la recogió"
276                );
277            }
278            g.pending = Some(ev);
279        }
280        shared.cv.notify_all();
281    }
282    if let Ok(mut g) = shared.inbox.lock() {
283        g.closed = true;
284        shared.cv.notify_all();
285    }
286}