fsqlite-mvcc 0.4.7

//! §5.10.6-5.10.8 History Compression, Intent Commutativity, and Merge Certificates.
//!
//! This module implements three tightly coupled subsystems:
//!
//! 1. **`PageHistory` Objects (§5.10.6):** Compressed version chains where the
//!    newest committed version is stored as a full page image. Older versions are
//!    currently stored as full images too. Intent and structured patches have
//!    lossless wire codecs, but are not generated automatically: semantic replay
//!    alone does not preserve every byte of a historical page. Encoded as ECS
//!    objects for repair and remote fetching.
//!
//! 2. **Intent Commutativity (§5.10.7):** Mazurkiewicz trace-monoid formalization
//!    of when intent operations commute. Defines the independence relation
//!    `I_intent`, column-level `UpdateExpression` refinement, the join-max
//!    exception for AUTOINCREMENT, and the canonical Foata normal form.
//!
//! 3. **Merge Certificates (§5.10.8):** Proof-carrying merge artifacts. Every
//!    commit accepted via the merge path produces a verifiable
//!    [`MergeCertificate`] containing op digests, footprint digest, canonical
//!    normal form, and post-state hashes. A circuit breaker disables SAFE
//!    merging on any verification failure.

use std::collections::{BTreeMap, BTreeSet};

use fsqlite_types::{
    BtreeRef, ColumnIdx, CommitSeq, IntentFootprint, IntentOp, IntentOpKind, PageNumber,
    RebaseExpr, SemanticKeyKind, SemanticKeyRef, SqliteValue, StructuralEffects,
};

use crate::physical_merge::StructuredPagePatch;

#[cfg(test)]
mod certificate_tests;
pub mod compact;
mod intent_codec;
mod structured_codec;

// ---------------------------------------------------------------------------
// §5.10.6: Compressed PageHistory
// ---------------------------------------------------------------------------

/// How an older version's content is stored in a compressed page history.
#[derive(Debug, Clone, PartialEq)]
pub enum CompressedVersionData {
    /// Full page image (used for the newest committed version).
    FullImage(Vec<u8>),
    /// Intent log patches: replay these against the next-newer version to
    /// reconstruct.
    IntentLogPatch(Vec<IntentOp>),
    /// Structured page patch: cell-level delta against the next-newer version.
    StructuredPatch(StructuredPagePatch),
}

/// A single version entry in a compressed page history chain.
#[derive(Debug, Clone, PartialEq)]
pub struct CompressedPageVersion {
    /// The commit sequence at which this version was created.
    pub commit_seq: CommitSeq,
    /// The version's data (full image or patch).
    pub data: CompressedVersionData,
}

/// Compressed page history: newest version is always a full image.
///
/// Older versions are full images in the current compressor. Caller-supplied
/// intent and structured patches have lossless, versioned payload codecs;
/// decoding a patch does not establish its replay eligibility.
///
/// This is the ECS-integrated representation from §5.10.6. Hot pages encode
/// their patch chains as these objects for bounded memory, repairability, and
/// remote fetching.
#[derive(Debug, Clone, PartialEq)]
pub struct CompressedPageHistory {
    /// The page number this history covers.
    pub pgno: PageNumber,
    /// Versions ordered newest-first. `versions[0]` MUST be `FullImage`.
    pub versions: Vec<CompressedPageVersion>,
}

/// Errors from page history compression and ECS encoding.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum HistoryCompressionError {
    /// No versions provided — cannot compress an empty history.
    EmptyHistory,
    /// The newest version must be a full page image.
    NewestNotFullImage,
    /// ECS decoding failed: buffer too short or corrupt.
    DecodeError(String),
}

impl std::fmt::Display for HistoryCompressionError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::EmptyHistory => f.write_str("cannot compress empty page history"),
            Self::NewestNotFullImage => f.write_str("newest version must be a full page image"),
            Self::DecodeError(msg) => write!(f, "ECS decode error: {msg}"),
        }
    }
}

impl std::error::Error for HistoryCompressionError {}

/// Compress a page history by keeping every retained version as a full image.
///
/// The input `full_images` must be ordered newest-first; each entry is
/// `(commit_seq, page_bytes)`. Patch-shaped variants remain part of the
/// serialized type, but this encoder does not emit them: reconstructing the
/// exact original image requires more than semantic cell-level replay.
///
/// # Errors
///
/// Returns [`HistoryCompressionError::EmptyHistory`] if `full_images` is empty.
#[allow(clippy::module_name_repetitions)]
pub fn compress_page_history(
    pgno: PageNumber,
    full_images: &[(CommitSeq, Vec<u8>)],
) -> Result<CompressedPageHistory, HistoryCompressionError> {
    if full_images.is_empty() {
        return Err(HistoryCompressionError::EmptyHistory);
    }

    let mut versions = Vec::with_capacity(full_images.len());

    // Newest version: full image.
    versions.push(CompressedPageVersion {
        commit_seq: full_images[0].0,
        data: CompressedVersionData::FullImage(full_images[0].1.clone()),
    });

    // Older versions: store as full images. A future optimization could
    // compute XOR diffs or structured cell-level patches between consecutive
    // versions, but that requires a reliable reconstruction path. Storing
    // full images guarantees zero data loss — any version can be retrieved
    // without depending on the integrity of other versions in the chain.
    //
    // Memory trade-off: N full images vs 1 full + (N-1) diffs. For typical
    // page sizes (4 KiB) and chain lengths (< 64), the difference is modest
    // and correctness takes priority over compression ratio.
    for &(seq, ref page_bytes) in &full_images[1..] {
        versions.push(CompressedPageVersion {
            commit_seq: seq,
            data: CompressedVersionData::FullImage(page_bytes.clone()),
        });
    }

    Ok(CompressedPageHistory { pgno, versions })
}

// ECS binary encoding: to_bytes / from_bytes
// Wire format:
//   [4] pgno (u32 LE)
//   [4] version_count (u32 LE)
//   For each version:
//     [8] commit_seq (u64 LE)
//     [1] data_tag: 0=FullImage, 1=IntentLogPatch, 2=StructuredPatch
//     [4] payload_len (u32 LE)
//     [payload_len] payload bytes

const TAG_FULL_IMAGE: u8 = 0;
const TAG_INTENT_LOG_PATCH: u8 = 1;
const TAG_STRUCTURED_PATCH: u8 = 2;

impl CompressedPageHistory {
    /// Encode this compressed page history to canonical ECS bytes.
    ///
    /// # Panics
    ///
    /// Panics if this history cannot be encoded losslessly. Use
    /// [`Self::try_to_bytes`] for caller-supplied histories or patch payloads.
    #[must_use]
    pub fn to_bytes(&self) -> Vec<u8> {
        self.try_to_bytes()
            .expect("page history must have a lossless wire representation")
    }

    /// Encode without dropping mutation evidence or truncating lengths.
    /// Full-image and empty-placeholder wire bytes remain unchanged. Nonempty
    /// structured and intent patches use versioned SPP1 and ILP1 payloads,
    /// respectively. Intent payloads preserve complete operations and footprints,
    /// not the separate digest-only encoding used by merge certificates.
    ///
    /// # Errors
    ///
    /// Returns an error for an empty history, a non-full newest version,
    /// unrepresentable lengths, oversized patches, unknown structural flags,
    /// or intent expressions exceeding the depth/node limits.
    pub fn try_to_bytes(&self) -> Result<Vec<u8>, HistoryCompressionError> {
        let newest = self
            .versions
            .first()
            .ok_or(HistoryCompressionError::EmptyHistory)?;
        if !matches!(newest.data, CompressedVersionData::FullImage(_)) {
            return Err(HistoryCompressionError::NewestNotFullImage);
        }
        let wire_len = |length: usize| {
            u32::try_from(length).map_err(|_| {
                HistoryCompressionError::DecodeError("history length exceeds u32".to_owned())
            })
        };
        let mut buf = Vec::with_capacity(128);
        buf.extend_from_slice(&self.pgno.get().to_le_bytes());
        let version_count = wire_len(self.versions.len())?;
        buf.extend_from_slice(&version_count.to_le_bytes());

        for v in &self.versions {
            buf.extend_from_slice(&v.commit_seq.get().to_le_bytes());
            match &v.data {
                CompressedVersionData::FullImage(img) => {
                    buf.push(TAG_FULL_IMAGE);
                    let len = wire_len(img.len())?;
                    buf.extend_from_slice(&len.to_le_bytes());
                    buf.extend_from_slice(img);
                }
                CompressedVersionData::IntentLogPatch(ops) => {
                    buf.push(TAG_INTENT_LOG_PATCH);
                    let payload = intent_codec::encode(ops)?;
                    let len = wire_len(payload.len())?;
                    buf.extend_from_slice(&len.to_le_bytes());
                    buf.extend_from_slice(&payload);
                }
                CompressedVersionData::StructuredPatch(patch) => {
                    buf.push(TAG_STRUCTURED_PATCH);
                    let payload = structured_codec::encode(patch)?;
                    buf.extend_from_slice(&wire_len(payload.len())?.to_le_bytes());
                    buf.extend_from_slice(&payload);
                }
            }
        }
        Ok(buf)
    }

    /// Decode a compressed page history from canonical ECS bytes.
    ///
    /// # Errors
    ///
    /// Returns [`HistoryCompressionError::DecodeError`] on malformed input.
    pub fn from_bytes(data: &[u8]) -> Result<Self, HistoryCompressionError> {
        let err = |msg: &str| HistoryCompressionError::DecodeError(msg.to_owned());

        if data.len() < 8 {
            return Err(err("buffer too short for header"));
        }
        let pgno_raw = u32::from_le_bytes(
            data[..4]
                .try_into()
                .map_err(|_| err("pgno decode failed"))?,
        );
        let pgno = PageNumber::new(pgno_raw).ok_or_else(|| err("page number 0 is invalid"))?;
        let version_count = u32::from_le_bytes(
            data[4..8]
                .try_into()
                .map_err(|_| err("version count decode failed"))?,
        );
        let version_count = usize::try_from(version_count)
            .map_err(|_| err("version count exceeds addressable size"))?;
        let remaining = data.len() - 8;
        let max_possible_versions = remaining / 13;
        if version_count > max_possible_versions {
            return Err(err("truncated version entry"));
        }

        let mut offset = 8usize;
        let mut versions = Vec::with_capacity(version_count);

        for _ in 0..version_count {
            if offset + 9 > data.len() {
                return Err(err("truncated version entry"));
            }
            let commit_seq = CommitSeq::new(u64::from_le_bytes(
                data[offset..offset + 8]
                    .try_into()
                    .map_err(|_| err("commit_seq decode"))?,
            ));
            offset += 8;

            let tag = data[offset];
            offset += 1;

            if offset + 4 > data.len() {
                return Err(err("truncated payload length"));
            }
            let payload_len = u32::from_le_bytes(
                data[offset..offset + 4]
                    .try_into()
                    .map_err(|_| err("payload_len decode"))?,
            ) as usize;
            offset += 4;

            let payload_end = offset
                .checked_add(payload_len)
                .ok_or_else(|| err("payload length overflow"))?;
            if payload_end > data.len() {
                return Err(err("truncated payload"));
            }
            let payload = &data[offset..payload_end];
            offset = payload_end;

            let version_data = match tag {
                TAG_FULL_IMAGE => CompressedVersionData::FullImage(payload.to_vec()),
                TAG_INTENT_LOG_PATCH => {
                    CompressedVersionData::IntentLogPatch(intent_codec::decode(payload)?)
                }
                TAG_STRUCTURED_PATCH => {
                    CompressedVersionData::StructuredPatch(structured_codec::decode(payload)?)
                }
                other => return Err(err(&format!("unknown version data tag: {other}"))),
            };

            versions.push(CompressedPageVersion {
                commit_seq,
                data: version_data,
            });
        }

        if offset != data.len() {
            return Err(err("trailing bytes after compressed page history"));
        }

        let Some(newest) = versions.first() else {
            return Err(HistoryCompressionError::EmptyHistory);
        };
        if !matches!(newest.data, CompressedVersionData::FullImage(_)) {
            return Err(HistoryCompressionError::NewestNotFullImage);
        }

        Ok(Self { pgno, versions })
    }
}

