Initial commit

crates/ocs_plugin_api/src/ipc/client.rs

@@ -0,0 +1,373 @@
+//! Plugin-side IPC client and `HostApi` proxy.
+
+use std::any::Any;
+use std::cell::{Cell, OnceCell, RefCell};
+use std::collections::HashMap;
+use std::rc::Rc;
+
+use acadrust::xdata::ExtendedDataRecord;
+use acadrust::{CadDocument, EntityType, Handle};
+use interprocess::local_socket::traits::Stream as StreamTrait;
+use interprocess::local_socket::{GenericNamespaced, Stream, ToNsName};
+
+use crate::host::{HostApi, InteractiveCommand};
+use crate::ipc::protocol::{
+    HostResponse, HostToPlugin, PluginRequest, PluginResponse, PluginToHost,
+};
+use crate::ipc::transport::{recv, send};
+
+/// Shared registry of active interactive commands, keyed by host-assigned id.
+pub type InteractiveRegistry = Rc<RefCell<HashMap<u64, Box<dyn InteractiveCommand>>>>;
+
+/// Plugin-side connection to the host.
+#[derive(Clone)]
+pub struct IpcClient {
+    stream: Rc<RefCell<Stream>>,
+}
+
+impl IpcClient {
+    pub fn connect(name: &str) -> std::io::Result<Self> {
+        let name = name
+            .to_ns_name::<GenericNamespaced>()
+            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e))?;
+        let stream = StreamTrait::connect(name)?;
+        Ok(Self::from_stream(stream))
+    }
+
+    pub(crate) fn from_stream(stream: Stream) -> Self {
+        Self {
+            stream: Rc::new(RefCell::new(stream)),
+        }
+    }
+
+    pub fn stream_ref(&self) -> std::cell::RefMut<'_, Stream> {
+        self.stream.borrow_mut()
+    }
+
+    /// Send a plugin request and wait for the matching response. Any nested
+    /// host requests that arrive while we are waiting are treated as errors.
+    pub fn request(
+        &self,
+        req: PluginRequest,
+    ) -> Result<PluginResponse, crate::ipc::transport::TransportError> {
+        send(&mut self.stream.borrow_mut(), &PluginToHost::Request(req))?;
+        loop {
+            match recv::<HostToPlugin>(&mut self.stream.borrow_mut())? {
+                HostToPlugin::Response(resp) => return Ok(resp),
+                HostToPlugin::Request(host_req) => {
+                    let resp = HostResponse::Error(format!(
+                        "unexpected nested host request: {host_req:?}"
+                    ));
+                    send(&mut self.stream.borrow_mut(), &PluginToHost::Response(resp))?;
+                }
+            }
+        }
+    }
+}
+
+/// `HostApi` implementation used inside the plugin process. Every host-mutating
+/// method is an RPC; `document()` / `document_mut()` return a local cached copy.
+pub struct PluginHostApi {
+    client: IpcClient,
+    tab_index: usize,
+    document_cache: OnceCell<CadDocument>,
+    interactive: InteractiveRegistry,
+    next_command_id: Cell<u64>,
+    /// Cache XDATA records so repeated reads for the same (handle, app) return
+    /// stable references without leaking on every call. Each distinct record is
+    /// leaked once per plugin dispatch/interactive session.
+    record_cache: RefCell<HashMap<(Handle, String), &'static ExtendedDataRecord>>,
+}
+
+impl PluginHostApi {
+    pub fn new(client: IpcClient, tab_index: usize, interactive: InteractiveRegistry) -> Self {
+        Self {
+            client,
+            tab_index,
+            document_cache: OnceCell::new(),
+            interactive,
+            next_command_id: Cell::new(1),
+            record_cache: RefCell::new(HashMap::new()),
+        }
+    }
+
+    fn fetch_document(&self) -> CadDocument {
+        match self.client.request(PluginRequest::DocumentSnapshot) {
+            Ok(PluginResponse::Document(doc)) => doc,
+            Ok(other) => {
+                eprintln!("[plugin] unexpected DocumentSnapshot response: {other:?}");
+                CadDocument::default()
+            }
+            Err(e) => {
+                eprintln!("[plugin] failed to fetch document snapshot: {e}");
+                CadDocument::default()
+            }
+        }
+    }
+}
+
+impl HostApi for PluginHostApi {
+    fn tab_index(&self) -> usize {
+        self.tab_index
+    }
+
+    fn document(&self) -> &CadDocument {
+        self.document_cache.get_or_init(|| self.fetch_document())
+    }
+
+    fn document_mut(&mut self) -> &mut CadDocument {
+        if self.document_cache.get().is_none() {
+            let doc = self.fetch_document();
+            let _ = self.document_cache.set(doc);
+        }
+        self.document_cache.get_mut().expect("document initialized")
+    }
+
+    fn add_entity(&mut self, entity: EntityType) -> Handle {
+        match self.client.request(PluginRequest::AddEntity(entity)) {
+            Ok(PluginResponse::Handle(h)) => h,
+            Ok(other) => {
+                eprintln!("[plugin] unexpected AddEntity response: {other:?}");
+                Handle::default()
+            }
+            Err(e) => {
+                eprintln!("[plugin] AddEntity failed: {e}");
+                Handle::default()
+            }
+        }
+    }
+
+    fn bump_geometry(&mut self) {
+        let _ = self.client.request(PluginRequest::BumpGeometry);
+    }
+
+    fn read_record(&self, handle: Handle, app_name: &str) -> Option<&ExtendedDataRecord> {
+        let key = (handle, app_name.to_string());
+        {
+            let cache = self.record_cache.borrow();
+            if let Some(&r) = cache.get(&key) {
+                return Some(r);
+            }
+        }
+        match self.client.request(PluginRequest::ReadRecord {
+            handle,
+            app_name: app_name.to_string(),
+        }) {
+            Ok(PluginResponse::Record(rec)) => rec.map(|r| {
+                // Leak once per distinct (handle, app_name) and reuse the
+                // reference for the lifetime of this PluginHostApi.
+                let leaked: &'static ExtendedDataRecord = Box::leak(Box::new(r));
+                self.record_cache.borrow_mut().insert(key, leaked);
+                leaked
+            }),
+            Ok(other) => {
+                eprintln!("[plugin] unexpected ReadRecord response: {other:?}");
+                None
+            }
+            Err(e) => {
+                eprintln!("[plugin] ReadRecord failed: {e}");
+                None
+            }
+        }
+    }
+
+    fn write_record(&mut self, handle: Handle, record: ExtendedDataRecord) -> bool {
+        let app = record.application_name.clone();
+        match self
+            .client
+            .request(PluginRequest::WriteRecord { handle, record })
+        {
+            Ok(PluginResponse::Bool(b)) => {
+                if b {
+                    self.record_cache
+                        .borrow_mut()
+                        .remove(&(handle, app));
+                }
+                b
+            }
+            Ok(other) => {
+                eprintln!("[plugin] unexpected WriteRecord response: {other:?}");
+                false
+            }
+            Err(e) => {
+                eprintln!("[plugin] WriteRecord failed: {e}");
+                false
+            }
+        }
+    }
+
+    fn remove_record(&mut self, handle: Handle, app_name: &str) -> bool {
+        match self.client.request(PluginRequest::RemoveRecord {
+            handle,
+            app_name: app_name.to_string(),
+        }) {
+            Ok(PluginResponse::Bool(b)) => {
+                if b {
+                    self.record_cache
+                        .borrow_mut()
+                        .remove(&(handle, app_name.to_string()));
+                }
+                b
+            }
+            Ok(other) => {
+                eprintln!("[plugin] unexpected RemoveRecord response: {other:?}");
+                false
+            }
+            Err(e) => {
+                eprintln!("[plugin] RemoveRecord failed: {e}");
+                false
+            }
+        }
+    }
+
+    fn push_undo(&mut self, label: &str) {
+        if let Err(e) = self
+            .client
+            .request(PluginRequest::PushUndo { label: label.to_string() })
+        {
+            eprintln!("[plugin] push_undo failed: {e}");
+        }
+    }
+
+    fn set_dirty(&mut self) {
+        if let Err(e) = self.client.request(PluginRequest::SetDirty) {
+            eprintln!("[plugin] set_dirty failed: {e}");
+        }
+    }
+
+    fn push_info(&mut self, msg: &str) {
+        if let Err(e) = self.client.request(PluginRequest::PushInfo(msg.to_string())) {
+            eprintln!("[plugin] push_info failed: {e}");
+        }
+    }
+
+    fn push_output(&mut self, msg: &str) {
+        if let Err(e) = self.client.request(PluginRequest::PushOutput(msg.to_string())) {
+            eprintln!("[plugin] push_output failed: {e}");
+        }
+    }
+
+    fn push_error(&mut self, msg: &str) {
+        if let Err(e) = self.client.request(PluginRequest::PushError(msg.to_string())) {
+            eprintln!("[plugin] push_error failed: {e}");
+        }
+    }
+
+    fn start_interactive(&mut self, command: Box<dyn InteractiveCommand>) {
+        let id = self.next_command_id.get();
+        self.next_command_id.set(id + 1);
+        self.interactive.borrow_mut().insert(id, command);
+        if let Err(e) = self
+            .client
+            .request(PluginRequest::StartInteractive { command_id: id })
+        {
+            eprintln!("[plugin] start_interactive failed: {e}");
+        }
+    }
+
+    fn plugin_state_any(&self, _plugin_id: &str) -> Option<&(dyn Any + Send + Sync)> {
+        // Per-tab plugin state stored in the host cannot cross the process
+        // boundary because `dyn Any` is not serializable. Plugins should keep
+        // their own state inside the plugin process.
+        None
+    }
+
+    fn plugin_state_any_mut(&mut self, _plugin_id: &str) -> Option<&mut (dyn Any + Send + Sync)> {
+        None
+    }
+
+    fn ensure_plugin_state_any(
+        &mut self,
+        _plugin_id: &'static str,
+        _init: &mut dyn FnMut() -> Box<dyn Any + Send + Sync>,
+    ) -> &mut (dyn Any + Send + Sync) {
+        // Same limitation as `plugin_state_any`. This would need a serializable
+        // state contract to work across processes.
+        panic!("ensure_plugin_state is not supported for out-of-process plugins; keep state in the plugin crate")
+    }
+}
+
+#[cfg(all(test, feature = "host"))]
+mod tests {
+    use std::sync::atomic::{AtomicU64, Ordering};
+    use std::thread;
+
+    use acadrust::{EntityType, Handle};
+    use acadrust::entities::Point;
+    use interprocess::local_socket::{
+        traits::{Listener, Stream as StreamTrait},
+        GenericNamespaced, ListenerOptions, Stream, ToNsName,
+    };
+
+    use crate::host::HostApi;
+    use crate::ipc::client::{IpcClient, PluginHostApi};
+    use crate::ipc::protocol::{HostToPlugin, PluginRequest, PluginResponse, PluginToHost};
+    use crate::ipc::transport::{recv, send};
+
+    fn unique_socket_name() -> String {
+        static COUNTER: AtomicU64 = AtomicU64::new(0);
+        let n = COUNTER.fetch_add(1, Ordering::Relaxed);
+        format!("ocs_plugin_client_test_{}_{}", std::process::id(), n)
+    }
+
+    fn make_client() -> (PluginHostApi, Stream) {
+        let name = unique_socket_name();
+        let name_ref = name
+            .clone()
+            .to_ns_name::<GenericNamespaced>()
+            .expect("valid name");
+        let listener = ListenerOptions::new()
+            .name(name_ref)
+            .create_sync()
+            .expect("listener");
+        let client_name = name.clone();
+        let client_thread = thread::spawn(move || {
+            StreamTrait::connect(client_name.to_ns_name::<GenericNamespaced>().unwrap())
+                .expect("connect")
+        });
+        let server = listener.accept().expect("accept");
+        let client_stream = client_thread.join().expect("client thread");
+        let client = IpcClient::from_stream(server);
+        let api = PluginHostApi::new(
+            client,
+            0,
+            std::rc::Rc::new(std::cell::RefCell::new(std::collections::HashMap::new())),
+        );
+        (api, client_stream)
+    }
+
+    #[test]
+    fn push_info_emits_request() {
+        let (mut api, mut peer) = make_client();
+        let peer_handle = thread::spawn(move || {
+            let msg = recv::<PluginToHost>(&mut peer).unwrap();
+            match msg {
+                PluginToHost::Request(PluginRequest::PushInfo(s)) => assert_eq!(s, "hello host"),
+                other => panic!("unexpected: {other:?}"),
+            }
+            send(&mut peer, &HostToPlugin::Response(PluginResponse::Ok)).unwrap();
+        });
+        api.push_info("hello host");
+        peer_handle.join().unwrap();
+    }
+
+    #[test]
+    fn add_entity_awaits_handle_response() {
+        let (mut api, mut peer) = make_client();
+        let peer_handle = thread::spawn(move || {
+            let msg = recv::<PluginToHost>(&mut peer).unwrap();
+            match msg {
+                PluginToHost::Request(PluginRequest::AddEntity(_)) => {}
+                other => panic!("unexpected: {other:?}"),
+            }
+            send(
+                &mut peer,
+                &HostToPlugin::Response(PluginResponse::Handle(Handle::new(42))),
+            )
+            .unwrap();
+        });
+        let handle = api.add_entity(EntityType::Point(Point::new()));
+        peer_handle.join().unwrap();
+        assert_eq!(handle, Handle::new(42));
+    }
+}

