hive-priv: test container.rs's start, stop and config decisions

hive-priv/src/container.rs

@@ -14,6 +14,45 @@ 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)
+}
+
+/// 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),
+    /// 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
+    /// shutdown that was going to finish on its own.
+    release_timeout: std::time::Duration,
+    /// Poll cadence while waiting for the registration to go away.
+    release_poll: std::time::Duration,
+    /// Cap on a single registration probe. The probe is a D-Bus round trip to
+    /// machined; if machined itself is wedged the call would otherwise sit on
+    /// the D-Bus method timeout, which is far longer than the whole stop
+    /// sequence should take.
+    probe_timeout: std::time::Duration,
+}
+
+impl Host<'static> {
+    const REAL: Self = Host {
+        spawn: &real_command,
+        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<'_> {
+    fn command(&self, program: &str) -> Command {
+        (self.spawn)(program)
+    }
+}
+
 /// Write one `PrivEvent::Line` to the client. Best-effort: a write
 /// failure is logged but doesn't abort the running subprocess.
 async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data: &str) {
@@ -34,19 +73,24 @@ async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data:
 /// this itself). Best-effort — if the unit doesn't exist yet (first-time
 /// create), `reset-failed` is a no-op and the start proceeds regardless.
 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)> {
     validate_agent_name(name)?;
     let machine = container_system_name(name);
     // Silenced: a first-time create means the unit doesn't exist yet, and
     // `systemctl`'s own "Unit ... not loaded" complaint for that case would
     // otherwise land in the journal at error priority on every such start —
     // routine, not a signal, since the result is already discarded below.
-    let _ = Command::new("systemctl")
+    let _ = host
+        .command("systemctl")
         .args(["reset-failed", &format!("container@{machine}.service")])
         .stdout(std::process::Stdio::null())
         .stderr(std::process::Stdio::null())
         .status()
         .await;
-    container_run(&["start", &machine]).await
+    container_run_with(host, &["start", &machine]).await
 }
 
 /// Most bytes of journal text `StartFailureLog` returns.
@@ -106,27 +150,41 @@ pub(crate) async fn start_failure_log(name: &str, since_unix: u64) -> Result<(St
     let now = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
         .map_or(0, |d| d.as_secs());
-    start_failure_log_with(name, since_unix, now, |argv| async move {
+    start_failure_log_with(name, since_unix, now, |argv| {
-        let child = Command::new("journalctl")
+        run_journalctl(&Host::REAL, START_LOG_TIMEOUT, argv)
-            .args(&argv)
-            .kill_on_drop(true)
-            .output();
-        let out = tokio::time::timeout(START_LOG_TIMEOUT, child)
-            .await
-            .context("journalctl timed out")?
-            .context("invoke journalctl")?;
-        if !out.status.success() {
-            bail!("journalctl failed ({})", out.status);
-        }
-        Ok(String::from_utf8_lossy(&out.stdout).into_owned())
     })
     .await
 }
 
+/// `journalctl` with `argv`: its stdout, if it exits 0 within `limit`.
+async fn run_journalctl(
+    host: &Host<'_>,
+    limit: std::time::Duration,
+    argv: Vec<String>,
+) -> Result<String> {
+    let child = host
+        .command("journalctl")
+        .args(&argv)
+        .kill_on_drop(true)
+        .output();
+    let out = tokio::time::timeout(limit, child)
+        .await
+        .context("journalctl timed out")?
+        .context("invoke journalctl")?;
+    if !out.status.success() {
+        bail!("journalctl failed ({})", out.status);
+    }
+    Ok(String::from_utf8_lossy(&out.stdout).into_owned())
+}
+
 /// `StopContainer`.
 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)> {
     validate_agent_name(name)?;
-    stop_and_release(&container_system_name(name)).await
+    stop_and_release(host, &container_system_name(name)).await
 }
 
 /// `KillContainer`: `nixos-container` has no kill verb, so this uses
@@ -134,15 +192,23 @@ pub(crate) async fn stop_container(name: &str) -> Result<(String, String)> {
 /// right semantics for a forced shutdown after a graceful stop has already
 /// been attempted.
 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)> {
     validate_agent_name(name)?;
     let machine = container_system_name(name);
-    machinectl_run(&["kill", &machine, "--signal=SIGKILL"]).await
+    machinectl_run(host, &["kill", &machine, "--signal=SIGKILL"]).await
 }
 
 /// `DestroyContainer`.
 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)> {
     validate_agent_name(name)?;
-    container_run(&["destroy", &container_system_name(name)]).await
+    container_run_with(host, &["destroy", &container_system_name(name)]).await
 }
 
 /// Shared body for `CreateContainer` / `UpdateContainer`: validate the name
