//! **E4 — upgrades atómicos con rollback** ([SDD 13 §E4](../../docs/13-release-engineering.md)).
//!
//! Aplicar un árbol Stage1/producto **nuevo** al root vivo sin reinstalar, de forma atómica y con
//! **rollback** al árbol anterior. El árbol nuevo es un artefacto sellado del store (CAS BLAKE3); el
//! "apply" lo proyecta sobre el FHS vivo (`target_root`, normalmente `/`), registra cada cambio en el
//! [diario](takana_journal) y deja una **generación** nueva. Si algo sale mal, `rollback` revierte
//! exactamente al árbol anterior.
//!
//! ## Modelo de generaciones (inspirado en NixOS, aplicado in-place al FHS)
//!
//! El sistema booteado lee el userland directamente del root (vda2), no de un symlink indirecto. Así
//! que el "switch" no es un único symlink: es la proyección de los ficheros del árbol nuevo sobre el
//! root, fichero a fichero (escritura a temporal + `rename` ⇒ cada fichero conmuta atómicamente), más
//! un **registro de generación** que es la fuente de verdad de "qué árbol está vivo" y guarda lo
//! necesario para deshacer.
//!
//! ```text
//! <state_root>/                       (default /var/lib/hammer/upgrades)
//!   generations/
//!     <N>/manifest.json               id, tree_dir (store), tree_content (of_tree), parent, cambios
//!     <N>/backup/<rel...>             bytes PREVIOS de cada fichero que el apply pisó o borró
//!   current                           fichero con el id de la generación viva (escritura atómica)
//! ```
//!
//! **Atomicidad y rollback.** El apply respalda el estado previo de todo path que toca *antes* de
//! tocarlo; el commit-point es el avance de `current`. Un corte a mitad deja `current` en la
//! generación vieja y los backups intactos ⇒ `rollback` (o un re-apply) restaura. El rollback deshace
//! en orden inverso: borra lo que el apply creó, restaura desde `backup/` lo que pisó o borró, y
//! retrocede `current` al padre.
//!
//! **Alcance del primer corte.** Maneja ficheros regulares y symlinks (los dos tipos de contenido que
//! `of_tree` hashea); crea los directorios intermedios que haga falta. El cableado del init de boot
//! para que el rollback sobreviva un kernel-panic a media escritura (journal de intención + replay) es
//! endurecimiento posterior, ortogonal a esta lógica.

use std::path::{Path, PathBuf};

use takana_core::{ArtifactHash, Store};
use takana_journal::{Actor, Journal, MutationEvent, MutationOp, Source};
use serde::{Deserialize, Serialize};

pub mod boot_graph;

pub const DEFAULT_STATE_ROOT: &str = "/var/lib/hammer/upgrades";

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("io: {0}")]
    Io(#[from] std::io::Error),
    #[error("core: {0}")]
    Core(#[from] takana_core::Error),
    #[error("journal: {0}")]
    Journal(#[from] takana_journal::Error),
    #[error("json: {0}")]
    Json(#[from] serde_json::Error),
    #[error(
        "integridad: el árbol '{dir}' del store hashea a {got} pero se esperaba {want} \
         — store corrupto o árbol equivocado; NO se aplica"
    )]
    Verify { dir: String, want: String, got: String },
    #[error("upgrade: no hay generación viva que revertir (sistema sin upgrades aplicados)")]
    NothingToRollback,
    #[error(
        "upgrade: hay un apply interrumpido pendiente (generación {generation}) — corré \
         `hammer upgrade recover` (completa) o `hammer upgrade recover --rollback` (deshace) antes de seguir"
    )]
    PendingExists { generation: u64 },
    #[error(
        "upgrade: el registro del apply pendiente ({path}) está VACÍO. Es la huella de un CORTE a \
         mitad de escritura: el `rename` llegó al disco y los bytes no. El plan se perdió, así que \
         no se puede completar ni deshacer automáticamente — pero el árbol de respaldos sigue en \
         {backups}: restaurá desde ahí, y borrá el `pending.json` y la generación a medias"
    )]
    PendingCorrupt { path: String, backups: String },
    #[error("upgrade: {0}")]
    Other(String),
}

pub type Result<T> = std::result::Result<T, Error>;

/// Qué le pasó a un path concreto al aplicar una generación. Determina cómo se deshace.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum ChangeOp {
    /// El path no existía antes ⇒ rollback lo **borra**.
    Added,
    /// El path existía y fue pisado ⇒ `backup/` guarda los bytes previos ⇒ rollback los **restaura**.
    Replaced,
    /// El path existía en el árbol previo y el nuevo no lo tiene ⇒ apply lo borra; `backup/` guarda
    /// los bytes previos ⇒ rollback los **restaura**.
    Removed,
}

/// Un cambio atómico sobre un path del FHS, relativo a `target_root`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Change {
    pub path: PathBuf,
    pub op: ChangeOp,
}

/// Manifiesto persistente de una generación — la unidad de upgrade/rollback.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GenerationManifest {
    pub id: u64,
    /// Directorio del artefacto en el store (`<hash>-<name>`) cuyo árbol esta generación proyectó.
    pub tree_dir: String,
    /// `of_tree` (`b3:…`) del árbol aplicado: ancla de integridad y desempate vs el store.
    pub tree_content: String,
    /// Generación anterior (a la que el rollback vuelve). `None` en la primera.
    pub parent: Option<u64>,
    /// Segundos desde UNIX epoch al apply.
    pub created_at: u64,
    /// Lista relativa (ordenada) de los ficheros que el árbol aporta — sirve para que el SIGUIENTE
    /// apply sepa qué retirar (paths del árbol previo ausentes en el nuevo).
    pub files: Vec<PathBuf>,
    /// Cada path tocado y cómo, para deshacer.
    pub changes: Vec<Change>,
}

/// Reporte de un [`apply`].
#[derive(Debug, Clone, Default)]
pub struct ApplyReport {
    pub generation: u64,
    pub added: Vec<PathBuf>,
    pub replaced: Vec<PathBuf>,
    pub removed: Vec<PathBuf>,
    /// `true` si el árbol pedido ya era el vivo (apply idempotente, no se creó generación nueva).
    pub already_current: bool,
}

/// Reporte de un [`recover`].
#[derive(Debug, Clone, Default)]
pub struct RecoverReport {
    /// `false` si no había nada que recuperar (no había intento pendiente).
    pub recovered: bool,
    /// `true` = se completó el apply (roll-forward); `false` = se deshizo (roll-back).
    pub rolled_forward: bool,
    /// Generación afectada por el intento pendiente.
    pub generation: u64,
}

/// Reporte de un [`rollback`].
#[derive(Debug, Clone, Default)]
pub struct RollbackReport {
    /// Generación que se revirtió (la que estaba viva).
    pub reverted: u64,
    /// Generación que quedó viva tras revertir (`None` si se volvió al estado pre-upgrades).
    pub now_current: Option<u64>,
    pub restored: Vec<PathBuf>,
    pub deleted: Vec<PathBuf>,
}

// ---------------------------------------------------------------------------------------------------
// Registro de generaciones (lectura del estado).
// ---------------------------------------------------------------------------------------------------

fn generations_dir(state_root: &Path) -> PathBuf {
    state_root.join("generations")
}

fn current_file(state_root: &Path) -> PathBuf {
    state_root.join("current")
}

fn manifest_path(state_root: &Path, id: u64) -> PathBuf {
    generations_dir(state_root).join(id.to_string()).join("manifest.json")
}

/// El **journal de intención**: el plan de un apply en curso. Existe sólo entre el inicio de la
/// proyección y el commit; si sobrevive a un reinicio, un apply quedó a medias y hay que [`recover`].
fn pending_path(state_root: &Path) -> PathBuf {
    state_root.join("pending.json")
}

