Merge v3 and package the builder closure and complete userland

host/src/api/cargo.rs

@@ -0,0 +1,60 @@
+//! A streaming Cargo response. Dropping the body cancels the builder's native process group.
+use crate::builder::Event;
+use bytes::Bytes;
+use http::{Request, Response};
+use http_body_util::BodyExt;
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
+struct Body(tokio::sync::mpsc::Receiver<Event>);
+impl http_body::Body for Body {
+    type Data = Bytes;
+    type Error = wasmtime_wasi_http::Error;
+    fn poll_frame(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<http_body::Frame<Bytes>, Self::Error>>> {
+        self.0.poll_recv(cx).map(|event| {
+            event.map(|event| {
+                let mut bytes = serde_json::to_vec(&event).expect("Cargo event");
+                bytes.push(b'\n');
+                Ok(http_body::Frame::data(Bytes::from(bytes)))
+            })
+        })
+    }
+}
+pub(crate) fn stream(
+    api: &super::Api,
+    request: Request<Bytes>,
+) -> Response<wasmtime_wasi_http::WasiBody> {
+    let (tx, rx) = tokio::sync::mpsc::channel(16);
+    let builder = api.builder.get().cloned();
+    let request = serde_json::from_slice::<crate::builder::Request>(request.body());
+    tokio::spawn(async move {
+        let result = async {
+            builder
+                .ok_or_else(|| anyhow::anyhow!("builder unavailable"))?
+                .execute(request?, &tx)
+                .await
+        }
+        .await;
+        if let Err(error) = result {
+            let _ = tx
+                .send(Event::Stderr {
+                    bytes: format!("cargo: {error:#}\n").into_bytes(),
+                })
+                .await;
+            let _ = tx
+                .send(Event::Exit {
+                    code: 1,
+                    executables: Vec::new(),
+                })
+                .await;
+        }
+    });
+    Response::builder()
+        .header("content-type", "application/x-ndjson")
+        .body(Body(rx).boxed_unsync())
+        .unwrap()
+}

host/src/api/mod.rs

@@ -1,9 +1,12 @@
 //! JSON routes for host.sausage. The transport supplies the trusted caller.
 //! Host-owned pages use passkeys and typed proofs for every authorization ceremony.
 mod auth;
+mod cargo;
+pub(crate) use cargo::stream;
 mod pages;
 mod server;
 mod ssh;
+mod undo;
 pub use server::{Listener, Options, run_with_userland};
 
 use crate::cred::{self, Caller, Cred, entry::EntryKey, webauthn::Passkeys};
@@ -19,10 +22,12 @@ use std::{
 
 pub struct Api {
     pub cred: Arc<Cred>,
+    pub(crate) builder: std::sync::OnceLock<Arc<crate::builder::Builder>>,
     script: cred::quickjs::QuickJs,
     pub passkeys: Mutex<Passkeys<rusqlite::Connection>>,
     pub entry: EntryKey,
     requests: Mutex<HashMap<u32, (Instant, u32)>>,
+    journal: Mutex<Option<Arc<sausage_vfs::journal::Journal>>>,
 }
 impl Api {
     /// Own the connection and ceremony state shared by both transports.
@@ -47,12 +52,17 @@ impl Api {
     ) -> Self {
         Self {
             cred: cred.into(),
+            builder: Default::default(),
             script: cred::quickjs::QuickJs::default(),
             passkeys: Mutex::new(passkeys),
             entry,
             requests: Mutex::new(HashMap::new()),
+            journal: Mutex::new(None),
         }
     }
+    pub(crate) fn attach_journal(&self, journal: Option<Arc<sausage_vfs::journal::Journal>>) {
+        *self.journal.lock().unwrap() = journal;
+    }
     fn rate_limit(&self, caller: &Caller) -> Result<()> {
         let now = Instant::now();
         let mut requests = self.requests.lock().map_err(|_| Error(500, "storage"))?;
@@ -132,6 +142,9 @@ fn request_view(request: cred::requests::Request) -> Result<Value> {
 
 /// Handle a request without a network transport. Errors contain only static error kinds.
 pub async fn handle(ctx: &Api, caller: Caller, request: Request<Bytes>) -> Response<Bytes> {
+    if request.uri().path() == "/undo" || request.uri().path().starts_with("/undo/") {
+        return undo::handle(ctx, caller, request).await;
+    }
     respond(ctx, caller, request)
 }
 fn respond(ctx: &Api, caller: Caller, request: Request<Bytes>) -> Response<Bytes> {

host/src/api/server.rs

@@ -1,5 +1,5 @@
 //! The host HTTP listener and console startup share one owned API and credentials store.
-use super::{Api, respond};
+use super::Api;
 use crate::cred::{Caller, Cred};
 use std::{net::SocketAddr, path::Path, sync::Arc};
 
@@ -37,6 +37,10 @@ impl Listener {
         );
         let serving = server.clone();
         let thread = std::thread::spawn(move || {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("API runtime");
             for mut request in serving.incoming_requests() {
                 let response = (|| -> anyhow::Result<_> {
                     let mut body = Vec::new();
@@ -45,7 +49,7 @@ impl Listener {
                         .method(request.method().as_str())
                         .uri(request.url())
                         .body(bytes::Bytes::from(body))?;
-                    Ok(respond(
+                    Ok(runtime.block_on(super::handle(
                         &api,
                         Caller {
                             program: "/host/browser".into(),
@@ -53,7 +57,7 @@ impl Listener {
                             conversations: Vec::new(),
                         },
                         request,
-                    ))
+                    )))
                 })();
                 let response = response.unwrap_or_else(|_| {
                     http::Response::builder()

host/src/api/undo.rs

@@ -0,0 +1,60 @@
+//! Undo routes share the world's journal. They require no credential approval.
+use super::*;
+use sausage_vfs::journal::{Tag, Undo};
+pub(super) async fn handle(ctx: &Api, caller: Caller, request: Request<Bytes>) -> Response<Bytes> {
+    let journal = ctx.journal.lock().unwrap().clone();
+    let result = async {
+        let journal = journal.ok_or_else(|| {
+            std::io::Error::new(std::io::ErrorKind::NotFound, "journal unavailable")
+        })?;
+        journal.retain().await?;
+        let query: HashMap<String, String> =
+            url::form_urlencoded::parse(request.uri().query().unwrap_or("").as_bytes())
+                .into_owned()
+                .collect();
+        match (request.method().as_str(), request.uri().path()) {
+            ("GET", "/undo" | "/undo/list") => Ok((
+                200,
+                serde_json::to_value(journal.list(query.get("conversation").cloned()).await?)?,
+            )),
+            ("GET", "/undo/show") => {
+                let tag = query.get("tag").ok_or_else(|| {
+                    std::io::Error::new(std::io::ErrorKind::InvalidInput, "tag required")
+                })?;
+                Ok((200, serde_json::to_value(journal.show(tag.clone()).await?)?))
+            }
+            ("POST", "/undo") => {
+                let request: Undo = serde_json::from_slice(request.body())?;
+                let result = journal
+                    .undo(
+                        request,
+                        Tag::process(caller.pid, caller.program, caller.conversations)?,
+                    )
+                    .await?;
+                let status = if result.tag.is_none() { 409 } else { 200 };
+                Ok((status, serde_json::to_value(result)?))
+            }
+            _ => Ok((404, json!({"error":"not_found"}))),
+        }
+    }
+    .await;
+    let (status, body) = match result {
+        Ok(result) => result,
+        Err(error) => {
+            let error: Box<dyn std::error::Error> = error;
+            let status = error
+                .downcast_ref::<std::io::Error>()
+                .map_or(400, |e| match e.kind() {
+                    std::io::ErrorKind::NotFound => 404,
+                    std::io::ErrorKind::InvalidInput => 400,
+                    _ => 500,
+                });
+            (status, json!({"error":error.to_string()}))
+        }
+    };
+    Response::builder()
+        .status(status)
+        .header("content-type", "application/json")
+        .body(Bytes::from(serde_json::to_vec(&body).expect("JSON value")))
+        .expect("valid response")
+}

host/src/builder/README.md

@@ -0,0 +1,68 @@
+# Rust builder
+
+The world `cargo` calls `POST http://host.sausage/cargo` with `argv` and `cwd`.
+The request also accepts optional `timeout_secs`.
+The response contains newline-delimited JSON events.
+`stdout` and `stderr` events contain byte arrays.
+The final `exit` event contains a code and component locations.
+Each location contains an executable path and working directory.
+
+The CLI forwards the bytes immediately.
+For `test` and `run`, it executes the artifacts through the process interface.
+Cargo selects run binaries through a capture runner, so `--bin` and `default-run` retain their normal behavior.
+The test runner uses each package's root directory, even when the caller runs Cargo from a subdirectory.
+For `test` and `run`, arguments after `--` belong to the executable, not the host build.
+Other commands retain their `--` arguments, such as `fmt -- --check` and Clippy lint options.
+
+The host finds the Cargo workspace above the requested directory and copies its sources through the shared VFS.
+It excludes `target` and `.git`.
+
+Each workspace owns a source copy and target directory.
+It also owns a registry and kache cache.
+These directories live under `<dir>/host/builds/`.
+
+Cargo fetches dependencies with network access.
+The command then runs offline inside bubblewrap namespaces.
+
+The source copy and target directory permit writes.
+The private kache cache also permits writes.
+The registry and Nix store remain read-only.
+Host home directories and the host network are absent.
+The PID namespace removes compiler descendants and the private kache daemon after Cargo exits.
+
+The builder copies changed and new source files back through the VFS.
+It preserves other source files, including concurrent edits in the world.
+It publishes WASI executables into the world's workspace target directory.
+After `cargo clean`, it removes the published artifacts that Cargo removed from the native target.
+Every target build uses `wasm32-wasip2` and links the shared cwd-init object, including its numeric exit-code interface.
+The host rejects other build targets and `--target-dir` overrides.
+
+The command deadline defaults to 30 minutes.
+The host kills timed-out commands before it sends their timeout message.
+A closed response cancels the native command and its process group.
+
+The host retains at most 20 GiB of build directories and evicts the oldest workspace first.
+
+The Nix wrapper supplies these tools on PATH:
+
+- Rust and the native compiler
+- The WASI C tools
+- Bubblewrap
+- Nushell
+- wasm-tools
+- Kache
+- Bash and Coreutils
+
+Kache comes from the upstream flake at revision `96052b4842239bb216f6c4812054203c21550788`.
+The development shell supplies the same builder tools.
+The builder resolves the wrapper and interpreter paths to their canonical Nix store paths.
+The embedded cwd-init sources produce a startup object and a linker script inside each workspace's tool folder.
+The linker script uses the canonical Nushell interpreter, so it needs no host `/usr/bin/env`.
+The sandbox exposes Bash at `/bin/sh` for native build scripts.
+It does not depend on a user-profile path or a host HOME file.
+Development commands retain the global Cargo configuration and its kache wrapper.
+The sandbox uses a private Cargo home and selects kache through `--config build.rustc-wrapper`.
+
+For `cargo add`, the network phase performs a dry run to populate the registry index.
+The offline command makes the actual manifest change inside the namespaces.
+The disk cap applies after each completed command, including compiler failures.

host/src/builder/files.rs

@@ -0,0 +1,156 @@
+//! Workspace discovery and source/artifact copies through the world's shared VFS.
+use anyhow::{Context, Result};
+use sausage_vfs::{Layer, Open};
+use std::{collections::HashMap, path::Path};
+
+/// Source hashes before the native command changes its copy.
+pub type Snapshot = HashMap<String, [u8; 32]>;
+
+pub async fn read(root: &dyn Layer, path: &str) -> Result<Vec<u8>> {
+    let file = root
+        .open(
+            path,
+            Open {
+                follow: true,
+                ..Default::default()
+            },
+        )
+        .await?;
+    let size = file.stat().await?.size;
+    Ok(file.read(0, usize::try_from(size)?).await?)
+}
+pub async fn workspace(root: &dyn Layer, cwd: &str) -> Result<String> {
+    let mut path = sausage_vfs::resolve("/", cwd);
+    let mut nearest = None;
+    loop {
+        let manifest = sausage_vfs::resolve(&path, "Cargo.toml");
+        if let Ok(bytes) = read(root, &manifest).await {
+            let value: toml::Value = toml::from_str(std::str::from_utf8(&bytes)?)?;
+            if value.get("workspace").is_some() {
+                return Ok(path);
+            }
+            if nearest.is_none() {
+                nearest = Some(path.clone());
+            }
+        }
+        if path == "/" {
+            break;
+        }
+        path = sausage_vfs::resolve(&path, "..");
+    }
+    nearest.context("no Cargo.toml above cwd")
+}
+pub async fn copy_out(root: &dyn Layer, source: &str, destination: &Path) -> Result<Snapshot> {
+    std::fs::create_dir_all(destination)?;
+    let mut before = Snapshot::new();
+    for entry in root.list(source).await? {
+        if matches!(entry.name.as_str(), "target" | ".git") {
+            continue;
+        }
+        let from = sausage_vfs::resolve(source, &entry.name);
+        let to = destination.join(&entry.name);
+        if entry.directory {
+            before.extend(Box::pin(copy_out(root, &from, &to)).await?);
+        } else {
+            let bytes = read(root, &from).await?;
+            before.insert(from, openssl::sha::sha256(&bytes));
+            std::fs::write(to, bytes)?;
+        }
+    }
+    Ok(before)
+}
+pub async fn mkdir(root: &dyn Layer, path: &str) -> Result<()> {
+    let mut current = String::new();
+    for part in path.split('/').filter(|p| !p.is_empty()) {
+        current.push('/');
+        current.push_str(part);
+        if let Err(error) = root.mkdir(&current).await
+            && error.kind() != std::io::ErrorKind::AlreadyExists
+        {
+            return Err(error.into());
+        }
+    }
+    Ok(())
+}
+pub async fn write(root: &dyn Layer, path: &str, bytes: Vec<u8>) -> Result<()> {
+    if read(root, path).await.ok().as_deref() == Some(bytes.as_slice()) {
+        return Ok(());
+    }
+    mkdir(root, path.rsplit_once('/').unwrap().0).await?;
+    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 written = file.write(offset as u64, bytes[offset..].to_vec()).await?;
+        anyhow::ensure!(written != 0, "source write made no progress");
+        offset += written;
+    }
+    Ok(())
+}
+/// A snapshot selects changed sources. Without a snapshot, publish only WASM artifacts.
+pub async fn copy_back(
+    root: &dyn Layer,
+    source: &Path,
+    destination: &str,
+    before: Option<&Snapshot>,
+) -> Result<()> {
+    for entry in std::fs::read_dir(source)? {
+        let entry = entry?;
+        let name = entry.file_name().to_string_lossy().into_owned();
+        if matches!(name.as_str(), "target" | ".git") {
+            continue;
+        }
+        let to = sausage_vfs::resolve(destination, &name);
+        if entry.file_type()?.is_dir() {
+            Box::pin(copy_back(root, &entry.path(), &to, before)).await?;
+        } else if before.is_some() || entry.path().extension().is_some_and(|e| e == "wasm") {
+            let bytes = std::fs::read(entry.path())?;
+            if before.is_none_or(|before| before.get(&to) != Some(&openssl::sha::sha256(&bytes))) {
+                write(root, &to, bytes).await?;
+            }
+        }
+    }
+    Ok(())
+}
+
+/// Remove published artifacts that Cargo clean removed from the native target.
+pub async fn clean_artifacts(root: &dyn Layer, native: &Path, world: &str) -> Result<()> {
+    let entries = match root.list(world).await {
+        Ok(entries) => entries,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
+        Err(error) => return Err(error.into()),
+    };
+    for entry in entries {
+        let path = sausage_vfs::resolve(world, &entry.name);
+        let native = native.join(&entry.name);
+        if entry.directory {
+            Box::pin(clean_artifacts(root, &native, &path)).await?;
+        } else if entry.name.ends_with(".wasm") && !native.exists() {
+            root.remove_file(&path).await?;
+        }
+    }
+    Ok(())
+}
+
+pub fn size(path: &Path) -> Result<u64> {
+    let mut total = 0;
+    for entry in std::fs::read_dir(path)? {
+        let entry = entry?;
+        total += if entry.file_type()?.is_dir() {
+            size(&entry.path())?
+        } else {
+            entry.metadata()?.len()
+        };
+    }
+    Ok(total)
+}

host/src/builder/mod.rs

@@ -0,0 +1,492 @@
+//! Native Cargo execution, isolated workspace copies, streamed output, and bounded disk use.
+mod files;
+#[cfg(test)]
+mod tests;
+use anyhow::{Context, Result, ensure};
+use sausage_vfs::Layer;
+use serde::{Deserialize, Serialize};
+use std::{
+    path::{Path, PathBuf},
+    process::Stdio,
+    sync::Arc,
+    time::Duration,
+};
+use tokio::{
+    io::AsyncReadExt,
+    process::Command,
+    sync::{Mutex, mpsc},
+    time::Instant,
+};
+
+#[derive(Deserialize)]
+pub struct Request {
+    pub argv: Vec<String>,
+    pub cwd: String,
+    pub timeout_secs: Option<u64>,
+}
+#[derive(Serialize, PartialEq, Eq, PartialOrd, Ord)]
+pub struct Executable {
+    pub path: String,
+    pub cwd: String,
+}
+#[derive(Serialize)]
+#[serde(tag = "event", rename_all = "snake_case")]
+pub enum Event {
+    Stdout {
+        bytes: Vec<u8>,
+    },
+    Stderr {
+        bytes: Vec<u8>,
+    },
+    Exit {
+        code: i32,
+        executables: Vec<Executable>,
+    },
+}
+pub struct Builder {
+    root: Arc<dyn Layer>,
+    directory: PathBuf,
+    lock: Mutex<()>,
+}
+impl Builder {
+    pub fn new(root: Arc<dyn Layer>, host: &Path) -> Result<Self> {
+        Ok(Self {
+            root,
+            directory: std::fs::canonicalize(host)?.join("builds"),
+            lock: Mutex::new(()),
+        })
+    }
+    pub async fn execute(&self, request: Request, tx: &mpsc::Sender<Event>) -> Result<()> {
+        let _guard = self.lock.lock().await;
+        let action = request
+            .argv
+            .first()
+            .context("missing Cargo command")?
+            .as_str();
+        ensure!(
+            [
+                "build", "check", "clippy", "test", "run", "fmt", "fetch", "add", "remove",
+                "update", "tree", "metadata", "clean"
+            ]
+            .contains(&action),
+            "unsupported Cargo command"
+        );
+        let compiles = ["build", "check", "clippy", "test", "run"].contains(&action);
+        let mut args = request.argv.iter().take_while(|arg| arg.as_str() != "--");
+        while let Some(arg) = args.next() {
+            ensure!(
+                arg != "--target-dir" && !arg.starts_with("--target-dir="),
+                "the builder controls the target directory"
+            );
+            let target = if arg == "--target" {
+                args.next().map(String::as_str)
+            } else {
+                arg.strip_prefix("--target=")
+            };
+            if compiles && let Some(target) = target {
+                ensure!(
+                    target == "wasm32-wasip2",
+                    "the build target is always wasm32-wasip2"
+                );
+            }
+        }
+        let seconds = request.timeout_secs.unwrap_or(1800);
+        ensure!(seconds > 0, "timeout must be positive");
+        let deadline = Instant::now()
+            .checked_add(Duration::from_secs(seconds))
+            .context("timeout too large")?;
+        let world_cwd = sausage_vfs::resolve("/", &request.cwd);
+        let workspace = files::workspace(self.root.as_ref(), &world_cwd).await?;
+        let world_target = sausage_vfs::resolve(&workspace, "target");
+        use std::hash::{Hash, Hasher};
+        let mut hash = std::collections::hash_map::DefaultHasher::new();
+        workspace.hash(&mut hash);
+        let id = format!("{:016x}", hash.finish());
+        let base = self.directory.join(id);
+        let source = base.join("source");
+        // Cargo clean must remove its directory, not a namespace mount point.
+        let target_mount = base.join("target");
+        let target = target_mount.join("cargo");
+        // Keep /tmp's namespace mountpoints outside the directory Cargo clean removes.
+        let temporary = target_mount.join("tmp");
+        let registry = base.join("cargo");
+        let cache = base.join("kache");
+        if source.exists() {
+            std::fs::remove_dir_all(&source)?;
+        }
+        let before = files::copy_out(self.root.as_ref(), &workspace, &source).await?;
+        for path in [&target, &registry, &cache, &temporary] {
+            std::fs::create_dir_all(path)?;
+        }
+        std::fs::write(
+            target.join("CACHEDIR.TAG"),
+            b"Signature: 8a477f597d28d172789f06886806bc55\n",
+        )?;
+        std::fs::write(base.join("last-used"), [])?;
+        let relative = world_cwd
+            .strip_prefix(&workspace)
+            .expect("workspace is an ancestor of cwd")
+            .trim_start_matches('/');
+        let cwd = if relative.is_empty() {
+            source.clone()
+        } else {
+            source.join(relative)
+        };
+        let mut fetch = Command::new("cargo");
+        if action == "add" {
+            fetch.args(&request.argv).arg("--dry-run");
+        } else {
+            fetch.arg("fetch");
+        }
+        fetch.current_dir(&cwd).env("CARGO_HOME", &registry);
+        let mut code = if matches!(action, "clean" | "fmt" | "remove") {
+            0
+        } else {
+            run(fetch, tx, deadline).await?.0
+        };
+        let mut executables = Vec::new();
+        if code == 0 {
+            let tools = base.join("tools");
+            std::fs::create_dir_all(&tools)?;
+            for (name, bytes) in [
+                (
+                    "init.rs",
+                    include_bytes!("../../../userland/cwd-init/init.rs").as_slice(),
+                ),
+                (
+                    "exit.wit",
+                    include_bytes!("../../../userland/cwd-init/exit.wit").as_slice(),
+                ),
+                (
+                    "build.nu",
+                    include_bytes!("../../../userland/cwd-init/build.nu").as_slice(),
+                ),
+                (
+                    "link.nu",
+                    include_bytes!("../../../userland/cwd-init/link.nu").as_slice(),
+                ),
+                ("run.nu", include_bytes!("run.nu").as_slice()),
+            ] {
+                std::fs::write(tools.join(name), bytes)?;
+            }
+            let linker = std::fs::read_to_string(tools.join("link.nu"))?.replacen(
+                "#!/usr/bin/env nu",
+                &format!("#!{}", executable("nu")?.display()),
+                1,
+            );
+            std::fs::write(tools.join("link.nu"), linker)?;
+            use std::os::unix::fs::PermissionsExt;
+            std::fs::set_permissions(
+                tools.join("link.nu"),
+                std::fs::Permissions::from_mode(0o755),
+            )?;
+            let mut startup = Command::new("nu");
+            startup.arg(tools.join("build.nu"));
+            let (startup_code, _) = run(startup, tx, deadline).await?;
+            if startup_code != 0 {
+                files::copy_back(self.root.as_ref(), &source, &workspace, Some(&before)).await?;
+                self.evict(20 * 1024 * 1024 * 1024)?;
+                tx.send(Event::Exit {
+                    code: startup_code,
+                    executables,
+                })
+                .await?;
+                return Ok(());
+            }
+            let kache = executable("kache")?;
+            let run_path = target_mount.join("run-path");
+            if action == "run" && run_path.exists() {
+                std::fs::remove_file(&run_path)?;
+            }
+            let mut cmd = Command::new("bwrap");
+            cmd.args([
+                "--unshare-user",
+                "--unshare-pid",
+                "--unshare-net",
+                "--die-with-parent",
+                "--ro-bind",
+                "/nix/store",
+                "/nix/store",
+            ])
+            .arg("--bind")
+            .arg(&temporary)
+            .arg("/tmp")
+            .arg("--ro-bind")
+            .arg(executable("sh")?)
+            .arg("/bin/sh");
+            for path in [&source, &target_mount, &cache] {
+                cmd.arg("--bind").arg(path).arg(path);
+            }
+            for path in [&registry, &tools] {
+                cmd.arg("--ro-bind").arg(path).arg(path);
+            }
+            cmd.args([
+                "--proc",
+                "/proc",
+                "--dev",
+                "/dev",
+                "--remount-ro",
+                "/",
+                "--chdir",
+            ])
+            .arg(&cwd)
+            .args(["--clearenv", "--setenv", "PATH"])
+            .arg(std::env::var("PATH")?)
+            .args(["--setenv", "HOME", "/nonexistent", "--setenv", "CARGO_HOME"])
+            .arg(&registry)
+            .args(["--setenv", "CARGO_TARGET_DIR"])
+            .arg(&target)
+            .args(["--setenv", "KACHE_CACHE_DIR"])
+            .arg(&cache)
+            .args(["--setenv", "SAUSAGE_CWD_INIT"])
+            .arg(tools.join("init.o"))
+            .args(["--setenv", "SAUSAGE_RUN_PATH"])
+            .arg(&run_path)
+            .args(["--setenv", "CARGO_BUILD_JOBS", "2"]);
+            for name in [
+                "CC_wasm32_wasip2",
+                "AR_wasm32_wasip2",
+                "CFLAGS_wasm32_wasip2",
+            ] {
+                if let Ok(value) = std::env::var(name) {
+                    cmd.args(["--setenv", name]).arg(value);
+                }
+            }
+            cmd.args(["cargo", "--offline", "--config"])
+                .arg(format!("build.rustc-wrapper={:?}", kache))
+                .arg("--config")
+                .arg(format!(
+                    "target.wasm32-wasip2.linker={:?}",
+                    tools.join("link.nu")
+                ))
+                .args([
+                    "--config",
+                    "target.wasm32-wasip2.rustflags=['-C','link-arg=sausage-cwd-init']",
+                ]);
+            let mut args = request.argv.clone();
+            if matches!(action, "run" | "test")
+                && let Some(index) = args.iter().position(|a| a == "--")
+            {
+                args.truncate(index);
+            }
+            if action == "test" {
+                // Cargo's positional test filter belongs to the world runner, not the compiler.
+                args = test_build_args(&args);
+                if !args.iter().any(|a| a == "--no-run") {
+                    args.push("--no-run".into());
+                }
+            }
+            let split = args.iter().position(|a| a == "--").unwrap_or(args.len());
+            if action == "run" {
+                cmd.arg("--config").arg(format!(
+                    "target.wasm32-wasip2.runner=[{:?},{:?}]",
+                    executable("nu")?,
+                    tools.join("run.nu")
+                ));
+            }
+            // Keep all global config flags before the subcommand so Cargo retains them.
+            cmd.args(&args[..split]);
+            if compiles {
+                cmd.args([
+                    "--target",
+                    "wasm32-wasip2",
+                    "--message-format=json-render-diagnostics",
+                ]);
+            }
+            cmd.args(&args[split..]);
+            let (status, stdout) = run(cmd, tx, deadline).await?;
+            code = status;
+            if action == "clean" && code == 0 {
+                files::clean_artifacts(self.root.as_ref(), &target, &world_target).await?;
+            }
+            let artifacts = target.join("wasm32-wasip2");
+            if artifacts.exists() {
+                files::copy_back(
+                    self.root.as_ref(),
+                    &artifacts,
+                    &format!("{world_target}/wasm32-wasip2"),
+                    None,
+                )
+                .await?;
+            }
+            if code == 0 && action == "run" && run_path.exists() {
+                let path = std::fs::read_to_string(&run_path)?;
+                let relative = Path::new(&path).strip_prefix(&target)?;
+                executables.push(Executable {
+                    path: format!("{world_target}/{}", relative.display()),
+                    cwd: world_cwd.clone(),
+                });
+            }
+            if code == 0
+                && action == "test"
+                && !request
+                    .argv
+                    .iter()
+                    .take_while(|a| a.as_str() != "--")
+                    .any(|a| a == "--no-run")
+            {
+                for line in stdout.split(|b| *b == b'\n') {
+                    if let Ok(value) = serde_json::from_slice::<serde_json::Value>(line)
+                        && value["reason"] == "compiler-artifact"
+                        && value["profile"]["test"] == true
+                        && let Some(path) = value["executable"].as_str()
+                        && let Ok(relative) = Path::new(path).strip_prefix(&target)
+                    {
+                        let manifest = Path::new(
+                            value["manifest_path"]
+                                .as_str()
+                                .context("missing test manifest")?,
+                        );
+                        let package = manifest.parent().unwrap().strip_prefix(&source)?;
+                        let cwd = sausage_vfs::resolve(&workspace, &package.to_string_lossy());
+                        executables.push(Executable {
+                            path: format!("{world_target}/{}", relative.display()),
+                            cwd,
+                        });
+                    }
+                }
+                executables.sort();
+                executables.dedup();
+            }
+        }
+        files::copy_back(self.root.as_ref(), &source, &workspace, Some(&before)).await?;
+        self.evict(20 * 1024 * 1024 * 1024)?;
+        tx.send(Event::Exit { code, executables }).await?;
+        Ok(())
+    }
+    fn evict(&self, limit: u64) -> Result<()> {
+        let mut entries = Vec::new();
+        let mut total = 0;
+        for entry in std::fs::read_dir(&self.directory)? {
+            let path = entry?.path();
+            let size = files::size(&path)?;
+            total += size;
+            entries.push((
+                std::fs::metadata(path.join("last-used"))?.modified()?,
+                path,
+                size,
+            ));
+        }
+        entries.sort_by_key(|e| e.0);
+        for (_, path, size) in entries {
+            if total <= limit {
+                break;
+            }
+            std::fs::remove_dir_all(path)?;
+            total -= size;
+        }
+        Ok(())
+    }
+}
+fn executable(name: &str) -> Result<PathBuf> {
+    let path = std::env::split_paths(&std::env::var_os("PATH").context("PATH missing")?)
+        .map(|p| p.join(name))
+        .find(|p| p.is_file())
+        .context(format!("missing {name}"))?;
+    Ok(std::fs::canonicalize(path)?)
+}
+fn test_build_args(args: &[String]) -> Vec<String> {
+    let mut out = vec!["test".into()];
+    let mut value = false;
+    for arg in &args[1..] {
+        if value || arg.starts_with('-') {
+            out.push(arg.clone());
+        }
+        value = [
+            "-p",
+            "--package",
+            "--test",
+            "--bin",
+            "--example",
+            "--features",
+            "--profile",
+            "--manifest-path",
+            "--exclude",
+            "-j",
+            "--jobs",
+            "-F",
+            "--config",
+            "--target",
+            "--target-dir",
+            "--message-format",
+            "--color",
+            "--lockfile-path",
+            "-Z",
+        ]
+        .contains(&arg.as_str());
+    }
+    out
+}
+// The guard kills native descendants when Cargo exits or the caller cancels the command.
+struct ProcessGroup(u32);
+impl Drop for ProcessGroup {
+    fn drop(&mut self) {
+        unsafe {
+            libc::kill(-(self.0 as i32), libc::SIGKILL);
+        }
+    }
+}
+async fn pump(
+    mut reader: impl tokio::io::AsyncRead + Unpin,
+    tx: &mpsc::Sender<Event>,
+    stderr: bool,
+) -> Result<Vec<u8>> {
+    let mut all = Vec::new();
+    let mut bytes = vec![0; 8192];
+    loop {
+        let n = reader.read(&mut bytes).await?;
+        if n == 0 {
+            return Ok(all);
+        }
+        if !stderr {
+            all.extend_from_slice(&bytes[..n]);
+        }
+        tx.send(if stderr {
+            Event::Stderr {
+                bytes: bytes[..n].to_vec(),
+            }
+        } else {
+            Event::Stdout {
+                bytes: bytes[..n].to_vec(),
+            }
+        })
+        .await?;
+    }
+}
+async fn run(
+    mut cmd: Command,
+    tx: &mpsc::Sender<Event>,
+    deadline: Instant,
+) -> Result<(i32, Vec<u8>)> {
+    cmd.stdin(Stdio::null())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .process_group(0)
+        .kill_on_drop(true);
+    let mut child = cmd.spawn()?;
+    let group = ProcessGroup(child.id().unwrap());
+    let out = child.stdout.take().unwrap();
+    let err = child.stderr.take().unwrap();
+    let collect = async {
+        let (status, out, _) = tokio::try_join!(
+            async { Ok::<_, anyhow::Error>(child.wait().await?) },
+            pump(out, tx, false),
+            pump(err, tx, true)
+        )?;
+        Ok::<_, anyhow::Error>((status.code().unwrap_or(1), out))
+    };
+    let (result, timed_out) = tokio::select! {
+        result = collect => (result, false),
+        _ = tx.closed() => anyhow::bail!("caller exited"),
+        _ = tokio::time::sleep_until(deadline) => (Ok((124, Vec::new())), true),
+    };
+    // Kill the command before a full output channel can delay its timeout message.
+    drop(group);
+    if timed_out {
+        tx.send(Event::Stderr {
+            bytes: b"cargo: command timed out\n".to_vec(),
+        })
+        .await?;
+    }
+    result
+}

host/src/builder/run.nu

@@ -0,0 +1,4 @@
+# Record Cargo's selected component without executing it on the host.
+def main [executable: string] {
+    $executable | save --raw --force $env.SAUSAGE_RUN_PATH
+}

host/src/builder/tests.rs

@@ -0,0 +1,236 @@
+//! Focused tests for source discovery, argument forwarding, eviction, and process cancellation.
+use super::*;
+#[tokio::test]
+async fn discovers_workspace_and_skips_target() -> Result<()> {
+    let temp = tempfile::tempdir()?;
+    let world = crate::world::World::open(temp.path())?;
+    files::write(
+        world.root.as_ref(),
+        "/project/Cargo.toml",
+        b"[workspace]\nmembers=['member']".to_vec(),
+    )
+    .await?;
+    files::write(
+        world.root.as_ref(),
+        "/project/member/Cargo.toml",
+        b"[package]\nname='member'\nversion='0.1.0'".to_vec(),
+    )
+    .await?;
+    files::write(
+        world.root.as_ref(),
+        "/project/target/junk",
+        b"junk".to_vec(),
+    )
+    .await?;
+    assert_eq!(
+        files::workspace(world.root.as_ref(), "/project/member/src").await?,
+        "/project"
+    );
+    let copy = temp.path().join("copy");
+    files::copy_out(world.root.as_ref(), "/project", &copy).await?;
+    assert!(copy.join("member/Cargo.toml").exists());
+    assert!(!copy.join("target").exists());
+    Ok(())
+}
+#[tokio::test]
+async fn copies_only_changed_and_new_sources() -> Result<()> {
+    let temp = tempfile::tempdir()?;
+    let world = crate::world::World::open(temp.path())?;
+    for name in ["unchanged.rs", "changed.rs"] {
+        files::write(
+            world.root.as_ref(),
+            &format!("/project/{name}"),
+            b"initial".to_vec(),
+        )
+        .await?;
+    }
+    let copy = temp.path().join("copy");
+    let before = files::copy_out(world.root.as_ref(), "/project", &copy).await?;
+    files::write(
+        world.root.as_ref(),
+        "/project/unchanged.rs",
+        b"concurrent edit".to_vec(),
+    )
+    .await?;
+    std::fs::write(copy.join("changed.rs"), b"changed by command")?;
+    std::fs::write(copy.join("Cargo.lock"), b"created by command")?;
+    files::copy_back(world.root.as_ref(), &copy, "/project", Some(&before)).await?;
+    for (name, expected) in [
+        ("unchanged.rs", "concurrent edit"),
+        ("changed.rs", "changed by command"),
+        ("Cargo.lock", "created by command"),
+    ] {
+        assert_eq!(
+            files::read(world.root.as_ref(), &format!("/project/{name}")).await?,
+            expected.as_bytes()
+        );
+    }
+    Ok(())
+}
+#[tokio::test]
+async fn clean_removes_only_deleted_artifacts() -> Result<()> {
+    let temp = tempfile::tempdir()?;
+    let world = crate::world::World::open(temp.path())?;
+    let native = temp.path().join("native");
+    std::fs::create_dir_all(&native)?;
+    std::fs::write(native.join("keep.wasm"), b"keep")?;
+    for name in ["keep.wasm", "remove.wasm", "note.txt"] {
+        files::write(
+            world.root.as_ref(),
+            &format!("/target/{name}"),
+            b"data".to_vec(),
+        )
+        .await?;
+    }
+    files::clean_artifacts(world.root.as_ref(), &native, "/target").await?;
+    assert!(
+        files::read(world.root.as_ref(), "/target/remove.wasm")
+            .await
+            .is_err()
+    );
+    for name in ["keep.wasm", "note.txt"] {
+        assert_eq!(
+            files::read(world.root.as_ref(), &format!("/target/{name}")).await?,
+            b"data"
+        );
+    }
+    Ok(())
+}
+#[test]
+fn relative_host_directory_uses_absolute_namespace_paths() -> Result<()> {
+    let temp = tempfile::tempdir()?;
+    let world = crate::world::World::open(temp.path())?;
+    let builder = Builder::new(world.root, Path::new("."))?;
+    assert!(builder.directory.is_absolute());
+    Ok(())
+}
+#[tokio::test]
+async fn target_options_preserve_the_platform_target_and_cache() -> Result<()> {
+    let temp = tempfile::tempdir()?;
+    let world = crate::world::World::open(temp.path())?;
+    let builder = Builder::new(world.root, &world.host_dir)?;
+    let (tx, _rx) = mpsc::channel(1);
+    for options in [
+        vec!["--target", "aarch64-unknown-linux-gnu"],
+        vec!["--target=aarch64-unknown-linux-gnu"],
+        vec!["--target-dir", "/other"],
+        vec!["--target-dir=/other"],
+    ] {
+        let error = builder
+            .execute(
+                Request {
+                    argv: std::iter::once("build")
+                        .chain(options)
+                        .map(str::to_string)
+                        .collect(),
+                    cwd: "/".into(),
+                    timeout_secs: None,
+                },
+                &tx,
+            )
+            .await
+            .unwrap_err();
+        assert!(error.to_string().contains("target"), "{error}");
+    }
+    for (action, options) in [
+        ("run", vec!["--target", "wasm32-wasip2"]),
+        ("run", vec!["--target=wasm32-wasip2"]),
+        ("run", vec!["--", "--target-dir", "/argument"]),
+        ("tree", vec!["--target", "all"]),
+    ] {
+        let error = builder
+            .execute(
+                Request {
+                    argv: std::iter::once(action)
+                        .chain(options)
+                        .map(str::to_string)
+                        .collect(),
+                    cwd: "/".into(),
+                    timeout_secs: None,
+                },
+                &tx,
+            )
+            .await
+            .unwrap_err();
+        assert_eq!(error.to_string(), "no Cargo.toml above cwd");
+    }
+    Ok(())
+}
+#[test]
+fn test_filter_is_not_a_build_argument() {
+    let args = ["test", "-p", "example", "filter", "--release"].map(str::to_string);
+    assert_eq!(
+        test_build_args(&args),
+        ["test", "-p", "example", "--release"]
+    );
+}
+#[test]
+fn disk_cap_removes_workspaces() -> Result<()> {
+    let temp = tempfile::tempdir()?;
+    let world = crate::world::World::open(temp.path())?;
+    let builder = Builder::new(world.root, &world.host_dir)?;
+    for name in ["a", "b"] {
+        let path = builder.directory.join(name);
+        std::fs::create_dir_all(&path)?;
+        std::fs::write(path.join("last-used"), [])?;
+        std::fs::write(path.join("data"), [0; 32])?;
+    }
+    std::fs::File::open(builder.directory.join("a/last-used"))?
+        .set_times(std::fs::FileTimes::new().set_modified(std::time::UNIX_EPOCH))?;
+    builder.evict(32)?;
+    assert!(!builder.directory.join("a").exists());
+    assert!(builder.directory.join("b").exists());
+    assert_eq!(files::size(&builder.directory)?, 32);
+    Ok(())
+}
+#[cfg(target_os = "linux")]
+#[tokio::test]
+async fn timeout_kills_before_output_can_drain() -> Result<()> {
+    use std::os::fd::{FromRawFd, OwnedFd};
+    let (tx, mut rx) = mpsc::channel(1);
+    let output = tx.clone();
+    let mut cmd = Command::new("sh");
+    cmd.args(["-c", "echo $$; exec sleep 1000"]);
+    let task =
+        tokio::spawn(
+            async move { run(cmd, &output, Instant::now() + Duration::from_secs(1)).await },
+        );
+    let Some(Event::Stdout { bytes }) =
+        tokio::time::timeout(Duration::from_secs(2), rx.recv()).await?
+    else {
+        anyhow::bail!("missing native PID");
+    };
+    let pid: u32 = std::str::from_utf8(&bytes)?.trim().parse()?;
+    let fd = unsafe { libc::syscall(libc::SYS_pidfd_open, pid, 0) };
+    ensure!(fd >= 0, "pidfd_open: {}", std::io::Error::last_os_error());
+    // OwnedFd keeps the PID descriptor valid and unchanged until AsyncFd drops it.
+    let exited = unsafe { tokio::io::unix::AsyncFd::register(OwnedFd::from_raw_fd(fd as i32)) }?;
+    // Fill the output channel, so the timeout message cannot enter it.
+    tx.send(Event::Stdout { bytes: Vec::new() }).await?;
+    let _ready = tokio::time::timeout(Duration::from_secs(2), exited.readable()).await??;
+    assert!(matches!(rx.recv().await, Some(Event::Stdout { .. })));
+    assert!(matches!(rx.recv().await, Some(Event::Stderr { bytes })
+        if bytes.as_slice() == b"cargo: command timed out\n"));
+    assert_eq!(task.await??.0, 124);
+    Ok(())
+}
+#[tokio::test]
+async fn closed_receiver_cancels_command() -> Result<()> {
+    let (tx, mut rx) = mpsc::channel(1);
+    let mut cmd = Command::new("sh");
+    cmd.args(["-c", "printf ready; exec sleep 1000"]);
+    let task =
+        tokio::spawn(async move { run(cmd, &tx, Instant::now() + Duration::from_secs(60)).await });
+    // Output must arrive before the command exits, not after collection completes.
+    let event = tokio::time::timeout(Duration::from_secs(2), rx.recv())
+        .await?
+        .context("missing streamed output")?;
+    assert!(matches!(event, Event::Stdout { bytes } if bytes == b"ready"));
+    drop(rx);
+    assert!(
+        tokio::time::timeout(Duration::from_secs(2), task)
+            .await??
+            .is_err()
+    );
+    Ok(())
+}

host/src/exec.rs

@@ -17,6 +17,8 @@ pub struct Options {
     pub env: Vec<(String, String)>,
     pub timeout: Option<Duration>,
     pub conversations: Vec<String>,
+    /// Undo tag shared by this tool call and all child processes.
+    pub tool_call: Option<String>,
 }
 
 /// The stream that one output chunk came from.

host/src/harness/environment.rs

@@ -18,8 +18,10 @@ pub async fn op_file(
     #[string] kind: String,
     #[string] path: String,
     #[serde] bytes: Vec<u8>,
+    #[serde] tag: Option<sausage_vfs::journal::Tag>,
 ) -> Result<serde_json::Value, JsErrorBox> {
     let root = state.borrow().borrow::<Environment>().root.clone();
+    let root = tag.and_then(|tag| root.tagged(tag)).unwrap_or(root);
     let result: std::io::Result<Value> = async {
         match kind.as_str() {
             "read" => {
@@ -40,9 +42,15 @@ pub async fn op_file(
                 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}))
+            "list" => Ok(json!(root.list(&path).await?.into_iter().map(|e| json!({"name":e.name,"kind":if e.symlink {"symlink"} else if e.directory {"directory"} else {"file"}})).collect::<Vec<_>>())),
+            "remove" => {
+                if root.stat(&path, false).await?.directory { root.remove_dir(&path).await?; } else { root.remove_file(&path).await?; }
+                Ok(Value::Null)
+            }
+            "rename" => { root.rename(&path, &String::from_utf8(bytes).map_err(|_| std::io::ErrorKind::InvalidInput)?).await?; Ok(Value::Null) }
+            "stat" | "lstat" => {
+                let s = root.stat(&path, kind == "stat").await?;
+                Ok(json!({"path": path, "name": path.rsplit('/').next().unwrap_or(""), "kind": if s.symlink {"symlink"} else if s.directory {"directory"} else {"file"}, "size": s.size, "mtimeMs": 0}))
             }
             _ => Err(std::io::ErrorKind::Unsupported.into()),
         }
@@ -50,7 +58,7 @@ pub async fn op_file(
     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}})
+            json!({"ok": false, "error": {"code": match e.kind() { std::io::ErrorKind::NotFound => "not_found", std::io::ErrorKind::AlreadyExists => "already_exists", std::io::ErrorKind::PermissionDenied => "permission_denied", _ => "unknown" }, "message": e.to_string(), "path": path}})
         }
     })
 }
@@ -77,8 +85,12 @@ pub fn op_exec(
     #[string] cwd: String,
     timeout: Option<f64>,
     #[string] conversation: Option<String>,
+    #[serde] env: Option<Vec<(String, String)>>,
+    #[serde] conversations: Option<Vec<String>>,
+    #[serde] tag: Option<sausage_vfs::journal::Tag>,
 ) -> Result<u32, JsErrorBox> {
     let exec = state.borrow::<Environment>().exec.clone();
+    let journal = state.borrow::<Environment>().root.journal();
     let command = if let Some(s) = command.as_str() {
         Command::Shell(s.to_owned())
     } else {
@@ -88,7 +100,14 @@ pub fn op_exec(
     };
     let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
     let task = tokio::spawn(async move {
-        let result = exec.exec(command, Options { cwd, conversations: conversation.into_iter().collect(), 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 tool_call = tag.as_ref().map(|tag| tag.id.clone());
+        if let (Some(journal), Some(tag)) = (journal, tag)
+            && let Err(error) = journal.register(tag).await
+        {
+            let _ = sender.send(json!({"error": error.to_string()}));
+            return;
+        }
+        let result = exec.exec(command, Options { cwd, tool_call, env: env.unwrap_or_default(), conversations: conversations.unwrap_or_else(|| conversation.into_iter().collect()), timeout: timeout.map(std::time::Duration::from_secs_f64) }, &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()}),
@@ -129,6 +148,51 @@ pub async fn op_exec_next(
     Ok(value)
 }
 
+struct Watch {
+    events: tokio::sync::Mutex<
+        tokio::sync::broadcast::Receiver<(String, sausage_vfs::observe::ChangeKind)>,
+    >,
+    closed: tokio::sync::Notify,
+}
+impl deno_core::Resource for Watch {
+    fn close(self: Rc<Self>) {
+        self.closed.notify_one();
+    }
+}
+#[op2]
+pub fn op_watch(state: &mut OpState, #[serde] paths: Vec<String>) -> Result<u32, JsErrorBox> {
+    let observer = state
+        .borrow::<Environment>()
+        .root
+        .observer()
+        .ok_or_else(|| JsErrorBox::generic("filesystem has no observer"))?;
+    Ok(state.resource_table.add(Watch {
+        events: tokio::sync::Mutex::new(observer.subscribe(paths)),
+        closed: Default::default(),
+    }))
+}
+#[op2]
+#[serde]
+pub async fn op_watch_next(
+    state: Rc<RefCell<OpState>>,
+    rid: u32,
+) -> Result<serde_json::Value, JsErrorBox> {
+    let watch = state
+        .borrow()
+        .resource_table
+        .get::<Watch>(rid)
+        .map_err(|e| JsErrorBox::generic(e.to_string()))?;
+    let mut events = watch.events.lock().await;
+    let result = tokio::select! { value = events.recv() => value, _ = watch.closed.notified() => return Ok(Value::Null) };
+    Ok(match result {
+        Ok((path, kind)) => json!({"path":path,"kind":format!("{kind:?}")}),
+        Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
+            json!({"path":"/","kind":"Overflow"})
+        }
+        Err(_) => Value::Null,
+    })
+}
+
 #[op2(fast)]
 pub fn op_output(#[string] text: &str) {
     print!("{text}");

host/src/harness/loader.rs

@@ -0,0 +1,144 @@
+//! Loads committed world modules and shares the bundle's package instances and status.
+use deno_core::{
+    ModuleLoadOptions, ModuleLoadReferrer, ModuleLoadResponse, ModuleLoader, ModuleSource,
+    ModuleSourceCode, ModuleSpecifier, ModuleType, OpState, ResolutionKind, op2,
+};
+use deno_error::JsErrorBox;
+use sausage_vfs::{Layer, Open};
+use std::{
+    cell::RefCell,
+    collections::BTreeMap,
+    path::{Path, PathBuf},
+    rc::Rc,
+    sync::{Arc, Mutex, OnceLock},
+};
+
+pub(crate) type Status = Arc<Mutex<String>>;
+pub(crate) fn status(path: &Path) -> Status {
+    static STATES: OnceLock<Mutex<BTreeMap<PathBuf, Status>>> = OnceLock::new();
+    STATES
+        .get_or_init(Default::default)
+        .lock()
+        .unwrap()
+        .entry(path.to_owned())
+        .or_insert_with(|| {
+            Arc::new(Mutex::new(
+                "{\"good\":null,\"current\":null,\"error\":null}".into(),
+            ))
+        })
+        .clone()
+}
+pub(super) struct Loader {
+    pub root: Arc<dyn Layer>,
+    pub modules: RefCell<BTreeMap<String, String>>,
+}
+#[op2]
+pub fn op_modules(state: &mut OpState, #[serde] modules: BTreeMap<String, Vec<String>>) {
+    let loader = state.borrow::<Rc<Loader>>();
+    for (name, exports) in modules {
+        let quoted = serde_json::to_string(&name).unwrap();
+        let code = exports.into_iter().enumerate().map(|(index,key)| format!("const e{index} = globalThis.__sausage.modules[{quoted}].{key}; export {{ e{index} as {key} }};\n")).collect();
+        loader
+            .modules
+            .borrow_mut()
+            .insert(format!("sausage:{name}"), code);
+    }
+}
+#[op2(fast)]
+pub fn op_extension_status(state: &mut OpState, #[string] value: String) {
+    *state.borrow::<Status>().lock().unwrap() = value;
+}
+pub(super) fn transpile(name: &str, source: String) -> Result<String, JsErrorBox> {
+    if !name.ends_with(".ts") {
+        return Ok(source);
+    }
+    let parsed = deno_ast::parse_module(deno_ast::ParseParams {
+        specifier: ModuleSpecifier::parse(name).map_err(|e| JsErrorBox::generic(e.to_string()))?,
+        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()))?;
+    Ok(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()
+        .text)
+}
+impl ModuleLoader for Loader {
+    fn resolve(
+        &self,
+        specifier: &str,
+        referrer: &str,
+        _: ResolutionKind,
+    ) -> Result<ModuleSpecifier, JsErrorBox> {
+        if specifier.starts_with("@earendil-works/") {
+            return ModuleSpecifier::parse(&format!("sausage:{specifier}"))
+                .map_err(|e| JsErrorBox::generic(e.to_string()));
+        }
+        deno_core::resolve_import(specifier, referrer)
+            .map_err(|e| JsErrorBox::generic(e.to_string()))
+    }
+    fn load(
+        &self,
+        specifier: &ModuleSpecifier,
+        _: Option<&ModuleLoadReferrer>,
+        _: ModuleLoadOptions,
+    ) -> ModuleLoadResponse {
+        if let Some(source) = self.modules.borrow().get(specifier.as_str()) {
+            return ModuleLoadResponse::Sync(Ok(ModuleSource::new(
+                ModuleType::JavaScript,
+                ModuleSourceCode::String(source.clone().into()),
+                specifier,
+                None,
+            )));
+        }
+        let root = self.root.clone();
+        let specifier = specifier.clone();
+        ModuleLoadResponse::Async(Box::pin(async move {
+            let work = async {
+                let path = specifier
+                    .to_file_path()
+                    .map_err(|_| JsErrorBox::generic(format!("Unknown module {specifier}")))?;
+                let file = root
+                    .open(
+                        path.to_str().unwrap(),
+                        Open {
+                            follow: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .map_err(|e| JsErrorBox::generic(e.to_string()))?;
+                let size = file
+                    .stat()
+                    .await
+                    .map_err(|e| JsErrorBox::generic(e.to_string()))?
+                    .size;
+                let source = String::from_utf8(
+                    file.read(0, size as usize)
+                        .await
+                        .map_err(|e| JsErrorBox::generic(e.to_string()))?,
+                )
+                .map_err(|e| JsErrorBox::generic(e.to_string()))?;
+                transpile(specifier.as_str(), source)
+            }
+            .await?;
+            Ok(ModuleSource::new(
+                ModuleType::JavaScript,
+                ModuleSourceCode::String(work.into()),
+                &specifier,
+                None,
+            ))
+        }))
+    }
+}

host/src/harness/mod.rs

@@ -2,6 +2,7 @@
 mod credentials;
 mod environment;
 pub mod input;
+pub(crate) mod loader;
 use crate::{exec::Exec, world::World};
 use deno_core::{JsRuntime, OpState, RuntimeOptions, op2};
 use deno_error::JsErrorBox;
@@ -87,6 +88,10 @@ deno_core::extension!(
         environment::op_exec,
         environment::op_exec_next,
         input::op_input,
+        environment::op_watch,
+        environment::op_watch_next,
+        loader::op_modules,
+        loader::op_extension_status,
         environment::op_output
     ]
 );
@@ -150,7 +155,12 @@ pub async fn run_module_with_input(
     source: &str,
 ) -> anyhow::Result<()> {
     let proxy = credentials::proxy(&cred)?;
+    let loader = std::rc::Rc::new(loader::Loader {
+        root: world.root.clone(),
+        modules: Default::default(),
+    });
     let mut runtime = JsRuntime::new(RuntimeOptions {
+        module_loader: Some(loader.clone()),
         extensions: vec![
             deno_telemetry::deno_telemetry::init(),
             deno_webidl::deno_webidl::init(),
@@ -166,33 +176,15 @@ pub async fn run_module_with_input(
             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))
+            Ok((loader::transpile(&name, source.to_string())?.into(), None))
         })),
         ..Default::default()
     });
+    runtime.op_state().borrow_mut().put(loader);
+    runtime
+        .op_state()
+        .borrow_mut()
+        .put(loader::status(&world.host_dir));
     runtime
         .op_state()
         .borrow_mut()

host/src/lib.rs

@@ -2,6 +2,7 @@
 //! and the durable agent harness. See `~/research/sausage-v3/PLAN.md` until the docs move here.
 
 pub mod api;
+pub mod builder;
 pub mod cred;
 pub mod exec;
 pub mod harness;

host/src/process/http.rs

@@ -45,9 +45,14 @@ impl WasiHttpHooks for Hooks {
                 .is_some_and(|h| h.eq_ignore_ascii_case("host.sausage"))
             {
                 let bytes = body.collect().await?.to_bytes();
-                return Ok(response(
-                    crate::api::handle(&router, caller, Request::from_parts(parts, bytes)).await,
-                ));
+                let request = Request::from_parts(parts, bytes);
+                if request.method() == http::Method::POST && request.uri().path() == "/cargo" {
+                    return Ok((
+                        crate::api::stream(&router, request),
+                        Box::new(async { Ok(()) }) as Io,
+                    ));
+                }
+                return Ok(response(crate::api::handle(&router, caller, request).await));
             }
             if !matches!(parts.uri.scheme_str(), Some("http" | "https")) {
                 while let Some(frame) = body.frame().await {

host/src/process/mod.rs

@@ -83,6 +83,7 @@ struct Entry {
     command: u32,
     group: u32,
     cancelled: Arc<AtomicBool>,
+    tag: sausage_vfs::journal::Tag,
     abort: AbortHandle,
 }
 /// One runtime keeps compiled components warm and shares the world's namespace across stores.
@@ -191,6 +192,11 @@ impl Runtime {
                 "http://localhost:8081",
             )?),
         };
+        router.attach_journal(world.root.journal());
+        let _ = router.builder.set(Arc::new(crate::builder::Builder::new(
+            world.root.clone(),
+            &world.host_dir,
+        )?));
         let network = Arc::new(net::NetPolicy::open(&world.host_dir)?);
         let engine = compiler.engine.clone();
         let alive = Arc::new(AtomicBool::new(true));
@@ -313,6 +319,7 @@ impl Runtime {
         self.kill_group(group);
         Ok(())
     }
+    #[allow(clippy::too_many_arguments)]
     async fn start(
         self: &Arc<Self>,
         mut request: Request,
@@ -321,6 +328,7 @@ impl Runtime {
         inherited_group: u32,
         conversations: Vec<String>,
         inherited_io: Option<Io>,
+        tool_call: Option<String>,
     ) -> anyhow::Result<Started> {
         anyhow::ensure!(!request.argv.is_empty(), "empty argv");
         let cancelled = if parent == 0 {
@@ -361,6 +369,27 @@ impl Runtime {
         let pre = self.compiler.get(bytes.clone()).await?;
         self.cache.lock().unwrap().insert(bytes);
         let pid = self.next.fetch_add(1, Ordering::Relaxed);
+        let tag = if parent == 0 {
+            let mut tag =
+                sausage_vfs::journal::Tag::process(pid, executable.clone(), conversations.clone())?;
+            if let Some(id) = tool_call {
+                tag.id = id;
+            }
+            tag
+        } else {
+            self.entries
+                .lock()
+                .unwrap()
+                .get(&parent)
+                .ok_or_else(|| anyhow::anyhow!("parent exited"))?
+                .tag
+                .clone()
+        };
+        let root = self
+            .world
+            .root
+            .tagged(tag.clone())
+            .unwrap_or_else(|| self.world.root.clone());
         let command = if command == 0 { pid } else { command };
         let group = match request.group {
             Some(0) => pid,
@@ -384,7 +413,7 @@ impl Runtime {
         };
         let preview1 = if matches!(pre, Compiled::Module(_)) {
             Some(sausage_vfs::preview1::State::new(
-                self.world.root.clone(),
+                root.clone(),
                 &request.cwd,
                 request.argv.clone(),
                 request.env.clone(),
@@ -396,7 +425,7 @@ impl Runtime {
             None
         };
         let mut mounts = Mounts::default();
-        mounts.mount("/", self.world.root.clone(), false);
+        mounts.mount("/", root, false);
         let mut wasi = WasiCtxBuilder::new();
         wasi.args(&request.argv)
             .envs(&request.env)
@@ -499,6 +528,7 @@ impl Runtime {
                     command,
                     group,
                     cancelled: cancelled.clone(),
+                    tag,
                     abort: task.abort_handle(),
                 },
             );
@@ -552,6 +582,7 @@ impl Host for State {
                 self.group,
                 self.conversations.clone(),
                 Some(self.stdio.clone()),
+                None,
             )
             .await
         {
@@ -706,6 +737,7 @@ impl Exec for Arc<Runtime> {
             0,
             options.conversations,
             None,
+            options.tool_call,
         );
         let mut child = if let Some(deadline) = deadline {
             tokio::time::timeout_at(deadline, start)
@@ -737,13 +769,17 @@ impl Exec for Arc<Runtime> {
                 child.code.changed().await?;
             }
         };
-        if let Some(deadline) = deadline {
+        let result = if let Some(deadline) = deadline {
             tokio::time::timeout_at(deadline, run)
                 .await
                 .map_err(|_| anyhow::anyhow!("command timed out"))?
         } else {
             run.await
+        };
+        if let Some(journal) = self.world.root.journal() {
+            journal.retain().await?;
         }
+        result
     }
 }
 /// Construct the harness's shared executor over the same open world.

host/src/process/services.rs

@@ -33,6 +33,7 @@ pub fn supervise(runtime: Arc<Runtime>) -> Supervisor {
                 0,
                 Vec::new(),
                 None,
+                None,
             );
             let child = tokio::select! { _ = &mut stop => return, child = start => child };
             if let Ok(mut child) = child {

host/src/world.rs

@@ -68,6 +68,10 @@ impl World {
             Arc::new(std::sync::Mutex::new(Default::default()));
         let snapshot = processes.clone();
         let mut proc = Generated::default();
+        let extensions = crate::harness::loader::status(&host_dir);
+        proc.insert("/extensions", move || {
+            Ok(extensions.lock().unwrap().as_bytes().to_vec())
+        });
         proc.insert("/processes", move || {
             let mut text = String::from("PID\tPPID\tSTATE\tCWD\tCONVERSATION\tARGV\n");
             for (pid, info) in snapshot.lock().unwrap().iter() {
@@ -84,7 +88,13 @@ impl World {
         });
         mounts.mount("/proc", Arc::new(proc), true);
         mounts.mount("/run", Arc::new(Memory::default()), false);
-        let root: Arc<dyn Layer> = Arc::new(mounts);
+        let journal =
+            sausage_vfs::journal::Journal::open(&host_dir.join("undo"), Arc::new(mounts))?;
+        let root = journal.layer(sausage_vfs::journal::Tag::process(
+            std::process::id(),
+            "/host".into(),
+            Vec::new(),
+        )?);
         Ok(World {
             root,
             host_dir,