sergio/takana

1//! El bucle agéntico. Ver `docs/08-ai-integration.md` §2.
2//!
3//! `Orchestrator::run(intent)` ejecuta:
4//!
5//! ```text
6//!   plan    → translator.translate(intent, ctx) → Swm
7//!   schema  → swm.verify_schema()
8//!   base    → swm.verify_base(local) → debe ser Ok
9//!   build   → para cada source_patch: AgentClient.compile() vía el bus  (o saltar si --no-bus)
10//!   try     → hammer_overlay::try_overlay()                              (o usar --prefix)
11//!   apply   → hammer-core::apply para config_edit/file_drop + hydrate para source_patch
12//!   verify  → spot-checks post-condición sobre disco
13//!   propose → devuelve Proposal con overlay_id + log de checks
14//! ```
15//!
16//! La IA NUNCA ejecuta `hammer commit`. El humano lo hace tras revisar el `Proposal`.
17
18use std::path::{Path, PathBuf};
19use std::time::Duration;
20
21use hammer_core::apply::{apply_config_edit, apply_file_drop, rebase_path};
22use hammer_core::proto::RecipeInline;
23use hammer_core::swm::{Mutation, Swm};
24use hammer_core::BaseRef;
25use hammer_overlay::OverlayId;
26
27use crate::client::AgentClient;
28use crate::translator::{IntentTranslator, SystemContext};
29use crate::{Error, Result};
30
31/// Cómo aplica el orquestador las mutaciones al sistema.
32#[derive(Debug, Clone)]
33pub enum ApplyTarget {
34    /// Abre un overlay sobre los targets por defecto del FHS (requiere root). Las
35    /// mutaciones caen sobre las rutas absolutas del `.swm`.
36    Overlay { state_root: PathBuf },
37    /// Re-rootea cada path bajo `prefix`. No abre overlay — útil para tests offline y
38    /// para staging en otro filesystem.
39    Prefix { prefix: PathBuf },
40}
41
42/// Cómo se construyen las `source_patch`. `ViaBus` requiere un `hammerd` corriendo;
43/// `Skip` ignora las source_patch (apropiado para escenarios sin lab).
44#[derive(Debug, Clone)]
45pub enum CompileMode {
46    /// Llama a `Compile` por el bus y espera `BuildReady`. Hidrata localmente desde el
47    /// store si está configurado.
48    ViaBus {
49        sock: PathBuf,
50        store: PathBuf,
51        timeout: Duration,
52    },
53    /// Salta cualquier `source_patch`. Se anota en `Proposal.skipped_source_patches`.
54    Skip,
55}
56
57pub struct Orchestrator<T: IntentTranslator> {
58    translator: T,
59    base: BaseRef,
60    apply: ApplyTarget,
61    compile: CompileMode,
62}
63
64impl<T: IntentTranslator> Orchestrator<T> {
65    pub fn new(translator: T, base: BaseRef, apply: ApplyTarget, compile: CompileMode) -> Self {
66        Self { translator, base, apply, compile }
67    }
68
69    pub fn run(&self, intent: &str) -> Result<Proposal> {
70        // 1. PLAN
71        let ctx = SystemContext {
72            base: self.base.clone(),
73            extras: serde_json::Value::Null,
74        };
75        let swm = self.translator.translate(intent, &ctx)?;
76
77        // 2. SCHEMA
78        swm.verify_schema()?;
79
80        // 3. BASE
81        match swm.verify_base(&self.base) {
82            hammer_core::BaseCompat::Ok => {}
83            other => {
84                return Err(Error::Orchestrate(format!(
85                    "base incompatible: {other:?}"
86                )));
87            }
88        }
89
90        // 4. TRY (overlay) — recolectamos el id si lo abrimos; en modo Prefix no hay id.
91        let (overlay_id, prefix): (Option<OverlayId>, Option<PathBuf>) = match &self.apply {
92            ApplyTarget::Overlay { state_root } => {
93                std::fs::create_dir_all(state_root)?;
94                let id = hammer_overlay::try_overlay(&[], state_root)?;
95                (Some(id), None)
96            }
97            ApplyTarget::Prefix { prefix } => {
98                std::fs::create_dir_all(prefix)?;
99                (None, Some(prefix.clone()))
100            }
101        };
102
103        // 5. APPLY — recorremos las mutaciones en orden. En source_patch, dispatch al bus
104        // o skip; en config_edit/file_drop, primitivas de hammer-core::apply.
105        let mut applied = AppliedCounts::default();
106        let mut checks: Vec<VerifyCheck> = Vec::new();
107        let mut skipped_source_patches = 0usize;
108
109        let mut maybe_client: Option<AgentClient> = None;
110        if matches!(self.compile, CompileMode::ViaBus { .. })
111            && swm.mutations.iter().any(|m| matches!(m, Mutation::SourcePatch { .. }))
112        {
113            if let CompileMode::ViaBus { sock, .. } = &self.compile {
114                maybe_client = Some(AgentClient::connect(sock).map_err(Error::Client)?);
115            }
116        }
117
118        for m in &swm.mutations {
119            match m {
120                Mutation::ConfigEdit { file, inline_diff } => {
121                    let target = rebase_path(file, prefix.as_deref());
122                    apply_config_edit(&target, inline_diff)?;
123                    applied.config_edit += 1;
124                    checks.push(VerifyCheck::pass(format!(
125                        "config_edit aplicado a {}",
126                        target.display()
127                    )));
128                }
129                Mutation::FileDrop { path, content_hash, content_b64, content_url } => {
130                    if let Some(_url) = content_url {
131                        return Err(Error::Orchestrate(format!(
132                            "file_drop con content_url no soportado en Fase 6: {path}"
133                        )));
134                    }
135                    let b64 = content_b64.as_deref().ok_or_else(|| {
136                        Error::Orchestrate(format!(
137                            "file_drop {path} no trae content_b64 (verify_schema falló)"
138                        ))
139                    })?;
140                    let target = rebase_path(path, prefix.as_deref());
141                    apply_file_drop(&target, b64, content_hash)?;
142                    applied.file_drop += 1;
143                    checks.push(VerifyCheck::pass(format!(
144                        "file_drop escrito en {} (hash verificado)",
145                        target.display()
146                    )));
147                }
148                Mutation::InitRule { action, service, command } => {
149                    let rules_dir =
150                        rebase_path(hammer_core::apply::INIT_RULES_DIR, prefix.as_deref());
151                    let p = hammer_core::apply::apply_init_rule(
152                        &rules_dir, action, service, command,
153                    )
154                    .map_err(|e| Error::Orchestrate(format!("init_rule '{action} {service}': {e}")))?;
155                    applied.init_rule += 1;
156                    checks.push(VerifyCheck::pass(format!(
157                        "init_rule '{action} {service}' → {}",
158                        p.display()
159                    )));
160                }
161                Mutation::SourcePatch {
162                    repo,
163                    commit,
164                    tarball,
165                    sha256,
166                    patch,
167                    patch_url: _,
168                    build,
169                    target_bin,
170                    expected_hash,
171                } => {
172                    let CompileMode::ViaBus { store, timeout, .. } = &self.compile else {
173                        skipped_source_patches += 1;
174                        checks.push(VerifyCheck::warn(format!(
175                            "source_patch {target_bin} omitido (CompileMode::Skip)"
176                        )));
177                        continue;
178                    };
179                    let name = Path::new(target_bin)
180                        .file_name()
181                        .and_then(|s| s.to_str())
182                        .unwrap_or("swm-bin")
183                        .to_string();
184                    let recipe = RecipeInline {
185                        name: name.clone(),
186                        repo: repo.clone(),
187                        commit: commit.clone(),
188                        tarball: tarball.clone(),
189                        sha256: sha256.clone(),
190                        patch: patch.clone(),
191                        compiler: build.compiler.clone(),
192                        target: build.target.clone(),
193                        link: build.link.clone(),
194                        flags: build.flags.clone(),
195                    };
196                    let client = maybe_client.as_mut().expect("client conectado arriba");
197                    let artifact = client
198                        .compile(recipe, *timeout)
199                        .map_err(Error::Client)?;
200                    if let Some(expected) = expected_hash {
201                        let want = expected.strip_prefix("b3:").unwrap_or(expected);
202                        let got = artifact.strip_prefix("b3:").unwrap_or(&artifact);
203                        if want != got {
204                            return Err(Error::Orchestrate(format!(
205                                "source_patch {target_bin}: expected_hash no coincide \
206                                 (declarado=b3:{want}, build=b3:{got})"
207                            )));
208                        }
209                    }
210                    // Hidratamos localmente desde el store (sin pasar por el bus, evita un
211                    // segundo Inject que duplicaría trabajo).
212                    let store = hammer_core::Store::open(store)?;
213                    let artifact_dir = store
214                        .find_by_hash(&artifact)
215                        .map_err(|e| Error::Orchestrate(format!("find_by_hash: {e}")))?;
216                    let into = rebase_path("/", prefix.as_deref());
217                    let report = hammer_build_run_hydrate(&artifact_dir, &into)?;
218                    let abs = rebase_path(target_bin, prefix.as_deref());
219                    if !abs.exists() {
220                        return Err(Error::Orchestrate(format!(
221                            "source_patch declara target_bin={target_bin} pero no quedó en {}",
222                            abs.display()
223                        )));
224                    }
225                    applied.source_patch += 1;
226                    checks.push(VerifyCheck::pass(format!(
227                        "source_patch {target_bin}: build {artifact}, {} archivo(s) hidratado(s)",
228                        report
229                    )));
230                }
231            }
232        }
233
234        Ok(Proposal {
235            intent: intent.to_string(),
236            swm,
237            overlay_id: overlay_id.map(|o| o.as_str().to_string()),
238            prefix: prefix.clone(),
239            applied,
240            skipped_source_patches,
241            checks,
242            repair_chain: Vec::new(),
243        })
244    }
245
246    /// Variante con auto-reparación. Tras cada aplicación, si la política trae un
247    /// `event_sock`, abre un cliente del bus y espera `crash_window` para ver si
248    /// llega un `Crashed`. Si llega, traduce `(service, code)` a una nueva intención
249    /// con `policy.format_intent` y vuelve a entrar al bucle, hasta `max_attempts`.
250    ///
251    /// El `Proposal` devuelto siempre lleva la cadena completa de intentos en
252    /// `repair_chain`. Si el último intento se estabilizó (sin crash dentro de la
253    /// ventana), su `triggered_crash` es `None`. Si se agotaron los intentos con un
254    /// crash todavía pendiente, la última entrada lo refleja — el humano lee la cadena
255    /// y decide si commit, discard o seguir manualmente.
256    pub fn run_with_repair(&self, intent: &str, policy: &RepairPolicy) -> Result<Proposal> {
257        let max = policy.max_attempts.max(1);
258        let mut chain: Vec<RepairAttempt> = Vec::new();
259        let mut current = intent.to_string();
260        let mut last: Option<Proposal> = None;
261
262        for attempt in 0..max {
263            let p = self.run(&current)?;
264            let overlay_id = p.overlay_id.clone();
265
266            let crash = wait_for_crash(policy);
267            chain.push(RepairAttempt {
268                attempt,
269                intent: current.clone(),
270                overlay_id,
271                triggered_crash: crash.clone(),
272            });
273            last = Some(p);
274
275            match crash {
276                None => break, // estabilizado
277                Some(c) => {
278                    current = (policy.format_intent)(&c.service, c.code);
279                }
280            }
281        }
282
283        let mut p = last.expect("max_attempts >= 1, así que hubo al menos un run");
284        p.repair_chain = chain;
285        Ok(p)
286    }
287}
288
289/// Conecta al bus y drena eventos hasta encontrar un `Crashed` dentro de `crash_window`.
290/// Devuelve `None` si:
291/// - la política no tiene socket;
292/// - la conexión falla (caso "no hay bus") — el bucle decide entonces "estabilizado";
293/// - expira la ventana sin recibir Crashed.
294fn wait_for_crash(policy: &RepairPolicy) -> Option<CrashInfo> {
295    let Some(sock) = &policy.event_sock else { return None };
296    let client = match AgentClient::connect(sock) {
297        Ok(c) => c,
298        Err(e) => {
299            tracing::debug!(error = %e, "auto-repair: no pude conectar al bus, ignoro");
300            return None;
301        }
302    };
303    let deadline = std::time::Instant::now() + policy.crash_window;
304    // Drenamos primero la cola asíncrona ya acumulada (eventos llegados durante el `run`).
305    for ev in client.drain_async() {
306        if let hammer_core::proto::Event::Crashed { service, code } = ev {
307            return Some(CrashInfo { service, code });
308        }
309    }
310    while let Some(remaining) = deadline.checked_duration_since(std::time::Instant::now()) {
311        match client.next_async(remaining) {
312            Ok(hammer_core::proto::Event::Crashed { service, code }) => {
313                return Some(CrashInfo { service, code });
314            }
315            Ok(_other) => continue,
316            Err(_) => break, // timeout o cierre
317        }
318    }
319    None
320}
321
322/// Stub: `hammer-build` no es dependencia directa de `hammer-agent` para no arrastrar el
323/// lab a clientes que sólo quieran hablar con el bus. Implementamos la hidratación local
324/// con `std::fs::hard_link` directamente porque es trivial. Si el árbol del artefacto crece
325/// (symlinks, perms más finos), promovemos a `hammer-build::run_hydrate` añadiendo la dep
326/// de manera condicional.
327fn hammer_build_run_hydrate(artifact_dir: &Path, target_fhs: &Path) -> Result<usize> {
328    let mut files = 0usize;
329    walk(artifact_dir, &mut |rel, kind| {
330        let dst = target_fhs.join(rel);
331        match kind {
332            EntryKind::Dir => {
333                std::fs::create_dir_all(&dst)?;
334            }
335            EntryKind::Symlink(target) => {
336                if let Some(parent) = dst.parent() {
337                    std::fs::create_dir_all(parent)?;
338                }
339                let _ = std::fs::remove_file(&dst);
340                std::os::unix::fs::symlink(&target, &dst)?;
341                files += 1;
342            }
343            EntryKind::File(src) => {
344                if let Some(parent) = dst.parent() {
345                    std::fs::create_dir_all(parent)?;
346                }
347                let _ = std::fs::remove_file(&dst);
348                std::fs::hard_link(&src, &dst).or_else(|_| {
349                    // Cross-fs ⇒ caemos a copia. No optimizamos: hammer-agent espera
350                    // mismo FS; sólo damos el fallback como cinturón de seguridad.
351                    std::fs::copy(&src, &dst).map(|_| ())
352                })?;
353                files += 1;
354            }
355        }
356        Ok::<(), std::io::Error>(())
357    })?;
358    Ok(files)
359}
360
361enum EntryKind {
362    Dir,
363    Symlink(PathBuf),
364    File(PathBuf),
365}
366
367fn walk<F>(root: &Path, f: &mut F) -> std::io::Result<()>
368where
369    F: FnMut(&Path, EntryKind) -> std::io::Result<()>,
370{
371    fn inner<F>(root: &Path, cur: &Path, f: &mut F) -> std::io::Result<()>
372    where
373        F: FnMut(&Path, EntryKind) -> std::io::Result<()>,
374    {
375        for entry in std::fs::read_dir(cur)? {
376            let entry = entry?;
377            let src = entry.path();
378            let rel = src.strip_prefix(root).unwrap();
379            let ft = entry.file_type()?;
380            if ft.is_symlink() {
381                let target = std::fs::read_link(&src)?;
382                f(rel, EntryKind::Symlink(target))?;
383            } else if ft.is_dir() {
384                f(rel, EntryKind::Dir)?;
385                inner(root, &src, f)?;
386            } else if ft.is_file() {
387                f(rel, EntryKind::File(src.clone()))?;
388            }
389        }
390        Ok(())
391    }
392    if !root.is_dir() {
393        return Err(std::io::Error::new(
394            std::io::ErrorKind::NotFound,
395            format!("artifact_dir no existe: {}", root.display()),
396        ));
397    }
398    inner(root, root, f)
399}
400
401#[derive(Debug, Default, Clone, serde::Serialize)]
402pub struct AppliedCounts {
403    pub source_patch: usize,
404    pub config_edit: usize,
405    pub init_rule: usize,
406    pub file_drop: usize,
407}
408
409#[derive(Debug, Clone, serde::Serialize)]
410pub struct VerifyCheck {
411    pub level: CheckLevel,
412    pub msg: String,
413}
414
415#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
416#[serde(rename_all = "kebab-case")]
417pub enum CheckLevel {
418    Pass,
419    Warn,
420    Fail,
421}
422
423impl VerifyCheck {
424    pub fn pass(msg: impl Into<String>) -> Self {
425        Self { level: CheckLevel::Pass, msg: msg.into() }
426    }
427    pub fn warn(msg: impl Into<String>) -> Self {
428        Self { level: CheckLevel::Warn, msg: msg.into() }
429    }
430}
431
432/// El resultado del bucle. El humano lo lee para decidir `hammer commit <overlay_id>` (si
433/// está en modo Overlay) o `hammer discard`.
434#[derive(Debug, Clone, serde::Serialize)]
435pub struct Proposal {
436    pub intent: String,
437    pub swm: Swm,
438    /// `Some` si abrimos un overlay. En modo Prefix, `None`.
439    pub overlay_id: Option<String>,
440    pub prefix: Option<PathBuf>,
441    pub applied: AppliedCounts,
442    pub skipped_source_patches: usize,
443    pub checks: Vec<VerifyCheck>,
444    /// Si la propuesta provino del bucle de auto-reparación, la cadena completa de
445    /// intentos. La última entrada corresponde a este `Proposal`. Para corridas sin
446    /// auto-reparación queda vacío.
447    #[serde(default, skip_serializing_if = "Vec::is_empty")]
448    pub repair_chain: Vec<RepairAttempt>,
449}
450
451/// Un intento dentro del bucle de auto-reparación. Sirve para que el humano vea
452/// cómo evolucionó la intención: original → reparación A → reparación B …
453#[derive(Debug, Clone, serde::Serialize)]
454pub struct RepairAttempt {
455    pub attempt: u32,
456    pub intent: String,
457    /// `Some` cuando el ciclo abrió un overlay; preservado para que el humano pueda hacer
458    /// `hammer discard <id>` sobre intentos intermedios si lo desea.
459    pub overlay_id: Option<String>,
460    /// Si tras este intento llegó un `Crashed`, el servicio y código. `None` ⇒ estabilizado.
461    pub triggered_crash: Option<CrashInfo>,
462}
463
464#[derive(Debug, Clone, serde::Serialize)]
465pub struct CrashInfo {
466    pub service: String,
467    pub code: i32,
468}
469
470/// Política del bucle de auto-reparación. La IA NUNCA decide por sí sola seguir
471/// reintentando: la política la fija el llamador (CLI o cliente) y acota el blast-radius.
472#[derive(Clone)]
473pub struct RepairPolicy {
474    /// Máximo de intentos en total (incluyendo el primero). Mínimo efectivo: 1.
475    pub max_attempts: u32,
476    /// Ventana de espera tras aplicar para detectar un `Crashed` por el bus.
477    pub crash_window: Duration,
478    /// Socket del bus de eventos. Si `None`, el orquestador no escucha y devuelve
479    /// el `Proposal` tras la primera aplicación (equivalente a `run`).
480    pub event_sock: Option<PathBuf>,
481    /// Convierte un crash `(service, code)` en la próxima intención. Por defecto:
482    /// `"repair service <s> crashed code <c>"`.
483    pub format_intent: fn(&str, i32) -> String,
484}
485
486impl std::fmt::Debug for RepairPolicy {
487    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
488        f.debug_struct("RepairPolicy")
489            .field("max_attempts", &self.max_attempts)
490            .field("crash_window", &self.crash_window)
491            .field("event_sock", &self.event_sock)
492            .field("format_intent", &"fn(&str, i32) -> String")
493            .finish()
494    }
495}
496
497pub fn default_crash_intent(service: &str, code: i32) -> String {
498    format!("repair service {service} crashed code {code}")
499}
500
501impl Default for RepairPolicy {
502    fn default() -> Self {
503        Self {
504            max_attempts: 3,
505            crash_window: Duration::from_secs(5),
506            event_sock: None,
507            format_intent: default_crash_intent,
508        }
509    }
510}
511
512#[cfg(test)]
513mod tests {
514    use super::*;
515    use crate::translator::MockTranslator;
516    use hammer_core::swm::{Base, Mutation};
517    use std::collections::BTreeMap;
518
519    fn baseref() -> BaseRef {
520        BaseRef { distro_version: "2026-06-06".into(), pins: BTreeMap::new() }
521    }
522
523    fn swm_config_only() -> Swm {
524        Swm {
525            swm_version: 1,
526            base: Base { distro_version: "2026-06-06".into(), pins: BTreeMap::new() },
527            mutations: vec![Mutation::ConfigEdit {
528                file: "/etc/network.conf".into(),
529                inline_diff: "- DHCP=yes\n+ STATIC=1\n".into(),
530            }],
531            signature: None,
532        }
533    }
534
535    #[test]
536    fn orchestrator_applies_config_edit_to_prefix() {
537        let d = tempfile::tempdir().unwrap();
538        let prefix = d.path().join("prefix");
539        std::fs::create_dir_all(prefix.join("etc")).unwrap();
540        std::fs::write(prefix.join("etc/network.conf"), "DHCP=yes\nMTU=1500\n").unwrap();
541
542        let t = MockTranslator::empty().with_intent("set static ip", swm_config_only());
543        let o = Orchestrator::new(
544            t,
545            baseref(),
546            ApplyTarget::Prefix { prefix: prefix.clone() },
547            CompileMode::Skip,
548        );
549        let p = o.run("set static ip").unwrap();
550        assert_eq!(p.applied.config_edit, 1);
551        assert!(p.overlay_id.is_none());
552        assert_eq!(p.checks.len(), 1);
553        assert_eq!(p.checks[0].level, CheckLevel::Pass);
554
555        let got = std::fs::read_to_string(prefix.join("etc/network.conf")).unwrap();
556        assert_eq!(got, "STATIC=1\nMTU=1500\n");
557    }
558
559    #[test]
560    fn orchestrator_rejects_base_mismatch() {
561        let mut swm = swm_config_only();
562        swm.base.distro_version = "2027-01-01".into();
563        let t = MockTranslator::empty().with_intent("x", swm);
564        let d = tempfile::tempdir().unwrap();
565        let o = Orchestrator::new(
566            t,
567            baseref(),
568            ApplyTarget::Prefix { prefix: d.path().to_path_buf() },
569            CompileMode::Skip,
570        );
571        let err = o.run("x").unwrap_err().to_string();
572        assert!(err.contains("base incompatible"), "{err}");
573    }
574
575    #[test]
576    fn orchestrator_skips_source_patch_with_warn() {
577        let swm = Swm {
578            swm_version: 1,
579            base: Base { distro_version: "2026-06-06".into(), pins: BTreeMap::new() },
580            mutations: vec![Mutation::SourcePatch {
581                repo: Some("git://x".into()),
582                commit: Some("abc".into()),
583                tarball: None,
584                sha256: None,
585                patch: None,
586                patch_url: None,
587                build: hammer_core::swm::SwmBuild {
588                    compiler: "zig-cc".into(),
589                    target: "x86_64-linux-musl".into(),
590                    link: "static".into(),
591                    flags: vec![],
592                },
593                target_bin: "/bin/x".into(),
594                expected_hash: None,
595            }],
596            signature: None,
597        };
598        let t = MockTranslator::empty().with_intent("compile x", swm);
599        let d = tempfile::tempdir().unwrap();
600        let o = Orchestrator::new(
601            t,
602            baseref(),
603            ApplyTarget::Prefix { prefix: d.path().to_path_buf() },
604            CompileMode::Skip,
605        );
606        let p = o.run("compile x").unwrap();
607        assert_eq!(p.skipped_source_patches, 1);
608        assert_eq!(p.checks[0].level, CheckLevel::Warn);
609    }
610
611    #[test]
612    fn orchestrator_translates_unknown_intent_errors() {
613        let t = MockTranslator::empty();
614        let d = tempfile::tempdir().unwrap();
615        let o = Orchestrator::new(
616            t,
617            baseref(),
618            ApplyTarget::Prefix { prefix: d.path().to_path_buf() },
619            CompileMode::Skip,
620        );
621        let err = o.run("haz lo que tú quieras").unwrap_err().to_string();
622        assert!(err.contains("no reconocida"), "{err}");
623    }
624}