crates/ocs_plugin_api/src/ipc/mod.rs

@@ -0,0 +1,113 @@
+//! Inter-process communication layer for out-of-process plugins.
+//!
+//! Built only with the `host` feature because it needs `acadrust`-typed
+//! messages and the plugin runner binary.
+
+#[cfg(feature = "host")]
+pub mod client;
+#[cfg(feature = "host")]
+pub mod protocol;
+#[cfg(feature = "host")]
+pub mod server;
+#[cfg(feature = "host")]
+pub mod transport;
+
+#[cfg(all(test, feature = "host"))]
+mod tests {
+    use std::sync::atomic::{AtomicU64, Ordering};
+    use std::thread;
+
+    use interprocess::local_socket::{
+        traits::{Listener, Stream as StreamTrait},
+        GenericNamespaced, ListenerOptions, Stream, ToNsName,
+    };
+
+    use crate::ipc::protocol::{HostRequest, HostResponse, HostToPlugin, PluginToHost};
+    use crate::ipc::transport::{recv, send};
+
+    fn unique_socket_name() -> String {
+        static COUNTER: AtomicU64 = AtomicU64::new(0);
+        let n = COUNTER.fetch_add(1, Ordering::Relaxed);
+        format!("ocs_plugin_test_{}_{}", std::process::id(), n)
+    }
+
+    fn connect_pair() -> (Stream, Stream) {
+        let name = unique_socket_name();
+        let name_ref = name
+            .clone()
+            .to_ns_name::<GenericNamespaced>()
+            .expect("valid namespaced name");
+        let listener = ListenerOptions::new()
+            .name(name_ref)
+            .create_sync()
+            .expect("create listener");
+        let client = thread::spawn(move || {
+            StreamTrait::connect(name.to_ns_name::<GenericNamespaced>().unwrap())
+                .expect("connect")
+        });
+        let server = listener.accept().expect("accept");
+        let client = client.join().expect("client thread");
+        (server, client)
+    }
+
+    #[test]
+    fn transport_round_trips_host_request() {
+        let (mut a, mut b) = connect_pair();
+        let req = HostRequest::Dispatch {
+            cmd: "LINE".to_string(),
+        };
+        send(&mut a, &HostToPlugin::Request(req)).unwrap();
+        let got = recv::<HostToPlugin>(&mut b).unwrap();
+        match got {
+            HostToPlugin::Request(HostRequest::Dispatch { cmd }) => assert_eq!(cmd, "LINE"),
+            other => panic!("unexpected: {other:?}"),
+        }
+    }
+
+    #[test]
+    fn transport_round_trips_plugin_request() {
+        let (mut a, mut b) = connect_pair();
+        let req = PluginToHost::Request(crate::ipc::protocol::PluginRequest::PushInfo(
+            "hello".to_string(),
+        ));
+        send(&mut a, &req).unwrap();
+        let got = recv::<PluginToHost>(&mut b).unwrap();
+        match got {
+            PluginToHost::Request(crate::ipc::protocol::PluginRequest::PushInfo(msg)) => {
+                assert_eq!(msg, "hello")
+            }
+            other => panic!("unexpected: {other:?}"),
+        }
+    }
+
+    #[test]
+    fn transport_rejects_oversized_message() {
+        let (mut a, _b) = connect_pair();
+        // A Vec<u8> larger than MAX_MESSAGE_SIZE should be rejected on send.
+        let huge = vec![0u8; 65 * 1024 * 1024];
+        let err = send(&mut a, &huge).unwrap_err();
+        assert!(format!("{err}").contains("too large"));
+    }
+
+    #[test]
+    fn protocol_host_response_serde_roundtrip() {
+        let resp = HostResponse::Text("pick a point".to_string());
+        let bytes = bincode::serialize(&resp).unwrap();
+        let got: HostResponse = bincode::deserialize(&bytes).unwrap();
+        match got {
+            HostResponse::Text(s) => assert_eq!(s, "pick a point"),
+            other => panic!("unexpected: {other:?}"),
+        }
+    }
+
+    #[test]
+    fn protocol_plugin_to_host_serde_roundtrip() {
+        let msg = PluginToHost::Response(HostResponse::Bool(true));
+        let bytes = bincode::serialize(&msg).unwrap();
+        let got: PluginToHost = bincode::deserialize(&bytes).unwrap();
+        match got {
+            PluginToHost::Response(HostResponse::Bool(b)) => assert!(b),
+            other => panic!("unexpected: {other:?}"),
+        }
+    }
+}

