capsule-runtime 0.11.0

//! A parent's runtime launching an approved child as an operating-system process of its own.
//!
//! Every case here runs the real thing: `mur-roost` on a loopback port over a real registry, the
//! built `mur` binary as the child, and the child's own runtime registering itself. What the
//! daemon does is referee; what this crate does is launch.
//!
//! One daemon, one registry and one `HOME` are shared by the whole suite, because `HOME` is
//! process-wide and a child resolves its artifacts through it. Each case gets its own parent
//! workdir, so nothing a case creates is visible to another.

#[path = "common/mod.rs"]
mod common;

use std::path::{Path, PathBuf};
use std::sync::OnceLock;
use std::time::{Duration, Instant};

use capsule_runtime::child_launch::CONTAINED_LAUNCH_FLAG;
use capsule_runtime::session_status::{SessionDeadline, INHERITED_DEADLINE_FLAG};
use capsule_runtime::{
    child_workdir_for, delegation::SPAWNER_ENV, formation::FORMATION_ID_ENV, launch_child_capsule,
    ChildLaunchRequest, FormationId, LaunchedChild, ParentWorkdirSource, SnapshotLimits,
    SpawnApproval, Spawner, WorkdirRoot, HOST_WARNINGS_REPORTED_ENV, PARENT_WORKDIR_SNAPSHOT_FLAG,
    SPAWNER_LIFELINE_ENV,
};
use common::{component, files_under, find_in_files, mur_binary, Roost, ScriptedServer};
use serde_json::{json, Value};
use tempfile::TempDir;

/// The capsule the parent session is registered as. Its manifest is the ceiling every child in
/// this suite is refereed against.
const PARENT_CAPSULE: &str = "local/suite-parent";
const PARENT_SESSION: &str = "ses_suite000000000000000000000parent";
const DRIVER: &str = "local/murmur-driver-anthropic";
const DRIVER_VERSION: &str = "0.1.4";

/// Two host variables the parent holds and each child declares at most one of.
const A_ONLY: &str = "WORKER_A_ONLY";
const B_ONLY: &str = "WORKER_B_ONLY";

struct Suite {
    roost: Roost,
    credential: String,
    /// Kept alive for the life of the process: `HOME` points inside it.
    _home: TempDir,
    /// Held for the life of the process: the agent children address it as their inference
    /// endpoint.
    _inference: ScriptedServer,
}

/// The one daemon, registry and `HOME` every case shares.
fn suite() -> &'static Suite {
    static SUITE: OnceLock<Suite> = OnceLock::new();
    SUITE.get_or_init(|| {
        let home = TempDir::new().unwrap();
        std::env::set_var("HOME", home.path());
        std::env::set_var(A_ONLY, "a-only-value");
        std::env::set_var(B_ONLY, "b-only-value");
        std::env::set_var("GITHUB_TOKEN", "parent-github-token");

        let inference = ScriptedServer::always_replying("done");
        let registry_path = home.path().join(".murmur").join("artifacts");
        std::fs::create_dir_all(&registry_path).unwrap();

        let roost = Roost::start_at(&registry_path, &[PARENT_CAPSULE]);
        common::publish_driver(
            &registry_path,
            DRIVER,
            DRIVER_VERSION,
            &common::fixture_path("drivers/anthropic/driver/murmur-driver-anthropic.wasm"),
        );

        // The parent's ceiling: every child below is within it on every axis. `containment` is
        // left undeclared, so `child-sealed` raising the floor is a rise rather than an escape.
        common::publish_capsule(
            &registry_path,
            PARENT_CAPSULE,
            "0.1.0",
            &format!(
                "artifacts: []\ncapabilities:\n  \
                 network:\n    allow: [{endpoint}]\n  \
                 env:\n    allow: [{A_ONLY}, {B_ONLY}, MURMUR_TEST_ALLOWED_VAR]\n  \
                 spawn:\n    allow: [local/child-agent, local/child-a, local/child-b, \
                 local/child-escape, local/child-sealed, local/child-leak, local/child-reader, \
                 local/child-repeat, local/child-handed, local/child-misreported, \
                 local/child-snapshot, local/child-small]\n",
                endpoint = inference.authority(),
            ),
            Some(&component("capsule-env-echo.wasm")),
        );

        // Script children that write into their own directory and try to leave it.
        for name in ["local/child-a", "local/child-b"] {
            let allow = match name {
                "local/child-a" => format!("  env:\n    allow: [{A_ONLY}]\n"),
                _ => format!("  env:\n    allow: [{B_ONLY}]\n"),
            };
            common::publish_capsule(
                &registry_path,
                name,
                "0.1.0",
                &format!("artifacts: []\ncapabilities:\n{allow}"),
                Some(&component("capsule-filesystem-escape.wasm")),
            );
        }
        common::publish_capsule(
            &registry_path,
            "local/child-escape",
            "0.1.0",
            "artifacts: []\n",
            Some(&component("capsule-filesystem-escape.wasm")),
        );

        // Script children that read the paths their parent named, and report what each attempt
        // was served.
        for name in [
            "local/child-reader",
            "local/child-repeat",
            "local/child-misreported",
            "local/child-snapshot",
        ] {
            common::publish_capsule(
                &registry_path,
                name,
                "0.1.0",
                "artifacts: []\n",
                Some(&component("capsule-path-reader.wasm")),
            );
        }
        // A reader whose work folder may hold 1 KiB: a copy of a larger parent folder cannot
        // start it.
        common::publish_capsule(
            &registry_path,
            "local/child-small",
            "0.1.0",
            "artifacts: []\ncapabilities:\n  resources:\n    workdir_max_bytes: 1024\n",
            Some(&component("capsule-path-reader.wasm")),
        );
        common::publish_capsule(
            &registry_path,
            "local/child-leak",
            "0.1.0",
            "artifacts: []\ncapabilities:\n  env:\n    allow: [MURMUR_TEST_ALLOWED_VAR]\n",
            Some(&component("capsule-env-echo.wasm")),
        );

        // Agent children: they bind a port and serve A2A.
        for (name, containment) in [
            ("local/child-agent", ""),
            ("local/child-sealed", "  containment: sealed\n"),
            ("local/child-handed", ""),
        ] {
            common::publish_capsule(
                &registry_path,
                name,
                "0.1.0",
                &format!(
                    "artifacts:\n  - name: {DRIVER}\n    version: {DRIVER_VERSION}\n    \
                     runtime: driver\n    gateway:\n      endpoint: {url}\n      \
                     api_key: test-key\ncapabilities:\n  network:\n    allow: [{endpoint}]\n\
                     {containment}inference:\n  transport: http\n  \
                     model: test-model\n  driver:\n    artifact: {DRIVER}\n",
                    endpoint = inference.authority(),
                    url = inference.endpoint,
                ),
                None,
            );
        }

        // The parent's own registration: this is where its credential comes from, exactly as a
        // real parent's runtime would have obtained it at launch.
        let credential = roost.register(PARENT_SESSION, PARENT_CAPSULE, "0.1.0");
        std::env::set_var(
            capsule_runtime::MUR_BINARY_ENV,
            recording_wrapper(home.path(), mur_binary()),
        );

        Suite {
            roost,
            credential,
            _home: home,
            _inference: inference,
        }
    })
}

