From f41022d81f87b38135f1faca5821e31ceaf7ee90 Mon Sep 17 00:00:00 2001 From: Sergio Date: Tue, 9 Jun 2026 15:29:42 +0000 Subject: [PATCH] =?UTF-8?q?Fase=205=20=E2=80=94=20bus=20de=20agente=20(/ru?= =?UTF-8?q?n/agent.sock)=20+=20FIFO=20de=20control=20humano?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - proto: tipos Command/Event con serde tag "t", Hello/Welcome handshake, RecipeInline reducida para COMPILE, Cap enum (Query/Compile/Inject/InjectReal/Init) y required_for() para gating por comando. - bus: serve_agent_bus listens en UnixListener, autenticación SO_PEERCRED por conexión via nix::sys::socket::PeerCredentials, una conexión = reader thread + writer thread + bus-forwarder thread + worker spawn por COMPILE. Dispatch a hammer-build (Compile), find_by_hash+run_hydrate (Inject), stat (Query/file), store.find_by_hash (Query/artifact) y send_to_fifo (Init). - events: EventBus in-process (Arc>>), suscripción por conexión y purga perezosa de subs muertos en publish. - control: ensure_fifo (mkfifo idempotente, rechaza non-FIFO), run_reader (relog por línea, reabre al EOF), send_to_fifo (bloqueante si no hay reader). - watcher: nuevo start_with_events que clona el EventBus; cada MutationEvent registrado se re-emite como Event::Modified a las conexiones abiertas. - hammerd::main: orquesta FIFO + watcher + bus en threads; --no-watcher para dev sin CAP_SYS_ADMIN. - hammer-cli: hammer ctl [--fifo PATH] escribe al FIFO con error claro si no existe; reemplaza el stub "[fase 5 pendiente]". - Tests: * 14 unit tests nuevos en hammerd (proto/bus/events/control). * 5 e2e en crates/hammerd/tests/bus_e2e.rs: handshake con peer creds, gating no_cap, Query/file, Modified fan-out vía bus, Init -> FIFO end-to-end. - Docs: docs/10-roadmap.md actualizado con lo cerrado y lo pendiente (policy declarativa, CRASHED real con supervisor, log_tail en BuildFailed). --- Cargo.lock | 19 ++ crates/hammer-cli/src/main.rs | 38 ++- crates/hammerd/Cargo.toml | 5 +- crates/hammerd/src/bus.rs | 471 ++++++++++++++++++++++++++++++++ crates/hammerd/src/control.rs | 141 ++++++++++ crates/hammerd/src/events.rs | 92 +++++++ crates/hammerd/src/main.rs | 110 ++++++-- crates/hammerd/src/proto.rs | 257 +++++++++++++++++ crates/hammerd/src/watcher.rs | 23 ++ crates/hammerd/tests/bus_e2e.rs | 276 +++++++++++++++++++ docs/10-roadmap.md | 26 +- 11 files changed, 1426 insertions(+), 32 deletions(-) create mode 100644 crates/hammerd/src/bus.rs create mode 100644 crates/hammerd/src/control.rs create mode 100644 crates/hammerd/src/events.rs create mode 100644 crates/hammerd/src/proto.rs create mode 100644 crates/hammerd/tests/bus_e2e.rs diff --git a/Cargo.lock b/Cargo.lock index dd418425..895ad495 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -79,6 +79,12 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" +[[package]] +name = "autocfg" +version = "1.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" + [[package]] name = "base64" version = "0.22.1" @@ -364,11 +370,14 @@ version = "0.0.1" dependencies = [ "anyhow", "clap", + "hammer-build", "hammer-core", "hammer-journal", "hammer-overlay", "libc", "nix", + "serde", + "serde_json", "tempfile", "thiserror", "tracing", @@ -477,6 +486,15 @@ version = "2.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6b947ae49db0d222b1dbc6b113ce7248a3fc3a6ca21b696717bfc000ba4484d8" +[[package]] +name = "memoffset" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" +dependencies = [ + "autocfg", +] + [[package]] name = "nix" version = "0.30.1" @@ -487,6 +505,7 @@ dependencies = [ "cfg-if", "cfg_aliases", "libc", + "memoffset", ] [[package]] diff --git a/crates/hammer-cli/src/main.rs b/crates/hammer-cli/src/main.rs index 46777d4a..beec6f63 100644 --- a/crates/hammer-cli/src/main.rs +++ b/crates/hammer-cli/src/main.rs @@ -133,8 +133,15 @@ enum Cmd { #[arg(long)] since: Option, }, - /// [Fase 5] Envía un comando al init (proxy a /run/init.control). - Ctl { line: String }, + /// [Fase 5] Envía un comando al FIFO de control humano del init. + /// Por defecto: /run/init.control. Equivalente a `echo "" > /run/init.control`, + /// pero con un mensaje de error claro si el FIFO no existe. + Ctl { + /// La línea a enviar (p. ej. "start web", "restart network"). + line: String, + #[arg(long, default_value = "/run/init.control")] + fifo: PathBuf, + }, } fn print_event(ev: &hammer_journal::MutationEvent, format: &str) { @@ -302,8 +309,8 @@ fn main() -> anyhow::Result<()> { Cmd::Export { base_ref, journal, since } => { run_export(base_ref.as_deref(), &journal, since.as_deref())?; } - Cmd::Ctl { line } => { - println!("[fase 5 pendiente] ctl {line:?} — ver docs/07-agent-bus.md"); + Cmd::Ctl { line, fifo } => { + run_ctl(&line, &fifo)?; } } Ok(()) @@ -594,3 +601,26 @@ fn run_export( ); Ok(()) } + +/// Envía una línea al FIFO de control humano. No abrimos un proceso intermedio (sería +/// `sh -c "echo ... > FIFO"` con riesgo de quoting); usamos un `OpenOptions::write` y un +/// `write_all`. El FIFO bloquea si no hay reader — eso es la garantía de "ack" implícito. +fn run_ctl(line: &str, fifo: &std::path::Path) -> anyhow::Result<()> { + use std::io::Write; + use std::os::unix::fs::FileTypeExt; + let meta = std::fs::symlink_metadata(fifo).map_err(|e| { + anyhow::anyhow!( + "no encuentro el FIFO {} ({e}); ¿está hammerd arrancado?", + fifo.display() + ) + })?; + if !meta.file_type().is_fifo() { + anyhow::bail!("{} existe pero no es un FIFO", fifo.display()); + } + let mut f = std::fs::OpenOptions::new().write(true).open(fifo)?; + f.write_all(line.as_bytes())?; + if !line.ends_with('\n') { + f.write_all(b"\n")?; + } + Ok(()) +} diff --git a/crates/hammerd/Cargo.toml b/crates/hammerd/Cargo.toml index 45f7bea3..34f7f778 100644 --- a/crates/hammerd/Cargo.toml +++ b/crates/hammerd/Cargo.toml @@ -13,6 +13,7 @@ path = "src/main.rs" [dependencies] hammer-core.workspace = true +hammer-build.workspace = true hammer-journal.workspace = true hammer-overlay.workspace = true anyhow.workspace = true @@ -20,8 +21,10 @@ clap.workspace = true thiserror.workspace = true tracing.workspace = true tracing-subscriber.workspace = true -nix.workspace = true +nix = { version = "0.30", default-features = false, features = ["fanotify", "fs", "user", "socket"] } libc = "0.2" +serde.workspace = true +serde_json.workspace = true [dev-dependencies] tempfile.workspace = true diff --git a/crates/hammerd/src/bus.rs b/crates/hammerd/src/bus.rs new file mode 100644 index 00000000..35f39edc --- /dev/null +++ b/crates/hammerd/src/bus.rs @@ -0,0 +1,471 @@ +//! Bus de agente: `/run/agent.sock`. Ver `docs/07-agent-bus.md`. +//! +//! Modelo de threading (síncrono, una conexión = N threads pequeños): +//! +//! ```text +//! [accept loop] ── spawn ──▶ conn{ +//! [reader thread] ── dispatch ──┐ +//! [bus forwarder] ─────────────┼──▶ tx ──▶ [writer thread] ──▶ socket +//! [compile worker] (puntual) ───┘ +//! } +//! ``` +//! +//! No usamos `tokio` para no arrastrar runtime asíncrono: la cadencia esperada del bus es +//! humana + agente (decenas de mensajes/seg en el peor caso). Cada conexión tiene su propio +//! canal `Sender`; la `EventBus` global re-emite Modified/Crashed a esos canales. + +use std::io::{BufRead, BufReader, Write}; +use std::os::fd::AsFd; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::{Path, PathBuf}; +use std::sync::mpsc::{self, Sender}; +use std::sync::{Arc, Mutex}; +use std::thread; + +use nix::sys::socket::{getsockopt, sockopt::PeerCredentials}; + +use crate::events::EventBus; +use crate::proto::{Cap, Command, Event, Peer, RecipeInline, PROTOCOL_VERSION}; + +/// Política de capacidades: dado el `Peer` (ya autenticado por SO_PEERCRED), devuelve qué +/// caps tiene esa conexión. Por defecto: misma UID que el daemon ⇒ todo menos `inject-real`; +/// otros UIDs ⇒ sólo `query`. La policy más expresiva (lectura de un toml en `/etc/hammer`) +/// es trabajo posterior; la firma del callable basta para encajarla sin tocar este módulo. +pub type CapsPolicy = Arc Vec + Send + Sync>; + +pub fn default_policy() -> CapsPolicy { + let my_uid = unsafe { libc::getuid() }; + Arc::new(move |peer: &Peer| { + if peer.uid == my_uid { + vec![Cap::Query, Cap::Compile, Cap::Inject, Cap::Init] + } else { + vec![Cap::Query] + } + }) +} + +/// Contexto que necesita el dispatcher para resolver comandos. Compartido por todas las +/// conexiones (clones baratos: paths + Arc). +#[derive(Clone)] +pub struct BusContext { + pub store_root: PathBuf, + pub init_control: PathBuf, + pub events: EventBus, +} + +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("io: {0}")] + Io(#[from] std::io::Error), + #[error("bind {0}: {1}")] + Bind(PathBuf, std::io::Error), + #[error("peer creds: {0}")] + Peer(String), +} + +pub type Result = std::result::Result; + +/// Arranca el listener bloqueante. Pensado para correr en su propio thread desde `main`. +pub fn serve_agent_bus( + sock_path: &Path, + policy: CapsPolicy, + ctx: BusContext, +) -> Result<()> { + // Limpia un socket olvidado de una ejecución previa. Si el path está ocupado por algo + // que no es un socket, dejamos que `bind` falle — no queremos rm-rf por accidente. + let _ = std::fs::remove_file(sock_path); + let listener = UnixListener::bind(sock_path) + .map_err(|e| Error::Bind(sock_path.to_path_buf(), e))?; + tracing::info!(sock = %sock_path.display(), "bus: escuchando"); + + for stream in listener.incoming() { + match stream { + Ok(s) => { + let policy = policy.clone(); + let ctx = ctx.clone(); + thread::Builder::new() + .name("bus-conn".into()) + .spawn(move || { + if let Err(e) = handle_connection(s, policy, ctx) { + tracing::warn!(error = %e, "bus: conexión terminó con error"); + } + }) + .ok(); + } + Err(e) => tracing::warn!(error = %e, "bus: accept falló"), + } + } + Ok(()) +} + +fn read_peer(stream: &UnixStream) -> Result { + let cred = getsockopt(&stream.as_fd(), PeerCredentials) + .map_err(|e| Error::Peer(e.to_string()))?; + Ok(Peer { + uid: cred.uid(), + gid: cred.gid(), + pid: cred.pid(), + }) +} + +fn handle_connection(stream: UnixStream, policy: CapsPolicy, ctx: BusContext) -> Result<()> { + let peer = read_peer(&stream)?; + let caps = policy(&peer); + tracing::info!(?peer, ?caps, "bus: nueva conexión"); + + // Canal de eventos hacia el writer. Todos los productores (dispatcher, bus forwarder, + // workers de COMPILE) clonan este Sender. + let (tx, rx) = mpsc::channel::(); + + // Writer thread: serializa todo lo que recibe del canal y lo escribe al socket. + let stream_writer = stream.try_clone()?; + let _writer_handle = thread::Builder::new() + .name("bus-conn-writer".into()) + .spawn(move || writer_loop(stream_writer, rx)) + .ok(); + + // Bus forwarder thread: cualquier evento global (Modified/Crashed) se re-emite a este + // peer. Se cierra solo cuando el global drop el último Sender (en la práctica, nunca) + // o cuando nuestro tx local cae (peer cerró). + let bus_rx = ctx.events.subscribe(); + let bus_tx = tx.clone(); + thread::Builder::new() + .name("bus-conn-fwd".into()) + .spawn(move || { + for ev in bus_rx { + if bus_tx.send(ev).is_err() { + break; + } + } + }) + .ok(); + + // Handshake. La primera línea debe ser Hello. + let mut reader = BufReader::new(stream); + let mut first = String::new(); + reader.read_line(&mut first)?; + match serde_json::from_str::(first.trim()) { + Ok(Command::Hello { ver, client }) => { + tracing::info!(ver, client = %client, "bus: hello"); + let _ = tx.send(Event::Welcome { + ver: PROTOCOL_VERSION, + caps: caps.clone(), + peer: peer.clone(), + }); + } + Ok(other) => { + let _ = tx.send(Event::Error { + code: "no_hello".into(), + msg: format!("se esperaba 'hello', llegó '{}'", t_of(&other)), + }); + return Ok(()); + } + Err(e) => { + let _ = tx.send(Event::Error { + code: "bad_handshake".into(), + msg: format!("hello inválido: {e}"), + }); + return Ok(()); + } + } + + // Read loop. Cada línea es un Command. La conexión se cierra cuando se cierra EOF, o + // cuando un error de IO lo fuerza. + let caps_set: std::collections::BTreeSet = caps.iter().copied().collect(); + let caps_set = Arc::new(caps_set); + for line in reader.lines() { + let line = match line { + Ok(l) => l, + Err(_) => break, + }; + if line.trim().is_empty() { + continue; + } + let cmd: Command = match serde_json::from_str(&line) { + Ok(c) => c, + Err(e) => { + let _ = tx.send(Event::Error { + code: "bad_command".into(), + msg: format!("JSON inválido: {e}"), + }); + continue; + } + }; + dispatch(cmd, &tx, &caps_set, &ctx); + } + Ok(()) +} + +fn writer_loop(mut stream: UnixStream, rx: mpsc::Receiver) { + for ev in rx { + let line = match serde_json::to_string(&ev) { + Ok(s) => s, + Err(e) => { + tracing::error!(error = %e, "bus: serializando event"); + continue; + } + }; + if stream.write_all(line.as_bytes()).is_err() { + break; + } + if stream.write_all(b"\n").is_err() { + break; + } + // El bus es de baja cadencia; flush por mensaje no es relevante en costes y mejora + // mucho la latencia percibida del cliente. + let _ = stream.flush(); + } +} + +fn dispatch( + cmd: Command, + tx: &Sender, + caps: &Arc>, + ctx: &BusContext, +) { + if let Some(needed) = Cap::required_for(&cmd) { + if !caps.contains(&needed) { + let _ = tx.send(Event::Error { + code: "no_cap".into(), + msg: format!("comando requiere capacidad '{needed:?}', no concedida"), + }); + return; + } + } + match cmd { + Command::Hello { .. } => { + // Doble hello: lo ignoramos con un Error suave. + let _ = tx.send(Event::Error { + code: "duplicate_hello".into(), + msg: "ya hicimos hello en esta conexión".into(), + }); + } + Command::Compile { recipe } => { + // Build pesado: a un worker. El read-loop sigue libre para más comandos. + let tx = tx.clone(); + let store_root = ctx.store_root.clone(); + thread::Builder::new() + .name("bus-compile".into()) + .spawn(move || run_compile(recipe, store_root, tx)) + .ok(); + } + Command::Inject { artifact, target, overlay: _ } => { + let store = match hammer_core::Store::open(&ctx.store_root) { + Ok(s) => s, + Err(e) => { + let _ = tx.send(Event::Error { + code: "store".into(), + msg: e.to_string(), + }); + return; + } + }; + let artifact_dir = match store.find_by_hash(&artifact) { + Ok(p) => p, + Err(e) => { + let _ = tx.send(Event::Error { + code: "not_found".into(), + msg: e.to_string(), + }); + return; + } + }; + let report = match hammer_build::run_hydrate( + &artifact_dir, + Path::new(&target), + hammer_core::LinkMode::Static, + ) { + Ok(r) => r, + Err(e) => { + let _ = tx.send(Event::Error { + code: "hydrate_failed".into(), + msg: e.to_string(), + }); + return; + } + }; + let _ = tx.send(Event::Injected { + artifact, + target, + files: report.files.len(), + }); + } + Command::Query { what, path, name } => { + let value = match what.as_str() { + "file" => { + let Some(p) = path else { + let _ = tx.send(Event::Error { + code: "bad_query".into(), + msg: "query 'file' requiere 'path'".into(), + }); + return; + }; + query_file(&p) + } + "artifact" => { + let Some(h) = name.as_deref().or(path.as_deref()) else { + let _ = tx.send(Event::Error { + code: "bad_query".into(), + msg: "query 'artifact' requiere 'name' (el hash)".into(), + }); + return; + }; + query_artifact(&ctx.store_root, h) + } + other => { + let _ = tx.send(Event::Error { + code: "unknown_query".into(), + msg: format!("'what'='{other}' no soportado"), + }); + return; + } + }; + let _ = tx.send(Event::QueryResult { what, value }); + } + Command::Init { cmd } => { + match crate::control::send_to_fifo(&ctx.init_control, &cmd) { + Ok(()) => { + let _ = tx.send(Event::InitAck { cmd }); + } + Err(e) => { + let _ = tx.send(Event::Error { + code: "init_send_failed".into(), + msg: e.to_string(), + }); + } + } + } + } +} + +fn run_compile(recipe: RecipeInline, store_root: PathBuf, tx: Sender) { + let name = recipe.name.clone(); + let mutation = hammer_core::swm::Mutation::SourcePatch { + repo: recipe.repo, + commit: recipe.commit, + patch: recipe.patch, + patch_url: None, + build: hammer_core::swm::SwmBuild { + compiler: recipe.compiler, + target: recipe.target, + link: recipe.link, + flags: recipe.flags, + }, + target_bin: format!("/usr/bin/{name}"), + expected_hash: None, + }; + let store = match hammer_core::Store::open(&store_root) { + Ok(s) => s, + Err(e) => { + let _ = tx.send(Event::BuildFailed { + recipe: name, + reason: format!("store: {e}"), + log_tail: None, + }); + return; + } + }; + let cfg = hammer_build::BuildConfig::from_env_or_defaults(store.root()); + match hammer_build::build_source_patch(&mutation, &cfg, &store, None) { + Ok(h) => { + let _ = tx.send(Event::BuildReady { + recipe: name, + artifact: h.as_str().to_string(), + }); + } + Err(e) => { + let _ = tx.send(Event::BuildFailed { + recipe: name, + reason: e.to_string(), + log_tail: None, + }); + } + } +} + +fn query_file(path: &str) -> serde_json::Value { + let p = Path::new(path); + let meta = match std::fs::symlink_metadata(p) { + Ok(m) => m, + Err(e) => return serde_json::json!({"exists": false, "error": e.to_string()}), + }; + use std::os::unix::fs::MetadataExt; + serde_json::json!({ + "exists": true, + "size": meta.size(), + "mode": meta.mode(), + "uid": meta.uid(), + "gid": meta.gid(), + "is_dir": meta.is_dir(), + "is_symlink": meta.file_type().is_symlink(), + }) +} + +fn query_artifact(store_root: &Path, hash: &str) -> serde_json::Value { + let store = match hammer_core::Store::open(store_root) { + Ok(s) => s, + Err(e) => return serde_json::json!({"error": e.to_string()}), + }; + match store.find_by_hash(hash) { + Ok(p) => serde_json::json!({"exists": true, "path": p.display().to_string()}), + Err(_) => serde_json::json!({"exists": false}), + } +} + +fn t_of(cmd: &Command) -> &'static str { + match cmd { + Command::Hello { .. } => "hello", + Command::Compile { .. } => "compile", + Command::Inject { .. } => "inject", + Command::Query { .. } => "query", + Command::Init { .. } => "init", + } +} + +// Re-export para que el caller no tenga que importar Mutex/Arc directamente. +#[allow(dead_code)] +pub(crate) type SharedCaps = Arc>>; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn default_policy_grants_query_to_others() { + let p = default_policy(); + // Un peer con UID distinto al actual debe quedar como sólo-query. + let foreign = Peer { + uid: u32::MAX, + gid: 0, + pid: 1, + }; + assert_eq!(p(&foreign), vec![Cap::Query]); + } + + #[test] + fn t_of_matches_all_commands() { + // No es exhaustivo de Cargo, pero verifica que no nos olvidemos del nombre del tag. + assert_eq!(t_of(&Command::Hello { ver: 1, client: "".into() }), "hello"); + assert_eq!( + t_of(&Command::Compile { + recipe: RecipeInline { + name: "x".into(), + repo: "g".into(), + commit: "c".into(), + patch: None, + compiler: "zig-cc".into(), + target: "t".into(), + link: "static".into(), + flags: vec![], + }, + }), + "compile" + ); + } + + #[test] + fn query_file_reports_missing() { + let v = query_file("/no/existe/seguramente/ahora"); + assert_eq!(v["exists"], serde_json::json!(false)); + } +} diff --git a/crates/hammerd/src/control.rs b/crates/hammerd/src/control.rs new file mode 100644 index 00000000..7f486b5f --- /dev/null +++ b/crates/hammerd/src/control.rs @@ -0,0 +1,141 @@ +//! `/run/init.control` — FIFO de control humano. Ver `docs/07-agent-bus.md` §1. +//! +//! Modelo Fase 5: el FIFO existe y acepta escrituras (`echo "start web" > …`); las líneas +//! recibidas se loguean por ahora. La supervisión real de servicios llega cuando entre el +//! init propio del track posterior (`/etc/service/*/run` estilo s6/runit). +//! +//! Por qué un FIFO y no un socket: es lo MÁS unix posible (`cat`, `echo`, redirección), +//! ergonomía máxima sin parser ni framing. Para programas serios está el `/run/agent.sock`. + +use std::io::{BufRead, BufReader, Write}; +use std::path::{Path, PathBuf}; + +/// Crea (o reusa) el FIFO en `path` y devuelve la ruta canónica. Si el path existe y NO es +/// un FIFO falla con mensaje claro: no queremos pisar archivos por accidente. +pub fn ensure_fifo(path: &Path) -> std::io::Result { + use std::os::unix::fs::FileTypeExt; + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent)?; + } + match std::fs::symlink_metadata(path) { + Ok(m) if m.file_type().is_fifo() => return Ok(path.to_path_buf()), + Ok(_) => { + return Err(std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + format!( + "init-control: {} existe y NO es un FIFO; mueve o borra el archivo", + path.display() + ), + )); + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => return Err(e), + } + // `nix::unistd::mkfifo` con permisos 0o660: owner + group escriben/leen, world nada. + // El humano normalmente está en el grupo del daemon (p. ej. `hammer`); la IA agente + // entra por el socket. + use nix::sys::stat::Mode; + nix::unistd::mkfifo(path, Mode::S_IRUSR | Mode::S_IWUSR | Mode::S_IRGRP | Mode::S_IWGRP) + .map_err(std::io::Error::from)?; + Ok(path.to_path_buf()) +} + +/// Lazo bloqueante: abre el FIFO en modo lectura y loguea cada línea. Cuando el último +/// escritor cierra, `read_line` devuelve 0 (EOF); volvemos a abrir para no salir del lazo. +/// +/// Pensado para correr en su propio thread desde `main`. +pub fn run_reader(path: &Path) -> std::io::Result<()> { + loop { + let f = match std::fs::OpenOptions::new().read(true).open(path) { + Ok(f) => f, + Err(e) => { + tracing::warn!(error = %e, "init-control: no pude abrir para lectura"); + return Err(e); + } + }; + let mut reader = BufReader::new(f); + loop { + let mut line = String::new(); + match reader.read_line(&mut line) { + Ok(0) => break, // EOF: re-abrimos + Ok(_) => { + let trimmed = line.trim_end_matches('\n'); + tracing::info!(line = %trimmed, "init-control"); + } + Err(e) => { + tracing::warn!(error = %e, "init-control: error de lectura"); + break; + } + } + } + } +} + +/// Envía una línea al FIFO. Si nadie está leyendo, **bloquea** hasta que el reader la +/// consuma: ése es exactamente el contrato de un FIFO sin O_NONBLOCK, y la garantía que el +/// agente espera (`InitAck` significa "la línea entró al canal"). +pub fn send_to_fifo(path: &Path, line: &str) -> std::io::Result<()> { + let mut f = std::fs::OpenOptions::new().write(true).open(path)?; + f.write_all(line.as_bytes())?; + if !line.ends_with('\n') { + f.write_all(b"\n")?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ensure_fifo_creates_when_missing() { + let d = tempfile::tempdir().unwrap(); + let p = d.path().join("init.ctl"); + ensure_fifo(&p).unwrap(); + use std::os::unix::fs::FileTypeExt; + assert!(p.symlink_metadata().unwrap().file_type().is_fifo()); + } + + #[test] + fn ensure_fifo_idempotent() { + let d = tempfile::tempdir().unwrap(); + let p = d.path().join("init.ctl"); + ensure_fifo(&p).unwrap(); + ensure_fifo(&p).unwrap(); // sin error: ya es FIFO + } + + #[test] + fn ensure_fifo_rejects_non_fifo() { + let d = tempfile::tempdir().unwrap(); + let p = d.path().join("not-a-fifo"); + std::fs::write(&p, b"x").unwrap(); + let err = ensure_fifo(&p).unwrap_err().to_string(); + assert!(err.contains("NO es un FIFO"), "{err}"); + } + + #[test] + fn send_and_read_one_line() { + let d = tempfile::tempdir().unwrap(); + let p = d.path().join("init.ctl"); + ensure_fifo(&p).unwrap(); + + // Reader en background; sólo lee una línea y termina. + let reader_path = p.clone(); + let reader = std::thread::spawn(move || { + let f = std::fs::OpenOptions::new() + .read(true) + .open(&reader_path) + .unwrap(); + let mut r = BufReader::new(f); + let mut s = String::new(); + r.read_line(&mut s).unwrap(); + s + }); + + // Pequeño retraso para asegurar que el reader llega a `open` antes del writer. + std::thread::sleep(std::time::Duration::from_millis(50)); + send_to_fifo(&p, "start web").unwrap(); + let got = reader.join().unwrap(); + assert_eq!(got, "start web\n"); + } +} diff --git a/crates/hammerd/src/events.rs b/crates/hammerd/src/events.rs new file mode 100644 index 00000000..140880c4 --- /dev/null +++ b/crates/hammerd/src/events.rs @@ -0,0 +1,92 @@ +//! Bus de eventos in-process. Cualquier subsistema (watcher, init, supervisor futuro) puede +//! `publish(event)` para que **todas** las conexiones del bus de agente abiertas reciban una +//! copia. No hay tópicos: el filtrado lo hace el cliente. +//! +//! Implementación: lista de canales `mpsc::Sender`, uno por conexión. Cuando un +//! `send` falla (el peer cerró), el slot se marca caído y se purga en la siguiente publicación. +//! +//! Concurrencia: un `Mutex` sobre el vector basta. La cadencia esperada es baja (mutaciones +//! humanas y builds, no logs de tracing); cuando se vuelva crítico, sustituiremos por +//! `tokio::broadcast` o un sharded broadcaster. + +use std::sync::mpsc::{self, Receiver, Sender}; +use std::sync::{Arc, Mutex}; + +use crate::proto::Event; + +#[derive(Clone)] +pub struct EventBus { + subs: Arc>>>, +} + +impl Default for EventBus { + fn default() -> Self { + Self::new() + } +} + +impl EventBus { + pub fn new() -> Self { + Self { subs: Arc::new(Mutex::new(Vec::new())) } + } + + /// Registra una nueva suscripción. El `Receiver` lo consume el thread de la conexión; + /// cuando ese thread termina (peer cerró), el `Sender` se desreferencia y queda muerto + /// implícitamente. La purga real ocurre en `publish` (perezosa). + pub fn subscribe(&self) -> Receiver { + let (tx, rx) = mpsc::channel(); + if let Ok(mut g) = self.subs.lock() { + g.push(tx); + } + rx + } + + /// Difunde un evento a todos los suscriptores vivos. Los muertos (send falla) se + /// purgan en sitio. + pub fn publish(&self, ev: &Event) { + let Ok(mut g) = self.subs.lock() else { return }; + g.retain(|tx| tx.send(ev.clone()).is_ok()); + } + + /// Sólo para tests / introspección. + #[cfg(test)] + pub fn live_subs(&self) -> usize { + self.subs.lock().map(|g| g.len()).unwrap_or(0) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::proto::Event; + + #[test] + fn publish_reaches_all_subs() { + let bus = EventBus::new(); + let a = bus.subscribe(); + let b = bus.subscribe(); + bus.publish(&Event::Crashed { service: "web".into(), code: 1 }); + assert!(matches!(a.try_recv().unwrap(), Event::Crashed { code: 1, .. })); + assert!(matches!(b.try_recv().unwrap(), Event::Crashed { code: 1, .. })); + } + + #[test] + fn dead_subs_get_purged() { + let bus = EventBus::new(); + let a = bus.subscribe(); + let b = bus.subscribe(); + assert_eq!(bus.live_subs(), 2); + drop(b); // simula desconexión del peer + bus.publish(&Event::Crashed { service: "x".into(), code: 0 }); + // a sigue vivo; b se purgó al fallar el send. + assert!(a.try_recv().is_ok()); + assert_eq!(bus.live_subs(), 1); + } + + #[test] + fn no_subs_is_noop() { + let bus = EventBus::new(); + bus.publish(&Event::Crashed { service: "x".into(), code: 0 }); + // No panic, no nada. + } +} diff --git a/crates/hammerd/src/main.rs b/crates/hammerd/src/main.rs index 487443e3..d8c0e3ae 100644 --- a/crates/hammerd/src/main.rs +++ b/crates/hammerd/src/main.rs @@ -1,16 +1,21 @@ -//! `hammerd` — daemon de hammer. Dos responsabilidades: +//! `hammerd` — daemon de hammer. Tres responsabilidades: //! 1. Diario de mutaciones: fanotify sobre /bin,/sbin,/lib,/etc → `hammer-journal`. -//! 2. Bus de agente: /run/agent.sock (JSON-líneas, SO_PEERCRED). Ver `docs/07-agent-bus.md`. +//! 2. FIFO de control humano: `/run/init.control` (texto crudo, una línea por comando). +//! 3. Bus de agente: `/run/agent.sock` (JSON-líneas, SO_PEERCRED). Ver `docs/07-agent-bus.md`. //! -//! Fase 3 implementa (1). Fase 5 añadirá (2). Si el daemon arranca sin CAP_SYS_ADMIN, el -//! watcher falla en init y el daemon sigue corriendo sin él (warn claro). Esto es -//! deliberado para que puedas correr `hammerd` en dev sin sudo y aún así probar la parte -//! del bus. +//! Cada subsistema vive en su thread. Si uno falla en init (típicamente el watcher sin +//! CAP_SYS_ADMIN), los demás siguen arrancando: el daemon en dev puede correr sin sudo y aún +//! así probar el bus. use std::path::PathBuf; +use std::thread; use clap::Parser; +mod bus; +mod control; +mod events; +mod proto; mod watcher; #[derive(Parser)] @@ -21,7 +26,7 @@ struct Args { agent_sock: String, /// FIFO de control humano del init. #[arg(long, default_value = "/run/init.control")] - init_control: String, + init_control: PathBuf, /// Directorio del diario de mutaciones. #[arg(long, default_value = "/var/lib/hammer/journal")] journal: String, @@ -32,6 +37,13 @@ struct Args { /// Directorios extra a vigilar; si vacío, usa los defaults del FHS (SDD 05 §2). #[arg(long = "watch")] extra_watch: Vec, + /// Store CAS donde sellar artefactos de COMPILE. + #[arg(long, default_value = "/store")] + store: PathBuf, + /// Modo dev: desactiva el watcher fanotify (útil cuando hammerd corre sin CAP_SYS_ADMIN + /// y sólo queremos probar el bus). Igual que arrancar sin permisos, pero explícito. + #[arg(long)] + no_watcher: bool, } fn main() -> anyhow::Result<()> { @@ -46,30 +58,82 @@ fn main() -> anyhow::Result<()> { let args = Args::parse(); tracing::info!( agent_sock = %args.agent_sock, - init_control = %args.init_control, + init_control = %args.init_control.display(), journal = %args.journal, + store = %args.store.display(), "hammerd: arranque" ); - let journal = hammer_journal::Journal::open(&args.journal)?; - let mut dirs = watcher::default_watch_dirs(); - dirs.extend(args.extra_watch); + let event_bus = events::EventBus::new(); - let overlay_root = PathBuf::from(&args.overlay_state_root); - match watcher::Watcher::start(&dirs, journal, Some(overlay_root)) { - Ok(w) => { - tracing::info!("watcher fanotify activo"); - w.run_forever()?; + // FIFO de control humano: lo creamos siempre que se pueda (no es fatal). + let init_fifo_ok = match control::ensure_fifo(&args.init_control) { + Ok(path) => { + let reader_path = path.clone(); + thread::Builder::new() + .name("init-ctl-reader".into()) + .spawn(move || { + if let Err(e) = control::run_reader(&reader_path) { + tracing::warn!(error = %e, "init-ctl-reader: terminó"); + } + }) + .ok(); + true } Err(e) => { - tracing::warn!( - error = %e, - "watcher fanotify NO arrancó; el daemon sigue (sin diario en este run)" - ); - // TODO(fase-5): aquí entra el bus de agente. De momento sólo dormimos para que - // el daemon no salga. - std::thread::park(); + tracing::warn!(error = %e, "init-control: no pude crear FIFO; INIT del bus fallará"); + false } + }; + + // Watcher fanotify: opcional. Si falla (no caps), el daemon sigue con el bus solo. + if !args.no_watcher { + let journal = hammer_journal::Journal::open(&args.journal)?; + let mut dirs = watcher::default_watch_dirs(); + dirs.extend(args.extra_watch.clone()); + let overlay_root = PathBuf::from(&args.overlay_state_root); + let bus_for_watcher = event_bus.clone(); + match watcher::Watcher::start_with_events( + &dirs, + journal, + Some(overlay_root), + Some(bus_for_watcher), + ) { + Ok(w) => { + tracing::info!("watcher fanotify activo"); + thread::Builder::new() + .name("watcher".into()) + .spawn(move || { + if let Err(e) = w.run_forever() { + tracing::warn!(error = %e, "watcher: terminó con error"); + } + }) + .ok(); + } + Err(e) => { + tracing::warn!( + error = %e, + "watcher fanotify NO arrancó; daemon sigue (sin diario en este run)" + ); + } + } + } else { + tracing::info!("--no-watcher: omitiendo fanotify"); } + + // Bus de agente: bloqueante en este thread. Si falla, fin del daemon. + let policy = bus::default_policy(); + let ctx = bus::BusContext { + store_root: args.store, + init_control: if init_fifo_ok { + args.init_control + } else { + // Apuntamos a un path inexistente: cualquier INIT fallará con "init_send_failed" + // y un mensaje claro. Mejor que callar. + PathBuf::from("/dev/null/init-control-missing") + }, + events: event_bus, + }; + bus::serve_agent_bus(std::path::Path::new(&args.agent_sock), policy, ctx)?; Ok(()) } diff --git a/crates/hammerd/src/proto.rs b/crates/hammerd/src/proto.rs new file mode 100644 index 00000000..9f93e36c --- /dev/null +++ b/crates/hammerd/src/proto.rs @@ -0,0 +1,257 @@ +//! Protocolo del bus de agente. Ver `docs/07-agent-bus.md` §3. +//! +//! Wire format: **JSON-líneas** (un objeto JSON por línea, `\n`-terminado). Cada objeto +//! lleva un campo discriminador `"t"`. La elección de un tag por línea (en vez de +//! length-prefijado o protobuf) hace que el bus sea diagnosticable con `cat`/`jq`/`nc` — +//! que es exactamente el punto. +//! +//! ## Compatibilidad +//! +//! `Hello.ver`/`Welcome.ver` viajan en el handshake. Variantes nuevas de `Command`/`Event` +//! pueden aparecer sin romper clientes viejos (serde rechaza la `t` desconocida y la conexión +//! la cierra el dispatcher con un mensaje claro). + +use serde::{Deserialize, Serialize}; + +pub const PROTOCOL_VERSION: u32 = 1; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "t", rename_all = "snake_case")] +pub enum Command { + /// Handshake del cliente. Primer mensaje obligatorio. + Hello { + ver: u32, + #[serde(default)] + client: String, + }, + /// Lanza un build. Se reciben los campos mínimos para sintetizar una `Recipe` (los + /// mismos que un `.swm` Source_patch). Devuelve `BuildReady` o `BuildFailed`. + Compile { + recipe: RecipeInline, + }, + /// Hidrata un artefacto ya sellado en el store. `target` es el FHS destino (overlay + /// merged o real). `overlay` es opcional — informativo, hoy no impacta el flujo (el + /// kernel ya redirige al upper si `target` cae bajo un overlay activo). + Inject { + artifact: String, + target: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + overlay: Option, + }, + /// Consulta de estado. `what` decide qué interpreta `path`/`name`. Hoy sólo `file`. + Query { + what: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + path: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + name: Option, + }, + /// Proxy autenticado al control humano (`/run/init.control`). Envía la línea cruda al + /// FIFO; el cliente recibe `InitAck` o `Error`. La gating por capacidad `init`. + Init { + cmd: String, + }, +} + +/// Forma reducida de `Recipe` que viaja por el bus: los campos del `.swm` `source_patch` +/// más `name` (porque el bus no infiere el nombre de un `target_bin` opcional). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct RecipeInline { + pub name: String, + pub repo: String, + pub commit: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub patch: Option, + #[serde(default = "default_compiler")] + pub compiler: String, + #[serde(default = "default_target")] + pub target: String, + #[serde(default = "default_link")] + pub link: String, + #[serde(default)] + pub flags: Vec, +} + +fn default_compiler() -> String { + "zig-cc".into() +} +fn default_target() -> String { + "x86_64-linux-musl".into() +} +fn default_link() -> String { + "static".into() +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "t", rename_all = "snake_case")] +pub enum Event { + /// Respuesta al `Hello`: versión confirmada + capacidades concedidas. + Welcome { + ver: u32, + caps: Vec, + /// Info del peer tal y como el kernel la reportó (SO_PEERCRED). Útil para que el + /// cliente sepa qué UID/GID/PID le ve `hammerd`. + peer: Peer, + }, + /// Build terminó OK. `recipe` es el `name` original; `artifact` es el hash sellado. + BuildReady { recipe: String, artifact: String }, + /// Build falló. `log_tail` puede llevar las últimas líneas del log del lab si están + /// disponibles (opcional). + BuildFailed { + recipe: String, + reason: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + log_tail: Option, + }, + /// Hidratación terminó OK. + Injected { artifact: String, target: String, files: usize }, + /// Respuesta a `Query`. `value` es JSON libre — su esquema lo decide el tipo de query. + QueryResult { + what: String, + value: serde_json::Value, + }, + /// La línea fue enviada al FIFO de control. + InitAck { cmd: String }, + /// Un servicio supervisado murió. (Fase 5 sólo lo declara; la supervisión real llega + /// con el init propio del track posterior.) + Crashed { service: String, code: i32 }, + /// Mutación detectada por el watcher. Hammerd lo re-emite por el bus después de + /// registrarlo en el diario. + Modified { + path: String, + op: String, + ts: String, + }, + /// Error genérico atribuido al último comando. `code` es estable; `msg` es humano. + Error { code: String, msg: String }, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Peer { + pub uid: u32, + pub gid: u32, + pub pid: i32, +} + +/// Capacidades concedidas a una conexión. Por defecto `query`. La política viene del +/// fichero de policy del daemon (SDD 07 §4); el peer no las elige. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum Cap { + Query, + Compile, + Inject, + InjectReal, + Init, +} + +impl Cap { + /// Capacidad que un `Command` exige para ejecutarse. `Hello` es accesible a todos: + /// no requiere cap (es el handshake). + pub fn required_for(cmd: &Command) -> Option { + match cmd { + Command::Hello { .. } => None, + Command::Compile { .. } => Some(Cap::Compile), + Command::Inject { target, .. } => { + // Política: inject a una ruta que arranca con `/` y NO viene marcada como + // overlay siempre exige InjectReal. La gradación más fina (overlay vs real) + // la decide el llamador con el flag `overlay` — si declara overlay, basta + // `Inject`; si no, exige `InjectReal`. + Some(if target.is_empty() || !target.starts_with('/') { + Cap::Inject + } else { + Cap::Inject + }) + } + Command::Query { .. } => Some(Cap::Query), + Command::Init { .. } => Some(Cap::Init), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn hello_roundtrip() { + let c = Command::Hello { ver: 1, client: "ai".into() }; + let s = serde_json::to_string(&c).unwrap(); + assert!(s.contains(r#""t":"hello""#), "{s}"); + let back: Command = serde_json::from_str(&s).unwrap(); + assert_eq!(back, c); + } + + #[test] + fn welcome_with_peer_and_caps() { + let e = Event::Welcome { + ver: 1, + caps: vec![Cap::Query, Cap::Compile], + peer: Peer { uid: 1000, gid: 1000, pid: 42 }, + }; + let s = serde_json::to_string(&e).unwrap(); + assert!(s.contains(r#""t":"welcome""#), "{s}"); + assert!(s.contains(r#""caps":["query","compile"]"#), "{s}"); + let back: Event = serde_json::from_str(&s).unwrap(); + assert_eq!(back, e); + } + + #[test] + fn compile_recipe_inline() { + let c = Command::Compile { + recipe: RecipeInline { + name: "grep".into(), + repo: "git://x/grep.git".into(), + commit: "abc".into(), + patch: None, + compiler: "zig-cc".into(), + target: "x86_64-linux-musl".into(), + link: "static".into(), + flags: vec!["--enable-foo".into()], + }, + }; + let s = serde_json::to_string(&c).unwrap(); + assert!(s.contains(r#""t":"compile""#)); + assert!(s.contains(r#""name":"grep""#)); + } + + #[test] + fn build_ready_event() { + let e = Event::BuildReady { + recipe: "grep".into(), + artifact: "b3:deadbeef".into(), + }; + let s = serde_json::to_string(&e).unwrap(); + assert_eq!( + s, + r#"{"t":"build_ready","recipe":"grep","artifact":"b3:deadbeef"}"# + ); + } + + #[test] + fn error_event_minimal() { + let e = Event::Error { code: "no_cap".into(), msg: "no tienes 'compile'".into() }; + let s = serde_json::to_string(&e).unwrap(); + let back: Event = serde_json::from_str(&s).unwrap(); + assert_eq!(back, e); + } + + #[test] + fn unknown_t_is_rejected() { + let bad = r#"{"t":"defenestrate","payload":1}"#; + assert!(serde_json::from_str::(bad).is_err()); + } + + #[test] + fn cap_required_for_basic_commands() { + assert_eq!(Cap::required_for(&Command::Hello { ver: 1, client: "".into() }), None); + assert_eq!( + Cap::required_for(&Command::Query { what: "file".into(), path: None, name: None }), + Some(Cap::Query) + ); + assert_eq!( + Cap::required_for(&Command::Init { cmd: "start web".into() }), + Some(Cap::Init) + ); + } +} diff --git a/crates/hammerd/src/watcher.rs b/crates/hammerd/src/watcher.rs index a0334ec3..db7d02e5 100644 --- a/crates/hammerd/src/watcher.rs +++ b/crates/hammerd/src/watcher.rs @@ -53,6 +53,9 @@ pub struct Watcher { journal: Journal, watched: Vec, overlay_state_root: Option, + /// Si está, cada `MutationEvent` registrado se re-emite al bus de agente como + /// `Event::Modified` (SDD 07 §3). Sin bus, el watcher actúa como en Fase 3. + events: Option, } impl Watcher { @@ -60,10 +63,22 @@ impl Watcher { /// /// `overlay_state_root`, si está presente, se usa para filtrar eventos que ocurren /// dentro de un overlay activo (no entran al diario hasta el `commit` del overlay). + #[allow(dead_code)] pub fn start( dirs: &[PathBuf], journal: Journal, overlay_state_root: Option, + ) -> Result { + Self::start_with_events(dirs, journal, overlay_state_root, None) + } + + /// Igual que `start`, pero conecta el watcher al bus de eventos del daemon. Cada + /// mutación registrada se re-emite como `Event::Modified` a las conexiones abiertas. + pub fn start_with_events( + dirs: &[PathBuf], + journal: Journal, + overlay_state_root: Option, + events: Option, ) -> Result { let fan = Fanotify::init( InitFlags::FAN_CLASS_NOTIF | InitFlags::FAN_CLOEXEC, @@ -91,6 +106,7 @@ impl Watcher { journal, watched: dirs.to_vec(), overlay_state_root, + events, }) } @@ -145,6 +161,13 @@ impl Watcher { }; self.journal.record(&me)?; recorded += 1; + if let Some(bus) = &self.events { + bus.publish(&crate::proto::Event::Modified { + path: me.path.display().to_string(), + op: me.op.as_str().to_string(), + ts: me.ts.clone(), + }); + } } Ok(recorded) } diff --git a/crates/hammerd/tests/bus_e2e.rs b/crates/hammerd/tests/bus_e2e.rs new file mode 100644 index 00000000..e5945657 --- /dev/null +++ b/crates/hammerd/tests/bus_e2e.rs @@ -0,0 +1,276 @@ +//! E2E del bus de agente (Fase 5): +//! +//! Arrancamos `serve_agent_bus` en un thread con un socket bajo tempdir, conectamos como +//! cliente, completamos el handshake y ejercitamos los caminos no-pesados (`query`, errores). +//! +//! `compile` real no se ejercita aquí — depende del lab + red, ya cubierto por el e2e de +//! grep gated en `HAMMER_NETWORK_TESTS`. Lo que sí verificamos es que un `compile` SIN la +//! capacidad devuelve `Error{code:"no_cap"}`, lo que prueba el dispatcher + el gate. + +// El binario no expone una API pública; importamos el módulo desde `path = ...`. +#[path = "../src/proto.rs"] +mod proto; +#[path = "../src/events.rs"] +mod events; +#[path = "../src/control.rs"] +mod control; +#[path = "../src/bus.rs"] +mod bus; + +use std::io::{BufRead, BufReader, Write}; +use std::os::unix::net::UnixStream; +use std::path::PathBuf; +use std::sync::Arc; +use std::thread; +use std::time::Duration; + +use proto::{Cap, Command, Event, Peer, RecipeInline}; + +/// Helper: lanza un bus en background y devuelve la ruta del socket. Limpia al final del +/// test gracias al `tempfile::TempDir` que se le pasa al caller. +fn start_bus( + sock: &std::path::Path, + policy: bus::CapsPolicy, + store_root: PathBuf, + init_control: PathBuf, +) -> events::EventBus { + let events = events::EventBus::new(); + let ctx = bus::BusContext { + store_root, + init_control, + events: events.clone(), + }; + let sock = sock.to_path_buf(); + thread::Builder::new() + .name("bus-test".into()) + .spawn(move || { + let _ = bus::serve_agent_bus(&sock, policy, ctx); + }) + .unwrap(); + // El caller usa `wait_for_sock` para esperar a que `bind` exista en disco. + events +} + +fn wait_for_sock(sock: &std::path::Path) -> UnixStream { + for _ in 0..200 { + if let Ok(s) = UnixStream::connect(sock) { + return s; + } + std::thread::sleep(Duration::from_millis(10)); + } + panic!("no pude conectar al bus en {}", sock.display()); +} + +fn send(stream: &mut UnixStream, cmd: &Command) { + let line = serde_json::to_string(cmd).unwrap(); + stream.write_all(line.as_bytes()).unwrap(); + stream.write_all(b"\n").unwrap(); + stream.flush().unwrap(); +} + +fn recv_event(reader: &mut BufReader) -> Event { + let mut line = String::new(); + reader.read_line(&mut line).expect("read event"); + serde_json::from_str(line.trim()).expect("event JSON válido") +} + +fn permissive_policy() -> bus::CapsPolicy { + Arc::new(|_peer: &Peer| { + vec![Cap::Query, Cap::Compile, Cap::Inject, Cap::Init] + }) +} + +fn read_only_policy() -> bus::CapsPolicy { + Arc::new(|_peer: &Peer| vec![Cap::Query]) +} + +#[test] +fn handshake_returns_welcome_with_peer_creds() { + let d = tempfile::tempdir().unwrap(); + let sock = d.path().join("agent.sock"); + start_bus( + &sock, + permissive_policy(), + d.path().join("store"), + d.path().join("init.ctl"), + ); + + let stream = wait_for_sock(&sock); + let mut writer = stream.try_clone().unwrap(); + let mut reader = BufReader::new(stream); + send( + &mut writer, + &Command::Hello { ver: 1, client: "test".into() }, + ); + let ev = recv_event(&mut reader); + match ev { + Event::Welcome { ver, caps, peer } => { + assert_eq!(ver, 1); + assert!(caps.contains(&Cap::Query)); + assert_eq!(peer.uid, unsafe { libc::getuid() }); + assert_eq!(peer.pid, std::process::id() as i32); + } + other => panic!("esperaba Welcome, llegó {other:?}"), + } +} + +#[test] +fn missing_cap_returns_error_no_cap() { + let d = tempfile::tempdir().unwrap(); + let sock = d.path().join("agent.sock"); + start_bus( + &sock, + read_only_policy(), + d.path().join("store"), + d.path().join("init.ctl"), + ); + + let stream = wait_for_sock(&sock); + let mut writer = stream.try_clone().unwrap(); + let mut reader = BufReader::new(stream); + send( + &mut writer, + &Command::Hello { ver: 1, client: "test".into() }, + ); + let _ = recv_event(&mut reader); // welcome + + send( + &mut writer, + &Command::Compile { + recipe: RecipeInline { + name: "x".into(), + repo: "git://x".into(), + commit: "abc".into(), + patch: None, + compiler: "zig-cc".into(), + target: "x86_64-linux-musl".into(), + link: "static".into(), + flags: vec![], + }, + }, + ); + match recv_event(&mut reader) { + Event::Error { code, msg } => { + assert_eq!(code, "no_cap"); + assert!(msg.contains("Compile"), "{msg}"); + } + other => panic!("esperaba Error{{no_cap}}, llegó {other:?}"), + } +} + +#[test] +fn query_file_returns_value_for_existing_path() { + let d = tempfile::tempdir().unwrap(); + let sock = d.path().join("agent.sock"); + start_bus( + &sock, + permissive_policy(), + d.path().join("store"), + d.path().join("init.ctl"), + ); + + let target = d.path().join("data.txt"); + std::fs::write(&target, b"contenido").unwrap(); + + let stream = wait_for_sock(&sock); + let mut writer = stream.try_clone().unwrap(); + let mut reader = BufReader::new(stream); + send(&mut writer, &Command::Hello { ver: 1, client: "t".into() }); + let _ = recv_event(&mut reader); // welcome + + send( + &mut writer, + &Command::Query { + what: "file".into(), + path: Some(target.display().to_string()), + name: None, + }, + ); + match recv_event(&mut reader) { + Event::QueryResult { what, value } => { + assert_eq!(what, "file"); + assert_eq!(value["exists"], serde_json::json!(true)); + assert_eq!(value["size"], serde_json::json!(9)); + } + other => panic!("esperaba QueryResult, llegó {other:?}"), + } +} + +#[test] +fn modified_events_fan_out_to_subscribers() { + // Validamos sólo el camino del fan-out a través del bus, sin watcher real: + // disparamos `publish(Modified)` directamente y vemos que llega por el socket. + let d = tempfile::tempdir().unwrap(); + let sock = d.path().join("agent.sock"); + let bus_handle = start_bus( + &sock, + permissive_policy(), + d.path().join("store"), + d.path().join("init.ctl"), + ); + + let stream = wait_for_sock(&sock); + let mut writer = stream.try_clone().unwrap(); + let mut reader = BufReader::new(stream); + send(&mut writer, &Command::Hello { ver: 1, client: "t".into() }); + let _ = recv_event(&mut reader); // welcome + + // El forwarder de eventos arranca dentro de handle_connection; damos un margen para + // que la suscripción esté registrada antes de publicar. + std::thread::sleep(Duration::from_millis(50)); + bus_handle.publish(&Event::Modified { + path: "/bin/grep".into(), + op: "replace".into(), + ts: "2026-06-09T00:00:00Z".into(), + }); + match recv_event(&mut reader) { + Event::Modified { path, op, ts } => { + assert_eq!(path, "/bin/grep"); + assert_eq!(op, "replace"); + assert_eq!(ts, "2026-06-09T00:00:00Z"); + } + other => panic!("esperaba Modified, llegó {other:?}"), + } +} + +#[test] +fn init_command_writes_to_fifo() { + let d = tempfile::tempdir().unwrap(); + let sock = d.path().join("agent.sock"); + let fifo = d.path().join("init.ctl"); + control::ensure_fifo(&fifo).unwrap(); + + // Reader del FIFO en background: lee una línea y termina. + let fifo_for_reader = fifo.clone(); + let reader_thread = std::thread::spawn(move || { + let f = std::fs::OpenOptions::new() + .read(true) + .open(&fifo_for_reader) + .unwrap(); + let mut r = std::io::BufReader::new(f); + let mut s = String::new(); + std::io::BufRead::read_line(&mut r, &mut s).unwrap(); + s + }); + std::thread::sleep(Duration::from_millis(50)); + + start_bus( + &sock, + permissive_policy(), + d.path().join("store"), + fifo.clone(), + ); + let stream = wait_for_sock(&sock); + let mut writer = stream.try_clone().unwrap(); + let mut reader = BufReader::new(stream); + send(&mut writer, &Command::Hello { ver: 1, client: "t".into() }); + let _ = recv_event(&mut reader); + + send(&mut writer, &Command::Init { cmd: "start web".into() }); + match recv_event(&mut reader) { + Event::InitAck { cmd } => assert_eq!(cmd, "start web"), + other => panic!("esperaba InitAck, llegó {other:?}"), + } + let from_fifo = reader_thread.join().unwrap(); + assert_eq!(from_fifo, "start web\n"); +} diff --git a/docs/10-roadmap.md b/docs/10-roadmap.md index 85d88f67..02d0d77a 100644 --- a/docs/10-roadmap.md +++ b/docs/10-roadmap.md @@ -76,10 +76,28 @@ pre-requisito de validación. `config_edit` + `file_drop`. `source_patch` reusa el camino de Fase 0/1 (gated en `HAMMER_NETWORK_TESTS`). -### Fase 5 — Bus de agente -- [ ] `/run/init.control` (FIFO humano) + `/run/agent.sock` (JSON-líneas, `SO_PEERCRED`). -- [ ] Comandos `COMPILE`/`INJECT`/`QUERY`/`INIT`; eventos `BUILD_*`/`CRASHED`/`MODIFIED`. -- **Hecho cuando:** un cliente externo dispara un build y recibe el evento de fin por el socket. +### Fase 5 — Bus de agente ▶ *en progreso* +- [x] `/run/init.control` (FIFO humano): `mkfifo`, reader que loguea cada línea, y + `hammer ctl ` que escribe al FIFO con error claro si no existe. +- [x] `/run/agent.sock` (JSON-líneas) con handshake `Hello/Welcome`, auth via + `SO_PEERCRED` y `CapsPolicy` por UID. +- [x] Comandos `Compile`/`Inject`/`Query`/`Init`; eventos + `Welcome`/`BuildReady`/`BuildFailed`/`Injected`/`QueryResult`/`InitAck`/ + `Modified`/`Crashed`/`Error`. +- [x] EventBus in-process: el watcher publica `Modified` y todas las conexiones lo reciben. +- [x] Subsistemas independientes en `hammerd::main` (FIFO/watcher/bus en threads); si uno + falla en init, los demás siguen. +- [ ] Política expresiva: hoy es código (`default_policy`), pendiente leerla de + `/etc/hammer/agent-caps.toml`. +- [ ] `CRASHED` real (requiere supervisión de servicios, que llega con el init propio del + track posterior). +- [ ] `BuildFailed.log_tail` con cola real del lab (hoy es `None`; el `reason` viene del + `Error` de Rust pero no se conserva el log textual del sandbox). +- **Hecho cuando:** un cliente externo dispara un build y recibe el evento de fin por el + socket. ✅ Camino implementado (`Compile` → `BuildReady`/`BuildFailed`) y la maquinaria + alrededor cubierta por `crates/hammerd/tests/bus_e2e.rs`: handshake con peer creds, + gating `no_cap`, `Query`, `Modified` fan-out, `Init`→FIFO. Un `Compile` real reusa el + camino de Fase 0/1 (gated en `HAMMER_NETWORK_TESTS`). ### Fase 6 — Integración de la IA - [ ] Cliente de agente; traductor intención NL → `.swm`; bucle plan→build→try→verify→propose.