/// Lee el plan pendiente (apply interrumpido), o `None` si no hay ninguno.
pub fn pending(state_root: &Path) -> Result<Option<GenerationManifest>> {
    let p = pending_path(state_root);
    if !p.is_file() {
        return Ok(None);
    }
    let bytes = std::fs::read(&p)?;
    // Un `pending.json` VACÍO o ilegible es la huella de un corte a mitad de escritura — justo el
    // caso para el que existe `recover`. Un `EOF while parsing a value` crudo no dice eso: dice que
    // el JSON está mal, y manda a mirar al lugar equivocado.
    if bytes.iter().all(|b| b.is_ascii_whitespace()) {
        return Err(Error::PendingCorrupt {
            path: p.display().to_string(),
            backups: generations_dir(state_root).display().to_string(),
        });
    }
    Ok(Some(serde_json::from_slice(&bytes)?))
}

fn write_pending(state_root: &Path, manifest: &GenerationManifest) -> Result<()> {
    write_atomic(&pending_path(state_root), &serde_json::to_vec_pretty(manifest)?)
}

fn clear_pending(state_root: &Path) -> Result<()> {
    match std::fs::remove_file(pending_path(state_root)) {
        Ok(()) => Ok(()),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(e) => Err(e.into()),
    }
}

/// Id de la generación viva, o `None` si nunca se aplicó un upgrade.
pub fn current(state_root: &Path) -> Result<Option<u64>> {
    let f = current_file(state_root);
    if !f.is_file() {
        return Ok(None);
    }
    let s = std::fs::read_to_string(&f)?;
    let s = s.trim();
    if s.is_empty() {
        return Ok(None);
    }
    s.parse::<u64>()
        .map(Some)
        .map_err(|_| Error::Other(format!("`current` ilegible: {s:?}")))
}

/// Lee el manifiesto de una generación concreta.
pub fn get(state_root: &Path, id: u64) -> Result<Option<GenerationManifest>> {
    let p = manifest_path(state_root, id);
    if !p.is_file() {
        return Ok(None);
    }
    Ok(Some(serde_json::from_slice(&std::fs::read(&p)?)?))
}

/// Lista todas las generaciones registradas, ordenadas por id ascendente.
pub fn list(state_root: &Path) -> Result<Vec<GenerationManifest>> {
    let dir = generations_dir(state_root);
    if !dir.is_dir() {
        return Ok(Vec::new());
    }
    let mut out: Vec<GenerationManifest> = Vec::new();
    for entry in std::fs::read_dir(&dir)? {
        let entry = entry?;
        let mp = entry.path().join("manifest.json");
        if mp.is_file() {
            out.push(serde_json::from_slice(&std::fs::read(&mp)?)?);
        }
    }
    out.sort_by_key(|m| m.id);
    Ok(out)
}

fn next_id(state_root: &Path) -> Result<u64> {
    Ok(list(state_root)?.iter().map(|m| m.id).max().map_or(1, |m| m + 1))
}

/// La **cadena viva**: la generación actual seguida de sus ancestros vía `parent`, de la más nueva a
/// la más vieja. Son las únicas generaciones que el `rollback` puede necesitar (no hay "redo": tras un
/// rollback las generaciones más nuevas quedan **huérfanas**). Vacía si no hay `current`.
pub fn live_chain(state_root: &Path) -> Result<Vec<u64>> {
    let mut chain = Vec::new();
    let mut cur = current(state_root)?;
    let mut guard = 0usize; // anti-ciclo defensivo (un manifiesto manipulado no debe colgar el proceso)
    while let Some(id) = cur {
        if chain.contains(&id) || guard > 100_000 {
            break;
        }
        chain.push(id);
        guard += 1;
        cur = get(state_root, id)?.and_then(|m| m.parent);
    }
    Ok(chain)
}

/// Reporte de un [`prune`].
#[derive(Debug, Clone, Default)]
pub struct PruneReport {
    pub removed: Vec<u64>,
    pub kept: Vec<u64>,
}

/// Recupera espacio borrando generaciones (manifiesto + `backup/`) que el rollback ya no puede
/// alcanzar. **Siempre** se conservan las de la [cadena viva](live_chain); el resto (huérfanas tras un
/// rollback) se borran. Si `keep` es `Some(n)`, además se **recorta la cadena** a sus `n` generaciones
/// más nuevas — esto limita la profundidad de rollback disponible (como `delete-generations` en NixOS),
/// así que `None` (conservar toda la cadena) es lo seguro por defecto. Idempotente.
pub fn prune(state_root: &Path, keep: Option<usize>) -> Result<PruneReport> {
    let chain = live_chain(state_root)?;
    let keep_set: std::collections::BTreeSet<u64> = match keep {
        Some(n) => chain.iter().take(n).copied().collect(),
        None => chain.iter().copied().collect(),
    };
    let mut report = PruneReport::default();
    for m in list(state_root)? {
        if keep_set.contains(&m.id) {
            report.kept.push(m.id);
            continue;
        }
        let gen_base = generations_dir(state_root).join(m.id.to_string());
        std::fs::remove_dir_all(&gen_base)?;
        report.removed.push(m.id);
    }
    report.removed.sort_unstable();
    report.kept.sort_unstable();
    Ok(report)
}

// ---------------------------------------------------------------------------------------------------
// Enumeración de árboles.
// ---------------------------------------------------------------------------------------------------

/// Un fichero de un árbol: su path relativo y su tipo de contenido (lo que `of_tree` distingue).
#[derive(Debug, Clone)]
struct TreeEntry {
    rel: PathBuf,
    kind: EntryKind,
}

#[derive(Debug, Clone)]
enum EntryKind {
    /// Fichero regular. `exec` = bit de ejecución (de_tree lo mira).
    File { exec: bool },
    /// Symlink → target literal (sin resolver).
    Symlink(PathBuf),
}

/// Camina `root` y devuelve sus ficheros (regulares + symlinks) con path relativo, ordenados. Los
/// directorios no se listan: se recrean al proyectar según haga falta.
fn enumerate_tree(root: &Path) -> Result<Vec<TreeEntry>> {
    let mut out = Vec::new();
    fn walk(root: &Path, dir: &Path, out: &mut Vec<TreeEntry>) -> Result<()> {
        use std::os::unix::fs::PermissionsExt;
        for entry in std::fs::read_dir(dir)? {
            let entry = entry?;
            let p = entry.path();
            let meta = std::fs::symlink_metadata(&p)?;
            let ft = meta.file_type();
            let rel = p.strip_prefix(root).expect("p ⊂ root").to_path_buf();
            if ft.is_symlink() {
                out.push(TreeEntry { rel, kind: EntryKind::Symlink(std::fs::read_link(&p)?) });
            } else if ft.is_dir() {
                walk(root, &p, out)?;
            } else if ft.is_file() {
                let exec = meta.permissions().mode() & 0o111 != 0;
                out.push(TreeEntry { rel, kind: EntryKind::File { exec } });
            }
        }
        Ok(())
    }
    walk(root, root, &mut out)?;
    out.sort_by(|a, b| a.rel.cmp(&b.rel));
    Ok(out)
}

// ---------------------------------------------------------------------------------------------------
// Apply.
// ---------------------------------------------------------------------------------------------------

