common/

copy.rs

use std::os::unix::fs::MetadataExt;

use anyhow::{Context, anyhow};
use async_recursion::async_recursion;
use throttle::get_file_iops_tokens;
use tracing::instrument;

use crate::config::DryRunMode;
use crate::filecmp;
use crate::preserve;
use crate::progress;
use crate::rm;
use crate::rm::{Settings as RmSettings, Summary as RmSummary};
use crate::walk::{self, EntryKind};

/// Error type for copy operations. See [`crate::error::OperationError`] for
/// logging conventions and rationale.
pub type Error = crate::error::OperationError<Summary>;

/// Filter condition for overwrite operations.
///
/// Used with `--overwrite-filter` to skip overwriting files that match
/// a directional condition (e.g., destination is newer than source).
#[derive(Copy, Clone, Debug, PartialEq, Eq, clap::ValueEnum)]
pub enum OverwriteFilter {
    /// Skip overwriting if the destination file is strictly newer (by mtime).
    #[value(name = "newer")]
    Newer,
}

impl std::fmt::Display for OverwriteFilter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            OverwriteFilter::Newer => write!(f, "newer"),
        }
    }
}

/// Settings controlling rsync-style `--delete` (mirror) behavior.
///
/// Present (`Some`) only when `--delete` was requested. `None` means the
/// destination is never enumerated and no pruning work is done, so the default
/// copy path pays nothing for this feature.
#[derive(Debug, Clone)]
pub struct DeleteSettings {
    /// Also remove destination entries that match an exclude pattern
    /// (rsync `--delete-excluded`). When false, excluded entries are protected.
    pub delete_excluded: bool,
}

#[derive(Debug, Clone)]
pub struct Settings {
    pub dereference: bool,
    pub fail_early: bool,
    pub overwrite: bool,
    pub overwrite_compare: filecmp::MetadataCmpSettings,
    pub overwrite_filter: Option<OverwriteFilter>,
    pub ignore_existing: bool,
    pub chunk_size: u64,
    /// Skip special files (sockets, FIFOs, devices) without error.
    pub skip_specials: bool,
    /// Buffer size for remote copy file transfer operations in bytes.
    ///
    /// This is only used for remote copy operations and controls the buffer size
    /// when copying data between files and network streams. The actual buffer is
    /// capped to the file size to avoid over-allocation for small files.
    pub remote_copy_buffer_size: usize,
    /// filter settings for include/exclude patterns
    pub filter: Option<crate::filter::FilterSettings>,
    /// dry-run mode for previewing operations
    pub dry_run: Option<crate::config::DryRunMode>,
    /// rsync-style `--delete` settings; `None` disables deletion entirely.
    pub delete: Option<DeleteSettings>,
}

/// Summary with the appropriate `*_skipped` counter set to 1 for the given entry kind.
/// Special files count as `files_skipped` to match the historical mapping used
/// when filters skip an entry (`specials_skipped` is reserved for `--skip-specials`).
fn skipped_summary_for(kind: EntryKind) -> Summary {
    match kind {
        EntryKind::Dir => Summary {
            directories_skipped: 1,
            ..Default::default()
        },
        EntryKind::Symlink => Summary {
            symlinks_skipped: 1,
            ..Default::default()
        },
        EntryKind::File | EntryKind::Special => Summary {
            files_skipped: 1,
            ..Default::default()
        },
    }
}

/// Result of checking if an empty directory should be cleaned up.
/// Used when filtering is active and a directory we created ended up empty.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EmptyDirAction {
    /// keep the directory (directly matched or no filter active)
    Keep,
    /// directory was only traversed, remove it
    Remove,
    /// dry-run mode, don't count this directory in summary
    DryRunSkip,
}

/// Determine what to do with an empty directory when filtering is active.
///
/// This is called when we created a directory but nothing was copied into it
/// (no files, symlinks, or child directories). The decision depends on whether
/// the directory itself was directly matched by an include pattern, or if we
/// only entered it to look for potential matches inside.
///
/// # Arguments
/// * `filter` - the active filter settings (None means no filtering)
/// * `we_created_dir` - whether we created this directory (vs it already existed)
/// * `anything_copied` - whether any content was copied into this directory
/// * `relative_path` - path relative to the source root (for pattern matching)
/// * `is_root` - whether this is the root (user-specified) source directory
/// * `is_dry_run` - whether we're in dry-run mode
pub fn check_empty_dir_cleanup(
    filter: Option<&crate::filter::FilterSettings>,
    we_created_dir: bool,
    anything_copied: bool,
    relative_path: &std::path::Path,
    is_root: bool,
    is_dry_run: bool,
) -> EmptyDirAction {
    // if no filter active or something was copied, keep the directory
    if filter.is_none() || anything_copied {
        return EmptyDirAction::Keep;
    }
    // if we didn't create this directory, don't remove it
    if !we_created_dir {
        return EmptyDirAction::Keep;
    }
    // never remove the root directory — it's the user-specified source
    if is_root {
        return EmptyDirAction::Keep;
    }
    // filter is guaranteed to be Some here (checked above)
    let f = filter.unwrap();
    // check if directory directly matches include pattern
    if f.directly_matches_include(relative_path, true) {
        return EmptyDirAction::Keep;
    }
    // directory was only traversed for potential matches
    if is_dry_run {
        EmptyDirAction::DryRunSkip
    } else {
        EmptyDirAction::Remove
    }
}

#[instrument]
pub fn is_file_type_same(md1: &std::fs::Metadata, md2: &std::fs::Metadata) -> bool {
    let ft1 = md1.file_type();
    let ft2 = md2.file_type();
    ft1.is_dir() == ft2.is_dir()
        && ft1.is_file() == ft2.is_file()
        && ft1.is_symlink() == ft2.is_symlink()
}

#[instrument(skip(prog_track, src_metadata, settings, preserve))]
pub async fn copy_file(
    prog_track: &'static progress::Progress,
    src: &std::path::Path,
    dst: &std::path::Path,
    src_metadata: &std::fs::Metadata,
    settings: &Settings,
    preserve: &preserve::Settings,
    is_fresh: bool,
) -> Result<Summary, Error> {
    // check --ignore-existing before dry-run so dry-run output reflects actual behavior.
    // use symlink_metadata to detect dangling symlinks too (Path::exists follows symlinks)
    if !is_fresh
        && settings.ignore_existing
        && crate::walk::run_metadata_probed(
            congestion::Side::Destination,
            congestion::MetadataOp::Stat,
            tokio::fs::symlink_metadata(dst),
        )
        .await
        .is_ok()
    {
        if let Some(mode) = settings.dry_run {
            match mode {
                DryRunMode::Brief => {}
                DryRunMode::All => println!("skip file {:?}", dst),
                DryRunMode::Explain => println!("skip file {:?} (destination exists)", dst),
            }
        }
        tracing::debug!("destination exists, skipping (--ignore-existing)");
        prog_track.files_unchanged.inc();
        return Ok(Summary {
            files_unchanged: 1,
            ..Default::default()
        });
    }
    // handle dry-run mode for files
    if settings.dry_run.is_some() {
        crate::dry_run::report_action("copy", src, Some(dst), "file");
        return Ok(Summary {
            files_copied: 1,
            bytes_copied: src_metadata.len(),
            ..Default::default()
        });
    }
    tracing::debug!("opening 'src' for reading and 'dst' for writing");
    get_file_iops_tokens(settings.chunk_size, src_metadata.size()).await;
    let mut rm_summary = RmSummary::default();
    if !is_fresh && dst.exists() {
        if settings.overwrite {
            tracing::debug!("file exists, check if it's identical");
            let dst_metadata = crate::walk::run_metadata_probed(
                congestion::Side::Destination,
                congestion::MetadataOp::Stat,
                tokio::fs::symlink_metadata(dst),
            )
            .await
            .with_context(|| format!("failed reading metadata from {:?}", &dst))
            .map_err(|err| Error::new(err, Default::default()))?;
            if is_file_type_same(src_metadata, &dst_metadata) {
                if filecmp::metadata_equal(&settings.overwrite_compare, src_metadata, &dst_metadata)
                {
                    tracing::debug!("file is identical, skipping");
                    prog_track.files_unchanged.inc();
                    return Ok(Summary {
                        files_unchanged: 1,
                        ..Default::default()
                    });
                }
                if let Some(OverwriteFilter::Newer) = settings.overwrite_filter
                    && filecmp::dest_is_newer(src_metadata, &dst_metadata)
                {
                    tracing::debug!("dest is newer than source, skipping");
                    prog_track.files_unchanged.inc();
                    return Ok(Summary {
                        files_unchanged: 1,
                        ..Default::default()
                    });
                }
            }
            tracing::info!("file is different, removing existing file");
            // note tokio::fs::overwrite cannot handle this path being e.g. a directory
            rm_summary = rm::rm(
                prog_track,
                dst,
                &RmSettings {
                    fail_early: settings.fail_early,
                    filter: None,
                    dry_run: None,
                    time_filter: None,
                },
            )
            .await
            .map_err(|err| {
                let rm_summary = err.summary;
                let copy_summary = Summary {
                    rm_summary,
                    ..Default::default()
                };
                Error::new(err.source, copy_summary)
            })?;
        } else {
            return Err(Error::new(
                anyhow!(
                    "destination {:?} already exists, did you intend to specify --overwrite?",
                    dst
                ),
                Default::default(),
            ));
        }
    }
    tracing::debug!("copying data");
    let mut copy_summary = Summary {
        rm_summary,
        ..Default::default()
    };
    tokio::fs::copy(src, dst)
        .await
        .with_context(|| format!("failed copying {:?} to {:?}", &src, &dst))
        .map_err(|err| Error::new(err, copy_summary))?;
    prog_track.files_copied.inc();
    prog_track.bytes_copied.add(src_metadata.len());
    tracing::debug!("setting permissions");
    preserve::set_file_metadata(preserve, src_metadata, dst)
        .await
        .map_err(|err| Error::new(err, copy_summary))?;
    // we mark files as "copied" only after all metadata is set as well
    copy_summary.bytes_copied += src_metadata.len();
    copy_summary.files_copied += 1;
    Ok(copy_summary)
}

#[derive(Copy, Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
pub struct Summary {
    pub bytes_copied: u64,
    pub files_copied: usize,
    pub symlinks_created: usize,
    pub directories_created: usize,
    pub files_unchanged: usize,
    pub symlinks_unchanged: usize,
    pub directories_unchanged: usize,
    pub files_skipped: usize,
    pub symlinks_skipped: usize,
    pub directories_skipped: usize,
    pub specials_skipped: usize,
    pub rm_summary: RmSummary,
}

impl std::ops::Add for Summary {
    type Output = Self;
    fn add(self, other: Self) -> Self {
        Self {
            bytes_copied: self.bytes_copied + other.bytes_copied,
            files_copied: self.files_copied + other.files_copied,
            symlinks_created: self.symlinks_created + other.symlinks_created,
            directories_created: self.directories_created + other.directories_created,
            files_unchanged: self.files_unchanged + other.files_unchanged,
            symlinks_unchanged: self.symlinks_unchanged + other.symlinks_unchanged,
            directories_unchanged: self.directories_unchanged + other.directories_unchanged,
            files_skipped: self.files_skipped + other.files_skipped,
            symlinks_skipped: self.symlinks_skipped + other.symlinks_skipped,
            directories_skipped: self.directories_skipped + other.directories_skipped,
            specials_skipped: self.specials_skipped + other.specials_skipped,
            rm_summary: self.rm_summary + other.rm_summary,
        }
    }
}

impl std::fmt::Display for Summary {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        write!(
            f,
            "copy:\n\
            -----\n\
            bytes copied: {}\n\
            files copied: {}\n\
            symlinks created: {}\n\
            directories created: {}\n\
            files unchanged: {}\n\
            symlinks unchanged: {}\n\
            directories unchanged: {}\n\
            files skipped: {}\n\
            symlinks skipped: {}\n\
            directories skipped: {}\n\
            specials skipped: {}\n\
            \n\
            delete:\n\
            -------\n\
            {}",
            bytesize::ByteSize(self.bytes_copied),
            self.files_copied,
            self.symlinks_created,
            self.directories_created,
            self.files_unchanged,
            self.symlinks_unchanged,
            self.directories_unchanged,
            self.files_skipped,
            self.symlinks_skipped,
            self.directories_skipped,
            self.specials_skipped,
            &self.rm_summary,
        )
    }
}

/// Public entry point for copy operations.
/// Internally delegates to copy_internal with source_root tracking for proper filter matching.
#[instrument(skip(prog_track, settings, preserve))]
pub async fn copy(
    prog_track: &'static progress::Progress,
    src: &std::path::Path,
    dst: &std::path::Path,
    settings: &Settings,
    preserve: &preserve::Settings,
    is_fresh: bool,
) -> Result<Summary, Error> {
    copy_with_filter_base(
        prog_track,
        src,
        dst,
        settings,
        preserve,
        is_fresh,
        std::path::Path::new(""),
    )
    .await
}

/// Like [`copy`], but treats `src` as living at `filter_base` relative to the original filter
/// root. Used when `rlink` delegates an update-only entry to `copy`: `--delete` pruning inside
/// the delegated subtree then matches the include/exclude filter at the entry's true relative
/// path (e.g. `cache/*.log`) instead of relative to the delegated root.
#[instrument(skip(prog_track, settings, preserve))]
#[allow(clippy::too_many_arguments)]
pub async fn copy_with_filter_base(
    prog_track: &'static progress::Progress,
    src: &std::path::Path,
    dst: &std::path::Path,
    settings: &Settings,
    preserve: &preserve::Settings,
    is_fresh: bool,
    filter_base: &std::path::Path,
) -> Result<Summary, Error> {
    // check filter for top-level source (files, directories, and symlinks)
    if let Some(ref filter) = settings.filter {
        let src_name = src.file_name().map(std::path::Path::new);
        if let Some(name) = src_name {
            let src_metadata = crate::walk::run_metadata_probed(
                congestion::Side::Source,
                congestion::MetadataOp::Stat,
                tokio::fs::symlink_metadata(src),
            )
            .await
            .with_context(|| format!("failed reading metadata from src: {:?}", &src))
            .map_err(|err| Error::new(err, Default::default()))?;
            let is_dir = src_metadata.is_dir();
            // for a delegated subtree (non-empty filter_base) the source is not the true filter
            // root, so match it at its logical path with nested semantics; for a normal copy use
            // root-item semantics (anchored patterns don't apply to the root itself).
            let result = if filter_base.as_os_str().is_empty() {
                filter.should_include_root_item(name, is_dir)
            } else {
                filter.should_include(filter_base, is_dir)
            };
            match result {
                crate::filter::FilterResult::Included => {}
                result => {
                    let kind = EntryKind::from_metadata(&src_metadata);
                    if let Some(mode) = settings.dry_run {
                        crate::dry_run::report_skip(src, &result, mode, kind.label_long());
                    }
                    kind.inc_skipped(prog_track);
                    return Ok(skipped_summary_for(kind));
                }
            }
        }
    }
    copy_internal(
        prog_track,
        src,
        dst,
        src,
        settings,
        preserve,
        is_fresh,
        None,
        filter_base,
    )
    .await
}

