From facaf1d9b1f3afc2c68e064e653a329e9df84e9d Mon Sep 17 00:00:00 2001 From: Sausage Date: Mon, 2 Nov 2026 00:10:34 +0000 Subject: [PATCH] Generate native core discovery from typed dispatch contracts --- Cargo.lock | 2 + PROTOCOL.md | 16 ++- cli/Cargo.toml | 2 + cli/src/activation.rs | 45 ++++--- cli/src/core.rs | 224 +++++++++++++++++++++---------- cli/src/core/types.rs | 200 +++++++++++++++++++++++++++ cli/src/filesystem.rs | 3 + cli/src/graph.rs | 9 ++ cli/src/journal.rs | 35 ++++- cli/src/locks.rs | 8 +- cli/tests/host/core_discovery.rs | 146 ++++++++++++++++++++ cli/tests/host/main.rs | 1 + interface/src/client.rs | 5 +- interface/src/compat.rs | 47 ++++++- interface/tests/compat_maps.rs | 57 ++++++++ 15 files changed, 702 insertions(+), 98 deletions(-) create mode 100644 cli/src/core/types.rs create mode 100644 cli/tests/host/core_discovery.rs create mode 100644 interface/tests/compat_maps.rs diff --git a/Cargo.lock b/Cargo.lock index 6277921..10ba42b 100644 --- a/Cargo.lock +++ b/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", diff --git a/PROTOCOL.md b/PROTOCOL.md index 1dd9509..a1762ba 100644 --- a/PROTOCOL.md +++ b/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 diff --git a/cli/Cargo.toml b/cli/Cargo.toml index dbebbf6..b04a7c0 100644 --- a/cli/Cargo.toml +++ b/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" diff --git a/cli/src/activation.rs b/cli/src/activation.rs index cbd561e..be4161f 100644 --- a/cli/src/activation.rs +++ b/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!({ - "plugin": candidate.plugin, - "wasm": format!("/plugins/{}/plugin.wasm", candidate.plugin), - "hash": candidate.interface.hash, - "previous": old.map(|old| &old.hash).filter(|hash| !hash.is_empty()), - "method_count": candidate.interface.document["methods"].as_array().map_or(0, Vec::len), - "added": added, - "removed": removed, - "changed": changed, - "interface_check": if checked { "passed" } else { "skipped" }, - })); + summaries.insert( + candidate.plugin.clone(), + ActivationSummary { + plugin: candidate.plugin.clone(), + wasm: format!("/plugins/{}/plugin.wasm", candidate.plugin), + hash: candidate.interface.hash.clone(), + previous: old + .map(|old| old.hash.clone()) + .filter(|hash| !hash.is_empty()), + 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)) } } diff --git a/cli/src/core.rs b/cli/src/core.rs index 0f6fc0c..1c99f1d 100644 --- a/cli/src/core.rs +++ b/cli/src/core.rs @@ -1,75 +1,159 @@ -//! The native core methods and their JSON parameters. -use crate::{Host, RpcResult, error}; -use serde_json::{Value, json}; +//! Native core dispatch and OpenRPC, generated from one typed method registry. +mod types; +use types::*; +pub(crate) use types::{ActivationSummary, Held, InterfaceCheck}; -impl Host { - /// Execute a core method without starting a guest process. - pub(crate) async fn core( - &self, - caller: &str, - id: Option, - method: &str, - p: Value, - ) -> RpcResult { - match method { - "watch" => self.watch(caller, id, required_str(&p, "path")?).await, - "call" => { - self.route( - caller.into(), - id, - required_str(&p, "method")?.into(), - p.get("params").cloned().unwrap_or(json!({})), - p["detach"].as_bool().unwrap_or(false), - ) - .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")), +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 +// type. The same entry generates its descriptor; neither list can drift. +macro_rules! core_methods { + ($host:ident, $caller:ident, $id:ident, $wire:ident, $p:ident; + $($name:literal: $params:ty => $result:ty, $docs:literal $body:block)*) => { + #[cfg(test)] + const METHOD_NAMES: &[&str] = &[$($name),*]; + + pub(crate) fn discover() -> &'static Value { + static DOCUMENT: OnceLock = OnceLock::new(); + DOCUMENT.get_or_init(|| { + let mut doc = document("core", env!("CARGO_PKG_VERSION")); + $(method::<$params, $result>(&mut doc, $name, $docs);)* + finish_document(doc) + }) } + + impl Host { + /// Execute a native method without starting a guest process. + pub(crate) async fn core( + &$host, + $caller: &str, + $id: Option, + $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, + "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] - .as_str() - .ok_or_else(|| error(-32602, &format!("{name} must be a string"))) + +#[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 + ); + host.stop().await; + } } diff --git a/cli/src/core/types.rs b/cli/src/core/types.rs new file mode 100644 index 0000000..d2addb9 --- /dev/null +++ b/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, +} +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, +} + +/// 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, + /// Exact process ID. + #[serde(default, deserialize_with = "optional")] + pub pid: Option, + /// Inclusive lower entry ID. + #[serde(default, deserialize_with = "optional")] + pub since: Option, + /// Inclusive upper entry ID. + #[serde(default, deserialize_with = "optional")] + pub until: Option, + /// Literal substring in the line. + #[serde(default, deserialize_with = "optional")] + pub grep: Option, + /// Maximum newest entries; default 100, negative means no limit. + #[serde(default, deserialize_with = "optional")] + pub limit: Option, +} + +// 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, 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, + /// Single candidate WASM world path, used when plugins is omitted. + pub wasm: Option, + /// Group of candidates; takes precedence over plugin and wasm. Must not be empty. + pub plugins: Option>, + /// Skip interface checks only when true; default false. + #[serde(default, deserialize_with = "optional")] + pub force: Option, +} + +/// 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, + /// Group of plugins; takes precedence over plugin. Must not be empty. + pub plugins: Option>, + /// Skip interface checks only when true; default false. + #[serde(default, deserialize_with = "optional")] + pub force: Option, +} + +/// 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, + /// Number of methods in the main interface. + pub method_count: usize, + /// Sorted names added to the interface. + pub added: Vec, + /// Sorted names removed from the interface. + pub removed: Vec, + /// Sorted names whose description or reachable schema changed. + pub changed: Vec, + /// 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); +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: Value) -> Result { + serde_json::from_value(p).map_err(|error| crate::error(-32602, &error.to_string())) +} + +pub(super) fn result(value: Value) -> Result { + serde_json::from_value(value).map_err(|error| crate::error(-32603, &error.to_string())) +} diff --git a/cli/src/filesystem.rs b/cli/src/filesystem.rs index d0d83f5..ed084b5 100644 --- a/cli/src/filesystem.rs +++ b/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(); diff --git a/cli/src/graph.rs b/cli/src/graph.rs index 38ed297..9a95dcd 100644 --- a/cli/src/graph.rs +++ b/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; diff --git a/cli/src/journal.rs b/cli/src/journal.rs index 5b26b14..e09301b 100644 --- a/cli/src/journal.rs +++ b/cli/src/journal.rs @@ -9,15 +9,23 @@ use std::{ sync::{Mutex, mpsc}, }; use tokio::sync::oneshot; -#[derive(Debug, Serialize, Deserialize)] -struct Entry { +/// One journal entry, with its original ID and timestamp. +#[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), @@ -233,9 +241,26 @@ fn query_database(db: &Connection, filters: &Value) -> rusqlite::Result { 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!({ - "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)?, - })))?; + 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!(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::>()?; entries.reverse(); Ok(json!(entries)) diff --git a/cli/src/locks.rs b/cli/src/locks.rs index 94218f4..d2f9c1e 100644 --- a/cli/src/locks.rs +++ b/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; diff --git a/cli/tests/host/core_discovery.rs b/cli/tests/host/core_discovery.rs new file mode 100644 index 0000000..f0099da --- /dev/null +++ b/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 { + 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; +} diff --git a/cli/tests/host/main.rs b/cli/tests/host/main.rs index d0d6732..16420d6 100644 --- a/cli/tests/host/main.rs +++ b/cli/tests/host/main.rs @@ -6,5 +6,6 @@ mod cli; mod common; mod console; mod core; +mod core_discovery; mod sdk; mod watches; diff --git a/interface/src/client.rs b/interface/src/client.rs index baf3a94..b11bae3 100644 --- a/interface/src/client.rs +++ b/interface/src/client.rs @@ -48,8 +48,9 @@ pub fn client_rs(name: &str, doc: &Value) -> Result { 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::(wire) - .or_else(|_| syn::parse_str::(&format!("r#{wire}"))) + let rust_name = wire.replace('.', "_"); + let ident = syn::parse_str::(&rust_name) + .or_else(|_| syn::parse_str::(&format!("r#{rust_name}"))) .map_err(|e| e.to_string())?; let mut rs = method["result"]["schema"].clone(); local_refs(&mut rs); diff --git a/interface/src/compat.rs b/interface/src/compat.rs index f7d1040..b3cb49d 100644 --- a/interface/src/compat.rs +++ b/interface/src/compat.rs @@ -204,7 +204,7 @@ fn supported(sender: &Value, document: &Value, seen: &mut HashSet) -> 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( diff --git a/interface/tests/compat_maps.rs b/interface/tests/compat_maps.rs new file mode 100644 index 0000000..762d488 --- /dev/null +++ b/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") + ); +}