/// Aplica el árbol `tree_dir` del store sobre `target_root`, dejando una generación nueva.
///
/// Pasos: (1) resuelve y **verifica** el árbol del store (su `of_tree` debe casar el del índice);
/// (2) diferencia contra el árbol de la generación viva ⇒ qué se añade/pisa/retira; (3) **respalda**
/// el estado previo de cada path tocado; (4) proyecta el árbol nuevo (escritura a temporal + rename)
/// y retira los ficheros sobrantes, registrando cada cambio en el diario; (5) escribe el manifiesto y
/// **avanza `current`** (commit-point).
pub fn apply(
    store_root: &Path,
    tree_dir: &str,
    expected: Option<&str>,
    target_root: &Path,
    state_root: &Path,
    journal: Option<&Journal>,
) -> Result<ApplyReport> {
    let store = Store::open(store_root)?;
    let tree_path = store.root().join(tree_dir);
    if !tree_path.is_dir() {
        return Err(Error::Other(format!(
            "el árbol '{tree_dir}' no existe en el store {}",
            store.root().display()
        )));
    }
    // El of_tree real ancla la identidad del árbol. Si el llamador anuncia uno esperado (de un índice
    // de mirror o una release firmada), lo exigimos: defiende de un store corrupto/manipulado entre el
    // sellado y el apply. Sin `expected`, el store local se confía (sellado read-only) y el of_tree
    // sólo se registra en el manifiesto.
    let got = ArtifactHash::of_tree(&tree_path)?;
    let tree_content = got.as_str().to_string();
    if let Some(want) = expected {
        if tree_content != want {
            return Err(Error::Verify {
                dir: tree_dir.to_string(),
                want: want.to_string(),
                got: tree_content,
            });
        }
    }

    // Un apply previo interrumpido deja un `pending.json` ⇒ exigimos `recover` explícito antes de tocar
    // nada (no pisamos un FHS a medias). El recover lo completa (roll-forward) o lo deshace (roll-back).
    if let Some(p) = pending(state_root)? {
        return Err(Error::PendingExists { generation: p.id });
    }

    // Idempotencia: si el árbol vivo ya es éste, no hacemos nada.
    if let Some(cur) = current(state_root)? {
        if let Some(m) = get(state_root, cur)? {
            if m.tree_content == tree_content {
                return Ok(ApplyReport { generation: cur, already_current: true, ..Default::default() });
            }
        }
    }

    // --- PLAN completo ANTES de tocar nada: entradas del árbol nuevo + ficheros a retirar + op/path ---
    let new_entries = enumerate_tree(&tree_path)?;
    let new_files: Vec<PathBuf> = new_entries.iter().map(|e| e.rel.clone()).collect();
    let new_set: std::collections::BTreeSet<&PathBuf> = new_files.iter().collect();
    let prev_id = current(state_root)?;
    let prev_files: Vec<PathBuf> = match prev_id {
        Some(cur) => get(state_root, cur)?.map(|m| m.files).unwrap_or_default(),
        None => Vec::new(),
    };
    let removed_files: Vec<PathBuf> =
        prev_files.into_iter().filter(|p| !new_set.contains(p)).collect();

    let mut changes: Vec<Change> = Vec::with_capacity(new_entries.len() + removed_files.len());
    for e in &new_entries {
        let existed = std::fs::symlink_metadata(target_root.join(&e.rel)).is_ok();
        changes.push(Change {
            path: e.rel.clone(),
            op: if existed { ChangeOp::Replaced } else { ChangeOp::Added },
        });
    }
    for rel in &removed_files {
        if std::fs::symlink_metadata(target_root.join(rel)).is_ok() {
            changes.push(Change { path: rel.clone(), op: ChangeOp::Removed });
        }
    }

    let id = next_id(state_root)?;
    let manifest = GenerationManifest {
        id,
        tree_dir: tree_dir.to_string(),
        tree_content,
        parent: prev_id,
        created_at: now_secs(),
        files: new_files,
        changes,
    };

    let backup_root = generations_dir(state_root).join(id.to_string()).join("backup");
    std::fs::create_dir_all(&backup_root)?;

    // (INTENTO) escribir el plan ANTES de proyectar: el commit-point sigue siendo `current`, pero ahora
    // un corte a media proyección es recuperable — `recover` re-ejecuta el plan (idempotente) o lo deshace.
    write_pending(state_root, &manifest)?;

    // (PROYECTAR) ejecutar el plan: backups idempotentes (no pisan el original) + escritura temporal+rename.
    project_plan(&tree_path, &manifest, target_root, &backup_root, journal)?;

    // (COMMIT) manifiesto definitivo + avance de `current` + limpiar el intento.
    write_atomic(&manifest_path(state_root, id), &serde_json::to_vec_pretty(&manifest)?)?;
    set_current(state_root, Some(id))?;
    clear_pending(state_root)?;
    Ok(report_from_changes(id, &manifest.changes))
}

/// Construye el [`ApplyReport`] a partir de la lista de cambios del plan.
fn report_from_changes(id: u64, changes: &[Change]) -> ApplyReport {
    let mut r = ApplyReport { generation: id, ..Default::default() };
    for c in changes {
        match c.op {
            ChangeOp::Added => r.added.push(c.path.clone()),
            ChangeOp::Replaced => r.replaced.push(c.path.clone()),
            ChangeOp::Removed => r.removed.push(c.path.clone()),
        }
    }
    r
}

/// La ruta REAL de `target_root`, resolviendo los symlinks del tramo que ya existe y dejando
/// pegado el tramo que todavía no (que `apply` va a crear como directorios de verdad). Sin esto no
/// habría con qué comparar cuando el root aún no existe — el caso de una imagen nueva.
fn raiz_real(target_root: &Path) -> PathBuf {
    let mut faltan: Vec<std::ffi::OsString> = Vec::new();
    let mut p = target_root.to_path_buf();
    loop {
        if let Ok(mut r) = p.canonicalize() {
            for c in faltan.iter().rev() {
                r.push(c);
            }
            return r;
        }
        match (p.file_name().map(|n| n.to_os_string()), p.parent().map(|x| x.to_path_buf())) {
            (Some(n), Some(par)) => {
                faltan.push(n);
                p = par;
            }
            _ => return target_root.to_path_buf(),
        }
    }
}

/// El destino no puede salirse del root. **Se mide el DIRECTORIO padre, no el destino**: un
/// `Replaced` sobre un symlink que apunta afuera es legítimo —`rename` pisa el symlink, no escribe
/// a través de él— y medir el destino lo rechazaría por error.
///
/// EL FALLO QUE TAPA (medido 2026-08-30, `apply_no_escribe_fuera_del_root_por_un_symlink`): igual
/// que en `takana-build/hydrate.rs`, proyectar es `target_root.join(rel)` + escribir por ruta, así
/// que un symlink de DIRECTORIO ya presente en el root se sigue. Una generación que deja puesto
/// `usr/share/pkg → /algún/lado` hacía que la siguiente escribiera `usr/share/pkg/archivo` FUERA
/// del root, con `ApplyReport` en verde. Con `--root /` no cambia nada (todo empieza por `/`); con
/// `--root /mnt/imagen`, que es el caso real de armar una imagen, sí.
fn dentro_de_raiz(raiz: &Path, dst: &Path) -> Result<()> {
    let padre = match dst.parent() {
        Some(p) => p,
        None => return Ok(()),
    };
    let real = raiz_real(padre);
    if !real.starts_with(raiz) {
        return Err(Error::Io(std::io::Error::other(format!(
            "la proyección se saldría del root: {} lleva a {}, fuera de {}.              Es un symlink de directorio ya presente en el root (probablemente de otra generación);              no se escribe a través de él.",
            padre.display(),
            real.display(),
            raiz.display()
        ))));
    }
    Ok(())
}

