cursor-setup-system 0.0.84

//! The bounded backup pool, and the reference that names one slot in it.
//!
//! Every mutation captures the exact pre-operation state before its first
//! write. That capture is what `restore` restores and what recovery from
//! [`Phase::Prepared`](crate::setup_core::journal::Phase::Prepared) rewinds to.
//!
//! A slot's completion marker is written **last**. A slot without one was
//! interrupted mid-capture, and [`Pool::partial_slots`] surfaces it so planning
//! can refuse instead of restoring from a half-copied tree — the failure mode a
//! marker written first would hide.
//!
//! Slots are numbered by a monotonic sequence, not by a timestamp. Ordering that
//! depends on a clock reorders itself when the clock moves, and "the last
//! backup" would then name a different slot than it did a moment earlier.

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

use serde::{Deserialize, Serialize};

use crate::setup_core::error::{Error, ReasonCode, Result};
use crate::setup_core::lock;

/// The pool directory name inside the control directory.
pub const POOL_DIRECTORY_NAME: &str = "backups";

/// The marker file written last inside a completed slot.
pub const SLOT_MARKER_NAME: &str = "slot.json";

/// The subdirectory holding the captured tree.
pub const SLOT_PAYLOAD_NAME: &str = "payload";

/// The marker naming a slot that retention may not reclaim.
///
/// Beside the record rather than inside it, on purpose. The record states what
/// was *captured*; a hold states whether retention may take it back. Those are
/// two different facts about one slot, and writing the second into the first
/// would mean editing a completed capture every time someone changed their mind
/// about keeping it.
///
/// The file holds the reason the hold was placed. A pool can be held by more
/// than one run at a time, and a refusal that says only *which* slots are held
/// leaves the next caller releasing one blind — possibly one another run is
/// still depending on.
pub const SLOT_HELD_NAME: &str = "HELD";

/// The schema this kernel writes and is willing to read.
pub const SLOT_SCHEMA: u32 = 1;

/// Complete native snapshots require a reader that understands their coverage.
pub const NATIVE_SLOT_SCHEMA: u32 = 2;

/// A target-bound reference to one backup slot.
///
/// The reference is meaningful only against the target it was captured from;
/// it is never a global identifier.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize)]
pub struct BackupRef(String);

/// A deserialized reference goes through [`BackupRef::parse`]: the value names
/// a directory under the pool, so a persisted record may not mint one this
/// kernel would refuse.
impl<'de> Deserialize<'de> for BackupRef {
    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let text = String::deserialize(deserializer)?;
        Self::parse(&text).map_err(serde::de::Error::custom)
    }
}

impl BackupRef {
    /// Build a reference from a sequence number.
    #[must_use]
    pub fn from_sequence(sequence: u64) -> Self {
        Self(format!("slot-{sequence:012}"))
    }

    /// Parse a reference, rejecting anything that is not one this kernel mints.
    ///
    /// The rejection matters: a reference is used as a directory name, so a
    /// value carrying separators or parent components would let a caller name a
    /// path outside the pool.
    ///
    /// # Errors
    ///
    /// Returns [`ReasonCode::IntegrityMismatch`] for a malformed reference.
    pub fn parse(text: &str) -> Result<Self> {
        let Some(digits) = text.strip_prefix("slot-") else {
            return Err(Error::new(
                ReasonCode::IntegrityMismatch,
                format!("{text:?} is not a backup reference"),
            ));
        };
        if digits.len() != 12 || !digits.bytes().all(|byte| byte.is_ascii_digit()) {
            return Err(Error::new(
                ReasonCode::IntegrityMismatch,
                format!("{text:?} is not a backup reference"),
            ));
        }
        Ok(Self(text.to_owned()))
    }

    /// The wire spelling.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }

    fn sequence(&self) -> u64 {
        self.0
            .strip_prefix("slot-")
            .and_then(|digits| digits.parse().ok())
            .unwrap_or(0)
    }
}

/// What one completed slot records about itself.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SlotRecord {
    /// Schema of this record.
    pub schema_version: u32,
    /// The reference naming this slot.
    pub backup_ref: BackupRef,
    /// The operation that captured it.
    pub operation: String,
    /// The operation id that captured it.
    pub operation_id: String,
    /// The target identity digest at capture time.
    pub target_identity_digest: String,
    /// The setup identity in effect at capture time, when one was stamped.
    pub setup_id: Option<String>,
    /// The definition digest that setup was identified by, when one was stamped.
    ///
    /// Added after the schema was in use, and deliberately not a schema bump: a
    /// slot written before it is still a complete, restorable capture, and
    /// refusing to read one would trade a recoverable target for a field.
    #[serde(default)]
    pub setup_definition_digest: Option<String>,
    /// Verified complete native coverage, absent for write-only backups.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub native_snapshot: Option<crate::setup_core::native_snapshot::NativeSnapshot>,
    /// Installation ownership before capture, independent of preserved coverage.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub previous_written_paths: Option<Vec<String>>,
    /// Exact pre-operation provider metadata for complete recovery.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub previous_provider_state: Option<Box<crate::setup_core::stamp::ProviderState>>,
}

/// The bounded pool of backup slots for one target.
#[derive(Debug, Clone)]
pub struct Pool {
    root: PathBuf,
    capacity: usize,
}

