sergio/takana
> Una distribución Linux construida con un laboratorio funcional y hermético en el > sótano, y una terminal mutable, clásica y anárquica en el piso de arriba — diseñada > desde el suelo para que una IA programadora entienda, modifique y co…
Fase 5: BuildFailed.log_tail con la cola real del log del sandbox
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 ) {
284 Ok(r) => r,
285 Err(e) => {
286 let _ = tx.send(Event::Error {
287 code: "hydrate_failed".into(),
288 msg: e.to_string(),
289 });
290 return;
291 }
292 };
293 let _ = tx.send(Event::Injected {
294 artifact,
295 target,
296 files: report.files.len(),
297 });
298 }
299 Command::Query { what, path, name, expr } => {
300 let value = match what.as_str() {
301 "file" => {
302 let Some(p) = path else {
303 let _ = tx.send(Event::Error {
304 code: "bad_query".into(),
305 msg: "query 'file' requiere 'path'".into(),
306 });
307 return;
308 };
309 query_file(&p)
310 }
311 "artifact" => {
312 let Some(h) = name.as_deref().or(path.as_deref()) else {
313 let _ = tx.send(Event::Error {
314 code: "bad_query".into(),
315 msg: "query 'artifact' requiere 'name' (el hash)".into(),
316 });
317 return;
318 };
319 query_artifact(&ctx.store_root, h)
320 }
321 "expr" => {
322 let Some(e) = expr.as_deref() else {
323 let _ = tx.send(Event::Error {
324 code: "bad_query".into(),
325 msg: "query 'expr' requiere 'expr' (la expresión)".into(),
326 });
327 return;
328 };
329 // El daemon evalúa con el `fs_root` del sistema host (None ⇒ rutas
330 // absolutas tal y como vienen). Si el .swm necesitara otra base, el
331 // cliente puede evaluar localmente con un EvalContext distinto.
332 hammer_core::query::eval_str(e, &hammer_core::query::EvalContext::new())
333 }
334 other => {
335 let _ = tx.send(Event::Error {
336 code: "unknown_query".into(),
337 msg: format!("'what'='{other}' no soportado"),
338 });
339 return;
340 }
341 };
342 let _ = tx.send(Event::QueryResult { what, value });
343 }
344 Command::Init { cmd } => {
345 match crate::control::send_to_fifo(&ctx.init_control, &cmd) {
346 Ok(()) => {
347 let _ = tx.send(Event::InitAck { cmd });
348 }
349 Err(e) => {
350 let _ = tx.send(Event::Error {
351 code: "init_send_failed".into(),
352 msg: e.to_string(),
353 });
354 }
355 }
356 }
357 }
358}
359
360fn run_compile(recipe: RecipeInline, store_root: PathBuf, tx: Sender<Event>) {
361 let name = recipe.name.clone();
362 let mutation = hammer_core::swm::Mutation::SourcePatch {
363 repo: recipe.repo,
364 commit: recipe.commit,
365 patch: recipe.patch,
366 patch_url: None,
367 build: hammer_core::swm::SwmBuild {
368 compiler: recipe.compiler,
369 target: recipe.target,
370 link: recipe.link,
371 flags: recipe.flags,
372 },
373 target_bin: format!("/usr/bin/{name}"),
374 expected_hash: None,
375 };
376 let store = match hammer_core::Store::open(&store_root) {
377 Ok(s) => s,
378 Err(e) => {
379 let _ = tx.send(Event::BuildFailed {
380 recipe: name,
381 reason: format!("store: {e}"),
382 log_tail: None,
383 });
384 return;
385 }
386 };
387 let cfg = hammer_build::BuildConfig::from_env_or_defaults(store.root());
388 match hammer_build::build_source_patch(&mutation, &cfg, &store, None) {
389 Ok(h) => {
390 let _ = tx.send(Event::BuildReady {
391 recipe: name,
392 artifact: h.as_str().to_string(),
393 });
394 }
395 Err(e) => {
396 // Si el fallo trae la cola del log del lab, la separamos de la razón para que la
397 // IA pueda leer el error real del compilador, no sólo el mensaje de Rust.
398 let (reason, log_tail) = match hammer_build::sandbox::BuildFailure::from_error(&e) {
399 Some(bf) => (bf.reason.clone(), bf.log_tail.clone()),
400 None => (e.to_string(), None),
401 };
402 let _ = tx.send(Event::BuildFailed {
403 recipe: name,
404 reason,
405 log_tail,
406 });
407 }
408 }
409}
410
411fn query_file(path: &str) -> serde_json::Value {
412 let p = Path::new(path);
413 let meta = match std::fs::symlink_metadata(p) {
414 Ok(m) => m,
415 Err(e) => return serde_json::json!({"exists": false, "error": e.to_string()}),
416 };
417 use std::os::unix::fs::MetadataExt;
418 serde_json::json!({
419 "exists": true,
420 "size": meta.size(),
421 "mode": meta.mode(),
422 "uid": meta.uid(),
423 "gid": meta.gid(),
424 "is_dir": meta.is_dir(),
425 "is_symlink": meta.file_type().is_symlink(),
426 })
427}
428
429fn query_artifact(store_root: &Path, hash: &str) -> serde_json::Value {
430 let store = match hammer_core::Store::open(store_root) {
431 Ok(s) => s,
432 Err(e) => return serde_json::json!({"error": e.to_string()}),
433 };
434 match store.find_by_hash(hash) {
435 Ok(p) => serde_json::json!({"exists": true, "path": p.display().to_string()}),
436 Err(_) => serde_json::json!({"exists": false}),
437 }
438}
439
440fn t_of(cmd: &Command) -> &'static str {
441 match cmd {
442 Command::Hello { .. } => "hello",
443 Command::Compile { .. } => "compile",
444 Command::Inject { .. } => "inject",
445 Command::Query { .. } => "query",
446 Command::Init { .. } => "init",
447 }
448}
449
450// Re-export para que el caller no tenga que importar Mutex/Arc directamente.
451#[allow(dead_code)]
452pub(crate) type SharedCaps = Arc<Mutex<std::collections::BTreeSet<Cap>>>;
453
454#[cfg(test)]
455mod tests {
456 use super::*;
457
458 #[test]
459 fn default_policy_grants_query_to_others() {
460 let p = default_policy();
461 // Un peer con UID distinto al actual debe quedar como sólo-query.
462 let foreign = Peer {
463 uid: u32::MAX,
464 gid: 0,
465 pid: 1,
466 };
467 assert_eq!(p(&foreign), vec![Cap::Query]);
468 }
469
470 #[test]
471 fn config_policy_resolves_by_uid_gid() {
472 let cfg = hammer_core::AgentCapsConfig::from_toml(
473 r#"
474default = ["query"]
475[[rule]]
476uid = 0
477caps = ["query", "compile", "inject", "inject-real", "init"]
478"#,
479 )
480 .unwrap();
481 let p = policy_from_config(cfg);
482 let root = Peer { uid: 0, gid: 0, pid: 1 };
483 let other = Peer { uid: 1234, gid: 1234, pid: 2 };
484 assert!(p(&root).contains(&Cap::InjectReal));
485 assert_eq!(p(&other), vec![Cap::Query]);
486 }
487
488 #[test]
489 fn t_of_matches_all_commands() {
490 // No es exhaustivo de Cargo, pero verifica que no nos olvidemos del nombre del tag.
491 assert_eq!(t_of(&Command::Hello { ver: 1, client: "".into() }), "hello");
492 assert_eq!(
493 t_of(&Command::Compile {
494 recipe: RecipeInline {
495 name: "x".into(),
496 repo: "g".into(),
497 commit: "c".into(),
498 patch: None,
499 compiler: "zig-cc".into(),
500 target: "t".into(),
501 link: "static".into(),
502 flags: vec![],
503 },
504 }),
505 "compile"
506 );
507 }
508
509 #[test]
510 fn query_file_reports_missing() {
511 let v = query_file("/no/existe/seguramente/ahora");
512 assert_eq!(v["exists"], serde_json::json!(false));
513 }
514}