lix 0.18.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::hash::{Hash, Hasher};
use std::sync::Arc;

use crate::{LixError, common::SharedStr, wasm::WasmLimits};
// The guest host contract exposes the engine-owned row payload.
pub use crate::row_payload::{
    CertifiedCreateRange as WasmCertifiedCreateRange, CertifiedRowBatch as WasmCertifiedRowBatch,
    RowCreateContext as WasmCreateContext, TypedRow as WasmTypedRow,
};
use async_trait::async_trait;
use bytes::Bytes;

pub const PACKET_FORMAT_V2: u16 = 2;
pub const CURRENT_PACKET_FORMAT: u16 = PACKET_FORMAT_V2;
/// Durable major version; the canonical WIT package is `lix:plugin-v2`.
pub const WASM_COMPONENT_API_VERSION: &str = "2";
/// 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 FILE_TRANSITION_MAX_TOTAL_BYTES: u64 = 2 * 1024 * MIB;
const FILE_TRANSITION_OUTPUT_EXPANSION: u64 = 16;
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 v2 typed-page 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_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_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_for_file_bytes(&mut self, file_bytes: u64) {
        // A source record expands into typed cells, identity/order columns,
        // and framing. Aggregate streamed traffic is independent of guest
        // linear memory; a million-row CSV must not hit the small-file budget.
        self.max_total_bytes = file_bytes
            .saturating_mul(FILE_TRANSITION_OUTPUT_EXPANSION)
            .clamp(self.max_total_bytes, FILE_TRANSITION_MAX_TOTAL_BYTES);
        // 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)
    }
}

/// Forbidden outer-row JSON work at the plugin boundary.
/// One canonical JSON row backed by cursor-page storage.
/// The typed-row hard cut has no production caller. This enum and
/// [`WasmTransitionCounters::record_outer_row_json_operation`] are the single
/// instrumentation choke point for any future implementation: recording is
/// required before an outer snapshot is parsed, serialized, canonicalized, or
/// materialized as a JSON DOM.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[doc(hidden)]
pub enum OuterRowJsonOperation {
    Parse,
    Serialize,
    Canonicalize,
    DomFallback,
}

