sausage core: modules per topic, journal writer thread with WAL and batches, hash cache by metadata, 49 tests per rule

Blank stdout lines give no warning. core.exit requests the exit in its
own handler.

cli/src/cargo.rs

@@ -11,7 +11,9 @@ use include_dir::{Dir, include_dir};
 
 use crate::world::{self, World};
 
+/// The name reserved for the native Rust build plugin.
 pub const ID: &str = "cargo";
+/// The WASI component target used for Rust plugin builds.
 pub const TARGET: &str = "wasm32-wasip2";
 
 /// The source of the `sausage-actor` SDK.
@@ -19,23 +21,34 @@ static SDK_SRC: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../sdk/src");
 static SDK_MACROS: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../sdk/macros/src");
 
 #[derive(Clone, Default, Debug)]
+/// The totals and output from a source tree's test run.
 pub struct TestResults {
+    /// The number of tests that passed.
     pub passed: u64,
+    /// The number of tests that failed.
     pub failed: u64,
+    /// The test output retained for the caller.
     pub output: String,
 }
 
+/// The component artifact and diagnostics from a successful build.
 pub struct BuildOutput {
     /// Host path of the published `.wasm` file.
     pub wasm: PathBuf,
+    /// The elapsed build time in seconds.
     pub seconds: f64,
+    /// The compiler diagnostics without progress lines.
     pub warnings: String,
+    /// The totals and output of the test phase.
     pub tests: TestResults,
 }
 
 #[derive(Debug)]
+/// The failed build or test phase and its diagnostics.
 pub struct BuildFailure {
+    /// The reason that the build could not complete.
     pub reason: String,
+    /// The totals and output of the test phase.
     pub tests: TestResults,
 }
 

cli/src/components.rs

@@ -0,0 +1,86 @@
+//! Component versions cached by file metadata, with independent compilation per hash.
+use crate::{error, runtime::Runtime};
+use sha2::{Digest, Sha256};
+use std::{
+    collections::HashMap,
+    path::PathBuf,
+    sync::{Arc, Mutex},
+    time::SystemTime,
+};
+use tokio::sync::{Mutex as AsyncMutex, OnceCell};
+use wasmtime::component::Component;
+
+struct Version {
+    size: u64,
+    modified: SystemTime,
+    hash: String,
+    component: Component,
+}
+type PathEntry = Arc<AsyncMutex<Option<Version>>>;
+type CompiledEntry = Arc<OnceCell<Component>>;
+
+#[derive(Default)]
+pub(crate) struct Components {
+    paths: Mutex<HashMap<PathBuf, PathEntry>>,
+    compiled: Mutex<HashMap<String, CompiledEntry>>,
+}
+impl Components {
+    /// Stat each path, but read and hash its bytes only after its metadata changes.
+    pub(crate) async fn load(
+        &self,
+        path: PathBuf,
+        runtime: Arc<Runtime>,
+    ) -> Result<(Component, String), serde_json::Value> {
+        let entry = self
+            .paths
+            .lock()
+            .unwrap()
+            .entry(path.clone())
+            .or_default()
+            .clone();
+        let mut cached = entry.lock().await;
+        let metadata = tokio::fs::metadata(&path)
+            .await
+            .map_err(|_| error(-32001, "no such plugin"))?;
+        let modified = metadata
+            .modified()
+            .map_err(|e| error(-32603, &e.to_string()))?;
+        if let Some(version) = cached.as_ref()
+            && version.size == metadata.len()
+            && version.modified == modified
+        {
+            return Ok((version.component.clone(), version.hash.clone()));
+        }
+        let (bytes, hash) = tokio::task::spawn_blocking(move || -> std::io::Result<_> {
+            let bytes = std::fs::read(path)?;
+            let hash = format!("{:x}", Sha256::digest(&bytes));
+            Ok((bytes, hash))
+        })
+        .await
+        .map_err(|e| error(-32603, &e.to_string()))?
+        .map_err(|e| error(-32603, &e.to_string()))?;
+        let compiled = self
+            .compiled
+            .lock()
+            .unwrap()
+            .entry(hash.clone())
+            .or_default()
+            .clone();
+        let component = compiled
+            .get_or_try_init(|| async move {
+                tokio::task::spawn_blocking(move || runtime.compile(&bytes))
+                    .await
+                    .map_err(|e| error(-32603, &e.to_string()))?
+                    .map_err(|e| error(-32603, &e.to_string()))
+            })
+            .await?
+            .clone();
+        *cached = Some(Version {
+            size: metadata.len(),
+            modified,
+            hash: hash.clone(),
+            component: component.clone(),
+        });
+        Ok((component, hash))
+    }
+}

cli/src/console.rs

@@ -0,0 +1,24 @@
+//! The minimal console reads explicit JSON-RPC calls and the quit command.
+use crate::Host;
+use tokio::io::{AsyncBufReadExt, BufReader};
+
+/// Read console lines until EOF or quit, and print each call's result.
+pub async fn run_input(host: Host) -> anyhow::Result<()> {
+    let mut lines = BufReader::new(tokio::io::stdin()).lines();
+    while let Some(line) = lines.next_line().await? {
+        if line == "quit" {
+            break;
+        }
+        let (method, params) = line.split_once(' ').unwrap_or((&line, "{}"));
+        match serde_json::from_str(params) {
+            Ok(params) => {
+                let value = match host.call(method, params).await {
+                    Ok(value) | Err(value) => value,
+                };
+                println!("{value}");
+            }
+            Err(error) => eprintln!("{error}"),
+        }
+    }
+    Ok(())
+}

cli/src/core.rs

@@ -0,0 +1,64 @@
+//! The native core methods and their JSON parameters.
+use crate::{Host, RpcResult, error};
+use serde_json::{Value, json};
+
+impl Host {
+    /// Execute a core method without starting a guest process.
+    pub(crate) async fn core(
+        &self,
+        caller: &str,
+        id: Option<Value>,
+        method: &str,
+        p: Value,
+    ) -> RpcResult {
+        match method {
+            "call" => {
+                self.route(
+                    caller.into(),
+                    id,
+                    required_str(&p, "method")?.into(),
+                    p.get("params").cloned().unwrap_or(json!({})),
+                    p["detach"].as_bool().unwrap_or(false),
+                )
+                .await
+            }
+            "notify" => {
+                self.notify(
+                    caller,
+                    required_str(&p, "name")?,
+                    p.get("params").cloned().unwrap_or(json!({})),
+                );
+                Ok(Value::Null)
+            }
+            "kill" => {
+                self.kill(required_str(&p, "pid")?);
+                Ok(json!({}))
+            }
+            "stop" => {
+                if let Some(mut n) = self.begin_stop(required_str(&p, "pid")?) {
+                    let _ = n.wait_for(|ended| *ended).await;
+                }
+                Ok(json!({}))
+            }
+            "lock" => {
+                self.lock(caller, required_str(&p, "path")?, p["wait"] != false)
+                    .await
+            }
+            "unlock" => {
+                self.unlock(caller, required_str(&p, "path")?);
+                Ok(Value::Null)
+            }
+            "log" => Ok(self.0.journal.query(p).await),
+            "exit" => {
+                self.0.exit.send_replace(true);
+                Ok(json!({}))
+            }
+            _ => Err(error(-32601, "method not found")),
+        }
+    }
+}
+fn required_str<'a>(params: &'a Value, name: &str) -> Result<&'a str, Value> {
+    params[name]
+        .as_str()
+        .ok_or_else(|| error(-32602, &format!("{name} must be a string")))
+}

cli/src/execution.rs

@@ -0,0 +1,134 @@
+//! Execution of native futures and WASI processes, including transport drainage.
+use crate::{
+    Host, NativeContext, NativePlugin,
+    process::{ProcessGuard, ProcessInfo},
+    runtime::{Instance, InstanceConfig},
+};
+use serde_json::{Value, json};
+use tokio::sync::mpsc;
+
+impl Host {
+    /// Run one process and retain its cleanup guard until all transport reads finish.
+    pub(crate) async fn run_process(
+        &self,
+        pid: String,
+        info: ProcessInfo,
+        component: Option<wasmtime::component::Component>,
+        native: Option<NativePlugin>,
+        incoming: mpsc::UnboundedReceiver<Value>,
+        has_id: bool,
+    ) {
+        let _guard = ProcessGuard {
+            host: self.clone(),
+            pid: pid.clone(),
+        };
+        if let Some(native) = native {
+            let context = NativeContext {
+                host: self.clone(),
+                pid: pid.clone(),
+            };
+            let result = native(context, info.method, info.params).await;
+            self.answer(&pid, result);
+            self.finish(&pid, "exit 0");
+            return;
+        }
+        let result = self
+            .run_guest(&pid, &info, &component.unwrap(), incoming, has_id)
+            .await;
+        self.finish(&pid, &result.unwrap_or_else(|error| error.to_string()));
+    }
+
+    /// Drain stdout before deciding whether the guest returned an answer.
+    async fn run_guest(
+        &self,
+        pid: &str,
+        info: &ProcessInfo,
+        component: &wasmtime::component::Component,
+        mut incoming: mpsc::UnboundedReceiver<Value>,
+        has_id: bool,
+    ) -> anyhow::Result<String> {
+        let config = InstanceConfig {
+            world: self.0.world.root.clone(),
+            network: true,
+            env: vec![
+                ("SAUSAGE_PID".into(), pid.into()),
+                ("SAUSAGE_PLUGIN".into(), info.plugin.clone()),
+                (
+                    "SAUSAGE_PARENT".into(),
+                    info.parent.clone().unwrap_or_else(|| "core".into()),
+                ),
+            ],
+        };
+        let mut instance = self.0.runtime.start(component, config).await?;
+        let mut initial = json!({
+            "jsonrpc": "2.0",
+            "method": info.method,
+            "params": info.params,
+        });
+        if has_id {
+            initial["id"] = json!(pid);
+        }
+        self.write_input(pid, &mut instance, initial).await?;
+        let mut stdin_open = true;
+        let mut stderr_open = true;
+        loop {
+            tokio::select! {
+                line = instance.stderr.recv(), if stderr_open => {
+                    stderr_open = self.stderr_line(pid,&info.plugin,line);
+                }
+                line = instance.stdout.recv() => {
+                    if !self.stdout_line(pid,line) { break; }
+                }
+                value = incoming.recv(), if stdin_open => {
+                    stdin_open = self.forward_input(pid,&mut instance,value).await?;
+                }
+            }
+        }
+        while let Some(line) = instance.stderr.recv().await {
+            self.0.journal.write(pid, &info.plugin, "stderr", &line);
+        }
+        Ok(instance.wait().await.to_string())
+    }
+
+    /// Close stdin only after the queued responses reach the guest.
+    async fn forward_input(
+        &self,
+        pid: &str,
+        instance: &mut Instance,
+        value: Option<Value>,
+    ) -> std::io::Result<bool> {
+        if let Some(value) = value {
+            self.write_input(pid, instance, value).await?;
+            Ok(true)
+        } else {
+            instance.stdin.take();
+            Ok(false)
+        }
+    }
+    async fn write_input(
+        &self,
+        pid: &str,
+        instance: &mut Instance,
+        value: Value,
+    ) -> std::io::Result<()> {
+        instance
+            .send_line(&value.to_string(), |line| self.output(pid, line))
+            .await
+    }
+    fn stdout_line(&self, pid: &str, line: Option<String>) -> bool {
+        if let Some(line) = line {
+            self.output(pid, line);
+            true
+        } else {
+            false
+        }
+    }
+    fn stderr_line(&self, pid: &str, plugin: &str, line: Option<String>) -> bool {
+        if let Some(line) = line {
+            self.0.journal.write(pid, plugin, "stderr", &line);
+            true
+        } else {
+            false
+        }
+    }
+}

cli/src/host.rs

