From 6d04be6c625edae32ef5eefdce501467e15546fd Mon Sep 17 00:00:00 2001
From: atlas <atlas@hyperhive.local>
Date: Tue, 3 Nov 2026 08:11:16 +0100
Subject: [PATCH] hive-priv: test seams are traits, not closure bags

---
 hive-priv/src/container.rs |  82 +++++++---------
 hive-priv/src/shared.rs    |  50 +++++++++-
 hive-priv/src/storage.rs   | 194 +++++++++++++++++++++++++------------
 3 files changed, 213 insertions(+), 113 deletions(-)

diff --git a/hive-priv/src/container.rs b/hive-priv/src/container.rs
index 40ee26e3..5fc7f1a1 100644
--- a/hive-priv/src/container.rs
+++ b/hive-priv/src/container.rs
@@ -12,18 +12,13 @@ use tokio::io::{AsyncWriteExt, BufReader};
 use tokio::net::unix::OwnedWriteHalf;
 use tokio::process::Command;
 
-use crate::shared::{container_system_name, validate_agent_name};
-
-/// The program `program` names, ready for arguments. The one place the real
-/// tools are resolved, so a test can substitute stand-ins.
-fn real_command(program: &str) -> Command {
-    Command::new(program)
-}
+use crate::shared::{RealSpawn, Spawn, container_system_name, validate_agent_name};
 
 /// What the lifecycle ops run and how long they wait for it: the host's
 /// tools and timings in [`Host::REAL`], stand-ins in tests.
-struct Host<'a> {
-    spawn: &'a (dyn Fn(&str) -> Command + Sync),
+struct Host<S> {
+    /// Starts the tools the ops run.
+    spawn: S,
     /// How long to wait for machined to drop a machine's registration after a
     /// shutdown has been asked for. Generous: a container with slow-stopping
     /// units legitimately takes a while, and escalating early would SIGKILL a
@@ -38,18 +33,18 @@ struct Host<'a> {
     probe_timeout: std::time::Duration,
 }
 
-impl Host<'static> {
+impl Host<RealSpawn> {
     const REAL: Self = Host {
-        spawn: &real_command,
+        spawn: RealSpawn,
         release_timeout: std::time::Duration::from_secs(20),
         release_poll: std::time::Duration::from_millis(500),
         probe_timeout: std::time::Duration::from_secs(5),
     };
 }
 
-impl Host<'_> {
+impl<S: Spawn> Host<S> {
     fn command(&self, program: &str) -> Command {
-        (self.spawn)(program)
+        self.spawn.command(program)
     }
 }
 
@@ -76,7 +71,7 @@ pub(crate) async fn start_container(name: &str) -> Result<(String, String)> {
     start_container_with(&Host::REAL, name).await
 }
 