#[instrument(skip(prog_track, settings, preserve, open_file_guard))]
#[async_recursion]
#[allow(clippy::too_many_arguments)]
async fn copy_internal(
    prog_track: &'static progress::Progress,
    src: &std::path::Path,
    dst: &std::path::Path,
    source_root: &std::path::Path,
    settings: &Settings,
    preserve: &preserve::Settings,
    mut is_fresh: bool,
    open_file_guard: Option<throttle::OpenFileGuard>,
    filter_base: &std::path::Path,
) -> Result<Summary, Error> {
    let _ops_guard = prog_track.ops.guard();
    tracing::debug!("reading source metadata");
    let src_metadata = crate::walk::run_metadata_probed(
        congestion::Side::Source,
        congestion::MetadataOp::Stat,
        tokio::fs::symlink_metadata(src),
    )
    .await
    .with_context(|| format!("failed reading metadata from src: {:?}", &src))
    .map_err(|err| Error::new(err, Default::default()))?;
    if settings.dereference && src_metadata.is_symlink() {
        debug_assert!(
            open_file_guard.is_none(),
            "open file guard should not be pre-acquired for symlinks"
        );
        let link = crate::walk::run_metadata_probed(
            congestion::Side::Source,
            congestion::MetadataOp::Stat,
            tokio::fs::canonicalize(&src),
        )
        .await
        .with_context(|| format!("failed reading src symlink {:?}", &src))
        .map_err(|err| Error::new(err, Default::default()))?;
        return copy(prog_track, &link, dst, settings, preserve, is_fresh).await;
    }
    if src_metadata.is_file() {
        // acquire permit if not pre-acquired by the caller
        let _guard = match open_file_guard {
            Some(g) => g,
            None => throttle::open_file_permit().await,
        };
        return copy_file(
            prog_track,
            src,
            dst,
            &src_metadata,
            settings,
            preserve,
            is_fresh,
        )
        .await;
    }
    debug_assert!(
        open_file_guard.is_none(),
        "open file guard should not be pre-acquired for directories or symlinks"
    );
    if src_metadata.is_symlink() {
        // check --ignore-existing before dry-run so dry-run output reflects actual behavior.
        // use symlink_metadata to detect dangling symlinks too (Path::exists follows symlinks)
        if !is_fresh
            && settings.ignore_existing
            && crate::walk::run_metadata_probed(
                congestion::Side::Destination,
                congestion::MetadataOp::Stat,
                tokio::fs::symlink_metadata(dst),
            )
            .await
            .is_ok()
        {
            if let Some(mode) = settings.dry_run {
                match mode {
                    DryRunMode::Brief => {}
                    DryRunMode::All => println!("skip symlink {:?}", dst),
                    DryRunMode::Explain => println!("skip symlink {:?} (destination exists)", dst),
                }
            }
            tracing::debug!("destination exists, skipping symlink (--ignore-existing)");
            prog_track.symlinks_unchanged.inc();
            return Ok(Summary {
                symlinks_unchanged: 1,
                ..Default::default()
            });
        }
        // handle dry-run mode for symlinks
        if settings.dry_run.is_some() {
            crate::dry_run::report_action("copy", src, Some(dst), "symlink");
            return Ok(Summary {
                symlinks_created: 1,
                ..Default::default()
            });
        }
        let mut rm_summary = RmSummary::default();
        let link = crate::walk::run_metadata_probed(
            congestion::Side::Source,
            congestion::MetadataOp::ReadLink,
            tokio::fs::read_link(src),
        )
        .await
        .with_context(|| format!("failed reading symlink {:?}", &src))
        .map_err(|err| Error::new(err, Default::default()))?;
        // try creating a symlink, if dst path exists and overwrite is set - remove and try again
        if let Err(error) = crate::walk::run_metadata_probed(
            congestion::Side::Destination,
            congestion::MetadataOp::Symlink,
            tokio::fs::symlink(&link, dst),
        )
        .await
        {
            if settings.ignore_existing && error.kind() == std::io::ErrorKind::AlreadyExists {
                tracing::debug!("destination exists, skipping symlink (--ignore-existing)");
                prog_track.symlinks_unchanged.inc();
                return Ok(Summary {
                    symlinks_unchanged: 1,
                    ..Default::default()
                });
            }
            if settings.overwrite && error.kind() == std::io::ErrorKind::AlreadyExists {
                let dst_metadata = crate::walk::run_metadata_probed(
                    congestion::Side::Destination,
                    congestion::MetadataOp::Stat,
                    tokio::fs::symlink_metadata(dst),
                )
                .await
                .with_context(|| format!("failed reading metadata from dst: {:?}", &dst))
                .map_err(|err| Error::new(err, Default::default()))?;
                if is_file_type_same(&src_metadata, &dst_metadata) {
                    let dst_link = crate::walk::run_metadata_probed(
                        congestion::Side::Destination,
                        congestion::MetadataOp::ReadLink,
                        tokio::fs::read_link(dst),
                    )
                    .await
                    .with_context(|| format!("failed reading dst symlink: {:?}", &dst))
                    .map_err(|err| Error::new(err, Default::default()))?;
                    if link == dst_link {
                        tracing::debug!(
                            "'dst' is a symlink and points to the same location as 'src'"
                        );
                        if preserve.symlink.any() {
                            // do we need to update the metadata for this symlink?
                            let dst_metadata = crate::walk::run_metadata_probed(
                                congestion::Side::Destination,
                                congestion::MetadataOp::Stat,
                                tokio::fs::symlink_metadata(dst),
                            )
                            .await
                            .with_context(|| {
                                format!("failed reading metadata from dst: {:?}", &dst)
                            })
                            .map_err(|err| Error::new(err, Default::default()))?;
                            if !filecmp::metadata_equal(
                                &settings.overwrite_compare,
                                &src_metadata,
                                &dst_metadata,
                            ) {
                                tracing::debug!("'dst' metadata is different, updating");
                                preserve::set_symlink_metadata(preserve, &src_metadata, dst)
                                    .await
                                    .map_err(|err| Error::new(err, Default::default()))?;
                                prog_track.symlinks_removed.inc();
                                prog_track.symlinks_created.inc();
                                return Ok(Summary {
                                    rm_summary: RmSummary {
                                        symlinks_removed: 1,
                                        ..Default::default()
                                    },
                                    symlinks_created: 1,
                                    ..Default::default()
                                });
                            }
                        }
                        tracing::debug!("symlink already exists, skipping");
                        prog_track.symlinks_unchanged.inc();
                        return Ok(Summary {
                            symlinks_unchanged: 1,
                            ..Default::default()
                        });
                    }
                    tracing::debug!("'dst' is a symlink but points to a different path, updating");
                } else {
                    tracing::info!("'dst' is not a symlink, updating");
                }
                rm_summary = rm::rm(
                    prog_track,
                    dst,
                    &RmSettings {
                        fail_early: settings.fail_early,
                        filter: None,
                        dry_run: None,
                        time_filter: None,
                    },
                )
                .await
                .map_err(|err| {
                    let rm_summary = err.summary;
                    let copy_summary = Summary {
                        rm_summary,
                        ..Default::default()
                    };
                    Error::new(err.source, copy_summary)
                })?;
                crate::walk::run_metadata_probed(
                    congestion::Side::Destination,
                    congestion::MetadataOp::Symlink,
                    tokio::fs::symlink(&link, dst),
                )
                .await
                .with_context(|| format!("failed creating symlink {:?}", &dst))
                .map_err(|err| {
                    let copy_summary = Summary {
                        rm_summary,
                        ..Default::default()
                    };
                    Error::new(err, copy_summary)
                })?;
            } else {
                return Err(Error::new(
                    anyhow!("failed creating symlink {:?}", &dst),
                    Default::default(),
                ));
            }
        }
        preserve::set_symlink_metadata(preserve, &src_metadata, dst)
            .await
            .map_err(|err| {
                let copy_summary = Summary {
                    rm_summary,
                    ..Default::default()
                };
                Error::new(err, copy_summary)
            })?;
        prog_track.symlinks_created.inc();
        return Ok(Summary {
            rm_summary,
            symlinks_created: 1,
            ..Default::default()
        });
    }
    if !src_metadata.is_dir() {
        if settings.skip_specials {
            tracing::debug!(
                "skipping special file {:?} (type: {:?})",
                src,
                src_metadata.file_type()
            );
            if let Some(mode) = settings.dry_run {
                match mode {
                    DryRunMode::Brief => {}
                    DryRunMode::All => println!("skip special {:?}", src),
                    DryRunMode::Explain => {
                        println!(
                            "skip special {:?} (unsupported file type: {:?})",
                            src,
                            src_metadata.file_type()
                        );
                    }
                }
            }
            prog_track.specials_skipped.inc();
            return Ok(Summary {
                specials_skipped: 1,
                ..Default::default()
            });
        }
        return Err(Error::new(
            anyhow!(
                "copy: {:?} -> {:?} failed, unsupported src file type: {:?}",
                src,
                dst,
                src_metadata.file_type()
            ),
            Default::default(),
        ));
    }
    // handle dry-run mode for directories at the top level
    if settings.dry_run.is_some() {
        if settings.ignore_existing
            && !is_fresh
            && crate::walk::run_metadata_probed(
                congestion::Side::Destination,
                congestion::MetadataOp::Stat,
                tokio::fs::symlink_metadata(dst),
            )
            .await
            .is_ok()
            && !dst.is_dir()
        {
            // destination is not a directory - would skip entire subtree
            if let Some(mode) = settings.dry_run {
                match mode {
                    DryRunMode::Brief => {}
                    DryRunMode::All => println!("skip dir {:?}", dst),
                    DryRunMode::Explain => {
                        println!("skip dir {:?} (destination exists, not a directory)", dst);
                    }
                }
            }
            return Ok(Summary {
                directories_unchanged: 1,
                ..Default::default()
            });
        }
        crate::dry_run::report_action("copy", src, Some(dst), "dir");
        // still need to recurse to show contents
    }
    tracing::debug!("process contents of 'src' directory");
    let mut entries = tokio::fs::read_dir(src)
        .await
        .with_context(|| format!("cannot open directory {src:?} for reading"))
        .map_err(|err| Error::new(err, Default::default()))?;
    // in dry-run mode, skip directory creation but still traverse contents
    let mut copy_summary = if settings.dry_run.is_some() {
        Summary {
            directories_created: 1, // report as would be created
            ..Default::default()
        }
    } else if let Err(error) = crate::walk::run_metadata_probed(
        congestion::Side::Destination,
        congestion::MetadataOp::MkDir,
        tokio::fs::create_dir(dst),
    )
    .await
    {
        assert!(
            !is_fresh,
            "unexpected error creating directory: {dst:?}: {error}"
        );
        if (settings.overwrite || settings.ignore_existing)
            && error.kind() == std::io::ErrorKind::AlreadyExists
        {
            // check if the destination is a directory - if so, leave it
            //
            // N.B. the permissions may prevent us from writing to it but the alternative is to open up the directory
            // while we're writing to it which isn't safe
            let dst_metadata = crate::walk::run_metadata_probed(
                congestion::Side::Destination,
                congestion::MetadataOp::Stat,
                tokio::fs::symlink_metadata(dst),
            )
            .await
            .with_context(|| format!("failed reading metadata from dst: {:?}", &dst))
            .map_err(|err| Error::new(err, Default::default()))?;
            if dst_metadata.is_dir() {
                tracing::debug!("'dst' is a directory, leaving it as is");
                prog_track.directories_unchanged.inc();
                Summary {
                    directories_unchanged: 1,
                    ..Default::default()
                }
            } else if settings.ignore_existing {
                // destination is not a directory but something exists at this path;
                // with --ignore-existing we skip the entire subtree
                tracing::debug!(
                    "destination exists but is not a directory, skipping subtree (--ignore-existing)"
                );
                prog_track.directories_unchanged.inc();
                return Ok(Summary {
                    directories_unchanged: 1,
                    ..Default::default()
                });
            } else {
                tracing::info!("'dst' is not a directory, removing and creating a new one");
                let rm_summary = rm::rm(
                    prog_track,
                    dst,
                    &RmSettings {
                        fail_early: settings.fail_early,
                        filter: None,
                        dry_run: None,
                        time_filter: None,
                    },
                )
                .await
                .map_err(|err| {
                    let rm_summary = err.summary;
                    let copy_summary = Summary {
                        rm_summary,
                        ..Default::default()
                    };
                    Error::new(err.source, copy_summary)
                })?;
                crate::walk::run_metadata_probed(
                    congestion::Side::Destination,
                    congestion::MetadataOp::MkDir,
                    tokio::fs::create_dir(dst),
                )
                .await
                .with_context(|| format!("cannot create directory {dst:?}"))
                .map_err(|err| {
                    let copy_summary = Summary {
                        rm_summary,
                        ..Default::default()
                    };
                    Error::new(err, copy_summary)
                })?;
                // anything copied into dst may assume they don't need to check for conflicts
                is_fresh = true;
                prog_track.directories_created.inc();
                Summary {
                    rm_summary,
                    directories_created: 1,
                    ..Default::default()
                }
            }
        } else {
            let error = Err::<(), std::io::Error>(error)
                .with_context(|| format!("cannot create directory {:?}", dst))
                .unwrap_err();
            tracing::error!("{:#}", &error);
            return Err(Error::new(error, Default::default()));
        }
    } else {
        // new directory created, anything copied into dst may assume they don't need to check for conflicts
        is_fresh = true;
        prog_track.directories_created.inc();
        Summary {
            directories_created: 1,
            ..Default::default()
        }
    };
    // track whether we created this directory (vs it already existing)
    // this is used later to decide if we should clean up an empty directory
    let we_created_this_dir = copy_summary.directories_created == 1;
    let mut join_set = tokio::task::JoinSet::new();
    // names of source entries that pass the filter; used by --delete to decide
    // which destination entries are extraneous. only populated when --delete is
    // active; an empty HashSet has zero heap cost until the first insert, so the
    // default path pays only a stack word.
    let mut keep_set: std::collections::HashSet<std::ffi::OsString> =
        std::collections::HashSet::new();
    let errors = crate::error_collector::ErrorCollector::default();
    loop {
        let Some((entry, entry_file_type)) =
            crate::walk::next_entry_probed(&mut entries, congestion::Side::Source, || {
                format!("failed traversing src directory {:?}", &src)
            })
            .await
            .map_err(|err| Error::new(err, copy_summary))?
        else {
            break;
        };
        let entry_path = entry.path();
        let entry_name = entry_path.file_name().unwrap();
        let dst_path = dst.join(entry_name);
        let entry_kind = EntryKind::from_file_type(entry_file_type.as_ref());
        let entry_is_dir = entry_kind == EntryKind::Dir;
        // compute the path relative to the original filter root: `filter_base` is empty for a
        // normal copy and non-empty for an rlink-delegated update subtree, so copy-side filtering
        // matches the same logical path that delete pruning uses (e.g. `cache/keep.txt`).
        let relative_path = filter_base.join(walk::relative_to_root(&entry_path, source_root));
        // apply filter if configured
        if let Some(skip_result) =
            walk::should_skip_entry(&settings.filter, &relative_path, entry_is_dir)
        {
            if let Some(mode) = settings.dry_run {
                crate::dry_run::report_skip(&entry_path, &skip_result, mode, entry_kind.label());
            }
            tracing::debug!("skipping {:?} due to filter", &entry_path);
            copy_summary = copy_summary + skipped_summary_for(entry_kind);
            entry_kind.inc_skipped(prog_track);
            continue;
        }
        // record this name as authoritative for --delete: it has a source counterpart, so its
        // destination counterpart must not be pruned — even when --skip-specials skips copying
        // it (computed before the skip-specials check below for exactly that reason).
        if settings.delete.is_some() {
            keep_set.insert(entry_name.to_owned());
        }
        // skip special files (sockets, FIFOs, devices) when --skip-specials is set
        if settings.skip_specials && entry_kind == EntryKind::Special {
            tracing::debug!("skipping special file {:?}", &entry_path);
            if let Some(mode) = settings.dry_run {
                match mode {
                    DryRunMode::Brief => {}
                    DryRunMode::All => println!("skip special {:?}", &entry_path),
                    DryRunMode::Explain => {
                        println!(
                            "skip special {:?} (unsupported file type: {:?})",
                            &entry_path,
                            entry_file_type.unwrap()
                        );
                    }
                }
            }
            copy_summary.specials_skipped += 1;
            prog_track.specials_skipped.inc();
            continue;
        }
        // for regular files (not dirs, not symlinks), acquire the open file permit before
        // spawning the task. this provides backpressure so we don't create unbounded tasks.
        // it's safe because files are leaf nodes that never recurse, so no deadlock is possible.
        // we don't acquire for symlinks because with --dereference they could resolve to
        // directories, which would risk deadlock. when file_type() fails we also skip
        // pre-acquisition since we can't be sure it's a file — copy_internal will acquire
        // the permit if needed after reading the actual metadata.
        let entry_is_regular_file = entry_file_type.as_ref().is_some_and(|ft| ft.is_file());
        let open_file_guard = if entry_is_regular_file {
            Some(throttle::open_file_permit().await)
        } else {
            None
        };
        // spawn recursive call - dry-run reporting is handled by copy_internal
        // (copy_file, symlink handling, and directory handling all have their own dry-run reporting)
        let settings = settings.clone();
        let preserve = *preserve;
        let source_root = source_root.to_owned();
        let filter_base = filter_base.to_owned();
        let do_copy = || async move {
            copy_internal(
                prog_track,
                &entry_path,
                &dst_path,
                &source_root,
                &settings,
                &preserve,
                is_fresh,
                open_file_guard,
                &filter_base,
            )
            .await
        };
        join_set.spawn(do_copy());
    }
    // unfortunately ReadDir is opening file-descriptors and there's not a good way to limit this,
    // one thing we CAN do however is to drop it as soon as we're done with it
    drop(entries);
    while let Some(res) = join_set.join_next().await {
        match res {
            Ok(result) => match result {
                Ok(summary) => copy_summary = copy_summary + summary,
                Err(error) => {
                    tracing::error!("copy: {:?} -> {:?} failed with: {:#}", src, dst, &error);
                    copy_summary = copy_summary + error.summary;
                    if settings.fail_early {
                        return Err(Error::new(error.source, copy_summary));
                    }
                    errors.push(error.source);
                }
            },
            Err(error) => {
                if settings.fail_early {
                    return Err(Error::new(error.into(), copy_summary));
                }
                errors.push(error.into());
            }
        }
    }
    // rsync-style --delete: remove destination entries with no source counterpart.
    // runs only when --delete was requested; reaching here means the full set of
    // source children for this directory is known (all spawned tasks have joined).
    if let Some(delete_settings) = &settings.delete {
        if errors.has_errors() {
            // rsync-style safety: skip pruning when this subtree's copy reported errors —
            // deleting based on a run that did not fully succeed could remove data
            // unexpectedly. (rsync likewise skips --delete on I/O errors.)
            tracing::warn!(
                "skipping --delete pruning of {:?} because the copy reported errors",
                dst
            );
        } else {
            let relative_dir = filter_base.join(walk::relative_to_root(src, source_root));
            match crate::delete::prune_extraneous(
                prog_track,
                dst,
                &relative_dir,
                &keep_set,
                settings.filter.as_ref(),
                delete_settings,
                settings.fail_early,
                settings.dry_run,
            )
            .await
            {
                Ok(rm_summary) => copy_summary.rm_summary = copy_summary.rm_summary + rm_summary,
                Err(err) => {
                    copy_summary.rm_summary = copy_summary.rm_summary + err.summary;
                    if settings.fail_early {
                        return Err(Error::new(err.source, copy_summary));
                    }
                    errors.push(err.source);
                }
            }
        }
    }
    // when filtering is active and we created this directory, check if anything was actually
    // copied into it. if nothing was copied, we may need to clean up the empty directory.
    let this_dir_count = usize::from(we_created_this_dir);
    let child_dirs_created = copy_summary
        .directories_created
        .saturating_sub(this_dir_count);
    let anything_copied = copy_summary.files_copied > 0
        || copy_summary.symlinks_created > 0
        || child_dirs_created > 0;
    let relative_path = filter_base.join(walk::relative_to_root(src, source_root));
    let is_root = src == source_root;
    match check_empty_dir_cleanup(
        settings.filter.as_ref(),
        we_created_this_dir,
        anything_copied,
        &relative_path,
        is_root,
        settings.dry_run.is_some(),
    ) {
        EmptyDirAction::Keep => { /* proceed with metadata application */ }
        EmptyDirAction::DryRunSkip => {
            tracing::debug!(
                "dry-run: directory {:?} would not be created (nothing to copy inside)",
                &dst
            );
            copy_summary.directories_created = 0;
            return Ok(copy_summary);
        }
        EmptyDirAction::Remove => {
            tracing::debug!(
                "directory {:?} has nothing to copy inside, removing empty directory",
                &dst
            );
            match crate::walk::run_metadata_probed(
                congestion::Side::Destination,
                congestion::MetadataOp::RmDir,
                tokio::fs::remove_dir(dst),
            )
            .await
            {
                Ok(()) => {
                    copy_summary.directories_created = 0;
                    return Ok(copy_summary);
                }
                Err(err) => {
                    // removal failed (not empty, permission error, etc.) — keep directory
                    tracing::debug!(
                        "failed to remove empty directory {:?}: {:#}, keeping",
                        &dst,
                        &err
                    );
                    // fall through to apply metadata
                }
            }
        }
    }
    // apply directory metadata regardless of whether all children copied successfully.
    // the directory itself was created earlier in this function (we would have returned
    // early if create_dir failed), so we should preserve the source metadata.
    // skip metadata setting in dry-run mode since directory wasn't actually created
    tracing::debug!("set 'dst' directory metadata");
    let metadata_result = if settings.dry_run.is_some() {
        Ok(()) // skip metadata setting in dry-run mode
    } else {
        preserve::set_dir_metadata(preserve, &src_metadata, dst).await
    };
    if errors.has_errors() {
        // child failures take precedence - log metadata error if it also failed
        if let Err(metadata_err) = metadata_result {
            tracing::error!(
                "copy: {:?} -> {:?} failed to set directory metadata: {:#}",
                src,
                dst,
                &metadata_err
            );
        }
        // unwrap is safe: has_errors() guarantees into_error() returns Some
        return Err(Error::new(errors.into_error().unwrap(), copy_summary));
    }
    // no child failures, so metadata error is the primary error
    metadata_result.map_err(|err| Error::new(err, copy_summary))?;
    Ok(copy_summary)
}

