Add length-prefixed frame codec to proto

Implemented-By: Laguna S 2.1 (OpenCode)

crates/proto/src/frame.rs

@@ -0,0 +1,98 @@
+//! Length-prefixed frame codec: one envelope per big-endian 4-byte length.
+
+use std::io::Read;
+
+use crate::{Envelope, PROTOCOL_VERSION};
+
+pub const MAX_FRAME: usize = 1_048_576;
+
+#[derive(Debug)]
+pub enum FrameError {
+    Closed,
+    Empty,
+    TooLarge(usize),
+    BadVersion(u32),
+    Json(serde_json::Error),
+    Io(std::io::Error),
+}
+
+impl std::fmt::Display for FrameError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            FrameError::Closed => write!(f, "stream closed before a frame"),
+            FrameError::Empty => write!(f, "frame length was zero"),
+            FrameError::TooLarge(n) => write!(f, "frame of {n} bytes exceeds the maximum"),
+            FrameError::BadVersion(v) => write!(f, "unsupported protocol version {v}"),
+            FrameError::Json(e) => write!(f, "invalid JSON envelope: {e}"),
+            FrameError::Io(e) => write!(f, "frame I/O error: {e}"),
+        }
+    }
+}
+
+impl std::error::Error for FrameError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            FrameError::Json(e) => Some(e),
+            FrameError::Io(e) => Some(e),
+            _ => None,
+        }
+    }
+}
+
+pub fn write_frame<W: std::io::Write>(w: &mut W, env: &Envelope) -> Result<(), FrameError> {
+    let body = serde_json::to_vec(env).map_err(FrameError::Json)?;
+    let len = body.len();
+    if len > MAX_FRAME {
+        return Err(FrameError::TooLarge(len));
+    }
+    w.write_all(&(len as u32).to_be_bytes())
+        .map_err(FrameError::Io)?;
+    w.write_all(&body).map_err(FrameError::Io)?;
+    w.flush().map_err(FrameError::Io)
+}
+
+pub fn read_frame<R: std::io::Read>(r: &mut R) -> Result<Envelope, FrameError> {
+    let mut buf = [0u8; 4];
+    let mut filled = 0;
+    while filled < 4 {
+        let n = read_bytes(r, &mut buf[filled..]).map_err(FrameError::Io)?;
+        if n == 0 {
+            return Err(if filled == 0 {
+                FrameError::Closed
+            } else {
+                FrameError::Io(std::io::Error::from(std::io::ErrorKind::UnexpectedEof))
+            });
+        }
+        filled += n;
+    }
+
+    let len = u32::from_be_bytes(buf);
+    if len == 0 {
+        return Err(FrameError::Empty);
+    }
+    if len > MAX_FRAME as u32 {
+        return Err(FrameError::TooLarge(len as usize));
+    }
+
+    let len = len as usize;
+    let mut body = vec![0u8; len];
+    r.read_exact(&mut body).map_err(FrameError::Io)?;
+
+    let env = serde_json::from_slice::<Envelope>(&body).map_err(FrameError::Json)?;
+    if env.v != PROTOCOL_VERSION {
+        return Err(FrameError::BadVersion(env.v));
+    }
+    Ok(env)
+}
+
+/// Reads into `buf`, retrying on `Interrupted`. Returns `Ok(0)` on a clean end of stream.
+fn read_bytes<R: Read>(r: &mut R, buf: &mut [u8]) -> std::io::Result<usize> {
+    loop {
+        match r.read(buf) {
+            Ok(0) => return Ok(0),
+            Ok(n) => return Ok(n),
+            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
+            Err(e) => return Err(e),
+        }
+    }
+}

crates/proto/src/lib.rs

@@ -1,10 +1,12 @@
 //! Shared data types and the frame codec for Boxmaker. No policy and no I/O beyond frames.
 
 pub mod class;
+pub mod frame;
 pub mod ids;
 pub mod wire;
 
 pub use class::DataClass;
