Add armed plugin-originated file watches across VFS mutations

Register core.watch before its armed notification, coalesce later changes
without losing post-notification writes, and clean up on caller death.
Observe mounted layers and open file handles, including descriptor/stream
writes, truncation, both rename paths and deletion. Track cancelled
mutations through process idle and retain visible paths across renames.
Document the protocol, expose native core discovery, and exercise watches
through SDK callbacks and raw WASI filesystem operations.

Card: SAU-359

cli/src/core.json

@@ -0,0 +1,19 @@
+{
+  "openrpc": "1.2.6",
+  "info": {"title":"core", "version":"0.1.0"},
+  "methods": [
+    {"name": "rpc.discover", "summary": "Describe the native core API.", "description": "Describe the native core API.", "paramStructure": "by-name", "params": [], "result": {"name": "result", "schema": {"type": "object"}}},
+    {"name": "watch", "summary": "Keep this call open until the caller process ends. First core.notify is armed, then changed after successful plugin-originated file mutations. Scan only after armed; rescan after changed. Events may coalesce; files are the truth. No polling or cancellation API. See PROTOCOL section 4.", "description": "Keep this call open until the caller process ends. First core.notify is armed, then changed after successful plugin-originated file mutations. Scan only after armed; rescan after changed. Events may coalesce; files are the truth. No polling or cancellation API. See PROTOCOL section 4.", "paramStructure": "by-name", "params": [{"name": "path", "required": true, "description": "Virtual file or directory path, including descendants; need not exist. /self is relative to the caller.", "schema": {"type": "string"}}], "result": {"name": "result", "schema": {"type": "null"}}},
+    {"name": "call", "summary": "Make a call and return its result; detach lets the new process outlive its caller.", "description": "Make a call and return its result; detach lets the new process outlive its caller.", "paramStructure": "by-name", "params": [{"name": "method", "required": true, "description": "Qualified method.", "schema": {"type": "string"}}, {"name": "params", "required": true, "description": "Named parameters.", "schema": {"type": "object"}}, {"name": "detach", "required": false, "description": "Detach the new process.", "schema": {"type": "boolean"}}], "result": {"name": "result", "schema": {}}},
+    {"name": "notify", "summary": "Notify this process's caller while its call is open.", "description": "Notify this process's caller while its call is open.", "paramStructure": "by-name", "params": [{"name": "name", "required": true, "description": "Event name.", "schema": {"type": "string"}}, {"name": "params", "required": false, "description": "Event payload.", "schema": {}}], "result": {"name": "result", "schema": {"type": "null"}}},
+    {"name": "stop", "summary": "End a process and await cleanup.", "description": "End a process and await cleanup.", "paramStructure": "by-name", "params": [{"name": "pid", "required": true, "description": "Process ID.", "schema": {"type": "string"}}], "result": {"name": "result", "schema": {"type": "object"}}},
+    {"name": "kill", "summary": "End a process and await cleanup.", "description": "End a process and await cleanup.", "paramStructure": "by-name", "params": [{"name": "pid", "required": true, "description": "Process ID.", "schema": {"type": "string"}}], "result": {"name": "result", "schema": {"type": "object"}}},
+    {"name": "lock", "summary": "Acquire a FIFO process-lifetime lock.", "description": "Acquire a FIFO process-lifetime lock.", "paramStructure": "by-name", "params": [{"name": "path", "required": true, "description": "Lock key.", "schema": {"type": "string"}}, {"name": "wait", "required": false, "description": "Wait for acquisition (default true).", "schema": {"type": "boolean"}}], "result": {"name": "result", "schema": {"type": "object", "properties": {"held": {"type": "boolean"}}, "required": ["held"]}}},
+    {"name": "unlock", "summary": "Release an owned lock.", "description": "Release an owned lock.", "paramStructure": "by-name", "params": [{"name": "path", "required": true, "description": "Lock key.", "schema": {"type": "string"}}], "result": {"name": "result", "schema": {"type": "null"}}},
+    {"name": "activate", "summary": "Install one or more checked plugin versions. See PROTOCOL section 10.", "description": "Install one or more checked plugin versions. See PROTOCOL section 10.", "paramStructure": "by-name", "params": [{"name": "plugin", "required": false, "description": "Single plugin name.", "schema": {"type": "string"}}, {"name": "wasm", "required": false, "description": "Candidate path.", "schema": {"type": "string"}}, {"name": "plugins", "required": false, "description": "Batch candidates.", "schema": {"type": "array", "items": {"type": "object", "properties": {"plugin": {"type": "string"}, "wasm": {"type": "string"}}, "required": ["plugin", "wasm"]}}}, {"name": "force", "required": false, "description": "Skip compatibility rejection.", "schema": {"type": "boolean"}}], "result": {"name": "result", "schema": {"type": "object"}}},
+    {"name": "revert", "summary": "Restore the previous plugin version.", "description": "Restore the previous plugin version.", "paramStructure": "by-name", "params": [{"name": "plugin", "required": true, "description": "Plugin name.", "schema": {"type": "string"}}, {"name": "force", "required": false, "description": "Skip compatibility rejection.", "schema": {"type": "boolean"}}], "result": {"name": "result", "schema": {"type": "object"}}},
+    {"name": "log", "summary": "Read the central journal (PROTOCOL section 7).", "description": "Read the central journal (PROTOCOL section 7).", "paramStructure": "by-name", "params": [{"name": "pid", "required": false, "description": "Process ID.", "schema": {"type": "string"}}, {"name": "plugin", "required": false, "description": "Plugin name.", "schema": {"type": "string"}}, {"name": "since", "required": false, "description": "ISO 8601 start time.", "schema": {"type": "string"}}, {"name": "until", "required": false, "description": "ISO 8601 end time.", "schema": {"type": "string"}}, {"name": "grep", "required": false, "description": "Line substring.", "schema": {"type": "string"}}, {"name": "limit", "required": false, "description": "Maximum entries (default 100).", "schema": {"type": "integer"}}], "result": {"name": "result", "schema": {"type": "array", "items": {"type": "object"}}}},
+    {"name": "restart", "summary": "Request clean host restart through the loop supervisor; needs explicit operator approval.", "description": "Request clean host restart through the loop supervisor; needs explicit operator approval.", "paramStructure": "by-name", "params": [], "result": {"name": "result", "schema": {"type": "object"}}},
+    {"name": "exit", "summary": "Request clean host restart through the loop supervisor; needs explicit operator approval.", "description": "Request clean host restart through the loop supervisor; needs explicit operator approval.", "paramStructure": "by-name", "params": [], "result": {"name": "result", "schema": {"type": "object"}}}
+  ]
+}

