Add armed file watches without serializing unrelated IO

cli/src/core.rs

@@ -12,6 +12,7 @@ impl Host {
         p: Value,
     ) -> RpcResult {
         match method {
+            "watch" => self.watch(caller, id, required_str(&p, "path")?).await,
             "call" => {
                 self.route(
                     caller.into(),

cli/src/filesystem.rs

@@ -47,23 +47,40 @@ impl Host {
         tracker: Process,
     ) -> anyhow::Result<Mounts> {
         let root = self.0.world.root.clone();
+        let operations = tracker.clone();
         let host = tracker
             .clone()
             .blocking(move || sausage_vfs::Host::new(root, tracker))
             .await?;
         let mut mounts = Mounts::default();
-        mounts.mount("/", Arc::new(host), false);
+        mounts.mount(
+            "/",
+            self.0
+                .observer
+                .layer(Arc::new(host), "/", operations.clone()),
+            false,
+        );
         mounts.mount("/proc", Arc::new(Proc(self.clone())), true);
         mounts.mount(
             "/self",
-            Arc::new(SelfFolder {
-                host: self.clone(),
-                pid,
-                memory,
-            }),
+            self.0.observer.layer(
+                Arc::new(SelfFolder {
+                    host: self.clone(),
+                    pid: pid.clone(),
+                    memory,
+                }),
+                &format!("/proc/{}", pid.as_deref().unwrap_or("console")),
+                operations.clone(),
+            ),
+            false,
+        );
+        mounts.mount(
+            "/run",
+            self.0
+                .observer
+                .layer(Arc::new(self.0.run.clone()), "/run", operations),
             false,
         );
-        mounts.mount("/run", Arc::new(self.0.run.clone()), false);
         Ok(mounts)
     }
     fn core_files(&self) -> Generated {

cli/src/host.rs

@@ -37,6 +37,8 @@ pub(crate) struct Inner {
     /// processes, and upgrade tasks do nothing.
     pub(crate) shutdown: std::sync::atomic::AtomicBool,
     pub(crate) run: sausage_vfs::Memory,
+    pub(crate) watches: crate::watches::Watches,
+    pub(crate) observer: sausage_vfs::observe::Observer,
     pub(crate) graph: Mutex<crate::graph::Graph>,
     pub(crate) activations: tokio::sync::Mutex<()>,
     pub(crate) watcher: Mutex<Option<notify::RecommendedWatcher>>,
@@ -91,6 +93,10 @@ impl Host {
         ))
     }
     fn with_parts(world: World, runtime: Arc<Runtime>, journal: journal::Journal) -> Self {
+        let watches = crate::watches::Watches::default();
+        let events = watches.clone();
+        let observer =
+            sausage_vfs::observe::Observer::new(move |path| crate::watches::changed(&events, path));
         Self(Arc::new(Inner {
             world,
             runtime,
@@ -109,6 +115,8 @@ impl Host {
             admission: tokio::sync::RwLock::new(()),
             shutdown: std::sync::atomic::AtomicBool::new(false),
             run: sausage_vfs::Memory::default(),
+            watches,
+            observer,
             graph: Mutex::new(crate::graph::Graph::default()),
             activations: tokio::sync::Mutex::new(()),
             watcher: Mutex::new(None),

cli/src/lib.rs

@@ -18,6 +18,7 @@ pub mod runtime;
 mod services;
 mod upgrades;
 mod watcher;
+mod watches;
 pub mod world;
 pub use host::Host;
 pub use native::{NativeContext, NativeFuture, NativePlugin};

cli/src/routing.rs

@@ -51,15 +51,20 @@ impl Host {
             .processes
             .get(pid)
             .and_then(|p| p.answer.as_ref().map(|_| (p.caller.clone(), p.id.clone())));
-        if let Some((caller, id)) = target {
-            let notification = json!({
-                "jsonrpc": "2.0",
-                "method": "core.notify",
-                "params": { "call": id, "name": name, "params": params }
-            });
-            self.send(&caller, notification);
+        if let Some((caller, Some(id))) = target {
+            self.call_notify(&caller, &id, name, params);
         }
     }
+    /// Deliver a notification on an open core call using the same envelope as plugin progress.
+    pub(crate) fn call_notify(&self, caller: &str, id: &Value, name: &str, params: Value) {
+        self.send(
+            caller,
+            json!({
+                "jsonrpc":"2.0", "method":"core.notify",
+                "params": {"call":id, "name":name, "params":params}
+            }),
+        );
+    }
     /// Validate and route a request, then start its native or WASI process.
     pub(crate) fn route(
         &self,

cli/src/watches.rs

@@ -0,0 +1,178 @@
+//! Armed plugin-file watches. A Notify permit coalesces changes without losing a later change.
+use crate::{Host, RpcResult, error};
+use sausage_vfs::observe::under;
+use serde_json::{Value, json};
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex},
+};
+use tokio::sync::Notify;
+use uuid::Uuid;
+
+pub(crate) struct Watch {
+    path: String,
+    changed: Arc<Notify>,
+}
+pub(crate) type Watches = Arc<Mutex<HashMap<Uuid, Watch>>>;
+pub(crate) fn changed(watches: &Watches, path: &str) {
+    let targets: Vec<_> = watches
+        .lock()
+        .unwrap()
+        .values()
+        // Removing/moving an ancestor also invalidates a watch inside that subtree.
+        .filter(|w| under(path, &w.path) || under(&w.path, path))
+        .map(|w| w.changed.clone())
+        .collect();
+    for changed in targets {
+        changed.notify_one();
+    }
+}
+struct Registration {
+    watches: Watches,
+    key: Uuid,
+}
+impl Drop for Registration {
+    fn drop(&mut self) {
+        self.watches.lock().unwrap().remove(&self.key);
+    }
+}
+fn path(caller: &str, path: &str) -> String {
+    let path = sausage_vfs::resolve("/", path);
+    let path = path.trim_end_matches('/');
+    if path.is_empty() {
+        return "/".into();
+    }
+    if under(path, "/self") {
+        format!("/proc/{caller}{}", &path[5..])
+    } else {
+        path.into()
+    }
+}
+impl Host {
+    pub(crate) async fn watch(&self, caller: &str, id: Option<Value>, target: &str) -> RpcResult {
+        let id = id.ok_or_else(|| error(-32600, "watch needs a request ID"))?;
+        let mut ended = self
+            .0
+            .state
+            .lock()
+            .unwrap()
+            .processes
+            .get(caller)
+            .filter(|p| p.info.state == "running")
+            .map(|p| p.ended.subscribe())
+            .ok_or_else(|| error(-32600, "watch needs a running caller process"))?;
+        let changed = Arc::new(Notify::new());
+        let registration = Registration {
+            watches: self.0.watches.clone(),
+            key: Uuid::now_v7(),
+        };
+        self.0.watches.lock().unwrap().insert(
+            registration.key,
+            Watch {
+                path: path(caller, target),
+                changed: changed.clone(),
+            },
+        );
+        self.call_notify(caller, &id, "armed", json!({}));
+        loop {
+            tokio::select! {
+                biased;
+                _ = ended.wait_for(|ended| *ended) => return Ok(Value::Null),
+                _ = changed.notified() => self.call_notify(caller, &id, "changed", json!({})),
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    /// Pending changes coalesce; a change after consuming the last permit is retained.
+    #[tokio::test]
+    async fn coalescing_and_cleanup() {
+        let watches = Watches::default();
+        let n = Arc::new(Notify::new());
+        let r = Registration {
+            watches: watches.clone(),
+            key: Uuid::now_v7(),
+        };
+        watches.lock().unwrap().insert(
+            r.key,
+            Watch {
+                path: "/mailbox".into(),
+                changed: n.clone(),
+            },
+        );
+        changed(&watches, "/mailboxes/a");
+        assert!(
+            tokio::time::timeout(std::time::Duration::ZERO, n.notified())
+                .await
+                .is_err()
+        );
+        changed(&watches, "/mailbox/a");
+        changed(&watches, "/mailbox/b");
+        n.notified().await;
+        assert!(
+            tokio::time::timeout(std::time::Duration::ZERO, n.notified())
+                .await
+                .is_err()
+        );
+        changed(&watches, "/mailbox/c");
+        n.notified().await;
+        drop(r);
+        assert!(watches.lock().unwrap().is_empty());
+        changed(&watches, "/mailbox/d");
+        assert!(
+            tokio::time::timeout(std::time::Duration::ZERO, n.notified())
+                .await
+                .is_err()
+        );
+    }
+    #[test]
+    fn self_alias_and_normalization() {
+        assert_eq!(path("writer", "/self/../self/a/"), "/proc/writer/a");
+        assert_eq!(path("writer", "/selfish/a"), "/selfish/a");
+        assert_eq!(path("writer", "/"), "/");
+    }
+}
+
+#[cfg(test)]
+mod lifetime_tests {
+    use super::*;
+    /// The real caller's ended event releases the registration, without a cancellation RPC.
+    #[tokio::test]
+    async fn caller_death_removes_watch() {
+        let temp = tempfile::tempdir().unwrap();
+        let runtime = crate::runtime::Runtime::new(&temp.path().join("cache")).unwrap();
+        let host = Host::test_world(&temp.path().join("world"), runtime)
+            .await
+            .unwrap();
+        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+        host.register_native(
+            "holder",
+            Arc::new(move |ctx, _, _| {
+                let tx = tx.clone();
+                Box::pin(async move {
+                    tx.send(ctx.pid.clone()).unwrap();
+                    std::future::pending().await
+                })
+            }),
+        );
+        let h = host.clone();
+        let task = tokio::spawn(async move { h.call("holder.hold", json!({})).await });
+        let pid = rx.recv().await.unwrap();
+        let mut watch = Box::pin(host.watch(&pid, Some(json!("watch")), "/mailbox"));
+        assert!(
+            watch
+                .as_mut()
+                .poll(&mut std::task::Context::from_waker(std::task::Waker::noop()))
+                .is_pending()
+        );
+        assert_eq!(host.0.watches.lock().unwrap().len(), 1);
+        host.call("core.kill", json!({"pid":pid})).await.unwrap();
+        watch.await.unwrap();
+        assert!(host.0.watches.lock().unwrap().is_empty());
+        assert!(task.await.unwrap().is_err());
+        host.stop().await;
+    }
+}

cli/tests/host/core.rs

@@ -208,7 +208,11 @@ async fn journal_recovers_after_sqlite_write_failure() {
     call(&h, "core.log", json!({})).await;
     let batches: Vec<Value> = std::fs::read_dir(w.0.join("logs/core/failed-batches"))
         .unwrap()
-        .map(|file| serde_json::from_slice(&std::fs::read(file.unwrap().path()).unwrap()).unwrap())
+        .map(Result::unwrap)
+        // The writer atomically publishes .json files; its temporary files can
+        // still be incomplete while later process-exit entries are spooling.
+        .filter(|file| file.path().extension().is_some_and(|ext| ext == "json"))
+        .map(|file| serde_json::from_slice(&std::fs::read(file.path()).unwrap()).unwrap())
         .collect();
     assert!(
         batches
@@ -1104,17 +1108,16 @@ async fn stop_closes_stdin_and_waits_for_exit() {
     let w = TestWorld::new();
     let h = w.start().await;
     let hc = h.clone();
+    let mut events = h.subscribe();
     let pending = tokio::spawn(async move { hc.call("probe.wait_eof", json!({})).await });
+    // Admission is not readiness: stop may otherwise close stdin before the
+    // request reaches the guest, so wait for the handler to enter its EOF loop.
     let pid = tokio::time::timeout(std::time::Duration::from_secs(20), async {
         loop {
-            if let Some((pid, _)) = h
-                .processes()
-                .into_iter()
-                .find(|(_, p)| p.method == "wait_eof")
-            {
-                break pid;
+            let event = events.recv().await.unwrap();
+            if event["params"]["name"] == "waiting-eof" {
+                break event["params"]["params"]["pid"].clone();
             }
-            tokio::task::yield_now().await;
         }
     })
     .await

cli/tests/host/main.rs

@@ -7,3 +7,4 @@ mod common;
 mod console;
 mod core;
 mod sdk;
+mod watches;

cli/tests/host/watches.rs

@@ -0,0 +1,174 @@
+//! Real SDK watch callbacks and raw WASI filesystem mutations share the host namespace.
+use super::common::{CACHE, probe, sdk_probe};
+use sausage::{Host, world::World};
+use serde_json::{Value, json};
+use std::time::Duration;
+
+// Same call bound as core.rs; no sleeps or polling.
+async fn call(h: &Host, method: &str, p: Value) -> Value {
+    tokio::time::timeout(Duration::from_secs(20), h.call(method, p))
+        .await
+        .unwrap()
+        .unwrap()
+}
+async fn event(h: &Host, rx: &mut tokio::sync::broadcast::Receiver<Value>, name: &str) -> Value {
+    let result = tokio::time::timeout(Duration::from_secs(20), async {
+        loop {
+            let e = rx.recv().await.unwrap();
+            if e["params"]["name"] == name {
+                return e["params"]["params"].clone();
+            }
+        }
+    })
+    .await;
+    match result {
+        Ok(v) => v,
+        Err(e) => {
+            eprintln!(
+                "JOURNAL {}",
+                h.call("core.log", json!({"limit":100})).await.unwrap()
+            );
+            panic!("{e}");
+        }
+    }
+}
+async fn start() -> (tempfile::TempDir, Host) {
+    let dir = tempfile::tempdir().unwrap();
+    let world = World::create(&dir.path().join("world")).unwrap();
+    for (name, wasm) in [("probe", probe()), ("sdk", sdk_probe())] {
+        std::fs::create_dir_all(world.plugin_dir(name)).unwrap();
+        std::fs::copy(wasm, world.plugin_dir(name).join("plugin.wasm")).unwrap();
+    }
+    let host = Host::start_with_cache(&world.root, std::path::Path::new(CACHE))
+        .await
+        .unwrap();
+    (dir, host)
+}
+fn watch(h: &Host, path: &str) -> tokio::task::JoinHandle<sausage::RpcResult> {
+    let (h, path) = (h.clone(), path.to_owned());
+    tokio::spawn(async move { h.call("sdk.watch", json!({"path":path})).await })
+}
+async fn mutate(h: &Host, op: &str, path: &str) {
+    call(h, "probe.mutate", json!({"op":op,"path":path})).await;
+}
+/// PROTOCOL 4: pre-arm state is in the first scan; each post-scan mutation wakes
+/// the SDK without completing the watch. Covers descriptor/stream/truncate paths.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn armed_scan_and_later_changes() {
+    let (_dir, h) = start().await;
+    mutate(&h, "mkdir", "/state/mailbox").await;
+    call(
+        &h,
+        "probe.write",
+        json!({"path":"/state/mailbox/before", "text":"before"}),
+    )
+    .await;
+    let mut rx = h.subscribe();
+    let task = watch(&h, "/state/mailbox/../mailbox/");
+    let armed = event(&h, &mut rx, "armed").await;
+    assert_eq!(armed["entries"], json!(["before"]));
+    assert!(!task.is_finished());
+    for (op, expected) in [
+        ("descriptor", "descriptor"),
+        ("stream", "streamptor"),
+        ("append", "streamptorappend"),
+        ("size", ""),
+        ("descriptor", "descriptor"),
+        ("truncate", ""),
+    ] {
+        mutate(&h, op, "/state/mailbox/before").await;
+        let mut scan = event(&h, &mut rx, "changed").await;
+        while scan["text"] != expected {
+            scan = event(&h, &mut rx, "changed").await;
+        }
+    }
+    mutate(&h, "burst", "/state/mailbox").await;
+    let mut scanned = event(&h, &mut rx, "changed").await;
+    while !scanned["entries"].as_array().unwrap().contains(&json!("4")) {
+        scanned = event(&h, &mut rx, "changed").await;
+    }
+    while rx.try_recv().is_ok() {}
+    mutate(&h, "delete", "/state/mailbox/4").await;
+    let mut scanned = event(&h, &mut rx, "changed").await;
+    while scanned["entries"].as_array().unwrap().contains(&json!("4")) {
+        scanned = event(&h, &mut rx, "changed").await;
+    }
+    call(&h, "core.kill", json!({"pid":armed["pid"]})).await;
+    assert!(task.await.unwrap().is_err());
+    h.stop().await;
+}
+/// PROTOCOL 4: atomic publication, rename source/destination, mounted paths,
+/// absent paths, and caller-death cleanup. No external host writes are required.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn rename_delete_and_mounts() {
+    let (_dir, h) = start().await;
+    for base in ["/state/inbox", "/run/inbox"] {
+        let mut rx = h.subscribe();
+        let task = watch(&h, base);
+        let armed = event(&h, &mut rx, "armed").await;
+        assert_eq!(armed["entries"], json!([]));
+        let staging = format!("{base}-staging");
+        mutate(&h, "mkdir", &staging).await;
+        call(
+            &h,
+            "probe.mutate",
+            json!({"op":"rename","path":staging,"to":base}),
+        )
+        .await;
+        event(&h, &mut rx, "changed").await;
+        mutate(&h, "atomic", &format!("{base}/published")).await;
+        let mut scan = event(&h, &mut rx, "changed").await;
+        while !scan["entries"]
+            .as_array()
+            .unwrap()
+            .contains(&json!("published"))
+        {
+            scan = event(&h, &mut rx, "changed").await;
+        }
+        assert_eq!(
+            call(
+                &h,
+                "probe.read",
+                json!({"path":format!("{base}/published")})
+            )
+            .await,
+            "complete"
+        );
+        while rx.try_recv().is_ok() {}
+        call(
+            &h,
+            "probe.mutate",
+            json!({"op":"rename","path":base,"to":format!("{base}-moved")}),
+        )
+        .await;
+        let mut scan = event(&h, &mut rx, "changed").await;
+        while scan["entries"] != json!([]) {
+            scan = event(&h, &mut rx, "changed").await;
+        }
+        call(&h, "core.kill", json!({"pid":armed["pid"]})).await;
+        assert!(task.await.unwrap().is_err());
+    }
+    // /self watches bind to this caller, and mutations are visible via /proc/<pid>.
+    let mut rx = h.subscribe();
+    let task = watch(&h, "/self");
+    let armed = event(&h, &mut rx, "armed").await;
+    assert!(
+        armed["entries"]
+            .as_array()
+            .unwrap()
+            .contains(&json!("owned"))
+    );
+    event(&h, &mut rx, "changed").await;
+    assert_eq!(
+        call(
+            &h,
+            "probe.read",
+            json!({"path":format!("/proc/{}/owned", armed["pid"].as_str().unwrap())})
+        )
+        .await,
+        "visible via proc"
+    );
+    call(&h, "core.kill", json!({"pid":armed["pid"]})).await;
+    assert!(task.await.unwrap().is_err());
+    h.stop().await;
+}

cli/tests/plugins/probe/src/main.rs

@@ -83,6 +83,9 @@ fn unknown_id() {
 }
 fn wait_eof(lines: &mut Lines) {
     std::fs::write("/state/waiting", "yes").unwrap();
+    send(json!({"jsonrpc":"2.0","method":"core.notify","params":{
+        "name":"waiting-eof", "params":{"pid":std::env::var("SAUSAGE_PID").unwrap()}
+    }}));
     for _ in lines {}
     std::fs::write("/state/stopped", "yes").unwrap();
 }
@@ -468,6 +471,7 @@ fn dispatch(request: &Value, lines: &mut Lines) -> Option<Value> {
             return None;
         }
         "network" => network(params),
+        "mutate" => mutate(params),
         "read" => read_file(params),
         "write" => write_file(params),
         "raw_call" => {
@@ -518,3 +522,68 @@ fn main() {
         }));
     }
 }
+
+/// Separate mutation calls let the host await a scan before each subsequent change.
+fn mutate(p: &Value) -> Value {
+    use wasip2::filesystem::{
+        preopens,
+        types::{DescriptorFlags as D, OpenFlags as O, PathFlags as P},
+    };
+    let path = p["path"].as_str().unwrap();
+    match p["op"].as_str().unwrap() {
+        "mkdir" => std::fs::create_dir_all(path).unwrap(),
+        "rename" => std::fs::rename(path, p["to"].as_str().unwrap()).unwrap(),
+        "delete" => std::fs::remove_file(path).unwrap(),
+        "rmdir" => std::fs::remove_dir(path).unwrap(),
+        "atomic" => {
+            let temp = format!("{path}.tmp");
+            std::fs::write(&temp, "complete").unwrap();
+            std::fs::rename(temp, path).unwrap();
+        }
+        "burst" => {
+            for i in 0..5 {
+                std::fs::write(format!("{path}/{i}"), "complete").unwrap();
+            }
+        }
+        op => {
+            let root = preopens::get_directories().remove(0).0;
+            let file = root
+                .open_at(
+                    P::empty(),
+                    path.trim_start_matches('/'),
+                    O::empty(),
+                    D::READ | D::WRITE,
+                )
+                .unwrap();
+            match op {
+                "descriptor" => {
+                    file.write(b"descriptor", 0).unwrap();
+                }
+                "stream" => {
+                    file.write_via_stream(0)
+                        .unwrap()
+                        .blocking_write_and_flush(b"stream")
+                        .unwrap();
+                }
+                "append" => {
+                    file.append_via_stream()
+                        .unwrap()
+                        .blocking_write_and_flush(b"append")
+                        .unwrap();
+                }
+                "size" => file.set_size(0).unwrap(),
+                "truncate" => {
+                    root.open_at(
+                        P::empty(),
+                        path.trim_start_matches('/'),
+                        O::TRUNCATE,
+                        D::WRITE,
+                    )
+                    .unwrap();
+                }
+                _ => panic!("unknown mutation"),
+            }
+        }
+    }
+    json!({})
+}