capsule-runtime 0.9.0

//! What a delegated child's ending does to the parent that launched it.
//!
//! Every case here runs the real thing: `mur-roost` on a loopback port over a real registry, a
//! real parent capsule as its own `mur run --json` process, and children launched as processes of
//! this one. Nothing is stubbed — what a test observes is what an operator observes, in the
//! parent's own `trace.jsonl` and in the child's own `completion.json`.
//!
//! The children are launched by this test process, not by a task of the parent, so no task in the
//! parent is waiting for any of them: the parent's door refuses every completion here with
//! `-31002` (`COMPLETION_NOT_AWAITED`), and the child records that refusal. Delivery into a
//! delegating task is exercised end to end by `murmur-cli`'s `delegation` and
//! `spawner_lifeline_kill` suites, where the parent's own model makes the delegation.
//!
//! 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
//! process and its own workdir.
//!
//! Every case needs a host that can isolate a capsule. The parent declares
//! `capabilities.spawn.allow`, so its launch is refused with `E-CAP-005` on a host that cannot
//! give a subprocess tree its own network namespace, and with `E-RUN-012` on one that cannot
//! delegate a cgroup v2 scope — before anything a case here observes has happened. Each case
//! stands down through [`capsule_runtime::skip_without_host_support`] rather than failing, which
//! is what a containerised runner under the AppArmor unprivileged-userns restriction hits.

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

use std::io::{BufRead, BufReader};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};

use capsule_runtime::delegation::{DelegationStatus, Reporter, SpawnerHandle, SPAWNER_ENV};
use capsule_runtime::{
    launch_child_capsule, ChildLaunchRequest, CompletionAddress, LaunchedChild, SpawnApproval,
    Spawner,
};
use common::{component, mur_binary, Roost, ScriptedServer};
use serde_json::Value;
use tempfile::TempDir;

/// The capsule every parent process in this suite runs. Its manifest is the ceiling each child is
/// refereed against, and it accepts a queue of tasks and sleeps between them, which keeps its door
/// answering for as long as a case needs it.
const PARENT_CAPSULE: &str = "local/completion-parent";
/// The session the suite registers for itself, to obtain the credential `POST /spawn` wants.
const APPROVER_SESSION: &str = "ses_completion00000000000approver";
const DRIVER: &str = "local/murmur-driver-anthropic";
const DRIVER_VERSION: &str = "0.1.4";

/// The variable the `worker` capsule echoes into its result file. Set once for the process, so a
/// child that writes it into `out/result.txt` wrote something no runtime composed.
const MARKER_VAR: &str = "MURMUR_TEST_ALLOWED_VAR";
const MARKER: &str = "child-result-marker-91f3";

/// How much of a parent's stderr is kept, to quote back when it never reports a launch line.
const PARENT_STDERR_TAIL_LINES: usize = 20;

