console: one sequential loop; host tests in one binary

The console reads a line, runs it to the end, and prints the result,
with and without a terminal. Notifications print while a command runs.
Ctrl-C during a command stops only the wait. 314 lines (was 480).

All host integration tests are modules of cli/tests/host/, so they run
in one parallel pool, and the probes build once. Warm, the 184 host
tests take 24 s (was about 45 s as seven binaries in sequence), and
cargo test --workspace takes 45 s.

AGENTS.md

@@ -5,7 +5,7 @@ Keep it that way.
 
 ## Priorities
 
-1. **Correct.** Every behavior in `PROTOCOL.md` has a test in `cli/tests/`.
+1. **Correct.** Every behavior in `PROTOCOL.md` has a test in `cli/tests/host/`.
 2. **Simple and short.** Prefer the obvious code. No abstraction for one
    caller. No configuration that nobody asked for.
 3. **Readable.** A new reader must understand a file without the others.
@@ -37,6 +37,7 @@ Keep it that way.
 ```sh
 nix flake check -L      # all tests below; CI runs this on every push
 nix develop -c cargo test --workspace     # fast with a warm cache: run it during development
+nix develop -c cargo test -p sausage --test host -- core::   # one topic of the host tests
 nix develop -c cargo build --release -p agent --target wasm32-wasip2
 nix develop -c cargo clippy --workspace
 ```

cli/src/console.rs

@@ -1,28 +1,22 @@
-//! Console commands, readable results, and line editing with live notifications.
+//! Sequential console commands, line editing, and live notifications.
 use crate::Host;
 use anyhow::Result;
 use rustyline::{DefaultEditor, ExternalPrinter, error::ReadlineError};
 use sausage_vfs::{Layer, Open};
 use serde_json::{Value, json};
-use std::time::{SystemTime, UNIX_EPOCH};
 
 const HELP: &str = "text                         chat with the agent
 plugin.method {json}          call a plugin
-ls [path]                    list a folder (default /)
+ls [path], cat <path>, methods <plugin>
-cat <path>                   read a file
+ps, log [plugin or pid] [-n N]
-methods <plugin>             show methods, params, types, and docs
+activate <plugin> <wasm>, revert <plugin>
-ps                           show pid, plugin, method, parent, state, age
+stop <pid>, kill <pid>, help, quit";
-log [plugin or pid] [-n N]    read the journal
-activate <plugin> <wasm>      activate a version
-revert <plugin>               restore the previous version
-stop <pid>                   request a graceful stop
-kill <pid>                   kill a process
-help                         show this help
-quit                         leave the console";
 
