sergio/takana

1//! Bus de agente: `/run/agent.sock`. Ver `docs/07-agent-bus.md`.
2//!
3//! Modelo de threading (síncrono, una conexión = N threads pequeños):
4//!
5//! ```text
6//!   [accept loop] ── spawn ──▶ conn{
7//!        [reader thread] ── dispatch ──┐
8//!        [bus forwarder]  ─────────────┼──▶ tx ──▶ [writer thread] ──▶ socket
9//!        [compile worker] (puntual) ───┘
10//!   }
11//! ```
12//!
13//! No usamos `tokio` para no arrastrar runtime asíncrono: la cadencia esperada del bus es
14//! humana + agente (decenas de mensajes/seg en el peor caso). Cada conexión tiene su propio
15//! canal `Sender<Event>`; la `EventBus` global re-emite Modified/Crashed a esos canales.
16
17use std::io::{BufRead, BufReader, Write};
18use std::os::fd::AsFd;
19use std::os::unix::net::{UnixListener, UnixStream};
20use std::path::{Path, PathBuf};
21use std::sync::mpsc::{self, Sender};
22use std::sync::{Arc, Mutex};
23use std::thread;
24
25use nix::sys::socket::{getsockopt, sockopt::PeerCredentials};
26
27use crate::events::EventBus;
28use hammer_core::proto::{Cap, Command, Event, Peer, RecipeInline, PROTOCOL_VERSION};
29
30/// Política de capacidades: dado el `Peer` (ya autenticado por SO_PEERCRED), devuelve qué
31/// caps tiene esa conexión. Por defecto: misma UID que el daemon ⇒ todo menos `inject-real`;
32/// otros UIDs ⇒ sólo `query`. La policy más expresiva (lectura de un toml en `/etc/hammer`)
33/// es trabajo posterior; la firma del callable basta para encajarla sin tocar este módulo.
34pub type CapsPolicy = Arc<dyn Fn(&Peer) -> Vec<Cap> + Send + Sync>;
35
36pub fn default_policy() -> CapsPolicy {
37    let my_uid = unsafe { libc::getuid() };
38    Arc::new(move |peer: &Peer| {
39        if peer.uid == my_uid {
40            vec![Cap::Query, Cap::Compile, Cap::Inject, Cap::Init]
41        } else {
42            vec![Cap::Query]
43        }
44    })
45}
46
47/// Política derivada de un fichero `agent-caps.toml` (SDD 07 §4). El peer no elige sus caps;
48/// las resuelve la config por `(uid, gid)`.
49pub fn policy_from_config(cfg: hammer_core::AgentCapsConfig) -> CapsPolicy {
50    Arc::new(move |peer: &Peer| cfg.caps_for(peer.uid, peer.gid))
51}
52
53/// Contexto que necesita el dispatcher para resolver comandos. Compartido por todas las
54/// conexiones (clones baratos: paths + Arc).
55#[derive(Clone)]
56pub struct BusContext {
57    pub store_root: PathBuf,
58    pub init_control: PathBuf,
59    pub events: EventBus,
60}
61
62#[derive(Debug, thiserror::Error)]
63pub enum Error {
64    #[error("io: {0}")]
65    Io(#[from] std::io::Error),
66    #[error("bind {0}: {1}")]
67    Bind(PathBuf, std::io::Error),
68    #[error("peer creds: {0}")]
69    Peer(String),
70}
71
72pub type Result<T> = std::result::Result<T, Error>;
73
74/// Arranca el listener bloqueante. Pensado para correr en su propio thread desde `main`.
75pub fn serve_agent_bus(
76    sock_path: &Path,
77    policy: CapsPolicy,
78    ctx: BusContext,
79) -> Result<()> {
80    // Limpia un socket olvidado de una ejecución previa. Si el path está ocupado por algo
81    // que no es un socket, dejamos que `bind` falle — no queremos rm-rf por accidente.
82    let _ = std::fs::remove_file(sock_path);
83    let listener = UnixListener::bind(sock_path)
84        .map_err(|e| Error::Bind(sock_path.to_path_buf(), e))?;
85    tracing::info!(sock = %sock_path.display(), "bus: escuchando");
86
87    for stream in listener.incoming() {
88        match stream {
89            Ok(s) => {
90                let policy = policy.clone();
91                let ctx = ctx.clone();
92                thread::Builder::new()
93                    .name("bus-conn".into())
94                    .spawn(move || {
95                        if let Err(e) = handle_connection(s, policy, ctx) {
96                            tracing::warn!(error = %e, "bus: conexión terminó con error");
97                        }
98                    })
99                    .ok();
100            }
101            Err(e) => tracing::warn!(error = %e, "bus: accept falló"),
102        }
103    }
104    Ok(())
105}
106
107fn read_peer(stream: &UnixStream) -> Result<Peer> {
108    let cred = getsockopt(&stream.as_fd(), PeerCredentials)
109        .map_err(|e| Error::Peer(e.to_string()))?;
110    Ok(Peer {
111        uid: cred.uid(),
112        gid: cred.gid(),
113        pid: cred.pid(),
114    })
115}
116
117fn handle_connection(stream: UnixStream, policy: CapsPolicy, ctx: BusContext) -> Result<()> {
118    let peer = read_peer(&stream)?;
119    let caps = policy(&peer);
120    tracing::info!(?peer, ?caps, "bus: nueva conexión");
121
122    // Canal de eventos hacia el writer. Todos los productores (dispatcher, bus forwarder,
123    // workers de COMPILE) clonan este Sender.
124    let (tx, rx) = mpsc::channel::<Event>();
125
126    // Writer thread: serializa todo lo que recibe del canal y lo escribe al socket.
127    let stream_writer = stream.try_clone()?;
128    let _writer_handle = thread::Builder::new()
129        .name("bus-conn-writer".into())
130        .spawn(move || writer_loop(stream_writer, rx))
131        .ok();
132
133    // Bus forwarder thread: cualquier evento global (Modified/Crashed) se re-emite a este
134    // peer. Se cierra solo cuando el global drop el último Sender (en la práctica, nunca)
135    // o cuando nuestro tx local cae (peer cerró).
136    let bus_rx = ctx.events.subscribe();
137    let bus_tx = tx.clone();
138    thread::Builder::new()
139        .name("bus-conn-fwd".into())
140        .spawn(move || {
141            for ev in bus_rx {
142                if bus_tx.send(ev).is_err() {
143                    break;
144                }
145            }
146        })
147        .ok();
148
149    // Handshake. La primera línea debe ser Hello.
150    let mut reader = BufReader::new(stream);
151    let mut first = String::new();
152    reader.read_line(&mut first)?;
153    match serde_json::from_str::<Command>(first.trim()) {
154        Ok(Command::Hello { ver, client }) => {
155            tracing::info!(ver, client = %client, "bus: hello");
156            let _ = tx.send(Event::Welcome {
157                ver: PROTOCOL_VERSION,
158                caps: caps.clone(),
159                peer: peer.clone(),
160            });
161        }
162        Ok(other) => {
163            let _ = tx.send(Event::Error {
164                code: "no_hello".into(),
165                msg: format!("se esperaba 'hello', llegó '{}'", t_of(&other)),
166            });
167            return Ok(());
168        }
169        Err(e) => {
170            let _ = tx.send(Event::Error {
171                code: "bad_handshake".into(),
172                msg: format!("hello inválido: {e}"),
173            });
174            return Ok(());
175        }
176    }
177
178    // Read loop. Cada línea es un Command. La conexión se cierra cuando se cierra EOF, o
179    // cuando un error de IO lo fuerza.
180    let caps_set: std::collections::BTreeSet<Cap> = caps.iter().copied().collect();
181    let caps_set = Arc::new(caps_set);
182    for line in reader.lines() {
183        let line = match line {
184            Ok(l) => l,
185            Err(_) => break,
186        };
187        if line.trim().is_empty() {
188            continue;
189        }
190        let cmd: Command = match serde_json::from_str(&line) {
191            Ok(c) => c,
192            Err(e) => {
193                let _ = tx.send(Event::Error {
194                    code: "bad_command".into(),
195                    msg: format!("JSON inválido: {e}"),
196                });
197                continue;
198            }
199        };
200        dispatch(cmd, &tx, &caps_set, &ctx);
201    }
202    Ok(())
203}
204
205fn writer_loop(mut stream: UnixStream, rx: mpsc::Receiver<Event>) {
206    for ev in rx {
207        let line = match serde_json::to_string(&ev) {
208            Ok(s) => s,
209            Err(e) => {
210                tracing::error!(error = %e, "bus: serializando event");
211                continue;
212            }
213        };
214        if stream.write_all(line.as_bytes()).is_err() {
215            break;
216        }
217        if stream.write_all(b"\n").is_err() {
218            break;
219        }
220        // El bus es de baja cadencia; flush por mensaje no es relevante en costes y mejora
221        // mucho la latencia percibida del cliente.
222        let _ = stream.flush();
223    }
224}
225
226fn dispatch(
227    cmd: Command,
228    tx: &Sender<Event>,
229    caps: &Arc<std::collections::BTreeSet<Cap>>,
230    ctx: &BusContext,
231) {
232    if let Some(needed) = Cap::required_for(&cmd) {
233        if !caps.contains(&needed) {
234            let _ = tx.send(Event::Error {
235                code: "no_cap".into(),
236                msg: format!("comando requiere capacidad '{needed:?}', no concedida"),
237            });
238            return;
239        }
240    }
241    match cmd {
242        Command::Hello { .. } => {
243            // Doble hello: lo ignoramos con un Error suave.
244            let _ = tx.send(Event::Error {
245                code: "duplicate_hello".into(),
246                msg: "ya hicimos hello en esta conexión".into(),
247            });
248        }
249        Command::Compile { recipe } => {
250            // Build pesado: a un worker. El read-loop sigue libre para más comandos.
251            let tx = tx.clone();
252            let store_root = ctx.store_root.clone();
253            thread::Builder::new()
254                .name("bus-compile".into())
255                .spawn(move || run_compile(recipe, store_root, tx))
256                .ok();
257        }
258        Command::Inject { artifact, target, overlay: _ } => {
259            let store = match hammer_core::Store::open(&ctx.store_root) {
260                Ok(s) => s,
261                Err(e) => {
262                    let _ = tx.send(Event::Error {
263                        code: "store".into(),
264                        msg: e.to_string(),
265                    });
266                    return;
267                }
268            };
269            let artifact_dir = match store.find_by_hash(&artifact) {
270                Ok(p) => p,
271                Err(e) => {
272                    let _ = tx.send(Event::Error {
273                        code: "not_found".into(),
274                        msg: e.to_string(),
275                    });
276                    return;
277                }
278            };
279            let report = match hammer_build::run_hydrate(
280                &artifact_dir,
281                Path::new(&target),
282                hammer_core::LinkMode::Static,
283                None,
284            ) {
285                Ok(r) => r,
286                Err(e) => {
287                    let _ = tx.send(Event::Error {
288                        code: "hydrate_failed".into(),
289                        msg: e.to_string(),
290                    });
291                    return;
292                }
293            };
294            let _ = tx.send(Event::Injected {
295                artifact,
296                target,
297                files: report.files.len(),
298            });
299        }
300        Command::Query { what, path, name, expr } => {
301            let value = match what.as_str() {
302                "file" => {
303                    let Some(p) = path else {
304                        let _ = tx.send(Event::Error {
305                            code: "bad_query".into(),
306                            msg: "query 'file' requiere 'path'".into(),
307                        });
308                        return;
309                    };
310                    query_file(&p)
311                }
312                "artifact" => {
313                    let Some(h) = name.as_deref().or(path.as_deref()) else {
314                        let _ = tx.send(Event::Error {
315                            code: "bad_query".into(),
316                            msg: "query 'artifact' requiere 'name' (el hash)".into(),
317                        });
318                        return;
319                    };
320                    query_artifact(&ctx.store_root, h)
321                }
322                "expr" => {
323                    let Some(e) = expr.as_deref() else {
324                        let _ = tx.send(Event::Error {
325                            code: "bad_query".into(),
326                            msg: "query 'expr' requiere 'expr' (la expresión)".into(),
327                        });
328                        return;
329                    };
330                    // El daemon evalúa con el `fs_root` del sistema host (None ⇒ rutas
331                    // absolutas tal y como vienen). Si el .swm necesitara otra base, el
332                    // cliente puede evaluar localmente con un EvalContext distinto.
333                    hammer_core::query::eval_str(e, &hammer_core::query::EvalContext::new())
334                }
335                other => {
336                    let _ = tx.send(Event::Error {
337                        code: "unknown_query".into(),
338                        msg: format!("'what'='{other}' no soportado"),
339                    });
340                    return;
341                }
342            };
343            let _ = tx.send(Event::QueryResult { what, value });
344        }
345        Command::Init { cmd } => {
346            match crate::control::send_to_fifo(&ctx.init_control, &cmd) {
347                Ok(()) => {
348                    let _ = tx.send(Event::InitAck { cmd });
349                }
350                Err(e) => {
351                    let _ = tx.send(Event::Error {
352                        code: "init_send_failed".into(),
353                        msg: e.to_string(),
354                    });
355                }
356            }
357        }
358    }
359}
360
361fn run_compile(recipe: RecipeInline, store_root: PathBuf, tx: Sender<Event>) {
362    let name = recipe.name.clone();
363    let mutation = hammer_core::swm::Mutation::SourcePatch {
364        repo: recipe.repo,
365        commit: recipe.commit,
366        tarball: recipe.tarball,
367        sha256: recipe.sha256,
368        patch: recipe.patch,
369        patch_url: None,
370        build: hammer_core::swm::SwmBuild {
371            compiler: recipe.compiler,
372            target: recipe.target,
373            link: recipe.link,
374            flags: recipe.flags,
375        },
376        target_bin: format!("/usr/bin/{name}"),
377        expected_hash: None,
378    };
379    let store = match hammer_core::Store::open(&store_root) {
380        Ok(s) => s,
381        Err(e) => {
382            let _ = tx.send(Event::BuildFailed {
383                recipe: name,
384                reason: format!("store: {e}"),
385                log_tail: None,
386            });
387            return;
388        }
389    };
390    let cfg = hammer_build::BuildConfig::from_env_or_defaults(store.root());
391    match hammer_build::build_source_patch(&mutation, &cfg, &store, None) {
392        Ok(h) => {
393            let _ = tx.send(Event::BuildReady {
394                recipe: name,
395                artifact: h.as_str().to_string(),
396            });
397        }
398        Err(e) => {
399            // Si el fallo trae la cola del log del lab, la separamos de la razón para que la
400            // IA pueda leer el error real del compilador, no sólo el mensaje de Rust.
401            let (reason, log_tail) = match hammer_build::sandbox::BuildFailure::from_error(&e) {
402                Some(bf) => (bf.reason.clone(), bf.log_tail.clone()),
403                None => (e.to_string(), None),
404            };
405            let _ = tx.send(Event::BuildFailed {
406                recipe: name,
407                reason,
408                log_tail,
409            });
410        }
411    }
412}
413
414fn query_file(path: &str) -> serde_json::Value {
415    let p = Path::new(path);
416    let meta = match std::fs::symlink_metadata(p) {
417        Ok(m) => m,
418        Err(e) => return serde_json::json!({"exists": false, "error": e.to_string()}),
419    };
420    use std::os::unix::fs::MetadataExt;
421    serde_json::json!({
422        "exists": true,
423        "size": meta.size(),
424        "mode": meta.mode(),
425        "uid": meta.uid(),
426        "gid": meta.gid(),
427        "is_dir": meta.is_dir(),
428        "is_symlink": meta.file_type().is_symlink(),
429    })
430}
431
432fn query_artifact(store_root: &Path, hash: &str) -> serde_json::Value {
433    let store = match hammer_core::Store::open(store_root) {
434        Ok(s) => s,
435        Err(e) => return serde_json::json!({"error": e.to_string()}),
436    };
437    match store.find_by_hash(hash) {
438        Ok(p) => serde_json::json!({"exists": true, "path": p.display().to_string()}),
439        Err(_) => serde_json::json!({"exists": false}),
440    }
441}
442
443fn t_of(cmd: &Command) -> &'static str {
444    match cmd {
445        Command::Hello { .. } => "hello",
446        Command::Compile { .. } => "compile",
447        Command::Inject { .. } => "inject",
448        Command::Query { .. } => "query",
449        Command::Init { .. } => "init",
450    }
451}
452
453// Re-export para que el caller no tenga que importar Mutex/Arc directamente.
454#[allow(dead_code)]
455pub(crate) type SharedCaps = Arc<Mutex<std::collections::BTreeSet<Cap>>>;
456
457#[cfg(test)]
458mod tests {
459    use super::*;
460
461    #[test]
462    fn default_policy_grants_query_to_others() {
463        let p = default_policy();
464        // Un peer con UID distinto al actual debe quedar como sólo-query.
465        let foreign = Peer {
466            uid: u32::MAX,
467            gid: 0,
468            pid: 1,
469        };
470        assert_eq!(p(&foreign), vec![Cap::Query]);
471    }
472
473    #[test]
474    fn config_policy_resolves_by_uid_gid() {
475        let cfg = hammer_core::AgentCapsConfig::from_toml(
476            r#"
477default = ["query"]
478[[rule]]
479uid = 0
480caps = ["query", "compile", "inject", "inject-real", "init"]
481"#,
482        )
483        .unwrap();
484        let p = policy_from_config(cfg);
485        let root = Peer { uid: 0, gid: 0, pid: 1 };
486        let other = Peer { uid: 1234, gid: 1234, pid: 2 };
487        assert!(p(&root).contains(&Cap::InjectReal));
488        assert_eq!(p(&other), vec![Cap::Query]);
489    }
490
491    #[test]
492    fn t_of_matches_all_commands() {
493        // No es exhaustivo de Cargo, pero verifica que no nos olvidemos del nombre del tag.
494        assert_eq!(t_of(&Command::Hello { ver: 1, client: "".into() }), "hello");
495        assert_eq!(
496            t_of(&Command::Compile {
497                recipe: RecipeInline {
498                    name: "x".into(),
499                    repo: Some("g".into()),
500                    commit: Some("c".into()),
501                    tarball: None,
502                    sha256: None,
503                    patch: None,
504                    compiler: "zig-cc".into(),
505                    target: "t".into(),
506                    link: "static".into(),
507                    flags: vec![],
508                },
509            }),
510            "compile"
511        );
512    }
513
514    #[test]
515    fn query_file_reports_missing() {
516        let v = query_file("/no/existe/seguramente/ahora");
517        assert_eq!(v["exists"], serde_json::json!(false));
518    }
519}