clawft_kernel/
mesh_framing.rs

1//! Length-prefix framing for mesh wire protocol (K6.1).
2//!
3//! Each mesh frame is: `[4-byte big-endian length][1-byte message type][payload]`
4//! Maximum frame size is [`MAX_MESSAGE_SIZE`] (16 MiB).
5//!
6//! ## Frame type vs IPC encoding (ADR-031 / WEFT-683)
7//!
8//! `FrameType` is the **message kind** discriminator. It is orthogonal to
9//! how an `IpcMessage` payload is serialized — that is
10//! [`crate::mesh_ipc::MeshIpcEncoding`] (JSON shipped; RVF deferred behind
11//! `mesh-rvf`). See ADR-031 for the clarified layering.
12
13use serde::{Deserialize, Serialize};
14
15use crate::mesh::{MAX_MESSAGE_SIZE, MeshError, MeshStream};
16
17/// Mesh message types for framing dispatch.
18///
19/// Type bytes identify *what* the payload is, not *how* it is encoded.
20/// IPC payload encoding is [`crate::mesh_ipc::MeshIpcEncoding`].
21///
22/// # Wire `msg_type` (WEFT-115)
23///
24/// This enum **is** the exhaustive wire `msg_type` set. Prefer the
25/// [`MsgType`] alias in documentation and external protocol maps.
26#[non_exhaustive]
27#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
28#[repr(u8)]
29#[serde(rename_all = "snake_case")]
30pub enum FrameType {
31    /// WeftOS handshake payload.
32    Handshake = 0x01,
33    /// KernelMessage IPC envelope (payload encoding: see `mesh_ipc`).
34    IpcMessage = 0x02,
35    /// Chain sync request/response.
36    ChainSync = 0x03,
37    /// Tree sync request/response.
38    TreeSync = 0x04,
39    /// Service advertisement.
40    ServiceAdvert = 0x05,
41    /// Process advertisement.
42    ProcessAdvert = 0x06,
43    /// Heartbeat ping/pong.
44    Heartbeat = 0x07,
45    /// Join request.
46    JoinRequest = 0x08,
47    /// Join response.
48    JoinResponse = 0x09,
49    /// Sync state digest.
50    SyncDigest = 0x0A,
51    /// Artifact control (mesh-placement-11): `meta_request`, `announce`
52    /// (have bitfield), `request` (piece indexes). See `mesh_artifact_wire`.
53    ArtifactRequest = 0x0B,
54    /// Artifact data (mesh-placement-11): `meta` (piece list), `piece`
55    /// blocks, `no_piece`, `reject`. See `mesh_artifact_wire`.
56    ArtifactResponse = 0x0C,
57    /// Log aggregation (K6-G2).
58    LogAggregation = 0x0D,
59    /// Assessment sync (K6.6 -- cross-project assessment mesh).
60    AssessmentSync = 0x0E,
61    /// LeWM sensor observation frame (WEFT-526): CBOR `mesh.sensor.v1.*`
62    /// signed payload (encoded / consensus / control). Topic lives in the
63    /// IPC envelope or publisher metadata; payload is observational-only.
64    SensorObservation = 0x0F,
65    /// Inference request (mesh-placement-19, ADR-101 section 5): one
66    /// OpenAI-compatible HTTP request for a role this node serves. See
67    /// `infer_proxy::wire`. Served only to admitted, verified peers.
68    InferRequest = 0x10,
69    /// Inference response (mesh-placement-19): `head`, body `chunk`s,
70    /// `end` or `error`. See `infer_proxy::wire`.
71    InferResponse = 0x11,
72}
73
74/// Wire protocol name for [`FrameType`] (WEFT-115).
75pub type MsgType = FrameType;
76
77/// Historical / protocol-doc name for a framed message (WEFT-115).
78pub type Frame = MeshFrame;
79
80impl FrameType {
81    /// All known frame types in discriminant order (exhaustive for tests).
82    pub const ALL: &'static [FrameType] = &[
83        Self::Handshake,
84        Self::IpcMessage,
85        Self::ChainSync,
86        Self::TreeSync,
87        Self::ServiceAdvert,
88        Self::ProcessAdvert,
89        Self::Heartbeat,
90        Self::JoinRequest,
91        Self::JoinResponse,
92        Self::SyncDigest,
93        Self::ArtifactRequest,
94        Self::ArtifactResponse,
95        Self::LogAggregation,
96        Self::AssessmentSync,
97        Self::SensorObservation,
98        Self::InferRequest,
99        Self::InferResponse,
100    ];
101
102    /// Parse a byte into a known frame type, returning `None` for
103    /// unrecognised discriminants.
104    pub fn from_byte(b: u8) -> Option<Self> {
105        match b {
106            0x01 => Some(Self::Handshake),
107            0x02 => Some(Self::IpcMessage),
108            0x03 => Some(Self::ChainSync),
109            0x04 => Some(Self::TreeSync),
110            0x05 => Some(Self::ServiceAdvert),
111            0x06 => Some(Self::ProcessAdvert),
112            0x07 => Some(Self::Heartbeat),
113            0x08 => Some(Self::JoinRequest),
114            0x09 => Some(Self::JoinResponse),
115            0x0A => Some(Self::SyncDigest),
116            0x0B => Some(Self::ArtifactRequest),
117            0x0C => Some(Self::ArtifactResponse),
118            0x0D => Some(Self::LogAggregation),
119            0x0E => Some(Self::AssessmentSync),
120            0x0F => Some(Self::SensorObservation),
121            0x10 => Some(Self::InferRequest),
122            0x11 => Some(Self::InferResponse),
123            _ => None,
124        }
125    }
126
127    /// Wire discriminant byte.
128    pub const fn as_byte(self) -> u8 {
129        self as u8
130    }
131}
132
133/// A framed mesh message.
134#[derive(Debug, Clone, Serialize, Deserialize)]
135pub struct MeshFrame {
136    /// Discriminant identifying the payload contents.
137    pub frame_type: FrameType,
138    /// Raw payload bytes (deserialized by the receiver).
139    pub payload: Vec<u8>,
140}
141
142impl MeshFrame {
143    /// Encode frame to wire format: `[4-byte len][1-byte type][payload]`.
144    ///
145    /// The 4-byte length covers the type byte **plus** the payload,
146    /// so `len = 1 + payload.len()`.
147    pub fn encode(&self) -> Result<Vec<u8>, MeshError> {
148        let payload_len = self.payload.len();
149        let total = 1 + payload_len; // type byte + payload
150        if total > MAX_MESSAGE_SIZE {
151            return Err(MeshError::MessageTooLarge {
152                size: total,
153                max: MAX_MESSAGE_SIZE,
154            });
155        }
156        let mut buf = Vec::with_capacity(4 + total);
157        buf.extend_from_slice(&(total as u32).to_be_bytes());
158        buf.push(self.frame_type as u8);
159        buf.extend_from_slice(&self.payload);
160        Ok(buf)
161    }
162
163    /// Decode frame from wire bytes (after the 4-byte length prefix
164    /// has already been consumed).
165    pub fn decode(data: &[u8]) -> Result<Self, MeshError> {
166        if data.is_empty() {
167            return Err(MeshError::Transport("empty frame".into()));
168        }
169        let frame_type = FrameType::from_byte(data[0]).ok_or_else(|| {
170            MeshError::Transport(format!("unknown frame type: 0x{:02x}", data[0]))
171        })?;
172        let payload = data[1..].to_vec();
173        Ok(Self {
174            frame_type,
175            payload,
176        })
177    }
178}
179
180/// Read a single framed message from a mesh stream.
181///
182/// The transport delivers one message per `recv`, carrying exactly what
183/// [`write_frame`] sent: `[4-byte len][type][payload]`. The declared length
184/// must match the message and stay under the mesh cap.
185pub async fn read_frame(stream: &mut dyn MeshStream) -> Result<MeshFrame, MeshError> {
186    let data = stream.recv().await?;
187    if data.len() > MAX_MESSAGE_SIZE + 4 {
188        return Err(MeshError::MessageTooLarge {
189            size: data.len(),
190            max: MAX_MESSAGE_SIZE,
191        });
192    }
193    if data.len() < 5 {
194        return Err(MeshError::Transport("short frame".into()));
195    }
196    let declared = u32::from_be_bytes([data[0], data[1], data[2], data[3]]) as usize;
197    if declared > MAX_MESSAGE_SIZE {
198        return Err(MeshError::MessageTooLarge {
199            size: declared,
200            max: MAX_MESSAGE_SIZE,
201        });
202    }
203    if declared != data.len() - 4 {
204        return Err(MeshError::Transport("frame length mismatch".into()));
205    }
206    MeshFrame::decode(&data[4..])
207}
208
209/// Write a single framed message to a mesh stream.
210pub async fn write_frame(stream: &mut dyn MeshStream, frame: &MeshFrame) -> Result<(), MeshError> {
211    let encoded = frame.encode()?;
212    stream.send(&encoded).await
213}
214
215#[cfg(test)]
216mod tests {
217    use super::*;
218
219    #[test]
220    fn frame_type_from_byte_all_variants() {
221        let expected = [
222            (0x01, FrameType::Handshake),
223            (0x02, FrameType::IpcMessage),
224            (0x03, FrameType::ChainSync),
225            (0x04, FrameType::TreeSync),
226            (0x05, FrameType::ServiceAdvert),
227            (0x06, FrameType::ProcessAdvert),
228            (0x07, FrameType::Heartbeat),
229            (0x08, FrameType::JoinRequest),
230            (0x09, FrameType::JoinResponse),
231            (0x0A, FrameType::SyncDigest),
232            (0x0B, FrameType::ArtifactRequest),
233            (0x0C, FrameType::ArtifactResponse),
234            (0x0D, FrameType::LogAggregation),
235            (0x0E, FrameType::AssessmentSync),
236            (0x0F, FrameType::SensorObservation),
237            (0x10, FrameType::InferRequest),
238            (0x11, FrameType::InferResponse),
239        ];
240        for (byte, variant) in expected {
241            assert_eq!(FrameType::from_byte(byte), Some(variant));
242        }
243    }
244
245    #[tokio::test]
246    async fn write_then_read_frame_round_trips_and_checks_the_length() {
247        use crate::mesh_test_support::connected_pair;
248        let (mut a, mut b) = connected_pair().await.unwrap();
249        let f = MeshFrame {
250            frame_type: FrameType::InferRequest,
251            payload: vec![9, 8, 7],
252        };
253        write_frame(&mut a, &f).await.unwrap();
254        let g = read_frame(&mut b).await.unwrap();
255        assert_eq!(g.frame_type, FrameType::InferRequest);
256        assert_eq!(g.payload, vec![9, 8, 7]);
257        // Wrong declared length, short message, and over-cap length are refused.
258        let mut bad = f.encode().unwrap();
259        bad[3] = bad[3].wrapping_add(1);
260        a.send(&bad).await.unwrap();
261        assert!(read_frame(&mut b).await.is_err());
262        a.send(&[0, 0, 0]).await.unwrap();
263        assert!(read_frame(&mut b).await.is_err());
264        let mut huge = vec![0u8; 8];
265        huge[..4].copy_from_slice(&(MAX_MESSAGE_SIZE as u32 + 1).to_be_bytes());
266        a.send(&huge).await.unwrap();
267        assert!(matches!(
268            read_frame(&mut b).await,
269            Err(MeshError::MessageTooLarge { .. })
270        ));
271    }
272
273    #[test]
274    fn frame_type_from_byte_unknown() {
275        assert!(FrameType::from_byte(0x00).is_none());
276        assert!(FrameType::from_byte(0x12).is_none());
277        assert!(FrameType::from_byte(0xFF).is_none());
278    }
279
280    #[test]
281    fn encode_decode_roundtrip() {
282        let frame = MeshFrame {
283            frame_type: FrameType::Heartbeat,
284            payload: vec![1, 2, 3, 4],
285        };
286        let encoded = frame.encode().unwrap();
287
288        // First 4 bytes are the length prefix: 1 (type) + 4 (payload) = 5
289        let len = u32::from_be_bytes([encoded[0], encoded[1], encoded[2], encoded[3]]);
290        assert_eq!(len, 5);
291
292        // Decode the data portion (after length prefix)
293        let decoded = MeshFrame::decode(&encoded[4..]).unwrap();
294        assert_eq!(decoded.frame_type, FrameType::Heartbeat);
295        assert_eq!(decoded.payload, vec![1, 2, 3, 4]);
296    }
297
298    #[test]
299    fn encode_decode_empty_payload() {
300        let frame = MeshFrame {
301            frame_type: FrameType::Handshake,
302            payload: vec![],
303        };
304        let encoded = frame.encode().unwrap();
305        let decoded = MeshFrame::decode(&encoded[4..]).unwrap();
306        assert_eq!(decoded.frame_type, FrameType::Handshake);
307        assert!(decoded.payload.is_empty());
308    }
309
310    #[test]
311    fn decode_empty_data_fails() {
312        let result = MeshFrame::decode(&[]);
313        assert!(result.is_err());
314        let err = result.unwrap_err();
315        assert!(err.to_string().contains("empty frame"));
316    }
317
318    #[test]
319    fn decode_unknown_type_fails() {
320        let result = MeshFrame::decode(&[0xFF, 0x01, 0x02]);
321        assert!(result.is_err());
322        let err = result.unwrap_err();
323        assert!(err.to_string().contains("unknown frame type: 0xff"));
324    }
325
326    #[test]
327    fn encode_oversized_frame_fails() {
328        let frame = MeshFrame {
329            frame_type: FrameType::IpcMessage,
330            payload: vec![0u8; MAX_MESSAGE_SIZE], // payload alone = MAX, +1 for type byte
331        };
332        let result = frame.encode();
333        assert!(result.is_err());
334        match result.unwrap_err() {
335            MeshError::MessageTooLarge { size, max } => {
336                assert_eq!(size, MAX_MESSAGE_SIZE + 1);
337                assert_eq!(max, MAX_MESSAGE_SIZE);
338            }
339            other => panic!("expected MessageTooLarge, got: {other:?}"),
340        }
341    }
342
343    #[test]
344    fn all_frame_types_encode_decode() {
345        for ft in FrameType::ALL {
346            let frame = MeshFrame {
347                frame_type: *ft,
348                payload: vec![0xCA, 0xFE],
349            };
350            let encoded = frame.encode().unwrap();
351            let decoded = MeshFrame::decode(&encoded[4..]).unwrap();
352            assert_eq!(decoded.frame_type, *ft);
353            assert_eq!(decoded.payload, vec![0xCA, 0xFE]);
354        }
355    }
356
357    #[test]
358    fn frame_type_msg_type_alias_and_serde() {
359        // MsgType is the protocol name for FrameType (WEFT-115).
360        let t: MsgType = FrameType::ChainSync;
361        assert_eq!(t.as_byte(), 0x03);
362        let json = serde_json::to_string(&t).unwrap();
363        let restored: FrameType = serde_json::from_str(&json).unwrap();
364        assert_eq!(restored, FrameType::ChainSync);
365    }
366
367    #[test]
368    fn mesh_frame_serde_roundtrip() {
369        let frame = MeshFrame {
370            frame_type: FrameType::IpcMessage,
371            payload: vec![1, 2, 3],
372        };
373        let json = serde_json::to_string(&frame).unwrap();
374        let restored: Frame = serde_json::from_str(&json).unwrap();
375        assert_eq!(restored.frame_type, FrameType::IpcMessage);
376        assert_eq!(restored.payload, vec![1, 2, 3]);
377    }
378
379    #[test]
380    fn msg_type_all_variants_serde_roundtrip() {
381        for ft in FrameType::ALL {
382            let json = serde_json::to_string(ft).unwrap();
383            let restored: MsgType = serde_json::from_str(&json).unwrap();
384            assert_eq!(restored, *ft);
385        }
386    }
387}
clawft_kernel/mesh_framing.rs

clawft_kernel/
mesh_framing.rs