struct Suite {
    roost: Roost,
    /// The credential the suite presents to `POST /spawn`. The daemon judges a spawn against the
    /// envelope of the session the credential names, which is a different question from where the
    /// completion is addressed — that is the spawner handle's business.
    credential: String,
    home: TempDir,
    /// Held for the life of the process: every agent capsule here addresses it as its inference
    /// endpoint.
    _inference: ScriptedServer,
    /// The inference endpoint of `failing-worker`: it answers a body cut off mid-JSON, which the
    /// anthropic driver cannot parse.
    _failing_inference: ScriptedServer,
}

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(MARKER_VAR, MARKER);

        let inference = ScriptedServer::always_replying("parent turn done");
        let failing_inference =
            ScriptedServer::always_answering(r#"{"content":[{"type":"text","text":"hel"#);
        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 covers every child below on every axis, and its lifecycle keeps it
        // alive between tasks — a parent that exited after one task would have nowhere for a
        // completion to land.
        common::publish_capsule(
            &registry_path,
            PARENT_CAPSULE,
            "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}, {failing}]\n  \
                 env:\n    allow: [{MARKER_VAR}]\n  \
                 spawn:\n    allow: [local/worker, local/waiting-worker, local/slow-worker, \
                 local/failing-worker]\n\
                 lifecycle:\n  task_acceptance: queue\n  after_task: sleep\n\
                 trace:\n  capture: content\n\
                 inference:\n  transport: http\n  model: test-model\n  \
                 driver:\n    artifact: {DRIVER}\n",
                endpoint = inference.authority(),
                failing = failing_inference.authority(),
                url = inference.endpoint,
            ),
            None,
        );

        // An agent child whose every inference call fails, and which exits after its one task:
        // the session runs and shuts down cleanly, and its task did not complete.
        common::publish_capsule(
            &registry_path,
            "local/failing-worker",
            "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\n\
                 capabilities:\n  network:\n    allow: [{endpoint}]\n\
                 lifecycle:\n  task_acceptance: queue\n  after_task: exit\n\
                 inference:\n  transport: http\n  model: test-model\n  \
                 driver:\n    artifact: {DRIVER}\n",
                endpoint = failing_inference.authority(),
                url = failing_inference.endpoint,
            ),
            None,
        );

        // The child of scenario 2, 6b and 7: it echoes the marker into `out/result.txt` and
        // exits, so the result is something only the child could have written.
        common::publish_capsule(
            &registry_path,
            "local/worker",
            "0.1.0",
            &format!("artifacts: []\ncapabilities:\n  env:\n    allow: [{MARKER_VAR}]\n"),
            Some(&component("capsule-env-echo.wasm")),
        );

        // A child that does not finish until its parent puts `targets.txt` in its directory,
        // which is what makes "kill the parent while the child is still working" a thing a test
        // can arrange rather than race.
        common::publish_capsule(
            &registry_path,
            "local/waiting-worker",
            "0.1.0",
            "artifacts: []\n",
            Some(&component("capsule-path-reader.wasm")),
        );

        // An agent child: it binds a port and waits, so it is still running when the launch
        // returns and can be killed. `stderr_marker` is an unrecognized top-level key, so `mur run`
        // writes one predictable `W-SEC-019` line to its stderr first — the line a crash quotes
        // back.
        common::publish_capsule(
            &registry_path,
            "local/slow-worker",
            "0.1.0",
            &format!(
                "stderr_marker: true\nartifacts:\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}]\nlifecycle:\n  task_acceptance: queue\n  after_task: sleep\n\
                 inference:\n  transport: http\n  model: test-model\n  \
                 driver:\n    artifact: {DRIVER}\n",
                endpoint = inference.authority(),
                url = inference.endpoint,
            ),
            None,
        );

        let credential = roost.register(APPROVER_SESSION, PARENT_CAPSULE, "0.1.0");
        std::env::set_var(capsule_runtime::MUR_BINARY_ENV, mur_binary());

        Suite {
            roost,
            credential,
            home,
            _inference: inference,
            _failing_inference: failing_inference,
        }
    })
}

// ── A real parent capsule, as its own process ─────────────────────────────────

/// One `mur run --json` process running [`PARENT_CAPSULE`], with the address and session id a
/// completion has to name to reach it.
struct RunningParent {
    process: Option<Child>,
    workdir: TempDir,
    session_id: String,
    url: String,
}

impl RunningParent {
    fn start() -> Self {
        let suite = suite();
        let workdir = TempDir::new().unwrap();
        let mut process = Command::new(mur_binary())
            .args([
                "run",
                "--capsule",
                PARENT_CAPSULE,
                "--capsule-version",
                "0.1.0",
                "--workdir",
                workdir.path().to_str().unwrap(),
                "--json",
                "--no-env-file",
            ])
            .current_dir(workdir.path())
            .env_clear()
            .env("HOME", suite.home.path())
            .env("PATH", std::env::var("PATH").unwrap_or_default())
            .env("MURMUR_ROOST_URL", &suite.roost.url)
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .expect("the parent capsule process starts");

        // Retained rather than inherited: a parent that dies before it reports says why on its
        // stderr, and inheriting scatters that across every other case running at the same time.
        let stderr_tail = Arc::new(Mutex::new(Vec::<String>::new()));
        let stderr = process.stderr.take().expect("the parent's stderr is piped");
        let collector = Arc::clone(&stderr_tail);
        std::thread::spawn(move || {
            for line in BufReader::new(stderr).lines().map_while(Result::ok) {
                let mut tail = collector.lock().unwrap_or_else(|e| e.into_inner());
                if tail.len() == PARENT_STDERR_TAIL_LINES {
                    tail.remove(0);
                }
                tail.push(line);
            }
        });

        let stdout = process.stdout.take().expect("the parent reports on stdout");
        let (tx, rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let mut reader = BufReader::new(stdout);
            let mut line = String::new();
            let first = match reader.read_line(&mut line) {
                Ok(0) | Err(_) => None,
                Ok(_) => Some(line.trim_end().to_string()),
            };
            let _ = tx.send(first);
            // Keep the pipe drained, or a parent that logs after its launch line blocks.
            let mut rest = String::new();
            while matches!(reader.read_line(&mut rest), Ok(read) if read > 0) {
                rest.clear();
            }
        });
        // Three different failures, told apart: a parent that died, a parent still running that
        // never reported, and a pipe that broke. Collapsing them into one message costs the only
        // evidence there is — most usefully when the `mur` binary itself is unrunnable, which
        // fails every case here at once and says nothing about itself.
        let tail = || {
            let lines = stderr_tail.lock().unwrap_or_else(|e| e.into_inner());
            if lines.is_empty() {
                "(its stderr was empty)".to_string()
            } else {
                format!("its last stderr lines were:\n{}", lines.join("\n"))
            }
        };
        let line = match rx.recv_timeout(Duration::from_secs(180)) {
            Ok(Some(line)) => line,
            Ok(None) => {
                let status = match process.wait() {
                    Ok(status) => status.to_string(),
                    Err(error) => format!("its status could not be read: {error}"),
                };
                panic!(
                    "the parent exited without printing its --json launch line ({status}); {}",
                    tail()
                )
            }
            Err(_) => panic!(
                "the parent printed no --json launch line within 180s and is still running; {}",
                tail()
            ),
        };
        let report: Value = serde_json::from_str(&line).unwrap_or_else(|error| {
            panic!("the parent's launch line did not parse: {error}: {line}")
        });
        let session_id = report["session_id"]
            .as_str()
            .expect("the launch line carries a session id")
            .to_string();
        let url = report["url"]
            .as_str()
            .filter(|url| !url.is_empty())
            .map(|url| format!("http://{url}"))
            .expect("the launch line carries an address");

