sergio/takana

1//! Adaptador de transporte arje-bus → bus de agente de hammer (último tramo de B.2).
2//!
3//! arje (el init/PID 1) supervisa cada Ente y, al morir, difunde `BusEvent` por su bus
4//! (`arje-bus`: `BusRequest::Subscribe` + `BusPayload::Event`). Este módulo se **suscribe** a
5//! ese stream y traduce cada evento al vocabulario de hammer (`crashes::Lifecycle` →
6//! `Event::Crashed`), publicándolo en el [`EventBus`] → `/run/agent.sock` → la capa de IA.
7//!
8//! ## Por qué un mirror y no una dependencia de `arje-bus`
9//!
10//! `arje-bus` arrastra el grafo de crates de arje (arje-card → card-core → …). Acoplarlo aquí
11//! rompería el build hermético/standalone de hammer. En su lugar **releemos el frame postcard**:
12//! el bus usa frames `u32-BE-len + postcard(BusMessage)`, y reproducimos sólo el subconjunto que
13//! un suscriptor necesita. El layout está verificado byte-a-byte contra `arje-bus` real (mismas
14//! versiones de `ulid` 1.2 y `postcard` 1.1; `ulid` serializa como string). Si arje reordena las
15//! variantes del enum, el contrato se rompe — por eso los discriminantes están documentados y
16//! cubiertos por un test de round-trip local.
17//!
18//! Varios campos del mirror (`from`, `seq`, `id`) no se leen: existen para **consumir** los
19//! bytes del wire al deserializar. Por eso el `allow(dead_code)` a nivel de módulo.
20#![allow(dead_code)]
21
22use std::io::{Read, Write};
23use std::os::unix::net::UnixStream;
24use std::path::PathBuf;
25use std::thread;
26
27use serde::Deserialize;
28use ulid::Ulid;
29
30use crate::crashes::{self, LifeStatus, Lifecycle};
31use crate::events::EventBus;
32
33/// Igual que `arje_bus::ENV_BUS_SOCK`: el env donde arje-zero publica la ruta del socket.
34pub const ENV_BUS_SOCK: &str = "ENTE_BUS_SOCK";
35
36/// Tope de frame, igual que `arje_bus::MAX_FRAME` (1 MiB) — protección contra OOM.
37const MAX_FRAME: usize = 1 << 20;
38
39// ── Mirror del wire de arje-bus (subconjunto de suscriptor) ──────────────────────────────
40//
41// Discriminantes (postcard = varint del índice de variante), verificados contra arje-bus:
42//   BusPayload: Request=0, Response=1, Event=2
43//   BusRequest::Subscribe = 13   (último; ver arje-bus/src/lib.rs)
44//   BusEvent: EnteCrashed=0, EnteRestarting=1, EnteExited=2
45//   LifecycleStatus: Exited=0, Killed=1
46
47#[derive(Deserialize, Debug)]
48struct BusMessage {
49    from: Option<Ulid>,
50    seq: u64,
51    payload: BusPayload,
52}
53
54#[derive(Deserialize, Debug)]
55enum BusPayload {
56    /// Índice 0. Nunca lo recibe un suscriptor (los Invoke van a proveedores, no a subs);
57    /// existe sólo para alinear el discriminante de `Response`/`Event`.
58    Request,
59    /// Índice 1. El único que recibimos por aquí es el `Ok` del ack de `Subscribe`.
60    Response(MiniResponse),
61    /// Índice 2. El stream de eventos de ciclo de vida.
62    Event(BusEvent),
63}
64
65#[derive(Deserialize, Debug)]
66enum MiniResponse {
67    Ok,
68    Error(String),
69}
70
71#[derive(Deserialize, Debug)]
72enum BusEvent {
73    EnteCrashed { id: Ulid, label: String, status: LifecycleStatus },
74    EnteRestarting { id: Ulid, label: String, delay_ms: u64 },
75    EnteExited { id: Ulid, label: String },
76}
77
78#[derive(Deserialize, Debug)]
79enum LifecycleStatus {
80    Exited(i32),
81    Killed(i32),
82}
83
84impl BusEvent {
85    /// Traduce el evento de arje a la señal normalizada de hammer. El `label` del Ente es el
86    /// `service`; una muerte por señal usa la convención de shell `128 + signum` para que el
87    /// código sea siempre ≠ 0.
88    fn to_lifecycle(self) -> Lifecycle {
89        match self {
90            BusEvent::EnteCrashed { label, status, .. } => {
91                let code = match status {
92                    LifecycleStatus::Exited(c) => c,
93                    LifecycleStatus::Killed(sig) => 128 + sig,
94                };
95                Lifecycle { service: label, status: LifeStatus::Crashed { code } }
96            }
97            BusEvent::EnteRestarting { label, delay_ms, .. } => {
98                Lifecycle { service: label, status: LifeStatus::Restarting { delay_ms } }
99            }
100            BusEvent::EnteExited { label, .. } => {
101                Lifecycle { service: label, status: LifeStatus::Exited }
102            }
103        }
104    }
105}
106
107/// Cuerpo del frame `Subscribe` (sin el prefijo de longitud), hand-encodeado: `from=None` +
108/// `seq` (varint) + `BusPayload::Request`(0) + `BusRequest::Subscribe`(13). Verificado
109/// byte-a-byte contra `postcard::to_stdvec` de arje-bus.
110fn subscribe_body(seq: u64) -> Vec<u8> {
111    let mut v = vec![0x00u8]; // Option::None
112    let mut s = seq; // u64 como varint LEB128
113    loop {
114        let b = (s & 0x7f) as u8;
115        s >>= 7;
116        if s == 0 {
117            v.push(b);
118            break;
119        } else {
120            v.push(b | 0x80);
121        }
122    }
123    v.push(0x00); // BusPayload::Request
124    v.push(0x0D); // BusRequest::Subscribe (índice 13)
125    v
126}
127
128fn write_frame(w: &mut impl Write, body: &[u8]) -> std::io::Result<()> {
129    w.write_all(&(body.len() as u32).to_be_bytes())?;
130    w.write_all(body)?;
131    w.flush()
132}
133
134fn read_frame(r: &mut impl Read) -> std::io::Result<Vec<u8>> {
135    let mut len = [0u8; 4];
136    r.read_exact(&mut len)?;
137    let n = u32::from_be_bytes(len) as usize;
138    if n > MAX_FRAME {
139        return Err(std::io::Error::new(
140            std::io::ErrorKind::InvalidData,
141            format!("frame oversize: {n} > {MAX_FRAME}"),
142        ));
143    }
144    let mut buf = vec![0u8; n];
145    r.read_exact(&mut buf)?;
146    Ok(buf)
147}
148
149/// Conecta al bus de arje, se suscribe y bombea eventos de ciclo de vida al `EventBus` hasta
150/// que la conexión se cierra. Bloqueante — pensado para correr en su propio thread.
151fn run(sock: &PathBuf, bus: &EventBus) -> std::io::Result<()> {
152    let mut stream = UnixStream::connect(sock)?;
153    write_frame(&mut stream, &subscribe_body(1))?;
154    tracing::info!(sock = %sock.display(), "arje-link: suscrito al bus de init");
155    loop {
156        let body = read_frame(&mut stream)?;
157        let msg: BusMessage = match postcard::from_bytes(&body) {
158            Ok(m) => m,
159            Err(e) => {
160                // Un frame que no entendemos no debe matar el puente: lo saltamos.
161                tracing::warn!(error = %e, "arje-link: frame indescifrable, ignoro");
162                continue;
163            }
164        };
165        match msg.payload {
166            BusPayload::Event(ev) => {
167                let life = ev.to_lifecycle();
168                if let Some(event) = crashes::to_event(&life) {
169                    bus.publish(&event);
170                }
171            }
172            // El ack del Subscribe y cualquier otra respuesta: nada que hacer.
173            BusPayload::Response(_) | BusPayload::Request => {}
174        }
175    }
176}
177
178/// Lanza el puente en un thread si `ENTE_BUS_SOCK` está definido (arje corriendo). Si no, es un
179/// no-op: hammerd sigue sin la fuente de crashes (los tests del bus no la necesitan). El thread
180/// es resiliente: si el bus muere, loguea y termina sin tumbar al daemon.
181pub fn spawn_if_configured(bus: EventBus) {
182    let Ok(path) = std::env::var(ENV_BUS_SOCK) else {
183        tracing::debug!("arje-link: {ENV_BUS_SOCK} no definido; sin fuente de crashes");
184        return;
185    };
186    let sock = PathBuf::from(path);
187    let _ = thread::Builder::new().name("arje-link".into()).spawn(move || {
188        if let Err(e) = run(&sock, &bus) {
189            tracing::warn!(error = %e, "arje-link: puente terminó (¿bus de init caído?)");
190        }
191    });
192}
193
194#[cfg(test)]
195mod tests {
196    use super::*;
197
198    #[test]
199    fn subscribe_body_es_el_frame_esperado() {
200        // Verificado contra arje-bus real: [from=None, seq=1, Request, Subscribe=13].
201        assert_eq!(subscribe_body(1), vec![0x00, 0x01, 0x00, 0x0D]);
202        // seq grande → varint multibyte (300 = 0xAC 0x02).
203        assert_eq!(subscribe_body(300), vec![0x00, 0xAC, 0x02, 0x00, 0x0D]);
204    }
205
206    /// Round-trip local con el MISMO `postcard`/`ulid` que arje-bus: serializamos un
207    /// `BusMessage::Event` con tipos espejo `Serialize` y lo deserializamos con los `Deserialize`
208    /// reales del módulo. Si los discriminantes/orden de campos se desalinean, esto rompe.
209    #[test]
210    fn evento_crashed_roundtrips_por_el_wire() {
211        use serde::Serialize;
212        #[derive(Serialize)]
213        enum TxPayload {
214            #[allow(dead_code)]
215            Request,
216            #[allow(dead_code)]
217            Response,
218            Event(TxEvent),
219        }
220        #[derive(Serialize)]
221        enum TxEvent {
222            EnteCrashed { id: Ulid, label: String, status: TxStatus },
223        }
224        #[derive(Serialize)]
225        enum TxStatus {
226            #[allow(dead_code)]
227            Exited(i32),
228            Killed(i32),
229        }
230        #[derive(Serialize)]
231        struct TxMsg {
232            from: Option<Ulid>,
233            seq: u64,
234            payload: TxPayload,
235        }
236
237        let id = Ulid::from_string("01ARZ3NDEKTSV4RRFFQ69G5FAV").unwrap();
238        let bytes = postcard::to_stdvec(&TxMsg {
239            from: Some(id),
240            seq: 7,
241            payload: TxPayload::Event(TxEvent::EnteCrashed {
242                id,
243                label: "demonio".into(),
244                status: TxStatus::Killed(11),
245            }),
246        })
247        .unwrap();
248
249        let msg: BusMessage = postcard::from_bytes(&bytes).unwrap();
250        match msg.payload {
251            BusPayload::Event(ev) => {
252                let life = ev.to_lifecycle();
253                assert_eq!(life.service, "demonio");
254                // Killed(11) → 128 + 11 = 139.
255                assert_eq!(life.status, LifeStatus::Crashed { code: 139 });
256            }
257            other => panic!("esperaba Event, fue {other:?}"),
258        }
259    }
260}