//! 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(&timestamp_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);
            }
        }
    }
}