Rename project from xserial to pipeview

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>

crates/pipeview-core/src/frame/cobs.rs

@@ -0,0 +1,315 @@
+use super::Framer;
+use tracing::{debug, warn};
+
+/// COBS (Consistent Overhead Byte Stuffing) framer.
+///
+/// Frames are delimited by `0x00`.  Everything between two `0x00` bytes
+/// is a COBS-encoded packet.  The framer accumulates until a `0x00` is
+/// seen, decodes the COBS data, and yields the original payload.
+///
+/// Reference: <https://en.wikipedia.org/wiki/Consistent_Overhead_Byte_Stuffing>
+#[derive(Debug, Clone)]
+pub struct CobsFramer {
+    buf: Vec<u8>,
+    max_frame: usize,
+}
+
+impl Default for CobsFramer {
+    fn default() -> Self {
+        Self {
+            buf: Vec::new(),
+            max_frame: 1024 * 1024,
+        }
+    }
+}
+
+impl CobsFramer {
+    pub fn new(max_frame: usize) -> Self {
+        Self {
+            buf: Vec::new(),
+            max_frame,
+        }
+    }
+}
+
+impl Framer for CobsFramer {
+    fn feed(&mut self, data: &[u8]) -> Vec<Vec<u8>> {
+        let mut frames = Vec::new();
+
+        for &byte in data {
+            if byte == 0x00 {
+                if !self.buf.is_empty() {
+                    let buf_len = self.buf.len();
+                    if let Some(decoded) = cobs_decode(&self.buf) {
+                        debug!(
+                            encoded_len = buf_len,
+                            decoded_len = decoded.len(),
+                            "COBS frame decoded"
+                        );
+                        frames.push(decoded);
+                    } else {
+                        warn!(len = buf_len, "corrupt COBS packet discarded");
+                    }
+                    self.buf.clear();
+                }
+                // Consecutive 0x00 bytes produce empty frames
+                // (skip them — no payload to decode)
+            } else {
+                if self.buf.len() < self.max_frame {
+                    self.buf.push(byte);
+                }
+                // Exceeding max_frame: drop silently (avoid OOM)
+            }
+        }
+
+        frames
+    }
+
+    fn flush(&mut self) -> Option<Vec<u8>> {
+        if self.buf.is_empty() {
+            None
+        } else {
+            let data = std::mem::take(&mut self.buf);
+            cobs_decode(&data).or(Some(data))
+        }
+    }
+
+    fn reset(&mut self) {
+        self.buf.clear();
+    }
+
+    fn pending_len(&self) -> usize {
+        self.buf.len()
+    }
+}
+
+/// Decode a COBS-encoded packet (without the trailing 0x00 delimiter).
+///
+/// Returns `None` if the encoded data is invalid (e.g. an overhead byte
+/// points past the end of the buffer).
+pub fn cobs_encode(payload: &[u8]) -> Vec<u8> {
+    if payload.is_empty() {
+        return vec![0x01];
+    }
+
+    let mut out = Vec::with_capacity(payload.len() + (payload.len() / 254) + 1);
+    let mut code_index = 0usize;
+    let mut code = 1u8;
+    out.push(0);
+
+    for &byte in payload {
+        if byte == 0 {
+            out[code_index] = code;
+            code_index = out.len();
+            out.push(0);
+            code = 1;
+        } else {
+            out.push(byte);
+            code = code.saturating_add(1);
+            if code == 0xFF {
+                out[code_index] = code;
+                code_index = out.len();
+                out.push(0);
+                code = 1;
+            }
+        }
+    }
+
+    out[code_index] = code;
+    out
+}
+
+pub fn cobs_decode(encoded: &[u8]) -> Option<Vec<u8>> {
+    if encoded.is_empty() {
+        return Some(Vec::new());
+    }
+
+    let mut out = Vec::with_capacity(encoded.len());
+    let mut pos = 0;
+
+    while pos < encoded.len() {
+        let code = encoded[pos] as usize;
+        if code == 0 {
+            // Invalid: overhead byte should never be 0
+            return None;
+        }
+        pos += 1;
+
+        if code > 1 {
+            let copy_start = pos;
+            let desired_end = pos + code - 1;
+            if desired_end > encoded.len() {
+                return None;
+            }
+            out.extend_from_slice(&encoded[copy_start..desired_end]);
+            pos = desired_end;
+        }
+
+        // If the code byte was < 0xFF, the next byte (if any) in the
+        // original stream was 0x00 → insert it.
+        if code < 0xFF && pos < encoded.len() {
+            out.push(0x00);
+        }
+    }
+
+    Some(out)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn cobs_roundtrip_with_zeros() {
+        let payload = [0x11, 0x00, 0x22, 0x33, 0x00, 0x44];
+        let encoded = cobs_encode(&payload);
+        let decoded = cobs_decode(&encoded).unwrap();
+        assert_eq!(decoded, payload);
+    }
+
+    // ── cobs_decode unit tests ──────────────────────────────────────
+
+    #[test]
+    fn cobs_decode_no_zeros() {
+        let decoded = cobs_decode(&[0x06, 0x68, 0x65, 0x6C, 0x6C, 0x6F]).unwrap();
+        assert_eq!(decoded, b"hello");
+    }
+
+    #[test]
+    fn cobs_decode_single_zero() {
+        let decoded = cobs_decode(&[0x03, 0x11, 0x22, 0x02, 0x33]).unwrap();
+        assert_eq!(decoded, vec![0x11, 0x22, 0x00, 0x33]);
+    }
+
+    #[test]
+    fn cobs_decode_leading_zero() {
+        let decoded = cobs_decode(&[0x01, 0x01, 0x01, 0x01]).unwrap();
+        assert_eq!(decoded, vec![0x00, 0x00, 0x00]);
+    }
+
+    #[test]
+    fn cobs_decode_all_zeros() {
+        // Input: [0x00, 0x00, 0x00] → encoded: [0x01, 0x01, 0x01, 0x01]
+        let decoded = cobs_decode(&[0x01, 0x01, 0x01, 0x01]).unwrap();
+        assert_eq!(decoded, vec![0x00, 0x00, 0x00]);
+    }
+
+    #[test]
+    fn cobs_decode_empty() {
+        let decoded = cobs_decode(&[]).unwrap();
+        assert!(decoded.is_empty());
+    }
+
+    #[test]
+    fn cobs_decode_invalid_zero_code() {
+        assert!(cobs_decode(&[0x00, 0x01]).is_none());
+    }
+
+    #[test]
+    fn cobs_decode_truncated() {
+        // Claims 5 bytes follow but only 3 remain
+        assert!(cobs_decode(&[0x06, 0x01, 0x02, 0x03]).is_none());
+    }
+
+    #[test]
+    fn cobs_decode_full_254_block() {
+        // 254 non-zero bytes
+        let payload: Vec<u8> = (1u8..=254).collect();
+        let mut encoded = vec![0xFF];
+        encoded.extend_from_slice(&payload);
+        let decoded = cobs_decode(&encoded).unwrap();
+        assert_eq!(decoded, payload);
+    }
+
+    // ── CobsFramer integration tests ─────────────────────────────────
+
+    #[test]
+    fn cobs_framer_single_packet() {
+        let mut f = CobsFramer::default();
+        let frames = f.feed(&[0x06, b'h', b'e', b'l', b'l', b'o', 0x00]);
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"hello");
+    }
+
+    #[test]
+    fn cobs_framer_two_packets() {
+        let mut f = CobsFramer::default();
+        let mut data = vec![0x04, b'f', b'o', b'o', 0x00];
+        data.extend_from_slice(&[0x04, b'b', b'a', b'r', 0x00]);
+        let frames = f.feed(&data);
+        assert_eq!(frames.len(), 2);
+        assert_eq!(frames[0], b"foo");
+        assert_eq!(frames[1], b"bar");
+    }
+
+    #[test]
+    fn cobs_framer_split_across_chunks() {
+        let mut f = CobsFramer::default();
+        let frames = f.feed(&[0x06, b'h', b'e']);
+        assert!(frames.is_empty());
+
+        let frames = f.feed(&[b'l', b'l', b'o', 0x00]);
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"hello");
+    }
+
+    #[test]
+    fn cobs_framer_consecutive_zeros() {
+        let mut f = CobsFramer::default();
+        let frames = f.feed(&[0x00, 0x00, 0x03, b'a', b'b', 0x00]);
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"ab");
+    }
+
+    #[test]
+    fn cobs_framer_corrupt_packet_discarded() {
+        let mut f = CobsFramer::default();
+        let frames = f.feed(&[0x06, 0x01, 0x02, 0x00]);
+        assert!(frames.is_empty());
+    }
+
+    #[test]
+    fn cobs_framer_flush_partial() {
+        let mut f = CobsFramer::default();
+        f.feed(&[0x03, b'h', b'e']);
+        let flushed = f.flush().unwrap();
+        assert_eq!(flushed, b"he");
+    }
+
+    #[test]
+    fn cobs_framer_flush_empty() {
+        let mut f = CobsFramer::default();
+        assert_eq!(f.flush(), None);
+    }
+
+    #[test]
+    fn cobs_framer_reset() {
+        let mut f = CobsFramer::default();
+        f.feed(&[0x05, b'h', b'e']);
+        assert!(f.pending_len() > 0);
+        f.reset();
+        assert_eq!(f.pending_len(), 0);
+    }
+
+    #[test]
+    fn cobs_framer_max_frame() {
+        let mut f = CobsFramer::new(3);
+        f.feed(&[0x05, b'a', b'b', b'c', b'd', b'e']);
+        assert_eq!(f.pending_len(), 3);
+    }
+
+    #[test]
+    fn cobs_framer_empty_payload_packet() {
+        let mut f = CobsFramer::default();
+        let frames = f.feed(&[0x01, 0x00]);
+        assert_eq!(frames.len(), 1);
+        assert!(frames[0].is_empty());
+    }
+
+    #[test]
+    fn cobs_framer_pending_len() {
+        let mut f = CobsFramer::default();
+        f.feed(&[0x05, b'h', b'e']);
+        assert_eq!(f.pending_len(), 3);
+    }
+}