@@ -0,0 +1,131 @@
+//! Host startup, boot calls, and the public library API.
+use crate::{RpcResult, components, journal, runtime::Runtime, world::World};
+use crate::{
+    locks::Lock,
+    native::NativePlugin,
+    process::{Process, ProcessInfo},
+};
+use serde_json::{Value, json};
+use std::{
+    collections::HashMap,
+    path::Path,
+    sync::{Arc, Mutex},
+};
+use tokio::sync::watch;
+use uuid::Uuid;
+pub(crate) struct State {
+    pub(crate) processes: HashMap<String, Process>,
+    pub(crate) locks: HashMap<String, Lock>,
+    pub(crate) native: HashMap<String, NativePlugin>,
+}
+pub(crate) struct Inner {
+    pub(crate) world: World,
+    pub(crate) runtime: Arc<Runtime>,
+    pub(crate) state: Mutex<State>,
+    pub(crate) journal: journal::Journal,
+    pub(crate) notifications: tokio::sync::broadcast::Sender<Value>,
+    pub(crate) exit: watch::Sender<bool>,
+    pub(crate) components: components::Components,
+}
+#[derive(Clone)]
+/// A cloneable handle to a world and its running processes.
+pub struct Host(pub(crate) Arc<Inner>);
+
+impl Host {
+    /// Start a world with the default compile cache and its ordered boot calls.
+    pub async fn start(path: &Path) -> anyhow::Result<Self> {
+        let base = std::env::var_os("XDG_CACHE_HOME")
+            .map(std::path::PathBuf::from)
+            .unwrap_or_else(|| {
+                std::path::PathBuf::from(std::env::var_os("HOME").unwrap_or_default())
+                    .join(".cache")
+            });
+        Self::start_with_cache(path, &base.join("sausage")).await
+    }
+    /// Start a world with an explicit compile cache folder.
+    pub async fn start_with_cache(path: &Path, cache: &Path) -> anyhow::Result<Self> {
+        let path = path.to_path_buf();
+        let cache = cache.to_path_buf();
+        let (world, runtime) = tokio::task::spawn_blocking(move || -> anyhow::Result<_> {
+            Ok((World::open(&path)?, Runtime::new(&cache)?))
+        })
+        .await??;
+        let journal =
+            journal::Journal::open(world.state_dir("core").join("journal.sqlite")).await?;
+        let host = Self(Arc::new(Inner {
+            world,
+            runtime,
+            journal,
+            state: Mutex::new(State {
+                processes: HashMap::new(),
+                locks: HashMap::new(),
+                native: HashMap::new(),
+            }),
+            notifications: tokio::sync::broadcast::channel(1024).0,
+            exit: watch::channel(false).0,
+            components: components::Components::default(),
+        }));
+        let boot: Vec<Value> = serde_json::from_slice(
+            &tokio::fs::read(host.0.world.root.join("config/boot.json")).await?,
+        )?;
+        for service in boot {
+            let result = host
+                .route(
+                    "core".into(),
+                    Some(json!(Uuid::now_v7().to_string())),
+                    service["method"].as_str().unwrap_or("").into(),
+                    service["params"].clone(),
+                    true,
+                )
+                .await;
+            if let Err(e) = result {
+                host.0.journal.write("core", "core", "core", &e.to_string());
+            }
+        }
+        Ok(host)
+    }
+    /// Register a native plugin for subsequent calls.
+    pub fn register_native(&self, name: &str, plugin: NativePlugin) {
+        self.0
+            .state
+            .lock()
+            .unwrap()
+            .native
+            .insert(name.into(), plugin);
+    }
+    /// Subscribe to notifications addressed to the console.
+    pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<Value> {
+        self.0.notifications.subscribe()
+    }
+    /// Snapshot the metadata of each process that has not ended.
+    pub fn processes(&self) -> HashMap<String, ProcessInfo> {
+        self.0
+            .state
+            .lock()
+            .unwrap()
+            .processes
+            .iter()
+            .map(|(p, s)| (p.clone(), s.info.clone()))
+            .collect()
+    }
+    /// Make a console-side call and wait for its JSON result.
+    pub async fn call(&self, method: &str, params: Value) -> RpcResult {
+        self.route(
+            "console".into(),
+            Some(json!(Uuid::now_v7().to_string())),
+            method.into(),
+            params,
+            false,
+        )
+        .await
+    }
+    /// Wait until a core.exit handler requests a restart.
+    pub async fn exit_requested(&self) {
+        let mut exit = self.0.exit.subscribe();
+        let _ = exit.wait_for(|requested| *requested).await;
+    }
+    /// Return whether a core.exit handler requested a restart.
+    pub fn restart_requested(&self) -> bool {
+        *self.0.exit.borrow()
+    }
+}

cli/src/journal.rs

@@ -1,42 +1,158 @@
-//! SQLite journal entries and filters.
+//! A dedicated SQLite writer batches entries and serves ordered journal queries.
 use rusqlite::{Connection, params};
 use serde_json::{Value, json};
