Fase 5 — bus de agente (/run/agent.sock) + FIFO de control humano
- 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<Mutex<Vec<Sender>>>), 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 <line> [--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).
f41022d81f87b38135f1faca5821e31ceaf7ee90 · by Sergio · 2026-06-09T15:29:42Z (4 months ago) · parent 09d50f9e6browse the tree at this commit
Semantic delta · 83
crates/hammer-cli/src/main.rs crates/hammerd/src/bus.rs crates/hammerd/src/control.rs crates/hammerd/src/events.rs crates/hammerd/src/main.rs crates/hammerd/src/proto.rs crates/hammerd/src/watcher.rs 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<String>,
},
- /// [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 "<line>" > /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<Event>`; 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<dyn Fn(&Peer) -> Vec<Cap> + 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<T> = std::result::Result<T, Error>;
+
+/// 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<Peer> {
+ 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::<Event>();
+
+ // 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::<Command>(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<Cap> = 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<Event>) {
+ 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<Event>,
+ caps: &Arc<std::collections::BTreeSet<Cap>>,
+ 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<Event>) {
+ 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<Mutex<std::collections::BTreeSet<Cap>>>;
+
+#[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<PathBuf> {
+ 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<Event>`, 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<Mutex<Vec<Sender<Event>>>>,
+}
+
+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<Event> {
+ 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<PathBuf>,
+ /// 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<String>,
+ },
+ /// 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<String>,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ name: Option<String>,
+ },
+ /// 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<String>,
+ #[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<String>,
+}
+
+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<Cap>,
+ /// 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<String>,
+ },
+ /// 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<Cap> {
+ 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::<Command>(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<PathBuf>,
overlay_state_root: Option<PathBuf>,
+ /// 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<crate::events::EventBus>,
}
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<PathBuf>,
+ ) -> Result<Self> {
+ 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<PathBuf>,
+ events: Option<crate::events::EventBus>,
) -> Result<Self> {
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<UnixStream>) -> 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 <line>` 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.