hammer/crates/hammerd/src/bus.rs

//! Bus de agente: `/run/agent.sock`. Ver `docs/07-agent-bus.md`.
//!
//! Modelo de threading (síncrono, una conexión = N threads pequeños):
//!
//! ```text
//!   [accept loop] ── spawn ──▶ conn{
//!        [reader thread] ── dispatch ──┐
//!        [bus forwarder]  ─────────────┼──▶ tx ──▶ [writer thread] ──▶ socket
//!        [compile worker] (puntual) ───┘
//!   }
//! ```
//!
//! No usamos `tokio` para no arrastrar runtime asíncrono: la cadencia esperada del bus es
//! humana + agente (decenas de mensajes/seg en el peor caso). Cada conexión tiene su propio
//! canal `Sender<Event>`; la `EventBus` global re-emite Modified/Crashed a esos canales.

use std::io::{BufRead, BufReader, Write};
use std::os::fd::AsFd;
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::sync::mpsc::{self, Sender};
use std::sync::{Arc, Mutex};
use std::thread;

use nix::sys::socket::{getsockopt, sockopt::PeerCredentials};

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

/// Política de capacidades: dado el `Peer` (ya autenticado por SO_PEERCRED), devuelve qué
/// caps tiene esa conexión. Por defecto: misma UID que el daemon ⇒ todo menos `inject-real`;
/// otros UIDs ⇒ sólo `query`. La policy más expresiva (lectura de un toml en `/etc/hammer`)
/// es trabajo posterior; la firma del callable basta para encajarla sin tocar este módulo.
pub type CapsPolicy = Arc<dyn Fn(&Peer) -> Vec<Cap> + Send + Sync>;

pub fn default_policy() -> CapsPolicy {
    let my_uid = unsafe { libc::getuid() };
    Arc::new(move |peer: &Peer| {
        if peer.uid == my_uid {
            vec![Cap::Query, Cap::Compile, Cap::Inject, Cap::Init]
        } else {
            vec![Cap::Query]
        }
    })
}

/// Política derivada de un fichero `agent-caps.toml` (SDD 07 §4). El peer no elige sus caps;
/// las resuelve la config por `(uid, gid)`.
pub fn policy_from_config(cfg: hammer_core::AgentCapsConfig) -> CapsPolicy {
    Arc::new(move |peer: &Peer| cfg.caps_for(peer.uid, peer.gid))
}

/// Contexto que necesita el dispatcher para resolver comandos. Compartido por todas las
/// conexiones (clones baratos: paths + Arc).
#[derive(Clone)]
pub struct BusContext {
    pub store_root: PathBuf,
    pub init_control: PathBuf,
    pub events: EventBus,
}

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("io: {0}")]
    Io(#[from] std::io::Error),
    #[error("bind {0}: {1}")]
    Bind(PathBuf, std::io::Error),
    #[error("peer creds: {0}")]
    Peer(String),
}

pub type Result<T> = std::result::Result<T, Error>;

/// Arranca el listener bloqueante. Pensado para correr en su propio thread desde `main`.
pub fn serve_agent_bus(
    sock_path: &Path,
    policy: CapsPolicy,
    ctx: BusContext,
) -> Result<()> {
    // Limpia un socket olvidado de una ejecución previa. Si el path está ocupado por algo
    // que no es un socket, dejamos que `bind` falle — no queremos rm-rf por accidente.
    let _ = std::fs::remove_file(sock_path);
    let listener = UnixListener::bind(sock_path)
        .map_err(|e| Error::Bind(sock_path.to_path_buf(), e))?;
    tracing::info!(sock = %sock_path.display(), "bus: escuchando");

    for stream in listener.incoming() {
        match stream {
            Ok(s) => {
                let policy = policy.clone();
                let ctx = ctx.clone();
                thread::Builder::new()
                    .name("bus-conn".into())
                    .spawn(move || {
                        if let Err(e) = handle_connection(s, policy, ctx) {
                            tracing::warn!(error = %e, "bus: conexión terminó con error");
                        }
                    })
                    .ok();
            }
            Err(e) => tracing::warn!(error = %e, "bus: accept falló"),
        }
    }
    Ok(())
}

fn read_peer(stream: &UnixStream) -> Result<Peer> {
    let cred = getsockopt(&stream.as_fd(), PeerCredentials)
        .map_err(|e| Error::Peer(e.to_string()))?;
    Ok(Peer {
        uid: cred.uid(),
        gid: cred.gid(),
        pid: cred.pid(),
    })
}

