eggup-core 0.1.3

use std::collections::{HashMap, HashSet};
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

use crate::domain::{AbsentPolicy, CommitOwnership, InstallPlan, MemberId, Ownership};
use crate::error::{Error, RecoveryError, RecoveryResult, Result};
use crate::integrity::hash_file;
use crate::lock::MutationLock;
use crate::stage::PreparedTransaction;

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

/// The terminal disposition of a commit attempt.
///
/// A rolled-back or recovery-required result is never equivalent to success;
/// see `examples/receipts.rs`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TransactionDisposition {
    /// Every member was installed and the live set is coherent.
    Committed,
    /// The attempt failed and all old members were restored and verified.
    RolledBack,
    /// The live state or cleanup could not be safely restored automatically.
    RecoveryRequired,
}

/// How transaction-owned temporary evidence was handled.
///
/// This describes transaction-owned temporary evidence. The separate
/// [`PostCommitFailurePolicy`] selects what to do when a post-commit check
/// fails.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CleanupDisposition {
    /// Owned temporary state was cleaned up.
    Cleaned,
    /// Evidence was retained for an operator because cleanup was unsafe or failed.
    RetainedForRecovery,
    /// Cleanup was safely arranged but cannot complete until this process exits.
    ///
    /// This is the truthful terminal state for a Windows self-update kept
    /// installed: the new generation is live and the previous generation is
    /// still mapped by this very process, so it cannot be unlinked yet. The old
    /// image is owned and scheduled for deletion rather than abandoned, and no
    /// operator action is required. It is never reported for an ordinary
    /// multi-member transaction, where no artifact is still mapped.
    DeferredToProcessExit,
}

/// Caller policy for a failed post-commit check.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PostCommitFailurePolicy {
    /// Keep the complete new artifact set installed and report the check failure.
    KeepInstalled,
    /// Restore and verify the complete previous artifact set.
    RollBack,
}

/// The transaction phase that produced a failure.
///
/// Phases are ordered by first occurrence in a commit attempt.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum FailurePhase {
    /// Mutation-lock acquisition or inspection.
    Lock,
    /// Destination ownership classification or parent containment.
    Ownership,
    /// Re-reading and re-hashing staged bytes under lock.
    StageRevalidation,
    /// Moving live destinations into the backup set.
    Backup,
    /// Renaming staged members into live destinations.
    Commit,
    /// A caller-supplied post-commit check failed.
    PostCommit,
    /// Restoring backups after a failure.
    Rollback,
    /// Removing the backup set after a successful commit.
    Finalize,
}

impl FailurePhase {
    /// Returns the stable phase name used in receipts and diagnostics.
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Lock => "lock",
            Self::Ownership => "ownership",
            Self::StageRevalidation => "stage-revalidation",
            Self::Backup => "backup",
            Self::Commit => "commit",
            Self::PostCommit => "post-commit",
            Self::Rollback => "rollback",
            Self::Finalize => "finalize",
        }
    }
}

/// A stable machine-readable failure category.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum FailureCategory {
    /// Another transaction or ambiguous lock owns the domain.
    LockContention,
    /// A destination is foreign, unknown, or violates containment/parent rules.
    OwnershipConflict,
    /// Staged bytes changed, are unreadable, or lack required integrity evidence.
    Verification,
    /// A live filesystem mutation failed.
    Filesystem,
    /// Caller input or transaction configuration is invalid.
    InvalidInput,
    /// The failure was injected by the test fault harness.
    Injected,
    /// Transaction-owned evidence was retained and manual recovery is required.
    RetainedEvidence,
    /// A caller-supplied post-commit check failed or panicked.
    PostCommitCheck,
}

impl FailureCategory {
    /// Returns the stable category name.
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::LockContention => "lock-contention",
            Self::OwnershipConflict => "ownership-conflict",
            Self::Verification => "verification",
            Self::Filesystem => "filesystem",
            Self::InvalidInput => "invalid-input",
            Self::Injected => "injected",
            Self::RetainedEvidence => "retained-evidence",
            Self::PostCommitCheck => "post-commit-check",
        }
    }
}

/// A bounded structured report describing one transaction failure.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FailureReport {
    phase: FailurePhase,
    member: Option<MemberId>,
    category: FailureCategory,
    detail: String,
}

impl FailureReport {
    pub(crate) fn new(
        phase: FailurePhase,
        member: Option<MemberId>,
        category: FailureCategory,
        detail: impl Into<String>,
    ) -> Self {
        let mut detail = detail.into();
        // Bound human-readable detail; receipts must never embed unbounded output.
        if detail.len() > 512 {
            let boundary = detail
                .char_indices()
                .map(|(index, _)| index)
                .take_while(|index| *index <= 512)
                .last()
                .unwrap_or(0);
            detail.truncate(boundary);
        }
        Self {
            phase,
            member,
            category,
            detail,
        }
    }