cli/src/core.rs

@@ -12,6 +12,8 @@ impl Host {
         p: Value,
     ) -> RpcResult {
         match method {
+            "rpc.discover" => Ok(serde_json::from_str(include_str!("core.json")).unwrap()),
+            "watch" => self.watch(caller, id, required_str(&p, "path")?).await,
             "call" => {
                 self.route(
                     caller.into(),

cli/src/filesystem.rs

@@ -47,27 +47,48 @@ 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 {
         let mut files = Generated::default();
+        files.insert(
+            "/discover.json",
+            || Ok(include_bytes!("core.json").to_vec()),
+        );
         let protocol = include_bytes!("../../PROTOCOL.md");
         files.insert("/PROTOCOL.md", move || Ok(protocol.to_vec()));
         let graph = self.graph_json();

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/main.rs

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

cli/tests/host/watches.rs

@@ -0,0 +1,195 @@
+//! 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;
+}
+
+/// PROTOCOL 9: core discovery includes the armed watch and matches its proc projection.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn discovery() {
+    let (_dir, h) = start().await;
+    let doc = call(&h, "core.rpc.discover", json!({})).await;
+    let projected = call(&h, "probe.read", json!({"path":"/proc/core/discover.json"})).await;
+    assert_eq!(
+        doc,
+        serde_json::from_str::<Value>(projected.as_str().unwrap()).unwrap()
+    );
+    let watch = doc["methods"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .find(|m| m["name"] == "watch")
+        .unwrap();
+    assert_eq!(watch["params"][0]["name"], "path");
+    assert!(watch["description"].as_str().unwrap().contains("armed"));
+    h.stop().await;
+}

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

@@ -468,6 +468,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 +519,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!({})
+}