//! Agent state storage ops: btrfs subvolumes (`EnsureAgentSubvolume`,
//! `DeleteAgentSubvolume`, `UpgradeAgentSubvolume`), qgroup accounting and
//! quotas (`EnsureBtrfsQuota`, `ReadSubvolumeUsage`, `SetSubvolumeQuota`),
//! read-only snapshots and their export (`SnapshotAgentSubvolume`,
//! `DeleteAgentSnapshot`, `SendAgentSnapshotToFile`, `SendAgentSnapshotToFd`)
//! and the `du`-based `MeasureAgentDisk`.
//!
//! Every path is derived from a validated agent name (and, for snapshots, a
//! `hive-`-prefixed label) under `AGENT_STATE_ROOT`.

use std::os::fd::OwnedFd;
use std::path::{Path, PathBuf};

use anyhow::{Context as _, Result, anyhow, bail};
use hive_priv_sock::{AGENT_STATE_ROOT, MEASURE_AGENT_DISK_TIMEOUT, MIGRATE_STAGING_ROOT};
use tokio::process::Command;

use crate::shared::{
    RealSpawn, Spawn, container_system_name, validate_agent_name, validate_credential_name,
};

/// `EnsureAgentSubvolume`.
pub(crate) async fn exec_ensure_agent_subvolume(agent_name: &str) -> Result<(String, String)> {
    validate_agent_name(agent_name)?;
    ensure_agent_subvolume(&Env::real(), agent_name).await
}

/// `DeleteAgentSubvolume`.
pub(crate) async fn exec_delete_agent_subvolume(agent_name: &str) -> Result<(String, String)> {
    validate_agent_name(agent_name)?;
    delete_agent_subvolume(&Env::real(), agent_name).await
}

/// `ReadSubvolumeUsage`.
pub(crate) async fn exec_read_subvolume_usage(agent_name: &str) -> Result<(String, String)> {
    validate_agent_name(agent_name)?;
    read_subvolume_usage(&Env::real(), agent_name).await
}

/// `MeasureAgentDisk`.
pub(crate) async fn exec_measure_agent_disk(agent_name: &str) -> Result<(String, String)> {
    validate_agent_name(agent_name)?;
    measure_agent_disk(agent_name).await
}

/// `SetSubvolumeQuota`.
pub(crate) async fn exec_set_subvolume_quota(
    agent_name: &str,
    limit_bytes: Option<u64>,
) -> Result<(String, String)> {
    validate_agent_name(agent_name)?;
    set_subvolume_quota(&Env::real(), agent_name, limit_bytes).await
}

/// `UpgradeAgentSubvolume`.
pub(crate) async fn exec_upgrade_agent_subvolume(agent_name: &str) -> Result<(String, String)> {
    validate_agent_name(agent_name)?;
    upgrade_agent_subvolume(&Env::real(), agent_name).await
}

/// `SnapshotAgentSubvolume`.
pub(crate) async fn exec_snapshot_agent_subvolume(
    agent_name: &str,
    snapshot_name: &str,
) -> Result<(String, String)> {
    validate_agent_name(agent_name)?;
    validate_snapshot_name(snapshot_name)?;
    snapshot_agent_subvolume(&Env::real(), agent_name, snapshot_name).await
}

/// `DeleteAgentSnapshot`.
pub(crate) async fn exec_delete_agent_snapshot(
    agent_name: &str,
    snapshot_name: &str,
) -> Result<(String, String)> {
    validate_agent_name(agent_name)?;
    validate_snapshot_name(snapshot_name)?;
    delete_agent_snapshot(&Env::real(), agent_name, snapshot_name).await
}

/// `SendAgentSnapshotToFile`.
pub(crate) async fn exec_send_agent_snapshot_to_file(
    agent_name: &str,
    snapshot_name: &str,
    parent_snapshot_name: Option<&str>,
    dest_file_name: &str,
) -> Result<(String, String)> {
    validate_agent_name(agent_name)?;
    validate_snapshot_name(snapshot_name)?;
    if let Some(parent) = parent_snapshot_name {
        validate_snapshot_name(parent)?;
    }
    validate_credential_name(dest_file_name)?;
    send_agent_snapshot_to_file(
        &Env::real(),
        agent_name,
        snapshot_name,
        parent_snapshot_name,
        dest_file_name,
    )
    .await
}

/// `SendAgentSnapshotToFd`.
pub(crate) async fn exec_send_agent_snapshot_to_fd(
    agent_name: &str,
    snapshot_name: &str,
    parent_snapshot_name: Option<&str>,
    fd: Option<OwnedFd>,
) -> Result<(String, String)> {
    validate_agent_name(agent_name)?;
    validate_snapshot_name(snapshot_name)?;
    if let Some(parent) = parent_snapshot_name {
        validate_snapshot_name(parent)?;
    }
    let dest = fd.context("no descriptor to stream into")?;
    send_agent_snapshot_to_fd(
        &Env::real(),
        agent_name,
        snapshot_name,
        parent_snapshot_name,
        dest,
    )
    .await
}

/// A btrfs snapshot label must start with `hive-` — this doubles as an
/// allow-list: only names hivectl itself constructs (or an operator who
/// knows the convention) can reach the `btrfs subvolume snapshot`/`delete`
/// shellouts, so an arbitrary caller can't use the snapshot ops to probe or
/// churn unrelated paths under `AGENT_STATE_ROOT`. Beyond the prefix, the
/// same charset restriction as [`validate_credential_name`] applies (it's
/// interpolated straight into a filesystem path).
fn validate_snapshot_name(name: &str) -> Result<()> {
    if !name.starts_with("hive-") {
        bail!("invalid snapshot label {name:?}: must start with \"hive-\"");
    }
    validate_credential_name(name)
}

/// btrfs superblock magic, as reported by `statfs(2)`'s `f_type`.
const BTRFS_SUPER_MAGIC: i64 = 0x9123_683E;

/// Inode number of a btrfs subvolume root (`BTRFS_FIRST_FREE_OBJECTID`).
/// Every subvolume's top directory has this inode; plain directories do
/// not, so `statfs == btrfs && st_ino == 256` reliably identifies a
/// subvolume root.
const BTRFS_SUBVOL_ROOT_INO: u64 = 256;

/// Whether `path` lives on a btrfs filesystem (via `statfs(2)`).
fn is_on_btrfs(path: &Path) -> Result<bool> {
    use std::os::unix::ffi::OsStrExt as _;
    let c_path = std::ffi::CString::new(path.as_os_str().as_bytes())
        .with_context(|| format!("path {} has an interior null byte", path.display()))?;
    // SAFETY: `c_path` is a valid NUL-terminated C string that outlives the
    // call; `statfs` only writes into the zero-initialised `buf`.
    let mut buf: libc::statfs = unsafe { std::mem::zeroed() };
    let rc = unsafe { libc::statfs(c_path.as_ptr(), &raw mut buf) };
    if rc != 0 {
        return Err(std::io::Error::last_os_error())
            .with_context(|| format!("statfs {}", path.display()));
    }
    Ok(buf.f_type == BTRFS_SUPER_MAGIC)
}

/// Whether `path` is the root of a btrfs subvolume (on btrfs and inode 256).
fn is_btrfs_subvolume(path: &Path) -> bool {
    use std::os::unix::fs::MetadataExt as _;
    let on_btrfs = is_on_btrfs(path).unwrap_or(false);
    let ino = std::fs::metadata(path).ok().map(|m| m.ino());
    is_subvolume_root(on_btrfs, ino)
}

/// A subvolume root is a directory on btrfs whose inode is
/// [`BTRFS_SUBVOL_ROOT_INO`]; a missing path (`None`) is none.
fn is_subvolume_root(on_btrfs: bool, ino: Option<u64>) -> bool {
    on_btrfs && ino == Some(BTRFS_SUBVOL_ROOT_INO)
}

/// The two questions the storage ops ask the state filesystem. [`RealProbes`]
/// is production; tests answer without a btrfs filesystem.
trait Probes {
    /// Whether `path` lives on a btrfs filesystem; an error when it cannot be
    /// told.
    fn on_btrfs(&self, path: &Path) -> Result<bool>;
    /// Whether `path` is the root of a btrfs subvolume.
    fn is_subvolume(&self, path: &Path) -> bool;
}

/// Asks the kernel, through [`is_on_btrfs`] and [`is_btrfs_subvolume`].
struct RealProbes;

impl Probes for RealProbes {
    fn on_btrfs(&self, path: &Path) -> Result<bool> {
        is_on_btrfs(path)
    }

    fn is_subvolume(&self, path: &Path) -> bool {
        is_btrfs_subvolume(path)
    }
}

/// What the storage ops act on and run. [`Env::real`] is production; tests
/// point the roots at a scratch tree and swap the program and the probes, so
/// no root, btrfs filesystem or `btrfs` binary is needed.
struct Env<'a, S, P> {
    /// Holds one state root per agent (`AGENT_STATE_ROOT`).
    state_root: &'a Path,
    /// Where snapshot exports are written (`MIGRATE_STAGING_ROOT`).
    staging_root: &'a Path,
    /// Starts `btrfs`.
    spawn: S,
    probes: P,
}

impl Env<'static, RealSpawn, RealProbes> {
    fn real() -> Self {
        Self {
            state_root: Path::new(AGENT_STATE_ROOT),
            staging_root: Path::new(MIGRATE_STAGING_ROOT),
            spawn: RealSpawn,
            probes: RealProbes,
        }
    }
}

impl<S: Spawn, P: Probes> Env<'_, S, P> {
    /// A fresh `btrfs` command; the ops append their own arguments.
    fn btrfs(&self) -> Command {
        self.spawn.command("btrfs")
    }
}