    /// Returns the phase that produced the failure.
    pub fn phase(&self) -> FailurePhase {
        self.phase
    }

    /// Returns the affected member, when the failure is member-specific.
    pub fn member(&self) -> Option<&MemberId> {
        self.member.as_ref()
    }

    /// Returns the stable failure category.
    pub fn category(&self) -> FailureCategory {
        self.category
    }

    /// Returns bounded human-readable detail.
    pub fn detail(&self) -> &str {
        &self.detail
    }
}

pub(crate) fn report_for_error(
    phase: FailurePhase,
    member: Option<MemberId>,
    error: &Error,
) -> FailureReport {
    let (category, detail) = match error {
        Error::UpdateInProgress { lock } => (
            FailureCategory::LockContention,
            format!("lock contention at {}", lock.display()),
        ),
        Error::DestinationConflict { destination } => (
            FailureCategory::OwnershipConflict,
            format!("destination conflict at {}", destination.display()),
        ),
        Error::VerificationFailed(m) => (FailureCategory::Verification, m.clone()),
        Error::CandidateExecution(m) => (FailureCategory::Verification, m.clone()),
        #[cfg(test)]
        Error::Injected(m) => (FailureCategory::Injected, m.clone()),
        Error::InvalidInput(m) => (FailureCategory::InvalidInput, m.clone()),
        Error::UnknownMember(m) => (FailureCategory::InvalidInput, format!("unknown member {m}")),
        Error::Io { operation, source } => (
            FailureCategory::Filesystem,
            format!("{operation}: {source}"),
        ),
    };
    FailureReport::new(phase, member, category, detail)
}

/// Evidence describing the terminal result of a transaction attempt.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TransactionReceipt {
    product: crate::domain::ProductId,
    release: crate::domain::ReleaseId,
    disposition: TransactionDisposition,
    rollback_performed: bool,
    rollback_verified: bool,
    cleanup: CleanupDisposition,
    recovery_path: Option<PathBuf>,
    failure: Option<FailureReport>,
    rollback_failure: Option<FailureReport>,
    post_commit_failure: Option<FailureReport>,
}

impl TransactionReceipt {
    /// Returns the product identity recorded by the transaction.
    pub fn product(&self) -> &crate::domain::ProductId {
        &self.product
    }

    /// Returns the release identity recorded by the transaction.
    pub fn release(&self) -> &crate::domain::ReleaseId {
        &self.release
    }

    /// Returns the terminal disposition.
    pub fn disposition(&self) -> TransactionDisposition {
        self.disposition
    }

    /// Returns whether rollback was attempted.
    pub fn rollback_performed(&self) -> bool {
        self.rollback_performed
    }

    /// Returns whether every restoration check passed.
    ///
    /// This describes restoration of live destinations only. A rolled-back
    /// transaction whose restoration fully succeeded stays `true` even when
    /// the backup set could not be removed; that cleanup problem is reported
    /// in [`Self::failure`] with a [`FailurePhase::Finalize`] phase and
    /// `false` is never reported without a [`Self::rollback_failure`] or a
    /// recorded cleanup failure to explain it.
    pub fn rollback_verified(&self) -> bool {
        self.rollback_verified
    }

    /// Returns how transaction-owned temporary evidence was handled.
    pub fn cleanup(&self) -> CleanupDisposition {
        self.cleanup
    }

    /// Returns retained backup/lock evidence when manual recovery is required.
    ///
    /// When present, the path always refers to real retained evidence. A
    /// cleanup failure after an otherwise successful commit — or after a fully
    /// verified rollback — retains the real backup root and reports the cleanup
    /// problem in [`Self::failure`]; no synthetic non-existent path is ever
    /// returned.
    pub fn recovery_path(&self) -> Option<&Path> {
        self.recovery_path.as_deref()
    }

    /// Returns the structured report for the triggering failure, if any.
    ///
    /// A rolled-back transaction remains a first-class terminal result, but
    /// the caller can answer "what failed?" from this report without scraping
    /// logs. Successful commits with cleanup trouble also report the finalize
    /// problem here, as does a fully verified rollback whose backup set could
    /// not be cleaned up.
    pub fn failure(&self) -> Option<&FailureReport> {
        self.failure.as_ref()
    }

