lix 0.12.0

//! Host-neutral contract for the Lix plugin Wasm Component protocol.
//!
//! The Component binding lives under `lix::plugin`; this module deliberately
//! contains no Wasmtime types. A compiled component factory is shared, while
//! each branch/file actor owns one isolated mutable instance and all document,
//! cursor, output-table, and transition handles created by that instance.

use std::any::Any;
use std::collections::BTreeSet;
use std::fmt;
use std::sync::{Arc, OnceLock};

use async_trait::async_trait;
use bytes::Bytes;
use serde_json::Value as JsonValue;
use smallvec::SmallVec;

use crate::{
    LixError, catalog::SchemaPlanFingerprint, common::SharedStr, row_pk::RowPk, wasm::WasmLimits,
};

pub const PACKET_FORMAT_V1: u16 = 1;
pub const WASM_COMPONENT_API_VERSION: &str = "1.0.0";
/// Canonical ABI page charge for one renderer splice before inline insert
/// bytes. Both inline and output-backed edits pay this fixed metadata cost.
pub const EDIT_SPLICE_METADATA_BYTES: u64 = 24;

const MIB: u64 = 1024 * 1024;
const MIB_U32: u32 = 1024 * 1024;
const TRANSITION_PAGE_BYTES: u32 = 2 * MIB_U32;
const COLD_TRANSITION_MAX_PAGE_BYTES: u64 = 16 * MIB;
const COLD_TRANSITION_RECORD_OVERHEAD_BYTES: u64 = 64 * 1024;
const COLD_FILE_MAX_DEADLINE_NANOSECONDS: u64 = 60_000_000_000;
const COLD_FILE_EXTRA_DEADLINE_NANOSECONDS_PER_MIB: u64 = 1_000_000_000;

/// Aggregate limits for one top-level component transition, including cursor and
/// attachment draining after the exported guest call returns.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WasmTransitionLimits {
    pub max_record_bytes: u32,
    pub max_page_bytes: u32,
    pub max_pages: u32,
    pub max_total_bytes: u64,
    pub max_inline_edits: u32,
    pub max_inline_input_bytes: u64,
    pub max_attachment_refs: u32,
    pub total_deadline_nanoseconds: u64,
}

impl Default for WasmTransitionLimits {
    fn default() -> Self {
        Self {
            max_record_bytes: MIB_U32,
            // Match the largest fixed page schedule that the v3 push sink
            // safely uses under the same Wasm linear-memory limit. Records
            // remain capped independently at one MiB.
            max_page_bytes: TRANSITION_PAGE_BYTES,
            max_pages: 1_024,
            // A 10 MiB recursive JSON import can legitimately carry roughly
            // one compact row per leaf plus packet keys. Keep the
            // transition bounded, but do not confuse aggregate paging traffic
            // with the component's independent linear-memory ceiling.
            max_total_bytes: 128 * MIB,
            max_inline_edits: 4_096,
            max_inline_input_bytes: MIB,
            max_attachment_refs: 4_096,
            total_deadline_nanoseconds: 5_000_000_000,
        }
    }
}

impl WasmTransitionLimits {
    /// Returns the normal transition budget with enough bounded page space for
    /// one semantic record derived from the supplied file bytes.
    pub fn for_file_bytes(file_bytes: u64) -> Self {
        let mut limits = Self::default();
        limits.scale_page_for_file_bytes(file_bytes);
        limits
    }

    /// Returns a bounded budget for a first cold parse of one submitted file.
    ///
    /// A fresh `open-file` must legitimately examine all input bytes, unlike a
    /// warm sparse edit or column merge. Keep the normal five
    /// second bound for a zero/small file, add one second per started MiB, and
    /// cap the result at one minute. This is admission policy, not an excuse
    /// for a plugin to make the hot path proportional to file size.
    pub fn for_cold_file_bytes(file_bytes: u64) -> Self {
        let mut limits = Self::default();
        limits.scale_page_for_file_bytes(file_bytes);
        let extra = file_bytes
            .div_ceil(MIB)
            .saturating_mul(COLD_FILE_EXTRA_DEADLINE_NANOSECONDS_PER_MIB);
        limits.total_deadline_nanoseconds = limits
            .total_deadline_nanoseconds
            .saturating_add(extra)
            .min(COLD_FILE_MAX_DEADLINE_NANOSECONDS);
        limits
    }

    fn scale_page_for_file_bytes(&mut self, file_bytes: u64) {
        // Text rows encode their byte-exact content as unpadded base64.
        // A generated/source-map file may legitimately be one long line, so
        // its single semantic record can be larger than the normal 2 MiB
        // scheduling page. Retain a hard 16 MiB page bound; small files and
        // ordinary sparse transitions keep the fixed 2 MiB schedule.
        let encoded_record_bytes = file_bytes
            .saturating_mul(4)
            .div_ceil(3)
            .saturating_add(COLD_TRANSITION_RECORD_OVERHEAD_BYTES);
        let cold_page_bytes = encoded_record_bytes
            .max(u64::from(TRANSITION_PAGE_BYTES))
            .min(COLD_TRANSITION_MAX_PAGE_BYTES)
            .min(self.max_total_bytes);
        self.max_page_bytes = u32::try_from(cold_page_bytes).unwrap_or(u32::MAX);
        self.max_record_bytes = self.max_page_bytes;
    }

    pub fn validate(self) -> Result<Self, LixError> {
        if self.max_record_bytes == 0
            || self.max_page_bytes == 0
            || self.max_pages == 0
            || self.max_total_bytes == 0
            || self.max_inline_edits == 0
            || self.max_attachment_refs == 0
            || self.total_deadline_nanoseconds == 0
        {
            return Err(invalid_param(
                "component transition limits must use positive record, page, count, byte, reference, and deadline bounds",
            ));
        }
        if self.max_record_bytes > self.max_page_bytes {
            return Err(invalid_param(
                "component max_record_bytes must not exceed max_page_bytes",
            ));
        }
        if u64::from(self.max_page_bytes) > self.max_total_bytes {
            return Err(invalid_param(
                "component max_page_bytes must not exceed max_total_bytes",
            ));
        }
        if self.max_inline_input_bytes > self.max_total_bytes {
            return Err(invalid_param(
                "component max_inline_input_bytes must not exceed max_total_bytes",
            ));
        }
        Ok(self)
    }
}

/// Measurable work for a component transition. Engine-owned counters and binding-
/// owned counters share one snapshot so benchmarks can fail on hidden
/// O(document) work even when wall-clock timing happens to improve.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct WasmTransitionCounters {
    pub source_read_calls: u64,
    pub source_bytes_read: u64,
    /// File-byte reads through the immutable snapshot capability.
    pub file_read_calls: u64,
    pub file_bytes_read: u64,
    /// Opaque plugin-state reads through the immutable snapshot capability.
    pub state_read_calls: u64,
    pub state_key_bytes: u64,
    pub state_value_bytes_read: u64,
    pub packet_pages: u64,
    pub packet_records: u64,
    /// Universal row pages sent from host to guest.
    pub row_input_pages: u64,
    pub row_input_records: u64,
    pub row_input_wire_bytes: u64,
    /// Universal row pages sent from guest to host.
    pub row_output_pages: u64,
    pub row_output_records: u64,
    pub row_output_wire_bytes: u64,
    pub attachment_reads: u64,
    pub attachment_bytes_read: u64,
    pub row_input_attachment_reads: u64,
    pub row_input_attachment_bytes: u64,
    pub row_output_attachment_writes: u64,
    pub row_output_attachment_bytes: u64,
    pub component_import_calls: u64,
    /// Exported guest entries. Prototype A requires exactly one for an initial
    /// file transition, independent of page and source-read counts.
    pub guest_export_calls: u64,
    /// Persistent executor threads created for the actor while serving this
    /// transition. This is one on the actor's first transition and zero for
    /// subsequent transitions; it must never scale with source/sink calls.
    pub actor_executor_threads_created: u64,
    pub component_boundary_bytes: u64,
    pub guest_linear_memory_high_water_bytes: u64,
    /// Bytes examined by the host-only full-blob diff fallback. Validated
    /// transport splice provenance keeps this zero.
    pub host_full_diff_bytes_compared: u64,
    /// Bytes examined by host-only content-type classification. A matcher may
    /// deliberately use a bounded predicate (for example the text plugin.s 8 KiB NUL
    /// window), while a UTF-8 parser examines the complete payload. Warm,
    /// provenance-backed edits keep this zero when their invariant proof is
    /// sufficient.
    pub host_content_classification_bytes: u64,
    pub full_state_semantic_rows_materialized: u64,
    pub change_payload_requests: u64,
    pub returned_change_payloads: u64,
    pub durable_semantic_changes: u64,
    pub private_document_cache_hits: u64,
    pub shared_renderer_cache_hits: u64,
    pub full_document_reparses: u64,
    pub full_renderer_invocations: u64,
    pub filesystem_sync_full_renders: u64,
    /// Static conflict-resolution guest calls. A merge batches all colliding
    /// rows for one file/plugin generation into one call, so this should
    /// remain O(files), not O(conflicts).
    pub conflict_resolution_calls: u64,
    /// Conflict triples delivered to a plugin resolver.
    pub conflict_resolution_records: u64,
    /// Resolutions that selected an immutable input version without returning
    /// a replacement snapshot through guest linear memory.
    pub conflict_resolution_takes: u64,
}

