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