/// Ejecuta el plan de una generación sobre `target_root`: proyecta los Added/Replaced del árbol y
/// retira los Removed, respaldando el ORIGINAL de cada path tocado **idempotentemente** (no pisa un
/// backup ya existente — clave para que `recover` pueda re-ejecutar sin perder el estado previo). Es
/// **re-entrante**: correrlo dos veces deja el mismo resultado.
fn project_plan(
    tree_path: &Path,
    manifest: &GenerationManifest,
    target_root: &Path,
    backup_root: &Path,
    journal: Option<&Journal>,
) -> Result<()> {
    let entries = enumerate_tree(tree_path)?;
    let by_rel: std::collections::BTreeMap<&PathBuf, &TreeEntry> =
        entries.iter().map(|e| (&e.rel, e)).collect();
    let raiz = raiz_real(target_root);
    for ch in &manifest.changes {
        let dst = target_root.join(&ch.path);
        dentro_de_raiz(&raiz, &dst)?;
        match ch.op {
            ChangeOp::Added | ChangeOp::Replaced => {
                if ch.op == ChangeOp::Replaced {
                    backup_existing_once(&dst, &backup_root.join(&ch.path))?;
                }
                if let Some(e) = by_rel.get(&ch.path) {
                    project_entry(tree_path, e, &dst)?;
                }
            }
            ChangeOp::Removed => {
                backup_existing_once(&dst, &backup_root.join(&ch.path))?;
                remove_path(&dst)?;
            }
        }
        journal_change(journal, &manifest.tree_dir, &dst, ch.op);
    }
    Ok(())
}

/// Deshace el plan de un manifiesto sobre `target_root` (orden inverso): borra los Added, restaura los
/// Replaced/Removed desde `backup_root`. Idempotente. Compartido por [`rollback`] y `recover --rollback`.
fn undo_changes(
    manifest: &GenerationManifest,
    target_root: &Path,
    backup_root: &Path,
    journal: Option<&Journal>,
    report_restored: &mut Vec<PathBuf>,
    report_deleted: &mut Vec<PathBuf>,
) -> Result<()> {
    let raiz = raiz_real(target_root);
    for ch in manifest.changes.iter().rev() {
        let dst = target_root.join(&ch.path);
        dentro_de_raiz(&raiz, &dst)?;
        match ch.op {
            ChangeOp::Added => {
                if std::fs::symlink_metadata(&dst).is_ok() {
                    remove_path(&dst)?;
                    report_deleted.push(ch.path.clone());
                    journal_change(journal, &manifest.tree_dir, &dst, ChangeOp::Removed);
                }
            }
            ChangeOp::Replaced | ChangeOp::Removed => {
                restore_from_backup(&backup_root.join(&ch.path), &dst)?;
                report_restored.push(ch.path.clone());
                journal_change(journal, &manifest.tree_dir, &dst, ChangeOp::Replaced);
            }
        }
    }
    Ok(())
}

// ---------------------------------------------------------------------------------------------------
// Rollback.
// ---------------------------------------------------------------------------------------------------

/// Revierte la generación viva: deshace sus cambios (borra lo añadido, restaura desde `backup/` lo
/// pisado/retirado) y retrocede `current` al padre. Idempotente respecto a paths ya restaurados.
pub fn rollback(
    target_root: &Path,
    state_root: &Path,
    journal: Option<&Journal>,
) -> Result<RollbackReport> {
    let cur = current(state_root)?.ok_or(Error::NothingToRollback)?;
    let m = get(state_root, cur)?
        .ok_or_else(|| Error::Other(format!("manifiesto de la generación {cur} ausente")))?;
    let backup_root = generations_dir(state_root).join(cur.to_string()).join("backup");

    let mut report = RollbackReport { reverted: cur, now_current: m.parent, ..Default::default() };
    undo_changes(&m, target_root, &backup_root, journal, &mut report.restored, &mut report.deleted)?;
    set_current(state_root, m.parent)?;
    Ok(report)
}

/// Completa o deshace un apply interrumpido (un `pending.json` que sobrevivió a un corte/reinicio).
///
/// - **roll-forward** (`rollback=false`, default): re-ejecuta el plan pendiente (idempotente) y commitea
///   — el FHS queda como si el apply hubiera terminado. Re-verifica el `of_tree` del árbol contra el que
///   el plan registró (defensa ante un store que cambió bajo los pies).
/// - **roll-back** (`rollback=true`): deshace el plan (restaura backups, borra lo añadido), elimina la
///   generación a medias y vuelve `current` al padre — el FHS queda como ANTES del apply.
///
/// No-op (sin error) si no hay intento pendiente. Idempotente: re-correrlo es seguro.
pub fn recover(
    store_root: &Path,
    target_root: &Path,
    state_root: &Path,
    journal: Option<&Journal>,
    rollback: bool,
) -> Result<RecoverReport> {
    let manifest = match pending(state_root)? {
        Some(m) => m,
        None => return Ok(RecoverReport::default()),
    };
    let backup_root = generations_dir(state_root).join(manifest.id.to_string()).join("backup");

    if rollback {
        let (mut restored, mut deleted) = (Vec::new(), Vec::new());
        undo_changes(&manifest, target_root, &backup_root, journal, &mut restored, &mut deleted)?;
        // El apply nunca avanzó `current` (eso era el commit-point); por las dudas, asegurarlo en el padre.
        if current(state_root)? == Some(manifest.id) {
            set_current(state_root, manifest.parent)?;
        }
        let _ = std::fs::remove_dir_all(generations_dir(state_root).join(manifest.id.to_string()));
        clear_pending(state_root)?;
        return Ok(RecoverReport { recovered: true, rolled_forward: false, generation: manifest.id });
    }

    // roll-forward: re-resolver y re-verificar el árbol del store, completar la proyección, commitear.
    let store = Store::open(store_root)?;
    let tree_path = store.root().join(&manifest.tree_dir);
    if !tree_path.is_dir() {
        return Err(Error::Other(format!(
            "recover: el árbol '{}' del intento ya no existe en el store; usá --rollback",
            manifest.tree_dir
        )));
    }
    let got = ArtifactHash::of_tree(&tree_path)?;
    if got.as_str() != manifest.tree_content {
        return Err(Error::Verify {
            dir: manifest.tree_dir.clone(),
            want: manifest.tree_content.clone(),
            got: got.as_str().to_string(),
        });
    }
    project_plan(&tree_path, &manifest, target_root, &backup_root, journal)?;
    write_atomic(&manifest_path(state_root, manifest.id), &serde_json::to_vec_pretty(&manifest)?)?;
    set_current(state_root, Some(manifest.id))?;
    clear_pending(state_root)?;
    Ok(RecoverReport { recovered: true, rolled_forward: true, generation: manifest.id })
}

// ---------------------------------------------------------------------------------------------------
// Primitivas de filesystem.
// ---------------------------------------------------------------------------------------------------

/// Proyecta una entrada del árbol sobre `dst` (escritura a temporal hermano + `rename` atómico).
fn project_entry(tree_root: &Path, e: &TreeEntry, dst: &Path) -> Result<()> {
    if let Some(parent) = dst.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let tmp = tmp_sibling(dst);
    let _ = remove_path(&tmp);
    match &e.kind {
        EntryKind::File { exec } => {
            let src = tree_root.join(&e.rel);
            std::fs::copy(&src, &tmp)?;
            if *exec {
                use std::os::unix::fs::PermissionsExt;
                let mode = std::fs::metadata(&src)?.permissions().mode();
                std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(mode))?;
            }
        }
        EntryKind::Symlink(target) => {
            std::os::unix::fs::symlink(target, &tmp)?;
        }
    }
    // rename pisa el destino atómicamente (mismo dir ⇒ mismo FS).
    std::fs::rename(&tmp, dst)?;
    Ok(())
}

