Remove private-world pointer from service timing comment

PROTOCOL.md

@@ -19,7 +19,7 @@ Rust plugins use the SDK `sausage-plugin`. Its source is at
 | call | A JSON-RPC request. `notes.add` calls the method `add` of the plugin `notes`. |
 | interface | A set of methods. A plugin has one main interface. It can also declare endpoint interfaces. |
 | endpoint | An interface that a running process serves on a TCP port. An endpoint file tells others where it is. |
-| service | A call in `/config/boot.yaml`. The core starts it at boot and again after an upgrade of its plugin. |
+| service | A call in `/config/boot.yaml`. The core starts it at boot, restarts it after a crash, and starts it again after an upgrade of its plugin. |
 | import | An interface of another plugin that a plugin uses. |
 | provider, consumer | The plugin that has an interface, and a plugin that imports it. |
 | journal | The central log of the world. |
@@ -70,7 +70,7 @@ Rust plugins use the SDK `sausage-plugin`. Its source is at
   `notes.rpc.discover` sends it `rpc.discover`.
 - `core.<method>` is a method of the core (section 9).
 - Method names that start with `rpc.` are reserved: `rpc.discover`,
-  `rpc.retired`, and `rpc.endpoint`.
+  `rpc.superseded`, `rpc.retired`, and `rpc.endpoint`.
 
 ### 3.3 What a call does
 
@@ -153,9 +153,17 @@ A process is an instance of one plugin version. When a new version becomes
 active (section 10.3):
 
 1. New calls get the new version at once.
-2. Each process of an older version gets a grace period to finish its work.
-   The default is 60 seconds. A plugin can declare its own value in
-   `rpc.discover` (`x-sausage-grace`). Then the core stops the process.
+2. At grace start, the core sends each process of an older version the
+   JSON-RPC notification `rpc.superseded {"current": "<sha256 of the new file>"}`
+   on stdin, without an ID. The process is still running: calls and responses
+   continue to work. It can stop accepting work, finish its in-flight calls,
+   and exit early. Ignoring the notice changes nothing. The grace period is
+   the upper bound before the core stops the process (section 3.6); its
+   duration is 5 seconds for every plugin. Discovery metadata, including an
+   explicit `x-sausage-grace`, does not change it. After grace, stdin closes
+   and the separate 5-second stop budget applies (section 3.6): at most two
+   5-second windows. Discovery-only processes do not receive this notice or
+   enter upgrade grace; they follow the discovery lifetime (section 10.1).
 3. When the old process of a service has ended, the core starts the service
    again with the new version.
 4. When the last process of an older version has ended, the core starts a
@@ -241,6 +249,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,
@@ -317,10 +359,29 @@ At start, the core starts each service in `/config/boot.yaml`, in order:
 - A service is a detached call with the parent `core`. The core waits for
   its result before it starts the next one. The result shows that the
   service is ready.
-- If a service fails, the core writes the error to the journal and continues.
-- After an upgrade of the plugin of a service, the core starts the service
-  again with the new version (section 3.7).
-- The core does not restart other detached processes.
+- If a service fails before answering, the core writes the error to the
+  journal and continues booting. A failure to admit a process (for example,
+  a missing plugin or invalid parameters) is logged, not retried in a loop.
+- The core supervises the process lifetime, not its readiness response. A
+  nonzero exit, trap or failed host task restarts the same service with its
+  original method and parameters, including a crash before readiness. A
+  readiness response alone never triggers a restart; a clean exit does not
+  count as a crash or trigger an immediate restart.
+- More than 5 crashes within 10 seconds gives up on that service and writes
+  `service gave up` to the journal. Each service has an independent sliding
+  window; crashes exactly 10 seconds old expire. There is no retry delay or
+  periodic liveness check. Giving up is latched for this host lifetime:
+  neither elapsed time nor an upgrade re-arms it.
+- After an upgrade of the plugin of a service, the same supervisor starts
+  it again with the current version after its old process ends (section
+  3.7). Superseded exits are retirement, not crashes. Admission and cleanup
+  reserve a single owner/attempt, so an upgrade racing a crash cannot start
+  two owners. A matching boot-service start from `rpc.retired` is ignored
+  (returns `null`), even while restart is pending or after giving up; older
+  plugins' fallback starts do not bypass supervision.
+- An explicit `core.stop` or `core.kill` disables automatic attempts for
+  that service, including during an upgrade. Host shutdown starts no new
+  processes. The core does not restart other detached processes.
 - A singleton is a service with a fixed endpoint in `/run`. Only the core
   starts it, so two instances cannot start at the same time.
 
@@ -329,6 +390,7 @@ At start, the core starts each service in `/config/boot.yaml`, in order:
 | Method | Params | Result |
 | --- | --- | --- |
 | `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` | `{}` |
@@ -386,7 +448,8 @@ Params are by name. The document can have these extension fields:
   `methods`.
 - `x-sausage-imports`: for each provider, the part of its document that the
   plugin uses.
-- `x-sausage-grace`: the grace period in seconds.
+- `x-sausage-grace`: legacy grace declaration in seconds. The core ignores
+  this field; upgrade grace is always 5 seconds (section 3.7).
 
 The schema of params describes what the provider can read. The schema of a
 result describes what the provider writes. Descriptions are part of the

README.md

@@ -93,10 +93,16 @@ plugins/<name>/plugin.wasm   the current version of a plugin
 plugins/<name>/source/       its source
 state/<plugin>/              the state of a plugin, by convention
 state/core/journal.sqlite    the journal: stderr of all processes, and core events