    /// Returns the rollback-phase failure when restoration itself failed.
    ///
    /// When present, [`Self::failure`] still carries the original triggering
    /// cause; this field carries the distinct recovery failure.
    pub fn rollback_failure(&self) -> Option<&FailureReport> {
        self.rollback_failure.as_ref()
    }

    /// Returns the failed post-commit check, if one occurred.
    pub fn post_commit_failure(&self) -> Option<&FailureReport> {
        self.post_commit_failure.as_ref()
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
#[cfg_attr(not(test), allow(dead_code))]
pub(crate) enum CommitFault {
    LockCreation,
    Backup(MemberId),
    BeforeFirstCommit,
    Commit(MemberId),
    CommitThenRollback(MemberId, MemberId),
    Finalize,
    PostCommitRollback,
}

#[derive(Debug)]
struct BackupEntry {
    member: MemberId,
    destination: PathBuf,
    backup: Option<PathBuf>,
}

type PostCommitCheck<'a> = (
    PostCommitFailurePolicy,
    Box<dyn FnOnce() -> std::result::Result<(), String> + 'a>,
);

impl PreparedTransaction {
    pub(crate) fn commit_inner(
        self,
        ownership: CommitOwnership<'_>,
        verified_digests: &HashMap<MemberId, [u8; 32]>,
        fault: Option<CommitFault>,
        post_commit: Option<PostCommitCheck<'_>>,
        stale_recovery: Option<&dyn crate::lock::StaleLockVerifier>,
    ) -> RecoveryResult<TransactionReceipt> {
        #[cfg(test)]
        if fault == Some(CommitFault::LockCreation) {
            return Err(Error::injected("injected lock creation failure").into());
        }
        // Preflight ownership classification before taking the lock so a
        // change between preflight and locked revalidation is detectable.
        let preflight = classify_all(&self, &ownership)?;
        let mut lock = match stale_recovery {
            // Recovery is opt-in and never the default: a caller that supplies
            // no proof keeps the fail-closed acquire that never displaces a
            // record.
            None => MutationLock::acquire(
                self.plan.installation_root(),
                self.plan.product(),
                self.plan.release(),
            )
            .map_err(RecoveryError::from),
            Some(verifier) => MutationLock::acquire_with_recovery(
                self.plan.installation_root(),
                self.plan.product(),
                self.plan.release(),
                verifier,
            ),
        }?;
        // Locked ownership revalidation: must match preflight and authorize
        // the intended mutation.
        if let Err(error) = revalidate_ownership_locked(&self, &ownership, &preflight) {
            let report = report_for_error(FailurePhase::Ownership, None, &error);
            let verified = true;
            return finish_failure_no_mutation(self, lock, verified, report);
        }
        // Verification-to-commit continuity: re-read and re-hash every staged
        // member under lock before backing up any live destination.
        if let Err((member, error)) = revalidate_staged_under_lock(&self, verified_digests) {
            let report = report_for_error(FailurePhase::StageRevalidation, member, &error);
            return finish_failure_no_mutation(self, lock, true, report);
        }
        let backup_root = match create_backup_directory(self.plan.installation_root()) {
            Ok(p) => p,
            Err(error) => {
                let report = report_for_error(FailurePhase::Backup, None, &error);
                return finish_failure_no_mutation(self, lock, true, report);
            }
        };
        let mut entries = Vec::with_capacity(self.plan.artifacts().len());
        if let Err((member, error)) = self.backup_members(&mut entries, &backup_root, fault.clone())
        {
            let phase = FailurePhase::Backup;
            let report = report_for_error(phase, member, &error);
            let (verified, rollback_failure) =
                restore_entries(&entries, &HashSet::new(), fault.clone());
            return finish_failure(self, lock, backup_root, verified, rollback_failure, report);
        }
        #[cfg(test)]
        if fault == Some(CommitFault::BeforeFirstCommit) {
            let error = Error::injected("injected pre-commit failure");
            let report = report_for_error(FailurePhase::Commit, None, &error);
            let (verified, rollback_failure) =
                restore_entries(&entries, &HashSet::new(), fault.clone());
            return finish_failure(self, lock, backup_root, verified, rollback_failure, report);
        }

        let mut committed = HashSet::new();
        let members = self.plan.artifacts().iter().cloned().collect::<Vec<_>>();
        for member in members {
            #[cfg(test)]
            let commit_should_fail = fault.as_ref().is_some_and(|candidate| match candidate {
                CommitFault::Commit(id) | CommitFault::CommitThenRollback(id, _) => {
                    id == member.id()
                }
                _ => false,
            });
            #[cfg(test)]
            if commit_should_fail {
                let error = Error::injected(format!("injected commit failure at {}", member.id()));
                let report =
                    report_for_error(FailurePhase::Commit, Some(member.id().clone()), &error);
                let (verified, rollback_failure) =
                    restore_entries(&entries, &committed, fault.clone());
                return finish_failure(self, lock, backup_root, verified, rollback_failure, report);
            }
            let destination = match self.plan.destination(member.id()) {
                Ok(destination) => destination,
                Err(error) => {
                    let report =
                        report_for_error(FailurePhase::Commit, Some(member.id().clone()), &error);
                    let (verified, rollback_failure) =
                        restore_entries(&entries, &committed, fault.clone());
                    return finish_failure(
                        self,
                        lock,
                        backup_root,
                        verified,
                        rollback_failure,
                        report,
                    );
                }
            };
            if let Err(error) = require_ready_parent(&destination, self.plan.installation_root()) {
                let report =
                    report_for_error(FailurePhase::Commit, Some(member.id().clone()), &error);
                let (verified, rollback_failure) =
                    restore_entries(&entries, &committed, fault.clone());
                return finish_failure(self, lock, backup_root, verified, rollback_failure, report);
            }
            let staged = match self.staged_path(member.id()) {
                Ok(staged) => staged,
                Err(error) => {
                    let report =
                        report_for_error(FailurePhase::Commit, Some(member.id().clone()), &error);
                    let (verified, rollback_failure) =
                        restore_entries(&entries, &committed, fault.clone());
                    return finish_failure(
                        self,
                        lock,
                        backup_root,
                        verified,
                        rollback_failure,
                        report,
                    );
                }
            };
            if let Err(source) = fs::rename(&staged, &destination) {
                let error = Error::io("replacing destination from prepared stage", source);
                let report =
                    report_for_error(FailurePhase::Commit, Some(member.id().clone()), &error);
                let (verified, rollback_failure) =
                    restore_entries(&entries, &committed, fault.clone());
                return finish_failure(self, lock, backup_root, verified, rollback_failure, report);
            }
            committed.insert(member.id().clone());
        }

        let mut post_commit_failure = None;
        if let Some((policy, check)) = post_commit {
            let checked = std::panic::catch_unwind(std::panic::AssertUnwindSafe(check));
            let failure_detail = match checked {
                Ok(Ok(())) => None,
                Ok(Err(detail)) => Some(detail),
                Err(_) => Some("post-commit check panicked".to_string()),
            };
            if let Some(detail) = failure_detail {
                let report = FailureReport::new(
                    FailurePhase::PostCommit,
                    None,
                    FailureCategory::PostCommitCheck,
                    detail,
                );
                match policy {
                    PostCommitFailurePolicy::KeepInstalled => {
                        post_commit_failure = Some(report);
                    }
                    PostCommitFailurePolicy::RollBack => {
                        let rollback_fault = if fault == Some(CommitFault::PostCommitRollback) {
                            Some(CommitFault::CommitThenRollback(
                                self.plan
                                    .artifacts()
                                    .iter()
                                    .next()
                                    .expect("non-empty artifact set")
                                    .id()
                                    .clone(),
                                self.plan
                                    .artifacts()
                                    .iter()
                                    .next()
                                    .expect("non-empty artifact set")
                                    .id()
                                    .clone(),
                            ))
                        } else {
                            fault.clone()
                        };
                        let (verified, rollback_failure) =
                            restore_entries(&entries, &committed, rollback_fault);
                        let mut receipt = finish_failure(
                            self,
                            lock,
                            backup_root,
                            verified,
                            rollback_failure,
                            report.clone(),
                        )?;
                        receipt.post_commit_failure = Some(report);
                        return Ok(receipt);
                    }
                }
            }
        }

        if fault == Some(CommitFault::Finalize) {
            lock.preserve();
            let report = FailureReport::new(
                FailurePhase::Finalize,
                None,
                FailureCategory::Injected,
                "injected finalize failure",
            );
            return Ok(TransactionReceipt {
                product: self.plan.product().clone(),
                release: self.plan.release().clone(),
                disposition: TransactionDisposition::Committed,
                rollback_performed: false,
                rollback_verified: false,
                cleanup: CleanupDisposition::RetainedForRecovery,
                recovery_path: Some(backup_root),
                failure: Some(report),
                rollback_failure: None,
                post_commit_failure,
            });
        }
        // Finalization is the only phase that discards the previous generation,
        // so it is the only place a running image may be scheduled for removal.
        // Reaching here means `KeepInstalled` already resolved and no rollback
        // remains possible, which is exactly the precondition the platform
        // helper requires.
        match finalize_backup_set(&self.plan, &backup_root, &entries) {
            Ok(FinalizeOutcome::Cleaned) => Ok(TransactionReceipt {
                product: self.plan.product().clone(),
                release: self.plan.release().clone(),
                disposition: TransactionDisposition::Committed,
                rollback_performed: false,
                rollback_verified: false,
                cleanup: CleanupDisposition::Cleaned,
                recovery_path: None,
                failure: None,
                rollback_failure: None,
                post_commit_failure,
            }),
            Ok(FinalizeOutcome::Deferred) => Ok(TransactionReceipt {
                product: self.plan.product().clone(),
                release: self.plan.release().clone(),
                disposition: TransactionDisposition::Committed,
                rollback_performed: false,
                rollback_verified: false,
                cleanup: CleanupDisposition::DeferredToProcessExit,
                // Nothing is retained for an operator: the old image is owned
                // and scheduled for deletion when this process exits.
                recovery_path: None,
                failure: None,
                rollback_failure: None,
                post_commit_failure,
            }),
            Err(source) => {
                lock.preserve();
                let error = Error::io("removing backup set after commit", source);
                let report = report_for_error(FailurePhase::Finalize, None, &error);
                Ok(TransactionReceipt {
                    product: self.plan.product().clone(),
                    release: self.plan.release().clone(),
                    disposition: TransactionDisposition::Committed,
                    rollback_performed: false,
                    rollback_verified: false,
                    cleanup: CleanupDisposition::RetainedForRecovery,
                    recovery_path: Some(backup_root),
                    failure: Some(report),
                    rollback_failure: None,
                    post_commit_failure,
                })
            }
        }
    }