crates/pipeview-core/src/frame/fixed.rs

@@ -0,0 +1,151 @@
+use super::Framer;
+
+/// Fixed-length framer — every N bytes forms one frame.
+#[derive(Debug, Clone)]
+pub struct FixedLengthFramer {
+    buf: Vec<u8>,
+    frame_len: usize,
+}
+
+impl FixedLengthFramer {
+    pub fn new(frame_len: usize) -> Self {
+        assert!(frame_len > 0, "frame_len must be > 0");
+        Self {
+            buf: Vec::new(),
+            frame_len,
+        }
+    }
+
+    pub fn frame_len(&self) -> usize {
+        self.frame_len
+    }
+}
+
+impl Framer for FixedLengthFramer {
+    fn feed(&mut self, data: &[u8]) -> Vec<Vec<u8>> {
+        let mut frames = Vec::new();
+        self.buf.extend_from_slice(data);
+
+        while self.buf.len() >= self.frame_len {
+            let frame: Vec<u8> = self.buf.drain(..self.frame_len).collect();
+            frames.push(frame);
+        }
+
+        frames
+    }
+
+    fn flush(&mut self) -> Option<Vec<u8>> {
+        if self.buf.is_empty() {
+            None
+        } else {
+            Some(std::mem::take(&mut self.buf))
+        }
+    }
+
+    fn reset(&mut self) {
+        self.buf.clear();
+    }
+
+    fn pending_len(&self) -> usize {
+        self.buf.len()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn fixed_exact_one_frame() {
+        let mut f = FixedLengthFramer::new(5);
+        let frames = f.feed(b"hello");
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"hello");
+    }
+
+    #[test]
+    fn fixed_multiple_frames() {
+        let mut f = FixedLengthFramer::new(3);
+        let frames = f.feed(b"abcdefghi");
+        assert_eq!(frames.len(), 3);
+        assert_eq!(frames[0], b"abc");
+        assert_eq!(frames[1], b"def");
+        assert_eq!(frames[2], b"ghi");
+    }
+
+    #[test]
+    fn fixed_partial_buffered() {
+        let mut f = FixedLengthFramer::new(5);
+        let frames = f.feed(b"abc");
+        assert!(frames.is_empty());
+        assert_eq!(f.pending_len(), 3);
+
+        let frames = f.feed(b"de");
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"abcde");
+    }
+
+    #[test]
+    fn fixed_more_than_one_frame_in_chunk() {
+        let mut f = FixedLengthFramer::new(4);
+        let frames = f.feed(b"abcdefghij"); // 10 bytes → 2 full + 2 pending
+        assert_eq!(frames.len(), 2);
+        assert_eq!(frames[0], b"abcd");
+        assert_eq!(frames[1], b"efgh");
+        assert_eq!(f.pending_len(), 2);
+    }
+
+    #[test]
+    fn fixed_flush_partial() {
+        let mut f = FixedLengthFramer::new(5);
+        f.feed(b"xy");
+        let flushed = f.flush();
+        assert_eq!(flushed, Some(b"xy".to_vec()));
+    }
+
+    #[test]
+    fn fixed_flush_empty() {
+        let mut f = FixedLengthFramer::new(3);
+        f.feed(b"abc");
+        assert_eq!(f.flush(), None);
+    }
+
+    #[test]
+    fn fixed_reset() {
+        let mut f = FixedLengthFramer::new(4);
+        f.feed(b"ab");
+        assert_eq!(f.pending_len(), 2);
+        f.reset();
+        assert_eq!(f.pending_len(), 0);
+    }
+
+    #[test]
+    fn fixed_empty_feed() {
+        let mut f = FixedLengthFramer::new(5);
+        let frames = f.feed(b"");
+        assert!(frames.is_empty());
+    }
+
+    #[test]
+    fn fixed_exact_multiple_drains() {
+        let mut f = FixedLengthFramer::new(2);
+        let frames = f.feed(b"abcd");
+        assert_eq!(frames.len(), 2);
+        assert!(f.pending_len() == 0);
+        let more = f.feed(b"ef");
+        assert_eq!(more.len(), 1);
+        assert_eq!(more[0], b"ef");
+    }
+
+    #[test]
+    #[should_panic(expected = "frame_len must be > 0")]
+    fn fixed_zero_frame_len_panics() {
+        FixedLengthFramer::new(0);
+    }
+
+    #[test]
+    fn fixed_frame_len_accessor() {
+        let f = FixedLengthFramer::new(128);
+        assert_eq!(f.frame_len(), 128);
+    }
+}

