From a23630c50d051377f60507ae267924e85b368f71 Mon Sep 17 00:00:00 2001 From: Sergio Date: Sun, 14 Jun 2026 03:30:04 +0000 Subject: [PATCH] =?UTF-8?q?feat(arje-link):=20transporte=20arje-bus=20?= =?UTF-8?q?=E2=86=92=20bus=20de=20agente,=20cierra=20B.2=20end-to-end?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit hammerd::arje_link se suscribe al bus del init (ENTE_BUS_SOCK), relee el frame postcard de arje-bus con un mirror mínimo de suscriptor (sin arrastrar el crate-graph de arje ⇒ hammer sigue standalone) y reenvía cada BusEvent → crashes → Event::Crashed → agent.sock. Wire verificado byte-a-byte contra arje-bus real (ulid string, frame Subscribe=[00,01,00,0d]); 2 tests de round-trip local + frame. Se lanza en thread si ENTE_BUS_SOCK está definido (no-op si no). Roadmap B.2 marcado ✅ (resta sólo el smoke contra init vivo). Co-Authored-By: Claude Opus 4.8 (1M context) --- Cargo.lock | 303 +++++++++++++++++++++++++++++++- Cargo.toml | 4 + crates/hammerd/Cargo.toml | 2 + crates/hammerd/src/arje_link.rs | 260 +++++++++++++++++++++++++++ crates/hammerd/src/main.rs | 5 + docs/10-roadmap.md | 23 +-- 6 files changed, 584 insertions(+), 13 deletions(-) create mode 100644 crates/hammerd/src/arje_link.rs diff --git a/Cargo.lock b/Cargo.lock index d00f3ccb..566d7b1b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -79,6 +79,15 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" +[[package]] +name = "atomic-polyfill" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8cf2bce30dfe09ef0bfaef228b9d414faaf7e563035494d7fe092dba54b300f4" +dependencies = [ + "critical-section", +] + [[package]] name = "autocfg" version = "1.5.1" @@ -126,6 +135,18 @@ dependencies = [ "generic-array", ] +[[package]] +name = "bumpalo" +version = "3.20.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" + +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + [[package]] name = "cc" version = "1.2.63" @@ -188,6 +209,15 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" +[[package]] +name = "cobs" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fa961b519f0b462e3a3b4a34b64d119eeaca1d59af726fe450bbba07a9fc0a1" +dependencies = [ + "thiserror", +] + [[package]] name = "colorchoice" version = "1.0.5" @@ -224,6 +254,12 @@ dependencies = [ "libc", ] +[[package]] +name = "critical-section" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" + [[package]] name = "crypto-common" version = "0.1.7" @@ -316,6 +352,18 @@ dependencies = [ "zeroize", ] +[[package]] +name = "embedded-io" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef1a6892d9eef45c8fa6b9e0086428a2cca8491aca8f787c534a3d6d0bcb3ced" + +[[package]] +name = "embedded-io" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" + [[package]] name = "equivalent" version = "1.0.2" @@ -365,6 +413,30 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures-core" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" + +[[package]] +name = "futures-task" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" + +[[package]] +name = "futures-util" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" +dependencies = [ + "futures-core", + "futures-task", + "pin-project-lite", + "slab", +] + [[package]] name = "generic-array" version = "0.14.7" @@ -386,6 +458,18 @@ dependencies = [ "wasi", ] +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "libc", + "r-efi 5.3.0", + "wasip2", +] + [[package]] name = "getrandom" version = "0.4.2" @@ -394,7 +478,7 @@ checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555" dependencies = [ "cfg-if", "libc", - "r-efi", + "r-efi 6.0.0", "wasip2", "wasip3", ] @@ -527,12 +611,23 @@ dependencies = [ "hammer-overlay", "libc", "nix", + "postcard", "serde", "serde_json", "tempfile", "thiserror", "tracing", "tracing-subscriber", + "ulid", +] + +[[package]] +name = "hash32" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0c35f58762feb77d74ebe43bdbc3210f09be9fe6742234d573bacc26ed92b67" +dependencies = [ + "byteorder", ] [[package]] @@ -550,6 +645,20 @@ version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +[[package]] +name = "heapless" +version = "0.7.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdc6457c0eb62c71aac4bc17216026d8410337c4126773b9c5daba343f17964f" +dependencies = [ + "atomic-polyfill", + "hash32", + "rustc_version", + "serde", + "spin", + "stable_deref_trait", +] + [[package]] name = "heck" version = "0.5.0" @@ -695,6 +804,17 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "js-sys" +version = "0.3.102" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03d04c30968dffe80775bd4d7fb676131cd04a1fb46d2686dbffbaec2d9dfd31" +dependencies = [ + "cfg-if", + "futures-util", + "wasm-bindgen", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -725,6 +845,15 @@ version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + [[package]] name = "log" version = "0.4.32" @@ -811,6 +940,19 @@ dependencies = [ "spki", ] +[[package]] +name = "postcard" +version = "1.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6764c3b5dd454e283a30e6dfe78e9b31096d9e32036b5d1eaac7a6119ccb9a24" +dependencies = [ + "cobs", + "embedded-io 0.4.0", + "embedded-io 0.6.1", + "heapless", + "serde", +] + [[package]] name = "potential_utf" version = "0.1.5" @@ -820,6 +962,15 @@ dependencies = [ "zerovec", ] +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + [[package]] name = "prettyplease" version = "0.2.37" @@ -848,12 +999,38 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + [[package]] name = "r-efi" version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rand" +version = "0.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" +dependencies = [ + "rand_chacha", + "rand_core 0.9.5", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", +] + [[package]] name = "rand_core" version = "0.6.4" @@ -863,6 +1040,15 @@ dependencies = [ "getrandom 0.2.17", ] +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "regex-automata" version = "0.4.14" @@ -951,12 +1137,24 @@ dependencies = [ "untrusted", ] +[[package]] +name = "rustversion" +version = "1.0.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" + [[package]] name = "ryu" version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + [[package]] name = "semver" version = "1.0.28" @@ -1060,15 +1258,30 @@ version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de" dependencies = [ - "rand_core", + "rand_core 0.6.4", ] +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + [[package]] name = "smallvec" version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +[[package]] +name = "spin" +version = "0.9.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +dependencies = [ + "lock_api", +] + [[package]] name = "spki" version = "0.7.3" @@ -1279,6 +1492,17 @@ version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" +[[package]] +name = "ulid" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "470dbf6591da1b39d43c14523b2b469c86879a53e8b758c8e090a470fe7b1fbe" +dependencies = [ + "rand", + "serde", + "web-time", +] + [[package]] name = "unicode-ident" version = "1.0.24" @@ -1380,6 +1604,51 @@ dependencies = [ "wit-bindgen 0.51.0", ] +[[package]] +name = "wasm-bindgen" +version = "0.2.125" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ddb3f79143bced6de84270411622a2699cee572fc0875aeaf1e7867cf9fca1a" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.125" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e21a184b13fb19e157296e2c46056aec9092264fab83e4ba59e68c61b323c3d" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.125" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fecefd9c35bd935a20fc3fc344b5f29138961e4f47fb03297d88f2587afb5ebd" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.125" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23939e44bb9a5d7576fa2b563dc2e136628f1224e88a8deed09e04858b77871f" +dependencies = [ + "unicode-ident", +] + [[package]] name = "wasm-encoder" version = "0.244.0" @@ -1414,6 +1683,16 @@ dependencies = [ "semver", ] +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "webpki-roots" version = "0.26.11" @@ -1652,6 +1931,26 @@ dependencies = [ "synstructure", ] +[[package]] +name = "zerocopy" +version = "0.8.52" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce1022995ff5ff5d841ad7d994facc23098cd40152f2c1d11cd607c6f530653f" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.52" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ae7f38b72ec2a254e2b87ef277cf2cd4fb97cbebf944faa6f33354da0867930" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "zerofrom" version = "0.1.8" diff --git a/Cargo.toml b/Cargo.toml index 41572682..4858884c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,6 +33,10 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" serde_yaml = "0.9" toml = "0.8" +# postcard + ulid: SÓLO para releer el frame del bus de arje (hammerd::arje_link). Versiones +# alineadas con las que usa arje-bus (postcard 1.1.x, ulid 1.2.x) para garantizar el mismo wire. +postcard = { version = "1", features = ["use-std"] } +ulid = { version = "1", features = ["serde"] } blake3 = "1" sha2 = "0.10" hex = "0.4" diff --git a/crates/hammerd/Cargo.toml b/crates/hammerd/Cargo.toml index 34f7f778..6e211c03 100644 --- a/crates/hammerd/Cargo.toml +++ b/crates/hammerd/Cargo.toml @@ -25,6 +25,8 @@ nix = { version = "0.30", default-features = false, features = ["fanotify", "fs" libc = "0.2" serde.workspace = true serde_json.workspace = true +postcard.workspace = true +ulid.workspace = true [dev-dependencies] tempfile.workspace = true diff --git a/crates/hammerd/src/arje_link.rs b/crates/hammerd/src/arje_link.rs new file mode 100644 index 00000000..c12f6f3e --- /dev/null +++ b/crates/hammerd/src/arje_link.rs @@ -0,0 +1,260 @@ +//! Adaptador de transporte arje-bus → bus de agente de hammer (último tramo de B.2). +//! +//! arje (el init/PID 1) supervisa cada Ente y, al morir, difunde `BusEvent` por su bus +//! (`arje-bus`: `BusRequest::Subscribe` + `BusPayload::Event`). Este módulo se **suscribe** a +//! ese stream y traduce cada evento al vocabulario de hammer (`crashes::Lifecycle` → +//! `Event::Crashed`), publicándolo en el [`EventBus`] → `/run/agent.sock` → la capa de IA. +//! +//! ## Por qué un mirror y no una dependencia de `arje-bus` +//! +//! `arje-bus` arrastra el grafo de crates de arje (arje-card → card-core → …). Acoplarlo aquí +//! rompería el build hermético/standalone de hammer. En su lugar **releemos el frame postcard**: +//! el bus usa frames `u32-BE-len + postcard(BusMessage)`, y reproducimos sólo el subconjunto que +//! un suscriptor necesita. El layout está verificado byte-a-byte contra `arje-bus` real (mismas +//! versiones de `ulid` 1.2 y `postcard` 1.1; `ulid` serializa como string). Si arje reordena las +//! variantes del enum, el contrato se rompe — por eso los discriminantes están documentados y +//! cubiertos por un test de round-trip local. +//! +//! Varios campos del mirror (`from`, `seq`, `id`) no se leen: existen para **consumir** los +//! bytes del wire al deserializar. Por eso el `allow(dead_code)` a nivel de módulo. +#![allow(dead_code)] + +use std::io::{Read, Write}; +use std::os::unix::net::UnixStream; +use std::path::PathBuf; +use std::thread; + +use serde::Deserialize; +use ulid::Ulid; + +use crate::crashes::{self, LifeStatus, Lifecycle}; +use crate::events::EventBus; + +/// Igual que `arje_bus::ENV_BUS_SOCK`: el env donde arje-zero publica la ruta del socket. +pub const ENV_BUS_SOCK: &str = "ENTE_BUS_SOCK"; + +/// Tope de frame, igual que `arje_bus::MAX_FRAME` (1 MiB) — protección contra OOM. +const MAX_FRAME: usize = 1 << 20; + +// ── Mirror del wire de arje-bus (subconjunto de suscriptor) ────────────────────────────── +// +// Discriminantes (postcard = varint del índice de variante), verificados contra arje-bus: +// BusPayload: Request=0, Response=1, Event=2 +// BusRequest::Subscribe = 13 (último; ver arje-bus/src/lib.rs) +// BusEvent: EnteCrashed=0, EnteRestarting=1, EnteExited=2 +// LifecycleStatus: Exited=0, Killed=1 + +#[derive(Deserialize, Debug)] +struct BusMessage { + from: Option, + seq: u64, + payload: BusPayload, +} + +#[derive(Deserialize, Debug)] +enum BusPayload { + /// Índice 0. Nunca lo recibe un suscriptor (los Invoke van a proveedores, no a subs); + /// existe sólo para alinear el discriminante de `Response`/`Event`. + Request, + /// Índice 1. El único que recibimos por aquí es el `Ok` del ack de `Subscribe`. + Response(MiniResponse), + /// Índice 2. El stream de eventos de ciclo de vida. + Event(BusEvent), +} + +#[derive(Deserialize, Debug)] +enum MiniResponse { + Ok, + Error(String), +} + +#[derive(Deserialize, Debug)] +enum BusEvent { + EnteCrashed { id: Ulid, label: String, status: LifecycleStatus }, + EnteRestarting { id: Ulid, label: String, delay_ms: u64 }, + EnteExited { id: Ulid, label: String }, +} + +#[derive(Deserialize, Debug)] +enum LifecycleStatus { + Exited(i32), + Killed(i32), +} + +impl BusEvent { + /// Traduce el evento de arje a la señal normalizada de hammer. El `label` del Ente es el + /// `service`; una muerte por señal usa la convención de shell `128 + signum` para que el + /// código sea siempre ≠ 0. + fn to_lifecycle(self) -> Lifecycle { + match self { + BusEvent::EnteCrashed { label, status, .. } => { + let code = match status { + LifecycleStatus::Exited(c) => c, + LifecycleStatus::Killed(sig) => 128 + sig, + }; + Lifecycle { service: label, status: LifeStatus::Crashed { code } } + } + BusEvent::EnteRestarting { label, delay_ms, .. } => { + Lifecycle { service: label, status: LifeStatus::Restarting { delay_ms } } + } + BusEvent::EnteExited { label, .. } => { + Lifecycle { service: label, status: LifeStatus::Exited } + } + } + } +} + +/// Cuerpo del frame `Subscribe` (sin el prefijo de longitud), hand-encodeado: `from=None` + +/// `seq` (varint) + `BusPayload::Request`(0) + `BusRequest::Subscribe`(13). Verificado +/// byte-a-byte contra `postcard::to_stdvec` de arje-bus. +fn subscribe_body(seq: u64) -> Vec { + let mut v = vec![0x00u8]; // Option::None + let mut s = seq; // u64 como varint LEB128 + loop { + let b = (s & 0x7f) as u8; + s >>= 7; + if s == 0 { + v.push(b); + break; + } else { + v.push(b | 0x80); + } + } + v.push(0x00); // BusPayload::Request + v.push(0x0D); // BusRequest::Subscribe (índice 13) + v +} + +fn write_frame(w: &mut impl Write, body: &[u8]) -> std::io::Result<()> { + w.write_all(&(body.len() as u32).to_be_bytes())?; + w.write_all(body)?; + w.flush() +} + +fn read_frame(r: &mut impl Read) -> std::io::Result> { + let mut len = [0u8; 4]; + r.read_exact(&mut len)?; + let n = u32::from_be_bytes(len) as usize; + if n > MAX_FRAME { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("frame oversize: {n} > {MAX_FRAME}"), + )); + } + let mut buf = vec![0u8; n]; + r.read_exact(&mut buf)?; + Ok(buf) +} + +/// Conecta al bus de arje, se suscribe y bombea eventos de ciclo de vida al `EventBus` hasta +/// que la conexión se cierra. Bloqueante — pensado para correr en su propio thread. +fn run(sock: &PathBuf, bus: &EventBus) -> std::io::Result<()> { + let mut stream = UnixStream::connect(sock)?; + write_frame(&mut stream, &subscribe_body(1))?; + tracing::info!(sock = %sock.display(), "arje-link: suscrito al bus de init"); + loop { + let body = read_frame(&mut stream)?; + let msg: BusMessage = match postcard::from_bytes(&body) { + Ok(m) => m, + Err(e) => { + // Un frame que no entendemos no debe matar el puente: lo saltamos. + tracing::warn!(error = %e, "arje-link: frame indescifrable, ignoro"); + continue; + } + }; + match msg.payload { + BusPayload::Event(ev) => { + let life = ev.to_lifecycle(); + if let Some(event) = crashes::to_event(&life) { + bus.publish(&event); + } + } + // El ack del Subscribe y cualquier otra respuesta: nada que hacer. + BusPayload::Response(_) | BusPayload::Request => {} + } + } +} + +/// Lanza el puente en un thread si `ENTE_BUS_SOCK` está definido (arje corriendo). Si no, es un +/// no-op: hammerd sigue sin la fuente de crashes (los tests del bus no la necesitan). El thread +/// es resiliente: si el bus muere, loguea y termina sin tumbar al daemon. +pub fn spawn_if_configured(bus: EventBus) { + let Ok(path) = std::env::var(ENV_BUS_SOCK) else { + tracing::debug!("arje-link: {ENV_BUS_SOCK} no definido; sin fuente de crashes"); + return; + }; + let sock = PathBuf::from(path); + let _ = thread::Builder::new().name("arje-link".into()).spawn(move || { + if let Err(e) = run(&sock, &bus) { + tracing::warn!(error = %e, "arje-link: puente terminó (¿bus de init caído?)"); + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn subscribe_body_es_el_frame_esperado() { + // Verificado contra arje-bus real: [from=None, seq=1, Request, Subscribe=13]. + assert_eq!(subscribe_body(1), vec![0x00, 0x01, 0x00, 0x0D]); + // seq grande → varint multibyte (300 = 0xAC 0x02). + assert_eq!(subscribe_body(300), vec![0x00, 0xAC, 0x02, 0x00, 0x0D]); + } + + /// Round-trip local con el MISMO `postcard`/`ulid` que arje-bus: serializamos un + /// `BusMessage::Event` con tipos espejo `Serialize` y lo deserializamos con los `Deserialize` + /// reales del módulo. Si los discriminantes/orden de campos se desalinean, esto rompe. + #[test] + fn evento_crashed_roundtrips_por_el_wire() { + use serde::Serialize; + #[derive(Serialize)] + enum TxPayload { + #[allow(dead_code)] + Request, + #[allow(dead_code)] + Response, + Event(TxEvent), + } + #[derive(Serialize)] + enum TxEvent { + EnteCrashed { id: Ulid, label: String, status: TxStatus }, + } + #[derive(Serialize)] + enum TxStatus { + #[allow(dead_code)] + Exited(i32), + Killed(i32), + } + #[derive(Serialize)] + struct TxMsg { + from: Option, + seq: u64, + payload: TxPayload, + } + + let id = Ulid::from_string("01ARZ3NDEKTSV4RRFFQ69G5FAV").unwrap(); + let bytes = postcard::to_stdvec(&TxMsg { + from: Some(id), + seq: 7, + payload: TxPayload::Event(TxEvent::EnteCrashed { + id, + label: "demonio".into(), + status: TxStatus::Killed(11), + }), + }) + .unwrap(); + + let msg: BusMessage = postcard::from_bytes(&bytes).unwrap(); + match msg.payload { + BusPayload::Event(ev) => { + let life = ev.to_lifecycle(); + assert_eq!(life.service, "demonio"); + // Killed(11) → 128 + 11 = 139. + assert_eq!(life.status, LifeStatus::Crashed { code: 139 }); + } + other => panic!("esperaba Event, fue {other:?}"), + } + } +} diff --git a/crates/hammerd/src/main.rs b/crates/hammerd/src/main.rs index 1e989c33..1a40bb49 100644 --- a/crates/hammerd/src/main.rs +++ b/crates/hammerd/src/main.rs @@ -12,6 +12,7 @@ use std::thread; use clap::Parser; +mod arje_link; mod bus; mod control; mod crashes; @@ -70,6 +71,10 @@ fn main() -> anyhow::Result<()> { let event_bus = events::EventBus::new(); + // Puente con el bus del init (arje): si ENTE_BUS_SOCK está definido, se suscribe a los + // eventos de ciclo de vida y publica los crashes en el bus de agente. No-op si no hay init. + arje_link::spawn_if_configured(event_bus.clone()); + // 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) => { diff --git a/docs/10-roadmap.md b/docs/10-roadmap.md index af9c1bc9..d79e4066 100644 --- a/docs/10-roadmap.md +++ b/docs/10-roadmap.md @@ -287,17 +287,18 @@ Diseño completo en [SDD 11 — Bootstrap from-scratch](11-bootstrap.md). Resume `SWAP_BWRAP=1`. **Pendiente:** correr in-VM acumulando los swaps (`KVM=1 MEM=24576 SWAP_MAKE=1 SWAP_BUSYBOX=1 SWAP_LINUX_HEADERS=1 SWAP_BWRAP=1 ./scripts/selfhost-verify.sh`) para el `✓ REPRODUCIBLE`, y seguir con la última pieza: rust/llvm (la grande — ya hay infra de deps). -- 🔨 **B.2 — `CRASHED` real a la capa de IA (en progreso):** el `Event::Crashed` del bus de - agente ya no es sólo declarativo. **Fuente (lado arje, ✅):** `arje-bus` ganó - `BusRequest::Subscribe` + `BusPayload::Event(BusEvent)`; arje-zero difunde en `on_death` - `EnteCrashed{id,label,status}` / `EnteRestarting{delay_ms}` / `EnteExited` a las conexiones - suscritas, purgando las muertas (4 tests). **Sink (lado hammer, ✅):** `hammerd::crashes` - traduce la señal normalizada → `Event::Crashed` y la bombea al `EventBus` → - `/run/agent.sock` (3 tests). **Falta el último tramo:** el *adaptador de transporte* —un - thread que conecte a `$ARJE_BUS_SOCK`, mande `Subscribe` y reenvíe cada `BusEvent` a - `crashes::pump` (vía `BusClient::{subscribe,next_event}`)— y la decisión de cómo los dos - repos comparten el wire (dep directa a `arje-bus` vs. proto compartido vs. relectura del - frame postcard). Es el único paso que necesita un init vivo para validarse end-to-end. +- ✅ **B.2 — `CRASHED` real a la capa de IA (cableado end-to-end):** el `Event::Crashed` del + bus de agente ya tiene fuente real. **Fuente (arje):** `arje-bus` ganó `BusRequest::Subscribe` + + `BusPayload::Event(BusEvent)`; arje-zero difunde en `on_death` `EnteCrashed{id,label,status}` + / `EnteRestarting{delay_ms}` / `EnteExited` a las conexiones suscritas, purgando las muertas + (4 tests). **Sink (hammer):** `hammerd::crashes` traduce la señal normalizada → `Event::Crashed` + → `EventBus` → `/run/agent.sock` (3 tests). **Transporte (hammer):** `hammerd::arje_link` se + suscribe al bus de arje (`$ENTE_BUS_SOCK`) y reenvía cada `BusEvent`. Decisión de wire = + **relectura del frame postcard** (mirror mínimo del subconjunto de suscriptor; no arrastra el + crate-graph de arje ⇒ hammer sigue standalone). El layout está **verificado byte-a-byte contra + `arje-bus` real** (mismas versiones ulid 1.2 / postcard 1.1; frame Subscribe = `[00,01,00,0d]`) + y cubierto por un round-trip local (2 tests). Resta sólo el smoke end-to-end contra un init + vivo (no reproducible sin arje corriendo). - ⏭️ **También pendiente (Stage 1):** atestación arje (A1/A2 — ya cableada en el repo arje, resta sólo enchufar su veredicto de boot a este roadmap).