sdk, agent: review fixes (T18R)

- The serve loop ends the process after EOF or exit_after_response,
  also with background tasks or listeners.
- One schema generator per direction for the whole document, so types
  with the same name get distinct names.
- The agent gives the complete error (code, message, data) to the
  model, writes SKIP only after a successful request, caps tool output
  at 64 KiB, writes each history line in one write, and skips a damaged
  line.
- Endpoints skip blank and non-RPC lines. client_rs does not panic.
- #[must_use] on builders and the lock guard; #[schemars] attributes
  and duplicate hooks are compile errors; raw identifiers; -32000 for
  local SDK errors; x-sausage-params only for methods with params.

sdk/src/calls.rs

@@ -1,6 +1,7 @@
 //! Outgoing core calls and notification callbacks.
 use super::*;
 /// An outgoing request builder.
+#[must_use = "await the call to send it"]
 pub struct Call {
     method: String,
     params: Value,
@@ -43,7 +44,7 @@ impl IntoFuture for Call {
                 let mut s = s.borrow_mut();
                 if !s.active || s.stopping {
                     return Err(Error::new(
-                        -32002,
+                        -32000,
                         "no core transport (or stdin reached EOF)",
                     ));
                 }
@@ -72,7 +73,7 @@ impl IntoFuture for Call {
                 return Err(e);
             }
             rx.await
-                .unwrap_or_else(|_| Err(Error::new(-32002, "core transport closed")))
+                .unwrap_or_else(|_| Err(Error::new(-32000, "core transport closed")))
         })
     }
 }

sdk/src/discovery.rs

@@ -31,16 +31,7 @@ impl<T: SchemaType> SchemaType for Option<T> {}
 
 /// Create an OpenRPC document builder for generated interfaces.
 #[doc(hidden)]