impl WasmTransitionCounters {
    /// Adds one completed transition snapshot to an engine-wide aggregate.
    ///
    /// Counters saturate instead of wrapping so diagnostic instrumentation can
    /// never report a deceptively small value after a long-running process.
    pub fn accumulate(&mut self, other: Self) {
        self.source_read_calls = self
            .source_read_calls
            .saturating_add(other.source_read_calls);
        self.source_bytes_read = self
            .source_bytes_read
            .saturating_add(other.source_bytes_read);
        self.file_read_calls = self.file_read_calls.saturating_add(other.file_read_calls);
        self.file_bytes_read = self.file_bytes_read.saturating_add(other.file_bytes_read);
        self.state_read_calls = self.state_read_calls.saturating_add(other.state_read_calls);
        self.state_key_bytes = self.state_key_bytes.saturating_add(other.state_key_bytes);
        self.state_value_bytes_read = self
            .state_value_bytes_read
            .saturating_add(other.state_value_bytes_read);
        self.packet_pages = self.packet_pages.saturating_add(other.packet_pages);
        self.packet_records = self.packet_records.saturating_add(other.packet_records);
        self.row_input_pages = self.row_input_pages.saturating_add(other.row_input_pages);
        self.row_input_records = self
            .row_input_records
            .saturating_add(other.row_input_records);
        self.row_input_wire_bytes = self
            .row_input_wire_bytes
            .saturating_add(other.row_input_wire_bytes);
        self.row_output_pages = self.row_output_pages.saturating_add(other.row_output_pages);
        self.row_output_records = self
            .row_output_records
            .saturating_add(other.row_output_records);
        self.row_output_wire_bytes = self
            .row_output_wire_bytes
            .saturating_add(other.row_output_wire_bytes);
        self.attachment_reads = self.attachment_reads.saturating_add(other.attachment_reads);
        self.attachment_bytes_read = self
            .attachment_bytes_read
            .saturating_add(other.attachment_bytes_read);
        self.row_input_attachment_reads = self
            .row_input_attachment_reads
            .saturating_add(other.row_input_attachment_reads);
        self.row_input_attachment_bytes = self
            .row_input_attachment_bytes
            .saturating_add(other.row_input_attachment_bytes);
        self.row_output_attachment_writes = self
            .row_output_attachment_writes
            .saturating_add(other.row_output_attachment_writes);
        self.row_output_attachment_bytes = self
            .row_output_attachment_bytes
            .saturating_add(other.row_output_attachment_bytes);
        self.component_import_calls = self
            .component_import_calls
            .saturating_add(other.component_import_calls);
        self.guest_export_calls = self
            .guest_export_calls
            .saturating_add(other.guest_export_calls);
        self.actor_executor_threads_created = self
            .actor_executor_threads_created
            .saturating_add(other.actor_executor_threads_created);
        self.component_boundary_bytes = self
            .component_boundary_bytes
            .saturating_add(other.component_boundary_bytes);
        self.guest_linear_memory_high_water_bytes = self
            .guest_linear_memory_high_water_bytes
            .max(other.guest_linear_memory_high_water_bytes);
        self.host_full_diff_bytes_compared = self
            .host_full_diff_bytes_compared
            .saturating_add(other.host_full_diff_bytes_compared);
        self.host_content_classification_bytes = self
            .host_content_classification_bytes
            .saturating_add(other.host_content_classification_bytes);
        self.full_state_semantic_rows_materialized = self
            .full_state_semantic_rows_materialized
            .saturating_add(other.full_state_semantic_rows_materialized);
        self.change_payload_requests = self
            .change_payload_requests
            .saturating_add(other.change_payload_requests);
        self.returned_change_payloads = self
            .returned_change_payloads
            .saturating_add(other.returned_change_payloads);
        self.durable_semantic_changes = self
            .durable_semantic_changes
            .saturating_add(other.durable_semantic_changes);
        self.private_document_cache_hits = self
            .private_document_cache_hits
            .saturating_add(other.private_document_cache_hits);
        self.shared_renderer_cache_hits = self
            .shared_renderer_cache_hits
            .saturating_add(other.shared_renderer_cache_hits);
        self.full_document_reparses = self
            .full_document_reparses
            .saturating_add(other.full_document_reparses);
        self.full_renderer_invocations = self
            .full_renderer_invocations
            .saturating_add(other.full_renderer_invocations);
        self.filesystem_sync_full_renders = self
            .filesystem_sync_full_renders
            .saturating_add(other.filesystem_sync_full_renders);
        self.conflict_resolution_calls = self
            .conflict_resolution_calls
            .saturating_add(other.conflict_resolution_calls);
        self.conflict_resolution_records = self
            .conflict_resolution_records
            .saturating_add(other.conflict_resolution_records);
        self.conflict_resolution_takes = self
            .conflict_resolution_takes
            .saturating_add(other.conflict_resolution_takes);
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WasmPluginSelection {
    pub plugin_key: String,
    /// Content-addressed component generation selected by the engine.
    pub generation: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WasmFileDescriptor {
    pub file_id: String,
    pub path: Option<String>,
    pub plugin: WasmPluginSelection,
}

impl WasmFileDescriptor {
    pub fn validate_warm_successor(&self, after: &Self) -> Result<(), LixError> {
        if self.file_id != after.file_id {
            return Err(invalid_param(
                "warm component transitions require the same stable file id",
            ));
        }
        if self.plugin != after.plugin {
            return Err(invalid_param(
                "warm component transitions require the same plugin key and generation",
            ));
        }
        Ok(())
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WasmSourceRange {
    pub offset: u64,
    pub length: u64,
}

impl WasmSourceRange {
    pub fn end(self) -> Result<u64, LixError> {
        self.offset
            .checked_add(self.length)
            .ok_or_else(|| invalid_param("component source range overflowed"))
    }
}

/// Immutable random-access bytes owned by the engine.
pub trait WasmByteSource: Send + Sync {
    fn len(&self) -> u64;

    fn is_empty(&self) -> bool {
        self.len() == 0
    }

    fn read(&self, offset: u64, length: u32) -> Result<Vec<u8>, LixError>;
}

#[derive(Clone)]
pub struct WasmSourceSlice {
    pub source: Arc<dyn WasmByteSource>,
    pub range: WasmSourceRange,
}

impl fmt::Debug for WasmSourceSlice {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("WasmSourceSlice")
            .field("source_len", &self.source.len())
            .field("range", &self.range)
            .finish()
    }
}

impl WasmSourceSlice {
    pub fn validate(&self) -> Result<(), LixError> {
        if self.range.end()? > self.source.len() {
            return Err(invalid_param("component source slice is out of bounds"));
        }
        Ok(())
    }
}

#[derive(Debug, Clone)]
pub enum WasmHostBytes {
    /// Immutable bytes that can retain a slice of a materialization batch
    /// without allocating a row-owned buffer.
    Inline(Bytes),
    Source(WasmSourceSlice),
    CanonicalJson(WasmCanonicalJson),
}

/// One canonical JSON row backed by cursor-page storage.
///
/// Cloning a row clones only this owner pointer and row ordinal. Snapshot
/// values and canonical bytes stay owned once by the bounded batch produced
/// while draining the enclosing component cursor page. Certified pages retain their
/// original shared UTF-8 row buffers; decoded or mixed pages use one compact
/// normalized arena.
#[derive(Debug, Clone)]
pub struct WasmCanonicalJson {
    batch: Arc<WasmCanonicalJsonBatch>,
    row: u32,
}

#[derive(Debug)]
struct WasmCanonicalJsonBatch {
    storage: WasmCanonicalJsonStorage,
    parse_count: usize,
    serialize_count: usize,
    arena_allocation_count: u8,
}

#[derive(Debug)]
enum WasmCanonicalJsonStorage {
    Arena {
        values: Box<[Option<JsonValue>]>,
        certificates: Box<[Option<WasmCanonicalJsonCertificate>]>,
        normalized: SharedStr,
        offsets: Box<[WasmCanonicalJsonOffset]>,
    },
    CertifiedRows {
        decoded_values: OnceLock<Box<[OnceLock<JsonValue>]>>,
        row_pks: Box<[RowPk]>,
        schema_fingerprints: Box<[Arc<SchemaPlanFingerprint>]>,
        schema_fingerprint_indices: Box<[u32]>,
        normalized: Box<[SharedStr]>,
        normalized_len: u32,
    },
}

/// Schema and identity facts proven while streaming one exact canonical guest
/// snapshot. The row key remains the protocol's owner of schema metadata;
/// this certificate retains the typed identity plus one shared fingerprint of
/// the exact schema plan against which the proof was issued.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct WasmCanonicalJsonCertificate {
    row_pk: RowPk,
    schema_fingerprint: Arc<SchemaPlanFingerprint>,
}

#[derive(Debug, Clone, Copy)]
pub(crate) struct WasmCanonicalJsonCertificateRef<'a> {
    row_pk: &'a RowPk,
    schema_fingerprint: &'a Arc<SchemaPlanFingerprint>,
}

#[cfg(test)]
impl WasmCanonicalJsonCertificateRef<'_> {
    pub(crate) fn row_pk(&self) -> &RowPk {
        self.row_pk
    }
}

impl WasmCanonicalJsonCertificate {
    pub(crate) fn new(row_pk: RowPk, schema_fingerprint: Arc<SchemaPlanFingerprint>) -> Self {
        Self {
            row_pk,
            schema_fingerprint,
        }
    }

    pub(crate) fn row_pk(&self) -> &RowPk {
        &self.row_pk
    }

    pub(crate) fn schema_fingerprint(&self) -> &SchemaPlanFingerprint {
        self.schema_fingerprint.as_ref()
    }

    fn borrowed(&self) -> WasmCanonicalJsonCertificateRef<'_> {
        WasmCanonicalJsonCertificateRef {
            row_pk: &self.row_pk,
            schema_fingerprint: &self.schema_fingerprint,
        }
    }
}

impl WasmCanonicalJsonCertificateRef<'_> {
    pub(crate) fn into_owned(self) -> WasmCanonicalJsonCertificate {
        WasmCanonicalJsonCertificate {
            row_pk: self.row_pk.clone(),
            schema_fingerprint: self.schema_fingerprint.clone(),
        }
    }
}

#[derive(Debug, Clone, Copy)]
struct WasmCanonicalJsonOffset {
    start: u32,
    end: u32,
}

impl WasmCanonicalJson {
    pub(crate) fn from_batch_parts(
        values: Vec<JsonValue>,
        normalized: Vec<u8>,
        offsets: Vec<(u32, u32)>,
        parse_count: usize,
        serialize_count: usize,
    ) -> Result<Vec<Self>, LixError> {
        let row_count = values.len();
        let mut optional_values = Vec::with_capacity(row_count);
        optional_values.extend(values.into_iter().map(Some));
        Self::from_mixed_batch_parts(
            optional_values,
            vec![None; row_count],
            normalized,
            offsets,
            parse_count,
            serialize_count,
        )
    }

    pub(crate) fn from_mixed_batch_parts(
        values: Vec<Option<JsonValue>>,
        certificates: Vec<Option<WasmCanonicalJsonCertificate>>,
        normalized: Vec<u8>,
        offsets: Vec<(u32, u32)>,
        parse_count: usize,
        serialize_count: usize,
    ) -> Result<Vec<Self>, LixError> {
        if values.len() != offsets.len() {
            return Err(invalid_param(
                "canonical JSON batch value and offset counts differ",
            ));
        }
        if certificates.len() != offsets.len() {
            return Err(invalid_param(
                "canonical JSON batch certificate and offset counts differ",
            ));
        }
        if values
            .iter()
            .zip(&certificates)
            .any(|(value, certificate)| value.is_some() == certificate.is_some())
        {
            return Err(invalid_param(
                "canonical JSON batch rows must own exactly one decoded value or certificate",
            ));
        }
        // `Bytes::from(Vec<u8>)` retains the Vec allocation even when it has
        // spare capacity. `SharedStr` then validates and retains that same
        // immutable allocation, so row materialization can take cheap slices
        // instead of copying each canonical payload.
        let arena_allocation_count = u8::from(normalized.capacity() != 0);
        let normalized = SharedStr::from_utf8(Bytes::from(normalized))
            .map_err(|_| invalid_param("canonical JSON batch arena is not UTF-8"))?;
        let arena_len = u32::try_from(normalized.len())
            .map_err(|_| invalid_param("canonical JSON batch arena exceeds u32"))?;
        let mut previous_end = 0_u32;
        let mut validated_offsets = Vec::with_capacity(offsets.len());
        for (start, end) in offsets {
            if start != previous_end || end < start || end > arena_len {
                return Err(invalid_param(
                    "canonical JSON batch offsets are invalid or non-contiguous",
                ));
            }
            if !normalized.as_str().is_char_boundary(start as usize)
                || !normalized.as_str().is_char_boundary(end as usize)
            {
                return Err(invalid_param(
                    "canonical JSON batch offsets split a UTF-8 scalar",
                ));
            }
            previous_end = end;
            validated_offsets.push(WasmCanonicalJsonOffset { start, end });
        }
        if previous_end != arena_len {
            return Err(invalid_param(
                "canonical JSON batch offsets do not cover the arena",
            ));
        }

        Self::from_validated_batch(WasmCanonicalJsonBatch {
            storage: WasmCanonicalJsonStorage::Arena {
                values: values.into_boxed_slice(),
                certificates: certificates.into_boxed_slice(),
                normalized,
                offsets: validated_offsets.into_boxed_slice(),
            },
            parse_count,
            serialize_count,
            arena_allocation_count,
        })
    }