/// `EnsureAgentSubvolume` — make the agent's state root a btrfs subvolume
/// when the FS supports it. Idempotent + progressive: no-op when the root
/// already exists or the FS isn't btrfs. See the wire doc on the variant.
async fn ensure_agent_subvolume(
    env: &Env<'_, impl Spawn, impl Probes>,
    agent_name: &str,
) -> Result<(String, String)> {
    use std::os::unix::fs::MetadataExt as _;

    let root = env.state_root;
    let agent_root = root.join(agent_name);

    // Progressive: existing agents (plain dir OR already a subvol) are left
    // untouched — never auto-migrated.
    if agent_root.exists() {
        return Ok((String::new(), String::new()));
    }

    // Only btrfs supports subvolumes; on anything else hive-c0re's normal
    // `create_dir_all` makes a plain dir (the pre-subvolume behaviour). The
    // parent must exist for both statfs and `btrfs subvolume create`.
    std::fs::create_dir_all(root)
        .with_context(|| format!("create agents root {}", root.display()))?;
    if !env.probes.on_btrfs(root)? {
        return Ok((String::new(), String::new()));
    }

    let out = env
        .btrfs()
        .args(["subvolume", "create"])
        .arg(&agent_root)
        .output()
        .await
        .with_context(|| format!("spawn btrfs subvolume create {}", agent_root.display()))?;
    if !out.status.success() {
        bail!(
            "btrfs subvolume create {} failed: {}",
            agent_root.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }

    // The subvol root is created root-owned, but hive-c0re (the `hive-core`
    // user) must be able to mkdir state/ claude/ harness/ inside it — exactly
    // as it would in a plain dir. Chown it to AGENT_STATE_ROOT's owner
    // (hive-core). This MUST succeed: a root-owned subvol would make the
    // downstream dir creation fail with a confusing permission error, and the
    // c0re-side exists-check would then skip re-running this op on retry,
    // wedging the agent. So on any failure roll the subvol back and bail — the
    // create path surfaces a clear error and a retry starts clean.
    let chown_result = std::fs::metadata(root)
        .with_context(|| format!("stat agents root {} for ownership", root.display()))
        .and_then(|meta| {
            std::os::unix::fs::chown(&agent_root, Some(meta.uid()), Some(meta.gid())).with_context(
                || format!("chown subvol {} to agents-root owner", agent_root.display()),
            )
        });
    if let Err(e) = chown_result {
        // Best-effort rollback so we never leave a root-owned subvol behind.
        let _ = env
            .btrfs()
            .args(["subvolume", "delete"])
            .arg(&agent_root)
            .output()
            .await;
        return Err(e.context(format!(
            "rolled back subvolume {} after chown failed",
            agent_root.display()
        )));
    }
    tracing::info!(agent = %agent_name, path = %agent_root.display(), "created agent state subvolume");
    Ok((String::new(), String::new()))
}

/// `DeleteAgentSubvolume` — delete the agent's state root iff it's a btrfs
/// subvolume (purge path only). No-op for plain dirs / missing paths;
/// hive-c0re's own `remove_dir_all` covers those. See the wire doc.
async fn delete_agent_subvolume(
    env: &Env<'_, impl Spawn, impl Probes>,
    agent_name: &str,
) -> Result<(String, String)> {
    let agent_root = env.state_root.join(agent_name);
    if !env.probes.is_subvolume(&agent_root) {
        return Ok((String::new(), String::new()));
    }
    let out = env
        .btrfs()
        .args(["subvolume", "delete"])
        .arg(&agent_root)
        .output()
        .await
        .with_context(|| format!("spawn btrfs subvolume delete {}", agent_root.display()))?;
    if !out.status.success() {
        bail!(
            "btrfs subvolume delete {} failed: {}",
            agent_root.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    tracing::info!(agent = %agent_name, path = %agent_root.display(), "deleted agent state subvolume");
    Ok((String::new(), String::new()))
}

/// `EnsureBtrfsQuota` — enable btrfs qgroup accounting on the filesystem
/// holding `AGENT_STATE_ROOT`. Idempotent + statfs-gated (no-op off btrfs).
/// Operator opt-in only; see the wire doc.
pub(crate) async fn ensure_btrfs_quota() -> Result<(String, String)> {
    ensure_btrfs_quota_in(&Env::real()).await
}

async fn ensure_btrfs_quota_in(env: &Env<'_, impl Spawn, impl Probes>) -> Result<(String, String)> {
    let root = env.state_root;
    std::fs::create_dir_all(root)
        .with_context(|| format!("create agents root {}", root.display()))?;
    if !env.probes.on_btrfs(root)? {
        // Non-btrfs host: quota/qgroups don't apply. No-op success so the
        // operator-facing verb degrades cleanly.
        return Ok((
            String::new(),
            "filesystem is not btrfs — quota not applicable".to_owned(),
        ));
    }
    let out = env
        .btrfs()
        .args(["quota", "enable"])
        .arg(root)
        .output()
        .await
        .with_context(|| format!("spawn btrfs quota enable {}", root.display()))?;
    if !out.status.success() {
        bail!(
            "btrfs quota enable {} failed: {}",
            root.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    tracing::info!(path = %root.display(), "enabled btrfs qgroup accounting");
    Ok((String::new(), String::new()))
}

/// `ReadSubvolumeUsage` — return an agent subvolume's qgroup rows
/// (`btrfs qgroup show -f --raw <…/agent_name>`) verbatim in stdout for
/// hive-c0re to parse. `-f` lists the qgroups impacting the given path,
/// excluding ancestral qgroups (per btrfs-qgroup-show(8)) — so it scopes
/// to this subvolume and never mixes in other agents'. hive-c0re then
/// selects the level-0 (`0/<subvolid>`) leaf row. See the wire doc.
async fn read_subvolume_usage(
    env: &Env<'_, impl Spawn, impl Probes>,
    agent_name: &str,
) -> Result<(String, String)> {
    let agent_root = env.state_root.join(agent_name);
    let out = env
        .btrfs()
        .args(["qgroup", "show", "-f", "--raw"])
        .arg(&agent_root)
        .output()
        .await
        .with_context(|| format!("spawn btrfs qgroup show {}", agent_root.display()))?;
    if !out.status.success() {
        // The common failure is "quota not enabled" — pass the stderr
        // through so hive-c0re can surface it gracefully.
        bail!(
            "btrfs qgroup show {} failed: {}",
            agent_root.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    Ok((
        String::from_utf8_lossy(&out.stdout).into_owned(),
        String::new(),
    ))
}

/// The host paths `MeasureAgentDisk` sums for `agent_name`: its `state/`
/// and `work/` dirs under `AGENT_STATE_ROOT` and its container rootfs.
/// These are the only paths that op ever walks.
// `state` / `work` must match hive-c0re's `agent_notes_dir` and
// `paths::WORK_SUBDIR`.
fn agent_disk_paths(agent_name: &str) -> [PathBuf; 3] {
    let agent_root = PathBuf::from(AGENT_STATE_ROOT).join(agent_name);
    [
        agent_root.join("state"),
        agent_root.join("work"),
        PathBuf::from("/var/lib/nixos-containers").join(container_system_name(agent_name)),
    ]
}

/// `MeasureAgentDisk` — one `du` over the [`agent_disk_paths`] that exist,
/// returning the grand total in bytes. An agent with none of them has no
/// measurable size, which is an error rather than a total of 0.
async fn measure_agent_disk(agent_name: &str) -> Result<(String, String)> {
    let paths = existing_paths(agent_disk_paths(agent_name))?;
    let total = du_total(agent_name, du_command(&paths)?, MEASURE_AGENT_DISK_TIMEOUT).await?;
    Ok((total.to_string(), String::new()))
}

/// The members of `candidates` that exist, in order. A dangling link counts
/// (it is looked at, not followed); only a missing path is skipped.
fn existing_paths(candidates: impl IntoIterator<Item = PathBuf>) -> Result<Vec<PathBuf>> {
    let mut paths = Vec::new();
    for path in candidates {
        match std::fs::symlink_metadata(&path) {
            Ok(_) => paths.push(path),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => return Err(e).with_context(|| format!("stat {}", path.display())),
        }
    }
    Ok(paths)
}

/// `du -s -x -b -c -P -- <paths>`. `-x` keeps each walk on its operand's
/// filesystem (so the bind-mounted nix store is never counted), `-P` never
/// follows a symlink, `-b` counts apparent size, `-c` adds the `total` line
/// [`parse_du_total`] reads. Errors on no paths: `du` with no operand would
/// measure its working directory instead.
fn du_command(paths: &[PathBuf]) -> Result<Command> {
    if paths.is_empty() {
        bail!("none of the agent's state, work or rootfs paths exist");
    }
    let mut cmd = Command::new("du");
    cmd.args(["-s", "-x", "-b", "-c", "-P", "--"]).args(paths);
    Ok(cmd)
}

/// Run `cmd` (a [`du_command`]) for at most `limit` and parse its total.
/// The child is killed on timeout. A timeout, a spawn failure, a non-zero
/// exit or output without a total is an error, never a partial total.
async fn du_total(agent_name: &str, mut cmd: Command, limit: std::time::Duration) -> Result<u64> {
    cmd.kill_on_drop(true);
    let Ok(out) = tokio::time::timeout(limit, cmd.output()).await else {
        bail!(
            "du for {agent_name} timed out after {}s; killed it",
            limit.as_secs()
        );
    };
    let out = out.with_context(|| format!("spawn du for {agent_name}"))?;
    if !out.status.success() {
        bail!(
            "du for {agent_name} failed ({}): {}",
            out.status,
            String::from_utf8_lossy(&out.stderr)
                .lines()
                .last()
                .unwrap_or("")
                .trim()
        );
    }
    let stdout = String::from_utf8_lossy(&out.stdout);
    parse_du_total(&stdout)
        .ok_or_else(|| anyhow!("no `total` line in du output for {agent_name}: {stdout:?}"))
}

/// The byte count on `du -c`'s final `<bytes>\ttotal` line. `None` when the
/// last line isn't that shape, so truncated or unexpected output is never
/// read as a smaller total.
fn parse_du_total(stdout: &str) -> Option<u64> {
    let last = stdout.lines().rev().find(|l| !l.trim().is_empty())?;
    let (bytes, label) = last.split_once('\t')?;
    if label != "total" {
        return None;
    }
    bytes.parse().ok()
}

/// Best-effort removal of a leftover migration path from a prior aborted
/// upgrade: try `btrfs subvolume delete` (in case it's a half-created
/// subvolume) then a plain recursive remove. Both failures are ignored —
/// the path may simply not exist.
async fn cleanup_stale_path(env: &Env<'_, impl Spawn, impl Probes>, path: &Path) {
    if path.exists() {
        let _ = env
            .btrfs()
            .args(["subvolume", "delete"])
            .arg(path)
            .output()
            .await;
        let _ = std::fs::remove_dir_all(path);
    }
}

/// Stage a populated subvolume at `tmp` mirroring `agent_root`: create the
/// subvolume, copy `agent_root`'s contents into it preserving
/// ownership/permissions/xattrs, then match the subvolume root's owner + mode
/// to the original. On any failure the partially-staged `tmp` is cleaned up
/// (so the caller can bail with the original dir still untouched).
async fn stage_upgrade_subvolume(
    env: &Env<'_, impl Spawn, impl Probes>,
    agent_root: &Path,
    tmp: &Path,
) -> Result<()> {
    use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};

    // Fresh subvolume to receive the copy.
    let out = env
        .btrfs()
        .args(["subvolume", "create"])
        .arg(tmp)
        .output()
        .await
        .with_context(|| format!("spawn btrfs subvolume create {}", tmp.display()))?;
    if !out.status.success() {
        bail!(
            "btrfs subvolume create {} failed: {}",
            tmp.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }

    // Copy contents preserving everything (`-a` = --preserve=all → mode,
    // ownership, timestamps, links, xattrs); reflink for fast CoW clones on the
    // same btrfs. `<src>/.` copies the directory's contents (incl. dotfiles)
    // into the subvolume rather than nesting it.
    let copy = Command::new("cp")
        .arg("-a")
        .arg("--reflink=auto")
        .arg(format!("{}/.", agent_root.display()))
        .arg(tmp)
        .output()
        .await
        .with_context(|| format!("spawn cp into {}", tmp.display()))?;
    if !copy.status.success() {
        cleanup_stale_path(env, tmp).await;
        bail!(
            "copy {} -> {} failed (original left untouched): {}",
            agent_root.display(),
            tmp.display(),
            String::from_utf8_lossy(&copy.stderr).trim()
        );
    }

    // Match the new subvolume root's ownership + mode to the original dir.
    // `cp -a <src>/.` copies the *contents* but the subvolume root keeps its
    // create-time root ownership, so set it explicitly — the swapped-in
    // subvolume must be indistinguishable from the original to hive-c0re.
    let apply = std::fs::metadata(agent_root)
        .with_context(|| format!("stat {} for ownership", agent_root.display()))
        .and_then(|m| {
            std::os::unix::fs::chown(tmp, Some(m.uid()), Some(m.gid()))
                .with_context(|| format!("chown {} to match original", tmp.display()))?;
            std::fs::set_permissions(tmp, std::fs::Permissions::from_mode(m.mode()))
                .with_context(|| format!("chmod {} to match original", tmp.display()))?;
            Ok(())
        });
    if let Err(e) = apply {
        cleanup_stale_path(env, tmp).await;
        return Err(e.context("upgrade aborted before swap; original left untouched"));
    }
    Ok(())
}

/// `UpgradeAgentSubvolume` — convert an existing plain-dir agent state root
/// into a btrfs subvolume in place. Operator opt-in; the caller (hivectl)
/// stops the agent first and restarts it after. See the wire doc.
///
/// Migration: stage a sibling subvolume mirroring the dir
/// ([`stage_upgrade_subvolume`]), then rename the original aside and the
/// subvolume into place, then remove the original. Any failure before the
/// rename-swap leaves the original dir untouched.
async fn upgrade_agent_subvolume(
    env: &Env<'_, impl Spawn, impl Probes>,
    agent_name: &str,
) -> Result<(String, String)> {
    let root = env.state_root;
    let agent_root = root.join(agent_name);

    if !agent_root.exists() {
        // A crash between the two swap renames (original → `.<name>.old`
        // succeeded, `.<name>.migrating` → agent_root did not) leaves the
        // agent root missing but the original data intact under `.<name>.old`.
        // Point at the recovery rather than a bare "nothing to do" so the
        // operator isn't left guessing where the data went.
        let old = root.join(format!(".{agent_name}.old"));
        if old.exists() {
            bail!(
                "no state dir at {agent} — but {old} holds the original data from an \
                 interrupted upgrade (host crashed mid-swap). Restore it with \
                 `mv {old} {agent}`, then re-run the upgrade.",
                agent = agent_root.display(),
                old = old.display(),
            );
        }
        bail!(
            "no state dir to upgrade at {} — nothing to do",
            agent_root.display()
        );
    }
    // Idempotent: already a subvolume → nothing to do.
    if env.probes.is_subvolume(&agent_root) {
        return Ok((String::new(), String::new()));
    }
    if !env.probes.on_btrfs(root)? {
        bail!(
            "{} is not on btrfs — subvolumes are unsupported, cannot upgrade",
            root.display()
        );
    }

    // Sibling temp paths on the same filesystem (so the copy can reflink and
    // the swap renames are atomic). Leading dots keep them out of the agent
    // namespace (`validate_agent_name` rejects dot-prefixed names).
    let tmp = root.join(format!(".{agent_name}.migrating"));
    let old = root.join(format!(".{agent_name}.old"));
    // Clear any debris from a previously interrupted run before starting.
    cleanup_stale_path(env, &tmp).await;
    cleanup_stale_path(env, &old).await;

    stage_upgrade_subvolume(env, &agent_root, &tmp).await?;

    // Swap. `rename` is atomic within a filesystem. The window between the
    //    two renames is the only unsafe point: a crash there leaves the agent
    //    root missing but both `.old` (original) and the new subvolume present
    //    — recoverable by hand, hence the loud logging.
    if let Err(e) = std::fs::rename(&agent_root, &old) {
        cleanup_stale_path(env, &tmp).await;
        return Err(anyhow::Error::new(e).context(format!(
            "rename {} -> {} failed; original left untouched",
            agent_root.display(),
            old.display()
        )));
    }
    if let Err(e) = std::fs::rename(&tmp, &agent_root) {
        // Restore the original from its renamed-aside copy.
        let restored = std::fs::rename(&old, &agent_root).is_ok();
        cleanup_stale_path(env, &tmp).await;
        return Err(anyhow::Error::new(e).context(format!(
            "rename {} -> {} failed; original {}",
            tmp.display(),
            agent_root.display(),
            if restored {
                "restored"
            } else {
                "COULD NOT BE RESTORED — manual recovery needed"
            }
        )));
    }

    // 5. Success: drop the original (a plain dir) and report.
    if let Err(e) = std::fs::remove_dir_all(&old) {
        // The migration succeeded; a leftover `.old` is cosmetic. Warn only.
        tracing::warn!(
            agent = %agent_name, path = %old.display(),
            "upgraded subvolume but failed to remove old dir: {e}"
        );
    }
    tracing::info!(
        agent = %agent_name, path = %agent_root.display(),
        "upgraded agent state dir to btrfs subvolume"
    );
    Ok((
        format!("upgraded {} to a btrfs subvolume", agent_root.display()),
        String::new(),
    ))
}

/// Derive a snapshot's path from the agent name + label: a dot-prefixed
/// sibling of the agent's state root so it can never collide with a real
/// agent directory (`validate_agent_name` rejects dot-prefixed names).
fn snapshot_path(root: &Path, agent_name: &str, snapshot_name: &str) -> PathBuf {
    root.join(format!(".{agent_name}.snapshot.{snapshot_name}"))
}

/// `SnapshotAgentSubvolume` — create a read-only btrfs snapshot of an
/// agent's state subvolume, for `btrfs send` to stream from during
/// inter-hive migration. See the wire doc.
async fn snapshot_agent_subvolume(
    env: &Env<'_, impl Spawn, impl Probes>,
    agent_name: &str,
    snapshot_name: &str,
) -> Result<(String, String)> {
    let agent_root = env.state_root.join(agent_name);
    if !env.probes.is_subvolume(&agent_root) {
        bail!(
            "{} is not a btrfs subvolume — nothing to snapshot (run `hivectl agent <name> subvol upgrade` first)",
            agent_root.display()
        );
    }
    let snap = snapshot_path(env.state_root, agent_name, snapshot_name);
    if snap.exists() {
        bail!(
            "snapshot {} already exists — delete it first or pick a different name",
            snap.display()
        );
    }
    let out = env
        .btrfs()
        .args(["subvolume", "snapshot", "-r"])
        .arg(&agent_root)
        .arg(&snap)
        .output()
        .await
        .with_context(|| {
            format!(
                "spawn btrfs subvolume snapshot -r {} {}",
                agent_root.display(),
                snap.display()
            )
        })?;
    if !out.status.success() {
        bail!(
            "btrfs subvolume snapshot -r {} {} failed: {}",
            agent_root.display(),
            snap.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    tracing::info!(
        agent = %agent_name, snapshot = %snap.display(),
        "created read-only agent state snapshot"
    );
    Ok((snap.display().to_string(), String::new()))
}

/// `DeleteAgentSnapshot` — delete a previously-created read-only snapshot.
/// No-op if the path doesn't exist. See the wire doc.
async fn delete_agent_snapshot(
    env: &Env<'_, impl Spawn, impl Probes>,
    agent_name: &str,
    snapshot_name: &str,
) -> Result<(String, String)> {
    let snap = snapshot_path(env.state_root, agent_name, snapshot_name);
    if !snap.exists() {
        return Ok((String::new(), String::new()));
    }
    let out = env
        .btrfs()
        .args(["subvolume", "delete"])
        .arg(&snap)
        .output()
        .await
        .with_context(|| format!("spawn btrfs subvolume delete {}", snap.display()))?;
    if !out.status.success() {
        bail!(
            "btrfs subvolume delete {} failed: {}",
            snap.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    tracing::info!(agent = %agent_name, snapshot = %snap.display(), "deleted agent state snapshot");
    Ok((String::new(), String::new()))
}

/// Both paths a `SendAgentSnapshotToFile` request reads from — the
/// snapshot itself, and its optional incremental parent — must exist
/// before we touch `dest`. Pulled out on its own so that invariant is
/// checkable without a `dest` path, a `MIGRATE_STAGING_ROOT`, or `btrfs`
/// at all: a caller cannot create the output file by accident while
/// still holding a reference to this check.
fn check_send_snapshot_preconditions(snap: &Path, parent_path: Option<&Path>) -> Result<()> {
    if !snap.exists() {
        bail!(
            "snapshot {} does not exist — create it with `subvol snapshot create` first",
            snap.display()
        );
    }
    if let Some(parent_path) = parent_path
        && !parent_path.exists()
    {
        bail!(
            "parent snapshot {} does not exist — pick an existing parent or omit it for a full send",
            parent_path.display()
        );
    }
    Ok(())
}

/// Validate preconditions, then atomically create `dest` for a snapshot
/// export. The two are combined in one function on purpose: keeping
/// `check_send_snapshot_preconditions` ahead of the `create_new` call is
/// the entire fix, and folding them together here means that ordering
/// can't drift back apart in a later edit to `send_agent_snapshot_to_file`.
/// No `btrfs` involved, so this is unit-testable against plain files.
fn open_export_dest(snap: &Path, parent_path: Option<&Path>, dest: &Path) -> Result<std::fs::File> {
    check_send_snapshot_preconditions(snap, parent_path)?;
    // `create_new` (O_CREAT|O_EXCL) makes the no-overwrite guarantee atomic
    // instead of a check-then-create race against a concurrent request.
    match std::fs::File::options()
        .write(true)
        .create_new(true)
        .open(dest)
    {
        Ok(f) => Ok(f),
        Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => bail!(
            "{} already exists — pick a different destination or remove it first \
             (send never overwrites an existing export)",
            dest.display()
        ),
        Err(e) => Err(e).with_context(|| format!("create {}", dest.display())),
    }
}

/// Deletes `path` when dropped, unless [`disarm`](Self::disarm)ed first.
/// Guards `dest` between the moment `send_agent_snapshot_to_file` creates
/// it and the moment the export fully succeeds, so any `?`/`bail!` on
/// that path — including ones a future edit adds — removes the partial
/// file instead of leaving a zero-byte export that then makes every
/// retry against the same `--dest` fail with "already exists".
struct PartialExportGuard<'a> {
    path: &'a Path,
    armed: bool,
}

impl<'a> PartialExportGuard<'a> {
    fn new(path: &'a Path) -> Self {
        Self { path, armed: true }
    }

    fn disarm(mut self) {
        self.armed = false;
    }
}

impl Drop for PartialExportGuard<'_> {
    fn drop(&mut self) {
        if !self.armed {
            return;
        }
        // Best-effort: warn (don't fail the whole call over it) if removal
        // itself fails, so a stuck partial file that later masquerades as a
        // completed export is at least visible in the log.
        if let Err(rm_err) = std::fs::remove_file(self.path) {
            tracing::warn!(
                dest = %self.path.display(), error = %rm_err,
                "failed to remove partial export after failure — \
                 next attempt at this dest will hit the already-exists guard"
            );
        }
    }
}

/// `SendAgentSnapshotToFile` — stream a read-only snapshot (optionally
/// incremental against `parent_name`) to a file under
/// `MIGRATE_STAGING_ROOT` via `btrfs send`. Local-file half of the
/// inter-hive migration transport; see `PrivRequest::SendAgentSnapshotToFile`
/// for the cross-hive follow-up.
async fn send_agent_snapshot_to_file(
    env: &Env<'_, impl Spawn, impl Probes>,
    agent_name: &str,
    snapshot_name: &str,
    parent_name: Option<&str>,
    dest_file_name: &str,
) -> Result<(String, String)> {
    let snap = snapshot_path(env.state_root, agent_name, snapshot_name);
    let parent_path = parent_name.map(|parent| snapshot_path(env.state_root, agent_name, parent));

    let staging = env.staging_root;
    std::fs::create_dir_all(staging).with_context(|| format!("create {}", staging.display()))?;
    let dest = staging.join(dest_file_name);
    // Preconditions are checked *inside* this call, before `dest` is
    // created — see `open_export_dest`'s doc comment for why that ordering
    // matters and is kept in one place.
    let dest_file = open_export_dest(&snap, parent_path.as_deref(), &dest)?;
    let cleanup = PartialExportGuard::new(&dest);

    let mut cmd = env.btrfs();
    cmd.arg("send");
    if let Some(parent_path) = &parent_path {
        cmd.arg("-p").arg(parent_path);
    }
    cmd.arg(&snap);
    cmd.stdout(std::process::Stdio::from(dest_file));
    cmd.stderr(std::process::Stdio::piped());

    let out = cmd
        .spawn()
        .with_context(|| format!("spawn btrfs send {}", snap.display()))?
        .wait_with_output()
        .await
        .with_context(|| format!("wait on btrfs send {}", snap.display()))?;
    if !out.status.success() {
        bail!(
            "btrfs send {} failed: {}",
            snap.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    cleanup.disarm();
    tracing::info!(
        agent = %agent_name, snapshot = %snap.display(), dest = %dest.display(),
        parent = ?parent_name, "exported agent snapshot to file"
    );
    Ok((dest.display().to_string(), String::new()))
}

/// `SendAgentSnapshotToFd` — stream a read-only snapshot (optionally
/// incremental against `parent_name`) straight into a descriptor the
/// caller passed us.
///
/// The network half of the inter-hive migration transport, arranged so
/// this helper never learns there *is* a network: hive-c0re connects to
/// the peer's snapshot store, writes the header itself, and hands the
/// connected socket over. We only ever see "a thing to write bytes into",
/// which keeps a root process out of any address, protocol or trust
/// decision — and keeps everyone out of the data path once `btrfs send`
/// starts, which matters at multi-gigabyte sizes.
async fn send_agent_snapshot_to_fd(
    env: &Env<'_, impl Spawn, impl Probes>,
    agent_name: &str,
    snapshot_name: &str,
    parent_name: Option<&str>,
    dest: OwnedFd,
) -> Result<(String, String)> {
    let snap = snapshot_path(env.state_root, agent_name, snapshot_name);
    if !snap.exists() {
        bail!(
            "snapshot {} does not exist — create it with `subvol snapshot create` first",
            snap.display()
        );
    }

    let mut cmd = env.btrfs();
    cmd.arg("send");
    if let Some(parent) = parent_name {
        let parent_path = snapshot_path(env.state_root, agent_name, parent);
        if !parent_path.exists() {
            bail!(
                "parent snapshot {} does not exist — pick an existing parent or omit it for a full send",
                parent_path.display()
            );
        }
        cmd.arg("-p").arg(&parent_path);
    }
    cmd.arg(&snap);
    cmd.stdout(std::process::Stdio::from(dest));
    cmd.stderr(std::process::Stdio::piped());

    let out = cmd
        .spawn()
        .with_context(|| format!("spawn btrfs send {}", snap.display()))?
        .wait_with_output()
        .await
        .with_context(|| format!("wait on btrfs send {}", snap.display()))?;
    if !out.status.success() {
        // Nothing to clean up: the destination isn't ours. A partial
        // stream is the receiving end's problem, and `btrfs receive`
        // refuses to commit an incomplete subvolume anyway.
        bail!(
            "btrfs send {} failed: {}",
            snap.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    tracing::info!(
        agent = %agent_name, snapshot = %snap.display(), parent = ?parent_name,
        "streamed agent snapshot into a passed descriptor"
    );
    Ok((String::new(), String::new()))
}

/// `SetSubvolumeQuota` — set or clear a qgroup size limit on an agent
/// subvolume (`btrfs qgroup limit <bytes|none> <…/agent_name>`). See the
/// wire doc.
async fn set_subvolume_quota(
    env: &Env<'_, impl Spawn, impl Probes>,
    agent_name: &str,
    limit_bytes: Option<u64>,
) -> Result<(String, String)> {
    let agent_root = env.state_root.join(agent_name);
    let limit = limit_bytes.map_or_else(|| "none".to_owned(), |n| n.to_string());
    let out = env
        .btrfs()
        .args(["qgroup", "limit", &limit])
        .arg(&agent_root)
        .output()
        .await
        .with_context(|| format!("spawn btrfs qgroup limit {}", agent_root.display()))?;
    if !out.status.success() {
        bail!(
            "btrfs qgroup limit {limit} {} failed: {}",
            agent_root.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    tracing::info!(agent = %agent_name, %limit, "set agent subvolume quota");
    Ok((String::new(), String::new()))
}
#[cfg(test)]
mod tests {
    use super::{
        Command, Env, Probes, agent_disk_paths, cleanup_stale_path, delete_agent_snapshot,
        delete_agent_subvolume, du_command, du_total, ensure_agent_subvolume,
        ensure_btrfs_quota_in, exec_delete_agent_snapshot, exec_delete_agent_subvolume,
        exec_ensure_agent_subvolume, exec_measure_agent_disk, exec_read_subvolume_usage,
        exec_send_agent_snapshot_to_fd, exec_send_agent_snapshot_to_file, exec_set_subvolume_quota,
        exec_snapshot_agent_subvolume, exec_upgrade_agent_subvolume, existing_paths,
        is_btrfs_subvolume, is_on_btrfs, is_subvolume_root, open_export_dest, parse_du_total,
        read_subvolume_usage, send_agent_snapshot_to_fd, send_agent_snapshot_to_file,
        set_subvolume_quota, snapshot_agent_subvolume, snapshot_path, upgrade_agent_subvolume,
        validate_snapshot_name,
    };
    use crate::shared::Spawn;
    use crate::shared::test_support::{MissingTool, ScriptSpawn, scratch};
    use crate::shared::validate_agent_name;
    use std::path::{Path, PathBuf};
    use std::sync::atomic::{AtomicU32, Ordering};
    use std::time::Duration;

    /// A snapshot label must additionally carry the `hive-` prefix (it
    /// doubles as the allow-list gating the btrfs snapshot/delete
    /// shellouts) on top of [`validate_credential_name`]'s charset rule.
    #[test]
    fn validate_snapshot_name_requires_hive_prefix() {
        assert!(validate_snapshot_name("not-hive-prefixed").is_err());
        assert!(validate_snapshot_name("hive-valid").is_ok());
    }

    /// Unique scratch dir per test, no external tempfile dep. Same pattern
    /// as `shared::test_support::scratch()`, kept separate so a rename of one doesn't
    /// collide with the other's directory names under parallel test runs.
    fn snapshot_export_scratch() -> PathBuf {
        static CTR: AtomicU32 = AtomicU32::new(0);
        let n = CTR.fetch_add(1, Ordering::Relaxed);
        let dir = std::env::temp_dir().join(format!(
            "hive-priv-snapshot-export-test-{}-{n}",
            std::process::id()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        dir
    }

    /// The fix: a missing parent must be caught before `dest` is created,
    /// not after. Fails against the pre-fix ordering (`dest` created via
    /// `create_new` first, parent checked second) — see the issue for the
    /// inverted run.
    #[test]
    fn missing_parent_leaves_no_file_at_dest() {
        let dir = snapshot_export_scratch();
        let snap = dir.join("snap");
        std::fs::write(&snap, b"").unwrap();
        let parent = dir.join("parent-that-does-not-exist");
        let dest = dir.join("dest");

        let err = open_export_dest(&snap, Some(&parent), &dest).unwrap_err();
        assert!(
            err.to_string().contains("does not exist"),
            "wrong error: {err}"
        );
        assert!(
            !dest.exists(),
            "a failed export must not leave a file at dest"
        );
    }

    /// A retry against the same `dest` after a precondition failure must
    /// not trip the no-overwrite guard — there is nothing there to
    /// overwrite, since the failed attempt never created `dest` (the test
    /// above). This is the behavior a "stale parent falls back to a full
    /// send, retried without -p" caller needs against a fixed `--dest`.
    #[test]
    fn retry_after_precondition_failure_succeeds() {
        let dir = snapshot_export_scratch();
        let snap = dir.join("snap");
        std::fs::write(&snap, b"").unwrap();
        let parent = dir.join("parent-that-does-not-exist");
        let dest = dir.join("dest");

        assert!(open_export_dest(&snap, Some(&parent), &dest).is_err());
        assert!(!dest.exists());

        // Retry without the missing parent, same dest.
        let file = open_export_dest(&snap, None, &dest);
        assert!(
            file.is_ok(),
            "retry against a dest no previous attempt touched must succeed: {:?}",
            file.err()
        );
        assert!(dest.exists());
    }

    /// Control for the two tests above: an export that actually landed at
    /// `dest` must still refuse a second one. Without this, a bug that
    /// simply stopped creating `dest` at all (rather than fixing the
    /// ordering) would also make the retry test above pass.
    #[test]
    fn existing_dest_still_refuses_overwrite() {
        let dir = snapshot_export_scratch();
        let snap = dir.join("snap");
        std::fs::write(&snap, b"").unwrap();
        let dest = dir.join("dest");
        std::fs::write(&dest, b"already here").unwrap();

        let err = open_export_dest(&snap, None, &dest).unwrap_err();
        assert!(
            err.to_string().contains("already exists"),
            "wrong error: {err}"
        );
        assert_eq!(
            std::fs::read(&dest).unwrap(),
            b"already here",
            "must not have touched the existing file"
        );
    }

    #[test]
    fn disk_paths_are_derived_from_the_name_only() {
        assert_eq!(
            agent_disk_paths("iris"),
            [
                PathBuf::from("/var/lib/hyperhive/agents/iris/state"),
                PathBuf::from("/var/lib/hyperhive/agents/iris/work"),
                PathBuf::from("/var/lib/nixos-containers/h-iris"),
            ]
        );
    }

    /// Every name that could steer the walk out of the agent's own paths is
    /// refused before any path is built, let alone `du` spawned.
    #[tokio::test]
    async fn measure_rejects_names_that_are_not_agent_names() {
        for bad in [
            "", "..", "../etc", "a/b", "/etc", "Iris", "a b", "a\0b", "foo_bar",
        ] {
            let err = exec_measure_agent_disk(bad)
                .await
                .expect_err(&format!("{bad:?} must be rejected"));
            assert!(
                err.to_string().contains("invalid name"),
                "{bad:?}: wrong error: {err}"
            );
        }
    }

    #[test]
    fn du_total_reads_the_total_line() {
        let out = "100\t/var/lib/hyperhive/agents/iris/state\n\
                   2000\t/var/lib/nixos-containers/h-iris\n\
                   2100\ttotal\n";
        assert_eq!(parse_du_total(out), Some(2100));
        assert_eq!(parse_du_total("0\ttotal"), Some(0));
    }

    #[test]
    fn du_total_rejects_output_without_a_total_line() {
        assert_eq!(parse_du_total(""), None);
        assert_eq!(parse_du_total("\n"), None);
        // A per-path line is not the total, even when it is the last one.
        assert_eq!(parse_du_total("100\t/var/lib/x\n"), None);
        assert_eq!(parse_du_total("abc\ttotal\n"), None);
        assert_eq!(parse_du_total("-1\ttotal\n"), None);
        assert_eq!(parse_du_total("100 total\n"), None);
    }

    #[test]
    fn no_existing_paths_is_an_error_not_zero() {
        let err = du_command(&[]).expect_err("no operands must not run du");
        assert!(err.to_string().contains("none of"), "{err}");
    }

    #[tokio::test]
    async fn non_zero_du_exit_is_an_error() {
        let err = du_total("iris", Command::new("false"), Duration::from_secs(5))
            .await
            .expect_err("a failed du must not yield a total");
        assert!(err.to_string().contains("failed"), "{err}");
    }

    #[tokio::test]
    async fn a_walk_past_the_limit_is_killed_and_errors() {
        let mut cmd = Command::new("sleep");
        cmd.arg("30");
        let started = std::time::Instant::now();
        let err = du_total("iris", cmd, Duration::from_millis(50))
            .await
            .expect_err("a walk past the limit must fail");
        assert!(err.to_string().contains("timed out"), "{err}");
        assert!(started.elapsed() < Duration::from_secs(10));
    }

    #[tokio::test]
    async fn du_total_sums_a_real_walk() {
        let dir = std::env::temp_dir().join(format!("hive-priv-du-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        std::fs::write(dir.join("f"), b"hello").unwrap();
        let cmd = du_command(std::slice::from_ref(&dir)).unwrap();
        let total = du_total("iris", cmd, Duration::from_secs(30)).await;
        std::fs::remove_dir_all(&dir).unwrap();
        assert!(total.unwrap() >= 5);
    }

    // --- name and path decisions -------------------------------------------

    #[test]
    fn snapshot_labels_are_a_strict_allow_list() {
        for ok in ["hive-x", "hive-", "hive-a_b-1", "hive-A9"] {
            validate_snapshot_name(ok).unwrap_or_else(|e| panic!("{ok:?} must pass: {e}"));
        }
        for bad in [
            "",
            "hive",
            "xhive-a",
            "Hive-a",
            "hive-a.b",
            "hive-..",
            "hive-../x",
            "hive-a/b",
            "hive-a b",
            "hive-a\0b",
            "hive-é",
        ] {
            assert!(
                validate_snapshot_name(bad).is_err(),
                "{bad:?} must be rejected"
            );
        }
    }

    /// A snapshot is a dot-prefixed sibling of the agent dirs, so no agent
    /// name can be the same entry, and it never leaves the state root.
    #[test]
    fn snapshot_path_is_a_dot_sibling_no_agent_can_own() {
        let root = Path::new("/state");
        let p = snapshot_path(root, "iris", "hive-x");
        assert_eq!(p, PathBuf::from("/state/.iris.snapshot.hive-x"));
        assert_eq!(p.parent(), Some(root));
        let leaf = p.file_name().unwrap().to_str().unwrap();
        assert!(
            validate_agent_name(leaf).is_err(),
            "{leaf:?} would be a valid agent dir"
        );
        assert_ne!(p, snapshot_path(root, "other", "hive-x"));
        assert_ne!(p, snapshot_path(root, "iris", "hive-y"));
    }

    #[test]
    fn a_subvolume_root_needs_btrfs_and_inode_256() {
        assert!(is_subvolume_root(true, Some(256)));
        assert!(!is_subvolume_root(false, Some(256)));
        assert!(!is_subvolume_root(true, Some(255)));
        assert!(!is_subvolume_root(true, Some(257)));
        assert!(!is_subvolume_root(true, None));
    }

    #[test]
    fn btrfs_probes_on_real_paths() {
        assert!(!is_on_btrfs(Path::new("/proc")).unwrap());
        let err = is_on_btrfs(Path::new("/nonexistent-hive-priv-test")).unwrap_err();
        assert!(err.to_string().contains("statfs"), "{err}");
        let err = is_on_btrfs(Path::new("/pro\0c")).unwrap_err();
        assert!(err.to_string().contains("interior null"), "{err}");
        assert!(!is_btrfs_subvolume(Path::new("/proc")));
        assert!(!is_btrfs_subvolume(Path::new(
            "/nonexistent-hive-priv-test"
        )));
    }

    /// Every op refuses an agent name that could leave its directory before
    /// any path is built. The control: a valid name gets past validation and
    /// stops at the next check instead.
    #[tokio::test]
    async fn every_op_validates_the_agent_name_first() {
        for bad in ["", "..", "../etc", "a/b", "/etc", "Iris", "a.b", "a\0b"] {
            let want = |r: anyhow::Result<(String, String)>, op: &str| {
                let err = r.expect_err(&format!("{op}({bad:?}) must be rejected"));
                assert!(
                    err.to_string().contains("invalid name"),
                    "{op}({bad:?}): {err}"
                );
            };
            want(exec_ensure_agent_subvolume(bad).await, "ensure");
            want(exec_delete_agent_subvolume(bad).await, "delete");
            want(exec_read_subvolume_usage(bad).await, "usage");
            want(exec_set_subvolume_quota(bad, Some(1)).await, "quota");
            want(exec_upgrade_agent_subvolume(bad).await, "upgrade");
            want(
                exec_snapshot_agent_subvolume(bad, "hive-x").await,
                "snapshot",
            );
            want(
                exec_delete_agent_snapshot(bad, "hive-x").await,
                "delete-snapshot",
            );
            want(
                exec_send_agent_snapshot_to_file(bad, "hive-x", None, "out").await,
                "send-file",
            );
            want(
                exec_send_agent_snapshot_to_fd(bad, "hive-x", None, None).await,
                "send-fd",
            );
        }
        let err = exec_send_agent_snapshot_to_fd("iris", "hive-x", None, None)
            .await
            .unwrap_err();
        assert!(err.to_string().contains("no descriptor"), "{err}");
    }

    #[tokio::test]
    async fn every_snapshot_op_validates_the_labels_and_destination() {
        for bad in ["iris", "hive", "hive-../x", "hive-a/b", "hive-a.b", ""] {
            let want = |r: anyhow::Result<(String, String)>, op: &str| {
                let err = r.expect_err(&format!("{op}({bad:?}) must be rejected"));
                let msg = err.to_string();
                assert!(
                    msg.contains("invalid snapshot label")
                        || msg.contains("invalid credential name"),
                    "{op}({bad:?}): {msg}"
                );
            };
            want(exec_snapshot_agent_subvolume("iris", bad).await, "snapshot");
            want(
                exec_delete_agent_snapshot("iris", bad).await,
                "delete-snapshot",
            );
            want(
                exec_send_agent_snapshot_to_file("iris", bad, None, "out").await,
                "send-file snapshot",
            );
            want(
                exec_send_agent_snapshot_to_file("iris", "hive-x", Some(bad), "out").await,
                "send-file parent",
            );
            want(
                exec_send_agent_snapshot_to_fd("iris", bad, None, None).await,
                "send-fd snapshot",
            );
            want(
                exec_send_agent_snapshot_to_fd("iris", "hive-x", Some(bad), None).await,
                "send-fd parent",
            );
        }
        for bad in ["", ".", "..", "../x", "a/b", "a.b", "/etc/passwd"] {
            let err = exec_send_agent_snapshot_to_file("iris", "hive-x", None, bad)
                .await
                .expect_err(&format!("dest {bad:?} must be rejected"));
            assert!(
                err.to_string().contains("invalid credential name"),
                "{bad:?}: {err}"
            );
        }
    }

    #[test]
    fn measured_paths_skip_only_the_missing_ones() {
        let dir = scratch();
        let file = dir.join("file");
        std::fs::write(&file, b"x").unwrap();
        let dangling = dir.join("dangling");
        std::os::unix::fs::symlink(dir.join("gone"), &dangling).unwrap();
        let missing = dir.join("missing");
        let got = existing_paths([missing.clone(), dangling.clone(), file.clone(), dir.clone()]);
        assert_eq!(got.unwrap(), [dangling, file, dir]);
        assert!(existing_paths([missing]).unwrap().is_empty());
    }

    // --- the ops against a scratch tree and a fake btrfs -------------------

    /// Scratch state root, staging root and the log the fake btrfs writes.
    struct Fx {
        state: PathBuf,
        staging: PathBuf,
        log: PathBuf,
    }

    impl Fx {
        fn new() -> Self {
            let top = scratch().canonicalize().unwrap();
            let state = top.join("agents");
            std::fs::create_dir_all(&state).unwrap();
            Self {
                state,
                staging: top.join("staging"),
                log: top.join("btrfs.log"),
            }
        }

        fn agent(&self, name: &str) -> PathBuf {
            self.state.join(name)
        }

        fn snap(&self, agent: &str, label: &str) -> PathBuf {
            self.state.join(format!(".{agent}.snapshot.{label}"))
        }

        /// The fake btrfs's invocations, one argument string each.
        fn calls(&self) -> Vec<String> {
            std::fs::read_to_string(&self.log)
                .unwrap_or_default()
                .lines()
                .map(str::to_owned)
                .collect()
        }

        /// An [`Env`] over this tree, running `btrfs` and the two probes given.
        fn env<'a, S: Spawn>(
            &'a self,
            spawn: &'a S,
            on_btrfs: fn(&Path) -> anyhow::Result<bool>,
            is_subvolume: fn(&Path) -> bool,
        ) -> Env<'a, &'a S, FnProbes> {
            Env {
                state_root: &self.state,
                staging_root: &self.staging,
                spawn,
                probes: FnProbes {
                    on_btrfs,
                    is_subvolume,
                },
            }
        }

        /// A `btrfs` that logs its arguments, then runs the shell fragment
        /// `body` (`$1…` are btrfs's arguments).
        fn btrfs(&self, body: &str) -> ScriptSpawn {
            ScriptSpawn(format!(
                "FAKE_LOG={log}\necho \"$*\" >> \"$FAKE_LOG\"\n{body}",
                log = self.log.display()
            ))
        }
    }

    /// Probes answered by the functions given.
    struct FnProbes {
        on_btrfs: fn(&Path) -> anyhow::Result<bool>,
        is_subvolume: fn(&Path) -> bool,
    }

    impl Probes for FnProbes {
        fn on_btrfs(&self, path: &Path) -> anyhow::Result<bool> {
            (self.on_btrfs)(path)
        }

        fn is_subvolume(&self, path: &Path) -> bool {
            (self.is_subvolume)(path)
        }
    }

    /// Creates, deletes and snapshots as btrfs does; `send` writes `STREAM`.
    const LIKE_BTRFS: &str = r#"case "$1 $2" in
  "subvolume create") mkdir "$3";;
  "subvolume delete") rm -rf "$3";;
  "subvolume snapshot") cp -a "$4" "$5";;
  send*) printf STREAM;;
esac"#;
    const FAIL: &str = r#"echo "boom: no space left" >&2; exit 1"#;
    const NOTHING: &str = ":";

    /// Probes that fail on a path they cannot stat, as `statfs` does.
    fn on_btrfs(path: &Path) -> anyhow::Result<bool> {
        path.try_exists()?;
        Ok(true)
    }
    fn off_btrfs(path: &Path) -> anyhow::Result<bool> {
        path.try_exists()?;
        Ok(false)
    }
    fn probe_fails(_: &Path) -> anyhow::Result<bool> {
        anyhow::bail!("statfs refused")
    }
    fn is_subvol(_: &Path) -> bool {
        true
    }
    fn not_subvol(_: &Path) -> bool {
        false
    }

    #[tokio::test]
    async fn ensure_leaves_an_existing_agent_root_alone() {
        let fx = Fx::new();
        std::fs::create_dir(fx.agent("iris")).unwrap();
        let btrfs = fx.btrfs(LIKE_BTRFS);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        assert_eq!(
            ensure_agent_subvolume(&env, "iris").await.unwrap(),
            (String::new(), String::new())
        );
        assert!(fx.calls().is_empty(), "{:?}", fx.calls());
    }

    #[tokio::test]
    async fn ensure_makes_no_subvolume_off_btrfs_but_creates_the_root() {
        let fx = Fx::new();
        std::fs::remove_dir(&fx.state).unwrap();
        let btrfs = fx.btrfs(LIKE_BTRFS);
        let env = fx.env(&btrfs, off_btrfs, not_subvol);
        ensure_agent_subvolume(&env, "iris").await.unwrap();
        assert!(fx.calls().is_empty());
        assert!(fx.state.is_dir(), "the root is created for the probe");
        assert!(!fx.agent("iris").exists());
    }

    #[tokio::test]
    async fn ensure_creates_the_subvolume_owned_like_the_root() {
        use std::os::unix::fs::MetadataExt as _;
        let fx = Fx::new();
        let btrfs = fx.btrfs(LIKE_BTRFS);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        ensure_agent_subvolume(&env, "iris").await.unwrap();
        assert_eq!(
            fx.calls(),
            [format!("subvolume create {}", fx.agent("iris").display())]
        );
        let (root, agent) = (
            std::fs::metadata(&fx.state).unwrap(),
            std::fs::metadata(fx.agent("iris")).unwrap(),
        );
        assert_eq!((agent.uid(), agent.gid()), (root.uid(), root.gid()));
    }

    /// The test user can chown only to its own groups, so the group of the
    /// root is the one id that can differ from what `mkdir` gives.
    #[tokio::test]
    async fn ensure_hands_the_subvolume_to_the_roots_group() {
        use std::os::unix::fs::MetadataExt as _;
        let mut groups = [0 as libc::gid_t; 64];
        // SAFETY: `groups` has room for the 64 ids it declares.
        let n = unsafe { libc::getgroups(64, groups.as_mut_ptr()) };
        let primary = std::fs::metadata("/proc/self").unwrap().gid();
        let Some(other) = groups[..usize::try_from(n).unwrap_or(0)]
            .iter()
            .copied()
            .find(|g| *g != primary)
        else {
            return;
        };
        let fx = Fx::new();
        std::os::unix::fs::chown(&fx.state, None, Some(other)).unwrap();
        let btrfs = fx.btrfs(LIKE_BTRFS);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        ensure_agent_subvolume(&env, "iris").await.unwrap();
        assert_eq!(std::fs::metadata(fx.agent("iris")).unwrap().gid(), other);
    }

    #[tokio::test]
    async fn ensure_rolls_the_subvolume_back_when_the_chown_fails() {
        let fx = Fx::new();
        // The fake creates nothing, so the chown has no target.
        let btrfs = fx.btrfs(NOTHING);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        let err = ensure_agent_subvolume(&env, "iris").await.unwrap_err();
        assert!(err.to_string().contains("rolled back"), "{err:#}");
        let agent = fx.agent("iris");
        assert_eq!(
            fx.calls(),
            [
                format!("subvolume create {}", agent.display()),
                format!("subvolume delete {}", agent.display()),
            ]
        );
    }

    #[tokio::test]
    async fn ensure_reports_a_failed_create_without_rolling_back() {
        let fx = Fx::new();
        let btrfs = fx.btrfs(FAIL);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        let err = ensure_agent_subvolume(&env, "iris").await.unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("btrfs subvolume create") && msg.contains("boom"),
            "{msg}"
        );
        assert_eq!(fx.calls().len(), 1, "no rollback of what never existed");
    }

    #[tokio::test]
    async fn ensure_surfaces_a_missing_btrfs_binary_and_a_failed_probe() {
        let fx = Fx::new();
        let missing = MissingTool;
        let env = fx.env(&missing, on_btrfs, not_subvol);
        let err = ensure_agent_subvolume(&env, "iris").await.unwrap_err();
        assert!(
            err.to_string().contains("spawn btrfs subvolume create"),
            "{err:#}"
        );

        let btrfs = fx.btrfs(LIKE_BTRFS);
        let env = fx.env(&btrfs, probe_fails, not_subvol);
        let err = ensure_agent_subvolume(&env, "iris").await.unwrap_err();
        assert!(err.to_string().contains("statfs refused"), "{err:#}");
        assert!(fx.calls().is_empty());
    }

    #[tokio::test]
    async fn delete_removes_only_a_real_subvolume() {
        let fx = Fx::new();
        std::fs::create_dir(fx.agent("iris")).unwrap();
        let btrfs = fx.btrfs(LIKE_BTRFS);

        // A plain dir (or a missing one) is hive-c0re's to remove.
        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        delete_agent_subvolume(&env, "iris").await.unwrap();
        delete_agent_subvolume(&env, "nobody").await.unwrap();
        assert!(fx.calls().is_empty());
        assert!(fx.agent("iris").is_dir());

        let env = fx.env(&btrfs, on_btrfs, is_subvol);
        delete_agent_subvolume(&env, "iris").await.unwrap();
        assert_eq!(
            fx.calls(),
            [format!("subvolume delete {}", fx.agent("iris").display())]
        );
        assert!(!fx.agent("iris").exists());
    }

    #[tokio::test]
    async fn delete_reports_a_btrfs_failure() {
        let fx = Fx::new();
        let btrfs = fx.btrfs(FAIL);
        let env = fx.env(&btrfs, on_btrfs, is_subvol);
        let err = delete_agent_subvolume(&env, "iris").await.unwrap_err();
        assert!(err.to_string().contains("boom"), "{err}");
    }

    #[tokio::test]
    async fn quota_enable_is_a_noop_off_btrfs() {
        let fx = Fx::new();
        std::fs::remove_dir(&fx.state).unwrap();
        let btrfs = fx.btrfs(NOTHING);
        let env = fx.env(&btrfs, off_btrfs, not_subvol);
        let (out, note) = ensure_btrfs_quota_in(&env).await.unwrap();
        assert!(out.is_empty() && note.contains("not btrfs"), "{note}");
        assert!(fx.calls().is_empty());
        assert!(fx.state.is_dir());
    }

    #[tokio::test]
    async fn quota_enable_targets_the_state_root() {
        let fx = Fx::new();
        let btrfs = fx.btrfs(NOTHING);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        ensure_btrfs_quota_in(&env).await.unwrap();
        assert_eq!(fx.calls(), [format!("quota enable {}", fx.state.display())]);

        let failing = fx.btrfs(FAIL);
        let env = fx.env(&failing, on_btrfs, not_subvol);
        let err = ensure_btrfs_quota_in(&env).await.unwrap_err();
        assert!(err.to_string().contains("boom"), "{err}");
        let env = fx.env(&btrfs, probe_fails, not_subvol);
        assert!(ensure_btrfs_quota_in(&env).await.is_err());
    }

    #[tokio::test]
    async fn usage_is_scoped_to_the_agent_and_returned_verbatim() {
        let fx = Fx::new();
        let btrfs = fx.btrfs(r"printf '0/257 4096 4096\n'");
        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        let (out, err) = read_subvolume_usage(&env, "iris").await.unwrap();
        assert_eq!((out.as_str(), err.as_str()), ("0/257 4096 4096\n", ""));
        assert_eq!(
            fx.calls(),
            [format!(
                "qgroup show -f --raw {}",
                fx.agent("iris").display()
            )]
        );

        let failing = fx.btrfs(FAIL);
        let env = fx.env(&failing, on_btrfs, not_subvol);
        let err = read_subvolume_usage(&env, "iris").await.unwrap_err();
        assert!(err.to_string().contains("boom"), "{err}");
    }

    #[tokio::test]
    async fn quota_limit_is_the_byte_count_or_none() {
        let fx = Fx::new();
        let btrfs = fx.btrfs(NOTHING);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        set_subvolume_quota(&env, "iris", Some(1_048_576))
            .await
            .unwrap();
        set_subvolume_quota(&env, "iris", None).await.unwrap();
        let a = fx.agent("iris");
        assert_eq!(
            fx.calls(),
            [
                format!("qgroup limit 1048576 {}", a.display()),
                format!("qgroup limit none {}", a.display()),
            ]
        );

        let failing = fx.btrfs(FAIL);
        let env = fx.env(&failing, on_btrfs, not_subvol);
        let err = set_subvolume_quota(&env, "iris", Some(7))
            .await
            .unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("limit 7") && msg.contains("boom"), "{msg}");
    }

    #[tokio::test]
    async fn snapshot_needs_a_subvolume_and_an_unused_label() {
        let fx = Fx::new();
        std::fs::create_dir(fx.agent("iris")).unwrap();
        let btrfs = fx.btrfs(LIKE_BTRFS);

        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        let err = snapshot_agent_subvolume(&env, "iris", "hive-a")
            .await
            .unwrap_err();
        assert!(err.to_string().contains("not a btrfs subvolume"), "{err}");

        let env = fx.env(&btrfs, on_btrfs, is_subvol);
        let (path, _) = snapshot_agent_subvolume(&env, "iris", "hive-a")
            .await
            .unwrap();
        let snap = fx.snap("iris", "hive-a");
        assert_eq!(path, snap.display().to_string());
        assert_eq!(
            fx.calls(),
            [format!(
                "subvolume snapshot -r {} {}",
                fx.agent("iris").display(),
                snap.display()
            )]
        );

        let err = snapshot_agent_subvolume(&env, "iris", "hive-a")
            .await
            .unwrap_err();
        assert!(err.to_string().contains("already exists"), "{err}");
        assert_eq!(fx.calls().len(), 1, "an existing snapshot is never re-run");
    }

    #[tokio::test]
    async fn snapshot_reports_a_btrfs_failure() {
        let fx = Fx::new();
        let btrfs = fx.btrfs(FAIL);
        let env = fx.env(&btrfs, on_btrfs, is_subvol);
        let err = snapshot_agent_subvolume(&env, "iris", "hive-a")
            .await
            .unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("snapshot -r") && msg.contains("boom"), "{msg}");
    }

    #[tokio::test]
    async fn deleting_a_snapshot_touches_only_that_snapshot() {
        let fx = Fx::new();
        std::fs::create_dir(fx.agent("iris")).unwrap();
        std::fs::create_dir(fx.snap("iris", "hive-a")).unwrap();
        std::fs::create_dir(fx.snap("iris", "hive-b")).unwrap();
        let btrfs = fx.btrfs(LIKE_BTRFS);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);

        delete_agent_snapshot(&env, "iris", "hive-gone")
            .await
            .unwrap();
        assert!(fx.calls().is_empty(), "a missing snapshot is a no-op");

        delete_agent_snapshot(&env, "iris", "hive-a").await.unwrap();
        assert_eq!(
            fx.calls(),
            [format!(
                "subvolume delete {}",
                fx.snap("iris", "hive-a").display()
            )]
        );
        assert!(!fx.snap("iris", "hive-a").exists());
        assert!(fx.snap("iris", "hive-b").exists());
        assert!(fx.agent("iris").exists());

        let failing = fx.btrfs(FAIL);
        let env = fx.env(&failing, on_btrfs, not_subvol);
        let err = delete_agent_snapshot(&env, "iris", "hive-b")
            .await
            .unwrap_err();
        assert!(err.to_string().contains("boom"), "{err}");
    }

    #[tokio::test]
    async fn stale_paths_are_cleared_only_when_present() {
        let fx = Fx::new();
        let btrfs = fx.btrfs(LIKE_BTRFS);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        let stale = fx.state.join(".iris.migrating");
        cleanup_stale_path(&env, &stale).await;
        assert!(fx.calls().is_empty());

        std::fs::create_dir_all(stale.join("debris")).unwrap();
        cleanup_stale_path(&env, &stale).await;
        assert_eq!(
            fx.calls(),
            [format!("subvolume delete {}", stale.display())]
        );
        assert!(!stale.exists());

        // The plain remove covers what `btrfs subvolume delete` refuses.
        std::fs::create_dir_all(stale.join("debris")).unwrap();
        let refusing = fx.btrfs(FAIL);
        let env = fx.env(&refusing, on_btrfs, not_subvol);
        cleanup_stale_path(&env, &stale).await;
        assert!(!stale.exists());
    }

    // --- upgrade -------------------------------------------------------------

    fn plain_agent(fx: &Fx) {
        use std::os::unix::fs::PermissionsExt as _;
        let a = fx.agent("iris");
        std::fs::create_dir_all(a.join("state")).unwrap();
        std::fs::write(a.join("state/notes"), b"keep me").unwrap();
        std::fs::set_permissions(&a, std::fs::Permissions::from_mode(0o750)).unwrap();
    }

    #[tokio::test]
    async fn upgrade_swaps_in_a_copy_with_the_same_data_and_mode() {
        use std::os::unix::fs::MetadataExt as _;
        let fx = Fx::new();
        plain_agent(&fx);
        let btrfs = fx.btrfs(LIKE_BTRFS);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        let (out, _) = upgrade_agent_subvolume(&env, "iris").await.unwrap();
        assert!(out.contains("upgraded"), "{out}");

        let a = fx.agent("iris");
        assert_eq!(std::fs::read(a.join("state/notes")).unwrap(), b"keep me");
        let meta = std::fs::metadata(&a).unwrap();
        assert_eq!(meta.mode() & 0o777, 0o750);
        assert!(!fx.state.join(".iris.migrating").exists());
        assert!(!fx.state.join(".iris.old").exists());
        assert_eq!(
            fx.calls(),
            [format!(
                "subvolume create {}",
                fx.state.join(".iris.migrating").display()
            )]
        );
    }

    #[tokio::test]
    async fn upgrade_clears_debris_from_an_interrupted_run_first() {
        let fx = Fx::new();
        plain_agent(&fx);
        for leftover in [".iris.migrating", ".iris.old"] {
            std::fs::create_dir_all(fx.state.join(leftover).join("junk")).unwrap();
        }
        let btrfs = fx.btrfs(LIKE_BTRFS);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        upgrade_agent_subvolume(&env, "iris").await.unwrap();
        assert!(!fx.agent("iris").join("junk").exists());
        assert!(!fx.state.join(".iris.old").exists());
        assert_eq!(fx.calls().len(), 3, "{:?}", fx.calls());
    }

    #[tokio::test]
    async fn upgrade_is_a_noop_on_a_subvolume_and_refuses_off_btrfs() {
        let fx = Fx::new();
        plain_agent(&fx);
        let btrfs = fx.btrfs(LIKE_BTRFS);

        let env = fx.env(&btrfs, on_btrfs, is_subvol);
        let (out, _) = upgrade_agent_subvolume(&env, "iris").await.unwrap();
        assert!(out.is_empty());

        let env = fx.env(&btrfs, off_btrfs, not_subvol);
        let err = upgrade_agent_subvolume(&env, "iris").await.unwrap_err();
        assert!(err.to_string().contains("not on btrfs"), "{err}");

        assert!(fx.calls().is_empty());
        assert!(fx.agent("iris").join("state/notes").exists());
    }

    #[tokio::test]
    async fn upgrade_leaves_the_original_when_the_subvolume_cannot_be_made() {
        let fx = Fx::new();
        plain_agent(&fx);
        let btrfs = fx.btrfs(FAIL);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        let err = upgrade_agent_subvolume(&env, "iris").await.unwrap_err();
        assert!(err.to_string().contains("boom"), "{err}");
        assert_eq!(
            std::fs::read(fx.agent("iris").join("state/notes")).unwrap(),
            b"keep me"
        );
        assert!(!fx.state.join(".iris.old").exists());
    }

    #[tokio::test]
    async fn upgrade_points_at_the_data_after_a_crash_mid_swap() {
        let fx = Fx::new();
        std::fs::create_dir_all(fx.state.join(".iris.old")).unwrap();
        std::fs::write(fx.state.join(".iris.old/data"), b"x").unwrap();
        let btrfs = fx.btrfs(LIKE_BTRFS);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);

        let err = upgrade_agent_subvolume(&env, "iris").await.unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("mv ") && msg.contains(".iris.old"), "{msg}");
        assert!(
            fx.state.join(".iris.old/data").exists(),
            "recovery data kept"
        );
        assert!(fx.calls().is_empty());

        let err = upgrade_agent_subvolume(&env, "nobody").await.unwrap_err();
        assert!(err.to_string().contains("nothing to do"), "{err}");
    }

    // --- export ---------------------------------------------------------------

    fn snapshots(fx: &Fx, labels: &[&str]) {
        for l in labels {
            std::fs::create_dir_all(fx.snap("iris", l)).unwrap();
        }
    }

    #[tokio::test]
    async fn export_to_file_streams_into_the_staging_root() {
        let fx = Fx::new();
        snapshots(&fx, &["hive-a", "hive-p"]);
        let btrfs = fx.btrfs(LIKE_BTRFS);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);

        let (dest, _) = send_agent_snapshot_to_file(&env, "iris", "hive-a", Some("hive-p"), "out")
            .await
            .unwrap();
        assert_eq!(Path::new(&dest), fx.staging.join("out"));
        assert_eq!(std::fs::read(&dest).unwrap(), b"STREAM");

        send_agent_snapshot_to_file(&env, "iris", "hive-a", None, "full")
            .await
            .unwrap();
        let (a, p) = (fx.snap("iris", "hive-a"), fx.snap("iris", "hive-p"));
        assert_eq!(
            fx.calls(),
            [
                format!("send -p {} {}", p.display(), a.display()),
                format!("send {}", a.display()),
            ]
        );
    }

    #[tokio::test]
    async fn export_to_file_never_overwrites_and_never_leaves_a_partial() {
        let fx = Fx::new();
        snapshots(&fx, &["hive-a"]);
        std::fs::create_dir_all(&fx.staging).unwrap();
        std::fs::write(fx.staging.join("taken"), b"earlier export").unwrap();
        let btrfs = fx.btrfs(LIKE_BTRFS);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);

        let err = send_agent_snapshot_to_file(&env, "iris", "hive-a", None, "taken")
            .await
            .unwrap_err();
        assert!(err.to_string().contains("already exists"), "{err}");
        assert_eq!(
            std::fs::read(fx.staging.join("taken")).unwrap(),
            b"earlier export"
        );
        assert!(
            fx.calls().is_empty(),
            "btrfs never ran against a taken dest"
        );

        let failing = fx.btrfs(FAIL);
        let env = fx.env(&failing, on_btrfs, not_subvol);
        let err = send_agent_snapshot_to_file(&env, "iris", "hive-a", None, "half")
            .await
            .unwrap_err();
        assert!(err.to_string().contains("boom"), "{err}");
        assert!(!fx.staging.join("half").exists(), "partial export removed");

        let err = send_agent_snapshot_to_file(&env, "iris", "hive-missing", None, "none")
            .await
            .unwrap_err();
        assert!(err.to_string().contains("does not exist"), "{err}");
        assert!(!fx.staging.join("none").exists());
    }

    fn fd_for(path: &Path) -> std::os::fd::OwnedFd {
        std::os::fd::OwnedFd::from(std::fs::File::create(path).unwrap())
    }

    #[tokio::test]
    async fn export_to_fd_streams_into_the_descriptor() {
        let fx = Fx::new();
        snapshots(&fx, &["hive-a", "hive-p"]);
        let btrfs = fx.btrfs(LIKE_BTRFS);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);
        let sink = fx.state.join("sink");

        let (out, err) =
            send_agent_snapshot_to_fd(&env, "iris", "hive-a", Some("hive-p"), fd_for(&sink))
                .await
                .unwrap();
        assert!(out.is_empty() && err.is_empty());
        assert_eq!(std::fs::read(&sink).unwrap(), b"STREAM");
        send_agent_snapshot_to_fd(&env, "iris", "hive-a", None, fd_for(&sink))
            .await
            .unwrap();
        let (a, p) = (fx.snap("iris", "hive-a"), fx.snap("iris", "hive-p"));
        assert_eq!(
            fx.calls(),
            [
                format!("send -p {} {}", p.display(), a.display()),
                format!("send {}", a.display()),
            ]
        );
    }

    #[tokio::test]
    async fn export_to_fd_needs_both_snapshots_and_reports_a_failed_send() {
        let fx = Fx::new();
        snapshots(&fx, &["hive-a"]);
        let sink = fx.state.join("sink");
        let btrfs = fx.btrfs(LIKE_BTRFS);
        let env = fx.env(&btrfs, on_btrfs, not_subvol);

        let err = send_agent_snapshot_to_fd(&env, "iris", "hive-no", None, fd_for(&sink))
            .await
            .unwrap_err();
        assert!(err.to_string().contains("does not exist"), "{err}");
        let err = send_agent_snapshot_to_fd(&env, "iris", "hive-a", Some("hive-no"), fd_for(&sink))
            .await
            .unwrap_err();
        assert!(err.to_string().contains("parent snapshot"), "{err}");
        assert!(fx.calls().is_empty());

        let failing = fx.btrfs(FAIL);
        let env = fx.env(&failing, on_btrfs, not_subvol);
        let err = send_agent_snapshot_to_fd(&env, "iris", "hive-a", None, fd_for(&sink))
            .await
            .unwrap_err();
        assert!(err.to_string().contains("boom"), "{err}");
    }
}