-config/boot.yaml             the services that start with the world
+config/boot.yaml             ordered boot services, supervised for crashes (PROTOCOL §8)
 config/agent.yaml            the configuration of the agent
 ```
 
+Boot services restart on process crashes, not on readiness responses. More
+than 5 crashes within 10 seconds stops automatic attempts for that service
+and journals `service gave up`; an upgrade does not reset that decision.
+Explicitly stopping or killing a service also leaves it stopped for this
+host lifetime. See [Boot and services](PROTOCOL.md#8-boot-and-services).
+
 Rust builds use the SDK in `plugins/cargo/sdk/` directly. The host seeds this
 folder only when it is missing; restarts and host upgrades preserve existing
 files. SDK edits take effect on the next `cargo.build`, without a restart.

cli/src/activation.rs

@@ -279,15 +279,10 @@ impl Host {
                 .get(&candidate.plugin)
                 .filter(|old| !old.hash.is_empty() && old.hash != candidate.interface.hash)
             {
-                let grace = old.document["x-sausage-grace"]
-                    .as_f64()
-                    .filter(|value| value.is_finite() && *value >= 0.0)
-                    .unwrap_or(60.0);
                 upgrades.push(self.snapshot_upgrade(
                     &candidate.plugin,
                     old.hash.clone(),
                     &candidate.interface.hash,
-                    grace,
                 ));
             }
             self.0.journal.write(

cli/src/core.rs

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

cli/src/filesystem.rs

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

cli/src/graph.rs

@@ -223,16 +223,7 @@ impl Host {
             if let Some(old) = old_plugins.get(plugin).filter(|old| {
                 !old.hash.is_empty() && old.hash != current.hash && current.error.is_none()
             }) {
-                let grace = old.document["x-sausage-grace"]
-                    .as_f64()
-                    .filter(|value| value.is_finite() && *value >= 0.0)
-                    .unwrap_or(60.0);
-                upgrades.push(self.snapshot_upgrade(
-                    plugin,
-                    old.hash.clone(),
-                    &current.hash,
-                    grace,
-                ));
+                upgrades.push(self.snapshot_upgrade(plugin, old.hash.clone(), &current.hash));
             }
         }
         self.publish_graph(plugins);

cli/src/host.rs

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

cli/src/lib.rs

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

cli/src/process.rs

@@ -33,6 +33,8 @@ pub(crate) struct Process {
     pub(crate) id: Option<Value>,
     pub(crate) answer: Option<oneshot::Sender<RpcResult>>,
     pub(crate) ended: watch::Sender<bool>,
+    pub(crate) stopped: bool,
+    pub(crate) superseded: bool,
 }
 impl Host {
     /// Called only after the execution task ends. Wait for the store and file operations before releasing locks.
@@ -57,6 +59,7 @@ impl Host {
                 .map(|_| self.0.journal.query_later(json!({"pid":pid,"limit":20})));
             self.0.journal.forget(pid);
             Self::release_locks(&mut state, pid);
+            self.service_ended(&mut state, pid, status, p.stopped, p.superseded);
             let status = status.to_string();
             tokio::spawn(async move {
                 if let Some(answer) = p.answer {
@@ -84,13 +87,24 @@ impl Host {
     }
     /// Close stdin and start the same five-second deadline for each attached child.
     pub(crate) fn begin_stop(&self, pid: &str) -> Option<watch::Receiver<bool>> {
+        self.stop_process(pid, false)
+    }
+    /// Grace expiry is retirement, not an operator stopping the service.
+    pub(crate) fn begin_retire(&self, pid: &str) -> Option<watch::Receiver<bool>> {
+        self.stop_process(pid, true)
+    }
+    fn stop_process(&self, pid: &str, retirement: bool) -> Option<watch::Receiver<bool>> {
         let mut state = self.0.state.lock().unwrap();
         let p = state.processes.get_mut(pid)?;
+        if !retirement {
+            p.superseded = false;
+        }
         let ended = p.ended.subscribe();
         if p.info.state == "stopping" {
             return Some(ended);
         }
         p.info.state = "stopping".into();
+        p.stopped = true;
         p.input.take();
         self.0.journal.write(pid, &p.info.plugin, "core", "stop");
         for lock in state.locks.values_mut() {
@@ -110,17 +124,24 @@ impl Host {
         let pid = pid.to_string();
         tokio::spawn(async move {
             tokio::time::sleep(std::time::Duration::from_secs(5)).await;
-            h.kill(&pid);
+            h.kill_process(&pid, retirement);
         });
         Some(ended)
     }
     /// Abort a process task and perform its ended-state cleanup.
     pub(crate) fn kill(&self, pid: &str) {
+        self.kill_process(pid, false);
+    }
+    fn kill_process(&self, pid: &str, retirement: bool) {
         let mut state = self.0.state.lock().unwrap();
         for lock in state.locks.values_mut() {
             lock.queue.retain(|(queued, _)| queued != pid);
         }
         if let Some(p) = state.processes.get_mut(pid) {
+            if !retirement {
+                p.superseded = false;
+            }
+            p.stopped = true;
             p.info.state = "stopping".into();
             p.input.take();
             self.0.journal.write(pid, &p.info.plugin, "core", "kill");

cli/src/routing.rs

@@ -51,15 +51,20 @@ impl Host {
             .processes
             .get(pid)
             .and_then(|p| p.answer.as_ref().map(|_| (p.caller.clone(), p.id.clone())));
-        if let Some((caller, id)) = target {
-            let notification = json!({
-                "jsonrpc": "2.0",
-                "method": "core.notify",
-                "params": { "call": id, "name": name, "params": params }
-            });
-            self.send(&caller, notification);
+        if let Some((caller, Some(id))) = target {
+            self.call_notify(&caller, &id, name, params);
         }
     }
+    /// Deliver a notification on an open core call using the same envelope as plugin progress.
+    pub(crate) fn call_notify(&self, caller: &str, id: &Value, name: &str, params: Value) {
+        self.send(
+            caller,
+            json!({
+                "jsonrpc":"2.0", "method":"core.notify",
+                "params": {"call":id, "name":name, "params":params}
+            }),
+        );
+    }
     /// Validate and route a request, then start its native or WASI process.
     pub(crate) fn route(
         &self,
@@ -166,6 +171,33 @@ impl Host {
                 {
                     return Err(error(-32600, "caller stopped or ended"));
                 }
+                // Older plugins may still try to start their boot service from
+                // rpc.retired. The supervisor owns it even while a restart is
+                // pending, or after it has given up. Do not create a second owner.
+                if state
+                    .processes
+                    .get(&caller)
+                    .is_some_and(|p| p.info.method == "rpc.retired")
+                    && state
+                        .services
+                        .iter()
+                        .any(|s| s.method == format!("{plugin}.{method}") && s.params == params)
+                {
+                    return Ok(Value::Null);
+                }
+                if caller == "core"
+                    && let Some(service) = state
+                        .services
+                        .iter_mut()
+                        .find(|s| Some(&s.request) == id.as_ref())
+                {
+                    if !service.enabled || service.owner.is_some() {
+                        return Err(error(-32600, "service attempt is no longer current"));
+                    }
+                    service.owner = Some(pid.clone());
+                    service.pending = false;
+                    service.version = Some(info.version.clone());
+                }
                 host.0
                     .journal
                     .write(&pid, plugin, "core", &format!("start {method} {params}"));
@@ -182,6 +214,8 @@ impl Host {
                         id: id.clone(),
                         answer: if id.is_some() { Some(tx) } else { None },
                         ended: watch::channel(false).0,
+                        stopped: false,
+                        superseded: false,
                     },
                 );
             }

cli/src/services.rs

@@ -1,12 +1,36 @@
-//! Ordered boot services and their identities for upgrade restarts (PROTOCOL.md section 8).
-use crate::Host;
+//! Ordered boot calls and the single restart owner for each service (PROTOCOL.md section 8).
+use crate::{Host, host::State};
 use serde_json::{Value, json};
-#[derive(Clone)]
+use std::{collections::VecDeque, sync::atomic::Ordering, time::Duration};
+use tokio::time::Instant;
+
+// 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);
+
 pub(crate) struct Service {
     pub(crate) method: String,
     pub(crate) params: Value,
     pub(crate) request: Value,
-    pub(crate) starting: std::sync::Arc<std::sync::atomic::AtomicBool>,
+    pub(crate) owner: Option<String>,
+    pub(crate) pending: bool,
+    pub(crate) version: Option<String>,
+    pub(crate) enabled: bool,
+    crashes: VecDeque<Instant>,
+}
+impl Service {
+    fn crashed(&mut self, now: Instant) -> bool {
+        while self
+            .crashes
+            .front()
+            .is_some_and(|t| now.duration_since(*t) >= CRASH_WINDOW)
+        {
+            self.crashes.pop_front();
+        }
+        self.crashes.push_back(now);
+        self.crashes.len() <= CRASH_LIMIT
+    }
 }
 impl Host {
     /// Starts discovery in the background, then waits for each boot service in order.
@@ -24,30 +48,119 @@ impl Host {
         }
         Ok(())
     }
-    /// Records a unique request identity before the service can answer or exit.
+    /// Register once, before admission or readiness. Restarts reuse this registration.
     pub(crate) async fn start_service(&self, method: String, params: Value) {
-        if self.0.shutdown.load(std::sync::atomic::Ordering::SeqCst) {
-            return;
-        }
         let request = json!(uuid::Uuid::now_v7().to_string());
-        let starting = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true));
-        self.0.state.lock().unwrap().services.push(Service {
-            method: method.clone(),
-            params: params.clone(),
-            request: request.clone(),
-            starting: starting.clone(),
-        });
+        {
+            let mut state = self.0.state.lock().unwrap();
+            if self.0.shutdown.load(Ordering::SeqCst) {
+                return;
+            }
+            state.services.push(Service {
+                method,
+                params,
+                request: request.clone(),
+                owner: None,
+                enabled: true,
+                pending: true,
+                version: None,
+                crashes: VecDeque::new(),
+            });
+        }
+        self.run_service(request).await;
+    }
+    /// Await readiness only; process cleanup, not this response, decides whether to restart.
+    async fn run_service(&self, request: Value) {
+        let (method, params) = {
+            let state = self.0.state.lock().unwrap();
+            if self.0.shutdown.load(Ordering::SeqCst) {
+                return;
+            }
+            let Some(service) = state
+                .services
+                .iter()
+                .find(|s| s.request == request && s.enabled)
+            else {
+                return;
+            };
+            (service.method.clone(), service.params.clone())
+        };
         self.0
             .journal
             .write("core", "core", "core", &format!("service start {method}"));
-        if let Err(error) = self
-            .route("core".into(), Some(request), method, params, true)
-            .await
+        let result = self
+            .route("core".into(), Some(request.clone()), method, params, true)
+            .await;
         {
+            let mut state = self.0.state.lock().unwrap();
+            if let Some(service) = state.services.iter_mut().find(|s| s.request == request) {
+                service.pending = false;
+            }
+        }
+        if let Err(error) = result {
+            // Admission failures have no process lifetime to supervise. A process crash
+            // before readiness is handled by finish just like one after readiness.
             self.0
                 .journal
                 .write("core", "core", "core", &format!("service failed {error}"));
         }
-        starting.store(false, std::sync::atomic::Ordering::SeqCst);
+    }
+    /// Called under the state lock after the old owner and its files/locks are gone.
+    pub(crate) fn service_ended(
+        &self,
+        state: &mut State,
+        pid: &str,
+        status: &str,
+        stopped: bool,
+        superseded: bool,
+    ) {
+        let Some(service) = state
+            .services
+            .iter_mut()
+            .find(|s| s.owner.as_deref() == Some(pid))
+        else {
+            return;
+        };
+        service.owner = None;
+        if self.0.shutdown.load(Ordering::SeqCst) || !service.enabled {
+            return;
+        }
+        if !superseded {
+            if stopped {
+                service.enabled = false;
+                return;
+            }
+            if status == "exit 0" {
+                return;
+            }
+            if !service.crashed(Instant::now()) {
+                service.enabled = false;
+                self.0.journal.write(
+                    "core",
+                    "core",
+                    "core",
+                    &format!(
+                        "service gave up {}: more than {CRASH_LIMIT} crashes within {} seconds",
+                        service.method,
+                        CRASH_WINDOW.as_secs(),
+                    ),
+                );
+                return;
+            }
+        }
+        self.queue_service(service);
+    }
+    /// Reserve the next attempt before spawning it; upgrades and crashes share this owner.
+    pub(crate) fn queue_service(&self, service: &mut Service) {
+        service.request = json!(uuid::Uuid::now_v7().to_string());
+        service.pending = true;
+        let request = service.request.clone();
+        let host = self.clone();
+        tokio::spawn(async move {
+            host.run_service(request).await;
+        });
     }
 }
+
+#[cfg(test)]
+mod tests;

cli/src/services/tests.rs

@@ -0,0 +1,356 @@
+//! Deterministic service supervision tests: native tasks stand in for guest lifetimes.
+use super::*;
+use crate::{NativePlugin, runtime::Runtime};
+use std::sync::Arc;
+use tokio::sync::{mpsc, oneshot, watch};
+
+struct Rig {
+    host: Host,
+    starts: mpsc::UnboundedReceiver<(String, oneshot::Sender<()>)>,
+    _world: tempfile::TempDir,
+}
+impl Rig {
+    async fn new() -> Self {
+        let world = tempfile::tempdir().unwrap();
+        let runtime = Runtime::new(world.path()).unwrap();
+        let host = Host::test_world(world.path(), runtime).await.unwrap();
+        let (tx, starts) = mpsc::unbounded_channel();
+        let plugin: NativePlugin = Arc::new(move |ctx, method, _| {
+            let tx = tx.clone();
+            Box::pin(async move {
+                if method != "serve" {
+                    return Ok(Value::Null);
+                }
+                let (end, ended) = oneshot::channel();
+                tx.send((ctx.pid, end)).unwrap();
+                let _ = ended.await;
+                Ok(Value::Null)
+            })
+        });
+        host.register_native("demo", plugin.clone());
+        host.register_native("other", plugin);
+        Self {
+            host,
+            starts,
+            _world: world,
+        }
+    }
+    async fn start(&mut self, method: &str) -> (String, oneshot::Sender<()>) {
+        let host = self.host.clone();
+        let method = method.to_string();
+        tokio::spawn(async move {
+            host.start_service(method, json!({"kept":true})).await;
+        });
+        self.next().await
+    }
+    async fn next(&mut self) -> (String, oneshot::Sender<()>) {
+        // Reuse the host integration call bound (activation.rs::call).
+        tokio::time::timeout(Duration::from_secs(60), self.starts.recv())
+            .await
+            .unwrap()
+            .unwrap()
+    }
+    fn ended(&self, pid: &str) -> watch::Receiver<bool> {
+        self.host.0.state.lock().unwrap().processes[pid]
+            .ended
+            .subscribe()
+    }
+    async fn crash(&self, pid: &str) {
+        let mut ended = self.ended(pid);
+        // Abort execution without asking the supervisor for an intentional stop.
+        self.host.0.state.lock().unwrap().processes[pid]
+            .abort
+            .as_ref()
+            .unwrap()
+            .abort();
+        ended.wait_for(|done| *done).await.unwrap();
+    }
+    async fn clean(&self, pid: &str, end: oneshot::Sender<()>) {
+        let mut ended = self.ended(pid);
+        end.send(()).unwrap();
+        ended.wait_for(|done| *done).await.unwrap();
+    }
+    fn assert_single(&self, pid: &str, registrations: usize) {
+        let state = self.host.0.state.lock().unwrap();
+        assert_eq!(state.services.len(), registrations);
+        let p = &state.processes[pid];
+        assert_eq!(p.info.params, json!({"kept":true}));
+        assert!(p.info.detached);
+        assert_eq!(p.caller, "core");
+        assert_eq!(
+            state
+                .services
+                .iter()
+                .filter(|s| s.owner.as_deref() == Some(pid))
+                .count(),
+            1
+        );
+    }
+    fn no_start(&mut self) {
+        assert!(self.starts.try_recv().is_err());
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn crash_before_and_after_readiness_observes_process_not_response() {
+    let mut rig = Rig::new().await;
+    let (pid, _end) = rig.start("demo.serve").await;
+    rig.assert_single(&pid, 1);
+    rig.crash(&pid).await; // No ready RPC response was sent.
+    let (pid, end) = rig.next().await;
+    rig.assert_single(&pid, 1);
+    rig.host.answer(&pid, Ok(json!({"ready":true})));
+    tokio::task::yield_now().await;
+    rig.no_start(); // Ready does not end the service lifetime.
+    rig.crash(&pid).await;
+    drop(end);
+    let (pid, end) = rig.next().await;
+    rig.assert_single(&pid, 1);
+    rig.clean(&pid, end).await;
+    rig.no_start(); // A clean, intentional return is not a crash.
+    rig.host.stop().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn fifth_restarts_sixth_gives_up_and_journals_once() {
+    let mut rig = Rig::new().await;
+    let (mut pid, mut end) = rig.start("demo.serve").await;
+    for crash in 1..=CRASH_LIMIT {
+        rig.crash(&pid).await;
+        drop(end);
+        (pid, end) = rig.next().await;
+        rig.assert_single(&pid, 1);
+        assert_eq!(
+            rig.host.0.state.lock().unwrap().services[0].crashes.len(),
+            crash
+        );
+    }
+    rig.crash(&pid).await;
+    drop(end);
+    rig.no_start();
+    let logs = rig
+        .host
+        .call("core.log", json!({"grep":"service gave up"}))
+        .await
+        .unwrap();
+    assert_eq!(logs.as_array().unwrap().len(), 1);
+    assert!(
+        logs[0]["line"]
+            .as_str()
+            .unwrap()
+            .contains("more than 5 crashes within 10 seconds")
+    );
+    assert!(!rig.host.0.state.lock().unwrap().services[0].enabled);
+    // A retired hook cannot bypass give-up while its supervisor has no owner.
+    rig.host.register_native(
+        "fallback",
+        Arc::new(move |ctx, _, _| {
+            Box::pin(async move {
+                ctx.call(
+                    "core.call",
+                    json!({"method":"demo.serve","params":{"kept":true},"detach":true}),
+                )
+                .await
+            })
+        }),
+    );
+    assert_eq!(
+        rig.host
+            .call("fallback.rpc.retired", json!({}))
+            .await
+            .unwrap(),
+        Value::Null
+    );
+    rig.no_start();
+    // Neither time passing nor an upgrade re-arms a given-up service.
+    tokio::time::advance(CRASH_WINDOW).await;
+    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
+    rig.host.begin_upgrade(upgrade);
+    tokio::task::yield_now().await;
+    rig.no_start();
+    rig.host.stop().await;
+}
+
+#[test]
+fn sliding_window_boundary_and_progress() {
+    let now = Instant::now();
+    let service = || Service {
+        method: "demo.serve".into(),
+        params: json!({}),
+        request: json!("request"),
+        owner: None,
+        enabled: true,
+        pending: false,
+        version: None,
+        crashes: VecDeque::new(),
+    };
+    let mut before = service();
+    let mut exact = service();
+    let mut after = service();
+    for _ in 0..CRASH_LIMIT {
+        assert!(before.crashed(now));
+        assert!(exact.crashed(now));
+        assert!(after.crashed(now));
+    }
+    assert!(!before.crashed(now + CRASH_WINDOW - Duration::from_nanos(1)));
+    assert!(exact.crashed(now + CRASH_WINDOW));
+    assert_eq!(exact.crashes.len(), 1);
+    assert!(after.crashed(now + CRASH_WINDOW + Duration::from_nanos(1)));
+    // Repeated progress beyond the window never accumulates old failures.
+    for step in 2..=CRASH_LIMIT as u32 + 1 {
+        assert!(exact.crashed(now + CRASH_WINDOW * step));
+        assert_eq!(exact.crashes.len(), 1);
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn services_have_independent_crash_budgets() {
+    let mut rig = Rig::new().await;
+    let (mut a, mut a_end) = rig.start("demo.serve").await;
+    let (b, b_end) = rig.start("other.serve").await;
+    for _ in 0..CRASH_LIMIT {
+        rig.crash(&a).await;
+        drop(a_end);
+        (a, a_end) = rig.next().await;
+    }
+    rig.crash(&a).await;
+    drop(a_end);
+    rig.crash(&b).await;
+    drop(b_end);
+    let (b, b_end) = rig.next().await;
+    rig.assert_single(&b, 2);
+    rig.clean(&b, b_end).await;
+    rig.no_start();
+    rig.host.stop().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn upgrade_crash_race_has_one_restart_and_retired_fallback_cannot_duplicate_it() {
+    let mut rig = Rig::new().await;
+    let (pid, end) = rig.start("demo.serve").await;
+    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
+    rig.crash(&pid).await; // Exits between snapshot and begin_upgrade.
+    drop(end);
+    rig.host.begin_upgrade(upgrade);
+    let (pid, end) = rig.next().await;
+    rig.assert_single(&pid, 1);
+    assert!(
+        rig.host.0.state.lock().unwrap().services[0]
+            .crashes
+            .is_empty()
+    );
+    // A native retired hook exercises the same routing path as a guest fallback.
+    rig.host.register_native(
+        "fallback",
+        Arc::new(move |ctx, _, _| {
+            Box::pin(async move {
+                ctx.call(
+                    "core.call",
+                    json!({"method":"demo.serve","params":{"kept":true},"detach":true}),
+                )
+                .await
+            })
+        }),
+    );
+    assert_eq!(
+        rig.host
+            .call("fallback.rpc.retired", json!({}))
+            .await
+            .unwrap(),
+        Value::Null
+    );
+    rig.no_start();
+    rig.clean(&pid, end).await;
+    rig.host.stop().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn crash_before_upgrade_snapshot_also_has_one_restart() {
+    let mut rig = Rig::new().await;
+    let (pid, end) = rig.start("demo.serve").await;
+    rig.crash(&pid).await;
+    drop(end);
+    // Restart may be pending admission or already admitted: snapshot handles both.
+    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "native");
+    rig.host.begin_upgrade(upgrade);
+    let (pid, end) = rig.next().await;
+    rig.assert_single(&pid, 1);
+    rig.no_start();
+    rig.clean(&pid, end).await;
+    rig.host.stop().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn pending_admission_is_single_owned_across_upgrade() {
+    let mut rig = Rig::new().await;
+    let (pid, end) = rig.start("demo.serve").await;
+    let host = rig.host.clone();
+    let admission = host.0.admission.write().await;
+    rig.crash(&pid).await;
+    drop(end);
+    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "native");
+    rig.host.begin_upgrade(upgrade);
+    assert!(rig.host.0.state.lock().unwrap().services[0].pending);
+    drop(admission);
+    let (pid, end) = rig.next().await;
+    rig.assert_single(&pid, 1);
+    rig.no_start();
+    rig.clean(&pid, end).await;
+    rig.host.stop().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn shutdown_rejects_restart_already_waiting_for_admission() {
+    let mut rig = Rig::new().await;
+    let (pid, end) = rig.start("demo.serve").await;
+    let host = rig.host.clone();
+    let admission = host.0.admission.write().await;
+    rig.crash(&pid).await;
+    drop(end);
+    rig.host.stop().await;
+    drop(admission);
+    tokio::task::yield_now().await;
+    rig.no_start();
+    assert!(rig.host.processes().is_empty());
+}
+
+#[tokio::test(start_paused = true)]
+async fn clean_service_stays_dormant_until_upgrade_and_explicit_stop_stays_disabled() {
+    let mut rig = Rig::new().await;
+    let (pid, end) = rig.start("demo.serve").await;
+    rig.clean(&pid, end).await;
+    rig.no_start();
+    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
+    rig.host.begin_upgrade(upgrade);
+    let (pid, end) = rig.next().await;
+    rig.assert_single(&pid, 1);
+    // An explicit stop stays disabled even if the task subsequently exits cleanly.
+    let mut ended = rig.host.begin_stop(&pid).unwrap();
+    end.send(()).unwrap();
+    ended.wait_for(|done| *done).await.unwrap();
+    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "newer");
+    rig.host.begin_upgrade(upgrade);
+    tokio::task::yield_now().await;
+    rig.no_start();
+    rig.host.stop().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn intentional_stop_and_shutdown_never_restart_even_during_upgrade() {
+    let mut rig = Rig::new().await;
+    let (pid, end) = rig.start("demo.serve").await;
+    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
+    // An explicit kill after supersession overrides the automatic upgrade restart.
+    rig.host
+        .call("core.kill", json!({"pid":pid}))
+        .await
+        .unwrap();
+    drop(end);
+    rig.host.begin_upgrade(upgrade);
+    tokio::task::yield_now().await;
+    rig.no_start();
+    let (_pid, end) = rig.start("other.serve").await;
+    rig.host.stop().await;
+    drop(end);
+    rig.no_start();
+}

cli/src/upgrades.rs

@@ -1,9 +1,15 @@
 //! Grace periods, service restarts, and retirement of replaced versions (PROTOCOL.md section 3.7).
-use crate::{Host, services::Service};
+use crate::Host;
 use serde_json::json;
 use std::{collections::BTreeMap, sync::atomic::Ordering, time::Duration};
 use tokio::sync::watch;
 
+/// Uniform host policy (PROTOCOL 3.7), independent of discovery metadata.
+const UPGRADE_GRACE: Duration = Duration::from_secs(5);
+
+#[cfg(test)]
+mod tests;
+
 pub(crate) struct OldProcess {
     pid: String,
     version: String,
@@ -12,9 +18,8 @@ pub(crate) struct OldProcess {
 pub(crate) struct Upgrade {
     plugin: String,
     previous: String,
-    grace: Duration,
+    current: String,
     processes: Vec<OldProcess>,
-    services: Vec<(Service, Option<watch::Receiver<bool>>)>,
 }
 impl Host {
     /// Snapshots old admissions while activation holds the admission write guard.
@@ -23,57 +28,59 @@ impl Host {
         plugin: &str,
         previous: String,
         current: &str,
-        grace: f64,
     ) -> Upgrade {
         let mut state = self.0.state.lock().unwrap();
         let processes = state
             .processes
-            .iter()
+            .iter_mut()
             .filter(|(_, process)| {
                 process.info.plugin == plugin
                     && process.info.version != current
                     && process.info.method != "rpc.discover"
             })
-            .map(|(pid, process)| OldProcess {
-                pid: pid.clone(),
-                version: process.info.version.clone(),
-                ended: process.ended.subscribe(),
+            .map(|(pid, process)| {
+                // The service supervisor owns the restart, including an exit on
+                // the superseded notice before the grace timer is installed.
+                process.superseded = !process.stopped;
+                OldProcess {
+                    pid: pid.clone(),
+                    version: process.info.version.clone(),
+                    ended: process.ended.subscribe(),
+                }
             })
             .collect();
-        let mut services = Vec::new();
-        let mut retained = Vec::new();
-        for service in std::mem::take(&mut state.services) {
-            if service
-                .method
-                .split_once('.')
-                .is_some_and(|(name, _)| name == plugin)
+        // A cleanly returned service is dormant until an upgrade. A pending
+        // admission will choose the current version itself. Disabled services
+        // (explicit stop or crash-limit give-up) must never be re-armed here.
+        for service in &mut state.services {
+            if service.enabled
+                && !service.pending
+                && service.owner.is_none()
+                && service.version.as_deref() != Some(current)
+                && service
+                    .method
+                    .split_once('.')
+                    .is_some_and(|(name, _)| name == plugin)
             {
-                let process = state
-                    .processes
-                    .values()
-                    .find(|process| process.id.as_ref() == Some(&service.request));
-                if process.is_some_and(|process| process.info.version == current)
-                    || (process.is_none() && service.starting.load(Ordering::SeqCst))
-                {
-                    retained.push(service);
-                } else {
-                    services.push((service, process.map(|process| process.ended.subscribe())));
-                }
-            } else {
-                retained.push(service);
+                self.queue_service(service);
             }
         }
-        state.services = retained;
         Upgrade {
             plugin: plugin.into(),
             previous,
-            grace: Duration::from_secs_f64(grace),
+            current: current.into(),
             processes,
-            services,
         }
     }
     /// Lets old calls finish, stops them after grace, and starts each lifecycle hook once.
     pub(crate) fn begin_upgrade(&self, upgrade: Upgrade) {
+        // Queue the notice before returning from activation, without waiting on guest code.
+        for process in &upgrade.processes {
+            self.send(
+                &process.pid,
+                json!({"jsonrpc":"2.0","method":"rpc.superseded","params":{"current":upgrade.current}}),
+            );
+        }
         let host = self.clone();
         tokio::spawn(async move {
             if host.0.shutdown.load(Ordering::SeqCst) {
@@ -83,7 +90,7 @@ impl Host {
                 "core",
                 &upgrade.plugin,
                 "core",
-                &format!("upgrade grace {} seconds", upgrade.grace.as_secs_f64()),
+                &format!("upgrade grace {} seconds", UPGRADE_GRACE.as_secs()),
             );
             let mut versions: BTreeMap<String, Vec<watch::Receiver<bool>>> = BTreeMap::new();
             versions.entry(upgrade.previous).or_default();
@@ -93,23 +100,11 @@ impl Host {
                     .or_default()
                     .push(process.ended.clone());
                 let host = host.clone();
-                let grace = upgrade.grace;
                 tokio::spawn(async move {
                     let mut ended = process.ended;
                     tokio::select! {
                         _ = ended.wait_for(|ended|*ended) => (),
-                        _ = tokio::time::sleep(grace) => { host.begin_stop(&process.pid); }
-                    }
-                });
-            }
-            for (service, ended) in upgrade.services {
-                let host = host.clone();
-                tokio::spawn(async move {
-                    if let Some(mut ended) = ended {
-                        let _ = ended.wait_for(|ended| *ended).await;
-                    }
-                    if !host.0.shutdown.load(Ordering::SeqCst) {
-                        host.start_service(service.method, service.params).await;
+                        _ = tokio::time::sleep(UPGRADE_GRACE) => { host.begin_retire(&process.pid); }
                     }
                 });
             }

cli/src/upgrades/tests.rs

@@ -0,0 +1,115 @@
+//! Controlled-clock coverage of both upgrade windows and discovery exclusion.
+use super::*;
+use crate::{
+    process::{Process, ProcessInfo},
+    runtime::Runtime,
+    world::World,
+};
+use tokio::{
+    sync::mpsc,
+    task::JoinHandle,
+    time::{Instant, advance},
+};
+
+/// A process with active work that ignores both superseded and EOF.
+fn process(
+    host: &Host,
+    pid: &str,
+    method: &str,
+) -> (mpsc::UnboundedReceiver<serde_json::Value>, JoinHandle<()>) {
+    let (input, incoming) = mpsc::unbounded_channel();
+    let task = tokio::spawn(std::future::pending());
+    host.0.state.lock().unwrap().processes.insert(
+        pid.into(),
+        Process {
+            info: ProcessInfo {
+                plugin: "notes".into(),
+                version: "old".into(),
+                method: method.into(),
+                params: json!({}),
+                parent: None,
+                detached: true,
+                start_time: 0,
+                state: "running".into(),
+            },
+            memory: sausage_vfs::Memory::default(),
+            files: sausage_vfs::Process::default(),
+            store_ended: watch::channel(true).0,
+            input: Some(input),
+            abort: Some(task.abort_handle()),
+            caller: "console".into(),
+            id: None,
+            answer: None,
+            ended: watch::channel(false).0,
+            stopped: false,
+            superseded: false,
+        },
+    );
+    (incoming, task)
+}
+
+/// PROTOCOL 3.7/3.6: five seconds running, then five seconds stopping; discovery is untouched.
+#[tokio::test]
+async fn two_five_second_windows_and_discovery_exclusion() {
+    let dir = tempfile::tempdir().unwrap();
+    let root = dir.path().join("world");
+    World::create(&root).unwrap();
+    let host = Host::test_world(&root, Runtime::new(&dir.path().join("cache")).unwrap())
+        .await
+        .unwrap();
+    let (mut input, work) = process(&host, "old-work", "work");
+    let (mut discovery_input, discovery) = process(&host, "old-discovery", "rpc.discover");
+    let upgrade = host.snapshot_upgrade("notes", "old".into(), "new");
+    assert_eq!(upgrade.processes.len(), 1);
+    tokio::time::pause();
+    let start = Instant::now();
+    host.begin_upgrade(upgrade);
+    assert_eq!(
+        input.recv().await.unwrap(),
+        json!({"jsonrpc":"2.0","method":"rpc.superseded","params":{"current":"new"}})
+    );
+    // Run the upgrade dispatcher, then its per-process task, so the deadline is armed.
+    tokio::task::yield_now().await;
+    tokio::task::yield_now().await;
+    let window = Duration::from_secs(5);
+    assert_eq!(UPGRADE_GRACE, window);
+    // tokio::time::sleep documents millisecond granularity; the paused driver
+    // may round a deadline up by one tick. Check each window independently.
+    let tick = Duration::from_millis(1);
+    let before_deadline = window - tick;
+    advance(before_deadline).await;
+    tokio::task::yield_now().await;
+    assert_eq!(host.processes()["old-work"].state, "running");
+    assert!(!work.is_finished());
+    advance(tick).await;
+    assert!(
+        input.recv().await.is_none(),
+        "EOF at grace, even with active work"
+    );
+    let stopped = Instant::now();
+    assert!((window..=window + tick).contains(&(stopped - start)));
+    assert_eq!(host.processes()["old-work"].state, "stopping");
+    // begin_retire schedules the independent PROTOCOL 3.6 deadline.
+    tokio::task::yield_now().await;
+    advance(before_deadline).await;
+    tokio::task::yield_now().await;
+    assert!(
+        !work.is_finished(),
+        "on_stop still has its full five seconds"
+    );
+    advance(tick).await;
+    assert!(work.await.unwrap_err().is_cancelled());
+    assert!((window..=window + tick).contains(&(Instant::now() - stopped)));
+    assert_eq!(host.processes()["old-discovery"].state, "running");
+    assert!(matches!(
+        discovery_input.try_recv(),
+        Err(mpsc::error::TryRecvError::Empty)
+    ));
+    assert!(!discovery.is_finished());
+    discovery.abort();
+    let _ = discovery.await;
+    // Synthetic processes have no execution cleanup task; discard only this test table.
+    host.0.shutdown.store(true, Ordering::SeqCst);
+    host.0.state.lock().unwrap().processes.clear();
+    host.stop().await;
+}

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

@@ -160,9 +160,9 @@ impl Drop for TestWorld {
         let _ = std::fs::remove_dir_all(&self.0);
     }
 }
-/// Builds a method-only document without an upgrade delay.
+/// Builds a method-only document; all plugins use the host's five-second grace.
 fn document(name: &str) -> Value {
-    json!({"methods":[{"name":name,"params":[],"result":{"schema":{"type":"string"}}}],"x-sausage-grace":0})
+    json!({"methods":[{"name":name,"params":[],"result":{"schema":{"type":"string"}}}]})
 }
 /// Adds an import of a provider's method list.
 fn consumer(provider: &str, expected: Value) -> Value {
@@ -484,7 +484,7 @@ async fn review_dropped_activation_completes() {
     host.stop().await;
 }
 /// PROTOCOL 10.1: discovery fails after its fixed ten-second deadline and receives EOF.
-/// PROTOCOL 3.7: old processes retain their declared grace while new calls use new code.
+/// PROTOCOL 3.7: old processes get five seconds regardless of declarations; new calls use new code.
 /// PROTOCOL 3.7: retirement waits for all live admissions, including reverted versions.
 /// PROTOCOL 3.7: rejected discovery candidates never receive retirement hooks.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -604,7 +604,7 @@ async fn source_changes_do_not_change_candidate() {
     );
     host.stop().await;
 }
-/// PROTOCOL 3.7: plugins without a declared grace period retain old calls before the default deadline.
+/// PROTOCOL 3.7: missing metadata also retains old calls until the uniform deadline.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn default_grace_retains_old_process() {
     let world = TestWorld::new();
@@ -623,6 +623,104 @@ async fn default_grace_retains_old_process() {
     assert_eq!(host.processes()[&pid].state, "running");
     host.stop().await;
 }
+/// PROTOCOL 3.7: metadata cannot shorten or extend the actual grace for active requests.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn uniform_grace_ignores_discovery_declarations() {
+    let world = TestWorld::new();
+    let declarations = [
+        ("missing", None),
+        ("default", Some(json!(60))),
+        ("long", Some(json!(3600))),
+        ("zero", Some(json!(0))),
+        ("negative", Some(json!(-1))),
+        ("invalid", Some(json!("long"))),
+        ("huge", Some(json!(1e300))),
+    ];
+    for (name, grace) in &declarations {
+        let mut old = document("work");
+        if let Some(grace) = grace {
+            old["x-sausage-grace"] = grace.clone();
+        }
+        world.plugin(name, old, document("work"));
+    }
+    let host = world.start().await;
+    let mut ready = host.subscribe();
+    let mut calls = Vec::new();
+    for (name, _) in &declarations {
+        let host = host.clone();
+        let method = format!("{name}.work");
+        calls.push(tokio::spawn(async move {
+            let result = host.call(&method, json!({})).await;
+            (result, std::time::Instant::now())
+        }));
+    }
+    // Guest notifications are events, not process/file polling.
+    for _ in &declarations {
+        let event = tokio::time::timeout(Duration::from_secs(30), ready.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(event["params"]["name"], "ready");
+    }
+    let before = std::time::Instant::now();
+    let plugins: Vec<_> = declarations
+        .iter()
+        .map(|(name, _)| json!({"plugin":name,"wasm":format!("/state/{name}/next.wasm")}))
+        .collect();
+    call(&host, "core.activate", json!({"plugins":plugins})).await;
+    // Use the existing discovery+stop test budget, not a new execution limit.
+    tokio::time::timeout(Duration::from_secs(15), async {
+        for task in calls {
+            let (result, ended) = task.await.unwrap();
+            assert_eq!(result.unwrap_err()["code"], -32002);
+            assert!(
+                ended.duration_since(before) >= Duration::from_secs(5),
+                "every declaration gets the full grace"
+            );
+        }
+    })
+    .await
+    .unwrap();
+    host.stop().await;
+}
+
+/// PROTOCOL 10.4/3.7: a watcher refresh uses the same policy as explicit activation.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn manual_copy_uses_uniform_grace() {
+    let world = TestWorld::new();
+    let mut old = document("work");
+    old["x-sausage-grace"] = json!(3600);
+    world.plugin("notes", old, document("work"));
+    let host = world.start().await;
+    // Wait for initial discovery without changing the installed version.
+    call(
+        &host,
+        "core.activate",
+        json!({"plugin":"notes","wasm":"/plugins/notes/plugin.wasm"}),
+    )
+    .await;
+    let mut ready = host.subscribe();
+    let work_host = host.clone();
+    let work = tokio::spawn(async move { work_host.call("notes.work", json!({})).await });
+    let event = tokio::time::timeout(Duration::from_secs(30), ready.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(event["params"]["name"], "ready");
+    let before = std::time::Instant::now();
+    std::fs::copy(fixture(2), world.0.join("plugins/notes/plugin.wasm")).unwrap();
+    assert_eq!(
+        tokio::time::timeout(Duration::from_secs(15), work)
+            .await
+            .unwrap()
+            .unwrap()
+            .unwrap_err()["code"],
+        -32002
+    );
+    assert!(before.elapsed() >= Duration::from_secs(5));
+    host.stop().await;
+}
+
 /// PROTOCOL 10.3: candidate imports must fit installed providers.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn checks_candidate_imports() {
@@ -818,11 +916,11 @@ async fn graph_missing_import() {
     assert!(dot.as_str().unwrap().contains("absent"));
     host.stop().await;
 }
-/// Checks the declared upgrade grace and dispatch to the new version.
+/// Checks that an explicit long declaration cannot extend grace; new calls use the new version.
 async fn old_process_grace_and_new_calls() {
     let world = TestWorld::new();
     let mut old = document("add");
-    old["x-sausage-grace"] = json!(0.5);
+    old["x-sausage-grace"] = json!(3600);
     world.plugin("notes", old, document("add"));
     let host = world.start().await;
     call(&host, "notes.hold", json!({})).await;
@@ -839,6 +937,76 @@ async fn old_process_grace_and_new_calls() {
         "2"
     );
     marker(world.0.join("state/notes/ended-1"), "ended").await;
+    let notice: Value = serde_json::from_slice(
+        &std::fs::read(world.0.join(format!("state/notes/superseded-{pid}"))).unwrap(),
+    )
+    .unwrap();
+    assert_eq!(notice["params"]["current"], fixture_hash(2));
+    host.stop().await;
+}
+/// PROTOCOL 3.7: every old process gets the new hash while still running, before grace ends.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn superseded_notifies_each_old_process() {
+    let world = TestWorld::new();
+    let mut old = document("hold");
+    old["x-sausage-grace"] = json!(60);
+    world.plugin("notes", old, document("hold"));
+    let host = world.start().await;
+    call(&host, "notes.hold", json!({})).await;
+    call(&host, "notes.hold", json!({})).await;
+    let pids: Vec<_> = host
+        .processes()
+        .into_iter()
+        .filter(|(_, process)| process.method == "hold")
+        .map(|(pid, _)| pid)
+        .collect();
+    assert_eq!(pids.len(), 2);
+    activate(&host, "notes").await.unwrap();
+    let expected =
+        json!({"jsonrpc":"2.0","method":"rpc.superseded","params":{"current":fixture_hash(2)}})
+            .to_string();
+    for pid in pids {
+        // The notice arrives while the old process is still in its five-second window.
+        marker(
+            world.0.join(format!("state/notes/superseded-{pid}")),
+            &expected,
+        )
+        .await;
+        assert_eq!(host.processes()[&pid].state, "running");
+    }
+    assert_eq!(
+        call(&host, "notes.version", json!({})).await["version"],
+        "2"
+    );
+    host.stop().await;
+}
+/// PROTOCOL 3.7: a service that exits on the notice restarts before its grace deadline.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn superseded_service_exits_early() {
+    let world = TestWorld::new();
+    let mut old = document("serve");
+    old["x-sausage-grace"] = json!(60);
+    world.plugin("notes", old, document("serve"));
+    std::fs::write(
+        world.0.join("config/boot.yaml"),
+        json!([{"method":"notes.serve","params":{"exit_on_superseded":true}}]).to_string(),
+    )
+    .unwrap();
+    let host = world.start().await;
+    marker(world.0.join("state/notes/service"), "1").await;
+    let pid = host
+        .processes()
+        .into_iter()
+        .find(|(_, p)| p.method == "serve")
+        .unwrap()
+        .0;
+    activate(&host, "notes").await.unwrap();
+    marker(world.0.join("state/notes/service"), "2").await;
+    assert!(!host.processes().contains_key(&pid));
+    assert!(
+        !world.0.join("state/notes/ended-1").exists(),
+        "old service must exit on the notice, not EOF"
+    );
     host.stop().await;
 }
 /// PROTOCOL 3.7: retirement receives the previous hash only after old processes end.
@@ -923,7 +1091,7 @@ async fn upgrades_enter_the_journal() {
                 .as_array()
                 .unwrap()
                 .iter()
-                .any(|e| e["line"] == "upgrade grace 60 seconds")
+                .any(|e| e["line"] == "upgrade grace 5 seconds")
             {
                 break entries;
             }
@@ -1047,6 +1215,36 @@ async fn old_consumer_can_call_removed_provider_method() {
     host.stop().await;
 }
 
+/// PROTOCOL 8: supervise actual WASI exits on either side of the readiness response.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn crashed_boot_service_restarts_before_and_after_ready() {
+    for when in ["before", "after"] {
+        let world = TestWorld::new();
+        world.plugin("notes", document("serve"), document("serve"));
+        std::fs::write(
+            world.0.join("config/boot.yaml"),
+            json!([{"method":"notes.serve","params":{"crash_once":when}}]).to_string(),
+        )
+        .unwrap();
+        let host = world.start().await;
+        marker(world.0.join("state/notes/service"), "1").await;
+        let entries = call(
+            &host,
+            "core.log",
+            json!({"grep":"service start notes.serve"}),
+        )
+        .await;
+        assert_eq!(entries.as_array().unwrap().len(), 2);
+        let processes = host.processes();
+        let services: Vec<_> = processes.values().filter(|p| p.method == "serve").collect();
+        assert_eq!(services.len(), 1);
+        assert_eq!(services[0].params, json!({"crash_once":when}));
+        let exits = call(&host, "core.log", json!({"plugin":"notes","grep":"exit 1"})).await;
+        assert_eq!(exits.as_array().unwrap().len(), 1);
+        host.stop().await;
+    }
+}
+
 /// PROTOCOL 10.2: activation and revert check interfaces but leave persistent state unchanged.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn state_is_outside_interface_checks() {

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
@@ -1104,17 +1108,16 @@ async fn stop_closes_stdin_and_waits_for_exit() {
     let w = TestWorld::new();
     let h = w.start().await;
     let hc = h.clone();
+    let mut events = h.subscribe();
     let pending = tokio::spawn(async move { hc.call("probe.wait_eof", json!({})).await });
+    // Admission is not readiness: stop may otherwise close stdin before the
+    // request reaches the guest, so wait for the handler to enter its EOF loop.
     let pid = tokio::time::timeout(std::time::Duration::from_secs(20), async {
         loop {
-            if let Some((pid, _)) = h
-                .processes()
-                .into_iter()
-                .find(|(_, p)| p.method == "wait_eof")
-            {
-                break pid;
+            let event = events.recv().await.unwrap();
+            if event["params"]["name"] == "waiting-eof" {
+                break event["params"]["params"]["pid"].clone();
             }
-            tokio::task::yield_now().await;
         }
     })
     .await

cli/tests/host/main.rs

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

cli/tests/host/sdk.rs

@@ -178,6 +178,56 @@ async fn stop_runs_hook() {
     assert_eq!(std::fs::read(world.0.join("stopped")).unwrap(), b"yes");
     host.stop().await;
 }
+/// PROTOCOL 3.7: an idle SDK listener opts into drain and exits before EOF/grace.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn superseded_idle_listener_drains_before_grace() {
+    let world = TestWorld::new();
+    let host = world.start().await;
+    let (notice, mut notices) = tokio::sync::mpsc::unbounded_channel();
+    host.register_native(
+        "other",
+        std::sync::Arc::new(move |_, method, _| {
+            let notice = notice.clone();
+            Box::pin(async move {
+                assert_eq!(method, "superseded");
+                notice.send(std::time::Instant::now()).unwrap();
+                Ok(json!({}))
+            })
+        }),
+    );
+    let pid = call(&host, "probe.drain_service", json!({})).await;
+    // Replace with a different real component; force only skips unrelated test consumers.
+    std::fs::copy(
+        fixtures().join("notes.wasm"),
+        world.0.join("state/next.wasm"),
+    )
+    .unwrap();
+    call(
+        &host,
+        "core.activate",
+        json!({"plugin":"probe","wasm":"/state/next.wasm","force":true}),
+    )
+    .await;
+    // Activation preparation/discovery is outside upgrade grace. Time from the
+    // hook's outgoing call, not from the start of candidate compilation.
+    let before = tokio::time::timeout(Duration::from_secs(60), notices.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    // Waiting on its lifetime lock is an exit event, not process-liveness polling.
+    call(&host, "core.lock", json!({"path":"/drain-owner"})).await;
+    assert!(before.elapsed() < Duration::from_secs(5));
+    assert!(!host.processes().contains_key(pid.as_str().unwrap()));
+    assert!(
+        !world
+            .0
+            .join(format!("stopped-{}", pid.as_str().unwrap()))
+            .exists(),
+        "drain must not run on_stop (discovery processes may run their own hooks)"
+    );
+    host.stop().await;
+}
+
 /// SDK framing carries payloads above 64 KiB in both directions.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn large_messages() {

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

@@ -84,12 +84,52 @@ fn main() {
             .unwrap();
             json!({"retired":version})
         }
+        "work" => {
+            // Keep an actual request unanswered through grace; notices do not drain it.
+            send(
+                json!({"jsonrpc":"2.0","method":"core.notify","params":{"name":"ready","params":{}}}),
+            );
+            for _ in lines {}
+            return;
+        }
         "serve" | "hold" => {
+            // One genuine process failure, followed by a healthy service, without
+            // timers or liveness polling in either the fixture or supervisor.
+            let crash = request["params"]["crash_once"].as_str();
+            let first = crash.is_some()
+                && std::fs::OpenOptions::new()
+                    .write(true)
+                    .create_new(true)
+                    .open(format!("/state/{plugin}/crashed-once"))
+                    .is_ok();
+            if first && crash == Some("before") {
+                std::process::exit(1);
+            }
             send(json!({"jsonrpc":"2.0","id":request["id"],"result":{"version":version}}));
+            if first && crash == Some("after") {
+                std::process::exit(1);
+            }
             if request["method"] == "serve" {
                 std::fs::write(format!("/state/{plugin}/service"), version).unwrap();
             }
-            for _ in lines {}
+            for line in lines {
+                let Ok(message) = serde_json::from_str::<Value>(&line.unwrap()) else {
+                    continue;
+                };
+                if message["method"] == "rpc.superseded" {
+                    std::fs::write(
+                        format!(
+                            "/state/{plugin}/superseded-{}",
+                            std::env::var("SAUSAGE_PID").unwrap()
+                        ),
+                        message.to_string(),
+                    )
+                    .unwrap();
+                    if request["params"]["exit_on_superseded"] == true {
+                        return;
+                    }
+                }
+            }
             std::fs::write(format!("/state/{plugin}/ended-{version}"), "ended").unwrap();
             return;
         }

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

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

sdk/macros/src/lib.rs

@@ -153,6 +153,7 @@ pub fn rpc(args: TokenStream, input: TokenStream) -> TokenStream {
         let mut stop = quote!();
         let mut has_stop = false;
         let mut has_retired = false;
+        let mut has_superseded = false;
         for member in &mut item.items {
             let ImplItem::Fn(f) = member else { continue };
             let retired = f.attrs.iter().any(|a| {
@@ -161,6 +162,12 @@ pub fn rpc(args: TokenStream, input: TokenStream) -> TokenStream {
                     .last()
                     .is_some_and(|s| s.ident == "on_retired")
             });
+            let superseded = f.attrs.iter().any(|a| {
+                a.path()
+                    .segments
+                    .last()
+                    .is_some_and(|s| s.ident == "on_superseded")
+            });
             let stopping = f.attrs.iter().any(|a| {
                 a.path()
                     .segments
@@ -179,11 +186,16 @@ pub fn rpc(args: TokenStream, input: TokenStream) -> TokenStream {
                     "only one on_retired method is allowed",
                 ));
             }
+            if superseded && std::mem::replace(&mut has_superseded, true) {
+                return Err(syn::Error::new(
+                    f.span(),
+                    "only one on_superseded method is allowed",
+                ));
+            }
             f.attrs.retain(|a| {
-                !a.path()
-                    .segments
-                    .last()
-                    .is_some_and(|s| s.ident == "on_retired" || s.ident == "on_stop")
+                !a.path().segments.last().is_some_and(|s| {
+                    s.ident == "on_retired" || s.ident == "on_stop" || s.ident == "on_superseded"
+                })
             });
             documented(&f.attrs, f.span())?;
             if f.sig.inputs.is_empty()
@@ -224,7 +236,9 @@ pub fn rpc(args: TokenStream, input: TokenStream) -> TokenStream {
                 stop = quote!(fn stop(&self)->::std::pin::Pin<Box<dyn ::std::future::Future<Output=()> + '_>> {Box::pin(async move {let _=self.#name() #await_;})});
                 continue;
             }
-            let wire = if retired {
+            let wire = if superseded {
+                "rpc.superseded".to_string()
+            } else if retired {
                 "rpc.retired".to_string()
             } else {
                 name.unraw().to_string()
@@ -253,7 +267,7 @@ pub fn rpc(args: TokenStream, input: TokenStream) -> TokenStream {
             };
             let doc = docs(&f.attrs);
             arms.push(quote!(#wire => { fn params_type<T: ::sausage_plugin::Params>() {} params_type::<#p>(); let result=self.#name(#(#args),*) #await_ ?; Ok(::sausage_plugin::serde_json::to_value(result)?) }));
-            if !retired {
+            if !retired && !superseded {
                 entries.push(quote!(::sausage_plugin::method::<#p,#output>(&mut doc,#wire,#doc);));
             }
         }
@@ -328,6 +342,12 @@ pub fn import(input: TokenStream) -> TokenStream {
 pub fn on_retired(_: TokenStream, input: TokenStream) -> TokenStream {
     input
 }
+/// Mark the method that handles `rpc.superseded` at the start of upgrade grace.
+/// Calls still work here. Use `sausage_plugin::drain` to finish active work and exit.
+#[proc_macro_attribute]
+pub fn on_superseded(_: TokenStream, input: TokenStream) -> TokenStream {
+    input
+}
 /// Mark the hook that runs at stdin EOF without a live call transport.
 #[proc_macro_attribute]
 pub fn on_stop(_: TokenStream, input: TokenStream) -> TokenStream {

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

@@ -65,7 +65,7 @@ impl<T: serde::de::DeserializeOwned + 'static> IntoFuture for EndpointCall<T> {
     }
 }
 
-/// Publish a declared interface and keep its listener alive until stdin EOF.
+/// Publish a declared interface and keep its listener alive until stdin EOF or [`drain`].
 #[cfg(target_os = "wasi")]
 pub async fn listen<T: Interface>(path: impl AsRef<Path>, value: Rc<T>) -> Result<()> {
     use wstd::iter::AsyncIterator;
@@ -80,7 +80,7 @@ pub async fn listen<T: Interface>(path: impl AsRef<Path>, value: Rc<T>) -> Resul
         while let Some(Ok(stream)) = incoming.next().await {
             let value = value.clone();
             let name = name.clone();
-            crate::transport::endpoint_task(async move {
+            background(async move {
                 let _ = serve_connection(stream, value, name).await;
             });
         }
@@ -211,7 +211,11 @@ async fn serve_connection<T: Interface>(
     use futures::{FutureExt, StreamExt};
     let (mut reader, mut writer) = stream.split();
     let mut lines = crate::lines::Lines::default();
-    while let Some(request) = lines.rpc(&mut reader).await? {
+    while let Some(request) = crate::transport::until_drained(lines.rpc(&mut reader))
+        .await
+        .transpose()?
+        .flatten()
+    {
         let method = request["method"].as_str().unwrap_or_default();
         let (tx, mut rx) = futures::channel::mpsc::unbounded();
         let ctx = Ctx {

sdk/src/lib.rs

@@ -356,13 +356,20 @@
 //!
 //! The interface check covers calls, not state. Old and new processes share files in `/state`.
 //! `core.revert` restores code, not state. Preserve compatibility or migrate the state.
-//! `#[rpc(grace_seconds = N)]` declares the grace period for old processes. The default is 60 seconds.
+//! Core gives every old process 5 seconds of upgrade grace, then closes stdin and allows
+//! a separate 5-second stop budget before killing it. Discovery-only processes are excluded.
+//! `#[rpc(grace_seconds = N)]` still emits legacy `x-sausage-grace` metadata, but core ignores it.
 //! New calls use the new version immediately. Old processes can still use old imports during their grace period.