    fn backup_members(
        &self,
        entries: &mut Vec<BackupEntry>,
        backup_root: &Path,
        fault: Option<CommitFault>,
    ) -> std::result::Result<(), (Option<MemberId>, Error)> {
        #[cfg(not(test))]
        let _ = &fault;
        let members = self.plan.artifacts().iter().cloned().collect::<Vec<_>>();
        for member in members {
            #[cfg(test)]
            if fault == Some(CommitFault::Backup(member.id().clone())) {
                return Err((
                    Some(member.id().clone()),
                    Error::injected(format!("injected backup failure at {}", member.id())),
                ));
            }
            let destination = self
                .plan
                .destination(member.id())
                .map_err(|e| (Some(member.id().clone()), e))?;
            revalidate_destination(&destination, self.plan.installation_root())
                .map_err(|e| (Some(member.id().clone()), e))?;
            // Last identity proof before the first live rename: the running
            // image is moved aside here, and that rename is the point of no
            // return for the previous generation.
            self.plan
                .revalidate_current_executable()
                .map_err(|e| (Some(member.id().clone()), e))?;
            let backup = match fs::symlink_metadata(&destination) {
                Ok(_) => {
                    let path = backup_root.join(member.destination());
                    if let Some(parent) = path.parent() {
                        fs::create_dir_all(parent).map_err(|source| {
                            (
                                Some(member.id().clone()),
                                Error::io("creating backup directory", source),
                            )
                        })?;
                    }
                    fs::rename(&destination, &path).map_err(|source| {
                        (
                            Some(member.id().clone()),
                            Error::io("backing up existing destination", source),
                        )
                    })?;
                    Some(path)
                }
                Err(source) if source.kind() == std::io::ErrorKind::NotFound => None,
                Err(source) => {
                    return Err((
                        Some(member.id().clone()),
                        Error::io("reading destination before backup", source),
                    ));
                }
            };
            entries.push(BackupEntry {
                member: member.id().clone(),
                destination,
                backup,
            });
        }
        Ok(())
    }
}

fn classify_all(
    txn: &PreparedTransaction,
    ownership: &CommitOwnership<'_>,
) -> Result<HashMap<MemberId, Ownership>> {
    let mut out = HashMap::new();
    for member in txn.artifacts().iter() {
        let destination = txn.destination(member.id())?;
        out.insert(
            member.id().clone(),
            ownership.verifier.verify(member.id(), &destination),
        );
    }
    Ok(out)
}

/// How the previous generation could be discarded after a successful commit.
#[cfg_attr(not(windows), allow(dead_code))]
enum FinalizeOutcome {
    /// Everything transaction-owned is gone.
    Cleaned,
    /// The old image is owned and scheduled for deletion after this process exits.
    Deferred,
}

/// Discards the previous generation after the caller's policy has resolved.
///
/// For an ordinary transaction this is a plain recursive removal. For a
/// self-update the previous generation is the image this very process is still
/// executing: on Unix it can be unlinked immediately, but on Windows a mapped
/// image cannot be deleted until the process exits. The old image has already
/// been renamed to a transaction-owned path by the time this runs, so the
/// platform helper can address it exactly — and it is only ever called here,
/// after `KeepInstalled` has resolved and rollback is no longer possible.
fn finalize_backup_set(
    plan: &InstallPlan,
    backup_root: &Path,
    entries: &[BackupEntry],
) -> std::io::Result<FinalizeOutcome> {
    // Only a self-update plan can hold a still-mapped generation, and only its
    // single member has a backup of the running image.
    let running_image = plan
        .current_executable()
        .and_then(|_| entries.iter().find_map(|entry| entry.backup.clone()));
    #[cfg(windows)]
    if let Some(backup) = running_image {
        // The image was renamed aside rather than deleted, which is what kept
        // it rollback-addressable until this point. Scheduling its removal is
        // irreversible, so it happens only once the policy decision is final.
        self_replace::self_delete_at(&backup).map_err(|source| {
            std::io::Error::other(format!(
                "scheduling removal of the replaced running image: {source}"
            ))
        })?;
        // The scheduled image still occupies its slot, so only an
        // empty-parent removal is available until the process exits.
        return match fs::remove_dir(backup_root) {
            Ok(()) => Ok(FinalizeOutcome::Deferred),
            Err(source) if source.kind() == std::io::ErrorKind::DirectoryNotEmpty => {
                Ok(FinalizeOutcome::Deferred)
            }
            Err(source) => Err(source),
        };
    }
    #[cfg(not(windows))]
    let _ = running_image;
    fs::remove_dir_all(backup_root)?;
    Ok(FinalizeOutcome::Cleaned)
}

fn revalidate_ownership_locked(
    txn: &PreparedTransaction,
    ownership: &CommitOwnership<'_>,
    preflight: &HashMap<MemberId, Ownership>,
) -> Result<()> {
    for member in txn.artifacts().iter() {
        let destination = txn.destination(member.id())?;
        let locked = ownership.verifier.verify(member.id(), &destination);
        let first = preflight
            .get(member.id())
            .copied()
            .unwrap_or(Ownership::Unknown);
        // Any change between preflight and locked classification fails: the
        // live deployment moved under us.
        if locked != first {
            return Err(Error::DestinationConflict { destination });
        }
        match locked {
            Ownership::Owned => {}
            Ownership::Absent if ownership.absent == AbsentPolicy::AllowCreate => {}
            Ownership::Absent | Ownership::Foreign | Ownership::Unknown => {
                return Err(Error::DestinationConflict { destination });
            }
        }
        // Parent containment is re-proven under lock; missing parents fail
        // without creating anything.
        require_ready_parent(&destination, txn.plan.installation_root())?;
        revalidate_destination(&destination, txn.plan.installation_root())?;
    }
    // The bound current-executable object is re-proven under the same lock, so
    // a swapped image between preflight and mutation is detected here too.
    txn.plan.revalidate_current_executable()?;
    Ok(())
}

fn revalidate_staged_under_lock(
    txn: &PreparedTransaction,
    verified_digests: &HashMap<MemberId, [u8; 32]>,
) -> std::result::Result<(), (Option<MemberId>, Error)> {
    for member in txn.artifacts().iter() {
        let expected = verified_digests.get(member.id()).ok_or_else(|| {
            (
                Some(member.id().clone()),
                Error::VerificationFailed(format!(
                    "member {} has no verified digest; commit requires verified integrity",
                    member.id()
                )),
            )
        })?;
        let staged = txn
            .staged_path(member.id())
            .map_err(|e| (Some(member.id().clone()), e))?;
        let metadata = fs::symlink_metadata(&staged).map_err(|source| {
            (
                Some(member.id().clone()),
                Error::io("reading staged candidate", source),
            )
        })?;
        if !metadata.is_file() {
            return Err((
                Some(member.id().clone()),
                Error::VerificationFailed(format!(
                    "staged member {} is not a regular file",
                    member.id()
                )),
            ));
        }
        #[cfg(unix)]
        {
            use std::os::unix::fs::MetadataExt;
            if metadata.nlink() != 1 {
                return Err((
                    Some(member.id().clone()),
                    Error::VerificationFailed(format!(
                        "staged member {} is hard-linked",
                        member.id()
                    )),
                ));
            }
        }
        let actual = hash_file(&staged).map_err(|e| (Some(member.id().clone()), e))?;
        if &actual != expected {
            return Err((
                Some(member.id().clone()),
                Error::VerificationFailed(format!(
                    "staged member {} changed after verification",
                    member.id()
                )),
            ));
        }
    }
    Ok(())
}

fn create_backup_directory(root: &Path) -> Result<PathBuf> {
    use std::time::{SystemTime, UNIX_EPOCH};
    for _ in 0..32 {
        let nonce = NEXT_BACKUP_NONCE.fetch_add(1, Ordering::Relaxed);
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0);
        let path = root.join(format!(
            ".eggup-backup-{}-{nonce}-{nanos}",
            std::process::id()
        ));
        match fs::create_dir(&path) {
            Ok(()) => {
                #[cfg(unix)]
                {
                    use std::os::unix::fs::PermissionsExt;
                    let _ = fs::set_permissions(&path, fs::Permissions::from_mode(0o700));
                }
                return Ok(path);
            }
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
            Err(source) => return Err(Error::io("creating backup set", source)),
        }
    }
    Err(Error::invalid("could not create a unique backup directory"))
}