-use std::{collections::HashSet, path::Path, sync::Mutex};
-pub struct Journal {
-    db: Mutex<Connection>,
+use std::{
+    collections::HashSet,
+    path::PathBuf,
+    sync::{Mutex, mpsc},
+};
+use tokio::sync::oneshot;
+
+struct Entry {
+    id: String,
+    pid: String,
+    plugin: String,
+    stream: String,
+    line: String,
+}
+enum Command {
+    Write(Entry),
+    Query(Value, oneshot::Sender<Value>),
+}
+
+pub(crate) struct Journal {
+    sender: mpsc::Sender<Command>,
     warned: Mutex<HashSet<String>>,
 }
 impl Journal {
-    pub fn open(path: &Path) -> anyhow::Result<Self> {
-        std::fs::create_dir_all(path.parent().unwrap())?;
-        let db = Connection::open(path)?;
-        db.execute_batch("CREATE TABLE IF NOT EXISTS journal (id TEXT PRIMARY KEY, pid TEXT, plugin TEXT, stream TEXT, line TEXT)")?;
+    /// Open SQLite on its owning thread and wait for its schema to become ready.
+    pub(crate) async fn open(path: PathBuf) -> anyhow::Result<Self> {
+        let (sender, receiver) = mpsc::channel();
+        let (ready, initialized) = oneshot::channel();
+        std::thread::spawn(move || {
+            let opened = open_database(path);
+            match opened {
+                Ok(mut db) => {
+                    let _ = ready.send(Ok(()));
+                    while let Ok(first) = receiver.recv() {
+                        let mut batch = vec![first];
+                        batch.extend(receiver.try_iter().take(255));
+                        write_batch(&mut db, batch);
+                    }
+                }
+                Err(error) => {
+                    let _ = ready.send(Err(error));
+                }
+            }
+        });
+        initialized.await??;
         Ok(Self {
-            db: Mutex::new(db),
+            sender,
             warned: Mutex::new(HashSet::new()),
         })
     }
-    pub fn write(&self, pid: &str, plugin: &str, stream: &str, line: &str) {
-        self.db
-            .lock()
-            .unwrap()
-            .execute(
-                "INSERT INTO journal VALUES (?1,?2,?3,?4,?5)",
-                params![uuid::Uuid::now_v7().to_string(), pid, plugin, stream, line],
-            )
-            .expect("write journal");
+    /// Queue an entry without file operations or SQLite work on the caller's thread.
+    pub(crate) fn write(&self, pid: &str, plugin: &str, stream: &str, line: &str) {
+        let entry = Entry {
+            id: uuid::Uuid::now_v7().to_string(),
+            pid: pid.into(),
+            plugin: plugin.into(),
+            stream: stream.into(),
+            line: line.into(),
+        };
+        self.sender
+            .send(Command::Write(entry))
+            .expect("journal writer stopped");
     }
-    pub fn warning(&self, pid: &str, plugin: &str) {
+    /// Queue one warning for the first non-blank discarded line of a process.
+    pub(crate) fn warning(&self, pid: &str, plugin: &str) {
         if self.warned.lock().unwrap().insert(pid.into()) {
             self.write(pid, plugin, "core", "warning: discarded stdout");
         }
     }
-    pub fn query(&self, p: &Value) -> Value {
-        let db = self.db.lock().unwrap();
-        let mut stmt=db.prepare("SELECT id,pid,plugin,stream,line FROM journal WHERE (?1 IS NULL OR plugin=?1) AND (?2 IS NULL OR pid=?2) AND (?3 IS NULL OR id>=?3) AND (?4 IS NULL OR id<=?4) AND (?5 IS NULL OR instr(line,?5)>0) ORDER BY id DESC LIMIT ?6").unwrap();
-        let rows=stmt.query_map(params![p["plugin"].as_str(),p["pid"].as_str(),p["since"].as_str(),p["until"].as_str(),p["grep"].as_str(),p["limit"].as_i64().unwrap_or(100)],|r|Ok(json!({"id":r.get::<_,String>(0)?,"pid":r.get::<_,String>(1)?,"plugin":r.get::<_,String>(2)?,"stream":r.get::<_,String>(3)?,"line":r.get::<_,String>(4)?}))).unwrap();
+    /// Queue a query after all entries that the caller already submitted.
+    pub(crate) fn query_later(&self, filters: Value) -> oneshot::Receiver<Value> {
+        let (tx, rx) = oneshot::channel();
+        self.sender
+            .send(Command::Query(filters, tx))
+            .expect("journal writer stopped");
+        rx
+    }
+    /// Wait asynchronously for the journal thread to execute a query.
+    pub(crate) async fn query(&self, filters: Value) -> Value {
+        self.query_later(filters)
+            .await
+            .expect("journal writer stopped")
+    }
+}
+fn open_database(path: PathBuf) -> anyhow::Result<Connection> {
+    std::fs::create_dir_all(path.parent().unwrap())?;
+    let db = Connection::open(path)?;
+    db.execute_batch(
+        "PRAGMA journal_mode=WAL;
+        PRAGMA synchronous=NORMAL;
+        CREATE TABLE IF NOT EXISTS journal (
+            id TEXT PRIMARY KEY, time TEXT, pid TEXT, plugin TEXT, stream TEXT, line TEXT
+        );
+        CREATE INDEX IF NOT EXISTS journal_pid ON journal(pid);
+        CREATE INDEX IF NOT EXISTS journal_plugin ON journal(plugin);",
+    )?;
+    Ok(db)
+}
+/// Commit queued writes together before answering the batch's queries.
+fn write_batch(db: &mut Connection, batch: Vec<Command>) {
+    let transaction = db.transaction().expect("journal transaction");
+    for command in &batch {
+        if let Command::Write(entry) = command {
+            transaction
+                .execute(
+                    "INSERT INTO journal (id,time,pid,plugin,stream,line)
+                 VALUES (?1,strftime('%Y-%m-%dT%H:%M:%fZ','now'),?2,?3,?4,?5)",
+                    params![entry.id, entry.pid, entry.plugin, entry.stream, entry.line],
+                )
+                .expect("write journal");
+        }
+    }
+    transaction.commit().expect("commit journal");
+    for command in batch {
+        if let Command::Query(filters, response) = command {
+            let _ = response.send(query_database(db, &filters));
+        }
+    }
+}
+fn query_database(db: &Connection, filters: &Value) -> Value {
+    let mut statement = db
+        .prepare(
+            "SELECT id,time,pid,plugin,stream,line FROM journal
+         WHERE (?1 IS NULL OR plugin=?1) AND (?2 IS NULL OR pid=?2)
+           AND (?3 IS NULL OR id>=?3) AND (?4 IS NULL OR id<=?4)
+           AND (?5 IS NULL OR instr(line,?5)>0)
+         ORDER BY id DESC LIMIT ?6",
+        )
+        .expect("prepare journal query");
+    let rows = statement
+        .query_map(
+            params![
+                filters["plugin"].as_str(),
+                filters["pid"].as_str(),
+                filters["since"].as_str(),
+                filters["until"].as_str(),
+                filters["grep"].as_str(),
+                filters["limit"].as_i64().unwrap_or(100)
+            ],
+            |row| {
+                Ok(json!({
+                    "id":row.get::<_,String>(0)?,
+                    "time":row.get::<_,String>(1)?,
+                    "pid":row.get::<_,String>(2)?,
+                    "plugin":row.get::<_,String>(3)?,
+                    "stream":row.get::<_,String>(4)?,
+                    "line":row.get::<_,String>(5)?,
+                }))
+            },
+        )
+        .expect("query journal");
     let mut entries: Vec<_> = rows.map(Result::unwrap).collect();
     entries.reverse();
     json!(entries)
 }
-}

cli/src/lib.rs

@@ -1,636 +1,23 @@
-//! Process ownership, JSON-RPC routing, native plugins, and host lifetime.
+//! A host for worlds of JSON-RPC plugins.
 pub mod cargo;
+mod components;
+pub mod console;
+mod core;
+mod execution;
+mod host;
 mod journal;
+mod locks;
+mod native;
+mod process;
+mod routing;
 pub mod runtime;
 pub mod world;
-use runtime::{InstanceConfig, Runtime};
+pub use host::Host;
+pub use native::{NativeContext, NativeFuture, NativePlugin};
+pub use process::ProcessInfo;
 use serde_json::{Value, json};
-use sha2::{Digest, Sha256};
-use std::{
-    collections::{HashMap, VecDeque},
-    future::Future,
-    path::Path,
-    pin::Pin,
-    sync::{Arc, Mutex},
-};
-use tokio::sync::{mpsc, oneshot, watch};
-use uuid::Uuid;
-use world::World;
+/// A JSON result or a JSON-RPC error object.
 pub type RpcResult = Result<Value, Value>;
-pub type NativeFuture = Pin<Box<dyn Future<Output = RpcResult> + Send>>;
-pub type NativePlugin = Arc<dyn Fn(NativeContext, String, Value) -> NativeFuture + Send + Sync>;
-#[derive(Clone)]
-pub struct NativeContext {
-    pub host: Host,
-    pub pid: String,
-}
-impl NativeContext {
-    pub async fn call(&self, method: &str, params: Value) -> RpcResult {
-        let exit = requests_exit(method, &params);
-        let result = self
-            .host
-            .route(
-                self.pid.clone(),
-                Some(json!(Uuid::now_v7().to_string())),
-                method.into(),
-                params,
-                false,
-            )
-            .await;
-        if exit && result.is_ok() {
-            self.host.0.exit.send_replace(true);
-        }
-        result
-    }
-    pub fn notify(&self, name: &str, params: Value) {
-        self.host.notify(&self.pid, name, params);
-    }
-}
-#[derive(Clone, serde::Serialize)]
-pub struct ProcessInfo {
-    pub plugin: String,
-    pub version: String,
-    pub method: String,
-    pub params: Value,
-    pub parent: Option<String>,
-    pub detached: bool,
-    pub start_time: u128,
-    pub state: String,
-}
-struct Process {
-    info: ProcessInfo,
-    input: Option<mpsc::UnboundedSender<Value>>,
-    abort: Option<tokio::task::AbortHandle>,
-    caller: String,
-    id: Option<Value>,
-    answer: Option<oneshot::Sender<RpcResult>>,
-    ended: watch::Sender<bool>,
-}
-struct Lock {
-    owner: String,
-    queue: VecDeque<(String, oneshot::Sender<RpcResult>)>,
-}
-struct State {
-    processes: HashMap<String, Process>,
-    locks: HashMap<String, Lock>,
-    components: HashMap<String, wasmtime::component::Component>,
-    native: HashMap<String, NativePlugin>,
-}
-struct Inner {
-    world: World,
-    runtime: Arc<Runtime>,
-    state: Mutex<State>,
-    journal: journal::Journal,
-    notifications: tokio::sync::broadcast::Sender<Value>,
-    exit: watch::Sender<bool>,
-    compilation: tokio::sync::Mutex<()>,
-}
-#[derive(Clone)]
-pub struct Host(Arc<Inner>);
-struct ProcessGuard {
-    host: Host,
-    pid: String,
-}
-impl Drop for ProcessGuard {
-    fn drop(&mut self) {
-        self.host.finish(&self.pid, "process task ended");
-    }
-}
 fn error(code: i64, text: &str) -> Value {
     json!({"code":code,"message":text})
 }
-impl Host {
-    pub async fn start(path: &Path) -> anyhow::Result<Self> {
-        let base = std::env::var_os("XDG_CACHE_HOME")
-            .map(std::path::PathBuf::from)
-            .unwrap_or_else(|| {
-                std::path::PathBuf::from(std::env::var_os("HOME").unwrap_or_default())
-                    .join(".cache")
-            });
-        Self::start_with_cache(path, &base.join("sausage")).await
-    }
-    pub async fn start_with_cache(path: &Path, cache: &Path) -> anyhow::Result<Self> {
-        let world = World::open(path)?;
-        let journal = journal::Journal::open(&world.state_dir("core").join("journal.sqlite"))?;
-        let host = Self(Arc::new(Inner {
-            world,
-            runtime: Runtime::new(cache)?,
-            journal,
-            state: Mutex::new(State {
-                processes: HashMap::new(),
-                locks: HashMap::new(),
-                components: HashMap::new(),
-                native: HashMap::new(),
-            }),
-            notifications: tokio::sync::broadcast::channel(1024).0,
-            exit: watch::channel(false).0,
-            compilation: tokio::sync::Mutex::new(()),
-        }));
-        let boot: Vec<Value> =
-            serde_json::from_slice(&std::fs::read(host.0.world.root.join("config/boot.json"))?)?;
-        for service in boot {
-            let result = host
-                .route(
-                    "core".into(),
-                    Some(json!(Uuid::now_v7().to_string())),
-                    service["method"].as_str().unwrap_or("").into(),
-                    service["params"].clone(),
-                    true,
-                )
-                .await;
-            if let Err(e) = result {
-                host.0.journal.write("core", "core", "core", &e.to_string());
-            }
-        }
-        Ok(host)
-    }
-    pub fn register_native(&self, name: &str, plugin: NativePlugin) {
-        self.0
-            .state
-            .lock()
-            .unwrap()
-            .native
-            .insert(name.into(), plugin);
-    }
-    pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<Value> {
-        self.0.notifications.subscribe()
-    }
-    pub fn processes(&self) -> HashMap<String, ProcessInfo> {
-        self.0
-            .state
-            .lock()
-            .unwrap()
-            .processes
-            .iter()
-            .map(|(p, s)| (p.clone(), s.info.clone()))
-            .collect()
-    }
-    pub async fn call(&self, method: &str, params: Value) -> RpcResult {
-        let exit = requests_exit(method, &params);
-        let result = self
-            .route(
-                "console".into(),
-                Some(json!(Uuid::now_v7().to_string())),
-                method.into(),
-                params,
-                false,
-            )
-            .await;
-        if exit && result.is_ok() {
-            self.0.exit.send_replace(true);
-        }
-        result
-    }
-    pub async fn exit_requested(&self) {
-        let mut exit = self.0.exit.subscribe();
-        let _ = exit.wait_for(|requested| *requested).await;
-    }
-    pub fn restart_requested(&self) -> bool {
-        *self.0.exit.borrow()
-    }
-    fn send(&self, pid: &str, value: Value) {
-        if pid == "console" {
-            let _ = self.0.notifications.send(value);
-        } else if let Some(tx) = self
-            .0
-            .state
-            .lock()
-            .unwrap()
-            .processes
-            .get(pid)
-            .and_then(|p| p.input.as_ref())
-        {
-            let _ = tx.send(value);
-        }
-    }
-    fn notify(&self, pid: &str, name: &str, params: Value) {
-        let target = self
-            .0
-            .state
-            .lock()
-            .unwrap()
-            .processes
-            .get(pid)
-            .and_then(|p| p.answer.as_ref().map(|_| (p.caller.clone(), p.id.clone())));
-        if let Some((caller, id)) = target {
-            self.send(&caller,json!({"jsonrpc":"2.0","method":"core.notify","params":{"call":id,"name":name,"params":params}}));
-        }
-    }
-    fn route(
-        &self,
-        caller: String,
-        id: Option<Value>,
-        method: String,
-        params: Value,
-        detached: bool,
-    ) -> NativeFuture {
-        let host = self.clone();
-        Box::pin(async move {
-            if !params.is_object() {
-                return Err(error(-32602, "params must be an object"));
-            }
-            if caller != "console" && caller != "core" {
-                let state = host.0.state.lock().unwrap();
-                let Some(process) = state.processes.get(&caller) else {
-                    return Err(error(-32600, "process ended"));
-                };
-                if process.info.state != "running" && method != "core.notify" {
-                    return Err(error(-32600, "process is stopping"));
-                }
-            }
-            let (plugin, method) = method
-                .split_once('.')
-                .ok_or_else(|| error(-32601, "method not found"))?;
-            if plugin == "core" {
-                return host.core(&caller, id, method, params).await;
-            }
-            if plugin.is_empty()
-                || !plugin
-                    .bytes()
-                    .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-' || b == b'_')
-            {
-                return Err(error(-32601, "invalid plugin name"));
-            }
-            let native = host.0.state.lock().unwrap().native.get(plugin).cloned();
-            let (component, version) = if native.is_some() {
-                (None, "native".into())
-            } else {
-                let _compilation = host.0.compilation.lock().await;
-                let bytes = std::fs::read(host.0.world.plugin_dir(plugin).join("plugin.wasm"))
-                    .map_err(|_| error(-32001, "no such plugin"))?;
-                let hash = format!("{:x}", Sha256::digest(&bytes));
-                let cached = host.0.state.lock().unwrap().components.get(&hash).cloned();
-                let component = match cached {
-                    Some(c) => c,
-                    None => {
-                        let rt = host.0.runtime.clone();
-                        let c = tokio::task::spawn_blocking(move || rt.compile(&bytes))
-                            .await
-                            .map_err(|e| error(-32603, &e.to_string()))?
-                            .map_err(|e| error(-32603, &e.to_string()))?;
-                        host.0
-                            .state
-                            .lock()
-                            .unwrap()
-                            .components
-                            .insert(hash.clone(), c.clone());
-                        c
-                    }
-                };
-                (Some(component), hash)
-            };
-            let pid = format!("{plugin}-{}", Uuid::now_v7());
-            let (tx, rx) = oneshot::channel();
-            let (input, mut incoming) = mpsc::unbounded_channel::<Value>();
-            let info = ProcessInfo {
-                plugin: plugin.into(),
-                version,
-                method: method.into(),
-                params: params.clone(),
-                parent: if detached { None } else { Some(caller.clone()) },
-                detached,
-                start_time: std::time::SystemTime::now()
-                    .duration_since(std::time::UNIX_EPOCH)
-                    .unwrap()
-                    .as_millis(),
-                state: "running".into(),
-            };
-            let parent_name = info.parent.clone().unwrap_or_else(|| "core".into());
-            {
-                let mut state = host.0.state.lock().unwrap();
-                if caller != "console"
-                    && caller != "core"
-                    && !state
-                        .processes
-                        .get(&caller)
-                        .is_some_and(|p| p.info.state == "running")
-                {
-                    return Err(error(-32600, "caller stopped or ended"));
-                }
-                host.0
-                    .journal
-                    .write(&pid, plugin, "core", &format!("start {method} {params}"));
-                state.processes.insert(
-                    pid.clone(),
-                    Process {
-                        info,
-                        input: Some(input),
-                        abort: None,
-                        caller,
-                        id: id.clone(),
-                        answer: Some(tx),
-                        ended: watch::channel(false).0,
-                    },
-                );
-            }
-            let plugin_name = plugin.to_string();
-            let h = host.clone();
-            let p = pid.clone();
-            let method = method.to_string();
-            let has_id = id.is_some();
-            let task = tokio::spawn(async move {
-                let _guard = ProcessGuard {
-                    host: h.clone(),
-                    pid: p.clone(),
-                };
-                if let Some(native) = native {
-                    let result = native(
-                        NativeContext {
-                            host: h.clone(),
-                            pid: p.clone(),
-                        },
-                        method,
-                        params,
-                    )
-                    .await;
-                    h.answer(&p, result);
-                    h.finish(&p, "exit 0");
-                    return;
-                }
-                let cfg = InstanceConfig {
-                    world: h.0.world.root.clone(),
-                    network: true,
-                    env: vec![
-                        ("SAUSAGE_PID".into(), p.clone()),
-                        ("SAUSAGE_PLUGIN".into(), plugin_name.clone()),
-                        ("SAUSAGE_PARENT".into(), parent_name),
-                    ],
-                };
-                let result = async {
-                    let mut instance = h.0.runtime.start(&component.unwrap(),cfg)?;
-                    let initial = if has_id { json!({"jsonrpc":"2.0","id":p,"method":method,"params":params}) } else {json!({"jsonrpc":"2.0","method":method,"params":params})};
-                    instance.send_line(&initial.to_string(),|line|h.output(&p,line)).await?;
-                    let mut stdin_open = true;
-                    let mut stderr_open = true;
-                    loop { tokio::select! {
-                        line = instance.stderr.recv(), if stderr_open => match line { Some(line) => h.0.journal.write(&p,&plugin_name,"stderr",&line), None => stderr_open=false },
-                        line = instance.stdout.recv() => match line { Some(line) => h.output(&p,line), None => break },
-                        value = incoming.recv(), if stdin_open => match value { Some(v) => { instance.send_line(&v.to_string(),|line|h.output(&p,line)).await?; }, None => { instance.stdin.take(); stdin_open=false; } }
-                    }}
-                    while let Some(line) = instance.stderr.recv().await { h.0.journal.write(&p,&plugin_name,"stderr",&line); }
-                    Ok::<_,anyhow::Error>(instance.wait().await.to_string())
-                }.await;
-                h.finish(&p, &result.unwrap_or_else(|e| e.to_string()));
-            });
-            if let Some(p) = host.0.state.lock().unwrap().processes.get_mut(&pid) {
-                p.abort = Some(task.abort_handle());
-            } else {
-                task.abort();
-            }
-            if id.is_none() {
-                return Ok(Value::Null);
-            }
-            rx.await
-                .unwrap_or_else(|_| Err(error(-32002, "process ended")))
-        })
-    }
-    fn answer(&self, pid: &str, result: RpcResult) {
-        if let Some(tx) = self
-            .0
-            .state
-            .lock()
-            .unwrap()
-            .processes
-            .get_mut(pid)
-            .and_then(|p| p.answer.take())
-        {
-            let _ = tx.send(result);
-        }
-    }
-    fn output(&self, pid: &str, line: String) {
-        let Ok(v) = serde_json::from_str::<Value>(&line) else {
-            self.discard(pid);
-            return;
-        };
-        if !v.is_object() || v["jsonrpc"] != "2.0" {
-            self.discard(pid);
-            return;
-        }
-        if let Some(method) = v["method"].as_str() {
-            let h = self.clone();
-            let caller = pid.to_string();
-            let exit = requests_exit(method, &v["params"]);
-            let method = method.to_string();
-            tokio::spawn(async move {
-                let id = v.get("id").cloned();
-                let result = h
-                    .route(
-                        caller.clone(),
-                        id.clone(),
-                        method,
-                        v.get("params").cloned().unwrap_or(json!({})),
-                        false,
-                    )
-                    .await;
-                let success = result.is_ok();
-                if let Some(id) = id {
-                    let value = match result {
-                        Ok(r) => json!({"jsonrpc":"2.0","id":id,"result":r}),
-                        Err(e) => json!({"jsonrpc":"2.0","id":id,"error":e}),
-                    };
-                    h.send(&caller, value);
-                }
-                if exit && success {
-                    h.0.exit.send_replace(true);
-                }
-            });
-        } else if v["id"] == pid {
-            self.answer(
-                pid,
-                if let Some(e) = v.get("error") {
-                    Err(e.clone())
-                } else {
-                    Ok(v["result"].clone())
-                },
-            );
-        }
-    }
-    fn discard(&self, pid: &str) {
-        // The journal deduplicates this warning for each process.
-        if let Some(info) = self.processes().get(pid) {
-            self.0.journal.warning(pid, &info.plugin);
-        }
-    }
-    fn finish(&self, pid: &str, status: &str) {
-        let mut state = self.0.state.lock().unwrap();
-        let process = state.processes.remove(pid);
-        if let Some(p) = process {
-            self.0.journal.write(pid, &p.info.plugin, "core", status);
-            if let Some(tx) = p.answer {
-                let _=tx.send(Err(json!({"code":-32002,"message":"process ended without an answer","data":{"status":status,"lines":self.0.journal.query(&json!({"pid":pid,"limit":20}))}})));
-            }
-            Self::release_locks(&mut state, pid);
-            let _ = p.ended.send(true);
-            let children: Vec<_> = state
-                .processes
-                .iter()
-                .filter(|(_, p)| p.info.parent.as_deref() == Some(pid))
-                .map(|(id, _)| id.clone())
-                .collect();
-            drop(state);
-            for child in children {
-                self.begin_stop(&child);
-            }
-        }
-    }
-    fn begin_stop(&self, pid: &str) -> Option<watch::Receiver<bool>> {
-        let mut state = self.0.state.lock().unwrap();
-        let p = state.processes.get_mut(pid)?;
-        let ended = p.ended.subscribe();
-        if p.info.state == "stopping" {
-            return Some(ended);
-        }
-        p.info.state = "stopping".into();
-        p.input.take();
-        self.0.journal.write(pid, &p.info.plugin, "core", "stop");
-        let children: Vec<_> = state
-            .processes
-            .iter()
-            .filter(|(_, p)| p.info.parent.as_deref() == Some(pid))
-            .map(|(id, _)| id.clone())
-            .collect();
-        drop(state);
-        for child in children {
-            self.begin_stop(&child);
-        }
-        let h = self.clone();
-        let pid = pid.to_string();
-        tokio::spawn(async move {
-            tokio::time::sleep(std::time::Duration::from_secs(5)).await;
-            h.kill(&pid);
-        });
-        Some(ended)
-    }
-    fn kill(&self, pid: &str) {
-        if let Some(p) = self.0.state.lock().unwrap().processes.get(pid) {
-            self.0.journal.write(pid, &p.info.plugin, "core", "kill");
-            if let Some(a) = &p.abort {
-                a.abort();
-            }
-        }
-        self.finish(pid, "killed");
-    }
-    pub async fn stop(&self) {
-        let ids: Vec<_> = self.processes().into_keys().collect();
-        let mut waits = Vec::new();
-        for pid in ids {
-            if let Some(n) = self.begin_stop(&pid) {
-                waits.push(n);
-            }
-        }
-        for mut n in waits {
-            let _ = n.wait_for(|ended| *ended).await;
-        }
-    }
-    fn release_locks(state: &mut State, pid: &str) {
-        for lock in state.locks.values_mut() {
-            lock.queue.retain(|(p, _)| p != pid);
-        }
-        let paths: Vec<_> = state
-            .locks
-            .iter()
-            .filter(|(_, l)| l.owner == pid)
-            .map(|(p, _)| p.clone())
-            .collect();
-        for path in paths {
-            grant(state, &path);
-        }
-    }
-    async fn core(&self, caller: &str, id: Option<Value>, method: &str, p: Value) -> RpcResult {
-        match method {
-            "call" => {
-                self.route(
-                    caller.into(),
-                    id,
-                    required_str(&p, "method")?.into(),
-                    p.get("params").cloned().unwrap_or(json!({})),
-                    p["detach"].as_bool().unwrap_or(false),
-                )
-                .await
-            }
-            "notify" => {
-                self.notify(
-                    caller,
-                    required_str(&p, "name")?,
-                    p.get("params").cloned().unwrap_or(json!({})),
-                );
-                Ok(Value::Null)
-            }
-            "kill" => {
-                self.kill(required_str(&p, "pid")?);
-                Ok(json!({}))
-            }
-            "stop" => {
-                if let Some(mut n) = self.begin_stop(required_str(&p, "pid")?) {
-                    let _ = n.wait_for(|ended| *ended).await;
-                }
-                Ok(json!({}))
-            }
-            "lock" => {
-                let path = p["path"]
-                    .as_str()
-                    .ok_or_else(|| error(-32602, "missing path"))?;
-                let rx = {
-                    let mut state = self.0.state.lock().unwrap();
-                    if let Some(lock) = state.locks.get_mut(path) {
-                        if p["wait"] == false {
-                            return Ok(json!({"held":false}));
-                        }
-                        let (tx, rx) = oneshot::channel();
-                        lock.queue.push_back((caller.into(), tx));
-                        Some(rx)
-                    } else {
-                        state.locks.insert(
-                            path.into(),
-                            Lock {
-                                owner: caller.into(),
-                                queue: VecDeque::new(),
-                            },
-                        );
-                        None
-                    }
-                };
-                match rx {
-                    Some(rx) => rx
-                        .await
-                        .unwrap_or_else(|_| Err(error(-32603, "lock canceled"))),
-                    None => Ok(json!({"held":true})),
-                }
-            }
-            "unlock" => {
-                let mut state = self.0.state.lock().unwrap();
-                let path = required_str(&p, "path")?;
-                if state.locks.get(path).is_some_and(|l| l.owner == caller) {
-                    grant(&mut state, path);
-                }
-                Ok(Value::Null)
-            }
-            "log" => Ok(self.0.journal.query(&p)),
-            "exit" => Ok(json!({})),
-            _ => Err(error(-32601, "method not found")),
-        }
-    }
-}
-fn grant(state: &mut State, path: &str) {
-    if let Some(mut lock) = state.locks.remove(path) {
-        while let Some((owner, tx)) = lock.queue.pop_front() {
-            if tx.send(Ok(json!({"held":true}))).is_ok() {
-                lock.owner = owner;
-                state.locks.insert(path.into(), lock);
-                break;
-            }
-        }
-    }
-}
-
-fn requests_exit(method: &str, params: &Value) -> bool {
-    method == "core.exit"
-        || (method == "core.call"
-            && requests_exit(params["method"].as_str().unwrap_or(""), &params["params"]))
-}
-
-fn required_str<'a>(params: &'a Value, name: &str) -> Result<&'a str, Value> {
-    params[name]
-        .as_str()
-        .ok_or_else(|| error(-32602, &format!("{name} must be a string")))
-}

cli/src/locks.rs

@@ -0,0 +1,81 @@
+//! Advisory FIFO locks and cleanup of ended processes.
+use crate::error;
+use crate::{Host, RpcResult, host::State};
+use serde_json::json;
+use std::collections::VecDeque;
+use tokio::sync::oneshot;
+
+pub(crate) struct Lock {
+    pub(crate) owner: String,
+    pub(crate) queue: VecDeque<(String, oneshot::Sender<RpcResult>)>,
+}
+
+impl Host {
+    /// Acquire a FIFO queue position, or report whether the lock is free.
+    pub(crate) async fn lock(&self, caller: &str, path: &str, wait: bool) -> RpcResult {
+        let rx = {
+            let mut state = self.0.state.lock().unwrap();
+            if let Some(lock) = state.locks.get_mut(path) {
+                if !wait {
+                    return Ok(json!({"held":false}));
+                }
+                let (tx, rx) = oneshot::channel();
+                lock.queue.push_back((caller.into(), tx));
+                Some(rx)
+            } else {
+                state.locks.insert(
+                    path.into(),
+                    Lock {
+                        owner: caller.into(),
+                        queue: VecDeque::new(),
+                    },
+                );
+                None
+            }
+        };
+        match rx {
+            Some(rx) => rx
+                .await
+                .unwrap_or_else(|_| Err(error(-32603, "lock canceled"))),
+            None => Ok(json!({"held":true})),
+        }
+    }
+    /// Release a held lock only when this process owns it.
+    pub(crate) fn unlock(&self, caller: &str, path: &str) {
+        let mut state = self.0.state.lock().unwrap();
+        if state
+            .locks
+            .get(path)
+            .is_some_and(|lock| lock.owner == caller)
+        {
+            grant(&mut state, path);
+        }
+    }
+
+    /// Remove queued acquisitions and grant the next owner of each held lock.
+    pub(crate) fn release_locks(state: &mut State, pid: &str) {
+        for lock in state.locks.values_mut() {
+            lock.queue.retain(|(p, _)| p != pid);
+        }
+        let paths: Vec<_> = state
+            .locks
+            .iter()
+            .filter(|(_, l)| l.owner == pid)
+            .map(|(p, _)| p.clone())
+            .collect();
+        for path in paths {
+            grant(state, &path);
+        }
+    }
+}
+pub(crate) fn grant(state: &mut State, path: &str) {
+    if let Some(mut lock) = state.locks.remove(path) {
+        while let Some((owner, tx)) = lock.queue.pop_front() {
+            if tx.send(Ok(json!({"held":true}))).is_ok() {
+                lock.owner = owner;
+                state.locks.insert(path.into(), lock);
+                break;
+            }
+        }
+    }
+}

cli/src/main.rs

@@ -4,7 +4,6 @@ use clap::{Parser, Subcommand};
 use sausage::{Host, world::World};
 use serde_json::json;
 use std::{io::Write, path::PathBuf};
-use tokio::io::{AsyncBufReadExt, BufReader};
 #[derive(Parser)]
 #[command(version)]
 struct Cli {
@@ -71,24 +70,15 @@ async fn main() -> Result<()> {
                     println!("{notification}");
                 }
             });
-            let h = host.clone();
-            let console = async move {
-                let mut lines = BufReader::new(tokio::io::stdin()).lines();
-                loop {
-                    tokio::select! {
-                            line=lines.next_line()=>{let Some(line)=line? else {break;};if line=="quit" {break;}
-                            let (method,params)=line.split_once(' ').unwrap_or((&line,"{}"));
-                            match serde_json::from_str(params) {Ok(p)=>println!("{}",match h.call(method,p).await {Ok(v)=>v,Err(e)=>e}),Err(e)=>eprintln!("{e}")}
-                        }
-                    }
-                }
-                Ok::<_, anyhow::Error>(())
-            };
+            let console = sausage::console::run_input(host.clone());
             let exit = tokio::select! {
                 _=host.exit_requested()=>true,
                 _=term.recv()=>false,
                 _=tokio::signal::ctrl_c()=>false,
-                r=console,if !cli.headless=>{r?;false},
+                result = console, if !cli.headless => {
+                    result?;
+                    false
+                },
             };
             let exit = exit || host.restart_requested();
             host.stop().await;

cli/src/native.rs

@@ -0,0 +1,35 @@
+//! Native plugin futures and their process context.
+use crate::{Host, RpcResult};
+use serde_json::{Value, json};
+use std::{future::Future, pin::Pin, sync::Arc};
+use uuid::Uuid;
+/// The cancellable future of a native process.
+pub type NativeFuture = Pin<Box<dyn Future<Output = RpcResult> + Send>>;
+/// A handler for the methods of one native plugin.
+pub type NativePlugin = Arc<dyn Fn(NativeContext, String, Value) -> NativeFuture + Send + Sync>;
+#[derive(Clone)]
+/// The process identity and host access supplied to a native handler.
+pub struct NativeContext {
+    /// The host that owns this native process.
+    pub host: Host,
+    /// The ID assigned to this native process.
+    pub pid: String,
+}
+impl NativeContext {
+    /// Make an outgoing call with this native process as its caller.
+    pub async fn call(&self, method: &str, params: Value) -> RpcResult {
+        self.host
+            .route(
+                self.pid.clone(),
+                Some(json!(Uuid::now_v7().to_string())),
+                method.into(),
+                params,
+                false,
+            )
+            .await
+    }
+    /// Send progress to this process's caller while its call remains unanswered.
+    pub fn notify(&self, name: &str, params: Value) {
+        self.host.notify(&self.pid, name, params);
+    }
+}

cli/src/process.rs

@@ -0,0 +1,131 @@
+//! Process metadata, process lifetime, and attached children.
+use crate::{Host, RpcResult};
+use serde_json::{Value, json};
+use tokio::sync::{mpsc, oneshot, watch};
+#[derive(Clone, serde::Serialize)]
+/// A snapshot of process metadata for inspection and the VFS.
+pub struct ProcessInfo {
+    /// The plugin that supplied the process code.
+    pub plugin: String,
+    /// The SHA-256 of its component, or native for a native plugin.
+    pub version: String,
+    /// The method without its plugin prefix.
+    pub method: String,
+    /// The object supplied by the caller.
+    pub params: Value,
+    /// The attached parent, or no parent for a detached process.
+    pub parent: Option<String>,
+    /// Whether this process survives the end of its caller.
+    pub detached: bool,
+    /// The start time in milliseconds since the Unix epoch.
+    pub start_time: u128,
+    /// The current lifecycle state.
+    pub state: String,
+}
+pub(crate) struct Process {
+    pub(crate) info: ProcessInfo,
+    pub(crate) input: Option<mpsc::UnboundedSender<Value>>,
+    pub(crate) abort: Option<tokio::task::AbortHandle>,
+    pub(crate) caller: String,
+    pub(crate) id: Option<Value>,
+    pub(crate) answer: Option<oneshot::Sender<RpcResult>>,
+    pub(crate) ended: watch::Sender<bool>,
+}
+pub(crate) struct ProcessGuard {
+    pub(crate) host: Host,
+    pub(crate) pid: String,
+}
+impl Drop for ProcessGuard {
+    fn drop(&mut self) {
+        self.host.finish(&self.pid, "process task ended");
+    }
+}
+
+impl Host {
+    /// Remove a process and release its locks before asynchronous journal completion.
+    pub(crate) fn finish(&self, pid: &str, status: &str) {
+        let mut state = self.0.state.lock().unwrap();
+        let process = state.processes.remove(pid);
+        if let Some(p) = process {
+            self.0.journal.write(pid, &p.info.plugin, "core", status);
+            let logs = self.0.journal.query_later(json!({"pid":pid,"limit":20}));
+            Self::release_locks(&mut state, pid);
+            let status = status.to_string();
+            tokio::spawn(async move {
+                let lines = logs.await.expect("journal writer stopped");
+                if let Some(answer) = p.answer {
+                    let failure = json!({
+                        "code": -32002,
+                        "message": "process ended without an answer",
+                        "data": { "status": status, "lines": lines }
+                    });
+                    let _ = answer.send(Err(failure));
+                }
+                p.ended.send_replace(true);
+            });
+            let children: Vec<_> = state
+                .processes
+                .iter()
+                .filter(|(_, p)| p.info.parent.as_deref() == Some(pid))
+                .map(|(id, _)| id.clone())
+                .collect();
+            drop(state);
+            for child in children {
+                self.begin_stop(&child);
+            }
+        }
+    }
+    /// Close stdin and start the same five-second deadline for each attached child.
+    pub(crate) fn begin_stop(&self, pid: &str) -> Option<watch::Receiver<bool>> {
+        let mut state = self.0.state.lock().unwrap();
+        let p = state.processes.get_mut(pid)?;
+        let ended = p.ended.subscribe();
+        if p.info.state == "stopping" {
+            return Some(ended);
+        }
+        p.info.state = "stopping".into();
+        p.input.take();
+        self.0.journal.write(pid, &p.info.plugin, "core", "stop");
+        let children: Vec<_> = state
+            .processes
+            .iter()
+            .filter(|(_, p)| p.info.parent.as_deref() == Some(pid))
+            .map(|(id, _)| id.clone())
+            .collect();
+        drop(state);
+        for child in children {
+            self.begin_stop(&child);
+        }
+        let h = self.clone();
+        let pid = pid.to_string();
+        tokio::spawn(async move {
+            tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+            h.kill(&pid);
+        });
+        Some(ended)
+    }
+    /// Abort a process task and perform its ended-state cleanup.
+    pub(crate) fn kill(&self, pid: &str) {
+        if let Some(p) = self.0.state.lock().unwrap().processes.get(pid) {
+            self.0.journal.write(pid, &p.info.plugin, "core", "kill");
+            if let Some(a) = &p.abort {
+                a.abort();
+            }
+        }
+        self.finish(pid, "killed");
+    }
+    /// Stop every process in parallel and wait for journal writes to finish.
+    pub async fn stop(&self) {
+        let ids: Vec<_> = self.processes().into_keys().collect();
+        let mut waits = Vec::new();
+        for pid in ids {
+            if let Some(n) = self.begin_stop(&pid) {
+                waits.push(n);
+            }
+        }
+        for mut n in waits {
+            let _ = n.wait_for(|ended| *ended).await;
+        }
+        self.0.journal.query(json!({"limit":0})).await;
+    }
+}

cli/src/routing.rs

@@ -0,0 +1,233 @@
+//! JSON-RPC routing and responses between callers and processes.
+use crate::{
+    Host, NativeFuture, RpcResult, error,
+    process::{Process, ProcessInfo},
+};
+use serde_json::{Value, json};
+use tokio::sync::{mpsc, oneshot, watch};
+use uuid::Uuid;
+
+impl Host {
+    /// Queue an incoming response or notification for its destination.
+    pub(crate) fn send(&self, pid: &str, value: Value) {
+        if pid == "console" {
+            let _ = self.0.notifications.send(value);
+        } else if let Some(tx) = self
+            .0
+            .state
+            .lock()
+            .unwrap()
+            .processes
+            .get(pid)
+            .and_then(|p| p.input.as_ref())
+        {
+            let _ = tx.send(value);
+        }
+    }
+    /// Remap progress to the caller's original request ID.
+    pub(crate) fn notify(&self, pid: &str, name: &str, params: Value) {
+        let target = self
+            .0
+            .state
+            .lock()
+            .unwrap()
+            .processes
+            .get(pid)
+            .and_then(|p| p.answer.as_ref().map(|_| (p.caller.clone(), p.id.clone())));
+        if let Some((caller, id)) = target {
+            let notification = json!({
+                "jsonrpc": "2.0",
+                "method": "core.notify",
+                "params": { "call": id, "name": name, "params": params }
+            });
+            self.send(&caller, notification);
+        }
+    }
+    /// Validate and route a request, then start its native or WASI process.
+    pub(crate) fn route(
+        &self,
+        caller: String,
+        id: Option<Value>,
+        method: String,
+        params: Value,
+        detached: bool,
+    ) -> NativeFuture {
+        let host = self.clone();
+        Box::pin(async move {
+            if !params.is_object() {
+                return Err(error(-32602, "params must be an object"));
+            }
+            if caller != "console" && caller != "core" {
+                let state = host.0.state.lock().unwrap();
+                let Some(process) = state.processes.get(&caller) else {
+                    return Err(error(-32600, "process ended"));
+                };
+                if process.info.state != "running" && method != "core.notify" {
+                    return Err(error(-32600, "process is stopping"));
+                }
+            }
+            let (plugin, method) = method
+                .split_once('.')
+                .ok_or_else(|| error(-32601, "method not found"))?;
+            if plugin == "core" {
+                return host.core(&caller, id, method, params).await;
+            }
+            if plugin.is_empty()
+                || !plugin
+                    .bytes()
+                    .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-' || b == b'_')
+            {
+                return Err(error(-32601, "invalid plugin name"));
+            }
+            let native = host.0.state.lock().unwrap().native.get(plugin).cloned();
+            let (component, version) = if native.is_some() {
+                (None, "native".into())
+            } else {
+                let (component, hash) = host
+                    .0
+                    .components
+                    .load(
+                        host.0.world.plugin_dir(plugin).join("plugin.wasm"),
+                        host.0.runtime.clone(),
+                    )
+                    .await?;
+                (Some(component), hash)
+            };
+            let pid = format!("{plugin}-{}", Uuid::now_v7());
+            let (tx, rx) = oneshot::channel();
+            let (input, incoming) = mpsc::unbounded_channel::<Value>();
+            let info = ProcessInfo {
+                plugin: plugin.into(),
+                version,
+                method: method.into(),
+                params: params.clone(),
+                parent: if detached { None } else { Some(caller.clone()) },
+                detached,
+                start_time: std::time::SystemTime::now()
+                    .duration_since(std::time::UNIX_EPOCH)
+                    .unwrap()
+                    .as_millis(),
+                state: "running".into(),
+            };
+            {
+                let mut state = host.0.state.lock().unwrap();
+                if caller != "console"
+                    && caller != "core"
+                    && !state
+                        .processes
+                        .get(&caller)
+                        .is_some_and(|p| p.info.state == "running")
+                {
+                    return Err(error(-32600, "caller stopped or ended"));
+                }
+                host.0
+                    .journal
+                    .write(&pid, plugin, "core", &format!("start {method} {params}"));
+                state.processes.insert(
+                    pid.clone(),
+                    Process {
+                        info: info.clone(),
+                        input: Some(input),
+                        abort: None,
+                        caller,
+                        id: id.clone(),
+                        answer: Some(tx),
+                        ended: watch::channel(false).0,
+                    },
+                );
+            }
+            let h = host.clone();
+            let p = pid.clone();
+            let has_id = id.is_some();
+            let task = tokio::spawn(async move {
+                h.run_process(p, info, component, native, incoming, has_id)
+                    .await;
+            });
+            if let Some(p) = host.0.state.lock().unwrap().processes.get_mut(&pid) {
+                p.abort = Some(task.abort_handle());
+            } else {
+                task.abort();
+            }
+            if id.is_none() {
+                return Ok(Value::Null);
+            }
+            rx.await
+                .unwrap_or_else(|_| Err(error(-32002, "process ended")))
+        })
+    }
+    /// Complete the initial call once and discard late responses.
+    pub(crate) fn answer(&self, pid: &str, result: RpcResult) {
+        if let Some(tx) = self
+            .0
+            .state
+            .lock()
+            .unwrap()
+            .processes
+            .get_mut(pid)
+            .and_then(|p| p.answer.take())
+        {
+            let _ = tx.send(result);
+        }
+    }
+    /// Classify a stdout line and route requests or the initial response.
+    pub(crate) fn output(&self, pid: &str, line: String) {
+        if line.trim().is_empty() {
+            return;
+        }
+        let Ok(v) = serde_json::from_str::<Value>(&line) else {
+            self.discard(pid);
+            return;
+        };
+        if !v.is_object() || v["jsonrpc"] != "2.0" {
+            self.discard(pid);
+            return;
+        }
+        if let Some(method) = v["method"].as_str() {
+            let h = self.clone();
+            let caller = pid.to_string();
+            let method = method.to_string();
+            tokio::spawn(async move {
+                let id = v.get("id").cloned();
+                let result = h
+                    .route(
+                        caller.clone(),
+                        id.clone(),
+                        method,
+                        v.get("params").cloned().unwrap_or(json!({})),
+                        false,
+                    )
+                    .await;
+                if let Some(id) = id {
+                    let value = match result {
+                        Ok(result) => json!({
+                            "jsonrpc": "2.0",
+                            "id": id,
+                            "result": result
+                        }),
+                        Err(error) => json!({
+                            "jsonrpc": "2.0",
+                            "id": id,
+                            "error": error
+                        }),
+                    };
+                    h.send(&caller, value);
+                }
+            });
+        } else if v["id"] == pid {
+            self.answer(
+                pid,
+                if let Some(e) = v.get("error") {
+                    Err(e.clone())
+                } else {
+                    Ok(v["result"].clone())
+                },
+            );
+        }
+    }
+    pub(crate) fn discard(&self, pid: &str) {
+        // The journal deduplicates this warning for each process.
+        if let Some(info) = self.processes().get(pid) {
+            self.0.journal.warning(pid, &info.plugin);
+        }
+    }
+}

cli/src/runtime.rs

@@ -20,6 +20,7 @@ use wasmtime_wasi::p2::bindings::Command;
 use wasmtime_wasi::{FsPerms, I32Exit, WasiCtx, WasiCtxView, WasiView};
 use wasmtime_wasi_http::{WasiHttpCtx, WasiHttpCtxView, WasiHttpHooks, WasiHttpView};
 
+/// The HTTP transport hooks supplied to WASI processes.
 pub struct HttpHooks {
     allow: bool,
 }
@@ -54,6 +55,7 @@ impl WasiHttpHooks for HttpHooks {
     }
 }
 
