magi-code 0.77.1

use super::{
    capture::{
        LedgerRead, SnapshotBlobRef, SnapshotCaptureStatus, SnapshotLedgerRecord, SnapshotTool,
        secret_like_path,
    },
    sanitized_diagnostic,
    store::{CheckpointStore, sha256_hex},
};
use crate::output::redact_sensitive_text;
use crate::path_utils::lexical_normalize;
use cap_fs_ext::{DirExt, FollowSymlinks, OpenOptionsFollowExt};
#[cfg(unix)]
use cap_std::fs::{OpenOptionsExt as CapOpenOptionsExt, PermissionsExt as CapPermissionsExt};
use cap_std::{
    ambient_authority,
    fs::{Dir as CapDir, File as CapFile, OpenOptions as CapOpenOptions},
};
#[cfg(any(target_os = "linux", target_os = "macos"))]
fn rename_noreplace(parent: &CapDir, source: &OsStr, target: &OsStr) -> io::Result<()> {
    use std::{
        ffi::CString,
        os::{fd::AsRawFd, unix::ffi::OsStrExt},
    };

    let source_c = CString::new(source.as_bytes())
        .map_err(|_| io::Error::new(ErrorKind::InvalidInput, "rename source contains NUL"))?;
    let target_c = CString::new(target.as_bytes())
        .map_err(|_| io::Error::new(ErrorKind::InvalidInput, "rename target contains NUL"))?;
    #[cfg(target_os = "linux")]
    {
        // SAFETY: `parent` is a live held capability directory. The two
        // `CString`s contain NUL-terminated relative names and remain alive
        // for the duration of this syscall; the raw fd is only borrowed.
        let result = unsafe {
            libc::renameat2(
                parent.as_raw_fd(),
                source_c.as_ptr(),
                parent.as_raw_fd(),
                target_c.as_ptr(),
                libc::RENAME_NOREPLACE,
            )
        };
        if result == 0 {
            Ok(())
        } else {
            Err(io::Error::last_os_error())
        }
    }
    #[cfg(target_os = "macos")]
    {
        // SAFETY: `parent` is a live held capability directory. The two
        // `CString`s contain NUL-terminated relative names and remain alive
        // for the duration of this syscall; the raw fd is only borrowed.
        let result = unsafe {
            libc::renameatx_np(
                parent.as_raw_fd(),
                source_c.as_ptr(),
                parent.as_raw_fd(),
                target_c.as_ptr(),
                libc::RENAME_EXCL,
            )
        };
        if result == 0 {
            Ok(())
        } else {
            Err(io::Error::last_os_error())
        }
    }
}

