Fase 6 — bucle de auto-reparación (cliente reacciona a Crashed)
`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<RepairAttempt>` 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).
f769d9173c036c3d9dc035bf1c7040d189f1b7c1 · by Sergio · 2026-06-10T16:27:59Z (4 months ago) · parent dbbc10e85 · browse the tree at this commit
Semantic delta · 22
crates/hammer-agent/src/orchestrator.rs
crates/hammer-agent/tests/repair_loop.rs
crates/hammer-cli/src/main.rs
signature changed enum Cmd body changed fn main signature changed fn run_ai
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<T: IntentTranslator> Orchestrator<T> {
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<Proposal> {
+ let max = policy.max_attempts.max(1);
+ let mut chain: Vec<RepairAttempt> = Vec::new();
+ let mut current = intent.to_string();
+ let mut last: Option<Proposal> = 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<CrashInfo> {
+ 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<VerifyCheck>,
+ /// 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<RepairAttempt>,
+}
+
+/// 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 <id>` sobre intentos intermedios si lo desea.
+ pub overlay_id: Option<String>,
+ /// Si tras este intento llegó un `Crashed`, el servicio y código. `None` ⇒ estabilizado.
+ pub triggered_crash: Option<CrashInfo>,
+}
+
+#[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<PathBuf>,
+ /// Convierte un crash `(service, code)` en la próxima intención. Por defecto:
+ /// `"repair service <s> crashed code <c>"`.
+ 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<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());
+}
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<PathBuf>,
+ /// 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<u32>,
+ /// 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 "<line>" > /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<u32>,
+ 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 <expr>`) 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