hammer/crates/hammer-agent/tests/client_against_stub_bus.rs

//! E2E del cliente del bus (Fase 6) contra un servidor stub.
//!
//! No depende de `hammerd` ni del lab: un mini-servidor JSON-líneas que conoce el
//! handshake y responde a `Compile` con `BuildReady` sintético. Demuestra que el cliente
//! síncrono multiplexa correctamente respuestas (BuildReady) y eventos asíncronos (Modified).

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

use hammer_core::proto::{Cap, Command, Event, Peer, RecipeInline, PROTOCOL_VERSION};

fn start_stub_bus(sock: PathBuf) -> std::sync::mpsc::Sender<Event> {
    let (tx_async, rx_async) = std::sync::mpsc::channel::<Event>();
    let _ = std::fs::remove_file(&sock);
    let listener = UnixListener::bind(&sock).expect("bind stub");
    thread::Builder::new()
        .name("stub-bus".into())
        .spawn(move || {
            // Sólo aceptamos una conexión: el Receiver no es Clone, se mueve al handler.
            if let Some(Ok(stream)) = listener.incoming().next() {
                thread::spawn(move || stub_serve(stream, rx_async));
            }
        })
        .unwrap();
    tx_async
}

fn stub_serve(stream: UnixStream, rx_async: std::sync::mpsc::Receiver<Event>) {
    let reader_stream = stream.try_clone().unwrap();
    let mut writer = stream;
    let writer_clone = writer.try_clone().unwrap();

    // Hilo que reenvía los eventos asíncronos al cliente. Mover el writer impide que
    // se mezclen los bytes con las respuestas (cada Sender los serializa atómicamente).
    let async_writer = std::sync::Arc::new(std::sync::Mutex::new(writer_clone));
    let aw_for_async = async_writer.clone();
    thread::spawn(move || {
        for ev in rx_async {
            let line = serde_json::to_string(&ev).unwrap();
            let mut g = aw_for_async.lock().unwrap();
            let _ = g.write_all(line.as_bytes());
            let _ = g.write_all(b"\n");
            let _ = g.flush();
        }
    });

    let mut reader = BufReader::new(reader_stream);
    let mut first = String::new();
    reader.read_line(&mut first).unwrap();
    let cmd: Command = serde_json::from_str(first.trim()).unwrap();
    let Command::Hello { ver: _, client: _ } = cmd else {
        return;
    };
    // welcome
    let w = Event::Welcome {
        ver: PROTOCOL_VERSION,
        caps: vec![Cap::Query, Cap::Compile, Cap::Inject, Cap::Init],
        peer: Peer { uid: 1000, gid: 1000, pid: 1 },
    };
    let mut g = async_writer.lock().unwrap();
    g.write_all(serde_json::to_string(&w).unwrap().as_bytes()).unwrap();
    g.write_all(b"\n").unwrap();
    g.flush().unwrap();
    drop(g);

    // Loop de comandos.
    loop {
        let mut line = String::new();
        match reader.read_line(&mut line) {
            Ok(0) => break,
            Ok(_) => {}
            Err(_) => break,
        }
        if line.trim().is_empty() {
            continue;
        }
        let cmd: Command = match serde_json::from_str(line.trim()) {
            Ok(c) => c,
            Err(_) => continue,
        };
        let response = match cmd {
            Command::Compile { recipe } => Event::BuildReady {
                recipe: recipe.name.clone(),
                artifact: format!("b3:stub-{}", recipe.commit),
            },
            Command::Query { what, .. } => Event::QueryResult {
                what,
                value: serde_json::json!({"stub": true}),
            },
            Command::Init { cmd } => Event::InitAck { cmd },
            Command::Inject { artifact, target, .. } => Event::Injected {
                artifact,
                target,
                files: 1,
            },
            Command::Hello { .. } => Event::Error {
                code: "duplicate_hello".into(),
                msg: "ya hicimos hello".into(),
            },
        };
        let mut g = async_writer.lock().unwrap();
        g.write_all(serde_json::to_string(&response).unwrap().as_bytes())
            .unwrap();
        g.write_all(b"\n").unwrap();
        g.flush().unwrap();
        drop(g);

        // Cierra solo si el cliente cierra; aquí seguimos.
        if !writer.flush().is_ok() {
            break;
        }
    }
}

#[test]
fn client_handshake_compile_and_async_modified() {
    let d = tempfile::tempdir().unwrap();
    let sock = d.path().join("stub.sock");
    let async_tx = start_stub_bus(sock.clone());

    // Pequeño wait para que el listener bindee.
    for _ in 0..50 {
        if sock.exists() { break }
        thread::sleep(Duration::from_millis(10));
    }

    let mut client = hammer_agent::AgentClient::connect(&sock).expect("connect");
    assert_eq!(client.welcome.ver, PROTOCOL_VERSION);
    assert!(client.welcome.caps.contains(&Cap::Compile));

    let recipe = RecipeInline {
        name: "grep".into(),
        repo: "git://example/grep.git".into(),
        commit: "abc123".into(),
        patch: None,
        compiler: "zig-cc".into(),
        target: "x86_64-linux-musl".into(),
        link: "static".into(),
        flags: vec![],
    };
    let artifact = client
        .compile(recipe, Duration::from_secs(2))
        .expect("BuildReady");
    assert_eq!(artifact, "b3:stub-abc123");

    // El stub también puede empujar eventos asíncronos. Inyectamos un Modified y vemos
    // que el cliente lo entrega por drain_async()/next_async().
    async_tx
        .send(Event::Modified {
            path: "/bin/grep".into(),
            op: "replace".into(),
            ts: "2026-06-09T00:00:00Z".into(),
        })
        .unwrap();
    let ev = client.next_async(Duration::from_secs(2)).expect("async");
    match ev {
        Event::Modified { path, .. } => assert_eq!(path, "/bin/grep"),
        other => panic!("esperaba Modified, llegó {other:?}"),
    }
}