/// Legacy digest stream retained only to test that it is not decoded as history.
#[cfg(test)]
fn canonical_intent_ops_bytes(ops: &[IntentOp]) -> Vec<u8> {
    let mut buf = Vec::with_capacity(ops.len() * 32);
    #[allow(clippy::cast_possible_truncation)]
    let count = ops.len() as u32;
    buf.extend_from_slice(&count.to_le_bytes());
    for op in ops {
        buf.extend_from_slice(&canonical_intent_bytes(op));
    }
    buf
}

// ---------------------------------------------------------------------------
// §5.10.7: Intent Commutativity (Trace-Normalized Merge)
// ---------------------------------------------------------------------------

/// Compute the stable op digest for an intent operation.
///
/// `op_digest := Trunc128(BLAKE3("fsqlite:intent:v2" || canonical_intent_bytes))`
/// V2 binds the complete semantic-key identity, not only its supplied digest.
#[must_use]
pub fn compute_op_digest(op: &IntentOp) -> [u8; 16] {
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"fsqlite:intent:v2");
    let canonical = canonical_intent_bytes(op);
    hasher.update(&canonical);
    let hash = hasher.finalize();
    let mut digest = [0u8; 16];
    digest.copy_from_slice(&hash.as_bytes()[..16]);
    digest
}

/// Compute the footprint digest over a set of intent footprints.
///
/// Digests all reads, writes, and structural effects from every footprint.
#[must_use]
pub fn compute_footprint_digest(footprints: &[&IntentFootprint]) -> [u8; 16] {
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"fsqlite:footprint:v2");
    for fp in footprints {
        // Encode reads.
        #[allow(clippy::cast_possible_truncation)]
        let reads_len = fp.reads.len() as u32;
        hasher.update(&reads_len.to_le_bytes());
        for r in &fp.reads {
            hasher.update(&canonical_semantic_key_bytes(r));
        }
        // Encode writes.
        #[allow(clippy::cast_possible_truncation)]
        let writes_len = fp.writes.len() as u32;
        hasher.update(&writes_len.to_le_bytes());
        for w in &fp.writes {
            hasher.update(&canonical_semantic_key_bytes(w));
        }
        // Encode structural effects.
        hasher.update(&fp.structural.bits().to_le_bytes());
    }
    let hash = hasher.finalize();
    let mut digest = [0u8; 16];
    digest.copy_from_slice(&hash.as_bytes()[..16]);
    digest
}

/// Check whether two intent operations are independent under the trace-monoid
/// independence relation `I_intent` (§5.10.7).
///
/// Two ops `(a, b)` are independent iff:
/// - `a.schema_epoch == b.schema_epoch`
/// - `a.footprint.structural == NONE` and `b.footprint.structural == NONE`
/// - `Writes(a) ∩ Writes(b) = ∅`
/// - `Writes(a) ∩ Reads(b) = ∅` and `Writes(b) ∩ Reads(a) = ∅`
///
/// With column-level refinement for `UpdateExpression` pairs on the same key,
/// and the join-max exception for AUTOINCREMENT. Expression reads participate
/// in the refinement; an explicit footprint read is never waived. Callers
/// must still establish deterministic expressions and schema/rebase eligibility.
#[must_use]
pub fn are_intent_ops_independent(a: &IntentOp, b: &IntentOp) -> bool {
    // Rule 1: Schema epochs must match.
    if a.schema_epoch != b.schema_epoch {
        return false;
    }

    // Rule 2: Both structural effects must be NONE.
    if a.footprint.structural != StructuralEffects::NONE
        || b.footprint.structural != StructuralEffects::NONE
    {
        return false;
    }

    // An operation necessarily writes its own key even if the supplied
    // footprint omits it. Empty footprints are not proof of distinct writes.
    let target_a = intent_target_digest(&a.op);
    let target_b = intent_target_digest(&b.op);
    let refinement = check_update_expression_pair(&a.op, &b.op);
    if refinement == Some(false) {
        return false;
    }
    let writes_a: BTreeSet<[u8; 16]> = a
        .footprint
        .writes
        .iter()
        .map(|key| key.key_digest)
        .chain(std::iter::once(target_a))
        .collect();
    let writes_b: BTreeSet<[u8; 16]> = b
        .footprint
        .writes
        .iter()
        .map(|key| key.key_digest)
        .chain(std::iter::once(target_b))
        .collect();

    // Refine ONLY the shared target-row write/write conflict. Additional
    // index/row writes and reads made before replay remain dependencies.
    writes_a
        .intersection(&writes_b)
        .all(|key| refinement == Some(true) && *key == target_a && target_a == target_b)
        && a.footprint
            .reads
            .iter()
            .all(|key| !writes_b.contains(&key.key_digest))
        && b.footprint
            .reads
            .iter()
            .all(|key| !writes_a.contains(&key.key_digest))
}

fn intent_target_digest(op: &IntentOpKind) -> [u8; 16] {
    match op {
        IntentOpKind::Insert { table, key, .. }
        | IntentOpKind::Delete { table, key }
        | IntentOpKind::Update { table, key, .. }
        | IntentOpKind::UpdateExpression { table, key, .. } => {
            table_row_key_digest(BtreeRef::Table(*table), key.get())
        }
        IntentOpKind::IndexInsert { index, key, .. }
        | IntentOpKind::IndexDelete { index, key, .. } => SemanticKeyRef::compute_digest(
            SemanticKeyKind::IndexEntry,
            BtreeRef::Index(*index),
            key,
        ),
    }
}

/// Refine same-row expression updates only; all other conflicts use key sets.
fn check_update_expression_pair(a: &IntentOpKind, b: &IntentOpKind) -> Option<bool> {
    let (
        IntentOpKind::UpdateExpression {
            table: ta,
            key: ka,
            column_updates: cols_a,
        },
        IntentOpKind::UpdateExpression {
            table: tb,
            key: kb,
            column_updates: cols_b,
        },
    ) = (a, b)
    else {
        return None;
    };
    if ta != tb || ka != kb {
        return None;
    }

    let written_a: BTreeSet<ColumnIdx> = cols_a.iter().map(|(column, _)| *column).collect();
    let written_b: BTreeSet<ColumnIdx> = cols_b.iter().map(|(column, _)| *column).collect();
    // Do not let find() select one convenient assignment while hiding another.
    if written_a.len() != cols_a.len() || written_b.len() != cols_b.len() {
        return Some(false);
    }
    let overlap: BTreeSet<ColumnIdx> = written_a.intersection(&written_b).copied().collect();
    if !cols_a
        .iter()
        .chain(cols_b.iter())
        .all(|(column, expr)| !overlap.contains(column) || is_join_max_int_update(*column, expr))
    {
        return Some(false);
    }

    // A shared MAX assignment may read its own column, but another assignment
    // reading that same column still creates an order dependency. For example,
    // SET a=MAX(a,10), b=a does not commute with SET a=MAX(a,20).
    let mut remaining = 4096;
    let independent = [(cols_a, &written_b), (cols_b, &written_a)]
        .into_iter()
        .all(|(updates, other_writes)| {
            updates.iter().all(|(column, expr)| {
                overlap.contains(column)
                    || !expr_reads_columns(expr, other_writes, &mut remaining, 0)
            })
        });
    Some(independent)
}

/// Conservative read analysis: inspect even currently unselected CASE arms.
/// Exhausting the bounded traversal means "may read", never independence.
fn expr_reads_columns(
    expr: &RebaseExpr,
    columns: &BTreeSet<ColumnIdx>,
    remaining: &mut usize,
    depth: usize,
) -> bool {
    if *remaining == 0 || depth >= 64 {
        return true;
    }
    *remaining -= 1;
    let mut reads = |expr: &RebaseExpr| expr_reads_columns(expr, columns, remaining, depth + 1);
    match expr {
        RebaseExpr::ColumnRef(column) => columns.contains(column),
        RebaseExpr::Literal(_) => false,
        RebaseExpr::UnaryOp { operand, .. } | RebaseExpr::Cast { expr: operand, .. } => {
            reads(operand)
        }
        RebaseExpr::BinaryOp { left, right, .. }
        | RebaseExpr::NullIf { left, right }
        | RebaseExpr::Concat { left, right } => reads(left) || reads(right),
        RebaseExpr::FunctionCall { args, .. } | RebaseExpr::Coalesce(args) => {
            args.iter().any(reads)
        }
        RebaseExpr::Case {
            operand,
            when_clauses,
            else_clause,
        } => {
            operand.as_ref().is_some_and(|expr| reads(expr))
                || when_clauses
                    .iter()
                    .any(|(when, then)| reads(when) || reads(then))
                || else_clause.as_ref().is_some_and(|expr| reads(expr))
        }
    }
}

/// Detect whether a `RebaseExpr` represents a monotone join update of the form
/// `col = max(col, c)` on INTEGER values (§5.10.7 join-max exception).
///
/// Recognizes both argument orders:
/// - `MAX(ColumnRef(col_idx), Literal(Integer(c)))`
/// - `MAX(Literal(Integer(c)), ColumnRef(col_idx))`
#[must_use]
pub fn is_join_max_int_update(col_idx: ColumnIdx, expr: &RebaseExpr) -> bool {
    if let RebaseExpr::FunctionCall { name, args } = expr
        && name.eq_ignore_ascii_case("MAX")
        && args.len() == 2
    {
        return is_column_ref_and_int_literal(col_idx, &args[0], &args[1])
            || is_column_ref_and_int_literal(col_idx, &args[1], &args[0]);
    }
    false
}

/// Check if one arg is `ColumnRef(col_idx)` and the other is `Literal(Integer(_))`.
fn is_column_ref_and_int_literal(
    col_idx: ColumnIdx,
    maybe_col: &RebaseExpr,
    maybe_lit: &RebaseExpr,
) -> bool {
    matches!(maybe_col, RebaseExpr::ColumnRef(c) if *c == col_idx)
        && matches!(maybe_lit, RebaseExpr::Literal(SqliteValue::Integer(_)))
}

/// Extract the integer constant `c` from a join-max expression `MAX(col, c)`.
///
/// Returns `None` if the expression is not a valid join-max form.
#[must_use]
pub fn extract_join_max_constant(col_idx: ColumnIdx, expr: &RebaseExpr) -> Option<i64> {
    if let RebaseExpr::FunctionCall { name, args } = expr
        && name.eq_ignore_ascii_case("MAX")
        && args.len() == 2
    {
        return extract_int_constant_pair(col_idx, &args[0], &args[1])
            .or_else(|| extract_int_constant_pair(col_idx, &args[1], &args[0]));
    }
    None
}

fn extract_int_constant_pair(
    col_idx: ColumnIdx,
    maybe_col: &RebaseExpr,
    maybe_lit: &RebaseExpr,
) -> Option<i64> {
    if matches!(maybe_col, RebaseExpr::ColumnRef(c) if *c == col_idx)
        && let RebaseExpr::Literal(SqliteValue::Integer(c)) = maybe_lit
    {
        return Some(*c);
    }
    None
}

/// Collapse multiple join-max updates on the same `(table, key, col_idx)` into
/// a single update with `c = max(c_1, c_2, ...)`.
///
/// This is justified because `max` is associative, commutative, and idempotent
/// on integers.
#[must_use]
pub fn collapse_join_max_updates(col_idx: ColumnIdx, exprs: &[&RebaseExpr]) -> Option<RebaseExpr> {
    // Collapsing a mixed list must not silently discard non-MAX operations.
    let constants: Vec<i64> = exprs
        .iter()
        .map(|expr| extract_join_max_constant(col_idx, expr))
        .collect::<Option<_>>()?;

    if constants.is_empty() {
        return None;
    }

    let max_c = constants
        .into_iter()
        .max()
        .expect("non-empty checked above");

    Some(RebaseExpr::FunctionCall {
        name: "MAX".to_owned(),
        args: vec![
            RebaseExpr::ColumnRef(col_idx),
            RebaseExpr::Literal(SqliteValue::Integer(max_c)),
        ],
    })
}