fn handle_connection(stream: UnixStream, policy: CapsPolicy, ctx: BusContext) -> Result<()> {
    let peer = read_peer(&stream)?;
    let caps = policy(&peer);
    tracing::info!(?peer, ?caps, "bus: nueva conexión");

    // Canal de eventos hacia el writer. Todos los productores (dispatcher, bus forwarder,
    // workers de COMPILE) clonan este Sender.
    let (tx, rx) = mpsc::channel::<Event>();

    // Writer thread: serializa todo lo que recibe del canal y lo escribe al socket.
    let stream_writer = stream.try_clone()?;
    let _writer_handle = thread::Builder::new()
        .name("bus-conn-writer".into())
        .spawn(move || writer_loop(stream_writer, rx))
        .ok();

    // Bus forwarder thread: cualquier evento global (Modified/Crashed) se re-emite a este
    // peer. Se cierra solo cuando el global drop el último Sender (en la práctica, nunca)
    // o cuando nuestro tx local cae (peer cerró).
    let bus_rx = ctx.events.subscribe();
    let bus_tx = tx.clone();
    thread::Builder::new()
        .name("bus-conn-fwd".into())
        .spawn(move || {
            for ev in bus_rx {
                if bus_tx.send(ev).is_err() {
                    break;
                }
            }
        })
        .ok();

    // Handshake. La primera línea debe ser Hello.
    let mut reader = BufReader::new(stream);
    let mut first = String::new();
    reader.read_line(&mut first)?;
    match serde_json::from_str::<Command>(first.trim()) {
        Ok(Command::Hello { ver, client }) => {
            tracing::info!(ver, client = %client, "bus: hello");
            let _ = tx.send(Event::Welcome {
                ver: PROTOCOL_VERSION,
                caps: caps.clone(),
                peer: peer.clone(),
            });
        }
        Ok(other) => {
            let _ = tx.send(Event::Error {
                code: "no_hello".into(),
                msg: format!("se esperaba 'hello', llegó '{}'", t_of(&other)),
            });
            return Ok(());
        }
        Err(e) => {
            let _ = tx.send(Event::Error {
                code: "bad_handshake".into(),
                msg: format!("hello inválido: {e}"),
            });
            return Ok(());
        }
    }

    // Read loop. Cada línea es un Command. La conexión se cierra cuando se cierra EOF, o
    // cuando un error de IO lo fuerza.
    let caps_set: std::collections::BTreeSet<Cap> = caps.iter().copied().collect();
    let caps_set = Arc::new(caps_set);
    for line in reader.lines() {
        let line = match line {
            Ok(l) => l,
            Err(_) => break,
        };
        if line.trim().is_empty() {
            continue;
        }
        let cmd: Command = match serde_json::from_str(&line) {
            Ok(c) => c,
            Err(e) => {
                let _ = tx.send(Event::Error {
                    code: "bad_command".into(),
                    msg: format!("JSON inválido: {e}"),
                });
                continue;
            }
        };
        dispatch(cmd, &tx, &caps_set, &ctx);
    }
    Ok(())
}

fn writer_loop(mut stream: UnixStream, rx: mpsc::Receiver<Event>) {
    for ev in rx {
        let line = match serde_json::to_string(&ev) {
            Ok(s) => s,
            Err(e) => {
                tracing::error!(error = %e, "bus: serializando event");
                continue;
            }
        };
        if stream.write_all(line.as_bytes()).is_err() {
            break;
        }
        if stream.write_all(b"\n").is_err() {
            break;
        }
        // El bus es de baja cadencia; flush por mensaje no es relevante en costes y mejora
        // mucho la latencia percibida del cliente.
        let _ = stream.flush();
    }
}