/// 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,
    /// Typed row transport work at the host/guest boundary.
    pub typed_row_decode_records: u64,
    pub typed_row_decode_bytes: u64,
    /// Host CPU time spent decoding native typed pages, measured per callback.
    pub typed_row_decode_nanos: u64,
    pub typed_row_encode_records: u64,
    pub typed_row_encode_bytes: u64,
    pub typed_row_schema_validation_calls: u64,
    pub typed_row_schema_validation_bytes: u64,
    /// Host CPU time spent authenticating decoded rows against Schema v1.
    pub typed_row_schema_validation_nanos: u64,
    /// Complete native rows observed by transaction preparation after either
    /// the component or SQL construction boundary. Already-certified rows are
    /// counted as typed-path evidence without being validated a second time.
    pub typed_transaction_validation_calls: u64,
    pub typed_transaction_validation_bytes: u64,
    pub row_page_callback_calls: u64,
    /// Input-source callbacks that returned the terminal `none` page. Together
    /// with input/output page counts this exactly accounts for every callback.
    pub row_input_page_eof_callbacks: u64,
    /// Hard-cut proof counters. Production typed-row workloads must leave all
    /// of these at zero; a positive-control test proves every counter can
    /// become nonzero through the centralized operation hook.
    pub outer_row_json_parse_calls: u64,
    pub outer_row_json_parse_bytes: u64,
    pub outer_row_json_serialize_calls: u64,
    pub outer_row_json_serialize_bytes: u64,
    pub outer_row_json_canonicalize_calls: u64,
    pub outer_row_json_canonicalize_bytes: u64,
    pub outer_row_json_dom_fallback_calls: u64,
    pub outer_row_json_dom_fallback_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 {
    #[doc(hidden)]
    pub fn record_outer_row_json_operation(
        &mut self,
        operation: OuterRowJsonOperation,
        bytes: u64,
    ) {
        let (calls, measured_bytes) = match operation {
            OuterRowJsonOperation::Parse => (
                &mut self.outer_row_json_parse_calls,
                &mut self.outer_row_json_parse_bytes,
            ),
            OuterRowJsonOperation::Serialize => (
                &mut self.outer_row_json_serialize_calls,
                &mut self.outer_row_json_serialize_bytes,
            ),
            OuterRowJsonOperation::Canonicalize => (
                &mut self.outer_row_json_canonicalize_calls,
                &mut self.outer_row_json_canonicalize_bytes,
            ),
            OuterRowJsonOperation::DomFallback => (
                &mut self.outer_row_json_dom_fallback_calls,
                &mut self.outer_row_json_dom_fallback_bytes,
            ),
        };
        *calls = calls.saturating_add(1);
        *measured_bytes = measured_bytes.saturating_add(bytes);
    }

    /// 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.typed_row_decode_records = self
            .typed_row_decode_records
            .saturating_add(other.typed_row_decode_records);
        self.typed_row_decode_bytes = self
            .typed_row_decode_bytes
            .saturating_add(other.typed_row_decode_bytes);
        self.typed_row_decode_nanos = self
            .typed_row_decode_nanos
            .saturating_add(other.typed_row_decode_nanos);
        self.typed_row_encode_records = self
            .typed_row_encode_records
            .saturating_add(other.typed_row_encode_records);
        self.typed_row_encode_bytes = self
            .typed_row_encode_bytes
            .saturating_add(other.typed_row_encode_bytes);
        self.typed_row_schema_validation_calls = self
            .typed_row_schema_validation_calls
            .saturating_add(other.typed_row_schema_validation_calls);
        self.typed_row_schema_validation_bytes = self
            .typed_row_schema_validation_bytes
            .saturating_add(other.typed_row_schema_validation_bytes);
        self.typed_row_schema_validation_nanos = self
            .typed_row_schema_validation_nanos
            .saturating_add(other.typed_row_schema_validation_nanos);
        self.typed_transaction_validation_calls = self
            .typed_transaction_validation_calls
            .saturating_add(other.typed_transaction_validation_calls);
        self.typed_transaction_validation_bytes = self
            .typed_transaction_validation_bytes
            .saturating_add(other.typed_transaction_validation_bytes);
        self.row_page_callback_calls = self
            .row_page_callback_calls
            .saturating_add(other.row_page_callback_calls);
        self.row_input_page_eof_callbacks = self
            .row_input_page_eof_callbacks
            .saturating_add(other.row_input_page_eof_callbacks);
        self.outer_row_json_parse_calls = self
            .outer_row_json_parse_calls
            .saturating_add(other.outer_row_json_parse_calls);
        self.outer_row_json_parse_bytes = self
            .outer_row_json_parse_bytes
            .saturating_add(other.outer_row_json_parse_bytes);
        self.outer_row_json_serialize_calls = self
            .outer_row_json_serialize_calls
            .saturating_add(other.outer_row_json_serialize_calls);
        self.outer_row_json_serialize_bytes = self
            .outer_row_json_serialize_bytes
            .saturating_add(other.outer_row_json_serialize_bytes);
        self.outer_row_json_canonicalize_calls = self
            .outer_row_json_canonicalize_calls
            .saturating_add(other.outer_row_json_canonicalize_calls);
        self.outer_row_json_canonicalize_bytes = self
            .outer_row_json_canonicalize_bytes
            .saturating_add(other.outer_row_json_canonicalize_bytes);
        self.outer_row_json_dom_fallback_calls = self
            .outer_row_json_dom_fallback_calls
            .saturating_add(other.outer_row_json_dom_fallback_calls);
        self.outer_row_json_dom_fallback_bytes = self
            .outer_row_json_dom_fallback_bytes
            .saturating_add(other.outer_row_json_dom_fallback_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(Debug, Clone)]
pub enum WasmHostBytes {
    /// Native Schema v1 values retained after typed-row page validation. This
    /// is the only row payload accepted by the plugin runtime.
    Typed(Arc<WasmTypedRow>),
}

impl WasmHostBytes {
    pub fn len(&self) -> u64 {
        match self {
            Self::Typed(row) => row.estimated_size(),
        }
    }

    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)]