/// Check whether an intent op belongs to a mergeable intent class (§5.10.7).
///
/// SAFE merging is deliberately narrow. Only these classes are permitted:
/// - Insert/Delete/Update on table B-tree leaf pages for distinct `RowId` keys
///   (no overflow, no multi-page balance)
/// - `UpdateExpression` on table B-tree leaf pages (column-disjointness rule)
/// - `IndexInsert`/`IndexDelete` on index B-tree leaf pages for distinct index
///   keys (no overflow, no balance)
/// - Any op with `structural != NONE` → NOT mergeable
#[must_use]
pub fn is_mergeable_intent(op: &IntentOp) -> bool {
    // Structural effects disqualify.
    if op.footprint.structural != StructuralEffects::NONE {
        return false;
    }

    matches!(
        op.op,
        IntentOpKind::Insert { .. }
            | IntentOpKind::Delete { .. }
            | IntentOpKind::Update { .. }
            | IntentOpKind::UpdateExpression { .. }
            | IntentOpKind::IndexInsert { .. }
            | IntentOpKind::IndexDelete { .. }
    )
}

/// Sort key for canonical merge ordering within a Foata layer.
///
/// Sorted by `(btree_id, kind, key_digest, op_kind_tag, op_digest)`.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
struct FoataSortKey {
    btree_id: u64,
    kind: u8,
    key_digest: [u8; 16],
    op_kind_tag: u8,
    op_digest: [u8; 16],
}

/// Extract the btree reference, semantic key kind, and key digest for an op.
fn op_sort_components(op: &IntentOp) -> FoataSortKey {
    let (btree_id, kind, key_digest) = match &op.op {
        IntentOpKind::Insert { table, key, .. }
        | IntentOpKind::Delete { table, key }
        | IntentOpKind::Update { table, key, .. }
        | IntentOpKind::UpdateExpression { table, key, .. } => {
            let btree = BtreeRef::Table(*table);
            let digest = table_row_key_digest(btree, key.get());
            (u64::from(table.get()), 0u8, digest)
        }
        IntentOpKind::IndexInsert { index, key, .. }
        | IntentOpKind::IndexDelete { index, key, .. } => {
            let btree = BtreeRef::Index(*index);
            let digest = SemanticKeyRef::compute_digest(SemanticKeyKind::IndexEntry, btree, key);
            (u64::from(index.get()), 1u8, digest)
        }
    };

    let op_kind_tag = match &op.op {
        IntentOpKind::Insert { .. } => 0,
        IntentOpKind::Delete { .. } => 1,
        IntentOpKind::Update { .. } => 2,
        IntentOpKind::UpdateExpression { .. } => 3,
        IntentOpKind::IndexInsert { .. } => 4,
        IntentOpKind::IndexDelete { .. } => 5,
    };

    FoataSortKey {
        btree_id,
        kind,
        key_digest,
        op_kind_tag,
        op_digest: compute_op_digest(op),
    }
}

fn table_row_key_digest(btree: BtreeRef, rowid: i64) -> [u8; 16] {
    crate::cell_visibility::CellKey::table_row(btree, rowid).key_digest
}

/// Compute the canonical Foata normal form for a set of intent operations.
///
/// Returns op digests in the canonical order used for merge certificates.
/// The algorithm:
/// 1. Build a dependency graph (edges for non-independent pairs).
/// 2. Layer by topological order (Foata layers).
/// 3. Within each layer, sort by `(btree_id, kind, key_digest, op_kind, op_digest)`.
#[must_use]
pub fn foata_normal_form(ops: &[IntentOp]) -> Vec<[u8; 16]> {
    let n = ops.len();
    if n == 0 {
        return Vec::new();
    }

    // Build dependency graph: edge (i → j) means i must precede j.
    // Two ops that are NOT independent have a dependency.
    // We use the index-order to break ties: lower index goes first.
    let mut in_degree = vec![0u32; n];
    let mut successors: Vec<Vec<usize>> = vec![Vec::new(); n];

    for i in 0..n {
        for j in (i + 1)..n {
            if !are_intent_ops_independent(&ops[i], &ops[j]) {
                // i must come before j (preserve original order for dependent ops).
                successors[i].push(j);
                in_degree[j] += 1;
            }
        }
    }

    // Foata layering: BFS-like topological sort collecting layers.
    let mut result = Vec::with_capacity(n);
    let mut remaining_in_degree = in_degree;

    loop {
        // Collect all ops with in_degree 0.
        let mut layer: Vec<usize> = (0..n).filter(|&i| remaining_in_degree[i] == 0).collect();

        // Mark collected ops so we don't pick them again.
        // Use sentinel value u32::MAX.
        for &i in &layer {
            remaining_in_degree[i] = u32::MAX;
        }

        if layer.is_empty() {
            break;
        }

        // Sort within layer by canonical sort key.
        let mut sort_keys: Vec<(FoataSortKey, usize)> = layer
            .iter()
            .map(|&i| (op_sort_components(&ops[i]), i))
            .collect();
        sort_keys.sort_by(|a, b| a.0.cmp(&b.0));
        layer = sort_keys.into_iter().map(|(_, i)| i).collect();

        // Emit op digests in this layer's order.
        for &i in &layer {
            result.push(compute_op_digest(&ops[i]));
        }

        // Decrease in_degree for successors.
        for &i in &layer {
            for &j in &successors[i] {
                if remaining_in_degree[j] != u32::MAX {
                    remaining_in_degree[j] -= 1;
                }
            }
        }
    }

    result
}

// ---------------------------------------------------------------------------
// §5.10.8: Merge Certificates
// ---------------------------------------------------------------------------

/// The kind of merge that produced a commit.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub enum MergeKind {
    /// Deterministic rebase only (§5.10.2).
    Rebase,
    /// Structured page patch merge only (§5.10.3).
    StructuredPatch,
    /// Both rebase and structured patch merge.
    RebaseAndPatch,
}

/// Post-merge state hashes for verification.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct MergeCertificatePostState {
    /// Hash of each affected page's repacked bytes after merge.
    pub page_hashes: Vec<(PageNumber, [u8; 16])>,
    /// Hash validating B-tree invariants hold across all affected pages.
    pub btree_invariant_hash: [u8; 16],
}

/// A proof-carrying merge certificate (§5.10.8).
///
/// Every commit accepted via the merge path MUST produce a verifiable
/// `MergeCertificate`. In native mode, this is attached to the `CommitProof`.
/// In compatibility mode, it is emitted to the evidence ledger.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct MergeCertificate {
    /// The kind of merge that produced this commit.
    pub merge_kind: MergeKind,
    /// Sequence number of the base commit.
    pub base_commit_seq: u64,
    /// Schema version at merge time.
    pub schema_epoch: u64,
    /// All page numbers affected by the merge.
    pub pages: Vec<PageNumber>,
    /// Stable 128-bit digests of all involved intent ops (unordered).
    pub intent_op_digests: Vec<[u8; 16]>,
    /// Digest computed over all `IntentFootprint` values.
    pub footprint_digest: [u8; 16],
    /// Op digests in canonical Foata-layered order (the ORDER that matters).
    pub normal_form: Vec<[u8; 16]>,
    /// Post-merge state hashes.
    pub post_state: MergeCertificatePostState,
    /// Version of the verification algorithm (for future compatibility).
    pub verifier_version: u32,
}

/// Trusted replay context supplied independently of the certificate being checked.
///
/// Do not construct this from untrusted certificate fields. The caller still
/// owns replay and B-tree validation; matching hashes alone is not a replay.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct MergeCertificateContext {
    /// The merge path selected by the caller's policy and replay.
    pub merge_kind: MergeKind,
    /// The actual base snapshot used for that replay.
    pub base_commit_seq: u64,
    /// The schema epoch under which the replay was performed.
    pub schema_epoch: u64,
}

impl MergeCertificate {
    /// Verify complete evidence against independently supplied replay context.
    ///
    /// `post_merge_pages` must contain every affected page produced by replay,
    /// and `btree_invariant_hash` must come from the caller's invariant checks.
    /// This method neither executes replay nor authenticates a certificate.
    ///
    /// # Errors
    ///
    /// Rejects mismatched context, unsupported verifier versions, inconsistent
    /// manifests, schema epochs, intent evidence or reconstructed page hashes.
    pub fn verify_in_context(
        &self,
        context: MergeCertificateContext,
        intent_ops: &[IntentOp],
        post_merge_pages: &[(PageNumber, Vec<u8>)],
        btree_invariant_hash: [u8; 16],
    ) -> Result<(), CertificateVerificationError> {
        if self.merge_kind != context.merge_kind
            || self.base_commit_seq != context.base_commit_seq
            || self.schema_epoch != context.schema_epoch
        {
            return Err(CertificateVerificationError::ContextMismatch);
        }
        verify_merge_certificate(intent_ops, post_merge_pages, btree_invariant_hash, self)
    }
}

/// V2 binds complete semantic keys and preserves exact context/page coverage.
/// V1 certificates must be independently regenerated, not relabeled as V2.
pub const VERIFIER_VERSION: u32 = 2;

fn digest_bytes_equal(left: &[u8; 16], right: &[u8; 16]) -> bool {
    left.iter()
        .zip(right.iter())
        .fold(0u8, |diff, (left_byte, right_byte)| {
            diff | (*left_byte ^ *right_byte)
        })
        == 0
}

/// Errors from merge certificate verification.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CertificateVerificationError {
    /// This verifier cannot interpret the certificate's algorithm version.
    UnsupportedVerifierVersion { expected: u32, actual: u32 },
    /// Certificate metadata disagrees with the independently supplied replay.
    ContextMismatch,
    /// An intent belongs to a different schema epoch.
    SchemaEpochMismatch { expected: u64, actual: u64 },
    /// A page appears more than once in a manifest or in the replay output.
    DuplicatePage { page: PageNumber },
    /// Declared pages, supplied hashes and replay output do not cover one set.
    PageSetMismatch,
    /// Recomputed op digests do not match the certificate.
    OpDigestMismatch {
        expected: Vec<[u8; 16]>,
        actual: Vec<[u8; 16]>,
    },
    /// Recomputed footprint digest does not match.
    FootprintDigestMismatch {
        expected: [u8; 16],
        actual: [u8; 16],
    },
    /// Normal form in the certificate is not valid.
    InvalidNormalForm,
    /// Post-state page hashes do not match recomputed values.
    PageHashMismatch {
        page: PageNumber,
        expected: [u8; 16],
        actual: [u8; 16],
    },
    /// B-tree invariant hash does not match.
    BtreeInvariantHashMismatch {
        expected: [u8; 16],
        actual: [u8; 16],
    },
}

impl std::fmt::Display for CertificateVerificationError {
    #[allow(clippy::too_many_lines)]
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::UnsupportedVerifierVersion { expected, actual } => {
                write!(
                    f,
                    "unsupported merge verifier version {actual}; expected {expected}"
                )
            }
            Self::ContextMismatch => f.write_str("merge certificate replay context mismatch"),
            Self::SchemaEpochMismatch { expected, actual } => {
                write!(
                    f,
                    "merge intent schema epoch {actual} differs from {expected}"
                )
            }
            Self::DuplicatePage { page } => write!(f, "duplicate page {page} in merge evidence"),
            Self::PageSetMismatch => f.write_str("merge certificate page coverage mismatch"),
            Self::OpDigestMismatch { .. } => f.write_str("op digest mismatch in merge certificate"),
            Self::FootprintDigestMismatch { .. } => {
                f.write_str("footprint digest mismatch in merge certificate")
            }
            Self::InvalidNormalForm => f.write_str("invalid normal form in merge certificate"),
            Self::PageHashMismatch { page, .. } => {
                write!(f, "page hash mismatch for page {page} in merge certificate")
            }
            Self::BtreeInvariantHashMismatch { .. } => {
                f.write_str("B-tree invariant hash mismatch in merge certificate")
            }
        }
    }
}

