broker: add stopped-preserving upgrade without actor execution

PROTOCOL.md

@@ -63,10 +63,12 @@ Each instance sees three folders:
 
 | Path | Access | Contents |
 | --- | --- | --- |
-| `/data` | read-write | the actor's state |
+| `/data` | read-write by its actor; readable by every actor | the actor's state |
 | `/public` | read-write | the files that the actor publishes |
 | `/actor` | read-only | the `actor/` folder of the world |
 
+An actor can write only its own `/data`, but can read every actor's `/data` through `/actor/<id>/data/`. `/data` is not private: any unprotected secret file there is readable by every actor.
+
 `/actor/<id>/` holds everything about an actor: `actor.wasm`, `actor.json`
 (installed version, running, subscriptions, tapped, watches), `source/`, `public/`, `data/`,
 `stderr.log`, and `versions/<n>/`. `ls /actor` lists all actors.
@@ -123,14 +125,16 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`,
 | `type` | Fields | Effect |
 | --- | --- | --- |
 | `spawn` | `actor`, `wasm`, `source`?, `built_by`? | test, then start a new actor |
-| `upgrade` | `actor`, `wasm`, `source`?, `built_by`? | test, then swap the version |
+| `upgrade` | `actor`, `wasm`, `source`?, `built_by`?, `preserve_stopped`? (boolean, default `false`) | test, then swap the version; with `preserve_stopped: true`, a stopped actor is compiled and installed without execution and stays stopped |
 | `rollback` | `actor`, `version` | go back to an older version |
 | `stop`, `start` | `actor` | stop or start an actor |
 | `remove` | `actor` | stop the actor (normal shutdown), then delete its whole folder `/actor/<id>/` (all versions, `data/`, `public/`, logs); answer `{"type": "removed", "actor"}`. Messages to it are then undeliverable, and the ID can be spawned again. An unknown actor or a native one (`broker`, `console`, `cargo`, `fs`) is an `error` |
 | `subscribe`, `unsubscribe` | `actor` | receive (or not) what `actor` publishes |
 | `tap`, `untap` | | receive (or not) a copy of every message the broker routes |
 | `watch`, `unwatch` | `path` | receive (or not) `fs_changed` for every change of `path` or anything under it; answer `{"type": "watching" \| "unwatched", "path"}` with the path as an `/actor/...` path |
-| `exit` | | answer `exiting`, stop every actor, and end the `sausage` process (a wrapper like `nix run .#loop` starts it again, updated) |
+| `exit` | | answer `exiting`, stop every actor, and end the `sausage` process with exit code 75 (a wrapper like `nix run .#loop` starts it again, updated) |
+
+**Headless.** `sausage run --headless [dir]` (or `sausage --headless`, or `SAUSAGE_HEADLESS=1`) turns the console input off: the host does not read stdin, so end of input (`/dev/null`) does not stop the world. `nix run .#loop -- --headless <world>` passes it through the loop. It runs until SIGTERM or SIGINT (stops every actor, exit code 0) or a broker `exit` request (exit code 75, as above). Without the flag, EOF on stdin or `/quit` ends the host.
 
 Paths are `/actor/...`, or the sender's `/public/...` or `/data/...`.
 
@@ -178,10 +182,11 @@ change of `path` or a path under it (`/actor/a` covers `/actor/a/x`, not
   restarts and upgrades. `unwatch` with the same path ends it.
 
 **Versions.** A new version (from `spawn`, `upgrade`, or a changed
-`actor.wasm` on disk) is tested first, while the old version keeps running.
+`actor.wasm` on disk) is tested first, while the old version keeps running
+(except the stopped-preserving upgrade described below).
 Only when its tests pass does the broker copy it to `versions/<n>/`
 (`actor.wasm`, `source/`, `meta.json`) and switch to it. So `versions/` holds
-exactly the versions that were installed, and instances always run from
+exactly the versions that were installed (including stopped installations), and instances always run from
 there. If the tests fail, nothing changes, and the sender gets an `error`
 with the reason. All versions share the actor's `data/` and `public/`: a
 new version must read the data of the old one.
@@ -210,6 +215,21 @@ instance. A version without test files passes.
 top-level `actor.wasm` and `source/` a copy of the new version, and starts
 it. A rollback does the same with the older version.
 