-async fn read(host: &Host, path: &str) -> Result<Vec<u8>> {
+/// Read a complete file through the host filesystem.
-    let fs = host.filesystem().await?;
+async fn read(host: &Host, path: &str) -> Result<String> {
-    let file = fs
+    let file = host
+        .filesystem()
+        .await?
         .open(
             path,
             Open {
@@ -35,27 +29,72 @@ async fn read(host: &Host, path: &str) -> Result<Vec<u8>> {
     loop {
         let chunk = file.read(bytes.len() as u64, 65536).await?;
         if chunk.is_empty() {
-            return Ok(bytes);
+            return Ok(String::from_utf8(bytes)?);
         }
         bytes.extend(chunk);
     }
 }
-fn pretty(value: &Value) -> Result<String> {
+
-    Ok(serde_json::to_string_pretty(value)?)
+/// Recognize the process IDs accepted by stop, kill, and log.
-}
+fn is_pid(text: &str) -> bool {
-async fn call(host: &Host, method: &str, params: Value, chat: bool) -> Result<String> {
+    text.rsplit_once('-').is_some()
-    match host.call(method, params).await {
+        && text
-        Ok(value) if chat => Ok(value["text"]
+            .get(text.len().saturating_sub(36)..)
-            .as_str()
+            .is_some_and(|id| uuid::Uuid::try_parse(id).is_ok())
-            .or_else(|| value.as_str())
-            .map(str::to_owned)
-            .unwrap_or(pretty(&value)?)),
-        Ok(value) => pretty(&value),
-        Err(error) => Ok(format!("error: {}", pretty(&error)?)),
-    }
 }
 
-/// Execute one input line without terminal IO. The caller handles quit after printing.
+/// Render method signatures and their documentation.
+fn methods(plugin: &str, doc: &Value) -> String {
+    let mut lines = Vec::new();
+    for method in doc["methods"].as_array().into_iter().flatten() {
+        let params = method["params"].as_array().cloned().unwrap_or_default();
+        let signature = params
+            .iter()
+            .map(|p| {
+                format!(
+                    "{}: {}",
+                    p["name"].as_str().unwrap_or("?"),
+                    p["schema"]["type"].as_str().unwrap_or("any")
+                )
+            })
+            .collect::<Vec<_>>()
+            .join(", ");
+        lines.push(format!(
+            "{plugin}.{}({signature}) -> {}",
+            method["name"].as_str().unwrap_or("?"),
+            method["result"]["schema"]["type"].as_str().unwrap_or("any")
+        ));
+        for field in ["summary", "description"] {
+            if let Some(text) = method[field].as_str() {
+                lines.push(format!("  {text}"));
+            }
+        }
+        for p in params {
+            if let Some(text) = p["description"].as_str() {
+                lines.push(format!("  {}: {text}", p["name"].as_str().unwrap_or("?")));
+            }
+        }
+    }
+    lines.join("\n")
+}
+
+/// Parse journal filters, or treat invalid arguments as chat.
+fn log_params(words: &[&str]) -> Option<Value> {
+    let mut params = json!({});
+    let mut words = words.iter();
+    while let Some(word) = words.next() {
+        if *word == "-n" {
+            params["limit"] = json!(words.next()?.parse::<u64>().ok()?);
+        } else if params.get("plugin").is_none() && params.get("pid").is_none() {
+            params[if is_pid(word) { "pid" } else { "plugin" }] = json!(word);
+        } else {
+            return None;
+        }
+    }
+    Some(params)
+}
+
+/// Execute one line and return its unlabelled result.
 pub async fn execute_line(host: &Host, line: &str) -> Result<String> {
     let line = line.trim();
     if line.is_empty() {
@@ -64,84 +103,45 @@ pub async fn execute_line(host: &Host, line: &str) -> Result<String> {
     let (command, rest) = line.split_once(char::is_whitespace).unwrap_or((line, ""));
     let rest = rest.trim();
     let words: Vec<_> = rest.split_whitespace().collect();
-    match command {
+    match (command, words.as_slice()) {
-        "help" if words.is_empty() => Ok(HELP.into()),
+        ("help", []) => return Ok(HELP.into()),
-        "quit" if words.is_empty() => Ok("Goodbye.".into()),
+        ("quit", []) => return Ok("Goodbye.".into()),
-        "ls" if words.len() <= 1 => {
+        ("cat", [path]) => return read(host, path).await,
-            let fs = host.filesystem().await?;
+        ("ls", [] | [_]) => {
-            let mut entries = fs.list(words.first().copied().unwrap_or("/")).await?;
+            let mut entries = host
+                .filesystem()
+                .await?
+                .list(words.first().copied().unwrap_or("/"))
+                .await?;
             entries.sort_by(|a, b| a.name.cmp(&b.name));
-            Ok(entries
+            return Ok(entries
-                .into_iter()
+                .iter()
                 .map(|e| format!("{}{}", e.name, if e.directory { "/" } else { "" }))
                 .collect::<Vec<_>>()
-                .join("\n"))
+                .join("\n"));
         }
-        "cat" if words.len() == 1 => Ok(String::from_utf8(read(host, rest).await?)?),
+        ("methods", [plugin]) => {
-        "methods" if words.len() == 1 => {
             let _activation = host.0.activations.lock().await;
             host.refresh_graph()
                 .await
-                .map_err(|error| anyhow::anyhow!("{error}"))?;
+                .map_err(|e| anyhow::anyhow!("{e}"))?;
-            let document: Value = serde_json::from_slice(
+            return Ok(methods(
-                &read(
+                plugin,
-                    host,
+                &serde_json::from_str(
-                    &format!("/proc/core/plugins/{}/discover.json", words[0]),
+                    &read(host, &format!("/proc/core/plugins/{plugin}/discover.json")).await?,
-                )
+                )?,
-                .await?,
+            ));
-            )?;
-            let mut lines = Vec::new();
-            for method in document["methods"]
-                .as_array()
-                .ok_or_else(|| anyhow::anyhow!("discovery document has no methods"))?
-            {
-                let params = method["params"]
-                    .as_array()
-                    .map(|params| {
-                        params
-                            .iter()
-                            .map(|p| {
-                                format!(
-                                    "{}: {}",
-                                    p["name"].as_str().unwrap_or("?"),
-                                    p["schema"]["type"].as_str().unwrap_or("any")
-                                )
-                            })
-                            .collect::<Vec<_>>()
-                            .join(", ")
-                    })
-                    .unwrap_or_default();
-                lines.push(format!(
-                    "{}.{}({}) -> {}",
-                    words[0],
-                    method["name"].as_str().unwrap_or("?"),
-                    params,
-                    method["result"]["schema"]["type"].as_str().unwrap_or("any")
-                ));
-                for field in ["summary", "description"] {
-                    if let Some(doc) = method[field].as_str() {
-                        lines.push(format!("  {doc}"));
-                    }
-                }
-                if let Some(params) = method["params"].as_array() {
-                    for p in params {
-                        if let Some(doc) = p["description"].as_str() {
-                            lines.push(format!("  {}: {doc}", p["name"].as_str().unwrap_or("?")));
-                        }
-                    }
-                }
-            }
-            Ok(lines.join("\n"))
         }
-        "ps" if words.is_empty() => {
+        ("ps", []) => {
-            let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis();
+            let now = std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)?
+                .as_millis();
             let mut processes: Vec<_> = host.processes().into_iter().collect();
             processes.sort_by(|a, b| a.0.cmp(&b.0));
             let mut lines = vec!["PID PLUGIN METHOD PARENT STATE AGE".into()];
-            lines.extend(processes.into_iter().map(|(pid, p)| {
+            lines.extend(processes.iter().map(|(pid, p)| {
                 format!(
-                    "{} {} {} {} {} {}s",
+                    "{pid} {} {} {} {} {}s",
-                    pid,
                     p.plugin,
                     p.method,
                     p.parent.as_deref().unwrap_or("-"),
@@ -149,332 +149,166 @@ pub async fn execute_line(host: &Host, line: &str) -> Result<String> {
                     now.saturating_sub(p.start_time) / 1000
                 )
             }));
-            Ok(lines.join("\n"))
+            return Ok(lines.join("\n"));
         }
-        "log" => {
+        _ => {}
-            let mut params = json!({});
-            let mut i = 0;
-            while i < words.len() {
-                if words[i] == "-n" {
-                    i += 1;
-                    let Some(count) = words.get(i).and_then(|value| value.parse::<u64>().ok())
-                    else {
-                        return call(host, "agent.chat", json!({"text":line}), true).await;
-                    };
-                    params["limit"] = json!(count);
-                } else if params.get("pid").is_none() && params.get("plugin").is_none() {
-                    let key = if words[i].len() > 36
-                        && words[i]
-                            .get(words[i].len() - 36..)
-                            .is_some_and(|suffix| uuid::Uuid::try_parse(suffix).is_ok())
-                    {
-                        "pid"
-                    } else {
-                        "plugin"
-                    };
-                    params[key] = json!(words[i]);
-                } else {
-                    return call(host, "agent.chat", json!({"text":line}), true).await;
-                }
-                i += 1;
-            }
-            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(
-                host,
-                "core.activate",
-                json!({"plugin":words[0],"wasm":words[1]}),
-                false,
-            )
-            .await
-        }
-        "revert" if words.len() == 1 => {
-            call(host, "core.revert", json!({"plugin":words[0]}), false).await
-        }
-        "stop" | "kill"
-            if words.len() == 1
-                && words[0].rsplit_once('-').is_some()
-                && words[0]
-                    .get(words[0].len().saturating_sub(36)..)
-                    .is_some_and(|id| uuid::Uuid::try_parse(id).is_ok()) =>
-        {
-            call(
-                host,
-                &format!("core.{command}"),
-                json!({"pid":words[0]}),
-                false,
-            )
-            .await
-        }
-        _ if (rest.is_empty() || rest.starts_with('{'))
-            && command
-                .split_once('.')
-                .is_some_and(|(plugin, method)| !plugin.is_empty() && !method.is_empty()) =>
-        {
-            call(
-                host,
-                command,
-                if rest.is_empty() {
-                    json!({})
-                } else {
-                    serde_json::from_str(rest)?
-                },
-                false,
-            )
-            .await
-        }
-        _ => call(host, "agent.chat", json!({"text":line}), true).await,
     }
-}
+    let explicit = (rest.is_empty() || rest.starts_with('{'))
-
+        && command
-/// Format text notifications dimly, while preserving other notification payloads.
+            .split_once('.')
-pub fn notification_text(notification: &Value) -> String {
+            .is_some_and(|(p, m)| !p.is_empty() && !m.is_empty());
-    let params = &notification["params"];
+    let (method, params) = match (command, words.as_slice()) {
-    if params["name"] == "text" {
+        ("activate", [plugin, wasm]) => {
-        let payload = &params["params"];
+            ("core.activate".into(), json!({"plugin":plugin,"wasm":wasm}))
-        let text = payload["text"]
+        }
+        ("revert", [plugin]) => ("core.revert".into(), json!({"plugin":plugin})),
+        ("stop" | "kill", [pid]) if is_pid(pid) => (format!("core.{command}"), json!({"pid":pid})),
+        ("log", _) if log_params(&words).is_some() => {
+            ("core.log".into(), log_params(&words).unwrap())
+        }
+        _ if explicit => (
+            command.into(),
+            if rest.is_empty() {
+                json!({})
+            } else {
+                serde_json::from_str(rest)?
+            },
+        ),
+        _ => ("agent.chat".into(), json!({"text":line})),
+    };
+    let value = match host.call(&method, params).await {
+        Ok(value) => value,
+        Err(error) => return Ok(format!("error: {}", serde_json::to_string_pretty(&error)?)),
+    };
+    if method == "agent.chat" && !explicit {
+        return Ok(value["text"]
             .as_str()
-            .or_else(|| payload.as_str())
+            .or_else(|| value.as_str())
             .map(str::to_owned)
-            .unwrap_or_else(|| payload.to_string());
+            .unwrap_or(serde_json::to_string_pretty(&value)?));
-        format!("\x1b[2m{text}\x1b[0m")
-    } else {
-        notification.to_string()
     }
-}
+    if method == "core.log" && command == "log" {
-fn indent(text: &str) -> String {
+        return value
-    text.lines()
+            .as_array()
-        .map(|line| format!("  {line}"))
+            .unwrap()
-        .collect::<Vec<_>>()
+            .iter()
-        .join("\n")
+            .map(|e| {
+                let time = chrono::DateTime::parse_from_rfc3339(e["time"].as_str().unwrap())?
+                    .with_timezone(&chrono::Local);
+                let pid = e["pid"].as_str().unwrap();
+                let prefix = format!("{}-", e["plugin"].as_str().unwrap());
+                let short = pid
+                    .strip_prefix(&prefix)
+                    .map(|id| format!("{prefix}{}", &id[..8]))
+                    .unwrap_or(pid.into());
+                Ok(format!(
+                    "{} {short} {} {}",
+                    time.format("%H:%M:%S%.3f"),
+                    e["stream"].as_str().unwrap(),
+                    e["line"].as_str().unwrap()
+                ))
+            })
+            .collect::<Result<Vec<_>>>()
+            .map(|lines| lines.join("\n"));
+    }
+    Ok(serde_json::to_string_pretty(&value)?)
 }
 
-/// Run the line editor on a dedicated thread, with notifications through its external printer.
+/// Dim text notifications and preserve other payloads.
-/// Without a terminal (for example, input from a pipe), read plain lines instead.
+pub fn notification_text(value: &Value) -> String {
+    let params = &value["params"];
+    if params["name"] != "text" {
+        return value.to_string();
+    }
+    let payload = &params["params"];
+    let text = payload["text"]
+        .as_str()
+        .or_else(|| payload.as_str())
+        .map(str::to_owned)
+        .unwrap_or(payload.to_string());
+    format!("\x1b[2m{text}\x1b[0m")
+}
+
+/// Read and execute one line at a time. Interrupting a wait does not cancel its host process.
 pub async fn run_input(host: Host) -> Result<()> {
-    if !std::io::IsTerminal::is_terminal(&std::io::stdin()) {
+    let terminal = 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 mut editor = if terminal {
-    let (sender, lines) = tokio::sync::mpsc::unbounded_channel::<Result<String>>();
+        Some(DefaultEditor::new()?)
-    // A detached terminal thread does not hold Tokio shutdown open at a blocked readline.
+    } else {
+        None
+    };
+    if let Some(editor) = &mut editor {
+        let _ = editor.load_history(&history);
+    }
+    let mut printer = editor
+        .as_mut()
+        .map(|e| e.create_external_printer())
+        .transpose()?;
+    let (request, requests) =
+        std::sync::mpsc::channel::<tokio::sync::oneshot::Sender<Result<Option<String>>>>();
     std::thread::spawn(move || {
-        let setup = (|| -> Result<_> {
+        for response in requests {
-            let mut editor = DefaultEditor::new()?;
+            let line = (|| -> Result<_> {
-            let _ = editor.load_history(&history);
+                if let Some(editor) = &mut editor {
-            let printer = editor.create_external_printer()?;
+                    match editor.readline("> ") {
-            Ok((editor, printer))
+                        Ok(line) => {
-        })();
+                            editor.add_history_entry(&line)?;
-        let (mut editor, printer) = match setup {
+                            editor.save_history(&history)?;
-            Ok(value) => value,
+                            Ok(Some(line))
-            Err(error) => {
-                let _ = ready.send(Err(error));
-                return;
-            }
-        };
-        if ready.send(Ok(printer)).is_err() {
-            return;
-        }
-        let result = (|| -> Result<()> {
-            loop {
-                match editor.readline("> ") {
-                    Ok(line) => {
-                        editor.add_history_entry(&line)?;
-                        editor.save_history(&history)?;
-                        if sender.send(Ok(line)).is_err() {
-                            break;
                         }
+                        Err(ReadlineError::Interrupted | ReadlineError::Eof) => Ok(None),
+                        Err(e) => Err(e.into()),
                     }
-                    Err(ReadlineError::Interrupted | ReadlineError::Eof) => {
+                } else {
-                        if sender.send(Ok("quit".into())).is_err() {
+                    std::io::stdin()
-                            break;
+                        .lines()
-                        }
+                        .next()
-                    }
+                        .transpose()
-                    Err(error) => return Err(error.into()),
+                        .map_err(Into::into)
                 }
-            }
+            })();
-            Ok(())
+            if response.send(line).is_err() {
-        })();
-        if let Err(error) = result {
-            let _ = sender.send(Err(error));
-        }
-    });
-    let mut printer = started.await??;
-    run_commands(host, lines, true, move |text| {
-        printer.print(format!("{text}\n"))?;
-        Ok(())
-    })
-    .await
-}
-
-/// 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, lines) = tokio::sync::mpsc::unbounded_channel::<Result<String>>();
-    std::thread::spawn(move || {
-        for line in std::io::stdin().lines() {
-            let line = line.map_err(Into::into);
-            if sender.send(line).is_err() {
                 break;
             }
         }
     });
-    run_commands(host, lines, false, |text| {
-        use std::io::Write;
-        writeln!(std::io::stdout(), "{text}")?;
-        Ok(())
-    })
-    .await
-}
-
-// Both input modes label results. Only terminal input starts concurrent commands.
-async fn run_commands(
-    host: Host,
-    mut lines: tokio::sync::mpsc::UnboundedReceiver<Result<String>>,
-    concurrent: bool,
-    mut print: impl FnMut(String) -> Result<()>,
-) -> Result<()> {
     let mut notifications = host.subscribe();
-    let mut commands = tokio::task::JoinSet::new();
+    let mut notification_task = tokio::task::JoinSet::new();
-    let mut input_open = true;
+    notification_task.spawn(async move {
+        loop {
+            match notifications.recv().await {
+                Ok(value) => {
+                    let text = notification_text(&value);
+                    if let Some(printer) = &mut printer {
+                        let _ = printer.print(format!("{text}\n"));
+                    } else {
+                        println!("{text}");
+                    }
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
+                Err(_) => break,
+            }
+        }
+    });
+    let mut interrupt = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())?;
     loop {
-        if !input_open && commands.is_empty() {
+        let (send, line) = tokio::sync::oneshot::channel();
+        request.send(send)?;
+        let line =
+            tokio::select! { result = line => result??, _ = interrupt.recv() => return Ok(()) };
+        let Some(line) = line else {
+            return Ok(());
+        };
+        let text = tokio::select! {
+            result = execute_line(&host, &line) => result.unwrap_or_else(|e| format!("error: {e}")),
+            _ = interrupt.recv() => "stopped waiting; the call continues, see ps".into(),
+        };
+        if !text.is_empty() {
+            println!("{text}");
+        }
+        if line.trim() == "quit" {
             return Ok(());
         }
-        tokio::select! {
-            biased;
-            notification = notifications.recv() => match notification {
-                Ok(value) => print(indent(&notification_text(&value)))?,
-                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
-                Err(tokio::sync::broadcast::error::RecvError::Closed) => return Ok(()),
-            },
-            result = commands.join_next(), if !commands.is_empty() => {
-                let (line,result): (String,Result<String>) = result.unwrap()?;
-                let text = result.unwrap_or_else(|error|format!("error: {error}"));
-                if !text.is_empty() { print(format!("> {line}\n{}",indent(&text)))?; }
-            },
-            line = lines.recv(), if input_open && (concurrent || commands.is_empty()) => match line {
-                Some(line) => {
-                    let line = line?;
-                    if line.trim()=="quit" { print("> quit\n  Goodbye.".into())?; return Ok(()); }
-                    let host = host.clone();
-                    commands.spawn(async move {
-                        let result=execute_line(&host,&line).await;
-                        (line,result)
-                    });
-                },
-                None => input_open=false,
-            },
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    #[tokio::test]
-    async fn methods_waits_for_activation() {
-        let folder = tempfile::tempdir().unwrap();
-        let world = crate::world::World::create(&folder.path().join("world")).unwrap();
-        let host = Host::start_with_cache(&world.root, &folder.path().join("cache"))
-            .await
-            .unwrap();
-        let guard = host.0.activations.lock().await;
-        let mut methods = Box::pin(execute_line(&host, "methods missing"));
-        assert!(
-            tokio::time::timeout(std::time::Duration::from_millis(100), &mut methods)
-                .await
-                .is_err()
-        );
-        drop(guard);
-        assert!(methods.await.is_err());
-        host.stop().await;
-    }
-    #[tokio::test]
-    async fn commands_remain_available_and_notifications_precede_results() {
-        let folder = tempfile::tempdir().unwrap();
-        let world = crate::world::World::create(&folder.path().join("world")).unwrap();
-        let host = Host::start_with_cache(&world.root, &folder.path().join("cache"))
-            .await
-            .unwrap();
-        host.register_native(
-            "slow",
-            std::sync::Arc::new(|ctx, _, _| {
-                Box::pin(async move {
-                    ctx.notify("ready", json!({"pid":ctx.pid}));
-                    std::future::pending().await
-                })
-            }),
-        );
-        host.register_native(
-            "fast",
-            std::sync::Arc::new(|ctx, _, _| {
-                Box::pin(async move {
-                    ctx.notify("text", json!("before"));
-                    Ok(json!("after"))
-                })
-            }),
-        );
-        let mut events = host.subscribe();
-        let (sender, lines) = tokio::sync::mpsc::unbounded_channel();
-        let (printed, mut output) = tokio::sync::mpsc::unbounded_channel();
-        let runner = tokio::spawn(run_commands(host.clone(), lines, true, move |text| {
-            printed.send(text).unwrap();
-            Ok(())
-        }));
-        sender.send(Ok("slow.wait".into())).unwrap();
-        let event = events.recv().await.unwrap();
-        let pid = event["params"]["params"]["pid"].as_str().unwrap();
-        sender.send(Ok("ps".into())).unwrap();
-        loop {
-            let text = output.recv().await.unwrap();
-            if text.starts_with("> ps") {
-                assert!(text.contains(pid));
-                break;
-            }
-        }
-        sender.send(Ok(format!("kill {pid}"))).unwrap();
-        let mut killed = false;
-        let mut answered = false;
-        while !killed || !answered {
-            let text = output.recv().await.unwrap();
-            killed |= text.starts_with("> kill ");
-            answered |= text.starts_with("> slow.wait");
-        }
-        sender.send(Ok("fast.go".into())).unwrap();
-        assert!(output.recv().await.unwrap().contains("before"));
-        assert!(output.recv().await.unwrap().contains("after"));
-        sender.send(Ok("quit".into())).unwrap();
-        runner.await.unwrap().unwrap();
-        host.stop().await;
     }
 }

cli/src/main.rs

@@ -186,7 +186,7 @@ async fn main() -> Result<std::process::ExitCode> {
             let exit = tokio::select! {
                 _=host.exit_requested()=>true,
                 _=term.recv()=>false,
-                _=interrupt.recv()=>false,
+                _=interrupt.recv(), if cli.headless=>false,
                 result = console, if !cli.headless => {
                     console_result=result;
                     false

cli/tests/host/agent.rs

@@ -111,40 +111,8 @@ impl Drop for Llm {
 fn fixtures() -> &'static PathBuf {
     static TARGET: OnceLock<PathBuf> = OnceLock::new();
     TARGET.get_or_init(|| {
-        let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        super::common::build("agent", "agent");
-            .parent()
+        super::common::sdk_probe().parent().unwrap().to_owned()
-            .unwrap()
-            .to_owned();
-        let target = PathBuf::from(concat!(env!("CARGO_TARGET_TMPDIR"), "/guests"));
-        for args in [
-            vec![
-                "build",
-                "--offline",
-                "-p",
-                "agent",
-                "--target",
-                "wasm32-wasip2",
-            ],
-            vec![
-                "build",
-                "--offline",
-                "--manifest-path",
-                "sdk/tests/plugins/probe/Cargo.toml",
-                "--target",
-                "wasm32-wasip2",
-            ],
-        ] {
-            assert!(
-                std::process::Command::new("cargo")
-                    .args(args)
-                    .current_dir(&root)
-                    .env("CARGO_TARGET_DIR", &target)
-                    .status()
-                    .unwrap()
-                    .success()
-            );
-        }
-        target.join("wasm32-wasip2/debug")
     })
 }
 struct TestWorld(PathBuf);

cli/tests/host/cargo.rs

@@ -1,9 +1,9 @@
 //! Native build and WASI libtest integration checks under PROTOCOL.md section 11.
+use super::common::CACHE;
 use sausage::{Host, world::World};
 use serde_json::{Value, json};
 use std::path::{Path, PathBuf};
 use tokio::sync::OnceCell;
-const CACHE: &str = concat!(env!("CARGO_TARGET_TMPDIR"), "/cache");
 struct TestWorld(PathBuf);
 impl TestWorld {
     /// Creates a fresh native build world.

cli/tests/host/cli.rs

@@ -4,7 +4,6 @@ use serde_json::{Value, json};
 use std::{
     path::{Path, PathBuf},
     process::Stdio,
-    sync::OnceLock,
     time::Duration,
 };
 use tokio::{
@@ -12,7 +11,7 @@ use tokio::{
     process::{Child, Command},
 };
 
-const CACHE: &str = concat!(env!("CARGO_TARGET_TMPDIR"), "/cache");
+use super::common::{CACHE, probe};
 
 fn binary() -> Command {
     let mut command = Command::new(env!("CARGO_BIN_EXE_sausage"));
@@ -50,31 +49,6 @@ fn signal(child: &Child, signal: nix::sys::signal::Signal) {
     )
     .unwrap();
 }
-fn probe() -> PathBuf {
-    static WASM: OnceLock<PathBuf> = OnceLock::new();
-    WASM.get_or_init(|| {
-        let root = Path::new(env!("CARGO_MANIFEST_DIR")).parent().unwrap();
-        let target = root.join("target/core/console-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()
-        );
-        target.join("wasm32-wasip2/debug/probe.wasm")
-    })
-    .clone()
-}
 fn install_probe(world: &Path) {
     std::fs::create_dir_all(world.join("plugins/probe")).unwrap();
     std::fs::copy(probe(), world.join("plugins/probe/plugin.wasm")).unwrap();
@@ -137,7 +111,7 @@ async fn init_builds_agent() {
     }
     assert_eq!(
         std::fs::read(world.join("plugins/agent/source/Cargo.lock")).unwrap(),
-        include_bytes!("../../Cargo.lock")
+        include_bytes!("../../../Cargo.lock")
     );
     assert!(!world.join("state/cargo/target").exists());
     assert!(
@@ -307,6 +281,60 @@ async fn console_error_stops_host() {
     assert!(stopped > 0);
 }
 
+/// SIGINT stops only the wait, so ps can still find the live call.
+#[tokio::test]
+async fn interrupt_leaves_call_running() {
+    let (_folder, world) = world();
+    install_probe(&world);
+    let mut child = binary()
+        .arg("run")
+        .arg(&world)
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .spawn()
+        .unwrap();
+    let mut input = child.stdin.take().unwrap();
+    input.write_all(b"probe.busy\n").await.unwrap();
+    let pid = tokio::time::timeout(Duration::from_secs(20), async {
+        loop {
+            if world.join("state/core/journal.sqlite").exists() {
+                let db =
+                    rusqlite::Connection::open(world.join("state/core/journal.sqlite")).unwrap();
+                if let Ok(pid) = db.query_row(
+                    "SELECT pid FROM journal WHERE plugin='probe' AND line='start busy {}'",
+                    [],
+                    |r| r.get::<_, String>(0),
+                ) {
+                    break pid;
+                }
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+    signal(&child, nix::sys::signal::Signal::SIGINT);
+    input
+        .write_all(format!("ps\nkill {pid}\nquit\n").as_bytes())
+        .await
+        .unwrap();
+    drop(input);
+    let output = tokio::time::timeout(Duration::from_secs(20), child.wait_with_output())
+        .await
+        .unwrap()
+        .unwrap();
+    let text = String::from_utf8(output.stdout).unwrap();
+    assert_eq!(output.status.code(), Some(0), "{text}");
+    assert!(
+        text.contains("stopped waiting; the call continues, see ps"),
+        "{text}"
+    );
+    assert!(
+        text.contains(&format!("{pid} probe busy console running")),
+        "{text}"
+    );
+}
+
 /// Piped calls wait for activation, and EOF or quit preserves the last result.
 #[tokio::test]
 async fn piped_console_runs_dependent_commands_in_order() {
@@ -345,14 +373,10 @@ async fn piped_console_runs_dependent_commands_in_order() {
         );
         assert!(!stdout.contains("error:"), "{stdout}");
         assert!(stdout.contains(r#""ordered": true"#), "{stdout}");
-        assert!(
+        assert!(!stdout.contains("> core.activate"), "{stdout}");
-            stdout.find(&format!("> {activate}")).unwrap()
-                < stdout.find(&format!("> {call}")).unwrap(),
-            "{stdout}"
-        );
         if !ending.is_empty() {
             assert!(
-                stdout.find(r#""ordered": true"#).unwrap() < stdout.find("> quit").unwrap(),
+                stdout.find(r#""ordered": true"#).unwrap() < stdout.find("Goodbye.").unwrap(),
                 "{stdout}"
             );
         }

cli/tests/host/common.rs

@@ -0,0 +1,43 @@
+//! Persistent caches and guest builds shared by the host integration modules.
+use std::{
+    path::{Path, PathBuf},
+    sync::OnceLock,
+};
+
+pub const CACHE: &str = concat!(env!("CARGO_TARGET_TMPDIR"), "/cache");
+
+pub fn build(package: &str, binary: &str) -> PathBuf {
+    let root = Path::new(env!("CARGO_MANIFEST_DIR")).parent().unwrap();
+    let target = Path::new(concat!(env!("CARGO_TARGET_TMPDIR"), "/guests"));
+    assert!(
+        std::process::Command::new("cargo")
+            .args([
+                "build",
+                "--offline",
+                "--target",
+                "wasm32-wasip2",
+                "--manifest-path"
+            ])
+            .arg(root.join(package).join("Cargo.toml"))
+            .arg("--target-dir")
+            .arg(target)
+            .status()
+            .unwrap()
+            .success()
+    );
+    target
+        .join("wasm32-wasip2/debug")
+        .join(format!("{binary}.wasm"))
+}
+
+pub fn probe() -> PathBuf {
+    static WASM: OnceLock<PathBuf> = OnceLock::new();
+    WASM.get_or_init(|| build("cli/tests/plugins/probe", "probe"))
+        .clone()
+}
+
+pub fn sdk_probe() -> PathBuf {
+    static WASM: OnceLock<PathBuf> = OnceLock::new();
+    WASM.get_or_init(|| build("sdk/tests/plugins/probe", "sdk-probe"))
+        .clone()
+}

cli/tests/host/console.rs

@@ -1,38 +1,12 @@
 //! Console command contracts against a temporary world and the raw probe component.
+use super::common::probe as fixture;
 use sausage::{
     Host,
     console::{execute_line, notification_text},
     world::World,
 };
 use serde_json::{Value, json};
-use std::{path::PathBuf, sync::OnceLock};
+use std::path::PathBuf;
-
-fn fixture() -> PathBuf {
-    static WASM: OnceLock<PathBuf> = OnceLock::new();
-    WASM.get_or_init(|| {
-        let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
-            .parent()
-            .unwrap()
-            .to_owned();
-        let target = PathBuf::from(concat!(env!("CARGO_TARGET_TMPDIR"), "/guests"));
-        let status = 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();
-        assert!(status.success());
-        target.join("wasm32-wasip2/debug/probe.wasm")
-    })
-    .clone()
-}
 struct TestWorld(PathBuf);
 impl TestWorld {
     fn new() -> Self {
@@ -75,6 +49,11 @@ async fn rpc_call() {
     let result = command(&host, "probe.echo {\"answer\":42}").await;
     assert!(result.contains("\n  \"answer\": 42\n"));
     assert!(command(&host, "absent.echo {}").await.contains("error:"));
+    assert_eq!(
+        serde_json::from_str::<Value>(&command(&host, r#"agent.chat {"text":"explicit"}"#).await)
+            .unwrap(),
+        json!({"text":"explicit"})
+    );
     assert_eq!(
         command(&host, "probe.echo not-json").await,
         "probe.echo not-json"
@@ -111,6 +90,7 @@ async fn prose_is_not_a_command() {
         "ls too many words",
         "log -n nope",
         "README.md says hello",
+        "agent.chat not-json",
         "e.g. write notes",
     ] {
         assert_eq!(command(&host, line).await, line);
@@ -185,6 +165,8 @@ 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();
+    // Finish discovery before comparing two snapshots of the journal.
+    command(&host, "methods data-logger").await;
     let pid = host.call("probe.serve", json!({})).await.unwrap();
     for (line, filters) in [
         ("log".to_owned(), json!({})),

cli/tests/host/core.rs

@@ -1,33 +1,8 @@
 //! End-to-end host behavior with a raw WASI guest.
+use super::common::probe as fixture;
 use sausage::{Host, world::World};
 use serde_json::{Value, json};
-use std::{path::PathBuf, sync::OnceLock};
+use std::path::PathBuf;
-fn fixture() -> PathBuf {
-    static WASM: OnceLock<PathBuf> = OnceLock::new();
-    WASM.get_or_init(|| {
-        let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
-            .parent()
-            .unwrap()
-            .to_path_buf();
-        let target = PathBuf::from(concat!(env!("CARGO_TARGET_TMPDIR"), "/guests"));
-        let status = 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();
-        assert!(status.success());
-        target.join("wasm32-wasip2/debug/probe.wasm")
-    })
-    .clone()
-}
 struct TestWorld(PathBuf);
 impl TestWorld {
     fn new() -> Self {

cli/tests/host/main.rs

@@ -0,0 +1,9 @@
+//! Host integration contracts run in one parallel libtest pool.
+mod activation;
+mod agent;
+mod cargo;
+mod cli;
+mod common;
+mod console;
+mod core;
+mod sdk;

cli/tests/host/sdk.rs

@@ -6,36 +6,16 @@ use std::{path::PathBuf, sync::OnceLock, time::Duration};
 fn fixtures() -> &'static PathBuf {
     static TARGET: OnceLock<PathBuf> = OnceLock::new();
     TARGET.get_or_init(|| {
-        let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        for (package, binary) in [
-            .parent()
+            ("examples/notes", "notes"),
-            .unwrap()
+            ("examples/consumer", "consumer"),
-            .to_owned();
+            ("examples/www", "www"),
-        let target = PathBuf::from(concat!(env!("CARGO_TARGET_TMPDIR"), "/guests"));
+            ("examples/secrets", "secrets"),
-        for package in [
+            ("sdk/tests/plugins/consumer", "sdk-consumer"),
-            "examples/notes",
-            "examples/consumer",
-            "examples/www",
-            "examples/secrets",
-            "sdk/tests/plugins/probe",
-            "sdk/tests/plugins/consumer",
         ] {
-            assert!(
+            super::common::build(package, binary);
-                std::process::Command::new("cargo")
-                    .args([
-                        "build",
-                        "--offline",
-                        "--target",
-                        "wasm32-wasip2",
-                        "--manifest-path"
-                    ])
-                    .arg(root.join(package).join("Cargo.toml"))
-                    .env("CARGO_TARGET_DIR", &target)
-                    .status()
-                    .unwrap()
-                    .success()
-            );
         }
-        target.join("wasm32-wasip2/debug")
+        super::common::sdk_probe().parent().unwrap().to_owned()
     })
 }
 struct TestWorld(PathBuf);
@@ -133,8 +113,10 @@ async fn discovery_is_compatible() {
     let host = world.start().await;
     let actual = call(&host, "notes.rpc.discover", json!({})).await;
     let consumer = call(&host, "consumer.rpc.discover", json!({})).await;
-    let expected: Value =
+    let expected: Value = serde_json::from_str(include_str!(
-        serde_json::from_str(include_str!("../../examples/consumer/imports/notes.json")).unwrap();
+        "../../../examples/consumer/imports/notes.json"
+    ))
+    .unwrap();
     assert_eq!(actual, expected);
     assert_eq!(actual["openrpc"], "1.2.6");
     assert!(actual["components"]["schemas"]["AddParams"].is_object());