impl std::error::Error for CertificateVerificationError {}

fn unique_certificate_pages(
    pages: impl IntoIterator<Item = PageNumber>,
) -> Result<BTreeSet<PageNumber>, CertificateVerificationError> {
    let mut unique = BTreeSet::new();
    for page in pages {
        if !unique.insert(page) {
            return Err(CertificateVerificationError::DuplicatePage { page });
        }
    }
    Ok(unique)
}

fn validate_certificate_epochs(
    intent_ops: &[IntentOp],
    schema_epoch: u64,
) -> Result<(), CertificateVerificationError> {
    if let Some(op) = intent_ops.iter().find(|op| op.schema_epoch != schema_epoch) {
        return Err(CertificateVerificationError::SchemaEpochMismatch {
            expected: schema_epoch,
            actual: op.schema_epoch,
        });
    }
    Ok(())
}

/// A circuit breaker event emitted when merge verification fails.
///
/// Production behavior: disable SAFE merging (`PRAGMA fsqlite.write_merge = OFF`),
/// emit evidence ledger entry, escalate supervision.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CircuitBreakerEvent {
    /// The verification error that triggered the circuit breaker.
    pub error: CertificateVerificationError,
    /// The certificate that failed verification.
    pub certificate_digest: [u8; 16],
    /// Whether SAFE merging should be disabled.
    pub disable_safe_merge: bool,
}

/// Generate a merge certificate for a successful merge operation.
///
/// # Arguments
///
/// * `merge_kind` — The kind of merge (rebase, structured patch, or both).
/// * `base_commit_seq` — The commit sequence of the base snapshot.
/// * `schema_epoch` — The schema epoch at merge time.
/// * `intent_ops` — All intent operations involved in the merge.
/// * `affected_pages` — Pages affected, with their post-merge bytes.
/// * `btree_invariant_hash` — Hash of B-tree invariant validation.
///
/// # Errors
///
/// Rejects duplicate affected pages or intents from another schema epoch.
pub fn generate_merge_certificate(
    merge_kind: MergeKind,
    base_commit_seq: u64,
    schema_epoch: u64,
    intent_ops: &[IntentOp],
    affected_pages: &[(PageNumber, Vec<u8>)],
    btree_invariant_hash: [u8; 16],
) -> Result<MergeCertificate, CertificateVerificationError> {
    // Refuse ambiguous evidence before hashing it or building a dependency graph.
    unique_certificate_pages(affected_pages.iter().map(|(page, _)| *page))?;
    validate_certificate_epochs(intent_ops, schema_epoch)?;
    // Compute op digests (unordered set).
    let intent_op_digests: Vec<[u8; 16]> = intent_ops.iter().map(compute_op_digest).collect();

    // Compute footprint digest.
    let footprints: Vec<&IntentFootprint> = intent_ops.iter().map(|op| &op.footprint).collect();
    let footprint_digest = compute_footprint_digest(&footprints);

    // Compute canonical Foata normal form.
    let normal_form = foata_normal_form(intent_ops);

    // Compute page hashes.
    let page_hashes: Vec<(PageNumber, [u8; 16])> = affected_pages
        .iter()
        .map(|(pgno, bytes)| {
            let hash = blake3::hash(bytes);
            let mut truncated = [0u8; 16];
            truncated.copy_from_slice(&hash.as_bytes()[..16]);
            (*pgno, truncated)
        })
        .collect();

    let pages: Vec<PageNumber> = affected_pages.iter().map(|(pgno, _)| *pgno).collect();

    Ok(MergeCertificate {
        merge_kind,
        base_commit_seq,
        schema_epoch,
        pages,
        intent_op_digests,
        footprint_digest,
        normal_form,
        post_state: MergeCertificatePostState {
            page_hashes,
            btree_invariant_hash,
        },
        verifier_version: VERIFIER_VERSION,
    })
}

/// Verify certificate evidence against caller-supplied replay results.
///
/// This function does NOT replay a merge or validate B-tree structure itself.
/// The caller must supply the complete replay output and invariant-check hash.
/// It rejects unsupported versions and requires identical, duplicate-free page
/// sets in the declaration, page hashes and replay output. It then:
/// 1. Recomputes all op digests from canonical intent encodings.
/// 2. Recomputes the footprint digest.
/// 3. Validates the normal form.
/// 4. Compares page hashes and B-tree invariant hash.
///
/// The legacy signature cannot independently bind `base_commit_seq` or
/// `merge_kind`. Use [`MergeCertificate::verify_in_context`] when accepting
/// evidence from another replay/context; never take those values on trust.
///
/// # Errors
///
/// Returns a [`CertificateVerificationError`] describing the first mismatch found.
pub fn verify_merge_certificate(
    intent_ops: &[IntentOp],
    post_merge_pages: &[(PageNumber, Vec<u8>)],
    btree_invariant_hash: [u8; 16],
    certificate: &MergeCertificate,
) -> Result<(), CertificateVerificationError> {
    if certificate.verifier_version != VERIFIER_VERSION {
        return Err(CertificateVerificationError::UnsupportedVerifierVersion {
            expected: VERIFIER_VERSION,
            actual: certificate.verifier_version,
        });
    }
    validate_certificate_epochs(intent_ops, certificate.schema_epoch)?;
    let declared = unique_certificate_pages(certificate.pages.iter().copied())?;
    let hashed = unique_certificate_pages(
        certificate
            .post_state
            .page_hashes
            .iter()
            .map(|(page, _)| *page),
    )?;
    let replayed = unique_certificate_pages(post_merge_pages.iter().map(|(page, _)| *page))?;
    if declared != hashed || declared != replayed {
        return Err(CertificateVerificationError::PageSetMismatch);
    }
    // Step 1: Recompute op digests and compare (order-insensitive).
    let mut recomputed_digests: Vec<[u8; 16]> = intent_ops.iter().map(compute_op_digest).collect();
    let mut cert_digests = certificate.intent_op_digests.clone();
    recomputed_digests.sort_unstable();
    cert_digests.sort_unstable();
    if recomputed_digests != cert_digests {
        return Err(CertificateVerificationError::OpDigestMismatch {
            expected: cert_digests,
            actual: recomputed_digests,
        });
    }

    // Step 2: Recompute footprint digest.
    let footprints: Vec<&IntentFootprint> = intent_ops.iter().map(|op| &op.footprint).collect();
    let recomputed_fp_digest = compute_footprint_digest(&footprints);
    if !digest_bytes_equal(&recomputed_fp_digest, &certificate.footprint_digest) {
        return Err(CertificateVerificationError::FootprintDigestMismatch {
            expected: certificate.footprint_digest,
            actual: recomputed_fp_digest,
        });
    }

    // Step 3: Validate normal form is valid Foata layering.
    let expected_normal_form = foata_normal_form(intent_ops);
    if expected_normal_form != certificate.normal_form {
        return Err(CertificateVerificationError::InvalidNormalForm);
    }

    // Step 4: Compare page hashes.
    let recomputed_page_hashes: BTreeMap<PageNumber, [u8; 16]> = post_merge_pages
        .iter()
        .map(|(pgno, bytes)| {
            let hash = blake3::hash(bytes);
            let mut truncated = [0u8; 16];
            truncated.copy_from_slice(&hash.as_bytes()[..16]);
            (*pgno, truncated)
        })
        .collect();

    for &(pgno, expected_hash) in &certificate.post_state.page_hashes {
        // Missing evidence is never represented by a synthetic all-zero hash.
        let actual_hash = *recomputed_page_hashes
            .get(&pgno)
            .ok_or(CertificateVerificationError::PageSetMismatch)?;
        if actual_hash != expected_hash {
            return Err(CertificateVerificationError::PageHashMismatch {
                page: pgno,
                expected: expected_hash,
                actual: actual_hash,
            });
        }
    }

    // Step 5: Compare B-tree invariant hash.
    if btree_invariant_hash != certificate.post_state.btree_invariant_hash {
        return Err(CertificateVerificationError::BtreeInvariantHashMismatch {
            expected: certificate.post_state.btree_invariant_hash,
            actual: btree_invariant_hash,
        });
    }

    Ok(())
}

/// Check for circuit breaker conditions after a verification failure.
///
/// If verification fails, the circuit breaker fires:
/// - Production: disable SAFE merging, emit evidence ledger entry, escalate.
/// - Lab mode: fail fast.
#[must_use]
pub fn circuit_breaker_check(
    verification_error: CertificateVerificationError,
    certificate: &MergeCertificate,
) -> CircuitBreakerEvent {
    // Bind the whole failed certificate so omitted/tampered page evidence and
    // different verifier contracts cannot share an incident identity.
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"fsqlite:merge-certificate:v2");
    hasher.update(&certificate.verifier_version.to_le_bytes());
    hasher.update(&[match certificate.merge_kind {
        MergeKind::Rebase => 0,
        MergeKind::StructuredPatch => 1,
        MergeKind::RebaseAndPatch => 2,
    }]);
    hasher.update(&certificate.base_commit_seq.to_le_bytes());
    hasher.update(&certificate.schema_epoch.to_le_bytes());
    hash_certificate_length(&mut hasher, certificate.pages.len());
    for page in &certificate.pages {
        hasher.update(&page.get().to_le_bytes());
    }
    hash_certificate_length(&mut hasher, certificate.intent_op_digests.len());
    for digest in &certificate.intent_op_digests {
        hasher.update(digest);
    }
    hasher.update(&certificate.footprint_digest);
    hash_certificate_length(&mut hasher, certificate.normal_form.len());
    for digest in &certificate.normal_form {
        hasher.update(digest);
    }
    hash_certificate_length(&mut hasher, certificate.post_state.page_hashes.len());
    for (page, digest) in &certificate.post_state.page_hashes {
        hasher.update(&page.get().to_le_bytes());
        hasher.update(digest);
    }
    hasher.update(&certificate.post_state.btree_invariant_hash);
    let hash = hasher.finalize();
    let mut cert_digest = [0u8; 16];
    cert_digest.copy_from_slice(&hash.as_bytes()[..16]);

    CircuitBreakerEvent {
        error: verification_error,
        certificate_digest: cert_digest,
        disable_safe_merge: true,
    }
}

// ---------------------------------------------------------------------------
// Canonical intent encoding for op_digest computation
// ---------------------------------------------------------------------------

fn hash_certificate_length(hasher: &mut blake3::Hasher, length: usize) {
    let length = u64::try_from(length).expect("certificate vector length fits u64");
    hasher.update(&length.to_le_bytes());
}

/// Produce canonical bytes for an `IntentOp` (deterministic, for hashing).
fn canonical_intent_bytes(op: &IntentOp) -> Vec<u8> {
    let mut buf = Vec::with_capacity(64);

    // Schema epoch.
    buf.extend_from_slice(&op.schema_epoch.to_le_bytes());

    // Footprint.
    canonical_footprint_bytes(&mut buf, &op.footprint);

    // Op kind.
    canonical_op_kind_bytes(&mut buf, &op.op);

    buf
}

fn canonical_footprint_bytes(buf: &mut Vec<u8>, fp: &IntentFootprint) {
    // Reads.
    #[allow(clippy::cast_possible_truncation)]
    let reads_len = fp.reads.len() as u32;
    buf.extend_from_slice(&reads_len.to_le_bytes());
    for r in &fp.reads {
        buf.extend_from_slice(&canonical_semantic_key_bytes(r));
    }

    // Writes.
    #[allow(clippy::cast_possible_truncation)]
    let writes_len = fp.writes.len() as u32;
    buf.extend_from_slice(&writes_len.to_le_bytes());
    for w in &fp.writes {
        buf.extend_from_slice(&canonical_semantic_key_bytes(w));
    }

    // Structural effects.
    buf.extend_from_slice(&fp.structural.bits().to_le_bytes());
}