+//! [`macro@on_superseded`] handles the notice at grace start, with normal calls still available.
+//! It is opt-in: without a hook the process continues as before until core stops it.
+//! Call [`drain`] in this hook to close SDK endpoint listeners and idle connections,
+//! let active handlers and background tasks finish, and exit without waiting for grace.
+//! Custom listeners and long-lived background tasks must be ended by the plugin itself.
 //! [`macro@on_retired`] handles `rpc.retired` after the last process of an old version ends.
 //! This hook is best effort. If the host stops first, the hook does not run.
 //! New processes must handle unmigrated state. Check a migration marker and migrate if it is absent.
 //! See section 3.7 of `/proc/core/PROTOCOL.md`.
-//! Declare at most one stop hook and one retirement hook per interface.
+//! Declare at most one hook of each kind per interface.
 //!
 //! ```no_run
 //! use sausage_plugin::{rpc, schema, state_dir, write_atomic, Result};
@@ -374,8 +381,21 @@
 //!     /// The SHA-256 of the previous WASM file.
 //!     previous: String,
 //! }
+//! /// The version now receiving new calls.
+//! #[schema]
+//! struct SupersededParams {
+//!     /// The SHA-256 of the newly active WASM file.
+//!     current: String,
+//! }
 //! #[rpc]
 //! impl Migrator {