+/// The WASI and HTTP resources owned by one store.
 pub struct ProcessState {
     ctx: WasiCtx,
     table: ResourceTable,
@@ -80,6 +82,7 @@ impl WasiHttpView for ProcessState {
     }
 }
 
+/// A Wasmtime engine and linker shared by the world.
 pub struct Runtime {
     engine: Engine,
     linker: Linker<ProcessState>,
@@ -106,18 +109,23 @@ impl std::fmt::Display for Exit {
     }
 }
 
+/// The filesystem, network, and environment supplied to one process.
 pub struct InstanceConfig {
+    /// The world folder to preopen as the guest root.
     pub world: PathBuf,
     /// Outgoing HTTP, and TCP, UDP, and DNS through `wasi:sockets`.
     pub network: bool,
+    /// The environment variables supplied to the guest.
     pub env: Vec<(String, String)>,
 }
 
 /// A running guest. Dropping it kills the guest.
 pub struct Instance {
+    /// The host end of stdin, removed when the process stops.
     pub stdin: Option<DuplexStream>,
     /// Lines that the guest writes to stdout. Closed when the guest ends.
     pub stdout: mpsc::Receiver<String>,
+    /// The stderr lines that the process writes.
     pub stderr: mpsc::Receiver<String>,
     task: JoinHandle<Exit>,
 }
