//! Bounded disk IO for source projects. Compilation stays in `source_project`.
//!
//! Saving checks every captured dependency, stages exact bytes, and publishes a
//! recovery journal before replacing files (the entry is last). This provides
//! recoverability, not multi-file atomic visibility or a cross-process CAS.
//! Symbolic links/junctions are rejected inside the project root.
//! Recovery metadata admits at most 4096 files and 32 MiB of JSON; source bytes
//! obey the caller's aggregate limits. Reserved `.arcadia-save-*` directories
//! may remain after interruption during staging/completed cleanup. They contain
//! no pending transaction unless named `.arcadia-save-ENTRY.journal`. Windows
//! has no portable directory fsync; this is not a power-loss durability promise.
//! Journal v2 records whether writes may have begun with a flushed marker.
//! Recovery of a v2 journal without that marker discards only journal metadata;
//! v1 journals retain their prior rollback rules.
//! Journal publication refuses replacement on Windows and Linux, including a
//! competing empty directory. Linux requires kernel/filesystem support for
//! `renameat2(RENAME_NOREPLACE)` and fails without a fallback when unavailable.
//! Other operating systems retain standard rename semantics.
use crate::{language, source_project::*};
use serde::{Deserialize, Serialize};
use std::{
    collections::{BTreeMap, BTreeSet},
    fs::{self, File},
    io::{self, Read, Write},
    path::{Path, PathBuf},
};
use thiserror::Error;

const MAX_PATH_BYTES: usize = 4096;
const MAX_JOURNAL_BYTES: usize = 32 * 1024 * 1024;
const MAX_JOURNAL_FILES: usize = 4096;
const WRITE_STARTED: &str = "write-started";
const WRITE_STARTED_BYTES: &[u8] = b"Arcadia journal v2: model writes may have started\n";

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DiskExpectation {
    Missing,
    Bytes(Vec<u8>),
}

/// `bundle` retains detached and invalid buffers. Compile `reachable_bundle()`;
/// use `diagnostics` to explain files that discovery could not read or parse.
/// Diagnostic spans describe `bundle` bytes, including before UUID stabilization.
/// Expectations always describe disk bytes before UUID stabilization.
#[derive(Debug, Clone)]
pub struct DiskProject {
    pub root: PathBuf,
    pub bundle: SourceBundle,
    pub expectations: BTreeMap<LogicalPath, DiskExpectation>,
    pub reachable: BTreeSet<LogicalPath>,
    pub diagnostics: Vec<SourceDiagnostic>,
}

impl DiskProject {
    pub fn reachable_bundle(&self) -> SourceBundle {
        SourceBundle {
            entry: self.bundle.entry.clone(),
            files: self
                .bundle
                .files
                .iter()
                .filter(|(path, _)| self.reachable.contains(*path))
                .map(|(path, text)| (path.clone(), text.clone()))
                .collect(),
        }
    }
}

#[derive(Debug, Error)]
pub enum SourceFileError {
    #[error("Source file operation failed at {path}: {source}")]
    Io {
        path: PathBuf,
        #[source]
        source: io::Error,
    },
    #[error("Unsafe source path: {0}")]
    Path(String),
    #[error("Source project admission limit: {0}")]
    Limit(String),
    #[error("Source files changed outside Arcadia Studio: {0:?}")]
    Conflict(Vec<LogicalPath>),
    #[error("No disk expectation was established for {0}. Review the destination before saving.")]
    UnknownExpectation(LogicalPath),
    #[error("Invalid source project: {0}")]
    Invalid(String),
    #[error("A source save needs recovery at {0}. Recover it before saving again.")]
    RecoveryRequired(PathBuf),
    #[error(
        "Save failed ({save}); recovery could not finish ({recovery}). The journal remains at {journal}."
    )]
    RecoveryFailed {
        save: String,
        recovery: String,
        journal: PathBuf,
    },
}

fn io_at(path: &Path, source: io::Error) -> SourceFileError {
    SourceFileError::Io {
        path: path.into(),
        source,
    }
}

fn root_entry(entry: &Path) -> Result<(PathBuf, String), SourceFileError> {
    let name = entry
        .file_name()
        .and_then(|name| name.to_str())
        .ok_or_else(|| SourceFileError::Path("Entry needs a UTF-8 file name.".into()))?;
    check_logical(name)?;
    let parent = entry
        .parent()
        .filter(|path| !path.as_os_str().is_empty())
        .unwrap_or(Path::new("."));
    let root = parent
        .canonicalize()
        .map_err(|error| io_at(parent, error))?;
    if !root.is_dir() {
        return Err(SourceFileError::Path(
            "Project root is not a directory.".into(),
        ));
    }
    Ok((root, name.into()))
}

fn check_logical(path: &str) -> Result<(), SourceFileError> {
    if path.len() > MAX_PATH_BYTES
        || resolve_include("", path).as_deref() != Ok(path)
        || path
            .split('/')
            .any(|part| part.to_ascii_lowercase().starts_with(".arcadia-save-"))
    {
        return Err(SourceFileError::Path(format!(
            "Non-canonical, reserved or overlong path '{path}'."
        )));
    }
    Ok(())
}

/// Reject links rather than quietly replacing a link instead of its target.
/// Existing parents are checked even for a new destination.
fn disk_path(root: &Path, logical: &str) -> Result<PathBuf, SourceFileError> {
    check_logical(logical)?;
    let canonical_root = root.canonicalize().map_err(|error| io_at(root, error))?;
    if canonical_root != root {
        return Err(SourceFileError::Path(
            "Project root changed physical location.".into(),
        ));
    }
    let mut current = root.to_path_buf();
    for component in logical.split('/') {
        current.push(component);
        match fs::symlink_metadata(&current) {
            Ok(metadata) => {
                if metadata.file_type().is_symlink() {
                    return Err(SourceFileError::Path(format!(
                        "Links are not source project files: '{logical}'."
                    )));
                }
                let physical = current
                    .canonicalize()
                    .map_err(|error| io_at(&current, error))?;
                if !physical.starts_with(root) {
                    return Err(SourceFileError::Path(format!(
                        "'{logical}' escapes the project root."
                    )));
                }
            }
            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
            Err(error) => return Err(io_at(&current, error)),
        }
    }
    Ok(current)
}

// Check before opening: opening a FIFO for reading can wait forever. The
// post-open check below still verifies the opened object. As with disk_path,
// this does not promise a CAS against concurrent path replacement.
fn regular_file_metadata(path: &Path) -> Result<fs::Metadata, SourceFileError> {
    let metadata = fs::symlink_metadata(path).map_err(|error| io_at(path, error))?;
    if !metadata.is_file() {
        return Err(SourceFileError::Path(format!(
            "{} is not a regular file.",
            path.display()
        )));
    }
    Ok(metadata)
}

fn read_bounded(path: &Path, maximum: usize) -> Result<Vec<u8>, SourceFileError> {
    regular_file_metadata(path)?;
    let file = File::open(path).map_err(|error| io_at(path, error))?;
    if !file
        .metadata()
        .map_err(|error| io_at(path, error))?
        .is_file()
    {
        return Err(SourceFileError::Path(format!(
            "{} is not a regular file.",
            path.display()
        )));
    }
    let mut bytes = Vec::new();
    file.take((maximum as u64).saturating_add(1))
        .read_to_end(&mut bytes)
        .map_err(|error| io_at(path, error))?;
    if bytes.len() > maximum {
        return Err(SourceFileError::Limit(format!(
            "{} exceeds the remaining {maximum} byte budget.",
            path.display()
        )));
    }
    Ok(bytes)
}

fn located(
    bundle: &SourceBundle,
    path: &str,
    span: language::Span,
    code: &'static str,
    message: impl Into<String>,
) -> SourceDiagnostic {
    let source = bundle.files.get(path).map_or("", String::as_str);
    let mut diagnostic = language::diagnostic(
        source,
        language::Error {
            code,
            offset: span.start.min(source.len()),
            message: message.into(),
        },
    );
    diagnostic.span.end = span.end.min(source.len()).max(diagnostic.span.start);
    SourceDiagnostic {
        path: path.into(),
        diagnostic,
        related: vec![],
    }
}