pub struct WasmRowKey {
    pub schema_key: SharedStr,
    pub schema_fingerprint: [u8; 32],
    pub row_pk: Arc<[lix_schema::Value]>,
}

impl PartialEq for WasmRowKey {
    #[inline]
    fn eq(&self, other: &Self) -> bool {
        self.schema_key == other.schema_key
            && self.schema_fingerprint == other.schema_fingerprint
            && typed_key_values_cmp(&self.row_pk, &other.row_pk).is_eq()
    }
}

impl Eq for WasmRowKey {}

impl PartialOrd for WasmRowKey {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for WasmRowKey {
    #[inline]
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.schema_key
            .cmp(&other.schema_key)
            .then_with(|| self.schema_fingerprint.cmp(&other.schema_fingerprint))
            .then_with(|| typed_key_values_cmp(&self.row_pk, &other.row_pk))
    }
}

impl Hash for WasmRowKey {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.schema_key.hash(state);
        self.schema_fingerprint.hash(state);
        for value in self.row_pk.iter() {
            hash_typed_key_value(value, state);
        }
    }
}

impl WasmRowKey {
    pub fn from_typed_parts(
        schema_key: impl Into<SharedStr>,
        schema_fingerprint: [u8; 32],
        row_pk: impl Into<Arc<[lix_schema::Value]>>,
    ) -> Result<Self, LixError> {
        let row_pk = row_pk.into();
        if row_pk.is_empty()
            || row_pk.iter().any(|value| {
                !matches!(
                    value,
                    lix_schema::Value::Text(_)
                        | lix_schema::Value::Uuid(_)
                        | lix_schema::Value::Int8(_)
                )
            })
        {
            return Err(LixError::new(
                LixError::CODE_SCHEMA_VALIDATION,
                "typed row keys must contain one or more text, uuid, or int8 values",
            ));
        }
        Ok(Self {
            schema_key: schema_key.into(),
            schema_fingerprint,
            row_pk,
        })
    }
}

#[inline]
fn typed_key_values_cmp(
    left: &[lix_schema::Value],
    right: &[lix_schema::Value],
) -> std::cmp::Ordering {
    use lix_schema::Value;

    match (left, right) {
        ([Value::Text(left)], [Value::Text(right)]) => return left.cmp(right),
        ([Value::Uuid(left)], [Value::Uuid(right)]) => return left.cmp(right),
        ([Value::Int8(left)], [Value::Int8(right)]) => return left.cmp(right),
        _ => {}
    }
    for (left, right) in left.iter().zip(right) {
        let ordering = typed_key_value_cmp(left, right);
        if !ordering.is_eq() {
            return ordering;
        }
    }
    left.len().cmp(&right.len())
}

fn typed_key_value_cmp(left: &lix_schema::Value, right: &lix_schema::Value) -> std::cmp::Ordering {
    fn tag(value: &lix_schema::Value) -> u8 {
        match value {
            lix_schema::Value::Text(_) => 0,
            lix_schema::Value::Uuid(_) => 1,
            lix_schema::Value::Int8(_) => 2,
            _ => unreachable!("typed row key construction rejects non-key values"),
        }
    }
    tag(left)
        .cmp(&tag(right))
        .then_with(|| match (left, right) {
            (lix_schema::Value::Text(left), lix_schema::Value::Text(right)) => left.cmp(right),
            (lix_schema::Value::Uuid(left), lix_schema::Value::Uuid(right)) => left.cmp(right),
            (lix_schema::Value::Int8(left), lix_schema::Value::Int8(right)) => left.cmp(right),
            _ => std::cmp::Ordering::Equal,
        })
}

fn hash_typed_key_value<H: Hasher>(value: &lix_schema::Value, state: &mut H) {
    match value {
        lix_schema::Value::Text(value) => {
            0_u8.hash(state);
            value.hash(state);
        }
        lix_schema::Value::Uuid(value) => {
            1_u8.hash(state);
            value.hash(state);
        }
        lix_schema::Value::Int8(value) => {
            2_u8.hash(state);
            value.hash(state);
        }
        _ => unreachable!("typed row key construction rejects non-key values"),
    }
}