/// Key kind, B-tree namespace, B-tree id, and the supplied semantic digest.
/// Changing routing metadata without recomputing a digest must still change
/// the proof identity. Fixed widths keep the encoding unambiguous.
fn canonical_semantic_key_bytes(key: &SemanticKeyRef) -> [u8; 22] {
    let mut bytes = [0; 22];
    bytes[0] = match key.kind {
        SemanticKeyKind::TableRow => 0,
        SemanticKeyKind::IndexEntry => 1,
    };
    let (tag, id) = match key.btree {
        BtreeRef::Table(table) => (0, table.get()),
        BtreeRef::Index(index) => (1, index.get()),
    };
    bytes[1] = tag;
    bytes[2..6].copy_from_slice(&id.to_le_bytes());
    bytes[6..].copy_from_slice(&key.key_digest);
    bytes
}

#[allow(clippy::too_many_lines)]
fn canonical_op_kind_bytes(buf: &mut Vec<u8>, op: &IntentOpKind) {
    match op {
        IntentOpKind::Insert { table, key, record } => {
            buf.push(0);
            buf.extend_from_slice(&table.get().to_le_bytes());
            buf.extend_from_slice(&key.get().to_le_bytes());
            #[allow(clippy::cast_possible_truncation)]
            let len = record.len() as u32;
            buf.extend_from_slice(&len.to_le_bytes());
            buf.extend_from_slice(record);
        }
        IntentOpKind::Delete { table, key } => {
            buf.push(1);
            buf.extend_from_slice(&table.get().to_le_bytes());
            buf.extend_from_slice(&key.get().to_le_bytes());
        }
        IntentOpKind::Update {
            table,
            key,
            new_record,
        } => {
            buf.push(2);
            buf.extend_from_slice(&table.get().to_le_bytes());
            buf.extend_from_slice(&key.get().to_le_bytes());
            #[allow(clippy::cast_possible_truncation)]
            let len = new_record.len() as u32;
            buf.extend_from_slice(&len.to_le_bytes());
            buf.extend_from_slice(new_record);
        }
        IntentOpKind::IndexInsert { index, key, rowid } => {
            buf.push(3);
            buf.extend_from_slice(&index.get().to_le_bytes());
            #[allow(clippy::cast_possible_truncation)]
            let len = key.len() as u32;
            buf.extend_from_slice(&len.to_le_bytes());
            buf.extend_from_slice(key);
            buf.extend_from_slice(&rowid.get().to_le_bytes());
        }
        IntentOpKind::IndexDelete { index, key, rowid } => {
            buf.push(4);
            buf.extend_from_slice(&index.get().to_le_bytes());
            #[allow(clippy::cast_possible_truncation)]
            let len = key.len() as u32;
            buf.extend_from_slice(&len.to_le_bytes());
            buf.extend_from_slice(key);
            buf.extend_from_slice(&rowid.get().to_le_bytes());
        }
        IntentOpKind::UpdateExpression {
            table,
            key,
            column_updates,
        } => {
            buf.push(5);
            buf.extend_from_slice(&table.get().to_le_bytes());
            buf.extend_from_slice(&key.get().to_le_bytes());
            #[allow(clippy::cast_possible_truncation)]
            let len = column_updates.len() as u32;
            buf.extend_from_slice(&len.to_le_bytes());
            for (col, expr) in column_updates {
                buf.extend_from_slice(&col.get().to_le_bytes());
                canonical_rebase_expr_bytes(buf, expr);
            }
        }
    }
}

fn canonical_rebase_expr_bytes(buf: &mut Vec<u8>, expr: &RebaseExpr) {
    match expr {
        RebaseExpr::ColumnRef(col) => {
            buf.push(0);
            buf.extend_from_slice(&col.get().to_le_bytes());
        }
        RebaseExpr::Literal(val) => {
            buf.push(1);
            canonical_sqlite_value_bytes(buf, val);
        }
        RebaseExpr::UnaryOp { op, operand } => {
            buf.push(2);
            buf.push(*op as u8);
            canonical_rebase_expr_bytes(buf, operand);
        }
        RebaseExpr::BinaryOp { op, left, right } => {
            buf.push(3);
            buf.push(*op as u8);
            canonical_rebase_expr_bytes(buf, left);
            canonical_rebase_expr_bytes(buf, right);
        }
        RebaseExpr::FunctionCall { name, args } => {
            buf.push(4);
            #[allow(clippy::cast_possible_truncation)]
            let name_len = name.len() as u32;
            buf.extend_from_slice(&name_len.to_le_bytes());
            buf.extend_from_slice(name.as_bytes());
            #[allow(clippy::cast_possible_truncation)]
            let args_len = args.len() as u32;
            buf.extend_from_slice(&args_len.to_le_bytes());
            for arg in args {
                canonical_rebase_expr_bytes(buf, arg);
            }
        }
        RebaseExpr::Cast { expr, type_name } => {
            buf.push(5);
            canonical_rebase_expr_bytes(buf, expr);
            #[allow(clippy::cast_possible_truncation)]
            let tn_len = type_name.len() as u32;
            buf.extend_from_slice(&tn_len.to_le_bytes());
            buf.extend_from_slice(type_name.as_bytes());
        }
        RebaseExpr::Case {
            operand,
            when_clauses,
            else_clause,
        } => {
            buf.push(6);
            if let Some(op) = operand {
                buf.push(1);
                canonical_rebase_expr_bytes(buf, op);
            } else {
                buf.push(0);
            }
            #[allow(clippy::cast_possible_truncation)]
            let when_len = when_clauses.len() as u32;
            buf.extend_from_slice(&when_len.to_le_bytes());
            for (when, then) in when_clauses {
                canonical_rebase_expr_bytes(buf, when);
                canonical_rebase_expr_bytes(buf, then);
            }
            if let Some(el) = else_clause {
                buf.push(1);
                canonical_rebase_expr_bytes(buf, el);
            } else {
                buf.push(0);
            }
        }
        RebaseExpr::Coalesce(args) => {
            buf.push(7);
            #[allow(clippy::cast_possible_truncation)]
            let args_len = args.len() as u32;
            buf.extend_from_slice(&args_len.to_le_bytes());
            for arg in args {
                canonical_rebase_expr_bytes(buf, arg);
            }
        }
        RebaseExpr::NullIf { left, right } => {
            buf.push(8);
            canonical_rebase_expr_bytes(buf, left);
            canonical_rebase_expr_bytes(buf, right);
        }
        RebaseExpr::Concat { left, right } => {
            buf.push(9);
            canonical_rebase_expr_bytes(buf, left);
            canonical_rebase_expr_bytes(buf, right);
        }
    }
}