/// Respalda el estado PREVIO de `src` (fichero o symlink) en `bak`, preservando el tipo.
fn backup_existing(src: &Path, bak: &Path) -> Result<()> {
    if let Some(parent) = bak.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let meta = std::fs::symlink_metadata(src)?;
    if meta.file_type().is_symlink() {
        let _ = remove_path(bak);
        std::os::unix::fs::symlink(std::fs::read_link(src)?, bak)?;
    } else if meta.is_file() {
        let _ = remove_path(bak);
        std::fs::copy(src, bak)?;
        std::fs::set_permissions(bak, meta.permissions())?;
    } else {
        return Err(Error::Other(format!(
            "no sé respaldar {} (no es fichero ni symlink)",
            src.display()
        )));
    }
    Ok(())
}

/// Respalda el original de `src` en `bak` **sólo si aún no hay backup** y `src` existe. Idempotente:
/// preserva el primer original capturado (clave para que un `recover` roll-forward re-ejecute la
/// proyección sin sobrescribir el backup con la versión ya proyectada).
fn backup_existing_once(src: &Path, bak: &Path) -> Result<()> {
    if std::fs::symlink_metadata(bak).is_ok() {
        return Ok(()); // ya respaldado
    }
    if std::fs::symlink_metadata(src).is_err() {
        return Ok(()); // nada que respaldar
    }
    backup_existing(src, bak)
}

/// Restaura `dst` desde su backup (atómico vía temporal). Si el backup no existe, no hace nada
/// (idempotencia: ya restaurado o nunca existió).
fn restore_from_backup(bak: &Path, dst: &Path) -> Result<()> {
    let meta = match std::fs::symlink_metadata(bak) {
        Ok(m) => m,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()),
        Err(e) => return Err(e.into()),
    };
    if let Some(parent) = dst.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let tmp = tmp_sibling(dst);
    let _ = remove_path(&tmp);
    if meta.file_type().is_symlink() {
        std::os::unix::fs::symlink(std::fs::read_link(bak)?, &tmp)?;
    } else {
        std::fs::copy(bak, &tmp)?;
        std::fs::set_permissions(&tmp, meta.permissions())?;
    }
    std::fs::rename(&tmp, dst)?;
    Ok(())
}

fn remove_path(p: &Path) -> Result<()> {
    match std::fs::symlink_metadata(p) {
        Ok(m) if m.is_dir() && !m.file_type().is_symlink() => std::fs::remove_dir_all(p)?,
        Ok(_) => std::fs::remove_file(p)?,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
        Err(e) => return Err(e.into()),
    }
    Ok(())
}

fn tmp_sibling(dst: &Path) -> PathBuf {
    let name = dst.file_name().map(|n| n.to_string_lossy().into_owned()).unwrap_or_default();
    dst.with_file_name(format!(".hammer-upgrade-tmp-{name}"))
}

/// Escribe `bytes` en `path` de forma atómica **y DURABLE** (temporal + fsync + rename + fsync del dir).
///
/// ⚠ Los dos `fsync` no son prolijidad: sin ellos «atómico» vale sólo frente a OTROS PROCESOS, no
/// frente a un corte. `rename` sobre un fichero cuyos datos siguen en la caché de página deja, tras
/// un corte de luz, la entrada nueva apuntando a bloques que nunca se escribieron — o sea un fichero
/// de **CERO BYTES**.
///
/// Medido el 2026-09-11 (SDD 28 §6.14): tras un `hcloud server reset` —que es un corte duro, no un
/// apagado limpio— la caja volvió con `pending.json` y `generations/2/manifest.json` en 0 bytes, el
/// upgrade perdido, y `upgrade recover` —cuyo único propósito es sobrevivir exactamente a eso—
/// abortando con `json: EOF while parsing a value`. La máquina de estados que promete resistir un
/// corte no puede depender de escrituras que no resisten un corte.
///
/// `fsync` del fichero garantiza los datos; `fsync` del DIRECTORIO garantiza que la entrada del
/// rename sobreviva. Hacen falta los dos.
fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let tmp = tmp_sibling(path);
    {
        let mut f = std::fs::File::create(&tmp)?;
        use std::io::Write;
        f.write_all(bytes)?;
        f.sync_all()?; // datos + metadatos del temporal, ANTES del rename
    }
    std::fs::rename(&tmp, path)?;
    if let Some(parent) = path.parent() {
        // La entrada de directorio que el rename creó también necesita llegar al disco.
        if let Ok(d) = std::fs::File::open(parent) {
            let _ = d.sync_all();
        }
    }
    Ok(())
}

/// Avanza/retrocede `current` de forma atómica. `None` ⇒ borra el puntero (estado pre-upgrades).
fn set_current(state_root: &Path, id: Option<u64>) -> Result<()> {
    let f = current_file(state_root);
    match id {
        Some(id) => write_atomic(&f, id.to_string().as_bytes())?,
        None => {
            let _ = std::fs::remove_file(&f);
        }
    }
    Ok(())
}

fn journal_change(journal: Option<&Journal>, tree_dir: &str, dst: &Path, op: ChangeOp) {
    let Some(j) = journal else { return };
    let mop = match op {
        ChangeOp::Added => MutationOp::Create,
        ChangeOp::Replaced => MutationOp::Replace,
        ChangeOp::Removed => MutationOp::Delete,
    };
    let ev = MutationEvent {
        ts: takana_journal::now_rfc3339(),
        op: mop,
        path: dst.to_path_buf(),
        by: Actor {
            // El upgrade es replay-able: el contenido viene del árbol `tree_dir` del store.
            source: Source::HammerHydrate { artifact: tree_dir.to_string() },
            pid: Some(std::process::id()),
            uid: None,
        },
        content_hash: None,
        note: Some("hammer upgrade".to_string()),
    };
    if let Err(e) = j.record(&ev) {
        tracing::warn!(error = %e, path = %dst.display(), "upgrade: fallo al registrar en el diario");
    }
}