fn dispatch(
    cmd: Command,
    tx: &Sender<Event>,
    caps: &Arc<std::collections::BTreeSet<Cap>>,
    ctx: &BusContext,
) {
    if let Some(needed) = Cap::required_for(&cmd) {
        if !caps.contains(&needed) {
            let _ = tx.send(Event::Error {
                code: "no_cap".into(),
                msg: format!("comando requiere capacidad '{needed:?}', no concedida"),
            });
            return;
        }
    }
    match cmd {
        Command::Hello { .. } => {
            // Doble hello: lo ignoramos con un Error suave.
            let _ = tx.send(Event::Error {
                code: "duplicate_hello".into(),
                msg: "ya hicimos hello en esta conexión".into(),
            });
        }
        Command::Compile { recipe } => {
            // Build pesado: a un worker. El read-loop sigue libre para más comandos.
            let tx = tx.clone();
            let store_root = ctx.store_root.clone();
            thread::Builder::new()
                .name("bus-compile".into())
                .spawn(move || run_compile(recipe, store_root, tx))
                .ok();
        }
        Command::Inject { artifact, target, overlay: _ } => {
            let store = match hammer_core::Store::open(&ctx.store_root) {
                Ok(s) => s,
                Err(e) => {
                    let _ = tx.send(Event::Error {
                        code: "store".into(),
                        msg: e.to_string(),
                    });
                    return;
                }
            };
            let artifact_dir = match store.find_by_hash(&artifact) {
                Ok(p) => p,
                Err(e) => {
                    let _ = tx.send(Event::Error {
                        code: "not_found".into(),
                        msg: e.to_string(),
                    });
                    return;
                }
            };
            let report = match hammer_build::run_hydrate(
                &artifact_dir,
                Path::new(&target),
                hammer_core::LinkMode::Static,
            ) {
                Ok(r) => r,
                Err(e) => {
                    let _ = tx.send(Event::Error {
                        code: "hydrate_failed".into(),
                        msg: e.to_string(),
                    });
                    return;
                }
            };
            let _ = tx.send(Event::Injected {
                artifact,
                target,
                files: report.files.len(),
            });
        }
        Command::Query { what, path, name, expr } => {
            let value = match what.as_str() {
                "file" => {
                    let Some(p) = path else {
                        let _ = tx.send(Event::Error {
                            code: "bad_query".into(),
                            msg: "query 'file' requiere 'path'".into(),
                        });
                        return;
                    };
                    query_file(&p)
                }
                "artifact" => {
                    let Some(h) = name.as_deref().or(path.as_deref()) else {
                        let _ = tx.send(Event::Error {
                            code: "bad_query".into(),
                            msg: "query 'artifact' requiere 'name' (el hash)".into(),
                        });
                        return;
                    };
                    query_artifact(&ctx.store_root, h)
                }
                "expr" => {
                    let Some(e) = expr.as_deref() else {
                        let _ = tx.send(Event::Error {
                            code: "bad_query".into(),
                            msg: "query 'expr' requiere 'expr' (la expresión)".into(),
                        });
                        return;
                    };
                    // El daemon evalúa con el `fs_root` del sistema host (None ⇒ rutas
                    // absolutas tal y como vienen). Si el .swm necesitara otra base, el
                    // cliente puede evaluar localmente con un EvalContext distinto.
                    hammer_core::query::eval_str(e, &hammer_core::query::EvalContext::new())
                }
                other => {
                    let _ = tx.send(Event::Error {
                        code: "unknown_query".into(),
                        msg: format!("'what'='{other}' no soportado"),
                    });
                    return;
                }
            };
            let _ = tx.send(Event::QueryResult { what, value });
        }
        Command::Init { cmd } => {
            match crate::control::send_to_fifo(&ctx.init_control, &cmd) {
                Ok(()) => {
                    let _ = tx.send(Event::InitAck { cmd });
                }
                Err(e) => {
                    let _ = tx.send(Event::Error {
                        code: "init_send_failed".into(),
                        msg: e.to_string(),
                    });
                }
            }
        }
    }
}

