takana/crates/hammer-agent/tests/repair_loop.rs

//! E2E del bucle de auto-reparación (Fase 6 cont.).
//!
//! Stub multi-conexión del bus: cada conexión hace handshake y luego empuja eventos
//! pre-programados por el test. El orquestador abre una nueva conexión por intento
//! (lo cuál nos da puntos limpios para "este intento crashea, este intento no").

use std::io::{BufRead, BufReader, Write};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

use hammer_core::proto::{Cap, Command, Event, Peer, PROTOCOL_VERSION};
use hammer_core::swm::{Base, Mutation};
use hammer_core::{BaseRef, Swm};

use hammer_agent::orchestrator::{ApplyTarget, CompileMode, RepairPolicy};
use hammer_agent::{MockTranslator, Orchestrator};

/// El plan por conexión: lista de eventos a empujar tras el Welcome. El stub itera
/// por las conexiones en orden y consume un slot del plan por cada una.
type ConnectionPlan = Vec<Event>;

fn start_stub_bus(sock: PathBuf, plans: Vec<ConnectionPlan>) {
    let _ = std::fs::remove_file(&sock);
    let listener = UnixListener::bind(&sock).expect("bind stub");
    let plans = Arc::new(Mutex::new(plans.into_iter()));
    thread::Builder::new()
        .name("stub-bus-repair".into())
        .spawn(move || {
            for incoming in listener.incoming() {
                let Ok(stream) = incoming else { continue };
                let next_plan = {
                    let mut p = plans.lock().unwrap();
                    p.next()
                };
                let plan = next_plan.unwrap_or_default();
                thread::spawn(move || stub_serve(stream, plan));
            }
        })
        .unwrap();
}

fn stub_serve(mut stream: UnixStream, plan: ConnectionPlan) {
    // Handshake: leer Hello, responder Welcome.
    let mut reader = BufReader::new(stream.try_clone().unwrap());
    let mut first = String::new();
    if reader.read_line(&mut first).is_err() || first.trim().is_empty() {
        return;
    }
    let cmd: Command = match serde_json::from_str(first.trim()) {
        Ok(c) => c,
        Err(_) => return,
    };
    let Command::Hello { .. } = cmd else { return };
    let welcome = Event::Welcome {
        ver: PROTOCOL_VERSION,
        caps: vec![Cap::Query, Cap::Compile, Cap::Inject, Cap::Init],
        peer: Peer { uid: 1000, gid: 1000, pid: 1 },
    };
    let line = serde_json::to_string(&welcome).unwrap();
    let _ = stream.write_all(line.as_bytes());
    let _ = stream.write_all(b"\n");
    let _ = stream.flush();

    // Empujamos cada evento del plan inmediatamente. El cliente los recibirá vía
    // `drain_async()` o `next_async()`. Los eventos del plan se ponen como asíncronos
    // (Crashed/Modified) porque el orquestador no envía comandos durante wait_for_crash.
    for ev in plan {
        let line = serde_json::to_string(&ev).unwrap();
        if stream.write_all(line.as_bytes()).is_err() {
            return;
        }
        let _ = stream.write_all(b"\n");
        let _ = stream.flush();
    }

    // Mantenemos abierto hasta que el cliente cierre.
    let mut buf = String::new();
    loop {
        buf.clear();
        match reader.read_line(&mut buf) {
            Ok(0) => return,
            Ok(_) => {}
            Err(_) => return,
        }
    }
}