+//!     /// Finish current work and exit early during upgrade grace.
+//!     #[sausage_plugin::on_superseded]
+//!     fn superseded(&self, p: SupersededParams) -> Result<()> {
+//!         eprintln!("replaced by {}", p.current);
+//!         sausage_plugin::drain();
+//!         Ok(())
+//!     }
 //!     /// Record a retired version. Normal methods must also handle missing markers.
 //!     #[sausage_plugin::on_retired]
 //!     fn retired(&self, p: RetiredParams) -> Result<()> {
@@ -401,6 +421,6 @@ pub use helpers::*;
 pub use imports::*;
 #[doc(hidden)]
 pub use inventory;
-pub use sausage_plugin_macros::{import, on_retired, on_stop, rpc, schema, test};
+pub use sausage_plugin_macros::{import, on_retired, on_stop, on_superseded, rpc, schema, test};
 pub use transport::*;
 pub use {schemars, serde, serde_json};

sdk/src/transport.rs

@@ -96,8 +96,11 @@ struct State {
     active: bool,
     endpoints: HashMap<std::any::TypeId, String>,
     stopping: bool,
+    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")]
@@ -108,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| {
@@ -125,17 +140,48 @@ pub(crate) fn endpoint_name<T: 'static>() -> Result<String> {
 }
 #[cfg(target_os = "wasi")]
 pub(crate) fn endpoint_task(future: impl Future<Output = ()> + 'static) {
+    background(async move {
+        let _ = until_drained(future).await;
+    });
+}
+/// Cancel an accept/read wait when draining, but never an active handler.
+#[cfg(target_os = "wasi")]
+pub(crate) async fn until_drained<F: Future>(future: F) -> Option<F::Output> {
     let (handle, registration) = futures::future::AbortHandle::new_pair();
     let id = STATE.with(|s| {
         let mut s = s.borrow_mut();
+        if s.draining {
+            return None;
+        }
         s.next += 1;
         let id = s.next;
         s.listeners.insert(id, handle);
-        id
-    });
-    background(async move {
-        let _ = futures::future::Abortable::new(future, registration).await;
-        STATE.with(|s| s.borrow_mut().listeners.remove(&id));
+        Some(id)
+    })?;
+    let result = futures::future::Abortable::new(future, registration)
+        .await
+        .ok();
+    STATE.with(|s| s.borrow_mut().listeners.remove(&id));
+    result
+}
+/// Stop accepting SDK endpoint work, then exit when handlers and background tasks finish.
+///
+/// Calls and notifications still work. This does not run the EOF stop hook or cancel
+/// background tasks: services with their own listeners/tasks must arrange for those
+/// to finish too. Repeated calls are harmless. Core's upgrade grace remains the bound.
+pub fn drain() {
+    STATE.with(|s| {
+        let mut s = s.borrow_mut();
+        s.draining = true;
+        #[cfg(target_os = "wasi")]
+        {
+            for (_, listener) in s.listeners.drain() {
+                listener.abort();
+            }
+            if let Some(w) = s.wake.take() {
+                w.wake();
+            }
+        }
     });
 }
 #[cfg(target_os = "wasi")]
@@ -192,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
     })
 }