/// A script that execs `mur` with its arguments, and first, for a launch of `child-handed` alone,
/// records into the child's directory what the process was handed: its environment
/// (`handed.env`) and its descriptors (`handed.fd`, and `handed.fdinfo` for the spawner lifeline).
///
/// `mur` makes itself non-dumpable as its first act, which closes its `/proc/<pid>/environ` and
/// `/proc/<pid>/fd` to every non-root reader, this test included. The script runs in the same
/// process before that `exec`, so what it records is exactly what the launch handed the child.
///
/// A launch of `child-misreported` never reaches `mur`: the script records its pid
/// (`misreported.pid`), prints a readiness line naming a records directory outside
/// `~/.murmur/records`, and sleeps in the same process.
///
/// Every launch, whatever it is, first appends its arguments as one line to [`LAUNCH_LOG`] in
/// `home`, so a case can show a launch started no process. Every other launch then passes straight
/// through.
fn recording_wrapper(home: &Path, mur: &Path) -> PathBuf {
    let wrapper = home.join("mur-recording-wrapper");
    std::fs::write(
        &wrapper,
        format!(
            "#!/bin/sh
             echo \"$*\" >> '{log}'
             case \" $* \" in
  *\" local/child-handed \"*)
                 env > handed.env
                 ls -l /proc/$$/fd > handed.fd
                 cat \"/proc/$$/fdinfo/$MURMUR_SPAWNER_LIFELINE\" > handed.fdinfo
                 ;;
  *\" local/child-misreported \"*)
                 echo $$ > misreported.pid
                 echo '{{\"session_id\":\"ses_misreported\",\"url\":\"\",\"records_dir\":\"/tmp/ses_misreported\"}}'
                 exec sleep 300
                 ;;
esac
             exec '{mur}' \"$@\"
",
            log = home.join(LAUNCH_LOG).display(),
            mur = mur.display()
        ),
    )
    .unwrap();
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&wrapper, std::fs::Permissions::from_mode(0o755)).unwrap();
    }
    wrapper
}

/// The file in `HOME` [`recording_wrapper`] appends every launch's arguments to.
const LAUNCH_LOG: &str = "launches.log";

/// One case's parent: a directory standing for its records directory, whose `children/` every
/// child's work folder is composed in.
struct Parent {
    workdir: TempDir,
}

impl Parent {
    fn new() -> Self {
        Self {
            workdir: TempDir::new().unwrap(),
        }
    }

    fn dir(&self) -> &Path {
        self.workdir.path()
    }

    /// `POST /spawn`: the parent's runtime asking whether it may spawn this capsule.
    fn approve(&self, name: &str) -> (SpawnApproval, Value) {
        let suite = suite();
        let answer = suite.roost.permission(&suite.credential, name, "0.1.0");
        let token = answer["approval"]
            .as_str()
            .unwrap_or_else(|| panic!("spawn was refused: {answer}"))
            .to_string();
        (SpawnApproval::new(token), answer)
    }

    fn launch(&self, name: &str, env_allow: &[&str]) -> LaunchedChild {
        self.try_launch(name, env_allow)
            .unwrap_or_else(|error| panic!("launching '{name}' failed: {error}"))
    }

    /// Launch a `capsule-path-reader` child, run `place` on the directory the runtime made for
    /// it, ask it to read `targets`, and return it alongside one outcome per target.
    ///
    /// The placing runs beside the launch rather than before it because composing the child's
    /// directory, creating it and starting the child are one call: the directory does not exist
    /// until that call is under way. The child waits for `targets.txt`, which goes in last, so
    /// what it reports is what the parent placed rather than what it raced.
    fn launch_reading(
        &self,
        name: &str,
        targets: &[String],
        place: impl FnOnce(&Path),
    ) -> (LaunchedChild, Vec<String>) {
        self.launch_reading_with(name, targets, None, place)
    }

    /// [`Parent::launch_reading`], copying `snapshot` into the child's work folder first.
    fn launch_reading_with(
        &self,
        name: &str,
        targets: &[String],
        snapshot: Option<ParentWorkdirSource>,
        place: impl FnOnce(&Path),
    ) -> (LaunchedChild, Vec<String>) {
        let children = self.dir().join("children");
        let before = dirs_under(&children);

        let child = std::thread::scope(|scope| {
            let launch = scope.spawn(|| {
                self.try_launch_with(name, &[], None, snapshot)
                    .unwrap_or_else(|error| panic!("launching '{name}' failed: {error}"))
            });
            if let Some(dir) = wait_for_new_dir(&children, &before) {
                place(&dir);
                // Renamed into place, so the child that is polling for the list never reads half
                // of one.
                std::fs::write(dir.join("targets.part"), targets.join("\n")).unwrap();
                std::fs::rename(dir.join("targets.part"), dir.join("targets.txt")).unwrap();
            }
            launch.join().unwrap()
        });

        let report = std::fs::read_to_string(child.workdir.join("out").join("result.txt"))
            .unwrap_or_else(|error| panic!("'{name}' wrote no result file: {error}"));
        let outcomes = report.lines().map(str::to_string).collect();
        (child, outcomes)
    }

    fn try_launch(
        &self,
        name: &str,
        env_allow: &[&str],
    ) -> Result<LaunchedChild, capsule_runtime::RuntimeError> {
        self.try_launch_bounded(name, env_allow, None)
    }

    /// [`Parent::try_launch`], handing the child `deadline` as its parent's.
    fn try_launch_bounded(
        &self,
        name: &str,
        env_allow: &[&str],
        deadline: Option<SessionDeadline>,
    ) -> Result<LaunchedChild, capsule_runtime::RuntimeError> {
        self.try_launch_with(name, env_allow, deadline, None)
    }

    /// [`Parent::try_launch_bounded`], copying `snapshot` into the child's work folder first.
    fn try_launch_with(
        &self,
        name: &str,
        env_allow: &[&str],
        deadline: Option<SessionDeadline>,
        parent_workdir_snapshot: Option<ParentWorkdirSource>,
    ) -> Result<LaunchedChild, capsule_runtime::RuntimeError> {
        let (grant, _) = self.approve(name);
        launch_child_capsule(ChildLaunchRequest {
            parent_records_root: capsule_runtime::WorkdirRoot::open(self.dir()).unwrap(),
            capsule_name: name.to_string(),
            capsule_version: "0.1.0".to_string(),
            grant,
            child_env_allow: env_allow.iter().map(|name| name.to_string()).collect(),
            roost_url: suite().roost.url.clone(),
            // Nothing in this suite delegates: no handle is injected and no watcher runs.
            spawner: None,
            completion_deadline: None,
            formation_id: None,
            deadline,
            parent_workdir_snapshot,
        })
    }
}