    pub(crate) fn from_certified_batch_parts(
        normalized: Vec<SharedStr>,
        row_pks: Vec<RowPk>,
        schema_fingerprints: Vec<Arc<SchemaPlanFingerprint>>,
        schema_fingerprint_indices: Vec<u32>,
        parse_count: usize,
    ) -> Result<Vec<Self>, LixError> {
        if normalized.len() != row_pks.len() || normalized.len() != schema_fingerprint_indices.len()
        {
            return Err(invalid_param(
                "certified canonical JSON batch row and metadata counts differ",
            ));
        }
        if schema_fingerprint_indices
            .iter()
            .any(|index| *index as usize >= schema_fingerprints.len())
        {
            return Err(invalid_param(
                "certified canonical JSON batch schema index is invalid",
            ));
        }

        let normalized_len = normalized.iter().try_fold(0_u32, |total, row| {
            let row_len = u32::try_from(row.len())
                .map_err(|_| invalid_param("certified canonical JSON row exceeds u32"))?;
            total
                .checked_add(row_len)
                .ok_or_else(|| invalid_param("certified canonical JSON batch exceeds u32"))
        })?;

        Self::from_validated_batch(WasmCanonicalJsonBatch {
            storage: WasmCanonicalJsonStorage::CertifiedRows {
                decoded_values: OnceLock::new(),
                row_pks: row_pks.into_boxed_slice(),
                schema_fingerprints: schema_fingerprints.into_boxed_slice(),
                schema_fingerprint_indices: schema_fingerprint_indices.into_boxed_slice(),
                normalized: normalized.into_boxed_slice(),
                normalized_len,
            },
            parse_count,
            serialize_count: 0,
            arena_allocation_count: 0,
        })
    }

    fn from_validated_batch(batch: WasmCanonicalJsonBatch) -> Result<Vec<Self>, LixError> {
        let batch = Arc::new(batch);
        let mut rows = Vec::with_capacity(batch.row_count());
        for row in 0..batch.row_count() {
            rows.push(Self {
                batch: batch.clone(),
                row: u32::try_from(row)
                    .map_err(|_| invalid_param("canonical JSON batch has too many rows"))?,
            });
        }
        Ok(rows)
    }

    pub fn value(&self) -> &JsonValue {
        match &self.batch.storage {
            WasmCanonicalJsonStorage::Arena { values, .. } => values[self.row_index()]
                .as_ref()
                .expect("certified canonical JSON rows do not own decoded values"),
            WasmCanonicalJsonStorage::CertifiedRows {
                decoded_values,
                normalized,
                ..
            } => decoded_values
                .get_or_init(|| (0..normalized.len()).map(|_| OnceLock::new()).collect())
                [self.row_index()]
            .get_or_init(|| {
                serde_json::from_str(normalized[self.row_index()].as_str())
                    .expect("certified canonical JSON must parse")
            }),
        }
    }

    pub(crate) fn certificate(&self) -> Option<WasmCanonicalJsonCertificateRef<'_>> {
        match &self.batch.storage {
            WasmCanonicalJsonStorage::Arena { certificates, .. } => certificates[self.row_index()]
                .as_ref()
                .map(WasmCanonicalJsonCertificate::borrowed),
            WasmCanonicalJsonStorage::CertifiedRows {
                row_pks,
                schema_fingerprints,
                schema_fingerprint_indices,
                ..
            } => Some(WasmCanonicalJsonCertificateRef {
                row_pk: &row_pks[self.row_index()],
                schema_fingerprint: &schema_fingerprints
                    [schema_fingerprint_indices[self.row_index()] as usize],
            }),
        }
    }

    pub fn normalized(&self) -> &str {
        match &self.batch.storage {
            WasmCanonicalJsonStorage::Arena {
                normalized,
                offsets,
                ..
            } => normalized
                .as_str()
                .get(offset_range(offsets[self.row_index()]))
                .expect("canonical JSON row offsets were validated at batch construction"),
            WasmCanonicalJsonStorage::CertifiedRows { normalized, .. } => {
                normalized[self.row_index()].as_str()
            }
        }
    }

    pub(crate) fn normalized_shared(&self) -> SharedStr {
        match &self.batch.storage {
            WasmCanonicalJsonStorage::Arena {
                normalized,
                offsets,
                ..
            } => normalized
                .slice(offset_range(offsets[self.row_index()]))
                .expect("canonical JSON row offsets were validated at batch construction"),
            WasmCanonicalJsonStorage::CertifiedRows { normalized, .. } => {
                normalized[self.row_index()].clone()
            }
        }
    }

    pub fn row_index(&self) -> usize {
        self.row as usize
    }

    pub fn batch_row_count(&self) -> usize {
        self.batch.row_count()
    }

    pub fn batch_arena_len(&self) -> usize {
        match &self.batch.storage {
            WasmCanonicalJsonStorage::Arena { normalized, .. } => normalized.len(),
            WasmCanonicalJsonStorage::CertifiedRows { normalized_len, .. } => {
                *normalized_len as usize
            }
        }
    }

    pub fn shares_batch_with(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.batch, &other.batch)
    }

    pub fn validation_counts(&self) -> (usize, usize) {
        (self.batch.parse_count, self.batch.serialize_count)
    }

    pub fn batch_arena_allocation_count(&self) -> usize {
        self.batch.arena_allocation_count as usize
    }

    #[cfg(test)]
    pub(crate) fn batch_decoded_value_count(&self) -> usize {
        match &self.batch.storage {
            WasmCanonicalJsonStorage::Arena { values, .. } => {
                values.iter().filter(|value| value.is_some()).count()
            }
            WasmCanonicalJsonStorage::CertifiedRows { decoded_values, .. } => decoded_values
                .get()
                .map(|values| values.iter().filter(|value| value.get().is_some()).count())
                .unwrap_or(0),
        }
    }

    #[cfg(test)]
    pub(crate) fn batch_certified_schema_count(&self) -> usize {
        match &self.batch.storage {
            WasmCanonicalJsonStorage::Arena { .. } => 0,
            WasmCanonicalJsonStorage::CertifiedRows {
                schema_fingerprints,
                ..
            } => schema_fingerprints.len(),
        }
    }
}

impl WasmCanonicalJsonBatch {
    fn row_count(&self) -> usize {
        match &self.storage {
            WasmCanonicalJsonStorage::Arena { offsets, .. } => offsets.len(),
            WasmCanonicalJsonStorage::CertifiedRows { normalized, .. } => normalized.len(),
        }
    }
}

fn offset_range(offset: WasmCanonicalJsonOffset) -> std::ops::Range<usize> {
    offset.start as usize..offset.end as usize
}