fn check_snapshot(snapshot: &DiskProject, limits: ProjectLimits) -> Result<(), SourceFileError> {
    if snapshot.bundle.entry.contains('/') {
        return Err(SourceFileError::Path(
            "Entry must be at the source root.".into(),
        ));
    }
    let mut paths = BTreeSet::new();
    let mut folded = BTreeMap::new();
    for path in std::iter::once(&snapshot.bundle.entry)
        .chain(snapshot.bundle.files.keys())
        .chain(snapshot.expectations.keys())
    {
        check_logical(path)?;
        paths.insert(path);
        if let Some(previous) = folded.insert(path.to_lowercase(), path)
            && previous != path
        {
            return Err(SourceFileError::Path(format!(
                "Case-colliding paths '{previous}' and '{path}'."
            )));
        }
    }
    if paths.len() > limits.max_files {
        return Err(SourceFileError::Limit(format!(
            "More than {} files.",
            limits.max_files
        )));
    }
    for total in [
        snapshot
            .bundle
            .files
            .values()
            .fold(0usize, |sum, text| sum.saturating_add(text.len())),
        snapshot
            .expectations
            .values()
            .fold(0usize, |sum, expected| {
                sum.saturating_add(match expected {
                    DiskExpectation::Missing => 0,
                    DiskExpectation::Bytes(bytes) => bytes.len(),
                })
            }),
    ] {
        if total > limits.model.max_source_bytes {
            return Err(SourceFileError::Limit(
                "Combined source bytes exceed the project budget.".into(),
            ));
        }
    }
    Ok(())
}

fn check_physical_aliases<'a>(
    root: &Path,
    paths: impl IntoIterator<Item = &'a String>,
) -> Result<(), SourceFileError> {
    let mut handles: Vec<(&str, same_file::Handle)> = Vec::new();
    let mut visited = BTreeSet::new();
    for logical in paths {
        if !visited.insert(logical) {
            continue;
        }
        let path = disk_path(root, logical)?;
        match regular_file_metadata(&path) {
            Ok(_) => {}
            Err(SourceFileError::Io { source, .. }) if source.kind() == io::ErrorKind::NotFound => {
                continue;
            }
            Err(error) => return Err(error),
        }
        match same_file::Handle::from_path(&path) {
            Ok(handle) => {
                if let Some((previous, _)) = handles.iter().find(|(_, old)| *old == handle) {
                    return Err(SourceFileError::Path(format!(
                        "'{previous}' and '{logical}' refer to the same physical file."
                    )));
                }
                handles.push((logical, handle));
            }
            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
            Err(error) => return Err(io_at(&path, error)),
        }
    }
    Ok(())
}

/// Load original bytes, including invalid UTF-8/syntax diagnostics. Missing
/// includes are diagnostics, not a reason to discard editable files.
pub fn load_project(entry: &Path, limits: ProjectLimits) -> Result<DiskProject, SourceFileError> {
    let (root, entry) = root_entry(entry)?;
    let snapshot = DiskProject {
        root,
        bundle: SourceBundle {
            entry,
            files: BTreeMap::new(),
        },
        expectations: BTreeMap::new(),
        reachable: BTreeSet::new(),
        diagnostics: vec![],
    };
    discover_missing(&snapshot, &BTreeMap::new(), limits)
}

/// Establish an unsaved project destination. An existing entry is a collision,
/// never implicit permission to overwrite it. Parent directory must exist.
pub fn new_project(entry: &Path, limits: ProjectLimits) -> Result<DiskProject, SourceFileError> {
    let (root, entry) = root_entry(entry)?;
    let mut snapshot = DiskProject {
        root,
        bundle: SourceBundle {
            entry: entry.clone(),
            files: BTreeMap::new(),
        },
        expectations: BTreeMap::new(),
        reachable: BTreeSet::from([entry.clone()]),
        diagnostics: vec![],
    };
    expect_new_file(&mut snapshot, &entry, limits)?;
    Ok(snapshot)
}

pub fn expect_new_file(
    snapshot: &mut DiskProject,
    path: &str,
    limits: ProjectLimits,
) -> Result<(), SourceFileError> {
    let destination = disk_path(&snapshot.root, path)?;
    match fs::symlink_metadata(&destination) {
        Ok(_) => return Err(SourceFileError::Conflict(vec![path.into()])),
        Err(error) if error.kind() == io::ErrorKind::NotFound => {}
        Err(error) => return Err(io_at(&destination, error)),
    }
    let mut candidate = snapshot.clone();
    candidate
        .expectations
        .insert(path.into(), DiskExpectation::Missing);
    check_snapshot(&candidate, limits)?;
    snapshot.expectations = candidate.expectations;
    Ok(())
}

/// Merge editor overlays and discover only files absent from retained buffers.
/// Existing drafts (including detached ones) and their disk baselines survive.
pub fn discover_missing(
    snapshot: &DiskProject,
    overlays: &BTreeMap<LogicalPath, String>,
    limits: ProjectLimits,
) -> Result<DiskProject, SourceFileError> {
    let mut result = snapshot.clone();
    result.bundle.files.extend(overlays.clone());
    result.reachable.clear();
    result.diagnostics.clear();
    check_snapshot(&result, limits)?;
    let mut stack = vec![(result.bundle.entry.clone(), 0usize, None::<SourceLocation>)];
    let mut edges = 0usize;
    let mut attempted = BTreeSet::new();
    while let Some((path, depth, origin)) = stack.pop() {
        let at = origin.unwrap_or(SourceLocation {
            path: path.clone(),
            span: language::Span::default(),
        });
        if depth > limits.max_include_depth {
            result.diagnostics.push(located(
                &result.bundle,
                &at.path,
                at.span,
                "project_limit",
                "Include depth exceeds the project budget.",
            ));
            continue;
        }
        result.reachable.insert(path.clone());
        if !attempted.insert(path.clone()) {
            continue;
        }
        if !result.bundle.files.contains_key(&path) {
            let unique = result
                .expectations
                .keys()
                .chain(result.bundle.files.keys())
                .chain(std::iter::once(&path))
                .collect::<BTreeSet<_>>();
            if unique.len() > limits.max_files {
                result.diagnostics.push(located(
                    &result.bundle,
                    &at.path,
                    at.span,
                    "project_limit",
                    "File count exceeds the project budget.",
                ));
                continue;
            }
            let used = result
                .bundle
                .files
                .values()
                .fold(0usize, |sum, text| sum.saturating_add(text.len()))
                .max(result.expectations.values().fold(0usize, |sum, expected| {
                    sum.saturating_add(match expected {
                        DiskExpectation::Missing => 0,
                        DiskExpectation::Bytes(bytes) => bytes.len(),
                    })
                }));
            let loaded = disk_path(&result.root, &path).and_then(|physical| {
                read_bounded(
                    &physical,
                    limits.model.max_source_bytes.saturating_sub(used),
                )
            });
            match loaded {
                Ok(bytes) => {
                    result
                        .expectations
                        .insert(path.clone(), DiskExpectation::Bytes(bytes.clone()));
                    match String::from_utf8(bytes) {
                        Ok(text) => {
                            result.bundle.files.insert(path.clone(), text);
                        }
                        Err(_) => {
                            result.diagnostics.push(located(
                                &result.bundle,
                                &at.path,
                                at.span,
                                "source_encoding",
                                format!("'{path}' is not UTF-8 source."),
                            ));
                            continue;
                        }
                    }
                }
                Err(error) => {
                    if matches!(&error, SourceFileError::Io { source, .. } if source.kind() == io::ErrorKind::NotFound)
                    {
                        result
                            .expectations
                            .entry(path.clone())
                            .or_insert(DiskExpectation::Missing);
                    }
                    let code = match &error {
                        SourceFileError::Limit(_) => "source_limit",
                        SourceFileError::Path(_) => "invalid_path",
                        _ => "missing_include",
                    };
                    result.diagnostics.push(located(
                        &result.bundle,
                        &at.path,
                        at.span,
                        code,
                        format!("Cannot load '{path}': {error}"),
                    ));
                    continue;
                }
            }
        }
        match inspect_includes(
            &path,
            &result.bundle.files[&path],
            path == result.bundle.entry,
            limits,
        ) {
            Ok(includes) => {
                edges = edges.saturating_add(includes.len());
                if edges > limits.max_include_edges {
                    result.diagnostics.push(located(
                        &result.bundle,
                        &path,
                        language::Span::default(),
                        "project_limit",
                        "Include edges exceed the project budget.",
                    ));
                    continue;
                }
                stack.extend(
                    includes
                        .into_iter()
                        .rev()
                        .map(|include| (include.target, depth + 1, Some(include.location))),
                );
            }
            Err(diagnostics) => result.diagnostics.extend(diagnostics),
        }
    }
    check_snapshot(&result, limits)?;
    check_physical_aliases(
        &result.root,
        result.expectations.keys().chain(result.bundle.files.keys()),
    )?;
    if result.diagnostics.is_empty() {
        // Discovery retains raw editor bytes. Compilation may insert UUIDs or
        // rewrite older syntax, so its returned locations belong to a different
        // bundle. Lint still performs semantic migration and model validation;
        // Open/Apply separately compile and install the rewritten source.
        result.diagnostics = lint_bundle(&result.reachable_bundle(), limits);
    }
    let pending = journal_path(&result.root, &result.bundle.entry);
    if fs::symlink_metadata(&pending).is_ok() {
        result.diagnostics.push(located(
            &result.bundle,
            &result.bundle.entry,
            language::Span::default(),
            "pending_save",
            format!(
                "A save was interrupted. Review and recover {} before saving this project.",
                pending.display()
            ),
        ));
    }
    Ok(result)
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Journal {
    version: u8,
    entry: String,
    #[serde(deserialize_with = "journal_files")]
    files: Vec<JournalFile>,
}

fn journal_files<'de, D: serde::Deserializer<'de>>(
    deserializer: D,
) -> Result<Vec<JournalFile>, D::Error> {
    struct Files;
    impl<'de> serde::de::Visitor<'de> for Files {
        type Value = Vec<JournalFile>;
        fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            formatter.write_str("at most 4096 recovery entries")
        }
        fn visit_seq<A: serde::de::SeqAccess<'de>>(
            self,
            mut sequence: A,
        ) -> Result<Self::Value, A::Error> {
            let mut files = Vec::new();
            while let Some(file) = sequence.next_element()? {
                if files.len() >= MAX_JOURNAL_FILES {
                    return Err(serde::de::Error::custom("Recovery entry budget exceeded."));
                }
                files.push(file);
            }
            Ok(files)
        }
    }
    deserializer.deserialize_seq(Files)
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct JournalFile {
    path: String,
    before: bool,
    write: bool,
}