-pub fn document(title: &str, version: &str) -> Value {
-    let mut doc = sausage_interface::Document::new(title, version).into_value();
-    doc["x-sdk-input"] = json!({});
-    doc["x-sdk-output"] = json!({});
-    doc
-}
-
-/// Add a method with independent deserialize and serialize contracts.
-#[doc(hidden)]
-pub fn method<P: Params, R: SchemaType>(doc: &mut Value, name: &str, docs: &str) {
+pub fn document(title: &str, version: &str) -> Discovery {
     fn generator(serialize: bool) -> schemars::SchemaGenerator {
         let mut settings = schemars::generate::SchemaSettings::draft07();
         settings.definitions_path = if serialize {
@@ -56,10 +47,44 @@ pub fn method<P: Params, R: SchemaType>(doc: &mut Value, name: &str, docs: &str)
         }
         .into_generator()
     }
-    let mut input = generator(false);
-    let params_ref = serde_json::to_value(input.subschema_for::<P>()).unwrap();
-    let definitions = input.take_definitions(true);
-    let schema = &definitions[P::schema_name().as_ref()];
+    Discovery {
+        value: sausage_interface::Document::new(title, version).into_value(),
+        input: generator(false),
+        output: generator(true),
+    }
+}
+
+/// Shared generators for all methods and endpoint types in one document.
+#[doc(hidden)]
+pub struct Discovery {
+    /// The OpenRPC document before reference resolution.
+    pub value: Value,
+    input: schemars::SchemaGenerator,
+    output: schemars::SchemaGenerator,
+}
+impl Discovery {
+    /// Add an endpoint with the same input and output generators.
+    pub fn endpoint<T: Interface>(&mut self, name: &str) {
+        let methods = std::mem::replace(&mut self.value["methods"], json!([]));
+        T::add_methods(self);
+        let endpoint = std::mem::replace(&mut self.value["methods"], methods);
+        self.value["x-sausage-endpoints"][name] = json!({"methods":endpoint});
+    }
+}
+
+/// Add a method with independent deserialize and serialize contracts.
+#[doc(hidden)]
+pub fn method<P: Params, R: SchemaType>(doc: &mut Discovery, name: &str, docs: &str) {
+    let params_ref = if P::schema_id() == <Empty as schemars::JsonSchema>::schema_id() {
+        Value::Null
+    } else {
+        serde_json::to_value(doc.input.subschema_for::<P>()).unwrap()
+    };
+    let schema = params_ref["$ref"]
+        .as_str()
+        .and_then(|reference| reference.strip_prefix("#/sdk-input/"))
+        .and_then(|name| doc.input.definitions().get(name))
+        .unwrap_or(&Value::Null);
     let params:Vec<_>=schema["properties"].as_object().into_iter().flatten().map(|(name,field)| {
         json!({
             "name": name,
@@ -68,18 +93,7 @@ pub fn method<P: Params, R: SchemaType>(doc: &mut Value, name: &str, docs: &str)
             "schema": field,
         })
     }).collect();
-    if !params.is_empty() {
-        doc["x-sdk-input"]
-            .as_object_mut()
-            .unwrap()
-            .extend(definitions);
-    }
-    let mut output = generator(true);
-    let result_schema = serde_json::to_value(output.subschema_for::<R>()).unwrap();
-    doc["x-sdk-output"]
-        .as_object_mut()
-        .unwrap()
-        .extend(output.take_definitions(true));
+    let result_schema = serde_json::to_value(doc.output.subschema_for::<R>()).unwrap();
     let has_params = !params.is_empty();
     let mut method = json!({
         "name":name,
@@ -92,14 +106,15 @@ pub fn method<P: Params, R: SchemaType>(doc: &mut Value, name: &str, docs: &str)
     if has_params {
         method["x-sausage-params"] = params_ref;
     }
-    doc["methods"].as_array_mut().unwrap().push(method);
+    doc.value["methods"].as_array_mut().unwrap().push(method);
 }
 
 /// Resolve shared schema names after all methods contribute their contracts.
 #[doc(hidden)]
-pub fn finish_document(mut doc: Value) -> Value {
-    let input = doc.as_object_mut().unwrap().remove("x-sdk-input").unwrap();
-    let output = doc.as_object_mut().unwrap().remove("x-sdk-output").unwrap();
+pub fn finish_document(mut discovery: Discovery) -> Value {
+    let input = serde_json::to_value(discovery.input.take_definitions(true)).unwrap();
+    let output = serde_json::to_value(discovery.output.take_definitions(true)).unwrap();
+    let mut doc = discovery.value;
     fn refs(value: &mut Value, names: &std::collections::HashMap<String, String>) {
         match value {
             Value::Object(map) => {

sdk/src/endpoints.rs

@@ -7,6 +7,9 @@ use serde_json::json;
 use std::{future::IntoFuture, marker::PhantomData, path::Path, rc::Rc};
 
 /// A verified endpoint address and identity.
+///
+/// Connect verifies and closes one connection. Each call opens a new connection
+/// and repeats the identity handshake before its request. The first call costs three round trips.
 #[derive(Clone)]
 #[cfg_attr(not(target_os = "wasi"), allow(dead_code))]
 pub struct Endpoint {
@@ -20,7 +23,7 @@ impl Endpoint {
         if self.interface == name {
             Ok(())
         } else {
-            Err(Error::new(-32002, "endpoint interface mismatch"))
+            Err(Error::new(-32000, "endpoint interface mismatch"))
         }
     }
     /// Build a typed endpoint request.
@@ -35,6 +38,7 @@ impl Endpoint {
     }
 }
 /// A lazy endpoint request with optional progress callbacks.
+#[must_use = "await the endpoint call to send it"]
 pub struct EndpointCall<T> {
     endpoint: Endpoint,
     method: String,
@@ -87,7 +91,7 @@ pub async fn listen<T: Interface>(path: impl AsRef<Path>, value: Rc<T>) -> Resul
 #[cfg(not(target_os = "wasi"))]
 pub async fn listen<T: Interface>(_path: impl AsRef<Path>, _value: Rc<T>) -> Result<()> {
     crate::transport::endpoint_name::<T>()?;
-    Err(Error::new(-32002, "endpoints require WASI"))
+    Err(Error::new(-32000, "endpoints require WASI"))
 }
 
 /// Read an endpoint file and verify its owner before returning a client.
@@ -122,7 +126,7 @@ pub async fn connect(path: impl AsRef<Path>) -> Result<Endpoint> {
 /// Connect to an endpoint. Native programs require the WASI runtime.
 #[cfg(not(target_os = "wasi"))]
 pub async fn connect(_path: impl AsRef<Path>) -> Result<Endpoint> {
-    Err(Error::new(-32002, "endpoints require WASI"))
+    Err(Error::new(-32000, "endpoints require WASI"))
 }
 
 #[cfg(target_os = "wasi")]
@@ -139,11 +143,10 @@ async fn handshake(endpoint: &Endpoint) -> Result<wstd::net::TcpStream> {
         json!({"jsonrpc":"2.0","id":0,"method":"rpc.endpoint","params":{}}),
     )
     .await?;
-    let bytes = crate::lines::Lines::default()
-        .next(&mut stream)
+    let reply = crate::lines::Lines::default()
+        .rpc(&mut stream)
         .await?
-        .ok_or_else(|| Error::new(-32002, "endpoint closed"))?;
-    let reply: Value = serde_json::from_slice(&bytes)?;
+        .ok_or_else(|| Error::new(-32000, "endpoint closed"))?;
     let result = &reply["result"];
     let plugin = endpoint
         .owner
@@ -154,7 +157,7 @@ async fn handshake(endpoint: &Endpoint) -> Result<wstd::net::TcpStream> {
         || result["interface"] != endpoint.interface
         || result["plugin"].as_str() != plugin
     {
-        return Err(Error::new(-32002, "endpoint identity mismatch"));
+        return Err(Error::new(-32000, "endpoint identity mismatch"));
     }
     Ok(stream)
 }
@@ -172,8 +175,7 @@ async fn endpoint_request(
         json!({"jsonrpc":"2.0","id":1,"method":method,"params":params}),
     )
     .await?;
-    while let Some(bytes) = lines.next(&mut stream).await? {
-        let value: Value = serde_json::from_slice(&bytes)?;
+    while let Some(value) = lines.rpc(&mut stream).await? {
         if value["method"] == "core.notify" {
             if let Some(f) = &mut notify {
                 f(
@@ -188,7 +190,7 @@ async fn endpoint_request(
             return Ok(value["result"].clone());
         }
     }
-    Err(Error::new(-32002, "endpoint closed before response"))
+    Err(Error::new(-32000, "endpoint closed before response"))
 }
 #[cfg(not(target_os = "wasi"))]
 async fn endpoint_request(
@@ -197,7 +199,7 @@ async fn endpoint_request(
     _params: Value,
     _notify: Option<Notify>,
 ) -> Result<Value> {
-    Err(Error::new(-32002, "endpoints require WASI"))
+    Err(Error::new(-32000, "endpoints require WASI"))
 }
 
 #[cfg(target_os = "wasi")]
@@ -209,11 +211,7 @@ async fn serve_connection<T: Interface>(
     use futures::{FutureExt, StreamExt};
     let (mut reader, mut writer) = stream.split();
     let mut lines = crate::lines::Lines::default();
-    while let Some(bytes) = lines.next(&mut reader).await? {
-        let request: Value = serde_json::from_slice(&bytes)?;
-        if request["jsonrpc"] != "2.0" {
-            continue;
-        }
+    while let Some(request) = lines.rpc(&mut reader).await? {
         let method = request["method"].as_str().unwrap_or_default();
         let (tx, mut rx) = futures::channel::mpsc::unbounded();
         let ctx = Ctx {

sdk/src/imports.rs

@@ -14,6 +14,7 @@ pub struct Import {
 inventory::collect!(Import);
 
 /// An outgoing call whose result decodes to a generated Rust type.
+#[must_use = "await the call to send it"]
 pub struct TypedCall<T> {
     call: Call,
     result: PhantomData<T>,

sdk/src/lib.rs

@@ -44,6 +44,8 @@
 //! [`listen`] publishes a declared interface at an endpoint file.
 //! [`connect`] verifies its owner before it returns an [`Endpoint`].
 //! Endpoint calls report progress through [`EndpointCall::on_notify`].
+//! Local transport errors use code -32000: no transport, stdin EOF, a closed endpoint,
+//! or an endpoint identity or interface mismatch. Code -32002 belongs to core process failures.
 //! Log with `eprintln!`. Stdout belongs to JSON-RPC.
 #![warn(missing_docs)]
 mod discovery;

sdk/src/lines.rs

@@ -1,22 +1,34 @@
 //! Buffered newline framing shared by stdin and endpoint connections.
-#[cfg(target_os = "wasi")]
+#[cfg(any(target_os = "wasi", test))]
 #[derive(Default)]
 pub(crate) struct Lines {
     bytes: Vec<u8>,
+    scanned: usize,
 }
-#[cfg(target_os = "wasi")]
+#[cfg(any(target_os = "wasi", test))]
 impl Lines {
+    fn frame(&mut self) -> Option<Vec<u8>> {
+        if let Some(end) = self.bytes[self.scanned..].iter().position(|b| *b == b'\n') {
+            let end = self.scanned + end;
+            self.scanned = 0;
+            return Some(self.bytes.drain(..=end).collect());
+        }
+        self.scanned = self.bytes.len();
+        None
+    }
+    #[cfg(target_os = "wasi")]
     pub(crate) async fn next(
         &mut self,
         input: &mut impl wstd::io::AsyncRead,
     ) -> std::io::Result<Option<Vec<u8>>> {
         loop {
-            if let Some(end) = self.bytes.iter().position(|b| *b == b'\n') {
-                return Ok(Some(self.bytes.drain(..=end).collect()));
+            if let Some(line) = self.frame() {
+                return Ok(Some(line));
             }
             let mut buffer = [0; 8192];
             let count = input.read(&mut buffer).await?;
             if count == 0 {
+                self.scanned = 0;
                 return Ok(if self.bytes.is_empty() {
                     None
                 } else {
@@ -26,4 +38,33 @@ impl Lines {
             self.bytes.extend_from_slice(&buffer[..count]);
         }
     }
+    #[cfg(target_os = "wasi")]
+    pub(crate) async fn rpc(
+        &mut self,
+        input: &mut impl wstd::io::AsyncRead,
+    ) -> std::io::Result<Option<serde_json::Value>> {
+        while let Some(bytes) = self.next(input).await? {
+            if let Ok(value) = serde_json::from_slice::<serde_json::Value>(&bytes)
+                && value.is_object()
+                && value["jsonrpc"] == "2.0"
+            {
+                return Ok(Some(value));
+            }
+        }
+        Ok(None)
+    }
+}
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn incomplete_frames_retain_the_scanned_length() {
+        let mut lines = super::Lines::default();
+        lines.bytes.extend_from_slice(b"first");
+        assert!(lines.frame().is_none());
+        assert_eq!(lines.scanned, 5);
+        lines.bytes.extend_from_slice(b"\nsecond\n");
+        assert_eq!(lines.frame().unwrap(), b"first\n");
+        assert_eq!(lines.frame().unwrap(), b"second\n");
+        assert!(lines.frame().is_none());
+    }
 }

sdk/src/locks.rs

@@ -1,6 +1,7 @@
 //! Advisory lock acquisition and guard lifetime.
 use super::*;
 /// An advisory lock acquisition.
+#[must_use = "await the lock to acquire it"]
 pub struct Lock {
     path: String,
     wait: bool,
@@ -20,6 +21,7 @@ impl Lock {
     }
 }
 /// Releases its lock on drop, unless kept until process exit.
+#[must_use = "bind the guard to hold the lock"]
 pub struct Guard {
     path: Option<String>,
 }

sdk/src/transport.rs

@@ -21,6 +21,11 @@ pub use locks::*;
 pub trait Interface: 'static {
     /// Describe this interface without constructing its value.
     fn describe() -> Value
+    where
+        Self: Sized;
+    /// Add this type's methods to shared document generators.
+    #[doc(hidden)]
+    fn add_methods(doc: &mut crate::Discovery)
     where
         Self: Sized;
     /// Describe only this type's methods, without endpoint declarations.
@@ -58,7 +63,7 @@ impl Ctx {
         let params = json!({"name":name,"params":params});
         if let Some(sink) = &self.sink {
             sink.unbounded_send(json!({"jsonrpc":"2.0","method":"core.notify","params":params}))
-                .map_err(|_| Error::new(-32002, "endpoint closed"))
+                .map_err(|_| Error::new(-32000, "endpoint closed"))
         } else {
             notify("core.notify", params)
         }
@@ -152,7 +157,7 @@ fn send(value: Value) -> Result<()> {
 }
 fn notify(method: &str, params: Value) -> Result<()> {
     if !STATE.with(|s| s.borrow().active) {
-        return Err(Error::new(-32002, "no core transport"));
+        return Err(Error::new(-32000, "no core transport"));
     }
     send(json!({"jsonrpc":"2.0","method":method,"params":params}))
 }
@@ -376,4 +381,6 @@ async fn serve<T: Interface>(value: T) {
         s.active = false;
         s.pending.clear();
     });
+    let _ = std::io::stdout().flush();
+    std::process::exit(0);
 }