impl Pool {
    /// Open the pool inside a control directory.
    ///
    /// # Errors
    ///
    /// Returns [`ReasonCode::StateUnavailable`] if the pool directory cannot be
    /// created, and [`ReasonCode::IntegrityMismatch`] if `capacity` is zero — a
    /// pool that keeps nothing would let a mutation proceed with no way back.
    pub fn open(control_directory: &Path, capacity: usize) -> Result<Self> {
        if capacity == 0 {
            return Err(Error::new(
                ReasonCode::IntegrityMismatch,
                "a backup pool with no slots cannot support restore",
            ));
        }
        let root = control_directory.join(POOL_DIRECTORY_NAME);
        fs::create_dir_all(&root).map_err(|source| {
            Error::new(
                ReasonCode::StateUnavailable,
                format!("cannot create the backup pool at {}", root.display()),
            )
            .with_source(source)
        })?;
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            fs::set_permissions(&root, fs::Permissions::from_mode(0o700)).map_err(|source| {
                Error::new(
                    ReasonCode::StateUnavailable,
                    "cannot protect local recovery storage",
                )
                .with_source(source)
            })?;
        }
        Ok(Self { root, capacity })
    }

    /// Read an existing pool without creating one.
    ///
    /// `open` makes the pool directory, which is right when a mutation is about
    /// to capture into it and wrong when a command is only reporting: a caller
    /// that asks what backups exist should not thereby create the place they
    /// would live. `list` and friends work on the returned value either way,
    /// because a pool whose root is absent simply has no slots.
    ///
    /// # Errors
    ///
    /// Returns [`ReasonCode::IntegrityMismatch`] if `capacity` is zero, for the
    /// same reason [`Pool::open`] does.
    pub fn observe(control_directory: &Path, capacity: usize) -> Result<Self> {
        if capacity == 0 {
            return Err(Error::new(
                ReasonCode::IntegrityMismatch,
                "a backup pool with no slots cannot support restore",
            ));
        }
        Ok(Self {
            root: control_directory.join(POOL_DIRECTORY_NAME),
            capacity,
        })
    }

    /// Completed slots, newest first.
    ///
    /// # Errors
    ///
    /// Returns [`ReasonCode::StateUnavailable`] if the pool cannot be listed.
    pub fn list(&self) -> Result<Vec<SlotRecord>> {
        let mut records = Vec::new();
        for slot in self.slot_directories()? {
            if let Some(record) = read_record(&slot)? {
                records.push(record);
            }
        }
        // Newest first: the reverse of the minted sequence, never a clock.
        records.sort_by_key(|record| std::cmp::Reverse(record.backup_ref.sequence()));
        Ok(records)
    }

    /// The newest completed slot, if the pool holds one.
    ///
    /// This is what `restore` without an explicit reference restores.
    ///
    /// # Errors
    ///
    /// Propagates the refusal from [`Pool::list`].
    pub fn latest(&self) -> Result<Option<SlotRecord>> {
        Ok(self.list()?.into_iter().next())
    }

    /// Slots that were interrupted before their marker was written.
    ///
    /// # Errors
    ///
    /// Returns [`ReasonCode::StateUnavailable`] if the pool cannot be listed.
    pub fn partial_slots(&self) -> Result<Vec<PathBuf>> {
        let mut partial = Vec::new();
        for slot in self.slot_directories()? {
            if !slot.join(SLOT_MARKER_NAME).exists() {
                partial.push(slot);
            }
        }
        partial.sort();
        Ok(partial)
    }

    /// Move incomplete captures aside without discarding their bytes.
    ///
    /// The caller must hold the target lock and establish that no journal is
    /// published. An unpublished capture cannot have authorized target writes.
    ///
    /// # Errors
    ///
    /// Returns [`ReasonCode::StateUnavailable`] if listing or renaming fails.
    pub fn quarantine_partial(&self) -> Result<usize> {
        let partial = self.partial_slots()?;
        if partial.is_empty() {
            return Ok(0);
        }
        let archive = self.root.join("incomplete");
        fs::create_dir_all(&archive).map_err(|source| {
            Error::new(
                ReasonCode::StateUnavailable,
                "cannot create incomplete capture archive",
            )
            .with_source(source)
        })?;
        for slot in &partial {
            let name = slot
                .file_name()
                .ok_or_else(|| {
                    Error::new(ReasonCode::IntegrityMismatch, "a capture slot has no name")
                })?
                .to_string_lossy();
            let mut sequence = 0_u64;
            let destination = loop {
                let candidate = archive.join(format!("{name}-{sequence}"));
                if !candidate.exists() {
                    break candidate;
                }
                sequence += 1;
            };
            fs::rename(slot, destination).map_err(|source| {
                Error::new(
                    ReasonCode::StateUnavailable,
                    "cannot quarantine incomplete capture",
                )
                .with_source(source)
            })?;
        }
        Ok(partial.len())
    }

    /// The payload directory of one completed slot.
    ///
    /// # Errors
    ///
    /// Returns [`ReasonCode::IntegrityMismatch`] if the slot is absent or
    /// incomplete. An incomplete slot is never offered as a restore source.
    pub fn payload_of(&self, backup_ref: &BackupRef) -> Result<PathBuf> {
        let slot = self.root.join(backup_ref.as_str());
        if !slot.join(SLOT_MARKER_NAME).exists() {
            return Err(Error::new(
                ReasonCode::IntegrityMismatch,
                format!("backup {} is absent or incomplete", backup_ref.as_str()),
            ));
        }
        let payload = slot.join(SLOT_PAYLOAD_NAME);
        let record = read_record(&slot)?.ok_or_else(|| {
            Error::new(
                ReasonCode::IntegrityMismatch,
                "backup completion record is absent",
            )
        })?;
        if record.backup_ref != *backup_ref {
            return Err(Error::new(
                ReasonCode::IntegrityMismatch,
                "backup reference does not match its slot",
            ));
        }
        if let Some(snapshot) = record.native_snapshot {
            snapshot.verify(&payload)?;
        }
        Ok(payload)
    }

    /// The reference the next capture will use.
    ///
    /// # Errors
    ///
    /// Returns [`ReasonCode::StateUnavailable`] if the pool cannot be listed.
    pub fn next_ref(&self) -> Result<BackupRef> {
        let highest = self
            .slot_directories()?
            .iter()
            .filter_map(|slot| slot.file_name().and_then(|name| name.to_str()))
            .filter_map(|name| BackupRef::parse(name).ok())
            .map(|reference| reference.sequence())
            .max()
            .unwrap_or(0);
        Ok(BackupRef::from_sequence(highest.saturating_add(1)))
    }

    /// Capture the named paths of `source` into a new slot and complete it.
    ///
    /// The payload is copied first and the marker written last, so an
    /// interruption leaves a slot [`Pool::partial_slots`] can name.
    ///
    /// # Only what the caller can undo
    ///
    /// `included` names the paths a restore can put back, and nothing else is
    /// copied. That is not an optimisation bolted on afterwards -- it is the
    /// only honest scope, because a restore reads exactly these paths out of
    /// the payload again. Anything else in the slot could never be restored by
    /// the code that wrote it.
    ///
    /// Capturing the whole directory instead was measured against a real Grok
    /// home: 517 MB copied -- the product's downloads, marketplace cache,
    /// vendored runtime and logs -- to protect four kilobytes of configuration,
    /// doubling the directory on the first operation and multiplying it by the
    /// slot count thereafter. It also failed outright, because a live install
    /// keeps a symlink at `bin/grok` and this copy refuses symlinks; the
    /// provider could not touch a real installation at all. Both problems were
    /// the same mistake, and both end here.
    ///
    /// A path that does not exist is skipped: a target legitimately holds only
    /// some of what a harness may own.
    ///
    /// # Errors
    ///
    /// Returns [`ReasonCode::StateUnavailable`] if the copy or the marker write
    /// fails.
    pub fn capture(
        &self,
        source: &Path,
        included: &[&str],
        record: impl FnOnce(BackupRef) -> SlotRecord,
    ) -> Result<SlotRecord> {
        let backup_ref = self.next_ref()?;
        let slot = self.root.join(backup_ref.as_str());
        let payload = slot.join(SLOT_PAYLOAD_NAME);
        fs::create_dir_all(&payload).map_err(|source_error| {
            Error::new(
                ReasonCode::StateUnavailable,
                format!("cannot create backup slot {}", slot.display()),
            )
            .with_source(source_error)
        })?;

        for relative in included {
            let from = source.join(relative);
            if !from.exists() {
                continue;
            }
            let to = payload.join(relative);
            if let Some(parent) = to.parent() {
                fs::create_dir_all(parent).map_err(|source_error| {
                    Error::new(
                        ReasonCode::StateUnavailable,
                        format!("cannot create {}", parent.display()),
                    )
                    .with_source(source_error)
                })?;
            }
            if from.is_dir() {
                copy_tree(&from, &to, &[])?;
            } else {
                copy_file(&from, &to)?;
            }
        }

        self.complete(&slot, record(backup_ref))
    }

    /// Capture and read back a complete native snapshot before publishing its marker.
    ///
    /// # Errors
    /// Refuses changed source state, failed copies and verification or retention errors.
    pub fn capture_native(
        &self,
        source: &Path,
        snapshot: &crate::setup_core::native_snapshot::NativeSnapshot,
        record: impl FnOnce(BackupRef) -> SlotRecord,
    ) -> Result<SlotRecord> {
        let backup_ref = self.next_ref()?;
        let slot = self.root.join(backup_ref.as_str());
        let payload = slot.join(SLOT_PAYLOAD_NAME);
        fs::create_dir_all(&payload).map_err(|error| {
            Error::new(
                ReasonCode::StateUnavailable,
                "cannot create native backup payload",
            )
            .with_source(error)
        })?;
        snapshot.copy_to(source, &payload)?;
        let mut record = record(backup_ref);
        record.schema_version = NATIVE_SLOT_SCHEMA;
        record.native_snapshot = Some(snapshot.clone());
        // Publish retention before completion: a completed preserved setup must
        // never enter the reclaimable rolling window, even after a lost response.
        lock::atomic_write(&slot.join(SLOT_HELD_NAME), b"preserved native setup")?;
        self.complete(&slot, record)
    }

    fn complete(&self, slot: &Path, record: SlotRecord) -> Result<SlotRecord> {
        let value = serde_json::to_value(&record).map_err(|source_error| {
            Error::new(
                ReasonCode::StateUnavailable,
                "cannot encode the backup record",
            )
            .with_source(source_error)
        })?;
        let bytes = crate::setup_core::canonical::to_canonical_bytes(&value)?;
        lock::atomic_write(&slot.join(SLOT_MARKER_NAME), &bytes)?;

        self.prune()?;
        Ok(record)
    }

    /// Keep one slot until it is explicitly released.
    ///
    /// The pool rolls: ten slots, oldest evicted. A long evidence series makes
    /// more captures than that, so the baseline it means to return to at the
    /// end is gone by the time it gets there — reported after a run of fifty
    /// captures could not restore the state it started from.
    ///
    /// A hold is the smallest thing that fixes it: one marker that eviction has
    /// to read. Against an export/import format, which is a second format to
    /// keep correct, and against a separate baseline store, which is a second
    /// place state lives. With a hold, *retention never silently changes what a
    /// `BackupRef` names* becomes a checkable statement rather than an
    /// intention.
    ///
    /// # Errors
    ///
    /// Refuses a reference this pool does not hold or an unreadable hold marker.
    /// Held snapshots are outside the rolling capacity and remain until released.
    pub fn hold(&self, backup_ref: &BackupRef, reason: &str) -> Result<bool> {
        let slot = self.root.join(backup_ref.as_str());
        if read_record(&slot)?.is_none() {
            return Err(Error::new(
                ReasonCode::InvalidTarget,
                format!(
                    "{} is not a completed slot in this pool",
                    backup_ref.as_str()
                ),
            ));
        }
        let already = self.held()?;
        if already.iter().any(|(held, _)| held == backup_ref) {
            return Ok(false);
        }
        // The reason travels with the hold. Without it a caller reading a full
        // pool knows what to release and not what releasing it would cost. The
        // write is atomic so a crash cannot leave an empty marker that reads as
        // "held for no reason".
        lock::atomic_write(&slot.join(SLOT_HELD_NAME), reason.as_bytes())?;
        Ok(true)
    }

    /// Let retention have a slot back.
    ///
    /// Answers `false` for a slot that was not held rather than refusing: a run
    /// cleaning up after itself should not have to tell "nothing to do" apart
    /// from "something is wrong", which is the same reason `remove` answers
    /// `removed: false` on a prefix it never wrote.
    ///
    /// # Errors
    ///
    /// Returns [`ReasonCode::StateUnavailable`] if the marker cannot be removed.
    pub fn release(&self, backup_ref: &BackupRef) -> Result<bool> {
        let marker = self.root.join(backup_ref.as_str()).join(SLOT_HELD_NAME);
        match lock::remove_file(&marker) {
            Ok(()) => Ok(true),
            Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(false),
            Err(source) => Err(Error::new(
                ReasonCode::StateUnavailable,
                format!("cannot release {}", marker.display()),
            )
            .with_source(source)),
        }
    }

    /// Every slot retention may not reclaim, newest first, with why.
    ///
    /// # Errors
    ///
    /// Propagates the refusal from [`Pool::list`].
    pub fn held(&self) -> Result<Vec<(BackupRef, String)>> {
        let mut out = Vec::new();
        for record in self.list()? {
            let marker = self
                .root
                .join(record.backup_ref.as_str())
                .join(SLOT_HELD_NAME);
            if let Ok(reason) = fs::read_to_string(&marker) {
                let reason = reason.trim().to_owned();
                let reason = if reason.is_empty() {
                    // A hold placed before reasons were carried, or one whose
                    // caller gave none. Saying so beats an empty parenthesis.
                    "no reason recorded".to_owned()
                } else {
                    reason
                };
                out.push((record.backup_ref, reason));
            }
        }
        Ok(out)
    }

    /// Why one slot is held, when it is.
    ///
    /// # Errors
    ///
    /// Propagates the refusal from [`Pool::list`].
    pub fn held_reason(&self, backup_ref: &BackupRef) -> Result<Option<String>> {
        Ok(self
            .held()?
            .into_iter()
            .find(|(held, _)| held == backup_ref)
            .map(|(_, reason)| reason))
    }

    /// Whether one slot is held.
    ///
    /// # Errors
    ///
    /// Never; the signature matches its neighbours so a caller reads them alike.
    pub fn is_held(&self, backup_ref: &BackupRef) -> Result<bool> {
        Ok(self
            .root
            .join(backup_ref.as_str())
            .join(SLOT_HELD_NAME)
            .is_file())
    }

    /// Drop the oldest completed slots beyond the pool capacity.
    ///
    /// Incomplete slots are never pruned: they are evidence recovery needs, and
    /// deleting them here would erase the reason planning is refusing.
    ///
    /// # Errors
    ///
    /// Returns [`ReasonCode::StateUnavailable`] if a slot cannot be removed.
    pub fn prune(&self) -> Result<()> {
        // Held slots are not counted against the bound and are never reclaimed.
        // Counting them would make a hold quietly shorten the rolling window
        // instead of protecting one capture, and the caller asked for the
        // second thing.
        let mut records = Vec::new();
        for record in self.list()? {
            // An unreadable hold marker must not read as "not held": pruning
            // past it would delete the snapshot somebody asked to keep.
            if !self.is_held(&record.backup_ref)? {
                records.push(record);
            }
        }
        for record in records.into_iter().skip(self.capacity) {
            let slot = self.root.join(record.backup_ref.as_str());
            lock::remove_dir_all(&slot).map_err(|source| {
                Error::new(
                    ReasonCode::StateUnavailable,
                    format!("cannot prune {}", slot.display()),
                )
                .with_source(source)
            })?;
        }
        Ok(())
    }

    fn slot_directories(&self) -> Result<Vec<PathBuf>> {
        let read = match fs::read_dir(&self.root) {
            Ok(read) => read,
            Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(source) => {
                return Err(Error::new(
                    ReasonCode::StateUnavailable,
                    format!("cannot list {}", self.root.display()),
                )
                .with_source(source));
            }
        };
        let mut slots = Vec::new();
        for entry in read {
            let entry = entry.map_err(|source| {
                Error::new(
                    ReasonCode::StateUnavailable,
                    format!("cannot read an entry of {}", self.root.display()),
                )
                .with_source(source)
            })?;
            let path = entry.path();
            let is_slot = path
                .file_name()
                .and_then(|name| name.to_str())
                .is_some_and(|name| BackupRef::parse(name).is_ok());
            if is_slot && path.is_dir() {
                slots.push(path);
            }
        }
        Ok(slots)
    }
}

