//! El bucle agéntico. Ver `docs/08-ai-integration.md` §2. //! //! `Orchestrator::run(intent)` ejecuta: //! //! ```text //! plan → translator.translate(intent, ctx) → Swm //! schema → swm.verify_schema() //! base → swm.verify_base(local) → debe ser Ok //! build → para cada source_patch: AgentClient.compile() vía el bus (o saltar si --no-bus) //! try → hammer_overlay::try_overlay() (o usar --prefix) //! apply → hammer-core::apply para config_edit/file_drop + hydrate para source_patch //! verify → spot-checks post-condición sobre disco //! propose → devuelve Proposal con overlay_id + log de checks //! ``` //! //! La IA NUNCA ejecuta `hammer commit`. El humano lo hace tras revisar el `Proposal`. use std::path::{Path, PathBuf}; use std::time::Duration; use hammer_core::apply::{apply_config_edit, apply_file_drop, rebase_path}; use hammer_core::proto::RecipeInline; use hammer_core::swm::{Mutation, Swm}; use hammer_core::BaseRef; use hammer_overlay::OverlayId; use crate::client::AgentClient; use crate::translator::{IntentTranslator, SystemContext}; use crate::{Error, Result}; /// Cómo aplica el orquestador las mutaciones al sistema. #[derive(Debug, Clone)] pub enum ApplyTarget { /// Abre un overlay sobre los targets por defecto del FHS (requiere root). Las /// mutaciones caen sobre las rutas absolutas del `.swm`. Overlay { state_root: PathBuf }, /// Re-rootea cada path bajo `prefix`. No abre overlay — útil para tests offline y /// para staging en otro filesystem. Prefix { prefix: PathBuf }, } /// Cómo se construyen las `source_patch`. `ViaBus` requiere un `hammerd` corriendo; /// `Skip` ignora las source_patch (apropiado para escenarios sin lab). #[derive(Debug, Clone)] pub enum CompileMode { /// Llama a `Compile` por el bus y espera `BuildReady`. Hidrata localmente desde el /// store si está configurado. ViaBus { sock: PathBuf, store: PathBuf, timeout: Duration, }, /// Salta cualquier `source_patch`. Se anota en `Proposal.skipped_source_patches`. Skip, } pub struct Orchestrator { translator: T, base: BaseRef, apply: ApplyTarget, compile: CompileMode, } impl Orchestrator { pub fn new(translator: T, base: BaseRef, apply: ApplyTarget, compile: CompileMode) -> Self { Self { translator, base, apply, compile } } pub fn run(&self, intent: &str) -> Result { // 1. PLAN let ctx = SystemContext { base: self.base.clone(), extras: serde_json::Value::Null, }; let swm = self.translator.translate(intent, &ctx)?; // 2. SCHEMA swm.verify_schema()?; // 3. BASE match swm.verify_base(&self.base) { hammer_core::BaseCompat::Ok => {} other => { return Err(Error::Orchestrate(format!( "base incompatible: {other:?}" ))); } } // 4. TRY (overlay) — recolectamos el id si lo abrimos; en modo Prefix no hay id. let (overlay_id, prefix): (Option, Option) = match &self.apply { ApplyTarget::Overlay { state_root } => { std::fs::create_dir_all(state_root)?; let id = hammer_overlay::try_overlay(&[], state_root)?; (Some(id), None) } ApplyTarget::Prefix { prefix } => { std::fs::create_dir_all(prefix)?; (None, Some(prefix.clone())) } }; // 5. APPLY — recorremos las mutaciones en orden. En source_patch, dispatch al bus // o skip; en config_edit/file_drop, primitivas de hammer-core::apply. let mut applied = AppliedCounts::default(); let mut checks: Vec = Vec::new(); let mut skipped_source_patches = 0usize; let mut maybe_client: Option = None; if matches!(self.compile, CompileMode::ViaBus { .. }) && swm.mutations.iter().any(|m| matches!(m, Mutation::SourcePatch { .. })) { if let CompileMode::ViaBus { sock, .. } = &self.compile { maybe_client = Some(AgentClient::connect(sock).map_err(Error::Client)?); } } for m in &swm.mutations { match m { Mutation::ConfigEdit { file, inline_diff } => { let target = rebase_path(file, prefix.as_deref()); apply_config_edit(&target, inline_diff)?; applied.config_edit += 1; checks.push(VerifyCheck::pass(format!( "config_edit aplicado a {}", target.display() ))); } Mutation::FileDrop { path, content_hash, content_b64, content_url } => { if let Some(_url) = content_url { return Err(Error::Orchestrate(format!( "file_drop con content_url no soportado en Fase 6: {path}" ))); } let b64 = content_b64.as_deref().ok_or_else(|| { Error::Orchestrate(format!( "file_drop {path} no trae content_b64 (verify_schema falló)" )) })?; let target = rebase_path(path, prefix.as_deref()); apply_file_drop(&target, b64, content_hash)?; applied.file_drop += 1; checks.push(VerifyCheck::pass(format!( "file_drop escrito en {} (hash verificado)", target.display() ))); } Mutation::InitRule { action, service, command } => { let rules_dir = rebase_path(hammer_core::apply::INIT_RULES_DIR, prefix.as_deref()); let p = hammer_core::apply::apply_init_rule( &rules_dir, action, service, command, ) .map_err(|e| Error::Orchestrate(format!("init_rule '{action} {service}': {e}")))?; applied.init_rule += 1; checks.push(VerifyCheck::pass(format!( "init_rule '{action} {service}' → {}", p.display() ))); } Mutation::SourcePatch { repo, commit, tarball, sha256, patch, patch_url: _, build, target_bin, expected_hash, } => { let CompileMode::ViaBus { store, timeout, .. } = &self.compile else { skipped_source_patches += 1; checks.push(VerifyCheck::warn(format!( "source_patch {target_bin} omitido (CompileMode::Skip)" ))); continue; }; let name = Path::new(target_bin) .file_name() .and_then(|s| s.to_str()) .unwrap_or("swm-bin") .to_string(); let recipe = RecipeInline { name: name.clone(), repo: repo.clone(), commit: commit.clone(), tarball: tarball.clone(), sha256: sha256.clone(), patch: patch.clone(), compiler: build.compiler.clone(), target: build.target.clone(), link: build.link.clone(), flags: build.flags.clone(), }; let client = maybe_client.as_mut().expect("client conectado arriba"); let artifact = client .compile(recipe, *timeout) .map_err(Error::Client)?; if let Some(expected) = expected_hash { let want = expected.strip_prefix("b3:").unwrap_or(expected); let got = artifact.strip_prefix("b3:").unwrap_or(&artifact); if want != got { return Err(Error::Orchestrate(format!( "source_patch {target_bin}: expected_hash no coincide \ (declarado=b3:{want}, build=b3:{got})" ))); } } // Hidratamos localmente desde el store (sin pasar por el bus, evita un // segundo Inject que duplicaría trabajo). let store = hammer_core::Store::open(store)?; let artifact_dir = store .find_by_hash(&artifact) .map_err(|e| Error::Orchestrate(format!("find_by_hash: {e}")))?; let into = rebase_path("/", prefix.as_deref()); let report = hammer_build_run_hydrate(&artifact_dir, &into)?; let abs = rebase_path(target_bin, prefix.as_deref()); if !abs.exists() { return Err(Error::Orchestrate(format!( "source_patch declara target_bin={target_bin} pero no quedó en {}", abs.display() ))); } applied.source_patch += 1; checks.push(VerifyCheck::pass(format!( "source_patch {target_bin}: build {artifact}, {} archivo(s) hidratado(s)", report ))); } } } Ok(Proposal { intent: intent.to_string(), swm, overlay_id: overlay_id.map(|o| o.as_str().to_string()), prefix: prefix.clone(), 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 /// lab a clientes que sólo quieran hablar con el bus. Implementamos la hidratación local /// con `std::fs::hard_link` directamente porque es trivial. Si el árbol del artefacto crece /// (symlinks, perms más finos), promovemos a `hammer-build::run_hydrate` añadiendo la dep /// de manera condicional. fn hammer_build_run_hydrate(artifact_dir: &Path, target_fhs: &Path) -> Result { let mut files = 0usize; walk(artifact_dir, &mut |rel, kind| { let dst = target_fhs.join(rel); match kind { EntryKind::Dir => { std::fs::create_dir_all(&dst)?; } EntryKind::Symlink(target) => { if let Some(parent) = dst.parent() { std::fs::create_dir_all(parent)?; } let _ = std::fs::remove_file(&dst); std::os::unix::fs::symlink(&target, &dst)?; files += 1; } EntryKind::File(src) => { if let Some(parent) = dst.parent() { std::fs::create_dir_all(parent)?; } let _ = std::fs::remove_file(&dst); std::fs::hard_link(&src, &dst).or_else(|_| { // Cross-fs ⇒ caemos a copia. No optimizamos: hammer-agent espera // mismo FS; sólo damos el fallback como cinturón de seguridad. std::fs::copy(&src, &dst).map(|_| ()) })?; files += 1; } } Ok::<(), std::io::Error>(()) })?; Ok(files) } enum EntryKind { Dir, Symlink(PathBuf), File(PathBuf), } fn walk(root: &Path, f: &mut F) -> std::io::Result<()> where F: FnMut(&Path, EntryKind) -> std::io::Result<()>, { fn inner(root: &Path, cur: &Path, f: &mut F) -> std::io::Result<()> where F: FnMut(&Path, EntryKind) -> std::io::Result<()>, { for entry in std::fs::read_dir(cur)? { let entry = entry?; let src = entry.path(); let rel = src.strip_prefix(root).unwrap(); let ft = entry.file_type()?; if ft.is_symlink() { let target = std::fs::read_link(&src)?; f(rel, EntryKind::Symlink(target))?; } else if ft.is_dir() { f(rel, EntryKind::Dir)?; inner(root, &src, f)?; } else if ft.is_file() { f(rel, EntryKind::File(src.clone()))?; } } Ok(()) } if !root.is_dir() { return Err(std::io::Error::new( std::io::ErrorKind::NotFound, format!("artifact_dir no existe: {}", root.display()), )); } inner(root, root, f) } #[derive(Debug, Default, Clone, serde::Serialize)] pub struct AppliedCounts { pub source_patch: usize, pub config_edit: usize, pub init_rule: usize, pub file_drop: usize, } #[derive(Debug, Clone, serde::Serialize)] pub struct VerifyCheck { pub level: CheckLevel, pub msg: String, } #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)] #[serde(rename_all = "kebab-case")] pub enum CheckLevel { Pass, Warn, Fail, } impl VerifyCheck { pub fn pass(msg: impl Into) -> Self { Self { level: CheckLevel::Pass, msg: msg.into() } } pub fn warn(msg: impl Into) -> Self { Self { level: CheckLevel::Warn, msg: msg.into() } } } /// El resultado del bucle. El humano lo lee para decidir `hammer commit ` (si /// está en modo Overlay) o `hammer discard`. #[derive(Debug, Clone, serde::Serialize)] pub struct Proposal { pub intent: String, pub swm: Swm, /// `Some` si abrimos un overlay. En modo Prefix, `None`. pub overlay_id: Option, pub prefix: Option, 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)] mod tests { use super::*; use crate::translator::MockTranslator; use hammer_core::swm::{Base, Mutation}; use std::collections::BTreeMap; fn baseref() -> BaseRef { BaseRef { distro_version: "2026-06-06".into(), pins: BTreeMap::new() } } fn swm_config_only() -> Swm { Swm { swm_version: 1, base: Base { distro_version: "2026-06-06".into(), pins: BTreeMap::new() }, mutations: vec![Mutation::ConfigEdit { file: "/etc/network.conf".into(), inline_diff: "- DHCP=yes\n+ STATIC=1\n".into(), }], signature: None, } } #[test] fn orchestrator_applies_config_edit_to_prefix() { let d = tempfile::tempdir().unwrap(); let prefix = d.path().join("prefix"); std::fs::create_dir_all(prefix.join("etc")).unwrap(); std::fs::write(prefix.join("etc/network.conf"), "DHCP=yes\nMTU=1500\n").unwrap(); let t = MockTranslator::empty().with_intent("set static ip", swm_config_only()); let o = Orchestrator::new( t, baseref(), ApplyTarget::Prefix { prefix: prefix.clone() }, CompileMode::Skip, ); let p = o.run("set static ip").unwrap(); assert_eq!(p.applied.config_edit, 1); assert!(p.overlay_id.is_none()); assert_eq!(p.checks.len(), 1); assert_eq!(p.checks[0].level, CheckLevel::Pass); let got = std::fs::read_to_string(prefix.join("etc/network.conf")).unwrap(); assert_eq!(got, "STATIC=1\nMTU=1500\n"); } #[test] fn orchestrator_rejects_base_mismatch() { let mut swm = swm_config_only(); swm.base.distro_version = "2027-01-01".into(); let t = MockTranslator::empty().with_intent("x", swm); let d = tempfile::tempdir().unwrap(); let o = Orchestrator::new( t, baseref(), ApplyTarget::Prefix { prefix: d.path().to_path_buf() }, CompileMode::Skip, ); let err = o.run("x").unwrap_err().to_string(); assert!(err.contains("base incompatible"), "{err}"); } #[test] fn orchestrator_skips_source_patch_with_warn() { let swm = Swm { swm_version: 1, base: Base { distro_version: "2026-06-06".into(), pins: BTreeMap::new() }, mutations: vec![Mutation::SourcePatch { repo: Some("git://x".into()), commit: Some("abc".into()), tarball: None, sha256: None, patch: None, patch_url: None, build: hammer_core::swm::SwmBuild { compiler: "zig-cc".into(), target: "x86_64-linux-musl".into(), link: "static".into(), flags: vec![], }, target_bin: "/bin/x".into(), expected_hash: None, }], signature: None, }; let t = MockTranslator::empty().with_intent("compile x", swm); let d = tempfile::tempdir().unwrap(); let o = Orchestrator::new( t, baseref(), ApplyTarget::Prefix { prefix: d.path().to_path_buf() }, CompileMode::Skip, ); let p = o.run("compile x").unwrap(); assert_eq!(p.skipped_source_patches, 1); assert_eq!(p.checks[0].level, CheckLevel::Warn); } #[test] fn orchestrator_translates_unknown_intent_errors() { let t = MockTranslator::empty(); let d = tempfile::tempdir().unwrap(); let o = Orchestrator::new( t, baseref(), ApplyTarget::Prefix { prefix: d.path().to_path_buf() }, CompileMode::Skip, ); let err = o.run("haz lo que tú quieras").unwrap_err().to_string(); assert!(err.contains("no reconocida"), "{err}"); } }