crates/pipeview-core/src/frame/length.rs

@@ -0,0 +1,377 @@
+use super::{Endian, Framer};
+use tracing::{debug, warn};
+
+/// Configuration for the length-prefixed framer.
+#[derive(Debug, Clone)]
+pub struct LengthConfig {
+    /// Size of the length field in bytes: 1, 2, or 4.
+    pub len_bytes: usize,
+    pub endian: Endian,
+    /// When true, the length value includes the length field itself.
+    pub length_includes_self: bool,
+    /// Maximum payload size (0 = unlimited).
+    pub max_payload: usize,
+}
+
+impl Default for LengthConfig {
+    fn default() -> Self {
+        Self {
+            len_bytes: 2,
+            endian: Endian::Big,
+            length_includes_self: false,
+            max_payload: 1024 * 1024,
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+enum State {
+    ReadingLength,
+    ReadingPayload { expected: usize },
+}
+
+#[derive(Debug, Clone)]
+pub struct LengthPrefixedFramer {
+    buf: Vec<u8>,
+    config: LengthConfig,
+    state: State,
+}
+
+impl LengthPrefixedFramer {
+    pub fn new(config: LengthConfig) -> Self {
+        assert!(
+            matches!(config.len_bytes, 1 | 2 | 4),
+            "len_bytes must be 1, 2, or 4"
+        );
+        Self {
+            buf: Vec::new(),
+            config,
+            state: State::ReadingLength,
+        }
+    }
+
+    pub fn config(&self) -> &LengthConfig {
+        &self.config
+    }
+
+    fn parse_length(&self, bytes: &[u8]) -> usize {
+        let raw = match self.config.len_bytes {
+            1 => bytes[0] as usize,
+            2 => {
+                let b = [bytes[0], bytes[1]];
+                match self.config.endian {
+                    Endian::Big => u16::from_be_bytes(b) as usize,
+                    Endian::Little => u16::from_le_bytes(b) as usize,
+                }
+            }
+            4 => {
+                let b = [bytes[0], bytes[1], bytes[2], bytes[3]];
+                match self.config.endian {
+                    Endian::Big => u32::from_be_bytes(b) as usize,
+                    Endian::Little => u32::from_le_bytes(b) as usize,
+                }
+            }
+            _ => unreachable!(),
+        };
+
+        if self.config.length_includes_self {
+            raw.saturating_sub(self.config.len_bytes)
+        } else {
+            raw
+        }
+    }
+}
+
+impl Framer for LengthPrefixedFramer {
+    fn feed(&mut self, data: &[u8]) -> Vec<Vec<u8>> {
+        let mut frames = Vec::new();
+        self.buf.extend_from_slice(data);
+
+        loop {
+            match self.state {
+                State::ReadingLength => {
+                    if self.buf.len() < self.config.len_bytes {
+                        break;
+                    }
+                    let payload_len = self.parse_length(&self.buf[..self.config.len_bytes]);
+
+                    // Discard length, re-sync on corrupt frame
+                    if self.config.max_payload > 0 && payload_len > self.config.max_payload {
+                        warn!(
+                            claimed = payload_len,
+                            max = self.config.max_payload,
+                            "corrupt length frame, skipping"
+                        );
+                        self.buf.drain(..self.config.len_bytes);
+                        continue;
+                    }
+
+                    self.buf.drain(..self.config.len_bytes);
+
+                    if payload_len == 0 {
+                        frames.push(Vec::new());
+                    } else {
+                        self.state = State::ReadingPayload {
+                            expected: payload_len,
+                        };
+                    }
+                }
+                State::ReadingPayload { expected } => {
+                    if self.buf.len() < expected {
+                        break;
+                    }
+                    let frame: Vec<u8> = self.buf.drain(..expected).collect();
+                    debug!(len = frame.len(), "length-prefixed frame extracted");
+                    frames.push(frame);
+                    self.state = State::ReadingLength;
+                }
+            }
+        }
+
+        frames
+    }
+
+    fn flush(&mut self) -> Option<Vec<u8>> {
+        let remainder = std::mem::take(&mut self.buf);
+        self.state = State::ReadingLength;
+        if remainder.is_empty() {
+            None
+        } else {
+            Some(remainder)
+        }
+    }
+
+    fn reset(&mut self) {
+        self.buf.clear();
+        self.state = State::ReadingLength;
+    }
+
+    fn pending_len(&self) -> usize {
+        self.buf.len()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn be2(len: u16) -> Vec<u8> {
+        len.to_be_bytes().to_vec()
+    }
+
+    fn le2(len: u16) -> Vec<u8> {
+        len.to_le_bytes().to_vec()
+    }
+
+    // ── 2-byte BE, length excludes self ───────────────────────────────
+
+    #[test]
+    fn length_single_frame() {
+        let mut f = LengthPrefixedFramer::new(LengthConfig::default());
+        let mut data = be2(5);
+        data.extend_from_slice(b"hello");
+        let frames = f.feed(&data);
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"hello");
+    }
+
+    #[test]
+    fn length_two_frames_in_one_chunk() {
+        let mut f = LengthPrefixedFramer::new(LengthConfig::default());
+        let mut data = Vec::new();
+        data.extend(&be2(3));
+        data.extend_from_slice(b"foo");
+        data.extend(&be2(3));
+        data.extend_from_slice(b"bar");
+        let frames = f.feed(&data);
+        assert_eq!(frames.len(), 2);
+        assert_eq!(frames[0], b"foo");
+        assert_eq!(frames[1], b"bar");
+    }
+
+    #[test]
+    fn length_split_header() {
+        let mut f = LengthPrefixedFramer::new(LengthConfig::default());
+        // Feed first half of 2-byte length header
+        assert!(f.feed(&be2(5)[..1]).is_empty());
+        assert_eq!(f.pending_len(), 1);
+        let mut rest = be2(5)[1..].to_vec();
+        rest.extend_from_slice(b"hello");
+        let frames = f.feed(&rest);
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"hello");
+    }
+
+    #[test]
+    fn length_split_payload() {
+        let mut f = LengthPrefixedFramer::new(LengthConfig::default());
+        let frames = f.feed(&be2(5));
+        assert!(frames.is_empty());
+        assert_eq!(f.pending_len(), 0); // length consumed, waiting for payload
+
+        let frames = f.feed(b"hel");
+        assert!(frames.is_empty());
+
+        let frames = f.feed(b"lo");
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"hello");
+    }
+
+    #[test]
+    fn length_zero_payload() {
+        let mut f = LengthPrefixedFramer::new(LengthConfig::default());
+        let frames = f.feed(&be2(0));
+        assert_eq!(frames.len(), 1);
+        assert!(frames[0].is_empty());
+    }
+
+    #[test]
+    fn length_chain_zero_then_data() {
+        let mut f = LengthPrefixedFramer::new(LengthConfig::default());
+        let mut data = Vec::new();
+        data.extend(&be2(0));
+        data.extend(&be2(3));
+        data.extend_from_slice(b"foo");
+        let frames = f.feed(&data);
+        assert_eq!(frames.len(), 2);
+        assert!(frames[0].is_empty());
+        assert_eq!(frames[1], b"foo");
+    }
+
+    // ── 2-byte LE ────────────────────────────────────────────────────
+
+    #[test]
+    fn length_little_endian() {
+        let cfg = LengthConfig {
+            endian: Endian::Little,
+            ..LengthConfig::default()
+        };
+        let mut f = LengthPrefixedFramer::new(cfg);
+        let mut data = le2(5);
+        data.extend_from_slice(b"hello");
+        let frames = f.feed(&data);
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"hello");
+    }
+
+    // ── 1-byte length ────────────────────────────────────────────────
+
+    #[test]
+    fn length_one_byte() {
+        let cfg = LengthConfig {
+            len_bytes: 1,
+            ..LengthConfig::default()
+        };
+        let mut f = LengthPrefixedFramer::new(cfg);
+        let data = [5, b'h', b'e', b'l', b'l', b'o'];
+        let frames = f.feed(&data);
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"hello");
+    }
+
+    // ── 4-byte length ────────────────────────────────────────────────
+
+    #[test]
+    fn length_four_byte() {
+        let cfg = LengthConfig {
+            len_bytes: 4,
+            ..LengthConfig::default()
+        };
+        let mut f = LengthPrefixedFramer::new(cfg);
+        let mut data = 5u32.to_be_bytes().to_vec();
+        data.extend_from_slice(b"hello");
+        let frames = f.feed(&data);
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"hello");
+    }
+
+    // ── length_includes_self ─────────────────────────────────────────
+
+    #[test]
+    fn length_includes_self_enabled() {
+        let cfg = LengthConfig {
+            length_includes_self: true,
+            ..LengthConfig::default()
+        };
+        let mut f = LengthPrefixedFramer::new(cfg);
+        let mut data = be2(5);
+        data.extend_from_slice(b"foo");
+        let frames = f.feed(&data);
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"foo");
+    }
+
+    #[test]
+    fn length_includes_self_zero_payload() {
+        let cfg = LengthConfig {
+            length_includes_self: true,
+            ..LengthConfig::default()
+        };
+        let mut f = LengthPrefixedFramer::new(cfg);
+        let frames = f.feed(&be2(2));
+        assert_eq!(frames.len(), 1);
+        assert!(frames[0].is_empty());
+    }
+
+    // ── max_payload safety ───────────────────────────────────────────
+
+    #[test]
+    fn length_max_payload_exceeded_skips() {
+        let cfg = LengthConfig {
+            max_payload: 10,
+            ..LengthConfig::default()
+        };
+        let mut f = LengthPrefixedFramer::new(cfg);
+        // Corrupt frame: claims 200 bytes, actual payload is 0xFF bytes
+        let mut data = be2(200);
+        data.extend_from_slice(&[0xFFu8; 200]);
+        // Followed by valid frame
+        data.extend(&be2(3));
+        data.extend_from_slice(b"foo");
+        let frames = f.feed(&data);
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"foo");
+    }
+
+    // ── flush / reset ────────────────────────────────────────────────
+
+    #[test]
+    fn length_flush_partial() {
+        let mut f = LengthPrefixedFramer::new(LengthConfig::default());
+        f.feed(&be2(10));
+        f.feed(b"abc");
+        let flushed = f.flush();
+        assert_eq!(flushed, Some(b"abc".to_vec()));
+    }
+
+    #[test]
+    fn length_flush_empty() {
+        let mut f = LengthPrefixedFramer::new(LengthConfig::default());
+        assert_eq!(f.flush(), None);
+    }
+
+    #[test]
+    fn length_reset_mid_frame() {
+        let mut f = LengthPrefixedFramer::new(LengthConfig::default());
+        f.feed(&be2(50));
+        f.feed(b"partial");
+        assert!(f.pending_len() > 0);
+        f.reset();
+        assert_eq!(f.pending_len(), 0);
+        // Should work normally after reset
+        let mut data = be2(3);
+        data.extend_from_slice(b"foo");
+        let frames = f.feed(&data);
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"foo");
+    }
+
+    #[test]
+    fn length_pending_len_reflects_buffer() {
+        let mut f = LengthPrefixedFramer::new(LengthConfig::default());
+        f.feed(&be2(5));
+        assert_eq!(f.pending_len(), 0);
+        f.feed(b"he");
+        assert_eq!(f.pending_len(), 2);
+    }
+}