/// Read one slot's completion marker, or report that the slot has none.
///
/// A slot without a marker is not an error here — [`Pool::partial_slots`] is
/// what names it, and [`Pool::list`] simply does not offer it.
fn read_record(slot: &Path) -> Result<Option<SlotRecord>> {
    let marker = slot.join(SLOT_MARKER_NAME);
    let bytes = match fs::read(&marker) {
        Ok(bytes) => bytes,
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(source) => {
            return Err(Error::new(
                ReasonCode::StateUnavailable,
                format!("cannot read {}", marker.display()),
            )
            .with_source(source));
        }
    };
    let record: SlotRecord = serde_json::from_slice(&bytes).map_err(|source| {
        Error::new(
            ReasonCode::StateUnavailable,
            format!("{} does not parse as a backup record", marker.display()),
        )
        .with_source(source)
    })?;
    let valid = match record.schema_version {
        SLOT_SCHEMA => record.native_snapshot.is_none(),
        NATIVE_SLOT_SCHEMA => record.native_snapshot.is_some(),
        _ => false,
    };
    if !valid {
        return Err(Error::new(
            ReasonCode::StateUnavailable,
            format!(
                "backup schema {} does not match a supported recovery record",
                record.schema_version
            ),
        ));
    }
    // Path members were written by an older self and are joined to the target
    // at recovery time; a record cannot mint members this kernel would refuse.
    if let Some(paths) = &record.previous_written_paths {
        for member in paths {
            crate::setup_core::native_snapshot::validate_path(member)?;
        }
    }
    if let Some(state) = &record.previous_provider_state {
        state.validate()?;
    }
    if let Some(snapshot) = &record.native_snapshot {
        snapshot.validate()?;
    }
    Ok(Some(record))
}