fn run_compile(recipe: RecipeInline, store_root: PathBuf, tx: Sender<Event>) {
    let name = recipe.name.clone();
    let mutation = hammer_core::swm::Mutation::SourcePatch {
        repo: recipe.repo,
        commit: recipe.commit,
        tarball: recipe.tarball,
        sha256: recipe.sha256,
        patch: recipe.patch,
        patch_url: None,
        build: hammer_core::swm::SwmBuild {
            compiler: recipe.compiler,
            target: recipe.target,
            link: recipe.link,
            flags: recipe.flags,
        },
        target_bin: format!("/usr/bin/{name}"),
        expected_hash: None,
    };
    let store = match hammer_core::Store::open(&store_root) {
        Ok(s) => s,
        Err(e) => {
            let _ = tx.send(Event::BuildFailed {
                recipe: name,
                reason: format!("store: {e}"),
                log_tail: None,
            });
            return;
        }
    };
    let cfg = hammer_build::BuildConfig::from_env_or_defaults(store.root());
    match hammer_build::build_source_patch(&mutation, &cfg, &store, None) {
        Ok(h) => {
            let _ = tx.send(Event::BuildReady {
                recipe: name,
                artifact: h.as_str().to_string(),
            });
        }
        Err(e) => {
            // Si el fallo trae la cola del log del lab, la separamos de la razón para que la
            // IA pueda leer el error real del compilador, no sólo el mensaje de Rust.
            let (reason, log_tail) = match hammer_build::sandbox::BuildFailure::from_error(&e) {
                Some(bf) => (bf.reason.clone(), bf.log_tail.clone()),
                None => (e.to_string(), None),
            };
            let _ = tx.send(Event::BuildFailed {
                recipe: name,
                reason,
                log_tail,
            });
        }
    }
}

fn query_file(path: &str) -> serde_json::Value {
    let p = Path::new(path);
    let meta = match std::fs::symlink_metadata(p) {
        Ok(m) => m,
        Err(e) => return serde_json::json!({"exists": false, "error": e.to_string()}),
    };
    use std::os::unix::fs::MetadataExt;
    serde_json::json!({
        "exists": true,
        "size": meta.size(),
        "mode": meta.mode(),
        "uid": meta.uid(),
        "gid": meta.gid(),
        "is_dir": meta.is_dir(),
        "is_symlink": meta.file_type().is_symlink(),
    })
}

fn query_artifact(store_root: &Path, hash: &str) -> serde_json::Value {
    let store = match hammer_core::Store::open(store_root) {
        Ok(s) => s,
        Err(e) => return serde_json::json!({"error": e.to_string()}),
    };
    match store.find_by_hash(hash) {
        Ok(p) => serde_json::json!({"exists": true, "path": p.display().to_string()}),
        Err(_) => serde_json::json!({"exists": false}),
    }
}

fn t_of(cmd: &Command) -> &'static str {
    match cmd {
        Command::Hello { .. } => "hello",
        Command::Compile { .. } => "compile",
        Command::Inject { .. } => "inject",
        Command::Query { .. } => "query",
        Command::Init { .. } => "init",
    }
}

// Re-export para que el caller no tenga que importar Mutex/Arc directamente.
#[allow(dead_code)]
pub(crate) type SharedCaps = Arc<Mutex<std::collections::BTreeSet<Cap>>>;

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn default_policy_grants_query_to_others() {
        let p = default_policy();
        // Un peer con UID distinto al actual debe quedar como sólo-query.
        let foreign = Peer {
            uid: u32::MAX,
            gid: 0,
            pid: 1,
        };
        assert_eq!(p(&foreign), vec![Cap::Query]);
    }

    #[test]
    fn config_policy_resolves_by_uid_gid() {
        let cfg = hammer_core::AgentCapsConfig::from_toml(
            r#"
default = ["query"]
[[rule]]
uid = 0
caps = ["query", "compile", "inject", "inject-real", "init"]
"#,
        )
        .unwrap();
        let p = policy_from_config(cfg);
        let root = Peer { uid: 0, gid: 0, pid: 1 };
        let other = Peer { uid: 1234, gid: 1234, pid: 2 };
        assert!(p(&root).contains(&Cap::InjectReal));
        assert_eq!(p(&other), vec![Cap::Query]);
    }

    #[test]
    fn t_of_matches_all_commands() {
        // No es exhaustivo de Cargo, pero verifica que no nos olvidemos del nombre del tag.
        assert_eq!(t_of(&Command::Hello { ver: 1, client: "".into() }), "hello");
        assert_eq!(
            t_of(&Command::Compile {
                recipe: RecipeInline {
                    name: "x".into(),
                    repo: Some("g".into()),
                    commit: Some("c".into()),
                    tarball: None,
                    sha256: None,
                    patch: None,
                    compiler: "zig-cc".into(),
                    target: "t".into(),
                    link: "static".into(),
                    flags: vec![],
                },
            }),
            "compile"
        );
    }

    #[test]
    fn query_file_reports_missing() {
        let v = query_file("/no/existe/seguramente/ahora");
        assert_eq!(v["exists"], serde_json::json!(false));
    }
}