From f769d9173c036c3d9dc035bf1c7040d189f1b7c1 Mon Sep 17 00:00:00 2001 From: Sergio Date: Wed, 10 Jun 2026 16:27:59 +0000 Subject: [PATCH] =?UTF-8?q?Fase=206=20=E2=80=94=20bucle=20de=20auto-repara?= =?UTF-8?q?ci=C3=B3n=20(cliente=20reacciona=20a=20Crashed)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Orchestrator::run_with_repair(intent, &policy)` corre el ciclo plan→try→apply→wait_for_crash→re-plan hasta `max_attempts` o estabilización (sin Crashed en `crash_window`). Cuando llega un Crashed asíncrono por el bus, la política convierte `(service, code)` en una nueva intención (`default_crash_intent` por defecto) y reentra. Diseño: - `RepairPolicy { max_attempts, crash_window, event_sock, format_intent }` acota el blast-radius desde el caller; la IA no decide reintentar. - `Proposal` gana `repair_chain: Vec` con la cadena de intentos (intent, overlay_id, triggered_crash). La IA NUNCA promueve al FHS; el humano lee el chain y decide commit/discard. - `wait_for_crash` abre un AgentClient sólo para drenar async events; reusa toda la maquinaria síncrona del cliente. CLI: `hammer ai --repair-max-attempts N --repair-window-ms M`. Sin el flag, `run` corre una sola vez (compatible hacia atrás). Tests (3): stub bus multi-conexión que pre-programa eventos por conexión. Cubre reacción a Crashed, cap por max_attempts con crashes persistentes, y modo sin socket (una sola corrida). --- crates/hammer-agent/src/orchestrator.rs | 142 +++++++++++++ crates/hammer-agent/tests/repair_loop.rs | 252 +++++++++++++++++++++++ crates/hammer-cli/src/main.rs | 60 +++++- docs/10-roadmap.md | 6 +- 4 files changed, 454 insertions(+), 6 deletions(-) create mode 100644 crates/hammer-agent/tests/repair_loop.rs diff --git a/crates/hammer-agent/src/orchestrator.rs b/crates/hammer-agent/src/orchestrator.rs index 78cd77ff..2d7cba8b 100644 --- a/crates/hammer-agent/src/orchestrator.rs +++ b/crates/hammer-agent/src/orchestrator.rs @@ -228,8 +228,84 @@ impl Orchestrator { applied, skipped_source_patches, checks, + repair_chain: Vec::new(), }) } + + /// Variante con auto-reparación. Tras cada aplicación, si la política trae un + /// `event_sock`, abre un cliente del bus y espera `crash_window` para ver si + /// llega un `Crashed`. Si llega, traduce `(service, code)` a una nueva intención + /// con `policy.format_intent` y vuelve a entrar al bucle, hasta `max_attempts`. + /// + /// El `Proposal` devuelto siempre lleva la cadena completa de intentos en + /// `repair_chain`. Si el último intento se estabilizó (sin crash dentro de la + /// ventana), su `triggered_crash` es `None`. Si se agotaron los intentos con un + /// crash todavía pendiente, la última entrada lo refleja — el humano lee la cadena + /// y decide si commit, discard o seguir manualmente. + pub fn run_with_repair(&self, intent: &str, policy: &RepairPolicy) -> Result { + let max = policy.max_attempts.max(1); + let mut chain: Vec = Vec::new(); + let mut current = intent.to_string(); + let mut last: Option = None; + + for attempt in 0..max { + let p = self.run(¤t)?; + let overlay_id = p.overlay_id.clone(); + + let crash = wait_for_crash(policy); + chain.push(RepairAttempt { + attempt, + intent: current.clone(), + overlay_id, + triggered_crash: crash.clone(), + }); + last = Some(p); + + match crash { + None => break, // estabilizado + Some(c) => { + current = (policy.format_intent)(&c.service, c.code); + } + } + } + + let mut p = last.expect("max_attempts >= 1, así que hubo al menos un run"); + p.repair_chain = chain; + Ok(p) + } +} + +/// Conecta al bus y drena eventos hasta encontrar un `Crashed` dentro de `crash_window`. +/// Devuelve `None` si: +/// - la política no tiene socket; +/// - la conexión falla (caso "no hay bus") — el bucle decide entonces "estabilizado"; +/// - expira la ventana sin recibir Crashed. +fn wait_for_crash(policy: &RepairPolicy) -> Option { + let Some(sock) = &policy.event_sock else { return None }; + let client = match AgentClient::connect(sock) { + Ok(c) => c, + Err(e) => { + tracing::debug!(error = %e, "auto-repair: no pude conectar al bus, ignoro"); + return None; + } + }; + let deadline = std::time::Instant::now() + policy.crash_window; + // Drenamos primero la cola asíncrona ya acumulada (eventos llegados durante el `run`). + for ev in client.drain_async() { + if let hammer_core::proto::Event::Crashed { service, code } = ev { + return Some(CrashInfo { service, code }); + } + } + while let Some(remaining) = deadline.checked_duration_since(std::time::Instant::now()) { + match client.next_async(remaining) { + Ok(hammer_core::proto::Event::Crashed { service, code }) => { + return Some(CrashInfo { service, code }); + } + Ok(_other) => continue, + Err(_) => break, // timeout o cierre + } + } + None } /// Stub: `hammer-build` no es dependencia directa de `hammer-agent` para no arrastrar el @@ -354,6 +430,72 @@ pub struct Proposal { pub applied: AppliedCounts, pub skipped_source_patches: usize, pub checks: Vec, + /// Si la propuesta provino del bucle de auto-reparación, la cadena completa de + /// intentos. La última entrada corresponde a este `Proposal`. Para corridas sin + /// auto-reparación queda vacío. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub repair_chain: Vec, +} + +/// Un intento dentro del bucle de auto-reparación. Sirve para que el humano vea +/// cómo evolucionó la intención: original → reparación A → reparación B … +#[derive(Debug, Clone, serde::Serialize)] +pub struct RepairAttempt { + pub attempt: u32, + pub intent: String, + /// `Some` cuando el ciclo abrió un overlay; preservado para que el humano pueda hacer + /// `hammer discard ` sobre intentos intermedios si lo desea. + pub overlay_id: Option, + /// Si tras este intento llegó un `Crashed`, el servicio y código. `None` ⇒ estabilizado. + pub triggered_crash: Option, +} + +#[derive(Debug, Clone, serde::Serialize)] +pub struct CrashInfo { + pub service: String, + pub code: i32, +} + +/// Política del bucle de auto-reparación. La IA NUNCA decide por sí sola seguir +/// reintentando: la política la fija el llamador (CLI o cliente) y acota el blast-radius. +#[derive(Clone)] +pub struct RepairPolicy { + /// Máximo de intentos en total (incluyendo el primero). Mínimo efectivo: 1. + pub max_attempts: u32, + /// Ventana de espera tras aplicar para detectar un `Crashed` por el bus. + pub crash_window: Duration, + /// Socket del bus de eventos. Si `None`, el orquestador no escucha y devuelve + /// el `Proposal` tras la primera aplicación (equivalente a `run`). + pub event_sock: Option, + /// Convierte un crash `(service, code)` en la próxima intención. Por defecto: + /// `"repair service crashed code "`. + pub format_intent: fn(&str, i32) -> String, +} + +impl std::fmt::Debug for RepairPolicy { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RepairPolicy") + .field("max_attempts", &self.max_attempts) + .field("crash_window", &self.crash_window) + .field("event_sock", &self.event_sock) + .field("format_intent", &"fn(&str, i32) -> String") + .finish() + } +} + +pub fn default_crash_intent(service: &str, code: i32) -> String { + format!("repair service {service} crashed code {code}") +} + +impl Default for RepairPolicy { + fn default() -> Self { + Self { + max_attempts: 3, + crash_window: Duration::from_secs(5), + event_sock: None, + format_intent: default_crash_intent, + } + } } #[cfg(test)] diff --git a/crates/hammer-agent/tests/repair_loop.rs b/crates/hammer-agent/tests/repair_loop.rs new file mode 100644 index 00000000..1bb899b8 --- /dev/null +++ b/crates/hammer-agent/tests/repair_loop.rs @@ -0,0 +1,252 @@ +//! 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; + +fn start_stub_bus(sock: PathBuf, plans: Vec) { + 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()); +} diff --git a/crates/hammer-cli/src/main.rs b/crates/hammer-cli/src/main.rs index a93ca2c5..a80a898e 100644 --- a/crates/hammer-cli/src/main.rs +++ b/crates/hammer-cli/src/main.rs @@ -156,6 +156,16 @@ enum Cmd { /// Raíz de overlays para `try`. Default coincide con `hammer try`. #[arg(long)] state_root: Option, + /// Si está, activa el bucle de auto-reparación: tras aplicar, espera en el bus + /// de eventos (mismo socket que `--bus` si no se da otro) un evento `Crashed` + /// durante `--repair-window-ms`. Si llega, traduce `(service, code)` a una nueva + /// intención y reaplica, hasta este máximo de intentos. Sin `--repair-max-attempts`, + /// el bucle agéntico corre una sola vez (compatible hacia atrás). + #[arg(long)] + repair_max_attempts: Option, + /// Ventana de espera por un Crashed tras cada intento. Default: 5000 ms. + #[arg(long, default_value = "5000")] + repair_window_ms: u64, }, /// [Fase 5] Envía un comando al FIFO de control humano del init. /// Por defecto: /run/init.control. Equivalente a `echo "" > /run/init.control`, @@ -351,7 +361,16 @@ fn main() -> anyhow::Result<()> { Cmd::Export { base_ref, journal, since } => { run_export(base_ref.as_deref(), &journal, since.as_deref())?; } - Cmd::Ai { intent, catalog, prefix, base_ref, bus, state_root } => { + Cmd::Ai { + intent, + catalog, + prefix, + base_ref, + bus, + state_root, + repair_max_attempts, + repair_window_ms, + } => { run_ai( &intent, &catalog, @@ -360,6 +379,8 @@ fn main() -> anyhow::Result<()> { bus.as_deref(), state_root.as_deref(), &cli.store, + repair_max_attempts, + repair_window_ms, )?; } Cmd::Ctl { line, fifo } => { @@ -669,9 +690,11 @@ fn run_ai( bus: Option<&std::path::Path>, state_root: Option<&std::path::Path>, store_root: &str, + repair_max_attempts: Option, + repair_window_ms: u64, ) -> anyhow::Result<()> { use hammer_agent::{ - orchestrator::{ApplyTarget, CompileMode}, + orchestrator::{ApplyTarget, CompileMode, RepairPolicy}, IntentCatalog, MockTranslator, Orchestrator, }; @@ -728,9 +751,21 @@ fn run_ai( }; let orch = Orchestrator::new(translator, base, apply, compile); - let proposal = orch - .run(intent) - .map_err(|e| anyhow::anyhow!("orchestrator: {e}"))?; + let proposal = match repair_max_attempts { + Some(max) => { + let policy = RepairPolicy { + max_attempts: max, + crash_window: std::time::Duration::from_millis(repair_window_ms), + event_sock: bus.map(|p| p.to_path_buf()), + ..RepairPolicy::default() + }; + orch.run_with_repair(intent, &policy) + .map_err(|e| anyhow::anyhow!("orchestrator (auto-repair): {e}"))? + } + None => orch + .run(intent) + .map_err(|e| anyhow::anyhow!("orchestrator: {e}"))?, + }; println!("--- proposal ---"); println!("intent: {}", proposal.intent); @@ -751,6 +786,21 @@ fn run_ai( for c in &proposal.checks { println!(" [{:?}] {}", c.level, c.msg); } + if !proposal.repair_chain.is_empty() { + println!("\nrepair_chain ({} intento(s)):", proposal.repair_chain.len()); + for r in &proposal.repair_chain { + match &r.triggered_crash { + Some(c) => println!( + " #{:>2} intent={:?} overlay={:?} → crash service={} code={}", + r.attempt, r.intent, r.overlay_id, c.service, c.code + ), + None => println!( + " #{:>2} intent={:?} overlay={:?} → estabilizado", + r.attempt, r.intent, r.overlay_id + ), + } + } + } match (&proposal.overlay_id, &proposal.prefix) { (Some(id), _) => { println!("\noverlay {id} listo."); diff --git a/docs/10-roadmap.md b/docs/10-roadmap.md index 2a753b80..04e67b64 100644 --- a/docs/10-roadmap.md +++ b/docs/10-roadmap.md @@ -122,7 +122,11 @@ pre-requisito de validación. archivos sin rutas frágiles. Forma `kind:value` (`bin`, `file`, `pin`, `service`, `depends`), evaluable local (`hammer query `) y remoto vía bus (`Command::Query{what:"expr"}`). Parser ELF64 mínimo para extraer `DT_NEEDED`. -- [ ] Bucle de auto-reparación (cliente reacciona a `Crashed` con un nuevo plan). +- [x] Bucle de auto-reparación (cliente reacciona a `Crashed` con un nuevo plan). + `Orchestrator::run_with_repair(intent, &RepairPolicy)` drena async events del bus + tras cada apply; si llega `Crashed`, formatea una nueva intención y reentra hasta + `max_attempts`. El `Proposal` final lleva el `repair_chain` completo para que el + humano vea la evolución antes de hacer commit. CLI: `hammer ai --repair-max-attempts`. - **Hecho cuando:** una intención en lenguaje natural produce un cambio probado en overlay, presentado para `commit` humano. ✅ Demostrado por `hammer ai` con `MockTranslator`: intent → `.swm` → mutaciones aplicadas → `Proposal` con `overlay_id` + checks. El humano