/// Copy a directory tree, skipping the named top-level entries.
///
/// Symbolic links are not followed. A link inside the source is refused rather
/// than dereferenced, because a backup that silently inlined a link's target
/// would restore different bytes than it captured.
///
/// # Errors
///
/// Returns [`ReasonCode::StateUnavailable`] on any I/O failure and
/// [`ReasonCode::IntegrityMismatch`] when a symbolic link is encountered.
pub fn copy_tree(source: &Path, destination: &Path, excluded_top_level: &[&str]) -> Result<()> {
    copy_inner(source, destination, excluded_top_level, true)
}

/// Every owned path a capture would meet and be unable to take.
///
/// A backup slot is a statement about *content*. A link is a pointer, and one
/// captured into a slot would restore as a link whose target may have moved,
/// been deleted, or been replaced -- so `copy_file` refuses it. That refusal is
/// right and stays; what was wrong is *when* it happened.
///
/// The refusal used to arrive mid-copy: the slot was created, files were
/// written into it, and then the walk met a link and stopped. The operation
/// became `partial` and left recovery and control artifacts behind, for a shape
/// that could have been recognised before anything moved. Reported from a real
/// Windows target where four Antigravity skills under `config/skills` were
/// Junctions.
///
/// This is that recognition, and it is a reading rather than a write: it walks
/// only what the provider owns, follows nothing, and names every relative path
/// it cannot take so a caller learns all of them at once instead of one per
/// attempt.
///
/// Windows Junctions and Unix symbolic links are both reparse-style entries and
/// `symlink_metadata` reports both through `is_symlink`, which is the same
/// question `copy_file` asks. One check, both systems, and no second opinion
/// about what is capturable.
///
/// # Errors
///
/// Returns [`ReasonCode::StateUnavailable`] if an owned path cannot be walked.
pub fn uncapturable(source: &Path, included: &[&str]) -> Result<Vec<String>> {
    let mut refused = Vec::new();
    for relative in included {
        let from = relative
            .split('/')
            .fold(source.to_path_buf(), |at, part| at.join(part));
        collect_uncapturable(&from, relative, &mut refused)?;
    }
    refused.sort();
    Ok(refused)
}