#[cfg(test)]
mod copy_tests {
    use crate::testutils;
    use anyhow::Context;
    use std::os::unix::fs::MetadataExt;
    use std::os::unix::fs::PermissionsExt;
    use tracing_test::traced_test;

    use super::*;

    static PROGRESS: std::sync::LazyLock<progress::Progress> =
        std::sync::LazyLock::new(progress::Progress::new);
    static NO_PRESERVE_SETTINGS: std::sync::LazyLock<preserve::Settings> =
        std::sync::LazyLock::new(preserve::preserve_none);
    static DO_PRESERVE_SETTINGS: std::sync::LazyLock<preserve::Settings> =
        std::sync::LazyLock::new(preserve::preserve_all);

    fn settings_with_delete(delete: Option<DeleteSettings>) -> Settings {
        Settings {
            dereference: false,
            fail_early: false,
            overwrite: delete.is_some(), // --delete implies --overwrite
            overwrite_compare: filecmp::MetadataCmpSettings {
                size: true,
                mtime: true,
                ..Default::default()
            },
            overwrite_filter: None,
            ignore_existing: false,
            chunk_size: 0,
            skip_specials: false,
            remote_copy_buffer_size: 0,
            filter: None,
            dry_run: None,
            delete,
        }
    }

    fn delete_on() -> Option<DeleteSettings> {
        Some(DeleteSettings {
            delete_excluded: false,
        })
    }

    #[tokio::test]
    #[traced_test]
    async fn delete_protects_skipped_special_name() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        let src = test_path.join("src_dir");
        let dst = test_path.join("dst_dir");
        tokio::fs::create_dir(&src).await?;
        tokio::fs::write(src.join("file.txt"), "hello").await?;
        // a special file in the source that --skip-specials will skip copying
        nix::unistd::mkfifo(
            &src.join("pipe"),
            nix::sys::stat::Mode::S_IRUSR | nix::sys::stat::Mode::S_IWUSR,
        )?;
        // pre-existing destination: a counterpart for the skipped special, plus a genuine extra
        tokio::fs::create_dir(&dst).await?;
        tokio::fs::write(dst.join("pipe"), "old").await?;
        tokio::fs::write(dst.join("stale.txt"), "junk").await?;

