mesh_llm_protocol/protocol/
mod.rs

1pub mod convert;
2pub mod v0;
3use anyhow::Result;
4pub use convert::*;
5use iroh::endpoint::{ConnectOptions, Connection};
6use iroh::{Endpoint, EndpointAddr};
7use prost::Message;
8pub use v0::*;
9pub const ALPN_CONTROL_V1: &[u8] = b"mesh-llm-control/1";
10pub const ALPN_V1: &[u8] = b"mesh-llm/1";
11pub const NODE_PROTOCOL_GENERATION: u32 = 1;
12pub const MAX_CONTROL_FRAME_BYTES: usize = 8 * 1024 * 1024;
13
14pub const STREAM_GOSSIP: u8 = 0x01;
15pub const STREAM_TUNNEL: u8 = 0x02;
16pub const STREAM_TUNNEL_MAP: u8 = 0x03;
17pub const STREAM_TUNNEL_HTTP: u8 = 0x04;
18pub const STREAM_ROUTE_REQUEST: u8 = 0x05;
19pub const STREAM_PEER_DOWN: u8 = 0x06;
20pub const STREAM_PEER_LEAVING: u8 = 0x07;
21pub const STREAM_PLUGIN_CHANNEL: u8 = 0x08;
22pub const STREAM_PLUGIN_BULK_TRANSFER: u8 = 0x09;
23/// Reserved legacy mesh-plane config subscription stream ID.
24///
25/// Config and inventory control now live exclusively on `mesh-llm-control/1`;
26/// keep 0x0b reserved so old wire values are not accidentally reused.
27pub const STREAM_CONFIG_SUBSCRIBE: u8 = 0x0b;
28/// Reserved legacy mesh-plane config push stream ID.
29///
30/// Config and inventory control now live exclusively on `mesh-llm-control/1`;
31/// keep 0x0c reserved so old wire values are not accidentally reused.
32pub const STREAM_CONFIG_PUSH: u8 = 0x0c;
33pub const STREAM_SUBPROTOCOL: u8 = 0x0d;
34const _: () = {
35    let _ = STREAM_CONFIG_SUBSCRIBE;
36    let _ = STREAM_CONFIG_PUSH;
37    let _ = STREAM_SUBPROTOCOL;
38};
39
40#[derive(Clone, Copy, Debug, Eq, PartialEq)]
41pub enum ControlProtocol {
42    ProtoV1,
43    JsonV0,
44}
45
46#[derive(Debug, PartialEq)]
47pub enum ControlFrameError {
48    OversizeFrame { size: usize },
49    BadGeneration { got: u32 },
50    InvalidEndpointId { got: usize },
51    InvalidSenderId { got: usize },
52    MissingHttpPort,
53    MissingOwnerId,
54    MissingControlOwnerId,
55    InvalidConfigHashLength { got: usize },
56    InvalidSubprotocol,
57    InvalidPublicKeyLength { got: usize },
58    MissingSignature,
59    InvalidSignatureLength { got: usize },
60    MissingConfig,
61    MissingControlEnvelope,
62    MissingControlCommand,
63    MissingControlResult,
64    MissingControlOwnership,
65    MissingRequestId,
66    InvalidOwnerControlErrorCode { got: i32 },
67    DecodeError(String),
68    WrongStreamType { expected: u8, got: u8 },
69    ForgedSender,
70}
71
72impl std::fmt::Display for ControlFrameError {
73    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
74        match self {
75            ControlFrameError::OversizeFrame { size } => write!(
76                f,
77                "control frame too large: {} bytes (max {})",
78                size, MAX_CONTROL_FRAME_BYTES
79            ),
80            ControlFrameError::BadGeneration { got } => write!(
81                f,
82                "bad protocol generation: expected {}, got {}",
83                NODE_PROTOCOL_GENERATION, got
84            ),
85            ControlFrameError::InvalidEndpointId { got } => {
86                write!(f, "invalid endpoint_id length: expected 32, got {}", got)
87            }
88            ControlFrameError::InvalidSenderId { got } => {
89                write!(f, "invalid sender_id length: expected 32, got {}", got)
90            }
91            ControlFrameError::MissingHttpPort => {
92                write!(f, "HOST-role peer annotation missing http_port")
93            }
94            ControlFrameError::MissingOwnerId => write!(f, "config frame missing owner_id"),
95            ControlFrameError::MissingControlOwnerId => {
96                write!(f, "owner control handshake missing owner_id")
97            }
98            ControlFrameError::InvalidConfigHashLength { got } => {
99                write!(f, "invalid config_hash length: expected 32, got {}", got)
100            }
101            ControlFrameError::InvalidSubprotocol => {
102                write!(f, "subprotocol entries require a non-empty name and major")
103            }
104            ControlFrameError::InvalidPublicKeyLength { got } => {
105                write!(f, "invalid public key length: expected 32, got {}", got)
106            }
107            ControlFrameError::MissingSignature => write!(f, "config push missing signature"),
108            ControlFrameError::InvalidSignatureLength { got } => {
109                write!(f, "invalid signature length: expected 64, got {got}")
110            }
111            ControlFrameError::MissingConfig => {
112                write!(f, "config field is required but missing")
113            }
114            ControlFrameError::MissingControlEnvelope => {
115                write!(f, "owner control envelope requires exactly one payload")
116            }
117            ControlFrameError::MissingControlCommand => {
118                write!(
119                    f,
120                    "owner control request requires exactly one command variant"
121                )
122            }
123            ControlFrameError::MissingControlResult => {
124                write!(
125                    f,
126                    "owner control response requires exactly one result variant"
127                )
128            }
129            ControlFrameError::MissingControlOwnership => {
130                write!(f, "owner control handshake missing ownership attestation")
131            }
132            ControlFrameError::MissingRequestId => {
133                write!(f, "owner control request_id must be non-zero")
134            }
135            ControlFrameError::InvalidOwnerControlErrorCode { got } => {
136                write!(f, "invalid owner control error code: {got}")
137            }
138            ControlFrameError::DecodeError(msg) => write!(f, "protobuf decode error: {}", msg),
139            ControlFrameError::WrongStreamType { expected, got } => write!(
140                f,
141                "wrong stream type: expected {:#04x}, got {:#04x}",
142                expected, got
143            ),
144            ControlFrameError::ForgedSender => {
145                write!(f, "frame peer_id does not match QUIC connection identity")
146            }
147        }
148    }
149}
150
151impl std::error::Error for ControlFrameError {}
152
153pub trait ValidateControlFrame: prost::Message + Default + Sized {
154    fn validate_frame(&self) -> Result<(), ControlFrameError> {
155        Ok(())
156    }
157}
158
159impl ValidateControlFrame for crate::proto::node::GossipFrame {
160    fn validate_frame(&self) -> Result<(), ControlFrameError> {
161        if self.r#gen != NODE_PROTOCOL_GENERATION {
162            return Err(ControlFrameError::BadGeneration { got: self.r#gen });
163        }
164        if self.sender_id.len() != 32 {
165            return Err(ControlFrameError::InvalidSenderId {
166                got: self.sender_id.len(),
167            });
168        }
169        for pa in &self.peers {
170            validate_peer_announcement(pa)?;
171        }
172        Ok(())
173    }
174}
175
176impl ValidateControlFrame for crate::proto::node::TunnelMap {
177    fn validate_frame(&self) -> Result<(), ControlFrameError> {
178        if self.owner_peer_id.len() != 32 {
179            return Err(ControlFrameError::InvalidEndpointId {
180                got: self.owner_peer_id.len(),
181            });
182        }
183        for entry in &self.entries {
184            if entry.target_peer_id.len() != 32 {
185                return Err(ControlFrameError::InvalidEndpointId {
186                    got: entry.target_peer_id.len(),
187                });
188            }
189        }
190        Ok(())
191    }
192}
193impl ValidateControlFrame for crate::proto::node::RouteTableRequest {
194    fn validate_frame(&self) -> Result<(), ControlFrameError> {
195        if self.r#gen != NODE_PROTOCOL_GENERATION {
196            return Err(ControlFrameError::BadGeneration { got: self.r#gen });
197        }
198        if !self.requester_id.is_empty() && self.requester_id.len() != 32 {
199            return Err(ControlFrameError::InvalidEndpointId {
200                got: self.requester_id.len(),
201            });
202        }
203        Ok(())
204    }
205}
206impl ValidateControlFrame for crate::proto::node::RouteTable {
207    fn validate_frame(&self) -> Result<(), ControlFrameError> {
208        if self.r#gen != NODE_PROTOCOL_GENERATION {
209            return Err(ControlFrameError::BadGeneration { got: self.r#gen });
210        }
211        for entry in &self.entries {
212            if entry.endpoint_id.len() != 32 {
213                return Err(ControlFrameError::InvalidEndpointId {
214                    got: entry.endpoint_id.len(),
215                });
216            }
217        }
218        Ok(())
219    }
220}
221impl ValidateControlFrame for crate::proto::node::PeerDown {
222    fn validate_frame(&self) -> Result<(), ControlFrameError> {
223        if self.r#gen != NODE_PROTOCOL_GENERATION {
224            return Err(ControlFrameError::BadGeneration { got: self.r#gen });
225        }
226        if self.peer_id.len() != 32 {
227            return Err(ControlFrameError::InvalidEndpointId {
228                got: self.peer_id.len(),
229            });
230        }
231        Ok(())
232    }
233}
234impl ValidateControlFrame for crate::proto::node::PeerLeaving {
235    fn validate_frame(&self) -> Result<(), ControlFrameError> {
236        if self.r#gen != NODE_PROTOCOL_GENERATION {
237            return Err(ControlFrameError::BadGeneration { got: self.r#gen });
238        }
239        if self.peer_id.len() != 32 {
240            return Err(ControlFrameError::InvalidEndpointId {
241                got: self.peer_id.len(),
242            });
243        }
244        Ok(())
245    }
246}
247
248impl ValidateControlFrame for crate::proto::node::OwnerControlEnvelope {
249    fn validate_frame(&self) -> Result<(), ControlFrameError> {
250        if self.r#gen != NODE_PROTOCOL_GENERATION {
251            return Err(ControlFrameError::BadGeneration { got: self.r#gen });
252        }
253        let payloads = [
254            self.handshake.is_some(),
255            self.request.is_some(),
256            self.response.is_some(),
257            self.error.is_some(),
258        ];
259        if payloads.into_iter().filter(|present| *present).count() != 1 {
260            return Err(ControlFrameError::MissingControlEnvelope);
261        }
262        if let Some(handshake) = &self.handshake {
263            handshake.validate_frame()?;
264        }
265        if let Some(request) = &self.request {
266            request.validate_frame()?;
267        }
268        if let Some(response) = &self.response {
269            response.validate_frame()?;
270        }
271        if let Some(error) = &self.error {
272            error.validate_frame()?;
273        }
274        Ok(())
275    }
276}
277
278impl ValidateControlFrame for crate::proto::node::OwnerControlHandshake {
279    fn validate_frame(&self) -> Result<(), ControlFrameError> {
280        let ownership = self
281            .ownership
282            .as_ref()
283            .ok_or(ControlFrameError::MissingControlOwnership)?;
284        if ownership.owner_id.trim().is_empty() {
285            return Err(ControlFrameError::MissingControlOwnerId);
286        }
287        validate_public_key_length(ownership.owner_sign_public_key.len())?;
288        validate_endpoint_id_length(ownership.node_endpoint_id.len())?;
289        if ownership.signature.is_empty() {
290            return Err(ControlFrameError::MissingSignature);
291        }
292        if ownership.signature.len() != 64 {
293            return Err(ControlFrameError::InvalidSignatureLength {
294                got: ownership.signature.len(),
295            });
296        }
297        Ok(())
298    }
299}
300
301impl ValidateControlFrame for crate::proto::node::OwnerControlRequest {
302    fn validate_frame(&self) -> Result<(), ControlFrameError> {
303        if self.request_id == 0 {
304            return Err(ControlFrameError::MissingRequestId);
305        }
306        let commands = [
307            self.get_config.is_some(),
308            self.watch_config.is_some(),
309            self.apply_config.is_some(),
310            self.refresh_inventory.is_some(),
311        ];
312        if commands.into_iter().filter(|present| *present).count() != 1 {
313            return Err(ControlFrameError::MissingControlCommand);
314        }
315        if let Some(request) = &self.get_config {
316            request.validate_frame()?;
317        }
318        if let Some(request) = &self.watch_config {
319            request.validate_frame()?;
320        }
321        if let Some(request) = &self.apply_config {
322            request.validate_frame()?;
323        }
324        if let Some(request) = &self.refresh_inventory {
325            request.validate_frame()?;
326        }
327        Ok(())
328    }
329}
330
331impl ValidateControlFrame for crate::proto::node::OwnerControlResponse {
332    fn validate_frame(&self) -> Result<(), ControlFrameError> {
333        if self.request_id == 0 {
334            return Err(ControlFrameError::MissingRequestId);
335        }
336        let results = [
337            self.get_config.is_some(),
338            self.watch_config.is_some(),
339            self.apply_config.is_some(),
340            self.refresh_inventory.is_some(),
341        ];
342        if results.into_iter().filter(|present| *present).count() != 1 {
343            return Err(ControlFrameError::MissingControlResult);
344        }
345        if let Some(response) = &self.get_config {
346            response.validate_frame()?;
347        }
348        if let Some(response) = &self.watch_config {
349            response.validate_frame()?;
350        }
351        if let Some(response) = &self.apply_config {
352            response.validate_frame()?;
353        }
354        if let Some(response) = &self.refresh_inventory {
355            response.validate_frame()?;
356        }
357        Ok(())
358    }
359}
360
361impl ValidateControlFrame for crate::proto::node::OwnerControlError {
362    fn validate_frame(&self) -> Result<(), ControlFrameError> {
363        if matches!(
364            crate::proto::node::OwnerControlErrorCode::try_from(self.code),
365            Err(_) | Ok(crate::proto::node::OwnerControlErrorCode::Unspecified)
366        ) {
367            return Err(ControlFrameError::InvalidOwnerControlErrorCode { got: self.code });
368        }
369        Ok(())
370    }
371}
372
373impl ValidateControlFrame for crate::proto::node::OwnerControlGetConfigRequest {
374    fn validate_frame(&self) -> Result<(), ControlFrameError> {
375        validate_endpoint_id_length(self.requester_node_id.len())?;
376        validate_endpoint_id_length(self.target_node_id.len())?;
377        Ok(())
378    }
379}
380
381impl ValidateControlFrame for crate::proto::node::OwnerControlGetConfigResponse {
382    fn validate_frame(&self) -> Result<(), ControlFrameError> {
383        self.snapshot
384            .as_ref()
385            .ok_or(ControlFrameError::MissingConfig)?
386            .validate_frame()
387    }
388}
389
390impl ValidateControlFrame for crate::proto::node::OwnerControlWatchConfigRequest {
391    fn validate_frame(&self) -> Result<(), ControlFrameError> {
392        validate_endpoint_id_length(self.requester_node_id.len())?;
393        validate_endpoint_id_length(self.target_node_id.len())?;
394        Ok(())
395    }
396}
397
398impl ValidateControlFrame for crate::proto::node::OwnerControlWatchConfigResponse {
399    fn validate_frame(&self) -> Result<(), ControlFrameError> {
400        let results = [
401            self.accepted.is_some(),
402            self.snapshot.is_some(),
403            self.update.is_some(),
404        ];
405        if results.into_iter().filter(|present| *present).count() != 1 {
406            return Err(ControlFrameError::MissingControlResult);
407        }
408        if let Some(accepted) = &self.accepted {
409            accepted.validate_frame()?;
410        }
411        if let Some(snapshot) = &self.snapshot {
412            snapshot.validate_frame()?;
413        }
414        if let Some(update) = &self.update {
415            update.validate_frame()?;
416        }
417        Ok(())
418    }
419}
420
421impl ValidateControlFrame for crate::proto::node::OwnerControlWatchAccepted {
422    fn validate_frame(&self) -> Result<(), ControlFrameError> {
423        validate_endpoint_id_length(self.target_node_id.len())?;
424        Ok(())
425    }
426}
427
428impl ValidateControlFrame for crate::proto::node::OwnerControlApplyConfigRequest {
429    fn validate_frame(&self) -> Result<(), ControlFrameError> {
430        validate_endpoint_id_length(self.requester_node_id.len())?;
431        validate_endpoint_id_length(self.target_node_id.len())?;
432        if self.config.is_none() {
433            return Err(ControlFrameError::MissingConfig);
434        }
435        Ok(())
436    }
437}
438
439impl ValidateControlFrame for crate::proto::node::OwnerControlApplyConfigResponse {
440    fn validate_frame(&self) -> Result<(), ControlFrameError> {
441        if self.success || !self.config_hash.is_empty() {
442            validate_config_hash_length(self.config_hash.len())?;
443        }
444        Ok(())
445    }
446}
447
448impl ValidateControlFrame for crate::proto::node::OwnerControlRefreshInventoryRequest {
449    fn validate_frame(&self) -> Result<(), ControlFrameError> {
450        validate_endpoint_id_length(self.requester_node_id.len())?;
451        validate_endpoint_id_length(self.target_node_id.len())?;
452        Ok(())
453    }
454}
455
456impl ValidateControlFrame for crate::proto::node::OwnerControlRefreshInventoryResponse {
457    fn validate_frame(&self) -> Result<(), ControlFrameError> {
458        self.snapshot
459            .as_ref()
460            .ok_or(ControlFrameError::MissingConfig)?
461            .validate_frame()
462    }
463}
464
465impl ValidateControlFrame for crate::proto::node::OwnerControlConfigSnapshot {
466    fn validate_frame(&self) -> Result<(), ControlFrameError> {
467        validate_endpoint_id_length(self.node_id.len())?;
468        validate_config_hash_length(self.config_hash.len())?;
469        if self.config.is_none() {
470            return Err(ControlFrameError::MissingConfig);
471        }
472        Ok(())
473    }
474}
475
476impl ValidateControlFrame for crate::proto::node::OwnerControlConfigUpdate {
477    fn validate_frame(&self) -> Result<(), ControlFrameError> {
478        validate_endpoint_id_length(self.node_id.len())?;
479        validate_config_hash_length(self.config_hash.len())?;
480        if self.config.is_none() {
481            return Err(ControlFrameError::MissingConfig);
482        }
483        Ok(())
484    }
485}
486
487impl ValidateControlFrame for crate::proto::node::MeshSubprotocolOpen {
488    fn validate_frame(&self) -> Result<(), ControlFrameError> {
489        if self.r#gen != NODE_PROTOCOL_GENERATION {
490            return Err(ControlFrameError::BadGeneration { got: self.r#gen });
491        }
492        if self.name.trim().is_empty() || self.major == 0 {
493            return Err(ControlFrameError::InvalidSubprotocol);
494        }
495        Ok(())
496    }
497}
498
499pub fn validate_peer_announcement(
500    pa: &crate::proto::node::PeerAnnouncement,
501) -> Result<(), ControlFrameError> {
502    if pa.endpoint_id.len() != 32 {
503        return Err(ControlFrameError::InvalidEndpointId {
504            got: pa.endpoint_id.len(),
505        });
506    }
507    if pa.role == crate::proto::node::NodeRole::Host as i32 && pa.http_port.is_none() {
508        return Err(ControlFrameError::MissingHttpPort);
509    }
510    for subprotocol in &pa.subprotocols {
511        if subprotocol.name.trim().is_empty() || subprotocol.major == 0 {
512            return Err(ControlFrameError::InvalidSubprotocol);
513        }
514    }
515    Ok(())
516}
517
518fn validate_endpoint_id_length(len: usize) -> Result<(), ControlFrameError> {
519    if len != 32 {
520        return Err(ControlFrameError::InvalidEndpointId { got: len });
521    }
522    Ok(())
523}
524
525fn validate_config_hash_length(len: usize) -> Result<(), ControlFrameError> {
526    if len != 32 {
527        return Err(ControlFrameError::InvalidConfigHashLength { got: len });
528    }
529    Ok(())
530}
531
532fn validate_public_key_length(len: usize) -> Result<(), ControlFrameError> {
533    if len != 32 {
534        return Err(ControlFrameError::InvalidPublicKeyLength { got: len });
535    }
536    Ok(())
537}
538
539pub fn protocol_from_alpn(alpn: &[u8]) -> ControlProtocol {
540    if alpn == ALPN_V0 {
541        ControlProtocol::JsonV0
542    } else {
543        ControlProtocol::ProtoV1
544    }
545}
546
547pub fn connection_protocol(conn: &Connection) -> ControlProtocol {
548    protocol_from_alpn(conn.alpn())
549}
550
551pub async fn connect_mesh(endpoint: &Endpoint, addr: EndpointAddr) -> Result<Connection> {
552    let opts = ConnectOptions::new().with_additional_alpns(vec![ALPN_V0.to_vec()]);
553    let connecting = endpoint.connect_with_opts(addr, ALPN_V1, opts).await?;
554    Ok(connecting.await?)
555}
556
557pub async fn write_len_prefixed(send: &mut iroh::endpoint::SendStream, body: &[u8]) -> Result<()> {
558    send.write_all(&(body.len() as u32).to_le_bytes()).await?;
559    send.write_all(body).await?;
560    Ok(())
561}
562
563pub async fn read_len_prefixed(recv: &mut iroh::endpoint::RecvStream) -> Result<Vec<u8>> {
564    let mut len_buf = [0u8; 4];
565    recv.read_exact(&mut len_buf).await?;
566    let len = u32::from_le_bytes(len_buf) as usize;
567    if len > MAX_CONTROL_FRAME_BYTES {
568        anyhow::bail!("control frame too large: {} bytes", len);
569    }
570    let mut buf = vec![0u8; len];
571    recv.read_exact(&mut buf).await?;
572    Ok(buf)
573}
574
575pub fn encode_control_frame(stream_type: u8, msg: &impl prost::Message) -> Vec<u8> {
576    let proto_bytes = msg.encode_to_vec();
577    let len = proto_bytes.len() as u32;
578    let mut buf = Vec::with_capacity(1 + 4 + proto_bytes.len());
579    buf.push(stream_type);
580    buf.extend_from_slice(&len.to_le_bytes());
581    buf.extend_from_slice(&proto_bytes);
582    buf
583}
584
585pub fn decode_control_frame<T: ValidateControlFrame>(
586    expected_stream_type: u8,
587    data: &[u8],
588) -> Result<T, ControlFrameError> {
589    const HEADER_LEN: usize = 5;
590    if data.len() < HEADER_LEN {
591        return Err(ControlFrameError::DecodeError(format!(
592            "frame too short: {} bytes (minimum {})",
593            data.len(),
594            HEADER_LEN
595        )));
596    }
597    let actual_type = data[0];
598    if actual_type != expected_stream_type {
599        return Err(ControlFrameError::WrongStreamType {
600            expected: expected_stream_type,
601            got: actual_type,
602        });
603    }
604    let len = u32::from_le_bytes(data[1..5].try_into().unwrap()) as usize;
605    if len > MAX_CONTROL_FRAME_BYTES {
606        return Err(ControlFrameError::OversizeFrame { size: len });
607    }
608    let proto_bytes = data.get(5..5 + len).ok_or_else(|| {
609        ControlFrameError::DecodeError(format!(
610            "frame truncated: header says {} bytes but only {} available",
611            len,
612            data.len().saturating_sub(5)
613        ))
614    })?;
615    let msg = T::decode(proto_bytes).map_err(|e| ControlFrameError::DecodeError(e.to_string()))?;
616    msg.validate_frame()?;
617    Ok(msg)
618}
619
620pub fn encode_owner_control_envelope(msg: &crate::proto::node::OwnerControlEnvelope) -> Vec<u8> {
621    msg.encode_to_vec()
622}
623
624pub fn decode_owner_control_envelope(
625    data: &[u8],
626) -> Result<crate::proto::node::OwnerControlEnvelope, ControlFrameError> {
627    let msg = crate::proto::node::OwnerControlEnvelope::decode(data)
628        .map_err(|e| ControlFrameError::DecodeError(e.to_string()))?;
629    msg.validate_frame()?;
630    Ok(msg)
631}
632
633pub fn owner_control_error_envelope(
634    code: crate::proto::node::OwnerControlErrorCode,
635    request_id: Option<u64>,
636    message: impl Into<String>,
637) -> crate::proto::node::OwnerControlEnvelope {
638    crate::proto::node::OwnerControlEnvelope {
639        r#gen: NODE_PROTOCOL_GENERATION,
640        handshake: None,
641        request: None,
642        response: None,
643        error: Some(crate::proto::node::OwnerControlError {
644            code: code as i32,
645            message: message.into(),
646            request_id,
647            current_revision: None,
648        }),
649    }
650}
651
652pub fn owner_control_rejection_envelope(
653    data: &[u8],
654    request_id: Option<u64>,
655    err: &ControlFrameError,
656) -> crate::proto::node::OwnerControlEnvelope {
657    let code = if matches!(err, ControlFrameError::MissingControlCommand) {
658        crate::proto::node::OwnerControlErrorCode::UnknownCommand
659    } else if serde_json::from_slice::<serde_json::Value>(data).is_ok() {
660        crate::proto::node::OwnerControlErrorCode::LegacyJsonUnsupported
661    } else {
662        crate::proto::node::OwnerControlErrorCode::BadRequest
663    };
664    owner_control_error_envelope(code, request_id, err.to_string())
665}
666
667#[cfg(test)]
668mod tests {
669    use super::*;
670    use crate::proto::node::{
671        ConfigApplyMode, NodeConfigSnapshot, NodeGpuConfig, NodeModelEntry,
672        OwnerControlApplyConfigRequest, OwnerControlApplyConfigResponse,
673        OwnerControlConfigSnapshot, OwnerControlConfigUpdate, OwnerControlEnvelope,
674        OwnerControlError, OwnerControlErrorCode, OwnerControlGetConfigRequest,
675        OwnerControlGetConfigResponse, OwnerControlHandshake, OwnerControlRefreshInventoryRequest,
676        OwnerControlRefreshInventoryResponse, OwnerControlRequest, OwnerControlResponse,
677        OwnerControlWatchAccepted, OwnerControlWatchConfigResponse, SignedNodeOwnership,
678    };
679
680    fn control_plane_test_config() -> NodeConfigSnapshot {
681        NodeConfigSnapshot {
682            version: 1,
683            gpu: Some(NodeGpuConfig {
684                assignment: crate::proto::node::GpuAssignment::Auto as i32,
685            }),
686            models: vec![NodeModelEntry {
687                model: "Qwen3-8B".to_string(),
688                mmproj: None,
689                ctx_size: Some(8192),
690                gpu_id: None,
691                model_ref: None,
692                mmproj_ref: None,
693            }],
694            plugins: vec![],
695            config_toml: None,
696        }
697    }
698
699    fn control_plane_test_snapshot() -> OwnerControlConfigSnapshot {
700        OwnerControlConfigSnapshot {
701            node_id: vec![0x55; 32],
702            revision: 7,
703            config_hash: vec![0xA5; 32],
704            config: Some(control_plane_test_config()),
705            hostname: Some("node-01".to_string()),
706        }
707    }
708
709    fn control_plane_test_handshake() -> OwnerControlEnvelope {
710        OwnerControlEnvelope {
711            r#gen: NODE_PROTOCOL_GENERATION,
712            handshake: Some(OwnerControlHandshake {
713                ownership: Some(SignedNodeOwnership {
714                    version: 1,
715                    cert_id: "cert-1".to_string(),
716                    owner_id: "owner-1".to_string(),
717                    owner_sign_public_key: vec![0x11; 32],
718                    node_endpoint_id: vec![0x22; 32],
719                    issued_at_unix_ms: 1,
720                    expires_at_unix_ms: 2,
721                    node_label: Some("node-01".to_string()),
722                    hostname_hint: Some("node-01".to_string()),
723                    signature: vec![0x33; 64],
724                }),
725            }),
726            request: None,
727            response: None,
728            error: None,
729        }
730    }
731
732    #[test]
733    fn control_plane_messages_constants_are_stable() {
734        assert_eq!(ALPN_CONTROL_V1, b"mesh-llm-control/1");
735        assert_eq!(ALPN_V1, b"mesh-llm/1");
736        assert_eq!(ALPN_V0, b"mesh-llm/0");
737        assert_eq!(STREAM_CONFIG_SUBSCRIBE, 0x0b);
738        assert_eq!(STREAM_CONFIG_PUSH, 0x0c);
739        assert_eq!(STREAM_SUBPROTOCOL, 0x0d);
740    }
741
742    #[test]
743    fn control_plane_messages_roundtrip_commands_and_responses() {
744        let handshake = control_plane_test_handshake();
745        let decoded = decode_owner_control_envelope(&encode_owner_control_envelope(&handshake))
746            .expect("handshake must decode");
747        assert!(decoded.handshake.is_some());
748
749        let get_request = OwnerControlEnvelope {
750            r#gen: NODE_PROTOCOL_GENERATION,
751            handshake: None,
752            request: Some(OwnerControlRequest {
753                request_id: 10,
754                get_config: Some(OwnerControlGetConfigRequest {
755                    requester_node_id: vec![0x10; 32],
756                    target_node_id: vec![0x20; 32],
757                }),
758                watch_config: None,
759                apply_config: None,
760                refresh_inventory: None,
761            }),
762            response: None,
763            error: None,
764        };
765        let decoded = decode_owner_control_envelope(&encode_owner_control_envelope(&get_request))
766            .expect("get-config request must decode");
767        assert_eq!(decoded.request.unwrap().request_id, 10);
768
769        let watch_response = OwnerControlEnvelope {
770            r#gen: NODE_PROTOCOL_GENERATION,
771            handshake: None,
772            request: None,
773            response: Some(OwnerControlResponse {
774                request_id: 11,
775                get_config: None,
776                watch_config: Some(OwnerControlWatchConfigResponse {
777                    accepted: Some(OwnerControlWatchAccepted {
778                        target_node_id: vec![0x21; 32],
779                    }),
780                    snapshot: None,
781                    update: None,
782                }),
783                apply_config: None,
784                refresh_inventory: None,
785            }),
786            error: None,
787        };
788        decode_owner_control_envelope(&encode_owner_control_envelope(&watch_response))
789            .expect("watch-config response must decode");
790
791        let apply_request = OwnerControlEnvelope {
792            r#gen: NODE_PROTOCOL_GENERATION,
793            handshake: None,
794            request: Some(OwnerControlRequest {
795                request_id: 12,
796                get_config: None,
797                watch_config: None,
798                apply_config: Some(OwnerControlApplyConfigRequest {
799                    requester_node_id: vec![0x30; 32],
800                    target_node_id: vec![0x40; 32],
801                    expected_revision: 7,
802                    config: Some(control_plane_test_config()),
803                }),
804                refresh_inventory: None,
805            }),
806            response: None,
807            error: None,
808        };
809        decode_owner_control_envelope(&encode_owner_control_envelope(&apply_request))
810            .expect("apply-config request must decode");
811
812        let apply_response = OwnerControlEnvelope {
813            r#gen: NODE_PROTOCOL_GENERATION,
814            handshake: None,
815            request: None,
816            response: Some(OwnerControlResponse {
817                request_id: 12,
818                get_config: None,
819                watch_config: None,
820                apply_config: Some(OwnerControlApplyConfigResponse {
821                    success: true,
822                    current_revision: 8,
823                    config_hash: vec![0x99; 32],
824                    error: None,
825                    apply_mode: ConfigApplyMode::Live as i32,
826                }),
827                refresh_inventory: None,
828            }),
829            error: None,
830        };
831        decode_owner_control_envelope(&encode_owner_control_envelope(&apply_response))
832            .expect("apply-config response must decode");
833
834        let refresh_request = OwnerControlEnvelope {
835            r#gen: NODE_PROTOCOL_GENERATION,
836            handshake: None,
837            request: Some(OwnerControlRequest {
838                request_id: 13,
839                get_config: None,
840                watch_config: None,
841                apply_config: None,
842                refresh_inventory: Some(OwnerControlRefreshInventoryRequest {
843                    requester_node_id: vec![0x50; 32],
844                    target_node_id: vec![0x60; 32],
845                }),
846            }),
847            response: None,
848            error: None,
849        };
850        decode_owner_control_envelope(&encode_owner_control_envelope(&refresh_request))
851            .expect("refresh-inventory request must decode");
852
853        let refresh_response = OwnerControlEnvelope {
854            r#gen: NODE_PROTOCOL_GENERATION,
855            handshake: None,
856            request: None,
857            response: Some(OwnerControlResponse {
858                request_id: 13,
859                get_config: None,
860                watch_config: None,
861                apply_config: None,
862                refresh_inventory: Some(OwnerControlRefreshInventoryResponse {
863                    snapshot: Some(control_plane_test_snapshot()),
864                }),
865            }),
866            error: None,
867        };
868        decode_owner_control_envelope(&encode_owner_control_envelope(&refresh_response))
869            .expect("refresh-inventory response must decode");
870
871        let get_response = OwnerControlEnvelope {
872            r#gen: NODE_PROTOCOL_GENERATION,
873            handshake: None,
874            request: None,
875            response: Some(OwnerControlResponse {
876                request_id: 14,
877                get_config: Some(OwnerControlGetConfigResponse {
878                    snapshot: Some(control_plane_test_snapshot()),
879                }),
880                watch_config: None,
881                apply_config: None,
882                refresh_inventory: None,
883            }),
884            error: None,
885        };
886        decode_owner_control_envelope(&encode_owner_control_envelope(&get_response))
887            .expect("get-config response must decode");
888
889        let update_response = OwnerControlEnvelope {
890            r#gen: NODE_PROTOCOL_GENERATION,
891            handshake: None,
892            request: None,
893            response: Some(OwnerControlResponse {
894                request_id: 15,
895                get_config: None,
896                watch_config: Some(OwnerControlWatchConfigResponse {
897                    accepted: None,
898                    snapshot: None,
899                    update: Some(OwnerControlConfigUpdate {
900                        node_id: vec![0x55; 32],
901                        revision: 8,
902                        config_hash: vec![0x77; 32],
903                        config: Some(control_plane_test_config()),
904                    }),
905                }),
906                apply_config: None,
907                refresh_inventory: None,
908            }),
909            error: None,
910        };
911        decode_owner_control_envelope(&encode_owner_control_envelope(&update_response))
912            .expect("watch update response must decode");
913    }
914
915    #[test]
916    fn control_plane_messages_unknown_command_rejects_with_structured_error() {
917        let envelope = OwnerControlEnvelope {
918            r#gen: NODE_PROTOCOL_GENERATION,
919            handshake: None,
920            request: Some(OwnerControlRequest {
921                request_id: 42,
922                get_config: None,
923                watch_config: None,
924                apply_config: None,
925                refresh_inventory: None,
926            }),
927            response: None,
928            error: None,
929        };
930        let bytes = encode_owner_control_envelope(&envelope);
931        let err = decode_owner_control_envelope(&bytes)
932            .expect_err("missing command variant must be rejected");
933        assert!(matches!(err, ControlFrameError::MissingControlCommand));
934
935        let rejection = owner_control_rejection_envelope(&bytes, Some(42), &err);
936        let error = rejection
937            .error
938            .expect("structured rejection must carry an error");
939        assert_eq!(
940            crate::proto::node::OwnerControlErrorCode::try_from(error.code).unwrap(),
941            OwnerControlErrorCode::UnknownCommand
942        );
943        assert_eq!(error.request_id, Some(42));
944    }
945
946    #[test]
947    fn owner_control_handshake_empty_owner_id_uses_handshake_error() {
948        let mut envelope = control_plane_test_handshake();
949        envelope
950            .handshake
951            .as_mut()
952            .and_then(|handshake| handshake.ownership.as_mut())
953            .expect("test handshake must include ownership")
954            .owner_id = "   ".to_string();
955
956        let err = decode_owner_control_envelope(&encode_owner_control_envelope(&envelope))
957            .expect_err("handshake with blank owner_id must be rejected");
958        assert!(matches!(err, ControlFrameError::MissingControlOwnerId));
959        assert_eq!(err.to_string(), "owner control handshake missing owner_id");
960    }
961
962    #[test]
963    fn owner_control_error_rejects_invalid_error_code() {
964        for code in [OwnerControlErrorCode::Unspecified as i32, 9999] {
965            let err = OwnerControlError {
966                code,
967                message: "invalid".to_string(),
968                request_id: Some(1),
969                current_revision: None,
970            }
971            .validate_frame()
972            .expect_err("invalid owner-control error code must be rejected");
973            assert!(matches!(
974                err,
975                ControlFrameError::InvalidOwnerControlErrorCode { got } if got == code
976            ));
977            assert_eq!(
978                err.to_string(),
979                format!("invalid owner control error code: {code}")
980            );
981        }
982    }
983
984    #[test]
985    fn control_plane_messages_legacy_json_rejects_with_structured_error() {
986        let legacy_json = br#"{"owner_id":"legacy","command":"GetConfig"}"#;
987        let err = decode_owner_control_envelope(legacy_json)
988            .expect_err("legacy json must not decode on protobuf-only control plane");
989        let rejection = owner_control_rejection_envelope(legacy_json, Some(99), &err);
990        let error = rejection
991            .error
992            .expect("structured rejection must carry an error");
993        assert_eq!(
994            crate::proto::node::OwnerControlErrorCode::try_from(error.code).unwrap(),
995            OwnerControlErrorCode::LegacyJsonUnsupported
996        );
997        assert_eq!(error.request_id, Some(99));
998    }
999}
mesh_llm_protocol/protocol/mod.rs

mesh_llm_protocol/protocol/
mod.rs
