sausage/cli/tests/host.rs

mod common;

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

use common::*;
use sausage::fs::TMP_PREFIX;
use sausage::{Envelope, Host, World};
use serde_json::{Value, json};
use tokio::sync::mpsc;

async fn filesystem_world(label: &str) -> (World, Host, mpsc::Receiver<Envelope>) {
    let world = create_world(label);
    let wasm = toolbox_wasm().clone();
    let (host, mut mailbox) = start(&world).await;
    for actor in ["writer", "reader"] {
        let source = passing_source(&world, actor);
        let spawned = spawn_actor(&host, &mut mailbox, actor, &wasm, &source).await;
        assert_eq!(spawned.body["type"], "spawned");
    }
    (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");
    let wasm = toolbox_wasm().clone();
    let (host, mut mailbox) = start(&world).await;

    let missing_id = host.broker.send(
        "console",
        Some("missing"),
        None,
        None,
        json!({"type": "hello"}),
    );
    let missing = receive_reply(&mut mailbox, &missing_id, Duration::from_secs(10)).await;
    assert_eq!(missing.body["type"], "undeliverable");
    assert_eq!(missing.body["to"], "missing");

    let source = passing_source(&world, "router");
    let spawned = spawn_actor(&host, &mut mailbox, "router", &wasm, &source).await;
    assert_eq!(spawned.body["type"], "spawned");

    let echo = ask(
        &host,
        &mut mailbox,
        "router",
        json!({"type": "echo", "value": "hello"}),
    )
    .await;
    assert_eq!(echo.from, "router");
    assert_eq!(echo.body, json!({"type": "echo", "value": "hello"}));

    let subscribed = request(
        &host,
        &mut mailbox,
        json!({"type": "subscribe", "actor": "router"}),
    )
    .await;
    assert_eq!(subscribed.body["type"], "subscribed");

    host.broker.send(
        "console",
        Some("router"),
        None,
        None,
        json!({"type": "publish", "value": "news"}),
    );
    let published = receive_message(&mut mailbox, Duration::from_secs(10)).await;
    assert_eq!(published.from, "router");
    assert_eq!(
        published.body,
        json!({"type": "published", "value": "news"})
    );
    assert_eq!(published.reply_to, None);

    host.stop().await;
}

#[tokio::test]
async fn filesystem_permissions_and_public_sharing() {
    let (_, host, mut mailbox) = filesystem_world("filesystem-permissions").await;
    for path in [
        "/actor/reader/public/denied.txt",
        "/actor/writer/public/denied.txt",
    ] {
        let denied = ask(
            &host,
            &mut mailbox,
            "writer",
            json!({"type": "write", "path": path, "text": "denied"}),
        )
        .await;
        assert_eq!(denied.body["type"], "error", "write to {path} succeeded");
    }

    let written = ask(
        &host,
        &mut mailbox,
        "writer",
        json!({"type": "write", "path": "/public/shared.txt", "text": "public"}),
    )
    .await;
    assert_eq!(written.body["type"], "ok");
    let shared = ask(
        &host,
        &mut mailbox,
        "reader",
        json!({"type": "read", "path": "/actor/writer/public/shared.txt"}),
    )
    .await;
    assert_eq!(shared.body, json!({"type": "content", "text": "public"}));
    host.stop().await;
}

#[tokio::test]
async fn filesystem_hides_new_and_truncated_writes_until_close() {
    let (_, host, mut mailbox) = filesystem_world("filesystem-commit").await;
    let opened = ask(
        &host,
        &mut mailbox,
        "writer",
        json!({"type": "open", "path": "/public/new.txt", "text": "new"}),
    )
    .await;
    assert_eq!(opened.body["type"], "opened");
    let invisible = ask(
        &host,
        &mut mailbox,
        "reader",
        json!({"type": "read", "path": "/actor/writer/public/new.txt"}),
    )
    .await;
    assert_eq!(invisible.body["type"], "error");
    let listing = ask(
        &host,
        &mut mailbox,
        "writer",
        json!({"type": "list", "path": "/public"}),
    )
    .await;
    let names = listing.body["names"].as_array().unwrap();
    assert!(
        names
            .iter()
            .all(|name| !name.as_str().unwrap().starts_with(TMP_PREFIX))
    );
    assert!(names.iter().all(|name| name.as_str() != Some("new.txt")));
    let closed = ask(
        &host,
        &mut mailbox,
        "writer",
        json!({"type": "close", "path": "/public/new.txt"}),
    )
    .await;
    assert_eq!(closed.body["closed"], true);
    let visible = ask(
        &host,
        &mut mailbox,
        "reader",
        json!({"type": "read", "path": "/actor/writer/public/new.txt"}),
    )
    .await;
    assert_eq!(visible.body["text"], "new");

    ask(
        &host,
        &mut mailbox,
        "writer",
        json!({"type": "write", "path": "/public/truncated.txt", "text": "old"}),
    )
    .await;
    ask(
        &host,
        &mut mailbox,
        "writer",
        json!({"type": "open", "path": "/public/truncated.txt", "text": "new"}),
    )
    .await;
    let old_during_truncate = ask(
        &host,
        &mut mailbox,
        "reader",
        json!({"type": "read", "path": "/actor/writer/public/truncated.txt"}),
    )
    .await;
    assert_eq!(old_during_truncate.body["text"], "old");
    ask(
        &host,
        &mut mailbox,
        "writer",
        json!({"type": "close", "path": "/public/truncated.txt"}),
    )
    .await;
    let new_after_truncate = ask(
        &host,
        &mut mailbox,
        "reader",
        json!({"type": "read", "path": "/actor/writer/public/truncated.txt"}),
    )
    .await;
    assert_eq!(new_after_truncate.body["text"], "new");
    host.stop().await;
}

#[tokio::test]
async fn filesystem_append_is_visible_before_close() {
    let (_, host, mut mailbox) = filesystem_world("filesystem-append").await;
    ask(
        &host,
        &mut mailbox,
        "writer",
        json!({"type": "write", "path": "/public/append.txt", "text": "base"}),
    )
    .await;
    ask(
        &host,
        &mut mailbox,
        "writer",
        json!({"type": "append", "path": "/public/append.txt", "text": "+tail"}),
    )
    .await;
    let appended = ask(
        &host,
        &mut mailbox,
        "reader",
        json!({"type": "read", "path": "/actor/writer/public/append.txt"}),
    )
    .await;
    assert_eq!(appended.body["text"], "base+tail");
    let closed = ask(
        &host,
        &mut mailbox,
        "writer",
        json!({"type": "close", "path": "/public/append.txt"}),
    )
    .await;
    assert_eq!(closed.body["closed"], true);
    host.stop().await;
}

#[tokio::test]
async fn filesystem_crash_discards_an_open_truncated_write() {
    let (world, host, mut mailbox) = filesystem_world("filesystem-crash").await;
    ask(
        &host,
        &mut mailbox,
        "writer",
        json!({"type": "write", "path": "/public/crash.txt", "text": "old"}),
    )
    .await;
    ask(
        &host,
        &mut mailbox,
        "writer",
        json!({"type": "open", "path": "/public/crash.txt", "text": "uncommitted"}),
    )
    .await;
    let crash_id = host.broker.send(
        "console",
        Some("writer"),
        None,
        None,
        json!({"type": "crash"}),
    );
    wait_for_parked(&world, "writer", &crash_id).await;
    let after_crash = ask(
        &host,
        &mut mailbox,
        "reader",
        json!({"type": "read", "path": "/actor/writer/public/crash.txt"}),
    )
    .await;
    assert_eq!(after_crash.body["text"], "old");
    host.stop().await;
}

#[tokio::test]
async fn filesystem_rejects_paths_that_escape_data() {
    let (_, host, mut mailbox) = filesystem_world("filesystem-path-escape").await;
    let escaped = ask(
        &host,
        &mut mailbox,
        "writer",
        json!({"type": "write", "path": "/data/../../..", "text": "outside"}),
    )
    .await;
    assert_eq!(escaped.body["type"], "error");
    host.stop().await;
}

#[tokio::test]
async fn upgrades_test_candidates_keep_old_versions_and_queue_messages_in_order() {
    let world = create_world("upgrade");
    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", "value": "candidate"},
                "expect": {"type": "wrong"}
            }]
        }),
    );
    assert_eq!(
        spawn_actor(&host, &mut mailbox, "queue", &wasm, &passing)
            .await
            .body["type"],
        "spawned"
    );

    let failed = upgrade_actor(&host, &mut mailbox, "queue", &wasm, &failing).await;
    assert_eq!(failed.body["type"], "error");
    let old_version = ask(&host, &mut mailbox, "queue", json!({"type": "version"})).await;
    assert_eq!(old_version.body["version"], "1");

    let upgrade_id = host.broker.send(
        "console",
        Some("broker"),
        None,
        None,
        json!({"type": "upgrade", "actor": "queue", "wasm": wasm.display().to_string(), "source": passing.display().to_string()}),
    );
    tokio::time::timeout(Duration::from_secs(10), async {
        while !world.data_dir("queue").join("shutdown").is_file() {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await
    .expect("old actor did not enter shutdown during upgrade");

    let mut message_ids = Vec::new();
    for n in 1..=3 {
        message_ids.push(host.broker.send(
            "console",
            Some("queue"),
            None,
            None,
            json!({"type": "ordered", "n": n}),
        ));
    }

    let mut upgrade_result = None;
    let mut actor_replies = Vec::new();
    tokio::time::timeout(Duration::from_secs(15), async {
        while upgrade_result.is_none() || actor_replies.len() < message_ids.len() {
            let envelope = mailbox.recv().await.expect("console mailbox closed");
            if envelope.reply_to.as_deref() == Some(upgrade_id.as_str()) {
                upgrade_result = Some(envelope);
            } else if message_ids
                .iter()
                .any(|id| envelope.reply_to.as_deref() == Some(id.as_str()))
            {
                actor_replies.push(envelope);
            } else {
                panic!("unexpected message during queued upgrade: {envelope:?}");
            }
        }
    })
    .await
    .expect("upgrade did not deliver queued messages");

    assert_eq!(upgrade_result.unwrap().body["type"], "upgraded");
    let actual_ids: Vec<String> = actor_replies
        .iter()
        .map(|message| message.reply_to.clone().unwrap())
        .collect();
    assert_eq!(actual_ids, message_ids);
    for (index, envelope) in actor_replies.iter().enumerate() {
        assert_eq!(envelope.from, "queue");
        assert_eq!(envelope.body["type"], "ordered");
        assert_eq!(envelope.body["n"], json!(index + 1));
        assert_eq!(envelope.body["version"], "3");
    }

    host.stop().await;
}

#[tokio::test]
async fn redelivers_a_crashed_message_after_the_next_upgrade() {
    let world = create_world("redelivery");
    let wasm = toolbox_wasm().clone();
    let (host, mut mailbox) = start(&world).await;
    let source = passing_source(&world, "source");
    assert_eq!(
        spawn_actor(&host, &mut mailbox, "parked", &wasm, &source)
            .await
            .body["type"],
        "spawned"
    );

    let crash_id = host.broker.send(
        "console",
        Some("parked"),
        None,
        None,
        json!({"type": "crash"}),
    );
    wait_for_parked(&world, "parked", &crash_id).await;
    assert!(matches!(
        mailbox.try_recv(),
        Err(tokio::sync::mpsc::error::TryRecvError::Empty)
    ));

    let upgrade_id = host.broker.send(
        "console",
        Some("broker"),
        None,
        None,
        json!({"type": "upgrade", "actor": "parked", "wasm": wasm.display().to_string(), "source": source.display().to_string()}),
    );
    let mut upgraded = None;
    let mut replayed = None;
    tokio::time::timeout(Duration::from_secs(15), async {
        while upgraded.is_none() || replayed.is_none() {
            let envelope = mailbox.recv().await.expect("console mailbox closed");
            if envelope.reply_to.as_deref() == Some(upgrade_id.as_str()) {
                upgraded = Some(envelope);
            } else if envelope.reply_to.as_deref() == Some(crash_id.as_str()) {
                replayed = Some(envelope);
            } else {
                panic!("unexpected message during parked delivery: {envelope:?}");
            }
        }
    })
    .await
    .expect("upgrade did not redeliver the parked message");

    assert_eq!(upgraded.unwrap().body["type"], "upgraded");
    assert_eq!(
        replayed.unwrap().body,
        json!({"type": "replayed", "version": "2"})
    );
    host.stop().await;
}

#[tokio::test]
async fn rolls_back_a_crash_loop_and_restores_its_data_snapshot() {
    let world = create_world("crash-loop");
    let wasm = toolbox_wasm().clone();
    let (host, mut mailbox) = start(&world).await;
    let source = passing_source(&world, "source");
    assert_eq!(
        spawn_actor(&host, &mut mailbox, "crashloop", &wasm, &source)
            .await
            .body["type"],
        "spawned"
    );
    assert_eq!(
        ask(
            &host,
            &mut mailbox,
            "crashloop",
            json!({"type": "write", "path": "/data/state", "text": "stable"})
        )
        .await
        .body["type"],
        "ok"
    );

    let upgrade_id = host.broker.send(
        "console",
        Some("broker"),
        None,
        None,
        json!({"type": "upgrade", "actor": "crashloop", "wasm": wasm.display().to_string(), "source": source.display().to_string()}),
    );
    let mut upgraded: Option<Value> = None;
    let mut rolled_back: Option<Value> = None;
    tokio::time::timeout(Duration::from_secs(20), async {
        while upgraded.is_none() || rolled_back.is_none() {
            let envelope = mailbox.recv().await.expect("console mailbox closed");
            let kind = envelope.body["type"].as_str().map(str::to_string);
            match kind.as_deref() {
                Some("upgraded") if envelope.reply_to.as_deref() == Some(upgrade_id.as_str()) => {
                    upgraded = Some(envelope.body);
                }
                Some("rolled_back") => rolled_back = Some(envelope.body),
                other => panic!("unexpected automatic rollback message: {other:?}, {envelope:?}"),
            }
        }
    })
    .await
    .expect("crash loop did not trigger automatic rollback");

    assert_eq!(upgraded.unwrap()["version"], 2);
    let rollback = rolled_back.unwrap();
    assert_eq!(rollback["actor"], "crashloop");
    assert_eq!(rollback["from"], 2);
    assert_eq!(rollback["to"], 1);
    assert_eq!(rollback["reason"], "crash loop after upgrade");

    let version = ask(&host, &mut mailbox, "crashloop", json!({"type": "version"})).await;
    assert_eq!(version.body["version"], "1");
    let restored = ask(
        &host,
        &mut mailbox,
        "crashloop",
        json!({"type": "read", "path": "/data/state"}),
    )
    .await;
    assert_eq!(restored.body["text"], "stable");

    host.stop().await;
}

#[tokio::test]
async fn explicit_rollback_restores_the_previous_version_and_snapshot() {
    let world = create_world("explicit-rollback");
    let wasm = toolbox_wasm().clone();
    let (host, mut mailbox) = start(&world).await;
    let source = passing_source(&world, "source");
    assert_eq!(
        spawn_actor(&host, &mut mailbox, "recover", &wasm, &source)
            .await
            .body["type"],
        "spawned"
    );
    ask(
        &host,
        &mut mailbox,
        "recover",
        json!({"type": "write", "path": "/data/state", "text": "version-one"}),
    )
    .await;

    let upgraded = upgrade_actor(&host, &mut mailbox, "recover", &wasm, &source).await;
    assert_eq!(upgraded.body["type"], "upgraded");
    ask(
        &host,
        &mut mailbox,
        "recover",
        json!({"type": "write", "path": "/data/state", "text": "version-two"}),
    )
    .await;

    let rolled_back = request(
        &host,
        &mut mailbox,
        json!({"type": "rollback", "actor": "recover", "version": 1}),
    )
    .await;
    assert_eq!(rolled_back.body["type"], "rolled_back");
    assert_eq!(rolled_back.body["from"], 2);
    assert_eq!(rolled_back.body["to"], 1);
    assert_eq!(rolled_back.body["restored_state"], true);
    let version = ask(&host, &mut mailbox, "recover", json!({"type": "version"})).await;
    assert_eq!(version.body["version"], "1");
    let restored = ask(
        &host,
        &mut mailbox,
        "recover",
        json!({"type": "read", "path": "/data/state"}),
    )
    .await;
    assert_eq!(restored.body["text"], "version-one");

    host.stop().await;
}

#[tokio::test]
async fn cargo_actor_builds_valid_crates_and_reports_compiler_errors() {
    let world = create_world("cargo-actor");
    let (host, mut mailbox) = start(&world).await;
    let valid_source = cargo_source("cargo-valid");
    let valid_id = host.broker.send(
        "console",
        Some("cargo"),
        None,
        None,
        json!({"type": "build", "source": valid_source.display().to_string()}),
    );
    let built = receive_reply(&mut mailbox, &valid_id, Duration::from_secs(120)).await;
    assert_eq!(built.body["type"], "built");
    let wasm = built.body["wasm"].as_str().unwrap();
    assert!(wasm.ends_with(".wasm"));
    assert!(host.world.resolve("console", wasm).unwrap().is_file());

    let invalid_source = cargo_source("cargo-fail");
    let invalid_id = host.broker.send(
        "console",
        Some("cargo"),
        None,
        None,
        json!({"type": "build", "source": invalid_source.display().to_string()}),
    );
    let failed = receive_reply(&mut mailbox, &invalid_id, Duration::from_secs(120)).await;
    assert_eq!(failed.body["type"], "build_failed");
    assert!(
        failed.body["errors"]
            .as_str()
            .unwrap()
            .contains("missing_function")
    );

    host.stop().await;
}

#[tokio::test]
async fn rejects_invalid_actor_ids() {
    let world = create_world("invalid-id");
    let wasm = toolbox_wasm().clone();
    let (host, mut mailbox) = start(&world).await;
    let invalid = request(
        &host,
        &mut mailbox,
        json!({"type": "spawn", "actor": "bad/actor", "wasm": wasm.display().to_string()}),
    )
    .await;
    assert_eq!(invalid.body["type"], "error");
    assert!(
        invalid.body["reason"]
            .as_str()
            .unwrap()
            .contains("invalid actor ID")
    );
    host.stop().await;
}

#[tokio::test]
async fn rejects_parent_paths() {
    let world = create_world("parent-path");
    let (host, mut mailbox) = start(&world).await;
    let parent_path = request(
        &host,
        &mut mailbox,
        json!({"type": "spawn", "actor": "safe", "wasm": "/public/../escape.wasm"}),
    )
    .await;
    assert_eq!(parent_path.body["type"], "error");
    assert!(
        parent_path.body["reason"]
            .as_str()
            .unwrap()
            .contains("must not contain '..'")
    );
    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;
}