fn collect_uncapturable(path: &Path, relative: &str, out: &mut Vec<String>) -> Result<()> {
    let metadata = match fs::symlink_metadata(path) {
        Ok(metadata) => metadata,
        // Absent is not unsupported: a namespace that is not there is simply
        // nothing to capture, which `capture` already skips.
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
        Err(error) => {
            return Err(Error::new(
                ReasonCode::StateUnavailable,
                format!("cannot stat {}", path.display()),
            )
            .with_source(error));
        }
    };

    if metadata.is_symlink() {
        out.push(relative.to_owned());
        // Never descend through it. Following a link out of the target is how a
        // reading of what we own becomes a reading of someone else's disk.
        return Ok(());
    }
    if !metadata.is_dir() {
        return Ok(());
    }

    let entries = fs::read_dir(path).map_err(|error| {
        Error::new(
            ReasonCode::StateUnavailable,
            format!("cannot list {}", path.display()),
        )
        .with_source(error)
    })?;
    for entry in entries {
        let entry = entry.map_err(|error| {
            Error::new(
                ReasonCode::StateUnavailable,
                format!("cannot read an entry of {}", path.display()),
            )
            .with_source(error)
        })?;
        let name = entry.file_name().to_string_lossy().into_owned();
        collect_uncapturable(&entry.path(), &format!("{relative}/{name}"), out)?;
    }
    Ok(())
}

/// Copy one regular file, refusing a symlink for the same reason a tree does.
///
/// A symlink captured into a backup slot is a pointer, not the bytes it points
/// at: restoring it would recreate a link whose target may have moved, been
/// deleted, or been replaced by something else entirely. Refusing keeps a slot
/// a statement about content.
fn copy_file(from: &Path, to: &Path) -> Result<()> {
    let metadata = fs::symlink_metadata(from).map_err(|error| {
        Error::new(
            ReasonCode::StateUnavailable,
            format!("cannot stat {}", from.display()),
        )
        .with_source(error)
    })?;
    if metadata.is_symlink() {
        return Err(Error::new(
            ReasonCode::IntegrityMismatch,
            format!("{} is a symbolic link and is not captured", from.display()),
        ));
    }
    refuse_linked_destination(to)?;
    fs::copy(from, to).map_err(|error| {
        Error::new(
            ReasonCode::StateUnavailable,
            format!("cannot copy {} to {}", from.display(), to.display()),
        )
        .with_source(error)
    })?;
    Ok(())
}

/// Refuse to place bytes at `to` when a symbolic link already sits there.
///
/// `fs::copy` and `fs::create_dir_all` both follow the final component of the
/// path they are given, so a link swapped in at the destination between the
/// clear and the copy would carry the payload out of the tree it was being
/// restored into. The question is asked of the destination exactly as it is
/// asked of the source.
fn refuse_linked_destination(to: &Path) -> Result<()> {
    match fs::symlink_metadata(to) {
        Ok(metadata) if metadata.is_symlink() => Err(Error::new(
            ReasonCode::IntegrityMismatch,
            format!(
                "{} is a symbolic link and is not written through",
                to.display()
            ),
        )),
        Ok(_) => Ok(()),
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(source) => Err(Error::new(
            ReasonCode::StateUnavailable,
            format!("cannot read {}", to.display()),
        )
        .with_source(source)),
    }
}