@@ -168,15 +234,28 @@ pub(crate) async fn container_toplevel_action(
 ) -> Result<(String, String)> {
     validate_agent_name(name)?;
     let toplevel = crate::closure::applicable_for(name, toplevel)?;
+    apply_toplevel(&Host::REAL, verb, name, toplevel, stream, writer).await
+}
+
+/// [`container_toplevel_action`] past the check that `toplevel` is the
+/// closure realised for `name`.
+async fn apply_toplevel(
+    host: &Host<'_>,
+    verb: &str,
+    name: &str,
+    toplevel: &str,
+    stream: bool,
+    writer: &mut OwnedWriteHalf,
+) -> Result<(String, String)> {
     let system_name = container_system_name(name);
     if verb == "update" {
-        return swap_container_profile(&system_name, toplevel, stream, writer).await;
+        return swap_container_profile(host, &system_name, toplevel, stream, writer).await;
     }
     let args = [verb, &system_name, "--system-path", toplevel];
     if stream {
-        container_run_streaming(&args, writer).await
+        container_run_streaming(host, &args, writer).await
     } else {
-        container_run(&args).await
+        container_run_with(host, &args).await
     }
 }
 
@@ -193,13 +272,15 @@ pub(crate) async fn container_toplevel_action(
 /// 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<'_>,
     system_name: &str,
     toplevel: &str,
     stream: bool,
     writer: &mut OwnedWriteHalf,
 ) -> Result<(String, String)> {
     let profile = format!("/nix/var/nix/profiles/per-container/{system_name}/system");
-    let set_out = Command::new("nix-env")
+    let set_out = host
+        .command("nix-env")
         .args(["-p", &profile, "--set", toplevel])
         .output()
         .await
@@ -216,14 +297,16 @@ async fn swap_container_profile(
     }
 
     let unit = format!("container@{system_name}");
-    let state_out = Command::new("systemctl")
+    let state_out = host
+        .command("systemctl")
         .args(["show", "--property=ActiveState", "--value", &unit])
         .output()
         .await
         .context("query container ActiveState")?;
     let active = String::from_utf8_lossy(&state_out.stdout).trim() == "active";
     if active {
-        let reload_out = Command::new("systemctl")
+        let reload_out = host
+            .command("systemctl")
             .args(["reload", &unit])
             .output()
             .await
@@ -362,7 +445,12 @@ pub(crate) async fn run_bounded(
 
 /// Invoke `nixos-container` with the given args, log output to journald.
 pub(crate) async fn container_run(args: &[&str]) -> Result<(String, String)> {
-    let out = Command::new("nixos-container")
+    container_run_with(&Host::REAL, args).await
+}
+
+async fn container_run_with(host: &Host<'_>, args: &[&str]) -> Result<(String, String)> {
+    let out = host
+        .command("nixos-container")
         .args(args)
         .output()
         .await
@@ -396,8 +484,9 @@ pub(crate) async fn container_run(args: &[&str]) -> Result<(String, String)> {
 /// 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(args: &[&str]) -> Result<(String, String)> {
+async fn machinectl_run(host: &Host<'_>, args: &[&str]) -> Result<(String, String)> {
-    let out = Command::new("machinectl")
+    let out = host
+        .command("machinectl")
         .args(args)
         .output()
         .await
@@ -421,21 +510,6 @@ async fn machinectl_run(args: &[&str]) -> Result<(String, String)> {
     Ok((stdout, stderr))
 }
 
-/// 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
-/// shutdown that was going to finish on its own.
-const NAME_RELEASE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20);
-
-/// Poll cadence while waiting for the registration to go away.
-const NAME_RELEASE_POLL: std::time::Duration = std::time::Duration::from_millis(500);
-
-/// Cap on a single registration probe. The probe is a D-Bus round trip to
-/// machined; if machined itself is wedged the call would otherwise sit on
-/// the D-Bus method timeout, which is far longer than the whole stop
-/// sequence should take.
-const MACHINE_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
-
 /// True when machined still holds a registration for `machine`.
 ///
 /// This asks machined the same question the registration itself answers —
@@ -451,13 +525,14 @@ const MACHINE_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_sec
 /// 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(machine: &str) -> bool {
+async fn machine_registered(host: &Host<'_>, machine: &str) -> bool {
-    let probe = Command::new("machinectl")
+    let probe = host
+        .command("machinectl")
         .args(["show", machine, "--property=Name"])
         // Don't leave a probe behind when the timeout below fires.
         .kill_on_drop(true)
         .output();
-    match tokio::time::timeout(MACHINE_PROBE_TIMEOUT, probe).await {
+    match tokio::time::timeout(host.probe_timeout, probe).await {
         Ok(Ok(out)) => out.status.success(),
         Ok(Err(e)) => {
             tracing::warn!(%machine, error = %e, "machinectl show failed to run; assuming still registered");
@@ -472,16 +547,16 @@ async fn machine_registered(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(machine: &str) -> bool {
+async fn wait_name_released(host: &Host<'_>, machine: &str) -> bool {
-    let deadline = tokio::time::Instant::now() + NAME_RELEASE_TIMEOUT;
+    let deadline = tokio::time::Instant::now() + host.release_timeout;
     loop {
-        if !machine_registered(machine).await {
+        if !machine_registered(host, machine).await {
             return true;
         }
         if tokio::time::Instant::now() >= deadline {
             return false;
         }
-        tokio::time::sleep(NAME_RELEASE_POLL).await;
+        tokio::time::sleep(host.release_poll).await;
     }
 }
 
@@ -514,9 +589,9 @@ async fn wait_name_released(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(machine: &str) -> Result<(String, String)> {
+async fn stop_and_release(host: &Host<'_>, machine: &str) -> Result<(String, String)> {
-    let stop = container_run(&["stop", machine]).await;
+    let stop = container_run_with(host, &["stop", machine]).await;
-    if wait_name_released(machine).await {
+    if wait_name_released(host, machine).await {
         // Happy path, and also the path where a stop that reported failure
         // nonetheless brought the machine down. Either way the caller gets
         // the original result untouched.
@@ -535,7 +610,7 @@ async fn stop_and_release(machine: &str) -> Result<(String, String)> {
          This container is likely wedged and starting it again will fail to \
          register — needs host-level intervention (a reboot has been the \
          only reliable fix in practice).",
-        NAME_RELEASE_TIMEOUT.as_secs()
+        host.release_timeout.as_secs()
     )
 }
 
@@ -545,13 +620,14 @@ async fn stop_and_release(machine: &str) -> Result<(String, String)> {
 /// on success (all output was streamed); the error string includes stderr
 /// tail on failure.
 async fn container_run_streaming(
+    host: &Host<'_>,
     args: &[&str],
     writer: &mut OwnedWriteHalf,
 ) -> Result<(String, String)> {
     use tokio::io::AsyncBufReadExt as _;
-    use tokio::process::Command;
 
-    let mut child = Command::new("nixos-container")
+    let mut child = host
+        .command("nixos-container")
         .args(args)
         .stdout(std::process::Stdio::piped())
         .stderr(std::process::Stdio::piped())
@@ -774,3 +850,544 @@ mod start_log_tests {
         assert_eq!(small, "a\n");
     }
 }
+
+#[cfg(test)]
+mod host_tests {
+    use std::path::PathBuf;
+    use std::time::Duration;
+
+    use tokio::io::AsyncReadExt as _;
+    use tokio::net::UnixStream;
+
+    use super::*;
+    use crate::shared::test_support::scratch;
+
+    const TOPLEVEL: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-bob";
+
+    /// Programs that are `sh -c <script> <program> <args…>`: every call is
+    /// appended to a log as `program args…`, then `behave` decides the exit
+    /// status and output from `$0` (the program) and `$@` (its arguments).
+    /// `$D` is the stub's scratch directory.
+    struct Stub {
+        dir: PathBuf,
+        spawn: Box<dyn Fn(&str) -> Command + Sync>,
+    }
+
+    impl Stub {
+        fn new(behave: &str) -> Self {
+            let dir = scratch();
+            let script = format!(
+                "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),
+            }
+        }
+
+        /// Timings short enough for a test that waits them out.
+        fn host(&self) -> Host<'_> {
+            Host {
+                spawn: &*self.spawn,
+                release_timeout: Duration::from_millis(300),
+                release_poll: Duration::from_millis(20),
+                probe_timeout: Duration::from_secs(10),
+            }
+        }
+
+        /// Every command run so far, in order.
+        fn calls(&self) -> Vec<String> {
+            std::fs::read_to_string(self.dir.join("calls"))
+                .map(|log| log.lines().map(str::to_owned).collect())
+                .unwrap_or_default()
+        }
+    }
+
+    impl Drop for Stub {
+        fn drop(&mut self) {
+            std::fs::remove_dir_all(&self.dir).ok();
+        }
+    }
+
+    /// A write half and the read end of its socket.
+    fn pipe() -> (OwnedWriteHalf, UnixStream) {
+        let (ours, theirs) = UnixStream::pair().unwrap();
+        (ours.into_split().1, theirs)
+    }
+
+    /// The `(stream, data)` of every `Line` event the writer sent. Takes the
+    /// writer so its end closes and the read finishes.
+    async fn lines_sent(writer: OwnedWriteHalf, mut peer: UnixStream) -> Vec<(PrivStream, String)> {
+        drop(writer);
+        let mut raw = String::new();
+        peer.read_to_string(&mut raw).await.unwrap();
+        raw.lines()
+            .map(|l| match serde_json::from_str::<PrivEvent>(l).unwrap() {
+                PrivEvent::Line(line) => (line.stream, line.data),
+                PrivEvent::Done(_) => panic!("unexpected terminal event"),
+            })
+            .collect()
+    }
+
+    const BAD_NAMES: [&str; 8] = [
+        "",
+        "../x",
+        "bob/../x",
+        "A",
+        "a b",
+        "x.service",
+        "h_bob",
+        "x\n--vacuum-size=1",
+    ];
+
+    fn err_text(r: Result<(String, String)>) -> String {
+        format!("{:#}", r.unwrap_err())
+    }
+
+    // ---- name validation: nothing runs for a refused name ----
+
+    #[tokio::test]
+    async fn every_lifecycle_verb_refuses_a_bad_name_before_running_anything() {
+        let stub = Stub::new("exit 0");
+        let host = stub.host();
+        for bad in BAD_NAMES {
+            let results = [
+                start_container_with(&host, bad).await,
+                stop_container_with(&host, bad).await,
+                kill_container_with(&host, bad).await,
+                destroy_container_with(&host, bad).await,
+            ];
+            for r in results {
+                assert!(
+                    err_text(r).contains("invalid name"),
+                    "{bad:?} not refused as a name"
+                );
+            }
+        }
+        assert_eq!(stub.calls(), Vec::<String>::new(), "a command ran");
+
+        // Control: a good name runs, so an empty call log means "refused".
+        kill_container_with(&host, "bob").await.unwrap();
+        assert_eq!(stub.calls().len(), 1);
+    }
+
+    #[tokio::test]
+    async fn the_apply_verbs_check_the_name_before_the_closure() {
+        for bad in BAD_NAMES {
+            let (writer, _peer) = &mut pipe();
+            let err = container_toplevel_action("create", bad, TOPLEVEL, false, writer)
+                .await
+                .unwrap_err();
+            assert!(
+                format!("{err:#}").contains("invalid name"),
+                "{bad:?}: {err:#}"
+            );
+        }
+    }
+
+    // ---- the apply gate ----
+
+    /// Only the closure realised for the agent is applied. Nothing is
+    /// realised for these names, so each request is refused at the gate: the
+    /// message is the gate's, not a `nixos-container` failure.
+    #[tokio::test]
+    async fn an_unrealised_or_foreign_closure_is_refused_at_the_gate() {
+        for verb in ["create", "update"] {
+            let (writer, _peer) = &mut pipe();
+            let err = container_toplevel_action(verb, "never-realised", TOPLEVEL, false, writer)
+                .await
+                .unwrap_err();
+            let msg = format!("{err:#}");
+            assert!(
+                msg.contains("is not the closure realised for this agent (none)"),
+                "{verb}: {msg}"
+            );
+        }
+        for not_a_store_path in ["/bin/sh", "", "/nix/store/../etc/passwd"] {
+            let (writer, _peer) = &mut pipe();
+            let err = container_toplevel_action(
+                "create",
+                "never-realised",
+                not_a_store_path,
+                false,
+                writer,
+            )
+            .await
+            .unwrap_err();
+            assert!(
+                format!("{err:#}").contains("is not a store path"),
+                "{not_a_store_path:?}: {err:#}"
+            );
+        }
+    }
+
+    // ---- start ----
+
+    #[tokio::test]
+    async fn start_clears_the_start_limit_then_starts_the_prefixed_machine() {
+        let stub = Stub::new("echo started");
+        let (stdout, _) = start_container_with(&stub.host(), "bob").await.unwrap();
+        assert_eq!(stdout, "started\n");
+        assert_eq!(
+            stub.calls(),
+            [
+                "systemctl reset-failed container@h-bob.service",
+                "nixos-container start h-bob"
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn a_failed_reset_does_not_stop_the_start_but_a_failed_start_is_an_error() {
+        let stub = Stub::new("[ \"$0\" = systemctl ] && exit 1; echo ok");
+        start_container_with(&stub.host(), "bob").await.unwrap();
+
+        let stub =
+            Stub::new("[ \"$0\" = nixos-container ] && { echo no unit >&2; exit 3; }; exit 0");
+        let msg = err_text(start_container_with(&stub.host(), "bob").await);
+        assert!(msg.contains("nixos-container start h-bob failed"), "{msg}");
+        assert!(
+            msg.contains("no unit"),
+            "stderr must be in the error: {msg}"
+        );
+    }
+
+    // ---- kill / destroy ----
+
+    #[tokio::test]
+    async fn kill_sends_sigkill_to_the_prefixed_machine_through_machinectl() {
+        let stub = Stub::new("echo killed");
+        let (stdout, _) = kill_container_with(&stub.host(), "bob").await.unwrap();
+        assert_eq!(stdout, "killed\n");
+        assert_eq!(stub.calls(), ["machinectl kill h-bob --signal=SIGKILL"]);
+
+        let stub = Stub::new("echo not running >&2; exit 1");
+        let msg = err_text(kill_container_with(&stub.host(), "bob").await);
+        assert!(
+            msg.contains("machinectl kill h-bob --signal=SIGKILL failed"),
+            "{msg}"
+        );
+        assert!(msg.contains("not running"), "{msg}");
+    }
+
+    #[tokio::test]
+    async fn destroy_names_only_the_prefixed_machine() {
+        let stub = Stub::new("exit 0");
+        destroy_container_with(&stub.host(), "bob").await.unwrap();
+        assert_eq!(stub.calls(), ["nixos-container destroy h-bob"]);
+    }
+
+    // ---- container_run / machinectl_run ----
+
+    #[tokio::test]
+    async fn container_run_returns_both_streams_and_fails_with_the_trimmed_stderr() {
+        let stub = Stub::new("echo out; echo err >&2");
+        let (out, err) = container_run_with(&stub.host(), &["list"]).await.unwrap();
+        assert_eq!((out.as_str(), err.as_str()), ("out\n", "err\n"));
+
+        let stub = Stub::new("echo bad thing >&2; exit 2");
+        let msg = err_text(container_run_with(&stub.host(), &["stop", "h-bob"]).await);
+        assert!(
+            msg.starts_with("nixos-container stop h-bob failed ("),
+            "{msg}"
+        );
+        assert!(msg.ends_with("): bad thing"), "{msg}");
+    }
+
+    #[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 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);
+        assert!(msg.contains("invoke machinectl"), "{msg}");
+    }
+
+    // ---- journalctl ----
+
+    #[tokio::test]
+    async fn journalctl_runs_with_exactly_the_given_argv() {
+        let stub = Stub::new("echo line");
+        let argv = start_log_argv("h-bob", 5);
+        let out = run_journalctl(&stub.host(), Duration::from_secs(10), argv)
+            .await
+            .unwrap();
+        assert_eq!(out, "line\n");
+        assert_eq!(
+            stub.calls(),
+            ["journalctl _SYSTEMD_UNIT=container@h-bob.service --since @5 -n 30 -o cat --no-pager"]
+        );
+    }
+
+    #[tokio::test]
+    async fn journalctl_failure_and_timeout_are_errors() {
+        let stub = Stub::new("exit 1");
+        let r = run_journalctl(&stub.host(), Duration::from_secs(10), vec![]).await;
+        assert!(format!("{:#}", r.unwrap_err()).contains("journalctl failed"));
+
+        let stub = Stub::new("exec sleep 30");
+        let r = run_journalctl(&stub.host(), Duration::from_millis(100), vec![]).await;
+        assert!(format!("{:#}", r.unwrap_err()).contains("journalctl timed out"));
+    }
+
+    // ---- stop: success means the name is free ----
+
+    #[tokio::test]
+    async fn a_stop_that_releases_the_name_returns_the_stop_output() {
+        let stub = Stub::new("[ \"$0\" = machinectl ] && exit 1; echo stopped");
+        let (stdout, _) = stop_container_with(&stub.host(), "bob").await.unwrap();
+        assert_eq!(stdout, "stopped\n");
+        assert_eq!(
+            stub.calls(),
+            [
+                "nixos-container stop h-bob",
+                "machinectl show h-bob --property=Name"
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn a_stop_that_leaves_the_name_registered_is_an_error_even_if_it_exited_zero() {
+        let stub = Stub::new("exit 0");
+        let msg = err_text(stop_container_with(&stub.host(), "bob").await);
+        assert!(
+            msg.contains("machined still holds the machine name"),
+            "{msg}"
+        );
+        assert!(msg.contains("stop h-bob"), "{msg}");
+        let probes = stub
+            .calls()
+            .iter()
+            .filter(|c| c.starts_with("machinectl show"))
+            .count();
+        assert!(probes > 1, "the name was polled, not probed once: {probes}");
+    }
+
+    #[tokio::test]
+    async fn the_name_may_free_up_after_a_few_polls() {
+        let stub = Stub::new(
+            "if [ \"$0\" = machinectl ]; then\n\
+               n=$(wc -l < \"$D/probes\" 2>/dev/null || echo 0)\n\
+               echo x >> \"$D/probes\"\n\
+               [ \"$n\" -ge 2 ] && exit 1\n\
+               exit 0\n\
+             fi\n\
+             echo stopped",
+        );
+        let host = Host {
+            release_timeout: Duration::from_secs(10),
+            ..stub.host()
+        };
+        let (stdout, _) = stop_container_with(&host, "bob").await.unwrap();
+        assert_eq!(stdout, "stopped\n");
+        let probes = stub
+            .calls()
+            .iter()
+            .filter(|c| c.starts_with("machinectl show"))
+            .count();
+        assert_eq!(probes, 3);
+    }
+
+    #[tokio::test]
+    async fn a_failed_stop_that_freed_the_name_keeps_its_own_error() {
+        let stub = Stub::new("[ \"$0\" = machinectl ] && exit 1; echo refused >&2; exit 1");
+        let msg = err_text(stop_container_with(&stub.host(), "bob").await);
+        assert!(msg.contains("nixos-container stop h-bob failed"), "{msg}");
+        assert!(msg.contains("refused"), "{msg}");
+        assert!(!msg.contains("machined still holds"), "{msg}");
+    }
+
+    #[tokio::test]
+    async fn the_registration_probe_reads_exit_status_and_fails_safe() {
+        let registered = Stub::new("exit 0");
+        assert!(machine_registered(&registered.host(), "h-bob").await);
+        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()
+        };
+        assert!(
+            machine_registered(&host, "h-bob").await,
+            "a probe that cannot run must answer registered"
+        );
+
+        let wedged = Stub::new("exec sleep 30");
+        let host = Host {
+            probe_timeout: Duration::from_millis(100),
+            ..wedged.host()
+        };
+        assert!(
+            machine_registered(&host, "h-bob").await,
+            "a probe that times out must answer registered"
+        );
+    }
+
+    // ---- create / update ----
+
+    #[tokio::test]
+    async fn create_passes_the_closure_as_system_path_not_a_flake() {
+        let stub = Stub::new("echo created");
+        let (writer, _peer) = &mut pipe();
+        let (stdout, _) = apply_toplevel(&stub.host(), "create", "bob", TOPLEVEL, false, writer)
+            .await
+            .unwrap();
+        assert_eq!(stdout, "created\n");
+        assert_eq!(
+            stub.calls(),
+            [format!(
+                "nixos-container create h-bob --system-path {TOPLEVEL}"
+            )]
+        );
+    }
+
+    #[tokio::test]
+    async fn a_streamed_create_forwards_each_line_and_returns_nothing() {
+        let stub = Stub::new("echo building; echo warning >&2");
+        let (writer, peer) = pipe();
+        let mut writer = writer;
+        let (stdout, stderr) =
+            apply_toplevel(&stub.host(), "create", "bob", TOPLEVEL, true, &mut writer)
+                .await
+                .unwrap();
+        assert_eq!((stdout.as_str(), stderr.as_str()), ("", ""));
+        let mut seen = lines_sent(writer, peer).await;
+        seen.sort_by_key(|(s, _)| *s == PrivStream::Stderr);
+        assert_eq!(
+            seen,
+            [
+                (PrivStream::Stdout, "building".to_owned()),
+                (PrivStream::Stderr, "warning".to_owned())
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn a_failed_streamed_create_carries_only_the_last_stderr_line() {
+        let stub = Stub::new("echo first >&2; echo last >&2; exit 4");
+        let (writer, peer) = pipe();
+        let mut writer = writer;
+        let err = apply_toplevel(&stub.host(), "create", "bob", TOPLEVEL, true, &mut writer)
+            .await
+            .unwrap_err();
+        let msg = format!("{err:#}");
+        assert!(
+            msg.contains("nixos-container create h-bob --system-path"),
+            "{msg}"
+        );
+        assert!(msg.ends_with(": last"), "{msg}");
+        let seen = lines_sent(writer, peer).await;
+        assert_eq!(
+            seen.len(),
+            2,
+            "every line is forwarded even though the error keeps one"
+        );
+    }
+
+    #[tokio::test]
+    async fn update_sets_the_per_container_profile_and_never_calls_nixos_container() {
+        let stub = Stub::new(
+            "[ \"$0\" = systemctl ] && [ \"$2\" = --property=ActiveState ] && echo inactive; exit 0",
+        );
+        let (writer, _peer) = &mut pipe();
+        apply_toplevel(&stub.host(), "update", "bob", TOPLEVEL, false, writer)
+            .await
+            .unwrap();
+        assert_eq!(
+            stub.calls(),
+            [
+                format!(
+                    "nix-env -p /nix/var/nix/profiles/per-container/h-bob/system --set {TOPLEVEL}"
+                ),
+                "systemctl show --property=ActiveState --value container@h-bob".to_owned(),
+            ],
+            "a stopped container is not reloaded"
+        );
+    }
+
+    #[tokio::test]
+    async fn update_reloads_a_running_container_and_streams_both_steps() {
+        let stub = Stub::new(
+            "case \"$0 $1\" in\n\
+             'systemctl show') echo active;;\n\
+             'nix-env -p') echo set-done;;\n\
+             'systemctl reload') echo reloaded; echo reload-warn >&2;;\n\
+             esac",
+        );
+        let (writer, peer) = pipe();
+        let mut writer = writer;
+        let (stdout, stderr) =
+            apply_toplevel(&stub.host(), "update", "bob", TOPLEVEL, true, &mut writer)
+                .await
+                .unwrap();
+        assert_eq!(stdout, "set-done\nreloaded\n");
+        assert_eq!(stderr, "reload-warn\n");
+        assert_eq!(
+            stub.calls().last().unwrap(),
+            "systemctl reload container@h-bob"
+        );
+        assert_eq!(
+            lines_sent(writer, peer).await,
+            [
+                (PrivStream::Stdout, "set-done".to_owned()),
+                (PrivStream::Stdout, "reloaded".to_owned()),
+                (PrivStream::Stderr, "reload-warn".to_owned()),
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn update_stops_at_a_failed_profile_swap_and_reports_a_failed_reload() {
+        let stub = Stub::new("[ \"$0\" = nix-env ] && { echo no space >&2; exit 1; }; echo active");
+        let (writer, _peer) = &mut pipe();
+        let msg =
+            err_text(apply_toplevel(&stub.host(), "update", "bob", TOPLEVEL, false, writer).await);
+        assert!(
+            msg.contains("/per-container/h-bob/system --set failed"),
+            "{msg}"
+        );
+        assert!(msg.contains("no space"), "{msg}");
+        assert_eq!(stub.calls().len(), 1, "nothing runs after a failed swap");
+
+        let stub =
+            Stub::new("case \"$1\" in reload) echo wedged >&2; exit 1;; show) echo active;; esac");
+        let (writer, _peer) = &mut pipe();
+        let msg =
+            err_text(apply_toplevel(&stub.host(), "update", "bob", TOPLEVEL, false, writer).await);
+        assert!(
+            msg.contains("systemctl reload container@h-bob failed"),
+            "{msg}"
+        );
+        assert!(msg.contains("wedged"), "{msg}");
+    }
+
+    // ---- client events ----
+
+    #[tokio::test]
+    async fn a_line_event_is_one_json_line_and_a_dead_client_is_not_fatal() {
+        let (mut writer, peer) = pipe();
+        write_line_event(&mut writer, PrivStream::Stderr, "boom").await;
+        assert_eq!(
+            lines_sent(writer, peer).await,
+            [(PrivStream::Stderr, "boom".to_owned())]
+        );
+
+        let (mut writer, peer) = pipe();
+        drop(peer);
+        // Best-effort: the failed write is logged, not propagated.
+        write_line_event(&mut writer, PrivStream::Stdout, "x").await;
+        write_line_event(&mut writer, PrivStream::Stdout, "y").await;
+    }
+}