surma/sausage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

host: review fixes for cargo.build, the test runner, the console, and main; tests for every PROTOCOL.md rule

Browse Source 
cargo.build and the runner (review T22R): builds use the network again;
core.unlock and core.notify in order under the runner; quiet test
output; manifest preparation keeps [workspace], target tables, renamed
and optional entries; cargo.rpc.discover; -32004 with data.output;
retention of 10 builds; light, empty test worlds with an in-memory
journal; only CARGO_/RUST_/SAUSAGE_ variables; stdin is null; the runner
through --config. A compile cache that cannot be created is not fatal.
Activation replaces plugin.wasm without a gap.

Console and main: init ships the workspace Cargo.lock and creates the
world atomically; each console command runs in its own task, so ps,
stop, and kill work during a slow call; SAUSAGE_HEADLESS; the terminal
mode is restored; signals during boot; chat text is not taken as a
command.

Tests: 20 more tests for PROTOCOL.md sections 2 to 10; the cargo tests
choose the libtest binary from cargo's JSON output and time out.
This commit is contained in:
surma
2026-10-23 07:47:39 +01:00
parent 61d5d5f817
commit a090e1fd57
24 changed files with 1889 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/activation.rs
+5 -1
View File
@@ -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(&current).await.map_err(io_error)? {
tokio::fs::rename(&current, directory.join("plugin.prev.wasm"))
let backup = candidate.copy.with_extension("previous");
tokio::fs::hard_link(&current, &backup)
.await
.map_err(io_error)?;
tokio::fs::rename(&backup, directory.join("plugin.prev.wasm"))
.await
.map_err(io_error)?;
}
cli/src/cargo.rs
+142 -71
View File
@@ -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<anyhow::Error> 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<anyhow::Error> 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<tokio::sync::Mutex<()>>,
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<BuildOutput, BuildFailure> {
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(&copy)?;
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, &copy)?;
std::fs::create_dir_all(&cleanup)
.with_context(|| format!("create {}", cleanup.display()))?;
if copy.exists() {
std::fs::remove_dir_all(&copy)
.with_context(|| format!("remove {}", copy.display()))?;
}
copy_source(&source, &copy, 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::<Value>(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::<Value>(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::<u64>().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<std::process::Output> {
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<std::process::Output> {
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<String> {
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<PathBuf> {
std::fs::create_dir_all(builds).with_context(|| format!("create {}", builds.display()))?;
let mut numbers: Vec<u64> = 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(())
}
cli/src/console.rs
+152 -68
View File
@@ -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<String> {
.collect::<Vec<_>>()
.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<String> {
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::<u64>().ok())
else {
return call(host, "agent.chat", json!({"text":line}), true).await;
};
params["limit"] = json!(count);
} else if params.get("pid").is_none() && params.get("plugin").is_none() {
let key = if words[i].len() > 36
@@ -173,7 +174,7 @@ pub async fn execute_line(host: &Host, line: &str) -> Result<String> {
};
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<String> {
"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<String> {
)
.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<bool>)>,
>();
let (sender, lines) = tokio::sync::mpsc::unbounded_channel::<Result<String>>();
// 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(&notification_text(&value))))?,
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
Err(tokio::sync::broadcast::error::RecvError::Closed) => return Ok(()),
}
}
};
let commands = async {
while let Some(line) = lines.recv().await {
let (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::<String>();
let (sender, lines) = tokio::sync::mpsc::unbounded_channel::<Result<String>>();
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<Result<String>>,
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(&notification_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(&notification_text(&value)))?,
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
Err(tokio::sync::broadcast::error::RecvError::Closed) => return Ok(()),
}
},
result = commands.join_next(), if !commands.is_empty() => {
let (line,result): (String,Result<String>) = result.unwrap()?;
let text = result.unwrap_or_else(|error|format!("error: {error}"));
if !text.is_empty() { print(format!("> {line}\n{}",indent(&text)))?; }
},
line = lines.recv(), if input_open => 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;
}
}
cli/src/host.rs
+20 -8
View File
@@ -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<Runtime>) -> anyhow::Result<Self> {
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<Runtime>, 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) {
cli/src/journal.rs
+6 -1
View File
@@ -23,6 +23,9 @@ pub(crate) struct Journal {
warned: Mutex<HashSet<String>>,
}
impl Journal {
pub(crate) async fn memory() -> anyhow::Result<Self> {
Self::open(PathBuf::from(":memory:")).await
}
pub(crate) async fn open(path: PathBuf) -> anyhow::Result<Self> {
let (sender, receiver) = mpsc::channel();
let (ready, initialized) = oneshot::channel();
@@ -80,7 +83,9 @@ impl Journal {
}
}
fn open_database(path: PathBuf) -> anyhow::Result<Connection> {
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;
cli/src/main.rs
+61 -11
View File
@@ -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<PathBuf>,
@@ -38,6 +38,28 @@ enum Cmd {
args: Vec<String>,
},
}
fn headless(value: &str) -> std::result::Result<bool, String> {
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<nix::sys::termios::Termios>);
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<String> {
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<std::process::ExitCode> {
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)
}
cli/src/runtime.rs
+15 -4
View File
@@ -221,10 +221,21 @@ impl Runtime {
pub fn new(cache: &Path) -> Result<Arc<Self>> {
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<ProcessState> = Linker::new(&engine);
cli/src/test_runner.rs
+64 -25
View File
@@ -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<Run> {
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::<Value>(&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<String> {
@@ -220,7 +247,19 @@ pub async fn run(wasm: &Path, args: Vec<String>) -> Result<bool> {
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}");