        Self {
            process: Some(process),
            workdir,
            session_id,
            url,
        }
    }

    /// Where the parent's completions are sent.
    fn spawner(&self) -> Spawner {
        Spawner {
            session_id: self.session_id.clone(),
            context_id: format!("ctx_{}", uuid_hex()),
            report_to: Some(CompletionAddress {
                url: self.url.clone(),
            }),
        }
    }

    /// The parent's `trace.jsonl`, in its records directory.
    fn trace_path(&self) -> PathBuf {
        capsule_runtime::records_group(self.workdir.path())
            .expect("the suite's HOME holds ~/.murmur")
            .join(&self.session_id)
            .join("trace.jsonl")
    }

    /// Every event the parent has written so far.
    fn trace(&self) -> Vec<Value> {
        std::fs::read_to_string(self.trace_path())
            .unwrap_or_default()
            .lines()
            .filter(|line| !line.is_empty())
            .filter_map(|line| serde_json::from_str(line).ok())
            .collect()
    }

    /// Whether anything in the parent's trace names `delegation_id`, or any task started with
    /// the `completion` origin.
    fn heard_of(&self, delegation_id: &str) -> bool {
        self.trace().iter().any(|event| {
            event["origin"] == "completion" || event.to_string().contains(delegation_id)
        })
    }

    fn kill(&mut self) {
        if let Some(mut process) = self.process.take() {
            let _ = process.kill();
            let _ = process.wait();
        }
    }
}

impl Drop for RunningParent {
    fn drop(&mut self) {
        self.kill();
    }
}

fn uuid_hex() -> String {
    uuid::Uuid::now_v7().simple().to_string()
}

// ── Launching children ────────────────────────────────────────────────────────

/// One agent child stages at a time.
///
/// A spawn approval is valid for 60 seconds from the `POST /spawn` that granted it, and an agent
/// child spends that window staging a driver and binding a port. Two of them doing it at once on a
/// loaded machine can outlast the approval they were launched under, and the child is then refused
/// at registration rather than run unrefereed. Script children are cheap and take no slot.
fn agent_child_slot() -> std::sync::MutexGuard<'static, ()> {
    static SLOT: std::sync::Mutex<()> = std::sync::Mutex::new(());
    SLOT.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
}

/// `POST /spawn`, then a launch — the part the model-facing delegate tool will play later.
fn launch(
    parent_dir: &Path,
    name: &str,
    env_allow: &[&str],
    spawner: Option<Spawner>,
) -> LaunchedChild {
    launch_bounded(parent_dir, name, env_allow, spawner, None)
}

/// The same launch under a completion deadline: how long the watcher observes the child before
/// ending it and reporting a `terminated` outcome. `None` watches for as long as it runs.
fn launch_bounded(
    parent_dir: &Path,
    name: &str,
    env_allow: &[&str],
    spawner: Option<Spawner>,
    completion_deadline: Option<Duration>,
) -> LaunchedChild {
    let suite = suite();
    let answer = suite.roost.permission(&suite.credential, name, "0.1.0");
    let approval = answer["approval"]
        .as_str()
        .unwrap_or_else(|| panic!("spawn was refused: {answer}"))
        .to_string();

    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: SpawnApproval::new(approval),
        child_env_allow: env_allow.iter().map(|name| name.to_string()).collect(),
        roost_url: suite.roost.url.clone(),
        spawner,
        completion_deadline,
        formation_id: None,
        deadline: None,
        parent_workdir_snapshot: None,
    })
    .unwrap_or_else(|error| panic!("launching '{name}' failed: {error}"))
}