crates/ocs_plugin_api/src/ipc/protocol.rs

@@ -0,0 +1,112 @@
+//! Request/response envelopes exchanged between the host and a plugin process.
+//!
+//! A single bidirectional socket is used. Each side sends either a request
+//! (expecting a response) or a response (to a previous request). This lets the
+//! host handle plugin RPCs inline while it waits for the result of a host→plugin
+//! request such as `Dispatch`, avoiding the need for two sockets or threads.
+
+use serde::{Deserialize, Serialize};
+
+use crate::host::CommandStep;
+use crate::manifest::ApiVersion;
+use crate::ribbon::owned::{OwnedPluginManifest, OwnedRibbonGroup};
+
+pub use acadrust::{CadDocument, EntityType, Handle};
+pub use acadrust::xdata::ExtendedDataRecord;
+
+/// Events the host forwards to an active plugin `InteractiveCommand`.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum InteractiveEvent {
+    Point([f64; 3]),
+    Enter,
+    ObjectPick { handle: Handle, pt: [f64; 3] },
+}
+
+/// Requests the host sends to the plugin runner.
+#[derive(Debug, Serialize, Deserialize)]
+pub enum HostRequest {
+    GetManifest,
+    GetRibbon,
+    Dispatch { cmd: String },
+    InteractiveEvent { command_id: u64, event: InteractiveEvent },
+    GetPrompt { command_id: u64 },
+    NeedsEntityPick { command_id: u64 },
+    Shutdown,
+}
+
+/// Responses the plugin runner sends back for `HostRequest`.
+#[derive(Debug, Serialize, Deserialize)]
+pub enum HostResponse {
+    Bool(bool),
+    CommandStep(CommandStep),
+    Text(String),
+    Ribbon(Vec<OwnedRibbonGroup>),
+    Manifest(OwnedPluginManifest),
+    Error(String),
+}
+
+/// Requests the plugin runner sends to the host.
+#[derive(Debug, Serialize, Deserialize)]
+pub enum PluginRequest {
+    PushInfo(String),
+    PushOutput(String),
+    PushError(String),
+    AddEntity(EntityType),
+    BumpGeometry,
+    ReadRecord { handle: Handle, app_name: String },
+    WriteRecord { handle: Handle, record: ExtendedDataRecord },
+    RemoveRecord { handle: Handle, app_name: String },
+    PushUndo { label: String },
+    SetDirty,
+    StartInteractive { command_id: u64 },
+    DocumentSnapshot,
+}
+
+/// Responses the host sends back for `PluginRequest`.
+#[derive(Debug, Serialize, Deserialize)]
+pub enum PluginResponse {
+    Ok,
+    Bool(bool),
+    Handle(Handle),
+    Record(Option<ExtendedDataRecord>),
+    Document(CadDocument),
+    Error(String),
+}
+
+/// Messages sent from the host to the plugin runner.
+#[derive(Debug, Serialize, Deserialize)]
+pub enum HostToPlugin {
+    Request(HostRequest),
+    Response(PluginResponse),
+}
+
+/// Messages sent from the plugin runner to the host.
+#[derive(Debug, Serialize, Deserialize)]
+pub enum PluginToHost {
+    Request(PluginRequest),
+    Response(HostResponse),
+}
+
+/// Convenience helper for manifest serialization.
+impl From<&'static crate::manifest::PluginManifest> for OwnedPluginManifest {
+    fn from(m: &'static crate::manifest::PluginManifest) -> Self {
+        Self {
+            id: m.id.to_string(),
+            name: m.name.to_string(),
+            version: m.version.to_string(),
+            description: m.description.to_string(),
+            api_version: m.api_version.major,
+            ribbon_order: m.ribbon_order,
+            xdata_apps: m.xdata_apps.iter().map(|s| s.to_string()).collect(),
+            command_prefixes: m.command_prefixes.iter().map(|s| s.to_string()).collect(),
+        }
+    }
+}
+
+impl OwnedPluginManifest {
+    pub fn api_version(&self) -> ApiVersion {
+        ApiVersion {
+            major: self.api_version,
+        }
+    }
+}

