Wait for busy guest readiness before measuring shutdown grace

Cargo.lock

@@ -2092,7 +2092,9 @@ dependencies = [
  "rusqlite",
  "rustyline",
  "sausage-interface",
+ "sausage-plugin",
  "sausage-vfs",
+ "schemars 1.2.3",
  "serde",
  "serde_json",
  "serde_yaml_ng",

PROTOCOL.md

@@ -222,6 +222,7 @@ these areas fails.
 /proc/
   core/
     PROTOCOL.md          this file
+    discover.json        generated core.rpc.discover document
     graph.json           plugins, imports, and the result of each check
     graph.dot            the same graph for Graphviz
     locks.json           held locks and waiting processes
@@ -249,6 +250,40 @@ Rules:
 
 Principle: **inspect through files, act through calls.**
 
+### File watching
+
+`core.watch {"path": "/state/session/mailbox"}` stays open for the caller
+process's lifetime. It needs a request ID and a running plugin caller.
+Its first `core.notify` is `"armed"`, with empty params. Only then scan the
+files. Writes before arming are visible in that scan; writes after arming
+produce `"changed"` notifications (also with empty params). Rescan after
+each notification; files, not event counts, are the truth.
+
+Notifications follow successful plugin-originated file mutations: creation,
+truncation, descriptor/stream writes, rename (both source and destination),
+and deletion, including mounted `/run` and `/self` files. Paths are normalized
+in the virtual namespace; `/self` means the caller's `/proc/<pid>` folder.
+A watch matches itself and descendants, not sibling prefixes. Moving or
+removing an ancestor also invalidates it. Paths need not exist when armed.
+Open file writes follow plugin renames; unlinked/replaced handles no longer
+name visible files. External host writes are outside this contract.
+
+Changes may coalesce, including a change already visible in the initial
+scan, but a change after the last notification always produces another.
+Delivery holds no filesystem locks. Observation does not serialize descriptor
+IO; only overlapping opens and rename/removal wait for one another.
+Watch the narrowest useful path: delivery is not rate-limited, so watching a
+busy parent can cause a rescan for every write.
+
+The host-side watch ends only when its caller process exits, not on a change
+notification; there is no cancellation API or polling/retry loop. SDK callers
+should poll the watch future alongside their handler's work and drop it on
+return, not keep it in a background task that prevents an idle process from
+exiting. Dropping the future removes the SDK's pending entry and callback, so
+late notifications and responses are ignored; it does not cancel the host
+watch. A long-lived service keeps one watch per purpose for its lifetime,
+rather than repeatedly opening and dropping watches.
+
 ## 5. Endpoints
 
 An endpoint is a JSON-RPC interface that a running process serves over TCP,
@@ -355,7 +390,9 @@ At start, the core starts each service in `/config/boot.yaml`, in order:
 
 | Method | Params | Result |
 | --- | --- | --- |
+| `core.rpc.discover` | none | generated OpenRPC 1.2 document for every native core method |
 | `core.call` | `method`, `params`, `detach`? | the result of the method |
+| `core.watch` | `path` | stays open; `armed`, then `changed` notifications (section 4) |
 | `core.notify` | `name`, `params` | none (a notification) |
 | `core.stop` | `pid` | `{}` when the process has ended |
 | `core.kill` | `pid` | `{}` |
@@ -367,6 +404,17 @@ At start, the core starts each service in `/config/boot.yaml`, in order:
 | `core.restart` | none | `{}`. With `SAUSAGE_LOOP=1`, schedule clean shutdown and exit code 75. |
 | `core.exit` | none | Alias of `core.restart`: same supervisor gate, response, coalescing, and exit code 75. |
 
+`core.rpc.discover` and `/proc/core/discover.json` expose the same document,
+with unprefixed method names, just like a plugin's discovery document. The
+core generates it from the parameter and result types used by its dispatch;
+there is no separate JSON schema file to maintain. Copy the proc file to
+`imports/core.json` and use `import!("core")` to get typed calls, including
+`core::lock` and `core::log`. Dotted wire names become underscores in Rust
+(`rpc.discover` becomes `rpc_discover`). Import checks recognize the native
+core as a provider without requiring a `plugins/core` directory. Forwarded
+results and discovery documents contain arbitrary JSON; other core results
+are typed, including activation summaries and journal entries.
+
 ### Host restart
 
 **HARD RULE: the mediator and all agents NEVER call `core.restart` without
@@ -437,7 +485,8 @@ imported method:
 - There are no version numbers. This check replaces them.
 - The check covers a fixed subset of JSON Schema: objects with `properties`
   and `required`, the primitive types, arrays, optional and nullable values,
-  string enums, integer ranges, and `$ref`. A construct outside the subset
+  string enums, integer ranges, typed dictionaries (`additionalProperties`
+  with a value schema), and `$ref`. A construct outside the subset
   fails the check.
 - The check covers calls, not state. Old and new processes share the files in
   `/state`, and `core.revert` restores only code. A plugin must keep its

cli/Cargo.toml

@@ -17,6 +17,8 @@ http-body-util = "0.1"
 include_dir = "0.7"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+schemars = "1"
+sausage-plugin = { path = "../sdk" }
 serde_yaml_ng = "0.10"
 tokio = { version = "1", features = ["full"] }
 toml_edit = "0.25"

cli/src/activation.rs

@@ -1,6 +1,8 @@
 //! Serialized candidate-copy activation and revert under PROTOCOL.md section 10.3.
 use crate::{
-    Host, RpcResult, error,
+    Host, RpcResult,
+    core::{ActivationSummary, InterfaceCheck},
+    error,
     graph::{Interface, io_error, problems, read_hashed},
 };
 use serde_json::{Value, json};
@@ -246,7 +248,7 @@ impl Host {
                 .map_err(io_error)?;
         }
         let mut upgrades = Vec::new();
-        let mut summaries = serde_json::Map::new();
+        let mut summaries = BTreeMap::new();
         for candidate in candidates {
             let old = previous.get(&candidate.plugin);
             let before = methods(old.map(|old| &old.document).unwrap_or(&Value::Null));
@@ -254,27 +256,40 @@ impl Host {
             let added: Vec<_> = after
                 .keys()
                 .filter(|name| !before.contains_key(*name))
+                .cloned()
                 .collect();
             let removed: Vec<_> = before
                 .keys()
                 .filter(|name| !after.contains_key(*name))
+                .cloned()
                 .collect();
             let changed: Vec<_> = after
                 .iter()
                 .filter(|(name, method)| before.get(*name).is_some_and(|old| old != *method))
-                .map(|(name, _)| name)
+                .map(|(name, _)| name.clone())
                 .collect();
-            summaries.insert(candidate.plugin.clone(), json!({
+            summaries.insert(
-                "plugin": candidate.plugin,
+                candidate.plugin.clone(),
-                "wasm": format!("/plugins/{}/plugin.wasm", candidate.plugin),
+                ActivationSummary {
-                "hash": candidate.interface.hash,
+                    plugin: candidate.plugin.clone(),
-                "previous": old.map(|old| &old.hash).filter(|hash| !hash.is_empty()),
+                    wasm: format!("/plugins/{}/plugin.wasm", candidate.plugin),
-                "method_count": candidate.interface.document["methods"].as_array().map_or(0, Vec::len),
+                    hash: candidate.interface.hash.clone(),
-                "added": added,
+                    previous: old
-                "removed": removed,
+                        .map(|old| old.hash.clone())
-                "changed": changed,
+                        .filter(|hash| !hash.is_empty()),
-                "interface_check": if checked { "passed" } else { "skipped" },
+                    method_count: candidate.interface.document["methods"]
-            }));
+                        .as_array()
+                        .map_or(0, Vec::len),
+                    added,
+                    removed,
+                    changed,
+                    interface_check: if checked {
+                        InterfaceCheck::Passed
+                    } else {
+                        InterfaceCheck::Skipped
+                    },
+                },
+            );
             if let Some(old) = previous
                 .get(&candidate.plugin)
                 .filter(|old| !old.hash.is_empty() && old.hash != candidate.interface.hash)
@@ -297,6 +312,6 @@ impl Host {
         for upgrade in upgrades {
             self.begin_upgrade(upgrade);
         }
-        Ok(Value::Object(summaries))
+        Ok(json!(summaries))
     }
 }

cli/src/core.rs

@@ -1,74 +1,159 @@
-//! The native core methods and their JSON parameters.
+//! Native core dispatch and OpenRPC, generated from one typed method registry.
-use crate::{Host, RpcResult, error};
+mod types;
-use serde_json::{Value, json};
+use types::*;
+pub(crate) use types::{ActivationSummary, Held, InterfaceCheck};
 
-impl Host {
+use crate::{Host, RpcResult, error};
-    /// Execute a core method without starting a guest process.
+use sausage_plugin::{Empty, document, finish_document, method};
+use serde_json::{Value, json};
+use std::sync::OnceLock;
+
+// Every dispatched method must supply docs, a parameter type, and a result
+// type. The same entry generates its descriptor; neither list can drift.
+macro_rules! core_methods {
+    ($host:ident, $caller:ident, $id:ident, $wire:ident, $p:ident;
+     $($name:literal: $params:ty => $result:ty, $docs:literal $body:block)*) => {
+        #[cfg(test)]
+        const METHOD_NAMES: &[&str] = &[$($name),*];
+
+        pub(crate) fn discover() -> &'static Value {
+            static DOCUMENT: OnceLock<Value> = OnceLock::new();
+            DOCUMENT.get_or_init(|| {
+                let mut doc = document("core", env!("CARGO_PKG_VERSION"));
+                $(method::<$params, $result>(&mut doc, $name, $docs);)*
+                finish_document(doc)
+            })
+        }
+
+        impl Host {
+            /// Execute a native method without starting a guest process.
             pub(crate) async fn core(
-        &self,
+                &$host,
-        caller: &str,
+                $caller: &str,
-        id: Option<Value>,
+                $id: Option<Value>,
-        method: &str,
+                $wire: &str,
-        p: Value,
+                $p: Value,
             ) -> RpcResult {
-        match method {
+                match $wire {
-            "call" => {
+                    $($name => {
-                self.route(
+                        let $p: $params = params($p)?;
-                    caller.into(),
+                        let result: $result = $body;
-                    id,
+                        Ok(json!(result))
-                    required_str(&p, "method")?.into(),
+                    },)*
-                    p.get("params").cloned().unwrap_or(json!({})),
-                    p["detach"].as_bool().unwrap_or(false),
-                )
-                .await
-            }
-            "notify" => {
-                self.notify(
-                    caller,
-                    required_str(&p, "name")?,
-                    p.get("params").cloned().unwrap_or(json!({})),
-                );
-                Ok(Value::Null)
-            }
-            "kill" => {
-                let pid = required_str(&p, "pid")?;
-                let ended = self
-                    .0
-                    .state
-                    .lock()
-                    .unwrap()
-                    .processes
-                    .get(pid)
-                    .map(|p| p.ended.subscribe());
-                self.kill(pid);
-                if let Some(mut ended) = ended {
-                    let _ = ended.wait_for(|ended| *ended).await;
-                }
-                Ok(json!({}))
-            }
-            "stop" => {
-                if let Some(mut n) = self.begin_stop(required_str(&p, "pid")?) {
-                    let _ = n.wait_for(|ended| *ended).await;
-                }
-                Ok(json!({}))
-            }
-            "lock" => {
-                self.lock(caller, required_str(&p, "path")?, p["wait"] != false)
-                    .await
-            }
-            "unlock" => {
-                self.unlock(caller, required_str(&p, "path")?);
-                Ok(Value::Null)
-            }
-            "log" => Ok(self.0.journal.query(p).await),
-            "activate" | "revert" => self.activation(caller, method, p).await,
-            "restart" | "exit" => self.request_restart(),
                     _ => Err(error(-32601, "method not found")),
                 }
             }
+        }
+    };
 }
-fn required_str<'a>(params: &'a Value, name: &str) -> Result<&'a str, Value> {
+
-    params[name]
+core_methods! { self, caller, id, wire, p;
-        .as_str()
+    "rpc.discover": Empty => Json,
-        .ok_or_else(|| error(-32602, &format!("{name} must be a string")))
+    "Describe every native core method as OpenRPC 1.2, generated from its wire types." {
+        let _ = p;
+        Json(discover().clone())
+    }
+    "watch": Path => (),
+    "Watch a path for the running caller's lifetime. Needs a request ID; sends armed, then changed notifications with empty params. Scan after armed and rescan after changed." {
+        self.watch(caller, id, &p.path).await?;
+    }
+    "call": Call => Json,
+    "Call a qualified method, optionally detached. Returns that method's result unchanged." {
+        Json(self.route(caller.into(), id, p.method, p.params.0, p.detach.unwrap_or(false)).await?)
+    }
+    "notify": Notify => (),
+    "Notify the caller about progress. Normally sent as a notification, without an ID." {
+        self.notify(caller, &p.name, p.params.0);
+    }
+    "kill": Pid => Empty,
+    "Kill a process immediately; return after it has ended. An unknown process is a no-op." {
+        let ended = self.0.state.lock().unwrap().processes.get(&p.pid).map(|p| p.ended.subscribe());
+        self.kill(&p.pid);
+        if let Some(mut ended) = ended {
+            let _ = ended.wait_for(|ended| *ended).await;
+        }
+        Empty {}
+    }
+    "stop": Pid => Empty,
+    "Close a process's stdin and wait until it ends, killing it after the five-second stop budget. An unknown process is a no-op." {
+        if let Some(mut ended) = self.begin_stop(&p.pid) {
+            let _ = ended.wait_for(|ended| *ended).await;
+        }
+        Empty {}
+    }
+    "lock": Lock => Held,
+    "Acquire an advisory FIFO lock. Wait unless wait is false; locks are released when their owner ends." {
+        result(self.lock(caller, &p.path, p.wait.unwrap_or(true)).await?)?
+    }
+    "unlock": Path => (),
+    "Release an advisory lock held by this caller. Normally sent as a notification, without an ID." {
+        self.unlock(caller, &p.path);
+    }
+    "log": Log => Vec<crate::journal::Entry>,
+    "Query the journal. Return the newest matching entries in ascending entry-ID order." {
+        result(self.0.journal.query(json!(p)).await)?
+    }
+    "activate": Activate => ActivationResult,
+    "Copy, discover, check and install one or more plugin WASMs. Return short summaries keyed by plugin name; force skips interface checks." {
+        result(self.activation(caller, wire, json!(p)).await?)?
+    }
+    "revert": Revert => ActivationResult,
+    "Restore one or more plugins from plugin.prev.wasm, using the same checks and summaries as activate." {
+        result(self.activation(caller, wire, json!(p)).await?)?
+    }
+    "restart": Empty => Empty,
+    "Request a clean host restart. Requires the person's explicit approval each time and the SAUSAGE_LOOP=1 supervisor marker; otherwise fails with -32600." {
+        let _ = p;
+        self.request_restart()?;
+        Empty {}
+    }
+    "exit": Empty => Empty,
+    "Legacy alias of restart: same explicit-approval rule, supervisor gate and coalesced shutdown request." {
+        let _ = p;
+        self.request_restart()?;
+        Empty {}
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{runtime::Runtime, world::World};
+    use std::collections::BTreeSet;
+
+    /// Every descriptor names a dispatched method, and vice versa.
+    #[tokio::test]
+    async fn discovery_covers_dispatch_exactly() {
+        let dir = tempfile::tempdir().unwrap();
+        let world = dir.path().join("world");
+        World::create(&world).unwrap();
+        let host = Host::test_world(&world, Runtime::new(&dir.path().join("cache")).unwrap())
+            .await
+            .unwrap();
+        let doc = host.call("core.rpc.discover", json!({})).await.unwrap();
+        let names: BTreeSet<_> = doc["methods"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .map(|m| {
+                assert!(!m["description"].as_str().unwrap().is_empty());
+                m["name"].as_str().unwrap()
+            })
+            .collect();
+        assert_eq!(names, METHOD_NAMES.iter().copied().collect());
+        assert_eq!(names.len(), doc["methods"].as_array().unwrap().len());
+        for name in names {
+            let response = host.call(&format!("core.{name}"), json!({})).await;
+            if let Err(error) = response {
+                assert_ne!(error["code"], -32601, "descriptor without dispatch: {name}");
+            }
+        }
+        assert_eq!(
+            host.call("core.no_such_method", json!({}))
+                .await
+                .unwrap_err()["code"],
+            -32601
+        );
+        host.stop().await;
+    }
 }

cli/src/core/types.rs

@@ -0,0 +1,200 @@
+//! Native wire types shared by dispatch, discovery, and result producers.
+use sausage_plugin::{Params, SchemaType, schema};
+use serde::{Deserialize, Deserializer, Serialize};
+use serde_json::Value;
+use std::collections::BTreeMap;
+
+// Core forwarding/discovery carry arbitrary JSON, and activation returns a map
+// keyed by plugin name. These native contracts are wider than #[schema]'s
+// plugin subset; schemars still derives them from their actual wire types.
+#[derive(Serialize, Deserialize, schemars::JsonSchema)]
+#[serde(transparent)]
+pub(super) struct Json(pub Value);
+impl SchemaType for Json {}
+
+/// Forward a call, optionally detached from its caller.
+#[derive(Serialize, Deserialize, schemars::JsonSchema)]
+pub(super) struct Call {
+    /// Qualified target method.
+    pub method: String,
+    /// Target parameters; omitted means an empty object.
+    #[serde(default = "empty_params")]
+    pub params: Json,
+    /// Detach the new process; default false.
+    #[serde(default, deserialize_with = "optional")]
+    pub detach: Option<bool>,
+}
+fn empty_params() -> Json {
+    Json(serde_json::json!({}))
+}
+
+/// Notify the caller of progress.
+#[schema]
+pub(super) struct Notify {
+    /// Notification name.
+    pub name: String,
+    /// Notification payload; omitted means an empty object.
+    #[serde(default = "empty_params")]
+    pub params: Json,
+}
+
+/// Select a process.
+#[schema]
+pub(super) struct Pid {
+    /// Process ID.
+    pub pid: String,
+}
+
+/// Select a virtual path or advisory lock name.
+#[schema]
+pub(super) struct Path {
+    /// Path to observe or unlock.
+    pub path: String,
+}
+
+/// Acquire an advisory FIFO lock.
+#[derive(Serialize, Deserialize, schemars::JsonSchema)]
+pub(super) struct Lock {
+    /// Advisory lock name (usually a state file path).
+    pub path: String,
+    /// Wait for ownership unless false; omitted or null means true.
+    #[serde(default, deserialize_with = "optional")]
+    pub wait: Option<bool>,
+}
+
+/// Lock ownership after acquisition.
+#[schema]
+pub(crate) struct Held {
+    /// Whether the caller holds the lock.
+    pub held: bool,
+}
+
+/// Query journal entries, returned in ascending ID order.
+#[derive(Serialize, Deserialize, schemars::JsonSchema)]
+pub(super) struct Log {
+    /// Exact plugin name.
+    #[serde(default, deserialize_with = "optional")]
+    pub plugin: Option<String>,
+    /// Exact process ID.
+    #[serde(default, deserialize_with = "optional")]
+    pub pid: Option<String>,
+    /// Inclusive lower entry ID.
+    #[serde(default, deserialize_with = "optional")]
+    pub since: Option<String>,
+    /// Inclusive upper entry ID.
+    #[serde(default, deserialize_with = "optional")]
+    pub until: Option<String>,
+    /// Literal substring in the line.
+    #[serde(default, deserialize_with = "optional")]
+    pub grep: Option<String>,
+    /// Maximum newest entries; default 100, negative means no limit.
+    #[serde(default, deserialize_with = "optional")]
+    pub limit: Option<i64>,
+}
+
+// Keep the existing core's forgiving optional filters/flags: values of the
+// wrong type act like omitted values, rather than adding new validation.
+fn optional<'de, T: serde::de::DeserializeOwned, D: Deserializer<'de>>(
+    deserializer: D,
+) -> Result<Option<T>, D::Error> {
+    Ok(serde_json::from_value(Value::deserialize(deserializer)?).ok())
+}
+
+/// One candidate for activation.
+#[schema]
+pub(super) struct ActivatePlugin {
+    /// Plugin name.
+    pub plugin: String,
+    /// Candidate WASM world path.
+    pub wasm: String,
+}
+
+/// Activate one candidate, or a group checked and installed together.
+#[derive(Serialize, Deserialize, schemars::JsonSchema)]
+pub(super) struct Activate {
+    /// Single plugin name, used when plugins is omitted.
+    pub plugin: Option<String>,
+    /// Single candidate WASM world path, used when plugins is omitted.
+    pub wasm: Option<String>,
+    /// Group of candidates; takes precedence over plugin and wasm. Must not be empty.
+    pub plugins: Option<Vec<ActivatePlugin>>,
+    /// Skip interface checks only when true; default false.
+    #[serde(default, deserialize_with = "optional")]
+    pub force: Option<bool>,
+}
+
+/// One plugin to restore from plugin.prev.wasm.
+#[schema]
+pub(super) struct RevertPlugin {
+    /// Plugin name.
+    pub plugin: String,
+}
+
+/// Revert one plugin or a group to their previous WASM files.
+#[derive(Serialize, Deserialize, schemars::JsonSchema)]
+pub(super) struct Revert {
+    /// Single plugin name, used when plugins is omitted.
+    pub plugin: Option<String>,
+    /// Group of plugins; takes precedence over plugin. Must not be empty.
+    pub plugins: Option<Vec<RevertPlugin>>,
+    /// Skip interface checks only when true; default false.
+    #[serde(default, deserialize_with = "optional")]
+    pub force: Option<bool>,
+}
+
+/// A short summary of one installed plugin.
+#[schema]
+pub(crate) struct ActivationSummary {
+    /// Plugin name.
+    pub plugin: String,
+    /// Installed world path.
+    pub wasm: String,
+    /// SHA-256 of the installed copy.
+    pub hash: String,
+    /// Replaced hash, or null on first installation.
+    pub previous: Option<String>,
+    /// Number of methods in the main interface.
+    pub method_count: usize,
+    /// Sorted names added to the interface.
+    pub added: Vec<String>,
+    /// Sorted names removed from the interface.
+    pub removed: Vec<String>,
+    /// Sorted names whose description or reachable schema changed.
+    pub changed: Vec<String>,
+    /// Whether interface checks passed or were skipped.
+    pub interface_check: InterfaceCheck,
+}
+
+/// Outcome of the activation's interface check.
+#[derive(Serialize, Deserialize, schemars::JsonSchema)]
+#[schemars(inline)]
+#[serde(rename_all = "lowercase")]
+pub(crate) enum InterfaceCheck {
+    /// The interface checks passed.
+    Passed,
+    /// The caller explicitly skipped the checks.
+    Skipped,
+}
+
+impl SchemaType for InterfaceCheck {}
+
+#[derive(Serialize, Deserialize, schemars::JsonSchema)]
+#[serde(transparent)]
+pub(super) struct ActivationResult(pub BTreeMap<String, ActivationSummary>);
+impl SchemaType for ActivationResult {}
+
+// Native optional fields retain the core's lenient deserializers, which are
+// intentionally outside the SDK attribute macro's plugin schema subset.
+macro_rules! native_params {
+    ($($name:ty),*) => { $(impl SchemaType for $name {} impl Params for $name {})* };
+}
+native_params!(Call, Lock, Log, Activate, Revert);
+
+// Parameter decoding stays at the core boundary, not in the schema builder.
+pub(super) fn params<P: Params + serde::de::DeserializeOwned>(p: Value) -> Result<P, Value> {
+    serde_json::from_value(p).map_err(|error| crate::error(-32602, &error.to_string()))
+}
+
+pub(super) fn result<R: serde::de::DeserializeOwned>(value: Value) -> Result<R, Value> {
+    serde_json::from_value(value).map_err(|error| crate::error(-32603, &error.to_string()))
+}

cli/src/filesystem.rs

@@ -47,29 +47,49 @@ 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",
+            self.0.observer.layer(
                 Arc::new(SelfFolder {
                     host: self.clone(),
-                pid,
+                    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();
         let protocol = include_bytes!("../../PROTOCOL.md");
         files.insert("/PROTOCOL.md", move || Ok(protocol.to_vec()));
+        files.insert("/discover.json", || {
+            Ok(serde_json::to_vec(crate::core::discover()).unwrap())
+        });
         let graph = self.graph_json();
         files.insert("/graph.json", move || Ok(graph.clone()));
         let dot = self.graph_dot();

cli/src/graph.rs

@@ -61,6 +61,15 @@ pub(crate) fn problems(
                 {
                     continue;
                 }
+                // Core is a native provider, not an installed plugin or cached WASM.
+                if provider == "core" {
+                    for problem in
+                        sausage_interface::compat::check(expected, crate::core::discover())
+                    {
+                        failures.push(json!({"consumer":consumer,"provider":provider,"status":"broken","method":problem.method,"message":problem.message,"counterexample":problem.counterexample,"display":problem.to_string()}));
+                    }
+                    continue;
+                }
                 let Some(actual) = plugins.get(provider) else {
                     failures.push(json!({"consumer":consumer,"provider":provider,"status":"missing","message":"the provider does not exist"}));
                     continue;

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

@@ -9,15 +9,23 @@ use std::{
     sync::{Mutex, mpsc},
 };
 use tokio::sync::oneshot;
-#[derive(Debug, Serialize, Deserialize)]
+/// One journal entry, with its original ID and timestamp.
-struct Entry {
+#[derive(Debug, Serialize, Deserialize, schemars::JsonSchema)]
+pub(crate) struct Entry {
+    /// UTC timestamp.
     time: String,
+    /// UUIDv7 entry ID, ordered by time.
     id: String,
+    /// Process ID.
     pid: String,
+    /// Plugin name.
     plugin: String,
+    /// stderr or core.
     stream: String,
+    /// Complete log line.
     line: String,
 }
+impl sausage_plugin::SchemaType for Entry {}
 enum Command {
     Write(Entry),
     Query(Value, oneshot::Sender<Value>),
@@ -233,9 +241,26 @@ fn query_database(db: &Connection, filters: &Value) -> rusqlite::Result<Value> {
         AND (?3 IS NULL OR id>=?3) AND (?4 IS NULL OR id<=?4)
         AND (?5 IS NULL OR instr(line,?5)>0) ORDER BY id DESC LIMIT ?6",
     )?;
-    let rows=statement.query_map(params![filters["plugin"].as_str(),filters["pid"].as_str(),filters["since"].as_str(),filters["until"].as_str(),filters["grep"].as_str(),filters["limit"].as_i64().unwrap_or(100)],|row| Ok(json!({
+    let rows = statement.query_map(
-        "id":row.get::<_,String>(0)?,"time":row.get::<_,String>(1)?,"pid":row.get::<_,String>(2)?,"plugin":row.get::<_,String>(3)?,"stream":row.get::<_,String>(4)?,"line":row.get::<_,String>(5)?,
+        params![
-    })))?;
+            filters["plugin"].as_str(),
+            filters["pid"].as_str(),
+            filters["since"].as_str(),
+            filters["until"].as_str(),
+            filters["grep"].as_str(),
+            filters["limit"].as_i64().unwrap_or(100)
+        ],
+        |row| {
+            Ok(json!(Entry {
+                id: row.get(0)?,
+                time: row.get(1)?,
+                pid: row.get(2)?,
+                plugin: row.get(3)?,
+                stream: row.get(4)?,
+                line: row.get(5)?,
+            }))
+        },
+    )?;
     let mut entries: Vec<_> = rows.collect::<rusqlite::Result<_>>()?;
     entries.reverse();
     Ok(json!(entries))

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

@@ -1,6 +1,6 @@
 //! Advisory FIFO locks and cleanup of ended processes.
-use crate::error;
 use crate::{Host, RpcResult, host::State};
+use crate::{core::Held, error};
 use serde_json::json;
 use std::collections::VecDeque;
 use tokio::sync::oneshot;
@@ -26,7 +26,7 @@ impl Host {
             }
             if let Some(lock) = state.locks.get_mut(path) {
                 if !wait {
-                    return Ok(json!({"held":false}));
+                    return Ok(json!(Held { held: false }));
                 }
                 let (tx, rx) = oneshot::channel();
                 lock.queue.push_back((caller.into(), tx));
@@ -46,7 +46,7 @@ impl Host {
             Some(rx) => rx
                 .await
                 .unwrap_or_else(|_| Err(error(-32603, "lock canceled"))),
-            None => Ok(json!({"held":true})),
+            None => Ok(json!(Held { held: true })),
         }
     }
     /// Release a held lock only when this process owns it.
@@ -80,7 +80,7 @@ impl Host {
 pub(crate) fn grant(state: &mut State, path: &str) {
     if let Some(mut lock) = state.locks.remove(path) {
         while let Some((owner, tx)) = lock.queue.pop_front() {
-            if tx.send(Ok(json!({"held":true}))).is_ok() {
+            if tx.send(Ok(json!(Held { held: true }))).is_ok() {
                 lock.owner = owner;
                 state.locks.insert(path.into(), lock);
                 break;

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 {
+        if let Some((caller, Some(id))) = target {
-            let notification = json!({
+            self.call_notify(&caller, &id, name, params);
-                "jsonrpc": "2.0",
-                "method": "core.notify",
-                "params": { "call": id, "name": name, "params": params }
-            });
-            self.send(&caller, notification);
         }
     }
+    /// 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/services.rs

@@ -4,7 +4,7 @@ use serde_json::{Value, json};
 use std::{collections::VecDeque, sync::atomic::Ordering, time::Duration};
 use tokio::time::Instant;
 
-// Agreed systemd-derived values: /docs/design/plugin-handover/design.md,
+// Systemd-derived values:
 // https://www.man7.org/linux/man-pages/man5/systemd-system.conf.5.html.
 const CRASH_LIMIT: usize = 5;
 const CRASH_WINDOW: Duration = Duration::from_secs(10);

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

@@ -741,7 +741,7 @@ async fn standalone_builder() {
         1
     );
 }
-/// SAU-126: a named CPU-bound WASI test times out, while other instances still run.
+/// a named CPU-bound WASI test times out, while other instances still run.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn named_test_timeout_and_mutex_release() {
     let world = TestWorld::new();
@@ -815,7 +815,7 @@ async fn child_ended(child: u32) {
     .expect("cargo descendant survived cancellation/timeout");
 }
 
-/// SAU-126: an explicit small total budget kills build.rs and unblocks a queued build.
+/// an explicit small total budget kills build.rs and unblocks a queued build.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn build_timeout_and_queued_build() {
     let world = TestWorld::new();
@@ -853,7 +853,7 @@ async fn build_timeout_and_queued_build() {
     host.stop().await;
 }
 
-/// SAU-126: the same total deadline covers release compilation and retains test results.
+/// the same total deadline covers release compilation and retains test results.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn release_build_timeout_retains_tests() {
     let world = TestWorld::new();
@@ -903,7 +903,7 @@ fn main() {
     host.stop().await;
 }
 
-/// SAU-126: omission selects the real 900-second deadline, without a 15-minute test.
+/// omission selects the real 900-second deadline, without a 15-minute test.
 #[tokio::test]
 async fn omitted_build_timeout_uses_default() {
     let world = TestWorld::new();
@@ -928,7 +928,7 @@ async fn omitted_build_timeout_uses_default() {
     host.stop().await;
 }
 
-/// SAU-126: a caller ending stops its attached native build and releases the mutex.
+/// a caller ending stops its attached native build and releases the mutex.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn ended_caller_releases_build_mutex() {
     let world = TestWorld::new();
@@ -970,7 +970,7 @@ async fn ended_caller_releases_build_mutex() {
     host.stop().await;
 }
 
-/// SAU-126: discovery exposes non-null defaults; invalid budgets cannot disable limits.
+/// discovery exposes non-null defaults; invalid budgets cannot disable limits.
 #[tokio::test]
 async fn timeout_params_discovery_and_validation() {
     let world = TestWorld::new();

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
@@ -1011,6 +1015,7 @@ async fn stop_deadlines() {
 async fn shutdown_stops_busy_processes_in_parallel_after_five_seconds() {
     let w = TestWorld::new();
     let h = w.start().await;
+    let mut notifications = h.subscribe();
 
     let mut tasks = Vec::new();
     for _ in 0..2 {
@@ -1020,14 +1025,14 @@ async fn shutdown_stops_busy_processes_in_parallel_after_five_seconds() {
         }));
     }
     tokio::time::timeout(std::time::Duration::from_secs(20), async {
-        while h
+        // Admission precedes guest startup. Stopping at admission can close
-            .processes()
+        // stdin before the guest receives its busy request, so there is then
-            .values()
+        // no busy work that needs the five-second kill budget.
-            .filter(|p| p.plugin == "probe" && p.method == "busy")
+        let mut ready = 0;
-            .count()
+        while ready != tasks.len() {
-            != 2
+            if notifications.recv().await.unwrap()["params"]["name"] == "busy-ready" {
-        {
+                ready += 1;
-            tokio::task::yield_now().await;
+            }
         }
     })
     .await
@@ -1104,17 +1109,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
+            let event = events.recv().await.unwrap();
-                .processes()
+            if event["params"]["name"] == "waiting-eof" {
-                .into_iter()
+                break event["params"]["params"]["pid"].clone();
-                .find(|(_, p)| p.method == "wait_eof")
-            {
-                break pid;
             }
-            tokio::task::yield_now().await;
         }
     })
     .await

cli/tests/host/core_discovery.rs

@@ -0,0 +1,146 @@
+//! Generated native discovery, proc projection, and a real SDK import consumer.
+use sausage::{Host, world::World};
+use sausage_vfs::{Layer, Open};
+use serde_json::{Value, json};
+use std::{path::Path, time::Duration};
+
+async fn call(host: &Host, method: &str, params: Value) -> Value {
+    // Same call budget as the core/watches host tests.
+    tokio::time::timeout(Duration::from_secs(20), host.call(method, params))
+        .await
+        .unwrap()
+        .unwrap()
+}
+
+async fn read(host: &Host, path: &str) -> Value {
+    let fs = host.filesystem().await.unwrap();
+    let file = fs.open(path, Open::default()).await.unwrap();
+    let bytes = file
+        .read(0, file.stat().await.unwrap().size as usize)
+        .await
+        .unwrap();
+    serde_json::from_slice(&bytes).unwrap()
+}
+
+/// No saved core.json: the RPC, proc file, importer and admission use generated types.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn generated_core_import_runs_under_wasi() {
+    let dir = tempfile::tempdir().unwrap();
+    let world = dir.path().join("world");
+    World::create(&world).unwrap();
+    let host = Host::start_with_cache(&world, Path::new(super::common::CACHE))
+        .await
+        .unwrap();
+    let doc = call(&host, "core.rpc.discover", json!({})).await;
+    assert_eq!(doc["openrpc"], "1.2.6");
+    assert_eq!(doc["info"]["title"], "core");
+    assert_eq!(read(&host, "/proc/core/discover.json").await, doc);
+    let fs = host.filesystem().await.unwrap();
+    assert!(
+        fs.list("/proc/core")
+            .await
+            .unwrap()
+            .iter()
+            .any(|e| e.name == "discover.json")
+    );
+    assert!(
+        fs.open(
+            "/proc/core/discover.json",
+            Open {
+                write: true,
+                ..Open::default()
+            }
+        )
+        .await
+        .is_err()
+    );
+    let schemas = &doc["components"]["schemas"];
+    assert_eq!(schemas["Held"]["properties"]["held"]["type"], "boolean");
+    assert_eq!(schemas["Entry"]["properties"]["line"]["type"], "string");
+    let problems = sausage_interface::compat::check(&doc, &doc);
+    assert!(problems.is_empty(), "{problems:?}");
+
+    let source = dir.path().join("consumer");
+    std::fs::create_dir_all(source.join("imports")).unwrap();
+    std::fs::create_dir_all(source.join("src")).unwrap();
+    // This is exactly the user's copy-from-proc workflow, not a JSON fixture.
+    std::fs::write(source.join("imports/core.json"), doc.to_string()).unwrap();
+    let sdk = Path::new(env!("CARGO_MANIFEST_DIR"))
+        .parent()
+        .unwrap()
+        .join("sdk");
+    std::fs::write(
+        source.join("Cargo.toml"),
+        format!(
+            r#"
+[package]
+name = "native-core-consumer"
+version = "0.1.0"
+edition = "2024"
+[workspace]
+[dependencies]
+sausage-plugin = {{ path = {sdk:?} }}
+"#
+        ),
+    )
+    .unwrap();
+    std::fs::write(
+        source.join("src/main.rs"),
+        r#"
+use sausage_plugin::{Result, rpc, serde_json::json};
+sausage_plugin::import!("core");
+struct Consumer;
+#[rpc]
+impl Consumer {
+    /// Exercise actual generated parameter and result types.
+    async fn run(&self) -> Result<bool> {
+        let held = core::lock(core::Lock { path: "/typed-lock".into(), wait: Some(false) }).await?;
+        let filters: core::Log = sausage_plugin::serde_json::from_value(json!({"limit":1}))?;
+        for entry in core::log(filters).await? {
+            let _: String = entry.line;
+        }
+        core::unlock(core::Path { path: "/typed-lock".into() }).await?;
+        let doc = core::rpc_discover().await?;
+        Ok(held.held && doc["info"]["title"] == "core")
+    }
+}
+fn main() { sausage_plugin::run(Consumer) }
+"#,
+    )
+    .unwrap();
+    let target = Path::new(concat!(env!("CARGO_TARGET_TMPDIR"), "/guests"));
+    assert!(
+        std::process::Command::new("cargo")
+            .args([
+                "build",
+                "--offline",
+                "--target",
+                "wasm32-wasip2",
+                "--manifest-path"
+            ])
+            .arg(source.join("Cargo.toml"))
+            .arg("--target-dir")
+            .arg(target)
+            .status()
+            .unwrap()
+            .success()
+    );
+    std::fs::copy(
+        target.join("wasm32-wasip2/debug/native-core-consumer.wasm"),
+        world.join("candidate.wasm"),
+    )
+    .unwrap();
+    let activation = call(
+        &host,
+        "core.activate",
+        json!({"plugin":"consumer","wasm":"/candidate.wasm"}),
+    )
+    .await;
+    assert_eq!(activation["consumer"]["interface_check"], "passed");
+    assert_eq!(call(&host, "consumer.run", json!({})).await, true);
+    assert_eq!(
+        read(&host, "/proc/core/graph.json").await["plugins"]["consumer"]["imports"]["core"]["status"],
+        "compatible"
+    );
+    host.stop().await;
+}

cli/tests/host/main.rs

@@ -6,4 +6,6 @@ mod cli;
 mod common;
 mod console;
 mod core;
+mod core_discovery;
 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,10 +83,16 @@ 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();
 }
 fn busy() -> Value {
+    send(json!({"jsonrpc":"2.0", "method":"core.notify", "params":{
+        "name":"busy-ready", "params":{}
+    }}));
     loop {
         std::hint::spin_loop();
     }
@@ -468,6 +474,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 +525,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!({})
+}

interface/src/client.rs

@@ -48,8 +48,9 @@ pub fn client_rs(name: &str, doc: &Value) -> Result<String, String> {
         let mut result = vec![];
         for method in methods.as_array().into_iter().flatten() {
             let wire = method["name"].as_str().ok_or("method name missing")?;
-            let ident = syn::parse_str::<syn::Ident>(wire)
+            let rust_name = wire.replace('.', "_");
-                .or_else(|_| syn::parse_str::<syn::Ident>(&format!("r#{wire}")))
+            let ident = syn::parse_str::<syn::Ident>(&rust_name)
+                .or_else(|_| syn::parse_str::<syn::Ident>(&format!("r#{rust_name}")))
                 .map_err(|e| e.to_string())?;
             let mut rs = method["result"]["schema"].clone();
             local_refs(&mut rs);

interface/src/compat.rs

@@ -204,7 +204,7 @@ fn supported(sender: &Value, document: &Value, seen: &mut HashSet<usize>) -> Res
                     supported(child, document, seen)?;
                 }
             }
-            "additionalProperties" if value.is_boolean() => (),
+            "additionalProperties" => supported(value, document, seen)?,
             "enum"
                 if value
                     .as_array()
@@ -447,6 +447,15 @@ fn compare_variants(
             {
                 let error = if let Some(other) = receiver["properties"].get(key) {
                     subset(child, other, sender_doc, receiver_doc, active).err()
+                } else if receiver["additionalProperties"].is_object() {
+                    subset(
+                        child,
+                        &receiver["additionalProperties"],
+                        sender_doc,
+                        receiver_doc,
+                        active,
+                    )
+                    .err()
                 } else if receiver["additionalProperties"] == false {
                     Some((
                         format!(
@@ -471,6 +480,42 @@ fn compare_variants(
                     ));
                 }
             }
+            // A typed dictionary can send arbitrary keys, unlike a struct's
+            // closed set of output fields. Check both receiver named fields
+            // and its catch-all value contract.
+            if let Some(extra) = sender.get("additionalProperties").filter(|v| v.is_object()) {
+                let mut key = "extra".to_string();
+                while sender["properties"].get(&key).is_some()
+                    || receiver["properties"].get(&key).is_some()
+                {
+                    key.push('!');
+                }
+                let catch_all = receiver
+                    .get("additionalProperties")
+                    .unwrap_or(&Value::Bool(true));
+                let targets = receiver["properties"]
+                    .as_object()
+                    .into_iter()
+                    .flatten()
+                    .filter(|(key, _)| sender["properties"].get(*key).is_none())
+                    .chain(std::iter::once((&key, catch_all)));
+                for (key, target) in targets {
+                    if let Err((message, example)) =
+                        subset(extra, target, sender_doc, receiver_doc, active)
+                    {
+                        let mut object = sample(sender, sender_doc, &mut HashSet::new());
+                        if let (Some(Value::Object(object)), Some(value)) = (&mut object, example) {
+                            object.insert(key.clone(), value);
+                        } else {
+                            object = None;
+                        }
+                        return Err((
+                            format!("the dictionary field {key:?} is incompatible: {message}"),
+                            object,
+                        ));
+                    }
+                }
+            }
         }
         "array" => {
             if let Err((message, example)) = subset(

interface/tests/compat_maps.rs

@@ -0,0 +1,57 @@
+//! Typed dictionary results retain their value contracts in both directions.
+use sausage_interface::compat::check;
+use serde_json::{Value, json};
+
+fn document(schema: Value) -> Value {
+    json!({"methods":[{"name":"values","params":[],"result":{"schema":schema}}]})
+}
+fn map(value: Value) -> Value {
+    json!({"type":"object","additionalProperties":value})
+}
+
+#[test]
+fn dictionary_values_are_covariant() {
+    let strings = document(map(json!({"type":"string"})));
+    let numbers = document(map(json!({"type":"integer"})));
+    let short = document(map(json!({"type":"string","enum":["ok"]})));
+    assert!(check(&strings, &strings).is_empty());
+    assert!(check(&strings, &short).is_empty());
+    assert!(!check(&short, &strings).is_empty());
+    let errors = check(&strings, &numbers);
+    assert_eq!(errors.len(), 1);
+    assert_eq!(errors[0].counterexample, Some(json!({"extra":0})));
+}
+
+#[test]
+fn dictionaries_and_named_fields_check_each_other() {
+    let strings = document(map(json!({"type":"string"})));
+    let named_string = document(json!({"type":"object","properties":{"name":{"type":"string"}}}));
+    let named_number = document(json!({"type":"object","properties":{"name":{"type":"integer"}}}));
+    assert!(check(&strings, &named_string).is_empty());
+    assert!(check(&named_string, &strings).is_empty());
+    assert!(!check(&strings, &named_number).is_empty());
+    assert!(!check(&named_number, &strings).is_empty());
+    let closed = document(json!({"type":"object","additionalProperties":false}));
+    assert!(!check(&closed, &strings).is_empty());
+    assert!(check(&strings, &closed).is_empty());
+    let required = document(
+        json!({"type":"object","properties":{"name":{"type":"string"}},"required":["name"]}),
+    );
+    assert!(!check(&required, &strings).is_empty());
+}
+
+#[test]
+fn dictionary_references_are_checked_not_ignored() {
+    let mut expected = document(map(json!({"$ref":"#/components/schemas/Value"})));
+    expected["components"]["schemas"]["Value"] = json!({"type":"string"});
+    assert!(check(&expected, &expected).is_empty());
+    let mut actual = expected.clone();
+    actual["components"]["schemas"]["Value"] = json!({"type":"boolean"});
+    assert!(!check(&expected, &actual).is_empty());
+    actual["components"]["schemas"]["Value"] = json!({"pattern":"x"});
+    assert!(
+        check(&expected, &actual)[0]
+            .message
+            .contains("unsupported schema construct pattern")
+    );
+}

sdk/src/calls.rs

@@ -1,5 +1,19 @@
 //! Outgoing core calls and notification callbacks.
 use super::*;
+/// Removes a dropped call's local routing state, without cancelling host work.
+struct PendingCall(u64);
+impl Drop for PendingCall {
+    fn drop(&mut self) {
+        let pending = STATE.with(|s| s.borrow_mut().pending.remove(&self.0));
+        if let Some(mut pending) = pending {
+            // A lock may still be granted: release it when its late response arrives.
+            if let Some(lease) = pending.unlock.take() {
+                STATE.with(|s| s.borrow_mut().abandoned_locks.insert(self.0, lease));
+            }
+            // Drop user-owned callback captures outside the STATE borrow.
+        }
+    }
+}
 /// An outgoing request builder.
 #[must_use = "await the call to send it"]
 pub struct Call {
@@ -26,6 +40,21 @@ impl Call {
         self
     }
     /// Receive progress notifications before the result arrives.
+    /// Dropping the call future removes its local pending entry and callback;
+    /// later notifications and responses are ignored. It does not cancel host work.
+    ///
+    /// For `core.watch`, poll the call future alongside work in the handler, then
+    /// drop it when the handler returns. Do not put it in `background()` just to
+    /// keep the watch alive: background tasks keep the process alive, while the
+    /// watch only ends when the process ends. Dropping the future does not cancel
+    /// the host watch, but it lets an otherwise idle process exit normally. A
+    /// long-lived service should keep one watch per purpose for its lifetime,
+    /// rather than repeatedly opening and dropping watches on the host.
+    ///
+    /// Scan only after `armed`, and rescan on `changed`. Events may coalesce, but
+    /// are not rate-limited: watch the narrowest useful path to avoid rescanning
+    /// for unrelated writes (for example, watch a mailbox, not all session data).
+    /// Do not use endpoint connect retries to wait for file publication.
     pub fn on_notify(mut self, f: impl FnMut(&str, Value) + 'static) -> Self {
         self.notify = Some(Box::new(f));
         self
@@ -60,6 +89,7 @@ impl IntoFuture for Call {
                 );
                 Ok(id)
             })?;
+            let _pending = PendingCall(id);
             let (method, params) = if self.detach {
                 (
                     "core.call".to_string(),
@@ -69,7 +99,8 @@ impl IntoFuture for Call {
                 (self.method, self.params)
             };
             if let Err(e) = send(json!({"jsonrpc":"2.0","id":id,"method":method,"params":params})) {
-                STATE.with(|s| s.borrow_mut().pending.remove(&id));
+                let pending = STATE.with(|s| s.borrow_mut().pending.remove(&id));
+                drop(pending);
                 return Err(e);
             }
             rx.await
@@ -77,3 +108,154 @@ impl IntoFuture for Call {
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{STATE, State, block_on, call, clear_pending, lock, receive};
+    use crate::Result;
+    use futures::{FutureExt, future::LocalBoxFuture};
+    use serde_json::{Value, json};
+    use std::{cell::RefCell, future::IntoFuture, rc::Rc, task::Context};
+
+    fn activate() {
+        STATE.with(|s| {
+            *s.borrow_mut() = State {
+                active: true,
+                ..State::default()
+            };
+        });
+    }
+    fn start(
+        mut call: LocalBoxFuture<'static, Result<Value>>,
+    ) -> (u64, LocalBoxFuture<'static, Result<Value>>) {
+        assert!(
+            call.as_mut()
+                .poll(&mut Context::from_waker(std::task::Waker::noop()))
+                .is_pending()
+        );
+        (STATE.with(|s| s.borrow().next), call)
+    }
+    fn notification(id: u64, name: &str) {
+        receive(json!({"method":"core.notify","params":{"call":id,"name":name,"params":{}}}));
+    }
+    fn response(id: u64, result: Value) {
+        receive(json!({"id":id,"result":result}));
+    }
+
+    #[test]
+    fn dropped_call_removes_pending_and_callback() {
+        activate();
+        let events = Rc::new(RefCell::new(Vec::new()));
+        let captured = events.clone();
+        let (id, future) = start(
+            call("core.watch", json!({"path":"/mailbox"}))
+                .on_notify(move |name, _| captured.borrow_mut().push(name.to_owned()))
+                .into_future(),
+        );
+        notification(id, "armed");
+        assert_eq!(&*events.borrow(), &["armed"]);
+        assert_eq!(Rc::strong_count(&events), 2);
+        drop(future);
+        STATE.with(|s| {
+            assert!(s.borrow().pending.is_empty());
+            assert!(s.borrow().abandoned_locks.is_empty());
+        });
+        assert_eq!(Rc::strong_count(&events), 1);
+        notification(id, "changed");
+        response(id, Value::Null);
+        receive(json!({"id":id,"error":{"code":-32000,"message":"late error"}}));
+        assert_eq!(&*events.borrow(), &["armed"]);
+    }
+
+    #[test]
+    fn kept_watch_delivers_until_result() {
+        activate();
+        let events = Rc::new(RefCell::new(Vec::new()));
+        let captured = events.clone();
+        let (id, future) = start(
+            call("core.watch", json!({"path":"/mailbox"}))
+                .on_notify(move |name, _| captured.borrow_mut().push(name.to_owned()))
+                .into_future(),
+        );
+        notification(id, "armed");
+        notification(id, "changed");
+        notification(id, "changed");
+        response(id, json!({"done":true}));
+        assert_eq!(block_on(future).unwrap(), json!({"done":true}));
+        notification(id, "late");
+        assert_eq!(&*events.borrow(), &["armed", "changed", "changed"]);
+        STATE.with(|s| assert!(s.borrow().pending.is_empty()));
+        assert_eq!(Rc::strong_count(&events), 1);
+    }
+
+    #[test]
+    fn callback_can_drop_its_own_call_without_being_restored() {
+        activate();
+        let holder = Rc::new(RefCell::new(None));
+        let captured = holder.clone();
+        let (id, future) = start(
+            call("core.watch", json!({"path":"/mailbox"}))
+                .on_notify(move |_, _| {
+                    // Panics on a second invocation; the first must remove the entry.
+                    drop(captured.borrow_mut().take().expect("live call"));
+                })
+                .into_future(),
+        );
+        *holder.borrow_mut() = Some(future);
+        notification(id, "armed");
+        STATE.with(|s| assert!(s.borrow().pending.is_empty()));
+        notification(id, "changed");
+        response(id, Value::Null);
+        assert_eq!(Rc::strong_count(&holder), 1);
+    }
+
+    #[test]
+    fn eof_cleanup_can_drop_callback_owned_futures() {
+        activate();
+        let holder = Rc::new(RefCell::new(None));
+        let weak = Rc::downgrade(&holder);
+        let captured = holder.clone();
+        let (_, future) = start(
+            call("core.watch", json!({"path":"/mailbox"}))
+                .on_notify(move |_, _| {
+                    let _ = &captured;
+                })
+                .into_future(),
+        );
+        *holder.borrow_mut() = Some(future);
+        drop(holder);
+        STATE.with(|s| s.borrow_mut().stopping = true);
+        clear_pending();
+        assert!(weak.upgrade().is_none());
+        STATE.with(|s| assert!(s.borrow().pending.is_empty()));
+    }
+
+    #[test]
+    fn unpolled_call_never_registers() {
+        activate();
+        drop(call("core.watch", json!({"path":"/mailbox"})).into_future());
+        STATE.with(|s| {
+            assert!(s.borrow().pending.is_empty());
+            assert_eq!(s.borrow().next, 0);
+        });
+    }
+
+    #[test]
+    fn dropped_lock_keeps_only_late_grant_cleanup() {
+        activate();
+        let (id, future) = start(
+            lock("/late")
+                .into_future()
+                .map(|r| r.map(|_| Value::Null))
+                .boxed_local(),
+        );
+        drop(future);
+        STATE.with(|s| {
+            assert!(s.borrow().pending.is_empty());
+            assert_eq!(s.borrow().abandoned_locks.len(), 1);
+        });
+        notification(id, "ignored");
+        response(id, json!({"held":true}));
+        STATE.with(|s| assert!(s.borrow().abandoned_locks.is_empty()));
+    }
+}

sdk/src/transport.rs

@@ -99,6 +99,8 @@ struct State {
     draining: bool,
     next: u64,
     pending: HashMap<u64, Pending>,
+    // Only leases, never callbacks or response senders, survive dropped lock acquisitions.
+    abandoned_locks: HashMap<u64, std::rc::Rc<Lease>>,
     #[cfg(target_os = "wasi")]
     tasks: usize,
     #[cfg(target_os = "wasi")]
@@ -109,6 +111,18 @@ struct State {
     listeners: HashMap<u64, futures::future::AbortHandle>,
 }
 thread_local! { static STATE: RefCell<State> = RefCell::new(State::default()); }
+#[cfg(any(target_os = "wasi", test))]
+fn clear_pending() {
+    let pending = STATE.with(|s| {
+        let mut s = s.borrow_mut();
+        (
+            std::mem::take(&mut s.pending),
+            std::mem::take(&mut s.abandoned_locks),
+        )
+    });
+    // Callback captures may own other call futures whose drop guards use STATE.
+    drop(pending);
+}
 pub(crate) fn endpoint_name<T: 'static>() -> Result<String> {
     STATE
         .with(|s| {
@@ -224,16 +238,16 @@ pub fn test_run<F: Future>(f: F) -> F::Output {
             STATE.with(|s| {
                 let mut s = s.borrow_mut();
                 s.stopping = true;
-                s.pending.clear();
             });
+            clear_pending();
         });
         let result = f.await;
         drop(reader);
         STATE.with(|s| {
             let mut s = s.borrow_mut();
             s.active = false;
-            s.pending.clear();
         });
+        clear_pending();
         result
     })
 }
@@ -305,9 +319,18 @@ pub fn receive(value: Value) {
                 .set(result.as_ref().is_ok_and(|v| v["held"] == true));
         }
         let _ = p.tx.send(result);
+    } else {
+        let lease = value["id"]
+            .as_u64()
+            .and_then(|id| STATE.with(|s| s.borrow_mut().abandoned_locks.remove(&id)));
+        if let Some(lease) = lease {
+            lease
+                .held
+                .set(value.get("error").is_none() && value["result"]["held"] == true);
         } else {
             eprintln!("note: response with unknown ID {}", value["id"]);
         }
+    }
 }
 
 /// Serve the main interface until its work ends or stdin reaches EOF.
@@ -356,11 +379,11 @@ async fn serve<T: Interface>(value: T) {
                 STATE.with(|s| {
                     let mut s = s.borrow_mut();
                     s.stopping = true;
-                    s.pending.clear();
                     for (_, listener) in s.listeners.drain() {
                         listener.abort();
                     }
                 });
+                clear_pending();
                 value.stop().await;
                 break;
             }
@@ -411,8 +434,8 @@ async fn serve<T: Interface>(value: T) {
     STATE.with(|s| {
         let mut s = s.borrow_mut();
         s.active = false;
-        s.pending.clear();
     });
+    clear_pending();
     let _ = std::io::stdout().flush();
     std::process::exit(0);
 }

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

@@ -2,6 +2,12 @@
 use sausage_plugin::serde_json::{Value, json};
 use sausage_plugin::{Ctx, Result, call, lock, rpc, schema};
 struct Probe;
+/// A virtual path to observe.
+#[schema]
+struct WatchPath {
+    /// Directory or file, not necessarily present yet.
+    path: String,
+}
 /// The values to add.
 #[schema]
 struct Add {
@@ -30,6 +36,98 @@ struct Echo {
 }
 #[rpc(grace_seconds = 12,endpoints(control=Control))]
 impl Probe {
+    /// Scan only after the watch is armed; keep the call open until process death.
+    async fn watch(&self, p: WatchPath) -> Result<()> {
+        let path = p.path.clone();
+        call("core.watch", json!({"path":p.path}))
+            .on_notify(move |name, _| {
+                if name == "armed" && path == "/self" {
+                    std::fs::write("/self/owned", "visible via proc").unwrap();
+                }
+                let mut entries: Vec<_> = std::fs::read_dir(&path)
+                    .into_iter()
+                    .flatten()
+                    .filter_map(std::result::Result::ok)
+                    .map(|e| e.file_name().into_string().unwrap())
+                    .collect();
+                entries.sort();
+                let text = std::fs::read_to_string(format!("{path}/before")).ok();
+                Ctx::default()
+                    .notify(
+                        name,
+                        json!({"pid":sausage_plugin::pid(), "entries":entries, "text":text}),
+                    )
+                    .unwrap();
+            })
+            .await?;
+        Ok(())
+    }
+
+    /// Poll the watch in the handler, then return idle without a background task.
+    async fn watch_once(&self, p: WatchPath) -> Result<()> {
+        use std::{
+            cell::RefCell,
+            rc::Rc,
+            task::{Poll, Waker},
+        };
+        let state = Rc::new(RefCell::new((false, None::<Waker>)));
+        let received = state.clone();
+        let mut watch = std::pin::pin!(std::future::IntoFuture::into_future(
+            call("core.watch", json!({"path":p.path})).on_notify(move |name, _| {
+                if name == "armed" {
+                    let mut state = received.borrow_mut();
+                    state.0 = true;
+                    if let Some(waker) = state.1.take() {
+                        waker.wake();
+                    }
+                }
+            })
+        ));
+        std::future::poll_fn(|cx| {
+            if let Poll::Ready(result) = watch.as_mut().poll(cx) {
+                return Poll::Ready(result.map(|_| ()));
+            }
+            let mut state = state.borrow_mut();
+            state.1 = Some(cx.waker().clone());
+            if state.0 {
+                Poll::Ready(Ok(()))
+            } else {
+                Poll::Pending
+            }
+        })
+        .await
+    }
+
+    /// Drop one watch while another still delivers, in a process that remains alive.
+    async fn drop_watch(&self) -> Result<Vec<String>> {
+        use std::{cell::RefCell, future::IntoFuture, rc::Rc, task::Poll};
+        let events = Rc::new(RefCell::new(Vec::new()));
+        let captured = events.clone();
+        let mut dropped = call("core.watch", json!({"path":"/dropped"}))
+            .on_notify(move |_, _| captured.borrow_mut().push("dropped".to_owned()))
+            .into_future();
+        std::future::poll_fn(|cx| {
+            assert!(dropped.as_mut().poll(cx).is_pending());
+            Poll::Ready(())
+        })
+        .await;
+        drop(dropped);
+        let captured = events.clone();
+        let mut kept = call("core.watch", json!({"path":"/kept"}))
+            .on_notify(move |name, _| captured.borrow_mut().push(name.to_owned()))
+            .into_future();
+        std::future::poll_fn(|cx| {
+            assert!(kept.as_mut().poll(cx).is_pending());
+            Poll::Ready(())
+        })
+        .await;
+        // The harness sends late traffic and kept-watch events before answering.
+        call("other.barrier", json!({})).await?;
+        drop(kept);
+        let result = events.borrow().clone();
+        Ok(result)
+    }
+
     /// Publish a control endpoint until EOF.
     async fn serve(&self) -> Result<()> {
         sausage_plugin::listen("/endpoint", std::rc::Rc::new(Control)).await

sdk/tests/wasi.rs

@@ -108,6 +108,34 @@ fn protocol() {
         }
         p.finishes();
     }
+    // A handler-owned watch does not keep an idle process alive after return,
+    // even though the host never answers the watch and stdin remains open.
+    let mut p = start(&wasm, &dir);
+    p.request("watch_once", json!({"path":"/mailbox"}));
+    let watch = p.read();
+    assert_eq!(watch["method"], "core.watch");
+    p.send(json!({"jsonrpc":"2.0","method":"core.notify","params":{"call":watch["id"],"name":"armed","params":{}}}));
+    assert_eq!(p.read()["result"], Value::Null);
+    p.finishes();
+    // Keep the process alive after dropping a call: late traffic is ignored,
+    // while the kept watch continues to deliver every notification.
+    let mut p = start(&wasm, &dir);
+    p.request("drop_watch", json!({}));
+    let dropped = p.read();
+    let kept = p.read();
+    let barrier = p.read();
+    assert_eq!(dropped["params"]["path"], "/dropped");
+    assert_eq!(kept["params"]["path"], "/kept");
+    assert_eq!(barrier["method"], "other.barrier");
+    for watch in [&dropped, &kept] {
+        for name in ["armed", "changed", "changed"] {
+            p.send(json!({"jsonrpc":"2.0","method":"core.notify","params":{"call":watch["id"],"name":name,"params":{}}}));
+        }
+    }
+    p.answer(&dropped, Value::Null);
+    p.answer(&barrier, json!({}));
+    assert_eq!(p.read()["result"], json!(["armed", "changed", "changed"]));
+    p.finishes();
     let mut p = start(&wasm, &dir);
     p.request("rpc.discover", json!({}));
     let d = p.read();
@@ -267,7 +295,21 @@ fn protocol() {
     assert_eq!(provider.read()["params"]["name"], "draining");
     line.clear();
     assert_eq!(idle.read_line(&mut line).unwrap(), 0);
-    assert!(std::net::TcpStream::connect(&address).is_err());
+    // drain() wakes independent accept/read cancellations. Idle EOF does not
+    // order listener destruction: TCP may still complete a handshake, but that
+    // connection must close without serving work. Do not retry the connect.
+    if let Ok(mut late) = std::net::TcpStream::connect(&address) {
+        eprintln!("TCP handshake raced listener cancellation; waiting for close");
+        // Reuse the four-second socket bound used by idle/active above.
+        late.set_read_timeout(Some(std::time::Duration::from_secs(4)))
+            .unwrap();
+        let closed = std::io::Read::read(&mut late, &mut [0]);
+        assert!(
+            matches!(closed, Ok(0))
+                || matches!(&closed, Err(e) if e.kind() == std::io::ErrorKind::ConnectionReset),
+            "late connection survived drain: {closed:?}"
+        );
+    }
     assert!(provider.child.try_wait().unwrap().is_none());
     provider.answer(&inflight, json!({}));
     line.clear();
@@ -279,6 +321,7 @@ fn protocol() {
     assert!(provider.child.try_wait().unwrap().is_none());
     provider.answer(&background, json!({}));
     provider.finishes();
+    assert!(std::net::TcpStream::connect(&address).is_err());
     assert!(
         !dir.join("stopped").exists(),
         "drain is not the EOF stop hook"

vfs/src/lib.rs

@@ -5,6 +5,7 @@ pub mod generated;
 pub mod host;
 pub mod memory;
 pub mod mounts;
+pub mod observe;
 pub mod process;
 mod readonly;
 mod streams;

vfs/src/observe.rs

@@ -0,0 +1,615 @@
+//! Observes successful namespace and open-file mutations, even if their caller is cancelled.
+use crate::*;
+use std::{
+    collections::BTreeMap,
+    sync::{Mutex, Weak},
+};
+use tokio::sync::{OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock};
+
+type Namespace = Mutex<BTreeMap<String, Weak<RwLock<()>>>>;
+enum Held {
+    Read { _guard: OwnedRwLockReadGuard<()> },
+    Write { _guard: OwnedRwLockWriteGuard<()> },
+}
+
+type TrackedPath = Mutex<Option<String>>;
+type Path = Arc<TrackedPath>;
+type Callback = dyn Fn(&str) + Send + Sync;
+/// Shared observation for mount namespaces. Paths supplied to the callback are virtual paths.
+/// Callbacks run after filesystem locks have been released; they must not block.
+#[derive(Clone)]
+pub struct Observer {
+    changed: Arc<Callback>,
+    paths: Arc<Mutex<Vec<Weak<TrackedPath>>>>,
+    // Only open-registration versus rename/remove needs ordering. Lock the affected
+    // subtrees, not all IO: a blocked open must not stall unrelated namespaces.
+    namespace: Arc<Namespace>,
+}
+impl Observer {
+    pub fn new(changed: impl Fn(&str) + Send + Sync + 'static) -> Self {
+        Self {
+            changed: Arc::new(changed),
+            paths: Arc::default(),
+            namespace: Arc::default(),
+        }
+    }
+    /// Wrap a layer at its canonical namespace prefix (e.g. `/run`).
+    pub fn layer(&self, layer: Arc<dyn Layer>, prefix: &str, process: Process) -> Arc<dyn Layer> {
+        Arc::new(Observed {
+            layer,
+            prefix: prefix.trim_end_matches('/').into(),
+            observer: self.clone(),
+            process,
+        })
+    }
+    /// Opens share their path and ancestors; rename/remove exclusively hold their
+    /// endpoints, sharing ancestors. Sorted acquisition avoids cross-rename deadlocks.
+    /// No descriptor IO or notification holds these guards.
+    async fn order(&self, endpoints: &[(&str, bool)]) -> Vec<Held> {
+        let mut modes = BTreeMap::new();
+        for &(path, exclusive) in endpoints {
+            modes
+                .entry(path.to_owned())
+                .and_modify(|mode| *mode |= exclusive)
+                .or_insert(exclusive);
+            let mut ancestor = path;
+            while let Some((parent, _)) = ancestor.rsplit_once('/') {
+                let parent = if parent.is_empty() { "/" } else { parent };
+                modes.entry(parent.to_owned()).or_insert(false);
+                if parent == "/" {
+                    break;
+                }
+                ancestor = parent;
+            }
+        }
+        let locks: Vec<_> = {
+            let mut namespace = self.namespace.lock().unwrap();
+            namespace.retain(|_, lock| lock.strong_count() != 0);
+            modes
+                .into_iter()
+                .map(|(path, exclusive)| {
+                    let lock = namespace.entry(path).or_default();
+                    let strong = lock.upgrade().unwrap_or_else(|| {
+                        let strong = Arc::new(RwLock::new(()));
+                        *lock = Arc::downgrade(&strong);
+                        strong
+                    });
+                    (strong, exclusive)
+                })
+                .collect()
+        };
+        let mut held = Vec::new();
+        for (lock, exclusive) in locks {
+            held.push(if exclusive {
+                Held::Write {
+                    _guard: lock.write_owned().await,
+                }
+            } else {
+                Held::Read {
+                    _guard: lock.read_owned().await,
+                }
+            });
+        }
+        held
+    }
+    fn path(&self, path: String) -> Path {
+        let path = Arc::new(Mutex::new(Some(path)));
+        let mut paths = self.paths.lock().unwrap();
+        paths.retain(|p| p.strong_count() != 0);
+        paths.push(Arc::downgrade(&path));
+        path
+    }
+    fn moved(&self, from: &str, to: Option<&str>) {
+        let mut paths = self.paths.lock().unwrap();
+        paths.retain(|p| {
+            let Some(p) = p.upgrade() else { return false };
+            let mut path = p.lock().unwrap();
+            if let Some(old) = path.as_deref() {
+                if under(old, from) {
+                    *path = to.map(|to| format!("{to}{}", &old[from.len()..]));
+                } else if to.is_some_and(|to| under(old, to)) {
+                    // A replaced destination's old descriptor no longer names a visible file.
+                    *path = None;
+                }
+            }
+            true
+        });
+    }
+}
+/// Component-aware descendant matching, including the path itself.
+pub fn under(path: &str, base: &str) -> bool {
+    base == "/" || path == base || path.strip_prefix(base).is_some_and(|s| s.starts_with('/'))
+}
+struct Observed {
+    layer: Arc<dyn Layer>,
+    prefix: String,
+    observer: Observer,
+    process: Process,
+}
+impl Observed {
+    fn path(&self, path: &str) -> String {
+        let path = format!("{}{}", self.prefix, path.trim_end_matches('/'));
+        if path.is_empty() { "/".into() } else { path }
+    }
+}
+struct Handle {
+    file: Arc<dyn File>,
+    path: Path,
+    observer: Observer,
+    process: Process,
+}
+impl Handle {
+    async fn mutate<T: Send + 'static>(
+        &self,
+        f: impl Future<Output = Result<T>> + Send + 'static,
+    ) -> Result<T> {
+        let (path, observer) = (self.path.clone(), self.observer.clone());
+        self.process
+            .operation(async move {
+                let result = f.await?;
+                let path = path.lock().unwrap().clone();
+                if let Some(path) = path {
+                    (observer.changed)(&path);
+                }
+                Ok(result)
+            })
+            .await
+    }
+}
+#[async_trait]
+impl Layer for Observed {
+    async fn stat(&self, path: &str, follow: bool) -> Result<Stat> {
+        self.layer.stat(path, follow).await
+    }
+    async fn readlink(&self, path: &str) -> Result<String> {
+        self.layer.readlink(path).await
+    }
+    async fn list(&self, path: &str) -> Result<Vec<Entry>> {
+        self.layer.list(path).await
+    }
+    async fn open(&self, path: &str, options: Open) -> Result<Arc<dyn File>> {
+        let (layer, local, path, observer, process) = (
+            self.layer.clone(),
+            path.to_owned(),
+            self.path(path),
+            self.observer.clone(),
+            self.process.clone(),
+        );
+        self.process
+            .operation(async move {
+                let order = observer.order(&[(&path, false)]).await;
+                let file = layer.open(&local, options).await?;
+                let handle = Arc::new(Handle {
+                    file,
+                    path: observer.path(path.clone()),
+                    observer: observer.clone(),
+                    process,
+                });
+                drop(order);
+                if options.create || options.truncate {
+                    (observer.changed)(&path);
+                }
+                Ok(handle as Arc<dyn File>)
+            })
+            .await
+    }
+    async fn mkdir(&self, path: &str) -> Result<()> {
+        let (layer, local, path, observer) = (
+            self.layer.clone(),
+            path.to_owned(),
+            self.path(path),
+            self.observer.clone(),
+        );
+        self.process
+            .operation(async move {
+                layer.mkdir(&local).await?;
+                (observer.changed)(&path);
+                Ok(())
+            })
+            .await
+    }
+    async fn remove_file(&self, path: &str) -> Result<()> {
+        self.remove(path, false).await
+    }
+    async fn remove_dir(&self, path: &str) -> Result<()> {
+        self.remove(path, true).await
+    }
+    async fn rename(&self, from: &str, to: &str) -> Result<()> {
+        let (layer, a, b, from, to, observer) = (
+            self.layer.clone(),
+            from.to_owned(),
+            to.to_owned(),
+            self.path(from),
+            self.path(to),
+            self.observer.clone(),
+        );
+        self.process
+            .operation(async move {
+                let order = observer.order(&[(&from, true), (&to, true)]).await;
+                layer.rename(&a, &b).await?;
+                if from != to {
+                    observer.moved(&from, Some(&to));
+                }
+                drop(order);
+                (observer.changed)(&from);
+                (observer.changed)(&to);
+                Ok(())
+            })
+            .await
+    }
+}
+impl Observed {
+    async fn remove(&self, path: &str, directory: bool) -> Result<()> {
+        let (layer, local, path, observer) = (
+            self.layer.clone(),
+            path.to_owned(),
+            self.path(path),
+            self.observer.clone(),
+        );
+        self.process
+            .operation(async move {
+                let order = observer.order(&[(&path, true)]).await;
+                if directory {
+                    layer.remove_dir(&local).await?;
+                } else {
+                    layer.remove_file(&local).await?;
+                }
+                observer.moved(&path, None);
+                drop(order);
+                (observer.changed)(&path);
+                Ok(())
+            })
+            .await
+    }
+}
+#[async_trait]
+impl File for Handle {
+    async fn stat(&self) -> Result<Stat> {
+        self.file.stat().await
+    }
+    async fn read(&self, offset: u64, length: usize) -> Result<Vec<u8>> {
+        self.file.read(offset, length).await
+    }
+    async fn write(&self, offset: u64, bytes: Vec<u8>) -> Result<usize> {
+        let f = self.file.clone();
+        self.mutate(async move { f.write(offset, bytes).await })
+            .await
+    }
+    async fn append(&self, bytes: Vec<u8>) -> Result<usize> {
+        let f = self.file.clone();
+        self.mutate(async move { f.append(bytes).await }).await
+    }
+    async fn set_size(&self, size: u64) -> Result<()> {
+        let f = self.file.clone();
+        self.mutate(async move { f.set_size(size).await }).await
+    }
+    async fn sync(&self) -> Result<()> {
+        self.file.sync().await
+    }
+    async fn set_times(
+        &self,
+        accessed: Option<SystemTime>,
+        modified: Option<SystemTime>,
+    ) -> Result<()> {
+        let f = self.file.clone();
+        self.mutate(async move { f.set_times(accessed, modified).await })
+            .await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::sync::mpsc;
+    #[test]
+    fn component_matching() {
+        assert!(under("/mailbox/a", "/mailbox"));
+        assert!(under("/mailbox", "/mailbox"));
+        assert!(under("/anything", "/"));
+        assert!(!under("/mailboxes/a", "/mailbox"));
+    }
+    /// Every visible mutation is delivered after the layer call, including rebased
+    /// open handles in another process's namespace. Removed handles stay usable.
+    #[tokio::test]
+    async fn handles_follow_rename_and_unlink() {
+        for host in [false, true] {
+            let temp = tempfile::tempdir().unwrap();
+            let process = Process::default();
+            let layer: Arc<dyn Layer> = if host {
+                Arc::new(Host::new(temp.path(), process.clone()).unwrap())
+            } else {
+                Arc::new(Memory::default())
+            };
+            let (tx, mut rx) = mpsc::unbounded_channel();
+            let observer = Observer::new(move |p| {
+                tx.send(p.to_owned()).unwrap();
+            });
+            let a = observer.layer(layer.clone(), "/run", process.clone());
+            let b = observer.layer(layer, "/run", process);
+            a.mkdir("/dir").await.unwrap();
+            assert_eq!(rx.recv().await.unwrap(), "/run/dir");
+            let f = a
+                .open(
+                    "/dir/a",
+                    Open {
+                        create: true,
+                        write: true,
+                        ..Open::default()
+                    },
+                )
+                .await
+                .unwrap();
+            assert_eq!(rx.recv().await.unwrap(), "/run/dir/a");
+            b.rename("/dir", "/moved").await.unwrap();
+            assert_eq!(rx.recv().await.unwrap(), "/run/dir");
+            assert_eq!(rx.recv().await.unwrap(), "/run/moved");
+            f.write(0, b"new".to_vec()).await.unwrap();
+            assert_eq!(rx.recv().await.unwrap(), "/run/moved/a");
+            assert_eq!(
+                b.open("/moved/a", Open::default())
+                    .await
+                    .unwrap()
+                    .read(0, 3)
+                    .await
+                    .unwrap(),
+                b"new"
+            );
+            let old = b
+                .open(
+                    "/old",
+                    Open {
+                        create: true,
+                        write: true,
+                        ..Open::default()
+                    },
+                )
+                .await
+                .unwrap();
+            rx.recv().await.unwrap();
+            b.rename("/moved/a", "/old").await.unwrap();
+            rx.recv().await.unwrap();
+            rx.recv().await.unwrap();
+            old.write(0, b"invisible".to_vec()).await.unwrap();
+            assert!(rx.try_recv().is_err());
+            f.append(b"!".to_vec()).await.unwrap();
+            assert_eq!(rx.recv().await.unwrap(), "/run/old");
+            b.remove_file("/old").await.unwrap();
+            assert_eq!(rx.recv().await.unwrap(), "/run/old");
+            f.set_size(0).await.unwrap();
+            assert!(rx.try_recv().is_err());
+            assert!(b.remove_file("/missing").await.is_err());
+            assert!(rx.try_recv().is_err());
+        }
+    }
+    /// A pending exclusive namespace change must not block a sibling's opens,
+    /// even though Tokio's RwLock prefers queued writers over new readers.
+    #[tokio::test]
+    async fn namespace_order_is_scoped_and_shared() {
+        async fn ready<T>(future: impl Future<Output = T>) -> T {
+            let mut future = std::pin::pin!(future);
+            std::future::poll_fn(|cx| match future.as_mut().poll(cx) {
+                std::task::Poll::Ready(value) => std::task::Poll::Ready(value),
+                std::task::Poll::Pending => panic!("unrelated namespace operation blocked"),
+            })
+            .await
+        }
+        let observer = Observer::new(|_| {});
+        let open = observer.order(&[("/run/slow/file", false)]).await;
+        let concurrent_open = ready(observer.order(&[("/run/slow/file", false)])).await;
+        let mut rename =
+            std::pin::pin!(observer.order(&[("/run/slow", true), ("/run/moved", true)]));
+        std::future::poll_fn(|cx| {
+            assert!(rename.as_mut().poll(cx).is_pending());
+            std::task::Poll::Ready(())
+        })
+        .await;
+        // Same mounted layer, different subtree; also a different mounted layer.
+        let sibling_open = ready(observer.order(&[("/run/sibling/file", false)])).await;
+        let host_open = ready(observer.order(&[("/state/file", false)])).await;
+        drop(sibling_open);
+        let sibling_rename =
+            ready(observer.order(&[("/run/sibling", true), ("/run/other", true)])).await;
+        drop((sibling_rename, host_open, concurrent_open, open));
+        let renamed = ready(rename).await;
+        // A descendant open waits for its ancestor's rename until rebasing is done.
+        let mut descendant = std::pin::pin!(observer.order(&[("/run/moved/file", false)]));
+        std::future::poll_fn(|cx| {
+            assert!(descendant.as_mut().poll(cx).is_pending());
+            std::task::Poll::Ready(())
+        })
+        .await;
+        drop(renamed);
+        drop(ready(descendant).await);
+        drop(
+            observer
+                .order(&[("/run/moved/file", true), ("/run/moved", true)])
+                .await,
+        );
+        drop(
+            observer
+                .order(&[("/run/moved", true), ("/run/moved", true)])
+                .await,
+        );
+        // Dead weak entries are pruned, rather than retaining every path ever opened.
+        observer.order(&[("/", false)]).await;
+        assert_eq!(observer.namespace.lock().unwrap().len(), 1);
+    }
+
+    /// Pause an open after obtaining its real identity, before registering its
+    /// observed handle, like a slow host syscall or /self/log query.
+    struct SlowOpen {
+        memory: Arc<Memory>,
+        gate: Mutex<
+            Option<(
+                tokio::sync::oneshot::Sender<()>,
+                tokio::sync::oneshot::Receiver<()>,
+            )>,
+        >,
+    }
+    #[async_trait]
+    impl Layer for SlowOpen {
+        async fn stat(&self, p: &str, follow: bool) -> Result<Stat> {
+            self.memory.stat(p, follow).await
+        }
+        async fn list(&self, p: &str) -> Result<Vec<Entry>> {
+            self.memory.list(p).await
+        }
+        async fn open(&self, p: &str, options: Open) -> Result<Arc<dyn File>> {
+            let f = self.memory.open(p, options).await?;
+            let gate = self.gate.lock().unwrap().take();
+            if let Some((started, release)) = gate {
+                started.send(()).unwrap();
+                release.await.unwrap();
+            }
+            Ok(f)
+        }
+        async fn mkdir(&self, p: &str) -> Result<()> {
+            self.memory.mkdir(p).await
+        }
+        async fn remove_file(&self, p: &str) -> Result<()> {
+            self.memory.remove_file(p).await
+        }
+        async fn remove_dir(&self, p: &str) -> Result<()> {
+            self.memory.remove_dir(p).await
+        }
+        async fn rename(&self, a: &str, b: &str) -> Result<()> {
+            self.memory.rename(a, b).await
+        }
+    }
+    #[tokio::test]
+    async fn blocked_io_does_not_stall_other_processes() {
+        let memory = Arc::new(Memory::default());
+        memory.mkdir("/slow").await.unwrap();
+        let options = Open {
+            create: true,
+            write: true,
+            ..Open::default()
+        };
+        let raw = memory.open("/slow/file", options).await.unwrap();
+        let (started, opening) = tokio::sync::oneshot::channel();
+        let (release, released) = tokio::sync::oneshot::channel();
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let observer = Observer::new(move |p| {
+            tx.send(p.to_owned()).unwrap();
+        });
+        let slow = observer.layer(
+            Arc::new(SlowOpen {
+                memory: memory.clone(),
+                gate: Mutex::new(Some((started, released))),
+            }),
+            "/run",
+            Process::default(),
+        );
+        let other = observer.layer(memory, "/run", Process::default());
+        let opening_task = tokio::spawn(async move {
+            slow.open(
+                "/slow/file",
+                Open {
+                    write: true,
+                    ..Open::default()
+                },
+            )
+            .await
+        });
+        opening.await.unwrap();
+        let renamer = other.clone();
+        let renamed = tokio::spawn(async move { renamer.rename("/slow", "/moved").await });
+        // These must complete while the first process's open is still blocked,
+        // and a conflicting rename is waiting to rebase its registered handle.
+        other
+            .open(
+                "/",
+                Open {
+                    directory: true,
+                    ..Open::default()
+                },
+            )
+            .await
+            .unwrap();
+        let f = other.open("/other", options).await.unwrap();
+        assert_eq!(rx.recv().await.unwrap(), "/run/other");
+        f.write(0, b"visible".to_vec()).await.unwrap();
+        assert_eq!(rx.recv().await.unwrap(), "/run/other");
+        other.mkdir("/sibling").await.unwrap();
+        assert_eq!(rx.recv().await.unwrap(), "/run/sibling");
+        other.rename("/other", "/sibling/file").await.unwrap();
+        assert_eq!(rx.recv().await.unwrap(), "/run/other");
+        assert_eq!(rx.recv().await.unwrap(), "/run/sibling/file");
+        other.remove_file("/sibling/file").await.unwrap();
+        assert_eq!(rx.recv().await.unwrap(), "/run/sibling/file");
+        assert!(!renamed.is_finished());
+        release.send(()).unwrap();
+        let opened = opening_task.await.unwrap().unwrap();
+        renamed.await.unwrap().unwrap();
+        rx.recv().await.unwrap();
+        rx.recv().await.unwrap();
+        opened.set_size(0).await.unwrap();
+        assert_eq!(rx.recv().await.unwrap(), "/run/moved/file");
+
+        // No namespace guard surrounds descriptor IO: a rename completes while
+        // this write is held in flight. Its notification uses the rebased path.
+        let handle = Handle {
+            file: raw.clone(),
+            path: observer.path("/run/moved/file".into()),
+            observer,
+            process: Process::default(),
+        };
+        let (started, writing) = tokio::sync::oneshot::channel();
+        let (release, released) = tokio::sync::oneshot::channel();
+        let write = tokio::spawn(async move {
+            handle
+                .mutate(async move {
+                    started.send(()).unwrap();
+                    released.await.unwrap();
+                    raw.write(0, b"after rename".to_vec()).await
+                })
+                .await
+        });
+        writing.await.unwrap();
+        other.rename("/moved", "/again").await.unwrap();
+        rx.recv().await.unwrap();
+        rx.recv().await.unwrap();
+        other.open("/again/file", Open::default()).await.unwrap();
+        release.send(()).unwrap();
+        write.await.unwrap().unwrap();
+        assert_eq!(rx.recv().await.unwrap(), "/run/again/file");
+    }
+
+    /// Killing a waiting writer does not discard observation of a completed mutation;
+    /// the process's idle barrier includes the callback, not only the system call.
+    #[tokio::test]
+    async fn cancellation_keeps_mutation_and_observation() {
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let observer = Observer::new(move |p| {
+            tx.send(p.to_owned()).unwrap();
+        });
+        let process = Process::default();
+        let layer = observer.layer(Arc::new(Memory::default()), "/", process.clone());
+        let order = observer.order(&[("/published", true)]).await;
+        let (started, ready) = tokio::sync::oneshot::channel();
+        let p = process.clone();
+        let task = tokio::spawn(async move {
+            // Signal from inside the tracked operation, not before its guard exists.
+            p.operation(async move {
+                started.send(()).unwrap();
+                layer
+                    .open(
+                        "/published",
+                        Open {
+                            create: true,
+                            write: true,
+                            ..Open::default()
+                        },
+                    )
+                    .await
+            })
+            .await
+        });
+        ready.await.unwrap();
+        task.abort();
+        let _ = task.await;
+        drop(order);
+        process.idle().await;
+        assert_eq!(rx.try_recv().unwrap(), "/published");
+    }
+}

vfs/src/process.rs

@@ -36,7 +36,22 @@ impl Process {
         .await
         .map_err(std::io::Error::other)?
     }
-    /// Wait for all outstanding system calls.
+    /// Complete an asynchronous mutation and its observation even if the guest is cancelled.
+    /// Its guard participates in idle(), so no mutation outlives process cleanup.
+    pub async fn operation<T: Send + 'static>(
+        &self,
+        operation: impl std::future::Future<Output = std::io::Result<T>> + Send + 'static,
+    ) -> std::io::Result<T> {
+        self.0.count.fetch_add(1, Ordering::SeqCst);
+        let guard = Guard(self.0.clone());
+        tokio::spawn(async move {
+            let _guard = guard;
+            operation.await
+        })
+        .await
+        .map_err(std::io::Error::other)?
+    }
+    /// Wait for all outstanding system calls and observed mutations.
     /// First abort the guest task. Then await its JoinHandle, or drop its store.
     /// Only then await idle before releasing locks. Abort alone does not stop a running poll.
     /// This ordering also prevents later memory writes, which need no blocking guard.