From 1617dfc1ee0fa1d4dcf18c38f694b1c9d1eb5926 Mon Sep 17 00:00:00 2001 From: Surma Date: Fri, 23 Oct 2026 09:14:26 +0100 Subject: [PATCH] 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 | 3 +- cli/src/console.rs | 640 +++++++++++------------------ cli/src/main.rs | 2 +- cli/tests/{ => host}/activation.rs | 0 cli/tests/{ => host}/agent.rs | 36 +- cli/tests/{ => host}/cargo.rs | 2 +- cli/tests/{ => host}/cli.rs | 92 +++-- cli/tests/host/common.rs | 43 ++ cli/tests/{ => host}/console.rs | 38 +- cli/tests/{ => host}/core.rs | 29 +- cli/tests/host/main.rs | 9 + cli/tests/{ => host}/sdk.rs | 42 +- 12 files changed, 377 insertions(+), 559 deletions(-) rename cli/tests/{ => host}/activation.rs (100%) rename cli/tests/{ => host}/agent.rs (95%) rename cli/tests/{ => host}/cargo.rs (99%) rename cli/tests/{ => host}/cli.rs (84%) create mode 100644 cli/tests/host/common.rs rename cli/tests/{ => host}/console.rs (92%) rename cli/tests/{ => host}/core.rs (98%) create mode 100644 cli/tests/host/main.rs rename cli/tests/{ => host}/sdk.rs (91%) diff --git a/AGENTS.md b/AGENTS.md index 49bc1f7..5ac9c98 100644 --- a/AGENTS.md +++ b/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 ``` diff --git a/cli/src/console.rs b/cli/src/console.rs index dc07113..4199f61 100644 --- a/cli/src/console.rs +++ b/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 /) -cat read a file -methods show methods, params, types, and docs -ps show pid, plugin, method, parent, state, age -log [plugin or pid] [-n N] read the journal -activate activate a version -revert restore the previous version -stop request a graceful stop -kill kill a process -help show this help -quit leave the console"; +ls [path], cat , methods +ps, log [plugin or pid] [-n N] +activate , revert +stop , kill , help, quit"; -async fn read(host: &Host, path: &str) -> Result> { - let fs = host.filesystem().await?; - let file = fs +/// Read a complete file through the host filesystem. +async fn read(host: &Host, path: &str) -> Result { + let file = host + .filesystem() + .await? .open( path, Open { @@ -35,27 +29,72 @@ async fn read(host: &Host, path: &str) -> Result> { 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 { - Ok(serde_json::to_string_pretty(value)?) -} -async fn call(host: &Host, method: &str, params: Value, chat: bool) -> Result { - match host.call(method, params).await { - Ok(value) if chat => Ok(value["text"] - .as_str() - .or_else(|| value.as_str()) - .map(str::to_owned) - .unwrap_or(pretty(&value)?)), - Ok(value) => pretty(&value), - Err(error) => Ok(format!("error: {}", pretty(&error)?)), - } + +/// Recognize the process IDs accepted by stop, kill, and log. +fn is_pid(text: &str) -> bool { + text.rsplit_once('-').is_some() + && text + .get(text.len().saturating_sub(36)..) + .is_some_and(|id| uuid::Uuid::try_parse(id).is_ok()) } -/// 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::>() + .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 { + let mut params = json!({}); + let mut words = words.iter(); + while let Some(word) = words.next() { + if *word == "-n" { + params["limit"] = json!(words.next()?.parse::().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 { let line = line.trim(); if line.is_empty() { @@ -64,84 +103,45 @@ pub async fn execute_line(host: &Host, line: &str) -> Result { 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 { - "help" if words.is_empty() => Ok(HELP.into()), - "quit" if words.is_empty() => Ok("Goodbye.".into()), - "ls" if words.len() <= 1 => { - let fs = host.filesystem().await?; - let mut entries = fs.list(words.first().copied().unwrap_or("/")).await?; + match (command, words.as_slice()) { + ("help", []) => return Ok(HELP.into()), + ("quit", []) => return Ok("Goodbye.".into()), + ("cat", [path]) => return read(host, path).await, + ("ls", [] | [_]) => { + 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 - .into_iter() + return Ok(entries + .iter() .map(|e| format!("{}{}", e.name, if e.directory { "/" } else { "" })) .collect::>() - .join("\n")) + .join("\n")); } - "cat" if words.len() == 1 => Ok(String::from_utf8(read(host, rest).await?)?), - "methods" if words.len() == 1 => { + ("methods", [plugin]) => { let _activation = host.0.activations.lock().await; host.refresh_graph() .await - .map_err(|error| anyhow::anyhow!("{error}"))?; - let document: Value = serde_json::from_slice( - &read( - host, - &format!("/proc/core/plugins/{}/discover.json", words[0]), - ) - .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::>() - .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")) + .map_err(|e| anyhow::anyhow!("{e}"))?; + return Ok(methods( + plugin, + &serde_json::from_str( + &read(host, &format!("/proc/core/plugins/{plugin}/discover.json")).await?, + )?, + )); } - "ps" if words.is_empty() => { - let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis(); + ("ps", []) => { + 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, + "{pid} {} {} {} {} {}s", p.plugin, p.method, p.parent.as_deref().unwrap_or("-"), @@ -149,332 +149,166 @@ pub async fn execute_line(host: &Host, line: &str) -> Result { 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::().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::>>() - .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, + _ => {} } -} - -/// Format text notifications dimly, while preserving other notification payloads. -pub fn notification_text(notification: &Value) -> String { - let params = ¬ification["params"]; - if params["name"] == "text" { - let payload = ¶ms["params"]; - let text = payload["text"] + let explicit = (rest.is_empty() || rest.starts_with('{')) + && command + .split_once('.') + .is_some_and(|(p, m)| !p.is_empty() && !m.is_empty()); + let (method, params) = match (command, words.as_slice()) { + ("activate", [plugin, wasm]) => { + ("core.activate".into(), json!({"plugin":plugin,"wasm":wasm})) + } + ("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()); - format!("\x1b[2m{text}\x1b[0m") - } else { - notification.to_string() + .unwrap_or(serde_json::to_string_pretty(&value)?)); } -} -fn indent(text: &str) -> String { - text.lines() - .map(|line| format!(" {line}")) - .collect::>() - .join("\n") + if method == "core.log" && command == "log" { + return value + .as_array() + .unwrap() + .iter() + .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::>>() + .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. -/// Without a terminal (for example, input from a pipe), read plain lines instead. +/// Dim text notifications and preserve other payloads. +pub fn notification_text(value: &Value) -> String { + let params = &value["params"]; + if params["name"] != "text" { + return value.to_string(); + } + let payload = ¶ms["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()) { - return run_plain(host).await; - } + let terminal = std::io::IsTerminal::is_terminal(&std::io::stdin()); let history = host.0.world.state_dir("core").join("history"); - let (ready, started) = tokio::sync::oneshot::channel(); - let (sender, lines) = tokio::sync::mpsc::unbounded_channel::>(); - // A detached terminal thread does not hold Tokio shutdown open at a blocked readline. + let mut editor = if terminal { + Some(DefaultEditor::new()?) + } 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::>>>(); std::thread::spawn(move || { - let setup = (|| -> Result<_> { - let mut editor = DefaultEditor::new()?; - let _ = editor.load_history(&history); - let printer = editor.create_external_printer()?; - Ok((editor, printer)) - })(); - let (mut editor, printer) = match setup { - Ok(value) => value, - 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; + for response in requests { + let line = (|| -> Result<_> { + if let Some(editor) = &mut editor { + match editor.readline("> ") { + Ok(line) => { + editor.add_history_entry(&line)?; + editor.save_history(&history)?; + Ok(Some(line)) } + Err(ReadlineError::Interrupted | ReadlineError::Eof) => Ok(None), + Err(e) => Err(e.into()), } - Err(ReadlineError::Interrupted | ReadlineError::Eof) => { - if sender.send(Ok("quit".into())).is_err() { - break; - } - } - Err(error) => return Err(error.into()), + } else { + std::io::stdin() + .lines() + .next() + .transpose() + .map_err(Into::into) } - } - Ok(()) - })(); - 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::>(); - std::thread::spawn(move || { - for line in std::io::stdin().lines() { - let line = line.map_err(Into::into); - if sender.send(line).is_err() { + })(); + if response.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>, - concurrent: bool, - mut print: impl FnMut(String) -> Result<()>, -) -> Result<()> { let mut notifications = host.subscribe(); - let mut commands = tokio::task::JoinSet::new(); - let mut input_open = true; + let mut notification_task = tokio::task::JoinSet::new(); + 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(¬ification_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) = 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; } } diff --git a/cli/src/main.rs b/cli/src/main.rs index a50cb3f..af1caa8 100644 --- a/cli/src/main.rs +++ b/cli/src/main.rs @@ -186,7 +186,7 @@ async fn main() -> Result { 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 diff --git a/cli/tests/activation.rs b/cli/tests/host/activation.rs similarity index 100% rename from cli/tests/activation.rs rename to cli/tests/host/activation.rs diff --git a/cli/tests/agent.rs b/cli/tests/host/agent.rs similarity index 95% rename from cli/tests/agent.rs rename to cli/tests/host/agent.rs index ddb2891..95979ba 100644 --- a/cli/tests/agent.rs +++ b/cli/tests/host/agent.rs @@ -111,40 +111,8 @@ impl Drop for Llm { fn fixtures() -> &'static PathBuf { static TARGET: OnceLock = OnceLock::new(); TARGET.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")); - 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") + super::common::build("agent", "agent"); + super::common::sdk_probe().parent().unwrap().to_owned() }) } struct TestWorld(PathBuf); diff --git a/cli/tests/cargo.rs b/cli/tests/host/cargo.rs similarity index 99% rename from cli/tests/cargo.rs rename to cli/tests/host/cargo.rs index 6711b18..78be9c1 100644 --- a/cli/tests/cargo.rs +++ b/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. diff --git a/cli/tests/cli.rs b/cli/tests/host/cli.rs similarity index 84% rename from cli/tests/cli.rs rename to cli/tests/host/cli.rs index 2edea04..f51c0d7 100644 --- a/cli/tests/cli.rs +++ b/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 = 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!( - stdout.find(&format!("> {activate}")).unwrap() - < stdout.find(&format!("> {call}")).unwrap(), - "{stdout}" - ); + assert!(!stdout.contains("> core.activate"), "{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}" ); } diff --git a/cli/tests/host/common.rs b/cli/tests/host/common.rs new file mode 100644 index 0000000..afb571c --- /dev/null +++ b/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 = OnceLock::new(); + WASM.get_or_init(|| build("cli/tests/plugins/probe", "probe")) + .clone() +} + +pub fn sdk_probe() -> PathBuf { + static WASM: OnceLock = OnceLock::new(); + WASM.get_or_init(|| build("sdk/tests/plugins/probe", "sdk-probe")) + .clone() +} diff --git a/cli/tests/console.rs b/cli/tests/host/console.rs similarity index 92% rename from cli/tests/console.rs rename to cli/tests/host/console.rs index 9dbba31..3c851ab 100644 --- a/cli/tests/console.rs +++ b/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}; - -fn fixture() -> PathBuf { - static WASM: OnceLock = 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() -} +use std::path::PathBuf; 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::(&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!({})), diff --git a/cli/tests/core.rs b/cli/tests/host/core.rs similarity index 98% rename from cli/tests/core.rs rename to cli/tests/host/core.rs index 0b6cec0..9e25a7e 100644 --- a/cli/tests/core.rs +++ b/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}; -fn fixture() -> PathBuf { - static WASM: OnceLock = 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() -} +use std::path::PathBuf; struct TestWorld(PathBuf); impl TestWorld { fn new() -> Self { diff --git a/cli/tests/host/main.rs b/cli/tests/host/main.rs new file mode 100644 index 0000000..f99b7a8 --- /dev/null +++ b/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; diff --git a/cli/tests/sdk.rs b/cli/tests/host/sdk.rs similarity index 91% rename from cli/tests/sdk.rs rename to cli/tests/host/sdk.rs index a9d74e4..dae662c 100644 --- a/cli/tests/sdk.rs +++ b/cli/tests/host/sdk.rs @@ -6,36 +6,16 @@ use std::{path::PathBuf, sync::OnceLock, time::Duration}; fn fixtures() -> &'static PathBuf { static TARGET: OnceLock = OnceLock::new(); TARGET.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")); - for package in [ - "examples/notes", - "examples/consumer", - "examples/www", - "examples/secrets", - "sdk/tests/plugins/probe", - "sdk/tests/plugins/consumer", + for (package, binary) in [ + ("examples/notes", "notes"), + ("examples/consumer", "consumer"), + ("examples/www", "www"), + ("examples/secrets", "secrets"), + ("sdk/tests/plugins/consumer", "sdk-consumer"), ] { - assert!( - 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() - ); + super::common::build(package, binary); } - 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 = - serde_json::from_str(include_str!("../../examples/consumer/imports/notes.json")).unwrap(); + let expected: Value = serde_json::from_str(include_str!( + "../../../examples/consumer/imports/notes.json" + )) + .unwrap(); assert_eq!(actual, expected); assert_eq!(actual["openrpc"], "1.2.6"); assert!(actual["components"]["schemas"]["AddParams"].is_object());