crates/pipeview-core/src/frame/line.rs

@@ -0,0 +1,226 @@
+use super::Framer;
+use tracing::debug;
+
+/// Line-delimited framer.
+///
+/// Splits incoming bytes on `\n` (LF).  Optional `\r` (CR) stripping is
+/// controlled via [`LineConfig::strip_cr`].
+#[derive(Debug, Clone)]
+pub struct LineConfig {
+    pub strip_cr: bool,
+    pub max_line_len: usize,
+}
+
+impl Default for LineConfig {
+    fn default() -> Self {
+        Self {
+            strip_cr: true,
+            max_line_len: 1024 * 1024,
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct LineFramer {
+    buf: Vec<u8>,
+    config: LineConfig,
+}
+
+impl LineFramer {
+    pub fn new(config: LineConfig) -> Self {
+        Self {
+            buf: Vec::new(),
+            config,
+        }
+    }
+
+    pub fn config(&self) -> &LineConfig {
+        &self.config
+    }
+}
+
+impl Framer for LineFramer {
+    fn feed(&mut self, data: &[u8]) -> Vec<Vec<u8>> {
+        let mut frames = Vec::new();
+        self.buf.extend_from_slice(data);
+
+        while let Some(pos) = self.buf.iter().position(|&b| b == b'\n') {
+            let mut line = self.buf.drain(..=pos).collect::<Vec<u8>>();
+            line.pop();
+            if self.config.strip_cr && line.last() == Some(&b'\r') {
+                line.pop();
+            }
+            if line.len() <= self.config.max_line_len {
+                frames.push(line);
+            }
+        }
+
+        if !frames.is_empty() {
+            debug!(
+                count = frames.len(),
+                pending = self.buf.len(),
+                "line frames extracted"
+            );
+        }
+        frames
+    }
+
+    fn flush(&mut self) -> Option<Vec<u8>> {
+        if self.buf.is_empty() {
+            None
+        } else {
+            Some(std::mem::take(&mut self.buf))
+        }
+    }
+
+    fn reset(&mut self) {
+        self.buf.clear();
+    }
+
+    fn pending_len(&self) -> usize {
+        self.buf.len()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn line_single_complete() {
+        let mut f = LineFramer::new(LineConfig::default());
+        let frames = f.feed(b"hello\n");
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"hello");
+    }
+
+    #[test]
+    fn line_multiple_in_one_chunk() {
+        let mut f = LineFramer::new(LineConfig::default());
+        let frames = f.feed(b"aaa\nbbb\nccc\n");
+        assert_eq!(frames.len(), 3);
+        assert_eq!(frames[0], b"aaa");
+        assert_eq!(frames[1], b"bbb");
+        assert_eq!(frames[2], b"ccc");
+    }
+
+    #[test]
+    fn line_split_across_chunks() {
+        let mut f = LineFramer::new(LineConfig::default());
+        let frames = f.feed(b"hel");
+        assert!(frames.is_empty());
+        let frames = f.feed(b"lo\n");
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"hello");
+    }
+
+    #[test]
+    fn line_crlf_stripping() {
+        let mut f = LineFramer::new(LineConfig::default());
+        let frames = f.feed(b"hello\r\nworld\r\n");
+        assert_eq!(frames.len(), 2);
+        assert_eq!(frames[0], b"hello");
+        assert_eq!(frames[1], b"world");
+    }
+
+    #[test]
+    fn line_crlf_no_strip() {
+        let cfg = LineConfig {
+            strip_cr: false,
+            ..LineConfig::default()
+        };
+        let mut f = LineFramer::new(cfg);
+        let frames = f.feed(b"hello\r\n");
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"hello\r");
+    }
+
+    #[test]
+    fn line_empty_lines() {
+        let mut f = LineFramer::new(LineConfig::default());
+        let frames = f.feed(b"\n\n\n");
+        assert_eq!(frames.len(), 3);
+        assert!(frames[0].is_empty());
+        assert!(frames[1].is_empty());
+        assert!(frames[2].is_empty());
+    }
+
+    #[test]
+    fn line_no_newline_buffered() {
+        let mut f = LineFramer::new(LineConfig::default());
+        let frames = f.feed(b"incomplete");
+        assert!(frames.is_empty());
+        assert_eq!(f.pending_len(), 10);
+    }
+
+    #[test]
+    fn line_flush_partial() {
+        let mut f = LineFramer::new(LineConfig::default());
+        f.feed(b"partial data");
+        let flushed = f.flush();
+        assert_eq!(flushed, Some(b"partial data".to_vec()));
+        assert_eq!(f.pending_len(), 0);
+    }
+
+    #[test]
+    fn line_flush_empty() {
+        let mut f = LineFramer::new(LineConfig::default());
+        assert_eq!(f.flush(), None);
+    }
+
+    #[test]
+    fn line_reset() {
+        let mut f = LineFramer::new(LineConfig::default());
+        f.feed(b"some data");
+        assert!(f.pending_len() > 0);
+        f.reset();
+        assert_eq!(f.pending_len(), 0);
+    }
+
+    #[test]
+    fn line_max_len_filtered() {
+        let cfg = LineConfig {
+            max_line_len: 5,
+            ..LineConfig::default()
+        };
+        let mut f = LineFramer::new(cfg);
+        let frames = f.feed(b"short\nvery long line\nok\n");
+        assert_eq!(frames.len(), 2);
+        assert_eq!(frames[0], b"short");
+        assert_eq!(frames[1], b"ok");
+    }
+
+    #[test]
+    fn line_feed_then_flush_chain() {
+        let mut f = LineFramer::new(LineConfig::default());
+        let frames = f.feed(b"complete\npartial");
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"complete");
+
+        let flushed = f.flush();
+        assert_eq!(flushed, Some(b"partial".to_vec()));
+
+        let frames = f.feed(b"new\n");
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0], b"new");
+    }
+
+    #[test]
+    fn line_binary_data_with_embedded_newlines() {
+        let mut f = LineFramer::new(LineConfig::default());
+        let data = [0x00, 0x01, b'\n', 0xFF, 0xFE, b'\n', 0x7F];
+        let frames = f.feed(&data);
+        assert_eq!(frames.len(), 2);
+        assert_eq!(frames[0], &[0x00, 0x01]);
+        assert_eq!(frames[1], &[0xFF, 0xFE]);
+    }
+
+    #[test]
+    fn line_lf_only_with_strip_cr() {
+        let mut f = LineFramer::new(LineConfig::default());
+        let frames = f.feed(b"line1\nline2\n");
+        assert_eq!(frames.len(), 2);
+        assert_eq!(frames[0], b"line1");
+        assert_eq!(frames[1], b"line2");
+    }
+}