#[derive(Debug, Clone)]
pub struct WasmRow<B> {
    pub key: WasmRowKey,
    /// Native Schema v1 row payload for the schema identified by `key`.
    pub payload: 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: SharedStr,
        local_ref: u64,
        /// The v2 host may resolve the host-namespaced primary key before canonical
        /// validation. The component leaves this absent and retains keyless semantics
        /// until create-context materialization.
        resolved_key: Option<WasmRowKey>,
        payload: 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<WasmGuestRowPayload>;

#[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
/// encoded in the current typed-page format.
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 and overlapping values are native Schema v1 values; no row snapshot
/// or JSON column representation crosses the component boundary.
#[derive(Debug, Clone)]
pub struct WasmColumnMerge {
    pub ordinal: u32,
    pub key: WasmRowKey,
    pub file_id: Option<String>,
    pub column: String,
    pub schema_fingerprint: [u8; 32],
    pub base: Option<lix_schema::Value>,
    pub a: Option<lix_schema::Value>,
    pub b: Option<lix_schema::Value>,
    pub base_row: Arc<WasmTypedRow>,
    pub a_row: Arc<WasmTypedRow>,
    pub b_row: Arc<WasmTypedRow>,
}

pub type WasmHostColumnMerge = WasmColumnMerge;

#[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 the native Schema v1
    /// value, including `Value::Jsonb(Null)` for an explicit JSON null.
    Replace(Option<B>),
}

pub type WasmGuestColumnMergeResult = WasmColumnMergeResult<WasmGuestColumnValue>;

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

pub struct WasmOpenFileInput {
    pub descriptor: WasmFileDescriptor,
    pub file: Arc<dyn WasmByteSource>,
    pub creates: WasmCreateContext,
}

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)
            .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);

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

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,
        }
    }

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

    pub fn retained_bytes(&self) -> u64 {
        self.retained_bytes
    }
}
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)]
pub enum WasmGuestRowPayload {
    /// Native Schema v1 values emitted by a typed-row page.
    Typed(Arc<WasmTypedRow>),
}

#[derive(Debug, Clone, PartialEq)]
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),
}

/// A typed column-merge replacement whose bytes are streamed through the
/// page-local output table and decoded as one native Schema v1 value by the
/// host.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum WasmGuestColumnValue {
    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)]
pub struct WasmOutputSplice {
    pub offset: u64,
    pub delete_len: u64,
    pub insert: WasmGuestBytes,
}

#[derive(Debug, Clone, PartialEq)]
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 != CURRENT_PACKET_FORMAT {
            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",
            ));
        }

        for change in &page.changes.changes {
            match change {
                WasmRowChange::Create {
                    payload: WasmGuestRowPayload::Typed(_),
                    ..
                }
                | WasmRowChange::Upsert {
                    row:
                        WasmRow {
                            payload: WasmGuestRowPayload::Typed(_),
                            ..
                        },
                    ..
                }
                | WasmRowChange::Delete(_) => {}
            }
        }
        validate_attachment_table_presence(0, page.outputs.is_some())?;
        let page_refs = 0u32;
        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,
}