/// The child's own record of how it ended, once the delivery it describes has settled.
///
/// A reporter writes the file before it posts and rewrites it with the result, so the record is
/// briefly readable in a state that says neither that the completion arrived nor why it did not.
/// Waiting for the settled one is what stops a case reading that intermediate write.
fn wait_for_completion(child: &LaunchedChild) -> Value {
    wait_for_completion_at(&child.records_dir)
}

/// The same, named by the child's records directory.
fn wait_for_completion_at(records_dir: &Path) -> Value {
    let path = records_dir.join("completion.json");
    let deadline = Instant::now() + Duration::from_secs(90);
    loop {
        if let Ok(raw) = std::fs::read_to_string(&path) {
            if let Ok(parsed) = serde_json::from_str::<Value>(&raw) {
                if is_settled(&parsed) {
                    return parsed;
                }
            }
        }
        assert!(
            Instant::now() < deadline,
            "no settled {} appeared for the child at {}; what is there: {}",
            "completion.json",
            records_dir.display(),
            std::fs::read_to_string(&path).unwrap_or_else(|_| "nothing".to_string())
        );
        std::thread::sleep(Duration::from_millis(100));
    }
}

/// The child's own record of how it ended, once it says the parent's door refused it.
///
/// A `terminated` outcome the watcher posts is written once before the post and rewritten after
/// it, and [`is_settled`] reads any `terminated` record as settled on sight, so this waits for the
/// refusal itself.
fn wait_for_refused_completion(records_dir: &Path) -> Value {
    let path = records_dir.join("completion.json");
    let deadline = Instant::now() + Duration::from_secs(90);
    loop {
        if let Ok(raw) = std::fs::read_to_string(&path) {
            if let Ok(parsed) = serde_json::from_str::<Value>(&raw) {
                if !parsed["delivery_error"].is_null() {
                    return parsed;
                }
                if parsed["delivered"] == Value::Bool(true) {
                    panic!("no task was waiting, yet the completion was delivered: {parsed}");
                }
            }
        }
        assert!(
            Instant::now() < deadline,
            "no refused completion appeared at {}; what is there: {}",
            records_dir.display(),
            std::fs::read_to_string(&path).unwrap_or_else(|_| "nothing".to_string())
        );
        std::thread::sleep(Duration::from_millis(100));
    }
}

/// The refusal a parent's door gives a completion no task of its is waiting for: `-31002`, naming
/// the delegation.
fn assert_refused_as_unwaited(completion: &Value, delegation_id: &str) {
    assert_eq!(completion["delivered"], false, "{completion}");
    let reason = completion["delivery_error"]
        .as_str()
        .unwrap_or_else(|| panic!("the refusal is recorded: {completion}"));
    assert!(
        reason.contains(delegation_id) && reason.contains("no task"),
        "the refusal names the delegation and says no task is waiting for it: {reason}"
    );
}

/// Whether a completion record says what became of the delivery: it arrived, it was refused for a
/// named reason, or it was a `terminated` delegation, which is posted to nobody.
fn is_settled(completion: &Value) -> bool {
    completion["delivered"] == Value::Bool(true)
        || !completion["delivery_error"].is_null()
        || completion["status"] == DelegationStatus::Terminated.as_str()
}

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}");
}

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()
}

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
}

// ── 1. A child knows its spawner, from an injected value ──────────────────────