impl WasmHostBytes {
    pub fn len(&self) -> u64 {
        match self {
            Self::Inline(bytes) => bytes.len() as u64,
            Self::Source(slice) => slice.range.length,
            Self::CanonicalJson(json) => json.normalized().len() as u64,
        }
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WasmInputBytes {
    Inline(Vec<u8>),
    /// A range in the `after` source of the enclosing file update.
    AfterRange(WasmSourceRange),
}

impl WasmInputBytes {
    fn len(&self) -> u64 {
        match self {
            Self::Inline(bytes) => bytes.len() as u64,
            Self::AfterRange(range) => range.length,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WasmInputSplice {
    pub offset: u64,
    pub delete_len: u64,
    pub insert: WasmInputBytes,
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct WasmRowKey {
    pub schema_key: SharedStr,
    pub row_pk: SmallVec<[SharedStr; 2]>,
}

impl WasmRowKey {
    pub fn from_owned_parts(schema_key: impl Into<SharedStr>, row_pk: Vec<String>) -> Self {
        Self {
            schema_key: schema_key.into(),
            row_pk: row_pk.into_iter().map(SharedStr::from).collect(),
        }
    }

    pub fn from_shared_parts(
        schema_key: SharedStr,
        row_pk: impl IntoIterator<Item = SharedStr>,
    ) -> Self {
        Self {
            schema_key,
            row_pk: row_pk.into_iter().collect(),
        }
    }
}

#[derive(Debug, Clone)]
pub struct WasmRow<B> {
    pub key: WasmRowKey,
    /// Complete Snapshot JSON for this schema. Production component is currently
    /// restricted to Lix's durable JSON subset; the packet-v1
    /// arbitrary-precision extension remains gated on a durable codec.
    pub snapshot_content: B,
}

pub type WasmHostRow = WasmRow<WasmHostBytes>;
pub type WasmGuestRow = WasmRow<WasmGuestBytes>;

#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub enum WasmChangeEffect {
    #[default]
    Content,
    FormatOnly,
}

#[derive(Debug, Clone)]
pub enum WasmRowChange<B> {
    Create {
        schema_key: String,
        local_ref: u64,
        /// v3 may resolve the host-namespaced primary key before canonical
        /// validation. component leaves this absent and retains keyless semantics
        /// until create-context materialization.
        resolved_key: Option<WasmRowKey>,
        snapshot_content: B,
    },
    Upsert {
        row: WasmRow<B>,
        effect: WasmChangeEffect,
    },
    Delete(WasmRowKey),
}

impl<B> WasmRowChange<B> {
    pub fn row_key(&self) -> Option<&WasmRowKey> {
        match self {
            Self::Create { resolved_key, .. } => resolved_key.as_ref(),
            Self::Upsert { row, .. } => Some(&row.key),
            Self::Delete(key) => Some(key),
        }
    }

    pub fn schema_key(&self) -> &str {
        match self {
            Self::Create { schema_key, .. } => schema_key,
            Self::Upsert { row, .. } => &row.key.schema_key,
            Self::Delete(key) => &key.schema_key,
        }
    }

    pub fn local_ref(&self) -> Option<u64> {
        match self {
            Self::Create { local_ref, .. } => Some(*local_ref),
            Self::Upsert { .. } | Self::Delete(_) => None,
        }
    }
}

#[derive(Debug, Clone)]
pub struct WasmRowChanges<B> {
    pub changes: Vec<WasmRowChange<B>>,
}

impl<B> Default for WasmRowChanges<B> {
    fn default() -> Self {
        Self {
            changes: Vec::new(),
        }
    }
}

impl<B> WasmRowChanges<B> {
    pub fn validate(&self) -> Result<(), LixError> {
        if change_keys_have_duplicates(&self.changes) {
            return Err(invalid_param(
                "a component row key may occur only once in one transition",
            ));
        }
        if create_refs_have_duplicates(&self.changes) {
            return Err(invalid_param(
                "a component create local reference may occur only once per schema in one transition",
            ));
        }
        Ok(())
    }

    pub fn row_change_count(&self) -> usize {
        self.changes.len()
    }
}

/// Builds the sole transition-wide duplicate-check index as borrowed key
/// references. Sorting never moves or clones a key owner, and the exact
/// capacity prevents geometric growth for a large cursor.
fn sorted_change_key_refs<B>(changes: &[WasmRowChange<B>]) -> Vec<&WasmRowKey> {
    let mut keys = Vec::with_capacity(changes.len());
    keys.extend(changes.iter().filter_map(WasmRowChange::row_key));
    keys.sort_unstable();
    keys
}

fn sorted_create_refs<B>(changes: &[WasmRowChange<B>]) -> Vec<(&str, u64)> {
    let mut creates = Vec::with_capacity(changes.len());
    creates.extend(changes.iter().filter_map(|change| match change {
        WasmRowChange::Create {
            schema_key,
            local_ref,
            ..
        } => Some((schema_key.as_str(), *local_ref)),
        WasmRowChange::Upsert { .. } | WasmRowChange::Delete(_) => None,
    }));
    creates.sort_unstable();
    creates
}

fn change_keys_have_duplicates<B>(changes: &[WasmRowChange<B>]) -> bool {
    sorted_change_key_refs(changes)
        .windows(2)
        .any(|pair| pair[0] == pair[1])
}

fn create_refs_have_duplicates<B>(changes: &[WasmRowChange<B>]) -> bool {
    sorted_create_refs(changes)
        .windows(2)
        .any(|pair| pair[0] == pair[1])
}

/// Validates cursor-wide uniqueness after all pages have been moved into their
/// stable host vector.
///
/// Keeping this separate from [`WasmChangeDrainValidator`] avoids cloning every
/// untrusted key into a transition-lived owning tree while preserving
/// arbitrary guest output order and the established rejection text.
pub(crate) fn validate_change_cursor_key_uniqueness<B>(
    changes: &[WasmRowChange<B>],
) -> Result<(), LixError> {
    if change_keys_have_duplicates(changes) {
        return Err(invalid_param(
            "a component row key may occur only once across a change cursor",
        ));
    }
    if create_refs_have_duplicates(changes) {
        return Err(invalid_param(
            "a component create local reference may occur only once per schema across a change cursor",
        ));
    }
    Ok(())
}

pub type WasmHostRowChanges = WasmRowChanges<WasmHostBytes>;
pub type WasmGuestRowChanges = WasmRowChanges<WasmGuestBytes>;

#[derive(Debug, Clone)]
pub struct WasmRowPage {
    pub rows: Vec<WasmHostRow>,
}

/// Bounded, complete host rows. `None` is permanent EOF and every
/// successful page must be non-empty and no larger than `max_bytes` once
/// packet-v1 encoded.
pub trait WasmRowSource: Send {
    fn next_page(&mut self, max_bytes: u32) -> Result<Option<WasmRowPage>, LixError>;
}

/// Lazily materialized predecessor identities used only when a warm
/// `parse_changes` transition elects to replace every row.
pub trait WasmRowKeySource: Send {
    fn into_keys(self: Box<Self>) -> Result<BTreeSet<WasmRowKey>, LixError>;
}

/// Bounded, merge-resolved host changes supplied to `rows_changed`.
pub trait WasmRowChangeSource: Send {
    fn next_page(&mut self, max_bytes: u32) -> Result<Option<WasmHostRowChanges>, LixError>;
}

/// One same-column concurrent overlap presented to a column merger. Complete
/// rows remain available as read-only context, but the plugin can replace only
/// the named column value.
#[derive(Debug, Clone)]
pub struct WasmColumnMerge<B> {
    pub ordinal: u32,
    pub key: WasmRowKey,
    pub file_id: Option<String>,
    pub column: String,
    pub base: Option<B>,
    pub a: Option<B>,
    pub b: Option<B>,
    pub base_row: B,
    pub a_row: B,
    pub b_row: B,
}

pub type WasmHostColumnMerge = WasmColumnMerge<WasmHostBytes>;

#[derive(Debug, Clone)]
pub struct WasmColumnMergePage {
    pub merges: Vec<WasmHostColumnMerge>,
}

pub trait WasmColumnMergeSource: Send {
    fn next_page(&mut self, max_bytes: u32) -> Result<Option<WasmColumnMergePage>, LixError>;
}

#[derive(Debug, Clone)]
pub enum WasmColumnMergeResult<B> {
    UseLww,
    /// `None` removes an optional column. `Some` contains its canonical JSON
    /// value, including JSON `null` when the column is explicitly null.
    Replace(Option<B>),
}

pub type WasmGuestColumnMergeResult = WasmColumnMergeResult<WasmGuestBytes>;

#[derive(Debug, Clone)]
pub struct WasmColumnMergeResultPage {
    pub format_version: u16,
    pub ordinals: Vec<u32>,
    pub results: Vec<WasmGuestColumnMergeResult>,
    pub outputs: Option<WasmByteOutputsHandle>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WasmCreateContext {
    pub high: u64,
    pub low: u32,
}

impl WasmCreateContext {
    pub fn row_pk(self, local_ref: u64) -> Result<Vec<String>, LixError> {
        Ok(vec![self.component(local_ref)?])
    }

    pub fn component(self, local_ref: u64) -> Result<String, LixError> {
        Ok(uuid::Uuid::from_bytes(self.component_uuid_bytes(local_ref)?).to_string())
    }

    pub(crate) fn component_uuid_bytes(self, local_ref: u64) -> Result<[u8; 16], LixError> {
        let local_ref = u32::try_from(local_ref).map_err(|_| {
            invalid_param(
                "component create local references must fit in an unsigned 32-bit integer",
            )
        })?;
        let mut bytes = [0_u8; 16];
        bytes[..8].copy_from_slice(&self.high.to_be_bytes());
        bytes[8..12].copy_from_slice(&self.low.to_be_bytes());
        bytes[12..].copy_from_slice(&local_ref.to_be_bytes());
        Ok(bytes)
    }
}

pub struct WasmOpenFileInput {
    pub descriptor: WasmFileDescriptor,
    pub file: Arc<dyn WasmByteSource>,
    pub creates: WasmCreateContext,
    /// Whether this file may represent a complete parse as a certified row
    /// packet instead of ordinary per-row changes.
    ///
    /// A certified packet is materialized by expanding it from the file's
    /// published commit root, so the caller must set this to `false` for any
    /// file that publishes no commit. The same rows are then produced through
    /// the ordinary change path instead. This is a representation choice made
    /// by the host; the guest emits the same rows either way.
    pub certified_packets_available: bool,
}

impl fmt::Debug for WasmOpenFileInput {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("WasmOpenFileInput")
            .field("descriptor", &self.descriptor)
            .field("file_len", &self.file.len())
            .field("creates", &self.creates)
            .field(
                "certified_packets_available",
                &self.certified_packets_available,
            )
            .finish()
    }
}

pub struct WasmOpenRowsInput {
    pub descriptor: WasmFileDescriptor,
    pub rows: Box<dyn WasmRowSource>,
    pub accepted: Option<Arc<dyn WasmByteSource>>,
}

impl fmt::Debug for WasmOpenRowsInput {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("WasmOpenRowsInput")
            .field("descriptor", &self.descriptor)
            .field(
                "accepted_len",
                &self.accepted.as_ref().map(|source| source.len()),
            )
            .finish_non_exhaustive()
    }
}

pub struct WasmFileUpdate {
    pub before_descriptor: WasmFileDescriptor,
    pub after_descriptor: WasmFileDescriptor,
    pub before: Arc<dyn WasmByteSource>,
    pub edits: Vec<WasmInputSplice>,
    pub after: Arc<dyn WasmByteSource>,
    pub creates: WasmCreateContext,
    pub rows: Option<Box<dyn WasmRowSource>>,
    pub prior_row_keys: Option<Box<dyn WasmRowKeySource>>,
}

pub struct WasmColdFileUpdate {
    pub before_descriptor: WasmFileDescriptor,
    pub after_descriptor: WasmFileDescriptor,
    pub before: Option<Arc<dyn WasmByteSource>>,
    pub edits: Vec<WasmInputSplice>,
    pub after: Arc<dyn WasmByteSource>,
    pub creates: WasmCreateContext,
    pub rows: Box<dyn WasmRowSource>,
}

impl fmt::Debug for WasmColdFileUpdate {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("WasmColdFileUpdate")
            .field("before_descriptor", &self.before_descriptor)
            .field("after_descriptor", &self.after_descriptor)
            .field(
                "before_len",
                &self.before.as_ref().map(|source| source.len()),
            )
            .field("edits", &self.edits)
            .field("after_len", &self.after.len())
            .field("creates", &self.creates)
            .field("rows", &"lazy complete row source")
            .finish()
    }
}

impl WasmColdFileUpdate {
    pub fn validate(&self, limits: WasmTransitionLimits) -> Result<(), LixError> {
        match &self.before {
            Some(before) => WasmFileUpdate {
                before_descriptor: self.before_descriptor.clone(),
                after_descriptor: self.after_descriptor.clone(),
                before: Arc::clone(before),
                edits: self.edits.clone(),
                after: Arc::clone(&self.after),
                creates: self.creates,
                rows: None,
                prior_row_keys: None,
            }
            .validate(limits),
            None => {
                self.before_descriptor
                    .validate_warm_successor(&self.after_descriptor)?;
                limits.validate()?;
                if !self.edits.is_empty() {
                    return Err(invalid_param(
                        "a derived cold successor must not carry host byte splices",
                    ));
                }
                if self.after.len() > limits.max_total_bytes {
                    return Err(invalid_param(
                        "a derived cold successor source exceeds max_total_bytes",
                    ));
                }
                Ok(())
            }
        }
    }
}

impl fmt::Debug for WasmFileUpdate {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("WasmFileUpdate")
            .field("before_descriptor", &self.before_descriptor)
            .field("after_descriptor", &self.after_descriptor)
            .field("before_len", &self.before.len())
            .field("edits", &self.edits)
            .field("after_len", &self.after.len())
            .field("creates", &self.creates)
            .finish()
    }
}

impl WasmFileUpdate {
    /// Validates attacker-controlled splice metadata before Canonical-ABI
    /// lowering allocates the guest-side edit list.
    pub fn validate(&self, limits: WasmTransitionLimits) -> Result<(), LixError> {
        self.before_descriptor
            .validate_warm_successor(&self.after_descriptor)?;
        limits.validate()?;
        if self.edits.len() > limits.max_inline_edits as usize {
            return Err(invalid_param(
                "component input splice count exceeds its limit",
            ));
        }

        let before_len = self.before.len();
        let after_len = self.after.len();
        let mut previous_start = None;
        let mut previous_end = 0u64;
        let mut deleted = 0u64;
        let mut inserted = 0u64;
        let mut inline = 0u64;
        for edit in &self.edits {
            let end = edit
                .offset
                .checked_add(edit.delete_len)
                .ok_or_else(|| invalid_param("component input splice deletion range overflowed"))?;
            if previous_start == Some(edit.offset) || edit.offset < previous_end || end > before_len
            {
                return Err(invalid_param(
                    "component input splices must have strictly increasing starts, be non-overlapping, and stay in the accepted base",
                ));
            }
            if let WasmInputBytes::AfterRange(range) = &edit.insert
                && range.end()? > after_len
            {
                return Err(invalid_param(
                    "component after-source range is out of bounds",
                ));
            }
            if let WasmInputBytes::Inline(bytes) = &edit.insert {
                inline = inline
                    .checked_add(bytes.len() as u64)
                    .ok_or_else(|| invalid_param("component inline input byte count overflowed"))?;
            }
            deleted = deleted
                .checked_add(edit.delete_len)
                .ok_or_else(|| invalid_param("component deleted byte count overflowed"))?;
            inserted = inserted
                .checked_add(edit.insert.len())
                .ok_or_else(|| invalid_param("component inserted byte count overflowed"))?;
            previous_start = Some(edit.offset);
            previous_end = end;
        }
        if inline > limits.max_inline_input_bytes {
            return Err(invalid_param(
                "component inline input bytes exceed their limit",
            ));
        }
        let reconstructed_len = before_len
            .checked_sub(deleted)
            .and_then(|len| len.checked_add(inserted))
            .ok_or_else(|| invalid_param("component reconstructed file length overflowed"))?;
        if reconstructed_len != after_len {
            return Err(invalid_param(
                "component input splices do not reconstruct the declared after source length",
            ));
        }
        Ok(())
    }
}

pub struct WasmRowUpdate {
    pub before_descriptor: WasmFileDescriptor,
    pub after_descriptor: WasmFileDescriptor,
    pub before: Arc<dyn WasmByteSource>,
    pub changes: Box<dyn WasmRowChangeSource>,
}

/// Input for one stateless, file-scoped conflict-resolution call. It has no
/// document handle on purpose: a one-row/one-paragraph merge must not force a
/// cold open of all semantic rows in the file.
pub struct WasmColumnMergeUpdate {
    pub merges: Box<dyn WasmColumnMergeSource>,
}

impl fmt::Debug for WasmColumnMergeUpdate {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("WasmColumnMergeUpdate")
            .finish_non_exhaustive()
    }
}

impl fmt::Debug for WasmRowUpdate {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("WasmRowUpdate")
            .field("before_descriptor", &self.before_descriptor)
            .field("after_descriptor", &self.after_descriptor)
            .field("before_len", &self.before.len())
            .finish_non_exhaustive()
    }
}

