repositories / smith
smith
There are many coding harnesses - but this one is fast
owned by admin
smith-core/src/frame.rs
Raw//! Structurally traversed typed session frames (`SMH-SPEC-SPEC0001`, Sessions).
//!
//! File: `[SMH3, format=1, reserved=0: 8 bytes][frames...]`.
//! Frame, little endian: `[body_len u32][kind u16][version u16][header_crc u32]`
//! `[body_crc u32][payload][body_crc XOR 0x534d4833 u32]`.
//! `body_len` counts body CRC, payload, and completion marker (not the 12-byte
//! header). Header CRC-32 covers the first eight header bytes; body CRC-32
//! covers kind, version, and payload. Complete intact-boundary records with
//! damaged bodies are skipped by length; an untrustworthy header stops the
//! walk rather than scanning payload for marker-shaped bytes. Entries carry
//! 62 bytes of tightly packed fixed metadata followed by length-checked typed
//! content; format has no implicit padding, absent parent IDs are zero-filled.
use crate::session::{Entry, EntryFrame, EntryKind, SessionHeader};
use smith::error::{FrameFault, Result, SmithError};
use smith::id::{BranchId, EntryId, MessageId, SessionId, ToolCallId};
use smith::message::Role;
use std::io::{Read, Write};
/// Fixed file prologue, written once before the first framed record.
pub const FILE_PROLOGUE: [u8; 8] = *b"SMH3\x01\0\0\0";
/// Default frame size ceiling when no configuration is supplied.
pub const DEFAULT_MAX_FRAME_BYTES: usize = 1024 * 1024;
// Permit brief signal bursts; a reader that never progresses must fail closed.
const INTERRUPTED_READS_MAX: usize = 16;
/// Length of the independently checked frame header.
pub const HEADER_LEN: usize = 12;
/// Body checksum and completion marker bytes excluding payload.
pub const BODY_PREFIX_LEN: usize = 8;
/// Frame kind of a session entry payload.
pub const KIND_ENTRY: u16 = 1;
/// Supported version of the [`KIND_ENTRY`] payload schema.
/// Version 4 uses fixed metadata followed by typed binary content and
/// accompanies durable navigation records in the session protocol.
pub const VERSION_ENTRY: u16 = 4;
/// Frame kind of the session header payload.
pub const KIND_SESSION_HEADER: u16 = 2;
/// Supported version of the [`KIND_SESSION_HEADER`] payload schema.
pub const VERSION_SESSION_HEADER: u16 = 2;
/// An embedded, reconstructible entry-location hint.
pub const KIND_INDEX: u16 = 3;
/// Durable selection of a branch tip.
pub const KIND_SELECTION: u16 = 4;
/// Reconstructible selected-tip checkpoint kind.
pub const KIND_CHECKPOINT: u16 = 5;
/// Version of index, selection and checkpoint records.
pub const VERSION_NAV: u16 = 1;
/// A decoded session frame.
#[derive(Clone, PartialEq, Eq, Debug)]
pub enum Frame {
/// Durable session identity; the first frame of a session file.
Header {
/// Decoded header.
header: SessionHeader,
},
/// A known session entry.
Known {
/// Decoded entry frame; its content stays typed binary bytes.
entry: EntryFrame,
},
/// Derived location hint; authoritative entry frames remain discoverable without it.
Index {
/// Entry identifier.
id: EntryId,
/// Absolute byte offset of its entry frame.
offset: u64,
/// Parent identity, independent of physical address.
parent: Option<EntryId>,
/// Absolute byte offset of its parent entry frame, if known.
parent_offset: Option<u64>,
},
/// Append-only selection of an already recorded entry.
Selection {
/// Selected entry identifier.
id: EntryId,
},
/// Reconstructible selected-tip location hint; never authoritative.
Checkpoint {
/// Selected entry identifier.
id: EntryId,
/// Its absolute entry-frame boundary.
offset: u64,
},
/// A checksum-valid frame of a kind this build does not know.
/// Kind, version, and payload are preserved so it round-trips unchanged.
Unknown {
/// Frame kind tag.
kind: u16,
/// Frame version tag.
version: u16,
/// Exact payload bytes as read.
payload: Vec<u8>,
},
}
impl Frame {
/// Encode the full frame (length prefix, body prefix, payload) enforcing
/// `max_frame_bytes` on the payload.
///
/// # Errors
///
/// Returns a [`FrameFault::TooLarge`] or [`FrameFault::Encode`] error.
pub fn encode(&self, frame_bytes_max: usize) -> Result<Vec<u8>> {
match self {
Self::Header { header } => seal(
KIND_SESSION_HEADER,
VERSION_SESSION_HEADER,
header_bytes(header),
frame_bytes_max,
),
Self::Known { entry } => encode_entry_fields(
entry.id,
entry.parent,
entry.timestamp_ms,
entry.kind,
&entry.content,
frame_bytes_max,
),
Self::Index {
id,
offset,
parent,
parent_offset,
} => {
if frame_bytes_max < 48 {
return Err(too_large(frame_bytes_max, 48));
}
Ok(index_bytes(*id, *offset, *parent, *parent_offset).to_vec())
}
Self::Selection { id } => {
let mut out = Vec::with_capacity(FRAME_PREFIX_LEN + 16 + 4);
out.resize(FRAME_PREFIX_LEN, 0);
out.extend_from_slice(&id.as_u128().to_le_bytes());
seal(KIND_SELECTION, VERSION_NAV, out, frame_bytes_max)
}
Self::Checkpoint { id, offset } => {
let mut out = Vec::with_capacity(FRAME_PREFIX_LEN + 24 + 4);
out.resize(FRAME_PREFIX_LEN, 0);
out.extend_from_slice(&id.as_u128().to_le_bytes());
out.extend_from_slice(&offset.to_le_bytes());
seal(KIND_CHECKPOINT, VERSION_NAV, out, frame_bytes_max)
}
Self::Unknown {
kind,
version,
payload,
} => {
if payload.len() > frame_bytes_max {
return Err(too_large(frame_bytes_max, payload.len()));
}
let mut out = Vec::with_capacity(FRAME_PREFIX_LEN + payload.len() + 4);
out.resize(FRAME_PREFIX_LEN, 0);
out.extend_from_slice(payload);
seal(*kind, *version, out, frame_bytes_max)
}
}
}
}
/// Encode one fixed-width location hint without allocating.
#[must_use]
pub fn index_bytes(
id: EntryId,
offset: u64,
parent: Option<EntryId>,
parent_offset: Option<u64>,
) -> [u8; 68] {
let mut out = [0u8; 68];
out[4..6].copy_from_slice(&KIND_INDEX.to_le_bytes());
out[6..8].copy_from_slice(&VERSION_NAV.to_le_bytes());
out[16..32].copy_from_slice(&id.as_u128().to_le_bytes());
out[32..40].copy_from_slice(&offset.to_le_bytes());
out[40..56].copy_from_slice(&parent.map_or(0, |id| id.as_u128()).to_le_bytes());
out[56..64].copy_from_slice(&parent_offset.unwrap_or(0).to_le_bytes());
out[..4].copy_from_slice(&56_u32.to_le_bytes());
let header_crc = crc32fast::hash(&out[..8]);
out[8..12].copy_from_slice(&header_crc.to_le_bytes());
let checksum = crc32(KIND_INDEX, VERSION_NAV, &out[16..64]);
out[12..16].copy_from_slice(&checksum.to_le_bytes());
out[64..68].copy_from_slice(&(checksum ^ COMPLETION).to_le_bytes());
out
}
/// Encode a session record and its arena content bytes as a full entry
/// frame, without copying the content out of the arena first.
///
/// # Errors
///
/// Returns a [`FrameFault::TooLarge`] or [`FrameFault::Encode`] error.
pub fn encode_entry(entry: &Entry, content: &[u8], frame_bytes_max: usize) -> Result<Vec<u8>> {
encode_entry_fields(
entry.id,
entry.parent,
entry.timestamp_ms,
entry.kind,
content,
frame_bytes_max,
)
}
/// Bytes before a frame payload: length prefix, kind, version, checksum.
const FRAME_PREFIX_LEN: usize = HEADER_LEN + 4;
/// Fixed-size entry metadata is independent of the variable content span.
const ENTRY_METADATA_LEN: usize = 16 + 16 + 8 + 1 + 16 + 1 + 4;
const _: () = assert!(HEADER_LEN == 12, "frame header wire width changed");
const _: () = assert!(FRAME_PREFIX_LEN == 16, "frame prefix wire width changed");
const _: () = assert!(
ENTRY_METADATA_LEN == 62,
"entry metadata wire width changed"
);
fn header_bytes(header: &SessionHeader) -> Vec<u8> {
let mut bytes = Vec::with_capacity(FRAME_PREFIX_LEN + 32 + 4);
bytes.resize(FRAME_PREFIX_LEN, 0);
bytes.extend_from_slice(&header.session_id.as_u128().to_le_bytes());
bytes.extend_from_slice(&header.branch_id.as_u128().to_le_bytes());
bytes
}
#[expect(
clippy::large_types_passed_by_value,
reason = "EntryKind is a compact Copy wire tag passed once to the encoder"
)]
fn encode_entry_fields(
id: EntryId,
parent: Option<EntryId>,
timestamp_ms: i64,
kind: EntryKind,
content: &[u8],
frame_bytes_max: usize,
) -> Result<Vec<u8>> {
let payload_len = ENTRY_METADATA_LEN
.checked_add(content.len())
.ok_or_else(|| too_large(frame_bytes_max, usize::MAX))?;
if payload_len > frame_bytes_max {
return Err(too_large(frame_bytes_max, payload_len));
}
let length =
u32::try_from(content.len()).map_err(|_| too_large(frame_bytes_max, payload_len))?;
let mut bytes = Vec::with_capacity(FRAME_PREFIX_LEN + payload_len + 4);
bytes.resize(FRAME_PREFIX_LEN, 0);
bytes.extend_from_slice(&id.as_u128().to_le_bytes());
bytes.extend_from_slice(&parent.map_or(0, |p| p.as_u128()).to_le_bytes());
bytes.extend_from_slice(×tamp_ms.to_le_bytes());
let (tag, bits, flag) = match kind {
EntryKind::Message { id, role } => (
0,
id.as_u128(),
match role {
Role::User => 0,
Role::Assistant => 1,
Role::System => 2,
Role::Tool => 3,
},
),
EntryKind::ToolCall { call_id } => (1, call_id.as_u128(), 0),
EntryKind::ToolResult { call_id, ok } => (2, call_id.as_u128(), u8::from(ok)),
EntryKind::Meta { compaction } => (3, 0, u8::from(compaction)),
};
bytes.push(tag);
bytes.extend_from_slice(&bits.to_le_bytes());
bytes.push(flag);
bytes.extend_from_slice(&length.to_le_bytes());
bytes.extend_from_slice(content);
seal(KIND_ENTRY, VERSION_ENTRY, bytes, frame_bytes_max)
}
fn decode_header(payload: &[u8]) -> std::result::Result<SessionHeader, FrameFault> {
let bytes: [u8; 32] = payload
.try_into()
.map_err(|_| schema("invalid session header length"))?;
Ok(SessionHeader {
session_id: SessionId::from_u128(u128::from_le_bytes(
bytes[..16]
.try_into()
.map_err(|_| schema("invalid session id"))?,
)),
branch_id: BranchId::from_u128(u128::from_le_bytes(
bytes[16..]
.try_into()
.map_err(|_| schema("invalid branch id"))?,
)),
})
}
fn schema(message: &str) -> FrameFault {
FrameFault::SchemaDecode {
message: message.into(),
}
}
#[expect(
clippy::expect_used,
reason = "fixed metadata width checked before slicing"
)]
fn decode_entry(payload: &[u8]) -> std::result::Result<EntryFrame, FrameFault> {
let metadata = payload
.get(..ENTRY_METADATA_LEN)
.ok_or_else(|| schema("short entry metadata"))?;
let bits = |slice: std::ops::Range<usize>| -> u128 {
u128::from_le_bytes(metadata[slice].try_into().expect("fixed metadata width"))
};
let id = EntryId::from_u128(bits(0..16));
let parent = match bits(16..32) {
0 => None,
bits => Some(EntryId::from_u128(bits)),
};
let timestamp_ms =
i64::from_le_bytes(metadata[32..40].try_into().expect("fixed timestamp width"));
let flag = metadata[57];
let bool_flag = || match flag {
0 => Ok(false),
1 => Ok(true),
_ => Err(schema("invalid metadata boolean")),
};
let kind = match metadata[40] {
0 => EntryKind::Message {
id: MessageId::from_u128(bits(41..57)),
role: match flag {
0 => Role::User,
1 => Role::Assistant,
2 => Role::System,
3 => Role::Tool,
_ => return Err(schema("invalid role")),
},
},
1 if flag == 0 => EntryKind::ToolCall {
call_id: ToolCallId::from_u128(bits(41..57)),
},
2 => EntryKind::ToolResult {
call_id: ToolCallId::from_u128(bits(41..57)),
ok: bool_flag()?,
},
3 if bits(41..57) == 0 => EntryKind::Meta {
compaction: bool_flag()?,
},
_ => return Err(schema("invalid entry kind metadata")),
};
let content_len =
u32::from_le_bytes(metadata[58..62].try_into().expect("fixed length width")) as usize;
let content = payload
.get(ENTRY_METADATA_LEN..)
.ok_or_else(|| schema("missing content"))?;
if content.len() != content_len {
return Err(schema("entry content length mismatch"));
}
Ok(EntryFrame {
id,
parent,
timestamp_ms,
kind,
content: content.to_vec(),
})
}
/// Fill the reserved prefix of `out` (length, kind, version, checksum) for
/// the payload that follows it.
fn seal(kind: u16, version: u16, mut out: Vec<u8>, frame_bytes_max: usize) -> Result<Vec<u8>> {
assert!(
out.len() >= FRAME_PREFIX_LEN,
"encoder must reserve the complete frame prefix before sealing"
);
let payload_len = out.len() - FRAME_PREFIX_LEN;
if payload_len > frame_bytes_max {
return Err(too_large(frame_bytes_max, payload_len));
}
let Ok(declared) = u32::try_from(BODY_PREFIX_LEN + payload_len) else {
return Err(too_large(frame_bytes_max, payload_len));
};
let checksum = crc32(kind, version, &out[FRAME_PREFIX_LEN..]);
out[..4].copy_from_slice(&declared.to_le_bytes());
out[4..6].copy_from_slice(&kind.to_le_bytes());
out[6..8].copy_from_slice(&version.to_le_bytes());
let header_checksum = crc32fast::hash(&out[..8]);
out[8..HEADER_LEN].copy_from_slice(&header_checksum.to_le_bytes());
out[HEADER_LEN..FRAME_PREFIX_LEN].copy_from_slice(&checksum.to_le_bytes());
out.extend_from_slice(&(checksum ^ COMPLETION).to_le_bytes());
Ok(out)
}
/// Classify one frame body: verify the checksum first, then the kind and
/// version, then the payload schema.
const COMPLETION: u32 = 0x53_4d_48_33;
fn decode_body(kind: u16, version: u16, body: &[u8]) -> std::result::Result<Frame, FrameFault> {
if body.len() < BODY_PREFIX_LEN {
return Err(FrameFault::Malformed {
declared_length: body.len(),
});
}
let checksum = u32::from_le_bytes(
body[..4]
.try_into()
.map_err(|_| schema("missing body checksum"))?,
);
let end = body.len() - 4;
let published = u32::from_le_bytes(
body[end..]
.try_into()
.map_err(|_| schema("missing completion"))?,
);
let payload = &body[4..end];
if published != checksum ^ COMPLETION || checksum != crc32(kind, version, payload) {
return Err(FrameFault::Checksum {
declared_length: body.len(),
});
}
match (kind, version) {
(KIND_ENTRY, VERSION_ENTRY) => decode_entry(payload).map(|entry| Frame::Known { entry }),
(KIND_SESSION_HEADER, VERSION_SESSION_HEADER) => {
decode_header(payload).map(|header| Frame::Header { header })
}
(KIND_INDEX, VERSION_NAV) if payload.len() == 48 => {
let bits = |range: std::ops::Range<usize>| -> u128 {
// Fixed width is checked by the payload length guard.
u128::from_le_bytes(payload[range].try_into().unwrap_or([0; 16]))
};
let word = |range: std::ops::Range<usize>| -> u64 {
u64::from_le_bytes(payload[range].try_into().unwrap_or([0; 8]))
};
let parent = bits(24..40);
let parent_offset = word(40..48);
if (parent == 0) != (parent_offset == 0) {
return Err(schema("index parent address mismatch"));
}
Ok(Frame::Index {
id: EntryId::from_u128(bits(0..16)),
offset: word(16..24),
parent: (parent != 0).then(|| EntryId::from_u128(parent)),
parent_offset: (parent_offset != 0).then_some(parent_offset),
})
}
(KIND_SELECTION, VERSION_NAV) if payload.len() == 16 => Ok(Frame::Selection {
id: EntryId::from_u128(u128::from_le_bytes(
payload.try_into().map_err(|_| schema("selection length"))?,
)),
}),
(KIND_CHECKPOINT, VERSION_NAV) if payload.len() == 24 => Ok(Frame::Checkpoint {
id: EntryId::from_u128(u128::from_le_bytes(
payload[..16]
.try_into()
.map_err(|_| schema("checkpoint id"))?,
)),
offset: u64::from_le_bytes(
payload[16..]
.try_into()
.map_err(|_| schema("checkpoint offset"))?,
),
}),
(KIND_INDEX | KIND_SELECTION | KIND_CHECKPOINT, VERSION_NAV) => {
Err(schema("navigation length"))
}
_ => Ok(Frame::Unknown {
kind,
version,
payload: payload.to_vec(),
}),
}
}
/// Frames recovered from a frame stream, plus explicit damage.
///
/// Damage never discards the surrounding frames: a corrupt frame is reported
/// and skipped by its declared length, and the frames after it stay readable.
#[derive(Clone, PartialEq, Eq, Debug, Default)]
pub struct Recovery {
/// Frames recovered in stream order.
pub frames: Vec<Frame>,
/// Damage reports, each locating its frame header by byte offset.
/// Bounded repair and reporting scans retain at most their first report.
pub damage: Vec<SmithError>,
/// Total complete damaged frames observed, including reports not retained.
pub damage_count: u64,
/// Offset just past the last complete frame.
pub good_end: u64,
/// Offset where an incomplete trailing frame starts, if any.
pub torn_tail_at: Option<u64>,
}
/// Read every complete frame from `reader` with per-frame bounded allocation.
///
/// A torn tail ends the walk and is reported through
/// [`Recovery::torn_tail_at`]; complete frames before it are always returned.
///
/// # Errors
/// Returns `FRAME_READ` when the reader fails for a reason other than EOF.
pub fn read_frames<R: Read>(reader: &mut R, frame_bytes_max: usize) -> Result<Recovery> {
let mut frames = Vec::new();
let mut recovery = visit_frames(reader, frame_bytes_max, |frame| {
frames.push(frame);
true
})?;
recovery.frames = frames;
Ok(recovery)
}
/// Walk complete frames with one bounded buffer, without retaining earlier frames.
///
/// Returning `false` from `visit` ends the walk at the last complete boundary.
/// Damage and incomplete tails are reported separately from the visited records.
/// Read failures other than EOF return explicitly, never as torn tails.
///
/// # Errors
/// Returns `FRAME_READ` when the reader fails for a reason other than EOF.
pub fn visit_frames<R: Read>(
reader: &mut R,
frame_bytes_max: usize,
visit: impl FnMut(Frame) -> bool,
) -> Result<Recovery> {
let mut damage = Vec::new();
let mut recovery = visit_frames_reporting(reader, frame_bytes_max, visit, |fault| {
damage.push(fault);
true
})?;
recovery.damage = damage;
Ok(recovery)
}
/// Walk frames while reporting damage immediately instead of retaining it.
///
/// Returning `false` from either callback ends the walk at the last complete
/// boundary. The returned recovery never retains damage, so streaming callers
/// can fail or emit each fault with bounded memory.
///
/// # Errors
/// Returns `FRAME_READ` when the reader fails for a reason other than EOF.
pub fn visit_frames_reporting<R: Read>(
reader: &mut R,
frame_bytes_max: usize,
mut visit: impl FnMut(Frame) -> bool,
mut report_damage: impl FnMut(SmithError) -> bool,
) -> Result<Recovery> {
let mut recovery = Recovery::default();
let declared_limit = frame_bytes_max.saturating_add(BODY_PREFIX_LEN);
loop {
let offset = recovery.good_end;
let mut header = [0u8; HEADER_LEN];
match read_exact_or_eof(reader, &mut header) {
Ok(true) => {}
Ok(false) => break,
Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => {
recovery.torn_tail_at = Some(offset);
break;
}
Err(error) => return Err(read_error(&error)),
}
let header_crc =
u32::from_le_bytes(header[8..12].try_into().map_err(|_| boundary(offset))?);
if header_crc != crc32fast::hash(&header[..8]) {
return Err(boundary(offset));
}
let declared =
u32::from_le_bytes(header[..4].try_into().map_err(|_| boundary(offset))?) as usize;
let kind = u16::from_le_bytes(header[4..6].try_into().map_err(|_| boundary(offset))?);
let version = u16::from_le_bytes(header[6..8].try_into().map_err(|_| boundary(offset))?);
let total = HEADER_LEN as u64 + declared as u64;
if declared > declared_limit {
match skip_exact(reader, declared) {
Ok(true) => {}
Ok(false) => {
recovery.torn_tail_at = Some(offset);
break;
}
Err(error) => return Err(read_error(&error)),
}
recovery.good_end = offset + total;
recovery.damage_count += 1;
if !report_damage(located(
FrameFault::TooLarge {
bytes_max: frame_bytes_max,
actual: declared.saturating_sub(BODY_PREFIX_LEN),
},
offset,
)) {
break;
}
continue;
}
let mut body = vec![0u8; declared];
match reader.read_exact(&mut body) {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => {
recovery.torn_tail_at = Some(offset);
break;
}
Err(error) => return Err(read_error(&error)),
}
let decoded = decode_body(kind, version, &body);
recovery.good_end = offset + total;
match decoded {
Ok(frame) => {
if !visit(frame) {
break;
}
}
Err(f) => {
recovery.damage_count += 1;
if !report_damage(located(f, offset)) {
break;
}
}
}
}
Ok(recovery)
}
/// Visit boundaries and fixed metadata with bounded checksum chunks.
///
/// Known frames carry empty content; read validated offsets for payloads.
///
/// # Errors
/// Returns an I/O or boundary error; damaged complete frames are reported.
pub fn visit_metadata_reporting(
reader: &mut impl Read,
max: usize,
mut visit: impl FnMut(u64, Frame) -> Result<()>,
mut damage: impl FnMut(SmithError) -> Result<()>,
) -> Result<Recovery> {
let mut recovery = Recovery::default();
let mut chunk = [0u8; 8192];
'frames: loop {
let at = recovery.good_end;
let mut header = [0; HEADER_LEN];
match read_exact_or_eof(reader, &mut header) {
Ok(true) => {}
Ok(false) => break,
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
recovery.torn_tail_at = Some(at);
break;
}
Err(e) => return Err(read_error(&e)),
}
if crc32fast::hash(&header[..8])
!= u32::from_le_bytes(header[8..12].try_into().map_err(|_| boundary(at))?)
{
return Err(boundary(at));
}
let len = u32::from_le_bytes(header[..4].try_into().map_err(|_| boundary(at))?) as usize;
let kind = u16::from_le_bytes(header[4..6].try_into().map_err(|_| boundary(at))?);
let version = u16::from_le_bytes(header[6..8].try_into().map_err(|_| boundary(at))?);
if len < BODY_PREFIX_LEN || len > max.saturating_add(BODY_PREFIX_LEN) {
let skipped = std::io::copy(&mut reader.take(len as u64), &mut std::io::sink())
.map_err(|e| read_error(&e))?;
if skipped != len as u64 {
recovery.torn_tail_at = Some(at);
break;
}
recovery.good_end += HEADER_LEN as u64 + len as u64;
metadata_fault(
&mut recovery,
&mut damage,
FrameFault::TooLarge {
bytes_max: max,
actual: len.saturating_sub(BODY_PREFIX_LEN),
},
at,
)?;
continue;
}
let mut checksum = [0; 4];
if !read_part(reader, &mut checksum, &mut recovery)? {
break;
}
let expected = u32::from_le_bytes(checksum);
let payload_len = len - BODY_PREFIX_LEN;
let fixed = match (kind, version) {
(KIND_ENTRY, VERSION_ENTRY) => payload_len.min(ENTRY_METADATA_LEN),
(KIND_SESSION_HEADER, VERSION_SESSION_HEADER) => payload_len.min(32),
(KIND_INDEX, VERSION_NAV) => payload_len.min(48),
(KIND_SELECTION, VERSION_NAV) => payload_len.min(16),
(KIND_CHECKPOINT, VERSION_NAV) => payload_len.min(24),
_ => 0,
};
let mut prefix = [0u8; ENTRY_METADATA_LEN];
if !read_part(reader, &mut prefix[..fixed], &mut recovery)? {
break;
}
let mut hasher = crc32fast::Hasher::new();
hasher.update(&header[4..8]);
hasher.update(&prefix[..fixed]);
let mut left = payload_len - fixed;
while left > 0 {
let n = metadata_chunk_size(left, chunk.len());
if !read_part(reader, &mut chunk[..n], &mut recovery)? {
break 'frames;
}
hasher.update(&chunk[..n]);
left -= n;
}
let mut marker = [0; 4];
if !read_part(reader, &mut marker, &mut recovery)? {
break;
}
recovery.good_end += HEADER_LEN as u64 + len as u64;
if hasher.finalize() != expected || u32::from_le_bytes(marker) != expected ^ COMPLETION {
metadata_fault(
&mut recovery,
&mut damage,
FrameFault::Checksum {
declared_length: len,
},
at,
)?;
continue;
}
match decode_metadata_frame(
kind,
version,
payload_len,
&prefix[..fixed],
checksum,
marker,
) {
Ok(frame) => visit(at, frame)?,
Err(fault) => metadata_fault(&mut recovery, &mut damage, fault, at)?,
}
}
Ok(recovery)
}
fn metadata_chunk_size(left: usize, buffer_len: usize) -> usize {
let size = left.min(buffer_len);
assert!(
size > 0,
"nonempty metadata remainder yields a positive chunk"
);
size
}
fn metadata_fault(
recovery: &mut Recovery,
report: &mut impl FnMut(SmithError) -> Result<()>,
fault: FrameFault,
at: u64,
) -> Result<()> {
recovery.damage_count += 1;
report(located(fault, at))
}
fn read_part(reader: &mut impl Read, bytes: &mut [u8], recovery: &mut Recovery) -> Result<bool> {
match reader.read_exact(bytes) {
Ok(()) => Ok(true),
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
recovery.torn_tail_at = Some(recovery.good_end);
Ok(false)
}
Err(e) => Err(read_error(&e)),
}
}
fn decode_metadata_frame(
kind: u16,
version: u16,
payload_len: usize,
prefix: &[u8],
checksum: [u8; 4],
marker: [u8; 4],
) -> std::result::Result<Frame, FrameFault> {
if kind == KIND_ENTRY && version == VERSION_ENTRY {
if prefix.len() != ENTRY_METADATA_LEN {
return Err(schema("short entry metadata"));
}
let actual = u32::from_le_bytes(
prefix[58..62]
.try_into()
.map_err(|_| schema("entry length"))?,
) as usize;
if actual != payload_len - ENTRY_METADATA_LEN {
return Err(schema("entry content length mismatch"));
}
let mut metadata = [0; ENTRY_METADATA_LEN];
metadata.copy_from_slice(prefix);
metadata[58..62].fill(0);
return decode_entry(&metadata).map(|entry| Frame::Known { entry });
}
if !prefix.is_empty() && prefix.len() == payload_len {
return decode_body(
kind,
version,
&[&checksum[..], prefix, &marker[..]].concat(),
);
}
if matches!(
kind,
KIND_SESSION_HEADER | KIND_INDEX | KIND_SELECTION | KIND_CHECKPOINT
) {
Err(schema("navigation or header length"))
} else {
Ok(Frame::Unknown {
kind,
version,
payload: Vec::new(),
})
}
}
/// Write a full frame to a writer with an enforced payload size limit.
///
/// # Errors
///
/// Returns a framing error when encoding or writing fails.
pub fn write_frame(writer: &mut impl Write, frame: &Frame, frame_bytes_max: usize) -> Result<()> {
let bytes = frame.encode(frame_bytes_max)?;
writer.write_all(&bytes).map_err(|e| {
fault(FrameFault::Encode {
message: e.to_string(),
})
})?;
Ok(())
}
const fn fault(f: FrameFault) -> SmithError {
SmithError::Frame {
fault: f,
offset: None,
}
}
const fn located(f: FrameFault, offset: u64) -> SmithError {
SmithError::Frame {
fault: f,
offset: Some(offset),
}
}
const fn too_large(bytes_max: usize, actual: usize) -> SmithError {
fault(FrameFault::TooLarge { bytes_max, actual })
}
fn boundary(offset: u64) -> SmithError {
SmithError::Session {
code: smith::error::SessionCode::FrameBoundary,
message: format!(
"untrustworthy frame header at byte {offset}; cannot locate the next record"
),
}
}
fn read_error(error: &std::io::Error) -> SmithError {
SmithError::Session {
code: smith::error::SessionCode::FrameRead,
message: error.to_string(),
}
}
/// Read exactly `buf.len()` bytes. `Ok(false)` means clean end of stream,
/// `Err` means a partial read, which is a torn tail.
fn read_exact_or_eof<R: Read>(reader: &mut R, buffer: &mut [u8]) -> std::io::Result<bool> {
let mut filled = 0;
let mut interrupted_reads = 0;
while filled < buffer.len() {
match reader.read(&mut buffer[filled..]) {
Ok(0) => {
if filled == 0 {
return Ok(false);
}
return Err(std::io::Error::new(
std::io::ErrorKind::UnexpectedEof,
"torn frame",
));
}
Ok(n) => {
assert!(n > 0, "successful nonempty read makes positive progress");
filled += n;
interrupted_reads = 0;
}
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {
if interrupted_reads == INTERRUPTED_READS_MAX {
return Err(e);
}
interrupted_reads += 1;
}
Err(e) => return Err(e),
}
}
Ok(true)
}
/// Discard `len` bytes through a bounded copy buffer. `false` means the
/// stream ended early; other read errors propagate.
fn skip_exact<R: Read>(reader: &mut R, len: usize) -> std::io::Result<bool> {
let len = len as u64;
std::io::copy(&mut reader.take(len), &mut std::io::sink()).map(|skipped| skipped == len)
}
/// CRC-32 (IEEE 802.3, reflected) over kind, version, and payload.
///
/// `crc32fast` is pure Rust with runtime CPU dispatch (SSE4.2 PCLMULQDQ
/// where present); the table loop it replaces measured 60-80x slower on
/// large frames, and this checksum covers every frame twice.
fn crc32(kind: u16, version: u16, payload: &[u8]) -> u32 {
let mut hasher = crc32fast::Hasher::new();
hasher.update(&kind.to_le_bytes());
hasher.update(&version.to_le_bytes());
hasher.update(payload);
hasher.finalize()
}
#[cfg(test)]
#[expect(
clippy::unwrap_used,
clippy::panic,
reason = "tests may panic on invariant violations"
)]
mod tests {
use super::*;
use crate::session::{EntryContent, EntryKind, Session};
use smith::id::ToolCallId;
use smith::message::{ContentBlock, Message, Role};
use smith::structured::ToolArguments;
fn message_frame(message: Message) -> EntryFrame {
EntryFrame::new(None, &EntryContent::Message(message)).unwrap()
}
fn sample_entry(thinking: bool) -> EntryFrame {
let mut msg = Message::with_text(Role::User, "hello");
if thinking {
msg.add_block(ContentBlock::thinking("internal"));
}
message_frame(msg)
}
fn known_bytes(text: &str) -> Vec<u8> {
Frame::Known {
entry: message_frame(Message::with_text(Role::User, text)),
}
.encode(DEFAULT_MAX_FRAME_BYTES)
.unwrap()
}
fn text_of(frame: &Frame) -> String {
match frame {
Frame::Known { entry } => match EntryContent::decode(&entry.content).unwrap() {
EntryContent::Message(m) => match &m.blocks[0] {
ContentBlock::Text(t) => t.clone(),
other => panic!("unexpected block {other:?}"),
},
other => panic!("unexpected content {other:?}"),
},
other => panic!("unexpected frame {other:?}"),
}
}
fn read_all(bytes: &[u8], frame_bytes_max: usize) -> Recovery {
read_frames(&mut &bytes[..], frame_bytes_max).unwrap()
}
struct FailingReader;
impl Read for FailingReader {
fn read(&mut self, _: &mut [u8]) -> std::io::Result<usize> {
Err(std::io::Error::from(std::io::ErrorKind::Other))
}
}
struct CountedBrokenReader {
reads: usize,
}
impl Read for CountedBrokenReader {
fn read(&mut self, _: &mut [u8]) -> std::io::Result<usize> {
self.reads += 1;
Err(std::io::ErrorKind::BrokenPipe.into())
}
}
struct AlwaysInterruptedReader {
reads: usize,
}
impl Read for AlwaysInterruptedReader {
fn read(&mut self, _: &mut [u8]) -> std::io::Result<usize> {
self.reads += 1;
Err(std::io::ErrorKind::Interrupted.into())
}
}
struct InterruptedOnceThenData {
interrupted: bool,
bytes: &'static [u8],
}
impl Read for InterruptedOnceThenData {
fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
if !self.interrupted {
self.interrupted = true;
return Err(std::io::ErrorKind::Interrupted.into());
}
let count = buffer.len().min(self.bytes.len());
buffer[..count].copy_from_slice(&self.bytes[..count]);
self.bytes = &self.bytes[count..];
Ok(count)
}
}
#[test]
fn non_interruption_errors_propagate_without_retry() {
let mut reader = CountedBrokenReader { reads: 0 };
let error = read_exact_or_eof(&mut reader, &mut [0; 1]).unwrap_err();
assert_eq!(error.kind(), std::io::ErrorKind::BrokenPipe);
assert_eq!(reader.reads, 1);
}
#[test]
fn repeated_interruptions_are_bounded_and_propagated() {
let mut reader = AlwaysInterruptedReader { reads: 0 };
let error = read_exact_or_eof(&mut reader, &mut [0; 1]).unwrap_err();
assert_eq!(error.kind(), std::io::ErrorKind::Interrupted);
assert_eq!(reader.reads, INTERRUPTED_READS_MAX + 1);
}
#[test]
fn metadata_reader_accepts_frame_with_no_variable_payload_remainder() {
let frame = Frame::Unknown {
kind: u16::MAX,
version: u16::MAX,
payload: Vec::new(),
};
let bytes = frame.encode(DEFAULT_MAX_FRAME_BYTES).unwrap();
let mut visited = Vec::new();
let recovery = visit_metadata_reporting(
&mut bytes.as_slice(),
DEFAULT_MAX_FRAME_BYTES,
|offset, frame| {
visited.push((offset, frame));
Ok(())
},
|_| Ok(()),
)
.unwrap();
assert_eq!(visited, vec![(0, frame)]);
assert_eq!(recovery.good_end, bytes.len() as u64);
}
#[test]
fn interruption_before_data_retries_successfully() {
let mut reader = InterruptedOnceThenData {
interrupted: false,
bytes: b"ok",
};
let mut bytes = [0; 2];
assert!(read_exact_or_eof(&mut reader, &mut bytes).unwrap());
assert_eq!(&bytes, b"ok");
}
#[test]
fn read_errors_are_not_torn_tails() {
let error =
visit_frames(&mut FailingReader, DEFAULT_MAX_FRAME_BYTES, |_| true).unwrap_err();
assert_eq!(error.code(), "FRAME_READ");
}
#[test]
fn corrupt_frame_is_reported_and_neighbours_survive() {
let mut bytes = known_bytes("before");
let corrupt_at = bytes.len() as u64;
let mut corrupt = known_bytes("corrupt");
// Flip one payload byte: the length prefix stays intact, the checksum fails.
let last = corrupt.len() - 1;
corrupt[last] ^= 0xFF;
bytes.extend_from_slice(&corrupt);
bytes.extend_from_slice(&known_bytes("after"));
let recovery = read_all(&bytes, DEFAULT_MAX_FRAME_BYTES);
let texts: Vec<String> = recovery.frames.iter().map(text_of).collect();
assert_eq!(texts, vec!["before".to_string(), "after".to_string()]);
assert_eq!(recovery.damage.len(), 1);
assert_eq!(recovery.damage[0].code(), "FRAME_CHECKSUM");
assert_eq!(
recovery.damage[0],
SmithError::Frame {
fault: FrameFault::Checksum {
declared_length: corrupt.len() - HEADER_LEN
},
offset: Some(corrupt_at),
}
);
assert_eq!(recovery.good_end, bytes.len() as u64);
assert_eq!(recovery.torn_tail_at, None);
}
#[test]
fn damaged_header_never_scans_payload_for_a_frame_boundary() {
let nested = known_bytes("spoofed boundary");
let mut bytes = Frame::Unknown {
kind: 99,
version: 1,
payload: nested,
}
.encode(DEFAULT_MAX_FRAME_BYTES)
.unwrap();
bytes[0] ^= 1;
bytes.extend_from_slice(&known_bytes("after"));
assert_eq!(
read_frames(&mut bytes.as_slice(), DEFAULT_MAX_FRAME_BYTES)
.unwrap_err()
.code(),
"FRAME_BOUNDARY"
);
}
#[test]
fn future_frame_round_trips_byte_for_byte() {
let future = Frame::Unknown {
kind: 0xBEEF,
version: 7,
payload: vec![0xDE, 0xAD, 0x01, 0x02],
};
let bytes = future.encode(DEFAULT_MAX_FRAME_BYTES).unwrap();
let recovery = read_all(&bytes, DEFAULT_MAX_FRAME_BYTES);
assert!(recovery.damage.is_empty());
assert_eq!(recovery.frames, vec![future]);
assert_eq!(
recovery.frames[0].encode(DEFAULT_MAX_FRAME_BYTES).unwrap(),
bytes
);
}
#[test]
fn known_frame_round_trips_through_fixed_metadata() {
let entry = sample_entry(true);
let bytes = Frame::Known {
entry: entry.clone(),
}
.encode(DEFAULT_MAX_FRAME_BYTES)
.unwrap();
let recovery = read_all(&bytes, DEFAULT_MAX_FRAME_BYTES);
assert_eq!(recovery.frames, vec![Frame::Known { entry }]);
assert_eq!(recovery.good_end, bytes.len() as u64);
}
#[test]
fn entry_frame_carries_fixed_metadata_and_raw_content() {
let call_id = ToolCallId::new();
let content = EntryContent::ToolResult {
ok: true,
output: "done".to_string(),
call_id,
};
let entry = EntryFrame::new(None, &content).unwrap();
assert_eq!(entry.kind, EntryKind::ToolResult { call_id, ok: true });
assert_eq!(EntryContent::decode(&entry.content).unwrap(), content);
let bytes = Frame::Known {
entry: entry.clone(),
}
.encode(DEFAULT_MAX_FRAME_BYTES)
.unwrap();
assert_eq!(
&bytes[FRAME_PREFIX_LEN + ENTRY_METADATA_LEN..bytes.len() - 4],
entry.content
);
assert_eq!(
u32::from_le_bytes(
bytes[FRAME_PREFIX_LEN + 58..FRAME_PREFIX_LEN + 62]
.try_into()
.unwrap()
) as usize,
entry.content.len()
);
// A session record frames straight from arena bytes, identically.
let mut session = Session::new();
let record = *session.append(&content).unwrap();
let from_arena =
encode_entry(&record, session.bytes(&record), DEFAULT_MAX_FRAME_BYTES).unwrap();
let owned = Frame::Known {
entry: EntryFrame {
id: record.id,
parent: record.parent,
timestamp_ms: record.timestamp_ms,
kind: record.kind,
content: session.bytes(&record).to_vec(),
},
};
assert_eq!(from_arena, owned.encode(DEFAULT_MAX_FRAME_BYTES).unwrap());
}
#[test]
fn torn_tail_is_located_and_earlier_frames_kept() {
let mut bytes = known_bytes("kept");
let tail_at = bytes.len() as u64;
bytes.extend_from_slice(&known_bytes("lost")[..6]);
let recovery = read_all(&bytes, DEFAULT_MAX_FRAME_BYTES);
assert_eq!(
recovery.frames.iter().map(text_of).collect::<Vec<_>>(),
vec!["kept".to_string()]
);
assert_eq!(recovery.torn_tail_at, Some(tail_at));
assert_eq!(recovery.good_end, tail_at);
// A partial length prefix is torn too, never a fabricated header.
let mut short = known_bytes("kept");
let short_tail = short.len() as u64;
short.extend_from_slice(&[0xFF, 0xFF]);
let recovery = read_all(&short, DEFAULT_MAX_FRAME_BYTES);
assert_eq!(recovery.frames.len(), 1);
assert_eq!(recovery.torn_tail_at, Some(short_tail));
}
#[test]
fn oversized_complete_frame_is_skipped_by_length() {
let mut bytes = Frame::Unknown {
kind: 9,
version: 1,
payload: vec![0xAB; 5000],
}
.encode(6000)
.unwrap();
let skipped = bytes.len() as u64;
bytes.extend_from_slice(&known_bytes("tail"));
let recovery = read_all(&bytes, 1024);
assert_eq!(
recovery.frames.iter().map(text_of).collect::<Vec<_>>(),
vec!["tail".to_string()]
);
assert_eq!(recovery.damage.len(), 1);
assert_eq!(recovery.damage[0].code(), "FRAME_TOO_LARGE");
assert_eq!(
recovery.damage[0],
SmithError::Frame {
fault: FrameFault::TooLarge {
bytes_max: 1024,
actual: 5000
},
offset: Some(0),
}
);
assert_eq!(recovery.good_end, bytes.len() as u64);
assert_eq!(skipped, (HEADER_LEN + BODY_PREFIX_LEN + 5000) as u64);
}
#[test]
fn oversized_truncated_frame_is_torn_not_scanned() {
let mut bytes = Frame::Unknown {
kind: 9,
version: 1,
payload: vec![0; 200],
}
.encode(300)
.unwrap();
bytes.truncate(HEADER_LEN + 10);
let recovery = read_all(&bytes, 50);
assert!(recovery.frames.is_empty());
assert!(recovery.damage.is_empty());
assert_eq!(recovery.torn_tail_at, Some(0));
}
#[test]
fn declared_length_below_body_prefix_is_malformed() {
let mut bytes = Frame::Unknown {
kind: 9,
version: 1,
payload: vec![],
}
.encode(100)
.unwrap();
bytes[..4].copy_from_slice(&3_u32.to_le_bytes());
let checksum = crc32fast::hash(&bytes[..8]);
bytes[8..12].copy_from_slice(&checksum.to_le_bytes());
bytes.truncate(HEADER_LEN + 3);
bytes.extend_from_slice(&known_bytes("after"));
let recovery = read_all(&bytes, DEFAULT_MAX_FRAME_BYTES);
assert_eq!(recovery.damage.len(), 1);
assert_eq!(recovery.damage[0].code(), "FRAME_MALFORMED");
assert_eq!(recovery.frames.len(), 1);
}
#[test]
fn future_entry_version_is_preserved_verbatim() {
let payload = vec![0xA0]; // empty CBOR map
let frame = Frame::Unknown {
kind: KIND_ENTRY,
version: VERSION_ENTRY + 1,
payload,
};
let bytes = frame.encode(DEFAULT_MAX_FRAME_BYTES).unwrap();
let recovery = read_all(&bytes, DEFAULT_MAX_FRAME_BYTES);
assert!(recovery.damage.is_empty());
assert_eq!(recovery.frames, vec![frame]);
assert_eq!(
recovery.frames[0].encode(DEFAULT_MAX_FRAME_BYTES).unwrap(),
bytes
);
}
#[test]
fn checksum_valid_entry_with_bad_schema_is_schema_damage() {
let frame = Frame::Unknown {
kind: KIND_ENTRY,
version: VERSION_ENTRY,
payload: vec![0xA0], // valid CBOR, not an Entry
};
let bytes = frame.encode(DEFAULT_MAX_FRAME_BYTES).unwrap();
let recovery = read_all(&bytes, DEFAULT_MAX_FRAME_BYTES);
assert!(recovery.frames.is_empty());
assert_eq!(recovery.damage[0].code(), "FRAME_SCHEMA");
}
#[test]
fn old_frames_without_header_checksum_are_rejected_as_ambiguous() {
// Old layout: [len][CBOR payload] with no kind, version, or checksum.
let entry = sample_entry(false);
let payload = Frame::Known { entry }
.encode(DEFAULT_MAX_FRAME_BYTES)
.unwrap();
let mut bytes = u32::try_from(payload.len()).unwrap().to_le_bytes().to_vec();
bytes.extend_from_slice(&payload);
assert_eq!(
read_frames(&mut bytes.as_slice(), DEFAULT_MAX_FRAME_BYTES)
.unwrap_err()
.code(),
"FRAME_BOUNDARY"
);
}
#[test]
fn sequential_frames_consume_the_whole_stream() {
let mut bytes = known_bytes("one");
bytes.extend_from_slice(&known_bytes("two"));
let recovery = read_all(&bytes, DEFAULT_MAX_FRAME_BYTES);
assert_eq!(
recovery.frames.iter().map(text_of).collect::<Vec<_>>(),
vec!["one".to_string(), "two".to_string()]
);
assert_eq!(recovery.good_end, bytes.len() as u64);
}
#[test]
fn encoding_rejects_frames_above_limit() {
let big = message_frame({
let mut m = Message::with_text(Role::User, "x");
m.add_block(ContentBlock::text("y".repeat(4096)));
m
});
let error = Frame::Known { entry: big }.encode(64).unwrap_err();
assert!(
matches!(error, SmithError::Frame { fault: FrameFault::TooLarge { bytes_max: 64, actual }, .. } if actual > 64)
);
}
#[test]
fn default_ceiling_admits_one_mebibyte_payloads_and_no_more() {
let frame = |len| Frame::Unknown {
kind: 9,
version: 1,
payload: vec![0; len],
};
let at_ceiling = frame(1 << 20).encode(DEFAULT_MAX_FRAME_BYTES).unwrap();
assert_eq!(at_ceiling.len(), HEADER_LEN + BODY_PREFIX_LEN + (1 << 20));
assert_eq!(
frame((1 << 20) + 1).encode(DEFAULT_MAX_FRAME_BYTES),
Err(too_large(1 << 20, (1 << 20) + 1))
);
}
#[test]
fn equivalent_inputs_encode_identically_despite_key_order() {
let a = serde_json::json!({"path": "p", "content": {"b": 1, "a": 2}});
let b = serde_json::json!({"content": {"a": 2, "b": 1}, "path": "p"});
let call_id = ToolCallId::new();
let e1 = EntryFrame::new(
None,
&EntryContent::ToolCall {
name: "write".into(),
input: ToolArguments::from_serializable(&a).unwrap(),
call_id,
},
)
.unwrap();
let mut e2 = EntryFrame::new(
None,
&EntryContent::ToolCall {
name: "write".into(),
input: ToolArguments::from_serializable(&b).unwrap(),
call_id,
},
)
.unwrap();
// Force identical ids/timestamps so only key order could differ.
e2.id = e1.id;
e2.timestamp_ms = e1.timestamp_ms;
let b1 = Frame::Known { entry: e1 }.encode(1 << 20).unwrap();
let b2 = Frame::Known { entry: e2 }.encode(1 << 20).unwrap();
assert_eq!(b1, b2);
}
#[test]
fn navigation_frames_round_trip_and_fixed_index_matches_heap_encoder() {
let id = EntryId::from_u128(13);
let parent = EntryId::from_u128(7);
let index = Frame::Index {
id,
offset: 4096,
parent: Some(parent),
parent_offset: Some(128),
};
let bytes = index.encode(DEFAULT_MAX_FRAME_BYTES).unwrap();
assert_eq!(bytes, index_bytes(id, 4096, Some(parent), Some(128)));
let selection = Frame::Selection { id };
let mut stream = bytes;
stream.extend(selection.encode(DEFAULT_MAX_FRAME_BYTES).unwrap());
let checkpoint = Frame::Checkpoint { id, offset: 4096 };
stream.extend(checkpoint.encode(DEFAULT_MAX_FRAME_BYTES).unwrap());
assert_eq!(
read_all(&stream, DEFAULT_MAX_FRAME_BYTES).frames,
vec![index, selection, checkpoint]
);
}
#[test]
fn header_frame_round_trips() {
let header = SessionHeader {
session_id: smith::id::SessionId::new(),
branch_id: smith::id::BranchId::new(),
};
let bytes = Frame::Header { header }
.encode(DEFAULT_MAX_FRAME_BYTES)
.unwrap();
let recovery = read_all(&bytes, DEFAULT_MAX_FRAME_BYTES);
assert_eq!(recovery.frames, vec![Frame::Header { header }]);
assert!(recovery.damage.is_empty());
}
#[test]
fn checksum_matches_known_reference_value() {
// CRC-32 of "123456789" is a published check value.
assert_eq!(crc32(0x3231, 0x3433, b"56789"), 0xCBF4_3926);
}
}
/// Frame stream round-trip law (`SMH-SPEC-SPEC0001`, Sessions).
#[cfg(test)]
mod properties {
use super::*;
use crate::session::EntryKind;
use crate::session::properties::role;
use proptest::collection::vec;
use proptest::prelude::*;
use smith::id::{BranchId, EntryId, MessageId, SessionId, ToolCallId};
fn kind() -> impl Strategy<Value = EntryKind> {
prop_oneof![
(any::<u128>(), role()).prop_map(|(id, role)| EntryKind::Message {
id: MessageId::from_u128(id),
role,
}),
any::<u128>().prop_map(|id| EntryKind::ToolCall {
call_id: ToolCallId::from_u128(id),
}),
(any::<u128>(), any::<bool>()).prop_map(|(id, ok)| EntryKind::ToolResult {
call_id: ToolCallId::from_u128(id),
ok,
}),
any::<bool>().prop_map(|compaction| EntryKind::Meta { compaction }),
]
}
/// Any frame this build writes: header, entry with opaque content bytes,
/// or a frame of a kind it does not know.
fn frame() -> impl Strategy<Value = Frame> {
prop_oneof![
(any::<u128>(), any::<u128>()).prop_map(|(session, branch)| Frame::Header {
header: SessionHeader {
session_id: SessionId::from_u128(session),
branch_id: BranchId::from_u128(branch),
},
}),
(
any::<u128>(),
proptest::option::of(any::<u128>()),
any::<i64>(),
kind(),
vec(any::<u8>(), 0..64),
)
.prop_map(|(id, parent, timestamp_ms, kind, content)| Frame::Known {
entry: EntryFrame {
id: EntryId::from_u128(id),
parent: parent.map(EntryId::from_u128),
timestamp_ms,
kind,
content,
},
}),
(
any::<u128>(),
any::<u64>(),
any::<u128>(),
any::<u64>(),
any::<bool>(),
)
.prop_map(|(id, offset, parent, parent_offset, has_parent)| {
Frame::Index {
id: EntryId::from_u128(id),
offset,
parent: has_parent.then(|| EntryId::from_u128(parent | 1)),
parent_offset: has_parent.then_some(parent_offset | 1),
}
}),
any::<u128>().prop_map(|id| Frame::Selection {
id: EntryId::from_u128(id),
}),
(any::<u128>(), any::<u64>()).prop_map(|(id, offset)| Frame::Checkpoint {
id: EntryId::from_u128(id),
offset,
}),
(
(any::<u16>(), any::<u16>()).prop_filter(
"known frame versions decode through their schemas",
|(kind, version)| {
!matches!(
(*kind, *version),
(KIND_ENTRY, VERSION_ENTRY)
| (KIND_SESSION_HEADER, VERSION_SESSION_HEADER)
| (KIND_INDEX, VERSION_NAV)
| (KIND_SELECTION, VERSION_NAV)
| (KIND_CHECKPOINT, VERSION_NAV)
)
},
),
vec(any::<u8>(), 0..64),
)
.prop_map(|((kind, version), payload)| Frame::Unknown {
kind,
version,
payload,
}),
]
}
proptest! {
#[test]
fn visitor_and_eager_reader_agree_on_every_frame(frames in vec(frame(), 0..8)) {
// SMH-SPEC-SPEC0001: Sessions, bounded sequential reads preserve semantics.
let mut bytes = Vec::new();
for frame in &frames {
bytes.extend(frame.encode(DEFAULT_MAX_FRAME_BYTES).map_err(|e| TestCaseError::fail(e.to_string()))?);
}
let eager = read_frames(&mut bytes.as_slice(), DEFAULT_MAX_FRAME_BYTES).unwrap();
let mut visited = Vec::new();
let scan = visit_frames(&mut bytes.as_slice(), DEFAULT_MAX_FRAME_BYTES, |frame| {
visited.push(frame);
true
}).unwrap();
prop_assert_eq!(visited, eager.frames);
prop_assert_eq!(scan.damage, eager.damage);
prop_assert_eq!(scan.good_end, eager.good_end);
prop_assert_eq!(scan.torn_tail_at, eager.torn_tail_at);
}
#[test]
fn encoded_frame_streams_decode_to_the_same_frames(frames in vec(frame(), 0..8)) {
let mut bytes = Vec::new();
for frame in &frames {
bytes.extend(
frame
.encode(DEFAULT_MAX_FRAME_BYTES)
.map_err(|e| TestCaseError::fail(e.to_string()))?,
);
}
let recovery = read_frames(&mut bytes.as_slice(), DEFAULT_MAX_FRAME_BYTES).unwrap();
prop_assert!(recovery.damage.is_empty(), "damage: {:?}", recovery.damage);
prop_assert_eq!(recovery.torn_tail_at, None);
prop_assert_eq!(recovery.good_end, bytes.len() as u64);
let mut again = Vec::new();
for frame in &recovery.frames {
again.extend(
frame
.encode(DEFAULT_MAX_FRAME_BYTES)
.map_err(|e| TestCaseError::fail(e.to_string()))?,
);
}
prop_assert_eq!(recovery.frames, frames);
prop_assert_eq!(again, bytes);
}
#[test]
fn frame_encoding_is_deterministic_and_zero_fills_absent_fields(frame in frame()) {
// SMH-SPEC-SPEC0001: Sessions, deterministic encoding and reserved bytes.
let first = frame
.encode(DEFAULT_MAX_FRAME_BYTES)
.map_err(|error| TestCaseError::fail(error.to_string()))?;
let second = frame
.encode(DEFAULT_MAX_FRAME_BYTES)
.map_err(|error| TestCaseError::fail(error.to_string()))?;
prop_assert_eq!(&first, &second);
prop_assert_eq!(&FILE_PROLOGUE[5..], &[0, 0, 0]);
match frame {
Frame::Known { ref entry } if entry.parent.is_none() => {
prop_assert_eq!(
&first[FRAME_PREFIX_LEN + 16..FRAME_PREFIX_LEN + 32],
&[0; 16]
);
}
Frame::Index { parent: None, .. } => {
prop_assert_eq!(&first[40..64], &[0; 24]);
}
_ => {}
}
}
#[test]
fn a_frame_exactly_at_the_limit_round_trips_and_one_byte_less_rejects_it(frame in frame()) {
let unlimited = frame
.encode(DEFAULT_MAX_FRAME_BYTES)
.map_err(|e| TestCaseError::fail(e.to_string()))?;
let payload_len = unlimited.len() - HEADER_LEN - BODY_PREFIX_LEN;
let declared = u32::from_le_bytes(unlimited[..4].try_into().unwrap());
prop_assert_eq!(declared as usize, BODY_PREFIX_LEN + payload_len);
let at_limit = frame
.encode(payload_len)
.map_err(|e| TestCaseError::fail(e.to_string()))?;
prop_assert_eq!(&at_limit, &unlimited);
let recovery = read_frames(&mut at_limit.as_slice(), payload_len).unwrap();
prop_assert!(recovery.damage.is_empty(), "damage: {:?}", recovery.damage);
prop_assert_eq!(recovery.frames, vec![frame.clone()]);
if let Some(below) = payload_len.checked_sub(1) {
let result = frame.encode(below);
prop_assert!(result.is_err(), "below-limit frame was accepted");
let recovery = read_frames(&mut at_limit.as_slice(), below).unwrap();
prop_assert!(recovery.frames.is_empty());
prop_assert_eq!(recovery.good_end, at_limit.len() as u64);
}
}
}
}