core: review fixes, init builds the agent, console without a terminal

Activation path (review T6R): a broken plugin or a stray file in
/plugins blocks no activation; candidates replace their own entries;
discovery has a 10 s deadline and its process is stopped after the
answer; an unchanged hash keeps plugin.prev.wasm; activations run in
their own task; the watcher is not recursive; no duplicate services;
rpc.retired waits for live processes of its version; reads and hashes
on blocking threads.

Guest I/O and per-call paths: a stdin write error is EOF, stdout and
stderr are drained before the status; stderr is lossy UTF-8 and goes
to the journal from its own reader; compiled components are freed;
shutdown starts no new processes; lock queues drop stopping processes;
compile before the admission gate; CommandPre per component;
core.unlock in order; null errors are successes; the journal thread
survives SQLite errors.

sausage init builds the agent with cargo.build; sausage without a
command creates ./world. The console reads plain lines without a
terminal. PROTOCOL.md and DECISIONS.md describe the decided rules. The
flake tests the whole workspace and checks the packaged cargo.build.

cli/Cargo.toml

@@ -31,6 +31,7 @@ sausage-interface = { version = "0.1.0", path = "../interface" }
 notify = "8.2.0"
 rustyline = "19"
 nix = { version = "0.31.3", features = ["signal", "process"] }
+chrono = { version = "0.4.42", default-features = false, features = ["clock"] }
 
 [dev-dependencies]
 hyper = { version = "1.12.0", features = ["server", "http1"] }

cli/src/activation.rs

@@ -1,7 +1,7 @@
 //! Serialized candidate-copy activation and revert under PROTOCOL.md section 10.3.
 use crate::{
     Host, RpcResult, error,
-    graph::{Interface, io_error, problems},
+    graph::{Interface, io_error, problems, read_hashed},
 };
 use serde_json::{Value, json};
 use std::{collections::BTreeMap, path::PathBuf};