fn wait_for_sock(sock: &std::path::Path) {
    for _ in 0..200 {
        if sock.exists() {
            return;
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    panic!("stub bus no bindeó {}", sock.display());
}

fn baseref() -> BaseRef {
    BaseRef {
        distro_version: "2026-06-06".into(),
        pins: Default::default(),
    }
}

fn swm_touch(path: &str) -> Swm {
    Swm {
        swm_version: 1,
        base: Base { distro_version: "2026-06-06".into(), pins: Default::default() },
        mutations: vec![Mutation::FileDrop {
            path: path.into(),
            // hash de "X" (un byte 0x58) — calculado con BLAKE3.
            content_hash: hammer_core::ArtifactHash::of_inputs(&[b"X"]).as_str().to_string(),
            content_b64: Some("WA==".into()),
            content_url: None,
        }],
        signature: None,
    }
}

#[test]
fn repair_loop_reacts_to_crashed_event() {
    let d = tempfile::tempdir().unwrap();
    let sock = d.path().join("repair.sock");
    let prefix = d.path().join("root");
    std::fs::create_dir_all(&prefix).unwrap();

    // Plan:
    //   conn 1 (tras primer apply): empujamos Crashed → el bucle debe reintentar.
    //   conn 2 (tras segundo apply): empujamos nada → la ventana expira → estabilizado.
    let plans = vec![
        vec![Event::Crashed { service: "web".into(), code: 137 }],
        vec![],
    ];
    start_stub_bus(sock.clone(), plans);
    wait_for_sock(&sock);

    // Catálogo: la intención original toca /flag.original; la reparación toca /flag.repair.
    let t = MockTranslator::empty()
        .with_intent("install web", swm_touch("/flag.original"))
        .with_intent(
            "repair service web crashed code 137",
            swm_touch("/flag.repair"),
        );

    let orch = Orchestrator::new(
        t,
        baseref(),
        ApplyTarget::Prefix { prefix: prefix.clone() },
        CompileMode::Skip,
    );

    let policy = RepairPolicy {
        max_attempts: 3,
        crash_window: Duration::from_millis(300),
        event_sock: Some(sock.clone()),
        ..RepairPolicy::default()
    };

    let proposal = orch
        .run_with_repair("install web", &policy)
        .expect("run_with_repair OK");

    // Debe haber dos intentos: el original (con crash) y la reparación (estabilizada).
    assert_eq!(proposal.repair_chain.len(), 2, "{:?}", proposal.repair_chain);
    assert_eq!(proposal.repair_chain[0].intent, "install web");
    assert_eq!(
        proposal.repair_chain[0].triggered_crash.as_ref().map(|c| (c.service.as_str(), c.code)),
        Some(("web", 137))
    );
    assert_eq!(
        proposal.repair_chain[1].intent,
        "repair service web crashed code 137"
    );
    assert!(proposal.repair_chain[1].triggered_crash.is_none());

    // Ambos efectos en disco son visibles (cada apply re-rooteó bajo prefix).
    assert!(prefix.join("flag.original").exists());
    assert!(prefix.join("flag.repair").exists());
}

#[test]
fn repair_loop_caps_attempts_when_crashes_persist() {
    let d = tempfile::tempdir().unwrap();
    let sock = d.path().join("persistent.sock");
    let prefix = d.path().join("root");
    std::fs::create_dir_all(&prefix).unwrap();

    // Plan: TODAS las conexiones empujan Crashed. El bucle nunca estabiliza y debe
    // parar al llegar a `max_attempts`.
    let plans = vec![
        vec![Event::Crashed { service: "x".into(), code: 1 }],
        vec![Event::Crashed { service: "x".into(), code: 1 }],
    ];
    start_stub_bus(sock.clone(), plans);
    wait_for_sock(&sock);

    // El catálogo retorna el MISMO swm para la intención original y la reparada para
    // que ambos intentos puedan resolverse. El propósito del test es el límite de intentos.
    let swm = swm_touch("/flag.x");
    let t = MockTranslator::empty()
        .with_intent("intent A", swm.clone())
        .with_intent("repair service x crashed code 1", swm);

    let orch = Orchestrator::new(
        t,
        baseref(),
        ApplyTarget::Prefix { prefix: prefix.clone() },
        CompileMode::Skip,
    );

    let policy = RepairPolicy {
        max_attempts: 2,
        crash_window: Duration::from_millis(200),
        event_sock: Some(sock.clone()),
        ..RepairPolicy::default()
    };

    let proposal = orch
        .run_with_repair("intent A", &policy)
        .expect("run_with_repair devuelve OK aún con crashes persistentes");

    assert_eq!(proposal.repair_chain.len(), 2);
    // Ambos intentos crashearon ⇒ el último entry refleja el crash pendiente para
    // que el humano lo vea.
    assert!(proposal.repair_chain.last().unwrap().triggered_crash.is_some());
}

#[test]
fn repair_loop_without_event_sock_runs_once() {
    let d = tempfile::tempdir().unwrap();
    let prefix = d.path().join("root");
    std::fs::create_dir_all(&prefix).unwrap();
    let t = MockTranslator::empty().with_intent("only one", swm_touch("/flag"));
    let orch = Orchestrator::new(
        t,
        baseref(),
        ApplyTarget::Prefix { prefix: prefix.clone() },
        CompileMode::Skip,
    );
    let policy = RepairPolicy {
        max_attempts: 5,
        crash_window: Duration::from_millis(200),
        event_sock: None,
        ..RepairPolicy::default()
    };
    let p = orch.run_with_repair("only one", &policy).unwrap();
    assert_eq!(p.repair_chain.len(), 1);
    assert!(p.repair_chain[0].triggered_crash.is_none());
}