Keep file-watch observation from serializing unrelated IO

Scope open-registration and rename/remove ordering to overlapping paths;
let descriptor mutations and directory creation proceed without that lock.
Prove blocked opens/writes cannot stall unrelated namespaces, including a
queued rename, and retain post-rename descriptor notifications. Document
handler-owned watch lifetimes and narrow paths; test natural idle exit.

Card: SAU-359

sdk/src/calls.rs

@@ -27,10 +27,16 @@ impl Call {
     }
     /// Receive progress notifications before the result arrives.
     ///
-    /// For `core.watch`, await this call in a background task for the process's
-    /// lifetime. Scan only after `armed`, and rescan on `changed`; events can
-    /// coalesce. The call has no cancellation API. Do not use endpoint connect
-    /// retries to wait for file publication.
+    /// 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.
+    ///
+    /// 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

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

@@ -63,6 +63,41 @@ impl Probe {
         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
+    }
+
     /// Publish a control endpoint until EOF.
     async fn serve(&self) -> Result<()> {
         sausage_plugin::listen("/endpoint", std::rc::Rc::new(Control)).await

sdk/tests/wasi.rs

@@ -108,6 +108,15 @@ 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();
     let mut p = start(&wasm, &dir);
     p.request("rpc.discover", json!({}));
     let d = p.read();