fn revalidate_destination(destination: &Path, root: &Path) -> Result<()> {
    let root = fs::canonicalize(root)
        .map_err(|source| Error::io("canonicalizing install root", source))?;
    let mut ancestor = destination;
    while !ancestor.exists() {
        ancestor = ancestor
            .parent()
            .ok_or_else(|| Error::invalid("destination has no existing ancestor"))?;
    }
    let canonical_ancestor = fs::canonicalize(ancestor)
        .map_err(|source| Error::io("canonicalizing destination ancestor", source))?;
    if !canonical_ancestor.starts_with(&root) {
        return Err(Error::DestinationConflict {
            destination: destination.to_path_buf(),
        });
    }
    match fs::symlink_metadata(destination) {
        Ok(metadata) => {
            if !metadata.is_file() {
                return Err(Error::DestinationConflict {
                    destination: destination.to_path_buf(),
                });
            }
            #[cfg(unix)]
            {
                use std::os::unix::fs::MetadataExt;
                if metadata.nlink() != 1 {
                    return Err(Error::DestinationConflict {
                        destination: destination.to_path_buf(),
                    });
                }
            }
        }
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => {}
        Err(source) => return Err(Error::io("reading destination during revalidation", source)),
    }
    Ok(())
}

/// Requires that a live destination's parent already exists, is a real
/// directory beneath the installation root, and contains no symlink ambiguity.
///
/// Unlike the pre-corrective engine, this never calls `create_dir_all` on an
/// unchecked live path. Missing parents are an actionable error; symlink or
/// otherwise ambiguous parents fail closed.
fn require_ready_parent(destination: &Path, root: &Path) -> Result<()> {
    let parent = destination
        .parent()
        .ok_or_else(|| Error::invalid("destination has no parent"))?;
    let parent_meta = fs::symlink_metadata(parent).map_err(|source| {
        if source.kind() == std::io::ErrorKind::NotFound {
            Error::invalid(format!(
                "destination parent does not exist: {}",
                parent.display()
            ))
        } else {
            Error::io("reading destination parent", source)
        }
    })?;
    if parent_meta.file_type().is_symlink() {
        return Err(Error::DestinationConflict {
            destination: destination.to_path_buf(),
        });
    }
    if !parent_meta.is_dir() {
        return Err(Error::DestinationConflict {
            destination: destination.to_path_buf(),
        });
    }
    let canonical_root = fs::canonicalize(root)
        .map_err(|source| Error::io("canonicalizing install root", source))?;
    let canonical_parent = fs::canonicalize(parent)
        .map_err(|source| Error::io("canonicalizing destination parent", source))?;
    if !canonical_parent.starts_with(&canonical_root) {
        return Err(Error::DestinationConflict {
            destination: destination.to_path_buf(),
        });
    }
    // Walk ancestors of the parent to reject any symlink component, even when
    // the canonical target remains beneath the root.
    let mut cursor = Some(parent);
    while let Some(p) = cursor {
        if p == root {
            break;
        }
        let meta = fs::symlink_metadata(p)
            .map_err(|source| Error::io("reading destination ancestor", source))?;
        if meta.file_type().is_symlink() {
            return Err(Error::DestinationConflict {
                destination: destination.to_path_buf(),
            });
        }
        cursor = p.parent();
        if let Some(next) = cursor {
            if next.as_os_str().is_empty() {
                break;
            }
        } else {
            break;
        }
    }
    Ok(())
}

