diff --git a/Cargo.lock b/Cargo.lock index ca5aa87..da4b1e7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2089,6 +2089,7 @@ dependencies = [ "serde", "serde_json", "sha2", + "tempfile", "tokio", "toml_edit", "uuid", diff --git a/PROTOCOL.md b/PROTOCOL.md index 08809dd..c03de8d 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -190,6 +190,7 @@ The standard JSON-RPC codes apply. The core adds these codes: | -32001 | There is no such plugin. | | -32002 | The process ended without an answer. `data` has the exit status and the last journal lines. | | -32003 | The activation was rejected. `data` has the reasons. | +| -32004 | `cargo.build` failed. `data` has the `output` of the failed step and the `tests`. | ## 4. The filesystem diff --git a/cli/Cargo.toml b/cli/Cargo.toml index 67046d3..317da6c 100644 --- a/cli/Cargo.toml +++ b/cli/Cargo.toml @@ -11,7 +11,7 @@ path = "src/main.rs" [dependencies] anyhow = "1" bytes = "1" -clap = { version = "4", features = ["derive"] } +clap = { version = "4", features = ["derive", "env"] } http = "1" http-body-util = "0.1" include_dir = "0.7" @@ -30,8 +30,9 @@ sausage-vfs = { version = "0.1.0", path = "../vfs" } sausage-interface = { version = "0.1.0", path = "../interface" } notify = "8.2.0" rustyline = "19" -nix = { version = "0.31.3", features = ["signal", "process"] } +nix = { version = "0.31.3", features = ["signal", "process", "term"] } chrono = { version = "0.4.42", default-features = false, features = ["clock"] } +tempfile = "3.27.0" [dev-dependencies] hyper = { version = "1.12.0", features = ["server", "http1"] } diff --git a/cli/src/activation.rs b/cli/src/activation.rs index 35302ed..8b75e03 100644 --- a/cli/src/activation.rs +++ b/cli/src/activation.rs @@ -170,7 +170,11 @@ impl Host { let directory = self.0.world.plugin_dir(&candidate.plugin); let current = directory.join("plugin.wasm"); if tokio::fs::try_exists(¤t).await.map_err(io_error)? { - tokio::fs::rename(¤t, directory.join("plugin.prev.wasm")) + let backup = candidate.copy.with_extension("previous"); + tokio::fs::hard_link(¤t, &backup) + .await + .map_err(io_error)?; + tokio::fs::rename(&backup, directory.join("plugin.prev.wasm")) .await .map_err(io_error)?; } diff --git a/cli/src/cargo.rs b/cli/src/cargo.rs index 16f1ea2..8ab23da 100644 --- a/cli/src/cargo.rs +++ b/cli/src/cargo.rs @@ -36,17 +36,19 @@ pub struct BuildOutput { pub struct BuildFailure { pub reason: String, pub tests: TestResults, + pub output: String, } impl std::fmt::Display for BuildFailure { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(formatter, "{}\n{}", self.reason, self.tests.output) + write!(formatter, "{}\n{}", self.reason, self.output) } } impl std::error::Error for BuildFailure {} impl From for BuildFailure { fn from(error: anyhow::Error) -> Self { Self { - reason: format!("{error:#}"), + reason: "build preparation failed".into(), + output: format!("{error:#}"), tests: TestResults::default(), } } @@ -54,7 +56,7 @@ impl From for BuildFailure { /// A host-independent builder. Each builder serializes access to its work and target folders. pub struct Cargo { world: World, - builds: tokio::sync::Mutex<()>, + builds: Arc>, runner: PathBuf, } impl Cargo { @@ -93,7 +95,9 @@ impl Cargo { ), ] { let mut document: toml_edit::DocumentMut = manifest.parse()?; - document["workspace"] = toml_edit::table(); + if !document.contains_key("workspace") { + document["workspace"] = toml_edit::table(); + } if let Some(path) = interface_path { for section in ["dependencies", "dev-dependencies"] { if let Some(dependency) = document @@ -109,7 +113,7 @@ impl Cargo { Ok(Self { world, runner, - builds: tokio::sync::Mutex::new(()), + builds: Arc::new(tokio::sync::Mutex::new(())), }) } /// Registers cargo.build as a cancellable native process. @@ -118,43 +122,61 @@ impl Cargo { host.register_native(ID,Arc::new(move |_context,method,params| { let builder=builder.clone(); Box::pin(async move { + if method=="rpc.discover" { return Ok(json!({"openrpc":"1.2.6","info":{"title":"cargo","version":"native"},"methods":[{"name":"build","paramStructure":"by-name","params":[{"name":"plugin","required":true,"schema":{"type":"string"}}],"result":{"name":"build","schema":{"type":"object","required":["wasm","tests","warnings"],"properties":{"wasm":{"type":"string"},"warnings":{"type":"string"},"tests":{"type":"object","required":["passed","failed","output"],"properties":{"passed":{"type":"integer","minimum":0},"failed":{"type":"integer","minimum":0},"output":{"type":"string"}}}}}},"errors":[{"code":-32004,"message":"Build failed"}]}]})); } if method!="build" { return Err(crate::error(-32601,"cargo method not found")); } let plugin=params["plugin"].as_str().ok_or_else(||crate::error(-32602,"plugin must be a string"))?; match builder.build(&builder.world.plugin_dir(plugin).join("source")).await { Ok(output)=>Ok(json!({"wasm":format!("/{}",output.wasm.strip_prefix(&builder.world.root).unwrap().display()),"tests":output.tests,"warnings":output.warnings})), - Err(failure)=>Err(json!({"code":-32004,"message":failure.reason,"data":{"tests":failure.tests}})), + Err(failure)=>Err(json!({"code":-32004,"message":failure.reason,"data":{"tests":failure.tests,"output":failure.output}})), } }) })); } /// Copies source once, runs WASI tests, then publishes a release component under state/cargo/builds. pub async fn build(&self, source: &Path) -> std::result::Result { - let _build = self.builds.lock().await; + let build_guard = self.builds.clone().lock_owned().await; let work = self.world.state_dir(ID).join("work"); let target = self.world.state_dir(ID).join("target"); + let temporary = self.world.state_dir(ID).join("tmp"); + let cleanup = temporary.clone(); let source = source.to_path_buf(); let copy = work.clone(); let sdk = self.world.plugin_dir(ID).join("sdk"); - tokio::task::spawn_blocking(move || -> Result<()> { - if copy.exists() { - std::fs::remove_dir_all(©)?; + let build_guard = tokio::task::spawn_blocking(move || -> Result<_> { + if cleanup.exists() { + std::fs::remove_dir_all(&cleanup) + .with_context(|| format!("remove {}", cleanup.display()))?; } - copy_source(&source, ©)?; + std::fs::create_dir_all(&cleanup) + .with_context(|| format!("create {}", cleanup.display()))?; + if copy.exists() { + std::fs::remove_dir_all(©) + .with_context(|| format!("remove {}", copy.display()))?; + } + copy_source(&source, ©, true)?; let manifest = copy.join("Cargo.toml"); - let text = std::fs::read_to_string(&manifest)?; - std::fs::write(manifest, prepare_manifest(&text, &sdk)?)?; - Ok(()) + let text = std::fs::read_to_string(&manifest) + .with_context(|| format!("read {}", manifest.display()))?; + std::fs::write( + &manifest, + prepare_manifest(&text, &sdk) + .with_context(|| format!("prepare {}", manifest.display()))?, + ) + .with_context(|| format!("write {}", manifest.display()))?; + Ok(build_guard) }) .await .map_err(anyhow::Error::from)??; let mut test = Command::new("cargo"); - test.args(["test", "--offline", "--target", TARGET, "--target-dir"]) + test.args(["test", "--target", TARGET, "--target-dir"]) .arg(&target) .current_dir(&work) - .env( - "CARGO_TARGET_WASM32_WASIP2_RUNNER", - format!("{} test-runner", self.runner.display()), - ); + .env("TMPDIR", &temporary) + .arg("--config") + .arg(format!( + "target.wasm32-wasip2.runner={}", + json!([self.runner, "test-runner"]) + )); let output = run_command(test).await?; let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); @@ -165,11 +187,12 @@ impl Cargo { } else { test_count(&stdout, "failed").max(1) }, - output: format!("{stdout}\n{stderr}"), + output: strip_progress(&format!("{stdout}\n{stderr}")), }; if !output.status.success() { return Err(BuildFailure { reason: "WASI tests failed".into(), + output: tests.output.clone(), tests, }); } @@ -178,7 +201,6 @@ impl Cargo { build .args([ "build", - "--offline", "--release", "--target", TARGET, @@ -190,35 +212,35 @@ impl Cargo { let output = run_command(build).await?; let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); - let mut diagnostics = stderr.to_string(); - let mut artifact = None; + let diagnostics = stderr.to_string(); + let mut artifacts = Vec::new(); for line in stdout.lines() { - if let Ok(message) = serde_json::from_str::(line) { - if message["reason"] == "compiler-artifact" - && let Some(executable) = message["executable"].as_str() - { - artifact = Some(PathBuf::from(executable)); - } - if let Some(rendered) = message["message"]["rendered"].as_str() { - diagnostics.push_str(rendered); - } + if let Ok(message) = serde_json::from_str::(line) + && message["reason"] == "compiler-artifact" + && message["target"]["kind"] + .as_array() + .is_some_and(|kinds| kinds.iter().any(|kind| kind == "bin")) + && let Some(executable) = message["executable"].as_str() + { + artifacts.push(PathBuf::from(executable)); } } if !output.status.success() { bail!("{diagnostics}"); } - let artifact = artifact.context("cargo built no binary")?; + if artifacts.len() != 1 { + bail!( + "cargo must build exactly one binary, found {}", + artifacts.len() + ); + } + let artifact = artifacts.pop().unwrap(); let builds = self.world.state_dir(ID).join("builds"); - std::fs::create_dir_all(&builds)?; - let number = std::fs::read_dir(&builds)? - .filter_map(|entry| entry.ok()?.file_name().to_str()?.parse::().ok()) - .max() - .unwrap_or(0) - + 1; - let destination = builds - .join(number.to_string()) - .join(artifact.file_name().unwrap()); - crate::world::write_atomic(&destination, &tokio::fs::read(artifact).await?)?; + let destination = tokio::task::spawn_blocking(move || { + let _guard = build_guard; + publish(&builds, &artifact) + }) + .await??; Ok(BuildOutput { wasm: destination, warnings: strip_progress(&diagnostics), @@ -227,7 +249,8 @@ impl Cargo { } .await; result.map_err(|error: anyhow::Error| BuildFailure { - reason: format!("{error:#}"), + reason: "release build failed".into(), + output: format!("{error:#}"), tests, }) } @@ -247,6 +270,7 @@ async fn run_command(mut command: Command) -> Result { command .process_group(0) .kill_on_drop(true) + .stdin(std::process::Stdio::null()) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()); let child = command.spawn().context("start cargo")?; @@ -254,17 +278,21 @@ async fn run_command(mut command: Command) -> Result { Ok(child.wait_with_output().await?) } /// Copies only source, excluding build outputs and git internals at every level. -fn copy_source(source: &Path, destination: &Path) -> Result<()> { - std::fs::create_dir_all(destination)?; - for entry in std::fs::read_dir(source)? { - let entry = entry?; - if matches!(entry.file_name().to_str(), Some("target" | ".git")) { +fn copy_source(source: &Path, destination: &Path, root: bool) -> Result<()> { + std::fs::create_dir_all(destination) + .with_context(|| format!("create {}", destination.display()))?; + for entry in std::fs::read_dir(source).with_context(|| format!("read {}", source.display()))? { + let entry = entry.with_context(|| format!("read entry in {}", source.display()))?; + if root && matches!(entry.file_name().to_str(), Some("target" | ".git")) { continue; } - if entry.file_type()?.is_dir() { - copy_source(&entry.path(), &destination.join(entry.file_name()))?; + let from = entry.path(); + let to = destination.join(entry.file_name()); + if from.is_dir() { + copy_source(&from, &to, false)?; } else { - std::fs::copy(entry.path(), destination.join(entry.file_name()))?; + std::fs::copy(&from, &to) + .with_context(|| format!("copy {} to {}", from.display(), to.display()))?; } } Ok(()) @@ -272,32 +300,75 @@ fn copy_source(source: &Path, destination: &Path) -> Result<()> { /// Repoints SDK dependencies without discarding feature flags and detaches the copied crate's workspace. fn prepare_manifest(text: &str, sdk: &Path) -> Result { let mut document: toml_edit::DocumentMut = text.parse()?; - for section in ["dependencies", "dev-dependencies", "build-dependencies"] { - if let Some(dependency) = document - .get_mut(section) - .and_then(|table| table.get_mut("sausage-plugin")) - { - let features = dependency.get("features").cloned(); - let defaults = dependency.get("default-features").cloned(); - let mut replacement = toml_edit::InlineTable::new(); - replacement.insert("path", sdk.display().to_string().into()); - if let Some(features) = features.and_then(|value| value.as_value().cloned()) { - replacement.insert("features", features); + rewrite_dependencies(document.as_table_mut(), sdk); + if !document.contains_key("workspace") { + document["workspace"] = toml_edit::table(); + } + Ok(document.to_string()) +} +/// Rewrites dependency entries recursively, including target and workspace tables. +fn rewrite_dependencies(table: &mut dyn toml_edit::TableLike, sdk: &Path) { + for (name, item) in table.iter_mut() { + if matches!( + &*name, + "dependencies" | "dev-dependencies" | "build-dependencies" + ) { + if let Some(dependencies) = item.as_table_like_mut() { + for (name, dependency) in dependencies.iter_mut() { + if dependency + .get("workspace") + .and_then(toml_edit::Item::as_bool) + == Some(true) + { + continue; + } + if &*name == "sausage-plugin" + || dependency.get("package").and_then(toml_edit::Item::as_str) + == Some("sausage-plugin") + { + if dependency.is_str() { + let version = dependency.as_str().unwrap().to_string(); + *dependency = toml_edit::value(toml_edit::InlineTable::new()); + dependency["version"] = toml_edit::value(version); + } + dependency["path"] = toml_edit::value(sdk.display().to_string()); + } + } } - if let Some(defaults) = defaults.and_then(|value| value.as_value().cloned()) { - replacement.insert("default-features", defaults); - } - *dependency = toml_edit::value(replacement); + } else if let Some(nested) = item.as_table_like_mut() { + rewrite_dependencies(nested, sdk); } } - document["workspace"] = toml_edit::table(); - Ok(document.to_string()) +} +/// Publishes a complete artifact and retains only the ten newest numeric build folders. +fn publish(builds: &Path, artifact: &Path) -> Result { + std::fs::create_dir_all(builds).with_context(|| format!("create {}", builds.display()))?; + let mut numbers: Vec = std::fs::read_dir(builds) + .with_context(|| format!("read {}", builds.display()))? + .filter_map(|entry| entry.ok()?.file_name().to_str()?.parse().ok()) + .collect(); + numbers.sort_unstable(); + let number = numbers.last().copied().unwrap_or(0) + 1; + let destination = builds + .join(number.to_string()) + .join(artifact.file_name().unwrap()); + crate::world::write_atomic( + &destination, + &std::fs::read(artifact).with_context(|| format!("read {}", artifact.display()))?, + ) + .with_context(|| format!("publish {}", destination.display()))?; + for old in numbers.iter().take(numbers.len().saturating_sub(9)) { + let path = builds.join(old.to_string()); + std::fs::remove_dir_all(&path).with_context(|| format!("remove {}", path.display()))?; + } + Ok(destination) } /// Writes embedded sources only when their bytes changed. fn write_changed(path: &Path, bytes: &[u8]) -> Result<()> { if std::fs::read(path).ok().as_deref() != Some(bytes) { - std::fs::create_dir_all(path.parent().unwrap())?; - std::fs::write(path, bytes)?; + std::fs::create_dir_all(path.parent().unwrap()) + .with_context(|| format!("create parent of {}", path.display()))?; + std::fs::write(path, bytes).with_context(|| format!("write {}", path.display()))?; } Ok(()) } diff --git a/cli/src/console.rs b/cli/src/console.rs index 37f32a9..4e24911 100644 --- a/cli/src/console.rs +++ b/cli/src/console.rs @@ -1,6 +1,6 @@ //! Console commands, readable results, and line editing with live notifications. use crate::Host; -use anyhow::{Result, bail}; +use anyhow::Result; use rustyline::{DefaultEditor, ExternalPrinter, error::ReadlineError}; use sausage_vfs::{Layer, Open}; use serde_json::{Value, json}; @@ -77,8 +77,9 @@ pub async fn execute_line(host: &Host, line: &str) -> Result { .collect::>() .join("\n")) } - "cat" if !rest.is_empty() => Ok(String::from_utf8(read(host, rest).await?)?), + "cat" if words.len() == 1 => Ok(String::from_utf8(read(host, rest).await?)?), "methods" if words.len() == 1 => { + let _activation = host.0.activations.lock().await; host.refresh_graph() .await .map_err(|error| anyhow::anyhow!("{error}"))?; @@ -156,10 +157,10 @@ pub async fn execute_line(host: &Host, line: &str) -> Result { while i < words.len() { if words[i] == "-n" { i += 1; - let count: u64 = words - .get(i) - .ok_or_else(|| anyhow::anyhow!("log -n requires a count"))? - .parse()?; + 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 @@ -173,7 +174,7 @@ pub async fn execute_line(host: &Host, line: &str) -> Result { }; params[key] = json!(words[i]); } else { - bail!("usage: log [plugin or pid] [-n N]"); + return call(host, "agent.chat", json!({"text":line}), true).await; } i += 1; } @@ -218,7 +219,13 @@ pub async fn execute_line(host: &Host, line: &str) -> Result { "revert" if words.len() == 1 => { call(host, "core.revert", json!({"plugin":words[0]}), false).await } - "stop" | "kill" if words.len() == 1 => { + "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}"), @@ -227,11 +234,10 @@ pub async fn execute_line(host: &Host, line: &str) -> Result { ) .await } - "help" | "quit" | "ls" | "methods" | "ps" | "activate" | "revert" | "stop" | "kill" - | "cat" => bail!("invalid arguments for {command}; use help"), - _ if command - .split_once('.') - .is_some_and(|(plugin, method)| !plugin.is_empty() && !method.is_empty()) => + _ if (rest.is_empty() || rest.starts_with('{')) + && command + .split_once('.') + .is_some_and(|(plugin, method)| !plugin.is_empty() && !method.is_empty()) => { call( host, @@ -279,9 +285,7 @@ pub async fn run_input(host: Host) -> Result<()> { } 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::< - Result<(String, tokio::sync::oneshot::Sender)>, - >(); + let (sender, lines) = tokio::sync::mpsc::unbounded_channel::>(); // A detached terminal thread does not hold Tokio shutdown open at a blocked readline. std::thread::spawn(move || { let setup = (|| -> Result<_> { @@ -306,15 +310,15 @@ pub async fn run_input(host: Host) -> Result<()> { Ok(line) => { editor.add_history_entry(&line)?; editor.save_history(&history)?; - let (done, wait) = tokio::sync::oneshot::channel(); - if sender.send(Ok((line, done))).is_err() - || !wait.blocking_recv().unwrap_or(false) - { + if sender.send(Ok(line)).is_err() { + break; + } + } + Err(ReadlineError::Interrupted | ReadlineError::Eof) => { + if sender.send(Ok("quit".into())).is_err() { break; } } - Err(ReadlineError::Interrupted) => break, - Err(ReadlineError::Eof) => break, Err(error) => return Err(error.into()), } } @@ -325,71 +329,151 @@ pub async fn run_input(host: Host) -> Result<()> { } }); let mut printer = started.await??; - let mut notifications = host.subscribe(); - let print_notifications = async move { - loop { - match notifications.recv().await { - Ok(value) => printer.print(format!("{}\n", indent(¬ification_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 (line, done) = line?; - let text = execute_line(&host, &line) - .await - .unwrap_or_else(|error| format!("error: {error}")); - if !text.is_empty() { - println!("{}", indent(&text)); - } - let quit = line.trim() == "quit"; - let _ = done.send(!quit); - if quit { - break; - } - } + run_commands(host, lines, move |text| { + printer.print(format!("{text}\n"))?; Ok(()) - }; - tokio::select! {result=commands=>result,result=print_notifications=>result} + }) + .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, mut lines) = tokio::sync::mpsc::unbounded_channel::(); + let (sender, lines) = tokio::sync::mpsc::unbounded_channel::>(); std::thread::spawn(move || { for line in std::io::stdin().lines() { - let Ok(line) = line else { break }; + let line = line.map_err(Into::into); if sender.send(line).is_err() { break; } } }); + run_commands(host, lines, |text| { + use std::io::Write; + writeln!(std::io::stdout(), "{text}")?; + Ok(()) + }) + .await +} + +// Both input modes keep the prompt available and label asynchronous results. +async fn run_commands( + host: Host, + mut lines: tokio::sync::mpsc::UnboundedReceiver>, + mut print: impl FnMut(String) -> Result<()>, +) -> Result<()> { let mut notifications = host.subscribe(); - let print_notifications = async move { - loop { - match notifications.recv().await { - Ok(value) => println!("{}", indent(¬ification_text(&value))), + let mut commands = tokio::task::JoinSet::new(); + let mut input_open = true; + loop { + if !input_open && commands.is_empty() { + 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 => 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, + }, } - }; - let commands = async { - while let Some(line) = lines.recv().await { - let text = execute_line(&host, &line) + } +} + +#[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 - .unwrap_or_else(|error| format!("error: {error}")); - if !text.is_empty() { - println!("{}", indent(&text)); - } - if line.trim() == "quit" { + .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, 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; } } - Ok(()) - }; - tokio::select! {result=commands=>result,result=print_notifications=>result} + 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/host.rs b/cli/src/host.rs index 59be425..eb595be 100644 --- a/cli/src/host.rs +++ b/cli/src/host.rs @@ -65,7 +65,25 @@ impl Host { .await??; let journal = journal::Journal::open(world.state_dir("core").join("journal.sqlite")).await?; - let host = Self(Arc::new(Inner { + let host = Self::with_parts(world, runtime, journal); + let cargo_world = host.0.world.clone(); + let cargo = + tokio::task::spawn_blocking(move || crate::cargo::Cargo::new(cargo_world)).await??; + cargo.register(&host); + host.boot().await?; + Ok(host) + } + /// Creates an empty test world with a shared engine and an in-memory journal. + pub async fn test_world(path: &Path, runtime: Arc) -> anyhow::Result { + std::fs::create_dir_all(path)?; + Ok(Self::with_parts( + World::open(path)?, + runtime, + journal::Journal::memory().await?, + )) + } + fn with_parts(world: World, runtime: Arc, journal: journal::Journal) -> Self { + Self(Arc::new(Inner { world, runtime, journal: Arc::new(journal), @@ -85,13 +103,7 @@ impl Host { activations: tokio::sync::Mutex::new(()), watcher: Mutex::new(None), retirements: Mutex::new(std::collections::HashSet::new()), - })); - let cargo_world = host.0.world.clone(); - let cargo = - tokio::task::spawn_blocking(move || crate::cargo::Cargo::new(cargo_world)).await??; - cargo.register(&host); - host.boot().await?; - Ok(host) + })) } /// Register a native plugin for subsequent calls. pub fn register_native(&self, name: &str, plugin: NativePlugin) { diff --git a/cli/src/journal.rs b/cli/src/journal.rs index 275f553..8d9f37a 100644 --- a/cli/src/journal.rs +++ b/cli/src/journal.rs @@ -23,6 +23,9 @@ pub(crate) struct Journal { warned: Mutex>, } impl Journal { + pub(crate) async fn memory() -> anyhow::Result { + Self::open(PathBuf::from(":memory:")).await + } pub(crate) async fn open(path: PathBuf) -> anyhow::Result { let (sender, receiver) = mpsc::channel(); let (ready, initialized) = oneshot::channel(); @@ -80,7 +83,9 @@ impl Journal { } } fn open_database(path: PathBuf) -> anyhow::Result { - std::fs::create_dir_all(path.parent().unwrap())?; + if path != std::path::Path::new(":memory:") { + std::fs::create_dir_all(path.parent().unwrap())?; + } let db = Connection::open(path)?; db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; diff --git a/cli/src/main.rs b/cli/src/main.rs index 0d52eb8..57312ab 100644 --- a/cli/src/main.rs +++ b/cli/src/main.rs @@ -14,7 +14,7 @@ mod test_runner; #[derive(Parser)] #[command(version)] struct Cli { - #[arg(long, global = true)] + #[arg(long, global = true, env = "SAUSAGE_HEADLESS", value_parser = headless)] headless: bool, #[arg(long, global = true)] cache_dir: Option, @@ -38,6 +38,28 @@ enum Cmd { args: Vec, }, } +fn headless(value: &str) -> std::result::Result { + match value.to_ascii_lowercase().as_str() { + "1" | "true" => Ok(true), + "" | "0" | "false" => Ok(false), + _ => Err("expected 1, true, 0, false, or an empty value".into()), + } +} + +// Readline can still wait on its detached thread when the host exits. +struct TerminalMode(Option); +impl Drop for TerminalMode { + fn drop(&mut self) { + if let Some(mode) = &self.0 { + let _ = nix::sys::termios::tcsetattr( + std::io::stdin(), + nix::sys::termios::SetArg::TCSANOW, + mode, + ); + } + } +} + fn setting(var: &str, question: &str, default: Option<&str>) -> Result { if let Ok(v) = std::env::var(var) && !v.is_empty() @@ -65,7 +87,18 @@ async fn init(dir: &Path) -> Result<()> { 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)?; + if dir.exists() { + bail!("world already exists: {}", dir.display()); + } + let parent = dir + .parent() + .filter(|p| !p.as_os_str().is_empty()) + .unwrap_or(Path::new(".")); + std::fs::create_dir_all(parent)?; + let staging = tempfile::Builder::new() + .prefix(".sausage-init-") + .tempdir_in(parent)?; + let world = World::create(&staging.path().join("world"))?; std::fs::write( world.root.join("config/agent.json"), serde_json::to_vec_pretty( @@ -80,15 +113,21 @@ async fn init(dir: &Path) -> Result<()> { source.join("Cargo.toml"), include_str!("../../agent/Cargo.toml"), )?; + std::fs::write( + source.join("Cargo.lock"), + include_bytes!("../../Cargo.lock"), + )?; 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"))?; + std::fs::rename(&world.root, dir)?; println!("Start the world with: sausage run {}", dir.display()); Ok(()) } #[tokio::main] -async fn main() -> Result<()> { +async fn main() -> Result { let cli = Cli::parse(); + let _terminal = TerminalMode(nix::sys::termios::tcgetattr(std::io::stdin()).ok()); let command = match cli.command { Some(command) => command, None => { @@ -102,16 +141,25 @@ async fn main() -> Result<()> { match command { Cmd::TestRunner { wasm, args } => { if !test_runner::run(&wasm, args).await? { - std::process::exit(1); + return Ok(std::process::ExitCode::FAILURE); } } Cmd::Init { dir } => init(&dir).await?, Cmd::Run { dir } => { let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?; - let host = match cli.cache_dir { - Some(cache) => Host::start_with_cache(&dir, &cache).await?, - None => Host::start(&dir).await?, + let mut interrupt = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())?; + let start = async { + match cli.cache_dir { + Some(cache) => Host::start_with_cache(&dir, &cache).await, + None => Host::start(&dir).await, + } + }; + let host = tokio::select! { + result=start=>result?, + _=term.recv()=>return Ok(std::process::ExitCode::SUCCESS), + _=interrupt.recv()=>return Ok(std::process::ExitCode::SUCCESS), }; let mut notifications = host.subscribe(); let notification_task = cli.headless.then(|| { @@ -126,12 +174,13 @@ async fn main() -> Result<()> { }) }); let console = sausage::console::run_input(host.clone()); + let mut console_result = Ok(()); let exit = tokio::select! { _=host.exit_requested()=>true, _=term.recv()=>false, - _=tokio::signal::ctrl_c()=>false, + _=interrupt.recv()=>false, result = console, if !cli.headless => { - result?; + console_result=result; false }, }; @@ -141,9 +190,10 @@ async fn main() -> Result<()> { task.abort(); } if exit { - std::process::exit(75); + return Ok(std::process::ExitCode::from(75)); } + console_result?; } } - Ok(()) + Ok(std::process::ExitCode::SUCCESS) } diff --git a/cli/src/runtime.rs b/cli/src/runtime.rs index 2dd2d42..5160ba6 100644 --- a/cli/src/runtime.rs +++ b/cli/src/runtime.rs @@ -221,10 +221,21 @@ impl Runtime { pub fn new(cache: &Path) -> Result> { let mut config = Config::new(); config.epoch_interruption(true); - std::fs::create_dir_all(cache)?; - let mut cache_config = wasmtime::CacheConfig::new(); - cache_config.with_directory(cache.canonicalize()?); - config.cache(Some(wasmtime::Cache::new(cache_config)?)); + let disk_cache = (|| -> Result<_> { + std::fs::create_dir_all(cache)?; + let mut cache_config = wasmtime::CacheConfig::new(); + cache_config.with_directory(cache.canonicalize()?); + Ok(wasmtime::Cache::new(cache_config)?) + })(); + match disk_cache { + Ok(cache) => { + config.cache(Some(cache)); + } + Err(error) => eprintln!( + "compile cache {} unavailable: {error}; running without disk cache", + cache.display() + ), + } let engine = Engine::new(&config)?; let mut linker: Linker = Linker::new(&engine); diff --git a/cli/src/test_runner.rs b/cli/src/test_runner.rs index 8e7fd70..f42cffd 100644 --- a/cli/src/test_runner.rs +++ b/cli/src/test_runner.rs @@ -3,14 +3,12 @@ use anyhow::{Result, bail}; use sausage::{ Host, runtime::{Exit, InstanceConfig, Runtime}, - world::World, }; use serde_json::{Value, json}; use std::{ path::{Path, PathBuf}, sync::Arc, }; -use tokio::io::AsyncWriteExt; use wasmtime::component::Component; /// A temporary world remains alive through the end of its WASI store. @@ -32,19 +30,36 @@ async fn instance( package: PathBuf, ) -> Result { let path = std::env::temp_dir().join(format!("sausage-test-{}", uuid::Uuid::now_v7())); - World::create(&path)?; let world = TestWorld(path); - let host = Host::start(&world.0).await?; + let host = Host::test_world(&world.0, runtime.clone()).await?; let tracker = sausage_vfs::Process::default(); let mut filesystem = host.filesystem().await?; + // Virtual ancestors keep directory entries of the absolute package mount traversable. + for ancestor in package + .ancestors() + .skip(1) + .filter(|path| *path != Path::new("/")) + { + filesystem.mount( + ancestor.to_str().unwrap(), + Arc::new(sausage_vfs::Memory::default()), + false, + ); + } filesystem.mount( package.to_str().unwrap(), Arc::new(sausage_vfs::Host::new(&package, tracker.clone())?), false, ); - let mut env: Vec<_> = std::env::vars().filter(|(key, _)| key != "PWD").collect(); + let mut env: Vec<_> = std::env::vars_os() + .filter_map(|(key, value)| Some((key.into_string().ok()?, value.into_string().ok()?))) + .filter(|(key, _)| { + ["CARGO_", "RUST_", "SAUSAGE_"] + .iter() + .any(|prefix| key.starts_with(prefix)) + }) + .collect(); env.extend([ - ("PWD".into(), package.display().to_string()), ("SAUSAGE_PID".into(), "test".into()), ("SAUSAGE_PLUGIN".into(), "test".into()), ("SAUSAGE_PARENT".into(), "core".into()), @@ -61,23 +76,38 @@ async fn instance( &args, ) .await?; - let mut stdin = guest.stdin.take().unwrap(); - let mut stdout = std::mem::replace(&mut guest.stdout, tokio::sync::mpsc::channel(1).1); + let mut pending = std::collections::VecDeque::new(); let mut stderr = std::mem::replace(&mut guest.stderr, tokio::sync::mpsc::channel(1).1); - let mut exit = None; + let (errors, mut stderr_lines) = tokio::sync::mpsc::unbounded_channel(); + let stderr_task = tokio::spawn(async move { + while let Some(line) = stderr.recv().await { + let _ = errors.send(line); + } + }); let mut output = String::new(); - let wait = guest.wait(); - tokio::pin!(wait); let mut stdout_open = true; let mut stderr_open = true; let mut handlers = tokio::task::JoinSet::new(); - while exit.is_none() || stdout_open || stderr_open { + while stdout_open || stderr_open || !pending.is_empty() { tokio::select! { - result=&mut wait, if exit.is_none()=>exit=Some(result), - line=stdout.recv(), if stdout_open=>match line { + line=async { if let Some(line) = pending.pop_front() { Some(line) } else { guest.stdout.recv().await } }, if stdout_open || !pending.is_empty()=>match line { Some(line)=> { // Libtest can prefix its status line before SDK output. The SDK starts JSON on a fresh line. if let Ok(request)=serde_json::from_str::(&line) && request["jsonrpc"]=="2.0" && request["method"].is_string() { + let method = request["method"].as_str().unwrap(); + if !matches!(method, "core.lock" | "core.log") { + let params = request.get("params").cloned().unwrap_or(json!({})); + let result = match method { + "core.unlock" => host.call(method, params).await, + "core.notify" => Ok(Value::Null), + _ => Err(json!({"code":-32601,"message":"a test has no other plugins"})), + }; + if let Some(id) = request.get("id") { + let response = match result { Ok(result)=>json!({"jsonrpc":"2.0","id":id,"result":result}), Err(error)=>json!({"jsonrpc":"2.0","id":id,"error":error}) }; + if guest.send_line(&response.to_string(), |line| pending.push_back(line)).await.is_err() { guest.stdin.take(); } + } + continue; + } let host = host.clone(); handlers.spawn(async move { let method=request["method"].as_str().unwrap(); @@ -89,25 +119,22 @@ async fn instance( }; request.get("id").map(|id|match result {Ok(result)=>json!({"jsonrpc":"2.0","id":id,"result":result}),Err(error)=>json!({"jsonrpc":"2.0","id":id,"error":error})}) }); - } else {output.push_str(&line);output.push('\n');} + } else {output.push_str(line.trim_end_matches(['\r', '\n']));output.push('\n');} },None=>stdout_open=false, }, - line=stderr.recv(), if stderr_open=>match line {Some(line)=>{output.push_str(&line);output.push('\n');},None=>stderr_open=false}, + line=stderr_lines.recv(), if stderr_open=>match line {Some(line)=>{output.push_str(&line);output.push('\n');},None=>stderr_open=false}, response=handlers.join_next(), if !handlers.is_empty()=> { - if let Some(response) = response.transpose()?.flatten() { - stdin.write_all(format!("{response}\n").as_bytes()).await?; - stdin.flush().await?; - } + if let Some(response) = response.transpose()?.flatten() + && guest.send_line(&response.to_string(), |line| pending.push_back(line)).await.is_err() { guest.stdin.take(); } }, } } handlers.abort_all(); + let exit = guest.wait().await; + let _ = stderr_task.await; tracker.idle().await; host.stop().await; - Ok(Run { - exit: exit.unwrap(), - output, - }) + Ok(Run { exit, output }) } /// Extracts libtest names from its stable list format. fn names(output: &str) -> Vec { @@ -220,7 +247,19 @@ pub async fn run(wasm: &Path, args: Vec) -> Result { for line in run .output .lines() - .filter(|line| !line.starts_with("test result:")) + .filter(|line| { + !line.trim().is_empty() + && !line.starts_with("test result:") + && !matches!(*line, "running 1 test" | "running 0 tests") + && !matches!(*line, "ok" | "FAILED") + }) + .map(|line| { + line.strip_prefix(&format!("test {name} ... ")) + .unwrap_or(line) + }) + .filter(|line| { + !matches!(*line, "ok" | "FAILED" | "ignored") && !line.is_empty() + }) { // Indentation keeps ordinary test output separate from our libtest status and totals. println!(" {line}"); diff --git a/cli/tests/activation.rs b/cli/tests/activation.rs index 23a6960..31cd6e9 100644 --- a/cli/tests/activation.rs +++ b/cli/tests/activation.rs @@ -313,6 +313,24 @@ async fn review_retirement_checks_live_version() { marker(world.0.join("state/notes/retired"), &previous).await; host.stop().await; } +/// T23: an unwritable compile cache does not prevent a world from running. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn unwritable_compile_cache_is_nonfatal() { + use std::os::unix::fs::PermissionsExt; + let world = TestWorld::new(); + world.plugin("notes", document("add"), document("add")); + let parent = world.0.join("unwritable"); + std::fs::create_dir(&parent).unwrap(); + std::fs::set_permissions(&parent, std::fs::Permissions::from_mode(0o500)).unwrap(); + let result = Host::start_with_cache(&world.0, &parent.join("cache")).await; + std::fs::set_permissions(&parent, std::fs::Permissions::from_mode(0o700)).unwrap(); + let host = result.unwrap(); + assert_eq!( + call(&host, "notes.version", json!({})).await["version"], + "1" + ); + 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() { @@ -322,7 +340,7 @@ async fn review_watcher_is_shallow() { 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; + let result = Host::start_with_cache(&world.0, &world.0.join("compile-cache")).await; std::fs::set_permissions(&source, std::fs::Permissions::from_mode(0o700)).unwrap(); let host = result.unwrap(); activate(&host, "notes").await.unwrap(); @@ -337,7 +355,7 @@ async fn review_watcher_failure_is_nonfatal() { 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; + let result = Host::start_with_cache(&world.0, &world.0.join("compile-cache")).await; std::fs::set_permissions(&inaccessible, std::fs::Permissions::from_mode(0o700)).unwrap(); let host = result.unwrap(); assert!( @@ -785,3 +803,165 @@ async fn does_not_restart_detached_call() { ); host.stop().await; } + +/// PROTOCOL 7: activation records the plugin's upgrade grace period in the journal. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn upgrades_enter_the_journal() { + let world = TestWorld::new(); + let mut old = document("hold"); + old.as_object_mut().unwrap().remove("x-sausage-grace"); + world.plugin("notes", old, document("hold")); + let host = world.start().await; + call(&host, "notes.hold", json!({})).await; + activate(&host, "notes").await.unwrap(); + let entries = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let entries = call( + &host, + "core.log", + json!({"plugin":"notes","grep":"upgrade grace"}), + ) + .await; + if entries + .as_array() + .unwrap() + .iter() + .any(|e| e["line"] == "upgrade grace 60 seconds") + { + break entries; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("the upgrade journal entry must arrive within 10 seconds"); + assert!( + entries + .as_array() + .unwrap() + .iter() + .all(|e| e["stream"] == "core") + ); + host.stop().await; +} + +/// PROTOCOL 10.2 and 10.3: breaking endpoint methods reject activation with counterexamples where possible. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn endpoint_contracts_and_activation_counterexamples() { + let method = json!({"name":"stats","params":[{"name":"value","required":true,"schema":{"type":"integer","minimum":0}}],"result":{"schema":{"type":"string"}}}); + let expected = json!({"methods":[],"x-sausage-endpoints":{"control":{"methods":[method]}}}); + for case in ["missing", "params", "result", "unsupported"] { + let world = TestWorld::new(); + let mut next = expected.clone(); + match case { + "missing" => next["x-sausage-endpoints"]["control"]["methods"] = json!([]), + "params" => { + next["x-sausage-endpoints"]["control"]["methods"][0]["params"][0]["schema"]["minimum"] = + json!(1) + } + "result" => { + next["x-sausage-endpoints"]["control"]["methods"][0]["result"]["schema"] = + json!({"type":"integer"}) + } + _ => { + next["x-sausage-endpoints"]["control"]["methods"][0]["result"]["schema"]["pattern"] = + json!("[a-z]+") + } + } + world.plugin("notes", expected.clone(), next); + world.plugin( + "agent", + consumer("notes", expected.clone()), + document("consume"), + ); + let host = world.start().await; + let error = activate(&host, "notes").await.unwrap_err(); + assert_eq!(error["code"], -32003, "{case}: {error}"); + assert!(!error["data"].as_array().unwrap().is_empty()); + if ["params", "result"].contains(&case) { + assert!( + error["data"] + .as_array() + .unwrap() + .iter() + .any(|p| !p["counterexample"].is_null()), + "{error}" + ); + } + if case == "unsupported" { + assert!(error["data"].to_string().contains("unsupported"), "{error}"); + } + assert_eq!( + std::fs::read(world.0.join("plugins/notes/plugin.wasm")).unwrap(), + std::fs::read(fixture(1)).unwrap() + ); + host.stop().await; + } +} + +/// PROTOCOL 10.2: providers can add methods, optional params, and result fields without breaking consumers. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn additive_contract_changes_activate() { + let old = json!({"methods":[{"name":"read","params":[],"result":{"schema":{"type":"object","properties":{"text":{"type":"string"}},"required":["text"]}}}]}); + let mut next = old.clone(); + next["methods"][0]["params"] = + json!([{"name":"limit","required":false,"schema":{"type":"integer"}}]); + next["methods"][0]["result"]["schema"]["properties"]["extra"] = json!({"type":"boolean"}); + next["methods"] + .as_array_mut() + .unwrap() + .push(json!({"name":"new","params":[],"result":{"schema":{"type":"null"}}})); + let world = TestWorld::new(); + world.plugin("notes", old.clone(), next); + world.plugin("agent", consumer("notes", old), document("consume")); + let host = world.start().await; + activate(&host, "notes").await.unwrap(); + host.stop().await; +} + +/// PROTOCOL 3.7: an old consumer can still send its old import during grace, even after the provider removes it. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn old_consumer_can_call_removed_provider_method() { + let world = TestWorld::new(); + world.plugin("notes", document("legacy"), document("new")); + let mut old = consumer("notes", document("legacy")); + old["x-sausage-grace"] = json!(60); + world.plugin("agent", old, document("consume")); + let host = world.start().await; + let pid = call(&host, "agent.deferred", json!({})).await; + activate(&host, "agent").await.unwrap(); + activate(&host, "notes").await.unwrap(); + assert_eq!(host.processes()[pid.as_str().unwrap()].state, "running"); + std::fs::write(world.0.join("state/agent/release-call"), "release").unwrap(); + let response: Value = tokio::time::timeout(Duration::from_secs(10), async { + loop { + if let Ok(bytes) = std::fs::read(world.0.join("state/agent/call-result")) + && let Ok(value) = serde_json::from_slice(&bytes) + { + break value; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert_eq!(response["error"]["code"], -32601); + assert_eq!(response["error"]["data"]["version"], "2"); + host.stop().await; +} + +/// PROTOCOL 10.2: activation and revert check interfaces but leave persistent state unchanged. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn state_is_outside_interface_checks() { + let world = TestWorld::new(); + world.plugin("notes", document("add"), document("add")); + let state = world.0.join("state/notes/user-state"); + let contents = b"a state format neither version understands\0"; + std::fs::write(&state, contents).unwrap(); + let host = world.start().await; + activate(&host, "notes").await.unwrap(); + assert_eq!(std::fs::read(&state).unwrap(), contents); + call(&host, "core.revert", json!({"plugin":"notes"})).await; + assert_eq!(std::fs::read(state).unwrap(), contents); + host.stop().await; +} diff --git a/cli/tests/cargo.rs b/cli/tests/cargo.rs index 7463256..e293011 100644 --- a/cli/tests/cargo.rs +++ b/cli/tests/cargo.rs @@ -46,6 +46,11 @@ fn fixture(name: &str) -> PathBuf { /// Builds through the real registered native plugin. async fn build(world: &TestWorld, name: &str) -> Result { let host = Host::start(&world.0).await.unwrap(); + let runner = world.0.join("state/runner with spaces"); + std::os::unix::fs::symlink(env!("CARGO_BIN_EXE_sausage"), &runner).unwrap(); + sausage::cargo::Cargo::with_runner(World::open(&world.0).unwrap(), runner) + .unwrap() + .register(&host); let result = host.call("cargo.build", json!({"plugin":name})).await; host.stop().await; result @@ -68,22 +73,62 @@ async fn valid() -> &'static Valid { "exclude", ) .unwrap(); + let nested = world.0.join("plugins/valid/source/src/target"); + std::fs::create_dir_all(&nested).unwrap(); + std::fs::write(nested.join("data"), "nested target").unwrap(); + let git = world.0.join("plugins/valid/source/src/.git"); + std::fs::create_dir_all(&git).unwrap(); + std::fs::write(git.join("data"), "nested git").unwrap(); + for number in 1..=20 { + std::fs::create_dir_all(world.0.join(format!("state/cargo/builds/{number}"))) + .unwrap(); + } + std::fs::create_dir_all(world.0.join("state/cargo/tmp/stale")).unwrap(); + std::fs::write(world.0.join("state/cargo/tmp/stale/file"), "stale").unwrap(); let result = build(&world, "valid") .await .unwrap_or_else(|error| panic!("build failed: {error}")); - let test = - std::fs::read_dir(world.0.join("state/cargo/target/wasm32-wasip2/debug/deps")) - .unwrap() - .filter_map(|entry| { - let path = entry.ok()?.path(); - (path - .extension() - .is_some_and(|extension| extension == "wasm") - && path.file_name()?.to_str()?.starts_with("cargo_valid-")) - .then_some(path) - }) - .next() - .unwrap(); + // Integration tests also build a normal binary. Cargo identifies the libtest artifact. + let artifacts = tokio::process::Command::new("cargo") + .args([ + "test", + "--no-run", + "--message-format=json", + "--bin", + "cargo-valid", + "--target", + "wasm32-wasip2", + "--target-dir", + ]) + .arg(world.0.join("state/cargo/target")) + .current_dir(world.0.join("state/cargo/work")) + .stdin(std::process::Stdio::null()) + .output() + .await + .unwrap(); + assert!( + artifacts.status.success(), + "{}", + String::from_utf8_lossy(&artifacts.stderr) + ); + let mut tests: Vec = String::from_utf8_lossy(&artifacts.stdout) + .lines() + .filter_map(|line| serde_json::from_str::(line).ok()) + .filter(|message| { + message["reason"] == "compiler-artifact" + && message["profile"]["test"] == true + && message["target"]["kind"] + .as_array() + .is_some_and(|kinds| kinds.iter().any(|kind| kind == "bin")) + }) + .filter_map(|message| message["executable"].as_str().map(PathBuf::from)) + .collect(); + assert_eq!( + tests.len(), + 1, + "expected exactly one libtest binary: {tests:?}" + ); + let test = tests.pop().unwrap(); Valid { world, result, @@ -94,16 +139,215 @@ async fn valid() -> &'static Valid { } /// Invokes the same runner subcommand that cargo uses. async fn runner(valid: &Valid, args: &[&str]) -> std::process::Output { - tokio::process::Command::new(env!("CARGO_BIN_EXE_sausage")) + let mut command = tokio::process::Command::new(env!("CARGO_BIN_EXE_sausage")); + command .arg("test-runner") .arg(&valid.test) .args(args) .current_dir(valid.world.0.join("state/cargo/work")) - .output() + .kill_on_drop(true); + tokio::time::timeout(std::time::Duration::from_secs(120), command.output()) .await + .expect("test runner exceeded 120 seconds") .unwrap() } +/// T23: unlocks stay ordered before the next lock over 300 acquisitions. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn unlock_order() { + assert!( + runner(valid().await, &["unlock_order"]) + .await + .status + .success() + ); +} +/// T23: each runner world is empty and has independent proc, self, and run mounts. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn empty_and_isolated_worlds() { + for name in ["empty_world", "isolation_", "environment", "late_response"] { + let output = runner(valid().await, &[name]).await; + assert!( + output.status.success(), + "{name}: {} {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + } +} +/// T23: successful builds preserve warnings and integration output without cargo progress noise. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn build_output_and_warnings() { + let valid = valid().await; + assert!( + valid.result["warnings"] + .as_str() + .unwrap() + .contains("fixture build warning") + ); + let output = valid.result["tests"]["output"].as_str().unwrap(); + assert!(output.contains("INTEGRATION_CASE")); + assert!(!output.contains("Compiling ")); + assert!(!output.contains(" running 1 test")); + assert!( + !output + .lines() + .any(|line| line.starts_with(" ") && line.trim().is_empty()) + ); +} +/// T23: publication retains ten builds and the next build removes abandoned test worlds. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn build_retention_and_temporary_cleanup() { + let valid = valid().await; + assert_eq!( + std::fs::read_dir(valid.world.0.join("state/cargo/builds")) + .unwrap() + .count(), + 10 + ); + assert!(!valid.world.0.join("state/cargo/tmp/stale").exists()); + assert_eq!( + std::fs::read_dir(valid.world.0.join("state/cargo/tmp")) + .unwrap() + .count(), + 0 + ); +} +/// T23: manifest preparation preserves workspaces, renamed target dependencies, and optional flags. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn manifest_and_nested_sources() { + let valid = valid().await; + let root = valid.world.0.join("state/cargo/work"); + let manifest = std::fs::read_to_string(root.join("Cargo.toml")).unwrap(); + assert!(manifest.contains("workspace.package")); + assert!(manifest.contains("optional = true")); + assert!(manifest.contains("/plugins/cargo/sdk")); + assert_eq!( + std::fs::read_to_string(root.join("src/target/data")).unwrap(), + "nested target" + ); + assert_eq!( + std::fs::read_to_string(root.join("src/.git/data")).unwrap(), + "nested git" + ); +} +/// T23: discovery describes the native build method and preparation errors name their paths. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn native_discovery_and_error_paths() { + let world = TestWorld::new(); + let host = Host::start(&world.0).await.unwrap(); + assert_eq!( + host.call("cargo.rpc.discover", json!({})).await.unwrap()["methods"][0]["name"], + "build" + ); + let error = host + .call("cargo.build", json!({"plugin":"missing"})) + .await + .unwrap_err(); + assert_eq!(error["code"], -32004); + assert!( + error["data"]["output"] + .as_str() + .unwrap() + .contains("plugins/missing/source") + ); + host.stop().await; +} +/// T23: release-only failures use data.output and retain successful WASI totals. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn release_only_failure() { + let world = TestWorld::new(); + let root = world.0.join("plugins/release/source/src"); + std::fs::create_dir_all(&root).unwrap(); + std::fs::write( + root.parent().unwrap().join("Cargo.toml"), + "[package]\nname='release-only'\nversion='0.1.0'\nedition='2024'\n", + ) + .unwrap(); + std::fs::write(root.join("main.rs"), "#[cfg(not(debug_assertions))] compile_error!(\"release-only error\"); fn main() {} #[test] fn passes() {}\n").unwrap(); + let error = build(&world, "release").await.unwrap_err(); + assert!( + error["data"]["output"] + .as_str() + .unwrap() + .contains("release-only error") + ); + assert_eq!(error["data"]["tests"]["passed"], 1); +} +/// T23: cancellation retains serialization until a blocking source copy completes. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn cancelled_preparation_is_serialized() { + let world = TestWorld::new(); + let first = world.0.join("first"); + let second = world.0.join("second"); + for (root, name) in [(&first, "first"), (&second, "second")] { + std::fs::create_dir_all(root.join("src")).unwrap(); + std::fs::write( + root.join("Cargo.toml"), + format!("[package]\nname='{name}'\nversion='0.1.0'\nedition='2024'\n"), + ) + .unwrap(); + std::fs::write(root.join("src/main.rs"), "fn main() {}\n").unwrap(); + } + for number in 0..10000 { + std::fs::write(first.join(format!("old-{number}")), "copy").unwrap(); + } + let builder = std::sync::Arc::new( + sausage::cargo::Cargo::with_runner( + World::open(&world.0).unwrap(), + PathBuf::from(env!("CARGO_BIN_EXE_sausage")), + ) + .unwrap(), + ); + let copying = builder.clone(); + let task = tokio::spawn(async move { copying.build(&first).await }); + tokio::time::timeout(std::time::Duration::from_secs(30), async { + loop { + if std::fs::read_dir(world.0.join("state/cargo/work")) + .is_ok_and(|entries| entries.count() > 10) + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + task.abort(); + let _ = task.await; + builder.build(&second).await.unwrap(); + assert!( + !std::fs::read_dir(world.0.join("state/cargo/work")) + .unwrap() + .any(|entry| entry + .unwrap() + .file_name() + .to_string_lossy() + .starts_with("old-")) + ); +} +/// T23: multiple binary artifacts fail clearly instead of publishing an arbitrary binary. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn multiple_binaries() { + let world = TestWorld::new(); + let root = world.0.join("plugins/multi/source/src/bin"); + std::fs::create_dir_all(&root).unwrap(); + std::fs::write( + root.parent().unwrap().parent().unwrap().join("Cargo.toml"), + "[package]\nname='multiple'\nversion='0.1.0'\nedition='2024'\n", + ) + .unwrap(); + for name in ["one", "two"] { + std::fs::write(root.join(format!("{name}.rs")), "fn main() {}\n").unwrap(); + } + let error = build(&world, "multi").await.unwrap_err(); + assert!( + error["data"]["output"] + .as_str() + .unwrap() + .contains("exactly one binary") + ); +} /// PROTOCOL 11: a valid SDK plugin builds and publishes a release component without activation. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn valid_build() { @@ -111,7 +355,7 @@ async fn valid_build() { let wasm = valid.result["wasm"].as_str().unwrap(); assert!(wasm.starts_with("/state/cargo/builds/")); assert!(valid.world.0.join(wasm.trim_start_matches('/')).is_file()); - assert_eq!(valid.result["tests"]["passed"], 5); + assert_eq!(valid.result["tests"]["passed"], 12); assert_eq!(valid.result["tests"]["failed"], 0); assert!(!valid.world.0.join("plugins/valid/plugin.wasm").exists()); } @@ -305,6 +549,10 @@ async fn standalone_builder() { /// PROTOCOL 11: killing a native build also kills its build-script descendants. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn kill_process_group() { + assert!( + Path::new("/proc/self/stat").is_file(), + "process-group checks require /proc" + ); let world = TestWorld::new(); world.source("hang", &fixture("cargo-hang")); let host = Host::start(&world.0).await.unwrap(); @@ -323,21 +571,68 @@ async fn kill_process_group() { }) .await .unwrap(); + let mut ancestor = child; + loop { + let command = std::fs::read(format!("/proc/{ancestor}/cmdline")).unwrap(); + let arguments: Vec<_> = command + .split(|byte| *byte == 0) + .map(|part| String::from_utf8_lossy(part).to_string()) + .collect(); + if arguments[0].ends_with("/cargo") || arguments[0] == "cargo" { + assert!(!arguments.iter().any(|arg| arg == "--offline")); + assert!( + arguments + .iter() + .any(|arg| arg.starts_with("target.wasm32-wasip2.runner=[")) + ); + assert_eq!( + std::fs::read_link(format!("/proc/{ancestor}/fd/0")).unwrap(), + PathBuf::from("/dev/null") + ); + break; + } + let status = std::fs::read_to_string(format!("/proc/{ancestor}/status")).unwrap(); + ancestor = status + .lines() + .find_map(|line| line.strip_prefix("PPid:")) + .unwrap() + .trim() + .parse() + .unwrap(); + assert!(ancestor > 1, "cargo ancestor not found"); + } let pid = host .processes() .into_iter() .find(|(_, process)| process.plugin == "cargo") .unwrap() .0; + let fs = host.filesystem().await.unwrap(); + use sausage_vfs::Layer; + assert!( + fs.stat(&format!("/proc/{pid}/info.json"), true) + .await + .is_ok() + ); + assert!( + !host + .call("core.log", json!({"pid":pid,"grep":"start build"})) + .await + .unwrap() + .as_array() + .unwrap() + .is_empty() + ); host.call("core.kill", json!({"pid":pid})).await.unwrap(); assert!(task.await.unwrap().is_err()); tokio::time::timeout(std::time::Duration::from_secs(10), async { loop { let status = std::fs::read_to_string(format!("/proc/{child}/stat")); - if status - .as_ref() - .map_or(true, |status| status.split_whitespace().nth(2) == Some("Z")) - { + if match status { + Ok(status) => status.split_whitespace().nth(2) == Some("Z"), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => true, + Err(error) => panic!("read /proc/{child}/stat: {error}"), + } { break; } tokio::task::yield_now().await; diff --git a/cli/tests/cli.rs b/cli/tests/cli.rs index df65dd4..9e08bf1 100644 --- a/cli/tests/cli.rs +++ b/cli/tests/cli.rs @@ -1,39 +1,120 @@ -//! Initialize the embedded agent and start the resulting world through the CLI. -use sausage::Host; +//! Binary contracts for atomic init, environment flags, signals, and the piped console. +use sausage::{Host, world::World}; use serde_json::{Value, json}; -use std::process::{Command, Stdio}; +use std::{ + path::{Path, PathBuf}, + process::Stdio, + sync::OnceLock, + time::Duration, +}; +use tokio::{ + io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, + process::{Child, Command}, +}; -/// Environment configuration creates a complete world with a callable agent. +fn binary() -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_sausage")); + command.env_remove("SAUSAGE_HEADLESS").kill_on_drop(true); + command +} +fn world() -> (tempfile::TempDir, PathBuf) { + let folder = tempfile::tempdir().unwrap(); + let world = World::create(&folder.path().join("world")).unwrap(); + (folder, world.root) +} +async fn status(child: &mut Child) -> std::process::ExitStatus { + tokio::time::timeout(Duration::from_secs(15), child.wait()) + .await + .expect("CLI did not exit") + .unwrap() +} +async fn journal_exists(world: &Path) { + tokio::time::timeout(Duration::from_secs(15), async { + while !world.join("state/core/journal.sqlite").exists() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); +} +fn signal(child: &Child, signal: nix::sys::signal::Signal) { + nix::sys::signal::kill( + nix::unistd::Pid::from_raw(child.id().unwrap() as i32), + 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(); +} + +/// A failed agent build leaves no world. Bare sausage retries and publishes a complete world. #[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")) + let temporary = tempfile::tempdir().unwrap(); + let world = temporary.path().join("world"); + let configured = || { + let mut command = binary(); + command + .current_dir(temporary.path()) + .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()); + command + }; + let failed = configured() .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()) + .env("RUSTC", "/nonexistent-sausage-rustc") .output() + .await .unwrap(); + assert!(!failed.status.success()); + assert!(!world.exists()); + let output = configured().output().await.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")); + assert!( + String::from_utf8_lossy(&output.stdout).contains("first agent build takes a few minutes") + ); 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!(String::from_utf8_lossy(&output.stdout).contains("sausage run")); assert_eq!( serde_json::from_slice::(&std::fs::read(world.join("config/boot.json")).unwrap()) .unwrap(), @@ -48,9 +129,16 @@ async fn init_builds_agent() { ] { 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(); + assert_eq!( + std::fs::read(world.join("plugins/agent/source/Cargo.lock")).unwrap(), + include_bytes!("../../Cargo.lock") + ); + let host = Host::start_with_cache( + &world, + &Path::new(env!("CARGO_MANIFEST_DIR")).join("../target/core/cli-cache"), + ) + .await + .unwrap(); let doc = host.call("agent.rpc.discover", json!({})).await.unwrap(); assert_eq!(doc["x-sausage-grace"], 600); assert!( @@ -61,6 +149,200 @@ async fn init_builds_agent() { .any(|method| method["name"] == "chat") ); host.stop().await; +} - std::fs::remove_dir_all(temporary).unwrap(); +/// Both headless forms ignore EOF. False and empty values preserve the plain console. +#[tokio::test] +async fn headless_flag_and_environment() { + for value in [None, Some("1"), Some("true")] { + let (_folder, world) = world(); + let mut command = binary(); + command.arg("run").arg(&world).stdin(Stdio::null()); + if let Some(value) = value { + command.env("SAUSAGE_HEADLESS", value); + } else { + command.arg("--headless"); + } + let mut child = command.spawn().unwrap(); + journal_exists(&world).await; + assert!( + tokio::time::timeout(Duration::from_millis(300), child.wait()) + .await + .is_err() + ); + signal(&child, nix::sys::signal::Signal::SIGTERM); + assert_eq!(status(&mut child).await.code(), Some(0)); + } + for value in ["0", ""] { + let (_folder, world) = world(); + let mut child = binary() + .arg("run") + .arg(world) + .env("SAUSAGE_HEADLESS", value) + .stdin(Stdio::null()) + .spawn() + .unwrap(); + assert_eq!(status(&mut child).await.code(), Some(0)); + } +} + +/// Signals interrupt an unanswered boot service instead of waiting forever. +#[tokio::test] +async fn signals_work_during_boot() { + for sig in [ + nix::sys::signal::Signal::SIGTERM, + nix::sys::signal::Signal::SIGINT, + ] { + let (_folder, world) = world(); + std::fs::write( + world.join("config/boot.json"), + json!([ + {"method":"core.lock","params":{"path":"boot"}}, + {"method":"core.lock","params":{"path":"boot"}} + ]) + .to_string(), + ) + .unwrap(); + let mut child = binary() + .arg("run") + .arg(&world) + .arg("--headless") + .stdin(Stdio::null()) + .spawn() + .unwrap(); + journal_exists(&world).await; + tokio::time::timeout(Duration::from_secs(15), async { + loop { + let db = + rusqlite::Connection::open(world.join("state/core/journal.sqlite")).unwrap(); + if let Ok(2) = db.query_row( + "SELECT count(*) FROM journal WHERE line='service start core.lock'", + [], + |row| row.get::<_, i64>(0), + ) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + signal(&child, sig); + assert_eq!(status(&mut child).await.code(), Some(0)); + } +} + +/// Piped input supports commands, quit, and the restart exit code. +#[tokio::test] +async fn piped_console_and_exit_codes() { + for (line, expected) in [("help\n", 0), ("quit\n", 0), ("core.exit\n", 75)] { + let (_folder, world) = world(); + let mut child = binary() + .arg("run") + .arg(world) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .spawn() + .unwrap(); + child + .stdin + .take() + .unwrap() + .write_all(line.as_bytes()) + .await + .unwrap(); + let output = tokio::time::timeout(Duration::from_secs(15), child.wait_with_output()) + .await + .unwrap() + .unwrap(); + assert_eq!(output.status.code(), Some(expected)); + if expected == 0 { + assert!(!output.stdout.is_empty()); + } + } +} + +/// A console input error still stops live actors and flushes the journal. +#[tokio::test] +async fn console_error_stops_host() { + let (_folder, world) = world(); + install_probe(&world); + std::fs::write( + world.join("config/boot.json"), + json!([{"method":"probe.serve","params":{}}]).to_string(), + ) + .unwrap(); + let mut child = binary() + .arg("run") + .arg(&world) + .stdin(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + child + .stdin + .take() + .unwrap() + .write_all(&[255, 10]) + .await + .unwrap(); + assert_eq!(status(&mut child).await.code(), Some(1)); + let db = rusqlite::Connection::open(world.join("state/core/journal.sqlite")).unwrap(); + let stopped: i64 = db + .query_row( + "SELECT count(*) FROM journal WHERE plugin='probe' AND line='stop'", + [], + |row| row.get(0), + ) + .unwrap(); + assert!(stopped > 0); +} + +/// Ps and kill remain available before a slow command returns. +#[tokio::test] +async fn piped_console_accepts_commands_during_a_call() { + 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(); + let mut output = BufReader::new(child.stdout.take().unwrap()).lines(); + input.write_all(b"probe.busy\n").await.unwrap(); + let mut pid = None; + tokio::time::timeout(Duration::from_secs(15), async { + while pid.is_none() { + input.write_all(b"ps\n").await.unwrap(); + while let Some(line) = output.next_line().await.unwrap() { + if line.contains("probe busy console") { + pid = line.split_whitespace().next().map(str::to_owned); + break; + } + if line.contains("PID PLUGIN METHOD") { + break; + } + } + } + }) + .await + .unwrap(); + input + .write_all(format!("kill {}\n", pid.unwrap()).as_bytes()) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(15), async { + while let Some(line) = output.next_line().await.unwrap() { + if line.starts_with("> kill ") { + break; + } + } + }) + .await + .unwrap(); + input.write_all(b"quit\n").await.unwrap(); + assert_eq!(status(&mut child).await.code(), Some(0)); } diff --git a/cli/tests/console.rs b/cli/tests/console.rs index 78becc3..d5fb6c7 100644 --- a/cli/tests/console.rs +++ b/cli/tests/console.rs @@ -14,7 +14,7 @@ fn fixture() -> PathBuf { .parent() .unwrap() .to_owned(); - let target = root.join("target/vfs/console-guests"); + let target = root.join("target/core/console-guests"); let status = std::process::Command::new("cargo") .args([ "build", @@ -54,7 +54,7 @@ impl TestWorld { let cache = PathBuf::from(env!("CARGO_MANIFEST_DIR")) .parent() .unwrap() - .join("target/vfs/console-cache"); + .join("target/core/console-cache"); Host::start_with_cache(&self.0, &cache).await.unwrap() } } @@ -78,7 +78,10 @@ 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!(execute_line(&host, "probe.echo not-json").await.is_err()); + assert_eq!( + command(&host, "probe.echo not-json").await, + "probe.echo not-json" + ); host.stop().await; } /// Plain text calls agent.chat and its interim notification uses dim styling. @@ -96,6 +99,28 @@ async fn chat() { ); host.stop().await; } +/// Invalid command arguments and dotted prose remain agent chat. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn prose_is_not_a_command() { + let world = TestWorld::new(); + let host = world.start().await; + for line in [ + "help me write a notes plugin", + "ps is useful", + "stop it", + "kill that", + "cat", + "methods probe extra", + "ls too many words", + "log -n nope", + "README.md says hello", + "e.g. write notes", + ] { + assert_eq!(command(&host, line).await, line); + } + host.stop().await; +} + /// Folder listings include the live mounts and accept an explicit folder. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn ls() { @@ -215,7 +240,7 @@ async fn log() { } } assert_eq!(command(&host, "log absent-plugin").await, ""); - assert!(execute_line(&host, "log -n").await.is_err()); + assert_eq!(command(&host, "log -n").await, "log -n"); host.stop().await; } /// Activate installs a candidate through core.activate. diff --git a/cli/tests/core.rs b/cli/tests/core.rs index 70bb829..25b1683 100644 --- a/cli/tests/core.rs +++ b/cli/tests/core.rs @@ -1438,7 +1438,7 @@ async fn core_lifetime_events_enter_the_journal() { let h = w.start().await; let stopped = call(&h, "probe.serve", json!({"n":1})).await; call(&h, "core.stop", json!({"pid":stopped})).await; - let killed = call(&h, "probe.serve", json!({"n":2})).await; + let killed = call(&h, "probe.tree", json!({"depth":0})).await; call(&h, "core.kill", json!({"pid":killed})).await; let entries = call(&h, "core.log", json!({"plugin":"probe","limit":100})).await; let lines: Vec<_> = entries @@ -1706,3 +1706,254 @@ async fn filesystem_native_processes_and_lock_snapshot() { assert!(view.stat(&format!("/proc/{pid}"), true).await.is_err()); h.stop().await; } + +/// PROTOCOL 2: invalid plugin names cannot route calls, and core cannot be shadowed. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn plugin_names_and_core_reservation() { + let w = TestWorld::new(); + let h = w.start().await; + for name in ["Upper", "bad!", "with space", "é", "", "../probe"] { + h.register_native( + name, + std::sync::Arc::new(|_, _, _| Box::pin(async { Ok(json!("bad")) })), + ); + assert_eq!( + h.call(&format!("{name}.echo"), json!({})) + .await + .unwrap_err()["code"], + -32601 + ); + } + h.register_native( + "core", + std::sync::Arc::new(|_, _, _| Box::pin(async { Ok(json!("bad")) })), + ); + assert_eq!( + h.call("core.unknown", json!({})).await.unwrap_err()["code"], + -32601 + ); + for name in ["a-z_09", "0"] { + h.register_native( + name, + std::sync::Arc::new(|_, _, _| Box::pin(async { Ok(json!(true)) })), + ); + assert_eq!(call(&h, &format!("{name}.echo"), json!({})).await, true); + } + h.stop().await; +} + +/// PROTOCOL 3.1: a batch on guest stdout does not execute any of its requests. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn batches_are_discarded() { + let w = TestWorld::new(); + let h = w.start().await; + assert_eq!(call(&h, "probe.batch", json!({})).await, true); + h.stop().await; + assert!(!w.0.join("state/batch").exists()); + assert_eq!( + call(&h, "core.log", json!({"grep":"discarded stdout"})) + .await + .as_array() + .unwrap() + .len(), + 1 + ); +} + +/// PROTOCOL 3.3: a second response to an answered call cannot replace its result. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn late_responses_are_dropped() { + let w = TestWorld::new(); + let h = w.start().await; + assert_eq!(call(&h, "probe.late_answer", json!({})).await, 42); + assert_eq!( + call(&h, "probe.echo", json!({"next":true})).await, + json!({"next":true}) + ); + h.stop().await; +} + +/// PROTOCOL 3.4: core.call notifications forget detached children after the caller ends. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn notification_core_call_forgets_detached_child() { + let w = TestWorld::new(); + let h = w.start().await; + let parent = call(&h, "probe.forget", json!({})).await; + let child: Value = tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + if let Ok(bytes) = std::fs::read(w.0.join("state/forgotten")) + && let Ok(value) = serde_json::from_slice::(&bytes) + { + break value; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(10), async { + while h.processes().contains_key(parent["pid"].as_str().unwrap()) { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(h.processes()[child["pid"].as_str().unwrap()].detached); + assert!( + h.processes()[child["pid"].as_str().unwrap()] + .parent + .is_none() + ); + call(&h, "core.stop", json!({"pid":child["pid"]})).await; + assert!(!h.processes().contains_key(child["pid"].as_str().unwrap())); + h.stop().await; +} + +/// PROTOCOL 3.6: three generations that ignore EOF stop together within one five-second deadline. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn attached_tree_stops_in_parallel() { + let w = TestWorld::new(); + let h = w.start().await; + let root = call(&h, "probe.tree", json!({"depth":2})).await; + assert_eq!( + h.processes() + .values() + .filter(|p| p.method == "tree") + .count(), + 3 + ); + let before = std::time::Instant::now(); + call(&h, "core.stop", json!({"pid":root})).await; + tokio::time::timeout(std::time::Duration::from_secs(2), async { + while h.processes().values().any(|p| p.method == "tree") { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(before.elapsed() >= std::time::Duration::from_millis(4900)); + assert!(before.elapsed() < std::time::Duration::from_secs(7)); + h.stop().await; +} + +/// PROTOCOL 3.8: a nested call receives its caller's process ID as SAUSAGE_PARENT. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn nested_parent_environment() { + let w = TestWorld::new(); + let h = w.start().await; + let result = call(&h, "probe.nested_environment", json!({})).await; + assert_eq!(result["child"]["parent"], result["caller"]); + assert_ne!(result["child"]["pid"], result["caller"]); + h.stop().await; +} + +/// PROTOCOL 8 and 3.8: boot services receive core as SAUSAGE_PARENT. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn service_parent_environment() { + let w = TestWorld::new(); + std::fs::write( + w.0.join("config/boot.json"), + br#"[{"method":"probe.forgotten","params":{}}]"#, + ) + .unwrap(); + let h = w.start().await; + let env: Value = + serde_json::from_slice(&std::fs::read(w.0.join("state/forgotten")).unwrap()).unwrap(); + assert_eq!(env["parent"], "core"); + assert_eq!(env["plugin"], "probe"); + assert!(h.processes()[env["pid"].as_str().unwrap()].detached); + h.stop().await; +} + +/// PROTOCOL 4: info.json exposes every field of the live process snapshot. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn info_json_has_all_process_fields() { + let w = TestWorld::new(); + let h = w.start().await; + let params = json!({"marker":"complete"}); + let pid = call(&h, "probe.serve", params.clone()).await; + let text = call( + &h, + "probe.read", + json!({"path":format!("/proc/{}/info.json", pid.as_str().unwrap())}), + ) + .await; + let info: Value = serde_json::from_str(text.as_str().unwrap()).unwrap(); + let expected = serde_json::to_value(&h.processes()[pid.as_str().unwrap()]).unwrap(); + assert_eq!(info, expected); + for field in [ + "plugin", + "version", + "method", + "params", + "parent", + "detached", + "start_time", + "state", + ] { + assert!(info.get(field).is_some(), "{field}"); + } + assert_eq!(info["params"], params); + assert_eq!(info["parent"], "console"); + assert_eq!(info["state"], "running"); + h.stop().await; +} + +/// PROTOCOL 3.6: detached processes persist until stop, kill, or host shutdown explicitly ends them. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn detached_processes_need_explicit_termination() { + let w = TestWorld::new(); + let h = w.start().await; + let mut pids = Vec::new(); + for _ in 0..3 { + let pid = call( + &h, + "core.call", + json!({"method":"probe.serve","params":{},"detach":true}), + ) + .await; + assert!(h.processes()[pid.as_str().unwrap()].parent.is_none()); + pids.push(pid); + } + call(&h, "probe.echo", json!({})).await; + for pid in &pids { + assert!(h.processes().contains_key(pid.as_str().unwrap())); + } + call(&h, "core.stop", json!({"pid":pids[0]})).await; + call(&h, "core.kill", json!({"pid":pids[1]})).await; + assert!(h.processes().contains_key(pids[2].as_str().unwrap())); + h.stop().await; + for pid in pids { + assert!(!h.processes().contains_key(pid.as_str().unwrap())); + } +} + +/// PROTOCOL 9: core.exit from boot ends the host executable with status 75. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn core_exit_sets_executable_status_75() { + let w = TestWorld::new(); + std::fs::write( + w.0.join("config/boot.json"), + br#"[{"method":"core.exit","params":{}}]"#, + ) + .unwrap(); + let output = tokio::time::timeout( + std::time::Duration::from_secs(30), + tokio::process::Command::new(env!("CARGO_BIN_EXE_sausage")) + .arg("--headless") + .arg("run") + .arg(&w.0) + .stdin(std::process::Stdio::null()) + .kill_on_drop(true) + .output(), + ) + .await + .unwrap() + .unwrap(); + assert_eq!( + output.status.code(), + Some(75), + "{}", + String::from_utf8_lossy(&output.stderr) + ); +} diff --git a/cli/tests/crates/cargo-valid/Cargo.toml b/cli/tests/crates/cargo-valid/Cargo.toml index 9dfbb1b..0cbbca2 100644 --- a/cli/tests/crates/cargo-valid/Cargo.toml +++ b/cli/tests/crates/cargo-valid/Cargo.toml @@ -1,9 +1,15 @@ [package] name = "cargo-valid" version = "0.1.0" -edition = "2024" +edition.workspace = true -[dependencies] -sausage-plugin = "0.1" +[target.'cfg(target_arch = "wasm32")'.dependencies] +sausage_plugin = { package = "sausage-plugin", version = "0.1", optional = true } + +[features] +default = ["dep:sausage_plugin"] [workspace] + +[workspace.package] +edition = "2024" diff --git a/cli/tests/crates/cargo-valid/build.rs b/cli/tests/crates/cargo-valid/build.rs new file mode 100644 index 0000000..7bdd766 --- /dev/null +++ b/cli/tests/crates/cargo-valid/build.rs @@ -0,0 +1,4 @@ +//! Produces a warning so the build response can retain diagnostics. +fn main() { + println!("cargo:warning=fixture build warning"); +} diff --git a/cli/tests/crates/cargo-valid/src/main.rs b/cli/tests/crates/cargo-valid/src/main.rs index 28ec5a8..aa81b93 100644 --- a/cli/tests/crates/cargo-valid/src/main.rs +++ b/cli/tests/crates/cargo-valid/src/main.rs @@ -34,6 +34,12 @@ mod tests { async fn sdk_lock() { let guard = sausage_plugin::lock("/state/lock").await.unwrap(); assert!(guard.held()); + let locks: sausage_plugin::serde_json::Value = sausage_plugin::serde_json::from_str( + &std::fs::read_to_string("/proc/core/locks.json").unwrap(), + ) + .unwrap(); + assert_eq!(locks.as_object().unwrap().len(), 1); + assert!(locks.get("/state/lock").is_some()); let mut guard = Some(guard); let mut waiting = Box::pin(std::future::IntoFuture::into_future(sausage_plugin::lock( "/state/lock", @@ -66,6 +72,80 @@ mod tests { .unwrap_err(); assert!(error.to_string().contains("a test has no other plugins")); } + #[sausage_plugin::test] + async fn unlock_order() { + for iteration in 0..300 { + let guard = sausage_plugin::lock("/state/order").await.unwrap(); + assert!(guard.held()); + drop(guard); + let guard = sausage_plugin::lock("/state/order") + .try_now() + .await + .unwrap(); + assert!(guard.held(), "iteration {iteration}"); + } + } + #[test] + fn empty_world() { + for path in [ + "/plugins", + "/config", + "/state", + "/state/core/journal.sqlite", + ] { + assert!(!std::path::Path::new(path).exists(), "{path}"); + } + let names: Vec<_> = std::fs::read_dir("/") + .unwrap() + .map(|entry| entry.unwrap().file_name()) + .collect(); + assert!(names.iter().all(|name| { + ["proc", "self", "run", "tmp"] + .iter() + .any(|allowed| name == allowed) + })); + for path in ["/proc", "/self", "/run"] { + assert!(std::path::Path::new(path).is_dir()); + } + assert!(std::fs::write("/proc/forbidden", "no").is_err()); + } + fn isolated() { + assert!(!std::path::Path::new("/self/own").exists()); + assert!(!std::path::Path::new("/run/own").exists()); + std::fs::write("/self/own", "self").unwrap(); + std::fs::write("/run/own", "run").unwrap(); + assert_eq!(std::fs::read_to_string("/self/own").unwrap(), "self"); + } + #[test] + fn isolation_one() { + isolated(); + } + #[test] + fn isolation_two() { + isolated(); + } + #[test] + fn environment() { + for (key, _) in std::env::vars() { + assert!( + ["CARGO_", "RUST_", "SAUSAGE_"] + .iter() + .any(|prefix| key.starts_with(prefix)), + "{key}" + ); + } + assert!(std::env::var("PWD").is_err()); + } + #[test] + fn late_response() { + use std::io::Write; + for id in 0..100 { + println!( + "\n{{\"jsonrpc\":\"2.0\",\"id\":{id},\"method\":\"core.log\",\"params\":{{}}}}" + ); + } + std::io::stdout().flush().unwrap(); + } #[test] #[ignore] fn ignored_case() { diff --git a/cli/tests/crates/cargo-valid/tests/integration.rs b/cli/tests/crates/cargo-valid/tests/integration.rs new file mode 100644 index 0000000..1b8cc50 --- /dev/null +++ b/cli/tests/crates/cargo-valid/tests/integration.rs @@ -0,0 +1,5 @@ +//! Checks that cargo invokes integration test binaries through the WASI runner. +#[test] +fn integration_case() { + println!("INTEGRATION_CASE"); +} diff --git a/cli/tests/plugins/activation/src/main.rs b/cli/tests/plugins/activation/src/main.rs index f6c6e34..8924562 100644 --- a/cli/tests/plugins/activation/src/main.rs +++ b/cli/tests/plugins/activation/src/main.rs @@ -6,6 +6,38 @@ fn send(value: Value) { println!("{value}"); io::stdout().flush().unwrap(); } +/// Calls a provider after activation releases an old consumer process. +fn deferred(request: &Value, plugin: &str, lines: &mut impl Iterator>) { + send( + json!({"jsonrpc":"2.0","id":request["id"],"result":std::env::var("SAUSAGE_PID").unwrap()}), + ); + while !std::path::Path::new(&format!("/state/{plugin}/release-call")).exists() { + std::thread::sleep(std::time::Duration::from_millis(1)); + } + send(json!({"jsonrpc":"2.0","id":"deferred","method":"notes.legacy","params":{}})); + for line in lines { + let response: Value = serde_json::from_str(&line.unwrap()).unwrap(); + if response["id"] == "deferred" { + std::fs::write(format!("/state/{plugin}/call-result"), response.to_string()).unwrap(); + break; + } + } +} +/// Answers a method that version two removes from its interface. +fn legacy(request: &Value, document: &Value, version: &str) { + if document["methods"] + .as_array() + .unwrap() + .iter() + .any(|m| m["name"] == "legacy") + { + send(json!({"jsonrpc":"2.0","id":request["id"],"result":version})); + } else { + send( + json!({"jsonrpc":"2.0","id":request["id"],"error":{"code":-32601,"message":"legacy removed","data":{"version":version}}}), + ); + } +} /// Reads a version-specific discovery document without the SDK. fn main() { let version = env!("FIXTURE_VERSION"); @@ -16,6 +48,14 @@ fn main() { serde_json::from_slice(&std::fs::read(format!("/state/{plugin}/{version}.json")).unwrap()) .unwrap(); let result = match request["method"].as_str().unwrap() { + "deferred" => { + deferred(&request, &plugin, &mut lines); + return; + } + "legacy" => { + legacy(&request, &document, version); + return; + } "rpc.discover" => { if document["x-test-wait"] == true { std::fs::write(format!("/state/{plugin}/discover-started"), "ready").unwrap(); diff --git a/cli/tests/plugins/probe/src/main.rs b/cli/tests/plugins/probe/src/main.rs index efd2339..c4ab3ca 100644 --- a/cli/tests/plugins/probe/src/main.rs +++ b/cli/tests/plugins/probe/src/main.rs @@ -324,9 +324,75 @@ fn notification_child(lines: &mut Lines) -> Value { } json!(count) } +fn batch(lines: &mut Lines) -> Value { + send( + json!([{"jsonrpc":"2.0","method":"probe.write","params":{"path":"/state/batch","text":"bad"}}]), + ); + outgoing_call( + &json!({"method":"probe.echo","params":{"barrier":true}}), + lines, + ); + json!(!std::path::Path::new("/state/batch").exists()) +} +fn late_answer(request: &Value, lines: &mut Lines) { + send(json!({"jsonrpc":"2.0","id":request["id"],"result":42})); + send(json!({"jsonrpc":"2.0","id":request["id"],"result":99})); + for _ in lines {} +} +fn nested_environment(lines: &mut Lines) -> Value { + let child = outgoing_call(&json!({"method":"probe.env","params":{}}), lines); + json!({"caller":std::env::var("SAUSAGE_PID").unwrap(),"child":child}) +} +fn forgotten(request: &Value, lines: &mut Lines) { + std::fs::write("/state/forgotten", environment().to_string()).unwrap(); + serve(request, lines); +} +fn forget(lines: &mut Lines) -> Value { + send( + json!({"jsonrpc":"2.0","method":"core.call","params":{"method":"probe.forgotten","params":{},"detach":true}}), + ); + while !std::path::Path::new("/state/forgotten").exists() { + std::thread::sleep(std::time::Duration::from_millis(1)); + } + send(json!({"jsonrpc":"2.0","id":"barrier","method":"probe.echo","params":{}})); + let response = receive(lines); + assert_eq!( + response["id"], "barrier", + "a forgotten call must not send a result" + ); + environment() +} +fn tree(request: &Value, lines: &mut Lines) { + let depth = request["params"]["depth"].as_u64().unwrap(); + if depth > 0 { + outgoing_call( + &json!({"method":"probe.tree","params":{"depth":depth-1}}), + lines, + ); + } + send( + json!({"jsonrpc":"2.0","id":request["id"],"result":std::env::var("SAUSAGE_PID").unwrap()}), + ); + busy(); +} fn dispatch(request: &Value, lines: &mut Lines) -> Option { let params = &request["params"]; let result = match request["method"].as_str().unwrap() { + "batch" => batch(lines), + "nested_environment" => nested_environment(lines), + "forget" => forget(lines), + "forgotten" => { + forgotten(request, lines); + return None; + } + "tree" => { + tree(request, lines); + return None; + } + "late_answer" => { + late_answer(request, lines); + return None; + } "invalid_stderr" => { io::stderr().write_all(&[255, 10]).unwrap(); eprintln!("after invalid UTF-8"); diff --git a/cli/tests/sdk.rs b/cli/tests/sdk.rs index 40bc77e..22cfa2a 100644 --- a/cli/tests/sdk.rs +++ b/cli/tests/sdk.rs @@ -246,3 +246,122 @@ async fn secrets_fixed_endpoint() { assert_eq!(call(&host, "consumer.vault", json!({})).await, "hello"); host.stop().await; } + +/// PROTOCOL 3.6: a background task retains its SDK process until the task completes. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn background_task_retains_process() { + let world = TestWorld::new(); + let host = world.start().await; + let pid = call(&host, "probe.background_hold", json!({})).await; + assert!(host.processes().contains_key(pid.as_str().unwrap())); + assert_eq!(call(&host, "notes.list", json!({})).await, json!([])); + assert!(host.processes().contains_key(pid.as_str().unwrap())); + std::fs::write(world.0.join("release-background"), "release").unwrap(); + tokio::time::timeout(Duration::from_secs(10), async { + while host.processes().contains_key(pid.as_str().unwrap()) { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert_eq!( + std::fs::read_to_string(world.0.join("background-ended")).unwrap(), + "done" + ); + host.stop().await; +} + +/// PROTOCOL 5: stale endpoint identities and interfaces fail through genuine SDK processes. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn endpoint_handshake_rejects_mismatch() { + let world = TestWorld::new(); + let host = world.start().await; + let pid = call(&host, "probe.service", json!({})).await; + let path = world.0.join("state/control"); + let original = std::fs::read_to_string(&path).unwrap(); + let words: Vec<_> = original.split_whitespace().collect(); + for changed in [ + format!("{} wrong-owner {}\n", words[0], words[2]), + format!("{} {} wrong-interface\n", words[0], words[1]), + ] { + std::fs::write(&path, changed).unwrap(); + let error = host + .call("typed.world_endpoint", json!({})) + .await + .unwrap_err(); + assert_eq!(error["code"], -32000); + assert!(error["message"].as_str().unwrap().contains("mismatch")); + assert!(host.processes().contains_key(pid.as_str().unwrap())); + } + std::fs::write(path, original).unwrap(); + assert_eq!(call(&host, "typed.world_endpoint", json!({})).await, 7); + host.stop().await; +} + +/// PROTOCOL 10.1 and 5: discovery through core retains params, endpoints, imports, and grace. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn discovery_extensions_through_core() { + let world = TestWorld::new(); + let host = world.start().await; + let provider = call(&host, "probe.rpc.discover", json!({})).await; + assert_eq!(provider["openrpc"], "1.2.6"); + assert_eq!(provider["x-sausage-grace"], 12); + assert!( + provider["x-sausage-endpoints"]["control"]["methods"] + .as_array() + .unwrap() + .iter() + .any(|m| m["name"] == "stats") + ); + let add = provider["methods"] + .as_array() + .unwrap() + .iter() + .find(|m| m["name"] == "add") + .unwrap(); + assert_eq!(add["paramStructure"], "by-name"); + let reference = add["x-sausage-params"]["$ref"].as_str().unwrap(); + assert_eq!( + provider.pointer(reference.trim_start_matches('#')).unwrap()["type"], + "object" + ); + let consumer = call(&host, "typed.rpc.discover", json!({})).await; + assert!( + consumer["x-sausage-imports"]["probe"]["methods"] + .as_array() + .unwrap() + .iter() + .any(|m| m["name"] == "add") + ); + host.stop().await; +} + +/// PROTOCOL 3.2: rpc.discover is SDK control traffic and endpoint control is not a main method. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn reserved_rpc_control_names() { + let world = TestWorld::new(); + let host = world.start().await; + assert!(call(&host, "probe.rpc.discover", json!({})).await["methods"].is_array()); + for method in ["probe.rpc.endpoint", "probe.rpc.unknown"] { + assert_eq!( + host.call(method, json!({})).await.unwrap_err()["code"], + -32601 + ); + } + host.stop().await; +} + +/// PROTOCOL 3.1: the SDK's leading newline isolates an unfinished diagnostic from its response. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn leading_newline_protects_sdk_response() { + let world = TestWorld::new(); + let host = world.start().await; + assert_eq!( + call(&host, "probe.framed_output", json!({})).await, + "intact" + ); + host.stop().await; + let warnings = call(&host, "core.log", json!({"grep":"discarded stdout"})).await; + assert_eq!(warnings.as_array().unwrap().len(), 1); + assert!(!warnings.to_string().contains("partial diagnostic")); +} diff --git a/sdk/tests/plugins/probe/src/main.rs b/sdk/tests/plugins/probe/src/main.rs index 623ee48..e6aae1a 100644 --- a/sdk/tests/plugins/probe/src/main.rs +++ b/sdk/tests/plugins/probe/src/main.rs @@ -123,6 +123,21 @@ impl Probe { }); Ok("started".into()) } + /// Leave an unfinished diagnostic before the SDK writes its framed response. + fn framed_output(&self) -> Result { + print!("partial diagnostic"); + Ok("intact".into()) + } + /// Keep the process alive until the test releases its background task. + fn background_hold(&self) -> Result { + sausage_plugin::background(async { + while !std::path::Path::new("/release-background").exists() { + wstd::task::sleep(wstd::time::Duration::from_millis(10)).await; + } + std::fs::write("/background-ended", "done").unwrap(); + }); + Ok(sausage_plugin::pid()) + } /// Keep a pollable alive for one hour. fn sleepy(&self) -> Result<()> { sausage_plugin::background(async {