/// Wait for a predicate to hold, up to `deadline`.
fn wait_for(what: &str, timeout: Duration, mut ready: impl FnMut() -> bool) {
    let deadline = Instant::now() + timeout;
    while Instant::now() < deadline {
        if ready() {
            return;
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    panic!("timed out waiting for {what}");
}

/// The directories directly under `parent`, or none if it does not exist yet.
fn dirs_under(parent: &Path) -> Vec<PathBuf> {
    let Ok(entries) = std::fs::read_dir(parent) else {
        return Vec::new();
    };
    entries.flatten().map(|entry| entry.path()).collect()
}

/// The one directory under `parent` that was not there in `before`.
fn wait_for_new_dir(parent: &Path, before: &[PathBuf]) -> Option<PathBuf> {
    let deadline = Instant::now() + Duration::from_secs(60);
    while Instant::now() < deadline {
        if let Some(fresh) = dirs_under(parent)
            .into_iter()
            .find(|dir| !before.contains(dir))
        {
            return Some(fresh);
        }
        std::thread::sleep(Duration::from_millis(20));
    }
    None
}

/// The child's `trace.jsonl`, in the records directory its readiness line reported.
fn trace_of(child: &LaunchedChild) -> PathBuf {
    child.records_dir.join("trace.jsonl")
}

// ── 1. The happy path ─────────────────────────────────────────────────────────

/// A spawn approved by the daemon, launched by this runtime, answering A2A from a process of its
/// own — and known to the daemon as running because it registered for itself.
#[test]
fn a_parents_runtime_launches_an_approved_child_as_a_separate_process() {
    let suite = suite();
    let parent = Parent::new();

    let (_, answer) = parent.approve("local/child-agent");
    // Permission, not a process: the daemon answered with an artifact and an expiry, and nothing
    // that could address anything running.
    assert_eq!(answer["name"], "local/child-agent");
    assert_eq!(answer["version"], "0.1.0");
    assert!(answer["sha256"].as_str().unwrap().len() == 64);
    assert!(answer.get("capsule_url").is_none());
    assert!(answer.get("session_id").is_none());

    let mut child = parent.launch("local/child-agent", &[]);

    assert_ne!(child.pid(), std::process::id());
    assert!(child.capsule_url.starts_with("http://"), "{child:?}");
    assert!(child.session_id.starts_with("ses_"), "{child:?}");
    #[cfg(target_os = "linux")]
    {
        // The runtime that owns the child's Wasmtime store is a different image entirely: this
        // process is a test binary, and the child is `mur`.
        let comm = std::fs::read_to_string(format!("/proc/{}/comm", child.pid())).unwrap();
        assert_eq!(comm.trim(), "mur");
    }

    // The daemon knows it, by the session id the child minted for itself.
    assert_eq!(
        suite.roost.status(&child.session_id).as_deref(),
        Some("running"),
        "the child did not register"
    );
    assert_eq!(
        suite.roost.status_over_http(&child.session_id)["status"],
        "running"
    );

    // And it answers A2A on the url it reported.
    let response = common::request(
        "POST",
        &child.capsule_url,
        Some(
            &json!({
                "jsonrpc": "2.0",
                "id": "child-launch-1",
                "method": "SendMessage",
                "params": {"message": {"messageId": "m-1", "role": "ROLE_USER",
                                       "parts": [{"text": "hello", "mediaType": "text/plain"}]}}
            })
            .to_string(),
        ),
        &[(
            capsule_runtime::A2A_VERSION_HEADER,
            capsule_runtime::A2A_PROTOCOL_VERSION,
        )],
    )
    .expect("the child answers SendMessage");
    assert!(
        response["result"]["task"]["id"].as_str().is_some(),
        "SendMessage returned {response}"
    );

    child.shutdown().unwrap();
}

// ── 2. Two children of the same parent are independent ────────────────────────

/// Each child's environment is built from that child's own declaration, and each child's
/// directory is a sibling of the other's rather than beneath it.
#[test]
fn two_children_of_the_same_parent_are_independent() {
    let parent = Parent::new();

    let a = parent.launch("local/child-a", &[A_ONLY]);
    let b = parent.launch("local/child-b", &[B_ONLY]);

    // The environment the parent built for each, which is the environment `execve` received.
    let names = |child: &LaunchedChild| -> Vec<String> {
        child.env.iter().map(|(key, _)| key.clone()).collect()
    };
    assert!(names(&a).contains(&A_ONLY.to_string()));
    assert!(!names(&a).contains(&B_ONLY.to_string()));
    assert!(names(&b).contains(&B_ONLY.to_string()));
    assert!(!names(&b).contains(&A_ONLY.to_string()));
    // Neither was handed the parent's own environment wholesale.
    assert!(!names(&a).contains(&"GITHUB_TOKEN".to_string()));

    // The same read through the kernel's own record, on a host that allows it. `mur` marks itself
    // non-dumpable at startup (`security::harden_process_dumpable`), so this is normally
    // unreadable even to the launching parent — that is the point of the flag, and the assertion
    // above is what stands in for it.
    for (child, present, absent) in [(&a, A_ONLY, B_ONLY), (&b, B_ONLY, A_ONLY)] {
        if let Ok(raw) = std::fs::read(format!("/proc/{}/environ", child.pid())) {
            let observed = String::from_utf8_lossy(&raw);
            assert!(observed.contains(present), "{observed}");
            assert!(!observed.contains(absent), "{observed}");
        }
    }

    // Siblings, neither beneath the other, neither the parent's own.
    assert_ne!(a.workdir, b.workdir);
    assert!(!a.workdir.starts_with(&b.workdir));
    assert!(!b.workdir.starts_with(&a.workdir));
    assert_ne!(a.workdir, parent.dir());
    assert_eq!(a.workdir.parent(), b.workdir.parent());
    assert!(a.workdir.starts_with(parent.dir()));

    // Writing into one leaves the other untouched.
    std::fs::write(a.workdir.join("only-in-a.txt"), "a").unwrap();
    assert!(!b.workdir.join("only-in-a.txt").exists());

    // Each child's records are its own, named by its own session id, and nothing of the runtime's
    // is in either work folder.
    assert_ne!(a.records_dir, b.records_dir);
    assert!(a.records_dir.ends_with(&a.session_id) && a.records_dir.is_dir());
    assert!(b.records_dir.ends_with(&b.session_id) && b.records_dir.is_dir());
    for child in [&a, &b] {
        assert!(!child.workdir.join(".murmur").exists());
        assert!(child.records_dir.join("home").is_dir());
    }
    assert!(!parent.dir().join(".murmur").exists());
}

/// The environment a child that declares nothing is handed: the names the runtime owns, and
/// nothing else. An empty `capabilities.env.allow` is a complete answer rather than a hole, so a
/// list that arrives empty — which is what a child declaring none of its parent's variables
/// yields — leaves the child holding the runtime-owned names alone.
#[test]
fn a_child_declaring_no_variables_gets_only_the_runtime_owned_names() {
    let parent = Parent::new();
    let keys = |child: &LaunchedChild| -> Vec<String> {
        child.env.iter().map(|(key, _)| key.clone()).collect()
    };
    // The account names follow `HOME`, each present only when this process holds it.
    let runtime_owned = |tail: &[&str]| -> Vec<String> {
        ["PATH", "HOME"]
            .into_iter()
            .chain(
                ["LOGNAME", "USER"]
                    .into_iter()
                    .filter(|name| std::env::var(name).is_ok()),
            )
            .chain(["MURMUR_ROOST_URL", HOST_WARNINGS_REPORTED_ENV])
            .chain(tail.iter().copied())
            .map(str::to_string)
            .collect()
    };

    let plain = parent.launch("local/child-a", &[]);
    assert_eq!(keys(&plain), runtime_owned(&[SPAWNER_LIFELINE_ENV]));

    // The same launch with lineage. It reports to nobody, so the handle is the whole of what the
    // spawner adds and no watcher runs.
    let (grant, _) = parent.approve("local/child-b");
    let delegated = launch_child_capsule(ChildLaunchRequest {
        parent_records_root: capsule_runtime::WorkdirRoot::open(parent.dir()).unwrap(),
        capsule_name: "local/child-b".to_string(),
        capsule_version: "0.1.0".to_string(),
        grant,
        child_env_allow: Vec::new(),
        roost_url: suite().roost.url.clone(),
        spawner: Some(Spawner {
            session_id: PARENT_SESSION.to_string(),
            context_id: "ctx_runtime_owned_names".to_string(),
            report_to: None,
        }),
        completion_deadline: None,
        formation_id: None,
        deadline: None,
        parent_workdir_snapshot: None,
    })
    .expect("a child that declares no variables launches");
    assert_eq!(
        keys(&delegated),
        runtime_owned(&[SPAWNER_LIFELINE_ENV, SPAWNER_ENV])
    );
}

/// A parent in a formation hands its id to the child, which joins it and says so on its readiness
/// line — with a lineage and without one, since the formation is not part of the spawner handle.
/// A parent in no formation hands none, and the child reports none.
#[test]
fn a_child_joins_its_parents_formation_with_or_without_a_spawner() {
    let parent = Parent::new();
    let formation = FormationId::mint();
    let lineage = Spawner {
        session_id: PARENT_SESSION.to_string(),
        context_id: "ctx_formation".to_string(),
        report_to: None,
    };

    for (name, spawner) in [("local/child-a", Some(lineage)), ("local/child-b", None)] {
        let (grant, _) = parent.approve(name);
        let child = launch_child_capsule(ChildLaunchRequest {
            parent_records_root: capsule_runtime::WorkdirRoot::open(parent.dir()).unwrap(),
            capsule_name: name.to_string(),
            capsule_version: "0.1.0".to_string(),
            grant,
            child_env_allow: Vec::new(),
            roost_url: suite().roost.url.clone(),
            spawner,
            completion_deadline: None,
            formation_id: Some(formation.clone()),
            deadline: None,
            parent_workdir_snapshot: None,
        })
        .unwrap_or_else(|error| panic!("launching '{name}' failed: {error}"));

        assert_eq!(
            child.env.last(),
            Some(&(FORMATION_ID_ENV.to_string(), formation.as_str().to_string())),
            "{name}: the formation id is the last runtime-owned name: {:?}",
            child.env
        );
        assert_eq!(child.formation_id, Some(formation.clone()), "{name}");
    }

    let outside = parent.launch("local/child-a", &[]);
    assert!(
        !outside.env.iter().any(|(key, _)| key == FORMATION_ID_ENV),
        "{:?}",
        outside.env
    );
    assert_eq!(outside.formation_id, None);
}

/// A launch handed its parent's deadline passes it on as one moment, and the agent child runs
/// under it, inherited; a launch handed none passes nothing.
#[test]
fn a_child_is_handed_its_parents_deadline_as_its_own() {
    let parent = Parent::new();
    let deadline = SessionDeadline::after(Instant::now(), std::time::SystemTime::now(), 300);
    let mut child = parent
        .try_launch_bounded("local/child-agent", &[], Some(deadline))
        .unwrap_or_else(|error| panic!("launching the bounded child failed: {error}"));
    let at = child
        .argv
        .iter()
        .position(|arg| arg == INHERITED_DEADLINE_FLAG)
        .unwrap_or_else(|| panic!("no {INHERITED_DEADLINE_FLAG}: {:?}", child.argv));
    assert_eq!(
        child.argv.get(at + 1),
        Some(&deadline.at_unix_millis().to_string()),
        "{:?}",
        child.argv
    );
    let trace = trace_of(&child);
    wait_for("the child's session_start", Duration::from_secs(30), || {
        std::fs::read_to_string(&trace).is_ok_and(|text| text.contains("session_start"))
    });
    let start: Value = serde_json::from_str(
        std::fs::read_to_string(&trace)
            .unwrap()
            .lines()
            .next()
            .unwrap(),
    )
    .unwrap();
    assert_eq!(start["deadline"]["at"], deadline.at_rfc3339(), "{start}");
    assert_eq!(start["deadline"]["source"], "inherited", "{start}");
    child.shutdown().unwrap();

    let unbounded = parent.launch("local/child-a", &[]);
    assert!(
        !unbounded
            .argv
            .iter()
            .any(|arg| arg == INHERITED_DEADLINE_FLAG),
        "{:?}",
        unbounded.argv
    );
    // Every launch is contained: the unconfined stance never spreads to a child.
    for argv in [&child.argv, &unbounded.argv] {
        assert!(
            argv.iter().any(|arg| arg == CONTAINED_LAUNCH_FLAG),
            "{argv:?}"
        );
        assert!(!argv.iter().any(|arg| arg == "--unconfined"), "{argv:?}");
    }
}

// ── 3. A child cannot reach out of its own directory ──────────────────────────

/// The child's only preopen is the directory the parent made for it. A file the parent placed
/// there is inside it; the parent's own workdir root and a sibling's directory are not, and the
/// attempt to traverse out is refused rather than served.
#[test]
fn a_child_cannot_reach_its_parents_workdir_or_its_siblings() {
    let parent = Parent::new();
    std::fs::write(parent.dir().join("parent-secret.txt"), "parent only").unwrap();

    let sibling = parent.launch("local/child-b", &[B_ONLY]);
    std::fs::write(sibling.workdir.join("sibling-secret.txt"), "sibling only").unwrap();

    // The escaping child gets its own input from the parent, placed in its own directory before it
    // is asked to do anything with it.
    let escaping_dir = child_workdir_for(parent.dir(), "local/child-escape");
    let child = parent.launch("local/child-escape", &[]);

    // Its attempt to reach `../../outside.txt` — out of its preopen and into the parent's own
    // directory above `children/` — was refused by the WASI layer rather than served.
    assert_eq!(
        std::fs::read_to_string(child.workdir.join("out").join("result.txt")).unwrap(),
        "blocked"
    );
    // Nothing it tried to write landed anywhere: not in the parent's root, not in `children/`, not
    // in the sibling's directory.
    assert!(!parent.dir().join("outside.txt").exists());
    assert!(!parent.dir().join("children").join("outside.txt").exists());
    assert!(!sibling.workdir.join("outside.txt").exists());
    assert_eq!(
        std::fs::read_to_string(sibling.workdir.join("sibling-secret.txt")).unwrap(),
        "sibling only"
    );
    assert_eq!(
        std::fs::read_to_string(parent.dir().join("parent-secret.txt")).unwrap(),
        "parent only"
    );

    // The child's own directory is the preopen it does reach: its output landed there, written
    // from inside the guest, while the traversal above was refused.
    assert_eq!(child.workdir.parent(), escaping_dir.parent());
    assert!(child.workdir.join("out").join("result.txt").is_file());

    // And reads run the same way round. A third child is pointed at existing files: its own input,
    // the parent's secret and the sibling's. Only the one in its own directory is served.
    let sibling_dir = sibling.workdir.file_name().unwrap().to_string_lossy();
    let mut targets = vec![
        "./input.txt".to_string(),
        "../../parent-secret.txt".to_string(),
        format!("../{sibling_dir}/sibling-secret.txt"),
    ];
    let mut expected = vec!["served placed by the parent", "blocked", "blocked"];

    // Symlinks are the other way out of a preopen, and a trailing slash on one is the shape
    // RUSTSEC-2026-0269 escaped through, so both forms are attempted alongside the plain
    // traversals above.
    #[cfg(unix)]
    {
        targets.extend([
            "secret-link".to_string(),
            "secret-link/".to_string(),
            "root-link/parent-secret.txt".to_string(),
        ]);
        expected.extend(["blocked", "blocked", "blocked"]);
    }

    let (reader, outcomes) = parent.launch_reading("local/child-reader", &targets, |dir| {
        std::fs::write(dir.join("input.txt"), "placed by the parent").unwrap();
        #[cfg(unix)]
        {
            std::os::unix::fs::symlink(
                parent.dir().join("parent-secret.txt"),
                dir.join("secret-link"),
            )
            .unwrap();
            std::os::unix::fs::symlink(parent.dir(), dir.join("root-link")).unwrap();
        }
    });
    assert_eq!(outcomes, expected, "reader in {}", reader.workdir.display());

    // Every refused path names content that is really there: resolved from the reader's own
    // directory, or followed from the links planted in it, they are the parent's secret and the
    // sibling's. So `blocked` is a refusal and not a miss.
    assert!(reader.workdir.join("../../parent-secret.txt").is_file());
    assert!(reader
        .workdir
        .join(format!("../{sibling_dir}/sibling-secret.txt"))
        .is_file());
    #[cfg(unix)]
    {
        assert_eq!(
            std::fs::read_to_string(reader.workdir.join("secret-link")).unwrap(),
            "parent only"
        );
        assert_eq!(
            std::fs::read_to_string(reader.workdir.join("root-link").join("parent-secret.txt"))
                .unwrap(),
            "parent only"
        );
    }
}

// ── 4. A directory per delegation ─────────────────────────────────────────────

/// Delegating the same capsule name and version twice yields two directories, both beneath the
/// parent's accessible workdir, both owner-only, and the second does not disturb the first.
#[test]
fn each_delegation_gets_a_directory_of_its_own() {
    let parent = Parent::new();

    let (first, first_read) =
        parent.launch_reading("local/child-repeat", &["./input.txt".to_string()], |dir| {
            std::fs::write(dir.join("input.txt"), "first only").unwrap();
        });
    std::fs::write(first.workdir.join("placed-by-the-parent.txt"), "kept").unwrap();

    let first_dir = first
        .workdir
        .file_name()
        .unwrap()
        .to_string_lossy()
        .to_string();
    let (second, second_read) = parent.launch_reading(
        "local/child-repeat",
        &[
            "./input.txt".to_string(),
            format!("../{first_dir}/input.txt"),
        ],
        |dir| {
            std::fs::write(dir.join("input.txt"), "second only").unwrap();
        },
    );

    // What the parent put in each directory belongs to that delegation alone: each child was
    // served its own input, and the second was refused the first's.
    assert_eq!(first_read, ["served first only"]);
    assert_eq!(second_read, ["served second only", "blocked"]);
    // The first child's input is still on disk where the second was pointed at it, so the second
    // was refused a file rather than sent after one that had gone.
    assert!(second
        .workdir
        .join(format!("../{first_dir}/input.txt"))
        .is_file());

    assert_ne!(first.workdir, second.workdir);
    assert_ne!(first.workdir, parent.dir());
    assert_ne!(second.workdir, parent.dir());
    assert!(first.workdir.starts_with(parent.dir()));
    assert!(second.workdir.starts_with(parent.dir()));
    assert_eq!(
        std::fs::read_to_string(first.workdir.join("placed-by-the-parent.txt")).unwrap(),
        "kept"
    );

    // The naming rule, and its permissions.
    for dir in [&first.workdir, &second.workdir] {
        assert_eq!(dir.parent(), Some(parent.dir().join("children").as_path()));
        let name = dir.file_name().unwrap().to_string_lossy();
        assert_eq!(name.len(), "child-repeat-".len() + 16);
        assert!(name.starts_with("child-repeat-"));
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            assert_eq!(
                std::fs::metadata(dir).unwrap().permissions().mode() & 0o777,
                0o700,
                "{}",
                dir.display()
            );
        }
    }
}

// ── 5. No token reaches anything a capsule can read ───────────────────────────

/// The credential and the approval appear in no file, no trace line, no guest environment and no
/// error string.
#[test]
fn neither_token_leaks_into_a_file_a_trace_an_environment_or_an_error() {
    let suite = suite();
    let parent = Parent::new();
    let (grant, _) = parent.approve("local/child-leak");
    let approval = grant.expose().to_string();

    let child = launch_child_capsule(ChildLaunchRequest {
        parent_records_root: capsule_runtime::WorkdirRoot::open(parent.dir()).unwrap(),
        capsule_name: "local/child-leak".to_string(),
        capsule_version: "0.1.0".to_string(),
        grant,
        child_env_allow: vec!["MURMUR_TEST_ALLOWED_VAR".to_string()],
        roost_url: suite.roost.url.clone(),
        spawner: None,
        completion_deadline: None,
        formation_id: None,
        deadline: None,
        parent_workdir_snapshot: None,
    })
    .expect("the leak fixture launches");

    // Not in the argument vector, and not in the environment: the grant travelled on standard
    // input, which nothing in `/proc` exposes.
    assert!(!child
        .argv
        .iter()
        .any(|argument| argument.contains(&approval)));
    assert!(!child
        .env
        .iter()
        .any(|(key, value)| key.contains(&approval) || value.contains(&approval)));
    assert!(!child
        .env
        .iter()
        .any(|(_, value)| value.contains(&suite.credential)));

    // Not in any file either process wrote.
    for root in [parent.dir(), child.workdir.as_path()] {
        for token in [&approval, &suite.credential] {
            assert_eq!(
                find_in_files(root, token),
                None,
                "a token appears under {}",
                root.display()
            );
        }
    }
    assert!(
        files_under(&child.workdir)
            .iter()
            .any(|path| path.ends_with("out/result.txt")),
        "the fixture wrote nothing, so this proves nothing"
    );

    // The guest's own view of its environment: the fixture reports what it saw for the two names
    // it probes, and the host's `GITHUB_TOKEN`, which the child does not declare, is absent.
    let observed = std::fs::read_to_string(child.workdir.join("out").join("result.txt")).unwrap();
    assert!(observed.contains("GITHUB_TOKEN=absent"), "{observed}");
    assert!(!observed.contains(&approval), "{observed}");
    assert!(!observed.contains(&suite.credential), "{observed}");

    // And not in a refusal: registering with a daemon that is not there names the daemon and the
    // reason, and nothing else.
    let error = launch_child_capsule(ChildLaunchRequest {
        parent_records_root: capsule_runtime::WorkdirRoot::open(parent.dir()).unwrap(),
        capsule_name: "local/child-leak".to_string(),
        capsule_version: "0.1.0".to_string(),
        grant: SpawnApproval::new(approval.clone()),
        child_env_allow: Vec::new(),
        roost_url: "http://127.0.0.1:1".to_string(),
        spawner: None,
        completion_deadline: None,
        formation_id: None,
        deadline: None,
        parent_workdir_snapshot: None,
    })
    .expect_err("a child cannot register with a daemon that is not listening");
    let message = error.to_string();
    assert!(!message.contains(&approval), "{message}");
    assert!(!message.contains(&suite.credential), "{message}");
}

/// The registration refusal a capsule that declares spawn meets when the daemon is unreachable
/// names the daemon and the reason, and carries no token.
#[test]
fn a_registration_failure_names_the_daemon_and_the_reason() {
    let error = capsule_runtime::register_session(
        "http://127.0.0.1:1",
        "ses_unreachable",
        "local/child-leak",
        "0.1.0",
        Some(&SpawnApproval::new("msa1.secret.token".to_string())),
    )
    .expect_err("nothing is listening on port 1");

    let message = error.to_string();
    assert!(message.contains("http://127.0.0.1:1"), "{message}");
    assert!(message.contains("mur-roost"), "{message}");
    assert!(message.contains("capabilities.spawn.allow"), "{message}");
    assert!(!message.contains("msa1.secret.token"), "{message}");
}

// ── 6. Containment, judged in the child's own process ─────────────────────────

/// A child declaring the strongest floor is judged by the host it actually runs on — which is its
/// own process, probed at its own launch. Both outcomes are asserted; neither is skipped.
#[test]
fn a_sealed_child_either_achieves_its_floor_or_is_refused() {
    let parent = Parent::new();
    let host_reaches_sealed =
        capsule_runtime::detect_containment_decision(murmur_artifact::ContainmentClass::Sealed)
            .0
            .achieved()
            == murmur_artifact::ContainmentClass::Sealed;

    match parent.try_launch("local/child-sealed", &[]) {
        Ok(mut child) => {
            assert!(
                host_reaches_sealed,
                "a sealed child launched on a host that cannot provide sealed"
            );
            let trace = trace_of(&child);
            wait_for("the child's trace", Duration::from_secs(30), || {
                trace.is_file()
                    && std::fs::read_to_string(&trace)
                        .unwrap()
                        .contains("session_start")
            });
            let session_start: Value = std::fs::read_to_string(&trace)
                .unwrap()
                .lines()
                .map(|line| serde_json::from_str::<Value>(line).unwrap())
                .find(|event| event["event_type"] == "session_start")
                .expect("the child's trace records a session_start");
            assert_eq!(
                session_start["effective_grants"]["achieved_containment"], "sealed",
                "{session_start}"
            );
            assert_eq!(
                session_start["effective_grants"]["enforcement_tier"],
                "mountns+pivot_root+landlock+seccomp",
                "{session_start}"
            );
            child.shutdown().unwrap();
        }
        Err(error) => {
            assert!(
                !host_reaches_sealed,
                "a sealed child was refused on a host that provides sealed: {error}"
            );
            let message = error.to_string();
            assert!(
                message.contains("declared containment class 'sealed'"),
                "{message}"
            );
            assert!(
                message.contains("is not achievable on this host"),
                "{message}"
            );
        }
    }
}

/// A child declaring no floor stages through `mur run` like any session, so its trace names the
/// class and tier its own decision reached — on a sealed-capable host `scoped` on
/// `landlock+seccomp`, never the `sealed` the host could back for a capsule asking for it.
#[test]
fn an_undeclared_child_reports_the_tier_its_commands_run_on() {
    let parent = Parent::new();
    let (expected, _) =
        capsule_runtime::detect_containment_decision(murmur_artifact::ContainmentClass::Advisory);

    let mut child = parent
        .try_launch("local/child-agent", &[])
        .expect("an advisory child launches on every host");
    let trace = trace_of(&child);
    wait_for("the child's trace", Duration::from_secs(30), || {
        trace.is_file()
            && std::fs::read_to_string(&trace)
                .unwrap()
                .contains("session_start")
    });
    let session_start: Value = std::fs::read_to_string(&trace)
        .unwrap()
        .lines()
        .map(|line| serde_json::from_str::<Value>(line).unwrap())
        .find(|event| event["event_type"] == "session_start")
        .expect("the child's trace records a session_start");
    let grants = &session_start["effective_grants"];
    assert_eq!(
        session_start["containment_achieved"],
        expected.achieved().to_string(),
        "{session_start}"
    );
    assert_eq!(
        grants["achieved_containment"],
        expected.achieved().to_string(),
        "{session_start}"
    );
    assert_eq!(
        grants["enforcement_tier"],
        expected.enforcement_tier(),
        "{session_start}"
    );
    assert_eq!(
        grants["host_containment"],
        expected.host_containment().to_string(),
        "{session_start}"
    );
    assert_ne!(grants["achieved_containment"], "sealed", "{session_start}");
    child.shutdown().unwrap();
}

// ── 7. The naming rule ────────────────────────────────────────────────────────

#[test]
fn the_child_directory_naming_rule_is_stable_and_unique() {
    let parent = TempDir::new().unwrap();
    let first = child_workdir_for(parent.path(), "worker");
    let second = child_workdir_for(parent.path(), "worker");

    assert_ne!(first, second);
    assert_eq!(
        first.parent(),
        Some(parent.path().join("children").as_path())
    );
    let suffix = first
        .file_name()
        .unwrap()
        .to_string_lossy()
        .strip_prefix("worker-")
        .unwrap()
        .to_string();
    assert_eq!(suffix.len(), 16);
    assert!(suffix.chars().all(|c| c.is_ascii_hexdigit()));
}

// ── 8. A parent that returns early leaves no orphan ───────────────────────────

/// `LaunchedChild`'s `Drop` terminates and reaps, so a parent that returns — or panics — before
/// shutting a child down leaves no process holding a port and a directory.
#[test]
fn dropping_a_launched_child_terminates_and_reaps_it() {
    let parent = Parent::new();
    let pid = {
        let child = parent.launch("local/child-agent", &[]);
        let pid = child.pid();
        assert!(process_is_alive(pid));
        pid
    };

    wait_for(
        "the dropped child to be reaped",
        Duration::from_secs(10),
        || !process_is_alive(pid),
    );
}

/// There is no way out of that: a launched child lives exactly as long as the handle that
/// launched it. Held, it keeps running and is not seen exited; dropped, it is ended and reaped.
/// A delegating task holds the handle until the outcome is delivered or the task ends the child.
#[test]
fn a_launched_child_lives_exactly_as_long_as_its_handle() {
    let parent = Parent::new();
    let child = parent.launch("local/child-agent", &[]);
    let pid = child.pid();
    assert!(process_is_alive(pid));

    // Long enough that a child which outlived nothing, or died on its own, would be observed.
    std::thread::sleep(Duration::from_millis(500));
    assert!(process_is_alive(pid), "a held child keeps running");
    assert!(
        !child.has_exited(),
        "a held, running child is not seen exited"
    );

    drop(child);
    wait_for(
        "the child to end with its handle",
        Duration::from_secs(10),
        || !process_is_alive(pid),
    );
}

/// A child whose readiness line names a records directory outside `~/.murmur/records` is refused,
/// and the launch that refuses it ends and reaps it before returning: there is no handle left to
/// do so later.
#[cfg(target_os = "linux")]
#[test]
fn a_child_refused_at_its_readiness_line_is_ended_and_reaped() {
    let parent = Parent::new();
    let error = parent
        .try_launch("local/child-misreported", &[])
        .expect_err("a records_dir outside ~/.murmur/records refuses the launch");
    let message = error.to_string();
    assert!(
        message.contains("named /tmp/ses_misreported as its records directory"),
        "{message}"
    );

    let children = parent.dir().join("children");
    let dir = dirs_under(&children)
        .into_iter()
        .next()
        .expect("the launch created the child's directory");
    let pid: u32 = std::fs::read_to_string(dir.join("misreported.pid"))
        .unwrap()
        .trim()
        .parse()
        .unwrap();
    assert!(
        !process_is_alive(pid),
        "the refused child is still in the process table"
    );
}

/// What `/proc/<pid>/fd/<fd>` points at, e.g. `pipe:[1234]`.
#[cfg(target_os = "linux")]
fn fd_target(pid: u32, fd: &str) -> Option<String> {
    std::fs::read_link(format!("/proc/{pid}/fd/{fd}"))
        .ok()
        .map(|target| target.display().to_string())
}

/// A delegated child is handed one descriptor of its parent's making: the read end of a spawner
/// lifeline of its own, named by `MURMUR_SPAWNER_LIFELINE` exactly once and open read-only, with
/// the pipe's write end held by this process and by nothing in the child. Neither lifeline
/// variable is copied from this process's environment, even allowlisted, and the formation id is
/// still handed on.
///
/// The child's side is read from what [`recording_wrapper`] wrote at its `exec`.
#[cfg(target_os = "linux")]
#[test]
fn a_child_holds_its_own_spawner_lifeline_read_end_and_no_formation_lifeline() {
    use capsule_runtime::FORMATION_LIFELINE_ENV;

    let parent = Parent::new();
    let formation = FormationId::mint();
    let (grant, _) = parent.approve("local/child-handed");
    // Decoys in this process's own environment, allowlisted for the child: neither may reach it.
    std::env::set_var(FORMATION_LIFELINE_ENV, "4097");
    std::env::set_var(SPAWNER_LIFELINE_ENV, "4098");
    let launched = launch_child_capsule(ChildLaunchRequest {
        parent_records_root: capsule_runtime::WorkdirRoot::open(parent.dir()).unwrap(),
        capsule_name: "local/child-handed".to_string(),
        capsule_version: "0.1.0".to_string(),
        grant,
        child_env_allow: vec![
            FORMATION_LIFELINE_ENV.to_string(),
            SPAWNER_LIFELINE_ENV.to_string(),
        ],
        roost_url: suite().roost.url.clone(),
        spawner: None,
        completion_deadline: None,
        formation_id: Some(formation.clone()),
        deadline: None,
        parent_workdir_snapshot: None,
    });
    std::env::remove_var(FORMATION_LIFELINE_ENV);
    std::env::remove_var(SPAWNER_LIFELINE_ENV);
    let child = launched.unwrap_or_else(|error| panic!("launching 'child-handed' failed: {error}"));

    // What the parent built.
    let handed: Vec<&String> = child
        .env
        .iter()
        .filter(|(key, _)| key == SPAWNER_LIFELINE_ENV)
        .map(|(_, value)| value)
        .collect();
    assert_eq!(handed.len(), 1, "{:?}", child.env);
    let fd = handed[0].clone();
    assert!(fd != "4098" && fd.parse::<i32>().unwrap() >= 3, "{fd}");
    assert!(
        !child
            .env
            .iter()
            .any(|(key, _)| key == FORMATION_LIFELINE_ENV),
        "{:?}",
        child.env
    );
    assert_eq!(
        child.env.last(),
        Some(&(FORMATION_ID_ENV.to_string(), formation.as_str().to_string())),
        "{:?}",
        child.env
    );

    // What the child was handed.
    let read = |name: &str| {
        std::fs::read_to_string(child.workdir.join(name))
            .unwrap_or_else(|error| panic!("the wrapper wrote no {name}: {error}"))
    };
    let environ = read("handed.env");
    let named: Vec<&str> = environ
        .lines()
        .filter(|line| line.starts_with(&format!("{SPAWNER_LIFELINE_ENV}=")))
        .collect();
    assert_eq!(named, [format!("{SPAWNER_LIFELINE_ENV}={fd}")], "{environ}");
    assert!(
        !environ
            .lines()
            .any(|line| line.starts_with(&format!("{FORMATION_LIFELINE_ENV}="))),
        "{environ}"
    );
    assert!(
        environ
            .lines()
            .any(|line| line == format!("{FORMATION_ID_ENV}={}", formation.as_str())),
        "{environ}"
    );

    // `ls -l /proc/<pid>/fd`: `... <n> -> <target>` per descriptor.
    let descriptors: Vec<(String, String)> = read("handed.fd")
        .lines()
        .filter_map(|line| {
            let (left, target) = line.split_once(" -> ")?;
            let number = left.split_whitespace().last()?;
            Some((number.to_string(), target.trim().to_string()))
        })
        .collect();
    let pipe = descriptors
        .iter()
        .find(|(number, _)| *number == fd)
        .map(|(_, target)| target.clone())
        .unwrap_or_else(|| panic!("descriptor {fd} was not handed: {descriptors:?}"));
    assert!(pipe.starts_with("pipe:["), "{pipe}");
    let flags = read("handed.fdinfo")
        .lines()
        .find_map(|line| line.strip_prefix("flags:"))
        .map(|octal| u32::from_str_radix(octal.trim(), 8).unwrap())
        .expect("fdinfo names the descriptor's flags");
    assert_eq!(flags & 0o3, 0, "descriptor {fd} is not read-only");

    // Both ends of a pipe share its inode: the read end is the one descriptor the child was
    // handed that names it, and this process holds the write end.
    let in_child: Vec<&String> = descriptors
        .iter()
        .filter(|(_, target)| *target == pipe)
        .map(|(number, _)| number)
        .collect();
    assert_eq!(in_child, [&fd], "the child holds a write end too");
    let own = std::process::id();
    let in_parent = std::fs::read_dir("/proc/self/fd")
        .unwrap()
        .flatten()
        .filter(|entry| {
            fd_target(own, &entry.file_name().to_string_lossy()).as_deref() == Some(pipe.as_str())
        })
        .count();
    assert_eq!(
        in_parent, 1,
        "this process holds the write end, and only that"
    );
}

#[cfg(target_os = "linux")]
fn process_is_alive(pid: u32) -> bool {
    // A reaped child has no `/proc` entry at all; a zombie would still have one, which is exactly
    // what `Drop`'s `wait` exists to prevent.
    Path::new(&format!("/proc/{pid}")).exists()
}

#[cfg(not(target_os = "linux"))]
fn process_is_alive(_pid: u32) -> bool {
    false
}

/// A WASM tool of a child started in a copy of its parent's work folder is preopened into the
/// copy alone: the copied file is served with the parent's content, and the parent's real file
/// is refused.
///
/// The parent's work folder here is its records directory, as on an `advisory`-capped launch
/// where that directory lies inside it, so the copy also has to skip the child's own new
/// directory rather than recurse into it.
#[test]
fn workdir_snapshot_a_wasm_tool_reads_the_copy_and_not_the_parents_folder() {
    let parent = Parent::new();
    std::fs::write(parent.dir().join("parent-file.txt"), "from the parent").unwrap();
    let source = ParentWorkdirSource {
        root: WorkdirRoot::open(parent.dir()).unwrap(),
        limits: SnapshotLimits {
            max_bytes: 1 << 20,
            excluded: Vec::new(),
        },
    };

    let (child, outcomes) = parent.launch_reading_with(
        "local/child-snapshot",
        &[
            "./parent-file.txt".to_string(),
            "../../parent-file.txt".to_string(),
        ],
        Some(source),
        |_| {},
    );

    assert_eq!(outcomes, ["served from the parent", "blocked"]);
    let snapshot = child
        .workdir_snapshot
        .clone()
        .expect("a launch with a source reports its copy");
    assert_eq!(snapshot.source, parent.dir().display().to_string());
    assert_eq!((snapshot.files, snapshot.bytes), (1, 15), "{snapshot:?}");
    assert!(
        snapshot
            .skipped
            .iter()
            .any(|entry| child.workdir.ends_with(&entry.path)),
        "the child's own directory is skipped: {snapshot:?}"
    );
    let flag = child
        .argv
        .iter()
        .position(|arg| arg == PARENT_WORKDIR_SNAPSHOT_FLAG)
        .expect("the child is told what it received");
    let told: Value = serde_json::from_str(&child.argv[flag + 1]).unwrap();
    assert_eq!(told, serde_json::to_value(&snapshot).unwrap());
    assert_eq!(
        std::fs::read_to_string(parent.dir().join("parent-file.txt")).unwrap(),
        "from the parent"
    );
}

/// A child launched without a source gets no flag and no copy: an empty folder but for what the
/// parent placed.
#[test]
fn workdir_snapshot_absent_starts_the_child_in_an_empty_folder() {
    let parent = Parent::new();
    std::fs::write(parent.dir().join("parent-file.txt"), "from the parent").unwrap();

    let (child, outcomes) = parent.launch_reading(
        "local/child-snapshot",
        &["./parent-file.txt".to_string()],
        |_| {},
    );

    assert_eq!(outcomes, ["blocked"]);
    assert!(child.workdir_snapshot.is_none());
    assert!(!child
        .argv
        .iter()
        .any(|arg| arg == PARENT_WORKDIR_SNAPSHOT_FLAG));
}

/// A copy past the child's own `workdir_max_bytes` fails the launch before any process exists,
/// naming the child and the ceiling, and leaves no directory for it behind. Through the plane it
/// is a `failed` delegation whose output is that text, read off the child's own manifest.
#[test]
fn workdir_snapshot_past_the_childs_ceiling_starts_nothing_and_leaves_no_directory() {
    let parent = Parent::new();
    let work = TempDir::new().unwrap();
    std::fs::write(work.path().join("big.bin"), vec![7u8; 4096]).unwrap();
    let launches = || {
        std::fs::read_to_string(suite()._home.path().join(LAUNCH_LOG))
            .unwrap_or_default()
            .lines()
            .filter(|line| line.contains("local/child-small"))
            .count()
    };
    let leftovers = || {
        dirs_under(&parent.dir().join("children"))
            .into_iter()
            .filter(|dir| {
                dir.file_name()
                    .is_some_and(|name| name.to_string_lossy().starts_with("child-small-"))
            })
            .collect::<Vec<_>>()
    };

    let error = parent
        .try_launch_with(
            "local/child-small",
            &[],
            None,
            Some(ParentWorkdirSource {
                root: WorkdirRoot::open(work.path()).unwrap(),
                limits: SnapshotLimits {
                    max_bytes: 1024,
                    excluded: Vec::new(),
                },
            }),
        )
        .expect_err("a copy past the ceiling starts no child");
    let message = error.to_string();
    for part in [
        "local/child-small",
        "1024",
        "capabilities.resources.workdir_max_bytes",
    ] {
        assert!(message.contains(part), "{part} missing from: {message}");
    }
    assert_eq!(launches(), 0, "no process was started");
    assert_eq!(leftovers(), Vec::<PathBuf>::new());

    // The same refusal through the plane, which reads the 1024 off the child's manifest.
    let suite = suite();
    let registry = suite._home.path().join(".murmur").join("artifacts");
    let records = WorkdirRoot::open(parent.dir()).unwrap();
    let plane = capsule_runtime::delegation_plane::DelegationPlane::new(
        suite.roost.url.clone(),
        capsule_runtime::SpawnCredential::new(suite.credential.clone()),
        records.clone(),
        PARENT_SESSION.to_string(),
        Duration::from_secs(60),
        std::sync::Arc::new(murmur_artifact::LocalRegistry::new(&registry)),
        Vec::new(),
        None,
    )
    .snapshotting_workdir(
        capsule_runtime::delegation_plane::WorkdirSnapshotGrant::new(
            vec!["local/child-small".to_string()],
            WorkdirRoot::open(work.path()).unwrap(),
            &[],
        ),
    );
    let result = plane.delegate(
        &capsule_runtime::delegation_plane::DelegationRequest {
            capsule: "local/child-small".to_string(),
            version: "0.1.0".to_string(),
            task: "read nothing".to_string(),
        },
        &capsule_runtime::delegation_plane::DelegationOrigin::default(),
        &records,
    );
    assert_eq!(
        result.status,
        capsule_runtime::delegation_plane::DelegationStatus::Failed,
        "{}",
        result.output
    );
    assert!(
        result.output.contains("1024")
            && result
                .output
                .contains("capabilities.resources.workdir_max_bytes"),
        "{}",
        result.output
    );
    assert_eq!(
        launches(),
        0,
        "no process was started through the plane either"
    );
    assert_eq!(leftovers(), Vec::<PathBuf>::new());
}