@@ -274,7 +320,16 @@ pub fn receive(value: Value) {
         }
         let _ = p.tx.send(result);
     } else {
-        eprintln!("note: response with unknown ID {}", value["id"]);
+        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"]);
+        }
     }
 }
 
@@ -324,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;
             }
@@ -379,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 {
@@ -16,6 +22,12 @@ struct Retired {
     /// The previous hash.
     previous: String,
 }
+/// The replacement version.
+#[schema]
+struct Superseded {
+    /// The current hash.
+    current: String,
+}
 /// A large payload.
 #[schema]
 struct Echo {
@@ -24,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
@@ -54,6 +158,11 @@ impl Probe {
         sausage_plugin::listen("/state/control", std::rc::Rc::new(Control)).await?;
         Ok(sausage_plugin::pid())
     }
+    /// Keep an idle listener and ownership lock until opt-in drain ends this process.
+    async fn drain_service(&self) -> Result<String> {
+        lock("/drain-owner").await?.keep();
+        self.service().await
+    }
     /// Add two numbers.
     ///
     /// The result is their sum.
@@ -111,6 +220,15 @@ impl Probe {
     fn retired(&self, p: Retired) -> Result<String> {
         Ok(p.previous)
     }
+    /// Drain on replacement while calls are still available.
+    #[sausage_plugin::on_superseded]
+    async fn superseded(&self, p: Superseded) -> Result<()> {
+        call("other.superseded", json!({"current":p.current})).await?;
+        sausage_plugin::drain();
+        sausage_plugin::drain();
+        Ctx::default().notify("draining", json!({}))?;
+        Ok(())
+    }
     /// Wait until EOF, then run the stop hook.
     async fn wait(&self) -> Result<String> {
         call("other.wait", json!({})).await?;
@@ -155,6 +273,7 @@ impl Probe {
     #[sausage_plugin::on_stop]
     async fn stop(&self) {
         std::fs::write("/stopped", b"yes").unwrap();
+        std::fs::write(format!("/stopped-{}", sausage_plugin::pid()), b"yes").unwrap();
         Ctx::default().notify("stopped", json!({})).unwrap();
         assert!(call("other.no", json!({})).await.is_err());
     }
@@ -193,6 +312,11 @@ impl Control {
         ctx.notify("endpoint-progress", json!({"value":7}))?;
         Ok(7)
     }
+    /// Finish an in-flight call after draining begins.
+    async fn wait(&self) -> Result<String> {
+        call("other.inflight", json!({})).await?;
+        Ok("finished".into())
+    }
     /// End the endpoint process after the response.
     fn stop(&self, ctx: &Ctx) -> Result<()> {
         ctx.exit_after_response();

sdk/tests/ui/superseded.rs

@@ -0,0 +1,11 @@
+struct Supersedes;
+#[sausage_plugin::rpc]
+impl Supersedes {
+    /// First superseded hook.
+    #[sausage_plugin::on_superseded]
+    fn a(&self) -> sausage_plugin::Result<()> { Ok(()) }
+    /// Second superseded hook.
+    #[sausage_plugin::on_superseded]
+    fn b(&self) -> sausage_plugin::Result<()> { Ok(()) }
+}
+fn main() {}

sdk/tests/ui/superseded.stderr

@@ -0,0 +1,5 @@
+error: only one on_superseded method is allowed
+ --> tests/ui/superseded.rs:7:5
+  |
+7 |     /// Second superseded hook.
+  |     ^^^^^^^^^^^^^^^^^^^^^^^^^^^

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();
@@ -201,6 +229,103 @@ fn protocol() {
     assert_eq!(outgoing["method"], "other.background");
     p.answer(&outgoing, json!({}));
     assert!(p.child.wait().unwrap().success());
+    // Replacement is a notification, not EOF. Drain keeps both endpoint handlers
+    // and background calls alive, while closing idle connections and the listener.
+    assert!(
+        !d["result"]["methods"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .any(|m| m["name"] == "rpc.superseded" || m["name"] == "superseded")
+    );
+    std::fs::remove_file(dir.join("stopped")).unwrap();
+    let mut provider = start(&wasm, &dir);
+    provider.request("serve", json!({}));
+    assert_eq!(provider.read()["result"], Value::Null);
+    provider.request("start", json!({}));
+    let first = provider.read();
+    let second = provider.read();
+    let background = if first["method"] == "other.background" {
+        first
+    } else {
+        second
+    };
+    assert_eq!(background["method"], "other.background");
+    let address = std::fs::read_to_string(dir.join("endpoint"))
+        .unwrap()
+        .split_whitespace()
+        .next()
+        .unwrap()
+        .trim_start_matches("tcp:")
+        .to_owned();
+    let mut idle = std::net::TcpStream::connect(&address).unwrap();
+    idle.set_read_timeout(Some(std::time::Duration::from_secs(4)))
+        .unwrap();
+    writeln!(
+        idle,
+        "{}",
+        json!({"jsonrpc":"2.0","id":0,"method":"rpc.endpoint","params":{}})
+    )
+    .unwrap();
+    let mut idle = BufReader::new(idle);
+    let mut line = String::new();
+    idle.read_line(&mut line).unwrap();
+    assert_eq!(
+        sausage_plugin::serde_json::from_str::<Value>(&line).unwrap()["id"],
+        0
+    );
+    let mut active = std::net::TcpStream::connect(&address).unwrap();
+    active
+        .set_read_timeout(Some(std::time::Duration::from_secs(4)))
+        .unwrap();
+    writeln!(
+        active,
+        "{}",
+        json!({"jsonrpc":"2.0","id":1,"method":"wait","params":{}})
+    )
+    .unwrap();
+    let inflight = provider.read();
+    assert_eq!(inflight["method"], "other.inflight");
+    provider
+        .send(json!({"jsonrpc":"2.0","method":"rpc.superseded","params":{"current":"new-hash"}}));
+    let hook = provider.read();
+    assert_eq!(hook["method"], "other.superseded");
+    assert_eq!(hook["params"]["current"], "new-hash");
+    provider.answer(&hook, json!({}));
+    assert_eq!(provider.read()["params"]["name"], "draining");
+    line.clear();
+    assert_eq!(idle.read_line(&mut line).unwrap(), 0);
+    // 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();
+    BufReader::new(active).read_line(&mut line).unwrap();
+    assert_eq!(
+        sausage_plugin::serde_json::from_str::<Value>(&line).unwrap()["result"],
+        "finished"
+    );
+    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"
+    );
     let provider_doc = &d["result"];
     let imports = target.join("imports");
     std::fs::create_dir_all(&imports).unwrap();

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.