@@ -14,6 +14,13 @@ struct Candidate {
 impl Host {
     /// Copies first, discovers that copy, checks the combined graph, then installs it.
     pub(crate) async fn activation(&self, _caller: &str, method: &str, params: Value) -> RpcResult {
+        let host = self.clone();
+        let method = method.to_owned();
+        tokio::spawn(async move { host.activation_body(&method, params).await })
+            .await
+            .map_err(|failure| error(-32603, &failure.to_string()))?
+    }
+    async fn activation_body(&self, method: &str, params: Value) -> RpcResult {
         let _activation = self.0.activations.lock().await;
         let mut candidates = Vec::new();
         let result = self
@@ -79,12 +86,12 @@ impl Host {
             };
             let copy = directory.join(format!("activate-{}.wasm", uuid::Uuid::now_v7()));
             // Read once. Discovery and installation both use the same private bytes.
-            let bytes = tokio::fs::read(source).await.map_err(io_error)?;
+            let (bytes, hash) = read_hashed(source).await.map_err(io_error)?;
             tokio::fs::write(&copy, &bytes).await.map_err(io_error)?;
             self.0
                 .journal
                 .write("core", plugin, "core", "activation copied candidate");
-            let discovery = self.discover_copy(plugin, &copy, &bytes).await;
+            let discovery = self.discover_copy(plugin, &copy, hash).await;
             let interface = match discovery {
                 Ok(interface) => interface,
                 Err(error) => {
@@ -101,19 +108,13 @@ impl Host {
                 interface,
             });
         }
-        self.refresh_graph().await?;
-        let previous = self.0.graph.lock().unwrap().plugins.clone();
-        if previous.values().any(|entry| entry.error.is_some()) {
-            return Err(error(
-                -32003,
-                "activation waits for successful discovery of every plugin",
-            ));
-        }
-        let mut proposed = previous.clone();
         let names: Vec<_> = candidates
             .iter()
             .map(|candidate| candidate.plugin.clone())
             .collect();
+        self.refresh_graph_except(&names).await?;
+        let previous = self.0.graph.lock().unwrap().plugins.clone();
+        let mut proposed = previous.clone();
         for candidate in candidates.iter() {
             proposed.insert(candidate.plugin.clone(), candidate.interface.clone());
         }
@@ -160,6 +161,12 @@ impl Host {
         let admission = self.0.admission.write().await;
         // Every candidate already has a temporary name before any installed file changes.
         for candidate in candidates {
+            if previous
+                .get(&candidate.plugin)
+                .is_some_and(|old| old.hash == candidate.interface.hash)
+            {
+                continue;
+            }
             let directory = self.0.world.plugin_dir(&candidate.plugin);
             let current = directory.join("plugin.wasm");
             if tokio::fs::try_exists(&current).await.map_err(io_error)? {
@@ -180,7 +187,7 @@ impl Host {
             );
             if let Some(old) = previous
                 .get(&candidate.plugin)
-                .filter(|old| old.hash != candidate.interface.hash)
+                .filter(|old| !old.hash.is_empty() && old.hash != candidate.interface.hash)
             {
                 let grace = old.document["x-sausage-grace"]
                     .as_f64()

cli/src/components.rs

@@ -1,55 +1,176 @@
-//! Component versions cached by file metadata, with independent compilation per hash.
-use crate::{error, runtime::Runtime};
+//! Metadata-keyed component loads and weak compiled versions; candidate copies never enter the path cache.
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    /// PROTOCOL 10.3: a pending component load does not occupy admission.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn loading_does_not_hold_admission() {
+        let folder =
+            std::env::temp_dir().join(format!("sausage-admission-{}", uuid::Uuid::now_v7()));
+        let world = crate::world::World::create(&folder).unwrap();
+        let path = world.plugin_dir("probe").join("plugin.wasm");
+        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
+        std::fs::write(&path, []).unwrap();
+        let host = crate::Host::start_with_cache(&folder, &folder.join("cache"))
+            .await
+            .unwrap();
+        let entry = host
+            .0
+            .components
+            .paths
+            .lock()
+            .unwrap()
+            .entry(path)
+            .or_default()
+            .clone();
+        let blocked = entry.lock().await;
+        let caller = host.clone();
+        let pending =
+            tokio::spawn(async move { caller.call("probe.echo", serde_json::json!({})).await });
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            while Arc::strong_count(&entry) < 3 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .unwrap();
+        let admitted = host.0.admission.try_write().is_ok();
+        drop(blocked);
+        assert!(pending.await.unwrap().is_err());
+        host.stop().await;
+        std::fs::remove_dir_all(folder).unwrap();
+        assert!(admitted, "a component load held the admission gate");
+    }
+    /// PROTOCOL 10.3: temporary discovery and activation copies never accumulate cached paths.
+    #[tokio::test]
+    async fn candidate_copies_do_not_enter_path_cache() {
+        let root = Path::new(env!("CARGO_MANIFEST_DIR")).parent().unwrap();
+        let target = root.join("target/core/guests");
+        assert!(
+            std::process::Command::new("cargo")
+                .args([
+                    "build",
+                    "--offline",
+                    "--target",
+                    "wasm32-wasip2",
+                    "--manifest-path"
+                ])
+                .arg(root.join("cli/tests/plugins/probe/Cargo.toml"))
+                .arg("--target-dir")
+                .arg(&target)
+                .status()
+                .unwrap()
+                .success()
+        );
+        let folder =
+            std::env::temp_dir().join(format!("sausage-component-cache-{}", uuid::Uuid::now_v7()));
+        std::fs::create_dir_all(&folder).unwrap();
+        let runtime = Runtime::new(&root.join("target/core/cache-test")).unwrap();
+        let cache = Components::default();
+        for name in ["discover-test.wasm", "activate-test.wasm"] {
+            let path = folder.join(name);
+            std::fs::copy(target.join("wasm32-wasip2/debug/probe.wasm"), &path).unwrap();
+            let loaded = cache.load(path.clone(), runtime.clone()).await.unwrap();
+            assert!(cache.paths.lock().unwrap().is_empty());
+            let weak = Arc::downgrade(&loaded.prepared);
+            drop(loaded);
+            std::fs::remove_file(path).unwrap();
+            assert!(weak.upgrade().is_none());
+        }
+        std::fs::remove_dir(folder).unwrap();
+    }
+}
+
+use crate::{
+    error,
+    runtime::{Prepared, Runtime},
+};
 use sha2::{Digest, Sha256};
 use std::{
     collections::HashMap,
-    path::PathBuf,
-    sync::{Arc, Mutex},
+    path::{Path, PathBuf},
+    sync::{Arc, Mutex, Weak},
     time::SystemTime,
 };
 use tokio::sync::{Mutex as AsyncMutex, OnceCell};
-use wasmtime::component::Component;
 
-struct Version {
+#[derive(Clone, PartialEq, Eq)]
+pub(crate) struct Fingerprint {
     size: u64,
     modified: SystemTime,
-    hash: String,
-    component: Component,
+    #[cfg(unix)]
+    identity: (u64, u64, i64, i64),
 }
-type PathEntry = Arc<AsyncMutex<Option<Version>>>;
-type CompiledEntry = Arc<OnceCell<Component>>;
-
+impl Fingerprint {
+    pub(crate) async fn read(path: &Path) -> Result<Self, serde_json::Value> {
+        let metadata = tokio::fs::metadata(path)
+            .await
+            .map_err(|_| error(-32001, "no such plugin"))?;
+        Ok(Self {
+            size: metadata.len(),
+            modified: metadata
+                .modified()
+                .map_err(|e| error(-32603, &e.to_string()))?,
+            #[cfg(unix)]
+            identity: {
+                use std::os::unix::fs::MetadataExt;
+                (
+                    metadata.dev(),
+                    metadata.ino(),
+                    metadata.ctime(),
+                    metadata.ctime_nsec(),
+                )
+            },
+        })
+    }
+}
+#[derive(Clone)]
+pub(crate) struct Loaded {
+    pub(crate) prepared: Arc<Prepared>,
+    pub(crate) hash: String,
+    pub(crate) fingerprint: Fingerprint,
+    // Retain the per-hash cell only while a cached path refers to this version.
+    _compiled: Arc<OnceCell<Arc<Prepared>>>,
+}
+type PathEntry = Arc<AsyncMutex<Option<Loaded>>>;
 #[derive(Default)]
 pub(crate) struct Components {
     paths: Mutex<HashMap<PathBuf, PathEntry>>,
-    compiled: Mutex<HashMap<String, CompiledEntry>>,
+    compiled: Mutex<HashMap<String, Weak<OnceCell<Arc<Prepared>>>>>,
 }
 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();
+    ) -> Result<Loaded, serde_json::Value> {
+        let candidate = path
+            .file_name()
+            .and_then(|n| n.to_str())
+            .is_some_and(|n| n.starts_with("activate-") || n.starts_with("discover-"));
+        let entry = if candidate {
+            Arc::default()
+        } else {
+            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()))?;
+        let fingerprint = match Fingerprint::read(&path).await {
+            Ok(metadata) => metadata,
+            Err(error) => {
+                *cached = None;
+                return Err(error);
+            }
+        };
         if let Some(version) = cached.as_ref()
-            && version.size == metadata.len()
-            && version.modified == modified
+            && version.fingerprint == fingerprint
         {
-            return Ok((version.component.clone(), version.hash.clone()));
+            return Ok(version.clone());
         }
         let (bytes, hash) = tokio::task::spawn_blocking(move || -> std::io::Result<_> {
             let bytes = std::fs::read(path)?;
@@ -59,28 +180,38 @@ impl Components {
         .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
+        let compiled = {
+            let mut versions = self.compiled.lock().unwrap();
+            versions.retain(|_, v| v.strong_count() > 0);
+            if let Some(cell) = versions.get(&hash).and_then(Weak::upgrade) {
+                cell
+            } else {
+                let cell = Arc::new(OnceCell::new());
+                versions.insert(hash.clone(), Arc::downgrade(&cell));
+                cell
+            }
+        };
+        let prepared = 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()))
+                tokio::task::spawn_blocking(move || -> anyhow::Result<_> {
+                    let component = runtime.compile(&bytes)?;
+                    Ok(Arc::new(runtime.prepare(&component)))
+                })
+                .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))
+        let loaded = Loaded {
+            prepared,
+            hash,
+            fingerprint,
+            _compiled: compiled,
+        };
+        if !candidate {
+            *cached = Some(loaded.clone());
+        }
+        Ok(loaded)
     }
 }

cli/src/console.rs

@@ -177,7 +177,34 @@ pub async fn execute_line(host: &Host, line: &str) -> Result<String> {
                 }
                 i += 1;
             }
-            call(host, "core.log", params, false).await
+            match host.call("core.log", params).await {
+                Ok(entries) => entries
+                    .as_array()
+                    .unwrap()
+                    .iter()
+                    .map(|entry| {
+                        let time =
+                            chrono::DateTime::parse_from_rfc3339(entry["time"].as_str().unwrap())?
+                                .with_timezone(&chrono::Local);
+                        let pid = entry["pid"].as_str().unwrap();
+                        let prefix = format!("{}-", entry["plugin"].as_str().unwrap());
+                        let short = pid
+                            .strip_prefix(&prefix)
+                            .and_then(|uuid| uuid.get(..8))
+                            .map(|uuid| format!("{prefix}{uuid}"))
+                            .unwrap_or_else(|| pid.to_owned());
+                        Ok(format!(
+                            "{} {} {} {}",
+                            time.format("%H:%M:%S%.3f"),
+                            short,
+                            entry["stream"].as_str().unwrap(),
+                            entry["line"].as_str().unwrap()
+                        ))
+                    })
+                    .collect::<Result<Vec<_>>>()
+                    .map(|lines| lines.join("\n")),
+                Err(error) => Ok(format!("error: {}", pretty(&error)?)),
+            }
         }
         "activate" if words.len() == 2 => {
             call(
@@ -245,7 +272,11 @@ fn indent(text: &str) -> String {
 }
 
 /// Run the line editor on a dedicated thread, with notifications through its external printer.
+/// Without a terminal (for example, input from a pipe), read plain lines instead.
 pub async fn run_input(host: Host) -> Result<()> {
+    if !std::io::IsTerminal::is_terminal(&std::io::stdin()) {
+        return run_plain(host).await;
+    }
     let history = host.0.world.state_dir("core").join("history");
     let (ready, started) = tokio::sync::oneshot::channel();
     let (sender, mut lines) = tokio::sync::mpsc::unbounded_channel::<
@@ -323,3 +354,42 @@ pub async fn run_input(host: Host) -> Result<()> {
     };
     tokio::select! {result=commands=>result,result=print_notifications=>result}
 }
+
+/// Read plain lines from stdin, without line editing or history. A detached
+/// thread reads them, so that a blocked read does not hold the runtime open.
+async fn run_plain(host: Host) -> Result<()> {
+    let (sender, mut lines) = tokio::sync::mpsc::unbounded_channel::<String>();
+    std::thread::spawn(move || {
+        for line in std::io::stdin().lines() {
+            let Ok(line) = line else { break };
+            if sender.send(line).is_err() {
+                break;
+            }
+        }
+    });
+    let mut notifications = host.subscribe();
+    let print_notifications = async move {
+        loop {
+            match notifications.recv().await {
+                Ok(value) => println!("{}", indent(&notification_text(&value))),
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
+                Err(tokio::sync::broadcast::error::RecvError::Closed) => return Ok(()),
+            }
+        }
+    };
+    let commands = async {
+        while let Some(line) = lines.recv().await {
+            let text = execute_line(&host, &line)
+                .await
+                .unwrap_or_else(|error| format!("error: {error}"));
+            if !text.is_empty() {
+                println!("{}", indent(&text));
+            }
+            if line.trim() == "quit" {
+                break;
+            }
+        }
+        Ok(())
+    };
+    tokio::select! {result=commands=>result,result=print_notifications=>result}
+}

cli/src/execution.rs

@@ -13,7 +13,7 @@ impl Host {
         &self,
         pid: String,
         info: ProcessInfo,
-        component: Option<wasmtime::component::Component>,
+        component: Option<std::sync::Arc<crate::runtime::Prepared>>,
         native: Option<NativePlugin>,
         incoming: mpsc::UnboundedReceiver<Value>,
         has_id: bool,
@@ -38,7 +38,7 @@ impl Host {
         &self,
         pid: &str,
         info: &ProcessInfo,
-        component: &wasmtime::component::Component,
+        component: &crate::runtime::Prepared,
         mut incoming: mpsc::UnboundedReceiver<Value>,
         has_id: bool,
     ) -> anyhow::Result<String> {
@@ -60,7 +60,17 @@ impl Host {
                 ),
             ],
         };
-        let mut instance = self.0.runtime.start(component, config).await?;
+        let mut instance = self
+            .0
+            .runtime
+            .start_logged(
+                component,
+                config,
+                self.0.journal.clone(),
+                pid.into(),
+                info.plugin.clone(),
+            )
+            .await?;
         let mut initial = json!({
             "jsonrpc": "2.0",
             "method": info.method,
@@ -69,25 +79,17 @@ impl Host {
         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;
+        let mut stdin_open = self.forward_input(pid, &mut instance, Some(initial)).await;
         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?;
+                    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())
     }
 
@@ -97,14 +99,14 @@ impl Host {
         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)
+    ) -> bool {
+        if let Some(value) = value
+            && self.write_input(pid, instance, value).await.is_ok()
+        {
+            return true;
         }
+        instance.stdin.take();
+        false
     }
     async fn write_input(
         &self,
@@ -124,12 +126,4 @@ impl Host {
             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/filesystem.rs

@@ -137,6 +137,8 @@ impl Proc {
 #[async_trait]
 impl Layer for Proc {
     async fn stat(&self, path: &str, follow: bool) -> Result<Stat> {
+        let path = path.trim_end_matches('/');
+        let path = if path.is_empty() { "/" } else { path };
         if matches!(path.trim_end_matches('/'), "" | "/core" | "/core/plugins") {
             return Ok(directory());
         }
@@ -147,6 +149,8 @@ impl Layer for Proc {
         folder.stat(&path, follow).await
     }
     async fn list(&self, path: &str) -> Result<Vec<Entry>> {
+        let path = path.trim_end_matches('/');
+        let path = if path.is_empty() { "/" } else { path };
         if path == "/" {
             let mut entries = vec![entry("core".into(), true)];
             entries.extend(self.0.processes().into_keys().map(|pid| entry(pid, true)));
@@ -176,6 +180,8 @@ impl Layer for Proc {
         folder.list(&path).await
     }
     async fn open(&self, path: &str, options: Open) -> Result<Arc<dyn File>> {
+        let path = path.trim_end_matches('/');
+        let path = if path.is_empty() { "/" } else { path };
         if options.write || options.create || options.truncate {
             return denied();
         }

cli/src/graph.rs

@@ -4,8 +4,8 @@ use crate::{Host, RpcResult};
 use serde_json::{Value, json};
 use sha2::{Digest, Sha256};
 use std::{
-    collections::{BTreeMap, HashMap},
-    path::Path,
+    collections::BTreeMap,
+    path::{Path, PathBuf},
 };
 
 #[derive(Clone)]
@@ -16,7 +16,6 @@ pub(crate) struct Interface {
 }
 pub(crate) struct Graph {
     pub(crate) plugins: BTreeMap<String, Interface>,
-    pub(crate) versions: HashMap<String, Value>,
     json: Vec<u8>,
     dot: Vec<u8>,
 }
@@ -26,7 +25,6 @@ impl Default for Graph {
     fn default() -> Self {
         Self {
             plugins: BTreeMap::new(),
-            versions: HashMap::new(),
             json: b"{\"plugins\":{}}".to_vec(),
             dot: b"digraph plugins {}\n".to_vec(),
         }
@@ -38,6 +36,17 @@ pub(crate) fn hash(bytes: &[u8]) -> String {
     format!("{:x}", Sha256::digest(bytes))
 }
 
+/// Reads and hashes one immutable snapshot without blocking an async worker.
+pub(crate) async fn read_hashed(path: PathBuf) -> std::io::Result<(Vec<u8>, String)> {
+    tokio::task::spawn_blocking(move || -> std::io::Result<_> {
+        let bytes = std::fs::read(path)?;
+        let hash = hash(&bytes);
+        Ok((bytes, hash))
+    })
+    .await
+    .map_err(std::io::Error::other)?
+}
+
 /// Lists broken imports, optionally restricted to changed consumers or providers.
 pub(crate) fn problems(
     plugins: &BTreeMap<String, Interface>,
@@ -105,32 +114,39 @@ impl Host {
         &self,
         plugin: &str,
         path: &Path,
-        bytes: &[u8],
+        hash: String,
     ) -> Result<Interface, Value> {
-        let hash = hash(bytes);
-        let cache_key = format!("{plugin}/{hash}");
-        let document = self
-            .route_from(
-                "core".into(),
-                Some(json!(uuid::Uuid::now_v7().to_string())),
-                format!("{plugin}.rpc.discover"),
-                json!({}),
-                true,
-                Some(path.to_path_buf()),
-            )
-            .await?;
+        let request = json!(uuid::Uuid::now_v7().to_string());
+        let discovery = self.route_from(
+            "core".into(),
+            Some(request.clone()),
+            format!("{plugin}.rpc.discover"),
+            json!({}),
+            true,
+            Some(path.to_path_buf()),
+        );
+        let result = tokio::time::timeout(std::time::Duration::from_secs(10), discovery).await;
+        let processes: Vec<_> = self
+            .0
+            .state
+            .lock()
+            .unwrap()
+            .processes
+            .iter()
+            .filter(|(_, process)| process.id.as_ref() == Some(&request))
+            .map(|(pid, _)| pid.clone())
+            .collect();
+        for pid in processes {
+            self.begin_stop(&pid);
+        }
+        let document = result
+            .map_err(|_| crate::error(-32003, "plugin discovery timed out after 10 seconds"))??;
         if !document["methods"].is_array() {
             return Err(crate::error(
                 -32003,
                 "rpc.discover did not return an OpenRPC document",
             ));
         }
-        self.0
-            .graph
-            .lock()
-            .unwrap()
-            .versions
-            .insert(cache_key, document.clone());
         Ok(Interface {
             hash,
             document,
@@ -139,21 +155,44 @@ impl Host {
     }
     /// Refreshes every installed plugin while the caller holds the activation mutex.
     pub(crate) async fn refresh_graph(&self) -> RpcResult {
+        self.refresh_graph_except(&[]).await
+    }
+    pub(crate) async fn refresh_graph_except(&self, candidates: &[String]) -> RpcResult {
         let mut directory = tokio::fs::read_dir(self.0.world.root.join("plugins"))
             .await
             .map_err(io_error)?;
         let mut plugins = BTreeMap::new();
         while let Some(entry) = directory.next_entry().await.map_err(io_error)? {
+            if !entry.file_type().await.map_err(io_error)?.is_dir() {
+                continue;
+            }
             let name = entry.file_name().to_string_lossy().into_owned();
             let source = entry.path().join("plugin.wasm");
-            let bytes = match tokio::fs::read(&source).await {
-                Ok(bytes) => bytes,
+            let (bytes, hash) = match read_hashed(source).await {
+                Ok(snapshot) => snapshot,
                 Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
-                Err(error) => return Err(io_error(error)),
+                Err(error) => {
+                    plugins.insert(
+                        name,
+                        Interface {
+                            hash: String::new(),
+                            document: Value::Null,
+                            error: Some(io_error(error)),
+                        },
+                    );
+                    continue;
+                }
             };
-            let hash = hash(&bytes);
             let previous = self.0.graph.lock().unwrap().plugins.get(&name).cloned();
-            let interface = if let Some(previous) =
+            let interface = if candidates.contains(&name) {
+                previous
+                    .filter(|entry| entry.hash == hash)
+                    .unwrap_or(Interface {
+                        hash,
+                        document: Value::Null,
+                        error: None,
+                    })
+            } else if let Some(previous) =
                 previous.filter(|entry| entry.hash == hash && entry.error.is_none())
             {
                 previous
@@ -161,8 +200,10 @@ impl Host {
                 let copy = entry
                     .path()
                     .join(format!("discover-{}.wasm", uuid::Uuid::now_v7()));
-                tokio::fs::write(&copy, &bytes).await.map_err(io_error)?;
-                let result = self.discover_copy(&name, &copy, &bytes).await;
+                let result = match tokio::fs::write(&copy, &bytes).await {
+                    Ok(()) => self.discover_copy(&name, &copy, hash.clone()).await,
+                    Err(error) => Err(io_error(error)),
+                };
                 let _ = tokio::fs::remove_file(&copy).await;
                 match result {
                     Ok(interface) => interface,
@@ -179,10 +220,9 @@ impl Host {
         let old_plugins = self.0.graph.lock().unwrap().plugins.clone();
         let mut upgrades = Vec::new();
         for (plugin, current) in &plugins {
-            if let Some(old) = old_plugins
-                .get(plugin)
-                .filter(|old| old.hash != current.hash && current.error.is_none())
-            {
+            if let Some(old) = old_plugins.get(plugin).filter(|old| {
+                !old.hash.is_empty() && old.hash != current.hash && current.error.is_none()
+            }) {
                 let grace = old.document["x-sausage-grace"]
                     .as_f64()
                     .filter(|value| value.is_finite() && *value >= 0.0)

cli/src/host.rs

@@ -23,7 +23,7 @@ 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) journal: Arc<journal::Journal>,
     pub(crate) notifications: tokio::sync::broadcast::Sender<Value>,
     pub(crate) exit: watch::Sender<bool>,
     pub(crate) components: components::Components,
@@ -31,6 +31,9 @@ pub(crate) struct Inner {
     /// process is in the table. Activation holds it for writing while it
     /// installs a version, so that it sees every process of the old version.
     pub(crate) admission: tokio::sync::RwLock<()>,
+    /// Set when the host starts to stop. After that, the core starts no new
+    /// processes, and upgrade tasks do nothing.
+    pub(crate) shutdown: std::sync::atomic::AtomicBool,
     pub(crate) run: sausage_vfs::Memory,
     pub(crate) graph: Mutex<crate::graph::Graph>,
     pub(crate) activations: tokio::sync::Mutex<()>,
@@ -65,7 +68,7 @@ impl Host {
         let host = Self(Arc::new(Inner {
             world,
             runtime,
-            journal,
+            journal: Arc::new(journal),
             state: Mutex::new(State {
                 processes: HashMap::new(),
                 locks: HashMap::new(),
@@ -76,6 +79,7 @@ impl Host {
             exit: watch::channel(false).0,
             components: components::Components::default(),
             admission: tokio::sync::RwLock::new(()),
+            shutdown: std::sync::atomic::AtomicBool::new(false),
             run: sausage_vfs::Memory::default(),
             graph: Mutex::new(crate::graph::Graph::default()),
             activations: tokio::sync::Mutex::new(()),

cli/src/journal.rs

@@ -1,4 +1,4 @@
-//! A dedicated SQLite writer batches entries and serves ordered journal queries.
+//! A dedicated SQLite writer batches entries and serves ordered queries without propagating disk failures into the host.
 use rusqlite::{Connection, params};
 use serde_json::{Value, json};
 use std::{
@@ -7,7 +7,6 @@ use std::{
     sync::{Mutex, mpsc},
 };
 use tokio::sync::oneshot;
-
 struct Entry {
     id: String,
     pid: String,
@@ -19,31 +18,27 @@ enum Command {
     Write(Entry),
     Query(Value, oneshot::Sender<Value>),
 }
-
 pub(crate) struct Journal {
     sender: mpsc::Sender<Command>,
     warned: Mutex<HashSet<String>>,
 }
 impl Journal {
-    /// 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));
+        std::thread::spawn(move || match open_database(path) {
+            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) => {
+                eprintln!("journal open failed: {error}");
+                let _ = ready.send(Err(error));
+            }
         });
         initialized.await??;
         Ok(Self {
@@ -51,7 +46,6 @@ impl Journal {
             warned: Mutex::new(HashSet::new()),
         })
     }
-    /// 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(),
@@ -60,99 +54,110 @@ impl Journal {
             stream: stream.into(),
             line: line.into(),
         };
-        self.sender
-            .send(Command::Write(entry))
-            .expect("journal writer stopped");
+        if self.sender.send(Command::Write(entry)).is_err() {
+            eprintln!("journal writer stopped");
+        }
     }
-    /// 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");
         }
     }
-    /// Queue a query after all entries that the caller already submitted.
+    pub(crate) fn forget(&self, pid: &str) {
+        self.warned.lock().unwrap().remove(pid);
+    }
     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");
+        if self.sender.send(Command::Query(filters, tx)).is_err() {
+            eprintln!("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")
+            .unwrap_or_else(|_| json!([]))
     }
 }
 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;
+    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 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);",
-    )?;
+        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");
+    let write = (|| -> rusqlite::Result<()> {
+        let transaction = db.transaction()?;
+        for command in &batch {
+            if let Command::Write(e) = 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![e.id,e.pid,e.plugin,e.stream,e.line])?;
+            }
         }
+        transaction.commit()
+    })();
+    if let Err(error) = write {
+        eprintln!("journal write failed: {error}");
     }
-    transaction.commit().expect("commit journal");
     for command in batch {
         if let Command::Query(filters, response) = command {
-            let _ = response.send(query_database(db, &filters));
+            let value = query_database(db, &filters).unwrap_or_else(|error| {
+                eprintln!("journal query failed: {error}");
+                json!([])
+            });
+            let _ = response.send(value);
         }
     }
 }
-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();
+fn query_database(db: &Connection, filters: &Value) -> rusqlite::Result<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",
+    )?;
+    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)?,
+    })))?;
+    let mut entries: Vec<_> = rows.collect::<rusqlite::Result<_>>()?;
     entries.reverse();
-    json!(entries)
+    Ok(json!(entries))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    /// PROTOCOL 7: completed processes do not leave warning identities in memory.
+    #[tokio::test]
+    async fn completion_forgets_warning_identity() {
+        let folder = std::env::temp_dir().join(format!("sausage-warning-{}", uuid::Uuid::now_v7()));
+        crate::world::World::create(&folder).unwrap();
+        let host = crate::Host::start_with_cache(&folder, &folder.join("cache"))
+            .await
+            .unwrap();
+        host.register_native(
+            "warn",
+            std::sync::Arc::new(|ctx, _, _| {
+                Box::pin(async move {
+                    ctx.host.output(&ctx.pid, "not JSON".into());
+                    Ok(json!(ctx.pid))
+                })
+            }),
+        );
+        let pid = host.call("warn.once", json!({})).await.unwrap();
+        host.call("core.stop", json!({"pid":pid})).await.unwrap();
+        let retained = host
+            .0
+            .journal
+            .warned
+            .lock()
+            .unwrap()
+            .contains(pid.as_str().unwrap());
+        host.stop().await;
+        std::fs::remove_dir_all(folder).unwrap();
+        assert!(!retained);
+    }
 }

cli/src/locks.rs

@@ -15,6 +15,15 @@ impl Host {
     pub(crate) async fn lock(&self, caller: &str, path: &str, wait: bool) -> RpcResult {
         let rx = {
             let mut state = self.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, "process is stopping or ended"));
+            }
             if let Some(lock) = state.locks.get_mut(path) {
                 if !wait {
                     return Ok(json!({"held":false}));

cli/src/main.rs

@@ -1,9 +1,15 @@
 //! Command parsing and the minimal line console.
 use anyhow::{Result, bail};
 use clap::{Parser, Subcommand};
+use include_dir::{Dir, include_dir};
 use sausage::{Host, world::World};
 use serde_json::json;
-use std::{io::Write, path::PathBuf};
+use std::{
+    io::Write,
+    path::{Path, PathBuf},
+};
+
+static AGENT: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../agent/src");
 mod test_runner;
 #[derive(Parser)]
 #[command(version)]
@@ -13,14 +19,16 @@ struct Cli {
     #[arg(long, global = true)]
     cache_dir: Option<PathBuf>,
     #[command(subcommand)]
-    command: Cmd,
+    command: Option<Cmd>,
 }
 #[derive(Subcommand)]
 enum Cmd {
     Init {
+        #[arg(default_value = "world")]
         dir: PathBuf,
     },
     Run {
+        #[arg(default_value = "world")]
         dir: PathBuf,
     },
     /// Run a WASI libtest component in isolated instances.
@@ -48,31 +56,56 @@ fn setting(var: &str, question: &str, default: Option<&str>) -> Result<String> {
         None => bail!("{question} is required"),
     }
 }
+/// Creates a configured world and builds its embedded agent through the standalone builder.
+async fn init(dir: &Path) -> Result<()> {
+    let api = setting("SAUSAGE_API", "API shape", Some("openai-responses"))?;
+    if api != "openai-responses" && api != "anthropic-messages" {
+        bail!("unknown API shape {api:?}");
+    }
+    let base_url = setting("SAUSAGE_BASE_URL", "Base URL", None)?;
+    let model = setting("SAUSAGE_MODEL", "Model", None)?;
+    let token = setting("SAUSAGE_API_TOKEN", "API token", None)?;
+    let world = World::create(dir)?;
+    std::fs::write(
+        world.root.join("config/agent.json"),
+        serde_json::to_vec_pretty(
+            &json!({"api":api,"base_url":base_url.trim_end_matches('/'),"model":model,"token":token}),
+        )?,
+    )?;
+    let builder = sausage::cargo::Cargo::new(world.clone())?;
+    let source = world.plugin_dir("agent").join("source");
+    std::fs::create_dir_all(source.join("src"))?;
+    AGENT.extract(source.join("src"))?;
+    std::fs::write(
+        source.join("Cargo.toml"),
+        include_str!("../../agent/Cargo.toml"),
+    )?;
+    println!("The first agent build takes a few minutes.");
+    let output = builder.build(&source).await?;
+    std::fs::copy(output.wasm, world.plugin_dir("agent").join("plugin.wasm"))?;
+    println!("Start the world with: sausage run {}", dir.display());
+    Ok(())
+}
 #[tokio::main]
 async fn main() -> Result<()> {
     let cli = Cli::parse();
-    match cli.command {
+    let command = match cli.command {
+        Some(command) => command,
+        None => {
+            let dir = PathBuf::from("world");
+            if !dir.join("config/boot.json").is_file() {
+                init(&dir).await?;
+            }
+            Cmd::Run { dir }
+        }
+    };
+    match command {
         Cmd::TestRunner { wasm, args } => {
             if !test_runner::run(&wasm, args).await? {
                 std::process::exit(1);
             }
         }
-        Cmd::Init { dir } => {
-            let api = setting("SAUSAGE_API", "API shape", Some("openai-responses"))?;
-            if api != "openai-responses" && api != "anthropic-messages" {
-                bail!("unknown API shape {api:?}");
-            }
-            let base_url = setting("SAUSAGE_BASE_URL", "Base URL", None)?;
-            let model = setting("SAUSAGE_MODEL", "Model", None)?;
-            let token = setting("SAUSAGE_API_TOKEN", "API token", None)?;
-            let world = World::create(&dir)?;
-            std::fs::write(
-                world.root.join("config/agent.json"),
-                serde_json::to_vec_pretty(
-                    &json!({"api":api,"base_url":base_url.trim_end_matches('/'),"model":model,"token":token}),
-                )?,
-            )?;
-        }
+        Cmd::Init { dir } => init(&dir).await?,
         Cmd::Run { dir } => {
             let mut term =
                 tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
@@ -83,8 +116,12 @@ async fn main() -> Result<()> {
             let mut notifications = host.subscribe();
             let notification_task = cli.headless.then(|| {
                 tokio::spawn(async move {
-                    while let Ok(notification) = notifications.recv().await {
-                        println!("{notification}");
+                    loop {
+                        match notifications.recv().await {
+                            Ok(notification) => println!("{notification}"),
+                            Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
+                            Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+                        }
                     }
                 })
             });

cli/src/process.rs

@@ -51,12 +51,16 @@ impl Host {
         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}));
+            let logs = p
+                .answer
+                .as_ref()
+                .map(|_| self.0.journal.query_later(json!({"pid":pid,"limit":20})));
+            self.0.journal.forget(pid);
             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 lines = logs.unwrap().await.unwrap_or_else(|_| json!([]));
                     let failure = json!({
                         "code": -32002,
                         "message": "process ended without an answer",
@@ -89,6 +93,9 @@ impl Host {
         p.info.state = "stopping".into();
         p.input.take();
         self.0.journal.write(pid, &p.info.plugin, "core", "stop");
+        for lock in state.locks.values_mut() {
+            lock.queue.retain(|(queued, _)| queued != pid);
+        }
         let children: Vec<_> = state
             .processes
             .iter()
@@ -109,7 +116,11 @@ impl Host {
     }
     /// 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_mut(pid) {
+        let mut state = self.0.state.lock().unwrap();
+        for lock in state.locks.values_mut() {
+            lock.queue.retain(|(queued, _)| queued != pid);
+        }
+        if let Some(p) = state.processes.get_mut(pid) {
             p.info.state = "stopping".into();
             p.input.take();
             self.0.journal.write(pid, &p.info.plugin, "core", "kill");
@@ -120,15 +131,26 @@ impl Host {
     }
     /// 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);
-            }
+        {
+            let _state = self.0.state.lock().unwrap();
+            self.0
+                .shutdown
+                .store(true, std::sync::atomic::Ordering::SeqCst);
         }
-        for mut n in waits {
-            let _ = n.wait_for(|ended| *ended).await;
+        loop {
+            let ids: Vec<_> = self.processes().into_keys().collect();
+            if ids.is_empty() {
+                break;
+            }
+            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

@@ -93,20 +93,31 @@ impl Host {
             {
                 return Err(error(-32601, "invalid plugin name"));
             }
-            let admission = host.0.admission.read().await;
-            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(
-                        wasm.unwrap_or_else(|| host.0.world.plugin_dir(plugin).join("plugin.wasm")),
-                        host.0.runtime.clone(),
+            let path = wasm.unwrap_or_else(|| host.0.world.plugin_dir(plugin).join("plugin.wasm"));
+            let (native, component, version, admission) = loop {
+                if host.0.shutdown.load(std::sync::atomic::Ordering::SeqCst) {
+                    return Err(error(-32600, "host is stopping"));
+                }
+                let native = host.0.state.lock().unwrap().native.get(plugin).cloned();
+                let loaded = if native.is_none() {
+                    Some(
+                        host.0
+                            .components
+                            .load(path.clone(), host.0.runtime.clone())
+                            .await?,
                     )
-                    .await?;
-                (Some(component), hash)
+                } else {
+                    None
+                };
+                let admission = host.0.admission.read().await;
+                if let Some(loaded) = loaded {
+                    if crate::components::Fingerprint::read(&path).await? != loaded.fingerprint {
+                        drop(admission);
+                        continue;
+                    }
+                    break (native, Some(loaded.prepared), loaded.hash, admission);
+                }
+                break (native, None, "native".into(), admission);
             };
             let pid = format!("{plugin}-{}", Uuid::now_v7());
             let (tx, rx) = oneshot::channel();
@@ -126,6 +137,9 @@ impl Host {
             };
             {
                 let mut state = host.0.state.lock().unwrap();
+                if host.0.shutdown.load(std::sync::atomic::Ordering::SeqCst) {
+                    return Err(error(-32600, "host is stopping"));
+                }
                 if caller != "console"
                     && caller != "core"
                     && !state
@@ -149,7 +163,7 @@ impl Host {
                         abort: None,
                         caller,
                         id: id.clone(),
-                        answer: Some(tx),
+                        answer: if id.is_some() { Some(tx) } else { None },
                         ended: watch::channel(false).0,
                     },
                 );
@@ -220,11 +234,18 @@ impl Host {
             return;
         }
         if let Some(method) = v["method"].as_str() {
-            if method == "core.notify" {
+            if method == "core.notify" || method == "core.unlock" {
                 // Deliver progress before a following response removes its destination.
                 let params = v.get("params").cloned().unwrap_or(json!({}));
                 let result = if !params.is_object() {
                     Err(error(-32602, "params must be an object"))
+                } else if method == "core.unlock" {
+                    if let Some(path) = params["path"].as_str() {
+                        self.unlock(pid, path);
+                        Ok(Value::Null)
+                    } else {
+                        Err(error(-32602, "path must be a string"))
+                    }
                 } else if let Some(name) = params["name"].as_str() {
                     self.notify(
                         pid,
@@ -277,7 +298,7 @@ impl Host {
         } else if v["id"] == pid {
             self.answer(
                 pid,
-                if let Some(e) = v.get("error") {
+                if let Some(e) = v.get("error").filter(|e| !e.is_null()) {
                     Err(e.clone())
                 } else {
                     Ok(v["result"].clone())
@@ -287,8 +308,39 @@ impl Host {
     }
     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);
+        if let Some(plugin) = self
+            .0
+            .state
+            .lock()
+            .unwrap()
+            .processes
+            .get(pid)
+            .map(|p| p.info.plugin.clone())
+        {
+            self.0.journal.warning(pid, &plugin);
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    /// PROTOCOL 6: an unlock from stdout completes before output returns.
+    #[tokio::test]
+    async fn stdout_unlock_is_synchronous() {
+        let folder = std::env::temp_dir().join(format!("sausage-unlock-{}", Uuid::now_v7()));
+        crate::world::World::create(&folder).unwrap();
+        let host = Host::start_with_cache(&folder, &folder.join("cache"))
+            .await
+            .unwrap();
+        host.lock("core", "ordered", true).await.unwrap();
+        host.output(
+            "core",
+            r#"{"jsonrpc":"2.0","method":"core.unlock","params":{"path":"ordered"}}"#.into(),
+        );
+        let held = host.lock("console", "ordered", false).await.unwrap();
+        host.stop().await;
+        std::fs::remove_dir_all(folder).unwrap();
+        assert_eq!(held, json!({"held":true}));
+    }
+}

cli/src/runtime.rs

@@ -15,7 +15,10 @@ use wasmtime::component::{Component, Linker, ResourceTable};
 use wasmtime::{Config, Engine, Store};
 use wasmtime_wasi::cli::{AsyncStdinStream, AsyncStdoutStream};
 
-use wasmtime_wasi::p2::bindings::Command;
+use wasmtime_wasi::p2::bindings::CommandPre;
+
+/// A component with its imports resolved once for this runtime.
+pub(crate) struct Prepared(Result<CommandPre<ProcessState>, String>);
 
 use wasmtime_wasi::{I32Exit, WasiCtx, WasiCtxView, WasiView};
 use wasmtime_wasi_http::{WasiHttpCtx, WasiHttpCtxView, WasiHttpHooks, WasiHttpView};
@@ -130,17 +133,27 @@ pub struct Instance {
     pub stdout: mpsc::Receiver<String>,
     /// The stderr lines that the process writes.
     pub stderr: mpsc::Receiver<String>,
+    stderr_task: Option<JoinHandle<()>>,
     task: Option<JoinHandle<Exit>>,
     store_ended: tokio::sync::watch::Sender<bool>,
 }
 
 impl Drop for Instance {
     fn drop(&mut self) {
-        if let Some(task) = self.task.take() {
+        let task = self.task.take();
+        let stderr = self.stderr_task.take();
+        if let Some(task) = &task {
             task.abort();
+        }
+        if task.is_some() || stderr.is_some() {
             let ended = self.store_ended.clone();
             tokio::spawn(async move {
-                let _ = task.await;
+                if let Some(task) = task {
+                    let _ = task.await;
+                }
+                if let Some(stderr) = stderr {
+                    let _ = stderr.await;
+                }
                 ended.send_replace(true);
             });
         }
@@ -179,6 +192,10 @@ impl Instance {
     pub async fn wait(&mut self) -> Exit {
         let result = self.task.as_mut().unwrap().await;
         self.task.take();
+        if let Some(stderr) = self.stderr_task.as_mut() {
+            let _ = stderr.await;
+        }
+        self.stderr_task.take();
         self.store_ended.send_replace(true);
         match result {
             Ok(exit) => exit,
@@ -238,6 +255,28 @@ impl Runtime {
         Ok(Component::new(&self.engine, bytes)?)
     }
 
+    pub(crate) fn prepare(&self, component: &Component) -> Prepared {
+        // Preserve link failures as process exits, rather than load errors.
+        Prepared(
+            self.linker
+                .instantiate_pre(component)
+                .and_then(CommandPre::new)
+                .map_err(|error| error.to_string()),
+        )
+    }
+
+    pub(crate) async fn start_logged(
+        self: &Arc<Self>,
+        prepared: &Prepared,
+        cfg: InstanceConfig,
+        journal: Arc<crate::journal::Journal>,
+        pid: String,
+        plugin: String,
+    ) -> Result<Instance> {
+        self.start_inner(prepared.0.clone(), cfg, &[], Some((journal, pid, plugin)))
+            .await
+    }
+
     /// Start a WASI task with the supplied virtual namespace.
     pub async fn start(
         self: &Arc<Self>,
@@ -253,6 +292,17 @@ impl Runtime {
         component: &Component,
         cfg: InstanceConfig,
         args: &[String],
+    ) -> Result<Instance> {
+        self.start_inner(self.prepare(component).0, cfg, args, None)
+            .await
+    }
+
+    async fn start_inner(
+        self: &Arc<Self>,
+        prepared: Result<CommandPre<ProcessState>, String>,
+        cfg: InstanceConfig,
+        args: &[String],
+        log: Option<(Arc<crate::journal::Journal>, String, String)>,
     ) -> Result<Instance> {
         let (host_in, guest_in) = tokio::io::duplex(64 * 1024);
         let (guest_out, host_out) = tokio::io::duplex(64 * 1024);
@@ -299,22 +349,34 @@ impl Runtime {
         });
 
         let (err_tx, err_rx) = mpsc::channel(1024);
-        tokio::spawn(async move {
-            let mut lines = BufReader::new(host_err).lines();
-            while let Ok(Some(line)) = lines.next_line().await {
-                if err_tx.send(line).await.is_err() {
+        let stderr_task = tokio::spawn(async move {
+            let mut reader = BufReader::new(host_err);
+            let mut bytes = Vec::new();
+            while let Ok(n) = reader.read_until(b'\n', &mut bytes).await {
+                if n == 0 {
+                    break;
+                }
+                let line = String::from_utf8_lossy(&bytes)
+                    .trim_end_matches(['\r', '\n'])
+                    .to_string();
+                bytes.clear();
+                if let Some((journal, pid, plugin)) = &log {
+                    journal.write(pid, plugin, "stderr", &line);
+                } else if err_tx.send(line).await.is_err() {
                     break;
                 }
             }
         });
         let rt = self.clone();
-        let component = component.clone();
         cfg.store_ended.send_replace(false);
         let task = tokio::spawn(async move {
             let mut store = Store::new(&rt.engine, state);
             store.epoch_deadline_async_yield_and_update(1);
             let result = async {
-                let cmd = Command::instantiate_async(&mut store, &component, &rt.linker).await?;
+                let cmd = prepared
+                    .map_err(wasmtime::Error::msg)?
+                    .instantiate_async(&mut store)
+                    .await?;
                 cmd.wasi_cli_run().call_run(&mut store).await
             }
             .await;
@@ -336,6 +398,7 @@ impl Runtime {
             stdin: Some(host_in),
             stdout: line_rx,
             stderr: err_rx,
+            stderr_task: Some(stderr_task),
             task: Some(task),
             store_ended: cfg.store_ended,
         })

cli/src/services.rs

@@ -6,6 +6,7 @@ pub(crate) struct Service {
     pub(crate) method: String,
     pub(crate) params: Value,
     pub(crate) request: Value,
+    pub(crate) starting: std::sync::Arc<std::sync::atomic::AtomicBool>,
 }
 impl Host {
     /// Starts discovery in the background, then waits for each boot service in order.
@@ -25,11 +26,16 @@ impl Host {
     }
     /// Records a unique request identity before the service can answer or exit.
     pub(crate) async fn start_service(&self, method: String, params: Value) {
+        if self.0.shutdown.load(std::sync::atomic::Ordering::SeqCst) {
+            return;
+        }
         let request = json!(uuid::Uuid::now_v7().to_string());
+        let starting = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true));
         self.0.state.lock().unwrap().services.push(Service {
             method: method.clone(),
             params: params.clone(),
             request: request.clone(),
+            starting: starting.clone(),
         });
         self.0
             .journal
@@ -42,5 +48,6 @@ impl Host {
                 .journal
                 .write("core", "core", "core", &format!("service failed {error}"));
         }
+        starting.store(false, std::sync::atomic::Ordering::SeqCst);
     }
 }

cli/src/upgrades.rs

@@ -1,7 +1,7 @@
 //! Grace periods, service restarts, and retirement of replaced versions (PROTOCOL.md section 3.7).
 use crate::{Host, services::Service};
 use serde_json::json;
-use std::{collections::BTreeMap, time::Duration};
+use std::{collections::BTreeMap, sync::atomic::Ordering, time::Duration};
 use tokio::sync::watch;
 
 pub(crate) struct OldProcess {
@@ -29,7 +29,11 @@ impl Host {
         let processes = state
             .processes
             .iter()
-            .filter(|(_, process)| process.info.plugin == plugin && process.info.version != current)
+            .filter(|(_, process)| {
+                process.info.plugin == plugin
+                    && process.info.version != current
+                    && process.info.method != "rpc.discover"
+            })
             .map(|(pid, process)| OldProcess {
                 pid: pid.clone(),
                 version: process.info.version.clone(),
@@ -44,12 +48,17 @@ impl Host {
                 .split_once('.')
                 .is_some_and(|(name, _)| name == plugin)
             {
-                let ended = state
+                let process = state
                     .processes
                     .values()
-                    .find(|process| process.id.as_ref() == Some(&service.request))
-                    .map(|process| process.ended.subscribe());
-                services.push((service, ended));
+                    .find(|process| process.id.as_ref() == Some(&service.request));
+                if process.is_some_and(|process| process.info.version == current)
+                    || (process.is_none() && service.starting.load(Ordering::SeqCst))
+                {
+                    retained.push(service);
+                } else {
+                    services.push((service, process.map(|process| process.ended.subscribe())));
+                }
             } else {
                 retained.push(service);
             }
@@ -67,6 +76,9 @@ impl Host {
     pub(crate) fn begin_upgrade(&self, upgrade: Upgrade) {
         let host = self.clone();
         tokio::spawn(async move {
+            if host.0.shutdown.load(Ordering::SeqCst) {
+                return;
+            }
             host.0.journal.write(
                 "core",
                 &upgrade.plugin,
@@ -96,7 +108,9 @@ impl Host {
                     if let Some(mut ended) = ended {
                         let _ = ended.wait_for(|ended| *ended).await;
                     }
-                    host.start_service(service.method, service.params).await;
+                    if !host.0.shutdown.load(Ordering::SeqCst) {
+                        host.start_service(service.method, service.params).await;
+                    }
                 });
             }
             for (version, processes) in versions {
@@ -116,6 +130,32 @@ impl Host {
                     for mut ended in processes {
                         let _ = ended.wait_for(|ended| *ended).await;
                     }
+                    loop {
+                        if host.0.shutdown.load(Ordering::SeqCst) {
+                            host.0.retirements.lock().unwrap().remove(&retirement);
+                            return;
+                        }
+                        let live: Vec<_> = host
+                            .0
+                            .state
+                            .lock()
+                            .unwrap()
+                            .processes
+                            .values()
+                            .filter(|process| {
+                                process.info.plugin == plugin
+                                    && process.info.version == version
+                                    && process.info.method != "rpc.discover"
+                            })
+                            .map(|process| process.ended.subscribe())
+                            .collect();
+                        if live.is_empty() {
+                            break;
+                        }
+                        for mut ended in live {
+                            let _ = ended.wait_for(|ended| *ended).await;
+                        }
+                    }
                     if host
                         .0
                         .graph

cli/src/watcher.rs

@@ -1,24 +1,42 @@
-//! Background discovery and file notifications for installed plugin versions.
+//! Background discovery and nonrecursive notifications for installed plugin versions.
 use crate::Host;
 use notify::{RecursiveMode, Watcher};
 impl Host {
-    /// Starts background discovery and retains the watcher for this host's lifetime.
+    /// Watches only the plugin root and its immediate folders. Discovery survives watcher failure.
     pub(crate) fn watch_plugins(&self) -> anyhow::Result<()> {
+        let root = self.0.world.root.join("plugins");
+        let watched_root = root.clone();
         let (events, mut incoming) = tokio::sync::mpsc::unbounded_channel();
-        let mut watcher =
-            notify::recommended_watcher(move |event: notify::Result<notify::Event>| {
-                if let Ok(event) = event
-                    && !matches!(event.kind, notify::EventKind::Access(_))
-                    && event
-                        .paths
-                        .iter()
-                        .any(|path| path.file_name().is_some_and(|name| name == "plugin.wasm"))
-                {
-                    let _ = events.send(());
+        let setup = (|| -> anyhow::Result<_> {
+            let mut watcher =
+                notify::recommended_watcher(move |event: notify::Result<notify::Event>| {
+                    if let Ok(event) = event
+                        && !matches!(event.kind, notify::EventKind::Access(_))
+                        && event.paths.iter().any(|path| {
+                            path.parent() == Some(&watched_root)
+                                || path.file_name().is_some_and(|name| name == "plugin.wasm")
+                        })
+                    {
+                        let _ = events.send(());
+                    }
+                })?;
+            watcher.watch(&root, RecursiveMode::NonRecursive)?;
+            for entry in std::fs::read_dir(&root)? {
+                let entry = entry?;
+                if entry.file_type()?.is_dir() {
+                    watcher.watch(&entry.path(), RecursiveMode::NonRecursive)?;
                 }
-            })?;
-        watcher.watch(&self.0.world.root.join("plugins"), RecursiveMode::Recursive)?;
-        *self.0.watcher.lock().unwrap() = Some(watcher);
+            }
+            Ok(watcher)
+        })();
+        match setup {
+            Ok(watcher) => *self.0.watcher.lock().unwrap() = Some(watcher),
+            Err(error) => {
+                self.0
+                    .journal
+                    .write("core", "core", "core", &format!("watcher failed {error}"))
+            }
+        }
         let weak = std::sync::Arc::downgrade(&self.0);
         tokio::spawn(async move {
             loop {
@@ -26,6 +44,24 @@ impl Host {
                     break;
                 };
                 let host = Host(inner);
+                // Root events also cover plugin folders created after startup.
+                if let Some(watcher) = host.0.watcher.lock().unwrap().as_mut()
+                    && let Ok(entries) = std::fs::read_dir(&root)
+                {
+                    for entry in entries.flatten() {
+                        if entry.file_type().is_ok_and(|kind| kind.is_dir())
+                            && let Err(error) =
+                                watcher.watch(&entry.path(), RecursiveMode::NonRecursive)
+                        {
+                            host.0.journal.write(
+                                "core",
+                                "core",
+                                "core",
+                                &format!("watcher failed {error}"),
+                            );
+                        }
+                    }
+                }
                 let _activation = host.0.activations.lock().await;
                 if let Err(error) = host.refresh_graph().await {
                     host.0.journal.write(

cli/tests/activation.rs

@@ -7,7 +7,7 @@ use std::{
     time::Duration,
 };
 
-/// Builds two genuine guest versions in this executor's target folder.
+/// Builds genuine guest versions in this executor's target folder.
 fn fixture(version: usize) -> PathBuf {
     static FILES: OnceLock<Vec<PathBuf>> = OnceLock::new();
     FILES.get_or_init(|| {
@@ -16,7 +16,7 @@ fn fixture(version: usize) -> PathBuf {
             "target/compat/activation-guests/{}",
             std::process::id()
         ));
-        (1..=2)
+        (1..=3)
             .map(|version| {
                 let status = std::process::Command::new("cargo")
                     .env("FIXTURE_VERSION", version.to_string())
@@ -132,6 +132,302 @@ async fn graph(host: &Host) -> Value {
     call(host, "notes.read", json!({"path":"/proc/core/graph.json"})).await
 }
 
+/// H1/H2: broken providers and stray files do not block repair, revert, or force.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn review_repairs_with_broken_entries() {
+    let world = TestWorld::new();
+    world.plugin("notes", document("add"), document("add"));
+    world.plugin("broken", document("bad"), document("bad"));
+    std::fs::write(world.0.join("plugins/broken/plugin.wasm"), b"broken").unwrap();
+    let host = world.start().await;
+    activate(&host, "notes").await.unwrap();
+    assert!(!graph(&host).await["plugins"]["broken"]["error"].is_null());
+    std::fs::copy(fixture(1), world.0.join("plugins/broken/plugin.prev.wasm")).unwrap();
+    call(&host, "core.revert", json!({"plugin":"broken"})).await;
+    std::fs::write(world.0.join("plugins/broken/plugin.wasm"), b"broken again").unwrap();
+    activate(&host, "broken").await.unwrap();
+    std::fs::write(world.0.join("plugins/broken/plugin.wasm"), b"broken again").unwrap();
+    std::fs::write(
+        world.0.join("state/notes/2.json"),
+        consumer("broken", document("bad")).to_string(),
+    )
+    .unwrap();
+    call(
+        &host,
+        "core.activate",
+        json!({"plugin":"notes","wasm":"/state/notes/next.wasm","force":true}),
+    )
+    .await;
+    assert_eq!(
+        graph(&host).await["plugins"]["notes"]["imports"]["broken"]["status"],
+        "unavailable"
+    );
+    host.stop().await;
+}
+/// H2: a stray file under plugins does not interrupt discovery or activation.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn review_stray_file() {
+    let world = TestWorld::new();
+    world.plugin("notes", document("add"), document("add"));
+    std::fs::write(world.0.join("plugins/notes.md"), "a note").unwrap();
+    let host = world.start().await;
+    activate(&host, "notes").await.unwrap();
+    assert!(graph(&host).await["plugins"]["notes"]["error"].is_null());
+    host.stop().await;
+}
+/// M4: an answered discovery receives EOF instead of leaking a detached process.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn review_discovery_answer_stops() {
+    let world = TestWorld::new();
+    let mut next = document("add");
+    next["x-test-linger"] = json!(true);
+    world.plugin("notes", document("add"), next);
+    let host = world.start().await;
+    activate(&host, "notes").await.unwrap();
+    marker(world.0.join("state/notes/discovery-ended-2"), "ended").await;
+    host.stop().await;
+}
+/// M4: a rejected candidate's discovery process never becomes a retired installed version.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn review_candidate_is_not_old_version() {
+    use sha2::{Digest, Sha256};
+    let world = TestWorld::new();
+    let mut rejected = consumer("missing", document("add"));
+    rejected["x-test-linger"] = json!(true);
+    rejected["x-test-stubborn"] = json!(true);
+    world.plugin("notes", document("add"), rejected);
+    std::fs::write(
+        world.0.join("state/notes/3.json"),
+        document("add").to_string(),
+    )
+    .unwrap();
+    std::fs::copy(fixture(3), world.0.join("state/notes/third.wasm")).unwrap();
+    let host = world.start().await;
+    assert!(activate(&host, "notes").await.is_err());
+    let candidate = host
+        .processes()
+        .into_iter()
+        .find(|(_, process)| process.method == "rpc.discover")
+        .unwrap()
+        .0;
+    call(
+        &host,
+        "core.activate",
+        json!({"plugin":"notes","wasm":"/state/notes/third.wasm"}),
+    )
+    .await;
+    call(&host, "core.kill", json!({"pid":candidate})).await;
+    let old = format!("{:x}", Sha256::digest(std::fs::read(fixture(1)).unwrap()));
+    marker(world.0.join("state/notes/retired"), &old).await;
+    let rejected_hash = format!("{:x}", Sha256::digest(std::fs::read(fixture(2)).unwrap()));
+    assert!(
+        tokio::time::timeout(
+            Duration::from_millis(300),
+            marker(world.0.join("state/notes/retired"), &rejected_hash)
+        )
+        .await
+        .is_err()
+    );
+    host.stop().await;
+}
+/// M6: shutdown suppresses service restarts and retirement hooks from pending upgrades.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn review_shutdown_suppresses_hooks() {
+    let world = TestWorld::new();
+    let mut old = document("serve");
+    old["x-sausage-grace"] = json!(60);
+    world.plugin("notes", old, document("serve"));
+    std::fs::write(
+        world.0.join("config/boot.json"),
+        json!([{"method":"notes.serve","params":{}}]).to_string(),
+    )
+    .unwrap();
+    let host = world.start().await;
+    marker(world.0.join("state/notes/service"), "1").await;
+    activate(&host, "notes").await.unwrap();
+    host.stop().await;
+    for _ in 0..100 {
+        tokio::task::yield_now().await;
+    }
+    assert!(!world.0.join("state/notes/retired").exists());
+    assert_eq!(
+        std::fs::read_to_string(world.0.join("state/notes/service")).unwrap(),
+        "1"
+    );
+    assert!(
+        call(&host, "core.log", json!({"grep":"retired start"}))
+            .await
+            .as_array()
+            .unwrap()
+            .is_empty()
+    );
+    assert_eq!(
+        call(&host, "core.log", json!({"grep":"service start"}))
+            .await
+            .as_array()
+            .unwrap()
+            .len(),
+        1
+    );
+}
+/// L15: retirement waits for new admissions of a reverted version, not just its first snapshot.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn review_retirement_checks_live_version() {
+    let world = TestWorld::new();
+    let mut old = document("hold");
+    old["x-sausage-grace"] = json!(60);
+    world.plugin("notes", old.clone(), old);
+    let host = world.start().await;
+    call(&host, "notes.hold", json!({})).await;
+    let first = host
+        .processes()
+        .into_iter()
+        .find(|(_, p)| p.method == "hold")
+        .unwrap()
+        .0;
+    activate(&host, "notes").await.unwrap();
+    call(&host, "core.revert", json!({"plugin":"notes"})).await;
+    call(&host, "notes.hold", json!({})).await;
+    let second = host
+        .processes()
+        .into_iter()
+        .find(|(pid, p)| p.method == "hold" && *pid != first)
+        .unwrap()
+        .0;
+    activate(&host, "notes").await.unwrap();
+    // The v2 retirement from the revert is unrelated to the pending v1 retirement.
+    let previous = {
+        use sha2::{Digest, Sha256};
+        format!("{:x}", Sha256::digest(std::fs::read(fixture(1)).unwrap()))
+    };
+    call(&host, "core.stop", json!({"pid":first})).await;
+    assert!(
+        tokio::time::timeout(
+            Duration::from_millis(300),
+            marker(world.0.join("state/notes/retired"), &previous)
+        )
+        .await
+        .is_err()
+    );
+    call(&host, "core.stop", json!({"pid":second})).await;
+    marker(world.0.join("state/notes/retired"), &previous).await;
+    host.stop().await;
+}
+/// L13: source subtrees require no watches, even when the host cannot read them.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn review_watcher_is_shallow() {
+    use std::os::unix::fs::PermissionsExt;
+    let world = TestWorld::new();
+    world.plugin("notes", document("add"), document("add"));
+    let source = world.0.join("plugins/notes/source");
+    std::fs::create_dir_all(&source).unwrap();
+    std::fs::set_permissions(&source, std::fs::Permissions::from_mode(0o0)).unwrap();
+    let result = Host::start(&world.0).await;
+    std::fs::set_permissions(&source, std::fs::Permissions::from_mode(0o700)).unwrap();
+    let host = result.unwrap();
+    activate(&host, "notes").await.unwrap();
+    host.stop().await;
+}
+/// L13: a watcher startup error reaches the journal without preventing host startup.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn review_watcher_failure_is_nonfatal() {
+    use std::os::unix::fs::PermissionsExt;
+    let world = TestWorld::new();
+    world.plugin("notes", document("add"), document("add"));
+    let inaccessible = world.0.join("plugins/inaccessible");
+    std::fs::create_dir_all(&inaccessible).unwrap();
+    std::fs::set_permissions(&inaccessible, std::fs::Permissions::from_mode(0o0)).unwrap();
+    let result = Host::start(&world.0).await;
+    std::fs::set_permissions(&inaccessible, std::fs::Permissions::from_mode(0o700)).unwrap();
+    let host = result.unwrap();
+    assert!(
+        !call(&host, "core.log", json!({"grep":"watcher failed"}))
+            .await
+            .as_array()
+            .unwrap()
+            .is_empty()
+    );
+    activate(&host, "notes").await.unwrap();
+    host.stop().await;
+}
+/// L14: rapid service upgrades never overlap singleton instances.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn review_singleton_restart() {
+    let world = TestWorld::new();
+    world.plugin("notes", document("serve"), document("serve"));
+    std::fs::write(
+        world.0.join("config/boot.json"),
+        json!([{"method":"notes.serve","params":{}}]).to_string(),
+    )
+    .unwrap();
+    let host = world.start().await;
+    for _ in 0..128 {
+        activate(&host, "notes").await.unwrap();
+        call(&host, "core.revert", json!({"plugin":"notes"})).await;
+    }
+    host.stop().await;
+    assert!(!world.0.join("state/notes/duplicate-service").exists());
+}
+/// H2: a plugin read error belongs to that entry, not the entire graph.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn review_read_error_is_local() {
+    let world = TestWorld::new();
+    world.plugin("notes", document("add"), document("add"));
+    std::fs::create_dir_all(world.0.join("plugins/broken/plugin.wasm")).unwrap();
+    let host = world.start().await;
+    activate(&host, "notes").await.unwrap();
+    assert!(!graph(&host).await["plugins"]["broken"]["error"].is_null());
+    host.stop().await;
+}
+/// L3: cancellation of the caller cannot cancel an activation body.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn review_dropped_activation_completes() {
+    let world = TestWorld::new();
+    let mut next = document("add");
+    next["x-test-wait"] = json!(true);
+    world.plugin("notes", document("add"), next);
+    let host = world.start().await;
+    let caller = host.clone();
+    let task = tokio::spawn(async move { activate(&caller, "notes").await });
+    marker(world.0.join("state/notes/discover-started"), "ready").await;
+    task.abort();
+    let _ = task.await;
+    std::fs::write(world.0.join("state/notes/discover-release"), "ready").unwrap();
+    tokio::time::timeout(Duration::from_secs(30), async {
+        while std::fs::read(world.0.join("plugins/notes/plugin.wasm")).unwrap()
+            != std::fs::read(fixture(2)).unwrap()
+        {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+    host.stop().await;
+}
+/// M4: discovery times out and its process stops after either an answer or a timeout.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn review_discovery_deadline_and_stop() {
+    let world = TestWorld::new();
+    let mut next = document("add");
+    next["x-test-wait"] = json!(true);
+    world.plugin("notes", document("add"), next);
+    let host = world.start().await;
+    let result = tokio::time::timeout(Duration::from_secs(15), activate(&host, "notes")).await;
+    assert!(
+        result.unwrap().unwrap_err()["message"]
+            .as_str()
+            .unwrap()
+            .contains("discovery")
+    );
+    assert!(
+        !host
+            .processes()
+            .values()
+            .any(|process| process.method == "rpc.discover" && process.state == "running")
+    );
+    host.stop().await;
+}
+
 /// PROTOCOL 10.4: proc graph files contain valid documents before discovery completes.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn graph_is_valid_immediately() {
@@ -321,7 +617,7 @@ async fn serializes_activations() {
     second.unwrap();
     assert_eq!(
         std::fs::read(world.0.join("plugins/notes/plugin.prev.wasm")).unwrap(),
-        std::fs::read(fixture(2)).unwrap()
+        std::fs::read(fixture(1)).unwrap()
     );
     host.stop().await;
 }

cli/tests/cli.rs

@@ -0,0 +1,66 @@
+//! Initialize the embedded agent and start the resulting world through the CLI.
+use sausage::Host;
+use serde_json::{Value, json};
+use std::process::{Command, Stdio};
+
+/// Environment configuration creates a complete world with a callable agent.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn init_builds_agent() {
+    let temporary = std::env::temp_dir().join(format!("sausage-cli-{}", uuid::Uuid::now_v7()));
+    std::fs::create_dir_all(&temporary).unwrap();
+    let world = temporary.join("world");
+    let output = Command::new(env!("CARGO_BIN_EXE_sausage"))
+        .arg("init")
+        .arg(&world)
+        .env("SAUSAGE_API", "anthropic-messages")
+        .env("SAUSAGE_BASE_URL", "http://127.0.0.1:1/")
+        .env("SAUSAGE_MODEL", "init-test")
+        .env("SAUSAGE_API_TOKEN", "fake-token")
+        .stdin(Stdio::null())
+        .output()
+        .unwrap();
+    assert!(
+        output.status.success(),
+        "stdout: {}\nstderr: {}",
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr)
+    );
+    let stdout = String::from_utf8(output.stdout).unwrap();
+    assert!(stdout.contains("first agent build takes a few minutes"));
+    assert!(stdout.contains("sausage run"));
+    let config: Value =
+        serde_json::from_slice(&std::fs::read(world.join("config/agent.json")).unwrap()).unwrap();
+    assert_eq!(
+        config,
+        json!({"api":"anthropic-messages","base_url":"http://127.0.0.1:1","model":"init-test","token":"fake-token"})
+    );
+    assert_eq!(
+        serde_json::from_slice::<Value>(&std::fs::read(world.join("config/boot.json")).unwrap())
+            .unwrap(),
+        json!([])
+    );
+    for file in [
+        "plugins/cargo/sdk/Cargo.toml",
+        "plugins/cargo/sdk/src/lib.rs",
+        "plugins/agent/source/Cargo.toml",
+        "plugins/agent/source/src/main.rs",
+        "plugins/agent/plugin.wasm",
+    ] {
+        assert!(world.join(file).is_file(), "missing {file}");
+    }
+    let cache =
+        std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../target/sdk/cli-cache");
+    let host = Host::start_with_cache(&world, &cache).await.unwrap();
+    let doc = host.call("agent.rpc.discover", json!({})).await.unwrap();
+    assert_eq!(doc["x-sausage-grace"], 600);
+    assert!(
+        doc["methods"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .any(|method| method["name"] == "chat")
+    );
+    host.stop().await;
+
+    std::fs::remove_dir_all(temporary).unwrap();
+}

cli/tests/console.rs

@@ -44,7 +44,10 @@ impl TestWorld {
         }
         std::fs::create_dir_all(world.state_dir("probe")).unwrap();
         std::fs::write(world.state_dir("probe").join("discover.json"),json!({"methods":[{"name":"echo","description":"Echo supplied text.","params":[{"name":"text","description":"The input text.","schema":{"type":"string"}}],"result":{"schema":{"type":"string"}}}]}).to_string()).unwrap();
-        std::fs::copy(fixture(), root.join("candidate.wasm")).unwrap();
+        let mut candidate = std::fs::read(fixture()).unwrap();
+        // A custom section creates a new version without changing probe behavior.
+        candidate.extend_from_slice(&[0, 2, 1, b'c']);
+        std::fs::write(root.join("candidate.wasm"), candidate).unwrap();
         Self(root)
     }
     async fn start(&self) -> Host {
@@ -152,7 +155,7 @@ async fn ps() {
     assert!(text.contains("probe serve console"));
     host.stop().await;
 }
-/// Log supports plugin filters, process filters, and a row count.
+/// Log formats local times and short pids, while core.log preserves JSON and filters.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn log() {
     let world = TestWorld::new();
@@ -160,22 +163,58 @@ async fn log() {
     std::fs::copy(fixture(), world.0.join("plugins/data-logger/plugin.wasm")).unwrap();
     let host = world.start().await;
     host.call("data-logger.echo", json!({})).await.unwrap();
-    let filtered: Value =
-        serde_json::from_str(&command(&host, "log data-logger -n 1").await).unwrap();
-    assert_eq!(filtered[0]["plugin"], "data-logger");
     let pid = host.call("probe.serve", json!({})).await.unwrap();
-    for line in [
-        "log".to_owned(),
-        "log probe -n 1".to_owned(),
-        format!("log -n 1 {}", pid.as_str().unwrap()),
+    for (line, filters) in [
+        ("log".to_owned(), json!({})),
+        (
+            "log data-logger -n 1".to_owned(),
+            json!({"plugin":"data-logger","limit":1}),
+        ),
+        (
+            "log probe -n 1".to_owned(),
+            json!({"plugin":"probe","limit":1}),
+        ),
+        (
+            format!("log -n 1 {}", pid.as_str().unwrap()),
+            json!({"pid":pid,"limit":1}),
+        ),
     ] {
-        let entries: Value = serde_json::from_str(&command(&host, &line).await).unwrap();
-        assert!(!entries.as_array().unwrap().is_empty());
+        let raw = command(&host, &format!("core.log {filters}")).await;
+        let entries: Value = serde_json::from_str(&raw).unwrap();
+        let entries = entries.as_array().unwrap();
+        assert!(!entries.is_empty());
+        let printed = command(&host, &line).await;
+        let lines: Vec<_> = printed.lines().collect();
+        assert_eq!(lines.len(), entries.len(), "{printed}");
         if line != "log" {
-            assert_eq!(entries.as_array().unwrap().len(), 1);
-            assert_eq!(entries[0]["plugin"], "probe");
+            assert_eq!(entries.len(), 1);
+        }
+        for (line, entry) in lines.into_iter().zip(entries) {
+            let time = chrono::DateTime::parse_from_rfc3339(entry["time"].as_str().unwrap())
+                .unwrap()
+                .with_timezone(&chrono::Local)
+                .format("%H:%M:%S%.3f")
+                .to_string();
+            let plugin = entry["plugin"].as_str().unwrap();
+            let pid = entry["pid"].as_str().unwrap();
+            let prefix = format!("{plugin}-");
+            let short = pid
+                .strip_prefix(&prefix)
+                .map(|uuid| format!("{prefix}{}", &uuid[..8]))
+                .unwrap_or_else(|| pid.to_owned());
+            assert_eq!(
+                line,
+                format!(
+                    "{} {} {} {}",
+                    time,
+                    short,
+                    entry["stream"].as_str().unwrap(),
+                    entry["line"].as_str().unwrap()
+                )
+            );
         }
     }
+    assert_eq!(command(&host, "log absent-plugin").await, "");
     assert!(execute_line(&host, "log -n").await.is_err());
     host.stop().await;
 }

cli/tests/core.rs

@@ -53,6 +53,314 @@ async fn call(h: &Host, m: &str, p: Value) -> Value {
         .unwrap()
 }
 
+/// PROTOCOL 3.3: a closed stdin does not discard a response already on stdout.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn closed_stdin_drains_stdout() {
+    let w = TestWorld::new();
+    std::fs::write(w.0.join("state/early"), "answer").unwrap();
+    let h = w.start().await;
+    assert_eq!(
+        call(&h, "probe.echo", json!({"large":"x".repeat(2*1024*1024)})).await,
+        "early"
+    );
+    h.stop().await;
+}
+
+/// PROTOCOL 3.3: failed imports remain process failures, even with initial input in flight.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn failed_imports_preserve_process_status() {
+    let w = TestWorld::new();
+    let path = w.0.join("plugins/probe/plugin.wasm");
+    let mut bytes = std::fs::read(&path).unwrap();
+    for index in 0..bytes.len() - 5 {
+        if &bytes[index..index + 5] == b"wasi:" {
+            bytes[index..index + 5].copy_from_slice(b"fake:");
+        }
+    }
+    std::fs::write(path, bytes).unwrap();
+    let h = w.start().await;
+    let error = tokio::time::timeout(
+        std::time::Duration::from_secs(20),
+        h.call("probe.echo", json!({"large":"x".repeat(2*1024*1024)})),
+    )
+    .await
+    .unwrap()
+    .unwrap_err();
+    assert_eq!(error["code"], -32002);
+    assert!(error["data"]["status"].as_str().unwrap().contains("fake:"));
+    h.stop().await;
+}
+
+/// PROTOCOL 3.3: a closed stdin preserves the actual exit status and final stderr.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn closed_stdin_preserves_failure_details() {
+    let w = TestWorld::new();
+    std::fs::write(w.0.join("state/early"), "fail").unwrap();
+    let h = w.start().await;
+    let error = tokio::time::timeout(
+        std::time::Duration::from_secs(20),
+        h.call("probe.echo", json!({"large":"x".repeat(2*1024*1024)})),
+    )
+    .await
+    .unwrap()
+    .unwrap_err();
+    assert_eq!(error["code"], -32002);
+    // WASIp2 command.run represents a nonzero exit as its unit error result.
+    assert_eq!(error["data"]["status"], "exit 1");
+    assert!(
+        error["data"]["lines"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .any(|line| line["line"] == "last early log")
+    );
+    h.stop().await;
+}
+
+/// PROTOCOL 7: invalid stderr bytes do not stop later log lines or the guest.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn stderr_uses_lossy_utf8_and_continues() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    let pid = call(&h, "probe.invalid_stderr", json!({})).await;
+    call(&h, "core.stop", json!({"pid":pid})).await;
+    let logs = call(&h, "core.log", json!({"pid":pid})).await;
+    assert!(
+        logs.as_array()
+            .unwrap()
+            .iter()
+            .any(|line| line["line"] == "\u{fffd}")
+    );
+    assert!(
+        logs.as_array()
+            .unwrap()
+            .iter()
+            .any(|line| line["line"] == "after invalid UTF-8")
+    );
+    h.stop().await;
+}
+
+/// PROTOCOL 3.3 and 7: stderr drains even while a large initial stdin write blocks.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn stderr_flood_does_not_block_stdin() {
+    let w = TestWorld::new();
+    std::fs::write(w.0.join("state/early"), "flood").unwrap();
+    let h = w.start().await;
+    let payload = json!({"large":"x".repeat(1024*1024)});
+    assert_eq!(call(&h, "probe.echo", payload.clone()).await, payload);
+    h.stop().await;
+    let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
+    let count: i64 = db
+        .query_row(
+            "SELECT count(*) FROM journal WHERE line LIKE 'flood %'",
+            [],
+            |row| row.get(0),
+        )
+        .unwrap();
+    assert!(count >= 5000);
+}
+
+/// PROTOCOL 3.6: shutdown rejects new processes, including calls after stop returns.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn shutdown_rejects_new_processes() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    h.stop().await;
+    assert_eq!(
+        h.call("probe.echo", json!({})).await.unwrap_err()["code"],
+        -32600
+    );
+    assert!(h.processes().is_empty());
+}
+
+/// PROTOCOL 6: stdout unlock takes effect before the next result reaches its caller.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn stdout_unlock_precedes_result() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    for _ in 0..10 {
+        let pid = call(&h, "probe.unlock_answer", json!({})).await;
+        assert_eq!(
+            call(
+                &h,
+                "core.lock",
+                json!({"path":"ordered-unlock","wait":false})
+            )
+            .await,
+            json!({"held":true})
+        );
+        call(&h, "core.unlock", json!({"path":"ordered-unlock"})).await;
+        call(&h, "core.stop", json!({"pid":pid})).await;
+    }
+    h.stop().await;
+}
+
+/// PROTOCOL 3.5: a notification-started process has no call to receive progress.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn notification_process_has_no_progress_destination() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    assert_eq!(call(&h, "probe.notification_child", json!({})).await, 0);
+    h.stop().await;
+}
+
+/// PROTOCOL 3.1: an explicit null error does not replace a successful result.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn null_error_is_success() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    assert_eq!(call(&h, "probe.null_error", json!({})).await, 42);
+    h.stop().await;
+}
+
+/// PROTOCOL 7: a journal insertion failure does not poison the host or stop later writes.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn journal_recovers_after_sqlite_write_failure() {
+    let w = TestWorld::new();
+    let h = w.start().await;
+    let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
+    db.busy_timeout(std::time::Duration::from_secs(5)).unwrap();
+    db.execute_batch("CREATE TRIGGER reject_writes BEFORE INSERT ON journal BEGIN SELECT RAISE(FAIL,'test failure'); END;").unwrap();
+    assert_eq!(
+        call(&h, "probe.echo", json!({"failed_log":true})).await,
+        json!({"failed_log":true})
+    );
+    call(&h, "core.log", json!({})).await;
+    db.execute_batch("DROP TRIGGER reject_writes;").unwrap();
+    assert_eq!(
+        call(&h, "probe.echo", json!({"recovered":true})).await,
+        json!({"recovered":true})
+    );
+    let entries = call(&h, "core.log", json!({"grep":"recovered"})).await;
+    assert!(!entries.as_array().unwrap().is_empty());
+    h.stop().await;
+}
+
+/// PROTOCOL 4: trailing slashes preserve live proc folders and generated plugin folders.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn proc_directories_accept_trailing_slashes() {
+    use sausage_vfs::{Layer, Open};
+    let w = TestWorld::new();
+    let document = json!({"methods":[]});
+    std::fs::create_dir_all(w.0.join("state/probe")).unwrap();
+    std::fs::write(w.0.join("state/probe/discover.json"), document.to_string()).unwrap();
+    let h = w.start().await;
+    let view = h.filesystem().await.unwrap();
+    tokio::time::timeout(std::time::Duration::from_secs(20), async {
+        loop {
+            if view.list("/proc/core/plugins/probe").await.is_ok() {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+    let pid = call(&h, "probe.serve", json!({})).await;
+    for path in [
+        "/proc/".to_string(),
+        "/proc/core/".into(),
+        "/proc/core/plugins/".into(),
+        "/proc/core/plugins/probe/".into(),
+        format!("/proc/{}/", pid.as_str().unwrap()),
+    ] {
+        assert!(view.stat(&path, true).await.unwrap().directory);
+        view.list(&path).await.unwrap();
+        view.open(
+            &path,
+            Open {
+                directory: true,
+                ..Open::default()
+            },
+        )
+        .await
+        .unwrap();
+    }
+    h.stop().await;
+}
+
+/// PROTOCOL 3.4: Unix replacements with the same size and mtime use the new component.
+#[cfg(unix)]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn component_identity_includes_inode_and_ctime() {
+    let w = TestWorld::new();
+    let path = w.0.join("plugins/probe/plugin.wasm");
+    let mut bytes = std::fs::read(&path).unwrap();
+    bytes.extend_from_slice(&[0, 2, 1, b'a']);
+    std::fs::write(&path, &bytes).unwrap();
+    let modified = std::fs::metadata(&path).unwrap().modified().unwrap();
+    let h = w.start().await;
+    let old = call(&h, "probe.serve", json!({})).await;
+    *bytes.last_mut().unwrap() = b'b';
+    std::fs::write(&path, &bytes).unwrap();
+    std::fs::File::options()
+        .write(true)
+        .open(&path)
+        .unwrap()
+        .set_times(std::fs::FileTimes::new().set_modified(modified))
+        .unwrap();
+    let new = call(&h, "probe.serve", json!({})).await;
+    let table = h.processes();
+    assert_ne!(
+        table[old.as_str().unwrap()].version,
+        table[new.as_str().unwrap()].version
+    );
+    h.stop().await;
+}
+
+/// PROTOCOL 3.6 and 6: stopping cancels a native process's queued lock request immediately.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn stopping_removes_queued_lock_acquisitions() {
+    use sausage_vfs::{Layer, Open};
+    let w = TestWorld::new();
+    let h = w.start().await;
+    call(&h, "core.lock", json!({"path":"queued-stop"})).await;
+    h.register_native(
+        "queued",
+        std::sync::Arc::new(|ctx, _, _| {
+            Box::pin(async move {
+                ctx.notify("ready", json!({"pid":ctx.pid}));
+                ctx.call("core.lock", json!({"path":"queued-stop"})).await
+            })
+        }),
+    );
+    let mut events = h.subscribe();
+    let host = h.clone();
+    let pending = tokio::spawn(async move { host.call("queued.wait", json!({})).await });
+    let event = events.recv().await.unwrap();
+    let pid = event["params"]["params"]["pid"].clone();
+    let view = h.filesystem().await.unwrap();
+    tokio::time::timeout(std::time::Duration::from_secs(10), async {
+        loop {
+            let f = view
+                .open("/proc/core/locks.json", Open::default())
+                .await
+                .unwrap();
+            let locks: Value = serde_json::from_slice(&f.read(0, 65536).await.unwrap()).unwrap();
+            if locks["queued-stop"]["waiting"] == json!([pid]) {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+    tokio::time::timeout(
+        std::time::Duration::from_secs(1),
+        h.call("core.stop", json!({"pid":pid})),
+    )
+    .await
+    .unwrap()
+    .unwrap();
+    assert!(pending.await.unwrap().is_err());
+    call(&h, "core.unlock", json!({"path":"queued-stop"})).await;
+    assert_eq!(
+        call(&h, "core.lock", json!({"path":"queued-stop","wait":false})).await,
+        json!({"held":true})
+    );
+    h.stop().await;
+}
+
 /// PROTOCOL 3.1: json rpc lines carry results.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn json_rpc_lines_carry_results() {
@@ -608,7 +916,13 @@ async fn shutdown_stops_busy_processes_in_parallel_after_five_seconds() {
         }));
     }
     tokio::time::timeout(std::time::Duration::from_secs(20), async {
-        while h.processes().len() != 2 {
+        while h
+            .processes()
+            .values()
+            .filter(|p| p.plugin == "probe" && p.method == "busy")
+            .count()
+            != 2
+        {
             tokio::task::yield_now().await;
         }
     })

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

@@ -41,13 +41,40 @@ fn main() {
         "serve" | "hold" => {
             send(json!({"jsonrpc":"2.0","id":request["id"],"result":{"version":version}}));
             if request["method"] == "serve" {
+                let exclusive = format!("/run/{plugin}-service");
+                if std::fs::OpenOptions::new()
+                    .write(true)
+                    .create_new(true)
+                    .open(&exclusive)
+                    .is_err()
+                {
+                    std::fs::write(format!("/state/{plugin}/duplicate-service"), "duplicate")
+                        .unwrap();
+                }
                 std::fs::write(format!("/state/{plugin}/service"), version).unwrap();
             }
             for _ in lines {}
+            if request["method"] == "serve" {
+                let _ = std::fs::remove_file(format!("/run/{plugin}-service"));
+            }
             std::fs::write(format!("/state/{plugin}/ended-{version}"), "ended").unwrap();
             return;
         }
         _ => json!({"version":version}),
     };
     send(json!({"jsonrpc":"2.0","id":request["id"],"result":result}));
+    if request["method"] == "rpc.discover" && result["x-test-linger"] == true {
+        if result["x-test-stubborn"] == true {
+            while !std::path::Path::new(&format!("/state/{plugin}/discovery-exit")).exists() {
+                std::thread::sleep(std::time::Duration::from_millis(1));
+            }
+        } else {
+            for _ in lines {}
+        }
+        std::fs::write(
+            format!("/state/{plugin}/discovery-ended-{version}"),
+            "ended",
+        )
+        .unwrap();
+    }
 }

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

@@ -292,9 +292,59 @@ fn locked_write(lines: &mut Lines) -> Value {
     assert_eq!(std::fs::read_to_string("/state/loop").unwrap(), "new");
     json!({})
 }
+fn unlock_answer(request: &Value, lines: &mut Lines) {
+    lock(&json!({"path":"ordered-unlock"}), lines);
+    let unlock = json!({"jsonrpc":"2.0","method":"core.unlock","params":{"path":"ordered-unlock"}});
+    let answer =
+        json!({"jsonrpc":"2.0","id":request["id"],"result":std::env::var("SAUSAGE_PID").unwrap()});
+    let mut stdout = io::stdout().lock();
+    writeln!(stdout, "{unlock}\n{answer}").unwrap();
+    stdout.flush().unwrap();
+    drop(stdout);
+    for _ in lines {}
+}
+fn notification_child(lines: &mut Lines) -> Value {
+    send(json!({"jsonrpc":"2.0","method":"probe.orphan_notify","params":{}}));
+    let pid = loop {
+        if let Ok(pid) = std::fs::read_to_string("/state/orphan") {
+            break pid;
+        }
+        std::thread::sleep(std::time::Duration::from_millis(1));
+    };
+    send(json!({"jsonrpc":"2.0","id":7,"method":"core.stop","params":{"pid":pid}}));
+    let mut count = 0;
+    loop {
+        let response = receive(lines);
+        if response["method"] == "core.notify" {
+            count += 1;
+        }
+        if response["id"] == 7 {
+            break;
+        }
+    }
+    json!(count)
+}
 fn dispatch(request: &Value, lines: &mut Lines) -> Option<Value> {
     let params = &request["params"];
     let result = match request["method"].as_str().unwrap() {
+        "invalid_stderr" => {
+            io::stderr().write_all(&[255, 10]).unwrap();
+            eprintln!("after invalid UTF-8");
+            json!(std::env::var("SAUSAGE_PID").unwrap())
+        }
+        "unlock_answer" => {
+            unlock_answer(request, lines);
+            return None;
+        }
+        "null_error" => {
+            send(json!({"jsonrpc":"2.0","id":request["id"],"error":null,"result":42}));
+            return None;
+        }
+        "notification_child" => notification_child(lines),
+        "orphan_notify" => {
+            std::fs::write("/state/orphan", std::env::var("SAUSAGE_PID").unwrap()).unwrap();
+            notify()
+        }
         "self_serve" => {
             std::fs::write("/self/endpoint", "endpoint").unwrap();
             serve(request, lines);
@@ -359,6 +409,21 @@ fn dispatch(request: &Value, lines: &mut Lines) -> Option<Value> {
     Some(result)
 }
 fn main() {
+    if let Ok(mode) = std::fs::read_to_string("/state/early") {
+        if mode == "flood" {
+            for index in 0..5000 {
+                eprintln!("flood {index} {}", "x".repeat(128));
+            }
+        } else {
+            eprintln!("last early log");
+            if mode == "answer" {
+                send(
+                    json!({"jsonrpc":"2.0","id":std::env::var("SAUSAGE_PID").unwrap(),"result":"early"}),
+                );
+            }
+            std::process::exit(23);
+        }
+    }
     let mut lines = io::BufReader::new(io::stdin()).lines();
     let request = receive(&mut lines);
     if let Some(result) = dispatch(&request, &mut lines)