        let mut settings = settings_with_delete(delete_on());
        settings.skip_specials = true;
        let summary = copy(
            &PROGRESS,
            &src,
            &dst,
            &settings,
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;

        assert_eq!(summary.specials_skipped, 1);
        assert!(dst.join("file.txt").exists());
        assert!(
            dst.join("pipe").exists(),
            "a destination entry matching a skipped special must not be pruned (it has a source counterpart)"
        );
        assert!(!dst.join("stale.txt").exists()); // genuine extra removed
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn delete_removes_extraneous_destination_entries() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        let src = test_path.join("foo");
        let dst = test_path.join("bar");
        // initial copy (no delete)
        copy(
            &PROGRESS,
            &src,
            &dst,
            &settings_with_delete(None),
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        // introduce extraneous entries at the destination
        tokio::fs::write(dst.join("extraneous.txt"), b"junk").await?;
        tokio::fs::create_dir(dst.join("extra_dir")).await?;
        tokio::fs::write(dst.join("extra_dir").join("nested.txt"), b"junk").await?;
        // re-copy with --delete
        let summary = copy(
            &PROGRESS,
            &src,
            &dst,
            &settings_with_delete(delete_on()),
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.rm_summary.files_removed, 2); // extraneous.txt + extra_dir/nested.txt
        assert_eq!(summary.rm_summary.directories_removed, 1); // extra_dir
        assert!(!dst.join("extraneous.txt").exists());
        assert!(!dst.join("extra_dir").exists());
        testutils::check_dirs_identical(&src, &dst, testutils::FileEqualityCheck::Basic).await?;
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn delete_prunes_extraneous_at_depth() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        let src = test_path.join("foo");
        let dst = test_path.join("bar");
        copy(
            &PROGRESS,
            &src,
            &dst,
            &settings_with_delete(None),
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        // foo/bar is a common subdirectory; place a stale file inside the dst copy of it
        let nested = dst.join("bar");
        assert!(
            nested.is_dir(),
            "expected common subdirectory bar/ to exist at destination"
        );
        tokio::fs::write(nested.join("stale_nested.txt"), b"junk").await?;
        let summary = copy(
            &PROGRESS,
            &src,
            &dst,
            &settings_with_delete(delete_on()),
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert!(
            !nested.join("stale_nested.txt").exists(),
            "stale entry inside a common subdirectory must be pruned"
        );
        assert!(summary.rm_summary.files_removed >= 1);
        testutils::check_dirs_identical(&src, &dst, testutils::FileEqualityCheck::Basic).await?;
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn delete_removes_extraneous_symlink() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        let src = test_path.join("foo");
        let dst = test_path.join("bar");
        copy(
            &PROGRESS,
            &src,
            &dst,
            &settings_with_delete(None),
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        // an extraneous symlink at the destination root (no source counterpart)
        tokio::fs::symlink("/nonexistent/target", dst.join("stale_link")).await?;
        let summary = copy(
            &PROGRESS,
            &src,
            &dst,
            &settings_with_delete(delete_on()),
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert!(
            tokio::fs::symlink_metadata(dst.join("stale_link"))
                .await
                .is_err(),
            "extraneous symlink must be removed"
        );
        assert_eq!(summary.rm_summary.symlinks_removed, 1);
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn delete_skips_pruning_when_copy_has_errors() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        let src = test_path.join("foo");
        let dst = test_path.join("bar");
        // baseline copy establishes the destination
        copy(
            &PROGRESS,
            &src,
            &dst,
            &settings_with_delete(None),
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        // an extraneous file that --delete would normally prune
        tokio::fs::write(dst.join("extraneous.txt"), b"junk").await?;
        // make a source sub-directory unreadable so traversal fails (fail_early is false).
        // a directory (not a file) is used because --overwrite with mtime-equal files skips
        // copying identical files; a directory's read_dir fails unconditionally when mode is 0o000.
        let unreadable = src.join("baz");
        let original = tokio::fs::metadata(&unreadable).await?.permissions();
        tokio::fs::set_permissions(&unreadable, std::fs::Permissions::from_mode(0o000)).await?;

        let result = copy(
            &PROGRESS,
            &src,
            &dst,
            &settings_with_delete(delete_on()),
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await;

        tokio::fs::set_permissions(&unreadable, original).await?;

        assert!(
            result.is_err(),
            "copy of the unreadable directory should fail"
        );
        assert!(
            dst.join("extraneous.txt").exists(),
            "pruning must be skipped when the copy reported errors"
        );
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn check_basic_copy() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        let summary = copy(
            &PROGRESS,
            &test_path.join("foo"),
            &test_path.join("bar"),
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &NO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.files_copied, 5);
        assert_eq!(summary.symlinks_created, 2);
        assert_eq!(summary.directories_created, 3);
        testutils::check_dirs_identical(
            &test_path.join("foo"),
            &test_path.join("bar"),
            testutils::FileEqualityCheck::Basic,
        )
        .await?;
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn no_read_permission() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        let filepaths = vec![
            test_path.join("foo").join("0.txt"),
            test_path.join("foo").join("baz"),
        ];
        for fpath in &filepaths {
            // change file permissions to not readable
            tokio::fs::set_permissions(&fpath, std::fs::Permissions::from_mode(0o000)).await?;
        }
        match copy(
            &PROGRESS,
            &test_path.join("foo"),
            &test_path.join("bar"),
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &NO_PRESERVE_SETTINGS,
            false,
        )
        .await
        {
            Ok(_) => panic!("Expected the copy to error!"),
            Err(error) => {
                tracing::info!("{}", &error);
                // foo
                // |- 0.txt  // <- no read permission
                // |- bar
                //    |- 1.txt
                //    |- 2.txt
                //    |- 3.txt
                // |- baz   // <- no read permission
                //    |- 4.txt
                //    |- 5.txt -> ../bar/2.txt
                //    |- 6.txt -> (absolute path) .../foo/bar/3.txt
                assert_eq!(error.summary.files_copied, 3);
                assert_eq!(error.summary.symlinks_created, 0);
                assert_eq!(error.summary.directories_created, 2);
            }
        }
        // make source directory same as what we expect destination to be
        for fpath in &filepaths {
            tokio::fs::set_permissions(&fpath, std::fs::Permissions::from_mode(0o700)).await?;
            if tokio::fs::symlink_metadata(fpath).await?.is_file() {
                tokio::fs::remove_file(fpath).await?;
            } else {
                tokio::fs::remove_dir_all(fpath).await?;
            }
        }
        testutils::check_dirs_identical(
            &test_path.join("foo"),
            &test_path.join("bar"),
            testutils::FileEqualityCheck::Basic,
        )
        .await?;
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn check_default_mode() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        // set file to executable
        tokio::fs::set_permissions(
            tmp_dir.join("foo").join("0.txt"),
            std::fs::Permissions::from_mode(0o700),
        )
        .await?;
        // set file executable AND also set sticky bit, setuid and setgid
        let exec_sticky_file = tmp_dir.join("foo").join("bar").join("1.txt");
        tokio::fs::set_permissions(&exec_sticky_file, std::fs::Permissions::from_mode(0o3770))
            .await?;
        let test_path = tmp_dir.as_path();
        let summary = copy(
            &PROGRESS,
            &test_path.join("foo"),
            &test_path.join("bar"),
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &NO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.files_copied, 5);
        assert_eq!(summary.symlinks_created, 2);
        assert_eq!(summary.directories_created, 3);
        // clear the setuid, setgid and sticky bit for comparison
        tokio::fs::set_permissions(
            &exec_sticky_file,
            std::fs::Permissions::from_mode(
                std::fs::symlink_metadata(&exec_sticky_file)?
                    .permissions()
                    .mode()
                    & 0o0777,
            ),
        )
        .await?;
        testutils::check_dirs_identical(
            &test_path.join("foo"),
            &test_path.join("bar"),
            testutils::FileEqualityCheck::Basic,
        )
        .await?;
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn no_write_permission() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        // directory - readable and non-executable
        let non_exec_dir = test_path.join("foo").join("bogey");
        tokio::fs::create_dir(&non_exec_dir).await?;
        tokio::fs::set_permissions(&non_exec_dir, std::fs::Permissions::from_mode(0o400)).await?;
        // directory - readable and executable
        tokio::fs::set_permissions(
            &test_path.join("foo").join("baz"),
            std::fs::Permissions::from_mode(0o500),
        )
        .await?;
        // file
        tokio::fs::set_permissions(
            &test_path.join("foo").join("baz").join("4.txt"),
            std::fs::Permissions::from_mode(0o440),
        )
        .await?;
        let summary = copy(
            &PROGRESS,
            &test_path.join("foo"),
            &test_path.join("bar"),
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &NO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.files_copied, 5);
        assert_eq!(summary.symlinks_created, 2);
        assert_eq!(summary.directories_created, 4);
        testutils::check_dirs_identical(
            &test_path.join("foo"),
            &test_path.join("bar"),
            testutils::FileEqualityCheck::Basic,
        )
        .await?;
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn dereference() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        // make files pointed to by symlinks have different permissions than the symlink itself
        let src1 = &test_path.join("foo").join("bar").join("2.txt");
        let src2 = &test_path.join("foo").join("bar").join("3.txt");
        let test_mode = 0o440;
        for f in [src1, src2] {
            tokio::fs::set_permissions(f, std::fs::Permissions::from_mode(test_mode)).await?;
        }
        let summary = copy(
            &PROGRESS,
            &test_path.join("foo"),
            &test_path.join("bar"),
            &Settings {
                dereference: true, // <- important!
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &NO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.files_copied, 7);
        assert_eq!(summary.symlinks_created, 0);
        assert_eq!(summary.directories_created, 3);
        // ...
        // |- baz
        //    |- 4.txt
        //    |- 5.txt -> ../bar/2.txt
        //    |- 6.txt -> (absolute path) .../foo/bar/3.txt
        let dst1 = &test_path.join("bar").join("baz").join("5.txt");
        let dst2 = &test_path.join("bar").join("baz").join("6.txt");
        for f in [dst1, dst2] {
            let metadata = tokio::fs::symlink_metadata(f)
                .await
                .with_context(|| format!("failed reading metadata from {:?}", &f))?;
            assert!(metadata.is_file());
            // check that the permissions are the same as the source file modulo no sticky bit, setuid and setgid
            assert_eq!(metadata.permissions().mode() & 0o777, test_mode);
        }
        Ok(())
    }

    async fn cp_compare(
        cp_args: &[&str],
        rcp_settings: &Settings,
        preserve: bool,
    ) -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        // run a cp command to copy the files
        let cp_output = tokio::process::Command::new("cp")
            .args(cp_args)
            .arg(test_path.join("foo"))
            .arg(test_path.join("bar"))
            .output()
            .await?;
        assert!(cp_output.status.success());
        // now run rcp
        let summary = copy(
            &PROGRESS,
            &test_path.join("foo"),
            &test_path.join("baz"),
            rcp_settings,
            if preserve {
                &DO_PRESERVE_SETTINGS
            } else {
                &NO_PRESERVE_SETTINGS
            },
            false,
        )
        .await?;
        if rcp_settings.dereference {
            assert_eq!(summary.files_copied, 7);
            assert_eq!(summary.symlinks_created, 0);
        } else {
            assert_eq!(summary.files_copied, 5);
            assert_eq!(summary.symlinks_created, 2);
        }
        assert_eq!(summary.directories_created, 3);
        testutils::check_dirs_identical(
            &test_path.join("bar"),
            &test_path.join("baz"),
            if preserve {
                testutils::FileEqualityCheck::Timestamp
            } else {
                testutils::FileEqualityCheck::Basic
            },
        )
        .await?;
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn test_cp_compat() -> Result<(), anyhow::Error> {
        cp_compare(
            &["-r"],
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            false,
        )
        .await?;
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn test_cp_compat_preserve() -> Result<(), anyhow::Error> {
        cp_compare(
            &["-r", "-p"],
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            true,
        )
        .await?;
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn test_cp_compat_dereference() -> Result<(), anyhow::Error> {
        cp_compare(
            &["-r", "-L"],
            &Settings {
                dereference: true,
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            false,
        )
        .await?;
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn test_cp_compat_preserve_and_dereference() -> Result<(), anyhow::Error> {
        cp_compare(
            &["-r", "-p", "-L"],
            &Settings {
                dereference: true,
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            true,
        )
        .await?;
        Ok(())
    }

    async fn setup_test_dir_and_copy() -> Result<std::path::PathBuf, anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        let summary = copy(
            &PROGRESS,
            &test_path.join("foo"),
            &test_path.join("bar"),
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.files_copied, 5);
        assert_eq!(summary.symlinks_created, 2);
        assert_eq!(summary.directories_created, 3);
        Ok(tmp_dir)
    }

    #[tokio::test]
    #[traced_test]
    async fn test_cp_overwrite_basic() -> Result<(), anyhow::Error> {
        let tmp_dir = setup_test_dir_and_copy().await?;
        let output_path = &tmp_dir.join("bar");
        {
            // bar
            // |- 0.txt
            // |- bar  <---------------------------------------- REMOVE
            //    |- 1.txt  <----------------------------------- REMOVE
            //    |- 2.txt  <----------------------------------- REMOVE
            //    |- 3.txt  <----------------------------------- REMOVE
            // |- baz
            //    |- 4.txt
            //    |- 5.txt -> ../bar/2.txt <-------------------- REMOVE
            //    |- 6.txt -> (absolute path) .../foo/bar/3.txt
            let summary = rm::rm(
                &PROGRESS,
                &output_path.join("bar"),
                &RmSettings {
                    fail_early: false,
                    filter: None,
                    dry_run: None,
                    time_filter: None,
                },
            )
            .await?
                + rm::rm(
                    &PROGRESS,
                    &output_path.join("baz").join("5.txt"),
                    &RmSettings {
                        fail_early: false,
                        filter: None,
                        dry_run: None,
                        time_filter: None,
                    },
                )
                .await?;
            assert_eq!(summary.files_removed, 3);
            assert_eq!(summary.symlinks_removed, 1);
            assert_eq!(summary.directories_removed, 1);
        }
        let summary = copy(
            &PROGRESS,
            &tmp_dir.join("foo"),
            output_path,
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: true, // <- important!
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.files_copied, 3);
        assert_eq!(summary.symlinks_created, 1);
        assert_eq!(summary.directories_created, 1);
        testutils::check_dirs_identical(
            &tmp_dir.join("foo"),
            output_path,
            testutils::FileEqualityCheck::Timestamp,
        )
        .await?;
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn test_cp_overwrite_dir_file() -> Result<(), anyhow::Error> {
        let tmp_dir = setup_test_dir_and_copy().await?;
        let output_path = &tmp_dir.join("bar");
        {
            // bar
            // |- 0.txt
            // |- bar
            //    |- 1.txt  <------------------------------------- REMOVE
            //    |- 2.txt
            //    |- 3.txt
            // |- baz  <------------------------------------------ REMOVE
            //    |- 4.txt  <------------------------------------- REMOVE
            //    |- 5.txt -> ../bar/2.txt <---------------------- REMOVE
            //    |- 6.txt -> (absolute path) .../foo/bar/3.txt <- REMOVE
            let summary = rm::rm(
                &PROGRESS,
                &output_path.join("bar").join("1.txt"),
                &RmSettings {
                    fail_early: false,
                    filter: None,
                    dry_run: None,
                    time_filter: None,
                },
            )
            .await?
                + rm::rm(
                    &PROGRESS,
                    &output_path.join("baz"),
                    &RmSettings {
                        fail_early: false,
                        filter: None,
                        dry_run: None,
                        time_filter: None,
                    },
                )
                .await?;
            assert_eq!(summary.files_removed, 2);
            assert_eq!(summary.symlinks_removed, 2);
            assert_eq!(summary.directories_removed, 1);
        }
        {
            // replace bar/1.txt file with a directory
            tokio::fs::create_dir(&output_path.join("bar").join("1.txt")).await?;
            // replace baz directory with a file
            tokio::fs::write(&output_path.join("baz"), "baz").await?;
        }
        let summary = copy(
            &PROGRESS,
            &tmp_dir.join("foo"),
            output_path,
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: true, // <- important!
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.rm_summary.files_removed, 1);
        assert_eq!(summary.rm_summary.symlinks_removed, 0);
        assert_eq!(summary.rm_summary.directories_removed, 1);
        assert_eq!(summary.files_copied, 2);
        assert_eq!(summary.symlinks_created, 2);
        assert_eq!(summary.directories_created, 1);
        testutils::check_dirs_identical(
            &tmp_dir.join("foo"),
            output_path,
            testutils::FileEqualityCheck::Timestamp,
        )
        .await?;
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn test_cp_overwrite_symlink_file() -> Result<(), anyhow::Error> {
        let tmp_dir = setup_test_dir_and_copy().await?;
        let output_path = &tmp_dir.join("bar");
        {
            // bar
            // |- 0.txt
            // |- baz
            //    |- 4.txt  <------------------------------------- REMOVE
            //    |- 5.txt -> ../bar/2.txt <---------------------- REMOVE
            // ...
            let summary = rm::rm(
                &PROGRESS,
                &output_path.join("baz").join("4.txt"),
                &RmSettings {
                    fail_early: false,
                    filter: None,
                    dry_run: None,
                    time_filter: None,
                },
            )
            .await?
                + rm::rm(
                    &PROGRESS,
                    &output_path.join("baz").join("5.txt"),
                    &RmSettings {
                        fail_early: false,
                        filter: None,
                        dry_run: None,
                        time_filter: None,
                    },
                )
                .await?;
            assert_eq!(summary.files_removed, 1);
            assert_eq!(summary.symlinks_removed, 1);
            assert_eq!(summary.directories_removed, 0);
        }
        {
            // replace baz/4.txt file with a symlink
            tokio::fs::symlink("../0.txt", &output_path.join("baz").join("4.txt")).await?;
            // replace baz/5.txt symlink with a file
            tokio::fs::write(&output_path.join("baz").join("5.txt"), "baz").await?;
        }
        let summary = copy(
            &PROGRESS,
            &tmp_dir.join("foo"),
            output_path,
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: true, // <- important!
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.rm_summary.files_removed, 1);
        assert_eq!(summary.rm_summary.symlinks_removed, 1);
        assert_eq!(summary.rm_summary.directories_removed, 0);
        assert_eq!(summary.files_copied, 1);
        assert_eq!(summary.symlinks_created, 1);
        assert_eq!(summary.directories_created, 0);
        testutils::check_dirs_identical(
            &tmp_dir.join("foo"),
            output_path,
            testutils::FileEqualityCheck::Timestamp,
        )
        .await?;
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn test_cp_overwrite_symlink_dir() -> Result<(), anyhow::Error> {
        let tmp_dir = setup_test_dir_and_copy().await?;
        let output_path = &tmp_dir.join("bar");
        {
            // bar
            // |- 0.txt
            // |- bar  <------------------------------------------ REMOVE
            //    |- 1.txt  <------------------------------------- REMOVE
            //    |- 2.txt  <------------------------------------- REMOVE
            //    |- 3.txt  <------------------------------------- REMOVE
            // |- baz
            //    |- 5.txt -> ../bar/2.txt <---------------------- REMOVE
            // ...
            let summary = rm::rm(
                &PROGRESS,
                &output_path.join("bar"),
                &RmSettings {
                    fail_early: false,
                    filter: None,
                    dry_run: None,
                    time_filter: None,
                },
            )
            .await?
                + rm::rm(
                    &PROGRESS,
                    &output_path.join("baz").join("5.txt"),
                    &RmSettings {
                        fail_early: false,
                        filter: None,
                        dry_run: None,
                        time_filter: None,
                    },
                )
                .await?;
            assert_eq!(summary.files_removed, 3);
            assert_eq!(summary.symlinks_removed, 1);
            assert_eq!(summary.directories_removed, 1);
        }
        {
            // replace bar directory with a symlink
            tokio::fs::symlink("0.txt", &output_path.join("bar")).await?;
            // replace baz/5.txt symlink with a directory
            tokio::fs::create_dir(&output_path.join("baz").join("5.txt")).await?;
        }
        let summary = copy(
            &PROGRESS,
            &tmp_dir.join("foo"),
            output_path,
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: true, // <- important!
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.rm_summary.files_removed, 0);
        assert_eq!(summary.rm_summary.symlinks_removed, 1);
        assert_eq!(summary.rm_summary.directories_removed, 1);
        assert_eq!(summary.files_copied, 3);
        assert_eq!(summary.symlinks_created, 1);
        assert_eq!(summary.directories_created, 1);
        assert_eq!(summary.files_unchanged, 2);
        assert_eq!(summary.symlinks_unchanged, 1);
        assert_eq!(summary.directories_unchanged, 2);
        testutils::check_dirs_identical(
            &tmp_dir.join("foo"),
            output_path,
            testutils::FileEqualityCheck::Timestamp,
        )
        .await?;
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn test_cp_overwrite_error() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        let summary = copy(
            &PROGRESS,
            &test_path.join("foo"),
            &test_path.join("bar"),
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &NO_PRESERVE_SETTINGS, // we want timestamps to differ!
            false,
        )
        .await?;
        assert_eq!(summary.files_copied, 5);
        assert_eq!(summary.symlinks_created, 2);
        assert_eq!(summary.directories_created, 3);
        let source_path = &test_path.join("foo");
        let output_path = &tmp_dir.join("bar");
        // unreadable
        tokio::fs::set_permissions(
            &source_path.join("bar"),
            std::fs::Permissions::from_mode(0o000),
        )
        .await?;
        tokio::fs::set_permissions(
            &source_path.join("baz").join("4.txt"),
            std::fs::Permissions::from_mode(0o000),
        )
        .await?;
        // bar
        // |- 0.txt
        // |- bar  <---------------------------------------- NON READABLE
        // |- baz
        //    |- 4.txt  <----------------------------------- NON READABLE
        //    |- 5.txt -> ../bar/2.txt
        //    |- 6.txt -> (absolute path) .../foo/bar/3.txt
        match copy(
            &PROGRESS,
            &tmp_dir.join("foo"),
            output_path,
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: true, // <- important!
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await
        {
            Ok(_) => panic!("Expected the copy to error!"),
            Err(error) => {
                tracing::info!("{}", &error);
                assert_eq!(error.summary.files_copied, 1);
                assert_eq!(error.summary.symlinks_created, 2);
                assert_eq!(error.summary.directories_created, 0);
                assert_eq!(error.summary.rm_summary.files_removed, 2);
                assert_eq!(error.summary.rm_summary.symlinks_removed, 2);
                assert_eq!(error.summary.rm_summary.directories_removed, 0);
            }
        }
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn overwrite_filter_newer_skips_when_dest_is_newer() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::create_temp_dir().await?;
        let test_path = tmp_dir.as_path();
        let src_file = test_path.join("src.txt");
        let dst_file = test_path.join("dst.txt");
        // create dest first with older content, then source
        tokio::fs::write(&dst_file, "newer content").await?;
        // set dest mtime to the future so it's strictly newer than source
        let future_time = filetime::FileTime::from_unix_time(2_000_000_000, 0);
        filetime::set_file_mtime(&dst_file, future_time)?;
        tokio::fs::write(&src_file, "older content").await?;
        let past_time = filetime::FileTime::from_unix_time(1_000_000_000, 0);
        filetime::set_file_mtime(&src_file, past_time)?;
        let summary = copy_file(
            &PROGRESS,
            &src_file,
            &dst_file,
            &tokio::fs::metadata(&src_file).await?,
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: true,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: Some(OverwriteFilter::Newer),
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &NO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.files_unchanged, 1);
        assert_eq!(summary.files_copied, 0);
        // dest should still have original content
        let content = tokio::fs::read_to_string(&dst_file).await?;
        assert_eq!(content, "newer content");
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn overwrite_filter_newer_copies_when_dest_is_older() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::create_temp_dir().await?;
        let test_path = tmp_dir.as_path();
        let src_file = test_path.join("src.txt");
        let dst_file = test_path.join("dst.txt");
        // create dest with old mtime, source with newer mtime
        tokio::fs::write(&dst_file, "old content").await?;
        let past_time = filetime::FileTime::from_unix_time(1_000_000_000, 0);
        filetime::set_file_mtime(&dst_file, past_time)?;
        tokio::fs::write(&src_file, "new content").await?;
        let future_time = filetime::FileTime::from_unix_time(2_000_000_000, 0);
        filetime::set_file_mtime(&src_file, future_time)?;
        let summary = copy_file(
            &PROGRESS,
            &src_file,
            &dst_file,
            &tokio::fs::metadata(&src_file).await?,
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: true,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: Some(OverwriteFilter::Newer),
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &NO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.files_copied, 1);
        assert_eq!(summary.files_unchanged, 0);
        // dest should now have source content
        let content = tokio::fs::read_to_string(&dst_file).await?;
        assert_eq!(content, "new content");
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn overwrite_filter_newer_copies_when_same_mtime() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::create_temp_dir().await?;
        let test_path = tmp_dir.as_path();
        let src_file = test_path.join("src.txt");
        let dst_file = test_path.join("dst.txt");
        // create both files with the same mtime but different size
        tokio::fs::write(&dst_file, "old").await?;
        tokio::fs::write(&src_file, "new content").await?;
        let same_time = filetime::FileTime::from_unix_time(1_500_000_000, 0);
        filetime::set_file_mtime(&dst_file, same_time)?;
        filetime::set_file_mtime(&src_file, same_time)?;
        let summary = copy_file(
            &PROGRESS,
            &src_file,
            &dst_file,
            &tokio::fs::metadata(&src_file).await?,
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: true,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: Some(OverwriteFilter::Newer),
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &NO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        // same mtime means NOT newer, so the file should be overwritten
        assert_eq!(summary.files_copied, 1);
        assert_eq!(summary.files_unchanged, 0);
        let content = tokio::fs::read_to_string(&dst_file).await?;
        assert_eq!(content, "new content");
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn overwrite_without_filter_copies_when_dest_is_newer() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::create_temp_dir().await?;
        let test_path = tmp_dir.as_path();
        let src_file = test_path.join("src.txt");
        let dst_file = test_path.join("dst.txt");
        // dest is newer, but no filter set so it should still overwrite
        tokio::fs::write(&dst_file, "newer content").await?;
        let future_time = filetime::FileTime::from_unix_time(2_000_000_000, 0);
        filetime::set_file_mtime(&dst_file, future_time)?;
        tokio::fs::write(&src_file, "older content").await?;
        let past_time = filetime::FileTime::from_unix_time(1_000_000_000, 0);
        filetime::set_file_mtime(&src_file, past_time)?;
        let summary = copy_file(
            &PROGRESS,
            &src_file,
            &dst_file,
            &tokio::fs::metadata(&src_file).await?,
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: true,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &NO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        // without filter, file should be overwritten regardless
        assert_eq!(summary.files_copied, 1);
        let content = tokio::fs::read_to_string(&dst_file).await?;
        assert_eq!(content, "older content");
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn ignore_existing_skips_when_dest_exists() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::create_temp_dir().await?;
        let test_path = tmp_dir.as_path();
        let src_file = test_path.join("src.txt");
        let dst_file = test_path.join("dst.txt");
        tokio::fs::write(&src_file, "source content").await?;
        tokio::fs::write(&dst_file, "dest content").await?;
        let summary = copy_file(
            &PROGRESS,
            &src_file,
            &dst_file,
            &tokio::fs::metadata(&src_file).await?,
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: false,
                overwrite_compare: Default::default(),
                overwrite_filter: None,
                ignore_existing: true,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &NO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.files_unchanged, 1);
        assert_eq!(summary.files_copied, 0);
        // dest should still have original content
        let content = tokio::fs::read_to_string(&dst_file).await?;
        assert_eq!(content, "dest content");
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn ignore_existing_skips_when_dest_is_different_type() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::create_temp_dir().await?;
        let test_path = tmp_dir.as_path();
        let src_file = test_path.join("src.txt");
        let dst_dir = test_path.join("dst.txt");
        tokio::fs::write(&src_file, "source content").await?;
        // destination is a directory, not a file
        tokio::fs::create_dir(&dst_dir).await?;
        let summary = copy_file(
            &PROGRESS,
            &src_file,
            &dst_dir,
            &tokio::fs::metadata(&src_file).await?,
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: false,
                overwrite_compare: Default::default(),
                overwrite_filter: None,
                ignore_existing: true,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &NO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.files_unchanged, 1);
        assert_eq!(summary.files_copied, 0);
        // dest directory should still exist
        assert!(dst_dir.is_dir());
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn ignore_existing_copies_when_dest_missing() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::create_temp_dir().await?;
        let test_path = tmp_dir.as_path();
        let src_file = test_path.join("src.txt");
        let dst_file = test_path.join("dst.txt");
        tokio::fs::write(&src_file, "source content").await?;
        let summary = copy_file(
            &PROGRESS,
            &src_file,
            &dst_file,
            &tokio::fs::metadata(&src_file).await?,
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: false,
                overwrite_compare: Default::default(),
                overwrite_filter: None,
                ignore_existing: true,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &NO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.files_copied, 1);
        let content = tokio::fs::read_to_string(&dst_file).await?;
        assert_eq!(content, "source content");
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn test_cp_dereference_symlink_chain() -> Result<(), anyhow::Error> {
        // Create a fresh temporary directory to avoid conflicts
        let tmp_dir = testutils::create_temp_dir().await?;
        let test_path = tmp_dir.as_path();
        // Create a chain of symlinks: foo -> bar -> baz (actual file)
        let baz_file = test_path.join("baz_file.txt");
        tokio::fs::write(&baz_file, "final content").await?;
        let bar_link = test_path.join("bar_link");
        let foo_link = test_path.join("foo_link");
        // Create chain: foo_link -> bar_link -> baz_file.txt
        tokio::fs::symlink(&baz_file, &bar_link).await?;
        tokio::fs::symlink(&bar_link, &foo_link).await?;
        // Create source directory with the symlink chain
        let src_dir = test_path.join("src_chain");
        tokio::fs::create_dir(&src_dir).await?;
        // Copy the chain into the source directory
        tokio::fs::symlink("../foo_link", &src_dir.join("foo")).await?;
        tokio::fs::symlink("../bar_link", &src_dir.join("bar")).await?;
        tokio::fs::symlink("../baz_file.txt", &src_dir.join("baz")).await?;
        // Test with dereference - should copy 3 files with same content
        let summary = copy(
            &PROGRESS,
            &src_dir,
            &test_path.join("dst_with_deref"),
            &Settings {
                dereference: true, // <- important!
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &NO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.files_copied, 3); // foo, bar, baz all copied as files
        assert_eq!(summary.symlinks_created, 0); // dereference is set
        assert_eq!(summary.directories_created, 1);
        let dst_dir = test_path.join("dst_with_deref");
        // Verify all three are now regular files with the same content
        let foo_content = tokio::fs::read_to_string(dst_dir.join("foo")).await?;
        let bar_content = tokio::fs::read_to_string(dst_dir.join("bar")).await?;
        let baz_content = tokio::fs::read_to_string(dst_dir.join("baz")).await?;
        assert_eq!(foo_content, "final content");
        assert_eq!(bar_content, "final content");
        assert_eq!(baz_content, "final content");
        // Verify they are all regular files, not symlinks
        assert!(dst_dir.join("foo").is_file());
        assert!(dst_dir.join("bar").is_file());
        assert!(dst_dir.join("baz").is_file());
        assert!(!dst_dir.join("foo").is_symlink());
        assert!(!dst_dir.join("bar").is_symlink());
        assert!(!dst_dir.join("baz").is_symlink());
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn test_cp_dereference_symlink_to_directory() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::create_temp_dir().await?;
        let test_path = tmp_dir.as_path();
        // Create a directory with specific permissions and content
        let target_dir = test_path.join("target_dir");
        tokio::fs::create_dir(&target_dir).await?;
        tokio::fs::set_permissions(&target_dir, std::fs::Permissions::from_mode(0o755)).await?;
        // Add some files to the directory
        tokio::fs::write(target_dir.join("file1.txt"), "content1").await?;
        tokio::fs::write(target_dir.join("file2.txt"), "content2").await?;
        tokio::fs::set_permissions(
            &target_dir.join("file1.txt"),
            std::fs::Permissions::from_mode(0o644),
        )
        .await?;
        tokio::fs::set_permissions(
            &target_dir.join("file2.txt"),
            std::fs::Permissions::from_mode(0o600),
        )
        .await?;
        // Create a symlink pointing to the directory
        let dir_symlink = test_path.join("dir_symlink");
        tokio::fs::symlink(&target_dir, &dir_symlink).await?;
        // Test copying the symlink with dereference - should copy as a directory
        let summary = copy(
            &PROGRESS,
            &dir_symlink,
            &test_path.join("copied_dir"),
            &Settings {
                dereference: true, // <- important!
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.files_copied, 2); // file1.txt, file2.txt
        assert_eq!(summary.symlinks_created, 0); // dereference is set
        assert_eq!(summary.directories_created, 1); // copied_dir
        let copied_dir = test_path.join("copied_dir");
        // Verify the directory and its contents were copied
        assert!(copied_dir.is_dir());
        assert!(!copied_dir.is_symlink()); // Should be a real directory, not a symlink
        // Verify files were copied with correct content
        let file1_content = tokio::fs::read_to_string(copied_dir.join("file1.txt")).await?;
        let file2_content = tokio::fs::read_to_string(copied_dir.join("file2.txt")).await?;
        assert_eq!(file1_content, "content1");
        assert_eq!(file2_content, "content2");
        // Verify permissions were preserved
        let copied_dir_metadata = tokio::fs::metadata(&copied_dir).await?;
        let file1_metadata = tokio::fs::metadata(copied_dir.join("file1.txt")).await?;
        let file2_metadata = tokio::fs::metadata(copied_dir.join("file2.txt")).await?;
        assert_eq!(copied_dir_metadata.permissions().mode() & 0o777, 0o755);
        assert_eq!(file1_metadata.permissions().mode() & 0o777, 0o644);
        assert_eq!(file2_metadata.permissions().mode() & 0o777, 0o600);
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn test_cp_dereference_permissions_preserved() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::create_temp_dir().await?;
        let test_path = tmp_dir.as_path();
        // Create files with specific permissions
        let file1 = test_path.join("file1.txt");
        let file2 = test_path.join("file2.txt");
        tokio::fs::write(&file1, "content1").await?;
        tokio::fs::write(&file2, "content2").await?;
        tokio::fs::set_permissions(&file1, std::fs::Permissions::from_mode(0o755)).await?;
        tokio::fs::set_permissions(&file2, std::fs::Permissions::from_mode(0o640)).await?;
        // Create symlinks pointing to these files
        let symlink1 = test_path.join("symlink1");
        let symlink2 = test_path.join("symlink2");
        tokio::fs::symlink(&file1, &symlink1).await?;
        tokio::fs::symlink(&file2, &symlink2).await?;
        // Test copying symlinks with dereference and preserve
        let summary1 = copy(
            &PROGRESS,
            &symlink1,
            &test_path.join("copied_file1.txt"),
            &Settings {
                dereference: true, // <- important!
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings::default(),
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &DO_PRESERVE_SETTINGS, // <- important!
            false,
        )
        .await?;
        let summary2 = copy(
            &PROGRESS,
            &symlink2,
            &test_path.join("copied_file2.txt"),
            &Settings {
                dereference: true,
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings::default(),
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary1.files_copied, 1);
        assert_eq!(summary1.symlinks_created, 0);
        assert_eq!(summary2.files_copied, 1);
        assert_eq!(summary2.symlinks_created, 0);
        let copied1 = test_path.join("copied_file1.txt");
        let copied2 = test_path.join("copied_file2.txt");
        // Verify files are regular files, not symlinks
        assert!(copied1.is_file());
        assert!(!copied1.is_symlink());
        assert!(copied2.is_file());
        assert!(!copied2.is_symlink());
        // Verify content was copied correctly
        let content1 = tokio::fs::read_to_string(&copied1).await?;
        let content2 = tokio::fs::read_to_string(&copied2).await?;
        assert_eq!(content1, "content1");
        assert_eq!(content2, "content2");
        // Verify permissions from the target files were preserved (not symlink permissions)
        let copied1_metadata = tokio::fs::metadata(&copied1).await?;
        let copied2_metadata = tokio::fs::metadata(&copied2).await?;
        assert_eq!(copied1_metadata.permissions().mode() & 0o777, 0o755);
        assert_eq!(copied2_metadata.permissions().mode() & 0o777, 0o640);
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn test_cp_dereference_dir() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        // symlink bar to bar-link
        tokio::fs::symlink("bar", &tmp_dir.join("foo").join("bar-link")).await?;
        // symlink bar-link to bar-link-link
        tokio::fs::symlink("bar-link", &tmp_dir.join("foo").join("bar-link-link")).await?;
        let summary = copy(
            &PROGRESS,
            &tmp_dir.join("foo"),
            &tmp_dir.join("bar"),
            &Settings {
                dereference: true, // <- important!
                fail_early: false,
                overwrite: false,
                overwrite_compare: filecmp::MetadataCmpSettings {
                    size: true,
                    mtime: true,
                    ..Default::default()
                },
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert_eq!(summary.files_copied, 13); // 0.txt, 3x bar/(1.txt, 2.txt, 3.txt), baz/(4.txt, 5.txt, 6.txt)
        assert_eq!(summary.symlinks_created, 0); // dereference is set
        assert_eq!(summary.directories_created, 5);
        // check_dirs_identical doesn't handle dereference so let's do it manually
        tokio::process::Command::new("cp")
            .args(["-r", "-L"])
            .arg(tmp_dir.join("foo"))
            .arg(tmp_dir.join("bar-cp"))
            .output()
            .await?;
        testutils::check_dirs_identical(
            &tmp_dir.join("bar"),
            &tmp_dir.join("bar-cp"),
            testutils::FileEqualityCheck::Basic,
        )
        .await?;
        Ok(())
    }

    /// Tests to verify error messages include root causes for debugging
    mod error_message_tests {
        use super::*;

        /// Helper to extract full error message with chain
        fn get_full_error_message(error: &Error) -> String {
            format!("{:#}", error.source)
        }

        #[tokio::test]
        #[traced_test]
        async fn test_permission_error_includes_root_cause() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::create_temp_dir().await?;
            let unreadable = tmp_dir.join("unreadable.txt");
            tokio::fs::write(&unreadable, "test").await?;
            tokio::fs::set_permissions(&unreadable, std::fs::Permissions::from_mode(0o000)).await?;

            // symlink_metadata succeeds even without read permission
            let src_metadata = tokio::fs::symlink_metadata(&unreadable).await?;
            let result = copy_file(
                &PROGRESS,
                &unreadable,
                &tmp_dir.join("dest.txt"),
                &src_metadata,
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: None,
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await;

            assert!(result.is_err(), "Should fail with permission error");
            let err_msg = get_full_error_message(&result.unwrap_err());

            // The error message MUST include the root cause
            assert!(
                err_msg.to_lowercase().contains("permission")
                    || err_msg.contains("EACCES")
                    || err_msg.contains("denied"),
                "Error message must include permission-related text. Got: {}",
                err_msg
            );
            Ok(())
        }

        #[tokio::test]
        #[traced_test]
        async fn test_nonexistent_source_includes_root_cause() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::create_temp_dir().await?;

            let result = copy(
                &PROGRESS,
                &tmp_dir.join("does_not_exist.txt"),
                &tmp_dir.join("dest.txt"),
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: None,
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await;

            assert!(result.is_err());
            let err_msg = get_full_error_message(&result.unwrap_err());

            assert!(
                err_msg.to_lowercase().contains("no such file")
                    || err_msg.to_lowercase().contains("not found")
                    || err_msg.contains("ENOENT"),
                "Error message must include file not found text. Got: {}",
                err_msg
            );
            Ok(())
        }

        #[tokio::test]
        #[traced_test]
        async fn test_unreadable_directory_includes_root_cause() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::create_temp_dir().await?;
            let unreadable_dir = tmp_dir.join("unreadable_dir");
            tokio::fs::create_dir(&unreadable_dir).await?;
            tokio::fs::set_permissions(&unreadable_dir, std::fs::Permissions::from_mode(0o000))
                .await?;

            let result = copy(
                &PROGRESS,
                &unreadable_dir,
                &tmp_dir.join("dest"),
                &Settings {
                    dereference: false,
                    fail_early: true,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: None,
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await;

            assert!(result.is_err());
            let err_msg = get_full_error_message(&result.unwrap_err());

            assert!(
                err_msg.to_lowercase().contains("permission")
                    || err_msg.contains("EACCES")
                    || err_msg.contains("denied"),
                "Error message must include permission-related text. Got: {}",
                err_msg
            );

            // Clean up - restore permissions so cleanup can remove it
            tokio::fs::set_permissions(&unreadable_dir, std::fs::Permissions::from_mode(0o700))
                .await?;
            Ok(())
        }

        #[tokio::test]
        #[traced_test]
        async fn test_destination_permission_error_includes_root_cause() -> Result<(), anyhow::Error>
        {
            let tmp_dir = testutils::setup_test_dir().await?;
            let test_path = tmp_dir.as_path();
            let readonly_parent = test_path.join("readonly_dest");
            tokio::fs::create_dir(&readonly_parent).await?;
            tokio::fs::set_permissions(&readonly_parent, std::fs::Permissions::from_mode(0o555))
                .await?;

            let result = copy(
                &PROGRESS,
                &test_path.join("foo"),
                &readonly_parent.join("copy"),
                &Settings {
                    dereference: false,
                    fail_early: true,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: None,
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await;

            // restore permissions so cleanup succeeds even when copy fails
            tokio::fs::set_permissions(&readonly_parent, std::fs::Permissions::from_mode(0o755))
                .await?;

            assert!(result.is_err(), "copy into read-only parent should fail");
            let err_msg = get_full_error_message(&result.unwrap_err());

            assert!(
                err_msg.to_lowercase().contains("permission denied") || err_msg.contains("EACCES"),
                "Error message must include permission denied text. Got: {}",
                err_msg
            );
            Ok(())
        }
    }

    mod empty_dir_cleanup_tests {
        use super::*;
        use crate::filter::FilterSettings;
        use std::path::Path;
        #[test]
        fn test_check_empty_dir_cleanup_no_filter() {
            // when no filter, always keep
            assert_eq!(
                check_empty_dir_cleanup(None, true, false, Path::new("any"), false, false),
                EmptyDirAction::Keep
            );
        }
        #[test]
        fn test_check_empty_dir_cleanup_something_copied() {
            // when content was copied, keep
            let mut filter = FilterSettings::new();
            filter.add_include("*.txt").unwrap();
            assert_eq!(
                check_empty_dir_cleanup(Some(&filter), true, true, Path::new("any"), false, false),
                EmptyDirAction::Keep
            );
        }
        #[test]
        fn test_check_empty_dir_cleanup_not_created() {
            // when we didn't create the directory, keep
            let mut filter = FilterSettings::new();
            filter.add_include("*.txt").unwrap();
            assert_eq!(
                check_empty_dir_cleanup(
                    Some(&filter),
                    false,
                    false,
                    Path::new("any"),
                    false,
                    false
                ),
                EmptyDirAction::Keep
            );
        }
        #[test]
        fn test_check_empty_dir_cleanup_directly_matched() {
            // when directory directly matches include pattern, keep
            let mut filter = FilterSettings::new();
            filter.add_include("target/").unwrap();
            assert_eq!(
                check_empty_dir_cleanup(
                    Some(&filter),
                    true,
                    false,
                    Path::new("target"),
                    false,
                    false
                ),
                EmptyDirAction::Keep
            );
        }
        #[test]
        fn test_check_empty_dir_cleanup_traversed_only() {
            // when directory was only traversed, remove
            let mut filter = FilterSettings::new();
            filter.add_include("*.txt").unwrap();
            assert_eq!(
                check_empty_dir_cleanup(Some(&filter), true, false, Path::new("src"), false, false),
                EmptyDirAction::Remove
            );
        }
        #[test]
        fn test_check_empty_dir_cleanup_dry_run() {
            // in dry-run mode, skip instead of remove
            let mut filter = FilterSettings::new();
            filter.add_include("*.txt").unwrap();
            assert_eq!(
                check_empty_dir_cleanup(Some(&filter), true, false, Path::new("src"), false, true),
                EmptyDirAction::DryRunSkip
            );
        }
        #[test]
        fn test_check_empty_dir_cleanup_root_always_kept() {
            // root directory is never removed, even with filter and nothing copied
            let mut filter = FilterSettings::new();
            filter.add_include("*.txt").unwrap();
            assert_eq!(
                check_empty_dir_cleanup(Some(&filter), true, false, Path::new(""), true, false),
                EmptyDirAction::Keep
            );
        }
        #[test]
        fn test_check_empty_dir_cleanup_root_kept_in_dry_run() {
            // root directory is kept even in dry-run mode
            let mut filter = FilterSettings::new();
            filter.add_include("*.txt").unwrap();
            assert_eq!(
                check_empty_dir_cleanup(Some(&filter), true, false, Path::new(""), true, true),
                EmptyDirAction::Keep
            );
        }
    }

    /// Verify that directory metadata is applied even when child operations fail.
    /// This is a regression test for a bug where directory permissions were not preserved
    /// when copying with fail_early=false and some children failed to copy.
    #[tokio::test]
    #[traced_test]
    async fn test_directory_metadata_applied_on_child_error() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::create_temp_dir().await?;
        let test_path = tmp_dir.as_path();
        // create source directory with specific permissions
        let src_dir = test_path.join("src");
        tokio::fs::create_dir(&src_dir).await?;
        tokio::fs::set_permissions(&src_dir, std::fs::Permissions::from_mode(0o750)).await?;
        // create a readable file and an unreadable file inside
        let readable_file = src_dir.join("readable.txt");
        tokio::fs::write(&readable_file, "content").await?;
        let unreadable_file = src_dir.join("unreadable.txt");
        tokio::fs::write(&unreadable_file, "secret").await?;
        tokio::fs::set_permissions(&unreadable_file, std::fs::Permissions::from_mode(0o000))
            .await?;
        let dst_dir = test_path.join("dst");
        // copy with fail_early=false and preserve=all
        let result = copy(
            &PROGRESS,
            &src_dir,
            &dst_dir,
            &Settings {
                dereference: false,
                fail_early: false,
                overwrite: false,
                overwrite_compare: Default::default(),
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await;
        // restore permissions so cleanup can succeed
        tokio::fs::set_permissions(&unreadable_file, std::fs::Permissions::from_mode(0o644))
            .await?;
        // verify the operation returned an error (unreadable file should fail)
        assert!(result.is_err(), "copy should fail due to unreadable file");
        let error = result.unwrap_err();
        // verify some files were copied (the readable one)
        assert_eq!(error.summary.files_copied, 1);
        assert_eq!(error.summary.directories_created, 1);
        // verify the destination directory exists and has the correct permissions
        let dst_metadata = tokio::fs::metadata(&dst_dir).await?;
        assert!(dst_metadata.is_dir());
        let actual_mode = dst_metadata.permissions().mode() & 0o7777;
        assert_eq!(
            actual_mode, 0o750,
            "directory should have preserved source permissions (0o750), got {:o}",
            actual_mode
        );
        Ok(())
    }

    /// Verify that fail-early does not apply parent directory metadata after a child fails.
    #[tokio::test]
    #[traced_test]
    async fn test_fail_early_does_not_apply_parent_directory_metadata_after_child_error()
    -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::create_temp_dir().await?;
        let test_path = tmp_dir.as_path();
        let src_dir = test_path.join("src");
        tokio::fs::create_dir(&src_dir).await?;
        tokio::fs::write(src_dir.join("readable.txt"), "content").await?;
        let unreadable_file = src_dir.join("unreadable.txt");
        tokio::fs::write(&unreadable_file, "secret").await?;
        tokio::fs::set_permissions(&unreadable_file, std::fs::Permissions::from_mode(0o000))
            .await?;
        let fixed_secs = 946684800;
        let fixed_nsec = 123_456_789;
        let fixed_time = nix::sys::time::TimeSpec::new(fixed_secs, fixed_nsec);
        nix::sys::stat::utimensat(
            nix::fcntl::AT_FDCWD,
            &src_dir,
            &fixed_time,
            &fixed_time,
            nix::sys::stat::UtimensatFlags::NoFollowSymlink,
        )?;
        let src_metadata = tokio::fs::metadata(&src_dir).await?;
        let dst_dir = test_path.join("dst");
        let result = copy(
            &PROGRESS,
            &src_dir,
            &dst_dir,
            &Settings {
                dereference: false,
                fail_early: true,
                overwrite: false,
                overwrite_compare: Default::default(),
                overwrite_filter: None,
                ignore_existing: false,
                chunk_size: 0,
                skip_specials: false,
                remote_copy_buffer_size: 0,
                filter: None,
                dry_run: None,
                delete: None,
            },
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await;
        tokio::fs::set_permissions(&unreadable_file, std::fs::Permissions::from_mode(0o644))
            .await?;
        assert!(result.is_err(), "copy should fail due to unreadable file");
        let dst_metadata = tokio::fs::metadata(&dst_dir).await?;
        assert!(dst_metadata.is_dir());
        assert_ne!(
            (dst_metadata.mtime(), dst_metadata.mtime_nsec()),
            (src_metadata.mtime(), src_metadata.mtime_nsec()),
            "fail-early should return before applying preserved directory timestamps"
        );
        Ok(())
    }
    mod filter_tests {
        use super::*;
        use crate::filter::FilterSettings;
        /// Test that path-based patterns (with /) work correctly with nested paths.
        /// This test exposes the bug where only entry_name is passed to the filter
        /// instead of the relative path.
        #[tokio::test]
        #[traced_test]
        async fn test_path_pattern_matches_nested_files() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::setup_test_dir().await?;
            let test_path = tmp_dir.as_path();
            // test directory structure from setup_test_dir:
            // foo/
            //   0.txt
            //   bar/
            //     1.txt
            //     2.txt
            //   baz/
            //     3.txt -> ../0.txt (symlink)
            //     4.txt
            //     5 -> ../bar (symlink)
            // create filter that should match bar/*.txt (files in bar directory)
            let mut filter = FilterSettings::new();
            filter.add_include("bar/*.txt").unwrap();
            let summary = copy(
                &PROGRESS,
                &test_path.join("foo"),
                &test_path.join("dst"),
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: Some(filter),
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            // should only copy files matching bar/*.txt pattern
            // bar/1.txt, bar/2.txt, and bar/3.txt should be copied
            assert_eq!(
                summary.files_copied, 3,
                "should copy 3 files matching bar/*.txt"
            );
            // verify the right files exist
            assert!(
                test_path.join("dst/bar/1.txt").exists(),
                "bar/1.txt should be copied"
            );
            assert!(
                test_path.join("dst/bar/2.txt").exists(),
                "bar/2.txt should be copied"
            );
            assert!(
                test_path.join("dst/bar/3.txt").exists(),
                "bar/3.txt should be copied"
            );
            // verify files outside the pattern don't exist
            assert!(
                !test_path.join("dst/0.txt").exists(),
                "0.txt should not be copied"
            );
            Ok(())
        }
        /// Test that anchored patterns (starting with /) match only at root.
        #[tokio::test]
        #[traced_test]
        async fn test_anchored_pattern_matches_only_at_root() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::setup_test_dir().await?;
            let test_path = tmp_dir.as_path();
            // create filter that should match /bar/** (bar directory and all its contents)
            let mut filter = FilterSettings::new();
            filter.add_include("/bar/**").unwrap();
            let summary = copy(
                &PROGRESS,
                &test_path.join("foo"),
                &test_path.join("dst"),
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: Some(filter),
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            // should only copy bar directory and its contents
            assert!(
                test_path.join("dst/bar").exists(),
                "bar directory should be copied"
            );
            assert!(
                !test_path.join("dst/baz").exists(),
                "baz directory should not be copied"
            );
            assert!(
                !test_path.join("dst/0.txt").exists(),
                "0.txt should not be copied"
            );
            // verify summary counts
            assert_eq!(
                summary.files_copied, 3,
                "should copy 3 files in bar (1.txt, 2.txt, 3.txt)"
            );
            assert_eq!(
                summary.directories_created, 2,
                "should create 2 directories (root dst + bar)"
            );
            // skipped: 0.txt (file) and baz (directory) - baz contents not counted since dir is skipped
            assert_eq!(summary.files_skipped, 1, "should skip 1 file (0.txt)");
            assert_eq!(
                summary.directories_skipped, 1,
                "should skip 1 directory (baz)"
            );
            Ok(())
        }
        /// Test that double-star patterns (**) match across directories.
        #[tokio::test]
        #[traced_test]
        async fn test_double_star_pattern_matches_nested() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::setup_test_dir().await?;
            let test_path = tmp_dir.as_path();
            // create filter that should match all .txt files at any depth
            let mut filter = FilterSettings::new();
            filter.add_include("**/*.txt").unwrap();
            let summary = copy(
                &PROGRESS,
                &test_path.join("foo"),
                &test_path.join("dst"),
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: Some(filter),
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            // should copy all .txt files: 0.txt, bar/1.txt, bar/2.txt, bar/3.txt, baz/4.txt
            assert_eq!(
                summary.files_copied, 5,
                "should copy all 5 .txt files with **/*.txt pattern"
            );
            Ok(())
        }
        /// Test that filters are applied to top-level file arguments.
        #[tokio::test]
        #[traced_test]
        async fn test_filter_applies_to_single_file_source() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::setup_test_dir().await?;
            let test_path = tmp_dir.as_path();
            // create filter that excludes .txt files
            let mut filter = FilterSettings::new();
            filter.add_exclude("*.txt").unwrap();
            let result = copy(
                &PROGRESS,
                &test_path.join("foo/0.txt"), // single file source
                &test_path.join("dst.txt"),
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: Some(filter),
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            // the file should NOT be copied because it matches the exclude pattern
            assert_eq!(
                result.files_copied, 0,
                "file matching exclude pattern should not be copied"
            );
            assert!(
                !test_path.join("dst.txt").exists(),
                "excluded file should not exist at destination"
            );
            Ok(())
        }
        /// Test that filters apply to root directories with simple exclude patterns.
        #[tokio::test]
        #[traced_test]
        async fn test_filter_applies_to_root_directory() -> Result<(), anyhow::Error> {
            let test_path = testutils::create_temp_dir().await?;
            // create a directory that should be excluded
            tokio::fs::create_dir_all(test_path.join("excluded_dir")).await?;
            tokio::fs::write(test_path.join("excluded_dir/file.txt"), "content").await?;
            // create filter that excludes *_dir/ directories
            let mut filter = FilterSettings::new();
            filter.add_exclude("*_dir/").unwrap();
            let result = copy(
                &PROGRESS,
                &test_path.join("excluded_dir"),
                &test_path.join("dst"),
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: Some(filter),
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            // directory should NOT be copied because it matches exclude pattern
            assert_eq!(
                result.directories_created, 0,
                "root directory matching exclude should not be created"
            );
            assert!(
                !test_path.join("dst").exists(),
                "excluded root directory should not exist at destination"
            );
            Ok(())
        }
        /// Test that filters apply to root symlinks with simple exclude patterns.
        #[tokio::test]
        #[traced_test]
        async fn test_filter_applies_to_root_symlink() -> Result<(), anyhow::Error> {
            let test_path = testutils::create_temp_dir().await?;
            // create a target file and a symlink to it
            tokio::fs::write(test_path.join("target.txt"), "content").await?;
            tokio::fs::symlink(
                test_path.join("target.txt"),
                test_path.join("excluded_link"),
            )
            .await?;
            // create filter that excludes *_link
            let mut filter = FilterSettings::new();
            filter.add_exclude("*_link").unwrap();
            let result = copy(
                &PROGRESS,
                &test_path.join("excluded_link"),
                &test_path.join("dst"),
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: Some(filter),
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            // symlink should NOT be copied because it matches exclude pattern
            assert_eq!(
                result.symlinks_created, 0,
                "root symlink matching exclude should not be created"
            );
            assert!(
                !test_path.join("dst").exists(),
                "excluded root symlink should not exist at destination"
            );
            Ok(())
        }
        /// Test combined include and exclude patterns (exclude takes precedence).
        #[tokio::test]
        #[traced_test]
        async fn test_combined_include_exclude_patterns() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::setup_test_dir().await?;
            let test_path = tmp_dir.as_path();
            // test structure from setup_test_dir:
            // foo/
            //   0.txt
            //   bar/ (1.txt, 2.txt, 3.txt)
            //   baz/ (4.txt, 5.txt symlink, 6.txt symlink)
            // include all .txt files, but exclude bar/2.txt specifically
            let mut filter = FilterSettings::new();
            filter.add_include("**/*.txt").unwrap();
            filter.add_exclude("bar/2.txt").unwrap();
            let summary = copy(
                &PROGRESS,
                &test_path.join("foo"),
                &test_path.join("dst"),
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: Some(filter),
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            // should copy: 0.txt, bar/1.txt, bar/3.txt, baz/4.txt = 4 files
            // should skip: bar/2.txt (excluded by pattern) = 1 file
            // symlinks 5.txt and 6.txt don't match *.txt include pattern (symlinks, not files)
            assert_eq!(summary.files_copied, 4, "should copy 4 .txt files");
            assert_eq!(
                summary.files_skipped, 1,
                "should skip 1 file (bar/2.txt excluded)"
            );
            // verify specific files
            assert!(
                test_path.join("dst/bar/1.txt").exists(),
                "bar/1.txt should be copied"
            );
            assert!(
                !test_path.join("dst/bar/2.txt").exists(),
                "bar/2.txt should be excluded"
            );
            assert!(
                test_path.join("dst/bar/3.txt").exists(),
                "bar/3.txt should be copied"
            );
            Ok(())
        }
        /// Test that skipped counts accurately reflect what was filtered.
        #[tokio::test]
        #[traced_test]
        async fn test_skipped_counts_comprehensive() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::setup_test_dir().await?;
            let test_path = tmp_dir.as_path();
            // test structure from setup_test_dir:
            // foo/
            //   0.txt
            //   bar/ (1.txt, 2.txt, 3.txt)
            //   baz/ (4.txt, 5.txt symlink, 6.txt symlink)
            // exclude bar/ directory entirely
            let mut filter = FilterSettings::new();
            filter.add_exclude("bar/").unwrap();
            let summary = copy(
                &PROGRESS,
                &test_path.join("foo"),
                &test_path.join("dst"),
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: Some(filter),
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            // copied: 0.txt (1 file), baz/4.txt (1 file), 5.txt symlink, 6.txt symlink
            // skipped: bar directory (1 dir) - contents not counted since whole dir skipped
            // directories: foo (root), baz = 2
            assert_eq!(summary.files_copied, 2, "should copy 2 files");
            assert_eq!(summary.symlinks_created, 2, "should copy 2 symlinks");
            assert_eq!(
                summary.directories_created, 2,
                "should create 2 directories"
            );
            assert_eq!(
                summary.directories_skipped, 1,
                "should skip 1 directory (bar)"
            );
            assert_eq!(
                summary.files_skipped, 0,
                "no files skipped (bar contents not counted)"
            );
            Ok(())
        }
        /// Test that empty directories are not created when they were only traversed to look
        /// for matches (regression test for bug where --include='foo' would create empty dir baz).
        #[tokio::test]
        #[traced_test]
        async fn test_empty_dir_not_created_when_only_traversed() -> Result<(), anyhow::Error> {
            let test_path = testutils::create_temp_dir().await?;
            // create structure:
            // src/
            //   foo (file)
            //   bar (file)
            //   baz/ (empty directory)
            let src_path = test_path.join("src");
            tokio::fs::create_dir(&src_path).await?;
            tokio::fs::write(src_path.join("foo"), "content").await?;
            tokio::fs::write(src_path.join("bar"), "content").await?;
            tokio::fs::create_dir(src_path.join("baz")).await?;
            // include only 'foo' file
            let mut filter = FilterSettings::new();
            filter.add_include("foo").unwrap();
            let summary = copy(
                &PROGRESS,
                &src_path,
                &test_path.join("dst"),
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: Some(filter),
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            // only 'foo' should be copied
            assert_eq!(summary.files_copied, 1, "should copy only 'foo' file");
            assert_eq!(
                summary.directories_created, 1,
                "should create only root directory (not empty 'baz')"
            );
            // verify foo was copied
            assert!(
                test_path.join("dst").join("foo").exists(),
                "foo should be copied"
            );
            // verify bar was not copied (not matching include pattern)
            assert!(
                !test_path.join("dst").join("bar").exists(),
                "bar should not be copied"
            );
            // verify empty baz directory was NOT created
            assert!(
                !test_path.join("dst").join("baz").exists(),
                "empty baz directory should NOT be created"
            );
            Ok(())
        }
        /// Test that directories with only non-matching content are not created at destination.
        /// This is different from empty directories - the source dir has content but none matches.
        #[tokio::test]
        #[traced_test]
        async fn test_dir_with_nonmatching_content_not_created() -> Result<(), anyhow::Error> {
            let test_path = testutils::create_temp_dir().await?;
            // create structure:
            // src/
            //   foo (file)
            //   baz/
            //     qux (file - doesn't match 'foo')
            //     quux (file - doesn't match 'foo')
            let src_path = test_path.join("src");
            tokio::fs::create_dir(&src_path).await?;
            tokio::fs::write(src_path.join("foo"), "content").await?;
            tokio::fs::create_dir(src_path.join("baz")).await?;
            tokio::fs::write(src_path.join("baz").join("qux"), "content").await?;
            tokio::fs::write(src_path.join("baz").join("quux"), "content").await?;
            // include only 'foo' file
            let mut filter = FilterSettings::new();
            filter.add_include("foo").unwrap();
            let summary = copy(
                &PROGRESS,
                &src_path,
                &test_path.join("dst"),
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: Some(filter),
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            // only 'foo' should be copied
            assert_eq!(summary.files_copied, 1, "should copy only 'foo' file");
            assert_eq!(
                summary.files_skipped, 2,
                "should skip 2 files (qux and quux)"
            );
            assert_eq!(
                summary.directories_created, 1,
                "should create only root directory (not 'baz' with non-matching content)"
            );
            // verify foo was copied
            assert!(
                test_path.join("dst").join("foo").exists(),
                "foo should be copied"
            );
            // verify baz directory was NOT created (even though source baz has content)
            assert!(
                !test_path.join("dst").join("baz").exists(),
                "baz directory should NOT be created (no matching content inside)"
            );
            Ok(())
        }
        /// Test that empty directories are not reported as created in dry-run mode
        /// when they were only traversed.
        #[tokio::test]
        #[traced_test]
        async fn test_dry_run_empty_dir_not_reported_as_created() -> Result<(), anyhow::Error> {
            let test_path = testutils::create_temp_dir().await?;
            // create structure:
            // src/
            //   foo (file)
            //   bar (file)
            //   baz/ (empty directory)
            let src_path = test_path.join("src");
            tokio::fs::create_dir(&src_path).await?;
            tokio::fs::write(src_path.join("foo"), "content").await?;
            tokio::fs::write(src_path.join("bar"), "content").await?;
            tokio::fs::create_dir(src_path.join("baz")).await?;
            // include only 'foo' file
            let mut filter = FilterSettings::new();
            filter.add_include("foo").unwrap();
            let summary = copy(
                &PROGRESS,
                &src_path,
                &test_path.join("dst"),
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: Some(filter),
                    dry_run: Some(crate::config::DryRunMode::Explain),
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            // only 'foo' should be reported as would-be-copied
            assert_eq!(
                summary.files_copied, 1,
                "should report only 'foo' would be copied"
            );
            assert_eq!(
                summary.directories_created, 1,
                "should report only root directory would be created (not empty 'baz')"
            );
            // verify nothing was actually created (dry-run mode)
            assert!(
                !test_path.join("dst").exists(),
                "dst should not exist in dry-run"
            );
            Ok(())
        }
        /// Test that existing directories are NOT removed when using --overwrite,
        /// even if nothing is copied into them due to filters.
        #[tokio::test]
        #[traced_test]
        async fn test_existing_dir_not_removed_with_overwrite() -> Result<(), anyhow::Error> {
            let test_path = testutils::create_temp_dir().await?;
            // create source structure:
            // src/
            //   foo (file)
            //   bar (file)
            //   baz/ (empty directory)
            let src_path = test_path.join("src");
            tokio::fs::create_dir(&src_path).await?;
            tokio::fs::write(src_path.join("foo"), "content").await?;
            tokio::fs::write(src_path.join("bar"), "content").await?;
            tokio::fs::create_dir(src_path.join("baz")).await?;
            // create destination with baz directory already existing
            let dst_path = test_path.join("dst");
            tokio::fs::create_dir(&dst_path).await?;
            tokio::fs::create_dir(dst_path.join("baz")).await?;
            // add a marker file inside dst/baz to verify we don't touch it
            tokio::fs::write(dst_path.join("baz").join("marker.txt"), "existing").await?;
            // include only 'foo' file - baz should not match
            let mut filter = FilterSettings::new();
            filter.add_include("foo").unwrap();
            let summary = copy(
                &PROGRESS,
                &src_path,
                &dst_path,
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: true, // enable overwrite mode
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: Some(filter),
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            // foo should be copied
            assert_eq!(summary.files_copied, 1, "should copy only 'foo' file");
            // dst and baz should be unchanged (both already existed)
            assert_eq!(
                summary.directories_unchanged, 2,
                "root dst and baz directories should be unchanged"
            );
            assert_eq!(
                summary.directories_created, 0,
                "should not create any directories"
            );
            // verify foo was copied
            assert!(dst_path.join("foo").exists(), "foo should be copied");
            // verify bar was NOT copied
            assert!(!dst_path.join("bar").exists(), "bar should not be copied");
            // verify existing baz directory still exists with its content
            assert!(
                dst_path.join("baz").exists(),
                "existing baz directory should still exist"
            );
            assert!(
                dst_path.join("baz").join("marker.txt").exists(),
                "existing content in baz should still exist"
            );
            Ok(())
        }
    }
    mod dry_run_tests {
        use super::*;
        use crate::filter::FilterSettings;
        /// Test that dry-run mode for directories doesn't create the destination
        /// and doesn't try to set metadata on non-existent directories.
        #[tokio::test]
        #[traced_test]
        async fn test_dry_run_directory_does_not_create_destination() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::setup_test_dir().await?;
            let test_path = tmp_dir.as_path();
            let dst_path = test_path.join("nonexistent_dst");
            // verify destination doesn't exist
            assert!(
                !dst_path.exists(),
                "destination should not exist before dry-run"
            );
            let summary = copy(
                &PROGRESS,
                &test_path.join("foo"),
                &dst_path,
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: None,
                    dry_run: Some(crate::config::DryRunMode::Brief),
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            // verify destination still doesn't exist
            assert!(
                !dst_path.exists(),
                "dry-run should not create destination directory"
            );
            // verify summary reports what would be created
            assert!(
                summary.directories_created > 0,
                "dry-run should report directories that would be created"
            );
            assert!(
                summary.files_copied > 0,
                "dry-run should report files that would be copied"
            );
            Ok(())
        }
        /// Test that root directory is always created even when nothing matches
        /// the include pattern. The root is the user-specified source — it should
        /// never be removed/skipped due to empty-dir cleanup.
        #[tokio::test]
        #[traced_test]
        async fn test_root_dir_preserved_when_nothing_matches() -> Result<(), anyhow::Error> {
            let test_path = testutils::create_temp_dir().await?;
            // create structure:
            // src/
            //   bar.log (doesn't match *.txt)
            //   baz/ (empty directory)
            let src_path = test_path.join("src");
            tokio::fs::create_dir(&src_path).await?;
            tokio::fs::write(src_path.join("bar.log"), "content").await?;
            tokio::fs::create_dir(src_path.join("baz")).await?;
            // include only *.txt - nothing in source matches
            let mut filter = FilterSettings::new();
            filter.add_include("*.txt").unwrap();
            let dst_path = test_path.join("dst");
            let summary = copy(
                &PROGRESS,
                &src_path,
                &dst_path,
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: Some(filter),
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            // no files should be copied
            assert_eq!(summary.files_copied, 0, "no files match *.txt");
            // root directory should still be created
            assert_eq!(
                summary.directories_created, 1,
                "root directory should always be created"
            );
            assert!(dst_path.exists(), "root destination directory should exist");
            // non-matching subdirectories should not be created
            assert!(
                !dst_path.join("baz").exists(),
                "empty baz should not be created"
            );
            Ok(())
        }
        /// Test that root directory is counted in dry-run even when nothing matches.
        #[tokio::test]
        #[traced_test]
        async fn test_root_dir_counted_in_dry_run_when_nothing_matches() -> Result<(), anyhow::Error>
        {
            let test_path = testutils::create_temp_dir().await?;
            let src_path = test_path.join("src");
            tokio::fs::create_dir(&src_path).await?;
            tokio::fs::write(src_path.join("bar.log"), "content").await?;
            // include only *.txt - nothing matches
            let mut filter = FilterSettings::new();
            filter.add_include("*.txt").unwrap();
            let dst_path = test_path.join("dst");
            let summary = copy(
                &PROGRESS,
                &src_path,
                &dst_path,
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: Some(filter),
                    dry_run: Some(crate::config::DryRunMode::Explain),
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            assert_eq!(summary.files_copied, 0, "no files match *.txt");
            assert_eq!(
                summary.directories_created, 1,
                "root directory should be counted in dry-run"
            );
            assert!(
                !dst_path.exists(),
                "nothing should be created in dry-run mode"
            );
            Ok(())
        }
    }

    /// stress tests exercising max-open-files saturation during copy
    mod max_open_files_tests {
        use super::*;

        /// wide copy: many files with a very low open-files limit.
        /// verifies all files are copied correctly under permit saturation.
        #[tokio::test]
        #[traced_test]
        async fn wide_copy_under_open_files_saturation() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::create_temp_dir().await?;
            let src = tmp_dir.join("src");
            let dst = tmp_dir.join("dst");
            tokio::fs::create_dir(&src).await?;
            let file_count = 200;
            for i in 0..file_count {
                tokio::fs::write(src.join(format!("{}.txt", i)), format!("content-{}", i)).await?;
            }
            // set a very low limit to force permit contention
            throttle::set_max_open_files(4);
            let summary = copy(
                &PROGRESS,
                &src,
                &dst,
                &Settings {
                    dereference: false,
                    fail_early: true,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: None,
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            assert_eq!(summary.files_copied, file_count);
            assert_eq!(summary.directories_created, 1);
            for i in 0..file_count {
                let content = tokio::fs::read_to_string(dst.join(format!("{}.txt", i))).await?;
                assert_eq!(content, format!("content-{}", i));
            }
            Ok(())
        }

        /// deep + wide copy: directory tree deeper than the open-files limit, with files
        /// at every level. verifies no deadlock occurs (directories don't consume permits).
        #[tokio::test]
        #[traced_test]
        async fn deep_tree_no_deadlock_under_open_files_saturation() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::create_temp_dir().await?;
            let src = tmp_dir.join("src");
            let dst = tmp_dir.join("dst");
            let depth = 20;
            let files_per_level = 5;
            let limit = 4;
            // create a directory chain deeper than the permit limit, with files at each level
            let mut dir = src.clone();
            for level in 0..depth {
                tokio::fs::create_dir_all(&dir).await?;
                for f in 0..files_per_level {
                    tokio::fs::write(
                        dir.join(format!("f{}_{}.txt", level, f)),
                        format!("L{}F{}", level, f),
                    )
                    .await?;
                }
                dir = dir.join(format!("d{}", level));
            }
            throttle::set_max_open_files(limit);
            let summary = tokio::time::timeout(
                std::time::Duration::from_secs(30),
                copy(
                    &PROGRESS,
                    &src,
                    &dst,
                    &Settings {
                        dereference: false,
                        fail_early: true,
                        overwrite: false,
                        overwrite_compare: Default::default(),
                        overwrite_filter: None,
                        ignore_existing: false,
                        chunk_size: 0,
                        skip_specials: false,
                        remote_copy_buffer_size: 0,
                        filter: None,
                        dry_run: None,
                        delete: None,
                    },
                    &NO_PRESERVE_SETTINGS,
                    false,
                ),
            )
            .await
            .context("copy timed out — possible deadlock")?
            .context("copy failed")?;
            assert_eq!(summary.files_copied, depth * files_per_level);
            assert_eq!(summary.directories_created, depth);
            // spot-check content at a few levels
            let mut check_dir = dst.clone();
            for level in 0..depth {
                let content =
                    tokio::fs::read_to_string(check_dir.join(format!("f{}_0.txt", level))).await?;
                assert_eq!(content, format!("L{}F0", level));
                check_dir = check_dir.join(format!("d{}", level));
            }
            Ok(())
        }

        /// Regression: copy_file → rm cross-pool deadlock.
        ///
        /// Scenario: many parallel copies overwrite destinations that are
        /// directories (so each copy_file path takes the
        /// "remove existing then copy" branch, and rm recurses). Each copy
        /// task holds an open-files permit during copy_file; if rm also
        /// drew permits from the open-files pool, a saturated pool would
        /// deadlock — every permit held by a copy task waiting for rm to
        /// release one. Decoupling rm onto pending-meta avoids that.
        #[tokio::test]
        #[traced_test]
        async fn parallel_overwrite_dir_with_file_no_deadlock() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::create_temp_dir().await?;
            let src = tmp_dir.join("src");
            let dst = tmp_dir.join("dst");
            tokio::fs::create_dir(&src).await?;
            tokio::fs::create_dir(&dst).await?;
            // 8 sources are regular files; the 8 corresponding destinations
            // are directories with nested files — copy with --overwrite
            // forces rm of each dst directory tree from inside copy_file.
            let n = 8;
            for i in 0..n {
                tokio::fs::write(src.join(format!("e{}", i)), format!("file-{}", i)).await?;
                let dst_subdir = dst.join(format!("e{}", i));
                tokio::fs::create_dir(&dst_subdir).await?;
                for j in 0..3 {
                    tokio::fs::write(
                        dst_subdir.join(format!("inner_{}.txt", j)),
                        format!("inner-{}-{}", i, j),
                    )
                    .await?;
                }
            }
            // Saturate the open-files pool: if rm shared this pool, every
            // outer copy task would hold its single permit and the inner rm
            // recursion would block forever.
            throttle::set_max_open_files(2);
            let summary = tokio::time::timeout(
                std::time::Duration::from_secs(30),
                copy(
                    &PROGRESS,
                    &src,
                    &dst,
                    &Settings {
                        dereference: false,
                        fail_early: true,
                        overwrite: true,
                        overwrite_compare: Default::default(),
                        overwrite_filter: None,
                        ignore_existing: false,
                        chunk_size: 0,
                        skip_specials: false,
                        remote_copy_buffer_size: 0,
                        filter: None,
                        dry_run: None,
                        delete: None,
                    },
                    &NO_PRESERVE_SETTINGS,
                    false,
                ),
            )
            .await
            .context(
                "copy timed out — deadlock between copy_file's open-files permit and inner rm",
            )?
            .context("copy failed")?;
            assert_eq!(summary.files_copied, n);
            assert_eq!(summary.rm_summary.files_removed, n * 3);
            assert_eq!(summary.rm_summary.directories_removed, n);
            for i in 0..n {
                let path = dst.join(format!("e{}", i));
                let content = tokio::fs::read_to_string(&path).await?;
                assert_eq!(content, format!("file-{}", i));
            }
            Ok(())
        }
    }

    mod skip_specials_tests {
        use super::*;

        #[tokio::test]
        #[traced_test]
        async fn skip_specials_skips_socket_in_directory() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::setup_test_dir().await?;
            let test_path = tmp_dir.as_path();
            let src = test_path.join("src_dir");
            let dst = test_path.join("dst_dir");
            tokio::fs::create_dir(&src).await?;
            tokio::fs::write(src.join("file.txt"), "hello").await?;
            // create a unix socket inside the source directory
            let _listener = std::os::unix::net::UnixListener::bind(src.join("test.sock"))?;
            let summary = copy(
                &PROGRESS,
                &src,
                &dst,
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: true,
                    remote_copy_buffer_size: 0,
                    filter: None,
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            assert_eq!(summary.files_copied, 1);
            assert_eq!(summary.specials_skipped, 1);
            assert!(dst.join("file.txt").exists());
            assert!(!dst.join("test.sock").exists());
            Ok(())
        }

        #[tokio::test]
        #[traced_test]
        async fn skip_specials_skips_fifo_in_directory() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::setup_test_dir().await?;
            let test_path = tmp_dir.as_path();
            let src = test_path.join("src_dir");
            let dst = test_path.join("dst_dir");
            tokio::fs::create_dir(&src).await?;
            tokio::fs::write(src.join("file.txt"), "hello").await?;
            // create a FIFO inside the source directory
            nix::unistd::mkfifo(
                &src.join("test.fifo"),
                nix::sys::stat::Mode::S_IRUSR | nix::sys::stat::Mode::S_IWUSR,
            )?;
            let summary = copy(
                &PROGRESS,
                &src,
                &dst,
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: true,
                    remote_copy_buffer_size: 0,
                    filter: None,
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            assert_eq!(summary.files_copied, 1);
            assert_eq!(summary.specials_skipped, 1);
            assert!(dst.join("file.txt").exists());
            assert!(!dst.join("test.fifo").exists());
            Ok(())
        }

        #[tokio::test]
        #[traced_test]
        async fn special_file_errors_without_skip_specials() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::setup_test_dir().await?;
            let test_path = tmp_dir.as_path();
            let src = test_path.join("src_dir");
            let dst = test_path.join("dst_dir");
            tokio::fs::create_dir(&src).await?;
            tokio::fs::write(src.join("file.txt"), "hello").await?;
            let _listener = std::os::unix::net::UnixListener::bind(src.join("test.sock"))?;
            let result = copy(
                &PROGRESS,
                &src,
                &dst,
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: false,
                    remote_copy_buffer_size: 0,
                    filter: None,
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await;
            assert!(result.is_err());
            let err = result.unwrap_err();
            assert!(
                format!("{:#}", err).contains("unsupported src file type"),
                "error should mention unsupported file type, got: {:#}",
                err
            );
            Ok(())
        }

        #[tokio::test]
        #[traced_test]
        async fn skip_specials_top_level_socket() -> Result<(), anyhow::Error> {
            let tmp_dir = testutils::setup_test_dir().await?;
            let test_path = tmp_dir.as_path();
            let src_socket = test_path.join("test.sock");
            let dst = test_path.join("dst.sock");
            let _listener = std::os::unix::net::UnixListener::bind(&src_socket)?;
            let summary = copy(
                &PROGRESS,
                &src_socket,
                &dst,
                &Settings {
                    dereference: false,
                    fail_early: false,
                    overwrite: false,
                    overwrite_compare: Default::default(),
                    overwrite_filter: None,
                    ignore_existing: false,
                    chunk_size: 0,
                    skip_specials: true,
                    remote_copy_buffer_size: 0,
                    filter: None,
                    dry_run: None,
                    delete: None,
                },
                &NO_PRESERVE_SETTINGS,
                false,
            )
            .await?;
            assert_eq!(summary.specials_skipped, 1);
            assert_eq!(summary.files_copied, 0);
            assert!(!dst.exists());
            Ok(())
        }
    }

    #[tokio::test]
    #[traced_test]
    async fn delete_protects_excluded_then_removes_with_delete_excluded()
    -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        let src = test_path.join("foo");
        let dst = test_path.join("bar");
        copy(
            &PROGRESS,
            &src,
            &dst,
            &settings_with_delete(None),
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        // a destination-only file that matches an exclude pattern
        tokio::fs::write(dst.join("keep.log"), b"protected").await?;

        let mut filter = crate::filter::FilterSettings::new();
        filter.add_exclude("*.log")?;

        // default --delete: keep.log is protected
        let mut settings = settings_with_delete(delete_on());
        settings.filter = Some(filter.clone());
        copy(
            &PROGRESS,
            &src,
            &dst,
            &settings,
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert!(
            dst.join("keep.log").exists(),
            "*.log must be protected by default"
        );

        // --delete-excluded: keep.log is removed
        let mut settings = settings_with_delete(Some(DeleteSettings {
            delete_excluded: true,
        }));
        settings.filter = Some(filter);
        copy(
            &PROGRESS,
            &src,
            &dst,
            &settings,
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        assert!(!dst.join("keep.log").exists());
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn delete_dry_run_reports_without_removing() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        let src = test_path.join("foo");
        let dst = test_path.join("bar");
        copy(
            &PROGRESS,
            &src,
            &dst,
            &settings_with_delete(None),
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        tokio::fs::write(dst.join("stale.txt"), b"junk").await?;

        let mut settings = settings_with_delete(delete_on());
        settings.dry_run = Some(crate::config::DryRunMode::Brief);
        let summary = copy(
            &PROGRESS,
            &src,
            &dst,
            &settings,
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;

        // the key invariant: dry-run must NOT remove anything
        assert!(
            dst.join("stale.txt").exists(),
            "dry-run must not remove anything"
        );
        // rm's dry-run does count would-be removals in files_removed
        assert_eq!(summary.rm_summary.files_removed, 1);
        Ok(())
    }

    #[tokio::test]
    #[traced_test]
    async fn delete_does_not_prune_when_source_unreadable() -> Result<(), anyhow::Error> {
        let tmp_dir = testutils::setup_test_dir().await?;
        let test_path = tmp_dir.as_path();
        let src = test_path.join("foo");
        let dst = test_path.join("bar");
        copy(
            &PROGRESS,
            &src,
            &dst,
            &settings_with_delete(None),
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await?;
        tokio::fs::write(dst.join("stale.txt"), b"junk").await?;
        // make the source directory unreadable so enumeration fails
        let original = tokio::fs::metadata(&src).await?.permissions();
        tokio::fs::set_permissions(&src, std::fs::Permissions::from_mode(0o000)).await?;

        let result = copy(
            &PROGRESS,
            &src,
            &dst,
            &settings_with_delete(delete_on()),
            &DO_PRESERVE_SETTINGS,
            false,
        )
        .await;

        // restore permissions before asserting (so the temp dir cleans up)
        tokio::fs::set_permissions(&src, original).await?;

        assert!(result.is_err(), "unreadable source must error");
        assert!(
            dst.join("stale.txt").exists(),
            "destination must not be pruned when source enumeration fails"
        );
        Ok(())
    }
}