fn canonical_sqlite_value_bytes(buf: &mut Vec<u8>, val: &SqliteValue) {
    match val {
        SqliteValue::Null => buf.push(0),
        SqliteValue::Integer(i) => {
            buf.push(1);
            buf.extend_from_slice(&i.to_le_bytes());
        }
        SqliteValue::Float(f) => {
            buf.push(2);
            buf.extend_from_slice(&f.to_le_bytes());
        }
        SqliteValue::Text(s) => {
            buf.push(3);
            #[allow(clippy::cast_possible_truncation)]
            let len = s.len() as u32;
            buf.extend_from_slice(&len.to_le_bytes());
            // Bind stored bytes, not SmallText's lossy str view of raw TEXT.
            buf.extend_from_slice(s.as_bytes_direct());
        }
        SqliteValue::Blob(b) => {
            buf.push(4);
            #[allow(clippy::cast_possible_truncation)]
            let len = b.len() as u32;
            buf.extend_from_slice(&len.to_le_bytes());
            buf.extend_from_slice(b);
        }
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use fsqlite_types::{
        BtreeRef, ColumnIdx, CommitSeq, IndexId, IntentFootprint, IntentOp, IntentOpKind,
        PageNumber, RebaseExpr, RowId, SemanticKeyKind, SemanticKeyRef, SqliteValue,
        StructuralEffects, TableId,
    };

    /// Helper: create an IntentOp with the given kind and empty footprint.
    fn make_op(schema_epoch: u64, kind: IntentOpKind) -> IntentOp {
        IntentOp {
            schema_epoch,
            footprint: IntentFootprint::empty(),
            op: kind,
        }
    }

    /// Helper: create an IntentOp with specific writes.
    fn make_op_with_writes(
        schema_epoch: u64,
        kind: IntentOpKind,
        writes: Vec<SemanticKeyRef>,
    ) -> IntentOp {
        IntentOp {
            schema_epoch,
            footprint: IntentFootprint {
                reads: Vec::new(),
                writes,
                structural: StructuralEffects::NONE,
            },
            op: kind,
        }
    }

    /// Helper: create a table row semantic key ref.
    fn table_key(table_id: u32, rowid: i64) -> SemanticKeyRef {
        let btree = BtreeRef::Table(TableId::new(table_id));
        SemanticKeyRef {
            btree,
            kind: SemanticKeyKind::TableRow,
            key_digest: table_row_key_digest(btree, rowid),
        }
    }

    fn expression_update(updates: Vec<(ColumnIdx, RebaseExpr)>) -> IntentOp {
        make_op(
            1,
            IntentOpKind::UpdateExpression {
                table: TableId::new(1),
                key: RowId::new(1),
                column_updates: updates,
            },
        )
    }

    fn max_update(column: ColumnIdx, value: i64) -> RebaseExpr {
        RebaseExpr::FunctionCall {
            name: "MAX".to_owned(),
            args: vec![
                RebaseExpr::ColumnRef(column),
                RebaseExpr::Literal(SqliteValue::Integer(value)),
            ],
        }
    }

    fn assert_dependent(left: &IntentOp, right: &IntentOp) {
        assert!(!are_intent_ops_independent(left, right));
        assert!(!are_intent_ops_independent(right, left));
    }

    #[test]
    fn expression_reads_prevent_reordering_disjoint_writes() {
        let left = expression_update(vec![(
            ColumnIdx::new(0),
            RebaseExpr::ColumnRef(ColumnIdx::new(1)),
        )]);
        let right = expression_update(vec![(
            ColumnIdx::new(1),
            RebaseExpr::Literal(SqliteValue::Integer(9)),
        )]);
        // From (a,b)=(1,2), a=b then b=9 yields (2,9); the reverse yields (9,9).
        assert_dependent(&left, &right);
        assert_eq!(
            foata_normal_form(&[left.clone(), right.clone()]),
            vec![compute_op_digest(&left), compute_op_digest(&right)]
        );
        assert_eq!(
            foata_normal_form(&[right.clone(), left.clone()]),
            vec![compute_op_digest(&right), compute_op_digest(&left)]
        );
    }

    #[test]
    fn dependency_walk_covers_every_expression_container() {
        use fsqlite_types::{RebaseBinaryOp, RebaseUnaryOp};

        let column = RebaseExpr::ColumnRef(ColumnIdx::new(1));
        let literal = RebaseExpr::Literal(SqliteValue::Integer(0));
        let expressions = vec![
            column.clone(),
            RebaseExpr::UnaryOp {
                op: RebaseUnaryOp::Not,
                operand: Box::new(column.clone()),
            },
            RebaseExpr::BinaryOp {
                op: RebaseBinaryOp::Add,
                left: Box::new(literal.clone()),
                right: Box::new(column.clone()),
            },
            RebaseExpr::FunctionCall {
                name: "ABS".to_owned(),
                args: vec![column.clone()],
            },
            RebaseExpr::Cast {
                expr: Box::new(column.clone()),
                type_name: "TEXT".to_owned(),
            },
            RebaseExpr::Case {
                operand: Some(Box::new(column.clone())),
                when_clauses: vec![(literal.clone(), literal.clone())],
                else_clause: None,
            },
            RebaseExpr::Case {
                operand: None,
                when_clauses: vec![(column.clone(), literal.clone())],
                else_clause: None,
            },
            RebaseExpr::Case {
                operand: None,
                when_clauses: vec![(literal.clone(), column.clone())],
                else_clause: None,
            },
            RebaseExpr::Case {
                operand: None,
                when_clauses: Vec::new(),
                else_clause: Some(Box::new(column.clone())),
            },
            RebaseExpr::Coalesce(vec![literal.clone(), column.clone()]),
            RebaseExpr::NullIf {
                left: Box::new(column.clone()),
                right: Box::new(literal.clone()),
            },
            RebaseExpr::Concat {
                left: Box::new(literal.clone()),
                right: Box::new(column),
            },
        ];
        let writer = expression_update(vec![(ColumnIdx::new(1), literal)]);
        for expr in expressions {
            assert_dependent(&expression_update(vec![(ColumnIdx::new(0), expr)]), &writer);
        }
    }

    #[test]
    fn join_max_does_not_hide_reads_in_other_assignments() {
        let a = ColumnIdx::new(0);
        let b = ColumnIdx::new(1);
        let left = expression_update(vec![(a, max_update(a, 10)), (b, RebaseExpr::ColumnRef(a))]);
        let right = expression_update(vec![(a, max_update(a, 20))]);
        assert_dependent(&left, &right);
        // Without the cross-column read the shared MAX remains admissible.
        let independent = expression_update(vec![
            (a, max_update(a, 10)),
            (b, RebaseExpr::Literal(SqliteValue::Integer(7))),
        ]);
        assert!(are_intent_ops_independent(&independent, &right));
        assert!(are_intent_ops_independent(&right, &independent));
    }

    #[test]
    fn duplicate_assignments_cannot_supply_a_false_join_max_witness() {
        let column = ColumnIdx::new(0);
        let duplicate = expression_update(vec![
            (column, max_update(column, 10)),
            (column, RebaseExpr::Literal(SqliteValue::Integer(0))),
        ]);
        let max = expression_update(vec![(column, max_update(column, 20))]);
        assert_dependent(&duplicate, &max);
        let other_column = expression_update(vec![(
            ColumnIdx::new(1),
            RebaseExpr::Literal(SqliteValue::Integer(1)),
        )]);
        assert_dependent(&duplicate, &other_column);
    }

    #[test]
    fn column_refinement_preserves_all_explicit_footprint_dependencies() {
        let mut left = expression_update(vec![(
            ColumnIdx::new(0),
            RebaseExpr::Literal(SqliteValue::Integer(1)),
        )]);
        let mut right = expression_update(vec![(
            ColumnIdx::new(1),
            RebaseExpr::Literal(SqliteValue::Integer(2)),
        )]);
        left.footprint.writes.push(table_key(1, 1));
        right.footprint.writes.push(table_key(1, 1));
        assert!(are_intent_ops_independent(&left, &right));
        let mut explicit_read = left.clone();
        explicit_read.footprint.reads.push(table_key(1, 1));
        assert_dependent(&explicit_read, &right);

        let extra = SemanticKeyRef::new(
            BtreeRef::Index(IndexId::new(9)),
            SemanticKeyKind::IndexEntry,
            &[7],
        );
        left.footprint.writes.push(extra.clone());
        right.footprint.reads.push(extra.clone());
        assert_dependent(&left, &right);
        right.footprint.reads.clear();
        right.footprint.writes.push(extra);
        assert_dependent(&left, &right);
        right.footprint.writes.pop();
        right.footprint.writes.push(table_key(9, 3));
        assert!(are_intent_ops_independent(&left, &right));
    }

    #[test]
    fn intrinsic_table_writes_do_not_depend_on_footprint_completeness() {
        let table = TableId::new(1);
        let key = RowId::new(1);
        let operations = [
            IntentOpKind::Insert {
                table,
                key,
                record: vec![1],
            },
            IntentOpKind::Delete { table, key },
            IntentOpKind::Update {
                table,
                key,
                new_record: vec![2],
            },
            IntentOpKind::UpdateExpression {
                table,
                key,
                column_updates: vec![(
                    ColumnIdx::new(0),
                    RebaseExpr::Literal(SqliteValue::Integer(1)),
                )],
            },
        ];
        for a in &operations {
            for b in &operations {
                assert_dependent(&make_op(1, a.clone()), &make_op(1, b.clone()));
            }
        }
        let writer = make_op(1, operations[0].clone());
        let mut reader = make_op(
            1,
            IntentOpKind::Delete {
                table,
                key: RowId::new(2),
            },
        );
        assert!(are_intent_ops_independent(&writer, &reader));
        reader.footprint.reads.push(table_key(1, 1));
        assert_dependent(&writer, &reader);
    }

    #[test]
    fn intrinsic_index_writes_are_separate_from_table_key_space() {
        let index = IndexId::new(1);
        let left = make_op(
            1,
            IntentOpKind::IndexInsert {
                index,
                key: vec![1],
                rowid: RowId::new(1),
            },
        );
        let right = make_op(
            1,
            IntentOpKind::IndexDelete {
                index,
                key: vec![1],
                rowid: RowId::new(1),
            },
        );
        assert_dependent(&left, &right);
        let table = make_op(
            1,
            IntentOpKind::Delete {
                table: TableId::new(1),
                key: RowId::new(1),
            },
        );
        assert!(are_intent_ops_independent(&left, &table));
    }

    #[test]
    fn bounded_expression_analysis_fails_closed() {
        let mut deep = RebaseExpr::Literal(SqliteValue::Integer(1));
        for _ in 0..65 {
            deep = RebaseExpr::Cast {
                expr: Box::new(deep),
                type_name: "INTEGER".to_owned(),
            };
        }
        let writer = expression_update(vec![(
            ColumnIdx::new(1),
            RebaseExpr::Literal(SqliteValue::Integer(2)),
        )]);
        assert_dependent(&expression_update(vec![(ColumnIdx::new(0), deep)]), &writer);
        let wide = RebaseExpr::Coalesce(vec![RebaseExpr::Literal(SqliteValue::Integer(1)); 4097]);
        assert_dependent(&expression_update(vec![(ColumnIdx::new(0), wide)]), &writer);
    }

    #[test]
    fn join_max_collapse_rejects_the_entire_mixed_input() {
        let column = ColumnIdx::new(0);
        let valid = max_update(column, 10);
        let other_column = max_update(ColumnIdx::new(1), 20);
        let non_max = RebaseExpr::Literal(SqliteValue::Integer(1));
        for invalid in [&other_column, &non_max] {
            assert_eq!(collapse_join_max_updates(column, &[&valid, invalid]), None);
            assert_eq!(collapse_join_max_updates(column, &[invalid, &valid]), None);
        }
        assert_eq!(collapse_join_max_updates(column, &[]), None);
        assert_eq!(
            collapse_join_max_updates(column, &[&valid, &valid]),
            Some(valid.clone())
        );
    }

    #[test]
    fn independent_small_updates_commute_under_snapshot_assignment() {
        // Independent test evaluator: every RHS sees the pre-statement row.
        fn apply(op: &IntentOp, row: [i64; 2]) -> [i64; 2] {
            let IntentOpKind::UpdateExpression { column_updates, .. } = &op.op else {
                panic!("test only builds expression updates");
            };
            let mut result = row;
            for (column, expr) in column_updates {
                let value = match expr {
                    RebaseExpr::ColumnRef(source) => row[usize::try_from(source.get()).unwrap()],
                    RebaseExpr::Literal(SqliteValue::Integer(value)) => *value,
                    _ => row[usize::try_from(column.get()).unwrap()]
                        .max(extract_join_max_constant(*column, expr).unwrap()),
                };
                result[usize::try_from(column.get()).unwrap()] = value;
            }
            result
        }
        let mut updates = Vec::new();
        for column in [ColumnIdx::new(0), ColumnIdx::new(1)] {
            for expr in [
                RebaseExpr::ColumnRef(ColumnIdx::new(0)),
                RebaseExpr::ColumnRef(ColumnIdx::new(1)),
                RebaseExpr::Literal(SqliteValue::Integer(-1)),
                RebaseExpr::Literal(SqliteValue::Integer(2)),
                max_update(column, 1),
                max_update(column, 3),
            ] {
                updates.push(expression_update(vec![(column, expr)]));
            }
        }
        for a in &updates {
            for b in &updates {
                assert_eq!(
                    are_intent_ops_independent(a, b),
                    are_intent_ops_independent(b, a)
                );
                if are_intent_ops_independent(a, b) {
                    for row in [[-2, 5], [0, 0], [9, -1]] {
                        assert_eq!(apply(b, apply(a, row)), apply(a, apply(b, row)));
                    }
                }
            }
        }
    }

    // -- Test 1: §5.10.6 PageHistory stores each retained version losslessly --

    #[test]
    fn test_page_history_retained_versions_are_full_images() {
        let pgno = PageNumber::new(5).unwrap();
        let images = vec![
            (CommitSeq::new(100), vec![0xAA; 4096]),
            (CommitSeq::new(50), vec![0xBB; 4096]),
            (CommitSeq::new(10), vec![0xCC; 4096]),
        ];

        let compressed = compress_page_history(pgno, &images).unwrap();

        assert_eq!(compressed.pgno, pgno);
        assert_eq!(compressed.versions.len(), 3);

        // Newest is full image.
        assert!(matches!(
            compressed.versions[0].data,
            CompressedVersionData::FullImage(ref img) if img == &vec![0xAA; 4096]
        ));
        assert_eq!(compressed.versions[0].commit_seq, CommitSeq::new(100));

        // Older retained versions are currently stored as full images. The
        // compression layer favors lossless reconstruction over patch density
        // until structured patch replay is fully wired.
        assert!(matches!(
            compressed.versions[1].data,
            CompressedVersionData::FullImage(ref img) if img == &vec![0xBB; 4096]
        ));
        assert!(matches!(
            compressed.versions[2].data,
            CompressedVersionData::FullImage(ref img) if img == &vec![0xCC; 4096]
        ));
    }

    // -- Test 2: §5.10.6 ECS encode/decode roundtrip --

    #[test]
    fn test_page_history_ecs_encode_decode_roundtrip() {
        let pgno = PageNumber::new(42).unwrap();
        let images = vec![
            (CommitSeq::new(200), vec![0x11; 512]),
            (CommitSeq::new(100), vec![0x22; 512]),
        ];

        let original = compress_page_history(pgno, &images).unwrap();
        let encoded = original.to_bytes();
        let decoded = CompressedPageHistory::from_bytes(&encoded).unwrap();

        assert_eq!(original.pgno, decoded.pgno);
        assert_eq!(original.versions.len(), decoded.versions.len());
        assert_eq!(
            original.versions[0].commit_seq,
            decoded.versions[0].commit_seq
        );

        // Full image roundtrips exactly.
        let newest_versions = (&original.versions[0].data, &decoded.versions[0].data);
        assert!(
            matches!(
                newest_versions,
                (
                    CompressedVersionData::FullImage(_),
                    CompressedVersionData::FullImage(_)
                )
            ),
            "expected FullImage for newest version"
        );
        let (CompressedVersionData::FullImage(orig), CompressedVersionData::FullImage(dec)) =
            newest_versions
        else {
            return;
        };
        assert_eq!(orig, dec);
    }

    #[test]
    fn test_page_history_decode_rejects_empty_history() {
        let mut encoded = Vec::new();
        encoded.extend_from_slice(&1_u32.to_le_bytes());
        encoded.extend_from_slice(&0_u32.to_le_bytes());

        assert!(matches!(
            CompressedPageHistory::from_bytes(&encoded),
            Err(HistoryCompressionError::EmptyHistory)
        ));
    }

    #[test]
    fn test_page_history_decode_rejects_non_full_newest_version() {
        let mut encoded = Vec::new();
        encoded.extend_from_slice(&1_u32.to_le_bytes());
        encoded.extend_from_slice(&1_u32.to_le_bytes());
        encoded.extend_from_slice(&10_u64.to_le_bytes());
        encoded.push(TAG_INTENT_LOG_PATCH);
        encoded.extend_from_slice(&0_u32.to_le_bytes());

        assert!(matches!(
            CompressedPageHistory::from_bytes(&encoded),
            Err(HistoryCompressionError::NewestNotFullImage)
        ));
    }

    #[test]
    fn test_page_history_decode_rejects_trailing_bytes() {
        let pgno = PageNumber::new(42).unwrap();
        let images = vec![(CommitSeq::new(200), vec![0x11; 512])];
        let original = compress_page_history(pgno, &images).unwrap();
        let mut encoded = original.to_bytes();
        encoded.extend_from_slice(b"extra");

        assert!(matches!(
            CompressedPageHistory::from_bytes(&encoded),
            Err(HistoryCompressionError::DecodeError(message))
                if message.contains("trailing bytes")
        ));
    }

    #[test]
    fn test_page_history_decode_rejects_impossible_version_count_before_allocating() {
        let mut encoded = Vec::new();
        encoded.extend_from_slice(&1_u32.to_le_bytes());
        encoded.extend_from_slice(&u32::MAX.to_le_bytes());

        assert!(matches!(
            CompressedPageHistory::from_bytes(&encoded),
            Err(HistoryCompressionError::DecodeError(message))
                if message.contains("truncated version entry")
        ));
    }

    #[test]
    fn test_page_history_decode_rejects_non_empty_intent_patch_payload() {
        let mut encoded = Vec::new();
        encoded.extend_from_slice(&1_u32.to_le_bytes());
        encoded.extend_from_slice(&2_u32.to_le_bytes());
        encoded.extend_from_slice(&20_u64.to_le_bytes());
        encoded.push(TAG_FULL_IMAGE);
        encoded.extend_from_slice(&1_u32.to_le_bytes());
        encoded.push(0xAA);
        encoded.extend_from_slice(&10_u64.to_le_bytes());
        encoded.push(TAG_INTENT_LOG_PATCH);
        encoded.extend_from_slice(&4_u32.to_le_bytes());
        encoded.extend_from_slice(&1_u32.to_le_bytes());

        assert!(matches!(
            CompressedPageHistory::from_bytes(&encoded),
            Err(HistoryCompressionError::DecodeError(message))
                if message.contains("intent log patch payload decode")
        ));
    }

    #[test]
    fn test_page_history_decode_rejects_non_empty_structured_patch_payload() {
        let mut encoded = Vec::new();
        encoded.extend_from_slice(&1_u32.to_le_bytes());
        encoded.extend_from_slice(&2_u32.to_le_bytes());
        encoded.extend_from_slice(&20_u64.to_le_bytes());
        encoded.push(TAG_FULL_IMAGE);
        encoded.extend_from_slice(&1_u32.to_le_bytes());
        encoded.push(0xAA);
        encoded.extend_from_slice(&10_u64.to_le_bytes());
        encoded.push(TAG_STRUCTURED_PATCH);
        encoded.extend_from_slice(&1_u32.to_le_bytes());
        encoded.push(0xBB);

        assert!(matches!(
            CompressedPageHistory::from_bytes(&encoded),
            Err(HistoryCompressionError::DecodeError(message))
                if message.contains("structured patch payload decode")
        ));
    }

    // -- Test 3: §5.10.7 Independence relation --

    #[allow(clippy::too_many_lines)]
    #[test]
    fn test_intent_independence_relation() {
        let t1 = TableId::new(1);
        let t2 = TableId::new(2);

        // Two inserts on different tables, different keys → independent.
        let insert_table1 = make_op_with_writes(
            1,
            IntentOpKind::Insert {
                table: t1,
                key: RowId::new(10),
                record: vec![1, 2, 3],
            },
            vec![table_key(1, 10)],
        );
        let insert_table2 = make_op_with_writes(
            1,
            IntentOpKind::Insert {
                table: t2,
                key: RowId::new(20),
                record: vec![4, 5, 6],
            },
            vec![table_key(2, 20)],
        );
        assert!(are_intent_ops_independent(&insert_table1, &insert_table2));

        // Two inserts on same table, same key → NOT independent (write overlap).
        let insert_same_key = make_op_with_writes(
            1,
            IntentOpKind::Insert {
                table: t1,
                key: RowId::new(10),
                record: vec![7, 8, 9],
            },
            vec![table_key(1, 10)],
        );
        assert!(!are_intent_ops_independent(
            &insert_table1,
            &insert_same_key
        ));

        // Different schema epochs → NOT independent.
        let different_epoch = make_op_with_writes(
            2,
            IntentOpKind::Insert {
                table: t2,
                key: RowId::new(30),
                record: vec![],
            },
            vec![table_key(2, 30)],
        );
        assert!(!are_intent_ops_independent(
            &insert_table1,
            &different_epoch
        ));

        // Structural effects → NOT independent.
        let structural_insert = IntentOp {
            schema_epoch: 1,
            footprint: IntentFootprint {
                reads: Vec::new(),
                writes: vec![table_key(2, 40)],
                structural: StructuralEffects::PAGE_SPLIT,
            },
            op: IntentOpKind::Insert {
                table: t2,
                key: RowId::new(40),
                record: vec![],
            },
        };
        assert!(!are_intent_ops_independent(
            &insert_table1,
            &structural_insert
        ));

        // Write/Read overlap → NOT independent.
        let read_write_overlap = IntentOp {
            schema_epoch: 1,
            footprint: IntentFootprint {
                reads: vec![table_key(1, 10)], // reads what `insert_table1` writes
                writes: vec![table_key(2, 50)],
                structural: StructuralEffects::NONE,
            },
            op: IntentOpKind::Update {
                table: t2,
                key: RowId::new(50),
                new_record: vec![],
            },
        };
        assert!(!are_intent_ops_independent(
            &insert_table1,
            &read_write_overlap
        ));

        // Index ops on different indices → independent.
        let idx_a = make_op_with_writes(
            1,
            IntentOpKind::IndexInsert {
                index: IndexId::new(1),
                key: vec![10],
                rowid: RowId::new(1),
            },
            vec![SemanticKeyRef::new(
                BtreeRef::Index(IndexId::new(1)),
                SemanticKeyKind::IndexEntry,
                &[10],
            )],
        );
        let idx_b = make_op_with_writes(
            1,
            IntentOpKind::IndexDelete {
                index: IndexId::new(2),
                key: vec![20],
                rowid: RowId::new(2),
            },
            vec![SemanticKeyRef::new(
                BtreeRef::Index(IndexId::new(2)),
                SemanticKeyKind::IndexEntry,
                &[20],
            )],
        );
        assert!(are_intent_ops_independent(&idx_a, &idx_b));
    }

    // -- Test 4: §5.10.7 UpdateExpression column disjoint commutativity --

    #[test]
    fn test_update_expression_column_disjoint_commutativity() {
        let t1 = TableId::new(1);
        let key = RowId::new(100);

        // Disjoint columns → independent.
        let update_col0 = make_op(
            1,
            IntentOpKind::UpdateExpression {
                table: t1,
                key,
                column_updates: vec![(
                    ColumnIdx::new(0),
                    RebaseExpr::Literal(SqliteValue::Integer(42)),
                )],
            },
        );
        let update_col1 = make_op(
            1,
            IntentOpKind::UpdateExpression {
                table: t1,
                key,
                column_updates: vec![(
                    ColumnIdx::new(1),
                    RebaseExpr::Literal(SqliteValue::Integer(99)),
                )],
            },
        );
        assert!(are_intent_ops_independent(&update_col0, &update_col1));

        // Overlapping columns (not join-max) → NOT independent.
        let overlapping_col0 = make_op(
            1,
            IntentOpKind::UpdateExpression {
                table: t1,
                key,
                column_updates: vec![(
                    ColumnIdx::new(0),
                    RebaseExpr::Literal(SqliteValue::Integer(77)),
                )],
            },
        );
        assert!(!are_intent_ops_independent(&update_col0, &overlapping_col0));

        // Different keys → fall through to general independence (independent).
        let different_key_update = make_op(
            1,
            IntentOpKind::UpdateExpression {
                table: t1,
                key: RowId::new(200),
                column_updates: vec![(
                    ColumnIdx::new(0),
                    RebaseExpr::Literal(SqliteValue::Integer(55)),
                )],
            },
        );
        assert!(are_intent_ops_independent(
            &update_col0,
            &different_key_update
        ));

        // UpdateExpression + materialized Update on same key → NOT independent.
        let materialized_update = make_op(
            1,
            IntentOpKind::Update {
                table: t1,
                key,
                new_record: vec![1, 2, 3],
            },
        );
        assert!(!are_intent_ops_independent(
            &update_col0,
            &materialized_update
        ));

        // UpdateExpression + Delete on same key → NOT independent.
        let delete_same_key = make_op(1, IntentOpKind::Delete { table: t1, key });
        assert!(!are_intent_ops_independent(&update_col0, &delete_same_key));
    }

    // -- Test 5: §5.10.7 Join-max-int-update recognition --

    #[test]
    fn test_join_max_int_update_recognized() {
        let col = ColumnIdx::new(3);

        // MAX(col, c) — standard order.
        let expr1 = RebaseExpr::FunctionCall {
            name: "MAX".to_owned(),
            args: vec![
                RebaseExpr::ColumnRef(col),
                RebaseExpr::Literal(SqliteValue::Integer(100)),
            ],
        };
        assert!(is_join_max_int_update(col, &expr1));

        // MAX(c, col) — reversed order.
        let expr2 = RebaseExpr::FunctionCall {
            name: "MAX".to_owned(),
            args: vec![
                RebaseExpr::Literal(SqliteValue::Integer(200)),
                RebaseExpr::ColumnRef(col),
            ],
        };
        assert!(is_join_max_int_update(col, &expr2));

        // Case-insensitive function name.
        let expr3 = RebaseExpr::FunctionCall {
            name: "max".to_owned(),
            args: vec![
                RebaseExpr::ColumnRef(col),
                RebaseExpr::Literal(SqliteValue::Integer(50)),
            ],
        };
        assert!(is_join_max_int_update(col, &expr3));

        // Wrong column → not recognized.
        assert!(!is_join_max_int_update(ColumnIdx::new(99), &expr1));

        // Not MAX function → not recognized.
        let expr4 = RebaseExpr::FunctionCall {
            name: "MIN".to_owned(),
            args: vec![
                RebaseExpr::ColumnRef(col),
                RebaseExpr::Literal(SqliteValue::Integer(10)),
            ],
        };
        assert!(!is_join_max_int_update(col, &expr4));

        // Non-integer literal → not recognized.
        let expr5 = RebaseExpr::FunctionCall {
            name: "MAX".to_owned(),
            args: vec![
                RebaseExpr::ColumnRef(col),
                RebaseExpr::Literal(SqliteValue::Text("hello".into())),
            ],
        };
        assert!(!is_join_max_int_update(col, &expr5));

        // Multiple join-max updates collapse correctly.
        let expressions: Vec<&RebaseExpr> = vec![&expr1, &expr2];
        let collapsed = collapse_join_max_updates(col, &expressions).unwrap();
        // max(100, 200) = 200.
        assert!(
            matches!(&collapsed, RebaseExpr::FunctionCall { .. }),
            "expected FunctionCall"
        );
        let RebaseExpr::FunctionCall { args, .. } = &collapsed else {
            return;
        };
        assert!(matches!(
            &args[1],
            RebaseExpr::Literal(SqliteValue::Integer(200))
        ));

        // Overlapping columns with join-max → independent.
        let op_a = make_op(
            1,
            IntentOpKind::UpdateExpression {
                table: TableId::new(1),
                key: RowId::new(1),
                column_updates: vec![(col, expr1)],
            },
        );
        let op_b = make_op(
            1,
            IntentOpKind::UpdateExpression {
                table: TableId::new(1),
                key: RowId::new(1),
                column_updates: vec![(col, expr2)],
            },
        );
        assert!(are_intent_ops_independent(&op_a, &op_b));
    }

    // -- Test 6: §5.10.8 MergeCertificate generation and verification --

    #[test]
    fn test_merge_certificate_generation_and_verification() {
        let ops = vec![
            make_op_with_writes(
                1,
                IntentOpKind::Insert {
                    table: TableId::new(1),
                    key: RowId::new(10),
                    record: vec![1, 2, 3],
                },
                vec![table_key(1, 10)],
            ),
            make_op_with_writes(
                1,
                IntentOpKind::Insert {
                    table: TableId::new(1),
                    key: RowId::new(20),
                    record: vec![4, 5, 6],
                },
                vec![table_key(1, 20)],
            ),
        ];

        let page_bytes = vec![0xAA; 4096];
        let pgno = PageNumber::new(3).unwrap();
        let affected_pages = vec![(pgno, page_bytes)];
        let btree_hash = blake3::hash(b"btree_invariant_ok");
        let mut btree_inv_hash = [0u8; 16];
        btree_inv_hash.copy_from_slice(&btree_hash.as_bytes()[..16]);

        let cert = generate_merge_certificate(
            MergeKind::StructuredPatch,
            50,
            1,
            &ops,
            &affected_pages,
            btree_inv_hash,
        )
        .unwrap();

        // Check certificate fields.
        assert_eq!(cert.merge_kind, MergeKind::StructuredPatch);
        assert_eq!(cert.base_commit_seq, 50);
        assert_eq!(cert.schema_epoch, 1);
        assert_eq!(cert.pages, vec![pgno]);
        assert_eq!(cert.intent_op_digests.len(), 2);
        assert_eq!(cert.normal_form.len(), 2);
        assert_eq!(cert.verifier_version, VERIFIER_VERSION);

        // Verification should pass.
        let result = verify_merge_certificate(&ops, &affected_pages, btree_inv_hash, &cert);
        assert!(result.is_ok());
    }

    // -- Test 7: §5.10.8 MergeCertificate replay deterministic --

    #[test]
    fn test_merge_certificate_replay_deterministic() {
        let ops = vec![
            make_op(
                1,
                IntentOpKind::Insert {
                    table: TableId::new(5),
                    key: RowId::new(1),
                    record: vec![10, 20],
                },
            ),
            make_op(
                1,
                IntentOpKind::Delete {
                    table: TableId::new(5),
                    key: RowId::new(2),
                },
            ),
            make_op(
                1,
                IntentOpKind::Insert {
                    table: TableId::new(5),
                    key: RowId::new(3),
                    record: vec![30, 40],
                },
            ),
        ];

        let pages = vec![
            (PageNumber::new(10).unwrap(), vec![0xBB; 2048]),
            (PageNumber::new(11).unwrap(), vec![0xCC; 2048]),
        ];
        let btree_hash = [0x42u8; 16];

        // Generate certificate.
        let cert = generate_merge_certificate(MergeKind::Rebase, 100, 1, &ops, &pages, btree_hash)
            .unwrap();

        // Re-execute: same inputs produce same hashes.
        let result = verify_merge_certificate(&ops, &pages, btree_hash, &cert);
        assert!(result.is_ok());

        // Determinism: generate again and compare.
        let cert2 = generate_merge_certificate(MergeKind::Rebase, 100, 1, &ops, &pages, btree_hash)
            .unwrap();
        assert_eq!(cert.intent_op_digests, cert2.intent_op_digests);
        assert_eq!(cert.footprint_digest, cert2.footprint_digest);
        assert_eq!(cert.normal_form, cert2.normal_form);
        assert_eq!(cert.post_state.page_hashes, cert2.post_state.page_hashes);
        assert_eq!(
            cert.post_state.btree_invariant_hash,
            cert2.post_state.btree_invariant_hash
        );
    }

    // -- Test 8: §5.10.8 Circuit breaker on verification failure --

    #[test]
    fn test_circuit_breaker_disables_merging_on_verification_failure() {
        let ops = vec![make_op(
            1,
            IntentOpKind::Insert {
                table: TableId::new(1),
                key: RowId::new(1),
                record: vec![1],
            },
        )];

        let pages = vec![(PageNumber::new(1).unwrap(), vec![0xFF; 4096])];
        let btree_hash = [0x00u8; 16];

        let cert =
            generate_merge_certificate(MergeKind::StructuredPatch, 10, 1, &ops, &pages, btree_hash)
                .unwrap();

        // Corrupt: tamper with the btree invariant hash.
        let tampered_btree_hash = [0xFFu8; 16];
        let result = verify_merge_certificate(&ops, &pages, tampered_btree_hash, &cert);
        assert!(result.is_err());

        let err = result.unwrap_err();
        assert!(matches!(
            err,
            CertificateVerificationError::BtreeInvariantHashMismatch { .. }
        ));

        // Circuit breaker fires.
        let event = circuit_breaker_check(err, &cert);
        assert!(event.disable_safe_merge);
        assert!(matches!(
            event.error,
            CertificateVerificationError::BtreeInvariantHashMismatch { .. }
        ));

        // Corrupt: tamper with page data.
        let tampered_pages = vec![(PageNumber::new(1).unwrap(), vec![0x00; 4096])];
        let result2 = verify_merge_certificate(&ops, &tampered_pages, btree_hash, &cert);
        assert!(result2.is_err());
        assert!(matches!(
            result2.unwrap_err(),
            CertificateVerificationError::PageHashMismatch { .. }
        ));

        // Corrupt: change an intent op.
        let tampered_ops = vec![make_op(
            1,
            IntentOpKind::Insert {
                table: TableId::new(1),
                key: RowId::new(1),
                record: vec![99], // different record
            },
        )];
        let result3 = verify_merge_certificate(&tampered_ops, &pages, btree_hash, &cert);
        assert!(result3.is_err());
        assert!(matches!(
            result3.unwrap_err(),
            CertificateVerificationError::OpDigestMismatch { .. }
        ));
    }

    // -- E2E: §5.10.6 history compression preserves retained query results --

    #[test]
    fn test_e2e_history_compression_preserves_query_results() {
        let pgno = PageNumber::new(7).unwrap();

        // Generate a 100-commit hot-page history (newest-first).
        let full_images: Vec<(CommitSeq, Vec<u8>)> = (1u64..=100)
            .rev()
            .map(|seq| {
                let mut page = vec![0u8; 256];
                page[..8].copy_from_slice(&seq.to_le_bytes());
                #[allow(clippy::cast_possible_truncation)]
                {
                    page[8] = (seq % 251) as u8;
                }
                (CommitSeq::new(seq), page)
            })
            .collect();

        let compressed = compress_page_history(pgno, &full_images).unwrap();
        let roundtripped = CompressedPageHistory::from_bytes(&compressed.to_bytes()).unwrap();

        // Current compression keeps every retained version materialized as a
        // full image. The retained query window therefore starts at the oldest
        // retained commit in this synthetic history.
        let retention_low = full_images
            .last()
            .expect("synthetic history should be non-empty")
            .0;

        let query_uncompressed = |snapshot: CommitSeq| -> Option<Vec<u8>> {
            if snapshot.get() < retention_low.get() {
                return None;
            }
            full_images
                .iter()
                .find(|(seq, _)| seq.get() <= snapshot.get())
                .map(|(_, page)| page.clone())
        };

        let query_compressed = |snapshot: CommitSeq| -> Option<Vec<u8>> {
            if snapshot.get() < retention_low.get() {
                return None;
            }
            roundtripped
                .versions
                .iter()
                .find(|version| version.commit_seq.get() <= snapshot.get())
                .and_then(|version| match &version.data {
                    CompressedVersionData::FullImage(page) => Some(page.clone()),
                    CompressedVersionData::IntentLogPatch(_)
                    | CompressedVersionData::StructuredPatch(_) => None,
                })
        };

        // Selected query points: two before retention (None expected), one at tip,
        // one beyond tip (should still see newest retained state).
        let snapshots = [
            CommitSeq::new(50),
            CommitSeq::new(99),
            CommitSeq::new(100),
            CommitSeq::new(101),
        ];

        for snapshot in snapshots {
            assert_eq!(
                query_uncompressed(snapshot),
                query_compressed(snapshot),
                "retained query mismatch at commit_seq={}",
                snapshot.get()
            );
        }
    }

    // -- Additional edge case tests --

    #[test]
    fn test_op_digest_deterministic() {
        let op = make_op(
            1,
            IntentOpKind::Insert {
                table: TableId::new(1),
                key: RowId::new(42),
                record: vec![1, 2, 3],
            },
        );

        let d1 = compute_op_digest(&op);
        let d2 = compute_op_digest(&op);
        assert_eq!(d1, d2, "op digest must be deterministic");

        // Different op produces different digest.
        let op2 = make_op(
            1,
            IntentOpKind::Insert {
                table: TableId::new(1),
                key: RowId::new(43),
                record: vec![1, 2, 3],
            },
        );
        let d3 = compute_op_digest(&op2);
        assert_ne!(d1, d3);
    }

    #[test]
    fn test_foata_sort_table_row_digest_matches_cell_key() {
        let table = TableId::new(7);
        let rowid = RowId::new(-42);
        let op = make_op(
            1,
            IntentOpKind::Update {
                table,
                key: rowid,
                new_record: vec![1, 2, 3],
            },
        );

        let components = op_sort_components(&op);
        let expected =
            crate::cell_visibility::CellKey::table_row(BtreeRef::Table(table), rowid.get())
                .key_digest;
        let little_endian = SemanticKeyRef::compute_digest(
            SemanticKeyKind::TableRow,
            BtreeRef::Table(table),
            &rowid.get().to_le_bytes(),
        );

        assert_eq!(components.key_digest, expected);
        assert_ne!(
            components.key_digest, little_endian,
            "Foata ordering must use canonical rowid varint bytes, not little-endian i64 bytes"
        );
    }

    #[test]
    fn test_foata_normal_form_independent_ops_single_layer() {
        // Three independent ops should all be in one Foata layer.
        let ops = vec![
            make_op_with_writes(
                1,
                IntentOpKind::Insert {
                    table: TableId::new(1),
                    key: RowId::new(1),
                    record: vec![],
                },
                vec![table_key(1, 1)],
            ),
            make_op_with_writes(
                1,
                IntentOpKind::Insert {
                    table: TableId::new(1),
                    key: RowId::new(2),
                    record: vec![],
                },
                vec![table_key(1, 2)],
            ),
            make_op_with_writes(
                1,
                IntentOpKind::Insert {
                    table: TableId::new(1),
                    key: RowId::new(3),
                    record: vec![],
                },
                vec![table_key(1, 3)],
            ),
        ];

        let nf = foata_normal_form(&ops);
        assert_eq!(nf.len(), 3);

        // All three digests should be present.
        let digests: BTreeSet<[u8; 16]> = ops.iter().map(compute_op_digest).collect();
        let nf_set: BTreeSet<[u8; 16]> = nf.into_iter().collect();
        assert_eq!(digests, nf_set);
    }

    #[test]
    fn test_foata_normal_form_dependent_ops_multiple_layers() {
        // Two ops on same key → dependent. One must come before the other.
        let ops = vec![
            make_op_with_writes(
                1,
                IntentOpKind::Insert {
                    table: TableId::new(1),
                    key: RowId::new(1),
                    record: vec![1],
                },
                vec![table_key(1, 1)],
            ),
            make_op_with_writes(
                1,
                IntentOpKind::Update {
                    table: TableId::new(1),
                    key: RowId::new(1),
                    new_record: vec![2],
                },
                vec![table_key(1, 1)],
            ),
        ];

        let nf = foata_normal_form(&ops);
        assert_eq!(nf.len(), 2);

        // First should be Insert's digest (lower index = preserved order).
        assert_eq!(nf[0], compute_op_digest(&ops[0]));
        assert_eq!(nf[1], compute_op_digest(&ops[1]));
    }

    #[test]
    fn test_empty_history_returns_error() {
        let pgno = PageNumber::new(1).unwrap();
        let result = compress_page_history(pgno, &[]);
        assert!(matches!(result, Err(HistoryCompressionError::EmptyHistory)));
    }

    #[test]
    fn test_is_mergeable_intent_structural_rejected() {
        let op = IntentOp {
            schema_epoch: 1,
            footprint: IntentFootprint {
                reads: Vec::new(),
                writes: Vec::new(),
                structural: StructuralEffects::PAGE_SPLIT,
            },
            op: IntentOpKind::Insert {
                table: TableId::new(1),
                key: RowId::new(1),
                record: vec![],
            },
        };
        assert!(!is_mergeable_intent(&op));
    }
}