fn now_secs() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Sella un árbol sintético en `store_root` y devuelve su `<hash>-<name>`.
    fn seal_tree(store_root: &Path, hex: &str, name: &str, build: impl Fn(&Path)) -> String {
        let store = Store::open(store_root).unwrap();
        let h = ArtifactHash::from_hex(hex);
        let staging = store_root.join(".bootstrap-tmp").join(h.store_dir_name(name));
        let _ = std::fs::remove_dir_all(&staging);
        std::fs::create_dir_all(&staging).unwrap();
        build(&staging);
        store.seal(&staging, &h, name).unwrap();
        h.store_dir_name(name)
    }

    /// Un `pending.json` de CERO BYTES es la huella de un corte a mitad de escritura — el caso para
    /// el que `recover` existe. Antes reventaba con `json: EOF while parsing a value`, que manda a
    /// mirar el JSON en vez del corte. Medido en la caja de producción el 2026-09-11 tras un
    /// `hcloud server reset` (corte duro): `pending.json` y el manifiesto de la generación, los dos
    /// en 0 bytes, y `upgrade recover` abortando.
    #[test]
    fn pending_vacio_se_reporta_como_corte_no_como_json_roto() {
        let tmp = tempfile::tempdir().unwrap();
        let state = tmp.path();
        std::fs::create_dir_all(state).unwrap();

        // (1) sin fichero ⇒ no hay pendiente, sin error
        assert!(matches!(pending(state), Ok(None)), "sin pending.json no hay intento pendiente");

        // (2) fichero VACÍO ⇒ error de CORTE, nombrando el árbol de respaldos
        std::fs::write(pending_path(state), b"").unwrap();
        match pending(state) {
            Err(Error::PendingCorrupt { path, backups }) => {
                assert!(path.contains("pending.json"), "el error nombra el fichero: {path}");
                assert!(!backups.is_empty(), "y dónde están los respaldos");
            }
            otro => panic!("esperaba PendingCorrupt, obtuve {otro:?}"),
        }

        // (3) sólo espacios ⇒ lo mismo (una escritura truncada puede dejar basura en blanco)
        std::fs::write(pending_path(state), b"  \n").unwrap();
        assert!(matches!(pending(state), Err(Error::PendingCorrupt { .. })));
    }

    /// El control que hace que el test de arriba valga: un `pending.json` VÁLIDO se lee igual que
    /// siempre. Si esto se rompiera, el chequeo nuevo estaría tragándose intentos reales.
    #[test]
    fn pending_valido_sigue_leyendose() {
        let tmp = tempfile::tempdir().unwrap();
        let state = tmp.path();
        let m = GenerationManifest {
            id: 7,
            tree_dir: "aaaa-x".into(),
            tree_content: "b3:aaaa".into(),
            parent: None,
            created_at: 0,
            files: vec![],
            changes: vec![],
        };
        write_pending(state, &m).unwrap();
        let leido = pending(state).unwrap().expect("un pending válido se lee");
        assert_eq!(leido.id, 7);
    }

    fn h64(prefix: &str) -> String {
        format!("{prefix}{}", "0".repeat(64 - prefix.len()))
    }

    fn write(p: &Path, bytes: &[u8]) {
        std::fs::create_dir_all(p.parent().unwrap()).unwrap();
        std::fs::write(p, bytes).unwrap();
    }

    /// EVIDENCIA (2026-08-30): ¿el proyector de `apply` escribe a través de un symlink de
    /// directorio que sale del root? Es el mismo patrón que se midió en `hydrate` — `root.join(rel)`
    /// y escribir por ruta — y con `--root /mnt/imagen` el root NO es `/`.
    #[test]
    fn apply_no_escribe_fuera_del_root_por_un_symlink() {
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");
        let fuera = tmp.path().join("fuera");
        std::fs::create_dir_all(&fuera).unwrap();

        // Generación 1: deja puesto `usr/share/pkg → <fuera>` (los symlinks se proyectan literales).
        let g1 = seal_tree(&store, &h64("a1"), "v1", |w| {
            std::fs::create_dir_all(w.join("usr/share")).unwrap();
            std::os::unix::fs::symlink(&fuera, w.join("usr/share/pkg")).unwrap();
        });
        apply(&store, &g1, None, &root, &state, None).unwrap();

        // Generación 2: trae un fichero DEBAJO de esa ruta.
        let g2 = seal_tree(&store, &h64("a2"), "v2", |w| {
            write(&w.join("usr/share/pkg/archivo"), b"contenido");
        });
        let r = apply(&store, &g2, None, &root, &state, None);

        assert!(
            !fuera.join("archivo").exists(),
            "ESCAPÓ: apply escribió fuera del root a través del symlink ({r:?})"
        );
        let err = format!("{}", r.unwrap_err());
        assert!(err.contains("se saldría del root"), "el error tiene que decir qué pasó: {err}");
    }

    /// El contrapeso del guardián anterior: un symlink de directorio que se queda DENTRO del root
    /// tiene que seguir proyectando. Es el caso de usr-merge (`/lib → usr/lib`) y el de los temas de
    /// iconos; romperlo sería peor que el agujero que se tapa.
    #[test]
    fn apply_sigue_proyectando_por_un_symlink_que_no_sale() {
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");

        // Generación 1: `usr/share/pkg → ../otro`, relativo y dentro del propio root.
        let g1 = seal_tree(&store, &h64("b1"), "v1", |w| {
            write(&w.join("usr/otro/.marca"), b"x");
            std::fs::create_dir_all(w.join("usr/share")).unwrap();
            std::os::unix::fs::symlink("../otro", w.join("usr/share/pkg")).unwrap();
        });
        apply(&store, &g1, None, &root, &state, None).unwrap();

        // Generación 2: un fichero debajo de esa ruta + lo de la 1 (si no, la 2 la retira).
        let g2 = seal_tree(&store, &h64("b2"), "v2", |w| {
            write(&w.join("usr/otro/.marca"), b"x");
            std::fs::create_dir_all(w.join("usr/share")).unwrap();
            std::os::unix::fs::symlink("../otro", w.join("usr/share/pkg")).unwrap();
            write(&w.join("usr/share/pkg/archivo"), b"contenido");
        });
        apply(&store, &g2, None, &root, &state, None).unwrap();

        assert!(
            root.join("usr/otro/archivo").exists(),
            "el fichero tiene que aterrizar donde apunta el symlink, dentro del root"
        );
    }

    #[test]
    fn apply_projects_tree_and_records_generation() {
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");
        std::fs::create_dir_all(&root).unwrap();
        // root ya tiene un /usr/bin/ls "viejo" que el upgrade pisará.
        write(&root.join("usr/bin/ls"), b"OLD-ls");

        let dir = seal_tree(&store, &h64("aa"), "v1", |w| {
            write(&w.join("usr/bin/ls"), b"NEW-ls");
            write(&w.join("usr/bin/cat"), b"NEW-cat");
            std::fs::create_dir_all(w.join("bin")).unwrap();
            std::os::unix::fs::symlink("busybox", w.join("bin/sh")).unwrap();
        });

        let r = apply(&store, &dir, None, &root, &state, None).unwrap();
        assert_eq!(r.generation, 1);
        assert!(!r.already_current);
        assert_eq!(r.replaced, vec![PathBuf::from("usr/bin/ls")]);
        assert!(r.added.contains(&PathBuf::from("usr/bin/cat")));
        assert!(r.added.contains(&PathBuf::from("bin/sh")));

        // los bytes nuevos están en el root vivo.
        assert_eq!(std::fs::read(root.join("usr/bin/ls")).unwrap(), b"NEW-ls");
        assert_eq!(std::fs::read(root.join("usr/bin/cat")).unwrap(), b"NEW-cat");
        assert_eq!(std::fs::read_link(root.join("bin/sh")).unwrap(), PathBuf::from("busybox"));

        assert_eq!(current(&state).unwrap(), Some(1));
        assert_eq!(list(&state).unwrap().len(), 1);
    }

    #[test]
    fn apply_is_idempotent_for_same_tree() {
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");
        let dir = seal_tree(&store, &h64("bb"), "v1", |w| write(&w.join("a"), b"A"));
        let r1 = apply(&store, &dir, None, &root, &state, None).unwrap();
        assert!(!r1.already_current && r1.generation == 1);
        let r2 = apply(&store, &dir, None, &root, &state, None).unwrap();
        assert!(r2.already_current, "mismo árbol ⇒ no-op");
        assert_eq!(list(&state).unwrap().len(), 1, "no se creó generación nueva");
    }

    #[test]
    fn rollback_restores_previous_tree_exactly() {
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");

        let v1 = seal_tree(&store, &h64("c1"), "v1", |w| {
            write(&w.join("usr/bin/ls"), b"LS-v1");
            write(&w.join("usr/bin/old"), b"OLD-only-in-v1");
        });
        let v2 = seal_tree(&store, &h64("c2"), "v2", |w| {
            write(&w.join("usr/bin/ls"), b"LS-v2");
            write(&w.join("usr/bin/new"), b"NEW-only-in-v2");
            // `old` desaparece ⇒ apply de v2 debe retirarlo.
        });

        apply(&store, &v1, None, &root, &state, None).unwrap();
        let r2 = apply(&store, &v2, None, &root, &state, None).unwrap();
        assert_eq!(r2.generation, 2);
        assert_eq!(r2.removed, vec![PathBuf::from("usr/bin/old")]);
        // estado tras v2.
        assert_eq!(std::fs::read(root.join("usr/bin/ls")).unwrap(), b"LS-v2");
        assert!(root.join("usr/bin/new").exists());
        assert!(!root.join("usr/bin/old").exists());

        // rollback ⇒ vuelve EXACTAMENTE a v1.
        let rb = rollback(&root, &state, None).unwrap();
        assert_eq!(rb.reverted, 2);
        assert_eq!(rb.now_current, Some(1));
        assert_eq!(current(&state).unwrap(), Some(1));
        assert_eq!(std::fs::read(root.join("usr/bin/ls")).unwrap(), b"LS-v1", "ls revertido");
        assert!(!root.join("usr/bin/new").exists(), "lo añadido por v2 se borró");
        assert_eq!(std::fs::read(root.join("usr/bin/old")).unwrap(), b"OLD-only-in-v1", "lo retirado por v2 se restauró");
    }

    #[test]
    fn rollback_to_pre_upgrade_state() {
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");
        std::fs::create_dir_all(&root).unwrap();
        write(&root.join("etc/motd"), b"PRISTINE");

        let v1 = seal_tree(&store, &h64("dd"), "v1", |w| {
            write(&w.join("etc/motd"), b"UPGRADED");
            write(&w.join("usr/bin/brand-new"), b"X");
        });
        apply(&store, &v1, None, &root, &state, None).unwrap();
        assert_eq!(std::fs::read(root.join("etc/motd")).unwrap(), b"UPGRADED");

        let rb = rollback(&root, &state, None).unwrap();
        assert_eq!(rb.now_current, None, "se volvió al estado pre-upgrades");
        assert_eq!(current(&state).unwrap(), None);
        assert_eq!(std::fs::read(root.join("etc/motd")).unwrap(), b"PRISTINE", "config original restaurada");
        assert!(!root.join("usr/bin/brand-new").exists());
    }

    #[test]
    fn rollback_without_generation_errors() {
        let tmp = tempfile::tempdir().unwrap();
        let err = rollback(&tmp.path().join("root"), &tmp.path().join("state"), None).unwrap_err();
        assert!(matches!(err, Error::NothingToRollback), "vino {err}");
    }

    #[test]
    fn apply_rejects_tree_that_does_not_match_expected() {
        // Si el llamador anuncia un of_tree esperado (de un índice de mirror / release firmada) y el
        // árbol del store no casa, apply rechaza ANTES de tocar el root.
        let tmp = tempfile::tempdir().unwrap();
        let store_root = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");
        let dir = seal_tree(&store_root, &h64("ee"), "v1", |w| write(&w.join("a"), b"A"));

        // expected MENTIROSO ⇒ Verify, y nada se aplicó.
        let err = apply(&store_root, &dir, Some("b3:deadbeef"), &root, &state, None).unwrap_err();
        assert!(matches!(err, Error::Verify { .. }), "vino {err}");
        assert_eq!(current(&state).unwrap(), None, "no se creó generación");
        assert!(!root.join("a").exists(), "el root quedó intacto");

        // expected CORRECTO (el of_tree real) ⇒ aplica.
        let want = ArtifactHash::of_tree(&store_root.join(&dir)).unwrap().as_str().to_string();
        let r = apply(&store_root, &dir, Some(&want), &root, &state, None).unwrap();
        assert_eq!(r.generation, 1);
    }

    #[test]
    fn journal_records_each_change() {
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");
        let j = Journal::open(tmp.path().join("journal")).unwrap();
        let dir = seal_tree(&store, &h64("ff"), "v1", |w| {
            write(&w.join("usr/bin/a"), b"A");
            write(&w.join("usr/bin/b"), b"B");
        });
        apply(&store, &dir, None, &root, &state, Some(&j)).unwrap();
        let evs = j.read_all().unwrap();
        assert_eq!(evs.len(), 2);
        assert!(evs.iter().all(|e| matches!(&e.by.source, Source::HammerHydrate { artifact } if artifact == &dir)));
    }

    /// Simula un apply INTERRUMPIDO a media proyección: escribe el `pending.json` con el plan y deja el
    /// FHS a medias (proyecta sólo el primer cambio), sin commitear. Devuelve el manifiesto del plan.
    fn interrupt_apply(
        store: &Path,
        tree_dir: &str,
        target_root: &Path,
        state_root: &Path,
        partial: usize,
    ) -> GenerationManifest {
        let store_h = Store::open(store).unwrap();
        let tree_path = store_h.root().join(tree_dir);
        let tree_content = ArtifactHash::of_tree(&tree_path).unwrap().as_str().to_string();
        let entries = enumerate_tree(&tree_path).unwrap();
        let mut changes = Vec::new();
        for e in &entries {
            let existed = std::fs::symlink_metadata(target_root.join(&e.rel)).is_ok();
            changes.push(Change { path: e.rel.clone(), op: if existed { ChangeOp::Replaced } else { ChangeOp::Added } });
        }
        let id = next_id(state_root).unwrap();
        let manifest = GenerationManifest {
            id, tree_dir: tree_dir.to_string(), tree_content,
            parent: current(state_root).unwrap(), created_at: 0,
            files: entries.iter().map(|e| e.rel.clone()).collect(), changes,
        };
        let backup_root = generations_dir(state_root).join(id.to_string()).join("backup");
        std::fs::create_dir_all(&backup_root).unwrap();
        write_pending(state_root, &manifest).unwrap();
        // proyectar SÓLO `partial` entradas (corte a media proyección).
        for e in entries.iter().take(partial) {
            let dst = target_root.join(&e.rel);
            if std::fs::symlink_metadata(&dst).is_ok() {
                backup_existing_once(&dst, &backup_root.join(&e.rel)).unwrap();
            }
            project_entry(&tree_path, e, &dst).unwrap();
        }
        manifest
    }

    #[test]
    fn apply_refuses_when_a_pending_apply_exists() {
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");
        let v1 = seal_tree(&store, &h64("e1"), "v1", |w| { write(&w.join("a"), b"A"); write(&w.join("b"), b"B"); });
        interrupt_apply(&store, &v1, &root, &state, 1);
        // con un intento pendiente, apply se NIEGA (exige recover).
        let err = apply(&store, &v1, None, &root, &state, None).unwrap_err();
        assert!(matches!(err, Error::PendingExists { .. }), "vino {err}");
        assert!(pending(&state).unwrap().is_some());
    }

    #[test]
    fn recover_roll_forward_completes_interrupted_apply() {
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");
        std::fs::create_dir_all(&root).unwrap();
        write(&root.join("etc/keep"), b"PRE");
        let v1 = seal_tree(&store, &h64("e2"), "v1", |w| {
            write(&w.join("usr/bin/ls"), b"NEW-ls");
            write(&w.join("usr/bin/cat"), b"NEW-cat");
            write(&w.join("usr/bin/grep"), b"NEW-grep");
        });
        // corte tras proyectar 1 de 3 ⇒ FHS a medias, sin generación viva.
        interrupt_apply(&store, &v1, &root, &state, 1);
        assert_eq!(current(&state).unwrap(), None, "el apply nunca commiteó");
        assert!(pending(&state).unwrap().is_some());

        let r = recover(&store, &root, &state, None, false).unwrap();
        assert!(r.recovered && r.rolled_forward && r.generation == 1);
        // el apply quedó COMPLETO: los 3 binarios, generación viva, intento limpio.
        assert_eq!(std::fs::read(root.join("usr/bin/ls")).unwrap(), b"NEW-ls");
        assert_eq!(std::fs::read(root.join("usr/bin/cat")).unwrap(), b"NEW-cat");
        assert_eq!(std::fs::read(root.join("usr/bin/grep")).unwrap(), b"NEW-grep");
        assert_eq!(current(&state).unwrap(), Some(1));
        assert!(pending(&state).unwrap().is_none());
        // y como quedó committeado, un rollback posterior restaura el estado pre-apply EXACTO.
        rollback(&root, &state, None).unwrap();
        assert!(!root.join("usr/bin/ls").exists(), "lo añadido se revierte");
        assert_eq!(std::fs::read(root.join("etc/keep")).unwrap(), b"PRE");
    }

    #[test]
    fn recover_roll_back_undoes_interrupted_apply() {
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");
        std::fs::create_dir_all(&root).unwrap();
        write(&root.join("usr/bin/ls"), b"ORIG-ls");   // existirá ⇒ Replaced (con backup)
        let v1 = seal_tree(&store, &h64("e3"), "v1", |w| {
            write(&w.join("usr/bin/ls"), b"NEW-ls");
            write(&w.join("usr/bin/new"), b"NEW-only"); // Added
        });
        // corte tras proyectar el 1er cambio (orden alfabético: usr/bin/ls Replaced primero).
        interrupt_apply(&store, &v1, &root, &state, 1);
        assert_eq!(std::fs::read(root.join("usr/bin/ls")).unwrap(), b"NEW-ls", "ls ya pisado a medias");

        let r = recover(&store, &root, &state, None, true).unwrap();
        assert!(r.recovered && !r.rolled_forward);
        // el FHS volvió al estado PRE-apply: ls original restaurado, nada de la generación a medias.
        assert_eq!(std::fs::read(root.join("usr/bin/ls")).unwrap(), b"ORIG-ls", "ls restaurado del backup");
        assert!(!root.join("usr/bin/new").exists());
        assert_eq!(current(&state).unwrap(), None);
        assert!(pending(&state).unwrap().is_none());
        assert!(!generations_dir(&state).join("1").exists(), "la generación a medias se borró");
    }

    #[test]
    fn recover_is_noop_without_pending() {
        let tmp = tempfile::tempdir().unwrap();
        let r = recover(&tmp.path().join("store"), &tmp.path().join("root"), &tmp.path().join("state"), None, false).unwrap();
        assert!(!r.recovered);
    }

    #[test]
    fn apply_clears_pending_on_success() {
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");
        let v1 = seal_tree(&store, &h64("e4"), "v1", |w| write(&w.join("a"), b"A"));
        apply(&store, &v1, None, &root, &state, None).unwrap();
        assert!(pending(&state).unwrap().is_none(), "el apply exitoso limpia el intento");
    }

    #[test]
    fn prune_removes_orphans_but_keeps_live_chain() {
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");
        let v1 = seal_tree(&store, &h64("a1"), "v1", |w| write(&w.join("f"), b"1"));
        let v2 = seal_tree(&store, &h64("a2"), "v2", |w| write(&w.join("f"), b"2"));
        let v3 = seal_tree(&store, &h64("a3"), "v3", |w| write(&w.join("f"), b"3"));

        apply(&store, &v1, None, &root, &state, None).unwrap(); // gen 1
        apply(&store, &v2, None, &root, &state, None).unwrap(); // gen 2 (parent 1)
        apply(&store, &v3, None, &root, &state, None).unwrap(); // gen 3 (parent 2)
        // rollback dos veces ⇒ current=1; las gens 2 y 3 quedan HUÉRFANAS (rollback no las alcanza).
        rollback(&root, &state, None).unwrap();
        rollback(&root, &state, None).unwrap();
        assert_eq!(current(&state).unwrap(), Some(1));
        assert_eq!(live_chain(&state).unwrap(), vec![1]);

        let r = prune(&state, None).unwrap();
        assert_eq!(r.removed, vec![2, 3], "huérfanas borradas");
        assert_eq!(r.kept, vec![1], "cadena viva conservada");
        assert_eq!(list(&state).unwrap().iter().map(|m| m.id).collect::<Vec<_>>(), vec![1]);
        // idempotente.
        assert!(prune(&state, None).unwrap().removed.is_empty());
    }

    #[test]
    fn boot_graph_builds_dag_and_activate_rolls_back_to_ancestor() {
        use crate::boot_graph::{activate, build, recovery_id, NodeKind};

        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");
        let v1 = seal_tree(&store, &h64("c1"), "v1", |w| write(&w.join("f"), b"1"));
        let v2 = seal_tree(&store, &h64("c2"), "v2", |w| write(&w.join("f"), b"2"));
        let v3 = seal_tree(&store, &h64("c3"), "v3", |w| write(&w.join("f"), b"3"));
        apply(&store, &v1, None, &root, &state, None).unwrap(); // gen 1 (Base)
        apply(&store, &v2, None, &root, &state, None).unwrap(); // gen 2
        apply(&store, &v3, None, &root, &state, None).unwrap(); // gen 3 (viva)

        let g = build(&state).unwrap();
        assert_eq!(g.version, 1);
        // 3 generaciones + el nodo de recuperación sintético.
        assert_eq!(g.nodes.len(), 4);
        let gen1 = &g.nodes[0];
        assert_eq!(gen1.kind, NodeKind::Base, "la primera generación es la raíz del linaje");
        assert!(gen1.parents.is_empty());
        let gen2 = &g.nodes[1];
        assert_eq!(gen2.kind, NodeKind::Generation);
        assert_eq!(gen2.parents, vec![gen1.id.clone()], "el DAG cuelga del of_tree del padre");
        // current/default = la generación viva (gen 3).
        let gen3 = &g.nodes[2];
        assert_eq!(g.current, gen3.id);
        assert_eq!(g.default, g.current);
        // el nodo de recuperación cuelga de la viva.
        let rec = g.nodes.iter().find(|n| n.kind == NodeKind::Recovery).unwrap();
        assert_eq!(rec.id, recovery_id());
        assert_eq!(rec.parents, vec![gen3.id.clone()]);

        // activar la viva ⇒ no-op.
        let r = activate(&root, &state, None, &gen3.id).unwrap();
        assert!(r.already_current);
        assert_eq!(current(&state).unwrap(), Some(3));

        // activar la gen 1 (ancestro) ⇒ rollback paso a paso hasta ella.
        let r = activate(&root, &state, None, &gen1.id).unwrap();
        assert!(!r.already_current);
        assert_eq!(r.rolled_back, vec![3, 2], "revierte de la más nueva a la más vieja");
        assert_eq!(current(&state).unwrap(), Some(1));
        assert_eq!(std::fs::read(root.join("f")).unwrap(), b"1", "el FHS vivo es el de la gen 1");

        // ahora las gens 2 y 3 son huérfanas (redo no soportado) ⇒ error honesto.
        let g2 = build(&state).unwrap();
        assert_eq!(g2.current, gen1.id);
        let err = activate(&root, &state, None, &gen3.id).unwrap_err();
        assert!(err.to_string().contains("redo"), "activar hacia adelante debe fallar claro: {err}");

        // recuperación desde la gen 1 ⇒ rollback al estado pre-upgrades (sin current).
        let r = activate(&root, &state, None, &recovery_id()).unwrap();
        assert_eq!(r.kind, NodeKind::Recovery);
        assert_eq!(r.rolled_back, vec![1]);
        assert_eq!(current(&state).unwrap(), None);
    }

    #[test]
    fn prune_with_keep_trims_chain_depth() {
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("store");
        let root = tmp.path().join("root");
        let state = tmp.path().join("state");
        for (hx, nm, by) in [("b1", "v1", b"1"), ("b2", "v2", b"2"), ("b3", "v3", b"3")] {
            let d = seal_tree(&store, &h64(hx), nm, |w| write(&w.join("f"), by));
            apply(&store, &d, None, &root, &state, None).unwrap();
        }
        // current=3, cadena=[3,2,1]. keep=Some(2) ⇒ conserva 3 y 2, borra 1.
        assert_eq!(live_chain(&state).unwrap(), vec![3, 2, 1]);
        let r = prune(&state, Some(2)).unwrap();
        assert_eq!(r.kept, vec![2, 3]);
        assert_eq!(r.removed, vec![1]);
        // gen 1 borrada del disco; 2 y 3 siguen.
        assert_eq!(list(&state).unwrap().iter().map(|m| m.id).collect::<Vec<_>>(), vec![2, 3]);
    }
}