sergio/takana

1//! E2E del bucle de auto-reparación (Fase 6 cont.).
2//!
3//! Stub multi-conexión del bus: cada conexión hace handshake y luego empuja eventos
4//! pre-programados por el test. El orquestador abre una nueva conexión por intento
5//! (lo cuál nos da puntos limpios para "este intento crashea, este intento no").
6
7use std::io::{BufRead, BufReader, Write};
8use std::os::unix::net::{UnixListener, UnixStream};
9use std::path::PathBuf;
10use std::sync::{Arc, Mutex};
11use std::thread;
12use std::time::Duration;
13
14use takana_core::proto::{Cap, Command, Event, Peer, PROTOCOL_VERSION};
15use takana_core::swm::{Base, Mutation};
16use takana_core::{BaseRef, Swm};
17
18use takana_agent::orchestrator::{ApplyTarget, CompileMode, RepairPolicy};
19use takana_agent::{MockTranslator, Orchestrator};
20
21/// El plan por conexión: lista de eventos a empujar tras el Welcome. El stub itera
22/// por las conexiones en orden y consume un slot del plan por cada una.
23type ConnectionPlan = Vec<Event>;
24
25fn start_stub_bus(sock: PathBuf, plans: Vec<ConnectionPlan>) {
26    let _ = std::fs::remove_file(&sock);
27    let listener = UnixListener::bind(&sock).expect("bind stub");
28    let plans = Arc::new(Mutex::new(plans.into_iter()));
29    thread::Builder::new()
30        .name("stub-bus-repair".into())
31        .spawn(move || {
32            for incoming in listener.incoming() {
33                let Ok(stream) = incoming else { continue };
34                let next_plan = {
35                    let mut p = plans.lock().unwrap();
36                    p.next()
37                };
38                let plan = next_plan.unwrap_or_default();
39                thread::spawn(move || stub_serve(stream, plan));
40            }
41        })
42        .unwrap();
43}
44
45fn stub_serve(mut stream: UnixStream, plan: ConnectionPlan) {
46    // Handshake: leer Hello, responder Welcome.
47    let mut reader = BufReader::new(stream.try_clone().unwrap());
48    let mut first = String::new();
49    if reader.read_line(&mut first).is_err() || first.trim().is_empty() {
50        return;
51    }
52    let cmd: Command = match serde_json::from_str(first.trim()) {
53        Ok(c) => c,
54        Err(_) => return,
55    };
56    let Command::Hello { .. } = cmd else { return };
57    let welcome = Event::Welcome {
58        ver: PROTOCOL_VERSION,
59        caps: vec![Cap::Query, Cap::Compile, Cap::Inject, Cap::Init],
60        peer: Peer { uid: 1000, gid: 1000, pid: 1 },
61    };
62    let line = serde_json::to_string(&welcome).unwrap();
63    let _ = stream.write_all(line.as_bytes());
64    let _ = stream.write_all(b"\n");
65    let _ = stream.flush();
66
67    // Empujamos cada evento del plan inmediatamente. El cliente los recibirá vía
68    // `drain_async()` o `next_async()`. Los eventos del plan se ponen como asíncronos
69    // (Crashed/Modified) porque el orquestador no envía comandos durante wait_for_crash.
70    for ev in plan {
71        let line = serde_json::to_string(&ev).unwrap();
72        if stream.write_all(line.as_bytes()).is_err() {
73            return;
74        }
75        let _ = stream.write_all(b"\n");
76        let _ = stream.flush();
77    }
78
79    // Mantenemos abierto hasta que el cliente cierre.
80    let mut buf = String::new();
81    loop {
82        buf.clear();
83        match reader.read_line(&mut buf) {
84            Ok(0) => return,
85            Ok(_) => {}
86            Err(_) => return,
87        }
88    }
89}
90
91fn wait_for_sock(sock: &std::path::Path) {
92    for _ in 0..200 {
93        if sock.exists() {
94            return;
95        }
96        std::thread::sleep(Duration::from_millis(10));
97    }
98    panic!("stub bus no bindeó {}", sock.display());
99}
100
101fn baseref() -> BaseRef {
102    BaseRef {
103        distro_version: "2026-06-06".into(),
104        pins: Default::default(),
105    }
106}
107
108fn swm_touch(path: &str) -> Swm {
109    Swm {
110        swm_version: 1,
111        base: Base { distro_version: "2026-06-06".into(), pins: Default::default() },
112        mutations: vec![Mutation::FileDrop {
113            path: path.into(),
114            // hash de "X" (un byte 0x58) — calculado con BLAKE3.
115            content_hash: takana_core::ArtifactHash::of_inputs(&[b"X"]).as_str().to_string(),
116            content_b64: Some("WA==".into()),
117            content_url: None,
118        }],
119        signature: None,
120    }
121}
122
123#[test]
124fn repair_loop_reacts_to_crashed_event() {
125    let d = tempfile::tempdir().unwrap();
126    let sock = d.path().join("repair.sock");
127    let prefix = d.path().join("root");
128    std::fs::create_dir_all(&prefix).unwrap();
129
130    // Plan:
131    //   conn 1 (tras primer apply): empujamos Crashed → el bucle debe reintentar.
132    //   conn 2 (tras segundo apply): empujamos nada → la ventana expira → estabilizado.
133    let plans = vec![
134        vec![Event::Crashed { service: "web".into(), code: 137 }],
135        vec![],
136    ];
137    start_stub_bus(sock.clone(), plans);
138    wait_for_sock(&sock);
139
140    // Catálogo: la intención original toca /flag.original; la reparación toca /flag.repair.
141    let t = MockTranslator::empty()
142        .with_intent("install web", swm_touch("/flag.original"))
143        .with_intent(
144            "repair service web crashed code 137",
145            swm_touch("/flag.repair"),
146        );
147
148    let orch = Orchestrator::new(
149        t,
150        baseref(),
151        ApplyTarget::Prefix { prefix: prefix.clone() },
152        CompileMode::Skip,
153    );
154
155    let policy = RepairPolicy {
156        max_attempts: 3,
157        crash_window: Duration::from_millis(300),
158        event_sock: Some(sock.clone()),
159        ..RepairPolicy::default()
160    };
161
162    let proposal = orch
163        .run_with_repair("install web", &policy)
164        .expect("run_with_repair OK");
165
166    // Debe haber dos intentos: el original (con crash) y la reparación (estabilizada).
167    assert_eq!(proposal.repair_chain.len(), 2, "{:?}", proposal.repair_chain);
168    assert_eq!(proposal.repair_chain[0].intent, "install web");
169    assert_eq!(
170        proposal.repair_chain[0].triggered_crash.as_ref().map(|c| (c.service.as_str(), c.code)),
171        Some(("web", 137))
172    );
173    assert_eq!(
174        proposal.repair_chain[1].intent,
175        "repair service web crashed code 137"
176    );
177    assert!(proposal.repair_chain[1].triggered_crash.is_none());
178
179    // Ambos efectos en disco son visibles (cada apply re-rooteó bajo prefix).
180    assert!(prefix.join("flag.original").exists());
181    assert!(prefix.join("flag.repair").exists());
182}
183
184#[test]
185fn repair_loop_caps_attempts_when_crashes_persist() {
186    let d = tempfile::tempdir().unwrap();
187    let sock = d.path().join("persistent.sock");
188    let prefix = d.path().join("root");
189    std::fs::create_dir_all(&prefix).unwrap();
190
191    // Plan: TODAS las conexiones empujan Crashed. El bucle nunca estabiliza y debe
192    // parar al llegar a `max_attempts`.
193    let plans = vec![
194        vec![Event::Crashed { service: "x".into(), code: 1 }],
195        vec![Event::Crashed { service: "x".into(), code: 1 }],
196    ];
197    start_stub_bus(sock.clone(), plans);
198    wait_for_sock(&sock);
199
200    // El catálogo retorna el MISMO swm para la intención original y la reparada para
201    // que ambos intentos puedan resolverse. El propósito del test es el límite de intentos.
202    let swm = swm_touch("/flag.x");
203    let t = MockTranslator::empty()
204        .with_intent("intent A", swm.clone())
205        .with_intent("repair service x crashed code 1", swm);
206
207    let orch = Orchestrator::new(
208        t,
209        baseref(),
210        ApplyTarget::Prefix { prefix: prefix.clone() },
211        CompileMode::Skip,
212    );
213
214    let policy = RepairPolicy {
215        max_attempts: 2,
216        crash_window: Duration::from_millis(200),
217        event_sock: Some(sock.clone()),
218        ..RepairPolicy::default()
219    };
220
221    let proposal = orch
222        .run_with_repair("intent A", &policy)
223        .expect("run_with_repair devuelve OK aún con crashes persistentes");
224
225    assert_eq!(proposal.repair_chain.len(), 2);
226    // Ambos intentos crashearon ⇒ el último entry refleja el crash pendiente para
227    // que el humano lo vea.
228    assert!(proposal.repair_chain.last().unwrap().triggered_crash.is_some());
229}
230
231#[test]
232fn repair_loop_without_event_sock_runs_once() {
233    let d = tempfile::tempdir().unwrap();
234    let prefix = d.path().join("root");
235    std::fs::create_dir_all(&prefix).unwrap();
236    let t = MockTranslator::empty().with_intent("only one", swm_touch("/flag"));
237    let orch = Orchestrator::new(
238        t,
239        baseref(),
240        ApplyTarget::Prefix { prefix: prefix.clone() },
241        CompileMode::Skip,
242    );
243    let policy = RepairPolicy {
244        max_attempts: 5,
245        crash_window: Duration::from_millis(200),
246        event_sock: None,
247        ..RepairPolicy::default()
248    };
249    let p = orch.run_with_repair("only one", &policy).unwrap();
250    assert_eq!(p.repair_chain.len(), 1);
251    assert!(p.repair_chain[0].triggered_crash.is_none());
252}