+**Stopped-preserving upgrade.** Send `upgrade` with `"preserve_stopped": true`
+to install a stopped actor's new program without starting it, running `main`
+(even as a candidate), or reading or sending any of its mail. The host compiles
+the WASM and copies the new version, source and metadata as usual, increments
+the version, and keeps `actor.json`'s `running: false`. Its mailbox, `data/`
+and `public/` stay untouched. Because candidate tests execute `main`, they are
+not run in this case: the `upgraded` reply contains
+`"tests": {"skipped": "preserve_stopped"}`. Test the build separately before
+installing it this way. An explicit `start` later runs the installed version
+and delivers the queued mail. No startup/crash probation runs for this stopped
+installation. Running actors still use the normal tested swap, even with this
+option. Omitting the option (or `false`) retains the existing behavior,
+including starting a stopped actor. Explicit stop/start requests serialize
+with upgrades of the same actor.
+
 **Crashes.** The broker restarts a crashed actor after 0.5 s, then doubles
 the delay up to 30 s. A new version that crashes 3 times within 30 seconds
 after an upgrade goes back to the previous version. The actor that asked for

README.md

@@ -48,6 +48,18 @@ From a clone (`git clone https://gitea.surma.technology/surma/sausage`), the
 same commands are `nix run`, `nix run -- init`, `nix run -- run` and
 `nix run .#loop`.
 
+### Headless
+
+`--headless` (or `SAUSAGE_HEADLESS=1`) turns the console input off: the host
+never reads stdin, so end of input does not stop the world. It runs until
+SIGTERM/SIGINT (clean stop, exit code 0) or a broker `exit` request (exit
+code 75, so the loop starts the newest version). Create the world first
+(`init`), then pass the flag through the loop:
+
+```sh
+nix run git+https://gitea.surma.technology/surma/sausage#loop -- --headless my-world
+```
+
 In the console:
 
 ```text
@@ -66,7 +78,7 @@ actor.wasm     the current version
 actor.json     installed version, running, subscriptions
 source/        the source of the current version
 public/        files that the actor publishes
-data/          the actor's state
+data/          actor state; owner-writable, world-readable via /actor/<id>/data/ (not private)
 stderr.log     what the actor and the broker write about it
 versions/<n>/  every installed version: actor.wasm, source/, meta.json
 ```

agent/src/prompt.md

@@ -3,3 +3,5 @@ People talk to you through other actors. Your text goes to your subscribers.
 Before writing an actor, read `/actor/broker/public/PROTOCOL.md` for messages, the broker, `cargo`, and tests.
 Read the SDK at `/actor/cargo/public/sdk/` before you write an actor.
 You can change your own source at `/actor/agent/source`.
+
+The global filesystem is readable by every actor, including every actor's `/data` at `/actor/<id>/data/`. An actor can write only its own `/data` and `/public`. `/data` is not private: any unprotected secret file there is readable by every actor.

cli/src/broker.rs

@@ -313,12 +313,16 @@ impl Broker {
         Ok(component)
     }
 