/// The handle a child holds is the one its parent composed for this launch. A child nobody
/// delegated holds none, and a child that allowlists the name still gets the injected one: the
/// runtime-owned value is applied last.
#[test]
fn a_child_knows_its_spawner_from_an_injected_value_and_not_an_inherited_one() {
    if capsule_runtime::skip_without_host_support(
        "a_child_knows_its_spawner_from_an_injected_value_and_not_an_inherited_one",
    ) {
        return;
    }
    let parent_dir = TempDir::new().unwrap();
    // The decoy the launching process itself holds. Nothing may reach a child from here. It is a
    // readable handle rather than nonsense because `SPAWNER_ENV` is process-wide, and an
    // unreadable one would refuse a launch running beside this case.
    const DECOY_HANDLE: &str =
        r#"{"session_id":"ses_decoy","context_id":"ctx_decoy","delegation_id":"dlg_decoy"}"#;
    std::env::set_var(SPAWNER_ENV, DECOY_HANDLE);

    let parent = RunningParent::start();
    let spawner = parent.spawner();

    let delegated = launch(
        parent_dir.path(),
        "local/worker",
        &[MARKER_VAR],
        Some(spawner.clone()),
    );
    let undelegated = launch(parent_dir.path(), "local/worker", &[MARKER_VAR], None);
    let allowlisting = launch(
        parent_dir.path(),
        "local/worker",
        &[MARKER_VAR, SPAWNER_ENV],
        Some(spawner.clone()),
    );

    fn injected(child: &LaunchedChild) -> Vec<&String> {
        child
            .env
            .iter()
            .filter(|(key, _)| key == SPAWNER_ENV)
            .map(|(_, value)| value)
            .collect()
    }

    let values = injected(&delegated);
    assert_eq!(values.len(), 1, "{:?}", delegated.env);
    let handle = SpawnerHandle::parse(values[0]).expect("the injected value parses as a handle");
    let address = handle.report_to.as_ref().expect("the parent wants telling");
    let expected = spawner
        .report_to
        .as_ref()
        .expect("the parent wants telling");
    assert_eq!(address.url, expected.url);
    assert_eq!(handle.session_id, parent.session_id);
    assert_eq!(handle.context_id, spawner.context_id);
    assert_eq!(
        Some(handle.delegation_id.clone()),
        delegated.delegation_id,
        "the launcher's delegation id is what was injected"
    );
    assert!(
        handle.delegation_id.starts_with("dlg_"),
        "{}",
        handle.delegation_id
    );

    assert!(
        injected(&undelegated).is_empty(),
        "a child nobody delegated holds no handle: {:?}",
        undelegated.env
    );
    assert_eq!(undelegated.delegation_id, None);

    let allowlisted = injected(&allowlisting);
    assert_eq!(allowlisted.len(), 1, "{:?}", allowlisting.env);
    assert_ne!(allowlisted[0], DECOY_HANDLE);
    assert_eq!(
        SpawnerHandle::parse(allowlisted[0])
            .expect("the injected value parses")
            .session_id,
        parent.session_id
    );

    // Two launches from one spawner report under two ids, never one.
    assert_ne!(delegated.delegation_id, allowlisting.delegation_id);
    std::env::remove_var(SPAWNER_ENV);
}

// ── 2. A finished child's completion reaches a real parent's door ─────────────

/// The child writes its outcome and posts it, and the parent's door answers. No task of the
/// parent's made this delegation, so the door refuses it with `-31002` naming the delegation, the
/// child records the refusal, and the parent starts no task and writes nothing about it. The result
/// stays in the child's own directory either way.
#[test]
fn a_finished_childs_completion_is_refused_by_a_parent_with_no_task_waiting_for_it() {
    if capsule_runtime::skip_without_host_support(
        "a_finished_childs_completion_is_refused_by_a_parent_with_no_task_waiting_for_it",
    ) {
        return;
    }
    let parent = RunningParent::start();
    let parent_dir = TempDir::new().unwrap();

    let child = launch(
        parent_dir.path(),
        "local/worker",
        &[MARKER_VAR],
        Some(parent.spawner()),
    );
    let delegation_id = child
        .delegation_id
        .clone()
        .expect("a delegated launch has an id");

    let completion = wait_for_completion(&child);
    assert_eq!(completion["status"], DelegationStatus::Ok.as_str());
    assert_eq!(completion["reported_by"], Reporter::Child.as_str());
    assert_refused_as_unwaited(&completion, &delegation_id);
    // A script child writes its own `out/result.txt` into its work folder, and the completion
    // names it absolutely.
    assert_eq!(
        completion["result_path"],
        child
            .workdir
            .join("out")
            .join("result.txt")
            .display()
            .to_string()
    );
    assert!(child
        .workdir
        .starts_with(parent_dir.path().join("children")));
    assert!(!parent_dir.path().join(".murmur").exists());
    assert_eq!(completion["delegation_id"], delegation_id);
    assert_eq!(completion["capsule_name"], "local/worker");
    assert_eq!(completion["session_id"], child.session_id);
    assert_eq!(completion["workdir"], child.workdir.display().to_string());

    // The result is where the completion said it was, and nowhere else.
    let result = std::fs::read_to_string(child.workdir.join("out").join("result.txt"))
        .expect("the child wrote the file the completion named");
    assert!(result.contains(MARKER), "{result}");

    // A completion is never turned into a task. Give the parent a moment to have done so wrongly.
    std::thread::sleep(Duration::from_secs(2));
    assert!(
        !parent.heard_of(&delegation_id),
        "the parent must start no task for a completion and record nothing of it; trace:\n{}",
        std::fs::read_to_string(parent.trace_path()).unwrap_or_default()
    );
    let trace = std::fs::read_to_string(parent.trace_path()).unwrap_or_default();
    assert!(
        !trace.contains(MARKER),
        "the child's own output must not reach the parent's trace:\n{trace}"
    );
}

// ── 4b. A child whose task failed ─────────────────────────────────────────────