crates/pipeview-core/src/frame/mixed.rs

@@ -0,0 +1,215 @@
+use super::Framer;
+use crate::frame::cobs::cobs_decode;
+use crate::protocol::mixed::{
+    MIXED_FRAME_KIND_PLOT, MIXED_FRAME_KIND_TEXT, MIXED_PLOT_ESCAPE, MIXED_PLOT_MARKER,
+};
+
+#[derive(Debug, Clone)]
+pub struct MixedTextPlotConfig {
+    pub strip_cr: bool,
+    pub max_line_len: usize,
+    pub max_plot_frame: usize,
+}
+
+impl Default for MixedTextPlotConfig {
+    fn default() -> Self {
+        Self {
+            strip_cr: true,
+            max_line_len: 1024 * 1024,
+            max_plot_frame: 1024 * 1024,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum State {
+    Text,
+    Escape,
+    Plot,
+}
+
+#[derive(Debug, Clone)]
+pub struct MixedTextPlotFramer {
+    state: State,
+    text_buf: Vec<u8>,
+    plot_buf: Vec<u8>,
+    plot_overflow: bool,
+    config: MixedTextPlotConfig,
+}
+
+impl MixedTextPlotFramer {
+    pub fn new(config: MixedTextPlotConfig) -> Self {
+        Self {
+            state: State::Text,
+            text_buf: Vec::new(),
+            plot_buf: Vec::new(),
+            plot_overflow: false,
+            config,
+        }
+    }
+
+    fn push_text_byte(&mut self, byte: u8, frames: &mut Vec<Vec<u8>>) {
+        if byte == b'\n' {
+            let mut line = std::mem::take(&mut self.text_buf);
+            if self.config.strip_cr && line.last() == Some(&b'\r') {
+                line.pop();
+            }
+            if line.len() <= self.config.max_line_len {
+                frames.push(tag_text_frame(line));
+            }
+            return;
+        }
+
+        if self.text_buf.len() < self.config.max_line_len.saturating_add(1) {
+            self.text_buf.push(byte);
+        }
+    }
+
+    fn finish_plot_frame(&mut self, frames: &mut Vec<Vec<u8>>) {
+        if self.plot_overflow || self.plot_buf.is_empty() {
+            self.plot_buf.clear();
+            self.plot_overflow = false;
+            return;
+        }
+
+        if let Some(decoded) = cobs_decode(&self.plot_buf)
+            && decoded.len() <= self.config.max_plot_frame
+        {
+            frames.push(tag_plot_frame(decoded));
+        }
+        self.plot_buf.clear();
+        self.plot_overflow = false;
+    }
+}
+
+impl Framer for MixedTextPlotFramer {
+    fn feed(&mut self, data: &[u8]) -> Vec<Vec<u8>> {
+        let mut frames = Vec::new();
+
+        for &byte in data {
+            match self.state {
+                State::Text => {
+                    if byte == MIXED_PLOT_ESCAPE {
+                        self.state = State::Escape;
+                    } else {
+                        self.push_text_byte(byte, &mut frames);
+                    }
+                }
+                State::Escape => {
+                    self.state = State::Text;
+                    match byte {
+                        MIXED_PLOT_ESCAPE => self.push_text_byte(MIXED_PLOT_ESCAPE, &mut frames),
+                        MIXED_PLOT_MARKER => {
+                            self.plot_buf.clear();
+                            self.plot_overflow = false;
+                            self.state = State::Plot;
+                        }
+                        _ => {
+                            self.push_text_byte(MIXED_PLOT_ESCAPE, &mut frames);
+                            self.push_text_byte(byte, &mut frames);
+                        }
+                    }
+                }
+                State::Plot => {
+                    if byte == 0x00 {
+                        self.finish_plot_frame(&mut frames);
+                        self.state = State::Text;
+                    } else if !self.plot_overflow {
+                        let max_encoded_len = self
+                            .config
+                            .max_plot_frame
+                            .saturating_add(self.config.max_plot_frame / 254)
+                            .saturating_add(8);
+                        if self.plot_buf.len() < max_encoded_len {
+                            self.plot_buf.push(byte);
+                        } else {
+                            self.plot_overflow = true;
+                        }
+                    }
+                }
+            }
+        }
+
+        frames
+    }
+
+    fn flush(&mut self) -> Option<Vec<u8>> {
+        if matches!(self.state, State::Escape) {
+            self.text_buf.push(MIXED_PLOT_ESCAPE);
+        }
+
+        self.state = State::Text;
+        self.plot_buf.clear();
+        self.plot_overflow = false;
+
+        if self.text_buf.is_empty() {
+            None
+        } else {
+            Some(tag_text_frame(std::mem::take(&mut self.text_buf)))
+        }
+    }
+
+    fn reset(&mut self) {
+        self.state = State::Text;
+        self.text_buf.clear();
+        self.plot_buf.clear();
+        self.plot_overflow = false;
+    }
+
+    fn pending_len(&self) -> usize {
+        self.text_buf.len() + self.plot_buf.len()
+    }
+}
+
+fn tag_text_frame(mut payload: Vec<u8>) -> Vec<u8> {
+    let mut frame = Vec::with_capacity(payload.len() + 1);
+    frame.push(MIXED_FRAME_KIND_TEXT);
+    frame.append(&mut payload);
+    frame
+}
+
+fn tag_plot_frame(mut payload: Vec<u8>) -> Vec<u8> {
+    let mut frame = Vec::with_capacity(payload.len() + 1);
+    frame.push(MIXED_FRAME_KIND_PLOT);
+    frame.append(&mut payload);
+    frame
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::frame::cobs::cobs_encode;
+
+    #[test]
+    fn mixed_text_lines_pass_through() {
+        let mut framer = MixedTextPlotFramer::new(MixedTextPlotConfig::default());
+        let frames = framer.feed(b"hello\nworld\n");
+        assert_eq!(frames, vec![b"\0hello".to_vec(), b"\0world".to_vec()]);
+    }
+
+    #[test]
+    fn mixed_escaped_rs_stays_in_text() {
+        let mut framer = MixedTextPlotFramer::new(MixedTextPlotConfig::default());
+        let frames = framer.feed(&[b'o', b'k', MIXED_PLOT_ESCAPE, MIXED_PLOT_ESCAPE, b'\n']);
+        assert_eq!(frames, vec![vec![0x00, b'o', b'k', MIXED_PLOT_ESCAPE]]);
+    }
+
+    #[test]
+    fn mixed_extracts_plot_packet() {
+        let mut framer = MixedTextPlotFramer::new(MixedTextPlotConfig::default());
+        let payload = [b'X', b'P', 1, 0, 8, 0, 1, 1, 0, 4, 0, 0, 0, 0, 0x80, 0x3f];
+        let encoded = cobs_encode(&payload);
+        let mut stream = vec![MIXED_PLOT_ESCAPE, MIXED_PLOT_MARKER];
+        stream.extend_from_slice(&encoded);
+        stream.push(0x00);
+        let frames = framer.feed(&stream);
+        assert_eq!(frames, vec![[vec![0x01], payload.to_vec()].concat()]);
+    }
+
+    #[test]
+    fn mixed_ignores_truncated_plot_on_flush() {
+        let mut framer = MixedTextPlotFramer::new(MixedTextPlotConfig::default());
+        let _ = framer.feed(&[MIXED_PLOT_ESCAPE, MIXED_PLOT_MARKER, 0x03, b'a']);
+        assert_eq!(framer.flush(), None);
+    }
+}

crates/pipeview-core/src/frame/mod.rs

@@ -0,0 +1,36 @@
+pub mod cobs;
+pub mod fixed;
+pub mod length;
+pub mod line;
+pub mod mixed;
+
+pub use crate::protocol::Endian;
+
+/// Stateful byte-stream framer.
+///
+/// A `Framer` accumulates raw bytes from a transport layer and yields
+/// complete frames once a boundary is detected.  Each call to [`feed`]
+/// may return zero, one, or many frames depending on how much data has
+/// arrived.
+///
+/// [`feed`]: Framer::feed
+pub trait Framer: Send {
+    /// Feed a chunk of newly arrived bytes.
+    ///
+    /// Any complete frames that can be extracted are returned.  The
+    /// framer keeps incomplete trailing data in its internal buffer so
+    /// it can be completed on the next call.
+    fn feed(&mut self, data: &[u8]) -> Vec<Vec<u8>>;
+
+    /// Drain any remaining buffered data as a final frame.
+    ///
+    /// This is typically called when the transport disconnects or the
+    /// user explicitly wants to flush a partial frame.
+    fn flush(&mut self) -> Option<Vec<u8>>;
+
+    /// Discard all buffered state and start fresh.
+    fn reset(&mut self);
+
+    /// Number of bytes currently buffered in the incomplete frame.
+    fn pending_len(&self) -> usize;
+}