Wait for busy guest readiness before measuring shutdown grace

Cargo.lock

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

PROTOCOL.md

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

cli/Cargo.toml

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

cli/src/activation.rs

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

cli/src/core.rs

@@ -1,75 +1,159 @@
-//! The native core methods and their JSON parameters.
+//! Native core dispatch and OpenRPC, generated from one typed method registry.
-use crate::{Host, RpcResult, error};
+mod types;
-use serde_json::{Value, json};
+use types::*;
+pub(crate) use types::{ActivationSummary, Held, InterfaceCheck};
 
-impl Host {
+use crate::{Host, RpcResult, error};
-    /// Execute a core method without starting a guest process.
+use sausage_plugin::{Empty, document, finish_document, method};
-    pub(crate) async fn core(
+use serde_json::{Value, json};
-        &self,
+use std::sync::OnceLock;
-        caller: &str,
+
-        id: Option<Value>,
+// Every dispatched method must supply docs, a parameter type, and a result
-        method: &str,
+// type. The same entry generates its descriptor; neither list can drift.
-        p: Value,
+macro_rules! core_methods {
-    ) -> RpcResult {
+    ($host:ident, $caller:ident, $id:ident, $wire:ident, $p:ident;
-        match method {
+     $($name:literal: $params:ty => $result:ty, $docs:literal $body:block)*) => {
-            "watch" => self.watch(caller, id, required_str(&p, "path")?).await,
+        #[cfg(test)]
-            "call" => {
+        const METHOD_NAMES: &[&str] = &[$($name),*];
-                self.route(
+
-                    caller.into(),
+        pub(crate) fn discover() -> &'static Value {
-                    id,
+            static DOCUMENT: OnceLock<Value> = OnceLock::new();
-                    required_str(&p, "method")?.into(),
+            DOCUMENT.get_or_init(|| {
-                    p.get("params").cloned().unwrap_or(json!({})),
+                let mut doc = document("core", env!("CARGO_PKG_VERSION"));
-                    p["detach"].as_bool().unwrap_or(false),
+                $(method::<$params, $result>(&mut doc, $name, $docs);)*
-                )
+                finish_document(doc)
-                .await
+            })
-            }
-            "notify" => {
-                self.notify(
-                    caller,
-                    required_str(&p, "name")?,
-                    p.get("params").cloned().unwrap_or(json!({})),
-                );
-                Ok(Value::Null)
-            }
-            "kill" => {
-                let pid = required_str(&p, "pid")?;
-                let ended = self
-                    .0
-                    .state
-                    .lock()
-                    .unwrap()
-                    .processes
-                    .get(pid)
-                    .map(|p| p.ended.subscribe());
-                self.kill(pid);
-                if let Some(mut ended) = ended {
-                    let _ = ended.wait_for(|ended| *ended).await;
-                }
-                Ok(json!({}))
-            }
-            "stop" => {
-                if let Some(mut n) = self.begin_stop(required_str(&p, "pid")?) {
-                    let _ = n.wait_for(|ended| *ended).await;
-                }
-                Ok(json!({}))
-            }
-            "lock" => {
-                self.lock(caller, required_str(&p, "path")?, p["wait"] != false)
-                    .await
-            }
-            "unlock" => {
-                self.unlock(caller, required_str(&p, "path")?);
-                Ok(Value::Null)
-            }
-            "log" => Ok(self.0.journal.query(p).await),
-            "activate" | "revert" => self.activation(caller, method, p).await,
-            "restart" | "exit" => self.request_restart(),
-            _ => Err(error(-32601, "method not found")),
         }
+
+        impl Host {
+            /// Execute a native method without starting a guest process.
+            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 {}
     }
 }
-fn required_str<'a>(params: &'a Value, name: &str) -> Result<&'a str, Value> {
+
-    params[name]
+#[cfg(test)]
-        .as_str()
+mod tests {
-        .ok_or_else(|| error(-32602, &format!("{name} must be a string")))
+    use super::*;
+    use crate::{runtime::Runtime, world::World};
+    use std::collections::BTreeSet;
+
+    /// Every descriptor names a dispatched method, and vice versa.
+    #[tokio::test]
+    async fn discovery_covers_dispatch_exactly() {
+        let dir = tempfile::tempdir().unwrap();
+        let world = dir.path().join("world");
+        World::create(&world).unwrap();
+        let host = Host::test_world(&world, Runtime::new(&dir.path().join("cache")).unwrap())
+            .await
+            .unwrap();
+        let doc = host.call("core.rpc.discover", json!({})).await.unwrap();
+        let names: BTreeSet<_> = doc["methods"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .map(|m| {
+                assert!(!m["description"].as_str().unwrap().is_empty());
+                m["name"].as_str().unwrap()
+            })
+            .collect();
+        assert_eq!(names, METHOD_NAMES.iter().copied().collect());
+        assert_eq!(names.len(), doc["methods"].as_array().unwrap().len());
+        for name in names {
+            let response = host.call(&format!("core.{name}"), json!({})).await;
+            if let Err(error) = response {
+                assert_ne!(error["code"], -32601, "descriptor without dispatch: {name}");
+            }
+        }
+        assert_eq!(
+            host.call("core.no_such_method", json!({}))
+                .await
+                .unwrap_err()["code"],
+            -32601
+        );
+        host.stop().await;
+    }
 }

cli/src/core/types.rs

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

cli/src/filesystem.rs

@@ -87,6 +87,9 @@ impl Host {
         let mut files = Generated::default();
         let protocol = include_bytes!("../../PROTOCOL.md");
         files.insert("/PROTOCOL.md", move || Ok(protocol.to_vec()));
+        files.insert("/discover.json", || {
+            Ok(serde_json::to_vec(crate::core::discover()).unwrap())
+        });
         let graph = self.graph_json();
         files.insert("/graph.json", move || Ok(graph.clone()));
         let dot = self.graph_dot();

cli/src/graph.rs

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

cli/src/journal.rs

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

cli/src/locks.rs

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

cli/tests/host/core.rs

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

cli/tests/host/core_discovery.rs

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

cli/tests/host/main.rs

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

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

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

interface/src/compat.rs

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

interface/tests/compat_maps.rs

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