/// An agent child whose inference fails ends its session cleanly, and reports the delegation as
/// `error` itself: a task that did not complete is not a delegation that succeeded.
#[test]
fn an_agent_child_whose_task_failed_reports_error() {
    if capsule_runtime::skip_without_host_support("an_agent_child_whose_task_failed_reports_error")
    {
        return;
    }
    let parent = RunningParent::start();
    let parent_dir = TempDir::new().unwrap();

    let child = {
        let _slot = agent_child_slot();
        launch(
            parent_dir.path(),
            "local/failing-worker",
            &[],
            Some(parent.spawner()),
        )
    };
    let sent = common::request(
        "POST",
        &child.capsule_url,
        Some(
            &serde_json::json!({
                "jsonrpc": "2.0",
                "id": 1,
                "method": "SendMessage",
                "params": {"message": {
                    "messageId": "msg-1",
                    "role": "ROLE_USER",
                    "parts": [{"text": "Say hello.", "mediaType": "text/plain"}]
                }}
            })
            .to_string(),
        ),
        &[(
            capsule_runtime::A2A_VERSION_HEADER,
            capsule_runtime::A2A_PROTOCOL_VERSION,
        )],
    );
    assert!(
        sent.as_ref()
            .is_some_and(|sent| sent["result"]["task"]["id"].is_string()),
        "the child took the task: {sent:?}"
    );

    let completion = wait_for_completion(&child);
    assert_eq!(
        completion["status"],
        DelegationStatus::Error.as_str(),
        "{completion}"
    );
    assert_eq!(completion["reported_by"], Reporter::Child.as_str());
}

// ── 5. A child that crashes without reporting ─────────────────────────────────

/// The launcher reports for a child that could not: the delegation is recorded as `crashed`, with
/// the exit status and the child's own stderr, and posted to the parent's door — which refuses it,
/// because no task of the parent's made this delegation.
#[test]
fn a_child_killed_without_reporting_is_reported_by_its_launcher() {
    if capsule_runtime::skip_without_host_support(
        "a_child_killed_without_reporting_is_reported_by_its_launcher",
    ) {
        return;
    }
    let parent = RunningParent::start();
    let parent_dir = TempDir::new().unwrap();

    let child = {
        let _slot = agent_child_slot();
        launch(
            parent_dir.path(),
            "local/slow-worker",
            &[],
            Some(parent.spawner()),
        )
    };
    let delegation_id = child
        .delegation_id
        .clone()
        .expect("a delegated launch has an id");
    let pid = child.pid();
    assert!(pid > 0);
    // The `W-SEC-019` line for `stderr_marker` is the child's own line, and it is on its stderr
    // before anything else the child does.
    wait_for(
        "the child to write its stderr line",
        Duration::from_secs(60),
        || {
            child
                .stderr_tail()
                .iter()
                .any(|line| line.contains("unrecognized key 'stderr_marker'"))
        },
    );

    let killed = Command::new("kill")
        .args(["-9", &pid.to_string()])
        .status()
        .expect("kill runs");
    assert!(killed.success(), "kill -9 {pid} failed");

    let completion = wait_for_completion(&child);
    assert_eq!(completion["status"], DelegationStatus::Crashed.as_str());
    assert_eq!(completion["reported_by"], Reporter::Launcher.as_str());
    assert_refused_as_unwaited(&completion, &delegation_id);
    let detail = completion["detail"]
        .as_str()
        .expect("a crash carries detail");
    assert!(detail.contains("signal: 9"), "{detail}");
    assert!(
        detail.contains("unrecognized key 'stderr_marker'"),
        "{detail}"
    );
    assert!(
        !parent.heard_of(&delegation_id),
        "the parent must start no task for a completion"
    );

    // The handle is still alive, and ending it now is not an error.
    drop(child);
}

// ── 6a. A completion whose parent is gone ─────────────────────────────────────