fn journal_path(root: &Path, entry: &str) -> PathBuf {
    root.join(format!(".arcadia-save-{entry}.journal"))
}
fn staged_path(journal: &Path, index: usize, suffix: &str) -> PathBuf {
    journal.join(format!("{index}.{suffix}"))
}

fn publish_journal(stage: &Path, journal: &Path) -> Result<(), SourceFileError> {
    #[cfg(windows)]
    {
        publish_journal_with_retry(
            stage,
            journal,
            rename_journal_without_replacement,
            std::thread::sleep,
        )
    }
    #[cfg(target_os = "linux")]
    {
        rename_journal_without_replacement(stage, journal).map_err(|error| io_at(journal, error))
    }
    // Other platforms retain their existing rename behavior; the native
    // no-replacement publication guarantee is currently Windows/Linux only.
    #[cfg(not(any(windows, target_os = "linux")))]
    {
        fs::rename(stage, journal).map_err(|error| io_at(journal, error))
    }
}

#[cfg(target_os = "linux")]
fn rename_journal_without_replacement(stage: &Path, journal: &Path) -> io::Result<()> {
    use std::{ffi::CString, os::unix::ffi::OsStrExt};

    let path_bytes = |path: &Path| {
        CString::new(path.as_os_str().as_bytes()).map_err(|_| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                "Journal path contains an interior NUL.",
            )
        })
    };
    let (stage, journal) = (path_bytes(stage)?, path_bytes(journal)?);
    // Use libc's architecture-specific syscall number, avoiding a dependency on
    // the newer glibc renameat2 wrapper. Both NUL-terminated byte buffers remain
    // alive for the call. RENAME_NOREPLACE makes even an empty competing directory
    // an error. Unsupported kernels/filesystems fail closed: never fall back to
    // ordinary rename, whose replacement semantics would lose journal ownership.
    let result = unsafe {
        libc::syscall(
            libc::SYS_renameat2,
            libc::AT_FDCWD as libc::c_long,
            stage.as_ptr(),
            libc::AT_FDCWD as libc::c_long,
            journal.as_ptr(),
            libc::RENAME_NOREPLACE as libc::c_ulong,
        )
    };
    if result == -1 {
        Err(io::Error::last_os_error())
    } else {
        Ok(())
    }
}

#[cfg(windows)]
fn rename_journal_without_replacement(stage: &Path, journal: &Path) -> io::Result<()> {
    use std::os::windows::ffi::OsStrExt;
    #[link(name = "kernel32")]
    unsafe extern "system" {
        fn MoveFileExW(existing: *const u16, destination: *const u16, flags: u32) -> i32;
    }
    let wide = |path: &Path| -> io::Result<Vec<u16>> {
        let mut value: Vec<_> = path.as_os_str().encode_wide().collect();
        if value.contains(&0) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "Journal path contains an interior NUL.",
            ));
        }
        value.push(0);
        Ok(value)
    };
    let (stage, journal) = (wide(stage)?, wide(journal)?);
    // Both terminated UTF-16 buffers live through the call. Flags=0 requests
    // neither replacement, cross-volume copying nor deferred execution. Unlike
    // fs::rename on modern Windows, an existing empty directory is not replaced.
    if unsafe { MoveFileExW(stage.as_ptr(), journal.as_ptr(), 0) } == 0 {
        Err(io::Error::last_os_error())
    } else {
        Ok(())
    }
}

// Windows scanners may briefly open newly written stage files without delete
// sharing. Retry only publication, before any model file has been replaced.
// Ownership still transfers only after one successful rename. Never overwrite
// a journal that appeared while waiting, and retain the final OS error/path.
#[cfg(any(windows, test))]
fn publish_journal_with_retry(
    stage: &Path,
    journal: &Path,
    mut rename: impl FnMut(&Path, &Path) -> io::Result<()>,
    mut pause: impl FnMut(std::time::Duration),
) -> Result<(), SourceFileError> {
    let mut delays = [5, 10, 25, 50, 100, 200].into_iter();
    loop {
        match fs::symlink_metadata(journal) {
            Ok(_) => return Err(SourceFileError::RecoveryRequired(journal.into())),
            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
            Err(error) => return Err(io_at(journal, error)),
        }
        match rename(stage, journal) {
            Ok(()) => return Ok(()),
            Err(error) => {
                // ERROR_ACCESS_DENIED, ERROR_SHARING_VIOLATION, ERROR_LOCK_VIOLATION.
                // A permanent ACL failure is still returned after at most 390 ms.
                if matches!(error.raw_os_error(), Some(5 | 32 | 33))
                    && let Some(delay) = delays.next()
                {
                    pause(std::time::Duration::from_millis(delay));
                    continue;
                }
                return Err(io_at(journal, error));
            }
        }
    }
}

fn write_synced(path: &Path, bytes: &[u8]) -> Result<(), SourceFileError> {
    let mut file = File::options()
        .write(true)
        .create_new(true)
        .open(path)
        .map_err(|error| io_at(path, error))?;
    file.write_all(bytes)
        .and_then(|()| file.sync_all())
        .map_err(|error| io_at(path, error))
}

fn sync_directory(path: &Path) -> Result<(), SourceFileError> {
    #[cfg(unix)]
    File::open(path)
        .and_then(|file| file.sync_all())
        .map_err(|error| io_at(path, error))?;
    #[cfg(not(unix))]
    let _ = path;
    Ok(())
}

fn atomic_write(root: &Path, logical: &str, bytes: &[u8]) -> Result<(), SourceFileError> {
    let destination = disk_path(root, logical)?;
    let parent = destination.parent().unwrap();
    fs::create_dir_all(parent).map_err(|error| io_at(parent, error))?;
    // Recheck after directory creation; an external process may replace parents.
    disk_path(root, logical)?;
    let permissions = match regular_file_metadata(&destination) {
        Ok(metadata) => Some(metadata.permissions()),
        Err(SourceFileError::Io { source, .. }) if source.kind() == io::ErrorKind::NotFound => None,
        Err(error) => return Err(error),
    };
    let mut staged =
        tempfile::NamedTempFile::new_in(parent).map_err(|error| io_at(parent, error))?;
    staged
        .write_all(bytes)
        .map_err(|error| io_at(&destination, error))?;
    // Preserve the existing file's portable permissions through replacement
    // and recovery. New files retain the private temporary-file default.
    if let Some(permissions) = permissions {
        staged
            .as_file()
            .set_permissions(permissions)
            .map_err(|error| io_at(&destination, error))?;
    }
    staged
        .as_file()
        .sync_all()
        .map_err(|error| io_at(&destination, error))?;
    staged
        .persist(&destination)
        .map_err(|error| io_at(&destination, error.error))?;
    sync_directory(parent)
}