-async fn start_container_with(host: &Host<'_>, name: &str) -> Result<(String, String)> {
+async fn start_container_with(host: &Host<impl Spawn>, name: &str) -> Result<(String, String)> {
     validate_agent_name(name)?;
     let machine = container_system_name(name);
     // Silenced: a first-time create means the unit doesn't exist yet, and
@@ -158,7 +153,7 @@ pub(crate) async fn start_failure_log(name: &str, since_unix: u64) -> Result<(St
 
 /// `journalctl` with `argv`: its stdout, if it exits 0 within `limit`.
 async fn run_journalctl(
-    host: &Host<'_>,
+    host: &Host<impl Spawn>,
     limit: std::time::Duration,
     argv: Vec<String>,
 ) -> Result<String> {
@@ -182,7 +177,7 @@ pub(crate) async fn stop_container(name: &str) -> Result<(String, String)> {
     stop_container_with(&Host::REAL, name).await
 }
 
-async fn stop_container_with(host: &Host<'_>, name: &str) -> Result<(String, String)> {
+async fn stop_container_with(host: &Host<impl Spawn>, name: &str) -> Result<(String, String)> {
     validate_agent_name(name)?;
     stop_and_release(host, &container_system_name(name)).await
 }
@@ -195,7 +190,7 @@ pub(crate) async fn kill_container(name: &str) -> Result<(String, String)> {
     kill_container_with(&Host::REAL, name).await
 }
 
-async fn kill_container_with(host: &Host<'_>, name: &str) -> Result<(String, String)> {
+async fn kill_container_with(host: &Host<impl Spawn>, name: &str) -> Result<(String, String)> {
     validate_agent_name(name)?;
     let machine = container_system_name(name);
     machinectl_run(host, &["kill", &machine, "--signal=SIGKILL"]).await
@@ -206,7 +201,7 @@ pub(crate) async fn destroy_container(name: &str) -> Result<(String, String)> {
     destroy_container_with(&Host::REAL, name).await
 }
 
-async fn destroy_container_with(host: &Host<'_>, name: &str) -> Result<(String, String)> {
+async fn destroy_container_with(host: &Host<impl Spawn>, name: &str) -> Result<(String, String)> {
     validate_agent_name(name)?;
     container_run_with(host, &["destroy", &container_system_name(name)]).await
 }
@@ -240,7 +235,7 @@ pub(crate) async fn container_toplevel_action(
 /// [`container_toplevel_action`] past the check that `toplevel` is the
 /// closure realised for `name`.
 async fn apply_toplevel(
-    host: &Host<'_>,
+    host: &Host<impl Spawn>,
     verb: &str,
     name: &str,
     toplevel: &str,
@@ -272,7 +267,7 @@ async fn apply_toplevel(
 /// wedged `systemctl reload`) is exactly when the caller most wants the
 /// live line in the build log, not just a summary error afterward.
 async fn swap_container_profile(
-    host: &Host<'_>,
+    host: &Host<impl Spawn>,
     system_name: &str,
     toplevel: &str,
     stream: bool,
@@ -448,7 +443,7 @@ pub(crate) async fn container_run(args: &[&str]) -> Result<(String, String)> {
     container_run_with(&Host::REAL, args).await
 }
 
-async fn container_run_with(host: &Host<'_>, args: &[&str]) -> Result<(String, String)> {
+async fn container_run_with(host: &Host<impl Spawn>, args: &[&str]) -> Result<(String, String)> {
     let out = host
         .command("nixos-container")
         .args(args)
@@ -484,7 +479,7 @@ async fn container_run_with(host: &Host<'_>, args: &[&str]) -> Result<(String, S
 /// Invoke `machinectl` with the given args, log output to journald.
 /// Used for operations that nixos-container doesn't expose (e.g. sending
 /// signals to running containers).
-async fn machinectl_run(host: &Host<'_>, args: &[&str]) -> Result<(String, String)> {
+async fn machinectl_run(host: &Host<impl Spawn>, args: &[&str]) -> Result<(String, String)> {
     let out = host
         .command("machinectl")
         .args(args)
@@ -525,7 +520,7 @@ async fn machinectl_run(host: &Host<'_>, args: &[&str]) -> Result<(String, Strin
 /// Errors and timeouts answer "still registered". Being wrong that way costs
 /// a redundant SIGKILL to something already gone; being wrong the other way
 /// hands back a stop that silently leaked the name.
-async fn machine_registered(host: &Host<'_>, machine: &str) -> bool {
+async fn machine_registered(host: &Host<impl Spawn>, machine: &str) -> bool {
     let probe = host
         .command("machinectl")
         .args(["show", machine, "--property=Name"])
@@ -547,7 +542,7 @@ async fn machine_registered(host: &Host<'_>, machine: &str) -> bool {
 
 /// Poll [`machine_registered`] until the name is free or the timeout expires.
 /// Returns true once the registration is gone.
-async fn wait_name_released(host: &Host<'_>, machine: &str) -> bool {
+async fn wait_name_released(host: &Host<impl Spawn>, machine: &str) -> bool {
     let deadline = tokio::time::Instant::now() + host.release_timeout;
     loop {
         if !machine_registered(host, machine).await {
@@ -589,7 +584,7 @@ async fn wait_name_released(host: &Host<'_>, machine: &str) -> bool {
 /// that every stop gets it: dashboard, reconcile, destroy, cold-start
 /// fallback. The start path already distrusts its own exit code the same way;
 /// this is the missing half of that pair.
-async fn stop_and_release(host: &Host<'_>, machine: &str) -> Result<(String, String)> {
+async fn stop_and_release(host: &Host<impl Spawn>, machine: &str) -> Result<(String, String)> {
     let stop = container_run_with(host, &["stop", machine]).await;
     if wait_name_released(host, machine).await {
         // Happy path, and also the path where a stop that reported failure
@@ -620,7 +615,7 @@ async fn stop_and_release(host: &Host<'_>, machine: &str) -> Result<(String, Str
 /// on success (all output was streamed); the error string includes stderr
 /// tail on failure.
 async fn container_run_streaming(
-    host: &Host<'_>,
+    host: &Host<impl Spawn>,
     args: &[&str],
     writer: &mut OwnedWriteHalf,
 ) -> Result<(String, String)> {
@@ -862,7 +857,7 @@ mod host_tests {
     use tokio::net::UnixStream;
 
     use super::*;
-    use crate::shared::test_support::scratch;
+    use crate::shared::test_support::{MissingTool, ScriptSpawn, scratch};
 
     const TOPLEVEL: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-bob";
 
@@ -872,7 +867,7 @@ mod host_tests {
     /// `$D` is the stub's scratch directory.
     struct Stub {
         dir: PathBuf,
-        spawn: Box<dyn Fn(&str) -> Command + Sync>,
+        script: String,
     }
 
     impl Stub {
@@ -882,21 +877,18 @@ mod host_tests {
                 "D={d}\necho \"$0 $*\" >> \"$D/calls\"\n{behave}",
                 d = dir.display()
             );
-            let spawn = move |program: &str| {
-                let mut cmd = Command::new("sh");
-                cmd.arg("-c").arg(&script).arg(program);
-                cmd
-            };
-            Self {
-                dir,
-                spawn: Box::new(spawn),
-            }
+            Self { dir, script }
         }
 
         /// Timings short enough for a test that waits them out.
-        fn host(&self) -> Host<'_> {
+        fn host(&self) -> Host<ScriptSpawn> {
+            Self::host_over(ScriptSpawn(self.script.clone()))
+        }
+
+        /// [`Stub::host`]'s timings over another `spawn`.
+        fn host_over<S>(spawn: S) -> Host<S> {
             Host {
-                spawn: &*self.spawn,
+                spawn,
                 release_timeout: Duration::from_millis(300),
                 release_poll: Duration::from_millis(20),
                 probe_timeout: Duration::from_secs(10),
@@ -1104,11 +1096,7 @@ mod host_tests {
 
     #[tokio::test]
     async fn a_missing_tool_is_reported_by_name() {
-        let gone = |_: &str| Command::new("/nonexistent/hive-priv-test-tool");
-        let host = Host {
-            spawn: &gone,
-            ..Stub::new("").host()
-        };
+        let host = Stub::host_over(MissingTool);
         let msg = err_text(container_run_with(&host, &["list"]).await);
         assert!(msg.contains("invoke nixos-container"), "{msg}");
         let msg = err_text(machinectl_run(&host, &["list"]).await);
@@ -1216,11 +1204,7 @@ mod host_tests {
         let free = Stub::new("exit 1");
         assert!(!machine_registered(&free.host(), "h-bob").await);
 
-        let gone = |_: &str| Command::new("/nonexistent/hive-priv-test-tool");
-        let host = Host {
-            spawn: &gone,
-            ..free.host()
-        };
+        let host = Stub::host_over(MissingTool);
         assert!(
             machine_registered(&host, "h-bob").await,
             "a probe that cannot run must answer registered"
diff --git a/hive-priv/src/shared.rs b/hive-priv/src/shared.rs
index a8ae7332..bda35d6d 100644
--- a/hive-priv/src/shared.rs
+++ b/hive-priv/src/shared.rs
@@ -7,6 +7,7 @@ use std::path::{Path, PathBuf};
 
 use anyhow::{Context as _, Result, bail};
 use hive_priv_sock::AGENT_PREFIX;
+use tokio::process::Command;
 
 /// Return the system container name for a logical agent name.
 /// All agents (including the manager) use the `h-` prefix.
@@ -210,13 +211,60 @@ impl Drop for StagedFile<'_> {
     }
 }
 
-/// Scratch-directory helpers shared by every module's tests. One counter for
+/// How an op starts a host program. [`RealSpawn`] in production; tests
+/// substitute stand-ins, so no root, container or btrfs filesystem is needed.
+pub(crate) trait Spawn {
+    /// The `program` command, ready for arguments.
+    fn command(&self, program: &str) -> Command;
+}
+
+/// Starts the program `program` names on `PATH`.
+pub(crate) struct RealSpawn;
+
+impl Spawn for RealSpawn {
+    fn command(&self, program: &str) -> Command {
+        Command::new(program)
+    }
+}
+
+/// Scratch directories and stand-in tools shared by every module's tests. One counter for
 /// the whole crate, so no two tests get the same directory.
 #[cfg(test)]
 pub(crate) mod test_support {
     use std::path::PathBuf;
     use std::sync::atomic::{AtomicU32, Ordering};
 
+    use tokio::process::Command;
+
+    use super::Spawn;
+
+    /// Runs every program as `sh -c <script> <program> <args…>`.
+    pub(crate) struct ScriptSpawn(pub(crate) String);
+
+    impl Spawn for ScriptSpawn {
+        fn command(&self, program: &str) -> Command {
+            let mut cmd = Command::new("sh");
+            cmd.arg("-c").arg(&self.0).arg(program);
+            cmd
+        }
+    }
+
+    /// Starts a program that does not exist, whatever is asked for.
+    pub(crate) struct MissingTool;
+
+    impl Spawn for MissingTool {
+        fn command(&self, _: &str) -> Command {
+            Command::new("/nonexistent/hive-priv-test-tool")
+        }
+    }
+
+    /// A borrowed stand-in, so one stub can back several fixtures.
+    impl<T: Spawn> Spawn for &T {
+        fn command(&self, program: &str) -> Command {
+            (**self).command(program)
+        }
+    }
+
     /// Unique scratch dir per test, no external tempfile dep.
     pub(crate) fn scratch() -> PathBuf {
         static CTR: AtomicU32 = AtomicU32::new(0);
diff --git a/hive-priv/src/storage.rs b/hive-priv/src/storage.rs
index cd50f0d7..6a7fbd64 100644
--- a/hive-priv/src/storage.rs
+++ b/hive-priv/src/storage.rs
@@ -15,7 +15,9 @@ 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::{container_system_name, validate_agent_name, validate_credential_name};
+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)> {
@@ -175,40 +177,67 @@ 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 btrfs probes and program,
-/// so no root, btrfs filesystem or `btrfs` binary is needed.
-struct Env<'a> {
+/// 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,
-    /// A fresh `btrfs` command; the ops append their own arguments.
-    btrfs: &'a (dyn Fn() -> Command + Sync),
-    on_btrfs: fn(&Path) -> Result<bool>,
-    is_subvolume: fn(&Path) -> bool,
+    /// Starts `btrfs`.
+    spawn: S,
+    probes: P,
 }
 
-fn real_btrfs() -> Command {
-    Command::new("btrfs")
-}
-
-impl Env<'static> {
+impl Env<'static, RealSpawn, RealProbes> {
     fn real() -> Self {
         Self {
             state_root: Path::new(AGENT_STATE_ROOT),
             staging_root: Path::new(MIGRATE_STAGING_ROOT),
-            btrfs: &real_btrfs,
-            on_btrfs: is_on_btrfs,
-            is_subvolume: is_btrfs_subvolume,
+            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<'_>, agent_name: &str) -> Result<(String, String)> {
+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;
@@ -225,11 +254,12 @@ async fn ensure_agent_subvolume(env: &Env<'_>, agent_name: &str) -> Result<(Stri
     // 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.on_btrfs)(root)? {
+    if !env.probes.on_btrfs(root)? {
         return Ok((String::new(), String::new()));
     }
 
-    let out = (env.btrfs)()
+    let out = env
+        .btrfs()
         .args(["subvolume", "create"])
         .arg(&agent_root)
         .output()
@@ -260,7 +290,8 @@ async fn ensure_agent_subvolume(env: &Env<'_>, agent_name: &str) -> Result<(Stri
         });
     if let Err(e) = chown_result {
         // Best-effort rollback so we never leave a root-owned subvol behind.
-        let _ = (env.btrfs)()
+        let _ = env
+            .btrfs()
             .args(["subvolume", "delete"])
             .arg(&agent_root)
             .output()
@@ -277,12 +308,16 @@ async fn ensure_agent_subvolume(env: &Env<'_>, agent_name: &str) -> Result<(Stri
 /// `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<'_>, agent_name: &str) -> Result<(String, String)> {
+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.is_subvolume)(&agent_root) {
+    if !env.probes.is_subvolume(&agent_root) {
         return Ok((String::new(), String::new()));
     }
-    let out = (env.btrfs)()
+    let out = env
+        .btrfs()
         .args(["subvolume", "delete"])
         .arg(&agent_root)
         .output()
@@ -306,11 +341,11 @@ 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<'_>) -> Result<(String, String)> {
+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.on_btrfs)(root)? {
+    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((
@@ -318,7 +353,8 @@ async fn ensure_btrfs_quota_in(env: &Env<'_>) -> Result<(String, String)> {
             "filesystem is not btrfs — quota not applicable".to_owned(),
         ));
     }
-    let out = (env.btrfs)()
+    let out = env
+        .btrfs()
         .args(["quota", "enable"])
         .arg(root)
         .output()
@@ -341,9 +377,13 @@ async fn ensure_btrfs_quota_in(env: &Env<'_>) -> Result<(String, String)> {
 /// 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<'_>, agent_name: &str) -> Result<(String, String)> {
+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)()
+    let out = env
+        .btrfs()
         .args(["qgroup", "show", "-f", "--raw"])
         .arg(&agent_root)
         .output()
@@ -459,9 +499,10 @@ fn parse_du_total(stdout: &str) -> Option<u64> {
 /// 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<'_>, path: &Path) {
+async fn cleanup_stale_path(env: &Env<'_, impl Spawn, impl Probes>, path: &Path) {
     if path.exists() {
-        let _ = (env.btrfs)()
+        let _ = env
+            .btrfs()
             .args(["subvolume", "delete"])
             .arg(path)
             .output()
@@ -475,11 +516,16 @@ async fn cleanup_stale_path(env: &Env<'_>, path: &Path) {
 /// 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<'_>, agent_root: &Path, tmp: &Path) -> Result<()> {
+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)()
+    let out = env
+        .btrfs()
         .args(["subvolume", "create"])
         .arg(tmp)
         .output()
@@ -543,7 +589,10 @@ async fn stage_upgrade_subvolume(env: &Env<'_>, agent_root: &Path, tmp: &Path) -
 /// ([`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<'_>, agent_name: &str) -> Result<(String, String)> {
+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);
 
@@ -569,10 +618,10 @@ async fn upgrade_agent_subvolume(env: &Env<'_>, agent_name: &str) -> Result<(Str
         );
     }
     // Idempotent: already a subvolume → nothing to do.
-    if (env.is_subvolume)(&agent_root) {
+    if env.probes.is_subvolume(&agent_root) {
         return Ok((String::new(), String::new()));
     }
-    if !(env.on_btrfs)(root)? {
+    if !env.probes.on_btrfs(root)? {
         bail!(
             "{} is not on btrfs — subvolumes are unsupported, cannot upgrade",
             root.display()
@@ -647,12 +696,12 @@ fn snapshot_path(root: &Path, agent_name: &str, snapshot_name: &str) -> PathBuf
 /// 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<'_>,
+    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.is_subvolume)(&agent_root) {
+    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()
@@ -665,7 +714,8 @@ async fn snapshot_agent_subvolume(
             snap.display()
         );
     }
-    let out = (env.btrfs)()
+    let out = env
+        .btrfs()
         .args(["subvolume", "snapshot", "-r"])
         .arg(&agent_root)
         .arg(&snap)
@@ -696,7 +746,7 @@ async fn snapshot_agent_subvolume(
 /// `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<'_>,
+    env: &Env<'_, impl Spawn, impl Probes>,
     agent_name: &str,
     snapshot_name: &str,
 ) -> Result<(String, String)> {
@@ -704,7 +754,8 @@ async fn delete_agent_snapshot(
     if !snap.exists() {
         return Ok((String::new(), String::new()));
     }
-    let out = (env.btrfs)()
+    let out = env
+        .btrfs()
         .args(["subvolume", "delete"])
         .arg(&snap)
         .output()
@@ -815,7 +866,7 @@ impl Drop for PartialExportGuard<'_> {
 /// inter-hive migration transport; see `PrivRequest::SendAgentSnapshotToFile`
 /// for the cross-hive follow-up.
 async fn send_agent_snapshot_to_file(
-    env: &Env<'_>,
+    env: &Env<'_, impl Spawn, impl Probes>,
     agent_name: &str,
     snapshot_name: &str,
     parent_name: Option<&str>,
@@ -833,7 +884,7 @@ async fn send_agent_snapshot_to_file(
     let dest_file = open_export_dest(&snap, parent_path.as_deref(), &dest)?;
     let cleanup = PartialExportGuard::new(&dest);
 
-    let mut cmd = (env.btrfs)();
+    let mut cmd = env.btrfs();
     cmd.arg("send");
     if let Some(parent_path) = &parent_path {
         cmd.arg("-p").arg(parent_path);
@@ -875,7 +926,7 @@ async fn send_agent_snapshot_to_file(
 /// 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<'_>,
+    env: &Env<'_, impl Spawn, impl Probes>,
     agent_name: &str,
     snapshot_name: &str,
     parent_name: Option<&str>,
@@ -889,7 +940,7 @@ async fn send_agent_snapshot_to_fd(
         );
     }
 
-    let mut cmd = (env.btrfs)();
+    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);
@@ -932,13 +983,14 @@ async fn send_agent_snapshot_to_fd(
 /// subvolume (`btrfs qgroup limit <bytes|none> <…/agent_name>`). See the
 /// wire doc.
 async fn set_subvolume_quota(
-    env: &Env<'_>,
+    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)()
+    let out = env
+        .btrfs()
         .args(["qgroup", "limit", &limit])
         .arg(&agent_root)
         .output()
@@ -957,7 +1009,7 @@ async fn set_subvolume_quota(
 #[cfg(test)]
 mod tests {
     use super::{
-        Command, Env, agent_disk_paths, cleanup_stale_path, delete_agent_snapshot,
+        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,
@@ -968,7 +1020,8 @@ mod tests {
         set_subvolume_quota, snapshot_agent_subvolume, snapshot_path, upgrade_agent_subvolume,
         validate_snapshot_name,
     };
-    use crate::shared::test_support::scratch;
+    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};
@@ -1360,31 +1413,46 @@ mod tests {
         }
 
         /// An [`Env`] over this tree, running `btrfs` and the two probes given.
-        fn env<'a>(
+        fn env<'a, S: Spawn>(
             &'a self,
-            btrfs: &'a (dyn Fn() -> Command + Sync),
+            spawn: &'a S,
             on_btrfs: fn(&Path) -> anyhow::Result<bool>,
             is_subvolume: fn(&Path) -> bool,
-        ) -> Env<'a> {
+        ) -> Env<'a, &'a S, FnProbes> {
             Env {
                 state_root: &self.state,
                 staging_root: &self.staging,
-                btrfs,
-                on_btrfs,
-                is_subvolume,
+                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) -> impl Fn() -> Command + Sync + use<> {
-            let log = self.log.clone();
-            let script = format!("echo \"$*\" >> \"$FAKE_LOG\"\n{body}");
-            move || {
-                let mut c = Command::new("sh");
-                c.arg("-c").arg(&script).arg("btrfs").env("FAKE_LOG", &log);
-                c
-            }
+        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)
         }
     }
 
@@ -1519,7 +1587,7 @@ esac"#;
     #[tokio::test]
     async fn ensure_surfaces_a_missing_btrfs_binary_and_a_failed_probe() {
         let fx = Fx::new();
-        let missing = || Command::new("/nonexistent-hive-priv-test/btrfs");
+        let missing = MissingTool;
         let env = fx.env(&missing, on_btrfs, not_subvol);
         let err = ensure_agent_subvolume(&env, "iris").await.unwrap_err();
         assert!(