1#![forbid(unsafe_code)]
32
33use std::{error::Error, fmt, path::PathBuf};
34
35use serde::{Deserialize, Serialize};
36
37pub mod frame;
38pub mod manifest;
39pub mod session;
40
41pub use frame::{Frame, FrameBuildError};
42
43#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
45pub struct BindIdentity {
46 pub project_root: PathBuf,
47 pub harness: String,
48 pub session: String,
49}
50
51#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
53#[serde(tag = "kind", rename_all = "snake_case")]
54pub enum Principal {
55 Reserved { module_id: String },
57 Direct,
59 Unverified,
61}
62
63#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
76#[serde(tag = "kind", rename_all = "snake_case")]
77pub enum RouteTarget {
78 ToolProvider {
79 module_id: String,
80 },
81 ManagementSurface {
82 module_id: String,
83 },
84 InternalService {
85 module_id: String,
86 service_id: String,
87 },
88}
89
90pub const PROTOCOL_VERSION: u8 = 2;
92
93pub const MIN_SUPPORTED_VERSION: u8 = 2;
95
96pub const SUBC_MODULE_ID_ENV: &str = "SUBC_MODULE_ID";
99
100pub const SUBC_LAUNCH_NONCE_ENV: &str = "SUBC_LAUNCH_NONCE";
105
106pub const HEADER_LEN: usize = 21;
108
109pub const FROZEN_PREFIX_LEN: usize = 5;
113
114pub const MAX_FRAME_BODY_LEN: u32 = 64 * 1024 * 1024;
120
121#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
123pub struct ErrorBody {
124 pub code: String,
125 pub message: String,
126}
127
128#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
130pub struct ModuleHelloBody {
131 pub manifest: manifest::ModuleManifest,
132 pub protocol_ver: u8,
133 #[serde(default)]
134 pub control_ops: Option<Vec<String>>,
135 #[serde(default, skip_serializing_if = "Option::is_none")]
143 pub launch_nonce: Option<String>,
144}
145
146#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
148pub struct ModuleHelloAckBody {
149 pub negotiated_ver: u8,
150 pub subc_ops: Vec<String>,
151 pub subc_capabilities: Vec<String>,
152 #[serde(default, skip_serializing_if = "Option::is_none")]
159 pub storage: Option<serde_json::Value>,
160}
161
162#[derive(Debug, Clone, Copy, PartialEq, Eq)]
167#[repr(u8)]
168pub enum FrameType {
169 Request = 0,
170 Response = 1,
171 Push = 2,
172 StreamData = 3,
173 StreamEnd = 4,
174 Error = 5,
175 Cancel = 6,
176 Ping = 7,
177 Pong = 8,
178 Hello = 9,
179 HelloAck = 10,
180 Goodbye = 11,
181}
182
183impl FrameType {
184 pub fn from_u8(b: u8) -> Option<Self> {
186 Some(match b {
187 0 => Self::Request,
188 1 => Self::Response,
189 2 => Self::Push,
190 3 => Self::StreamData,
191 4 => Self::StreamEnd,
192 5 => Self::Error,
193 6 => Self::Cancel,
194 7 => Self::Ping,
195 8 => Self::Pong,
196 9 => Self::Hello,
197 10 => Self::HelloAck,
198 11 => Self::Goodbye,
199 _ => return None,
200 })
201 }
202
203 pub fn is_pure_header(self) -> bool {
204 matches!(self, Self::Cancel | Self::Ping | Self::Pong | Self::Goodbye)
205 }
206}
207
208#[derive(Debug, Clone, Copy, PartialEq, Eq)]
211#[repr(u8)]
212pub enum Priority {
213 Passive = 0,
214 Interactive = 1,
215 Background = 2,
216}
217
218impl Priority {
219 fn from_bits(bits: u8) -> Option<Self> {
220 Some(match bits {
221 0 => Self::Passive,
222 1 => Self::Interactive,
223 2 => Self::Background,
224 _ => return None,
225 })
226 }
227}
228
229#[derive(Debug, Clone, Copy, PartialEq, Eq)]
231#[repr(u8)]
232pub enum AdmissionClass {
233 Normal = 0,
234 Expedite = 1,
235 Sheddable = 2,
236}
237
238impl AdmissionClass {
239 fn from_bits(bits: u8) -> Option<Self> {
240 Some(match bits {
241 0 => Self::Normal,
242 1 => Self::Expedite,
243 2 => Self::Sheddable,
244 _ => return None,
245 })
246 }
247}
248
249const FLAG_BINARY: u8 = 0b0000_0001; const FLAG_PRIORITY_MASK: u8 = 0b0000_0110; const FLAG_PRIORITY_SHIFT: u8 = 1;
252const FLAG_LAST: u8 = 0b0000_1000; const FLAG_ADMISSION_MASK: u8 = 0b0011_0000; const FLAG_ADMISSION_SHIFT: u8 = 4;
255const FLAG_RESERVED_MASK: u8 = 0b1100_0000; #[derive(Debug, Clone, Copy, PartialEq, Eq)]
259pub struct Flags(pub u8);
260
261impl Flags {
262 pub fn new(binary: bool, priority: Priority, last: bool) -> Self {
264 let mut b = 0u8;
265 if binary {
266 b |= FLAG_BINARY;
267 }
268 b |= (priority as u8) << FLAG_PRIORITY_SHIFT;
269 if last {
270 b |= FLAG_LAST;
271 }
272 Flags(b)
273 }
274
275 pub fn with_admission_class(mut self, admission_class: AdmissionClass) -> Self {
277 self.0 =
278 (self.0 & !FLAG_ADMISSION_MASK) | ((admission_class as u8) << FLAG_ADMISSION_SHIFT);
279 self
280 }
281
282 pub fn is_binary(self) -> bool {
284 self.0 & FLAG_BINARY != 0
285 }
286
287 pub fn is_last(self) -> bool {
289 self.0 & FLAG_LAST != 0
290 }
291
292 pub fn priority(self) -> Option<Priority> {
294 Priority::from_bits((self.0 & FLAG_PRIORITY_MASK) >> FLAG_PRIORITY_SHIFT)
295 }
296
297 pub fn admission_class(self) -> Option<AdmissionClass> {
299 AdmissionClass::from_bits((self.0 & FLAG_ADMISSION_MASK) >> FLAG_ADMISSION_SHIFT)
300 }
301
302 pub fn has_reserved_bits(self) -> bool {
304 self.0 & FLAG_RESERVED_MASK != 0
305 }
306}
307
308#[derive(Debug, Clone, Copy, PartialEq, Eq)]
310pub struct EnvelopeHeader {
311 pub len: u32,
313 pub ver: u8,
315 pub ty: FrameType,
317 pub flags: Flags,
319 pub channel: u16,
321 pub epoch: u32,
323 pub corr: u64,
325}
326
327impl EnvelopeHeader {
328 pub fn encode(&self) -> [u8; HEADER_LEN] {
330 let mut buf = [0u8; HEADER_LEN];
331 buf[0..4].copy_from_slice(&self.len.to_le_bytes());
332 buf[4] = self.ver;
333 buf[5] = self.ty as u8;
334 buf[6] = self.flags.0;
335 buf[7..9].copy_from_slice(&self.channel.to_le_bytes());
336 buf[9..13].copy_from_slice(&self.epoch.to_le_bytes());
337 buf[13..21].copy_from_slice(&self.corr.to_le_bytes());
338 buf
339 }
340}
341
342#[derive(Debug, Clone, Copy, PartialEq, Eq)]
344pub enum DecodeError {
345 TooShortForPrefix { have: usize },
347 UnsupportedVersion { ver: u8 },
349 TooShortForHeader { have: usize, need: usize },
351 UnknownFrameType { byte: u8 },
353 ReservedFlagBits { flags: u8 },
355 ReservedPriorityBits { flags: u8 },
357 ReservedAdmissionClass { flags: u8 },
359 SheddableIllegalFrameType { ty: FrameType, flags: u8 },
361 NonzeroEpochOnControlChannel { epoch: u32 },
363 PureHeaderFrameWithBody { ty: FrameType, len: u32 },
365}
366
367impl fmt::Display for DecodeError {
368 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
369 match self {
370 Self::TooShortForPrefix { have } => {
371 write!(f, "header shorter than frozen prefix: have {have} bytes")
372 }
373 Self::UnsupportedVersion { ver } => write!(f, "unsupported envelope version {ver}"),
374 Self::TooShortForHeader { have, need } => {
375 write!(
376 f,
377 "header too short for version: have {have} bytes, need {need}"
378 )
379 }
380 Self::UnknownFrameType { byte } => write!(f, "unknown frame type byte {byte}"),
381 Self::ReservedFlagBits { flags } => {
382 write!(f, "reserved flag bits set in flags 0b{flags:08b}")
383 }
384 Self::ReservedPriorityBits { flags } => {
385 write!(f, "reserved priority bits set in flags 0b{flags:08b}")
386 }
387 Self::ReservedAdmissionClass { flags } => {
388 write!(f, "reserved admission class set in flags 0b{flags:08b}")
389 }
390 Self::SheddableIllegalFrameType { ty, flags } => write!(
391 f,
392 "SHEDDABLE admission class is illegal on {ty:?} in flags 0b{flags:08b}"
393 ),
394 Self::NonzeroEpochOnControlChannel { epoch } => {
395 write!(f, "control channel carried nonzero epoch {epoch}")
396 }
397 Self::PureHeaderFrameWithBody { ty, len } => {
398 write!(
399 f,
400 "pure-header frame {ty:?} declared non-zero body length {len}"
401 )
402 }
403 }
404 }
405}
406
407impl Error for DecodeError {}
408
409fn header_len_for_version(ver: u8) -> Option<usize> {
412 match ver {
413 PROTOCOL_VERSION => Some(HEADER_LEN),
414 _ => None,
415 }
416}
417
418pub fn decode_header(bytes: &[u8]) -> Result<EnvelopeHeader, DecodeError> {
426 if bytes.len() < FROZEN_PREFIX_LEN {
427 return Err(DecodeError::TooShortForPrefix { have: bytes.len() });
428 }
429 let ver = bytes[4];
430 let need = header_len_for_version(ver).ok_or(DecodeError::UnsupportedVersion { ver })?;
431 if bytes.len() < need {
432 return Err(DecodeError::TooShortForHeader {
433 have: bytes.len(),
434 need,
435 });
436 }
437
438 let len = u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
439 let ty =
440 FrameType::from_u8(bytes[5]).ok_or(DecodeError::UnknownFrameType { byte: bytes[5] })?;
441 let flags = Flags(bytes[6]);
442 if flags.has_reserved_bits() {
443 return Err(DecodeError::ReservedFlagBits { flags: bytes[6] });
444 }
445 if flags.priority().is_none() {
446 return Err(DecodeError::ReservedPriorityBits { flags: bytes[6] });
447 }
448 let admission_class = flags
449 .admission_class()
450 .ok_or(DecodeError::ReservedAdmissionClass { flags: bytes[6] })?;
451 if admission_class == AdmissionClass::Sheddable
452 && !matches!(ty, FrameType::Push | FrameType::StreamData)
453 {
454 return Err(DecodeError::SheddableIllegalFrameType {
455 ty,
456 flags: bytes[6],
457 });
458 }
459 if ty.is_pure_header() && len != 0 {
460 return Err(DecodeError::PureHeaderFrameWithBody { ty, len });
461 }
462 let channel = u16::from_le_bytes([bytes[7], bytes[8]]);
463 let epoch = u32::from_le_bytes([bytes[9], bytes[10], bytes[11], bytes[12]]);
464 if channel == 0 && epoch != 0 {
465 return Err(DecodeError::NonzeroEpochOnControlChannel { epoch });
466 }
467 let corr = u64::from_le_bytes([
468 bytes[13], bytes[14], bytes[15], bytes[16], bytes[17], bytes[18], bytes[19], bytes[20],
469 ]);
470
471 Ok(EnvelopeHeader {
472 len,
473 ver,
474 ty,
475 flags,
476 channel,
477 epoch,
478 corr,
479 })
480}
481
482#[cfg(test)]
483mod tests {
484 use super::*;
485
486 fn hdr(len: u32, ty: FrameType, flags: Flags, channel: u16, corr: u64) -> EnvelopeHeader {
487 hdr_with_epoch(len, ty, flags, channel, u32::from(channel != 0), corr)
488 }
489
490 fn hdr_with_epoch(
491 len: u32,
492 ty: FrameType,
493 flags: Flags,
494 channel: u16,
495 epoch: u32,
496 corr: u64,
497 ) -> EnvelopeHeader {
498 EnvelopeHeader {
499 len,
500 ver: PROTOCOL_VERSION,
501 ty,
502 flags,
503 channel,
504 epoch,
505 corr,
506 }
507 }
508
509 #[test]
510 fn bind_identity_round_trips_json() {
511 let identity = BindIdentity {
512 project_root: PathBuf::from("/tmp/project"),
513 harness: "opencode".to_string(),
514 session: "session-1".to_string(),
515 };
516
517 let encoded = serde_json::to_vec(&identity).unwrap();
518 let decoded: BindIdentity = serde_json::from_slice(&encoded).unwrap();
519
520 assert_eq!(decoded, identity);
521 }
522
523 #[test]
524 fn route_target_variants_round_trip_json() {
525 let targets = [
526 RouteTarget::ToolProvider {
527 module_id: "aft".to_string(),
528 },
529 RouteTarget::ManagementSurface {
530 module_id: "memory".to_string(),
531 },
532 RouteTarget::InternalService {
533 module_id: "bus".to_string(),
534 service_id: "dm".to_string(),
535 },
536 ];
537
538 for target in targets {
539 let encoded = serde_json::to_vec(&target).unwrap();
540 let decoded: RouteTarget = serde_json::from_slice(&encoded).unwrap();
541 assert_eq!(decoded, target);
542 }
543 }
544
545 #[test]
546 fn error_body_round_trips_json() {
547 let body = ErrorBody {
548 code: "config_divergence".to_string(),
549 message: "active config differs".to_string(),
550 };
551
552 let encoded = serde_json::to_vec(&body).unwrap();
553 let decoded: ErrorBody = serde_json::from_slice(&encoded).unwrap();
554
555 assert_eq!(decoded, body);
556 }
557
558 #[test]
559 fn round_trip_request() {
560 let h = hdr(
561 1234,
562 FrameType::Request,
563 Flags::new(false, Priority::Interactive, false),
564 42,
565 0xDEAD_BEEF_0000_0001,
566 );
567 let decoded = decode_header(&h.encode()).unwrap();
568 assert_eq!(h, decoded);
569 }
570
571 #[test]
572 fn round_trip_all_frame_types() {
573 for b in 0u8..=11 {
574 let ty = FrameType::from_u8(b).unwrap();
575 let h = hdr(0, ty, Flags::new(false, Priority::Passive, false), 0, 0);
576 assert_eq!(decode_header(&h.encode()).unwrap().ty, ty);
577 }
578 }
579
580 #[test]
581 fn pure_header_frame_has_zero_len() {
582 let h = hdr(
584 0,
585 FrameType::Cancel,
586 Flags::new(false, Priority::Passive, false),
587 7,
588 99,
589 );
590 let d = decode_header(&h.encode()).unwrap();
591 assert_eq!(d.len, 0);
592 assert_eq!(d.corr, 99);
593 }
594
595 #[test]
596 fn flags_round_trip() {
597 let f = Flags::new(true, Priority::Background, true)
598 .with_admission_class(AdmissionClass::Expedite);
599 assert!(f.is_binary());
600 assert!(f.is_last());
601 assert_eq!(f.priority(), Some(Priority::Background));
602 assert_eq!(f.admission_class(), Some(AdmissionClass::Expedite));
603 let h = hdr(8, FrameType::StreamData, f, 1, 1);
604 assert_eq!(decode_header(&h.encode()).unwrap().flags, f);
605 }
606
607 #[test]
608 fn little_endian_and_frozen_prefix_layout() {
609 let h = hdr_with_epoch(
610 0x0403_0201,
611 FrameType::Request,
612 Flags(0),
613 0x0605,
614 0x0a09_0807,
615 0x1211_100f_0e0d_0c0b,
616 );
617 let buf = h.encode();
618 assert_eq!(&buf[0..4], &[1, 2, 3, 4]);
619 assert_eq!(buf[4], PROTOCOL_VERSION);
620 assert_eq!(&buf[7..9], &[5, 6]);
621 assert_eq!(&buf[9..13], &[7, 8, 9, 10]);
622 assert_eq!(&buf[13..21], &[11, 12, 13, 14, 15, 16, 17, 18]);
623 assert_eq!(buf.len(), HEADER_LEN);
624 }
625
626 #[test]
627 fn reject_too_short_for_prefix() {
628 assert_eq!(
629 decode_header(&[0, 0, 0, 0]),
630 Err(DecodeError::TooShortForPrefix { have: 4 })
631 );
632 }
633
634 #[test]
635 fn reject_too_short_for_header() {
636 let mut b = [0u8; 10];
638 b[4] = PROTOCOL_VERSION;
639 assert_eq!(
640 decode_header(&b),
641 Err(DecodeError::TooShortForHeader {
642 have: 10,
643 need: HEADER_LEN
644 })
645 );
646 }
647
648 #[test]
649 fn reject_unsupported_version() {
650 let mut b = [0u8; HEADER_LEN];
651 b[4] = 1;
652 assert_eq!(
653 decode_header(&b),
654 Err(DecodeError::UnsupportedVersion { ver: 1 })
655 );
656 }
657
658 #[test]
659 fn reject_unknown_frame_type() {
660 let mut b = [0u8; HEADER_LEN];
661 b[4] = PROTOCOL_VERSION;
662 b[5] = 99;
663 assert_eq!(
664 decode_header(&b),
665 Err(DecodeError::UnknownFrameType { byte: 99 })
666 );
667 }
668
669 #[test]
670 fn reject_reserved_flag_bits() {
671 let mut b = [0u8; HEADER_LEN];
672 b[4] = PROTOCOL_VERSION;
673 b[5] = FrameType::Request as u8;
674 b[6] = 0b1000_0000; assert_eq!(
676 decode_header(&b),
677 Err(DecodeError::ReservedFlagBits { flags: 0b1000_0000 })
678 );
679 }
680
681 #[test]
682 fn reject_reserved_priority_bits() {
683 let mut b = [0u8; HEADER_LEN];
684 b[4] = PROTOCOL_VERSION;
685 b[5] = FrameType::Request as u8;
686 b[6] = 0b0000_0110; assert_eq!(
688 decode_header(&b),
689 Err(DecodeError::ReservedPriorityBits { flags: 0b0000_0110 })
690 );
691 }
692
693 #[test]
694 fn reject_pure_header_frame_with_body_len() {
695 let h = hdr(
696 1,
697 FrameType::Ping,
698 Flags::new(false, Priority::Passive, false),
699 0,
700 1,
701 );
702 assert_eq!(
703 decode_header(&h.encode()),
704 Err(DecodeError::PureHeaderFrameWithBody {
705 ty: FrameType::Ping,
706 len: 1
707 })
708 );
709 }
710
711 #[test]
712 fn epoch_boundaries_round_trip() {
713 for (channel, epoch) in [(0, 0), (1, 1), (u16::MAX, u32::MAX)] {
714 let h = hdr_with_epoch(
715 0,
716 FrameType::Request,
717 Flags::new(false, Priority::Passive, false),
718 channel,
719 epoch,
720 9,
721 );
722 assert_eq!(decode_header(&h.encode()).unwrap(), h);
723 }
724 }
725
726 #[test]
727 fn admission_classes_accept_three_values_and_reject_reserved_value() {
728 for (ty, admission_class) in [
729 (FrameType::Request, AdmissionClass::Normal),
730 (FrameType::Request, AdmissionClass::Expedite),
731 (FrameType::Push, AdmissionClass::Sheddable),
732 (FrameType::StreamData, AdmissionClass::Sheddable),
733 ] {
734 let flags = Flags::new(false, Priority::Interactive, false)
735 .with_admission_class(admission_class);
736 let h = hdr(0, ty, flags, 1, 2);
737 assert_eq!(decode_header(&h.encode()).unwrap().flags, flags);
738 }
739
740 let mut h = hdr(
741 0,
742 FrameType::Push,
743 Flags::new(false, Priority::Passive, false),
744 1,
745 2,
746 )
747 .encode();
748 h[6] |= 0b0011_0000;
749 assert_eq!(
750 decode_header(&h),
751 Err(DecodeError::ReservedAdmissionClass { flags: h[6] })
752 );
753 }
754
755 #[test]
756 fn sheddable_rejected_on_every_illegal_frame_type() {
757 let flags = Flags::new(false, Priority::Passive, false)
758 .with_admission_class(AdmissionClass::Sheddable);
759 for ty in [
760 FrameType::Request,
761 FrameType::Response,
762 FrameType::StreamEnd,
763 FrameType::Error,
764 FrameType::Cancel,
765 FrameType::Ping,
766 FrameType::Pong,
767 FrameType::Hello,
768 FrameType::HelloAck,
769 FrameType::Goodbye,
770 ] {
771 let h = hdr(0, ty, flags, 1, 2);
772 assert_eq!(
773 decode_header(&h.encode()),
774 Err(DecodeError::SheddableIllegalFrameType { ty, flags: flags.0 })
775 );
776 }
777 }
778
779 #[test]
780 fn nonzero_epoch_on_control_channel_is_rejected() {
781 let h = hdr_with_epoch(
782 0,
783 FrameType::Request,
784 Flags::new(false, Priority::Passive, false),
785 0,
786 u32::MAX,
787 2,
788 );
789 assert_eq!(
790 decode_header(&h.encode()),
791 Err(DecodeError::NonzeroEpochOnControlChannel { epoch: u32::MAX })
792 );
793 }
794}