/// The parent's process is killed while its child is still working. The child finishes, cannot
/// deliver, and records that — the delivery failure is not the child's session failing.
#[test]
fn a_completion_with_no_parent_left_is_recorded_rather_than_dropped() {
    if capsule_runtime::skip_without_host_support(
        "a_completion_with_no_parent_left_is_recorded_rather_than_dropped",
    ) {
        return;
    }
    let mut parent = RunningParent::start();
    let parent_dir = TempDir::new().unwrap();
    let spawner = parent.spawner();
    let children_dir = parent_dir.path().join("children");
    let before = dirs_under(&children_dir);

    // The child waits for `targets.txt` before it finishes, so the parent can be killed while it
    // is demonstrably still working rather than in a race with its exit.
    let child = std::thread::scope(|scope| {
        let launch_thread = scope.spawn(|| {
            launch(
                parent_dir.path(),
                "local/waiting-worker",
                &[],
                Some(spawner),
            )
        });
        let child_dir = wait_for_new_dir(&children_dir, &before).expect("the child's directory");
        parent.kill();
        std::fs::write(child_dir.join("targets.part"), "targets.txt").unwrap();
        std::fs::rename(
            child_dir.join("targets.part"),
            child_dir.join("targets.txt"),
        )
        .unwrap();
        launch_thread.join().unwrap()
    });

    let completion = wait_for_completion(&child);
    assert_eq!(
        completion["status"],
        DelegationStatus::Ok.as_str(),
        "the child's own session succeeded; only the delivery failed: {completion}"
    );
    assert_eq!(completion["reported_by"], Reporter::Child.as_str());
    assert_eq!(completion["delivered"], false, "{completion}");
    let reason = completion["delivery_error"]
        .as_str()
        .expect("the refusal is recorded");
    assert!(reason.contains("failed to connect"), "{reason}");

    wait_for(
        "the child to say on its stderr that the completion went nowhere",
        Duration::from_secs(60),
        || {
            child
                .stderr_tail()
                .iter()
                .any(|line| line.contains("could not be delivered"))
        },
    );
}

// ── 6b. A completion addressed to a session that is not there ─────────────────

/// One capsule's address, another capsule's session id — the shape a parent that restarted onto
/// the same port leaves behind. The door refuses, the addressed capsule files no completion task,
/// and the child records the refusal.
#[test]
fn a_completion_addressed_to_another_session_is_refused_and_recorded() {
    if capsule_runtime::skip_without_host_support(
        "a_completion_addressed_to_another_session_is_refused_and_recorded",
    ) {
        return;
    }
    let addressed = RunningParent::start();
    let elsewhere = RunningParent::start();
    let parent_dir = TempDir::new().unwrap();
    let before = addressed
        .trace()
        .iter()
        .filter(|event| event["origin"] == "completion")
        .count();

    let misaddressed = Spawner {
        session_id: elsewhere.session_id.clone(),
        context_id: format!("ctx_{}", uuid_hex()),
        report_to: Some(CompletionAddress {
            url: addressed.url.clone(),
        }),
    };
    let child = launch(
        parent_dir.path(),
        "local/worker",
        &[MARKER_VAR],
        Some(misaddressed),
    );

    let completion = wait_for_completion(&child);
    assert_eq!(completion["status"], DelegationStatus::Ok.as_str());
    assert_eq!(completion["delivered"], false, "{completion}");
    let reason = completion["delivery_error"]
        .as_str()
        .expect("the refusal is recorded");
    assert!(
        reason.contains(&elsewhere.session_id),
        "the refusal names the addressed session: {reason}"
    );
    assert!(
        !reason.contains(&addressed.session_id),
        "the refusal must not name the session actually running here: {reason}"
    );

    // Nothing reached the addressed capsule's queue. Give it a moment to have done so wrongly.
    std::thread::sleep(Duration::from_secs(2));
    let after = addressed
        .trace()
        .iter()
        .filter(|event| event["origin"] == "completion")
        .count();
    assert_eq!(
        after,
        before,
        "the addressed capsule must file no completion task; trace:\n{}",
        std::fs::read_to_string(addressed.trace_path()).unwrap_or_default()
    );
}

// ── 7. A child is told where to report, and no trust class ────────────────────

/// The handle a reporting child is handed names the parent's session, the conversation, the
/// delegation and the address, and no trust class: the outcome enters the delegating task under
/// that task's own provenance, so a class carried by the completion would decide nothing. The
/// child reads the handle it was handed and reports through it.
#[test]
fn a_childs_spawner_handle_carries_no_trust_class() {
    if capsule_runtime::skip_without_host_support("a_childs_spawner_handle_carries_no_trust_class")
    {
        return;
    }
    let parent = RunningParent::start();
    let parent_dir = TempDir::new().unwrap();

    let child = launch(
        parent_dir.path(),
        "local/worker",
        &[MARKER_VAR],
        Some(parent.spawner()),
    );
    let delegation_id = child.delegation_id.clone().expect("a delegated launch");
    let recorded = child
        .env
        .iter()
        .find(|(key, _)| key == SPAWNER_ENV)
        .map(|(_, value)| value.clone())
        .expect("a delegated child is handed a spawner handle");
    let keys: Vec<String> = serde_json::from_str::<Value>(&recorded)
        .expect("the handle is JSON")
        .as_object()
        .expect("the handle is an object")
        .keys()
        .cloned()
        .collect();
    assert_eq!(
        keys,
        ["context_id", "delegation_id", "session_id", "url"],
        "{recorded}"
    );

    // The child read that handle and reported through it, to the address it names.
    let completion = wait_for_completion(&child);
    assert_eq!(completion["status"], DelegationStatus::Ok.as_str());
    assert_refused_as_unwaited(&completion, &delegation_id);
}

