sergio/takana

1//! Adaptador de transporte arje-bus → bus de agente de takana (ú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 takana (`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 takana. 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 = 14   (arje insertó StopCardFromDisk el 2026-06-26 y corrió
44//                                 el índice 13→14; re-sincronizado 2026-07-16 — ver el
45//                                 golden test de arje-bus `contrato_wire`)
46//   BusEvent: EnteCrashed=0, EnteRestarting=1, EnteExited=2, EnteParked=3, EnteRefloored=4
47//   LifecycleStatus: Exited=0, Killed=1
48
49#[derive(Deserialize, Debug)]
50struct BusMessage {
51    from: Option<Ulid>,
52    seq: u64,
53    payload: BusPayload,
54}
55
56#[derive(Deserialize, Debug)]
57enum BusPayload {
58    /// Índice 0. Nunca lo recibe un suscriptor (los Invoke van a proveedores, no a subs);
59    /// existe sólo para alinear el discriminante de `Response`/`Event`.
60    Request,
61    /// Índice 1. El único que recibimos por aquí es el `Ok` del ack de `Subscribe`.
62    Response(MiniResponse),
63    /// Índice 2. El stream de eventos de ciclo de vida.
64    Event(BusEvent),
65}
66
67#[derive(Deserialize, Debug)]
68enum MiniResponse {
69    Ok,
70    Error(String),
71}
72
73#[derive(Deserialize, Debug)]
74enum BusEvent {
75    EnteCrashed { id: Ulid, label: String, status: LifecycleStatus },
76    EnteRestarting { id: Ulid, label: String, delay_ms: u64 },
77    EnteExited { id: Ulid, label: String },
78    /// Ente `Restart` aparcado: su "piso" (una capability) perdió proveedor.
79    /// No es un crash — no hay señal para la capa de IA; se consume del wire
80    /// para no tirar el frame como indescifrable.
81    EnteParked { id: Ulid, label: String },
82    /// El piso volvió y arje re-erigió al Ente aparcado. Ídem: sólo consumo.
83    EnteRefloored { id: Ulid, label: String },
84}
85
86#[derive(Deserialize, Debug)]
87enum LifecycleStatus {
88    Exited(i32),
89    Killed(i32),
90}
91
92impl BusEvent {
93    /// Traduce el evento de arje a la señal normalizada de takana. El `label` del Ente es el
94    /// `service`; una muerte por señal usa la convención de shell `128 + signum` para que el
95    /// código sea siempre ≠ 0. `None` = evento consumido sin señal (Parked/Refloored: son
96    /// vaivén del piso gráfico, no ciclo de vida que le importe a la capa de IA).
97    fn to_lifecycle(self) -> Option<Lifecycle> {
98        match self {
99            BusEvent::EnteCrashed { label, status, .. } => {
100                let code = match status {
101                    LifecycleStatus::Exited(c) => c,
102                    LifecycleStatus::Killed(sig) => 128 + sig,
103                };
104                Some(Lifecycle { service: label, status: LifeStatus::Crashed { code } })
105            }
106            BusEvent::EnteRestarting { label, delay_ms, .. } => {
107                Some(Lifecycle { service: label, status: LifeStatus::Restarting { delay_ms } })
108            }
109            BusEvent::EnteExited { label, .. } => {
110                Some(Lifecycle { service: label, status: LifeStatus::Exited })
111            }
112            BusEvent::EnteParked { .. } | BusEvent::EnteRefloored { .. } => None,
113        }
114    }
115}
116
117/// Cuerpo del frame `Subscribe` (sin el prefijo de longitud), hand-encodeado: `from=None` +
118/// `seq` (varint) + `BusPayload::Request`(0) + `BusRequest::Subscribe`(14). Verificado
119/// byte-a-byte contra `postcard::to_stdvec` de arje-bus (golden test `contrato_wire`).
120fn subscribe_body(seq: u64) -> Vec<u8> {
121    let mut v = vec![0x00u8]; // Option::None
122    let mut s = seq; // u64 como varint LEB128
123    loop {
124        let b = (s & 0x7f) as u8;
125        s >>= 7;
126        if s == 0 {
127            v.push(b);
128            break;
129        } else {
130            v.push(b | 0x80);
131        }
132    }
133    v.push(0x00); // BusPayload::Request
134    v.push(0x0E); // BusRequest::Subscribe (índice 14 desde 2026-06-26)
135    v
136}
137
138fn write_frame(w: &mut impl Write, body: &[u8]) -> std::io::Result<()> {
139    w.write_all(&(body.len() as u32).to_be_bytes())?;
140    w.write_all(body)?;
141    w.flush()
142}
143
144fn read_frame(r: &mut impl Read) -> std::io::Result<Vec<u8>> {
145    let mut len = [0u8; 4];
146    r.read_exact(&mut len)?;
147    let n = u32::from_be_bytes(len) as usize;
148    if n > MAX_FRAME {
149        return Err(std::io::Error::new(
150            std::io::ErrorKind::InvalidData,
151            format!("frame oversize: {n} > {MAX_FRAME}"),
152        ));
153    }
154    let mut buf = vec![0u8; n];
155    r.read_exact(&mut buf)?;
156    Ok(buf)
157}
158
159/// Conecta al bus de arje, se suscribe y bombea eventos de ciclo de vida al `EventBus` hasta
160/// que la conexión se cierra. Bloqueante — pensado para correr en su propio thread.
161/// `pub(crate)` para que el e2e (`tests/arje_link_e2e.rs`) lo maneje contra un bus falso.
162pub(crate) fn run(sock: &PathBuf, bus: &EventBus) -> std::io::Result<()> {
163    let mut stream = UnixStream::connect(sock)?;
164    write_frame(&mut stream, &subscribe_body(1))?;
165    tracing::info!(sock = %sock.display(), "arje-link: suscrito al bus de init");
166    loop {
167        let body = read_frame(&mut stream)?;
168        let msg: BusMessage = match postcard::from_bytes(&body) {
169            Ok(m) => m,
170            Err(e) => {
171                // Un frame que no entendemos no debe matar el puente: lo saltamos.
172                tracing::warn!(error = %e, "arje-link: frame indescifrable, ignoro");
173                continue;
174            }
175        };
176        match msg.payload {
177            BusPayload::Event(ev) => {
178                if let Some(life) = ev.to_lifecycle() {
179                    if let Some(event) = crashes::to_event(&life) {
180                        bus.publish(&event);
181                    }
182                }
183            }
184            // El ack del Subscribe y cualquier otra respuesta: nada que hacer.
185            BusPayload::Response(_) | BusPayload::Request => {}
186        }
187    }
188}
189
190/// Lanza el puente en un thread si `ENTE_BUS_SOCK` está definido (arje corriendo). Si no, es un
191/// no-op: hammerd sigue sin la fuente de crashes (los tests del bus no la necesitan). El thread
192/// es resiliente: si el bus muere, loguea y termina sin tumbar al daemon.
193pub fn spawn_if_configured(bus: EventBus) {
194    let Ok(path) = std::env::var(ENV_BUS_SOCK) else {
195        tracing::debug!("arje-link: {ENV_BUS_SOCK} no definido; sin fuente de crashes");
196        return;
197    };
198    let sock = PathBuf::from(path);
199    let _ = thread::Builder::new().name("arje-link".into()).spawn(move || {
200        if let Err(e) = run(&sock, &bus) {
201            tracing::warn!(error = %e, "arje-link: puente terminó (¿bus de init caído?)");
202        }
203    });
204}
205
206#[cfg(test)]
207mod tests {
208    use super::*;
209
210    #[test]
211    fn subscribe_body_es_el_frame_esperado() {
212        // Verificado contra arje-bus real: [from=None, seq=1, Request, Subscribe=14].
213        assert_eq!(subscribe_body(1), vec![0x00, 0x01, 0x00, 0x0E]);
214        // seq grande → varint multibyte (300 = 0xAC 0x02).
215        assert_eq!(subscribe_body(300), vec![0x00, 0xAC, 0x02, 0x00, 0x0E]);
216    }
217
218    /// Round-trip local con el MISMO `postcard`/`ulid` que arje-bus: serializamos un
219    /// `BusMessage::Event` con tipos espejo `Serialize` y lo deserializamos con los `Deserialize`
220    /// reales del módulo. Si los discriminantes/orden de campos se desalinean, esto rompe.
221    #[test]
222    fn evento_crashed_roundtrips_por_el_wire() {
223        use serde::Serialize;
224        #[derive(Serialize)]
225        enum TxPayload {
226            #[allow(dead_code)]
227            Request,
228            #[allow(dead_code)]
229            Response,
230            Event(TxEvent),
231        }
232        #[derive(Serialize)]
233        enum TxEvent {
234            EnteCrashed { id: Ulid, label: String, status: TxStatus },
235        }
236        #[derive(Serialize)]
237        enum TxStatus {
238            #[allow(dead_code)]
239            Exited(i32),
240            Killed(i32),
241        }
242        #[derive(Serialize)]
243        struct TxMsg {
244            from: Option<Ulid>,
245            seq: u64,
246            payload: TxPayload,
247        }
248
249        let id = Ulid::from_string("01ARZ3NDEKTSV4RRFFQ69G5FAV").unwrap();
250        let bytes = postcard::to_stdvec(&TxMsg {
251            from: Some(id),
252            seq: 7,
253            payload: TxPayload::Event(TxEvent::EnteCrashed {
254                id,
255                label: "demonio".into(),
256                status: TxStatus::Killed(11),
257            }),
258        })
259        .unwrap();
260
261        let msg: BusMessage = postcard::from_bytes(&bytes).unwrap();
262        match msg.payload {
263            BusPayload::Event(ev) => {
264                let life = ev.to_lifecycle().expect("un crash sí produce señal");
265                assert_eq!(life.service, "demonio");
266                // Killed(11) → 128 + 11 = 139.
267                assert_eq!(life.status, LifeStatus::Crashed { code: 139 });
268            }
269            other => panic!("esperaba Event, fue {other:?}"),
270        }
271    }
272}