fn restore_entries(
    entries: &[BackupEntry],
    committed: &HashSet<MemberId>,
    fault: Option<CommitFault>,
) -> (bool, Option<FailureReport>) {
    let mut verified = true;
    let mut rollback_failure: Option<FailureReport> = None;
    for entry in entries.iter().rev() {
        let rollback_target = match fault.as_ref() {
            Some(CommitFault::CommitThenRollback(_, id)) => Some(id),
            _ => None,
        };
        if rollback_target == Some(&entry.member) {
            verified = false;
            rollback_failure = Some(FailureReport::new(
                FailurePhase::Rollback,
                Some(entry.member.clone()),
                FailureCategory::Injected,
                format!("injected rollback failure at {}", entry.member),
            ));
            continue;
        }
        if entry.backup.is_some() || committed.contains(&entry.member) {
            if let Ok(metadata) = fs::symlink_metadata(&entry.destination) {
                if !metadata.is_file() || fs::remove_file(&entry.destination).is_err() {
                    verified = false;
                    if rollback_failure.is_none() {
                        rollback_failure = Some(FailureReport::new(
                            FailurePhase::Rollback,
                            Some(entry.member.clone()),
                            FailureCategory::Filesystem,
                            format!("could not remove partial member {}", entry.member),
                        ));
                    }
                    continue;
                }
            }
        }
        if let Some(backup) = &entry.backup {
            if fs::rename(backup, &entry.destination).is_err() {
                verified = false;
                if rollback_failure.is_none() {
                    rollback_failure = Some(FailureReport::new(
                        FailurePhase::Rollback,
                        Some(entry.member.clone()),
                        FailureCategory::Filesystem,
                        format!("could not restore backup for {}", entry.member),
                    ));
                }
            }
        }
    }
    for entry in entries {
        let restored = match &entry.backup {
            Some(_) => fs::symlink_metadata(&entry.destination).is_ok(),
            None => {
                !committed.contains(&entry.member)
                    || fs::symlink_metadata(&entry.destination).is_err()
            }
        };
        if !restored {
            verified = false;
        }
    }
    (verified, rollback_failure)
}