macro_rules! handle_type {
    ($name:ident) => {
        #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
        pub struct $name(pub u64);
    };
}

handle_type!(WasmDocumentHandle);

/// Runtime-private immutable document state retained outside a Wasm Store.
///
/// The engine treats the payload as opaque. A compatible component runtime
/// may restore it into a fresh Store after actor eviction, avoiding durable
/// row hydration while preserving the same semantic authority checks.
#[derive(Debug, Clone)]
pub struct WasmDurableDocumentCheckpoint {
    bytes: crate::Blob,
}

impl WasmDurableDocumentCheckpoint {
    const MAGIC: &'static [u8; 8] = b"LIXDPR01";
    pub const MAX_DECODED_BYTES: usize = 128 * 1024 * 1024;

    pub fn new(bytes: crate::Blob) -> Result<Self, LixError> {
        if bytes.len() > Self::MAX_DECODED_BYTES {
            return Err(LixError::new(
                LixError::CODE_PLUGIN_RESOURCE_LIMIT,
                "plugin runtime checkpoint exceeds the 128 MiB encode limit",
            ));
        }
        let decoded_len = u64::try_from(bytes.len()).map_err(|_| {
            LixError::new(
                LixError::CODE_PLUGIN_RESOURCE_LIMIT,
                "plugin runtime checkpoint exceeds u64",
            )
        })?;
        let compressed = crate::compression::compress_zstd_level_1(&bytes).map_err(|error| {
            LixError::new(
                LixError::CODE_INTERNAL_ERROR,
                format!("failed to compress plugin runtime checkpoint: {error}"),
            )
        })?;
        let mut encoded = Vec::with_capacity(16 + compressed.len());
        encoded.extend_from_slice(Self::MAGIC);
        encoded.extend_from_slice(&decoded_len.to_le_bytes());
        encoded.extend_from_slice(&compressed);
        Ok(Self {
            bytes: encoded.into(),
        })
    }

    pub fn bytes(&self) -> crate::Blob {
        self.bytes.clone()
    }

    pub(crate) fn decode(bytes: &[u8]) -> Result<crate::Blob, LixError> {
        let header = bytes.get(..16).ok_or_else(|| {
            LixError::new(
                LixError::CODE_INVALID_PLUGIN,
                "plugin runtime checkpoint is truncated",
            )
        })?;
        if &header[..8] != Self::MAGIC {
            return Err(LixError::new(
                LixError::CODE_INVALID_PLUGIN,
                "plugin runtime checkpoint has an unsupported format",
            ));
        }
        let decoded_len = usize::try_from(u64::from_le_bytes(
            header[8..16].try_into().expect("fixed checkpoint length"),
        ))
        .map_err(|_| {
            LixError::new(
                LixError::CODE_PLUGIN_RESOURCE_LIMIT,
                "plugin runtime checkpoint exceeds host address space",
            )
        })?;
        if decoded_len > Self::MAX_DECODED_BYTES {
            return Err(LixError::new(
                LixError::CODE_PLUGIN_RESOURCE_LIMIT,
                "plugin runtime checkpoint exceeds the 128 MiB decode limit",
            ));
        }
        crate::compression::decompress_zstd(&bytes[16..], decoded_len)
            .map(crate::Blob::from)
            .map_err(|error| {
                LixError::new(
                    LixError::CODE_INVALID_PLUGIN,
                    format!("plugin runtime checkpoint failed decompression: {error}"),
                )
            })
    }
}

#[derive(Clone)]
pub struct WasmDocumentCheckpoint {
    payload: Arc<dyn Any + Send + Sync>,
    retained_bytes: u64,
    durable: Option<WasmDurableDocumentCheckpoint>,
}

impl fmt::Debug for WasmDocumentCheckpoint {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("WasmDocumentCheckpoint")
            .field("retained_bytes", &self.retained_bytes)
            .finish_non_exhaustive()
    }
}

impl WasmDocumentCheckpoint {
    pub fn new<T>(payload: T, retained_bytes: u64) -> Self
    where
        T: Any + Send + Sync,
    {
        Self {
            payload: Arc::new(payload),
            retained_bytes,
            durable: None,
        }
    }

    pub fn new_with_durable<T>(
        payload: T,
        retained_bytes: u64,
        durable: WasmDurableDocumentCheckpoint,
    ) -> Self
    where
        T: Any + Send + Sync,
    {
        Self {
            payload: Arc::new(payload),
            retained_bytes,
            durable: Some(durable),
        }
    }

    pub fn downcast_ref<T: Any>(&self) -> Option<&T> {
        self.payload.downcast_ref()
    }

    pub fn retained_bytes(&self) -> u64 {
        self.retained_bytes
    }