-    /// Test a new version where it lies. Only when the tests pass, copy it
+    /// Compile a new version where it lies and, normally, test it. Copy it
-    /// to `versions/<n>/`: that folder holds only versions that run(ran).
+    /// to `versions/<n>/` only on success. A stopped-preserving install skips
-    /// Returns the component and the test report.
+    /// execution, including candidate tests. Return the component and report.
-    async fn try_version(&self, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<(Component, Value)> {
+    async fn try_version(&self, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>, run_tests: bool) -> Result<(Component, Value)> {
         let component = self.load(id, wasm)?;
-        let tests = candidate::test(self, id, meta.version, &component, source).await;
+        let tests = if run_tests {
+            candidate::test(self, id, meta.version, &component, source).await
+        } else {
+            Ok(json!({"skipped": "preserve_stopped"}))
+        };
         if let Err(e) = &tests {
             self.log(id, &format!("version {} from {} failed its tests: {e:#}", meta.version, meta.wasm));
         }
@@ -332,11 +336,11 @@ impl Broker {
     }
 
     /// Make `version` the current `actor.wasm`, `source/`, and `actor.json`.
-    pub fn install(&self, id: &str, version: u64) -> Result<()> {
+    pub fn install(&self, id: &str, version: u64, running: bool) -> Result<()> {
         self.world.install_files(id, version)?;
         self.update_meta(id, |m| {
             m.version = Some(version);
-            m.running = true;
+            m.running = running;
         })
     }
 
@@ -359,26 +363,29 @@ impl Broker {
     }
 
     async fn create_version(self: &Arc<Self>, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<Value> {
-        let (_, tests) = self.try_version(id, meta, wasm, source).await?;
+        let (_, tests) = self.try_version(id, meta, wasm, source, true).await?;
-        self.install(id, meta.version)?;
+        self.install(id, meta.version, true)?;
         self.start_supervisor(id, meta.version, true);
         Ok(json!({"type": "spawned", "actor": id, "version": meta.version, "tests": tests}))
     }
 
-    pub async fn upgrade_actor(self: &Arc<Self>, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta) -> Result<Value> {
+    pub async fn upgrade_actor(self: &Arc<Self>, id: &str, wasm: &Path, source: Option<&Path>, meta: VersionMeta, preserve_stopped: bool) -> Result<Value> {
         let lock = self.op_lock(id);
         let _guard = lock.lock().await;
         if !self.is_supervised(id) {
             bail!("no WASM actor with ID {id}");
         }
         let version = self.world.next_version(id);
-        self.upgrade_version(id, &VersionMeta { version, ..meta }, wasm, source).await
+        self.upgrade_version(id, &VersionMeta { version, ..meta }, wasm, source, preserve_stopped).await
     }
 
-    async fn upgrade_version(self: &Arc<Self>, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<Value> {
+    async fn upgrade_version(self: &Arc<Self>, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>, preserve_stopped: bool) -> Result<Value> {
-        let (component, tests) = self.try_version(id, meta, wasm, source).await?;
+        // Lifecycle requests hold the same op lock, so a stopped actor cannot
+        // start between this decision and the supervisor's installation.
+        let run_tests = !preserve_stopped || self.meta(id).is_some_and(|m| m.running);
+        let (component, tests) = self.try_version(id, meta, wasm, source, run_tests).await?;
         let (version, requester) = (meta.version, meta.requested_by.clone());
-        let mut result = self.control(id, |reply| Control::Upgrade { version, component, requester, reply }).await?;
+        let mut result = self.control(id, |reply| Control::Upgrade { version, component, requester, preserve_stopped, reply }).await?;
         result["tests"] = tests;
         Ok(result)
     }
@@ -517,7 +524,7 @@ impl Broker {
         let source = source.is_dir().then_some(source.as_path());
         let result = match installed {
             None => self.create_version(id, &vmeta, &wasm, source).await,
-            Some(_) => self.upgrade_version(id, &vmeta, &wasm, source).await,
+            Some(_) => self.upgrade_version(id, &vmeta, &wasm, source, false).await,
         };
         Some(result.unwrap_or_else(|e| json!({"type": "error", "actor": id, "reason": format!("{e:#}")})))
     }
@@ -564,7 +571,11 @@ impl Broker {
             }
             "upgrade" => {
                 let (wasm, source, meta) = version_meta()?;
-                self.upgrade_actor(&actor()?, &wasm, source.as_deref(), meta).await
+                let preserve_stopped = match b.get("preserve_stopped") {
+                    None => false,
+                    Some(value) => value.as_bool().ok_or_else(|| anyhow!("preserve_stopped must be a boolean"))?,
+                };
+                self.upgrade_actor(&actor()?, &wasm, source.as_deref(), meta, preserve_stopped).await
             }
             "rollback" => {
                 let id = actor()?;
@@ -573,8 +584,15 @@ impl Broker {
                 let _guard = lock.lock().await;
                 self.control(&id, |reply| Control::Rollback { version, reply }).await
             }
-            "stop" => self.control(&actor()?, |reply| Control::Stop { reply }).await,
+            "stop" | "start" => {
-            "start" => self.control(&actor()?, |reply| Control::Start { reply }).await,
+                let id = actor()?;
+                let lock = self.op_lock(&id);
+                let _guard = lock.lock().await;
+                self.control(&id, |reply| match ty {
+                    "stop" => Control::Stop { reply },
+                    _ => Control::Start { reply },
+                }).await
+            },
             "remove" => self.remove_actor(&actor()?).await,
             "subscribe" | "unsubscribe" => {
                 let publisher = actor()?;

cli/src/console.rs

@@ -3,7 +3,8 @@
 //! - `@<actor> <json>` sends a JSON body. `@<actor> <text>` sends
 //!   `{"type": "text", "text": ...}`.
 //! - A line without `@` goes to `agent` as text.
-//! - `/help`, `/quit`. EOF on stdin also quits.
+//! - `/help`, `/quit`. EOF on stdin also quits (not with `--headless`, which
+//!   never reads stdin).
 
 use std::sync::Arc;
 

cli/src/main.rs

@@ -5,6 +5,7 @@ use anyhow::{Context, Result, bail};
 use clap::{Parser, Subcommand};
 use include_dir::{Dir, include_dir};
 use serde_json::json;
+use tokio::signal::unix::Signal;
 
 use sausage::world::{ActorMeta, World};
 use sausage::{Host, cargo_actor, console};
@@ -17,6 +18,11 @@ static AGENT_SOURCE: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../agent");
 #[derive(Parser)]
 #[command(version, about)]
 struct Cli {
+    /// Do not read stdin, so end of input does not stop the world. It runs
+    /// until SIGTERM (or SIGINT), or until an actor sends
+    /// `@broker {"type":"exit"}`. Also set by SAUSAGE_HEADLESS=1.
+    #[arg(long, global = true)]
+    headless: bool,
     #[command(subcommand)]
     command: Option<Cmd>,
 }
@@ -91,13 +97,41 @@ async fn init(dir: PathBuf) -> Result<()> {
     Ok(())
 }
 
-async fn run(dir: PathBuf) -> Result<()> {
+/// Install the SIGTERM and SIGINT handlers now (not when first polled).
+fn termination_signals() -> Result<(Signal, Signal)> {
+    use tokio::signal::unix::{SignalKind, signal};
+    let term = signal(SignalKind::terminate()).context("install SIGTERM handler")?;
+    let int = signal(SignalKind::interrupt()).context("install SIGINT handler")?;
+    Ok((term, int))
+}
+
+/// Resolves on SIGTERM or SIGINT.
+async fn termination((mut term, mut int): (Signal, Signal)) {
+    tokio::select! {
+        _ = term.recv() => {}
+        _ = int.recv() => {}
+    }
+}
+
+async fn run(dir: PathBuf, headless: bool) -> Result<()> {
+    // Register the signal handlers before the host starts, so a signal that
+    // arrives early is never the default (kill) action.
+    let signals = if headless { Some(termination_signals()?) } else { None };
     let host = Host::start(&dir).await?;
     tokio::spawn(console::run_mailbox(host.mailbox(console::ID)));
     let broker = host.broker.clone();
-    let exit = tokio::select! {
+    let exit = match signals {
-        _ = console::run_input(host.broker.clone()) => false,
+        Some(signals) => {
-        _ = broker.exit_requested() => true,
+            println!("Sausage headless: running until SIGTERM or an exit request.");
+            tokio::select! {
+                _ = termination(signals) => false,
+                _ = broker.exit_requested() => true,
+            }
+        }
+        None => tokio::select! {
+            _ = console::run_input(host.broker.clone()) => false,
+            _ = broker.exit_requested() => true,
+        },
     };
     println!("Stopping actors …");
     host.stop().await;
@@ -113,16 +147,18 @@ async fn run(dir: PathBuf) -> Result<()> {
 
 #[tokio::main]
 async fn main() -> Result<()> {
-    match Cli::parse().command {
+    let cli = Cli::parse();
+    let headless = cli.headless || std::env::var("SAUSAGE_HEADLESS").is_ok_and(|v| !v.is_empty() && v != "0");
+    match cli.command {
         Some(Cmd::Init { dir }) => init(dir).await,
-        Some(Cmd::Run { dir }) => run(dir).await,
+        Some(Cmd::Run { dir }) => run(dir, headless).await,
         None => {
             let dir = PathBuf::from("world");
             if !dir.join("actor").is_dir() {
                 println!("No world in ./world yet, creating it (same as `sausage init`).");
                 init(dir.clone()).await?;
             }
-            run(dir).await
+            run(dir, headless).await
         }
     }
 }

cli/src/supervisor.rs

@@ -23,7 +23,7 @@ pub type Reply = oneshot::Sender<Result<Value>>;
 
 pub enum Control {
     /// Swap to `version` (already in `versions/`, compiled as `component`).
-    Upgrade { version: u64, component: Component, requester: String, reply: Reply },
+    Upgrade { version: u64, component: Component, requester: String, preserve_stopped: bool, reply: Reply },
     Rollback { version: u64, reply: Reply },
     Stop { reply: Reply },
     Start { reply: Reply },
@@ -221,24 +221,29 @@ impl Supervisor {
     /// Make `version` current. It is compiled first (unless `component` is
     /// given): if that fails, nothing changes. `data/` and `public/` stay as
     /// they are.
-    async fn install(&mut self, version: u64, component: Option<Component>) -> Result<()> {
+    async fn install(&mut self, version: u64, component: Option<Component>, running: bool) -> Result<()> {
         let component = match component {
             Some(c) => c,
             None => self.broker.load_version(&self.id, version)?,
         };
         self.stop().await;
-        self.broker.install(&self.id, version)?;
+        self.broker.install(&self.id, version, running)?;
         self.version = version;
         self.component = Some(component);
-        self.running = true;
+        self.running = running;
         self.restarts = 0;
         self.restart_at = None;
         Ok(())
     }
 
-    async fn upgrade(&mut self, version: u64, component: Component, requester: String) -> Result<Value> {
+    async fn upgrade(&mut self, version: u64, component: Component, requester: String, preserve_stopped: bool) -> Result<Value> {
         let previous = self.version;
-        self.install(version, Some(component)).await?;
+        let running = self.running || !preserve_stopped;
+        self.install(version, Some(component), running).await?;
+        if !running {
+            self.probation = None;
+            return Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous}));
+        }
         self.probation = Some(Probation { previous, since: Instant::now(), crashes: 0, requester });
         if let Err(e) = self.start().await {
             self.rollback(previous).await?;
@@ -249,7 +254,7 @@ impl Supervisor {
 
     async fn rollback(&mut self, version: u64) -> Result<Value> {
         let from = self.version;
-        self.install(version, None).await?;
+        self.install(version, None, true).await?;
         self.probation = None;
         self.start().await?;
         Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": version}))
@@ -258,8 +263,8 @@ impl Supervisor {
     /// Handle a control request. Returns `true` when the supervisor ends.
     async fn handle(&mut self, ctl: Control) -> bool {
         match ctl {
-            Control::Upgrade { version, component, requester, reply } => {
+            Control::Upgrade { version, component, requester, preserve_stopped, reply } => {
-                let _ = reply.send(self.upgrade(version, component, requester).await);
+                let _ = reply.send(self.upgrade(version, component, requester, preserve_stopped).await);
             }
             Control::Rollback { version, reply } => {
                 let _ = reply.send(self.rollback(version).await);

cli/tests/actors/toolbox/src/main.rs

@@ -38,6 +38,10 @@ fn main() {
         panic!("version 2 crashes on startup");
     }
 
+    if actor == "stopped-upgrade" {
+        std::fs::write("/data/main-version", &version).unwrap();
+    }
+
     let mut open = HashMap::<String, std::fs::File>::new();
     // Files that this actor serves at /actor/<id>/fs/ (path -> text).
     let mut served = std::collections::BTreeMap::<String, String>::new();

cli/tests/cli.rs

@@ -39,3 +39,47 @@ fn bare_command_without_a_world_starts_init() {
     // It asks before it creates anything.
     assert!(!dir.join("world").exists());
 }
+
+/// Start a headless host on a fresh world with stdin at EOF; returns once it
+/// printed its ready line (printed after the signal handlers are installed).
+fn start_headless(label: &str) -> (std::process::Child, std::io::Lines<std::io::BufReader<std::process::ChildStdout>>) {
+    use std::io::{BufRead, BufReader};
+    let dir = empty_dir(label);
+    let world = dir.join("world");
+    sausage::World::create(&world).unwrap();
+    let mut child = Command::new(env!("CARGO_BIN_EXE_sausage"))
+        .args(["run", "--headless"])
+        .arg(&world)
+        .stdin(Stdio::null())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::null())
+        .spawn()
+        .unwrap();
+    let mut lines = BufReader::new(child.stdout.take().unwrap()).lines();
+    let line = lines.next().expect("host exited before it started").unwrap();
+    assert!(line.contains("headless"), "unexpected first line: {line}");
+    (child, lines)
+}
+
+/// With stdin at EOF (as soon as the host starts), a headless host stays up.
+#[test]
+fn headless_ignores_stdin_eof() {
+    let (mut child, _lines) = start_headless("headless-eof");
+    std::thread::sleep(std::time::Duration::from_millis(500));
+    let alive = child.try_wait().unwrap().is_none();
+    let _ = child.kill();
+    let _ = child.wait();
+    assert!(alive, "headless host exited on stdin EOF");
+}
+
+/// SIGTERM right after the ready line stops the actors and exits with 0.
+#[test]
+fn headless_stops_cleanly_on_sigterm_right_after_ready() {
+    let (mut child, lines) = start_headless("headless-term");
+    let pid = child.id().to_string();
+    assert!(Command::new("kill").args(["-TERM", &pid]).status().unwrap().success());
+    let rest: Vec<String> = lines.map(|l| l.unwrap()).collect();
+    let status = child.wait().unwrap();
+    assert!(status.success(), "exit status {status:?}, output {rest:?}");
+    assert!(rest.iter().any(|l| l.contains("Stopping actors")), "output {rest:?}");
+}

cli/tests/host.rs

@@ -436,6 +436,113 @@ async fn upgrades_test_candidates_keep_old_versions_and_queue_messages_in_order(
     host.stop().await;
 }
 
+#[tokio::test]
+async fn stopped_preserving_upgrade_installs_without_execution_and_keeps_all_mail() {
+    let world = create_world("stopped-upgrade");
+    let wasm = toolbox_wasm().clone();
+    let (host, mut mailbox) = start(&world).await;
+    let source = passing_source(&world, "initial");
+    spawn_type(&host, &mut mailbox, "stopped-upgrade", &wasm, &source).await;
+    ask_type(&host, &mut mailbox, "stopped-upgrade", json!({"type": "version"}), "version").await;
+    let stopped = request(&host, &mut mailbox, json!({"type": "stop", "actor": "stopped-upgrade"})).await;
+    assert_eq!(stopped.body["type"], "stopped");
+
+    // A distinct valid component: append an empty custom section named x.
+    let mut bytes = std::fs::read(&wasm).unwrap();
+    bytes.extend_from_slice(&[0, 2, 1, b'x']);
+    let new_wasm = world.public_dir("console").join("new.wasm");
+    std::fs::write(&new_wasm, &bytes).unwrap();
+    // This would fail if candidate main/tests were executed.
+    let unrun = source_with_test(&world, "unrun", json!({"steps": [{
+        "send": {"type": "echo"}, "expect": {"type": "must-not-run"}
+    }]}));
+    std::fs::write(unrun.join("new-source"), "version two").unwrap();
+    std::fs::write(world.public_dir("stopped-upgrade").join("keep"), "public state").unwrap();
+
+    let mut ids = Vec::new();
+    for n in 0..sausage::broker::MAILBOX_SIZE {
+        ids.push(host.broker.send("console", Some("stopped-upgrade"), None, None, json!({"type": "ordered", "n": n})));
+    }
+    let upgrade_id = host.broker.send("console", Some("broker"), None, None, json!({
+        "type": "upgrade", "actor": "stopped-upgrade", "wasm": new_wasm.display().to_string(),
+        "source": unrun.display().to_string(), "preserve_stopped": true
+    }));
+    // Do not silently discard unexpected actor mail while awaiting the reply.
+    let upgraded = receive_message(&mut mailbox, Duration::from_secs(10)).await;
+    assert_eq!(upgraded.reply_to.as_deref(), Some(upgrade_id.as_str()));
+    assert_eq!(upgraded.body["type"], "upgraded");
+    assert_eq!(upgraded.body["version"], 2);
+    assert_eq!(upgraded.body["previous"], 1);
+    assert_eq!(upgraded.body["tests"], json!({"skipped": "preserve_stopped"}));
+    let meta = world.read_meta("stopped-upgrade");
+    assert_eq!(meta.version, Some(2));
+    assert!(!meta.running);
+    assert_eq!(std::fs::read(world.wasm_path("stopped-upgrade")).unwrap(), bytes);
+    assert_eq!(std::fs::read(world.version_dir("stopped-upgrade", 2).join("actor.wasm")).unwrap(), bytes);
+    assert_ne!(std::fs::read(world.version_dir("stopped-upgrade", 1).join("actor.wasm")).unwrap(), bytes);
+    assert_eq!(std::fs::read_to_string(world.source_dir("stopped-upgrade").join("new-source")).unwrap(), "version two");
+    assert_eq!(std::fs::read_to_string(world.data_dir("stopped-upgrade").join("main-version")).unwrap(), "1");
+    assert_eq!(std::fs::read_to_string(world.public_dir("stopped-upgrade").join("keep")).unwrap(), "public state");
+    assert!(mailbox.try_recv().is_err());
+
+    // The full mailbox is still full: nothing was consumed or added by the upgrade.
+    let overflow_id = host.broker.send("console", Some("stopped-upgrade"), None, None, json!({"type": "echo"}));
+    let overflow = receive_message(&mut mailbox, Duration::from_secs(10)).await;
+    assert_eq!(overflow.reply_to.as_deref(), Some(overflow_id.as_str()));
+    assert_eq!(overflow.body["type"], "undeliverable");
+    assert!(overflow.body["reason"].as_str().unwrap().contains("full"));
+
+    // Only an explicit start executes v2. Every original envelope survives in order.
+    let start_id = host.broker.send("console", Some("broker"), None, None, json!({"type": "start", "actor": "stopped-upgrade"}));
+    let mut started = false;
+    let mut count = 0;
+    tokio::time::timeout(Duration::from_secs(10), async {
+        while !started || count < ids.len() {
+            let reply = mailbox.recv().await.unwrap();
+            if reply.reply_to.as_deref() == Some(start_id.as_str()) {
+                assert_eq!(reply.body["type"], "started");
+                started = true;
+            } else {
+                assert_eq!(reply.from, "stopped-upgrade");
+                assert_eq!(reply.reply_to.as_deref(), Some(ids[count].as_str()));
+                assert_eq!(reply.body, json!({"type": "ordered", "n": count, "version": "2"}));
+                count += 1;
+            }
+        }
+    }).await.expect("queued mail was not delivered after start");
+    assert_eq!(std::fs::read_to_string(world.data_dir("stopped-upgrade").join("main-version")).unwrap(), "2");
+    host.stop().await;
+}
+
+#[tokio::test]
+async fn stopped_preserving_option_keeps_running_upgrades_tested_and_default_upgrades_start() {
+    let world = create_world("preserve-running");
+    let wasm = toolbox_wasm().clone();
+    let (host, mut mailbox) = start(&world).await;
+    let passing = passing_source(&world, "passing");
+    let failing = source_with_test(&world, "failing", json!({"steps": [{
+        "send": {"type": "echo"}, "expect": {"type": "wrong"}
+    }]}));
+    spawn_type(&host, &mut mailbox, "running", &wasm, &passing).await;
+    for (option, source, expected) in [(json!("true"), &passing, "error"), (json!(true), &failing, "error"), (json!(true), &passing, "upgraded")] {
+        let reply = request(&host, &mut mailbox, json!({
+            "type": "upgrade", "actor": "running", "wasm": wasm.display().to_string(),
+            "source": source.display().to_string(), "preserve_stopped": option
+        })).await;
+        assert_eq!(reply.body["type"], expected);
+    }
+    assert!(world.read_meta("running").running);
+    let version = ask_type(&host, &mut mailbox, "running", json!({"type": "version"}), "version").await;
+    assert_eq!(version.body["version"], "2");
+    request(&host, &mut mailbox, json!({"type": "stop", "actor": "running"})).await;
+    let default = upgrade_actor(&host, &mut mailbox, "running", &wasm, &passing).await;
+    assert_eq!(default.body["type"], "upgraded");
+    assert!(world.read_meta("running").running);
+    let version = ask_type(&host, &mut mailbox, "running", json!({"type": "version"}), "version").await;
+    assert_eq!(version.body["version"], "3");
+    host.stop().await;
+}
+
 #[tokio::test]
 async fn rolls_back_a_crash_loop_to_the_previous_version_without_restoring_data() {
     let world = create_world("crash-loop");