host: process table, shell, coreutils and harness (T8, T9)
T8: sausage:process WIT (spawn, wait, kill), one store per process over the shared root, /proc/processes, /run, pipeline and cancellation groups, Exec. brush and 78 uutils commands; the shared cwd-init object sets the cwd from PWD and keeps numeric exit codes. The corpus matches native bash. T9: V8 (deno_core) runs pi-durable, pi-ai and chord; SQLite storage in <dir>/host/harness.sqlite (WAL, synchronous=FULL); file tools on the VFS, bash through Exec; resumes after SIGKILL.
This commit is contained in:
@@ -0,0 +1,147 @@
|
||||
//! Rust operations for tools. File access uses the world's shared VFS layer.
|
||||
use crate::exec::{Command, Exec, Options, Stream};
|
||||
use deno_core::{OpState, op2};
|
||||
use deno_error::JsErrorBox;
|
||||
use sausage_vfs::{Layer, Open};
|
||||
use serde_json::{Value, json};
|
||||
use std::{cell::RefCell, io::Write, rc::Rc, sync::Arc};
|
||||
|
||||
pub(super) struct Environment {
|
||||
pub root: Arc<dyn Layer>,
|
||||
pub exec: Arc<dyn Exec>,
|
||||
}
|
||||
|
||||
#[op2]
|
||||
#[serde]
|
||||
pub async fn op_file(
|
||||
state: Rc<RefCell<OpState>>,
|
||||
#[string] kind: String,
|
||||
#[string] path: String,
|
||||
#[serde] bytes: Vec<u8>,
|
||||
) -> Result<serde_json::Value, JsErrorBox> {
|
||||
let root = state.borrow().borrow::<Environment>().root.clone();
|
||||
let result: std::io::Result<Value> = async {
|
||||
match kind.as_str() {
|
||||
"read" => {
|
||||
let file = root.open(&path, Open { follow: true, ..Default::default() }).await?;
|
||||
let size = file.stat().await?.size;
|
||||
Ok(json!(file.read(0, size as usize).await?))
|
||||
}
|
||||
"write" => {
|
||||
let file = root.open(&path, Open { follow: true, write: true, create: true, truncate: true, ..Default::default() }).await?;
|
||||
let mut offset = 0;
|
||||
while offset < bytes.len() {
|
||||
let count = file.write(offset as u64, bytes[offset..].to_vec()).await?;
|
||||
if count == 0 { return Err(std::io::ErrorKind::WriteZero.into()); }
|
||||
offset += count;
|
||||
}
|
||||
file.sync().await?;
|
||||
Ok(Value::Null)
|
||||
}
|
||||
"mkdir" => { root.mkdir(&path).await?; Ok(Value::Null) }
|
||||
"stat" => {
|
||||
let s = root.stat(&path, true).await?;
|
||||
Ok(json!({"path": path, "name": path.rsplit('/').next().unwrap_or(""), "kind": if s.directory {"directory"} else {"file"}, "size": s.size, "mtimeMs": 0}))
|
||||
}
|
||||
_ => Err(std::io::ErrorKind::Unsupported.into()),
|
||||
}
|
||||
}.await;
|
||||
Ok(match result {
|
||||
Ok(value) => json!({"ok": true, "value": value}),
|
||||
Err(e) => {
|
||||
json!({"ok": false, "error": {"code": match e.kind() { std::io::ErrorKind::NotFound => "not_found", std::io::ErrorKind::PermissionDenied => "permission_denied", _ => "unknown" }, "message": e.to_string(), "path": path}})
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
struct Execution {
|
||||
receiver: tokio::sync::Mutex<tokio::sync::mpsc::UnboundedReceiver<Value>>,
|
||||
task: tokio::task::AbortHandle,
|
||||
}
|
||||
impl deno_core::Resource for Execution {
|
||||
fn close(self: Rc<Self>) {
|
||||
self.task.abort();
|
||||
}
|
||||
}
|
||||
impl Drop for Execution {
|
||||
fn drop(&mut self) {
|
||||
self.task.abort();
|
||||
}
|
||||
}
|
||||
|
||||
#[op2]
|
||||
pub fn op_exec(
|
||||
state: &mut OpState,
|
||||
#[serde] command: serde_json::Value,
|
||||
#[string] cwd: String,
|
||||
timeout: Option<f64>,
|
||||
) -> Result<u32, JsErrorBox> {
|
||||
let exec = state.borrow::<Environment>().exec.clone();
|
||||
let command = if let Some(s) = command.as_str() {
|
||||
Command::Shell(s.to_owned())
|
||||
} else {
|
||||
Command::Argv(
|
||||
serde_json::from_value(command).map_err(|e| JsErrorBox::generic(e.to_string()))?,
|
||||
)
|
||||
};
|
||||
let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
|
||||
let task = tokio::spawn(async move {
|
||||
let result = exec.exec(command, Options { cwd, timeout: timeout.map(std::time::Duration::from_secs_f64), ..Default::default() }, &mut |stream, bytes| { let _ = sender.send(json!({"stream": if stream == Stream::Stdout {"stdout"} else {"stderr"}, "bytes": bytes})); }).await;
|
||||
let _ = sender.send(match result {
|
||||
Ok(exit) => json!({"exitCode": exit}),
|
||||
Err(e) => json!({"error": e.to_string()}),
|
||||
});
|
||||
});
|
||||
Ok(state.resource_table.add(Execution {
|
||||
receiver: tokio::sync::Mutex::new(receiver),
|
||||
task: task.abort_handle(),
|
||||
}))
|
||||
}
|
||||
|
||||
#[op2]
|
||||
#[serde]
|
||||
pub async fn op_exec_next(
|
||||
state: Rc<RefCell<OpState>>,
|
||||
rid: u32,
|
||||
) -> Result<serde_json::Value, JsErrorBox> {
|
||||
let execution = state
|
||||
.borrow()
|
||||
.resource_table
|
||||
.get::<Execution>(rid)
|
||||
.map_err(|e| JsErrorBox::generic(e.to_string()))?;
|
||||
let value = execution
|
||||
.receiver
|
||||
.lock()
|
||||
.await
|
||||
.recv()
|
||||
.await
|
||||
.unwrap_or(Value::Null);
|
||||
if value.get("stream").is_none() {
|
||||
let resource = state
|
||||
.borrow_mut()
|
||||
.resource_table
|
||||
.take::<Execution>(rid)
|
||||
.map_err(|e| JsErrorBox::generic(e.to_string()))?;
|
||||
deno_core::Resource::close(resource);
|
||||
}
|
||||
Ok(value)
|
||||
}
|
||||
|
||||
#[op2]
|
||||
#[string]
|
||||
pub async fn op_input() -> Result<Option<String>, JsErrorBox> {
|
||||
use tokio::io::AsyncBufReadExt;
|
||||
let mut reader = tokio::io::BufReader::with_capacity(1, tokio::io::stdin());
|
||||
let mut line = String::new();
|
||||
let n = reader
|
||||
.read_line(&mut line)
|
||||
.await
|
||||
.map_err(|e| JsErrorBox::generic(e.to_string()))?;
|
||||
Ok((n != 0).then_some(line))
|
||||
}
|
||||
|
||||
#[op2(fast)]
|
||||
pub fn op_output(#[string] text: &str) {
|
||||
print!("{text}");
|
||||
let _ = std::io::stdout().flush();
|
||||
}
|
||||
+209
-5
@@ -1,8 +1,212 @@
|
||||
//! The durable agent harness: V8 with pi-durable, harness storage, and the tools' environment.
|
||||
//! Owned by task T9.
|
||||
//! Durable V8 harness, host-side SQLite storage, and world tool operations.
|
||||
mod environment;
|
||||
use crate::{exec::Exec, world::World};
|
||||
use deno_core::{JsRuntime, OpState, RuntimeOptions, op2};
|
||||
use deno_error::JsErrorBox;
|
||||
use rusqlite::{Connection, types::Value as SqlValue};
|
||||
use serde_json::{Value, json};
|
||||
use std::path::Path;
|
||||
use std::sync::Arc;
|
||||
|
||||
/// Start the world at `world` with the harness and a console.
|
||||
pub async fn run(_world: &Path) -> anyhow::Result<()> {
|
||||
anyhow::bail!("the harness does not exist yet (T9)")
|
||||
#[op2]
|
||||
#[serde]
|
||||
fn op_sql(
|
||||
state: &mut OpState,
|
||||
#[string] kind: String,
|
||||
#[string] sql: String,
|
||||
#[serde] params: Vec<serde_json::Value>,
|
||||
) -> Result<serde_json::Value, JsErrorBox> {
|
||||
let work = || -> Result<Value, Box<dyn std::error::Error>> {
|
||||
if kind == "open" {
|
||||
let db = Connection::open(sql)?;
|
||||
db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;")?;
|
||||
state.put(db);
|
||||
return Ok(Value::Null);
|
||||
}
|
||||
if kind == "close" {
|
||||
state.take::<Connection>();
|
||||
return Ok(Value::Null);
|
||||
}
|
||||
let db = state.borrow::<Connection>();
|
||||
if kind == "exec" {
|
||||
db.execute_batch(&sql)?;
|
||||
return Ok(Value::Null);
|
||||
}
|
||||
let params: Vec<SqlValue> = params
|
||||
.into_iter()
|
||||
.map(|v| match v {
|
||||
Value::Null => SqlValue::Null,
|
||||
Value::Number(n) => n
|
||||
.as_i64()
|
||||
.map(SqlValue::Integer)
|
||||
.unwrap_or_else(|| SqlValue::Real(n.as_f64().unwrap())),
|
||||
Value::String(s) => SqlValue::Text(s),
|
||||
Value::Array(a) => {
|
||||
SqlValue::Blob(a.into_iter().map(|n| n.as_u64().unwrap() as u8).collect())
|
||||
}
|
||||
_ => panic!("unsupported SQL parameter"),
|
||||
})
|
||||
.collect();
|
||||
let mut stmt = db.prepare_cached(&sql)?;
|
||||
if kind == "run" {
|
||||
let changes = stmt.execute(rusqlite::params_from_iter(params))?;
|
||||
return Ok(json!({"changes": changes, "lastInsertRowid": db.last_insert_rowid()}));
|
||||
}
|
||||
let names: Vec<String> = stmt.column_names().into_iter().map(str::to_owned).collect();
|
||||
let mut rows = stmt.query(rusqlite::params_from_iter(params))?;
|
||||
let mut result = vec![];
|
||||
while let Some(row) = rows.next()? {
|
||||
let mut obj = serde_json::Map::new();
|
||||
for (i, name) in names.iter().enumerate() {
|
||||
let v = match row.get::<_, SqlValue>(i)? {
|
||||
SqlValue::Null => Value::Null,
|
||||
SqlValue::Integer(n) => json!(n),
|
||||
SqlValue::Real(n) => json!(n),
|
||||
SqlValue::Text(s) => json!(s),
|
||||
SqlValue::Blob(b) => json!(b),
|
||||
};
|
||||
obj.insert(name.clone(), v);
|
||||
}
|
||||
result.push(Value::Object(obj));
|
||||
}
|
||||
Ok(json!(result))
|
||||
};
|
||||
work().map_err(|e| JsErrorBox::generic(e.to_string()))
|
||||
}
|
||||
|
||||
deno_core::extension!(
|
||||
storage,
|
||||
ops = [
|
||||
op_sql,
|
||||
environment::op_file,
|
||||
environment::op_exec,
|
||||
environment::op_exec_next,
|
||||
environment::op_input,
|
||||
environment::op_output
|
||||
]
|
||||
);
|
||||
|
||||
/// Start the console with the process implementation supplied by the host.
|
||||
pub async fn run_with_exec(world: World, exec: Arc<dyn Exec>) -> anyhow::Result<()> {
|
||||
// Deno's unsynchronized operations require a current-thread Tokio runtime.
|
||||
tokio::task::spawn_blocking(move || {
|
||||
tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()?
|
||||
.block_on(run_module(world, exec, include_str!("../../js/bundle.js")))
|
||||
})
|
||||
.await?
|
||||
.map_err(|error: anyhow::Error| {
|
||||
let mut text = error.to_string();
|
||||
for key in ["PI_PROXY_BASE_URL", "PI_PROXY_API_KEY"] {
|
||||
if let Ok(value) = std::env::var(key) {
|
||||
if !value.is_empty() {
|
||||
text = text.replace(&value, "[redacted]");
|
||||
}
|
||||
}
|
||||
}
|
||||
anyhow::anyhow!(text)
|
||||
})
|
||||
}
|
||||
|
||||
/// Evaluate a bundled harness module with the same storage and tool operations as the console.
|
||||
/// The caller must use a current-thread Tokio runtime.
|
||||
pub async fn run_module(world: World, exec: Arc<dyn Exec>, source: &str) -> anyhow::Result<()> {
|
||||
let mut runtime = JsRuntime::new(RuntimeOptions {
|
||||
extensions: vec![
|
||||
deno_telemetry::deno_telemetry::init(),
|
||||
deno_webidl::deno_webidl::init(),
|
||||
deno_web::deno_web::init(
|
||||
Arc::new(deno_web::BlobStore::default()),
|
||||
None,
|
||||
false,
|
||||
Default::default(),
|
||||
),
|
||||
deno_crypto::deno_crypto::init(None),
|
||||
deno_net::deno_net::init(None, None),
|
||||
deno_fetch::deno_fetch::init(Default::default()),
|
||||
storage::init(),
|
||||
],
|
||||
extension_transpiler: Some(std::rc::Rc::new(|name, source| {
|
||||
if !name.ends_with(".ts") {
|
||||
return Ok((source, None));
|
||||
}
|
||||
let parsed = deno_ast::parse_module(deno_ast::ParseParams {
|
||||
specifier: deno_core::url::Url::parse(&name).unwrap(),
|
||||
text: source.into(),
|
||||
media_type: deno_ast::MediaType::TypeScript,
|
||||
capture_tokens: false,
|
||||
scope_analysis: false,
|
||||
maybe_syntax: None,
|
||||
})
|
||||
.map_err(|e| JsErrorBox::generic(e.to_string()))?;
|
||||
let emitted = parsed
|
||||
.transpile(
|
||||
&Default::default(),
|
||||
&Default::default(),
|
||||
&deno_ast::EmitOptions {
|
||||
source_map: deno_ast::SourceMapOption::None,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.map_err(|e| JsErrorBox::generic(e.to_string()))?
|
||||
.into_source();
|
||||
Ok((emitted.text.into(), None))
|
||||
})),
|
||||
..Default::default()
|
||||
});
|
||||
runtime
|
||||
.op_state()
|
||||
.borrow_mut()
|
||||
.put(deno_permissions::PermissionsContainer::allow_all(Arc::new(
|
||||
deno_permissions::RuntimePermissionDescriptorParser::new(sys_traits::impls::RealSys),
|
||||
)));
|
||||
runtime.execute_script("bootstrap.js", include_str!("../../js/bootstrap.js"))?;
|
||||
let env: serde_json::Map<String, Value> = ["PI_PROXY_BASE_URL", "PI_PROXY_API_KEY"]
|
||||
.into_iter()
|
||||
.filter_map(|key| std::env::var(key).ok().map(|v| (key.to_owned(), json!(v))))
|
||||
.collect();
|
||||
runtime.execute_script("env.js", format!("process.env = {};", Value::Object(env)))?;
|
||||
runtime
|
||||
.op_state()
|
||||
.borrow_mut()
|
||||
.put(environment::Environment {
|
||||
root: world.root,
|
||||
exec,
|
||||
});
|
||||
runtime.execute_script(
|
||||
"config.js",
|
||||
format!(
|
||||
"globalThis.storagePath = {};",
|
||||
serde_json::to_string(&world.host_dir.join("harness.sqlite"))?
|
||||
),
|
||||
)?;
|
||||
let specifier = deno_core::ModuleSpecifier::parse("file:///harness.js")?;
|
||||
let module = runtime
|
||||
.load_main_es_module_from_code(&specifier, source.to_owned())
|
||||
.await?;
|
||||
let result = runtime.mod_evaluate(module);
|
||||
runtime
|
||||
.with_event_loop_promise(result, Default::default())
|
||||
.await?;
|
||||
runtime.run_event_loop(Default::default()).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
struct Unavailable;
|
||||
#[async_trait::async_trait]
|
||||
impl Exec for Unavailable {
|
||||
async fn exec(
|
||||
&self,
|
||||
_: crate::exec::Command,
|
||||
_: crate::exec::Options,
|
||||
_: &mut (dyn for<'chunk> FnMut(crate::exec::Stream, &'chunk [u8]) + Send),
|
||||
) -> anyhow::Result<i32> {
|
||||
anyhow::bail!("process execution is not wired yet")
|
||||
}
|
||||
}
|
||||
|
||||
/// Open a world and resume its durable root conversation.
|
||||
pub async fn run(path: &Path) -> anyhow::Result<()> {
|
||||
run_with_exec(World::open(path)?, Arc::new(Unavailable)).await
|
||||
}
|
||||
|
||||
+546
-6
@@ -1,8 +1,548 @@
|
||||
//! The process table: WASI programs in their own stores, the process interface (spawn, wait, kill),
|
||||
//! `/proc`, and the `Exec` implementation. Owned by task T8.
|
||||
use std::path::Path;
|
||||
//! Runs components in independent stores over the world root. Owns process groups, stdio, and Exec.
|
||||
use crate::{
|
||||
exec::{Command, Exec, Options, Stream},
|
||||
world::World,
|
||||
};
|
||||
use sausage_vfs::{
|
||||
Mounts,
|
||||
adapter::{self, Adapter, View},
|
||||
};
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
path::Path,
|
||||
sync::{
|
||||
Arc, Mutex,
|
||||
atomic::{AtomicBool, AtomicU32, Ordering},
|
||||
},
|
||||
};
|
||||
use tokio::{
|
||||
io::{AsyncReadExt, DuplexStream},
|
||||
sync::watch,
|
||||
task::AbortHandle,
|
||||
};
|
||||
use wasmtime::{
|
||||
Engine, Store,
|
||||
component::{Component, Linker, Resource, ResourceTable},
|
||||
};
|
||||
use wasmtime_wasi::{
|
||||
WasiCtx, WasiCtxBuilder, WasiCtxView, WasiView,
|
||||
cli::{AsyncStdinStream, AsyncStdoutStream},
|
||||
p2::{
|
||||
DynInputStream, DynOutputStream,
|
||||
bindings::CommandPre,
|
||||
pipe::{AsyncReadStream, AsyncWriteStream},
|
||||
},
|
||||
};
|
||||
|
||||
/// Run a shell in the world at `world`, or one command with `command`. Returns the exit status.
|
||||
pub async fn sh(_world: &Path, _command: Option<String>) -> anyhow::Result<i32> {
|
||||
anyhow::bail!("the process table does not exist yet (T8)")
|
||||
mod bindings {
|
||||
wasmtime::component::bindgen!({
|
||||
path: "wit", world: "program",
|
||||
imports: { default: async | trappable },
|
||||
with: {
|
||||
"wasi:io": wasmtime_wasi::p2::bindings::io,
|
||||
"sausage:process/host.child": super::Child,
|
||||
},
|
||||
});
|
||||
}
|
||||
use bindings::sausage::process::host::{Host, HostChild, Request};
|
||||
|
||||
/// A finite snapshot of process metadata exposed through /proc/processes.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct ProcessInfo {
|
||||
pub parent: u32,
|
||||
pub argv: Vec<String>,
|
||||
pub cwd: String,
|
||||
pub state: String,
|
||||
}
|
||||
pub type Processes = Arc<Mutex<std::collections::BTreeMap<u32, ProcessInfo>>>;
|
||||
|
||||
struct Entry {
|
||||
command: u32,
|
||||
group: u32,
|
||||
cancelled: Arc<AtomicBool>,
|
||||
abort: AbortHandle,
|
||||
}
|
||||
/// One runtime keeps compiled components warm and shares the world's namespace across stores.
|
||||
pub struct Runtime {
|
||||
world: World,
|
||||
engine: Engine,
|
||||
cache: Mutex<HashMap<Vec<u8>, CommandPre<State>>>,
|
||||
entries: Mutex<HashMap<u32, Entry>>,
|
||||
next: AtomicU32,
|
||||
alive: Arc<AtomicBool>,
|
||||
}
|
||||
impl Drop for Runtime {
|
||||
fn drop(&mut self) {
|
||||
self.alive.store(false, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
struct State {
|
||||
wasi: WasiCtx,
|
||||
table: ResourceTable,
|
||||
fs: Adapter,
|
||||
runtime: Arc<Runtime>,
|
||||
pid: u32,
|
||||
command: u32,
|
||||
group: u32,
|
||||
}
|
||||
impl WasiView for State {
|
||||
fn ctx(&mut self) -> WasiCtxView<'_> {
|
||||
WasiCtxView {
|
||||
ctx: &mut self.wasi,
|
||||
table: &mut self.table,
|
||||
}
|
||||
}
|
||||
}
|
||||
pub struct Child {
|
||||
pid: u32,
|
||||
group: u32,
|
||||
code: watch::Receiver<Option<i32>>,
|
||||
input: Option<DynOutputStream>,
|
||||
output: Option<DynInputStream>,
|
||||
error: Option<DynInputStream>,
|
||||
}
|
||||
struct Started {
|
||||
pid: u32,
|
||||
group: u32,
|
||||
code: watch::Receiver<Option<i32>>,
|
||||
input: DuplexStream,
|
||||
output: DuplexStream,
|
||||
error: DuplexStream,
|
||||
}
|
||||
impl Runtime {
|
||||
pub fn open(path: &Path) -> anyhow::Result<Arc<Self>> {
|
||||
Self::new(World::open(path)?)
|
||||
}
|
||||
pub fn new(world: World) -> anyhow::Result<Arc<Self>> {
|
||||
let mut config = wasmtime::Config::new();
|
||||
config.epoch_interruption(true);
|
||||
let mut cache = wasmtime::CacheConfig::new();
|
||||
cache.with_directory(world.host_dir.join("compile-cache"));
|
||||
config.cache(Some(wasmtime::Cache::new(cache)?));
|
||||
let engine = Engine::new(&config)?;
|
||||
let alive = Arc::new(AtomicBool::new(true));
|
||||
let ticking = alive.clone();
|
||||
let clock = engine.clone();
|
||||
std::thread::spawn(move || {
|
||||
while ticking.load(Ordering::Relaxed) {
|
||||
std::thread::sleep(std::time::Duration::from_millis(10));
|
||||
clock.increment_epoch();
|
||||
}
|
||||
});
|
||||
Ok(Arc::new(Self {
|
||||
world,
|
||||
engine,
|
||||
alive,
|
||||
cache: Mutex::new(HashMap::new()),
|
||||
entries: Mutex::new(HashMap::new()),
|
||||
next: AtomicU32::new(1),
|
||||
}))
|
||||
}
|
||||
fn kill_command(&self, command: u32) {
|
||||
for entry in self.entries.lock().unwrap().values() {
|
||||
if entry.command == command {
|
||||
entry.cancelled.store(true, Ordering::Relaxed);
|
||||
entry.abort.abort();
|
||||
}
|
||||
}
|
||||
}
|
||||
fn kill_group(&self, group: u32) {
|
||||
for entry in self.entries.lock().unwrap().values() {
|
||||
if entry.group == group {
|
||||
entry.abort.abort();
|
||||
}
|
||||
}
|
||||
}
|
||||
fn kill_pid(&self, pid: u32) -> anyhow::Result<()> {
|
||||
let group = self
|
||||
.entries
|
||||
.lock()
|
||||
.unwrap()
|
||||
.get(&pid)
|
||||
.map(|e| e.group)
|
||||
.ok_or_else(|| anyhow::anyhow!("no live process {pid}"))?;
|
||||
self.kill_group(group);
|
||||
Ok(())
|
||||
}
|
||||
async fn start(
|
||||
self: &Arc<Self>,
|
||||
mut request: Request,
|
||||
parent: u32,
|
||||
command: u32,
|
||||
inherited_group: u32,
|
||||
) -> anyhow::Result<Started> {
|
||||
anyhow::ensure!(!request.argv.is_empty(), "empty argv");
|
||||
let cancelled = if parent == 0 {
|
||||
Arc::new(AtomicBool::new(false))
|
||||
} else {
|
||||
self.entries
|
||||
.lock()
|
||||
.unwrap()
|
||||
.get(&parent)
|
||||
.map(|entry| entry.cancelled.clone())
|
||||
.ok_or_else(|| anyhow::anyhow!("parent exited"))?
|
||||
};
|
||||
request.cwd = sausage_vfs::resolve("/", &request.cwd);
|
||||
request.env.retain(|(name, _)| name != "PWD");
|
||||
request.env.push(("PWD".into(), request.cwd.clone()));
|
||||
let executable = sausage_vfs::resolve(&request.cwd, &request.program);
|
||||
let file = self
|
||||
.world
|
||||
.root
|
||||
.open(
|
||||
&executable,
|
||||
sausage_vfs::Open {
|
||||
follow: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
let bytes = file
|
||||
.read(0, usize::try_from(file.stat().await?.size)?)
|
||||
.await?;
|
||||
let pre = {
|
||||
let mut cache = self.cache.lock().unwrap();
|
||||
if let Some(pre) = cache.get(&bytes) {
|
||||
pre.clone()
|
||||
} else {
|
||||
let component = Component::new(&self.engine, &bytes)?;
|
||||
let mut linker = Linker::new(&self.engine);
|
||||
wasmtime_wasi::p2::add_to_linker_async(&mut linker)?;
|
||||
// The library linker includes exit-with-code. Only Wasmtime's CLI gates it.
|
||||
adapter::add_to_linker(&mut linker, |s: &mut State| View {
|
||||
adapter: &mut s.fs,
|
||||
table: &mut s.table,
|
||||
})?;
|
||||
bindings::sausage::process::host::add_to_linker::<
|
||||
_,
|
||||
wasmtime::component::HasSelf<_>,
|
||||
>(&mut linker, |s| s)?;
|
||||
let pre = CommandPre::new(linker.instantiate_pre(&component)?)?;
|
||||
cache.insert(bytes, pre.clone());
|
||||
pre
|
||||
}
|
||||
};
|
||||
let pid = self.next.fetch_add(1, Ordering::Relaxed);
|
||||
let command = if command == 0 { pid } else { command };
|
||||
let group = match request.group {
|
||||
Some(0) => pid,
|
||||
Some(group) => group,
|
||||
None if inherited_group != 0 => inherited_group,
|
||||
None => pid,
|
||||
};
|
||||
request.env.push(("PID".into(), pid.to_string()));
|
||||
request.env.push(("PPID".into(), parent.to_string()));
|
||||
let (input, child_input) = tokio::io::duplex(8192);
|
||||
let (output, child_output) = tokio::io::duplex(8192);
|
||||
let (error, child_error) = tokio::io::duplex(8192);
|
||||
let mut mounts = Mounts::default();
|
||||
mounts.mount("/", self.world.root.clone(), false);
|
||||
let mut wasi = WasiCtxBuilder::new();
|
||||
wasi.args(&request.argv)
|
||||
.envs(&request.env)
|
||||
.initial_cwd(&request.cwd)
|
||||
.stdin(AsyncStdinStream::new(child_input))
|
||||
.stdout(AsyncStdoutStream::new(8192, child_output))
|
||||
.stderr(AsyncStdoutStream::new(8192, child_error));
|
||||
let mut store = Store::new(
|
||||
&self.engine,
|
||||
State {
|
||||
wasi: wasi.build(),
|
||||
table: ResourceTable::new(),
|
||||
fs: Adapter::new(mounts),
|
||||
runtime: self.clone(),
|
||||
pid,
|
||||
command,
|
||||
group,
|
||||
},
|
||||
);
|
||||
store.epoch_deadline_async_yield_and_update(1);
|
||||
self.world.processes.lock().unwrap().insert(
|
||||
pid,
|
||||
ProcessInfo {
|
||||
parent,
|
||||
argv: request.argv,
|
||||
cwd: request.cwd,
|
||||
state: "running".into(),
|
||||
},
|
||||
);
|
||||
let (tx, code) = watch::channel(None);
|
||||
// Capture cleanup before the first poll, so an early abort still removes the entry.
|
||||
let cleanup = Cleanup {
|
||||
runtime: self.clone(),
|
||||
pid,
|
||||
tx,
|
||||
};
|
||||
// The start gate makes the entry visible before a guest can spawn children or exit.
|
||||
let (gate, ready) = tokio::sync::oneshot::channel();
|
||||
let task = tokio::spawn(async move {
|
||||
let _cleanup = cleanup;
|
||||
if ready.await.is_err() {
|
||||
return;
|
||||
}
|
||||
let result = async {
|
||||
let command = pre.instantiate_async(&mut store).await?;
|
||||
match command.wasi_cli_run().call_run(&mut store).await {
|
||||
Ok(Ok(())) => Ok(0),
|
||||
Ok(Err(())) => Ok(1),
|
||||
Err(e) => match e.downcast_ref::<wasmtime_wasi::I32Exit>() {
|
||||
Some(exit) => Ok(exit.0),
|
||||
None => Err(e),
|
||||
},
|
||||
}
|
||||
}
|
||||
.await;
|
||||
_cleanup.tx.send_replace(Some(result.unwrap_or(125)));
|
||||
});
|
||||
{
|
||||
let mut entries = self.entries.lock().unwrap();
|
||||
entries.insert(
|
||||
pid,
|
||||
Entry {
|
||||
command,
|
||||
group,
|
||||
cancelled: cancelled.clone(),
|
||||
abort: task.abort_handle(),
|
||||
},
|
||||
);
|
||||
// A spawn already in progress must observe a command-wide cancellation.
|
||||
if cancelled.load(Ordering::Relaxed) {
|
||||
task.abort();
|
||||
} else {
|
||||
let _ = gate.send(());
|
||||
}
|
||||
}
|
||||
Ok(Started {
|
||||
pid,
|
||||
group,
|
||||
code,
|
||||
input,
|
||||
output,
|
||||
error,
|
||||
})
|
||||
}
|
||||
}
|
||||
struct Cleanup {
|
||||
runtime: Arc<Runtime>,
|
||||
pid: u32,
|
||||
tx: watch::Sender<Option<i32>>,
|
||||
}
|
||||
impl Drop for Cleanup {
|
||||
fn drop(&mut self) {
|
||||
if self.tx.borrow().is_none() {
|
||||
self.tx.send_replace(Some(137));
|
||||
}
|
||||
self.runtime.entries.lock().unwrap().remove(&self.pid);
|
||||
self.runtime
|
||||
.world
|
||||
.processes
|
||||
.lock()
|
||||
.unwrap()
|
||||
.remove(&self.pid);
|
||||
}
|
||||
}
|
||||
impl Host for State {
|
||||
async fn spawn(
|
||||
&mut self,
|
||||
request: Request,
|
||||
) -> wasmtime::Result<Result<Resource<Child>, String>> {
|
||||
let child = match self
|
||||
.runtime
|
||||
.start(request, self.pid, self.command, self.group)
|
||||
.await
|
||||
{
|
||||
Ok(child) => child,
|
||||
Err(e) => return Ok(Err(e.to_string())),
|
||||
};
|
||||
Ok(Ok(self.table.push(Child {
|
||||
pid: child.pid,
|
||||
group: child.group,
|
||||
code: child.code,
|
||||
input: Some(Box::new(AsyncWriteStream::new(8192, child.input))),
|
||||
output: Some(Box::new(AsyncReadStream::new(child.output))),
|
||||
error: Some(Box::new(AsyncReadStream::new(child.error))),
|
||||
})?))
|
||||
}
|
||||
async fn kill(&mut self, pid: u32) -> wasmtime::Result<Result<(), String>> {
|
||||
Ok(self.runtime.kill_pid(pid).map_err(|e| e.to_string()))
|
||||
}
|
||||
}
|
||||
impl HostChild for State {
|
||||
async fn pid(&mut self, child: Resource<Child>) -> wasmtime::Result<u32> {
|
||||
Ok(self.table.get(&child)?.pid)
|
||||
}
|
||||
async fn group(&mut self, child: Resource<Child>) -> wasmtime::Result<u32> {
|
||||
Ok(self.table.get(&child)?.group)
|
||||
}
|
||||
async fn stdin(
|
||||
&mut self,
|
||||
child: Resource<Child>,
|
||||
) -> wasmtime::Result<Resource<DynOutputStream>> {
|
||||
let stream = self
|
||||
.table
|
||||
.get_mut(&child)?
|
||||
.input
|
||||
.take()
|
||||
.ok_or_else(|| wasmtime::format_err!("stdin already taken"))?;
|
||||
Ok(self.table.push(stream)?)
|
||||
}
|
||||
async fn stdout(
|
||||
&mut self,
|
||||
child: Resource<Child>,
|
||||
) -> wasmtime::Result<Resource<DynInputStream>> {
|
||||
let stream = self
|
||||
.table
|
||||
.get_mut(&child)?
|
||||
.output
|
||||
.take()
|
||||
.ok_or_else(|| wasmtime::format_err!("stdout already taken"))?;
|
||||
Ok(self.table.push(stream)?)
|
||||
}
|
||||
async fn stderr(
|
||||
&mut self,
|
||||
child: Resource<Child>,
|
||||
) -> wasmtime::Result<Resource<DynInputStream>> {
|
||||
let stream = self
|
||||
.table
|
||||
.get_mut(&child)?
|
||||
.error
|
||||
.take()
|
||||
.ok_or_else(|| wasmtime::format_err!("stderr already taken"))?;
|
||||
Ok(self.table.push(stream)?)
|
||||
}
|
||||
async fn status(&mut self, child: Resource<Child>) -> wasmtime::Result<Option<i32>> {
|
||||
Ok(*self.table.get(&child)?.code.borrow())
|
||||
}
|
||||
async fn wait(&mut self, child: Resource<Child>) -> wasmtime::Result<i32> {
|
||||
let mut code = self.table.get(&child)?.code.clone();
|
||||
loop {
|
||||
if let Some(code) = *code.borrow_and_update() {
|
||||
return Ok(code);
|
||||
}
|
||||
code.changed().await?;
|
||||
}
|
||||
}
|
||||
async fn kill(&mut self, child: Resource<Child>) -> wasmtime::Result<()> {
|
||||
self.runtime
|
||||
.kill_pid(self.table.get(&child)?.pid)
|
||||
.map_err(|e| wasmtime::format_err!("{e}"))
|
||||
}
|
||||
async fn drop(&mut self, child: Resource<Child>) -> wasmtime::Result<()> {
|
||||
self.table.delete(child)?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
struct GroupGuard {
|
||||
runtime: Arc<Runtime>,
|
||||
group: u32,
|
||||
}
|
||||
impl Drop for GroupGuard {
|
||||
fn drop(&mut self) {
|
||||
self.runtime.kill_command(self.group);
|
||||
}
|
||||
}
|
||||
/// An Arc runtime implements Exec so the harness can retain one warm process table.
|
||||
#[async_trait::async_trait]
|
||||
impl Exec for Arc<Runtime> {
|
||||
async fn exec(
|
||||
&self,
|
||||
command: Command,
|
||||
options: Options,
|
||||
output: &mut (dyn for<'chunk> FnMut(Stream, &'chunk [u8]) + Send),
|
||||
) -> anyhow::Result<i32> {
|
||||
let deadline = options
|
||||
.timeout
|
||||
.map(|duration| tokio::time::Instant::now() + duration);
|
||||
let argv = match command {
|
||||
Command::Shell(script) => vec![
|
||||
"/bin/sh".into(),
|
||||
"--noprofile".into(),
|
||||
"--norc".into(),
|
||||
"-c".into(),
|
||||
script,
|
||||
],
|
||||
Command::Argv(argv) => argv,
|
||||
};
|
||||
let mut env = options.env;
|
||||
if !env.iter().any(|(k, _)| k == "PATH") {
|
||||
env.push(("PATH".into(), "/bin".into()));
|
||||
}
|
||||
let start = self.start(
|
||||
Request {
|
||||
program: argv.first().cloned().unwrap_or_default(),
|
||||
argv,
|
||||
env,
|
||||
cwd: options.cwd,
|
||||
group: None,
|
||||
},
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
);
|
||||
let mut child = if let Some(deadline) = deadline {
|
||||
tokio::time::timeout_at(deadline, start)
|
||||
.await
|
||||
.map_err(|_| anyhow::anyhow!("command timed out"))??
|
||||
} else {
|
||||
start.await?
|
||||
};
|
||||
let _guard = GroupGuard {
|
||||
runtime: self.clone(),
|
||||
group: child.pid,
|
||||
};
|
||||
drop(child.input);
|
||||
let run = async {
|
||||
let (mut out_open, mut err_open) = (true, true);
|
||||
let (mut out, mut err) = ([0; 8192], [0; 8192]);
|
||||
while out_open || err_open {
|
||||
tokio::select! {
|
||||
n = child.output.read(&mut out), if out_open => { let n = n?; out_open = n != 0; if n != 0 { output(Stream::Stdout, &out[..n]); } }
|
||||
n = child.error.read(&mut err), if err_open => { let n = n?; err_open = n != 0; if n != 0 { output(Stream::Stderr, &err[..n]); } }
|
||||
}
|
||||
}
|
||||
loop {
|
||||
if let Some(code) = *child.code.borrow_and_update() {
|
||||
return Ok(code);
|
||||
}
|
||||
child.code.changed().await?;
|
||||
}
|
||||
};
|
||||
if let Some(deadline) = deadline {
|
||||
tokio::time::timeout_at(deadline, run)
|
||||
.await
|
||||
.map_err(|_| anyhow::anyhow!("command timed out"))?
|
||||
} else {
|
||||
run.await
|
||||
}
|
||||
}
|
||||
}
|
||||
/// Construct the harness's shared executor over the same open world.
|
||||
pub fn exec(world: World) -> anyhow::Result<Arc<dyn Exec>> {
|
||||
Ok(Arc::new(Runtime::new(world)?))
|
||||
}
|
||||
|
||||
/// Run the world's shell. Noninteractive shell scripts stream directly to host stdio.
|
||||
pub async fn sh(world: &Path, command: Option<String>) -> anyhow::Result<i32> {
|
||||
let runtime = Runtime::open(world)?;
|
||||
let command = command.ok_or_else(|| anyhow::anyhow!("interactive shell is not implemented"))?;
|
||||
runtime
|
||||
.exec(
|
||||
Command::Shell(command),
|
||||
Options {
|
||||
cwd: "/".into(),
|
||||
..Options::default()
|
||||
},
|
||||
&mut |stream, bytes| {
|
||||
use std::io::Write;
|
||||
match stream {
|
||||
Stream::Stdout => {
|
||||
let _ = std::io::stdout().write_all(bytes);
|
||||
}
|
||||
Stream::Stderr => {
|
||||
let _ = std::io::stderr().write_all(bytes);
|
||||
}
|
||||
}
|
||||
},
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
+33
-5
@@ -3,7 +3,7 @@
|
||||
//! Layout of a world directory:
|
||||
//! - `world/`: the virtual root `/` that every guest sees. Agents change everything in it.
|
||||
//! - `host/`: host state outside the sandbox (harness storage, later the vault). No guest sees it.
|
||||
use sausage_vfs::{Host, Layer, Mounts, Process};
|
||||
use sausage_vfs::{Generated, Host, Layer, Memory, Mounts, Process};
|
||||
use std::{
|
||||
path::{Path, PathBuf},
|
||||
sync::Arc,
|
||||
@@ -15,18 +15,46 @@ pub struct World {
|
||||
pub root: Arc<dyn Layer>,
|
||||
/// The host state folder, `<dir>/host`.
|
||||
pub host_dir: PathBuf,
|
||||
/// Live process metadata shared with the process table.
|
||||
pub processes: crate::process::Processes,
|
||||
}
|
||||
|
||||
impl World {
|
||||
/// Open (and create if missing) the world directory `dir`. The process table adds more mounts
|
||||
/// (`/proc`, `/run`).
|
||||
/// Open (and create if missing) the world directory `dir`, including `/proc` and `/run` mounts.
|
||||
pub fn open(dir: &Path) -> anyhow::Result<Self> {
|
||||
let root_dir = dir.join("world");
|
||||
let host_dir = dir.join("host");
|
||||
std::fs::create_dir_all(&root_dir)?;
|
||||
std::fs::create_dir_all(&host_dir)?;
|
||||
let mut mounts = Mounts::default();
|
||||
mounts.mount("/", Arc::new(Host::new(&root_dir, Process::default())?), false);
|
||||
Ok(World { root: Arc::new(mounts), host_dir })
|
||||
mounts.mount(
|
||||
"/",
|
||||
Arc::new(Host::new(&root_dir, Process::default())?),
|
||||
false,
|
||||
);
|
||||
let processes: crate::process::Processes =
|
||||
Arc::new(std::sync::Mutex::new(Default::default()));
|
||||
let snapshot = processes.clone();
|
||||
let mut proc = Generated::default();
|
||||
proc.insert("/processes", move || {
|
||||
let mut text = String::from("PID\tPPID\tSTATE\tCWD\tARGV\n");
|
||||
for (pid, info) in snapshot.lock().unwrap().iter() {
|
||||
text.push_str(&format!(
|
||||
"{pid}\t{}\t{}\t{}\t{}\n",
|
||||
info.parent,
|
||||
info.state,
|
||||
info.cwd,
|
||||
info.argv.join(" ")
|
||||
));
|
||||
}
|
||||
Ok(text.into_bytes())
|
||||
});
|
||||
mounts.mount("/proc", Arc::new(proc), true);
|
||||
mounts.mount("/run", Arc::new(Memory::default()), false);
|
||||
Ok(World {
|
||||
root: Arc::new(mounts),
|
||||
host_dir,
|
||||
processes,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user