fn copy_inner(
    source: &Path,
    destination: &Path,
    excluded_top_level: &[&str],
    at_root: bool,
) -> Result<()> {
    refuse_linked_destination(destination)?;
    fs::create_dir_all(destination).map_err(|error| {
        Error::new(
            ReasonCode::StateUnavailable,
            format!("cannot create {}", destination.display()),
        )
        .with_source(error)
    })?;
    let read = fs::read_dir(source).map_err(|error| {
        Error::new(
            ReasonCode::StateUnavailable,
            format!("cannot list {}", source.display()),
        )
        .with_source(error)
    })?;
    for entry in read {
        let entry = entry.map_err(|error| {
            Error::new(
                ReasonCode::StateUnavailable,
                format!("cannot read an entry of {}", source.display()),
            )
            .with_source(error)
        })?;
        let from = entry.path();
        let Some(name) = from.file_name().and_then(|name| name.to_str()) else {
            return Err(Error::new(
                ReasonCode::StateUnavailable,
                format!("{} has a name this kernel cannot represent", from.display()),
            ));
        };
        if at_root && excluded_top_level.contains(&name) {
            continue;
        }
        let to = destination.join(name);
        let metadata = fs::symlink_metadata(&from).map_err(|error| {
            Error::new(
                ReasonCode::StateUnavailable,
                format!("cannot stat {}", from.display()),
            )
            .with_source(error)
        })?;
        if metadata.is_symlink() {
            return Err(Error::new(
                ReasonCode::IntegrityMismatch,
                format!("{} is a symbolic link and is not captured", from.display()),
            ));
        }
        refuse_linked_destination(&to)?;
        if metadata.is_dir() {
            copy_inner(&from, &to, excluded_top_level, false)?;
        } else {
            fs::copy(&from, &to).map_err(|error| {
                Error::new(
                    ReasonCode::StateUnavailable,
                    format!("cannot copy {} to {}", from.display(), to.display()),
                )
                .with_source(error)
            })?;
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    #![allow(clippy::unwrap_used, clippy::panic)]

    use super::*;

    fn scratch(name: &str) -> PathBuf {
        let base =
            std::env::temp_dir().join(format!("setup-core-backup-{name}-{}", std::process::id()));
        let _ = fs::remove_dir_all(&base);
        fs::create_dir_all(&base).unwrap();
        base
    }

    fn record_for(backup_ref: BackupRef) -> SlotRecord {
        SlotRecord {
            schema_version: SLOT_SCHEMA,
            backup_ref,
            operation: "install".to_owned(),
            operation_id: "op_test".to_owned(),
            target_identity_digest: "sha256:target".to_owned(),
            setup_id: Some("full-auto".to_owned()),
            setup_definition_digest: Some("sha256:definition".to_owned()),
            native_snapshot: None,
            previous_written_paths: None,
            previous_provider_state: None,
        }
    }

    #[test]
    fn complete_native_slots_verify_payload_integrity_on_read() {
        let root = scratch("native-integrity");
        let source = root.join("source");
        fs::create_dir_all(source.join("skills/empty")).unwrap();
        fs::write(source.join("skills/tool.sh"), b"echo captured\n").unwrap();
        let snapshot =
            crate::setup_core::native_snapshot::NativeSnapshot::inspect(&source, &["skills"], &[]).unwrap();
        let pool = Pool::open(&root.join("control"), 3).unwrap();
        let record = pool.capture_native(&source, &snapshot, record_for).unwrap();
        assert_eq!(record.schema_version, NATIVE_SLOT_SCHEMA);
        assert_eq!(record.native_snapshot, Some(snapshot));
        let payload = pool.payload_of(&record.backup_ref).unwrap();
        assert!(payload.join("skills/empty").is_dir());
        fs::write(payload.join("skills/tool.sh"), b"corrupted").unwrap();
        assert!(pool.payload_of(&record.backup_ref).is_err());
        assert_eq!(
            fs::read(source.join("skills/tool.sh")).unwrap(),
            b"echo captured\n"
        );
    }

    /// A link swapped in where the copy would land cannot carry the payload
    /// out of the destination tree: `fs::copy` follows the final component,
    /// and the destination is checked the way the source already is.
    #[cfg(unix)]
    #[test]
    fn copy_tree_refuses_a_destination_swapped_for_a_link() {
        let root = scratch("linked-destination");
        let source = root.join("source");
        fs::create_dir_all(&source).unwrap();
        fs::write(source.join("a.md"), b"payload").unwrap();
        let outside = root.join("outside");
        fs::create_dir_all(&outside).unwrap();
        fs::write(outside.join("a.md"), b"not ours").unwrap();
        let destination = root.join("destination");
        std::os::unix::fs::symlink(&outside, &destination).unwrap();

        let error = copy_tree(&source, &destination, &[]).unwrap_err();
        assert_eq!(error.reason(), ReasonCode::IntegrityMismatch);
        assert_eq!(fs::read(outside.join("a.md")).unwrap(), b"not ours");
    }

    /// The same refusal one level down: a *member* swapped for a link, so
    /// the copy's destination is a file inside a real directory.
    #[cfg(unix)]
    #[test]
    fn copy_tree_refuses_a_member_swapped_for_a_link() {
        let root = scratch("linked-member");
        let source = root.join("source");
        fs::create_dir_all(&source).unwrap();
        fs::write(source.join("a.md"), b"payload").unwrap();
        let destination = root.join("destination");
        fs::create_dir_all(&destination).unwrap();
        std::os::unix::fs::symlink(root.join("outside.md"), destination.join("a.md")).unwrap();

        let error = copy_tree(&source, &destination, &[]).unwrap_err();
        assert_eq!(error.reason(), ReasonCode::IntegrityMismatch);
        assert!(!root.join("outside.md").exists());
    }

    #[test]
    fn a_slot_written_before_the_definition_digest_existed_still_reads() {
        // The field was added without a schema bump, so a slot captured by an
        // earlier build must still restore. Refusing one would trade a
        // recoverable target for a field that is allowed to be absent.
        let older = serde_json::json!({
            "schema_version": SLOT_SCHEMA,
            "backup_ref": "slot-000000000001",
            "operation": "install",
            "operation_id": "op_test",
            "target_identity_digest": "sha256:target",
            "setup_id": "full-auto",
        });
        let read: SlotRecord = serde_json::from_value(older).unwrap();
        assert_eq!(read.setup_id.as_deref(), Some("full-auto"));
        assert_eq!(read.setup_definition_digest, None);
    }

    #[test]
    fn a_reference_that_could_escape_the_pool_is_refused() {
        for hostile in [
            "slot-../../etc",
            "slot-1",
            "../slot-000000000001",
            "slot-00000000000a",
        ] {
            assert!(BackupRef::parse(hostile).is_err(), "accepted {hostile:?}");
        }
        assert!(BackupRef::parse("slot-000000000001").is_ok());
    }

    /// Persisted records go through `parse` too: deserialization is not a side
    /// door that mints references the parser would refuse.
    #[test]
    fn a_recorded_reference_that_could_escape_the_pool_is_refused_on_read() {
        let root = scratch("record-escape-ref");
        let slot = root.join("control/backups/slot-000000000001");
        fs::create_dir_all(&slot).unwrap();
        let marker = serde_json::json!({
            "schema_version": SLOT_SCHEMA,
            "backup_ref": "../escape",
            "operation": "install",
            "operation_id": "op_test",
            "target_identity_digest": "sha256:target",
            "setup_id": "full-auto",
        });
        fs::write(
            slot.join(SLOT_MARKER_NAME),
            serde_json::to_vec(&marker).unwrap(),
        )
        .unwrap();
        let pool = Pool::observe(&root.join("control"), 3).unwrap();
        let error = pool.list().unwrap_err();
        assert_eq!(error.reason(), ReasonCode::StateUnavailable);
    }

    /// Path members a record carries are joined to the target at recovery; a
    /// member that could escape it refuses the record, not just the member.
    #[test]
    fn a_record_listing_a_path_that_escapes_the_target_is_refused_on_read() {
        let root = scratch("record-escape-path");
        let slot = root.join("control/backups/slot-000000000001");
        fs::create_dir_all(&slot).unwrap();
        let escaped_state = serde_json::to_value(&crate::setup_core::stamp::ProviderState {
            state_schema: crate::setup_core::stamp::STATE_SCHEMA,
            protocol_version: 3,
            provider_id: "test".to_owned(),
            provider_version: "0.0.0".to_owned(),
            provider_build_digest: "sha256:build".to_owned(),
            provider_release_digest: None,
            harness_id: "test".to_owned(),
            canonical_target: "/target".to_owned(),
            target_identity_digest: "sha256:after".to_owned(),
            setup_stable_id: None,
            setup_version: None,
            setup_version_passport_digest: None,
            setup_definition_digest: None,
            component_refs: Vec::new(),
            bundle_format: None,
            bundle_digest: None,
            artifact_digest: None,
            projection_profile_digest: None,
            provider_plan_digest: None,
            operation_id: "op_test".to_owned(),
            target_precondition_digest: "sha256:target".to_owned(),
            native_ownership: Vec::new(),
            written_paths: vec!["../escape".to_owned()],
            backup_ref: None,
            previous_verified_identity: None,
            drift_state: crate::setup_core::stamp::DriftState::Unknown,
        })
        .unwrap();
        for (field, hostile) in [
            ("previous_written_paths", serde_json::json!(["../escape"])),
            ("previous_provider_state", escaped_state.clone()),
        ] {
            let mut marker = serde_json::json!({
                "schema_version": NATIVE_SLOT_SCHEMA,
                "backup_ref": "slot-000000000001",
                "operation": "install",
                "operation_id": "op_test",
                "target_identity_digest": "sha256:target",
                "native_snapshot": {
                    "roots": ["skills"],
                    "excluded": [],
                    "entries": {}
                },
            });
            marker[field] = hostile;
            fs::write(
                slot.join(SLOT_MARKER_NAME),
                serde_json::to_vec(&marker).unwrap(),
            )
            .unwrap();
            let pool = Pool::observe(&root.join("control"), 3).unwrap();
            let error = pool.list().unwrap_err();
            assert_eq!(
                error.reason(),
                ReasonCode::IntegrityMismatch,
                "{field} carried an escaping path"
            );
        }
    }

    /// A snapshot inside a record is inspected data on write, but plain JSON on
    /// read; its members get the same gate the writer applied.
    #[test]
    fn a_recorded_native_snapshot_with_an_escaping_member_is_refused() {
        let root = scratch("record-escape-snapshot");
        let slot = root.join("control/backups/slot-000000000001");
        fs::create_dir_all(&slot).unwrap();
        for hostile in [
            serde_json::json!({"roots": ["../escape"], "excluded": [], "entries": {}}),
            serde_json::json!({"roots": ["skills"], "excluded": ["../escape"], "entries": {}}),
            serde_json::json!({"roots": ["skills"], "excluded": [], "entries": {"../escape": {"digest": null, "permissions": 420}}}),
        ] {
            let marker = serde_json::json!({
                "schema_version": NATIVE_SLOT_SCHEMA,
                "backup_ref": "slot-000000000001",
                "operation": "install",
                "operation_id": "op_test",
                "target_identity_digest": "sha256:target",
                "native_snapshot": hostile,
            });
            fs::write(
                slot.join(SLOT_MARKER_NAME),
                serde_json::to_vec(&marker).unwrap(),
            )
            .unwrap();
            let pool = Pool::observe(&root.join("control"), 3).unwrap();
            assert_eq!(
                pool.list().unwrap_err().reason(),
                ReasonCode::IntegrityMismatch,
                "accepted {hostile}"
            );
        }
    }

    #[test]
    fn a_pool_with_no_slots_is_refused_at_construction() {
        let control = scratch("zero");
        let error = Pool::open(&control, 0).unwrap_err();
        assert_eq!(error.reason(), ReasonCode::IntegrityMismatch);
    }

    /// The claim `#428` asks for: a held baseline survives a run longer than
    /// the pool.
    ///
    /// The pool rolls at ten. A series that captures fifty times loses the
    /// state it started from long before it gets back to it — reported after
    /// exactly that, where the final restore had nothing to restore to.
    #[test]
    fn a_held_slot_outlives_far_more_captures_than_the_pool_holds() {
        let base = scratch("held-baseline");
        let target = base.join("target");
        fs::create_dir_all(&target).unwrap();
        fs::write(target.join("a.txt"), "the baseline").unwrap();

        let pool = Pool::open(&base.join("control"), 10).unwrap();
        let baseline = pool.capture(&target, &["a.txt"], record_for).unwrap();
        assert!(
            pool.hold(&baseline.backup_ref, "E00 baseline for the evidence series")
                .unwrap()
        );

        // Fifty more, which is five times the pool.
        for round in 0..50 {
            fs::write(target.join("a.txt"), format!("round {round}")).unwrap();
            pool.capture(&target, &["a.txt"], record_for).unwrap();
            pool.prune().unwrap();
        }

        let held = pool.held().unwrap();
        assert_eq!(held.len(), 1);
        assert_eq!(held[0].0, baseline.backup_ref);
        assert_eq!(held[0].1, "E00 baseline for the evidence series");
        assert!(
            pool.payload_of(&baseline.backup_ref).is_ok(),
            "the held baseline was reclaimed by retention"
        );
        assert_eq!(
            fs::read_to_string(pool.payload_of(&baseline.backup_ref).unwrap().join("a.txt"))
                .unwrap(),
            "the baseline",
            "the held slot survived but no longer names what it named"
        );

        // And the hold did not quietly shorten the rolling window: ten
        // unheld slots are still there beside it.
        let unheld: Vec<_> = pool
            .list()
            .unwrap()
            .into_iter()
            .filter(|r| !pool.is_held(&r.backup_ref).unwrap())
            .collect();
        assert_eq!(unheld.len(), 10);

        // Released, it becomes ordinary and the next prune reclaims it.
        assert!(pool.release(&baseline.backup_ref).unwrap());
        assert!(!pool.release(&baseline.backup_ref).unwrap());
        pool.prune().unwrap();
        assert!(pool.payload_of(&baseline.backup_ref).is_err());
    }

    #[test]
    fn preserved_slots_do_not_consume_the_rolling_capacity() {
        let base = scratch("held-full");
        let target = base.join("target");
        fs::create_dir_all(&target).unwrap();
        fs::write(target.join("a.txt"), "x").unwrap();
        let pool = Pool::open(&base.join("control"), 3).unwrap();
        let mut saved = Vec::new();
        for _ in 0..7 {
            let record = pool.capture(&target, &["a.txt"], record_for).unwrap();
            assert!(pool.hold(&record.backup_ref, "saved setup").unwrap());
            saved.push(record.backup_ref);
        }
        for _ in 0..8 {
            pool.capture(&target, &["a.txt"], record_for).unwrap();
        }
        assert_eq!(pool.list().unwrap().len(), saved.len() + 3);
        for reference in saved {
            assert!(pool.payload_of(&reference).is_ok());
            assert!(!pool.hold(&reference, "retry").unwrap());
        }
    }

    /// A reference this pool never minted is refused rather than marked.
    #[test]
    fn holding_a_slot_that_is_not_here_is_refused() {
        let base = scratch("held-absent");
        let pool = Pool::open(&base.join("control"), 3).unwrap();
        let absent = BackupRef::parse("slot-000000000009").unwrap();
        let error = pool.hold(&absent, "nothing").unwrap_err();
        assert!(error.to_string().contains("slot-000000000009"), "{error}");
        // And releasing one that is not here is still not an error.
        assert!(!pool.release(&absent).unwrap());
    }

    #[test]
    fn capture_records_the_slot_and_latest_names_it() {
        let base = scratch("capture");
        let target = base.join("target");
        fs::create_dir_all(&target).unwrap();
        fs::write(target.join("a.txt"), "one").unwrap();

        let pool = Pool::open(&base.join("control"), 3).unwrap();
        let record = pool.capture(&target, &["a.txt"], record_for).unwrap();

        assert_eq!(pool.latest().unwrap().unwrap(), record);
        let payload = pool.payload_of(&record.backup_ref).unwrap();
        assert_eq!(fs::read_to_string(payload.join("a.txt")).unwrap(), "one");
    }

    #[test]
    fn slots_are_ordered_by_sequence_so_latest_is_the_last_captured() {
        let base = scratch("order");
        let target = base.join("target");
        fs::create_dir_all(&target).unwrap();
        let pool = Pool::open(&base.join("control"), 5).unwrap();

        fs::write(target.join("a.txt"), "one").unwrap();
        let first = pool.capture(&target, &["a.txt"], record_for).unwrap();
        fs::write(target.join("a.txt"), "two").unwrap();
        let second = pool.capture(&target, &["a.txt"], record_for).unwrap();

        assert_eq!(
            pool.latest().unwrap().unwrap().backup_ref,
            second.backup_ref
        );
        let listed = pool.list().unwrap();
        assert_eq!(listed.len(), 2);
        assert_eq!(listed[1].backup_ref, first.backup_ref);

        // The chosen older slot still restores the bytes it captured.
        let payload = pool.payload_of(&first.backup_ref).unwrap();
        assert_eq!(fs::read_to_string(payload.join("a.txt")).unwrap(), "one");
    }

    #[test]
    fn a_slot_without_its_marker_is_partial_and_never_offered_for_restore() {
        let base = scratch("partial");
        let target = base.join("target");
        fs::create_dir_all(&target).unwrap();
        fs::write(target.join("a.txt"), "one").unwrap();

        let control = base.join("control");
        let pool = Pool::open(&control, 3).unwrap();
        let record = pool.capture(&target, &["a.txt"], record_for).unwrap();

        // Simulate an interruption between payload copy and marker write.
        let slot = control
            .join(POOL_DIRECTORY_NAME)
            .join(record.backup_ref.as_str());
        fs::remove_file(slot.join(SLOT_MARKER_NAME)).unwrap();

        assert_eq!(pool.partial_slots().unwrap(), vec![slot]);
        assert!(pool.list().unwrap().is_empty());
        assert_eq!(
            pool.payload_of(&record.backup_ref).unwrap_err().reason(),
            ReasonCode::IntegrityMismatch
        );
    }

    #[test]
    fn pruning_bounds_completed_slots_and_leaves_partial_evidence_alone() {
        let base = scratch("prune");
        let target = base.join("target");
        fs::create_dir_all(&target).unwrap();
        let control = base.join("control");
        let pool = Pool::open(&control, 2).unwrap();

        for index in 0..4 {
            fs::write(target.join("a.txt"), format!("{index}")).unwrap();
            pool.capture(&target, &["a.txt"], record_for).unwrap();
        }
        let listed = pool.list().unwrap();
        assert_eq!(listed.len(), 2);
        assert_eq!(listed[0].backup_ref, BackupRef::from_sequence(4));
        assert_eq!(listed[1].backup_ref, BackupRef::from_sequence(3));

        // A partial slot survives pruning because recovery still needs it.
        let partial = control.join(POOL_DIRECTORY_NAME).join("slot-000000000009");
        fs::create_dir_all(partial.join(SLOT_PAYLOAD_NAME)).unwrap();
        pool.prune().unwrap();
        assert!(partial.exists());
    }

    #[test]
    fn only_the_named_paths_are_captured() {
        // The scope is what a restore can put back, and nothing else. Capturing
        // the whole directory once copied 517 MB of a Grok home's downloads and
        // caches to protect four kilobytes of configuration -- bytes no restore
        // in this kernel would ever read again.
        let base = scratch("included");
        let target = base.join("target");
        fs::create_dir_all(target.join("skills")).unwrap();
        fs::create_dir_all(target.join("downloads")).unwrap();
        fs::write(target.join("config.toml"), "kept").unwrap();
        fs::write(target.join("skills/a.md"), "kept too").unwrap();
        fs::write(target.join("downloads/huge.bin"), "not ours").unwrap();
        fs::write(target.join("auth.json"), "secret").unwrap();

        let pool = Pool::open(&base.join("ctl"), 3).unwrap();
        let record = pool
            .capture(&target, &["config.toml", "skills"], record_for)
            .unwrap();
        let payload = pool.payload_of(&record.backup_ref).unwrap();

        assert_eq!(
            fs::read_to_string(payload.join("config.toml")).unwrap(),
            "kept"
        );
        assert_eq!(
            fs::read_to_string(payload.join("skills/a.md")).unwrap(),
            "kept too"
        );
        assert!(!payload.join("downloads").exists());
        assert!(!payload.join("auth.json").exists());
    }

    #[test]
    fn a_path_the_target_does_not_hold_is_skipped_not_fatal() {
        // A harness may own more than any one target happens to contain.
        let base = scratch("included-absent");
        let target = base.join("target");
        fs::create_dir_all(&target).unwrap();
        fs::write(target.join("config.toml"), "here").unwrap();

        let pool = Pool::open(&base.join("ctl"), 3).unwrap();
        let record = pool
            .capture(
                &target,
                &["config.toml", "never-created", "skills"],
                record_for,
            )
            .unwrap();
        let payload = pool.payload_of(&record.backup_ref).unwrap();
        assert_eq!(
            fs::read_to_string(payload.join("config.toml")).unwrap(),
            "here"
        );
        assert!(!payload.join("skills").exists());
    }

    // Creating a symlink on Windows needs a privilege a runner does not have,
    // so this is asserted where it can be.
    #[cfg(unix)]
    #[test]
    fn a_symbolic_link_is_refused_rather_than_inlined() {
        // Both shapes: named directly, and found while walking a named
        // directory. A slot has to be a statement about content, and a link is
        // a pointer whose target may since have moved or been replaced.
        let base = scratch("symlink");
        let target = base.join("target");
        fs::create_dir_all(target.join("skills")).unwrap();
        fs::write(base.join("outside.txt"), "secret").unwrap();
        std::os::unix::fs::symlink(base.join("outside.txt"), target.join("link.txt")).unwrap();
        std::os::unix::fs::symlink(base.join("outside.txt"), target.join("skills/deep.md"))
            .unwrap();

        let pool = Pool::open(&base.join("control"), 2).unwrap();
        let named = pool
            .capture(&target, &["link.txt"], record_for)
            .unwrap_err();
        assert_eq!(named.reason(), ReasonCode::IntegrityMismatch);

        let walked = pool.capture(&target, &["skills"], record_for).unwrap_err();
        assert_eq!(walked.reason(), ReasonCode::IntegrityMismatch);
    }

    // Creating a symlink on Windows needs a privilege a runner does not have,
    // so this is asserted where it can be.
    #[cfg(unix)]
    #[test]
    fn a_symbolic_link_outside_the_captured_paths_is_simply_not_seen() {
        // A live Grok install keeps a symlink at `bin/grok`, which no harness
        // here owns. Capturing the whole directory met it and refused, so the
        // provider could not touch a real installation at all. Scoped to what a
        // restore can use, the link is never reached.
        let base = scratch("symlink-outside");
        let target = base.join("target");
        fs::create_dir_all(target.join("bin")).unwrap();
        fs::write(base.join("outside.txt"), "runtime").unwrap();
        std::os::unix::fs::symlink(base.join("outside.txt"), target.join("bin/grok")).unwrap();
        fs::write(target.join("config.toml"), "ours").unwrap();

        let pool = Pool::open(&base.join("control"), 2).unwrap();
        let record = pool.capture(&target, &["config.toml"], record_for).unwrap();
        let payload = pool.payload_of(&record.backup_ref).unwrap();
        assert_eq!(
            fs::read_to_string(payload.join("config.toml")).unwrap(),
            "ours"
        );
        assert!(!payload.join("bin").exists());
    }
}