fn matches_disk(
    root: &Path,
    logical: &str,
    expected: &DiskExpectation,
) -> Result<bool, SourceFileError> {
    let path = disk_path(root, logical)?;
    match expected {
        DiskExpectation::Missing => match fs::symlink_metadata(&path) {
            Ok(_) => Ok(false),
            Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(true),
            Err(error) => Err(io_at(&path, error)),
        },
        DiskExpectation::Bytes(expected) => match read_bounded(&path, expected.len()) {
            Ok(actual) => Ok(actual == *expected),
            Err(SourceFileError::Limit(_)) => Ok(false),
            Err(SourceFileError::Io { source, .. }) if source.kind() == io::ErrorKind::NotFound => {
                Ok(false)
            }
            Err(error) => Err(error),
        },
    }
}

fn check_conflicts(
    root: &Path,
    expectations: &BTreeMap<LogicalPath, DiskExpectation>,
) -> Result<(), SourceFileError> {
    check_physical_aliases(root, expectations.keys())?;
    let mut conflicts = Vec::new();
    for (path, expected) in expectations {
        if !matches_disk(root, path, expected)? {
            conflicts.push(path.clone());
        }
    }
    if conflicts.is_empty() {
        Ok(())
    } else {
        Err(SourceFileError::Conflict(conflicts))
    }
}

/// Save an already accepted, identity-stabilized bundle. Every destination needs
/// an explicit original/Missing expectation. Detached files are never deleted.
pub fn save_project(
    snapshot: &DiskProject,
    compiled: &CompiledSourceProject,
    limits: ProjectLimits,
) -> Result<DiskProject, SourceFileError> {
    save_with_hook(snapshot, compiled, limits, |_| Ok(()))
}

fn save_with_hook(
    snapshot: &DiskProject,
    compiled: &CompiledSourceProject,
    limits: ProjectLimits,
    mut after_replace: impl FnMut(usize) -> io::Result<()>,
) -> Result<DiskProject, SourceFileError> {
    save_with_phases(snapshot, compiled, limits, |phase| match phase {
        SavePhase::AfterReplace(count) => after_replace(count),
        _ => Ok(()),
    })
}

enum SavePhase {
    BeforePublish,
    AfterPublish,
    AfterWriteMarker,
    AfterReplace(usize),
    BeforeRetire,
}

fn save_with_phases(
    snapshot: &DiskProject,
    compiled: &CompiledSourceProject,
    limits: ProjectLimits,
    mut phase: impl FnMut(SavePhase) -> io::Result<()>,
) -> Result<DiskProject, SourceFileError> {
    check_snapshot(snapshot, limits)?;
    let checked = compile_bundle(&compiled.bundle, limits).map_err(|diagnostics| {
        SourceFileError::Invalid(
            diagnostics
                .iter()
                .map(|d| d.diagnostic.to_string())
                .collect::<Vec<_>>()
                .join("\n"),
        )
    })?;
    if checked.bundle != compiled.bundle
        || checked.decoded.project != compiled.decoded.project
        || checked.index != compiled.index
        || snapshot.bundle.entry != compiled.bundle.entry
    {
        return Err(SourceFileError::Invalid(
            "Compiled model, source, index or entry is stale.".into(),
        ));
    }
    let compiled = &checked;
    for path in compiled.bundle.files.keys() {
        disk_path(&snapshot.root, path)?;
        if !snapshot.expectations.contains_key(path) {
            return Err(SourceFileError::UnknownExpectation(path.clone()));
        }
    }
    check_snapshot(&saved_snapshot(snapshot, compiled), limits)?;
    let journal_dir = journal_path(&snapshot.root, &snapshot.bundle.entry);
    if fs::symlink_metadata(&journal_dir).is_ok() {
        return Err(SourceFileError::RecoveryRequired(journal_dir));
    }
    check_conflicts(&snapshot.root, &snapshot.expectations)?;
    let mut entries = snapshot.expectations.iter().map(|(path, before)| JournalFile {
        path: path.clone(), before: matches!(before, DiskExpectation::Bytes(_)),
        write: compiled.bundle.files.get(path).is_some_and(|after| !matches!(before, DiskExpectation::Bytes(bytes) if bytes == after.as_bytes())),
    }).collect::<Vec<_>>();
    // Stable dependency order, with the project entry committed last.
    entries.sort_by_key(|file| (file.path == snapshot.bundle.entry, file.path.clone()));
    let journal = Journal {
        version: 2,
        entry: snapshot.bundle.entry.clone(),
        files: entries,
    };
    if journal.files.len() > MAX_JOURNAL_FILES {
        return Err(SourceFileError::Limit(
            "Recovery is limited to 4096 files.".into(),
        ));
    }
    if !journal.files.iter().any(|file| file.write) {
        return Ok(saved_snapshot(snapshot, compiled));
    }
    let metadata = serde_json::to_vec(&journal)
        .map_err(|error| SourceFileError::Invalid(error.to_string()))?;
    if metadata.len() > MAX_JOURNAL_BYTES {
        return Err(SourceFileError::Limit(
            "Recovery metadata is too large.".into(),
        ));
    }
    let stage = tempfile::Builder::new()
        .prefix(".arcadia-save-stage-")
        .tempdir_in(&snapshot.root)
        .map_err(|error| io_at(&snapshot.root, error))?;
    for (index, file) in journal.files.iter().enumerate() {
        if let DiskExpectation::Bytes(bytes) = &snapshot.expectations[&file.path] {
            write_synced(&staged_path(stage.path(), index, "before"), bytes)?;
        }
        if file.write {
            write_synced(
                &staged_path(stage.path(), index, "after"),
                compiled.bundle.files[&file.path].as_bytes(),
            )?;
        }
    }
    write_synced(&stage.path().join("journal.json"), &metadata)?;
    sync_directory(stage.path())?;
    phase(SavePhase::BeforePublish).map_err(|error| io_at(&journal_dir, error))?;
    // Publishing a complete directory makes interrupted staging harmless.
    publish_journal(stage.path(), &journal_dir)?;
    let _ = stage.keep(); // The renamed directory now belongs to recovery.
    sync_directory(&snapshot.root)?;
    phase(SavePhase::AfterPublish).map_err(|error| io_at(&journal_dir, error))?;
    // Publication may have waited while another writer completed the same
    // candidate. Nothing has been written by this save: never interpret their
    // matching after-bytes as ours and roll them back. The absent v2 marker
    // also preserves this distinction after interruption or retirement failure.
    if let Err(error) = check_conflicts(&snapshot.root, &snapshot.expectations) {
        let retirement = phase(SavePhase::BeforeRetire)
            .map_err(|error| io_at(&journal_dir, error))
            .and_then(|()| retire_journal(&snapshot.root, &journal_dir, &journal));
        return match retirement {
            Ok(()) => Err(error),
            Err(retirement) => Err(SourceFileError::RecoveryFailed {
                save: error.to_string(),
                recovery: retirement.to_string(),
                journal: journal_dir,
            }),
        };
    }
    write_synced(&journal_dir.join(WRITE_STARTED), WRITE_STARTED_BYTES)?;
    sync_directory(&journal_dir)?;
    phase(SavePhase::AfterWriteMarker).map_err(|error| io_at(&journal_dir, error))?;
    let result = (|| {
        let mut current = snapshot.expectations.clone();
        let mut replacements = 0;
        for file in journal.files.iter().filter(|file| file.write) {
            // The first check is above, before the durable write-attempt marker.
            // Further checks follow real write attempts and may need rollback.
            if replacements != 0 {
                check_conflicts(&snapshot.root, &current)?;
            }
            let bytes = compiled.bundle.files[&file.path].as_bytes();
            atomic_write(&snapshot.root, &file.path, bytes)?;
            current.insert(file.path.clone(), DiskExpectation::Bytes(bytes.to_vec()));
            replacements += 1;
            phase(SavePhase::AfterReplace(replacements))
                .map_err(|error| io_at(&snapshot.root.join(&file.path), error))?;
        }
        check_conflicts(&snapshot.root, &current)?;
        Ok(())
    })();
    if let Err(error) = result {
        return match recover_project(&snapshot.root.join(&snapshot.bundle.entry), limits) {
            Ok(_) => Err(error),
            Err(recovery) => Err(SourceFileError::RecoveryFailed {
                save: error.to_string(),
                recovery: recovery.to_string(),
                journal: journal_dir,
            }),
        };
    }
    phase(SavePhase::BeforeRetire).map_err(|error| io_at(&journal_dir, error))?;
    retire_journal(&snapshot.root, &journal_dir, &journal)?;
    Ok(saved_snapshot(snapshot, compiled))
}

