Host: fixes from the review, with regression tests

- one live instance per actor: check after the lock; dropping an instance kills it - state snapshots keep target/ and .git/; only source copies skip them - one shutdown deadline; output during shutdown is still routed - deliveries route stdout meanwhile; a stuck stdin kills the instance (no partial lines) and the sender gets undeliverable; control requests are not blocked by a stuck delivery - Host::stop aborts running broker and cargo requests first - compile a version before installing it; roll back if it does not start - unique archive names; a failed archive keeps the actor manageable - commit on close: truncate without create; directory renames - candidates see SAUSAGE_VERSION and the existing state of a new actor - the host publishes PROTOCOL.md at /actor/broker/public/
2026-10-15 12:03:26 +01:00
parent 69cf80a114
commit 9d575b6a62
11 changed files with 811 additions and 220 deletions
@@ -1,7 +1,8 @@
 # The Sausage protocol

 This document is the contract between the host and the actors. An actor in
-any language that follows it works in Sausage.
+any language that follows it works in Sausage. The host publishes this file at
+`/actor/broker/public/PROTOCOL.md`, so every actor can read it.

 ## Actors

@@ -43,10 +44,14 @@ A line on **stdout**:
  mailbox keeps messages while the actor restarts or is upgraded.
 - If the broker cannot deliver a message, the sender gets
  `{"type": "undeliverable", "reason": "...", "to": "..."}` from `broker`.
+- An actor must keep reading stdin. If a message waits 30 seconds for the
+  actor to read it, the broker kills and restarts the actor, and the sender
+  gets `undeliverable`.

 Before the broker stops an instance, it sends
 `{"id": "broker/shutdown", "from": "broker", "body": {"type": "shutdown"}}`
-and then closes stdin. After 5 seconds, it kills the instance.
+and then closes stdin. The broker still routes what the actor writes until
+it exits. After 5 seconds, it kills the instance.

 ## Filesystem

@@ -93,7 +98,9 @@ sender's).
 has `"rejected"` in its `meta.json`.

 **Tests.** Before a spawn or an upgrade, the broker runs the candidate on a