#[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",
        ))
    }

    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>;

    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::*;

    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(),
            },
        }
    }

    #[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"))
        }
    }

    #[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 outer_row_json_counter_positive_control_covers_every_forbidden_operation() {
        let mut counters = WasmTransitionCounters::default();
        for operation in [
            OuterRowJsonOperation::Parse,
            OuterRowJsonOperation::Serialize,
            OuterRowJsonOperation::Canonicalize,
            OuterRowJsonOperation::DomFallback,
        ] {
            counters.record_outer_row_json_operation(operation, 17);
        }
        let mut aggregate = WasmTransitionCounters::default();
        aggregate.accumulate(counters);

        assert_eq!(aggregate.outer_row_json_parse_calls, 1);
        assert_eq!(aggregate.outer_row_json_parse_bytes, 17);
        assert_eq!(aggregate.outer_row_json_serialize_calls, 1);
        assert_eq!(aggregate.outer_row_json_serialize_bytes, 17);
        assert_eq!(aggregate.outer_row_json_canonicalize_calls, 1);
        assert_eq!(aggregate.outer_row_json_canonicalize_bytes, 17);
        assert_eq!(aggregate.outer_row_json_dom_fallback_calls, 1);
        assert_eq!(aggregate.outer_row_json_dom_fallback_bytes, 17);
    }

    #[test]
    fn duplicate_sort_borrows_native_key_owners() {
        let schema = SharedStr::from_static("csv_row");
        let changes = (0..10_000)
            .map(|ordinal| {
                let row_pk = if ordinal % 2 == 0 {
                    vec![lix_schema::Value::Text("namespace".to_owned())]
                } else {
                    vec![
                        lix_schema::Value::Text("namespace".to_owned()),
                        lix_schema::Value::Text("row".to_owned()),
                    ]
                };
                WasmRowChange::<WasmGuestBytes>::Delete(WasmRowKey {
                    schema_key: schema.clone(),
                    schema_fingerprint: [7; 32],
                    row_pk: row_pk.into(),
                })
            })
            .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.schema_key.shares_buffer_with(&schema));
        }

        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 typed_row_key_retains_native_components() {
        let id =
            uuid::Uuid::parse_str("01920000-0000-7000-8000-0000000000aa").expect("fixture UUID");
        let values = vec![lix_schema::Value::Uuid(id), lix_schema::Value::Int8(42)];
        let key =
            WasmRowKey::from_typed_parts("typed", [7; 32], values.clone()).expect("typed key");
        let same_native_key =
            WasmRowKey::from_typed_parts("typed", [7; 32], values.clone()).expect("typed key");
        assert_eq!(key.row_pk.as_ref(), values.as_slice());
        assert_eq!(key.schema_fingerprint, [7; 32]);
        assert_eq!(key, same_native_key);
        assert_eq!(key.cmp(&same_native_key), std::cmp::Ordering::Equal);
    }

    #[test]
    fn cursor_key_uniqueness_accepts_arbitrary_unique_order() {
        let changes = vec![
            WasmRowChange::<WasmGuestBytes>::Delete(
                WasmRowKey::from_typed_parts(
                    "schema-z",
                    [1; 32],
                    vec![lix_schema::Value::Text("row-z".to_owned())],
                )
                .unwrap(),
            ),
            WasmRowChange::<WasmGuestBytes>::Delete(
                WasmRowKey::from_typed_parts(
                    "schema-a",
                    [2; 32],
                    vec![lix_schema::Value::Text("row-a".to_owned())],
                )
                .unwrap(),
            ),
        ];

        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(),
            uuid::Uuid::parse_str("01920000-0000-7000-8000-00000000002a").unwrap()
        );
        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_typed_parts(
            "csv_row",
            [0; 32],
            vec![lix_schema::Value::Text("row".to_owned())],
        )
        .unwrap();
        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 file_output_budget_admits_large_row_streams_and_remains_capped() {
        for make in [
            WasmTransitionLimits::for_file_bytes,
            WasmTransitionLimits::for_cold_file_bytes,
        ] {
            assert_eq!(
                make(0).max_total_bytes,
                WasmTransitionLimits::default().max_total_bytes
            );
            // The 45 MB million-row CSV emits more than the old 128 MiB cap.
            let limits = make(45_000_000);
            assert!(limits.max_total_bytes >= 512 * MIB);
            assert_eq!(
                make(u64::MAX).max_total_bytes,
                FILE_TRANSITION_MAX_TOTAL_BYTES
            );
            assert_eq!(limits.max_pages, WasmTransitionLimits::default().max_pages);
            limits.validate().unwrap();
            make(u64::MAX).validate().unwrap();
        }
        // Enlarging streamed output does not extend hot-operation time.
        assert_eq!(
            WasmTransitionLimits::for_file_bytes(u64::MAX).total_deadline_nanoseconds,
            WasmTransitionLimits::default().total_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_typed_parts(
            "csv_row",
            [0; 32],
            vec![lix_schema::Value::Text("row".to_owned())],
        )
        .unwrap();
        let page = WasmChangePage {
            format_version: CURRENT_PACKET_FORMAT,
            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-v2;"));
        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"));
    }
}