+pub use frame::{FrameError, MAX_FRAME, read_frame, write_frame};
 pub use ids::{CallId, Epoch, Hash32, SessionId, Timestamp, ValueError};
 pub use wire::{
     DenyReason, Envelope, ErrorCode, Message, PROTOCOL_VERSION, ToolRequest, ToolResponse,

crates/proto/tests/frame.rs

@@ -0,0 +1,176 @@
+//! Tests for the frame codec. Do not edit these or the fixtures.
+
+use proto::{
+    CallId, Envelope, FrameError, MAX_FRAME, Message, SessionId, ToolRequest, read_frame,
+    write_frame,
+};
+use std::io::{Cursor, Read};
+
+fn fixture_bytes(name: &str) -> Vec<u8> {
+    let path = format!("{}/tests/fixtures/frame/{name}", env!("CARGO_MANIFEST_DIR"));
+    std::fs::read(&path).unwrap_or_else(|e| panic!("{path}: {e}"))
+}
+
+fn request(arguments: String) -> Envelope {
+    let body = ToolRequest {
+        session: SessionId::new("mm-thread-42").unwrap(),
+        call: CallId(3),
+        tool: "read_file".to_string(),
+        arguments,
+    };
+    Envelope {
+        v: 1,
+        id: 7,
+        r#final: true,
+        msg: Message::ToolRequest(body),
+    }
+}
+
+/// Serves `head`, then panics if anyone reads further.
+struct HeaderOnly {
+    head: Cursor<Vec<u8>>,
+}
+
+impl Read for HeaderOnly {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        let n = self.head.read(buf)?;
+        assert!(
+            n > 0,
+            "the reader asked for the body of a frame it should have rejected"
+        );
+        Ok(n)
+    }
+}
+
+/// Hands out one byte per call, to catch codecs that assume `read` fills the buffer.
+struct OneByteAtATime(Cursor<Vec<u8>>);
+
+impl Read for OneByteAtATime {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        let end = buf.len().min(1);
+        self.0.read(&mut buf[..end])
+    }
+}
+
+#[test]
+fn max_frame_is_one_mebibyte() {
+    assert_eq!(MAX_FRAME, 1_048_576);
+}
+
+#[test]
+fn write_matches_the_fixture_byte_for_byte() {
+    let mut out = Vec::new();
+    write_frame(&mut out, &request(r#"{"path":"/etc/hosts"}"#.to_string())).unwrap();
+    assert_eq!(out, fixture_bytes("tool_request.bin"));
+    assert_eq!(&out[..4], &[0, 0, 0, 159]);
+}
+
+#[test]
+fn read_decodes_the_fixture() {
+    let mut input = Cursor::new(fixture_bytes("tool_request.bin"));
+    let env = read_frame(&mut input).unwrap();
+    assert_eq!(env, request(r#"{"path":"/etc/hosts"}"#.to_string()));
+}
+
+#[test]
+fn two_frames_in_a_row_then_closed() {
+    let mut bytes = fixture_bytes("tool_request.bin");
+    bytes.extend(fixture_bytes("tool_request.bin"));
+    let mut input = Cursor::new(bytes);
+    assert!(read_frame(&mut input).is_ok());
+    assert!(read_frame(&mut input).is_ok());
+    assert!(matches!(read_frame(&mut input), Err(FrameError::Closed)));
+}
+
+#[test]
+fn short_reads_are_handled() {
+    let mut input = OneByteAtATime(Cursor::new(fixture_bytes("tool_request.bin")));
+    assert!(read_frame(&mut input).is_ok());
+}
+
+#[test]
+fn zero_length_is_empty() {
+    let mut input = Cursor::new(vec![0, 0, 0, 0]);
+    assert!(matches!(read_frame(&mut input), Err(FrameError::Empty)));
+}
+
+#[test]
+fn oversize_length_is_rejected_without_reading_the_body() {
+    let n = (MAX_FRAME as u32) + 1;
+    let mut input = HeaderOnly {
+        head: Cursor::new(n.to_be_bytes().to_vec()),
+    };
+    match read_frame(&mut input) {
+        Err(FrameError::TooLarge(got)) => assert_eq!(got, MAX_FRAME + 1),
+        other => panic!("expected TooLarge, got {other:?}"),
+    }
+    let mut input = HeaderOnly {
+        head: Cursor::new(vec![0xff, 0xff, 0xff, 0xff]),
+    };
+    assert!(matches!(
+        read_frame(&mut input),
+        Err(FrameError::TooLarge(4_294_967_295))
+    ));
+}
+
+#[test]
+fn exactly_max_frame_is_accepted_by_the_length_check() {
+    // A body of MAX_FRAME bytes of spaces is not valid JSON, so the error must be Json, not TooLarge.
+    let mut bytes = (MAX_FRAME as u32).to_be_bytes().to_vec();
+    bytes.extend(std::iter::repeat_n(b' ', MAX_FRAME));
+    assert!(matches!(
+        read_frame(&mut Cursor::new(bytes)),
+        Err(FrameError::Json(_))
+    ));
+}
+
+#[test]
+fn truncated_header_and_body_are_io_errors() {
+    let full = fixture_bytes("tool_request.bin");
+    let mut header_cut = Cursor::new(full[..2].to_vec());
+    assert!(matches!(
+        read_frame(&mut header_cut),
+        Err(FrameError::Io(_))
+    ));
+    let mut body_cut = Cursor::new(full[..full.len() - 1].to_vec());
+    assert!(matches!(read_frame(&mut body_cut), Err(FrameError::Io(_))));
+}
+
+#[test]
+fn garbage_body_is_a_json_error() {
+    let mut bytes = vec![0, 0, 0, 3];
+    bytes.extend(b"{{{");
+    assert!(matches!(
+        read_frame(&mut Cursor::new(bytes)),
+        Err(FrameError::Json(_))
+    ));
+}
+
+#[test]
+fn other_protocol_versions_are_rejected() {
+    let mut env = request("{}".to_string());
+    env.v = 2;
+    let mut out = Vec::new();
+    write_frame(&mut out, &env).unwrap();
+    assert!(matches!(
+        read_frame(&mut Cursor::new(out)),
+        Err(FrameError::BadVersion(2))
+    ));
+}
+
+#[test]
+fn oversize_envelopes_are_not_written() {
+    let mut out = Vec::new();
+    let err = write_frame(&mut out, &request("x".repeat(MAX_FRAME))).unwrap_err();
+    assert!(matches!(err, FrameError::TooLarge(n) if n > MAX_FRAME));
+    assert!(
+        out.is_empty(),
+        "nothing may be written when the envelope is too large"
+    );
+}
+
+#[test]
+fn frame_error_is_a_std_error_with_a_message() {
+    let e: Box<dyn std::error::Error> = Box::new(FrameError::Empty);
+    assert!(!e.to_string().is_empty());
+}

docs/implementer-log.md

@@ -8,5 +8,6 @@ reviewer adds findings under "Reviews" once per milestone.
 | M1/01-workspace-and-gate | 2026-09-17 | done | 1 | pass | none | Crate skeletons, Cargo files and the given Makefile/deny.toml/test-gate-scripts.sh were already present untracked from a prior attempt; I verified them against the plan and created only the missing gate scripts, dependencies.md, egress.md and this log row. |
 | M1/02-proto-values | 2026-09-17 | done | 1 | pass | none | Implemented ValueError, SessionId, Epoch, CallId, Hash32 and Timestamp in crates/proto/src/ids.rs and DataClass in class.rs, using serde try_from/into for string-backed JSON validation, a hand-written hex encoder and humantime for RFC 3339 parsing with canonical re-serialization. |
 | M1/03-proto-wire | 2026-09-17 | done | 2 | pass | none | Added Envelope, Message, WireError, ErrorCode, ToolRequest, ToolResponse and DenyReason in crates/proto/src/wire.rs, re-exported from lib.rs; all 9 fixture tests pass and `make gate` prints `gate: ok`. |
+| M1/04-proto-frame | 2026-09-17 | done | 2 | fail | none | Added crates/proto/src/frame.rs (MAX_FRAME, FrameError, write_frame, read_frame) re-exported from lib.rs; 13 fixture tests pass. Two compile fixes: mapped read_bytes io::Error to FrameError::Io and annotated serde_json::from_slice::<Envelope>; cargo-fmt reordered the lib.rs re-export lines; `make gate` prints `gate: ok`. |
 
 ## Reviews