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        patch: recipe.patch,
373        patch_url: None,
374        build: takana_core::swm::SwmBuild {
375            compiler: recipe.compiler,
376            target: recipe.target,
377            link: recipe.link,
378            flags: recipe.flags,
379            phases: Default::default(),
380            zig_version: None,
381            strip_debug: None,
382        },
383        target_bin: format!("/usr/bin/{name}"),
384        expected_hash: None,
385        deps: Default::default(),
386        // [H1c] La evidencia declarada por el cliente viaja a la mutación para ejecutarse tras
387        // sellar el artefacto (ver `evidence_verdict`).
388        evidence: recipe.evidence,
389        // [H4b] El camino del daemon (agente propone → construye) no declara superficies: el gate
390        // de compatibilidad vive en `install` (CLI) contra el estado instalado. Vacío aquí.
391        slots: Default::default(),
392    };
393    let store = match takana_core::Store::open(&store_root) {
394        Ok(s) => s,
395        Err(e) => {
396            let _ = tx.send(Event::BuildFailed {
397                recipe: name,
398                reason: format!("store: {e}"),
399                log_tail: None,
400            });
401            return;
402        }
403    };
404    let cfg = takana_build::BuildConfig::from_env_or_defaults(store.root());
405    match takana_build::build_source_patch(&mutation, &cfg, &store, None, None) {
406        Ok(h) => {
407            // [H1c] El artefacto está sellado; ahora EJECUTAMOS la evidencia declarada y
408            // adjuntamos el veredicto. El gate ("no proponer" si falla) vive en el orquestador
409            // —el lab sólo construye y reporta—; por eso enviamos `BuildReady` con veredicto
410            // aunque algún check falle.
411            let verdict = evidence_verdict(&mutation, &cfg, &store);
412            let _ = tx.send(Event::BuildReady {
413                recipe: name,
414                artifact: h.as_str().to_string(),
415                verdict,
416            });
417        }
418        Err(e) => {
419            // Si el fallo trae la cola del log del lab, la separamos de la razón para que la
420            // IA pueda leer el error real del compilador, no sólo el mensaje de Rust.
421            let (reason, log_tail) = match takana_build::sandbox::BuildFailure::from_error(&e) {
422                Some(bf) => (bf.reason.clone(), bf.log_tail.clone()),
423                None => (e.to_string(), None),
424            };
425            let _ = tx.send(Event::BuildFailed {
426                recipe: name,
427                reason,
428                log_tail,
429            });
430        }
431    }
432}
433
434/// [H1c] Ejecuta la evidencia declarada de una mutación `source_patch` y la reduce al veredicto
435/// serializable del bus. Devuelve:
436/// - `None` si la receta no declara evidencia (nada que correr).
437/// - `Some(all_passed=true/false, …)` con el resultado por check.
438/// - `Some(all_passed=false, …)` sintético si la evidencia ni siquiera pudo ejecutarse (falta una
439///   build-dep, error de sandbox…): eso también debe bloquear la propuesta, no pasar en silencio.
440fn evidence_verdict(
441    mutation: &takana_core::swm::Mutation,
442    cfg: &takana_build::BuildConfig,
443    store: &takana_core::Store,
444) -> Option<EvidenceVerdict> {
445    let has_evidence = matches!(
446        mutation,
447        takana_core::swm::Mutation::SourcePatch { evidence, .. } if !evidence.is_empty()
448    );
449    if !has_evidence {
450        return None;
451    }
452    match run_evidence_inner(mutation, cfg, store) {
453        Ok(v) => Some(v),
454        Err(e) => {
455            tracing::warn!(error = %e, "evidence: no se pudo ejecutar; veredicto = fallo");
456            Some(EvidenceVerdict {
457                all_passed: false,
458                max_level: None,
459                checks: vec![EvidenceCheckVerdict {
460                    kind: takana_core::EvidenceKind::CmdExit,
461                    cmd: "<runner>".into(),
462                    passed: false,
463                    detail: format!("la evidencia no pudo ejecutarse: {e}"),
464                }],
465            })
466        }
467    }
468}
469
470fn run_evidence_inner(
471    mutation: &takana_core::swm::Mutation,
472    cfg: &takana_build::BuildConfig,
473    store: &takana_core::Store,
474) -> takana_core::Result<EvidenceVerdict> {
475    let catalog_dir = takana_build::swm_bridge::catalog_dir_for(cfg, None);
476    std::fs::create_dir_all(&catalog_dir)?;
477    let recipe = takana_build::swm_bridge::recipe_from_source_patch(mutation, None, &catalog_dir)?;
478    let report = takana_build::run_evidence(&recipe, cfg, store)?;
479    Ok(EvidenceVerdict {
480        all_passed: report.all_passed(),
481        max_level: report.max_level_passed(),
482        checks: report
483            .results
484            .iter()
485            .map(|r| EvidenceCheckVerdict {
486                kind: r.kind,
487                cmd: r.cmd.clone(),
488                passed: r.outcome.passed,
489                detail: r.outcome.detail.clone(),
490            })
491            .collect(),
492    })
493}
494
495fn query_file(path: &str) -> serde_json::Value {
496    let p = Path::new(path);
497    let meta = match std::fs::symlink_metadata(p) {
498        Ok(m) => m,
499        Err(e) => return serde_json::json!({"exists": false, "error": e.to_string()}),
500    };
501    use std::os::unix::fs::MetadataExt;
502    serde_json::json!({
503        "exists": true,
504        "size": meta.size(),
505        "mode": meta.mode(),
506        "uid": meta.uid(),
507        "gid": meta.gid(),
508        "is_dir": meta.is_dir(),
509        "is_symlink": meta.file_type().is_symlink(),
510    })
511}
512
513fn query_artifact(store_root: &Path, hash: &str) -> serde_json::Value {
514    let store = match takana_core::Store::open(store_root) {
515        Ok(s) => s,
516        Err(e) => return serde_json::json!({"error": e.to_string()}),
517    };
518    match store.find_by_hash(hash) {
519        Ok(p) => serde_json::json!({"exists": true, "path": p.display().to_string()}),
520        Err(_) => serde_json::json!({"exists": false}),
521    }
522}
523
524fn t_of(cmd: &Command) -> &'static str {
525    match cmd {
526        Command::Hello { .. } => "hello",
527        Command::Compile { .. } => "compile",
528        Command::Inject { .. } => "inject",
529        Command::Query { .. } => "query",
530        Command::Init { .. } => "init",
531    }
532}
533
534// Re-export para que el caller no tenga que importar Mutex/Arc directamente.
535#[allow(dead_code)]
536pub(crate) type SharedCaps = Arc<Mutex<std::collections::BTreeSet<Cap>>>;
537
538#[cfg(test)]
539mod tests {
540    use super::*;
541
542    #[test]
543    fn default_policy_grants_query_to_others() {
544        let p = default_policy();
545        // Un peer con UID distinto al actual debe quedar como sólo-query.
546        let foreign = Peer {
547            uid: u32::MAX,
548            gid: 0,
549            pid: 1,
550        };
551        assert_eq!(p(&foreign), vec![Cap::Query]);
552    }
553
554    #[test]
555    fn config_policy_resolves_by_uid_gid() {
556        let cfg = takana_core::AgentCapsConfig::from_toml(
557            r#"
558default = ["query"]
559[[rule]]
560uid = 0
561caps = ["query", "compile", "inject", "inject-real", "init"]
562"#,
563        )
564        .unwrap();
565        let p = policy_from_config(cfg);
566        let root = Peer { uid: 0, gid: 0, pid: 1 };
567        let other = Peer { uid: 1234, gid: 1234, pid: 2 };
568        assert!(p(&root).contains(&Cap::InjectReal));
569        assert_eq!(p(&other), vec![Cap::Query]);
570    }
571
572    #[test]
573    fn t_of_matches_all_commands() {
574        // No es exhaustivo de Cargo, pero verifica que no nos olvidemos del nombre del tag.
575        assert_eq!(t_of(&Command::Hello { ver: 1, client: "".into() }), "hello");
576        assert_eq!(
577            t_of(&Command::Compile {
578                recipe: RecipeInline {
579                    name: "x".into(),
580                    repo: Some("g".into()),
581                    commit: Some("c".into()),
582                    tarball: None,
583                    sha256: None,
584                    patch: None,
585                    compiler: "zig-cc".into(),
586                    target: "t".into(),
587                    link: "static".into(),
588                    flags: vec![],
589                    evidence: Default::default(),
590                },
591            }),
592            "compile"
593        );
594    }
595
596    #[test]
597    fn query_file_reports_missing() {
598        let v = query_file("/no/existe/seguramente/ahora");
599        assert_eq!(v["exists"], serde_json::json!(false));
600    }
601}