crates/ocs_plugin_api/src/ipc/server.rs

@@ -0,0 +1,58 @@
+//! Host-side IPC request handler.
+
+use crate::host::HostApi;
+use crate::ipc::protocol::{PluginRequest, PluginResponse};
+
+/// Apply one plugin request to the host's `HostApi` implementation.
+///
+/// `on_start_interactive` is called when the plugin starts an interactive
+/// command; the host should install an adapter that sends
+/// `HostRequest::InteractiveEvent` back to the plugin process.
+pub fn handle_plugin_request(
+    host: &mut dyn HostApi,
+    req: PluginRequest,
+    on_start_interactive: &mut dyn FnMut(u64),
+) -> PluginResponse {
+    use PluginRequest::*;
+    match req {
+        PushInfo(msg) => {
+            host.push_info(&msg);
+            PluginResponse::Ok
+        }
+        PushOutput(msg) => {
+            host.push_output(&msg);
+            PluginResponse::Ok
+        }
+        PushError(msg) => {
+            host.push_error(&msg);
+            PluginResponse::Ok
+        }
+        AddEntity(entity) => PluginResponse::Handle(host.add_entity(entity)),
+        BumpGeometry => {
+            host.bump_geometry();
+            PluginResponse::Ok
+        }
+        ReadRecord { handle, app_name } => {
+            PluginResponse::Record(host.read_record(handle, &app_name).cloned())
+        }
+        WriteRecord { handle, record } => {
+            PluginResponse::Bool(host.write_record(handle, record))
+        }
+        RemoveRecord { handle, app_name } => {
+            PluginResponse::Bool(host.remove_record(handle, &app_name))
+        }
+        PushUndo { label } => {
+            host.push_undo(&label);
+            PluginResponse::Ok
+        }
+        SetDirty => {
+            host.set_dirty();
+            PluginResponse::Ok
+        }
+        StartInteractive { command_id } => {
+            on_start_interactive(command_id);
+            PluginResponse::Ok
+        }
+        DocumentSnapshot => PluginResponse::Document(host.document().clone()),
+    }
+}

