//! `/run/init.control` — FIFO de control humano. Ver `docs/07-agent-bus.md` §1. //! //! Modelo Fase 5: el FIFO existe y acepta escrituras (`echo "start web" > …`); las líneas //! recibidas se loguean por ahora. La supervisión real de servicios llega cuando entre el //! init propio del track posterior (`/etc/service/*/run` estilo s6/runit). //! //! Por qué un FIFO y no un socket: es lo MÁS unix posible (`cat`, `echo`, redirección), //! ergonomía máxima sin parser ni framing. Para programas serios está el `/run/agent.sock`. use std::io::{BufRead, BufReader, Write}; use std::path::{Path, PathBuf}; /// Crea (o reusa) el FIFO en `path` y devuelve la ruta canónica. Si el path existe y NO es /// un FIFO falla con mensaje claro: no queremos pisar archivos por accidente. pub fn ensure_fifo(path: &Path) -> std::io::Result { use std::os::unix::fs::FileTypeExt; if let Some(parent) = path.parent() { std::fs::create_dir_all(parent)?; } match std::fs::symlink_metadata(path) { Ok(m) if m.file_type().is_fifo() => return Ok(path.to_path_buf()), Ok(_) => { return Err(std::io::Error::new( std::io::ErrorKind::AlreadyExists, format!( "init-control: {} existe y NO es un FIFO; mueve o borra el archivo", path.display() ), )); } Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} Err(e) => return Err(e), } // `nix::unistd::mkfifo` con permisos 0o660: owner + group escriben/leen, world nada. // El humano normalmente está en el grupo del daemon (p. ej. `takana`); la IA agente // entra por el socket. use nix::sys::stat::Mode; nix::unistd::mkfifo(path, Mode::S_IRUSR | Mode::S_IWUSR | Mode::S_IRGRP | Mode::S_IWGRP) .map_err(std::io::Error::from)?; Ok(path.to_path_buf()) } /// Lazo bloqueante: abre el FIFO en modo lectura y loguea cada línea. Cuando el último /// escritor cierra, `read_line` devuelve 0 (EOF); volvemos a abrir para no salir del lazo. /// /// Pensado para correr en su propio thread desde `main`. pub fn run_reader(path: &Path) -> std::io::Result<()> { loop { let f = match std::fs::OpenOptions::new().read(true).open(path) { Ok(f) => f, Err(e) => { tracing::warn!(error = %e, "init-control: no pude abrir para lectura"); return Err(e); } }; let mut reader = BufReader::new(f); loop { let mut line = String::new(); match reader.read_line(&mut line) { Ok(0) => break, // EOF: re-abrimos Ok(_) => { let trimmed = line.trim_end_matches('\n'); tracing::info!(line = %trimmed, "init-control"); } Err(e) => { tracing::warn!(error = %e, "init-control: error de lectura"); break; } } } } } /// Envía una línea al FIFO. Si nadie está leyendo, **bloquea** hasta que el reader la /// consuma: ése es exactamente el contrato de un FIFO sin O_NONBLOCK, y la garantía que el /// agente espera (`InitAck` significa "la línea entró al canal"). pub fn send_to_fifo(path: &Path, line: &str) -> std::io::Result<()> { let mut f = std::fs::OpenOptions::new().write(true).open(path)?; f.write_all(line.as_bytes())?; if !line.ends_with('\n') { f.write_all(b"\n")?; } Ok(()) } #[cfg(test)] mod tests { use super::*; #[test] fn ensure_fifo_creates_when_missing() { let d = tempfile::tempdir().unwrap(); let p = d.path().join("init.ctl"); ensure_fifo(&p).unwrap(); use std::os::unix::fs::FileTypeExt; assert!(p.symlink_metadata().unwrap().file_type().is_fifo()); } #[test] fn ensure_fifo_idempotent() { let d = tempfile::tempdir().unwrap(); let p = d.path().join("init.ctl"); ensure_fifo(&p).unwrap(); ensure_fifo(&p).unwrap(); // sin error: ya es FIFO } #[test] fn ensure_fifo_rejects_non_fifo() { let d = tempfile::tempdir().unwrap(); let p = d.path().join("not-a-fifo"); std::fs::write(&p, b"x").unwrap(); let err = ensure_fifo(&p).unwrap_err().to_string(); assert!(err.contains("NO es un FIFO"), "{err}"); } #[test] fn send_and_read_one_line() { let d = tempfile::tempdir().unwrap(); let p = d.path().join("init.ctl"); ensure_fifo(&p).unwrap(); // Reader en background; sólo lee una línea y termina. let reader_path = p.clone(); let reader = std::thread::spawn(move || { let f = std::fs::OpenOptions::new() .read(true) .open(&reader_path) .unwrap(); let mut r = BufReader::new(f); let mut s = String::new(); r.read_line(&mut s).unwrap(); s }); // Pequeño retraso para asegurar que el reader llega a `open` antes del writer. std::thread::sleep(std::time::Duration::from_millis(50)); send_to_fifo(&p, "start web").unwrap(); let got = reader.join().unwrap(); assert_eq!(got, "start web\n"); } }