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

Cargo.lock

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

PROTOCOL.md

@@ -222,7 +222,7 @@ these areas fails.
 /proc/
   core/
     PROTOCOL.md          this file
-    discover.json        generated core.rpc.discover document
+    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
@@ -390,7 +390,7 @@ At start, the core starts each service in `/config/boot.yaml`, in order:
 
 | Method | Params | Result |
 | --- | --- | --- |
-| `core.rpc.discover` | none | generated OpenRPC 1.2 document for every native core method |
+| `core.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) |
@@ -404,17 +404,6 @@ At start, the core starts each service in `/config/boot.yaml`, in order:
 | `core.restart` | none | `{}`. With `SAUSAGE_LOOP=1`, schedule clean shutdown and exit code 75. |
 | `core.exit` | none | Alias of `core.restart`: same supervisor gate, response, coalescing, and exit code 75. |
 
-`core.rpc.discover` and `/proc/core/discover.json` expose the same document,
-with unprefixed method names, just like a plugin's discovery document. The
-core generates it from the parameter and result types used by its dispatch;
-there is no separate JSON schema file to maintain. Copy the proc file to
-`imports/core.json` and use `import!("core")` to get typed calls, including
-`core::lock` and `core::log`. Dotted wire names become underscores in Rust
-(`rpc.discover` becomes `rpc_discover`). Import checks recognize the native
-core as a provider without requiring a `plugins/core` directory. Forwarded
-results and discovery documents contain arbitrary JSON; other core results
-are typed, including activation summaries and journal entries.
-
 ### Host restart
 
 **HARD RULE: the mediator and all agents NEVER call `core.restart` without
@@ -485,8 +474,7 @@ imported method:
 - There are no version numbers. This check replaces them.
 - The check covers a fixed subset of JSON Schema: objects with `properties`
   and `required`, the primitive types, arrays, optional and nullable values,
-  string enums, integer ranges, typed dictionaries (`additionalProperties`
+  string enums, integer ranges, and `$ref`. A construct outside the subset
-  with a value schema), and `$ref`. A construct outside the subset
   fails the check.
 - The check covers calls, not state. Old and new processes share the files in
   `/state`, and `core.revert` restores only code. A plugin must keep its

cli/Cargo.toml

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

cli/src/activation.rs

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

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

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

cli/src/core/types.rs

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

cli/src/filesystem.rs

@@ -85,11 +85,12 @@ 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()));
-        files.insert("/discover.json", || {
-            Ok(serde_json::to_vec(crate::core::discover()).unwrap())
-        });
         let graph = self.graph_json();
         files.insert("/graph.json", move || Ok(graph.clone()));
         let dot = self.graph_dot();

cli/src/graph.rs

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

cli/src/journal.rs

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

cli/src/locks.rs

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

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

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

cli/tests/host/core_discovery.rs

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

cli/tests/host/main.rs

@@ -6,6 +6,5 @@ mod cli;
 mod common;
 mod console;
 mod core;
-mod core_discovery;
 mod sdk;
 mod watches;

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;
+}

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

@@ -90,9 +90,6 @@ fn wait_eof(lines: &mut Lines) {
     std::fs::write("/state/stopped", "yes").unwrap();
 }
 fn busy() -> Value {
-    send(json!({"jsonrpc":"2.0", "method":"core.notify", "params":{
-        "name":"busy-ready", "params":{}
-    }}));
     loop {
         std::hint::spin_loop();
     }

interface/src/client.rs

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

interface/src/compat.rs

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

interface/tests/compat_maps.rs

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

sdk/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"