crates/ocs_plugin_api/src/ipc/transport.rs

@@ -0,0 +1,52 @@
+//! Length-framed transport over `interprocess::local_socket` streams.
+
+use std::io::{Read, Write};
+
+use interprocess::local_socket::Stream;
+use serde::{de::DeserializeOwned, Serialize};
+
+/// Maximum serialized message size accepted over the wire (64 MiB). Prevents
+/// a malicious or buggy peer from exhausting host/runner memory.
+const MAX_MESSAGE_SIZE: usize = 64 * 1024 * 1024;
+
+/// Errors that can occur during transport.
+#[derive(Debug, thiserror::Error)]
+pub enum TransportError {
+    #[error("IO error: {0}")]
+    Io(#[from] std::io::Error),
+    #[error("serialization error: {0}")]
+    Encode(#[from] bincode::Error),
+    #[error("empty message")]
+    Empty,
+    #[error("message too large: {0} bytes")]
+    TooLarge(usize),
+}
+
+/// Send a length-framed serialized message.
+pub fn send<T: Serialize>(stream: &mut Stream, msg: &T) -> Result<(), TransportError> {
+    let bytes = bincode::serialize(msg)?;
+    if bytes.len() > MAX_MESSAGE_SIZE {
+        return Err(TransportError::TooLarge(bytes.len()));
+    }
+    let len = bytes.len() as u64;
+    stream.write_all(&len.to_le_bytes())?;
+    stream.write_all(&bytes)?;
+    stream.flush()?;
+    Ok(())
+}
+
+/// Receive a length-framed serialized message.
+pub fn recv<T: DeserializeOwned>(stream: &mut Stream) -> Result<T, TransportError> {
+    let mut len_buf = [0u8; 8];
+    stream.read_exact(&mut len_buf)?;
+    let len = u64::from_le_bytes(len_buf) as usize;
+    if len == 0 {
+        return Err(TransportError::Empty);
+    }
+    if len > MAX_MESSAGE_SIZE {
+        return Err(TransportError::TooLarge(len));
+    }
+    let mut buf = vec![0u8; len];
+    stream.read_exact(&mut buf)?;
+    Ok(bincode::deserialize(&buf)?)
+}