    pub fn durable_checkpoint(&self) -> Option<WasmDurableDocumentCheckpoint> {
        self.durable.clone()
    }
}
handle_type!(WasmChangeCursorHandle);
handle_type!(WasmColumnMergeCursorHandle);
handle_type!(WasmEditCursorHandle);
handle_type!(WasmByteOutputsHandle);
handle_type!(WasmTransitionHandle);

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WasmOutputRange {
    pub index: u32,
    pub offset: u64,
    pub length: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WasmGuestBytes {
    /// A shared slice of the guest packet payload. Every inline value decoded
    /// from one packet page retains the same backing allocation.
    Inline(Bytes),
    Output(WasmOutputRange),
}

#[derive(Debug, Clone)]
pub struct WasmChangePage {
    pub format_version: u16,
    pub changes: WasmGuestRowChanges,
    /// Exactly one page-local table supplies all `Output` values in `changes`.
    pub outputs: Option<WasmByteOutputsHandle>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WasmOutputSplice {
    pub offset: u64,
    pub delete_len: u64,
    pub insert: WasmGuestBytes,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WasmEditPage {
    pub edits: Vec<WasmOutputSplice>,
    /// Exactly one page-local table supplies all output ranges in `edits`.
    pub outputs: Option<WasmByteOutputsHandle>,
}

/// Cross-page validator for a guest change cursor. Raw packet framing and
/// inline byte bounds are checked by the binding before it constructs a typed
/// page; this validator owns page, reference, and permanent-EOF invariants.
/// Cursor-wide uniqueness is checked once over borrowed keys after the stable
/// output vector has been assembled.
#[derive(Debug, Clone, Copy)]
pub struct WasmChangeDrainValidator {
    limits: WasmTransitionLimits,
    pages: u32,
    attachment_refs: u32,
    reached_eof: bool,
}

impl WasmChangeDrainValidator {
    pub fn new(limits: WasmTransitionLimits) -> Result<Self, LixError> {
        Ok(Self {
            limits: limits.validate()?,
            pages: 0,
            attachment_refs: 0,
            reached_eof: false,
        })
    }

    pub fn accept_page(&mut self, page: &WasmChangePage) -> Result<(), LixError> {
        if self.reached_eof {
            return Err(invalid_param(
                "a component change cursor advanced after EOF",
            ));
        }
        if page.format_version != PACKET_FORMAT_V1 {
            return Err(invalid_param(
                "unsupported component change packet format version",
            ));
        }
        if page.changes.changes.is_empty() {
            return Err(invalid_param("a component change page must not be empty"));
        }
        self.pages = self
            .pages
            .checked_add(1)
            .ok_or_else(|| invalid_param("component change page count overflowed"))?;
        if self.pages > self.limits.max_pages {
            return Err(invalid_param(
                "component change page count exceeds its limit",
            ));
        }

        let mut page_refs = 0u32;
        for change in &page.changes.changes {
            let output = match change {
                WasmRowChange::Create {
                    snapshot_content, ..
                } => match snapshot_content {
                    WasmGuestBytes::Output(range) => Some(range),
                    WasmGuestBytes::Inline(_) => None,
                },
                WasmRowChange::Upsert { row, .. } => match &row.snapshot_content {
                    WasmGuestBytes::Output(range) => Some(range),
                    WasmGuestBytes::Inline(_) => None,
                },
                WasmRowChange::Delete(_) => None,
            };
            if let Some(range) = output {
                range
                    .offset
                    .checked_add(range.length)
                    .ok_or_else(|| invalid_param("component change output range overflowed"))?;
                page_refs = page_refs.checked_add(1).ok_or_else(|| {
                    invalid_param("component attachment reference count overflowed")
                })?;
            }
        }
        validate_attachment_table_presence(page_refs, page.outputs.is_some())?;
        self.attachment_refs = self
            .attachment_refs
            .checked_add(page_refs)
            .ok_or_else(|| invalid_param("component attachment reference count overflowed"))?;
        if self.attachment_refs > self.limits.max_attachment_refs {
            return Err(invalid_param(
                "component attachment reference count exceeds its limit",
            ));
        }
        Ok(())
    }

    pub fn accept_eof(&mut self) {
        self.reached_eof = true;
    }
}

/// Cross-page validator for renderer splices whose offsets all address the
/// same accepted base document.
#[derive(Debug, Clone, Copy)]
pub struct WasmEditDrainValidator {
    limits: WasmTransitionLimits,
    base_len: u64,
    pages: u32,
    attachment_refs: u32,
    previous_start: Option<u64>,
    previous_end: u64,
    reached_eof: bool,
}

impl WasmEditDrainValidator {
    pub fn new(base_len: u64, limits: WasmTransitionLimits) -> Result<Self, LixError> {
        Ok(Self {
            limits: limits.validate()?,
            base_len,
            pages: 0,
            attachment_refs: 0,
            previous_start: None,
            previous_end: 0,
            reached_eof: false,
        })
    }

    pub fn accept_page(&mut self, page: &WasmEditPage) -> Result<(), LixError> {
        if self.reached_eof {
            return Err(invalid_param("a component edit cursor advanced after EOF"));
        }
        if page.edits.is_empty() {
            return Err(invalid_param("a component edit page must not be empty"));
        }
        if page.edits.len() > self.limits.max_inline_edits as usize {
            return Err(invalid_param("component edit page count exceeds its limit"));
        }
        self.pages = self
            .pages
            .checked_add(1)
            .ok_or_else(|| invalid_param("component edit page count overflowed"))?;
        if self.pages > self.limits.max_pages {
            return Err(invalid_param("component edit page count exceeds its limit"));
        }

        let mut page_record_bytes = 0u64;
        let mut page_refs = 0u32;
        for edit in &page.edits {
            let mut record_bytes = EDIT_SPLICE_METADATA_BYTES;
            let end = edit.offset.checked_add(edit.delete_len).ok_or_else(|| {
                invalid_param("component output splice deletion range overflowed")
            })?;
            if self.previous_start == Some(edit.offset)
                || edit.offset < self.previous_end
                || end > self.base_len
            {
                return Err(invalid_param(
                    "component output splices must have globally increasing starts, be non-overlapping, and stay in the accepted base",
                ));
            }
            match &edit.insert {
                WasmGuestBytes::Inline(bytes) => {
                    record_bytes =
                        record_bytes
                            .checked_add(bytes.len() as u64)
                            .ok_or_else(|| {
                                invalid_param("component output edit record bytes overflowed")
                            })?;
                }
                WasmGuestBytes::Output(range) => {
                    range
                        .offset
                        .checked_add(range.length)
                        .ok_or_else(|| invalid_param("component edit output range overflowed"))?;
                    page_refs = page_refs.checked_add(1).ok_or_else(|| {
                        invalid_param("component attachment reference count overflowed")
                    })?;
                }
            }
            if record_bytes > u64::from(self.limits.max_record_bytes) {
                return Err(invalid_param(
                    "component output edit record exceeds max_record_bytes",
                ));
            }
            page_record_bytes = page_record_bytes
                .checked_add(record_bytes)
                .ok_or_else(|| invalid_param("component output edit page bytes overflowed"))?;
            self.previous_start = Some(edit.offset);
            self.previous_end = end;
        }
        if page_record_bytes > u64::from(self.limits.max_page_bytes) {
            return Err(invalid_param(
                "component output edit page exceeds max_page_bytes",
            ));
        }
        validate_attachment_table_presence(page_refs, page.outputs.is_some())?;
        self.attachment_refs = self
            .attachment_refs
            .checked_add(page_refs)
            .ok_or_else(|| invalid_param("component attachment reference count overflowed"))?;
        if self.attachment_refs > self.limits.max_attachment_refs {
            return Err(invalid_param(
                "component attachment reference count exceeds its limit",
            ));
        }
        Ok(())
    }

    pub fn accept_eof(&mut self) {
        self.reached_eof = true;
    }
}

fn validate_attachment_table_presence(
    reference_count: u32,
    has_table: bool,
) -> Result<(), LixError> {
    if (reference_count == 0) == has_table {
        return Err(invalid_param(
            "a component page must own an output table exactly when it contains output references",
        ));
    }
    Ok(())
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WasmFileTransition {
    pub transition: WasmTransitionHandle,
    pub document: WasmDocumentHandle,
    pub changes: WasmChangeCursorHandle,
    pub replace_all_rows: bool,
}

/// One immutable, host-validated semantic batch which remains encoded until
/// storage/query consumption.
///
/// This is deliberately format-neutral at the runtime boundary. The format
/// id selects a registered engine codec; pages retain the guest's bounded
/// framing and share their backing buffers through transaction commit.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WasmCertifiedCreateRange {
    pub schema_key: String,
    pub first_local_ref: u32,
    pub last_local_ref: u32,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WasmCertifiedRowBatch {
    pub format: u16,
    pub schema_keys: Vec<String>,
    pub row_count: u64,
    pub creates: WasmCreateContext,
    /// Host-validated compact identities created by this batch. Keeping ranges
    /// here avoids reparsing or expanding certified packet payloads merely to
    /// establish successor authority.
    pub create_ranges: Vec<WasmCertifiedCreateRange>,
    /// True when this batch replaces every prior segment for the file.
    /// Incremental transitions emit sparse overlays instead.
    pub complete_file_state: bool,
    pub pages: Vec<Bytes>,
}

/// Dense snapshot packets synthesized only after ordinary component guest output has
/// completed host validation. Guest-emitted certified batches intentionally
/// cannot claim this format.
pub(crate) const HOST_CERTIFIED_PACKET_FORMAT: u16 = 3;
pub(crate) const HOST_CERTIFIED_ZSTD_PACKET_FORMAT: u16 = 4;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WasmRowTransition {
    pub transition: WasmTransitionHandle,
    pub document: WasmDocumentHandle,
    pub edits: WasmEditCursorHandle,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WasmColumnMergeTransition {
    pub transition: WasmTransitionHandle,
    pub results: WasmColumnMergeCursorHandle,
}

/// A compiled Component. Implementations share this factory but create one
/// isolated Store/instance for every branch/file actor.
#[async_trait]
pub trait WasmComponentFactory: Send + Sync {
    async fn instantiate_actor(&self) -> Result<Box<dyn WasmComponentActor>, LixError>;
}

/// One serial branch/file actor. Handles are instance-local and invalid after
/// `retire`; callers must retire the complete actor after traps, timeouts,
/// cancellation, or uncertain completion.
#[async_trait]
pub trait WasmComponentActor: Send {
    /// True when `open_rows` retains accepted bytes only and returns them
    /// unchanged without invoking a semantic renderer.
    fn cold_open_hydrates_without_render(&self) -> bool {
        false
    }

    /// False when a cold actor reconstructs its transition state from the
    /// accepted file bytes and does not consume durable semantic rows.
    fn cold_open_requires_rows(&self) -> bool {
        true
    }

    async fn fork_document(
        &mut self,
        document: WasmDocumentHandle,
    ) -> Result<WasmDocumentHandle, LixError>;

    /// Captures runtime-private immutable state without serializing it through
    /// the Component boundary. Implementations that cannot externalize a
    /// document return `None` and retain the normal hydration path.
    async fn checkpoint_document(
        &mut self,
        _document: WasmDocumentHandle,
    ) -> Result<Option<WasmDocumentCheckpoint>, LixError> {
        Ok(None)
    }

    /// Restores a compatible runtime-private checkpoint into this fresh actor.
    async fn restore_document(
        &mut self,
        _checkpoint: &WasmDocumentCheckpoint,
    ) -> Result<WasmDocumentHandle, LixError> {
        Err(invalid_param(
            "this component actor does not support decoded document checkpoints",
        ))
    }

    /// Restores a process-independent runtime checkpoint against the exact
    /// accepted file bytes named by the visible materialization.
    async fn restore_durable_document(
        &mut self,
        _checkpoint: &[u8],
        _accepted: &[u8],
    ) -> Result<WasmDocumentHandle, LixError> {
        Err(LixError::new(
            LixError::CODE_INVALID_PLUGIN,
            "this component actor does not support durable document checkpoints",
        ))
    }

    async fn open_file(
        &mut self,
        limits: WasmTransitionLimits,
        input: WasmOpenFileInput,
    ) -> Result<WasmFileTransition, LixError>;

    async fn open_rows(
        &mut self,
        limits: WasmTransitionLimits,
        input: WasmOpenRowsInput,
    ) -> Result<WasmRowTransition, LixError>;

    async fn file_changed(
        &mut self,
        document: WasmDocumentHandle,
        limits: WasmTransitionLimits,
        update: WasmFileUpdate,
    ) -> Result<WasmFileTransition, LixError>;

    async fn cold_file_changed(
        &mut self,
        limits: WasmTransitionLimits,
        update: WasmColdFileUpdate,
    ) -> Result<WasmFileTransition, LixError> {
        let _ = (limits, update);
        Err(invalid_param(
            "this component actor does not implement cold successor reconciliation",
        ))
    }

    async fn rows_changed(
        &mut self,
        document: WasmDocumentHandle,
        limits: WasmTransitionLimits,
        update: WasmRowUpdate,
    ) -> Result<WasmRowTransition, LixError>;

    async fn merge_columns(
        &mut self,
        limits: WasmTransitionLimits,
        update: WasmColumnMergeUpdate,
    ) -> Result<WasmColumnMergeTransition, LixError> {
        let _ = (limits, update);
        Err(invalid_param(
            "this component actor does not implement column merging",
        ))
    }

    async fn next_change_page(
        &mut self,
        transition: WasmTransitionHandle,
        cursor: WasmChangeCursorHandle,
        max_bytes: u32,
    ) -> Result<Option<WasmChangePage>, LixError>;

    /// Takes encoded semantic batches accumulated while draining `transition`.
    ///
    /// Ordinary component actors never produce these. A v3 adapter may return a
    /// batch only after every bounded page has passed its format validator.
    fn take_certified_row_batches(
        &mut self,
        _transition: WasmTransitionHandle,
    ) -> Vec<WasmCertifiedRowBatch> {
        Vec::new()
    }

    async fn next_column_merge_result_page(
        &mut self,
        transition: WasmTransitionHandle,
        cursor: WasmColumnMergeCursorHandle,
        max_bytes: u32,
    ) -> Result<Option<WasmColumnMergeResultPage>, LixError> {
        let _ = (transition, cursor, max_bytes);
        Err(invalid_param(
            "this component actor does not expose column merge results",
        ))
    }

    async fn next_edit_page(
        &mut self,
        transition: WasmTransitionHandle,
        cursor: WasmEditCursorHandle,
        max_edits: u32,
        max_inline_bytes: u32,
    ) -> Result<Option<WasmEditPage>, LixError>;

    async fn output_len(
        &mut self,
        transition: WasmTransitionHandle,
        outputs: WasmByteOutputsHandle,
        index: u32,
    ) -> Result<u64, LixError>;

    async fn read_output(
        &mut self,
        transition: WasmTransitionHandle,
        outputs: WasmByteOutputsHandle,
        index: u32,
        offset: u64,
        length: u32,
    ) -> Result<Vec<u8>, LixError>;

    /// Ends the aggregate budget after all output has been validated/drained.
    async fn finish_transition(
        &mut self,
        transition: WasmTransitionHandle,
    ) -> Result<WasmTransitionCounters, LixError>;

    /// Deterministically rejects one prospective transition, dropping its
    /// input fork, successor document, cursors, outputs, and budget without
    /// revoking separately accepted documents. Implementations should make
    /// this idempotent so a guest-returned error that already cleaned itself
    /// can pass through host drain cleanup safely.
    async fn discard_transition(
        &mut self,
        transition: WasmTransitionHandle,
    ) -> Result<(), LixError>;

    /// True once a trap, deadline, cancellation cleanup, or other uncertain
    /// failure has made every instance-local handle unusable.
    fn is_retired(&self) -> bool;

    async fn drop_document(&mut self, _document: WasmDocumentHandle) -> Result<(), LixError> {
        Ok(())
    }

    /// Revokes every instance-local handle. The default supports lightweight
    /// test runtimes; production runtimes should eagerly destroy their Store.
    async fn retire(&mut self) -> Result<(), LixError> {
        Ok(())
    }
}

pub fn validate_component_limits(
    component: WasmLimits,
    transition: WasmTransitionLimits,
) -> Result<(), LixError> {
    if component.max_memory_bytes == 0 {
        return Err(invalid_param(
            "component component memory limit must be positive",
        ));
    }
    transition.validate()?;
    Ok(())
}

fn invalid_param(message: impl Into<String>) -> LixError {
    LixError::new(LixError::CODE_INVALID_PARAM, message)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[derive(Debug)]
    struct MemorySource(Vec<u8>);

    impl WasmByteSource for MemorySource {
        fn len(&self) -> u64 {
            self.0.len() as u64
        }

        fn read(&self, offset: u64, length: u32) -> Result<Vec<u8>, LixError> {
            let start = usize::try_from(offset).map_err(|_| invalid_param("offset"))?;
            let end = start
                .checked_add(length as usize)
                .ok_or_else(|| invalid_param("range"))?;
            self.0
                .get(start..end)
                .map(<[u8]>::to_vec)
                .ok_or_else(|| invalid_param("range"))
        }
    }

    fn descriptor(generation: &str) -> WasmFileDescriptor {
        WasmFileDescriptor {
            file_id: "file-1".to_owned(),
            path: Some("data.csv".to_owned()),
            plugin: WasmPluginSelection {
                plugin_key: "plugin_csv".to_owned(),
                generation: generation.to_owned(),
            },
        }
    }

    #[test]
    fn warm_successor_rejects_a_different_stable_file_id() {
        let before = descriptor("generation-1");
        let mut after = before.clone();
        after.file_id = "file-2".to_owned();
        let error = before
            .validate_warm_successor(&after)
            .expect_err("a warm transition cannot cross file identities");
        assert!(error.message.contains("stable file id"));
    }

    #[test]
    fn transition_counter_aggregation_keeps_directional_page_and_source_metrics() {
        let mut total = WasmTransitionCounters::default();
        total.accumulate(WasmTransitionCounters {
            file_read_calls: 2,
            file_bytes_read: 11,
            state_read_calls: 3,
            state_key_bytes: 13,
            state_value_bytes_read: 17,
            row_input_pages: 5,
            row_input_records: 19,
            row_input_wire_bytes: 23,
            row_output_pages: 7,
            row_output_records: 29,
            row_output_wire_bytes: 31,
            row_input_attachment_reads: 37,
            row_input_attachment_bytes: 41,
            row_output_attachment_writes: 43,
            row_output_attachment_bytes: 47,
            ..WasmTransitionCounters::default()
        });

        assert_eq!(total.file_read_calls, 2);
        assert_eq!(total.file_bytes_read, 11);
        assert_eq!(total.state_read_calls, 3);
        assert_eq!(total.state_key_bytes, 13);
        assert_eq!(total.state_value_bytes_read, 17);
        assert_eq!(total.row_input_pages, 5);
        assert_eq!(total.row_input_records, 19);
        assert_eq!(total.row_input_wire_bytes, 23);
        assert_eq!(total.row_output_pages, 7);
        assert_eq!(total.row_output_records, 29);
        assert_eq!(total.row_output_wire_bytes, 31);
        assert_eq!(total.row_input_attachment_reads, 37);
        assert_eq!(total.row_input_attachment_bytes, 41);
        assert_eq!(total.row_output_attachment_writes, 43);
        assert_eq!(total.row_output_attachment_bytes, 47);
    }

    #[test]
    fn durable_document_checkpoint_roundtrips_and_bounds_decompression() {
        let original: crate::Blob = b"repeated plugin state".repeat(4_096).into();
        let checkpoint = WasmDurableDocumentCheckpoint::new(original.clone()).unwrap();
        assert_eq!(
            WasmDurableDocumentCheckpoint::decode(&checkpoint.bytes()).unwrap(),
            original
        );

        let mut oversized = Vec::from(WasmDurableDocumentCheckpoint::MAGIC.as_slice());
        oversized.extend_from_slice(
            &u64::try_from(WasmDurableDocumentCheckpoint::MAX_DECODED_BYTES + 1)
                .unwrap()
                .to_le_bytes(),
        );
        assert_eq!(
            WasmDurableDocumentCheckpoint::decode(&oversized)
                .expect_err("oversized checkpoint must fail before decompression")
                .code,
            LixError::CODE_PLUGIN_RESOURCE_LIMIT
        );
    }

    #[test]
    fn canonical_json_rows_share_one_utf8_arena_without_copying() {
        let first = r#"{"label":"é"}"#;
        let second = r#"{"label":"雪"}"#;
        let first_end = u32::try_from(first.len()).unwrap();
        let arena_len = first.len() + second.len();
        let mut normalized = Vec::with_capacity(arena_len + 64);
        normalized.extend_from_slice(first.as_bytes());
        normalized.extend_from_slice(second.as_bytes());
        let arena_pointer = normalized.as_ptr();

        let rows = WasmCanonicalJson::from_batch_parts(
            vec![
                serde_json::from_str(first).unwrap(),
                serde_json::from_str(second).unwrap(),
            ],
            normalized,
            vec![(0, first_end), (first_end, arena_len as u32)],
            2,
            2,
        )
        .unwrap();

        let first_shared = rows[0].normalized_shared();
        let second_shared = rows[1].normalized_shared();
        assert_eq!(rows[0].normalized(), first);
        assert_eq!(rows[1].normalized(), second);
        assert_eq!(first_shared.as_str(), first);
        assert_eq!(second_shared.as_str(), second);
        assert_eq!(first_shared.as_bytes().as_ptr(), arena_pointer);
        assert!(first_shared.shares_buffer_with(&second_shared));
        assert_eq!(first_shared.retained_buffer_len(), arena_len);
        assert_eq!(rows[0].batch_arena_allocation_count(), 1);
        assert_eq!(rows[0].validation_counts(), (2, 2));
    }

    #[test]
    fn canonical_json_rejects_offsets_inside_unicode_scalars() {
        let normalized = "é{}".as_bytes().to_vec();
        let error = WasmCanonicalJson::from_batch_parts(
            vec![JsonValue::Null, JsonValue::Object(Default::default())],
            normalized,
            vec![(0, 1), (1, 4)],
            2,
            2,
        )
        .unwrap_err();

        assert!(error.message.contains("UTF-8 scalar"));
    }

    #[test]
    fn canonical_json_large_batch_reserves_its_handle_vector_once() {
        let row_count = 10_000usize;
        let normalized = b"null".repeat(row_count);
        let offsets = (0..row_count)
            .map(|row| {
                let start = u32::try_from(row * 4).expect("test arena fits u32");
                (start, start + 4)
            })
            .collect();

        let rows = WasmCanonicalJson::from_batch_parts(
            vec![JsonValue::Null; row_count],
            normalized,
            offsets,
            row_count,
            row_count,
        )
        .expect("large canonical batch");

        assert_eq!(rows.len(), row_count);
        assert_eq!(
            rows.capacity(),
            row_count,
            "the handle column must be reserved at its final size"
        );
        assert_eq!(rows.last().expect("last row").row_index(), row_count - 1);
        assert!(rows[0].shares_batch_with(rows.last().expect("last row")));
    }

    #[test]
    fn large_common_keys_stay_inline_and_duplicate_sort_borrows_original_owners() {
        let schema = SharedStr::from_static("csv_row");
        let namespace = SharedStr::from_static("namespace");
        let row = SharedStr::from_static("row");
        let changes = (0..10_000)
            .map(|ordinal| {
                let row_pk = if ordinal % 2 == 0 {
                    [namespace.clone()].into_iter().collect()
                } else {
                    [namespace.clone(), row.clone()].into_iter().collect()
                };
                WasmRowChange::<WasmGuestBytes>::Delete(WasmRowKey {
                    schema_key: schema.clone(),
                    row_pk,
                })
            })
            .collect::<Vec<_>>();

        for (ordinal, change) in changes.iter().enumerate() {
            let key = change.row_key().expect("delete carries a row key");
            assert_eq!(key.row_pk.len(), 1 + (ordinal % 2));
            assert!(
                !key.row_pk.spilled(),
                "one- and two-component protocol keys must stay inline"
            );
            assert!(key.schema_key.shares_buffer_with(&schema));
            assert!(key.row_pk[0].shares_buffer_with(&namespace));
            if ordinal % 2 == 1 {
                assert!(key.row_pk[1].shares_buffer_with(&row));
            }
        }

        let mut original_owner_addresses = changes
            .iter()
            .map(|change| {
                let key: *const WasmRowKey = change.row_key().expect("delete carries a row key");
                key as usize
            })
            .collect::<Vec<_>>();
        let sorted = sorted_change_key_refs(&changes);
        let mut sorted_owner_addresses = sorted
            .iter()
            .map(|key| {
                let key: *const WasmRowKey = *key;
                key as usize
            })
            .collect::<Vec<_>>();
        original_owner_addresses.sort_unstable();
        sorted_owner_addresses.sort_unstable();

        assert_eq!(sorted.len(), changes.len());
        assert_eq!(
            sorted_owner_addresses, original_owner_addresses,
            "the duplicate index must contain references to original key owners"
        );
        assert!(change_keys_have_duplicates(&changes));
        assert_eq!(
            validate_change_cursor_key_uniqueness(&changes)
                .expect_err("the repeated structural keys are duplicates")
                .message,
            "a component row key may occur only once across a change cursor"
        );
    }

    #[test]
    fn cursor_key_uniqueness_accepts_arbitrary_unique_order() {
        let changes = vec![
            WasmRowChange::<WasmGuestBytes>::Delete(WasmRowKey::from_shared_parts(
                SharedStr::from_static("schema-z"),
                [SharedStr::from_static("row-z")],
            )),
            WasmRowChange::<WasmGuestBytes>::Delete(WasmRowKey::from_shared_parts(
                SharedStr::from_static("schema-a"),
                [SharedStr::from_static("row-a")],
            )),
        ];

        validate_change_cursor_key_uniqueness(&changes)
            .expect("cursor duplicate validation must not impose key order");
    }

    #[test]
    fn create_context_produces_canonical_uuid_components() {
        let creates = WasmCreateContext {
            high: 0x0192_0000_0000_7000,
            low: 0x8000_0000,
        };
        assert_eq!(
            creates.component(42).unwrap(),
            "01920000-0000-7000-8000-00000000002a"
        );
        assert_eq!(creates.row_pk(7).unwrap().len(), 1);
        assert!(creates.component(u64::from(u32::MAX) + 1).is_err());
    }

    #[test]
    fn splice_validation_is_pre_lowering_and_base_relative() {
        let before: Arc<dyn WasmByteSource> = Arc::new(MemorySource(b"abc".to_vec()));
        let after: Arc<dyn WasmByteSource> = Arc::new(MemorySource(b"aXYZc".to_vec()));
        let update = WasmFileUpdate {
            before_descriptor: descriptor("hash-a"),
            after_descriptor: descriptor("hash-a"),
            before,
            edits: vec![WasmInputSplice {
                offset: 1,
                delete_len: 1,
                insert: WasmInputBytes::AfterRange(WasmSourceRange {
                    offset: 1,
                    length: 3,
                }),
            }],
            after,
            creates: WasmCreateContext { high: 1, low: 2 },
            rows: None,
            prior_row_keys: None,
        };
        update.validate(WasmTransitionLimits::default()).unwrap();

        let mut wrong_generation = update;
        wrong_generation.after_descriptor = descriptor("hash-b");
        assert!(
            wrong_generation
                .validate(WasmTransitionLimits::default())
                .is_err()
        );
    }

    #[test]
    fn rejects_duplicate_row_keys() {
        let key = WasmRowKey::from_owned_parts("csv_row", vec!["row".to_owned()]);
        let duplicate = WasmRowChanges::<WasmGuestBytes> {
            changes: vec![
                WasmRowChange::Delete(key.clone()),
                WasmRowChange::Delete(key),
            ],
        };
        assert!(duplicate.validate().is_err());
        assert!(
            WasmRowChanges::<WasmGuestBytes>::default()
                .validate()
                .is_ok()
        );
    }

    #[test]
    fn transition_limits_reject_unbounded_or_inverted_values() {
        assert!(WasmTransitionLimits::default().validate().is_ok());
        assert!(
            WasmTransitionLimits {
                max_record_bytes: 2,
                max_page_bytes: 1,
                ..WasmTransitionLimits::default()
            }
            .validate()
            .is_err()
        );
        assert!(
            WasmTransitionLimits {
                total_deadline_nanoseconds: 0,
                ..WasmTransitionLimits::default()
            }
            .validate()
            .is_err()
        );
    }

    #[test]
    fn cold_file_budget_scales_with_input_and_stays_bounded() {
        assert_eq!(
            WasmTransitionLimits::for_cold_file_bytes(0).total_deadline_nanoseconds,
            WasmTransitionLimits::default().total_deadline_nanoseconds
        );
        assert_eq!(
            WasmTransitionLimits::for_cold_file_bytes(10 * MIB).total_deadline_nanoseconds,
            15_000_000_000
        );
        assert_eq!(
            WasmTransitionLimits::for_cold_file_bytes(128 * MIB).total_deadline_nanoseconds,
            COLD_FILE_MAX_DEADLINE_NANOSECONDS
        );
    }

    #[test]
    fn cold_file_page_fits_one_base64_encoded_source_map_line() {
        const OPENCLAW_SOURCE_MAP_BYTES: u64 = 5_298_078;
        let limits = WasmTransitionLimits::for_cold_file_bytes(OPENCLAW_SOURCE_MAP_BYTES);
        let encoded_line_bytes = OPENCLAW_SOURCE_MAP_BYTES.saturating_mul(4).div_ceil(3);

        assert!(u64::from(limits.max_record_bytes) > encoded_line_bytes);
        assert_eq!(limits.max_record_bytes, limits.max_page_bytes);
        assert!(u64::from(limits.max_page_bytes) <= COLD_TRANSITION_MAX_PAGE_BYTES);
        limits
            .validate()
            .expect("scaled cold limits should validate");
    }

    #[test]
    fn warm_file_page_also_fits_one_base64_encoded_source_map_line() {
        const OPENCLAW_SOURCE_MAP_BYTES: u64 = 5_298_078;
        let limits = WasmTransitionLimits::for_file_bytes(OPENCLAW_SOURCE_MAP_BYTES);
        let encoded_line_bytes = OPENCLAW_SOURCE_MAP_BYTES.saturating_mul(4).div_ceil(3);

        assert!(u64::from(limits.max_record_bytes) > encoded_line_bytes);
        assert_eq!(
            limits.total_deadline_nanoseconds,
            WasmTransitionLimits::default().total_deadline_nanoseconds
        );
        limits
            .validate()
            .expect("scaled warm limits should validate");
    }

    #[test]
    fn change_drain_validator_owns_framing_but_not_key_copies() {
        let key = WasmRowKey::from_owned_parts("csv_row", vec!["row".to_owned()]);
        let page = WasmChangePage {
            format_version: PACKET_FORMAT_V1,
            changes: WasmRowChanges {
                changes: vec![WasmRowChange::Delete(key)],
            },
            outputs: None,
        };
        let mut validator = WasmChangeDrainValidator::new(WasmTransitionLimits::default()).unwrap();
        validator.accept_page(&page).unwrap();
        validator.accept_page(&page).unwrap();
        validator.accept_eof();
        assert!(validator.accept_page(&page).is_err());
    }

    #[test]
    fn edit_drain_validation_requires_exact_page_attachment_table() {
        let range_edit = WasmOutputSplice {
            offset: 0,
            delete_len: 0,
            insert: WasmGuestBytes::Output(WasmOutputRange {
                index: 0,
                offset: 0,
                length: 10,
            }),
        };
        let missing_table = WasmEditPage {
            edits: vec![range_edit.clone()],
            outputs: None,
        };
        let mut validator =
            WasmEditDrainValidator::new(0, WasmTransitionLimits::default()).unwrap();
        assert!(validator.accept_page(&missing_table).is_err());

        let with_table = WasmEditPage {
            edits: vec![range_edit],
            outputs: Some(WasmByteOutputsHandle(1)),
        };
        let mut validator =
            WasmEditDrainValidator::new(0, WasmTransitionLimits::default()).unwrap();
        validator.accept_page(&with_table).unwrap();
    }

    #[test]
    fn edit_drain_validation_charges_metadata_to_record_and_page_limits() {
        let inline = |offset, bytes: &[u8]| WasmOutputSplice {
            offset,
            delete_len: 0,
            insert: WasmGuestBytes::Inline(bytes.to_vec().into()),
        };

        let record_limits = WasmTransitionLimits {
            max_record_bytes: u32::try_from(EDIT_SPLICE_METADATA_BYTES)
                .expect("edit splice metadata size should fit u32"),
            ..WasmTransitionLimits::default()
        };
        let mut validator = WasmEditDrainValidator::new(0, record_limits).unwrap();
        assert!(
            validator
                .accept_page(&WasmEditPage {
                    edits: vec![inline(0, b"x")],
                    outputs: None,
                })
                .expect_err("metadata plus inline bytes must fit one record")
                .message
                .contains("max_record_bytes")
        );

        let output_limits = WasmTransitionLimits {
            max_record_bytes: u32::try_from(EDIT_SPLICE_METADATA_BYTES - 1)
                .expect("edit splice metadata size should fit u32"),
            ..WasmTransitionLimits::default()
        };
        let mut validator = WasmEditDrainValidator::new(0, output_limits).unwrap();
        assert!(
            validator
                .accept_page(&WasmEditPage {
                    edits: vec![WasmOutputSplice {
                        offset: 0,
                        delete_len: 0,
                        insert: WasmGuestBytes::Output(WasmOutputRange {
                            index: 0,
                            offset: 0,
                            length: 1,
                        }),
                    }],
                    outputs: Some(WasmByteOutputsHandle(1)),
                })
                .expect_err("output-backed edits still pay fixed record metadata")
                .message
                .contains("max_record_bytes")
        );

        let page_limits = WasmTransitionLimits {
            max_record_bytes: 30,
            max_page_bytes: 49,
            ..WasmTransitionLimits::default()
        };
        let mut validator = WasmEditDrainValidator::new(1, page_limits).unwrap();
        assert!(
            validator
                .accept_page(&WasmEditPage {
                    edits: vec![inline(0, b"x"), inline(1, b"y")],
                    outputs: None,
                })
                .expect_err("the page pays metadata for every edit")
                .message
                .contains("max_page_bytes")
        );
    }

    #[test]
    fn production_wit_is_versioned_and_row_first() {
        let wit = include_str!("../../../wit/lix-plugin.wit");
        assert!(wit.starts_with("package lix:plugin@1.0.0;"));
        assert!(wit.contains("resource transition"));
        assert!(wit.contains("interface column-merger"));
        assert!(wit.contains("interface file-projection"));
        assert!(wit.contains("parse-changes: func("));
        assert!(wit.contains("serialize-changes: func("));
        assert!(wit.contains("world column-merger-plugin"));
        assert!(wit.contains("world file-projection-plugin"));
        assert!(!wit.contains("resolve-conflicts"));
        for packaged_copy in [
            include_str!("../../../../lix-plugin-bindings-column-merger/wit/lix-plugin.wit"),
            include_str!("../../../../lix-plugin-bindings-combined/wit/lix-plugin.wit"),
            include_str!("../../../../lix-plugin-bindings-file-projection/wit/lix-plugin.wit"),
        ] {
            assert_eq!(packaged_copy, wit, "published binding WIT must stay exact");
        }
    }
}