feat: connection handshake — Client::connect / Server::serve_handshaked

runtime/core/src/client.rs

@@ -19,7 +19,7 @@ use core::time::Duration;
 
 use serde::Serialize;
 
-use crate::contract::{Envelope, RuntimeError, WireFormat};
+use crate::contract::{Envelope, Handshake, RuntimeError, WireFormat};
 use crate::transport::Transport;
 use crate::wire;
 
@@ -33,6 +33,11 @@ pub struct Client<T, W> {
 }
 
 impl<T: Transport, W: WireFormat> Client<T, W> {
+    /// Bind a client to a transport **without a handshake** — "misaligned
+    /// mode". Use when the peer can't take part (a legacy server); a
+    /// wire-format or schema mismatch then surfaces later as a decode / framing
+    /// error instead of up front. [`connect`](Self::connect) is the checked
+    /// path.
     pub fn new(transport: T, format: W) -> Self {
         Self {
             transport,
@@ -43,6 +48,26 @@ impl<T: Transport, W: WireFormat> Client<T, W> {
         }
     }
 
+    /// Bind a client and run the connection [`Handshake`]: send `local`, read
+    /// the peer's, and refuse (`RuntimeError::Handshake`) if they disagree on
+    /// schema hash, wire format, or framing.
+    pub fn connect(transport: T, format: W, local: Handshake) -> Result<Self, RuntimeError> {
+        let mut client = Self::new(transport, format);
+        client.exchange_handshake(local)?;
+        Ok(client)
+    }
+
+    fn exchange_handshake(&mut self, local: Handshake) -> Result<(), RuntimeError> {
+        self.request.clear();
+        local.encode(&mut self.request);
+        self.transport.send(&self.request)?;
+
+        self.response.clear();
+        self.transport.recv(&mut self.response)?;
+        let peer = Handshake::decode(&self.response).ok_or(RuntimeError::Handshake)?;
+        local.check(&peer)
+    }
+
     /// Make call `call_id` with `params`, block for the response, and return
     /// its [`Envelope`] (borrowing this client's receive buffer) together with
     /// the format to decode it: `Ok(payload)` for the stub to read as `R`, or

runtime/core/src/contract/buf.rs

@@ -13,6 +13,9 @@ pub trait BufMut {
     fn put_u16_le(&mut self, value: u16) {
         self.put_slice(&value.to_le_bytes());
     }
+    fn put_u32_le(&mut self, value: u32) {
+        self.put_slice(&value.to_le_bytes());
+    }
     fn put_u64_le(&mut self, value: u64) {
         self.put_slice(&value.to_le_bytes());
     }

runtime/core/src/contract/error.rs

@@ -26,6 +26,10 @@ pub enum RuntimeError {
     /// knows — a newer schema. `id` is the schema-global error ordinal; the
     /// still-encoded body is not retained.
     Remote { id: u16 },
+    /// The connection handshake disagreed on schema hash, wire format, or
+    /// framing — or none arrived. Never raised when the handshake is skipped
+    /// (`Client::new` / `Server::serve` — "misaligned mode").
+    Handshake,
 }
 
 impl fmt::Display for RuntimeError {
@@ -37,6 +41,7 @@ impl fmt::Display for RuntimeError {
             RuntimeError::Timeout => f.write_str("call timed out"),
             RuntimeError::UnknownCall => f.write_str("unknown call"),
             RuntimeError::Remote { id } => write!(f, "unrecognised remote error (ordinal {id})"),
+            RuntimeError::Handshake => f.write_str("connection handshake mismatch"),
         }
     }
 }

runtime/core/src/contract/handshake.rs

@@ -0,0 +1,154 @@
+use crate::contract::{BufMut, RuntimeError};
+
+const MAGIC: [u8; 2] = *b"CO";
+const VERSION: u8 = 1;
+/// `[MAGIC:2][VERSION:1][ir_hash:u64 LE][wire_format:u64 LE][framing:u64 LE][capabilities:u32 LE]`
+const LEN: usize = 2 + 1 + 8 + 8 + 8 + 4;
+
+/// Name of the Comline datagram framing (`wire::encode_request` /
+/// `encode_response`). Pass it to [`Handshake::new`].
+pub const FRAMING_DATAGRAM: &str = "comline.datagram";
+
+/// 64-bit FNV-1a. Folds a wire-format / framing **name** into the fixed-size
+/// handshake without a central id registry — a user's add-on format picks a
+/// namespaced name (`"com.acme.myformat"`) and its hash won't collide with a
+/// built-in. Stable across platforms.
+pub fn name_hash(name: &str) -> u64 {
+    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
+    for b in name.as_bytes() {
+        h ^= u64::from(*b);
+        h = h.wrapping_mul(0x0000_0100_0000_01b3);
+    }
+    h
+}
+
+/// What each end declares when a connection opens: the schema fingerprint the
+/// generated code was built from, the serialization + framing it speaks, and
+/// its transport capability bits.
+///
+/// [`Client::connect`](crate::client::Client::connect) /
+/// [`Server::serve_handshaked`](crate::serve::Server::serve_handshaked) exchange
+/// these and refuse on a mismatch — catching "one end MessagePack, one end
+/// JSON" or two ends built from different schema versions. `Client::new` /
+/// `Server::serve` skip it ("misaligned mode" — for legacy peers that can't
+/// take part).
+///
+/// Lifetime-free and `Copy`: the format / framing are carried as
+/// [`name_hash`]es of their names, not the strings.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Handshake {
+    /// Fingerprint of the frozen IR the two ends generated from. A generator
+    /// emits this as a constant.
+    pub ir_hash: u64,
+    /// [`name_hash`] of the [`WireFormat::name`](crate::contract::WireFormat::name).
+    pub wire_format: u64,
+    /// [`name_hash`] of the framing name ([`FRAMING_DATAGRAM`], …).
+    pub framing: u64,
+    /// Transport capability bits (reliable / ordered / duplex / …). Advisory:
+    /// a difference here is not a mismatch.
+    pub capabilities: u32,
+}
+
+impl Handshake {
+    /// Build one from the format / framing **names** (hashed in).
+    pub fn new(ir_hash: u64, wire_format: &str, framing: &str, capabilities: u32) -> Self {
+        Self {
+            ir_hash,
+            wire_format: name_hash(wire_format),
+            framing: name_hash(framing),
+            capabilities,
+        }
+    }
+
+    /// Write the fixed-size handshake frame.
+    pub fn encode(&self, out: &mut dyn BufMut) {
+        out.put_slice(&MAGIC);
+        out.put_u8(VERSION);
+        out.put_u64_le(self.ir_hash);
+        out.put_u64_le(self.wire_format);
+        out.put_u64_le(self.framing);
+        out.put_u32_le(self.capabilities);
+    }
+
+    /// Parse one. `None` if it is truncated, or the magic / version is wrong
+    /// (a peer that never sent a handshake, or a different protocol).
+    pub fn decode(frame: &[u8]) -> Option<Self> {
+        let frame: [u8; LEN] = frame.get(..LEN)?.try_into().ok()?;
+        if frame[0] != MAGIC[0] || frame[1] != MAGIC[1] || frame[2] != VERSION {
+            return None;
+        }
+        Some(Self {
+            ir_hash: u64::from_le_bytes(frame[3..11].try_into().unwrap()),
+            wire_format: u64::from_le_bytes(frame[11..19].try_into().unwrap()),
+            framing: u64::from_le_bytes(frame[19..27].try_into().unwrap()),
+            capabilities: u32::from_le_bytes(frame[27..31].try_into().unwrap()),
+        })
+    }
+
+    /// Check a peer's declaration against ours. `Err(RuntimeError::Handshake)`
+    /// if `ir_hash`, `wire_format`, or `framing` disagree; capability bits are
+    /// allowed to differ.
+    pub fn check(&self, peer: &Handshake) -> Result<(), RuntimeError> {
+        let agree = self.ir_hash == peer.ir_hash
+            && self.wire_format == peer.wire_format
+            && self.framing == peer.framing;
+        if agree {
+            Ok(())
+        } else {
+            Err(RuntimeError::Handshake)
+        }
+    }
+}
+
+#[cfg(all(test, feature = "alloc"))]
+mod tests {
+    use super::*;
+    use alloc::vec::Vec;
+
+    fn sample() -> Handshake {
+        Handshake::new(0xdead_beef_0102_0304, "msgpack", FRAMING_DATAGRAM, 0b101)
+    }
+
+    #[test]
+    fn round_trips() {
+        let mut frame = Vec::new();
+        sample().encode(&mut frame);
+        assert_eq!(frame.len(), LEN);
+        assert_eq!(Handshake::decode(&frame), Some(sample()));
+    }
+
+    #[test]
+    fn name_hash_is_deterministic_and_name_specific() {
+        assert_eq!(name_hash("msgpack"), name_hash("msgpack"));
+        assert_ne!(name_hash("msgpack"), name_hash("json"));
+        assert_ne!(name_hash("msgpack"), name_hash("com.acme.msgpack"));
+    }
+
+    #[test]
+    fn rejects_truncated_or_foreign() {
+        assert_eq!(Handshake::decode(&[]), None);
+        assert_eq!(Handshake::decode(&[0u8; LEN]), None); // bad magic
+        let mut frame = Vec::new();
+        sample().encode(&mut frame);
+        frame.truncate(LEN - 1);
+        assert_eq!(Handshake::decode(&frame), None);
+    }
+
+    #[test]
+    fn check_agrees_and_disagrees() {
+        let ours = sample();
+        assert!(ours.check(&sample()).is_ok());
+
+        let mut caps_differ = sample();
+        caps_differ.capabilities = 0;
+        assert!(ours.check(&caps_differ).is_ok(), "capability bits may differ");
+
+        let fmt_differ =
+            Handshake::new(ours.ir_hash, "json", FRAMING_DATAGRAM, ours.capabilities);
+        assert_eq!(ours.check(&fmt_differ), Err(RuntimeError::Handshake));
+
+        let mut schema_differ = sample();
+        schema_differ.ir_hash = 0;
+        assert_eq!(ours.check(&schema_differ), Err(RuntimeError::Handshake));
+    }
+}

runtime/core/src/contract/mod.rs

@@ -11,11 +11,13 @@ mod call;
 mod dispatch;
 mod envelope;
 mod error;
+mod handshake;
 mod wire;
 
 pub use buf::{BufMut, SliceBuf};
 pub use call::CallError;
 pub use dispatch::{Dispatch, Kind};
 pub use envelope::Envelope;
 pub use error::RuntimeError;
+pub use handshake::{name_hash, Handshake, FRAMING_DATAGRAM};
 pub use wire::WireFormat;

runtime/core/src/contract/wire.rs

@@ -9,6 +9,13 @@ use crate::contract::{BufMut, RuntimeError};
 /// `decode` borrows from the input, so a generated `Msg<'de>` points straight
 /// into the receive buffer.
 pub trait WireFormat {
+    /// A stable name for this format — the connection
+    /// [`Handshake`](crate::contract::Handshake) folds it in (hashed) so the
+    /// two ends can catch "one speaks MessagePack, one speaks JSON". Built-ins
+    /// are bare (`"msgpack"`); a third-party format should namespace it
+    /// (`"com.acme.myformat"`) to avoid a clash.
+    fn name(&self) -> &'static str;
+
     fn encode<T: Serialize + ?Sized>(
         &self,
         value: &T,

runtime/core/src/format/msgpack.rs

@@ -26,6 +26,10 @@ impl Write for BufWriter<'_> {
 }
 
 impl WireFormat for MsgPack {
+    fn name(&self) -> &'static str {
+        "msgpack"
+    }
+
     fn encode<T: Serialize + ?Sized>(
         &self,
         value: &T,

runtime/core/src/serve.rs

@@ -3,7 +3,7 @@
 
 use alloc::vec::Vec;
 
-use crate::contract::{Dispatch, Kind, RuntimeError, WireFormat};
+use crate::contract::{Dispatch, Handshake, Kind, RuntimeError, WireFormat};
 use crate::transport::Transport;
 use crate::wire;
 
@@ -56,9 +56,30 @@ impl<D: Dispatch, W: WireFormat> Server<D, W> {
         Ok(true)
     }
 
-    /// Serve calls until the transport closes.
+    /// Serve calls until the transport closes. **No handshake** — pair with a
+    /// `Client::new` peer, or an aligned setup where a mismatch can't happen.
     pub fn serve<T: Transport>(&mut self, transport: &mut T) -> Result<(), RuntimeError> {
         while self.serve_one(transport)? {}
         Ok(())
     }
+
+    /// Run the connection [`Handshake`] against the connecting peer — send
+    /// `local`, read theirs, refuse (`RuntimeError::Handshake`) on a schema /
+    /// wire-format / framing mismatch — then [`serve`](Self::serve).
+    pub fn serve_handshaked<T: Transport>(
+        &mut self,
+        transport: &mut T,
+        local: Handshake,
+    ) -> Result<(), RuntimeError> {
+        self.response.clear();
+        local.encode(&mut self.response);
+        transport.send(&self.response)?;
+
+        self.recv.clear();
+        transport.recv(&mut self.recv)?;
+        let peer = Handshake::decode(&self.recv).ok_or(RuntimeError::Handshake)?;
+        local.check(&peer)?;
+
+        self.serve(transport)
+    }
 }