sergio/takana

1//! Puente de crashes: la **fuente real** del `Event::Crashed` del bus de agente.
2//!
3//! Hasta ahora `Event::Crashed` sólo lo declaraba el protocolo y lo disparaban los
4//! tests: la Fase 5 difirió la "supervisión real" al init propio del track posterior.
5//! Ese init es **arje** (ver `docs/adr/0007-arje-como-init-propio.md`): como PID 1
6//! supervisa cada Ente, detecta su muerte en `on_death` y la difunde por su bus como
7//! `BusEvent::EnteCrashed` (crate `arje-bus`, `BusRequest::Subscribe` + `BusPayload::Event`).
8//!
9//! Este módulo es el **sink** de ese flujo (B.2 "exponer el CRASHED a la capa de IA"):
10//! traduce la señal de ciclo de vida normalizada al vocabulario del bus de agente de
11//! takana (`Event::Crashed`) y la bombea al [`EventBus`] → toda conexión del
12//! `/run/agent.sock` (la capa de IA) la recibe.
13//!
14//! ## Frontera de acoplamiento
15//!
16//! La **traducción** y el **bucle de bombeo** viven acá y son agnósticos del transporte
17//! —se testean sin un init vivo—. Lo único que falta para cerrar B.2 end-to-end es el
18//! *adaptador de transporte*: un thread que conecte al socket de arje
19//! (`$ARJE_BUS_SOCK`), mande un `Subscribe` y reenvíe cada `BusEvent` a un
20//! [`std::sync::mpsc::Sender<Lifecycle>`] hacia [`pump`]. Ese adaptador es lo que
21//! decide cómo los dos repos comparten el wire (dep directa a `arje-bus` vs. crate de
22//! proto compartido vs. relectura del frame postcard) — decisión registrada en
23//! `docs/09-trust-model.md` / el PLAN de arje, y validable sólo contra un init real.
24//!
25//! Las APIs quedan listas (`pub`) aunque el transporte aún no las invoque: ese es el
26//! patrón "vocabulario declarado, flujo por cablear" que arje-zero usa en su `events.rs`.
27#![allow(dead_code)]
28
29use std::sync::mpsc::Receiver;
30
31use takana_core::proto::Event;
32
33use crate::events::EventBus;
34
35/// Señal de ciclo de vida normalizada desde el supervisor (arje), independiente del
36/// wire concreto para no acoplar `hammerd` al crate `arje-bus`. El adaptador de
37/// transporte construye estos valores a partir de `arje_bus::BusEvent`.
38#[derive(Debug, Clone, PartialEq, Eq)]
39pub struct Lifecycle {
40    /// Identidad humana del Ente que cambió de estado (su `label` en arje). Es lo que
41    /// la capa de IA verá como `service`.
42    pub service: String,
43    pub status: LifeStatus,
44}
45
46/// Estado terminal/transicional de un Ente, ya normalizado.
47#[derive(Debug, Clone, PartialEq, Eq)]
48pub enum LifeStatus {
49    /// Terminó de forma anómala. `code` es el exit-code; para muertes por señal el
50    /// adaptador usa la convención de shell `128 + signum` para que sea siempre ≠ 0.
51    Crashed { code: i32 },
52    /// arje programó un reinicio tras backoff. Observabilidad — el `proto` actual no
53    /// tiene evento dedicado, así que no se traduce (todavía).
54    Restarting { delay_ms: u64 },
55    /// Terminó limpio (exit 0) y no se reinicia. Sin evento en el `proto` actual.
56    Exited,
57}
58
59/// Traduce una señal de ciclo de vida al evento del bus de agente. Devuelve `None`
60/// para los estados que el `proto` actual no expone (reinicios / salidas limpias):
61/// hoy sólo el crash es accionable por la IA.
62pub fn to_event(life: &Lifecycle) -> Option<Event> {
63    match &life.status {
64        LifeStatus::Crashed { code } => Some(Event::Crashed {
65            service: life.service.clone(),
66            code: *code,
67        }),
68        LifeStatus::Restarting { .. } | LifeStatus::Exited => None,
69    }
70}
71
72/// Bucle de bombeo: drena señales de la fuente (el adaptador de transporte de arje) y
73/// publica las accionables en el `EventBus`. Bloquea hasta que la fuente se cierra;
74/// pensado para correr en su propio thread, como los demás subsistemas de `hammerd`.
75pub fn pump(source: Receiver<Lifecycle>, bus: EventBus) {
76    for life in source {
77        if let Some(ev) = to_event(&life) {
78            bus.publish(&ev);
79        }
80    }
81}
82
83#[cfg(test)]
84mod tests {
85    use super::*;
86
87    #[test]
88    fn crash_traduce_a_event_crashed() {
89        let life = Lifecycle {
90            service: "web".into(),
91            status: LifeStatus::Crashed { code: 139 }, // 128 + SIGSEGV
92        };
93        assert_eq!(
94            to_event(&life),
95            Some(Event::Crashed { service: "web".into(), code: 139 })
96        );
97    }
98
99    #[test]
100    fn restarting_y_exited_no_producen_evento() {
101        let r = Lifecycle { service: "x".into(), status: LifeStatus::Restarting { delay_ms: 100 } };
102        let e = Lifecycle { service: "x".into(), status: LifeStatus::Exited };
103        assert_eq!(to_event(&r), None);
104        assert_eq!(to_event(&e), None);
105    }
106
107    #[test]
108    fn pump_publica_solo_los_crashes_en_el_bus() {
109        let bus = EventBus::new();
110        let sub = bus.subscribe();
111
112        let (tx, rx) = std::sync::mpsc::channel();
113        // Mezcla: un exited (descartado), un crash (publicado), un restarting (descartado).
114        tx.send(Lifecycle { service: "a".into(), status: LifeStatus::Exited }).unwrap();
115        tx.send(Lifecycle { service: "b".into(), status: LifeStatus::Crashed { code: 1 } }).unwrap();
116        tx.send(Lifecycle { service: "c".into(), status: LifeStatus::Restarting { delay_ms: 5 } }).unwrap();
117        drop(tx); // cierra la fuente → pump termina.
118
119        pump(rx, bus);
120
121        // Sólo el crash de "b" llegó al suscriptor.
122        match sub.try_recv().expect("debía haber un Crashed") {
123            Event::Crashed { service, code } => {
124                assert_eq!(service, "b");
125                assert_eq!(code, 1);
126            }
127            other => panic!("esperaba Crashed, fue {other:?}"),
128        }
129        assert!(sub.try_recv().is_err(), "no debía publicarse nada más");
130    }
131}