fn saved_snapshot(snapshot: &DiskProject, compiled: &CompiledSourceProject) -> DiskProject {
    let mut saved = snapshot.clone();
    for (path, text) in &compiled.bundle.files {
        saved.bundle.files.insert(path.clone(), text.clone());
        saved.expectations.insert(
            path.clone(),
            DiskExpectation::Bytes(text.as_bytes().to_vec()),
        );
    }
    saved.reachable = compiled.bundle.files.keys().cloned().collect();
    saved.diagnostics = compiled.diagnostics.clone();
    saved
}

fn remove_journal(directory: &Path, journal: &Journal) -> Result<(), SourceFileError> {
    // Never recursively delete an untrusted journal directory. Only validated,
    // fixed-name files are removed; unexpected content keeps the directory.
    for (index, file) in journal.files.iter().enumerate() {
        for (present, suffix) in [(file.before, "before"), (file.write, "after")] {
            if present {
                let path = staged_path(directory, index, suffix);
                fs::remove_file(&path).map_err(|error| io_at(&path, error))?;
            }
        }
    }
    if journal.version == 2 {
        let marker = directory.join(WRITE_STARTED);
        match fs::remove_file(&marker) {
            Ok(()) => {}
            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
            Err(error) => return Err(io_at(&marker, error)),
        }
    }
    let metadata = directory.join("journal.json");
    fs::remove_file(&metadata).map_err(|error| io_at(&metadata, error))?;
    fs::remove_dir(directory).map_err(|error| io_at(directory, error))
}

fn retire_journal(root: &Path, directory: &Path, journal: &Journal) -> Result<(), SourceFileError> {
    // Atomic retirement precedes cleanup: interruption during cleanup cannot
    // leave a pending journal with half its recovery information missing.
    let completed = root.join(format!(".arcadia-save-completed-{}", crate::Id::new_v4()));
    fs::rename(directory, &completed).map_err(|error| io_at(directory, error))?;
    // Cleanup failures leave a uniquely named completed directory, never a
    // misleading pending transaction. It is safe to remove that directory later.
    let _ = sync_directory(root);
    let _ = remove_journal(&completed, journal);
    Ok(())
}

/// Explicit interrupted-save recovery. All original/written bytes are checked
/// before rollback, and each restoration rechecks its target. External changes
/// are never knowingly overwritten. `false` means there was no pending journal.
pub fn recover_project(entry: &Path, limits: ProjectLimits) -> Result<bool, SourceFileError> {
    recover_with_hook(entry, limits, |_| Ok(()))
}