#[cfg(not(any(target_os = "linux", target_os = "macos")))]
fn rename_noreplace(parent: &CapDir, source: &OsStr, target: &OsStr) -> io::Result<()> {
    let _ = (parent, source, target);
    Err(io::Error::new(
        ErrorKind::Unsupported,
        "atomic no-replace rename is unsupported on this platform",
    ))
}
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::{
    collections::BTreeMap,
    ffi::{OsStr, OsString},
    io::{self, ErrorKind, Read, Write},
    path::{Component, Path, PathBuf},
    sync::atomic::{AtomicU64, Ordering},
    time::{SystemTime, UNIX_EPOCH},
};

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ChangesReadModel {
    pub(crate) turns: Vec<ChangeTurn>,
    pub(crate) diagnostics: Vec<String>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ChangeTurn {
    pub(crate) user_turn: u64,
    pub(crate) changes: Vec<ChangeEntry>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ChangeEntry {
    pub(crate) tool: SnapshotTool,
    pub(crate) relative_path: PathBuf,
    pub(crate) classification: ChangeClassification,
    pub(crate) rewindable: bool,
    pub(crate) unavailable_reason: Option<String>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ChangeClassification {
    Created,
    Modified,
    Deleted,
    NonRewindable,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct RestorePlan {
    pub(crate) session_id: String,
    pub(crate) target_turn: u64,
    pub(crate) latest_turn: Option<u64>,
    pub(crate) operations: Vec<RestoreOperation>,
    pub(crate) diagnostics: Vec<String>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct RestoreOperation {
    pub(crate) relative_path: PathBuf,
    pub(crate) kind: RestoreOperationKind,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum RestoreOperationKind {
    Restore {
        pre: SnapshotBlobRef,
        expected_current: SnapshotBlobRef,
    },
    DeleteCreated {
        expected_current: SnapshotBlobRef,
    },
    ResurrectDeleted {
        pre: SnapshotBlobRef,
    },
    SkipConflict {
        reason: String,
    },
    SkipUnavailable {
        reason: String,
    },
}

impl RestoreOperationKind {
    pub(crate) fn label(&self) -> &'static str {
        match self {
            Self::Restore { .. } => "restore",
            Self::DeleteCreated { .. } => "delete-created",
            Self::ResurrectDeleted { .. } => "resurrect-deleted",
            Self::SkipConflict { .. } => "skip-conflict",
            Self::SkipUnavailable { .. } => "skip-unavailable",
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct RewindExecution {
    pub(crate) results: Vec<RestoreOperationResult>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct RestoreOperationResult {
    pub(crate) relative_path: PathBuf,
    pub(crate) status: RestoreStatus,
    pub(crate) reason: Option<String>,
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
#[serde(rename_all = "snake_case")]
pub(crate) enum RestoreStatus {
    Restored,
    DeletedCreated,
    ResurrectedDeleted,
    SkipConflict,
    SkipUnavailable,
}

const REDACTED_PATH: &str = "[redacted]";

impl CheckpointStore {
    pub(crate) fn changes(&self, session_id: &str) -> ChangesReadModel {
        changes_from_records(self, self.read_records(session_id))
    }

    pub(crate) fn plan_rewind(
        &self,
        session_id: &str,
        cwd: &Path,
        target_turn: u64,
    ) -> RestorePlan {
        plan_rewind(self, session_id, cwd, target_turn)
    }

    pub(crate) fn execute_plan(&self, plan: &RestorePlan, cwd: &Path) -> RewindExecution {
        execute_plan(self, plan, cwd)
    }

    pub(crate) fn changes_from_read(&self, read: &LedgerRead) -> ChangesReadModel {
        changes_from_records(self, read.clone())
    }

    pub(crate) fn plan_rewind_from_read(
        &self,
        session_id: &str,
        cwd: &Path,
        target_turn: u64,
        read: &LedgerRead,
    ) -> RestorePlan {
        plan_rewind_from_records(self, session_id, cwd, target_turn, read)
    }
}

fn secret_like_path_case_insensitive(path: &Path) -> bool {
    path.file_name()
        .and_then(OsStr::to_str)
        .is_some_and(|name| {
            name.eq_ignore_ascii_case("auth.json")
                || name.eq_ignore_ascii_case("id_rsa")
                || name.eq_ignore_ascii_case("id_ed25519")
                || name
                    .get(..4)
                    .is_some_and(|prefix| prefix.eq_ignore_ascii_case(".env"))
                || name
                    .get(name.len().saturating_sub(4)..)
                    .is_some_and(|suffix| suffix.eq_ignore_ascii_case(".pem"))
                || name
                    .get(name.len().saturating_sub(4)..)
                    .is_some_and(|suffix| suffix.eq_ignore_ascii_case(".key"))
        })
}

fn secret_path_for_rewind(path: &Path) -> bool {
    secret_like_path(path) || secret_like_path_case_insensitive(path)
}

fn ascii_case_insensitive_component_eq(left: &OsStr, right: &OsStr) -> bool {
    match (left.to_str(), right.to_str()) {
        (Some(left), Some(right)) => left.eq_ignore_ascii_case(right),
        // Comparison-only fallback: never turn an arbitrary OS component into
        // a lossy path used for filesystem access.
        _ => left == right,
    }
}

fn path_starts_with_ascii_case_insensitive(path: &Path, prefix: &Path) -> bool {
    let mut path_components = path.components();
    for prefix_component in prefix.components() {
        let Some(path_component) = path_components.next() else {
            return false;
        };
        if !ascii_case_insensitive_component_eq(
            path_component.as_os_str(),
            prefix_component.as_os_str(),
        ) {
            return false;
        }
    }
    true
}
fn redacted_relative_path(reason: &SnapshotCaptureStatus, relative_path: PathBuf) -> PathBuf {
    if matches!(
        reason,
        SnapshotCaptureStatus::Excluded {
            reason: super::SnapshotExclusionReason::SecretPath
        }
    ) || secret_path_for_rewind(&relative_path)
    {
        PathBuf::from(REDACTED_PATH)
    } else {
        relative_path
    }
}

pub(crate) fn redacted_path_for_display(path: &Path) -> String {
    safe_provider_path(&path.to_string_lossy())
}

fn safe_rewind_target_path(
    store: &CheckpointStore,
    cwd: &Path,
    relative_path: &Path,
) -> Result<(), &'static str> {
    if relative_path.is_absolute() || relative_path.as_os_str().is_empty() {
        return Err("denied path");
    }
    let mut has_normal_component = false;
    for component in relative_path.components() {
        match component {
            Component::Normal(name) => {
                has_normal_component = true;
                if name
                    .to_str()
                    .is_some_and(|name| name.eq_ignore_ascii_case(".git"))
                {
                    return Err("denied path");
                }
            }
            Component::CurDir => {}
            Component::ParentDir | Component::Prefix(_) | Component::RootDir => {
                return Err("denied path");
            }
        }
    }
    if !has_normal_component || secret_path_for_rewind(relative_path) {
        return Err("denied path");
    }

    let root = lexical_normalize(cwd);
    let path = lexical_normalize(&root.join(relative_path));
    if !path_starts_with_ascii_case_insensitive(&path, &root) || path == root {
        return Err("denied path");
    }

    let checkpoint_root = store
        .root
        .canonicalize()
        .unwrap_or_else(|_| lexical_normalize(&store.root));
    if path_starts_with_ascii_case_insensitive(&path, &checkpoint_root) {
        return Err("denied path");
    }
    if let Some(mc_root) = checkpoint_root.parent() {
        let session_root = lexical_normalize(&mc_root.join("sessions"));
        if path_starts_with_ascii_case_insensitive(&path, &session_root) {
            return Err("denied path");
        }
    }

    Ok(())
}

fn open_rewind_root(cwd: &Path) -> (PathBuf, Option<CapDir>) {
    let canonical_cwd = cwd
        .canonicalize()
        .unwrap_or_else(|_| lexical_normalize(cwd));
    let root = CapDir::open_ambient_dir(&canonical_cwd, ambient_authority()).ok();
    (canonical_cwd, root)
}

#[derive(Debug)]
struct RewindTarget {
    parent: CapDir,
    name: OsString,
}

#[derive(Debug)]
struct TargetSnapshot {
    hash: String,
    #[cfg(unix)]
    mode: u32,
}

fn open_target_parent(
    root: &CapDir,
    relative_path: &Path,
    create_missing: bool,
    inject_ancestor_sync_failure: bool,
) -> io::Result<RewindTarget> {
    let mut components = Vec::new();
    for component in relative_path.components() {
        match component {
            Component::Normal(name) => components.push(name),
            Component::CurDir => {}
            Component::ParentDir | Component::Prefix(_) | Component::RootDir => {
                return Err(io::Error::new(
                    ErrorKind::InvalidInput,
                    "unsafe rewind target path",
                ));
            }
        }
    }
    let name = components
        .pop()
        .ok_or_else(|| io::Error::new(ErrorKind::InvalidInput, "rewind target has no name"))?;
    let mut parent = root.try_clone()?;
    for component in components {
        parent = open_parent_component(
            &parent,
            component,
            create_missing,
            inject_ancestor_sync_failure,
        )?;
    }
    Ok(RewindTarget {
        parent,
        name: name.to_os_string(),
    })
}

fn open_parent_component(
    parent: &CapDir,
    name: &OsStr,
    create_missing: bool,
    inject_ancestor_sync_failure: bool,
) -> io::Result<CapDir> {
    let component = Path::new(name);
    match parent.symlink_metadata(component) {
        Ok(metadata) => {
            if metadata.is_symlink() || !metadata.is_dir() {
                return Err(io::Error::new(
                    ErrorKind::InvalidInput,
                    "rewind target parent is not a safe directory",
                ));
            }
            parent.open_dir_nofollow(component)
        }
        Err(error) if error.kind() == ErrorKind::NotFound && create_missing => {
            let created = match parent.create_dir(component) {
                Ok(()) => true,
                Err(error) if error.kind() == ErrorKind::AlreadyExists => false,
                Err(error) => return Err(error),
            };
            if created {
                let sync_result = if inject_ancestor_sync_failure {
                    Err(io::Error::other("injected ancestor directory sync failure"))
                } else {
                    sync_capability_directory(parent)
                };
                if let Err(error) = sync_result {
                    // This directory is still empty unless an external actor raced us.
                    // `remove_dir` is deliberately non-recursive; never remove someone
                    // else's contents during rollback.
                    let _ = parent.remove_dir(component);
                    return Err(error);
                }
            }
            parent.open_dir_nofollow(component)
        }
        Err(error) => Err(error),
    }
}

fn target_snapshot(target: &RewindTarget) -> io::Result<Option<TargetSnapshot>> {
    snapshot_named(&target.parent, &target.name)
}

fn snapshot_named(parent: &CapDir, name: &OsStr) -> io::Result<Option<TargetSnapshot>> {
    let metadata = match parent.symlink_metadata(name) {
        Ok(metadata) => metadata,
        Err(error) if error.kind() == ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(error),
    };
    if metadata.is_symlink() {
        return Err(io::Error::new(
            ErrorKind::InvalidInput,
            "rewind target is a symlink",
        ));
    }
    if !metadata.is_file() {
        return Err(io::Error::new(
            ErrorKind::InvalidInput,
            "rewind target is not a regular file",
        ));
    }

    let mut options = CapOpenOptions::new();
    options.read(true).follow(FollowSymlinks::No);
    let mut file = match parent.open_with(name, &options) {
        Ok(file) => file,
        Err(error) if error.kind() == ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(error),
    };
    let metadata = file.metadata()?;
    if metadata.is_symlink() || !metadata.is_file() {
        return Err(io::Error::new(
            ErrorKind::InvalidInput,
            "rewind target is not a regular file",
        ));
    }
    let mut bytes = Vec::new();
    file.read_to_end(&mut bytes)?;
    #[cfg(unix)]
    let mode = metadata.permissions().mode() & 0o777;
    Ok(Some(TargetSnapshot {
        hash: sha256_hex(&bytes),
        #[cfg(unix)]
        mode,
    }))
}

fn target_snapshot_for_plan(
    root: &CapDir,
    relative_path: &Path,
    missing_parent_is_missing: bool,
) -> anyhow::Result<Option<TargetSnapshot>> {
    let target = match open_target_parent(root, relative_path, false, false) {
        Ok(target) => target,
        Err(error) if missing_parent_is_missing && error.kind() == ErrorKind::NotFound => {
            return Ok(None);
        }
        Err(error) => return Err(error.into()),
    };
    Ok(target_snapshot(&target)?)
}

fn target_mode(snapshot: &TargetSnapshot) -> Option<u32> {
    #[cfg(unix)]
    {
        Some(snapshot.mode)
    }
    #[cfg(not(unix))]
    {
        let _ = snapshot;
        None
    }
}

fn reject_final_symlink(parent: &CapDir, name: &OsStr) -> io::Result<()> {
    match parent.symlink_metadata(Path::new(name)) {
        Ok(metadata) if metadata.is_symlink() => Err(io::Error::new(
            ErrorKind::InvalidInput,
            "rewind target is a symlink",
        )),
        Ok(_) => Ok(()),
        Err(error) if error.kind() == ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error),
    }
}

fn sync_capability_directory(directory: &CapDir) -> io::Result<()> {
    #[cfg(unix)]
    {
        directory.try_clone()?.into_std_file().sync_all()
    }
    #[cfg(not(unix))]
    {
        let _ = directory;
        Ok(())
    }
}

const COMMITTED_UNDURABLE_WARNING: &str = "mutation committed but durability is uncertain";
const COMMITTED_CLEANUP_WARNING: &str = "mutation committed but recovery cleanup is incomplete";
const TARGET_CHANGED_DURING_REWIND: &str = "target changed during rewind";

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum MutationOutcome {
    Committed { warning: Option<&'static str> },
    Conflict(&'static str),
    Failed(&'static str),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ValidatedRemoval {
    Removed,
    Missing,
    Preserved,
}

static REWIND_TEMP_COUNTER: AtomicU64 = AtomicU64::new(0);

fn move_entry_to_quarantine(parent: &CapDir, source: &OsStr, kind: &str) -> io::Result<OsString> {
    move_entry_to_quarantine_with_attempts(parent, source, kind, 128, |_| {})
}

fn move_entry_to_quarantine_with_attempts(
    parent: &CapDir,
    source: &OsStr,
    kind: &str,
    attempt_limit: u64,
    mut before_move: impl FnMut(&OsStr),
) -> io::Result<OsString> {
    let stamp = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();
    let pid = std::process::id();
    for attempt in 0..attempt_limit {
        let sequence = REWIND_TEMP_COUNTER.fetch_add(1, Ordering::Relaxed);
        let name = OsString::from(format!(
            ".magi-rewind-{kind}-{pid}-{stamp}-{sequence}-{attempt}"
        ));
        before_move(name.as_os_str());
        // The native operation checks the destination and moves the source
        // atomically. Never replace a raced destination or use check-then-rename.
        match rename_noreplace(parent, source, name.as_os_str()) {
            Ok(()) => return Ok(name),
            // A generated name can still be occupied. Try another name, but
            // leave the occupied entry untouched.
            Err(error) if error.kind() == ErrorKind::AlreadyExists => {}
            Err(error) => return Err(error),
        }
    }
    Err(io::Error::new(
        ErrorKind::AlreadyExists,
        "could not allocate a rewind quarantine name",
    ))
}

fn move_target_to_quarantine(target: &RewindTarget) -> io::Result<OsString> {
    reject_final_symlink(&target.parent, &target.name)?;
    move_entry_to_quarantine(&target.parent, &target.name, "quarantine")
}

fn remove_validated_entry(
    parent: &CapDir,
    source: &OsStr,
    expected_hash: &str,
) -> ValidatedRemoval {
    let recovery_name = match move_entry_to_quarantine(parent, source, "cleanup") {
        Ok(name) => name,
        Err(error) if error.kind() == ErrorKind::NotFound => return ValidatedRemoval::Missing,
        Err(_) => return ValidatedRemoval::Preserved,
    };
    let snapshot = match snapshot_named(parent, &recovery_name) {
        Ok(Some(snapshot)) => snapshot,
        Ok(None) => return ValidatedRemoval::Missing,
        Err(_) => return ValidatedRemoval::Preserved,
    };
    if snapshot.hash != expected_hash {
        return ValidatedRemoval::Preserved;
    }
    match parent.remove_file_or_symlink(Path::new(recovery_name.as_os_str())) {
        Ok(()) => ValidatedRemoval::Removed,
        Err(_) => ValidatedRemoval::Preserved,
    }
}

fn recover_quarantine(
    parent: &CapDir,
    quarantine_name: &OsStr,
    target_name: &OsStr,
    quarantine_hash: &str,
) {
    let Ok(Some(snapshot)) = snapshot_named(parent, quarantine_name) else {
        return;
    };
    if snapshot.hash != quarantine_hash {
        return;
    }
    // The native operation checks the destination and rolls the quarantine
    // entry back atomically. On conflict or unsupported platforms, retain both
    // entries rather than replacing or cleaning either one.
    let _ = rename_noreplace(parent, quarantine_name, target_name);
}

fn finish_committed(
    parent: &CapDir,
    inject_final_sync_failure: bool,
    cleanup_incomplete: bool,
) -> MutationOutcome {
    let final_sync_failed = inject_final_sync_failure || sync_capability_directory(parent).is_err();
    let warning = if final_sync_failed {
        Some(COMMITTED_UNDURABLE_WARNING)
    } else if cleanup_incomplete {
        Some(COMMITTED_CLEANUP_WARNING)
    } else {
        None
    };
    MutationOutcome::Committed { warning }
}

fn create_rewind_temp_file(
    parent: &CapDir,
    target_mode: Option<u32>,
) -> anyhow::Result<(OsString, CapFile)> {
    #[cfg(not(unix))]
    let _ = target_mode;
    let stamp = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();
    let pid = std::process::id();
    for attempt in 0..128_u64 {
        let sequence = REWIND_TEMP_COUNTER.fetch_add(1, Ordering::Relaxed);
        let name = OsString::from(format!(
            ".magi-rewind-{pid}-{stamp}-{sequence}-{attempt}.tmp"
        ));
        let mut options = CapOpenOptions::new();
        options
            .write(true)
            .create_new(true)
            .truncate(true)
            .follow(FollowSymlinks::No);
        #[cfg(unix)]
        options.mode(target_mode.unwrap_or(0o666));
        match parent.open_with(Path::new(name.as_os_str()), &options) {
            Ok(file) => return Ok((name, file)),
            Err(error) if error.kind() == ErrorKind::AlreadyExists => {}
            Err(error) => return Err(error.into()),
        }
    }
    anyhow::bail!("could not allocate a unique rewind temporary file")
}

fn prepare_rewind_temp_file(
    parent: &CapDir,
    bytes: &[u8],
    target_mode: Option<u32>,
    inject_write_failure: bool,
) -> Result<(OsString, String), &'static str> {
    let (name, mut file) =
        create_rewind_temp_file(parent, target_mode).map_err(|_| "write failed")?;
    let expected_hash = sha256_hex(bytes);
    let prepare_result = (|| -> anyhow::Result<()> {
        file.write_all(bytes)?;
        if inject_write_failure {
            anyhow::bail!("injected write failure");
        }
        file.flush()?;
        #[cfg(unix)]
        if let Some(mode) = target_mode {
            file.set_permissions(cap_std::fs::Permissions::from_mode(mode))?;
        }
        file.sync_all()?;
        Ok(())
    })();
    drop(file);
    if prepare_result.is_err() {
        // The entry is removed only after it has been moved and verified. A
        // replacement of the temporary name is retained as recovery evidence.
        let _ = remove_validated_entry(parent, &name, &expected_hash);
        return Err("write failed");
    }
    Ok((name, expected_hash))
}

struct TempFileCleanupGuard<'a> {
    parent: &'a CapDir,
    name: Option<OsString>,
    expected_hash: String,
}

impl<'a> TempFileCleanupGuard<'a> {
    fn new(parent: &'a CapDir, name: OsString, expected_hash: String) -> Self {
        Self {
            parent,
            name: Some(name),
            expected_hash,
        }
    }

    fn disarm(&mut self) {
        self.name = None;
    }
}

impl Drop for TempFileCleanupGuard<'_> {
    fn drop(&mut self) {
        if let Some(name) = self.name.take() {
            let _ = remove_validated_entry(self.parent, &name, &self.expected_hash);
        }
    }
}

fn atomic_replace_target(
    target: &RewindTarget,
    bytes: &[u8],
    expected_current_hash: &str,
    target_mode: Option<u32>,
    inject_write_failure: bool,
    inject_final_sync_failure: bool,
    after_quarantine_hashed: &mut dyn FnMut(),
) -> MutationOutcome {
    let (temp_name, temp_hash) =
        match prepare_rewind_temp_file(&target.parent, bytes, target_mode, inject_write_failure) {
            Ok(prepared) => prepared,
            Err(reason) => return MutationOutcome::Failed(reason),
        };
    let mut temp_cleanup =
        TempFileCleanupGuard::new(&target.parent, temp_name.clone(), temp_hash.clone());

    let quarantine_name = match move_target_to_quarantine(target) {
        Ok(name) => name,
        Err(error) if matches!(error.kind(), ErrorKind::NotFound | ErrorKind::AlreadyExists) => {
            return MutationOutcome::Conflict(TARGET_CHANGED_DURING_REWIND);
        }
        Err(_) => return MutationOutcome::Failed("write failed"),
    };
    let quarantine_snapshot = match snapshot_named(&target.parent, &quarantine_name) {
        Ok(Some(snapshot)) => snapshot,
        Ok(None) | Err(_) => return MutationOutcome::Failed("hash check failed"),
    };
    if quarantine_snapshot.hash != expected_current_hash {
        recover_quarantine(
            &target.parent,
            &quarantine_name,
            &target.name,
            &quarantine_snapshot.hash,
        );
        return MutationOutcome::Conflict(TARGET_CHANGED_DURING_REWIND);
    }
    after_quarantine_hashed();

    match target.parent.hard_link(
        Path::new(temp_name.as_os_str()),
        &target.parent,
        Path::new(target.name.as_os_str()),
    ) {
        Ok(()) => {}
        Err(error) if error.kind() == ErrorKind::AlreadyExists => {
            recover_quarantine(
                &target.parent,
                &quarantine_name,
                &target.name,
                &quarantine_snapshot.hash,
            );
            return MutationOutcome::Conflict(TARGET_CHANGED_DURING_REWIND);
        }
        Err(_) => {
            recover_quarantine(
                &target.parent,
                &quarantine_name,
                &target.name,
                &quarantine_snapshot.hash,
            );
            return MutationOutcome::Failed("write failed");
        }
    }

    let temp_cleanup_incomplete = !matches!(
        remove_validated_entry(&target.parent, &temp_name, &temp_hash),
        ValidatedRemoval::Removed | ValidatedRemoval::Missing
    );
    temp_cleanup.disarm();
    let quarantine_cleanup_incomplete = !matches!(
        remove_validated_entry(&target.parent, &quarantine_name, &quarantine_snapshot.hash,),
        ValidatedRemoval::Removed | ValidatedRemoval::Missing
    );
    finish_committed(
        &target.parent,
        inject_final_sync_failure,
        temp_cleanup_incomplete || quarantine_cleanup_incomplete,
    )
}

fn delete_target(
    target: &RewindTarget,
    expected_current_hash: &str,
    inject_delete_failure: bool,
    inject_final_sync_failure: bool,
    after_quarantine_hashed: &mut impl FnMut(),
) -> MutationOutcome {
    let quarantine_name = match move_target_to_quarantine(target) {
        Ok(name) => name,
        Err(error) if matches!(error.kind(), ErrorKind::NotFound | ErrorKind::AlreadyExists) => {
            return MutationOutcome::Conflict(TARGET_CHANGED_DURING_REWIND);
        }
        Err(_) => return MutationOutcome::Failed("delete failed"),
    };
    let quarantine_snapshot = match snapshot_named(&target.parent, &quarantine_name) {
        Ok(Some(snapshot)) => snapshot,
        Ok(None) | Err(_) => return MutationOutcome::Failed("hash check failed"),
    };
    if quarantine_snapshot.hash != expected_current_hash {
        recover_quarantine(
            &target.parent,
            &quarantine_name,
            &target.name,
            &quarantine_snapshot.hash,
        );
        return MutationOutcome::Conflict(TARGET_CHANGED_DURING_REWIND);
    }
    after_quarantine_hashed();
    if inject_delete_failure {
        recover_quarantine(
            &target.parent,
            &quarantine_name,
            &target.name,
            &quarantine_snapshot.hash,
        );
        return MutationOutcome::Failed("delete failed");
    }

    match remove_validated_entry(&target.parent, &quarantine_name, &quarantine_snapshot.hash) {
        ValidatedRemoval::Removed => {
            finish_committed(&target.parent, inject_final_sync_failure, false)
        }
        ValidatedRemoval::Missing => MutationOutcome::Failed("delete failed"),
        ValidatedRemoval::Preserved => MutationOutcome::Conflict(TARGET_CHANGED_DURING_REWIND),
    }
}

fn changes_from_records(store: &CheckpointStore, read: LedgerRead) -> ChangesReadModel {
    let mut by_turn: BTreeMap<u64, Vec<ChangeEntry>> = BTreeMap::new();
    for record in read.records {
        let event = record.event;
        let visible_path = redacted_relative_path(&event.status, event.relative_path.clone());
        let (classification, rewindable, unavailable_reason) = match event.status {
            SnapshotCaptureStatus::Captured => {
                let available = event
                    .pre
                    .as_ref()
                    .into_iter()
                    .chain(event.post.as_ref())
                    .all(|blob| store.blob_path(&blob.sha256).exists());
                let classification = match (&event.pre, &event.post) {
                    (None, Some(_)) => ChangeClassification::Created,
                    (Some(_), Some(_)) => ChangeClassification::Modified,
                    (Some(_), None) => ChangeClassification::Deleted,
                    (None, None) => ChangeClassification::NonRewindable,
                };
                (
                    classification,
                    available,
                    (!available).then_some("missing checkpoint blob".to_string()),
                )
            }
            SnapshotCaptureStatus::Excluded { reason } => (
                ChangeClassification::NonRewindable,
                false,
                Some(format!("excluded: {reason:?}")),
            ),
            SnapshotCaptureStatus::Oversized { bytes, max_bytes } => (
                ChangeClassification::NonRewindable,
                false,
                Some(format!("oversized: {bytes} > {max_bytes}")),
            ),
            SnapshotCaptureStatus::Unavailable { reason } => {
                (ChangeClassification::NonRewindable, false, Some(reason))
            }
        };
        by_turn
            .entry(event.user_turn)
            .or_default()
            .push(ChangeEntry {
                tool: event.tool,
                relative_path: visible_path,
                classification,
                rewindable,
                unavailable_reason,
            });
    }
    ChangesReadModel {
        turns: by_turn
            .into_iter()
            .map(|(user_turn, mut changes)| {
                changes.sort_by(|left, right| left.relative_path.cmp(&right.relative_path));
                ChangeTurn { user_turn, changes }
            })
            .collect(),
        diagnostics: read.diagnostics,
    }
}

fn plan_rewind(
    store: &CheckpointStore,
    session_id: &str,
    cwd: &Path,
    target_turn: u64,
) -> RestorePlan {
    let read = store.read_records(session_id);
    plan_rewind_from_records(store, session_id, cwd, target_turn, &read)
}

fn plan_rewind_from_records(
    store: &CheckpointStore,
    session_id: &str,
    cwd: &Path,
    target_turn: u64,
    read: &LedgerRead,
) -> RestorePlan {
    let latest_turn = read
        .records
        .iter()
        .map(|record| record.event.user_turn)
        .max();
    let selected = read
        .records
        .iter()
        .filter(|record| record.event.user_turn >= target_turn)
        .cloned()
        .collect::<Vec<_>>();
    let mut by_path: BTreeMap<PathBuf, Vec<SnapshotLedgerRecord>> = BTreeMap::new();
    for record in selected {
        by_path
            .entry(record.event.relative_path.clone())
            .or_default()
            .push(record);
    }

    // The ambient open is the trusted boundary. Every target below is then
    // resolved relative to this held directory, never through `cwd` again.
    let (canonical_cwd, root) = open_rewind_root(cwd);
    let mut operations = Vec::new();
    for (relative_path, mut records) in by_path {
        records.sort_by_key(|record| record.event.user_turn);
        if let Err(reason) = safe_rewind_target_path(store, &canonical_cwd, &relative_path) {
            operations.push(RestoreOperation {
                relative_path: PathBuf::from(REDACTED_PATH),
                kind: RestoreOperationKind::SkipUnavailable {
                    reason: reason.to_string(),
                },
            });
            continue;
        }

        let first = &records.first().expect("group non-empty").event;
        let last = &records.last().expect("group non-empty").event;
        if !records
            .iter()
            .all(|record| record.event.status == SnapshotCaptureStatus::Captured)
        {
            operations.push(RestoreOperation {
                relative_path,
                kind: RestoreOperationKind::SkipUnavailable {
                    reason: "non-rewindable snapshot".to_string(),
                },
            });
            continue;
        }

        let expected_current = last.post.clone();
        let current = match root.as_ref() {
            Some(root) => {
                match target_snapshot_for_plan(root, &relative_path, expected_current.is_none()) {
                    Ok(current) => current,
                    Err(_) => {
                        operations.push(RestoreOperation {
                            relative_path,
                            kind: RestoreOperationKind::SkipUnavailable {
                                reason: "hash check failed".to_string(),
                            },
                        });
                        continue;
                    }
                }
            }
            None => {
                operations.push(RestoreOperation {
                    relative_path,
                    kind: RestoreOperationKind::SkipUnavailable {
                        reason: "hash check failed".to_string(),
                    },
                });
                continue;
            }
        };
        if let Some(expected) = expected_current.as_ref() {
            if current.as_ref().map(|snapshot| snapshot.hash.as_str())
                != Some(expected.sha256.as_str())
            {
                operations.push(RestoreOperation {
                    relative_path,
                    kind: RestoreOperationKind::SkipConflict {
                        reason: "current file hash differs from checkpoint post-image".to_string(),
                    },
                });
                continue;
            }
        } else if current.is_some() {
            operations.push(RestoreOperation {
                relative_path,
                kind: RestoreOperationKind::SkipConflict {
                    reason: "expected deleted file exists".to_string(),
                },
            });
            continue;
        }

        let kind = match (&first.pre, expected_current) {
            (Some(pre), Some(expected_current)) => {
                if !store.blob_path(&pre.sha256).exists() {
                    RestoreOperationKind::SkipUnavailable {
                        reason: "missing pre-image blob".to_string(),
                    }
                } else {
                    RestoreOperationKind::Restore {
                        pre: pre.clone(),
                        expected_current,
                    }
                }
            }
            (None, Some(expected_current)) => {
                RestoreOperationKind::DeleteCreated { expected_current }
            }
            (Some(pre), None) => {
                if store.blob_path(&pre.sha256).exists() {
                    RestoreOperationKind::ResurrectDeleted { pre: pre.clone() }
                } else {
                    RestoreOperationKind::SkipUnavailable {
                        reason: "missing pre-image blob".to_string(),
                    }
                }
            }
            (None, None) => RestoreOperationKind::SkipUnavailable {
                reason: "missing pre and post images".to_string(),
            },
        };
        operations.push(RestoreOperation {
            relative_path,
            kind,
        });
    }
    operations.sort_by(|left, right| left.relative_path.cmp(&right.relative_path));
    RestorePlan {
        session_id: session_id.to_string(),
        target_turn,
        latest_turn,
        operations,
        diagnostics: read.diagnostics.clone(),
    }
}

fn execute_plan(store: &CheckpointStore, plan: &RestorePlan, cwd: &Path) -> RewindExecution {
    execute_plan_internal(store, plan, cwd, None)
}

#[cfg(test)]
fn execute_plan_with_failure(
    store: &CheckpointStore,
    plan: &RestorePlan,
    cwd: &Path,
    phase: &'static str,
) -> RewindExecution {
    execute_plan_internal(store, plan, cwd, Some(phase))
}

#[cfg(all(test, unix))]
fn execute_plan_with_parent_hook(
    store: &CheckpointStore,
    plan: &RestorePlan,
    cwd: &Path,
    hook: impl FnMut(),
) -> RewindExecution {
    execute_plan_internal_with_hooks(store, plan, cwd, None, hook, || {}, || {})
}

#[cfg(test)]
fn execute_plan_with_target_hook(
    store: &CheckpointStore,
    plan: &RestorePlan,
    cwd: &Path,
    hook: impl FnMut(),
) -> RewindExecution {
    execute_plan_internal_with_hooks(store, plan, cwd, None, || {}, hook, || {})
}

#[cfg(test)]
fn execute_plan_with_quarantine_hook(
    store: &CheckpointStore,
    plan: &RestorePlan,
    cwd: &Path,
    hook: impl FnMut(),
) -> RewindExecution {
    execute_plan_internal_with_hooks(store, plan, cwd, None, || {}, || {}, hook)
}

fn execute_plan_internal(
    store: &CheckpointStore,
    plan: &RestorePlan,
    cwd: &Path,
    injected_failure: Option<&str>,
) -> RewindExecution {
    execute_plan_internal_with_hooks(store, plan, cwd, injected_failure, || {}, || {}, || {})
}

fn mutation_status(
    outcome: MutationOutcome,
    committed_status: RestoreStatus,
    reason: &mut Option<String>,
) -> RestoreStatus {
    match outcome {
        MutationOutcome::Committed { warning } => {
            *reason = warning.map(str::to_string);
            committed_status
        }
        MutationOutcome::Conflict(operation_reason) => {
            *reason = Some(operation_reason.to_string());
            RestoreStatus::SkipConflict
        }
        MutationOutcome::Failed(operation_reason) => {
            *reason = Some(operation_reason.to_string());
            RestoreStatus::SkipUnavailable
        }
    }
}

fn execute_plan_internal_with_hooks(
    store: &CheckpointStore,
    plan: &RestorePlan,
    cwd: &Path,
    injected_failure: Option<&str>,
    mut after_parent_acquired: impl FnMut(),
    mut after_target_hashed: impl FnMut(),
    mut after_quarantine_hashed: impl FnMut(),
) -> RewindExecution {
    // This is the only ambient open in the target-operation path. All later
    // reads, creates, renames, and removals use descendants of this handle.
    let (canonical_cwd, root) = open_rewind_root(cwd);
    let mut results = Vec::new();
    for operation in &plan.operations {
        let mut reason = None;
        let status = if safe_rewind_target_path(store, &canonical_cwd, &operation.relative_path)
            .is_err()
        {
            RestoreStatus::SkipUnavailable
        } else {
            match &operation.kind {
                RestoreOperationKind::Restore {
                    pre,
                    expected_current,
                } => {
                    let target = match root.as_ref() {
                        Some(root) => {
                            open_target_parent(root, &operation.relative_path, false, false)
                        }
                        None => Err(io::Error::new(ErrorKind::NotFound, "cwd unavailable")),
                    };
                    match target {
                        Err(_) => {
                            reason = Some("hash check failed".to_string());
                            RestoreStatus::SkipUnavailable
                        }
                        Ok(target) => {
                            after_parent_acquired();
                            let current = if injected_failure == Some("hash") {
                                Err(anyhow::anyhow!("injected failure"))
                            } else {
                                target_snapshot(&target).map_err(anyhow::Error::from)
                            };
                            match current {
                                Err(_) => {
                                    reason = Some("hash check failed".to_string());
                                    RestoreStatus::SkipUnavailable
                                }
                                Ok(current)
                                    if current.as_ref().map(|snapshot| snapshot.hash.as_str())
                                        != Some(expected_current.sha256.as_str()) =>
                                {
                                    RestoreStatus::SkipConflict
                                }
                                Ok(Some(current)) => {
                                    after_target_hashed();
                                    let outcome = write_rewind_blob(
                                        store,
                                        pre,
                                        &target,
                                        RewindBlobOptions {
                                            expected_current_hash: Some(&expected_current.sha256),
                                            target_mode: target_mode(&current),
                                            injected_failure,
                                            inject_final_sync_failure: injected_failure
                                                == Some("final sync"),
                                            after_quarantine_hashed: &mut after_quarantine_hashed,
                                        },
                                    );
                                    mutation_status(outcome, RestoreStatus::Restored, &mut reason)
                                }
                                Ok(None) => RestoreStatus::SkipConflict,
                            }
                        }
                    }
                }
                RestoreOperationKind::DeleteCreated { expected_current } => {
                    let target = match root.as_ref() {
                        Some(root) => {
                            open_target_parent(root, &operation.relative_path, false, false)
                        }
                        None => Err(io::Error::new(ErrorKind::NotFound, "cwd unavailable")),
                    };
                    match target {
                        Err(_) => {
                            reason = Some("hash check failed".to_string());
                            RestoreStatus::SkipUnavailable
                        }
                        Ok(target) => {
                            after_parent_acquired();
                            let current = if injected_failure == Some("hash") {
                                Err(anyhow::anyhow!("injected failure"))
                            } else {
                                target_snapshot(&target).map_err(anyhow::Error::from)
                            };
                            match current {
                                Err(_) => {
                                    reason = Some("hash check failed".to_string());
                                    RestoreStatus::SkipUnavailable
                                }
                                Ok(current)
                                    if current.as_ref().map(|snapshot| snapshot.hash.as_str())
                                        != Some(expected_current.sha256.as_str()) =>
                                {
                                    RestoreStatus::SkipConflict
                                }
                                Ok(Some(_)) => {
                                    after_target_hashed();
                                    let outcome = delete_target(
                                        &target,
                                        &expected_current.sha256,
                                        injected_failure == Some("delete"),
                                        injected_failure == Some("final sync"),
                                        &mut after_quarantine_hashed,
                                    );
                                    mutation_status(
                                        outcome,
                                        RestoreStatus::DeletedCreated,
                                        &mut reason,
                                    )
                                }
                                Ok(None) => RestoreStatus::SkipConflict,
                            }
                        }
                    }
                }
                RestoreOperationKind::ResurrectDeleted { pre } => {
                    let target = match root.as_ref() {
                        Some(root) => {
                            open_target_parent(root, &operation.relative_path, false, false)
                        }
                        None => Err(io::Error::new(ErrorKind::NotFound, "cwd unavailable")),
                    };
                    match target {
                        Ok(target) => {
                            after_parent_acquired();
                            let current = target_snapshot(&target).map_err(anyhow::Error::from);
                            match current {
                                Err(_) => {
                                    reason = Some("hash check failed".to_string());
                                    RestoreStatus::SkipUnavailable
                                }
                                Ok(Some(_)) => RestoreStatus::SkipConflict,
                                Ok(None) => {
                                    let outcome = write_rewind_blob(
                                        store,
                                        pre,
                                        &target,
                                        RewindBlobOptions {
                                            expected_current_hash: None,
                                            target_mode: None,
                                            injected_failure,
                                            inject_final_sync_failure: injected_failure
                                                == Some("final sync"),
                                            after_quarantine_hashed: &mut after_quarantine_hashed,
                                        },
                                    );
                                    mutation_status(
                                        outcome,
                                        RestoreStatus::ResurrectedDeleted,
                                        &mut reason,
                                    )
                                }
                            }
                        }
                        Err(error) if error.kind() == ErrorKind::NotFound => {
                            // Do not create missing parents until the blob has
                            // been read successfully. This keeps blob failures
                            // side-effect free while retaining capability safety.
                            let bytes = match read_rewind_blob(store, pre, injected_failure) {
                                Ok(bytes) => bytes,
                                Err(failure) => {
                                    reason = Some(failure.to_string());
                                    Vec::new()
                                }
                            };
                            if reason.is_some() {
                                RestoreStatus::SkipUnavailable
                            } else {
                                let target = match root.as_ref() {
                                    Some(root) => open_target_parent(
                                        root,
                                        &operation.relative_path,
                                        true,
                                        injected_failure == Some("ancestor sync"),
                                    ),
                                    None => {
                                        Err(io::Error::new(ErrorKind::NotFound, "cwd unavailable"))
                                    }
                                };
                                match target {
                                    Err(_) => {
                                        reason = Some("write failed".to_string());
                                        RestoreStatus::SkipUnavailable
                                    }
                                    Ok(target) => {
                                        after_parent_acquired();
                                        match target_snapshot(&target) {
                                            Err(_) => {
                                                reason = Some("hash check failed".to_string());
                                                RestoreStatus::SkipUnavailable
                                            }
                                            Ok(Some(_)) => RestoreStatus::SkipConflict,
                                            Ok(None) => {
                                                let outcome = install_new_target(
                                                    &target,
                                                    &bytes,
                                                    injected_failure == Some("write"),
                                                    injected_failure == Some("final sync"),
                                                );
                                                mutation_status(
                                                    outcome,
                                                    RestoreStatus::ResurrectedDeleted,
                                                    &mut reason,
                                                )
                                            }
                                        }
                                    }
                                }
                            }
                        }
                        Err(_) => {
                            reason = Some("write failed".to_string());
                            RestoreStatus::SkipUnavailable
                        }
                    }
                }
                RestoreOperationKind::SkipConflict {
                    reason: operation_reason,
                } => {
                    reason = Some(sanitized_diagnostic(operation_reason.clone()));
                    RestoreStatus::SkipConflict
                }
                RestoreOperationKind::SkipUnavailable {
                    reason: operation_reason,
                } => {
                    reason = Some(sanitized_diagnostic(operation_reason.clone()));
                    RestoreStatus::SkipUnavailable
                }
            }
        };
        results.push(RestoreOperationResult {
            relative_path: operation.relative_path.clone(),
            status,
            reason,
        });
    }
    RewindExecution { results }
}

fn read_rewind_blob(
    store: &CheckpointStore,
    blob: &SnapshotBlobRef,
    injected_failure: Option<&str>,
) -> Result<Vec<u8>, &'static str> {
    if injected_failure == Some("blob read") {
        return Err("blob read failed");
    }
    store.read_blob(blob).map_err(|_| "blob read failed")
}

fn install_new_target(
    target: &RewindTarget,
    bytes: &[u8],
    inject_write_failure: bool,
    inject_final_sync_failure: bool,
) -> MutationOutcome {
    let (temp_name, temp_hash) =
        match prepare_rewind_temp_file(&target.parent, bytes, None, inject_write_failure) {
            Ok(prepared) => prepared,
            Err(reason) => return MutationOutcome::Failed(reason),
        };
    let mut temp_cleanup =
        TempFileCleanupGuard::new(&target.parent, temp_name.clone(), temp_hash.clone());
    match target.parent.hard_link(
        Path::new(temp_name.as_os_str()),
        &target.parent,
        Path::new(target.name.as_os_str()),
    ) {
        Ok(()) => {}
        Err(error) if error.kind() == ErrorKind::AlreadyExists => {
            return MutationOutcome::Conflict(TARGET_CHANGED_DURING_REWIND);
        }
        Err(_) => return MutationOutcome::Failed("write failed"),
    }
    let cleanup_incomplete = !matches!(
        remove_validated_entry(&target.parent, &temp_name, &temp_hash),
        ValidatedRemoval::Removed | ValidatedRemoval::Missing
    );
    temp_cleanup.disarm();
    finish_committed(
        &target.parent,
        inject_final_sync_failure,
        cleanup_incomplete,
    )
}

struct RewindBlobOptions<'a> {
    expected_current_hash: Option<&'a str>,
    target_mode: Option<u32>,
    injected_failure: Option<&'a str>,
    inject_final_sync_failure: bool,
    after_quarantine_hashed: &'a mut dyn FnMut(),
}

fn write_rewind_blob(
    store: &CheckpointStore,
    blob: &SnapshotBlobRef,
    target: &RewindTarget,
    options: RewindBlobOptions<'_>,
) -> MutationOutcome {
    let RewindBlobOptions {
        expected_current_hash,
        target_mode,
        injected_failure,
        inject_final_sync_failure,
        after_quarantine_hashed,
    } = options;
    let bytes = match read_rewind_blob(store, blob, injected_failure) {
        Ok(bytes) => bytes,
        Err(reason) => return MutationOutcome::Failed(reason),
    };
    match expected_current_hash {
        Some(expected_current_hash) => atomic_replace_target(
            target,
            &bytes,
            expected_current_hash,
            target_mode,
            injected_failure == Some("write"),
            inject_final_sync_failure,
            after_quarantine_hashed,
        ),
        None => install_new_target(
            target,
            &bytes,
            injected_failure == Some("write"),
            inject_final_sync_failure,
        ),
    }
}

pub(crate) fn rewind_event_payload(plan: &RestorePlan, execution: &RewindExecution) -> Value {
    let mut counts: BTreeMap<RestoreStatus, usize> = BTreeMap::new();
    let paths = execution
        .results
        .iter()
        .map(|result| {
            *counts.entry(result.status).or_default() += 1;
            let mut item = serde_json::Map::from_iter([
                (
                    "path".to_string(),
                    json!(redacted_path_for_display(&result.relative_path)),
                ),
                ("status".to_string(), json!(result.status)),
            ]);
            if let Some(reason) = &result.reason {
                item.insert("reason".to_string(), json!(reason));
            }
            Value::Object(item)
        })
        .collect::<Vec<_>>();
    serde_json::json!({
        "target_turn": plan.target_turn,
        "latest_turn": plan.latest_turn,
        "paths": paths,
        "counts": counts.into_iter().map(|(status, count)| (format!("{status:?}"), count)).collect::<BTreeMap<_, _>>(),
    })
}

pub(crate) fn replay_rewind_context(payload: &Value) -> Option<String> {
    let target = payload.get("target_turn").and_then(Value::as_u64)?;
    let latest = payload
        .get("latest_turn")
        .and_then(Value::as_u64)
        .unwrap_or(target);
    let mut lines = vec![format!(
        "Local filesystem rewind applied for user turns {target}..{latest}. File contents and diffs omitted."
    )];
    if let Some(paths) = payload.get("paths").and_then(Value::as_array) {
        for item in paths.iter().take(64) {
            let path = item
                .get("path")
                .and_then(Value::as_str)
                .unwrap_or("<redacted>");
            let status = item
                .get("status")
                .and_then(Value::as_str)
                .unwrap_or("unknown");
            lines.push(format!("- {}: {}", safe_provider_path(path), status));
        }
    }
    Some(lines.join("\n"))
}

fn safe_provider_path(path: &str) -> String {
    let sanitized = redact_sensitive_text(path);
    if sanitized.contains("..")
        || sanitized.starts_with('/')
        || secret_path_for_rewind(Path::new(&sanitized))
    {
        "<redacted>".to_string()
    } else {
        sanitized
    }
}

pub(crate) fn parse_rewind_target(arg: Option<&str>) -> Result<ParsedRewindTarget, String> {
    let Some(arg) = arg.map(str::trim).filter(|arg| !arg.is_empty()) else {
        return Ok(ParsedRewindTarget::NeedsSelection);
    };
    let mut dry_run = false;
    let mut target = None;
    let mut mode = super::RewindMode::default();
    let mut tokens = arg.split_whitespace();
    while let Some(token) = tokens.next() {
        match token {
            "--dry-run" => dry_run = true,
            "--mode" => {
                mode = match tokens.next() {
                    Some("conversation") => super::RewindMode::Conversation,
                    Some("files") => super::RewindMode::Files,
                    Some("both") => super::RewindMode::Both,
                    _ => return Err("mode must be conversation, files, or both".to_string()),
                };
            }
            "--to" => {
                let Some(value) = tokens.next() else {
                    return Err("missing turn after --to".to_string());
                };
                target = Some(parse_positive_turn(value)?);
            }
            value if target.is_none() => target = Some(parse_positive_turn(value)?),
            _ => return Err("unexpected rewind argument".to_string()),
        }
    }
    let Some(target_turn) = target else {
        return Err("missing rewind target turn".to_string());
    };
    Ok(ParsedRewindTarget::Target {
        target_turn,
        dry_run,
        mode,
    })
}

fn parse_positive_turn(value: &str) -> Result<u64, String> {
    if value.contains("..") || value.starts_with('-') {
        return Err("rewind target must be a positive turn number".to_string());
    }
    let turn = value
        .parse::<u64>()
        .map_err(|_| "rewind target must be a positive turn number".to_string())?;
    if turn == 0 {
        return Err("rewind target must be positive".to_string());
    }
    Ok(turn)
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ParsedRewindTarget {
    NeedsSelection,
    Target {
        target_turn: u64,
        dry_run: bool,
        mode: super::RewindMode,
    },
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        checkpoints::{
            SnapshotContext, SnapshotExclusionReason, capture::FileSnapshotEvent,
            capture_file_snapshot,
        },
        config::McPaths,
    };
    use std::fs;
    #[cfg(unix)]
    use std::os::unix::fs::symlink;
    use tempfile::TempDir;

    fn store(temp: &TempDir) -> CheckpointStore {
        CheckpointStore::new(temp.path().join("mc/checkpoints"))
    }

    fn context(temp: &TempDir) -> SnapshotContext {
        let paths = McPaths::from_root(temp.path().join("mc"));
        SnapshotContext {
            store: CheckpointStore::from_paths(&paths),
            paths,
            session_id: "session".to_string(),
            user_turn: 1,
        }
    }

    fn captured_record(
        store: &CheckpointStore,
        relative_path: impl Into<PathBuf>,
    ) -> SnapshotLedgerRecord {
        SnapshotLedgerRecord::new(FileSnapshotEvent {
            session_id: "session".to_string(),
            user_turn: 1,
            tool: SnapshotTool::Edit,
            cwd: PathBuf::new(),
            relative_path: relative_path.into(),
            pre: Some(store.write_blob(b"before").unwrap()),
            post: Some(store.write_blob(b"after").unwrap()),
            status: SnapshotCaptureStatus::Captured,
        })
    }

    fn assert_no_secret_path(text: &str) {
        let text = text.to_ascii_lowercase();
        assert!(!text.contains("auth.json"), "leaked auth.json: {text}");
        assert!(!text.contains(".env"), "leaked .env: {text}");
        assert!(!text.contains("id_rsa"), "leaked id_rsa: {text}");
    }

    #[cfg(unix)]
    fn all_operation_plan(
        relative_path: impl Into<PathBuf>,
        pre: SnapshotBlobRef,
        expected_current: SnapshotBlobRef,
    ) -> RestorePlan {
        let relative_path = relative_path.into();
        RestorePlan {
            session_id: "session".to_string(),
            target_turn: 1,
            latest_turn: Some(1),
            diagnostics: Vec::new(),
            operations: vec![
                RestoreOperation {
                    relative_path: relative_path.clone(),
                    kind: RestoreOperationKind::Restore {
                        pre: pre.clone(),
                        expected_current: expected_current.clone(),
                    },
                },
                RestoreOperation {
                    relative_path: relative_path.clone(),
                    kind: RestoreOperationKind::DeleteCreated { expected_current },
                },
                RestoreOperation {
                    relative_path,
                    kind: RestoreOperationKind::ResurrectDeleted { pre },
                },
            ],
        }
    }

    #[test]
    fn snapshots_group_by_user_turn_ordinal() {
        let temp = TempDir::new().unwrap();
        let store = store(&temp);
        for turn in [2, 1] {
            store
                .append_record(&SnapshotLedgerRecord::new(FileSnapshotEvent {
                    session_id: "session".to_string(),
                    user_turn: turn,
                    tool: SnapshotTool::Edit,
                    cwd: temp.path().to_path_buf(),
                    relative_path: PathBuf::from(format!("{turn}.txt")),
                    pre: None,
                    post: None,
                    status: SnapshotCaptureStatus::Excluded {
                        reason: SnapshotExclusionReason::SecretPath,
                    },
                }))
                .unwrap();
        }
        let changes = store.changes("session");
        assert_eq!(
            changes
                .turns
                .iter()
                .map(|turn| turn.user_turn)
                .collect::<Vec<_>>(),
            vec![1, 2]
        );
    }

    #[test]
    fn changes_lists_rewindable_turns_without_bytes() {
        let temp = TempDir::new().unwrap();
        let context = context(&temp);
        let file = temp.path().join("file.txt");
        fs::write(&file, "after").unwrap();
        capture_file_snapshot(
            &context,
            temp.path(),
            SnapshotTool::WriteFile,
            &file,
            Some(b"before"),
            Some(b"after"),
        )
        .unwrap();
        let changes = context.store.changes("session");
        assert!(changes.turns[0].changes[0].rewindable);
        assert!(!format!("{changes:?}").contains("before"));
    }

    #[test]
    fn changes_marks_missing_blob_unavailable() {
        let temp = TempDir::new().unwrap();
        let store = store(&temp);
        store
            .append_record(&SnapshotLedgerRecord::new(FileSnapshotEvent {
                session_id: "session".to_string(),
                user_turn: 1,
                tool: SnapshotTool::Edit,
                cwd: temp.path().to_path_buf(),
                relative_path: PathBuf::from("file.txt"),
                pre: Some(SnapshotBlobRef {
                    sha256: "a".repeat(64),
                    bytes: 1,
                }),
                post: None,
                status: SnapshotCaptureStatus::Captured,
            }))
            .unwrap();
        assert!(!store.changes("session").turns[0].changes[0].rewindable);
    }

    #[test]
    fn parse_rewind_target_matrix() {
        assert_eq!(
            parse_rewind_target(None).unwrap(),
            ParsedRewindTarget::NeedsSelection
        );
        assert_eq!(
            parse_rewind_target(Some("3")).unwrap(),
            ParsedRewindTarget::Target {
                target_turn: 3,
                dry_run: false,
                mode: super::super::RewindMode::Conversation,
            }
        );
        assert_eq!(
            parse_rewind_target(Some("--dry-run --to 3")).unwrap(),
            ParsedRewindTarget::Target {
                target_turn: 3,
                dry_run: true,
                mode: super::super::RewindMode::Conversation,
            }
        );
        assert!(parse_rewind_target(Some("0")).is_err());
        assert!(parse_rewind_target(Some("1..2")).is_err());
        assert!(parse_rewind_target(Some("--to")).is_err());
    }

    #[test]
    fn planner_merges_same_file_edits() {
        let temp = TempDir::new().unwrap();
        let context = context(&temp);
        let file = temp.path().join("file.txt");
        fs::write(&file, "c").unwrap();
        capture_file_snapshot(
            &context,
            temp.path(),
            SnapshotTool::Edit,
            &file,
            Some(b"a"),
            Some(b"b"),
        )
        .unwrap();
        let mut context2 = context.clone();
        context2.user_turn = 2;
        capture_file_snapshot(
            &context2,
            temp.path(),
            SnapshotTool::Edit,
            &file,
            Some(b"b"),
            Some(b"c"),
        )
        .unwrap();
        let plan = context.store.plan_rewind("session", temp.path(), 1);
        assert_eq!(plan.operations.len(), 1);
        assert!(matches!(
            plan.operations[0].kind,
            RestoreOperationKind::Restore { .. }
        ));
    }

    #[test]
    fn planner_emits_skip_conflict_on_hash_mismatch() {
        let temp = TempDir::new().unwrap();
        let context = context(&temp);
        let file = temp.path().join("file.txt");
        fs::write(&file, "after").unwrap();
        capture_file_snapshot(
            &context,
            temp.path(),
            SnapshotTool::Edit,
            &file,
            Some(b"before"),
            Some(b"after"),
        )
        .unwrap();
        fs::write(&file, "changed").unwrap();
        let plan = context.store.plan_rewind("session", temp.path(), 1);
        assert!(matches!(
            plan.operations[0].kind,
            RestoreOperationKind::SkipConflict { .. }
        ));
    }

    #[test]
    fn planner_emits_delete_created_and_resurrect_deleted() {
        let temp = TempDir::new().unwrap();
        let context = context(&temp);
        let created = temp.path().join("created.txt");
        fs::write(&created, "created").unwrap();
        capture_file_snapshot(
            &context,
            temp.path(),
            SnapshotTool::WriteFile,
            &created,
            None,
            Some(b"created"),
        )
        .unwrap();
        context
            .store
            .append_record(&SnapshotLedgerRecord::new(FileSnapshotEvent {
                session_id: "session".to_string(),
                user_turn: 1,
                tool: SnapshotTool::Edit,
                cwd: temp.path().to_path_buf(),
                relative_path: PathBuf::from("deleted.txt"),
                pre: Some(context.store.write_blob(b"old").unwrap()),
                post: None,
                status: SnapshotCaptureStatus::Captured,
            }))
            .unwrap();
        let plan = context.store.plan_rewind("session", temp.path(), 1);
        assert!(
            plan.operations
                .iter()
                .any(|op| matches!(op.kind, RestoreOperationKind::DeleteCreated { .. }))
        );
        assert!(
            plan.operations
                .iter()
                .any(|op| matches!(op.kind, RestoreOperationKind::ResurrectDeleted { .. }))
        );
    }

    #[test]
    fn planner_tolerates_missing_ledger_and_blobs() {
        let temp = TempDir::new().unwrap();
        assert!(
            store(&temp)
                .plan_rewind("missing", temp.path(), 1)
                .operations
                .is_empty()
        );
    }

    #[test]
    fn planner_skips_unsafe_ledger_paths() {
        let temp = TempDir::new().unwrap();
        let store = store(&temp);
        for path in [
            PathBuf::from("../outside"),
            PathBuf::from("/etc/passwd"),
            PathBuf::from(".git/config"),
            PathBuf::from("auth.json"),
            PathBuf::from(".env"),
            PathBuf::from("id_rsa"),
            PathBuf::from("mc/checkpoints/blob"),
            PathBuf::from("mc/sessions/session.jsonl"),
        ] {
            store.append_record(&captured_record(&store, path)).unwrap();
        }
        let plan = store.plan_rewind("session", temp.path(), 1);
        assert_eq!(plan.operations.len(), 8);
        assert!(plan.operations.iter().all(|operation| matches!(
            operation.kind,
            RestoreOperationKind::SkipUnavailable { .. }
        )));
        let text = format!("{plan:?}");
        assert!(!text.contains("../outside"));
        assert!(!text.contains("/etc/passwd"));
        assert!(!text.contains(".git/config"));
        assert_no_secret_path(&text);
    }

    #[test]
    fn planner_denies_case_insensitive_protected_paths() {
        let temp = TempDir::new().unwrap();
        let store = store(&temp);
        for path in [
            PathBuf::from(".GIT/config"),
            PathBuf::from("AUTH.JSON"),
            PathBuf::from(".ENV.production"),
            PathBuf::from("ID_RSA"),
            PathBuf::from("MC/CHECKPOINTS/blob"),
            PathBuf::from("MC/SESSIONS/session.jsonl"),
        ] {
            store.append_record(&captured_record(&store, path)).unwrap();
        }

        let plan = store.plan_rewind("session", temp.path(), 1);
        assert_eq!(plan.operations.len(), 6);
        assert!(plan.operations.iter().all(|operation| matches!(
            operation.kind,
            RestoreOperationKind::SkipUnavailable { .. }
        )));
        assert_no_secret_path(&format!("{plan:?}"));
    }

    #[cfg(unix)]
    #[test]
    fn non_utf8_components_are_compared_without_lossy_access_paths() {
        use std::os::unix::ffi::OsStringExt;

        let temp = TempDir::new().unwrap();
        let store = store(&temp);
        let non_utf8_name = OsString::from_vec(vec![b'c', 0x80]);
        let path = PathBuf::from(non_utf8_name).join("file.txt");
        assert!(safe_rewind_target_path(&store, temp.path(), &path).is_ok());
    }

    #[test]
    fn executor_skips_unsafe_operation_paths_before_io() {
        let temp = TempDir::new().unwrap();
        let store = store(&temp);
        let outside = temp
            .path()
            .parent()
            .unwrap()
            .join("outside-rewind-victim.txt");
        fs::write(&outside, "unchanged").unwrap();
        let pre = store.write_blob(b"before").unwrap();
        let expected_current = SnapshotBlobRef {
            sha256: sha256_hex(b"unchanged"),
            bytes: 9,
        };
        let plan = RestorePlan {
            session_id: "session".to_string(),
            target_turn: 1,
            latest_turn: Some(1),
            diagnostics: Vec::new(),
            operations: vec![
                RestoreOperation {
                    relative_path: PathBuf::from("../outside-rewind-victim.txt"),
                    kind: RestoreOperationKind::Restore {
                        pre: pre.clone(),
                        expected_current: expected_current.clone(),
                    },
                },
                RestoreOperation {
                    relative_path: PathBuf::from("/etc/passwd"),
                    kind: RestoreOperationKind::DeleteCreated {
                        expected_current: expected_current.clone(),
                    },
                },
                RestoreOperation {
                    relative_path: PathBuf::from("auth.json"),
                    kind: RestoreOperationKind::ResurrectDeleted { pre },
                },
            ],
        };
        let execution = store.execute_plan(&plan, temp.path());
        assert!(
            execution
                .results
                .iter()
                .all(|result| result.status == RestoreStatus::SkipUnavailable)
        );
        assert_eq!(fs::read_to_string(outside).unwrap(), "unchanged");
    }

    #[test]
    fn executor_reports_sanitized_filesystem_failure() {
        let temp = TempDir::new().unwrap();
        let context = context(&temp);
        let file = temp.path().join("safe.txt");
        fs::write(&file, "after").unwrap();
        capture_file_snapshot(
            &context,
            temp.path(),
            SnapshotTool::Edit,
            &file,
            Some(b"before"),
            Some(b"after"),
        )
        .unwrap();
        let plan = context.store.plan_rewind("session", temp.path(), 1);
        let execution = execute_plan_with_failure(&context.store, &plan, temp.path(), "write");
        let result = &execution.results[0];
        assert_eq!(result.status, RestoreStatus::SkipUnavailable);
        assert_eq!(result.reason.as_deref(), Some("write failed"));
        let payload = rewind_event_payload(&plan, &execution).to_string();
        assert!(payload.contains("write failed"));
        assert!(!payload.contains(temp.path().to_string_lossy().as_ref()));
        assert!(!payload.contains("before"));
        assert!(!fs::read_dir(temp.path()).unwrap().any(|entry| {
            entry
                .unwrap()
                .file_name()
                .to_string_lossy()
                .starts_with(".magi-rewind-")
        }));
    }

    #[test]
    fn executor_deletes_created_directly_and_keeps_post_blob() {
        let temp = TempDir::new().unwrap();
        let context = context(&temp);
        let file = temp.path().join("created.txt");
        fs::write(&file, "created").unwrap();
        capture_file_snapshot(
            &context,
            temp.path(),
            SnapshotTool::WriteFile,
            &file,
            None,
            Some(b"created"),
        )
        .unwrap();
        let plan = context.store.plan_rewind("session", temp.path(), 1);
        let expected = match &plan.operations[0].kind {
            RestoreOperationKind::DeleteCreated { expected_current } => expected_current.clone(),
            kind => panic!("expected delete-created operation, got {kind:?}"),
        };
        let execution = context.store.execute_plan(&plan, temp.path());
        assert_eq!(execution.results[0].status, RestoreStatus::DeletedCreated);
        assert!(!file.exists());
        assert_eq!(context.store.read_blob(&expected).unwrap(), b"created");
    }

    #[cfg(unix)]
    #[test]
    fn executor_rejects_symlink_parent_for_every_operation() {
        let temp = TempDir::new().unwrap();
        let outside = TempDir::new().unwrap();
        let store = store(&temp);
        let outside_target = outside.path().join("target.txt");
        fs::write(&outside_target, "outside").unwrap();
        symlink(outside.path(), temp.path().join("link")).unwrap();
        let plan = all_operation_plan(
            "link/target.txt",
            store.write_blob(b"before").unwrap(),
            store.write_blob(b"after").unwrap(),
        );

        let execution = store.execute_plan(&plan, temp.path());
        assert_eq!(execution.results.len(), 3);
        assert!(
            execution
                .results
                .iter()
                .all(|result| result.status == RestoreStatus::SkipUnavailable)
        );
        assert_eq!(fs::read_to_string(&outside_target).unwrap(), "outside");
    }

    #[cfg(unix)]
    #[test]
    fn executor_rejects_final_symlink_for_every_operation() {
        let temp = TempDir::new().unwrap();
        let outside = TempDir::new().unwrap();
        let store = store(&temp);
        let parent = temp.path().join("parent");
        fs::create_dir(&parent).unwrap();
        let outside_target = outside.path().join("target.txt");
        fs::write(&outside_target, "outside").unwrap();
        symlink(&outside_target, parent.join("target.txt")).unwrap();
        let plan = all_operation_plan(
            "parent/target.txt",
            store.write_blob(b"before").unwrap(),
            store.write_blob(b"after").unwrap(),
        );

        let execution = store.execute_plan(&plan, temp.path());
        assert_eq!(execution.results.len(), 3);
        assert!(
            execution
                .results
                .iter()
                .all(|result| result.status == RestoreStatus::SkipUnavailable)
        );
        assert_eq!(fs::read_to_string(&outside_target).unwrap(), "outside");
        assert!(
            fs::symlink_metadata(parent.join("target.txt"))
                .unwrap()
                .is_symlink()
        );
    }

    #[cfg(unix)]
    #[test]
    fn executor_uses_held_parent_after_ambient_parent_replacement() {
        let temp = TempDir::new().unwrap();
        let outside = TempDir::new().unwrap();
        let store = store(&temp);
        let cwd = temp.path().join("workspace");
        let parent = cwd.join("parent");
        let moved_parent = cwd.join("parent-moved");
        fs::create_dir_all(&parent).unwrap();
        let target = parent.join("target.txt");
        fs::write(&target, "after").unwrap();
        let outside_target = outside.path().join("target.txt");
        fs::write(&outside_target, "outside").unwrap();
        let plan = RestorePlan {
            session_id: "session".to_string(),
            target_turn: 1,
            latest_turn: Some(1),
            diagnostics: Vec::new(),
            operations: vec![RestoreOperation {
                relative_path: PathBuf::from("parent/target.txt"),
                kind: RestoreOperationKind::Restore {
                    pre: store.write_blob(b"before").unwrap(),
                    expected_current: store.write_blob(b"after").unwrap(),
                },
            }],
        };
        let hook_called = std::cell::Cell::new(false);
        let execution = execute_plan_with_parent_hook(&store, &plan, &cwd, || {
            hook_called.set(true);
            fs::rename(&parent, &moved_parent).unwrap();
            symlink(outside.path(), &parent).unwrap();
        });

        assert!(hook_called.get());
        assert_eq!(execution.results[0].status, RestoreStatus::Restored);
        assert_eq!(
            fs::read_to_string(moved_parent.join("target.txt")).unwrap(),
            "before"
        );
        assert_eq!(fs::read_to_string(&outside_target).unwrap(), "outside");
        assert!(fs::symlink_metadata(&parent).unwrap().is_symlink());
    }

    #[test]
    fn replacement_after_initial_hash_is_restored_as_conflict() {
        {
            let temp = TempDir::new().unwrap();
            let store = store(&temp);
            let file = temp.path().join("restore.txt");
            fs::write(&file, "after").unwrap();
            let plan = RestorePlan {
                session_id: "session".to_string(),
                target_turn: 1,
                latest_turn: Some(1),
                diagnostics: Vec::new(),
                operations: vec![RestoreOperation {
                    relative_path: PathBuf::from("restore.txt"),
                    kind: RestoreOperationKind::Restore {
                        pre: store.write_blob(b"before").unwrap(),
                        expected_current: store.write_blob(b"after").unwrap(),
                    },
                }],
            };
            let execution = execute_plan_with_target_hook(&store, &plan, temp.path(), || {
                fs::write(&file, "replacement").unwrap();
            });
            assert_eq!(execution.results[0].status, RestoreStatus::SkipConflict);
            assert_eq!(fs::read_to_string(&file).unwrap(), "replacement");
        }
        {
            let temp = TempDir::new().unwrap();
            let store = store(&temp);
            let file = temp.path().join("delete.txt");
            fs::write(&file, "created").unwrap();
            let plan = RestorePlan {
                session_id: "session".to_string(),
                target_turn: 1,
                latest_turn: Some(1),
                diagnostics: Vec::new(),
                operations: vec![RestoreOperation {
                    relative_path: PathBuf::from("delete.txt"),
                    kind: RestoreOperationKind::DeleteCreated {
                        expected_current: SnapshotBlobRef {
                            sha256: sha256_hex(b"created"),
                            bytes: 7,
                        },
                    },
                }],
            };
            let execution = execute_plan_with_target_hook(&store, &plan, temp.path(), || {
                fs::write(&file, "replacement").unwrap();
            });
            assert_eq!(execution.results[0].status, RestoreStatus::SkipConflict);
            assert_eq!(fs::read_to_string(&file).unwrap(), "replacement");
        }
    }

    #[test]
    fn replacement_after_quarantine_is_never_overwritten_or_deleted() {
        {
            let temp = TempDir::new().unwrap();
            let store = store(&temp);
            let file = temp.path().join("restore.txt");
            fs::write(&file, "after").unwrap();
            let plan = RestorePlan {
                session_id: "session".to_string(),
                target_turn: 1,
                latest_turn: Some(1),
                diagnostics: Vec::new(),
                operations: vec![RestoreOperation {
                    relative_path: PathBuf::from("restore.txt"),
                    kind: RestoreOperationKind::Restore {
                        pre: store.write_blob(b"before").unwrap(),
                        expected_current: store.write_blob(b"after").unwrap(),
                    },
                }],
            };
            let execution = execute_plan_with_quarantine_hook(&store, &plan, temp.path(), || {
                fs::write(&file, "replacement").unwrap();
            });
            assert_eq!(execution.results[0].status, RestoreStatus::SkipConflict);
            assert_eq!(fs::read_to_string(&file).unwrap(), "replacement");
        }
        {
            let temp = TempDir::new().unwrap();
            let store = store(&temp);
            let file = temp.path().join("delete.txt");
            fs::write(&file, "created").unwrap();
            let plan = RestorePlan {
                session_id: "session".to_string(),
                target_turn: 1,
                latest_turn: Some(1),
                diagnostics: Vec::new(),
                operations: vec![RestoreOperation {
                    relative_path: PathBuf::from("delete.txt"),
                    kind: RestoreOperationKind::DeleteCreated {
                        expected_current: SnapshotBlobRef {
                            sha256: sha256_hex(b"created"),
                            bytes: 7,
                        },
                    },
                }],
            };
            let execution = execute_plan_with_quarantine_hook(&store, &plan, temp.path(), || {
                fs::write(&file, "replacement").unwrap();
            });
            assert_eq!(execution.results[0].status, RestoreStatus::DeletedCreated);
            assert_eq!(fs::read_to_string(&file).unwrap(), "replacement");
        }
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    #[test]
    fn quarantine_destination_race_preserves_source_and_occupant() {
        let temp = TempDir::new().unwrap();
        let parent = CapDir::open_ambient_dir(temp.path(), ambient_authority()).unwrap();
        let source_path = temp.path().join("source.txt");
        fs::write(&source_path, b"source").unwrap();
        let (temp_name, temp_hash) =
            prepare_rewind_temp_file(&parent, b"temporary", None, false).unwrap();
        let temp_cleanup = TempFileCleanupGuard::new(&parent, temp_name, temp_hash);
        let occupied_path = std::cell::RefCell::new(None);

        let result = move_entry_to_quarantine_with_attempts(
            &parent,
            OsStr::new("source.txt"),
            "quarantine",
            1,
            |destination| {
                let path = temp.path().join(Path::new(destination));
                fs::write(&path, b"occupant").unwrap();
                *occupied_path.borrow_mut() = Some(path);
            },
        );
        let error = result.expect_err("occupied destination must be a conflict");
        assert_eq!(error.kind(), ErrorKind::AlreadyExists);

        // Cleanup of the failed move may remove only its own temporary file.
        drop(temp_cleanup);
        let occupied_path = occupied_path.into_inner().unwrap();
        assert_eq!(fs::read(&source_path).unwrap(), b"source");
        assert_eq!(fs::read(&occupied_path).unwrap(), b"occupant");
    }

    #[test]
    fn executor_reports_each_injected_filesystem_failure_phase() {
        let temp = TempDir::new().unwrap();
        let context = context(&temp);
        let file = temp.path().join("safe.txt");
        fs::write(&file, "after").unwrap();
        capture_file_snapshot(
            &context,
            temp.path(),
            SnapshotTool::Edit,
            &file,
            Some(b"before"),
            Some(b"after"),
        )
        .unwrap();
        let plan = context.store.plan_rewind("session", temp.path(), 1);
        for (phase, reason) in [
            ("hash", "hash check failed"),
            ("blob read", "blob read failed"),
            ("write", "write failed"),
        ] {
            let result =
                &execute_plan_with_failure(&context.store, &plan, temp.path(), phase).results[0];
            assert_eq!(result.status, RestoreStatus::SkipUnavailable);
            assert_eq!(result.reason.as_deref(), Some(reason));
        }
    }

    #[test]
    fn executor_reports_delete_failure() {
        let temp = TempDir::new().unwrap();
        let store = store(&temp);
        let file = temp.path().join("created.txt");
        fs::write(&file, "created").unwrap();
        let plan = RestorePlan {
            session_id: "session".to_string(),
            target_turn: 1,
            latest_turn: Some(1),
            diagnostics: Vec::new(),
            operations: vec![RestoreOperation {
                relative_path: PathBuf::from("created.txt"),
                kind: RestoreOperationKind::DeleteCreated {
                    expected_current: SnapshotBlobRef {
                        sha256: sha256_hex(b"created"),
                        bytes: 7,
                    },
                },
            }],
        };
        let result = &execute_plan_with_failure(&store, &plan, temp.path(), "delete").results[0];
        assert_eq!(result.status, RestoreStatus::SkipUnavailable);
        assert_eq!(result.reason.as_deref(), Some("delete failed"));
    }

    #[test]
    fn ancestor_sync_failure_is_precommit_and_does_not_create_target() {
        let temp = TempDir::new().unwrap();
        let store = store(&temp);
        let plan = RestorePlan {
            session_id: "session".to_string(),
            target_turn: 1,
            latest_turn: Some(1),
            diagnostics: Vec::new(),
            operations: vec![RestoreOperation {
                relative_path: PathBuf::from("new/nested/file.txt"),
                kind: RestoreOperationKind::ResurrectDeleted {
                    pre: store.write_blob(b"recovered").unwrap(),
                },
            }],
        };

        let execution = execute_plan_with_failure(&store, &plan, temp.path(), "ancestor sync");
        assert_eq!(execution.results[0].status, RestoreStatus::SkipUnavailable);
        assert_eq!(execution.results[0].reason.as_deref(), Some("write failed"));
        assert!(!temp.path().join("new").exists());
    }

    #[test]
    fn final_sync_failure_reports_committed_mutation() {
        let temp = TempDir::new().unwrap();
        let context = context(&temp);
        let file = temp.path().join("safe.txt");
        fs::write(&file, "after").unwrap();
        capture_file_snapshot(
            &context,
            temp.path(),
            SnapshotTool::Edit,
            &file,
            Some(b"before"),
            Some(b"after"),
        )
        .unwrap();
        let plan = context.store.plan_rewind("session", temp.path(), 1);

        let execution = execute_plan_with_failure(&context.store, &plan, temp.path(), "final sync");
        let result = &execution.results[0];
        assert_eq!(result.status, RestoreStatus::Restored);
        assert_eq!(result.reason.as_deref(), Some(COMMITTED_UNDURABLE_WARNING));
        assert_eq!(fs::read_to_string(file).unwrap(), "before");
    }

    #[test]
    fn rewind_payload_omits_reason_for_success_and_bounds_diagnostics() {
        let plan = RestorePlan {
            session_id: "session".to_string(),
            target_turn: 1,
            latest_turn: Some(1),
            diagnostics: Vec::new(),
            operations: Vec::new(),
        };
        let execution = RewindExecution {
            results: vec![RestoreOperationResult {
                relative_path: PathBuf::from("safe.txt"),
                status: RestoreStatus::Restored,
                reason: None,
            }],
        };
        let item = &rewind_event_payload(&plan, &execution)["paths"][0];
        assert_eq!(item, &json!({"path":"safe.txt", "status":"restored"}));
        assert_eq!(sanitized_diagnostic("x".repeat(600)).chars().count(), 500);
    }

    #[test]
    fn secret_paths_are_redacted_from_changes_payload_and_replay() {
        let temp = TempDir::new().unwrap();
        let store = store(&temp);
        for path in ["auth.json", ".env", "id_rsa"] {
            store
                .append_record(&SnapshotLedgerRecord::new(FileSnapshotEvent {
                    session_id: "session".to_string(),
                    user_turn: 1,
                    tool: SnapshotTool::WriteFile,
                    cwd: temp.path().to_path_buf(),
                    relative_path: PathBuf::from(path),
                    pre: None,
                    post: None,
                    status: SnapshotCaptureStatus::Excluded {
                        reason: SnapshotExclusionReason::SecretPath,
                    },
                }))
                .unwrap();
        }
        let changes = store.changes("session");
        assert_no_secret_path(&format!("{changes:?}"));
        assert!(
            changes.turns[0]
                .changes
                .iter()
                .all(|change| change.relative_path == Path::new(REDACTED_PATH))
        );
        let plan = RestorePlan {
            session_id: "session".to_string(),
            target_turn: 1,
            latest_turn: Some(1),
            diagnostics: Vec::new(),
            operations: Vec::new(),
        };
        let execution = RewindExecution {
            results: ["auth.json", ".env", "id_rsa"]
                .into_iter()
                .map(|path| RestoreOperationResult {
                    relative_path: PathBuf::from(path),
                    status: RestoreStatus::SkipUnavailable,
                    reason: None,
                })
                .collect(),
        };
        let payload = rewind_event_payload(&plan, &execution);
        assert_no_secret_path(&payload.to_string());
        assert_no_secret_path(&replay_rewind_context(&payload).unwrap());
    }

    #[test]
    fn executor_rechecks_hash_before_restore() {
        let temp = TempDir::new().unwrap();
        let context = context(&temp);
        let file = temp.path().join("file.txt");
        fs::write(&file, "after").unwrap();
        capture_file_snapshot(
            &context,
            temp.path(),
            SnapshotTool::Edit,
            &file,
            Some(b"before"),
            Some(b"after"),
        )
        .unwrap();
        let plan = context.store.plan_rewind("session", temp.path(), 1);
        fs::write(&file, "changed").unwrap();
        let execution = context.store.execute_plan(&plan, temp.path());
        assert_eq!(execution.results[0].status, RestoreStatus::SkipConflict);
    }

    #[test]
    fn executor_skips_denied_path() {
        let temp = TempDir::new().unwrap();
        let store = store(&temp);
        let plan = RestorePlan {
            session_id: "session".to_string(),
            target_turn: 1,
            latest_turn: Some(1),
            diagnostics: Vec::new(),
            operations: vec![RestoreOperation {
                relative_path: PathBuf::from(".git/config"),
                kind: RestoreOperationKind::SkipUnavailable {
                    reason: "denied".to_string(),
                },
            }],
        };
        let execution = store.execute_plan(&plan, temp.path());
        assert_eq!(execution.results[0].status, RestoreStatus::SkipUnavailable);
    }
}