// ── The launcher reports once, and only for a child that did not ──────────────

/// A delegation the parent ends itself is recorded as `terminated` and posted to nobody: the only
/// party that would be told is the party that did it.
///
/// The child winds down from `SIGTERM` and may record and post its own account first; the
/// launcher's record is written over it once the watcher has seen the child reaped, so the record
/// is read only after the watcher has finished.
#[test]
fn a_delegation_the_parent_ends_is_recorded_and_not_announced() {
    if capsule_runtime::skip_without_host_support(
        "a_delegation_the_parent_ends_is_recorded_and_not_announced",
    ) {
        return;
    }
    let parent = RunningParent::start();
    let parent_dir = TempDir::new().unwrap();

    let mut child = {
        let _slot = agent_child_slot();
        launch(
            parent_dir.path(),
            "local/slow-worker",
            &[],
            Some(parent.spawner()),
        )
    };
    let delegation_id = child.delegation_id.clone().expect("a delegated launch");
    let records_dir = child.records_dir.clone();
    child.shutdown().expect("the parent ends the delegation");
    wait_for(
        "the watcher to record the ending",
        Duration::from_secs(60),
        || child.watcher_finished(),
    );

    let raw = std::fs::read_to_string(records_dir.join("completion.json"))
        .expect("the watcher recorded completion.json");
    let completion: Value = serde_json::from_str(&raw).expect("completion.json parses");

    assert_eq!(completion["status"], DelegationStatus::Terminated.as_str());
    assert_eq!(completion["reported_by"], Reporter::Launcher.as_str());
    assert_eq!(completion["delivered"], false, "{completion}");
    assert!(completion["delivery_error"].is_null(), "{completion}");

    // Nothing was posted, so the parent has heard nothing of it.
    std::thread::sleep(Duration::from_secs(2));
    assert!(
        !parent.heard_of(&delegation_id),
        "a terminated delegation is announced to nobody"
    );
}

// ── 8. A started child that never ends ────────────────────────────────────────

/// A child that outruns the delegation deadline is ended by its watcher, which posts a
/// `terminated` outcome — the only thing that stops a wedged sub-capsule running for as long as its
/// parent does, and what a delegating task continues with. Here no task is waiting, so the door
/// refuses it, and the child's record says so.
///
/// Unlike an ending the parent chose by hand, this one is posted: the delegating agent asked for
/// an outcome, and this is the outcome.
#[test]
fn a_started_child_that_never_ends_is_stopped_at_its_deadline() {
    if capsule_runtime::skip_without_host_support(
        "a_started_child_that_never_ends_is_stopped_at_its_deadline",
    ) {
        return;
    }
    let parent = RunningParent::start();
    let parent_dir = TempDir::new().unwrap();

    // `slow-worker` binds a port and sleeps between tasks, so nothing it does ever ends its
    // session. The handle is held throughout, as a delegating task holds it, so the watcher's
    // deadline is what ends the child.
    let _slot = agent_child_slot();
    let child = launch_bounded(
        parent_dir.path(),
        "local/slow-worker",
        &[],
        Some(parent.spawner()),
        Some(Duration::from_secs(5)),
    );
    let delegation_id = child.delegation_id.clone().expect("a delegated launch");
    let pid = child.pid();

    // The child winds down from the watcher's `SIGTERM` and may record its own account first; the
    // watcher's record is written over it, so it is read once the watcher has finished.
    wait_for(
        "the watcher to record the ending",
        Duration::from_secs(90),
        || child.watcher_finished(),
    );
    let completion = wait_for_refused_completion(&child.records_dir);
    assert_eq!(completion["status"], DelegationStatus::Terminated.as_str());
    assert_eq!(completion["reported_by"], Reporter::Launcher.as_str());
    let detail = completion["detail"].as_str().unwrap_or_default();
    assert!(
        detail.contains("5s") && detail.contains("deadline"),
        "the detail names the bound in seconds: {completion}"
    );
    assert_refused_as_unwaited(&completion, &delegation_id);
    assert!(
        child.has_exited(),
        "the watcher ended the child while its handle was still held"
    );
    assert!(!parent.heard_of(&delegation_id));

    // And the process is gone, reaped by the watcher before the handle is let go.
    #[cfg(target_os = "linux")]
    wait_for(
        "the ended child to be reaped",
        Duration::from_secs(30),
        || !Path::new(&format!("/proc/{pid}")).exists(),
    );
    let _ = pid;
    drop(child);
}