fn recover_with_hook(
    entry: &Path,
    limits: ProjectLimits,
    mut after_restore: impl FnMut(usize) -> io::Result<()>,
) -> Result<bool, SourceFileError> {
    let (root, entry) = root_entry(entry)?;
    let directory = journal_path(&root, &entry);
    match fs::symlink_metadata(&directory) {
        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(false),
        Err(error) => return Err(io_at(&directory, error)),
        Ok(metadata) if !metadata.is_dir() || metadata.file_type().is_symlink() => {
            return Err(SourceFileError::Path(
                "Recovery journal is not a real directory.".into(),
            ));
        }
        Ok(_) => {}
    }
    if directory
        .canonicalize()
        .map_err(|error| io_at(&directory, error))?
        .parent()
        != Some(root.as_path())
    {
        return Err(SourceFileError::Path(
            "Recovery directory escapes project root.".into(),
        ));
    }
    let metadata_path = directory.join("journal.json");
    let read_journal_file = |path: &Path, budget| {
        if fs::symlink_metadata(path)
            .map_err(|error| io_at(path, error))?
            .file_type()
            .is_symlink()
        {
            return Err(SourceFileError::Path("Recovery file is a link.".into()));
        }
        read_bounded(path, budget)
    };
    let bytes = read_journal_file(&metadata_path, MAX_JOURNAL_BYTES)?;
    let journal: Journal = serde_json::from_slice(&bytes)
        .map_err(|error| SourceFileError::Invalid(format!("Invalid recovery journal: {error}")))?;
    if !matches!(journal.version, 1 | 2)
        || journal.entry != entry
        || journal.files.len() > limits.max_files
    {
        return Err(SourceFileError::Invalid(
            "Recovery journal version, entry or file budget is invalid.".into(),
        ));
    }
    let mut original = BTreeMap::new();
    let mut written = BTreeMap::new();
    let mut folded = BTreeSet::new();
    let (mut before_bytes, mut after_bytes) = (0usize, 0usize);
    let mut allowed = BTreeSet::from(["journal.json".to_string()]);
    let write_started = if journal.version == 2 {
        let marker = directory.join(WRITE_STARTED);
        match fs::symlink_metadata(&marker) {
            Ok(_) => {
                if read_journal_file(&marker, WRITE_STARTED_BYTES.len())? != WRITE_STARTED_BYTES {
                    return Err(SourceFileError::Invalid(
                        "Invalid journal write-started marker.".into(),
                    ));
                }
                allowed.insert(WRITE_STARTED.into());
                true
            }
            Err(error) if error.kind() == io::ErrorKind::NotFound => false,
            Err(error) => return Err(io_at(&marker, error)),
        }
    } else {
        // Frozen v1 journals predate the marker and retain their rollback rules.
        true
    };
    for (index, file) in journal.files.iter().enumerate() {
        disk_path(&root, &file.path)?;
        if !folded.insert(file.path.to_lowercase()) {
            return Err(SourceFileError::Path("Duplicate recovery paths.".into()));
        }
        let before = if file.before {
            let bytes = read_journal_file(
                &staged_path(&directory, index, "before"),
                limits.model.max_source_bytes.saturating_sub(before_bytes),
            )?;
            before_bytes += bytes.len();
            allowed.insert(format!("{index}.before"));
            DiskExpectation::Bytes(bytes)
        } else {
            DiskExpectation::Missing
        };
        if file.write {
            let bytes = read_journal_file(
                &staged_path(&directory, index, "after"),
                limits.model.max_source_bytes.saturating_sub(after_bytes),
            )?;
            after_bytes += bytes.len();
            allowed.insert(format!("{index}.after"));
            written.insert(file.path.clone(), DiskExpectation::Bytes(bytes));
        }
        original.insert(file.path.clone(), before);
    }
    if !original.contains_key(&entry) {
        return Err(SourceFileError::Invalid(
            "Recovery journal has no entry expectation.".into(),
        ));
    }
    for file in fs::read_dir(&directory).map_err(|error| io_at(&directory, error))? {
        let file = file.map_err(|error| io_at(&directory, error))?;
        if !file
            .file_name()
            .to_str()
            .is_some_and(|name| allowed.contains(name))
        {
            return Err(SourceFileError::Invalid(
                "Recovery directory contains unexpected files.".into(),
            ));
        }
    }
    if !write_started {
        retire_journal(&root, &directory, &journal)?;
        return Ok(true);
    }
    let mut current = BTreeMap::new();
    let mut conflicts = Vec::new();
    for (path, before) in &original {
        if matches_disk(&root, path, before)? {
            current.insert(path.clone(), before.clone());
        } else if let Some(after) = written.get(path)
            && matches_disk(&root, path, after)?
        {
            current.insert(path.clone(), after.clone());
        } else {
            conflicts.push(path.clone());
        }
    }
    if !conflicts.is_empty() {
        return Err(SourceFileError::Conflict(conflicts));
    }
    // Reverse save order; a second recovery can continue after interruption.
    let mut restored = 0;
    for file in journal.files.iter().rev().filter(|file| file.write) {
        check_conflicts(&root, &current)?;
        if current[&file.path] == original[&file.path] {
            continue;
        }
        match &original[&file.path] {
            DiskExpectation::Bytes(bytes) => atomic_write(&root, &file.path, bytes)?,
            DiskExpectation::Missing => {
                let path = disk_path(&root, &file.path)?;
                fs::remove_file(&path).map_err(|error| io_at(&path, error))?;
                sync_directory(path.parent().unwrap())?;
            }
        }
        current.insert(file.path.clone(), original[&file.path].clone());
        restored += 1;
        after_restore(restored).map_err(|error| io_at(&directory, error))?;
    }
    retire_journal(&root, &directory, &journal)?;
    Ok(true)
}

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

    #[cfg(target_os = "linux")]
    #[test]
    fn linux_journal_publication_preserves_competing_destinations_and_our_stage() {
        use std::os::unix::{fs::MetadataExt, fs::symlink};

        for kind in ["empty directory", "nonempty directory", "file", "symlink"] {
            let directory = tempfile::tempdir().unwrap();
            let stage = directory.path().join("stage");
            let journal = directory.path().join("journal");
            fs::create_dir(&stage).unwrap();
            fs::write(stage.join("journal.json"), b"our complete metadata").unwrap();
            // Simulate a competitor arriving after the caller's absence check.
            assert!(!journal.exists());
            match kind {
                "empty directory" => fs::create_dir(&journal).unwrap(),
                "nonempty directory" => {
                    fs::create_dir(&journal).unwrap();
                    fs::write(journal.join("journal.json"), b"competitor metadata").unwrap();
                }
                "file" => fs::write(&journal, b"competitor file").unwrap(),
                "symlink" => symlink("absent competitor target", &journal).unwrap(),
                _ => unreachable!(),
            }
            let before = fs::symlink_metadata(&journal).unwrap();
            let error = publish_journal(&stage, &journal).unwrap_err();
            assert!(
                matches!(error, SourceFileError::Io { path, source }
                    if path == journal && source.raw_os_error() == Some(libc::EEXIST)),
                "{kind} must refuse publication"
            );
            let after = fs::symlink_metadata(&journal).unwrap();
            assert_eq!((before.dev(), before.ino()), (after.dev(), after.ino()));
            assert_eq!(
                fs::read(stage.join("journal.json")).unwrap(),
                b"our complete metadata"
            );
            match kind {
                "empty directory" => assert_eq!(fs::read_dir(&journal).unwrap().count(), 0),
                "nonempty directory" => assert_eq!(
                    fs::read(journal.join("journal.json")).unwrap(),
                    b"competitor metadata"
                ),
                "file" => assert_eq!(fs::read(&journal).unwrap(), b"competitor file"),
                "symlink" => assert_eq!(
                    fs::read_link(&journal).unwrap(),
                    Path::new("absent competitor target")
                ),
                _ => unreachable!(),
            }
        }
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn linux_journal_publication_preserves_raw_path_bytes_and_rejects_nul() {
        use std::{ffi::OsStr, os::unix::ffi::OsStrExt};

        let directory = tempfile::tempdir().unwrap();
        let stage = directory.path().join(OsStr::from_bytes(b"stage-\xff"));
        let journal = directory.path().join(OsStr::from_bytes(b"journal-\xfe"));
        fs::create_dir(&stage).unwrap();
        fs::write(stage.join("journal.json"), b"complete metadata").unwrap();
        for (a, b) in [
            (Path::new("bad\0path"), journal.as_path()),
            (stage.as_path(), Path::new("bad\0path")),
        ] {
            let error = rename_journal_without_replacement(a, b).unwrap_err();
            assert_eq!(error.kind(), io::ErrorKind::InvalidInput);
            assert!(stage.join("journal.json").exists());
            assert!(!journal.exists());
        }
        publish_journal(&stage, &journal).unwrap();
        assert!(!stage.exists());
        assert_eq!(
            fs::read(journal.join("journal.json")).unwrap(),
            b"complete metadata"
        );
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn linux_journal_publication_has_one_winner_for_concurrent_stages() {
        use std::sync::Barrier;

        let directory = tempfile::tempdir().unwrap();
        let journal = directory.path().join("journal");
        let stages = [
            directory.path().join("stage-a"),
            directory.path().join("stage-b"),
        ];
        for (i, stage) in stages.iter().enumerate() {
            fs::create_dir(stage).unwrap();
            fs::write(stage.join("journal.json"), format!("complete writer {i}")).unwrap();
        }
        let barrier = Barrier::new(2);
        let results = std::thread::scope(|scope| {
            let journal = &journal;
            let barrier = &barrier;
            let handles: Vec<_> = stages
                .iter()
                .map(|stage| {
                    scope.spawn(move || {
                        assert!(!journal.exists());
                        barrier.wait();
                        publish_journal(stage, journal)
                    })
                })
                .collect();
            handles
                .into_iter()
                .map(|handle| handle.join().unwrap())
                .collect::<Vec<_>>()
        });
        assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1);
        for (i, result) in results.into_iter().enumerate() {
            match result {
                Ok(()) => {
                    assert!(!stages[i].exists());
                    assert_eq!(
                        fs::read_to_string(journal.join("journal.json")).unwrap(),
                        format!("complete writer {i}")
                    );
                }
                Err(SourceFileError::Io { path, source }) => {
                    assert_eq!(path, journal);
                    assert_eq!(source.raw_os_error(), Some(libc::EEXIST));
                    assert_eq!(
                        fs::read_to_string(stages[i].join("journal.json")).unwrap(),
                        format!("complete writer {i}")
                    );
                }
                Err(error) => panic!("unexpected publication error: {error}"),
            }
        }
    }

    #[cfg(target_os = "linux")]
    #[test]
    #[ignore = "requires strace renameat2 error injection and OA_TEST_RENAMEAT2_ERRNO"]
    fn linux_journal_publication_injected_unsupported_error_is_fail_closed() {
        let errno: i32 = std::env::var("OA_TEST_RENAMEAT2_ERRNO")
            .expect("set OA_TEST_RENAMEAT2_ERRNO to the injected Linux errno")
            .parse()
            .unwrap();
        assert!(matches!(
            errno,
            libc::ENOSYS | libc::EINVAL | libc::EOPNOTSUPP
        ));
        let (directory, snapshot, compiled) = fixture();
        let journal = journal_path(&snapshot.root, &snapshot.bundle.entry);
        let original: Vec<_> = snapshot
            .bundle
            .files
            .keys()
            .map(|path| (path.clone(), fs::read(directory.path().join(path)).unwrap()))
            .collect();
        let error = save_with_phases(&snapshot, &compiled, ProjectLimits::default(), |phase| {
            if matches!(
                phase,
                SavePhase::AfterPublish | SavePhase::AfterWriteMarker | SavePhase::AfterReplace(_)
            ) {
                panic!("unsupported no-replace publication must stop before ownership transfer");
            }
            Ok(())
        })
        .unwrap_err();
        assert!(matches!(error, SourceFileError::Io { path, source }
            if path == journal && source.raw_os_error() == Some(errno)));
        assert!(!journal.exists());
        for (path, bytes) in original {
            assert_eq!(fs::read(directory.path().join(path)).unwrap(), bytes);
        }
        check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn linux_journal_publication_refusal_never_starts_model_writes() {
        let (directory, snapshot, compiled) = fixture();
        let journal = journal_path(&snapshot.root, &snapshot.bundle.entry);
        let original: Vec<_> = snapshot
            .bundle
            .files
            .keys()
            .map(|path| (path.clone(), fs::read(directory.path().join(path)).unwrap()))
            .collect();
        let error = save_with_phases(&snapshot, &compiled, ProjectLimits::default(), |phase| {
            match phase {
                SavePhase::BeforePublish => fs::create_dir(&journal).unwrap(),
                SavePhase::AfterPublish
                | SavePhase::AfterWriteMarker
                | SavePhase::AfterReplace(_) => {
                    panic!("a rejected publisher must not obtain ownership or write source files")
                }
                _ => {}
            }
            Ok(())
        })
        .unwrap_err();
        assert!(matches!(error, SourceFileError::Io { path, source }
            if path == journal && source.raw_os_error() == Some(libc::EEXIST)));
        assert_eq!(fs::read_dir(&journal).unwrap().count(), 0);
        assert!(!journal.join(WRITE_STARTED).exists());
        for (path, bytes) in original {
            assert_eq!(fs::read(directory.path().join(path)).unwrap(), bytes);
        }
        check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
    }

    #[cfg(windows)]
    #[test]
    fn journal_publication_does_not_replace_empty_target_created_after_precheck() {
        let directory = tempfile::tempdir().unwrap();
        let stage = directory.path().join("stage");
        let journal = directory.path().join("journal");
        fs::create_dir(&stage).unwrap();
        fs::write(stage.join("journal.json"), b"our journal").unwrap();
        let mut called = false;
        let result = publish_journal_with_retry(
            &stage,
            &journal,
            |a, b| {
                assert!(!called);
                called = true;
                fs::create_dir(b).unwrap();
                rename_journal_without_replacement(a, b)
            },
            |_| {},
        );
        assert!(result.is_err());
        assert!(called);
        assert_eq!(
            fs::read(stage.join("journal.json")).unwrap(),
            b"our journal"
        );
        assert_eq!(fs::read_dir(&journal).unwrap().count(), 0);
        assert!(
            matches!(rename_journal_without_replacement(Path::new("bad\0path"), &journal), Err(error) if error.kind() == io::ErrorKind::InvalidInput)
        );
    }

    #[test]
    fn journal_publication_retries_only_transient_windows_errors_before_transferring_ownership() {
        let directory = tempfile::tempdir().unwrap();
        let stage = directory.path().join("stage");
        let journal = directory.path().join("journal");
        fs::create_dir(&stage).unwrap();
        fs::write(stage.join("journal.json"), b"complete staged metadata").unwrap();
        let mut errors = [5, 32, 33].into_iter();
        let mut waits = vec![];
        publish_journal_with_retry(
            &stage,
            &journal,
            |a, b| {
                assert!(a.join("journal.json").exists());
                assert!(!b.exists());
                match errors.next() {
                    Some(code) => Err(io::Error::from_raw_os_error(code)),
                    None => fs::rename(a, b),
                }
            },
            |delay| waits.push(delay.as_millis()),
        )
        .unwrap();
        assert_eq!(waits, [5, 10, 25]);
        assert!(!stage.exists());
        assert_eq!(
            fs::read(journal.join("journal.json")).unwrap(),
            b"complete staged metadata"
        );
    }

    #[test]
    fn journal_publication_exhaustion_and_permanent_errors_keep_stage_and_report_original_path() {
        for (code, expected_calls, expected_wait) in [(5, 7, 390), (123, 1, 0)] {
            let directory = tempfile::tempdir().unwrap();
            let stage = directory.path().join("stage");
            let journal = directory.path().join("journal");
            fs::create_dir(&stage).unwrap();
            fs::write(stage.join("journal.json"), b"before publication").unwrap();
            let (mut calls, mut waited) = (0, 0);
            let error = publish_journal_with_retry(
                &stage,
                &journal,
                |_, _| {
                    calls += 1;
                    Err(io::Error::from_raw_os_error(code))
                },
                |delay| waited += delay.as_millis(),
            )
            .unwrap_err();
            assert_eq!((calls, waited), (expected_calls, expected_wait));
            assert!(
                matches!(error, SourceFileError::Io { path, source } if path == journal && source.raw_os_error() == Some(code))
            );
            assert_eq!(
                fs::read(stage.join("journal.json")).unwrap(),
                b"before publication"
            );
            assert!(!journal.exists());
        }
    }

    #[test]
    fn journal_publication_never_replaces_another_journal_that_appears_during_retry() {
        let directory = tempfile::tempdir().unwrap();
        let stage = directory.path().join("stage");
        let journal = directory.path().join("journal");
        fs::create_dir(&stage).unwrap();
        fs::write(stage.join("journal.json"), b"our staged journal").unwrap();
        let mut calls = 0;
        let error = publish_journal_with_retry(
            &stage,
            &journal,
            |_, _| {
                calls += 1;
                Err(io::Error::from_raw_os_error(32))
            },
            |_| {
                fs::create_dir(&journal).unwrap();
                fs::write(journal.join("journal.json"), b"other save owns this").unwrap();
            },
        )
        .unwrap_err();
        assert_eq!(calls, 1);
        assert!(matches!(error, SourceFileError::RecoveryRequired(path) if path == journal));
        assert_eq!(
            fs::read(journal.join("journal.json")).unwrap(),
            b"other save owns this"
        );
        assert_eq!(
            fs::read(stage.join("journal.json")).unwrap(),
            b"our staged journal"
        );
    }

    fn fixture() -> (tempfile::TempDir, DiskProject, CompiledSourceProject) {
        let directory = tempfile::tempdir().unwrap();
        fs::write(
            directory.path().join("project.arcadia"),
            "arcadia 5; project p \"P\" { include \"a.arcadia\"; include \"b.arcadia\"; }",
        )
        .unwrap();
        fs::write(directory.path().join("a.arcadia"), "oa_activity a \"A\" {}").unwrap();
        fs::write(directory.path().join("b.arcadia"), "oa_activity b \"B\" {}").unwrap();
        let snapshot = load_project(
            &directory.path().join("project.arcadia"),
            ProjectLimits::default(),
        )
        .unwrap();
        let compiled =
            compile_bundle(&snapshot.reachable_bundle(), ProjectLimits::default()).unwrap();
        (directory, snapshot, compiled)
    }

    #[test]
    fn unwritten_journal_never_rolls_back_a_completed_writer_with_identical_after_bytes() {
        for interrupt_retirement in [false, true] {
            let (directory, snapshot, compiled) = fixture();
            let error = save_with_phases(&snapshot, &compiled, ProjectLimits::default(), |phase| {
                match phase {
                    SavePhase::BeforePublish => {
                        // A competing save finishes during our publication wait.
                        save_project(&snapshot, &compiled, ProjectLimits::default()).unwrap();
                    }
                    SavePhase::BeforeRetire if interrupt_retirement => {
                        return Err(io::Error::other("injected no-write retirement failure"));
                    }
                    SavePhase::AfterWriteMarker | SavePhase::AfterReplace(_) => {
                        panic!("losing writer must not start writes")
                    }
                    _ => {}
                }
                Ok(())
            })
            .unwrap_err();
            assert!(matches!(
                error,
                SourceFileError::Conflict(_) | SourceFileError::RecoveryFailed { .. }
            ));
            for (path, bytes) in &compiled.bundle.files {
                assert_eq!(
                    fs::read_to_string(directory.path().join(path)).unwrap(),
                    *bytes
                );
            }
            let journal = journal_path(&snapshot.root, &snapshot.bundle.entry);
            assert_eq!(journal.exists(), interrupt_retirement);
            if interrupt_retirement {
                assert!(!journal.join(WRITE_STARTED).exists());
                // Explicit recovery after a failed retirement must also leave
                // the completed other writer untouched, even matching after bytes.
                assert!(
                    recover_project(
                        &directory.path().join("project.arcadia"),
                        ProjectLimits::default()
                    )
                    .unwrap()
                );
                for (path, bytes) in &compiled.bundle.files {
                    assert_eq!(
                        fs::read_to_string(directory.path().join(path)).unwrap(),
                        *bytes
                    );
                }
            }
            assert!(!journal.exists());
        }
    }

    #[test]
    fn recovery_before_any_write_accepts_v2_marker_and_keeps_v1_rollback_compatibility() {
        for version in [1, 2] {
            let (directory, snapshot, compiled) = fixture();
            let interrupted = std::panic::catch_unwind(|| {
                let _ = save_with_phases(&snapshot, &compiled, ProjectLimits::default(), |phase| {
                    if matches!(phase, SavePhase::AfterWriteMarker) {
                        panic!("interrupted before atomic write");
                    }
                    Ok(())
                });
            });
            assert!(interrupted.is_err());
            check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
            let journal = journal_path(&snapshot.root, &snapshot.bundle.entry);
            assert_eq!(
                fs::read(journal.join(WRITE_STARTED)).unwrap(),
                WRITE_STARTED_BYTES
            );
            if version == 1 {
                // A frozen v1 journal has no marker and means writes may have begun.
                let path = journal.join("journal.json");
                let mut metadata: Journal =
                    serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
                metadata.version = 1;
                fs::write(path, serde_json::to_vec(&metadata).unwrap()).unwrap();
                fs::remove_file(journal.join(WRITE_STARTED)).unwrap();
                for (path, bytes) in &compiled.bundle.files {
                    fs::write(directory.path().join(path), bytes).unwrap();
                }
            }
            assert!(
                recover_project(
                    &directory.path().join("project.arcadia"),
                    ProjectLimits::default()
                )
                .unwrap()
            );
            check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
        }
    }

    #[test]
    fn interrupted_publication_without_write_marker_never_rolls_back_external_matching_bytes() {
        let (directory, snapshot, compiled) = fixture();
        let interrupted = std::panic::catch_unwind(|| {
            let _ = save_with_phases(&snapshot, &compiled, ProjectLimits::default(), |phase| {
                if matches!(phase, SavePhase::AfterPublish) {
                    panic!("interrupted before write marker");
                }
                Ok(())
            });
        });
        assert!(interrupted.is_err());
        let journal = journal_path(&snapshot.root, &snapshot.bundle.entry);
        assert!(!journal.join(WRITE_STARTED).exists());
        for (path, bytes) in &compiled.bundle.files {
            fs::write(directory.path().join(path), bytes).unwrap();
        }
        assert!(
            recover_project(
                &directory.path().join("project.arcadia"),
                ProjectLimits::default()
            )
            .unwrap()
        );
        for (path, bytes) in &compiled.bundle.files {
            assert_eq!(
                fs::read_to_string(directory.path().join(path)).unwrap(),
                *bytes
            );
        }
        assert!(!journal.exists());
    }

    #[test]
    fn malformed_v2_write_marker_refuses_recovery_without_changing_source_bytes() {
        let (directory, snapshot, compiled) = fixture();
        let _ = std::panic::catch_unwind(|| {
            let _ = save_with_phases(&snapshot, &compiled, ProjectLimits::default(), |phase| {
                if matches!(phase, SavePhase::AfterWriteMarker) {
                    panic!("interrupted");
                }
                Ok(())
            });
        });
        let journal = journal_path(&snapshot.root, &snapshot.bundle.entry);
        fs::write(journal.join(WRITE_STARTED), b"partial").unwrap();
        assert!(matches!(
            recover_project(
                &directory.path().join("project.arcadia"),
                ProjectLimits::default()
            ),
            Err(SourceFileError::Invalid(_))
        ));
        check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
        assert!(journal.exists());
    }

    #[test]
    fn partial_save_failures_restore_every_original_and_write_entry_last() {
        for stop in 1..=3 {
            let (directory, snapshot, compiled) = fixture();
            let error = save_with_hook(&snapshot, &compiled, ProjectLimits::default(), |count| {
                if count < 3 {
                    assert!(
                        matches_disk(
                            &snapshot.root,
                            &snapshot.bundle.entry,
                            &snapshot.expectations[&snapshot.bundle.entry]
                        )
                        .unwrap()
                    );
                }
                if count == stop {
                    Err(io::Error::other("injected replacement failure"))
                } else {
                    Ok(())
                }
            })
            .unwrap_err();
            assert!(matches!(error, SourceFileError::Io { .. }), "{error}");
            check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
            assert!(
                !recover_project(
                    &directory.path().join("project.arcadia"),
                    ProjectLimits::default()
                )
                .unwrap()
            );
        }
    }

    #[test]
    fn interrupted_save_and_interrupted_recovery_can_both_be_resumed() {
        let (directory, snapshot, compiled) = fixture();
        let interrupted = std::panic::catch_unwind(|| {
            let _ = save_with_hook(&snapshot, &compiled, ProjectLimits::default(), |count| {
                assert!(count < 3, "simulated process interruption");
                Ok(())
            });
        });
        assert!(interrupted.is_err());
        assert!(
            load_project(
                &directory.path().join("project.arcadia"),
                ProjectLimits::default()
            )
            .unwrap()
            .diagnostics
            .iter()
            .any(|d| d.diagnostic.code == "pending_save")
        );
        assert!(matches!(
            save_project(&snapshot, &compiled, ProjectLimits::default()),
            Err(SourceFileError::RecoveryRequired(_))
        ));
        let entry = directory.path().join("project.arcadia");
        let interrupted = recover_with_hook(&entry, ProjectLimits::default(), |_| {
            Err(io::Error::other("recovery interrupted"))
        });
        assert!(interrupted.is_err());
        assert!(recover_project(&entry, ProjectLimits::default()).unwrap());
        check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
        assert!(!recover_project(&entry, ProjectLimits::default()).unwrap());
    }

    #[test]
    fn rollback_refuses_external_edits_even_in_an_unchanged_dependency() {
        let (directory, mut snapshot, initial) = fixture();
        snapshot = save_project(&snapshot, &initial, ProjectLimits::default()).unwrap();
        let mut bundle = initial.bundle.clone();
        bundle
            .files
            .get_mut("a.arcadia")
            .unwrap()
            .push_str("\n// accepted change\n");
        let compiled = compile_bundle(&bundle, ProjectLimits::default()).unwrap();
        let error = save_with_hook(&snapshot, &compiled, ProjectLimits::default(), |_| {
            fs::write(directory.path().join("b.arcadia"), "// external edit").unwrap();
            Err(io::Error::other("injected failure after external edit"))
        })
        .unwrap_err();
        assert!(matches!(error, SourceFileError::RecoveryFailed { .. }));
        assert_eq!(
            fs::read_to_string(directory.path().join("b.arcadia")).unwrap(),
            "// external edit"
        );
        assert_eq!(
            fs::read_to_string(directory.path().join("a.arcadia")).unwrap(),
            compiled.bundle.files["a.arcadia"]
        );
        fs::write(
            directory.path().join("b.arcadia"),
            &initial.bundle.files["b.arcadia"],
        )
        .unwrap();
        assert!(
            recover_project(
                &directory.path().join("project.arcadia"),
                ProjectLimits::default()
            )
            .unwrap()
        );
        check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
    }

    #[test]
    fn recovery_rechecks_external_changes_between_restorations() {
        let (directory, snapshot, compiled) = fixture();
        let _ = std::panic::catch_unwind(|| {
            let _ = save_with_hook(&snapshot, &compiled, ProjectLimits::default(), |count| {
                assert!(count < 3, "interrupted after all replacements");
                Ok(())
            });
        });
        let entry = directory.path().join("project.arcadia");
        let error = recover_with_hook(&entry, ProjectLimits::default(), |_| {
            fs::write(
                directory.path().join("b.arcadia"),
                "// external edit during recovery",
            )
            .unwrap();
            Ok(())
        })
        .unwrap_err();
        assert!(matches!(error, SourceFileError::Conflict(_)));
        assert_eq!(
            fs::read_to_string(directory.path().join("b.arcadia")).unwrap(),
            "// external edit during recovery"
        );
        assert!(journal_path(&snapshot.root, &snapshot.bundle.entry).exists());
        fs::write(
            directory.path().join("b.arcadia"),
            &compiled.bundle.files["b.arcadia"],
        )
        .unwrap();
        assert!(recover_project(&entry, ProjectLimits::default()).unwrap());
        check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
    }

    #[test]
    fn rollback_removes_only_our_new_file_and_retains_detached_files() {
        let (directory, mut snapshot, _) = fixture();
        fs::write(directory.path().join("detached.arcadia"), "// retained").unwrap();
        let mut bundle = snapshot.reachable_bundle();
        let source = bundle.files.get_mut("project.arcadia").unwrap();
        *source = source.replacen(
            "include \"a.arcadia\";",
            "include \"new/first.arcadia\"; include \"a.arcadia\";",
            1,
        );
        bundle
            .files
            .insert("new/first.arcadia".into(), "oa_activity c \"C\" {}".into());
        expect_new_file(&mut snapshot, "new/first.arcadia", ProjectLimits::default()).unwrap();
        let compiled = compile_bundle(&bundle, ProjectLimits::default()).unwrap();
        let error = save_with_hook(&snapshot, &compiled, ProjectLimits::default(), |count| {
            if count == 3 {
                Err(io::Error::other("after new file"))
            } else {
                Ok(())
            }
        })
        .unwrap_err();
        assert!(matches!(error, SourceFileError::Io { .. }));
        assert!(!directory.path().join("new/first.arcadia").exists());
        assert_eq!(
            fs::read_to_string(directory.path().join("detached.arcadia")).unwrap(),
            "// retained"
        );
        check_conflicts(&snapshot.root, &snapshot.expectations).unwrap();
    }

    #[test]
    fn recovery_rejects_untrusted_escape_paths_before_touching_files() {
        let (directory, snapshot, compiled) = fixture();
        let _ = std::panic::catch_unwind(|| {
            let _ = save_with_hook(&snapshot, &compiled, ProjectLimits::default(), |_| {
                panic!("stop")
            });
        });
        let journal_dir = journal_path(&snapshot.root, &snapshot.bundle.entry);
        let metadata = journal_dir.join("journal.json");
        let mut journal: Journal = serde_json::from_slice(&fs::read(&metadata).unwrap()).unwrap();
        journal.files[0].path = "../outside.arcadia".into();
        fs::write(&metadata, serde_json::to_vec(&journal).unwrap()).unwrap();
        assert!(matches!(
            recover_project(
                &directory.path().join("project.arcadia"),
                ProjectLimits::default()
            ),
            Err(SourceFileError::Path(_))
        ));
        assert!(journal_dir.exists());
    }
}