fn finish_failure_no_mutation(
    transaction: PreparedTransaction,
    lock: MutationLock,
    nothing_mutated: bool,
    report: FailureReport,
) -> RecoveryResult<TransactionReceipt> {
    debug_assert!(nothing_mutated);
    drop(lock);
    Ok(TransactionReceipt {
        product: transaction.plan.product().clone(),
        release: transaction.plan.release().clone(),
        disposition: TransactionDisposition::RolledBack,
        rollback_performed: false,
        rollback_verified: true,
        cleanup: CleanupDisposition::Cleaned,
        recovery_path: None,
        failure: Some(report),
        rollback_failure: None,
        post_commit_failure: None,
    })
}

fn finish_failure(
    transaction: PreparedTransaction,
    mut lock: MutationLock,
    backup_root: PathBuf,
    rollback_verified: bool,
    rollback_failure: Option<FailureReport>,
    report: FailureReport,
) -> RecoveryResult<TransactionReceipt> {
    if rollback_verified && rollback_failure.is_none() {
        match fs::remove_dir_all(&backup_root) {
            Ok(()) => {
                return Ok(TransactionReceipt {
                    product: transaction.plan.product().clone(),
                    release: transaction.plan.release().clone(),
                    disposition: TransactionDisposition::RolledBack,
                    rollback_performed: true,
                    rollback_verified: true,
                    cleanup: CleanupDisposition::Cleaned,
                    recovery_path: None,
                    failure: Some(report),
                    rollback_failure: None,
                    post_commit_failure: None,
                });
            }
            // Restoration itself fully succeeded; only the backup-set cleanup
            // failed. Mirror the committed path: keep the receipt `RolledBack`,
            // carry the verified flag truthfully, and attribute the problem to
            // the `Finalize` phase in `failure()` rather than leaving a
            // `RecoveryRequired` receipt that claims unverified rollback with
            // no rollback failure recorded.
            Err(source) => {
                let error = Error::io("removing backup set after rollback", source);
                return Ok(TransactionReceipt {
                    product: transaction.plan.product().clone(),
                    release: transaction.plan.release().clone(),
                    disposition: TransactionDisposition::RolledBack,
                    rollback_performed: true,
                    rollback_verified: true,
                    cleanup: CleanupDisposition::RetainedForRecovery,
                    recovery_path: Some(backup_root),
                    failure: Some(report_for_error(FailurePhase::Finalize, None, &error)),
                    rollback_failure: None,
                    post_commit_failure: None,
                });
            }
        }
    }
    lock.preserve();
    Ok(TransactionReceipt {
        product: transaction.plan.product().clone(),
        release: transaction.plan.release().clone(),
        disposition: TransactionDisposition::RecoveryRequired,
        rollback_performed: true,
        rollback_verified,
        cleanup: CleanupDisposition::RetainedForRecovery,
        recovery_path: Some(backup_root),
        failure: Some(report),
        rollback_failure,
        post_commit_failure: None,
    })
}