Add tagged VFS undo journal, host API, and world CLI

host/src/api/mod.rs

@@ -4,6 +4,7 @@ mod auth;
 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};
@@ -23,6 +24,7 @@ pub struct Api {
     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.
@@ -51,8 +53,12 @@ impl Api {
             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 +138,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/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/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,7 @@ impl Runtime {
                 "http://localhost:8081",
             )?),
         };
+        router.attach_journal(world.root.journal());
         let network = Arc::new(net::NetPolicy::open(&world.host_dir)?);
         let engine = compiler.engine.clone();
         let alive = Arc::new(AtomicBool::new(true));
@@ -313,6 +315,7 @@ impl Runtime {
         self.kill_group(group);
         Ok(())
     }
+    #[allow(clippy::too_many_arguments)]
     async fn start(
         self: &Arc<Self>,
         mut request: Request,
@@ -321,6 +324,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 +365,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 +409,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 +421,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 +524,7 @@ impl Runtime {
                     command,
                     group,
                     cancelled: cancelled.clone(),
+                    tag,
                     abort: task.abort_handle(),
                 },
             );
@@ -552,6 +578,7 @@ impl Host for State {
                 self.group,
                 self.conversations.clone(),
                 Some(self.stdio.clone()),
+                None,
             )
             .await
         {
@@ -706,6 +733,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 +765,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

@@ -77,7 +77,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,