-copy of the actor's `/data` and `/public`, without HTTP. The tests are JSON
+copy of the actor's `/data` and `/public` (empty for a new actor), without
+HTTP. The candidate sees its future `SAUSAGE_VERSION` and
+`SAUSAGE_CANDIDATE=1`. The tests are JSON
 files in `source/tests/*.json`:

 ```json
@@ -241,9 +241,17 @@ impl Broker {
        let dir = self.world.version_dir(id, meta.version);
        std::fs::create_dir_all(&dir)?;
        std::fs::copy(wasm, dir.join("actor.wasm")).with_context(|| format!("copy {}", wasm.display()))?;
-        world::replace_tree(source.unwrap_or(&dir.join("no-source")), &dir.join("source"))?;
+        world::replace_source(source.unwrap_or(&dir.join("no-source")), &dir.join("source"))?;
        world::write_atomic(&dir.join("meta.json"), serde_json::to_string_pretty(meta)?.as_bytes())?;
-        self.runtime.load(&dir.join("actor.wasm"))
+        self.load_version(id, meta.version)
+    }
+
+    /// Compile `versions/<version>/actor.wasm`.
+    pub fn load_version(&self, id: &str, version: u64) -> Result<Component> {
+        let started = Instant::now();
+        let component = self.runtime.load(&self.world.version_dir(id, version).join("actor.wasm"))?;
+        self.log(id, &format!("compiled version {version} in {} ms", started.elapsed().as_millis()));
+        Ok(component)
    }

    /// Record in `versions/<n>/meta.json` why the broker refused a version.
@@ -282,11 +290,11 @@ impl Broker {
        if !world::valid_id(id) {
            bail!("invalid actor ID {id:?}: use a-z, 0-9, '-' and '_'");
        }
+        let lock = self.op_lock(id);
+        let _guard = lock.lock().await;
        if NATIVE.contains(&id) || self.is_supervised(id) || self.world.wasm_path(id).exists() {
            bail!("an actor with ID {id} already exists");
        }
-        let lock = self.op_lock(id);
-        let _guard = lock.lock().await;
        let version = self.world.next_version(id);
        let meta = VersionMeta { version, ..meta };
        self.create_version(id, version, &meta, wasm, source, start).await
@@ -295,7 +303,8 @@ impl Broker {
    async fn create_version(self: &Arc<Self>, id: &str, version: u64, meta: &VersionMeta, wasm: &Path, source: Option<&Path>, start: bool) -> Result<Value> {
        let tested = async {
            let component = self.store_version(id, meta, wasm, source)?;
-            candidate::test(self, id, &component, &self.world.version_dir(id, version).join("source"), None, None).await
+            let source = self.world.version_dir(id, version).join("source");
+            candidate::test(self, id, version, &component, &source, &self.world.data_dir(id), &self.world.public_dir(id)).await
        }
        .await;
        let report = match tested {
@@ -328,7 +337,7 @@ impl Broker {
        let tested = async {
            let component = self.store_version(id, meta, wasm, source)?;
            let source = self.world.version_dir(id, version).join("source");
-            candidate::test(self, id, &component, &source, Some(&self.world.data_dir(id)), Some(&self.world.public_dir(id))).await
+            candidate::test(self, id, version, &component, &source, &self.world.data_dir(id), &self.world.public_dir(id)).await
        }
        .await;
        let report = match tested {
@@ -405,10 +414,13 @@ impl Broker {
        let source = source.is_dir().then_some(source.as_path());
        let result = match installed {
            None => self.create_version(id, version, &vmeta, &wasm, source, true).await,
-            Some(current) => {
+            Some(_) => {
                let r = self.upgrade_version(id, &vmeta, &wasm, source).await;
-                if r.is_err() {
-                    // Put the installed version back in place.
+                // Put the installed version back in place. Read it again: an
+                // automatic rollback can have changed it meanwhile.
+                if r.is_err()
+                    && let Some(current) = self.meta(id).and_then(|m| m.version)
+                {
                    let _ = self.restore_files(id, current);
                }
                r
@@ -527,10 +539,18 @@ impl Broker {
        }
    }

-    /// The native `broker` actor.
+    /// The native `broker` actor. Requests run concurrently. Aborting this
+    /// task aborts the requests that still run.
    pub async fn run_actor(self: Arc<Self>, mut mailbox: mpsc::Receiver<Envelope>) {
-        while let Some(env) = mailbox.recv().await {
-            tokio::spawn(self.clone().handle(env));
+        let mut requests = tokio::task::JoinSet::new();
+        loop {
+            tokio::select! {
+                env = mailbox.recv() => match env {
+                    Some(env) => { requests.spawn(self.clone().handle(env)); }
+                    None => break,
+                },
+                Some(_) = requests.join_next() => {}
+            }
        }
    }
 }
@@ -586,6 +606,8 @@ struct Supervisor {
    swapped_at: Option<Instant>,
    crashes_since_swap: usize,
    upgrade_requester: Option<String>,
+    /// A control request that arrived during a blocked delivery.
+    next_control: Option<Control>,
 }

 async fn recv_line(instance: &mut Option<Instance>) -> Option<String> {
@@ -632,6 +654,7 @@ impl Supervisor {
            swapped_at: None,
            crashes_since_swap: 0,
            upgrade_requester: None,
+            next_control: None,
        }
    }

@@ -658,9 +681,15 @@ impl Supervisor {
    async fn run(mut self) {
        self.status();
        loop {
+            if let Some(ctl) = self.next_control.take() {
+                if self.handle_control(ctl).await {
+                    break;
+                }
+                continue;
+            }
            if self.desired_running && self.instance.is_none() && self.restart_at.is_none_or(|t| Instant::now() >= t) {
                self.restart_at = None;
-                self.start_instance().await;
+                let _ = self.start_instance().await;
            }
            let has_instance = self.instance.is_some();
            let waiting = !has_instance && self.desired_running;
@@ -690,36 +719,65 @@ impl Supervisor {
        }
    }

+    /// Write a message to the instance's stdin, and route its output
+    /// meanwhile. An instance that does not read its stdin for
+    /// `STDIN_WRITE_TIMEOUT`, or while a control request waits, is killed and
+    /// restarted: a partly written line would corrupt the next message.
    async fn deliver(&mut self, msg: Envelope) {
        let Some(inst) = self.instance.as_mut() else { return };
-        let line = msg.to_line();
-        match tokio::time::timeout(STDIN_WRITE_TIMEOUT, inst.send_line(&line)).await {
-            Ok(Ok(())) => self.last_delivered = Some(msg),
-            Ok(Err(_)) => {} // The instance is ending. The exit path handles it.
-            Err(_) => {
-                let note = format!("actor did not read stdin for {STDIN_WRITE_TIMEOUT:?}; message {} dropped", msg.id);
-                self.broker.log(&self.id, &note);
+        let (broker, id, line) = (self.broker.clone(), self.id.clone(), msg.to_line());
+        let write = tokio::time::timeout(STDIN_WRITE_TIMEOUT, inst.send_line(&line, |l| broker.route_line(&id, &l)));
+        let failure = tokio::select! {
+            written = write => match written {
+                Ok(Ok(())) => {
+                    self.last_delivered = Some(msg);
+                    return;
+                }
+                // The instance is ending. The main loop sees its closed stdout.
+                Ok(Err(_)) => None,
+                Err(_) => Some(format!("the actor did not read stdin for {STDIN_WRITE_TIMEOUT:?}")),
+            },
+            Some(ctl) = self.control.recv() => {
+                self.next_control = Some(ctl);
+                Some("the actor did not read stdin when the broker needed it".to_string())
            }
+        };
+        if msg.from != "broker" {
+            let reason = failure.as_deref().unwrap_or("the actor stopped");
+            let body = json!({"type": "undeliverable", "reason": reason, "to": self.id});
+            self.broker.send("broker", Some(&msg.from), None, Some(&msg.id), body);
+        }
+        if let (Some(failure), Some(inst)) = (failure, &self.instance) {
+            // The main loop sees the end of the instance and restarts it.
+            self.broker.log(&self.id, &format!("killed: {failure}"));
+            inst.kill();
        }
    }

-    async fn start_instance(&mut self) {
+    /// Start an instance of the current version. On failure, the actor is
+    /// "broken" until the next start, upgrade, or rollback.
+    async fn start_instance(&mut self) -> Result<()> {
+        let started = self.try_start();
+        if let Err(e) = &started {
+            self.broker.log(&self.id, &format!("cannot start version {}: {e:#}", self.version));
+            self.state = "broken";
+            self.desired_running = false;
+            self.status();
+            return started;
+        }
+        self.state = "running";
+        self.started_at = Instant::now();
+        self.status();
+        if let Some(msg) = self.redeliver.take() {
+            self.broker.log(&self.id, &format!("delivering parked message {} again", msg.id));
+            self.deliver(msg).await;
+        }
+        Ok(())
+    }
+
+    fn try_start(&mut self) -> Result<()> {
        if self.component.is_none() {
-            let path = self.world().version_dir(&self.id, self.version).join("actor.wasm");
-            let t = Instant::now();
-            match self.broker.runtime.load(&path) {
-                Ok(c) => {
-                    self.broker.log(&self.id, &format!("compiled version {} in {} ms", self.version, t.elapsed().as_millis()));
-                    self.component = Some(c);
-                }
-                Err(e) => {
-                    self.broker.log(&self.id, &format!("cannot load version {}: {e:#}", self.version));
-                    self.state = "broken";
-                    self.desired_running = false;
-                    self.status();
-                    return;
-                }
-            }
+            self.component = Some(self.broker.load_version(&self.id, self.version)?);
        }
        let cfg = InstanceConfig {
            mounts: self.broker.mounts(&self.id),
@@ -730,29 +788,15 @@ impl Supervisor {
                ("SAUSAGE_VERSION".into(), self.version.to_string()),
            ],
        };
-        match self.broker.runtime.start(self.component.as_ref().unwrap(), cfg) {
-            Ok(inst) => {
-                self.instance = Some(inst);
-                self.state = "running";
-                self.started_at = Instant::now();
-                self.status();
-                if let Some(msg) = self.redeliver.take() {
-                    self.broker.log(&self.id, &format!("delivering parked message {} again", msg.id));
-                    self.deliver(msg).await;
-                }
-            }
-            Err(e) => {
-                self.broker.log(&self.id, &format!("cannot start: {e:#}"));
-                self.state = "broken";
-                self.desired_running = false;
-                self.status();
-            }
-        }
+        self.instance = Some(self.broker.runtime.start(self.component.as_ref().unwrap(), cfg)?);
+        Ok(())
    }

+    /// Stop the instance gracefully. Its last messages still get routed.
    async fn stop_instance(&mut self) -> Option<Exit> {
        let inst = self.instance.take()?;
-        let exit = inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE).await;
+        let (broker, id) = (self.broker.clone(), self.id.clone());
+        let exit = inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE, |line| broker.route_line(&id, &line)).await;
        self.broker.log(&self.id, &format!("stopped: {exit}"));
        self.last_exit = Some(exit.to_string());
        Some(exit)
@@ -801,37 +845,48 @@ impl Supervisor {
        Ok(())
    }

-    /// Make `version` the current files: `actor.wasm`, `source/`, `actor.json`.
+    /// Make `version` current: compile it first, then write `actor.wasm`,
+    /// `source/`, and `actor.json`. If compiling fails, nothing changes.
    fn install(&mut self, version: u64) -> Result<()> {
+        let component = self.broker.load_version(&self.id, version)?;
        let dir = self.world().version_dir(&self.id, version);
        world::write_atomic(&self.world().wasm_path(&self.id), &std::fs::read(dir.join("actor.wasm"))?)?;
        world::replace_tree(&dir.join("source"), &self.world().source_dir(&self.id))?;
-        self.version = version;
-        self.component = None;
        self.broker.update_meta(&self.id, |m| {
            m.version = Some(version);
            m.running = true;
-        })
+        })?;
+        self.version = version;
+        self.component = Some(component);
+        Ok(())
    }

    fn set_running(&self, running: bool) -> Result<()> {
        self.broker.update_meta(&self.id, |m| m.running = running)
    }

+    /// Clear the restart and crash-loop state before a swap.
+    fn reset_restarts(&mut self) {
+        self.restarts = 0;
+        self.restart_at = None;
+        self.crashes_since_swap = 0;
+        self.desired_running = true;
+    }
+
    async fn upgrade(&mut self, version: u64, requester: String) -> Result<Value> {
        self.stop_instance().await;
        self.snapshot_current()?;
        let previous = self.version;
        self.install(version)?;
+        self.reset_restarts();
        self.previous = Some(previous);
        self.swapped_at = Some(Instant::now());
-        self.crashes_since_swap = 0;
-        self.restarts = 0;
-        self.restart_at = None;
        self.upgrade_requester = Some(requester);
        self.redeliver = self.parked.take();
-        self.desired_running = true;
-        self.start_instance().await;
+        if let Err(e) = self.start_instance().await {
+            self.rollback(Some(previous)).await?;
+            bail!("version {version} did not start ({e:#}); rolled back to {previous}");
+        }
        Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous}))
    }

@@ -847,6 +902,7 @@ impl Supervisor {
        if !dir.join("actor.wasm").is_file() {
            bail!("{} has no version {target}", self.id);
        }
+        self.broker.load_version(&self.id, target)?;
        self.stop_instance().await;
        self.snapshot_current()?;
        let from = self.version;
@@ -856,12 +912,10 @@ impl Supervisor {
            world::replace_tree(&snap.join("public"), &self.world().public_dir(&self.id))?;
        }
        self.install(target)?;
+        self.reset_restarts();
        self.previous = None;
        self.swapped_at = None;
-        self.restarts = 0;
-        self.restart_at = None;
-        self.desired_running = true;
-        self.start_instance().await;
+        self.start_instance().await?;
        Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": target, "restored_state": snap.is_dir()}))
    }

@@ -894,8 +948,12 @@ impl Supervisor {
                self.desired_running = false;
                self.stop_instance().await;
                let r = self.archive();
+                let removed = r.is_ok();
                let _ = reply.send(r);
-                return true;
+                if removed {
+                    return true;
+                }
+                self.state = "stopped";
            }
            Control::Shutdown { reply } => {
                self.stop_instance().await;
@@ -913,7 +971,8 @@ impl Supervisor {
    fn archive(&self) -> Result<Value> {
        let archive = self.world().root.join("archive");
        std::fs::create_dir_all(&archive)?;
-        let dest = archive.join(format!("{}-{}", self.id, world::now()));
+        let nanos = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_nanos();
+        let dest = archive.join(format!("{}-{nanos}", self.id));
        std::fs::rename(self.world().actor_dir(&self.id), &dest)?;
        Ok(json!({"type": "removed", "actor": self.id, "archive": dest.display().to_string()}))
    }
@@ -18,7 +18,7 @@
 //! also go to that recipient.
 //!
 //! Each file runs in a fresh instance on a copy of the actor's `/data` and
-//! `/public`. The candidate has no HTTP access. Without test files, the
+//! `/public` (empty folders for a new actor). The candidate has no HTTP access. Without test files, the
 //! candidate must start and stay up for `SMOKE_TIME`.

 use std::path::Path;
@@ -66,14 +66,26 @@ fn stderr_tail(path: &Path) -> String {
    text[s..].to_string()
 }

+/// The version under test and the state that its tests start from.
+struct Candidate<'a> {
+    broker: &'a Broker,
+    id: &'a str,
+    version: u64,
+    component: &'a Component,
+    data: &'a Path,
+    public: &'a Path,
+}
+
 pub async fn test(
    broker: &Broker,
    id: &str,
+    version: u64,
    component: &Component,
    source: &Path,
-    data: Option<&Path>,
-    public: Option<&Path>,
+    data: &Path,
+    public: &Path,
 ) -> Result<Value> {
+    let candidate = Candidate { broker, id, version, component, data, public };
    let mut files: Vec<_> = std::fs::read_dir(source.join("tests"))
        .map(|rd| {
            rd.filter_map(|e| e.ok())
@@ -85,7 +97,7 @@ pub async fn test(
    files.sort();

    if files.is_empty() {
-        let (mut inst, scratch) = start(broker, id, component, data, public)?;
+        let (mut inst, scratch) = candidate.start()?;
        tokio::time::sleep(SMOKE_TIME).await;
        if inst.is_finished() {
            let exit = inst.wait().await;
@@ -94,94 +106,93 @@ pub async fn test(
                stderr_tail(&scratch.dir.join("stderr.log"))
            );
        }
-        inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE).await;
+        inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE, |_| {}).await;
        return Ok(json!({"smoke_test": "passed", "files": 0}));
    }

    let mut passed = Vec::new();
    for file in &files {
        let name = file.file_name().unwrap().to_string_lossy().to_string();
-        run_file(broker, id, component, data, public, file)
-            .await
-            .map_err(|e| anyhow!("test {name} failed: {e:#}"))?;
+        candidate.run_file(file).await.map_err(|e| anyhow!("test {name} failed: {e:#}"))?;
        passed.push(name);
    }
    Ok(json!({"passed": passed}))
 }

-fn start(broker: &Broker, id: &str, component: &Component, data: Option<&Path>, public: Option<&Path>) -> Result<(Instance, Scratch)> {
-    static N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
-    let n = N.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-    let dir = broker.world.scratch_dir().join(format!("{id}-{}-{n}", std::process::id()));
-    let scratch = Scratch { dir: dir.clone() };
-    std::fs::create_dir_all(dir.join("data"))?;
-    std::fs::create_dir_all(dir.join("public"))?;
-    if let Some(d) = data.filter(|d| d.is_dir()) {
-        world::copy_tree(d, &dir.join("data"))?;
-    }
-    if let Some(p) = public.filter(|p| p.is_dir()) {
-        world::copy_tree(p, &dir.join("public"))?;
-    }
-    let cfg = InstanceConfig {
-        mounts: vec![
-            Mount { guest: "/actor".into(), host: broker.world.actors_root(), writable: false },
-            Mount { guest: "/public".into(), host: dir.join("public"), writable: true },
-            Mount { guest: "/data".into(), host: dir.join("data"), writable: true },
-        ],
-        allow_http: false,
-        stderr_log: dir.join("stderr.log"),
-        env: vec![("SAUSAGE_ACTOR".into(), id.to_string()), ("SAUSAGE_CANDIDATE".into(), "1".into())],
-    };
-    Ok((broker.runtime.start(component, cfg)?, scratch))
-}
-
-async fn run_file(
-    broker: &Broker,
-    id: &str,
-    component: &Component,
-    data: Option<&Path>,
-    public: Option<&Path>,
-    file: &Path,
-) -> Result<()> {
-    let test: Value = serde_json::from_str(&std::fs::read_to_string(file)?).map_err(|e| anyhow!("invalid JSON: {e}"))?;
-    let steps = test["steps"].as_array().ok_or_else(|| anyhow!("missing \"steps\" list"))?;
-    let (mut inst, scratch) = start(broker, id, component, data, public)?;
-    let stderr = scratch.dir.join("stderr.log");
-    for (n, step) in steps.iter().enumerate() {
-        let n = n + 1;
-        if let Some(body) = step.get("send") {
-            let from = step["from"].as_str().unwrap_or("test");
-            let env = json!({"id": format!("test/{n}"), "from": from, "body": body});
-            inst.send_line(&env.to_string())
-                .await
-                .map_err(|e| anyhow!("step {n}: candidate does not read stdin: {e}"))?;
+impl Candidate<'_> {
+    /// Start an instance on a fresh copy of the state, without HTTP.
+    fn start(&self) -> Result<(Instance, Scratch)> {
+        static N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
+        let n = N.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        let dir = self.broker.world.scratch_dir().join(format!("{}-{}-{n}", self.id, std::process::id()));
+        let scratch = Scratch { dir: dir.clone() };
+        for (from, name) in [(self.data, "data"), (self.public, "public")] {
+            std::fs::create_dir_all(dir.join(name))?;
+            if from.is_dir() {
+                world::copy_tree(from, &dir.join(name))?;
+            }
        }
-        let expected = match step.get("expect") {
-            None => vec![],
-            Some(Value::Array(list)) => list.clone(),
-            Some(one) => vec![one.clone()],
+        let cfg = InstanceConfig {
+            mounts: vec![
+                Mount { guest: "/actor".into(), host: self.broker.world.actors_root(), writable: false },
+                Mount { guest: "/public".into(), host: dir.join("public"), writable: true },
+                Mount { guest: "/data".into(), host: dir.join("data"), writable: true },
+            ],
+            allow_http: false,
+            stderr_log: dir.join("stderr.log"),
+            env: vec![
+                ("SAUSAGE_ACTOR".into(), self.id.to_string()),
+                ("SAUSAGE_VERSION".into(), self.version.to_string()),
+                ("SAUSAGE_CANDIDATE".into(), "1".into()),
+            ],
        };
-        for expected in expected {
-            let line = match tokio::time::timeout(EXPECT_TIMEOUT, inst.stdout.recv()).await {
-                Err(_) => bail!("step {n}: no message within {EXPECT_TIMEOUT:?}; expected {expected}"),
-                Ok(None) => {
-                    let exit = inst.wait().await;
-                    bail!("step {n}: candidate ended ({exit}); expected {expected}. stderr:\n{}", stderr_tail(&stderr))
-                }
-                Ok(Some(l)) => l,
-            };
-            let msg: Value = serde_json::from_str(&line)
-                .map_err(|_| anyhow!("step {n}: candidate wrote a non-JSON line: {}", truncate(&line, 300)))?;
-            if !subset_match(&expected, &msg["body"]) {
-                bail!("step {n}: expected body {expected}, got {}", truncate(&msg["body"].to_string(), 1000));
+        Ok((self.broker.runtime.start(self.component, cfg)?, scratch))
+    }
+
+    async fn run_file(&self, file: &Path) -> Result<()> {
+        let test: Value = serde_json::from_str(&std::fs::read_to_string(file)?).map_err(|e| anyhow!("invalid JSON: {e}"))?;
+        let steps = test["steps"].as_array().ok_or_else(|| anyhow!("missing \"steps\" list"))?;
+        let (mut inst, scratch) = self.start()?;
+        let stderr = scratch.dir.join("stderr.log");
+        // Lines that the candidate wrote while a step was sent.
+        let mut early = std::collections::VecDeque::new();
+        for (n, step) in steps.iter().enumerate() {
+            let n = n + 1;
+            if let Some(body) = step.get("send") {
+                let from = step["from"].as_str().unwrap_or("test");
+                let env = json!({"id": format!("test/{n}"), "from": from, "body": body});
+                inst.send_line(&env.to_string(), |line| early.push_back(line))
+                    .await
+                    .map_err(|e| anyhow!("step {n}: candidate does not read stdin: {e}"))?;
            }
-            if let Some(to) = step.get("to")
-                && msg.get("to") != Some(to)
-            {
-                bail!("step {n}: expected recipient {to}, got {}", msg.get("to").unwrap_or(&Value::Null));
+            let expected = match step.get("expect") {
+                None => vec![],
+                Some(Value::Array(list)) => list.clone(),
+                Some(one) => vec![one.clone()],
+            };
+            for expected in expected {
+                let next = async { if early.is_empty() { inst.stdout.recv().await } else { early.pop_front() } };
+                let line = match tokio::time::timeout(EXPECT_TIMEOUT, next).await {
+                    Err(_) => bail!("step {n}: no message within {EXPECT_TIMEOUT:?}; expected {expected}"),
+                    Ok(None) => {
+                        let exit = inst.wait().await;
+                        bail!("step {n}: candidate ended ({exit}); expected {expected}. stderr:\n{}", stderr_tail(&stderr))
+                    }
+                    Ok(Some(l)) => l,
+                };
+                let msg: Value = serde_json::from_str(&line)
+                    .map_err(|_| anyhow!("step {n}: candidate wrote a non-JSON line: {}", truncate(&line, 300)))?;
+                if !subset_match(&expected, &msg["body"]) {
+                    bail!("step {n}: expected body {expected}, got {}", truncate(&msg["body"].to_string(), 1000));
+                }
+                if let Some(to) = step.get("to")
+                    && msg.get("to") != Some(to)
+                {
+                    bail!("step {n}: expected recipient {to}, got {}", msg.get("to").unwrap_or(&Value::Null));
+                }
            }
        }
+        inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE, |_| {}).await;
+        Ok(())
    }
-    inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE).await;
-    Ok(())
 }
@@ -18,7 +18,7 @@ use std::time::Instant;
 use anyhow::{Context, Result, anyhow, bail};
 use serde_json::{Value, json};
 use tokio::process::Command;
-use tokio::sync::{Mutex, mpsc};
+use tokio::sync::mpsc;

 use include_dir::{Dir, include_dir};

@@ -38,17 +38,16 @@ pub struct BuildOutput {
    pub warnings: String,
 }

+/// All builds share one target folder, so they run one at a time.
 pub struct Cargo {
    world: World,
-    /// One build at a time: all builds share one target folder.
-    lock: Mutex<()>,
 }

 impl Cargo {
    /// Create the actor and install the SDK of this host into `data/sdk/`.
    pub fn new(world: World) -> Result<Self> {
        write_dir(&SDK, &world.data_dir(ID).join("sdk"))?;
-        Ok(Self { world, lock: Mutex::new(()) })
+        Ok(Self { world })
    }

    /// Build the crate in `source` (a host path) and publish the `.wasm` file
@@ -57,12 +56,11 @@ impl Cargo {
        if !source.join("Cargo.toml").is_file() {
            bail!("{} has no Cargo.toml", source.display());
        }
-        let _guard = self.lock.lock().await;
        let started = Instant::now();
        let data = self.world.data_dir(ID);
        let key = format!("{:016x}", fxhash(source.to_string_lossy().as_bytes()));
        let build_dir = data.join("work").join(&key);
-        world::replace_tree(source, &build_dir)?;
+        world::replace_source(source, &build_dir)?;
        let manifest = build_dir.join("Cargo.toml");
        std::fs::write(&manifest, prepare_manifest(&std::fs::read_to_string(&manifest)?, Some(&data.join("sdk")))?)?;

@@ -100,16 +98,13 @@ impl Cargo {
        Ok(BuildOutput { wasm: dest, seconds: started.elapsed().as_secs_f64(), warnings: strip_progress(&stderr) })
    }

-    pub async fn run_actor(self: Arc<Self>, broker: Arc<Broker>, mut mailbox: mpsc::Receiver<Envelope>) {
+    /// The native `cargo` actor. Builds run one at a time, in order.
+    pub async fn run_actor(self, broker: Arc<Broker>, mut mailbox: mpsc::Receiver<Envelope>) {
        while let Some(env) = mailbox.recv().await {
-            let this = self.clone();
-            let broker = broker.clone();
-            tokio::spawn(async move {
-                let body = this.handle(&broker, &env).await.unwrap_or_else(|e| {
-                    json!({"type": "build_failed", "source": env.body.get("source"), "errors": format!("{e:#}")})
-                });
-                broker.send(ID, Some(&env.from), None, Some(&env.id), body);
+            let body = self.handle(&broker, &env).await.unwrap_or_else(|e| {
+                json!({"type": "build_failed", "source": env.body.get("source"), "errors": format!("{e:#}")})
            });
+            broker.send(ID, Some(&env.from), None, Some(&env.id), body);
        }
    }

@@ -316,6 +316,8 @@ impl HostDescriptor for ActorFs<'_> {
                    None => tmp_name.clone(),
                };
                let tmp = target.with_file_name(&tmp_name);
+                // The temporary file is always new, also for `truncate` without `create`.
+                let oflags = oflags | types::OpenFlags::CREATE;
                let r = HostDescriptor::open_at(&mut self.inner, fd, path_flags, tmp_rel, oflags, flags).await?;
                self.ext.pending.insert(r.rep(), Pending { tmp, target });
                return Ok(r);
@@ -363,7 +365,25 @@ impl HostDescriptor for ActorFs<'_> {
        if is_tmp_name(&old_path) || is_tmp_name(&new_path) {
            return Err(ErrorCode::NoEntry.into());
        }
-        HostDescriptor::rename_at(&mut self.inner, fd, old_path, new_fd, new_path).await
+        let host = |ext: &FsExt, fd: u32, path: &str| ext.dirs.get(&fd).and_then(|d| resolve(d, path));
+        let moved = host(self.ext, fd.rep(), &old_path).zip(host(self.ext, new_fd.rep(), &new_path));
+        HostDescriptor::rename_at(&mut self.inner, fd, old_path, new_fd, new_path).await?;
+        // Open writes and directories below a renamed folder move with it.
+        if let Some((from, to)) = moved {
+            let rebase = |p: &mut PathBuf| {
+                if let Ok(rest) = p.strip_prefix(&from) {
+                    *p = to.join(rest);
+                }
+            };
+            for pending in self.ext.pending.values_mut() {
+                rebase(&mut pending.tmp);
+                rebase(&mut pending.target);
+            }
+            for dir in self.ext.dirs.values_mut() {
+                rebase(&mut dir.host);
+            }
+        }
+        Ok(())
    }

    async fn symlink_at(
@@ -20,10 +20,15 @@ use tokio::sync::mpsc;
 pub use broker::{Broker, Envelope};
 pub use world::World;

+/// The contract between the host and the actors. The host publishes it at
+/// `/actor/broker/public/PROTOCOL.md`, so that every actor can read it.
+const PROTOCOL: &str = include_str!("../../PROTOCOL.md");
+
 /// A running world.
 pub struct Host {
    pub world: World,
    pub broker: Arc<Broker>,
+    native: Vec<tokio::task::JoinHandle<()>>,
    _lock: std::fs::File,
 }

@@ -40,16 +45,17 @@ impl Host {
            bail!("another sausage process runs {}", root.display());
        }
        let broker = Broker::new(world.clone(), runtime::Runtime::new()?)?;
+        world::write_atomic(&world.public_dir("broker").join("PROTOCOL.md"), PROTOCOL.as_bytes())?;

-        let rx = broker.open_mailbox("broker");
-        tokio::spawn(broker.clone().run_actor(rx));
-        let rx = broker.open_mailbox(cargo_actor::ID);
-        tokio::spawn(Arc::new(cargo_actor::Cargo::new(world.clone())?).run_actor(broker.clone(), rx));
-
+        let cargo = cargo_actor::Cargo::new(world.clone())?;
+        let native = vec![
+            tokio::spawn(broker.clone().run_actor(broker.open_mailbox("broker"))),
+            tokio::spawn(cargo.run_actor(broker.clone(), broker.open_mailbox(cargo_actor::ID))),
+        ];
        for change in broker.scan().await {
            eprintln!("sausage: {change}");
        }
-        Ok(Host { world, broker, _lock: lock })
+        Ok(Host { world, broker, native, _lock: lock })
    }

    /// Register a native actor with the ID `id` and return its mailbox.
@@ -57,8 +63,15 @@ impl Host {
        self.broker.open_mailbox(id)
    }

-    /// Stop every actor gracefully.
+    /// Stop the broker and `cargo` (with the requests they still run), then
+    /// every actor, gracefully.
    pub async fn stop(self) {
+        for task in &self.native {
+            task.abort();
+        }
+        for task in self.native {
+            let _ = task.await;
+        }
        self.broker.shutdown_all().await;
    }
 }
@@ -131,19 +131,42 @@ pub struct InstanceConfig {
    pub env: Vec<(String, String)>,
 }

+/// A running guest. Dropping it kills the guest.
 pub struct Instance {
-    stdin: DuplexStream,
+    stdin: Option<DuplexStream>,
    /// Lines that the guest writes to stdout. Closed when the guest ends.
    pub stdout: mpsc::Receiver<String>,
    task: JoinHandle<Exit>,
 }

+impl Drop for Instance {
+    fn drop(&mut self) {
+        self.task.abort();
+    }
+}
+
 impl Instance {
-    /// Write one line to the guest's stdin.
-    pub async fn send_line(&mut self, line: &str) -> std::io::Result<()> {
-        self.stdin.write_all(line.as_bytes()).await?;
-        self.stdin.write_all(b"\n").await?;
-        self.stdin.flush().await
+    /// Write one line to the guest's stdin, in one write. Meanwhile, pass the
+    /// lines that the guest writes to `route`: a guest that blocks on a full
+    /// stdout would never read its stdin.
+    pub async fn send_line(&mut self, line: &str, mut route: impl FnMut(String)) -> std::io::Result<()> {
+        let stdin = self.stdin.as_mut().ok_or(std::io::ErrorKind::BrokenPipe)?;
+        let bytes = format!("{line}\n");
+        let write = async {
+            stdin.write_all(bytes.as_bytes()).await?;
+            stdin.flush().await
+        };
+        tokio::pin!(write);
+        let mut open = true;
+        loop {
+            tokio::select! {
+                done = &mut write => return done,
+                out = self.stdout.recv(), if open => match out {
+                    Some(out) => route(out),
+                    None => open = false,
+                },
+            }
+        }
    }

    /// Wait until the guest ends.
@@ -159,19 +182,35 @@ impl Instance {
        self.task.is_finished()
    }

-    /// Send `shutdown`, close stdin, and wait. Kill the guest after `deadline`.
-    pub async fn stop(mut self, shutdown_line: &str, deadline: Duration) -> Exit {
-        let _ = self.send_line(shutdown_line).await;
-        let _ = self.stdin.shutdown().await;
-        drop(self.stdin);
-        match tokio::time::timeout(deadline, &mut self.task).await {
-            Ok(Ok(exit)) => exit,
-            Ok(Err(e)) => Exit::Trap(format!("host task failed: {e}")),
-            Err(_) => {
-                self.task.abort();
-                let _ = (&mut self.task).await;
-                Exit::Trap("killed after shutdown deadline".into())
-            }
+    /// Kill the guest. Its stdout closes when it has ended.
+    pub fn kill(&self) {
+        self.task.abort();
+    }
+
+    /// Send `shutdown_line`, close stdin, and wait until the guest ends.
+    /// Pass every line that the guest writes meanwhile to `route`. Kill the
+    /// guest when `deadline` has passed.
+    pub async fn stop(mut self, shutdown_line: &str, deadline: Duration, mut route: impl FnMut(String)) -> Exit {
+        let stdin = self.stdin.take();
+        let stdout = &mut self.stdout;
+        let finished = tokio::time::timeout(deadline, async {
+            let close = async {
+                if let Some(mut stdin) = stdin {
+                    let _ = stdin.write_all(format!("{shutdown_line}\n").as_bytes()).await;
+                    let _ = stdin.shutdown().await;
+                }
+            };
+            let drain = async {
+                while let Some(line) = stdout.recv().await {
+                    route(line);
+                }
+            };
+            tokio::join!(close, drain);
+        })
+        .await;
+        match finished {
+            Ok(()) => self.wait().await,
+            Err(_) => Exit::Trap("killed after shutdown deadline".into()),
        }
    }
 }
@@ -299,6 +338,6 @@ impl Runtime {
            exit
        });

-        Ok(Instance { stdin: host_in, stdout: line_rx, task })
+        Ok(Instance { stdin: Some(host_in), stdout: line_rx, task })
    }
 }
@@ -227,8 +227,15 @@ pub fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> {
    Ok(())
 }

-/// Copy a directory tree. Skips `target/` folders and uncommitted files.
+/// Folders that a copy of a source folder leaves out.
+const BUILD_DIRS: &[&str] = &["target", ".git"];
+
+/// Copy a directory tree, except uncommitted files.
 pub fn copy_tree(from: &Path, to: &Path) -> Result<()> {
+    copy_filtered(from, to, &[])
+}
+
+fn copy_filtered(from: &Path, to: &Path, skip: &[&str]) -> Result<()> {
    std::fs::create_dir_all(to)?;
    for entry in std::fs::read_dir(from).with_context(|| format!("read {}", from.display()))? {
        let entry = entry?;
@@ -239,8 +246,8 @@ pub fn copy_tree(from: &Path, to: &Path) -> Result<()> {
        }
        let dst = to.join(&*name);
        let ty = entry.file_type()?;
-        if ty.is_dir() && name != "target" && name != ".git" {
-            copy_tree(&entry.path(), &dst)?;
+        if ty.is_dir() && !skip.contains(&&*name) {
+            copy_filtered(&entry.path(), &dst, skip)?;
        } else if ty.is_file() {
            std::fs::copy(entry.path(), &dst)?;
        }
@@ -251,6 +258,15 @@ pub fn copy_tree(from: &Path, to: &Path) -> Result<()> {
 /// Replace the directory `dst` with a copy of `src` (an empty folder if `src`
 /// does not exist): copy to a sibling, then swap with two renames.
 pub fn replace_tree(src: &Path, dst: &Path) -> Result<()> {
+    replace_filtered(src, dst, &[])
+}
+
+/// Like `replace_tree`, for a source folder: leaves out `target/` and `.git/`.
+pub fn replace_source(src: &Path, dst: &Path) -> Result<()> {
+    replace_filtered(src, dst, BUILD_DIRS)
+}
+
+fn replace_filtered(src: &Path, dst: &Path, skip: &[&str]) -> Result<()> {
    let parent = dst.parent().context("path has no parent")?;
    let name = dst.file_name().and_then(|n| n.to_str()).unwrap_or("dir");
    let fresh = parent.join(format!("{TMP_PREFIX}new-{name}"));
@@ -258,7 +274,7 @@ pub fn replace_tree(src: &Path, dst: &Path) -> Result<()> {
    let _ = std::fs::remove_dir_all(&fresh);
    let _ = std::fs::remove_dir_all(&old);
    if src.is_dir() {
-        copy_tree(src, &fresh)?;
+        copy_filtered(src, &fresh, skip)?;
    } else {
        std::fs::create_dir_all(&fresh)?;
    }
@@ -1,19 +1,21 @@
 use std::collections::HashMap;
 use std::io::{self, BufRead, Write};
-use std::time::Duration;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};

 use serde_json::{Value, json};

-fn reply(out: &mut impl Write, message: &Value, body: Value) {
-    writeln!(
-        out,
-        "{}",
-        json!({"to": message["from"], "reply_to": message["id"], "body": body})
-    )
-    .unwrap();
+fn send(out: &mut impl Write, message: Value) {
+    writeln!(out, "{message}").unwrap();
    out.flush().unwrap();
 }

+fn reply(out: &mut impl Write, message: &Value, body: Value) {
+    send(
+        out,
+        json!({"to": message["from"], "reply_to": message["id"], "body": body}),
+    );
+}
+
 fn io_error(error: io::Error) -> Value {
    json!({"type": "error", "error": error.to_string()})
 }
@@ -21,7 +23,8 @@ fn io_error(error: io::Error) -> Value {
 fn main() {
    let actor = std::env::var("SAUSAGE_ACTOR").unwrap_or_default();
    let version = std::env::var("SAUSAGE_VERSION").unwrap_or_else(|_| "candidate".to_string());
-    if actor == "crashloop" && version == "2" {
+    let candidate = std::env::var_os("SAUSAGE_CANDIDATE").is_some();
+    if actor == "crashloop" && version == "2" && !candidate {
        std::fs::write("/data/state", "bad-v2").unwrap();
        panic!("version 2 crashes on startup");
    }
@@ -29,6 +32,14 @@ fn main() {
    let mut open = HashMap::<String, std::fs::File>::new();
    let stdin = io::stdin();
    let mut out = io::stdout().lock();
+    let mut emit_shutdown = false;
+    if actor == "spawn-race" && !candidate {
+        let stamp = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_nanos();
+        std::fs::write(format!("/data/start-{stamp}"), "started").unwrap();
+    }
    for line in stdin.lock().lines() {
        let Ok(line) = line else { break };
        let Ok(message) = serde_json::from_str::<Value>(&line) else {
@@ -43,17 +54,21 @@ fn main() {
                    let _ = std::fs::write("/data/shutdown", "stopping");
                    std::thread::sleep(Duration::from_millis(800));
                }
+                if emit_shutdown {
+                    send(
+                        &mut out,
+                        json!({"to": "console", "reply_to": message["id"], "body": {"type": "shutdown-reply"}}),
+                    );
+                    send(&mut out, json!({"body": {"type": "shutdown-publish"}}));
+                }
                break;
            }
            Some("echo") => json!({"type": "echo", "value": body["value"]}),
            Some("publish") => {
-                writeln!(
-                    out,
-                    "{}",
-                    json!({"body": {"type": "published", "value": body["value"]}})
-                )
-                .unwrap();
-                out.flush().unwrap();
+                send(
+                    &mut out,
+                    json!({"body": {"type": "published", "value": body["value"]}}),
+                );
                continue;
            }
            Some("version") => json!({"type": "version", "version": version}),
@@ -62,6 +77,15 @@ fn main() {
                Ok(()) => json!({"type": "ok"}),
                Err(error) => io_error(error),
            },
+            Some("mkdir") => match std::fs::create_dir_all(&path) {
+                Ok(()) => json!({"type": "ok"}),
+                Err(error) => io_error(error),
+            },
+            Some("rename") => match std::fs::rename(&path, body["to"].as_str().unwrap_or_default())
+            {
+                Ok(()) => json!({"type": "ok"}),
+                Err(error) => io_error(error),
+            },
            Some("open") => {
                let result = (|| -> io::Result<()> {
                    let mut file = std::fs::File::create(&path)?;
@@ -74,6 +98,21 @@ fn main() {
                    Err(error) => io_error(error),
                }
            }
+            Some("open-truncate") => {
+                let result = (|| -> io::Result<()> {
+                    let mut file = std::fs::OpenOptions::new()
+                        .write(true)
+                        .truncate(true)
+                        .open(&path)?;
+                    file.write_all(text.as_bytes())?;
+                    open.insert(path.clone(), file);
+                    Ok(())
+                })();
+                match result {
+                    Ok(()) => json!({"type": "opened"}),
+                    Err(error) => io_error(error),
+                }
+            }
            Some("append") => {
                let result = (|| -> io::Result<()> {
                    let mut file = std::fs::OpenOptions::new().append(true).open(&path)?;
@@ -87,6 +126,11 @@ fn main() {
                }
            }
            Some("close") => json!({"type": "closed", "closed": open.remove(&path).is_some()}),
+            Some("arm-shutdown") => {
+                emit_shutdown = true;
+                json!({"type": "armed"})
+            }
+            Some("ignore-stdin") => json!({"type": "ignoring"}),
            Some("read") => match std::fs::read_to_string(&path) {
                Ok(text) => json!({"type": "content", "text": text}),
                Err(error) => io_error(error),
@@ -107,5 +151,10 @@ fn main() {
            _ => json!({"type": "error", "error": "unknown message"}),
        };
        reply(&mut out, &message, response);
+        if body["type"] == "ignore-stdin" {
+            loop {
+                std::thread::sleep(Duration::from_secs(60));
+            }
+        }
    }
 }
@@ -142,6 +142,28 @@ pub async fn receive_reply(
    .unwrap_or_else(|_| panic!("timed out waiting for reply to {id}"))
 }

+pub async fn receive_replies(
+    mailbox: &mut mpsc::Receiver<Envelope>,
+    ids: &[String],
+    timeout: Duration,
+) -> Vec<Envelope> {
+    let mut replies = Vec::with_capacity(ids.len());
+    tokio::time::timeout(timeout, async {
+        while replies.len() < ids.len() {
+            let message = mailbox.recv().await.expect("console mailbox closed");
+            assert!(
+                ids.iter()
+                    .any(|id| message.reply_to.as_deref() == Some(id.as_str())),
+                "unexpected message while waiting for replies: {message:?}"
+            );
+            replies.push(message);
+        }
+    })
+    .await
+    .expect("timed out waiting for replies");
+    replies
+}
+
 pub async fn ask(
    host: &Host,
    mailbox: &mut mpsc::Receiver<Envelope>,
@@ -152,16 +174,24 @@ pub async fn ask(
    receive_reply(mailbox, &id, REPLY_TIMEOUT).await
 }

-pub async fn spawn_actor(
+pub async fn ask_type(
    host: &Host,
    mailbox: &mut mpsc::Receiver<Envelope>,
    actor: &str,
-    wasm: &Path,
-    source: &Path,
+    body: Value,
+    expected: &str,
 ) -> Envelope {
-    request(
-        host,
-        mailbox,
+    let reply = ask(host, mailbox, actor, body).await;
+    assert_eq!(reply.body["type"], expected, "unexpected reply: {reply:?}");
+    reply
+}
+
+pub fn spawn_request(host: &Host, actor: &str, wasm: &Path, source: &Path) -> String {
+    host.broker.send(
+        "console",
+        Some("broker"),
+        None,
+        None,
        json!({
            "type": "spawn",
            "actor": actor,
@@ -169,7 +199,32 @@ pub async fn spawn_actor(
            "source": source.display().to_string()
        }),
    )
-    .await
+}
+
+pub async fn spawn_actor(
+    host: &Host,
+    mailbox: &mut mpsc::Receiver<Envelope>,
+    actor: &str,
+    wasm: &Path,
+    source: &Path,
+) -> Envelope {
+    let id = spawn_request(host, actor, wasm, source);
+    receive_reply(mailbox, &id, REPLY_TIMEOUT).await
+}
+
+pub async fn spawn_type(
+    host: &Host,
+    mailbox: &mut mpsc::Receiver<Envelope>,
+    actor: &str,
+    wasm: &Path,
+    source: &Path,
+) -> Envelope {
+    let reply = spawn_actor(host, mailbox, actor, wasm, source).await;
+    assert_eq!(
+        reply.body["type"], "spawned",
+        "unexpected spawn reply: {reply:?}"
+    );
+    reply
 }

 pub async fn upgrade_actor(
@@ -1,6 +1,6 @@
 mod common;

-use std::time::Duration;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};

 use common::*;
 use sausage::fs::TMP_PREFIX;
@@ -20,6 +20,41 @@ async fn filesystem_world(label: &str) -> (World, Host, mpsc::Receiver<Envelope>
    (world, host, mailbox)
 }

+async fn receive_shutdown_messages(
+    mailbox: &mut mpsc::Receiver<Envelope>,
+    id: &str,
+    expected: &str,
+) {
+    let (mut result, mut reply, mut publish) = (false, false, false);
+    tokio::time::timeout(Duration::from_secs(10), async {
+        while !result || !reply || !publish {
+            let message = mailbox.recv().await.expect("console mailbox closed");
+            if message.reply_to.as_deref() == Some(id) {
+                assert_eq!(message.body["type"], expected);
+                result = true;
+            } else if message.reply_to.as_deref() == Some("broker/shutdown") {
+                assert_eq!(message.from, "shutdown-output");
+                assert_eq!(message.body["type"], "shutdown-reply");
+                reply = true;
+            } else {
+                assert_eq!(message.from, "shutdown-output");
+                assert_eq!(message.reply_to, None);
+                assert_eq!(message.body["type"], "shutdown-publish");
+                publish = true;
+            }
+        }
+    })
+    .await
+    .expect("shutdown output or lifecycle reply did not arrive");
+}
+
+fn epoch_second() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap()
+        .as_secs()
+}
+
 #[tokio::test]
 async fn routes_replies_publishes_and_reports_undeliverable() {
    let world = create_world("routing");
@@ -639,3 +674,295 @@ async fn rejects_parent_paths() {
    );
    host.stop().await;
 }
+
+#[tokio::test]
+async fn concurrent_spawns_of_one_id_leave_one_live_instance() {
+    let world = create_world("concurrent-spawn");
+    let wasm = toolbox_wasm().clone();
+    let (host, mut mailbox) = start(&world).await;
+    let source = passing_source(&world, "source");
+    let ids = [
+        spawn_request(&host, "spawn-race", &wasm, &source),
+        spawn_request(&host, "spawn-race", &wasm, &source),
+    ];
+    let replies = receive_replies(&mut mailbox, &ids, Duration::from_secs(20)).await;
+    let spawned = replies
+        .iter()
+        .filter(|r| r.body["type"] == "spawned")
+        .count();
+    let failed = replies.iter().filter(|r| r.body["type"] == "error").count();
+    assert_eq!((spawned, failed), (1, 1));
+
+    let echo = ask_type(
+        &host,
+        &mut mailbox,
+        "spawn-race",
+        json!({"type": "echo", "value": "live"}),
+        "echo",
+    )
+    .await;
+    assert_eq!(echo.body["value"], "live");
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    let list = ask_type(
+        &host,
+        &mut mailbox,
+        "spawn-race",
+        json!({"type": "list", "path": "/data"}),
+        "entries",
+    )
+    .await;
+    let starts = list.body["names"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .filter(|n| n.as_str().unwrap().starts_with("start-"))
+        .count();
+    assert_eq!(starts, 1);
+    host.stop().await;
+}
+
+#[tokio::test]
+async fn rollback_preserves_target_and_git_files_in_data_and_public() {
+    let world = create_world("snapshot-special-directories");
+    let wasm = toolbox_wasm().clone();
+    let (host, mut mailbox) = start(&world).await;
+    let source = passing_source(&world, "source");
+    spawn_type(&host, &mut mailbox, "snapshot", &wasm, &source).await;
+    let files = [
+        ("/data/target", "/data/target/x", "data-target"),
+        ("/data/.git", "/data/.git/x", "data-git"),
+        ("/public/target", "/public/target/x", "public-target"),
+        ("/public/.git", "/public/.git/x", "public-git"),
+    ];
+    for (dir, path, text) in files {
+        let mkdir = json!({"type": "mkdir", "path": dir});
+        ask_type(&host, &mut mailbox, "snapshot", mkdir, "ok").await;
+        let write = json!({"type": "write", "path": path, "text": text});
+        ask_type(&host, &mut mailbox, "snapshot", write, "ok").await;
+    }
+    let upgrade = json!({"type": "upgrade", "actor": "snapshot", "wasm": wasm.display().to_string(), "source": source.display().to_string()});
+    ask_type(&host, &mut mailbox, "broker", upgrade, "upgraded").await;
+    let rollback = json!({"type": "rollback", "actor": "snapshot", "version": 1});
+    ask_type(&host, &mut mailbox, "broker", rollback, "rolled_back").await;
+    for (_, path, text) in files {
+        let read = json!({"type": "read", "path": path});
+        let reply = ask_type(&host, &mut mailbox, "snapshot", read, "content").await;
+        assert_eq!(reply.body["text"], text);
+    }
+    host.stop().await;
+}
+
+#[tokio::test]
+async fn stop_finishes_when_actor_stdin_is_full() {
+    let world = create_world("full-stdin-stop");
+    let wasm = toolbox_wasm().clone();
+    let (host, mut mailbox) = start(&world).await;
+    let source = passing_source(&world, "source");
+    spawn_type(&host, &mut mailbox, "full-stdin", &wasm, &source).await;
+    let ignore = json!({"type": "ignore-stdin"});
+    ask_type(&host, &mut mailbox, "full-stdin", ignore, "ignoring").await;
+    let large = json!({"type": "echo", "value": "x".repeat(256 * 1024)});
+    host.broker
+        .send("console", Some("full-stdin"), None, None, large);
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    let stop = json!({"type": "stop", "actor": "full-stdin"});
+    let id = host
+        .broker
+        .send("console", Some("broker"), None, None, stop);
+    let stopped = tokio::time::timeout(
+        Duration::from_secs(6),
+        receive_reply(&mut mailbox, &id, Duration::from_secs(10)),
+    )
+    .await
+    .expect("stop exceeded the five-second shutdown deadline");
+    assert_eq!(stopped.body["type"], "stopped");
+    host.stop().await;
+}
+
+#[tokio::test]
+async fn shutdown_reply_and_publish_survive_stop_and_upgrade() {
+    let world = create_world("shutdown-output");
+    let wasm = toolbox_wasm().clone();
+    let (host, mut mailbox) = start(&world).await;
+    let source = passing_source(&world, "source");
+    spawn_type(&host, &mut mailbox, "shutdown-output", &wasm, &source).await;
+    let subscribe = json!({"type": "subscribe", "actor": "shutdown-output"});
+    ask_type(&host, &mut mailbox, "broker", subscribe, "subscribed").await;
+
+    for operation in ["stop", "upgrade"] {
+        let arm = json!({"type": "arm-shutdown"});
+        ask_type(&host, &mut mailbox, "shutdown-output", arm, "armed").await;
+        let body = match operation {
+            "stop" => json!({"type": "stop", "actor": "shutdown-output"}),
+            _ => {
+                json!({"type": "upgrade", "actor": "shutdown-output", "wasm": wasm.display().to_string(), "source": source.display().to_string()})
+            }
+        };
+        let id = host
+            .broker
+            .send("console", Some("broker"), None, None, body);
+        receive_shutdown_messages(
+            &mut mailbox,
+            &id,
+            if operation == "stop" {
+                "stopped"
+            } else {
+                "upgraded"
+            },
+        )
+        .await;
+        if operation == "stop" {
+            let start = json!({"type": "start", "actor": "shutdown-output"});
+            ask_type(&host, &mut mailbox, "broker", start, "started").await;
+        }
+    }
+    host.stop().await;
+}
+
+#[tokio::test]
+async fn rapid_remove_recreate_remove_keeps_actor_manageable() {
+    let world = create_world("rapid-remove");
+    let wasm = toolbox_wasm().clone();
+    let (host, mut mailbox) = start(&world).await;
+    let source = passing_source(&world, "source");
+    spawn_type(&host, &mut mailbox, "rapid", &wasm, &source).await;
+    let stop = json!({"type": "stop", "actor": "rapid"});
+    ask_type(&host, &mut mailbox, "broker", stop, "stopped").await;
+
+    while SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap()
+        .subsec_millis()
+        >= 100
+    {
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    let archive_second = epoch_second();
+    let remove = json!({"type": "remove", "actor": "rapid"});
+    let first = ask_type(&host, &mut mailbox, "broker", remove.clone(), "removed").await;
+    assert_eq!(
+        epoch_second(),
+        archive_second,
+        "first remove crossed an epoch-second boundary"
+    );
+
+    // Restore the stopped actor from its archive to avoid a candidate smoke test.
+    sausage::world::copy_tree(
+        first.body["archive"]
+            .as_str()
+            .map(std::path::Path::new)
+            .unwrap(),
+        &world.actor_dir("rapid"),
+    )
+    .unwrap();
+    let scan = json!({"type": "scan"});
+    ask_type(&host, &mut mailbox, "broker", scan.clone(), "scanned").await;
+    let second = ask_type(&host, &mut mailbox, "broker", remove.clone(), "removed").await;
+    assert_eq!(
+        epoch_second(),
+        archive_second,
+        "remove and recreate crossed an epoch-second boundary"
+    );
+
+    sausage::world::copy_tree(
+        second.body["archive"]
+            .as_str()
+            .map(std::path::Path::new)
+            .unwrap(),
+        &world.actor_dir("rapid"),
+    )
+    .unwrap();
+    ask_type(&host, &mut mailbox, "broker", scan, "scanned").await;
+    let start = json!({"type": "start", "actor": "rapid"});
+    ask_type(&host, &mut mailbox, "broker", start, "started").await;
+    let echo = ask_type(
+        &host,
+        &mut mailbox,
+        "rapid",
+        json!({"type": "echo", "value": "managed"}),
+        "echo",
+    )
+    .await;
+    assert_eq!(echo.body["value"], "managed");
+    host.stop().await;
+}
+
+#[tokio::test]
+async fn close_commits_open_file_after_parent_directory_rename() {
+    let (_, host, mut mailbox) = filesystem_world("rename-open-file").await;
+    let mkdir = json!({"type": "mkdir", "path": "/public/a"});
+    ask_type(&host, &mut mailbox, "writer", mkdir, "ok").await;
+    let open = json!({"type": "open", "path": "/public/a/file", "text": "content"});
+    ask_type(&host, &mut mailbox, "writer", open, "opened").await;
+    let rename = json!({"type": "rename", "path": "/public/a", "to": "/public/b"});
+    ask_type(&host, &mut mailbox, "writer", rename, "ok").await;
+    let close = json!({"type": "close", "path": "/public/a/file"});
+    assert_eq!(
+        ask(&host, &mut mailbox, "writer", close).await.body["closed"],
+        true
+    );
+    let read = json!({"type": "read", "path": "/public/b/file"});
+    assert_eq!(
+        ask_type(&host, &mut mailbox, "writer", read, "content")
+            .await
+            .body["text"],
+        "content"
+    );
+    host.stop().await;
+}
+
+#[tokio::test]
+async fn truncate_without_create_commits_existing_file_on_close() {
+    let (_, host, mut mailbox) = filesystem_world("truncate-without-create").await;
+    let write = json!({"type": "write", "path": "/public/file", "text": "old"});
+    ask_type(&host, &mut mailbox, "writer", write, "ok").await;
+    let open = json!({"type": "open-truncate", "path": "/public/file", "text": "new"});
+    ask_type(&host, &mut mailbox, "writer", open, "opened").await;
+    let path = "/actor/writer/public/file";
+    let read = json!({"type": "read", "path": path});
+    assert_eq!(
+        ask_type(&host, &mut mailbox, "reader", read, "content")
+            .await
+            .body["text"],
+        "old"
+    );
+    let close = json!({"type": "close", "path": "/public/file"});
+    assert_eq!(
+        ask(&host, &mut mailbox, "writer", close).await.body["closed"],
+        true
+    );
+    let read = json!({"type": "read", "path": path});
+    assert_eq!(
+        ask_type(&host, &mut mailbox, "reader", read, "content")
+            .await
+            .body["text"],
+        "new"
+    );
+    host.stop().await;
+}
+
+#[tokio::test]
+async fn candidates_receive_version_and_preexisting_spawn_state() {
+    let world = create_world("candidate-context");
+    let wasm = toolbox_wasm().clone();
+    let (host, mut mailbox) = start(&world).await;
+    let version_step =
+        json!({"send": {"type": "version"}, "expect": {"type": "version", "version": "1"}});
+    let version_source = source_with_test(&world, "version", json!({"steps": [version_step]}));
+    let state_step = json!({"send": {"type": "read", "path": "/data/state.json"}, "expect": {"type": "content", "text": "seeded"}});
+    let state_source = source_with_test(&world, "state", json!({"steps": [state_step]}));
+    let state_dir = world.data_dir("stateful");
+    std::fs::create_dir_all(&state_dir).unwrap();
+    std::fs::write(state_dir.join("state.json"), "seeded").unwrap();
+    let ids = [
+        spawn_request(&host, "versioned", &wasm, &version_source),
+        spawn_request(&host, "stateful", &wasm, &state_source),
+    ];
+    let replies = receive_replies(&mut mailbox, &ids, Duration::from_secs(20)).await;
+    let successful = replies
+        .iter()
+        .filter(|reply| reply.body["type"] == "spawned")
+        .count();
+    assert_eq!(successful, 2);
+    host.stop().await;
+}