@@ -178,6 +186,7 @@ fn short_trap(full: &str) -> String {
 }
 
 impl Runtime {
+    /// Create an engine with disk caching and a ten-millisecond epoch tick.
     pub fn new(cache: &Path) -> Result<Arc<Self>> {
         let mut config = Config::new();
         config.epoch_interruption(true);
@@ -206,11 +215,17 @@ impl Runtime {
         Ok(runtime)
     }
 
+    /// Compile component bytes through Wasmtime's disk cache.
     pub fn compile(&self, bytes: &[u8]) -> Result<Component> {
         Ok(Component::new(&self.engine, bytes)?)
     }
 
-    pub fn start(self: &Arc<Self>, component: &Component, cfg: InstanceConfig) -> Result<Instance> {
+    /// Prepare a world preopen off the async workers, then start the WASI task.
+    pub async fn start(
+        self: &Arc<Self>,
+        component: &Component,
+        cfg: InstanceConfig,
+    ) -> Result<Instance> {
         let (host_in, guest_in) = tokio::io::duplex(64 * 1024);
         let (guest_out, host_out) = tokio::io::duplex(64 * 1024);
 
@@ -229,7 +244,12 @@ impl Runtime {
             b.env(k, v);
         }
         // Filesystem setup lives here. T6 replaces this preopen with the VFS.
-        b.preopened_dir(&cfg.world, "/", FsPerms::ReadWrite)?;
+        let world = cfg.world;
+        let mut b = tokio::task::spawn_blocking(move || -> Result<_> {
+            b.preopened_dir(world, "/", FsPerms::ReadWrite)?;
+            Ok(b)
+        })
+        .await??;
         let state = ProcessState {
             ctx: b.build(),
             table: ResourceTable::new(),
@@ -245,7 +265,9 @@ impl Runtime {
                 if n == 0 {
                     break;
                 }
-                let line = String::from_utf8(std::mem::take(&mut bytes)).unwrap_or_default();
+                // Preserve a non-blank invalid line so routing emits its warning.
+                let line = String::from_utf8(std::mem::take(&mut bytes))
+                    .unwrap_or_else(|_| "\u{fffd}".into());
                 if line_tx.send(line).await.is_err() {
                     break;
                 }

cli/src/world.rs

@@ -2,10 +2,13 @@
 use anyhow::{Result, bail};
 use std::path::{Path, PathBuf};
 #[derive(Clone)]
+/// The absolute host folder shared by all processes.
 pub struct World {
+    /// The absolute path of the world on the host.
     pub root: PathBuf,
 }
 impl World {
+    /// Validate reserved root names and resolve the world path.
     pub fn open(path: &Path) -> Result<Self> {
         for name in ["proc", "run", "self"] {
             if path.join(name).symlink_metadata().is_ok() {
@@ -16,6 +19,7 @@ impl World {
             root: path.canonicalize()?,
         })
     }
+    /// Create the folders and empty boot configuration of a new world.
     pub fn create(path: &Path) -> Result<Self> {
         if path.exists() {
             bail!("world already exists: {}", path.display());
@@ -26,13 +30,16 @@ impl World {
         std::fs::write(path.join("config/boot.json"), "[]\n")?;
         Self::open(path)
     }
+    /// Return the folder that contains a plugin component.
     pub fn plugin_dir(&self, name: &str) -> PathBuf {
         self.root.join("plugins").join(name)
     }
+    /// Return the persistent state folder of a plugin.
     pub fn state_dir(&self, name: &str) -> PathBuf {
         self.root.join("state").join(name)
     }
 }
+/// Write a temporary sibling and rename it over the destination.
 pub fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> {
     std::fs::create_dir_all(path.parent().unwrap())?;
     let tmp = path.with_extension("tmp");
@@ -40,6 +47,7 @@ pub fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> {
     std::fs::rename(tmp, path)?;
     Ok(())
 }
+/// Replace a build copy with a source tree, without target folders.
 pub fn replace_source(from: &Path, to: &Path) -> Result<()> {
     if to.exists() {
         std::fs::remove_dir_all(to)?;

cli/tests/core.rs

@@ -52,15 +52,35 @@ async fn call(h: &Host, m: &str, p: Value) -> Value {
         .unwrap()
         .unwrap()
 }
-#[tokio::test(flavor = "multi_thread")]
-async fn routing_environment_files_and_errors() {
+
+/// PROTOCOL 3.1: json rpc lines carry results.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn json_rpc_lines_carry_results() {
     let w = TestWorld::new();
     let h = w.start().await;
     assert_eq!(call(&h, "probe.echo", json!({"x":1})).await, json!({"x":1}));
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.2: method names split at the first dot.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn method_names_split_at_the_first_dot() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     assert_eq!(
         call(&h, "probe.rpc.discover", json!({"x":2})).await,
         json!({"x":2})
     );
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.3: processes can call other plugins.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn processes_can_call_other_plugins() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     assert_eq!(
         call(
             &h,
@@ -70,10 +90,28 @@ async fn routing_environment_files_and_errors() {
         .await,
         json!({"nested":true})
     );
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.3: pending ids distinguish numbers from strings.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn pending_ids_distinguish_numbers_from_strings() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     assert_eq!(
         call(&h, "probe.multicall", json!({})).await,
         json!({"count":2})
     );
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.3: notifications start attached processes.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn notifications_start_attached_processes() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     call(
         &h,
         "probe.notification",
@@ -84,14 +122,41 @@ async fn routing_environment_files_and_errors() {
         std::fs::read_to_string(w.0.join("state/notification")).unwrap(),
         "delivered"
     );
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.3: unknown response ids are dropped.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn unknown_response_ids_are_dropped() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     assert_eq!(
         h.call("probe.unknownid", json!({})).await.unwrap_err()["code"],
         -32002
     );
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.8: process environment identifies the caller.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn process_environment_identifies_the_caller() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     let env = call(&h, "probe.env", json!({})).await;
     assert_eq!(env["plugin"], "probe");
     assert_eq!(env["parent"], "console");
     assert!(env["pid"].as_str().unwrap().starts_with("probe-"));
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.6: processes read and write the world.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn processes_read_and_write_the_world() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     call(
         &h,
         "probe.write",
@@ -102,18 +167,54 @@ async fn routing_environment_files_and_errors() {
         call(&h, "probe.read", json!({"path":"/state/test"})).await,
         "hello"
     );
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.9: missing plugins return minus 32001.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn missing_plugins_return_minus_32001() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     assert_eq!(
         h.call("absent.run", json!({})).await.unwrap_err()["code"],
         -32001
     );
+
+    h.stop().await;
+}
+
+/// PROTOCOL 9: unimplemented core methods return method not found.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn unimplemented_core_methods_return_method_not_found() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     assert_eq!(
         h.call("core.activate", json!({})).await.unwrap_err()["code"],
         -32601
     );
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.1: params must be objects.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn params_must_be_objects() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     assert_eq!(
         h.call("probe.echo", json!([])).await.unwrap_err()["code"],
         -32602
     );
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.9: exit without answer reports status and last logs.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exit_without_answer_reports_status_and_last_logs() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     let e = h.call("probe.exit", json!({})).await.unwrap_err();
     assert_eq!(e["code"], -32002);
     assert!(
@@ -123,10 +224,13 @@ async fn routing_environment_files_and_errors() {
             .iter()
             .any(|l| l["line"] == "last line")
     );
+
     h.stop().await;
 }
-#[tokio::test(flavor = "multi_thread")]
-async fn journal_notifications_locks_and_kill() {
+
+/// PROTOCOL 3.5: progress notifications reach the caller.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn progress_notifications_reach_the_caller() {
     let w = TestWorld::new();
     let h = w.start().await;
     let mut notifications = h.subscribe();
@@ -135,6 +239,15 @@ async fn journal_notifications_locks_and_kill() {
         notifications.recv().await.unwrap()["params"]["name"],
         "progress"
     );
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.1: non message stdout warns once without its content.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn non_message_stdout_warns_once_without_its_content() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     call(&h, "probe.junk", json!({})).await;
     tokio::time::timeout(std::time::Duration::from_secs(5), async {
         while !h.processes().is_empty() {
@@ -148,8 +261,28 @@ async fn journal_notifications_locks_and_kill() {
     let warnings = call(&h, "core.log", json!({"pid":pid,"grep":"discarded stdout"})).await;
     assert_eq!(warnings.as_array().unwrap().len(), 1);
     assert!(!warnings.to_string().contains("junk"));
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.1: stderr lines enter the journal.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn stderr_lines_enter_the_journal() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    call(&h, "probe.junk", json!({})).await;
+    h.stop().await;
     let log = call(&h, "core.log", json!({"grep":"stderr line"})).await;
     assert_eq!(log.as_array().unwrap().len(), 1);
+
+    h.stop().await;
+}
+
+/// PROTOCOL 6: try lock reports held without waiting.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn try_lock_reports_held_without_waiting() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     assert_eq!(
         call(&h, "core.lock", json!({"path":"x"})).await,
         json!({"held":true})
@@ -163,13 +296,25 @@ async fn journal_notifications_locks_and_kill() {
         call(&h, "probe.lock", json!({"path":"x"})).await,
         json!({"held":true})
     );
+
+    h.stop().await;
+}
+
+/// PROTOCOL 9: kill ends a process at once.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn kill_ends_a_process_at_once() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     let pid = call(&h, "probe.serve", json!({})).await;
     call(&h, "core.kill", json!({"pid":pid})).await;
     assert!(!h.processes().contains_key(pid.as_str().unwrap()));
+
     h.stop().await;
 }
-#[tokio::test(flavor = "multi_thread")]
-async fn native_and_reserved_world_paths() {
+
+/// PROTOCOL 9: native calls have process ids and results.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn native_calls_have_process_ids_and_results() {
     let w = TestWorld::new();
     let h = w.start().await;
     h.register_native(
@@ -184,21 +329,30 @@ async fn native_and_reserved_world_paths() {
     let result = call(&h, "native.echo", json!({"x":1})).await;
     assert_eq!(result["method"], "echo");
     assert!(result["pid"].as_str().unwrap().starts_with("native-"));
+
     h.stop().await;
+}
+
+/// PROTOCOL 2: reserved world roots prevent startup.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn reserved_world_roots_prevent_startup() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     for name in ["proc", "run", "self"] {
         std::fs::create_dir(w.0.join(name)).unwrap();
         assert!(World::open(&w.0).is_err());
         std::fs::remove_dir(w.0.join(name)).unwrap();
     }
+    h.stop().await;
 }
 
-#[tokio::test(flavor = "multi_thread")]
-async fn tree_detach_stop_boot_and_scheduler() {
+/// PROTOCOL 8: boot waits for each readiness result before the next service.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn boot_waits_for_readiness_in_order() {
     let w = TestWorld::new();
     std::fs::write(
         w.0.join("config/boot.json"),
         serde_json::to_vec(&json!([
-            {"method":"absent.start","params":{}},
             {"method":"probe.write","params":{"path":"/state/boot","text":"ready"}},
             {"method":"probe.read","params":{"path":"/state/boot"}}
         ]))
@@ -210,6 +364,15 @@ async fn tree_detach_stop_boot_and_scheduler() {
         std::fs::read_to_string(w.0.join("state/boot")).unwrap(),
         "ready"
     );
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.6: attached children stop when the parent ends.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn attached_children_stop_when_the_parent_ends() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     let attached = call(
         &h,
         "probe.call",
@@ -223,6 +386,15 @@ async fn tree_detach_stop_boot_and_scheduler() {
     })
     .await
     .unwrap();
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.4: detached calls have no parent and outlive the caller.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn detached_calls_have_no_parent_and_outlive_the_caller() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     let detached = call(
         &h,
         "probe.call",
@@ -231,6 +403,15 @@ async fn tree_detach_stop_boot_and_scheduler() {
     .await;
     assert!(h.processes()[detached.as_str().unwrap()].detached);
     call(&h, "core.stop", json!({"pid":detached})).await;
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.8: busy guests do not block other calls.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn busy_guests_do_not_block_other_calls() {
+    let w = TestWorld::new();
+    let h = w.start().await;
     let hc = h.clone();
     let busy = tokio::spawn(async move { hc.call("probe.busy", json!({})).await });
     let pid = tokio::time::timeout(std::time::Duration::from_secs(20), async {
@@ -249,13 +430,16 @@ async fn tree_detach_stop_boot_and_scheduler() {
     );
     call(&h, "core.kill", json!({"pid":pid})).await;
     assert_eq!(busy.await.unwrap().unwrap_err()["code"], -32002);
+
     h.stop().await;
 }
 
-#[tokio::test(flavor = "multi_thread")]
+/// PROTOCOL 6: same process lock acquisitions queue.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn same_process_lock_acquisitions_queue() {
     let w = TestWorld::new();
     let h = w.start().await;
+
     h.register_native(
         "locks",
         std::sync::Arc::new(|ctx, _, _| {
@@ -290,46 +474,16 @@ async fn same_process_lock_acquisitions_queue() {
         call(&h, "core.lock", json!({"path":"fifo","wait":false})).await,
         json!({"held":true})
     );
-}
-
-#[tokio::test(flavor = "multi_thread")]
-async fn stop_closes_stdin_deadline_and_exit() {
-    let w = TestWorld::new();
-    let h = w.start().await;
-    let hc = h.clone();
-    let pending = tokio::spawn(async move { hc.call("probe.wait_eof", json!({})).await });
-    let pid = tokio::time::timeout(std::time::Duration::from_secs(20), async {
-        loop {
-            if let Some((pid, _)) = h
-                .processes()
-                .into_iter()
-                .find(|(_, p)| p.method == "wait_eof")
-            {
-                break pid;
-            }
-            tokio::task::yield_now().await;
-        }
-    })
-    .await
-    .unwrap();
-    call(&h, "core.stop", json!({"pid":pid})).await;
-    assert_eq!(
-        std::fs::read_to_string(w.0.join("state/stopped")).unwrap(),
-        "yes"
-    );
-    assert_eq!(pending.await.unwrap().unwrap_err()["code"], -32002);
-    assert_eq!(call(&h, "core.exit", json!({})).await, json!({}));
-    tokio::time::timeout(std::time::Duration::from_secs(1), h.exit_requested())
-        .await
-        .unwrap();
     h.stop().await;
 }
 
-#[tokio::test(flavor = "multi_thread")]
+/// PROTOCOL 3.8: tcp udp and dns.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn tcp_udp_and_dns() {
-    use tokio::io::AsyncWriteExt;
     let w = TestWorld::new();
     let h = w.start().await;
+
+    use tokio::io::AsyncWriteExt;
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
     let tcp = format!("localhost:{}", listener.local_addr().unwrap().port());
     let udp = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
@@ -346,11 +500,16 @@ async fn tcp_udp_and_dns() {
     udp.recv_from(&mut buf).await.unwrap();
     assert_eq!(&buf, b"udp");
     server.await.unwrap();
+
     h.stop().await;
 }
 
-#[tokio::test(flavor = "multi_thread")]
+/// PROTOCOL 6: native kill drops future and releases locks.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn native_kill_drops_future_and_releases_locks() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+
     use std::sync::{
         Arc,
         atomic::{AtomicBool, Ordering},
@@ -361,8 +520,6 @@ async fn native_kill_drops_future_and_releases_locks() {
             self.0.store(true, Ordering::SeqCst);
         }
     }
-    let w = TestWorld::new();
-    let h = w.start().await;
     let dropped = Arc::new(AtomicBool::new(false));
     let flag = dropped.clone();
     h.register_native(
@@ -405,13 +562,16 @@ async fn native_kill_drops_future_and_releases_locks() {
         json!({"held":true})
     );
     assert_eq!(pending.await.unwrap().unwrap_err()["code"], -32002);
+
     h.stop().await;
 }
 
-#[tokio::test(flavor = "multi_thread")]
+/// PROTOCOL 3.6: shutdown stops busy processes in parallel after five seconds.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn shutdown_stops_busy_processes_in_parallel_after_five_seconds() {
     let w = TestWorld::new();
     let h = w.start().await;
+
     let mut tasks = Vec::new();
     for _ in 0..2 {
         let hc = h.clone();
@@ -434,61 +594,16 @@ async fn shutdown_stops_busy_processes_in_parallel_after_five_seconds() {
         assert_eq!(task.await.unwrap().unwrap_err()["code"], -32002);
     }
     assert!(h.processes().is_empty());
+    h.stop().await;
 }
 
-#[tokio::test(flavor = "multi_thread")]
-async fn journal_filters_persist_and_process_versions_are_hashes() {
-    use sha2::{Digest, Sha256};
-    let w = TestWorld::new();
-    let h = w.start().await;
-    let pid = call(&h, "probe.serve", json!({"version":1})).await;
-    let pid = pid.as_str().unwrap();
-    let info = h.processes()[pid].clone();
-    assert_eq!(
-        info.version,
-        format!("{:x}", Sha256::digest(std::fs::read(fixture()).unwrap()))
-    );
-    assert_eq!(info.plugin, "probe");
-    assert_eq!(info.method, "serve");
-    assert_eq!(info.params, json!({"version":1}));
-    assert_eq!(info.parent.as_deref(), Some("console"));
-    assert_eq!(info.state, "running");
-    assert!(!info.detached);
-    assert!(info.start_time > 0);
-    let uuid = uuid::Uuid::parse_str(pid.strip_prefix("probe-").unwrap()).unwrap();
-    assert_eq!(uuid.get_version_num(), 7);
-    call(&h, "core.stop", json!({"pid":pid})).await;
-    let entries = call(
-        &h,
-        "core.log",
-        json!({"plugin":"probe","pid":pid,"limit":2}),
-    )
-    .await;
-    assert_eq!(entries.as_array().unwrap().len(), 2);
-    let entry = &entries[0];
-    let id = &entry["id"];
-    let single = call(&h, "core.log", json!({"since":id,"until":id})).await;
-    assert_eq!(single, json!([entry]));
-    h.stop().await;
-    drop(h);
-    let h = w.start().await;
-    assert_eq!(
-        call(&h, "core.log", json!({"since":id,"until":id})).await,
-        single
-    );
-    h.stop().await;
-    assert!(
-        std::fs::read(w.0.join("state/core/journal.sqlite"))
-            .unwrap()
-            .starts_with(b"SQLite format 3")
-    );
-}
-
-#[tokio::test(flavor = "multi_thread")]
+/// PROTOCOL 3.6: stopping rejects calls but allows notifications.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn stopping_rejects_calls_but_allows_notifications() {
-    use std::sync::{Arc, Mutex};
     let w = TestWorld::new();
     let h = w.start().await;
+
+    use std::sync::{Arc, Mutex};
     let (tx, rx) = tokio::sync::oneshot::channel();
     let tx = Arc::new(Mutex::new(Some(tx)));
     let gate = Arc::new(tokio::sync::Notify::new());
@@ -533,5 +648,465 @@ async fn stopping_rejects_calls_but_allows_notifications() {
     release.notify_one();
     assert_eq!(pending.await.unwrap().unwrap(), json!({}));
     assert_eq!(stop.await.unwrap().unwrap(), json!({}));
+
+    h.stop().await;
+}
+
+/// PROTOCOL 9: stop closes stdin and waits for exit.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn stop_closes_stdin_and_waits_for_exit() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    let hc = h.clone();
+    let pending = tokio::spawn(async move { hc.call("probe.wait_eof", json!({})).await });
+    let pid = tokio::time::timeout(std::time::Duration::from_secs(20), async {
+        loop {
+            if let Some((pid, _)) = h
+                .processes()
+                .into_iter()
+                .find(|(_, p)| p.method == "wait_eof")
+            {
+                break pid;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+    call(&h, "core.stop", json!({"pid":pid})).await;
+    assert_eq!(
+        std::fs::read_to_string(w.0.join("state/stopped")).unwrap(),
+        "yes"
+    );
+    assert_eq!(pending.await.unwrap().unwrap_err()["code"], -32002);
+
+    h.stop().await;
+}
+
+/// PROTOCOL 9: core exit requests restart from its handler.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn core_exit_requests_restart_from_its_handler() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    assert_eq!(call(&h, "core.exit", json!({})).await, json!({}));
+    tokio::time::timeout(std::time::Duration::from_secs(1), h.exit_requested())
+        .await
+        .unwrap();
+
+    h.stop().await;
+}
+
+/// PROTOCOL 3.3: process ids are uuidv7 and versions are sha256.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn process_ids_are_uuidv7_and_versions_are_sha256() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    use sha2::{Digest, Sha256};
+    let pid = call(&h, "probe.serve", json!({"version":1})).await;
+    let pid = pid.as_str().unwrap();
+    let info = h.processes()[pid].clone();
+    assert_eq!(
+        info.version,
+        format!("{:x}", Sha256::digest(std::fs::read(fixture()).unwrap()))
+    );
+    assert_eq!(info.plugin, "probe");
+    assert_eq!(info.method, "serve");
+    assert_eq!(info.params, json!({"version":1}));
+    assert_eq!(info.parent.as_deref(), Some("console"));
+    assert_eq!(info.state, "running");
+    assert!(!info.detached);
+    assert!(info.start_time > 0);
+    let uuid = uuid::Uuid::parse_str(pid.strip_prefix("probe-").unwrap()).unwrap();
+    assert_eq!(uuid.get_version_num(), 7);
+    call(&h, "core.stop", json!({"pid":pid})).await;
+
+    h.stop().await;
+}
+
+/// PROTOCOL 7: journal filters and entries persist in sqlite.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn journal_filters_and_entries_persist_in_sqlite() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    let pid = call(&h, "probe.serve", json!({})).await;
+    call(&h, "core.stop", json!({"pid":pid})).await;
+    let entries = call(
+        &h,
+        "core.log",
+        json!({"plugin":"probe","pid":pid,"limit":2}),
+    )
+    .await;
+    assert_eq!(entries.as_array().unwrap().len(), 2);
+    let entry = &entries[0];
+    let id = &entry["id"];
+    let single = call(&h, "core.log", json!({"since":id,"until":id})).await;
+    assert_eq!(single, json!([entry]));
+    h.stop().await;
+    drop(h);
+    let h = w.start().await;
+    assert_eq!(
+        call(&h, "core.log", json!({"since":id,"until":id})).await,
+        single
+    );
+    h.stop().await;
+    assert!(
+        std::fs::read(w.0.join("state/core/journal.sqlite"))
+            .unwrap()
+            .starts_with(b"SQLite format 3")
+    );
+    h.stop().await;
+}
+
+/// PROTOCOL 3.1: blank stdout lines do not warn.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn blank_stdout_lines_do_not_warn() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    call(&h, "probe.blank", json!({})).await;
+    h.stop().await;
+    let warnings = call(&h, "core.log", json!({"grep":"discarded stdout"})).await;
+    assert_eq!(warnings, json!([]));
+    h.stop().await;
+}
+
+/// PROTOCOL 3.1: large requests and responses cross pipe capacity.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn large_requests_and_responses_cross_pipe_capacity() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    let params = json!({"text":"x".repeat(192*1024)});
+    assert_eq!(call(&h, "probe.echo", params.clone()).await, params);
+    let outgoing = json!({"method":"probe.echo","params":params});
+    assert_eq!(call(&h, "probe.call", outgoing).await, params);
+    h.stop().await;
+}
+
+/// PROTOCOL 3.8: two plugins execute calls at the same time.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn two_plugins_execute_calls_at_the_same_time() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    let other = w.0.join("plugins/other");
+    std::fs::create_dir_all(&other).unwrap();
+    std::fs::copy(fixture(), other.join("plugin.wasm")).unwrap();
+    let hc = h.clone();
+    let slow = tokio::spawn(async move { hc.call("probe.sleep", json!({"ms":1500})).await });
+    while !h.processes().values().any(|p| p.method == "sleep") {
+        tokio::task::yield_now().await;
+    }
+    assert_eq!(
+        call(&h, "other.echo", json!({"ready":true})).await,
+        json!({"ready":true})
+    );
+    assert!(!slow.is_finished());
+    slow.await.unwrap().unwrap();
+    h.stop().await;
+}
+
+/// PROTOCOL 3.3: request id is the process id.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn request_id_is_the_process_id() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    let pid = call(&h, "probe.request_id", json!({})).await;
+    let uuid =
+        uuid::Uuid::parse_str(pid.as_str().unwrap().strip_prefix("probe-").unwrap()).unwrap();
+    assert_eq!(uuid.get_version_num(), 7);
+    h.stop().await;
+}
+
+/// PROTOCOL 9: core call returns the target result.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn core_call_returns_the_target_result() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    assert_eq!(
+        call(
+            &h,
+            "core.call",
+            json!({"method":"probe.echo","params":{"n":1}})
+        )
+        .await,
+        json!({"n":1})
+    );
+    h.stop().await;
+}
+
+/// PROTOCOL 9: core unlock releases the named lock.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn core_unlock_releases_the_named_lock() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    call(&h, "core.lock", json!({"path":"unlock"})).await;
+    call(&h, "core.unlock", json!({"path":"unlock"})).await;
+    assert_eq!(
+        call(&h, "core.lock", json!({"path":"unlock","wait":false})).await,
+        json!({"held":true})
+    );
+    h.stop().await;
+}
+
+/// PROTOCOL 6: locks do not block file writes.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn locks_do_not_block_file_writes() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    call(&h, "core.lock", json!({"path":"/state/advisory"})).await;
+    call(
+        &h,
+        "probe.write",
+        json!({"path":"/state/advisory","text":"allowed"}),
+    )
+    .await;
+    assert_eq!(
+        std::fs::read_to_string(w.0.join("state/advisory")).unwrap(),
+        "allowed"
+    );
+    h.stop().await;
+}
+
+/// PROTOCOL 6: crashes release all process locks.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn crashes_release_all_process_locks() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    h.register_native(
+        "crash",
+        std::sync::Arc::new(|ctx, _, _| {
+            Box::pin(async move {
+                ctx.call("core.lock", json!({"path":"crash"}))
+                    .await
+                    .unwrap();
+                panic!("test crash");
+            })
+        }),
+    );
+    assert_eq!(
+        h.call("crash.run", json!({})).await.unwrap_err()["code"],
+        -32002
+    );
+    assert_eq!(
+        call(&h, "core.lock", json!({"path":"crash","wait":false})).await,
+        json!({"held":true})
+    );
+    h.stop().await;
+}
+
+/// PROTOCOL 7: journal entries have uuid time identity stream and line.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn journal_entries_have_uuid_time_identity_stream_and_line() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    call(&h, "probe.junk", json!({})).await;
+    h.stop().await;
+    let entries = call(&h, "core.log", json!({"plugin":"probe"})).await;
+    for entry in entries.as_array().unwrap() {
+        assert_eq!(
+            uuid::Uuid::parse_str(entry["id"].as_str().unwrap())
+                .unwrap()
+                .get_version_num(),
+            7
+        );
+        assert!(entry["time"].as_str().unwrap().contains('T'));
+        assert!(entry["time"].as_str().unwrap().ends_with('Z'));
+        assert!(entry["pid"].as_str().unwrap().starts_with("probe-"));
+        assert!(["stderr", "core"].contains(&entry["stream"].as_str().unwrap()));
+        assert!(entry["line"].is_string());
+    }
+    h.stop().await;
+}
+
+/// PROTOCOL 7: journal has wal and identity indexes.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn journal_has_wal_and_identity_indexes() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    h.stop().await;
+    let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
+    assert_eq!(
+        db.query_row("PRAGMA journal_mode", [], |r| r.get::<_, String>(0))
+            .unwrap(),
+        "wal"
+    );
+    for name in ["journal_pid", "journal_plugin"] {
+        assert_eq!(
+            db.query_row(
+                "SELECT count(*) FROM sqlite_master WHERE type='index' AND name=?1",
+                [name],
+                |r| r.get::<_, i64>(0)
+            )
+            .unwrap(),
+            1
+        );
+    }
+    h.stop().await;
+}
+
+/// PROTOCOL 9: core exit works through core call.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn core_exit_works_through_core_call() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    assert_eq!(
+        call(&h, "core.call", json!({"method":"core.exit","params":{}})).await,
+        json!({})
+    );
+    assert!(h.restart_requested());
+    h.stop().await;
+}
+
+/// PROTOCOL 6: acquisitions from different processes keep their FIFO positions.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn lock_acquisitions_from_different_processes_are_fifo() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+    let gate = std::sync::Arc::new(tokio::sync::Semaphore::new(0));
+    let release = gate.clone();
+    h.register_native(
+        "queued",
+        std::sync::Arc::new(move |ctx, _, _| {
+            let tx = tx.clone();
+            let gate = gate.clone();
+            Box::pin(async move {
+                tx.send(ctx).unwrap();
+                gate.acquire().await.unwrap().forget();
+                Ok(json!({}))
+            })
+        }),
+    );
+    let first_host = h.clone();
+    let first = tokio::spawn(async move { first_host.call("queued.wait", json!({})).await });
+    let second_host = h.clone();
+    let second = tokio::spawn(async move { second_host.call("queued.wait", json!({})).await });
+    let first_ctx = rx.recv().await.unwrap();
+    let second_ctx = rx.recv().await.unwrap();
+    call(&h, "core.lock", json!({"path":"ordered"})).await;
+    let mut first_lock = Box::pin(first_ctx.call("core.lock", json!({"path":"ordered"})));
+    let mut second_lock = Box::pin(second_ctx.call("core.lock", json!({"path":"ordered"})));
+    std::future::poll_fn(|cx| {
+        assert!(first_lock.as_mut().poll(cx).is_pending());
+        assert!(second_lock.as_mut().poll(cx).is_pending());
+        std::task::Poll::Ready(())
+    })
+    .await;
+    call(&h, "core.unlock", json!({"path":"ordered"})).await;
+    assert_eq!(first_lock.await.unwrap(), json!({"held":true}));
+    std::future::poll_fn(|cx| {
+        assert!(second_lock.as_mut().poll(cx).is_pending());
+        std::task::Poll::Ready(())
+    })
+    .await;
+    first_ctx
+        .call("core.unlock", json!({"path":"ordered"}))
+        .await
+        .unwrap();
+    assert_eq!(second_lock.await.unwrap(), json!({"held":true}));
+    release.add_permits(2);
+    first.await.unwrap().unwrap();
+    second.await.unwrap().unwrap();
+    h.stop().await;
+}
+
+/// PROTOCOL 3.5: notifications after the result do not reach the caller.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn notifications_after_the_result_are_dropped() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    let (tx, rx) = tokio::sync::oneshot::channel();
+    let tx = std::sync::Mutex::new(Some(tx));
+    h.register_native(
+        "native",
+        std::sync::Arc::new(move |ctx, _, _| {
+            tx.lock().unwrap().take().unwrap().send(ctx).ok();
+            Box::pin(async { Ok(json!({})) })
+        }),
+    );
+    let mut notifications = h.subscribe();
+    call(&h, "native.answer", json!({})).await;
+    let ctx = rx.await.unwrap();
+    ctx.notify("late", json!({}));
+    assert!(matches!(
+        notifications.try_recv(),
+        Err(tokio::sync::broadcast::error::TryRecvError::Empty)
+    ));
+    h.stop().await;
+}
+
+/// PROTOCOL 3.1: non-UTF-8 stdout is discarded, not treated as blank framing.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn non_utf8_stdout_is_not_blank_framing() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    call(&h, "probe.invalid_utf8", json!({})).await;
+    h.stop().await;
+    let warnings = call(&h, "core.log", json!({"grep":"discarded stdout"})).await;
+    assert_eq!(warnings.as_array().unwrap().len(), 1);
+}
+
+/// PROTOCOL 8: a failed service enters the journal and does not prevent later services.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn failed_boot_services_are_logged_and_boot_continues() {
+    let w = TestWorld::new();
+    let services = json!([
+        {"method":"absent.start","params":{}},
+        {"method":"probe.write","params":{"path":"/state/continued","text":"yes"}}
+    ]);
+    std::fs::write(
+        w.0.join("config/boot.json"),
+        serde_json::to_vec(&services).unwrap(),
+    )
+    .unwrap();
+    let h = w.start().await;
+    assert_eq!(
+        std::fs::read_to_string(w.0.join("state/continued")).unwrap(),
+        "yes"
+    );
+    let errors = call(
+        &h,
+        "core.log",
+        json!({"plugin":"core","grep":"no such plugin"}),
+    )
+    .await;
+    assert_eq!(errors.as_array().unwrap().len(), 1);
+    h.stop().await;
+}
+
+/// PROTOCOL 8: boot services are detached and remain alive after their readiness result.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn boot_services_are_detached() {
+    let w = TestWorld::new();
+    std::fs::write(
+        w.0.join("config/boot.json"),
+        br#"[{"method":"probe.serve","params":{}}]"#,
+    )
+    .unwrap();
+    let h = w.start().await;
+    let processes = h.processes();
+    assert_eq!(processes.len(), 1);
+    assert!(processes.values().next().unwrap().detached);
+    h.stop().await;
+}
+
+/// PROTOCOL 7: the core records start parameters, stop, kill, and exit status.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn core_lifetime_events_enter_the_journal() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    let stopped = call(&h, "probe.serve", json!({"n":1})).await;
+    call(&h, "core.stop", json!({"pid":stopped})).await;
+    let killed = call(&h, "probe.serve", json!({"n":2})).await;
+    call(&h, "core.kill", json!({"pid":killed})).await;
+    let entries = call(&h, "core.log", json!({"plugin":"probe","limit":100})).await;
+    let lines: Vec<_> = entries
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|entry| entry["line"].as_str().unwrap())
+        .collect();
+    assert!(lines.contains(&"start serve {\"n\":1}"));
+    for line in ["stop", "kill", "exit 0", "killed"] {
+        assert!(lines.contains(&line), "missing {line}");
+    }
     h.stop().await;
 }

cli/tests/plugins/probe/src/main.rs

@@ -1,47 +1,219 @@
-//! A raw JSON-RPC guest for host integration tests.
-use std::io::{self,BufRead,Write};
+//! Raw JSON-RPC methods used by the host's protocol tests.
 use serde_json::{Value, json};
-fn send(v:Value) { let mut out=io::stdout().lock();writeln!(out,"\n{v}").unwrap();out.flush().unwrap(); }
-fn main() {
-    let mut lines=io::stdin().lock().lines();
-    let request:Value=serde_json::from_str(&lines.next().unwrap().unwrap()).unwrap();
-    let id=request.get("id").cloned();let p=&request["params"];
+use std::io::{self, BufRead, Read, Write};
+type Lines = io::Lines<io::BufReader<io::Stdin>>;
+
+fn send(value: Value) {
+    let mut output = io::stdout().lock();
+    writeln!(output, "\n{value}").unwrap();
+    output.flush().unwrap();
+}
+fn receive(lines: &mut Lines) -> Value {
+    serde_json::from_str(&lines.next().unwrap().unwrap()).unwrap()
+}
+fn echo(params: &Value) -> Value {
+    params.clone()
+}
+fn environment() -> Value {
+    json!({
+        "pid": std::env::var("SAUSAGE_PID").unwrap(),
+        "plugin": std::env::var("SAUSAGE_PLUGIN").unwrap(),
+        "parent": std::env::var("SAUSAGE_PARENT").unwrap(),
+    })
+}
+fn multicall(lines: &mut Lines) -> Value {
+    send(json!({
+        "jsonrpc": "2.0",
+        "id": 1,
+        "method": "probe.echo",
+        "params": {"n": 1}
+    }));
+    send(json!({
+        "jsonrpc": "2.0",
+        "id": "1",
+        "method": "probe.echo",
+        "params": {"n": 2}
+    }));
+    for _ in 0..2 {
+        let response = receive(lines);
+        if response["id"] == 1 {
+            assert_eq!(response["result"]["n"], 1);
+        } else {
+            assert_eq!(response["id"], "1");
+            assert_eq!(response["result"]["n"], 2);
+        }
+    }
+    json!({"count":2})
+}
+fn notification(params: &Value) -> Value {
+    send(json!({
+        "jsonrpc": "2.0",
+        "method": "probe.write",
+        "params": params
+    }));
+    for _ in 0..200 {
+        let contents = std::fs::read_to_string(params["path"].as_str().unwrap()).ok();
+        if contents.as_deref() == params["text"].as_str() {
+            break;
+        }
+        std::thread::sleep(std::time::Duration::from_millis(10));
+    }
+    json!({})
+}
+fn unknown_id() {
+    send(json!({
+        "jsonrpc": "2.0",
+        "id": "late",
+        "result": 7
+    }));
+}
+fn wait_eof(lines: &mut Lines) {
+    for _ in lines {}
+    std::fs::write("/state/stopped", "yes").unwrap();
+}
+fn busy() -> Value {
+    loop {
+        std::hint::spin_loop();
+    }
+}
+fn sleep(params: &Value) -> Value {
+    let duration = std::time::Duration::from_millis(params["ms"].as_u64().unwrap_or(100));
+    std::thread::sleep(duration);
+    json!({})
+}
+fn exit_without_answer() {
+    eprintln!("last line");
+}
+fn junk() -> Value {
+    io::stdout().write_all(&[255, 10]).unwrap();
+    println!("junk\n{{}}\n[]");
+    eprintln!("stderr line");
+    json!({})
+}
+fn invalid_utf8() -> Value {
+    io::stdout().write_all(&[255, 10]).unwrap();
+    json!({})
+}
+fn blank() -> Value {
+    println!("\n \t\n");
+    json!({})
+}
+fn notify() -> Value {
+    send(json!({
+        "jsonrpc": "2.0",
+        "method": "core.notify",
+        "params": {
+            "name": "progress",
+            "params": {"n": 1}
+        }
+    }));
+    std::thread::sleep(std::time::Duration::from_millis(30));
+    json!({})
+}
+fn serve(request: &Value, lines: &mut Lines) {
+    send(json!({
+        "jsonrpc": "2.0",
+        "id": request["id"],
+        "result": std::env::var("SAUSAGE_PID").unwrap()
+    }));
+    for _ in lines {}
+}
+fn network(params: &Value) -> Value {
+    let mut socket = std::net::TcpStream::connect(params["tcp"].as_str().unwrap()).unwrap();
+    let mut text = String::new();
+    socket.read_to_string(&mut text).unwrap();
+    let udp = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
+    udp.send_to(b"udp", params["udp"].as_str().unwrap())
+        .unwrap();
+    json!(text)
+}
+fn read_file(params: &Value) -> Value {
+    json!(std::fs::read_to_string(params["path"].as_str().unwrap()).unwrap())
+}
+fn write_file(params: &Value) -> Value {
+    std::fs::write(
+        params["path"].as_str().unwrap(),
+        params["text"].as_str().unwrap(),
+    )
+    .unwrap();
+    json!({})
+}
+fn outgoing_call(params: &Value, lines: &mut Lines) -> Value {
+    send(json!({
+        "jsonrpc": "2.0",
+        "id": {"nested": 7},
+        "method": params["method"],
+        "params": params["params"]
+    }));
+    loop {
+        let response = receive(lines);
+        if response["id"] == json!({"nested":7}) {
+            return response["result"].clone();
+        }
+    }
+}
+fn lock(params: &Value, lines: &mut Lines) -> Value {
+    outgoing_call(&json!({"method":"core.lock", "params":params}), lines)
+}
+fn dispatch(request: &Value, lines: &mut Lines) -> Option<Value> {
+    let params = &request["params"];
     let result = match request["method"].as_str().unwrap() {
-        "echo"|"rpc.discover"=>p.clone(),
-        "env"=>json!({"pid":std::env::var("SAUSAGE_PID").unwrap(),"plugin":std::env::var("SAUSAGE_PLUGIN").unwrap(),"parent":std::env::var("SAUSAGE_PARENT").unwrap()}),
-        "multicall"=>{
-            send(json!({"jsonrpc":"2.0","id":1,"method":"probe.echo","params":{"n":1}}));
-            send(json!({"jsonrpc":"2.0","id":"1","method":"probe.echo","params":{"n":2}}));
-            let mut seen=Vec::new();while seen.len()<2 {let r:Value=serde_json::from_str(&lines.next().unwrap().unwrap()).unwrap();if r["id"]==1 {assert_eq!(r["result"]["n"],1);} else {assert_eq!(r["id"],"1");assert_eq!(r["result"]["n"],2);}seen.push(r["id"].clone());}json!({"count":seen.len()})
-        },
-        "notification"=>{
-            send(json!({"jsonrpc":"2.0","method":"probe.write","params":p}));
-            for _ in 0..200 {if std::fs::exists(p["path"].as_str().unwrap()).unwrap() {break;}std::thread::sleep(std::time::Duration::from_millis(10));}json!({})
-        },
-        "unknownid"=>{send(json!({"jsonrpc":"2.0","id":"late","result":7}));return;},
+        "echo" | "rpc.discover" => echo(params),
+        "env" => environment(),
+        "multicall" => multicall(lines),
+        "notification" => notification(params),
+        "unknownid" => {
+            unknown_id();
+            return None;
+        }
         "request_id" => request["id"].clone(),
-        "wait_eof"=>{for _ in lines {} std::fs::write("/state/stopped","yes").unwrap(); return;},
-        "busy"=>loop {std::hint::spin_loop();},
-        "sleep"=>{std::thread::sleep(std::time::Duration::from_millis(p["ms"].as_u64().unwrap_or(100)));json!({})},
-        "exit"=>{eprintln!("last line");return;},
-        "junk"=>{io::stdout().write_all(&[255,10]).unwrap();println!("junk\n{{}}\n[]");eprintln!("stderr line");json!({})},
-        "notify"=>{send(json!({"jsonrpc":"2.0","method":"core.notify","params":{"name":"progress","params":{"n":1}}}));std::thread::sleep(std::time::Duration::from_millis(30));json!({})},
-        "serve"=>{send(json!({"jsonrpc":"2.0","id":id,"result":std::env::var("SAUSAGE_PID").unwrap()}));for _ in lines {} return;},
-        "network"=>{
-            use std::io::Read;
-            let mut socket=std::net::TcpStream::connect(p["tcp"].as_str().unwrap()).unwrap();
-            let mut text=String::new();socket.read_to_string(&mut text).unwrap();
-            let udp=std::net::UdpSocket::bind("127.0.0.1:0").unwrap();udp.send_to(b"udp",p["udp"].as_str().unwrap()).unwrap();json!(text)
-        },
-        "read"=>json!(std::fs::read_to_string(p["path"].as_str().unwrap()).unwrap()),
-        "write"=>{std::fs::write(p["path"].as_str().unwrap(),p["text"].as_str().unwrap()).unwrap();json!({})},
-        "call"|"lock"=>{
-            let method=if request["method"]=="lock" {"core.lock"} else {p["method"].as_str().unwrap()};
-            let params=if request["method"]=="lock" {p.clone()} else {p["params"].clone()};
-            send(json!({"jsonrpc":"2.0","id":{"nested":7},"method":method,"params":params}));
-            loop {let response:Value=serde_json::from_str(&lines.next().unwrap().unwrap()).unwrap();if response["id"]==json!({"nested":7}) {break response["result"].clone();}}
-        },
-        _=>{send(json!({"jsonrpc":"2.0","id":id,"error":{"code":-32601,"message":"method not found"}}));return;}
+        "wait_eof" => {
+            wait_eof(lines);
+            return None;
+        }
+        "busy" => busy(),
+        "sleep" => sleep(params),
+        "exit" => {
+            exit_without_answer();
+            return None;
+        }
+        "junk" => junk(),
+        "blank" => blank(),
+        "invalid_utf8" => invalid_utf8(),
+        "notify" => notify(),
+        "serve" => {
+            serve(request, lines);
+            return None;
+        }
+        "network" => network(params),
+        "read" => read_file(params),
+        "write" => write_file(params),
+        "call" => outgoing_call(params, lines),
+        "lock" => lock(params, lines),
+        _ => {
+            send(json!({
+                "jsonrpc": "2.0",
+                "id": request["id"],
+                "error": {
+                    "code": -32601,
+                    "message": "method not found"
+                }
+            }));
+            return None;
+        }
     };
-    if let Some(id)=id {send(json!({"jsonrpc":"2.0","id":id,"result":result}));}
+    Some(result)
+}
+fn main() {
+    let mut lines = io::BufReader::new(io::stdin()).lines();
+    let request = receive(&mut lines);
+    if let Some(result) = dispatch(&request, &mut lines)
+        && let Some(id) = request.get("id")
+    {
+        send(json!({
+            "jsonrpc": "2.0",
+            "id": id,
+            "result": result
+        }));
+    }
 }