radixdb-plugin 1.1.0

use std::{cmp::Ordering, fmt::Debug};

use radixdb_plugin_abi as abi;
use sha2::{Digest, Sha256};

use crate::{
    BoundedBytes, BoundedText, CodecReader, CodecWriter, PluginError, PluginResult, RadixType,
};

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TypeTestReport {
    pub corpus_values: usize,
    pub codec_vectors: Vec<Vec<u8>>,
    pub hash_vectors: Vec<[u8; 32]>,
}

/// Core scalar ordering used to prove that an operator-class key encoder
/// preserves the semantic relation declared by its external input type.
pub trait OperatorClassKey {
    fn key_compare(&self, other: &Self) -> Ordering;
}

macro_rules! ordered_key {
    ($($type:ty),+ $(,)?) => {
        $(
            impl OperatorClassKey for $type {
                fn key_compare(&self, other: &Self) -> Ordering {
                    self.cmp(other)
                }
            }
        )+
    };
}

ordered_key!(i8, i16, i32, i64, u8, u16, u32, u64, bool);

impl OperatorClassKey for f32 {
    fn key_compare(&self, other: &Self) -> Ordering {
        self.total_cmp(other)
    }
}

impl OperatorClassKey for f64 {
    fn key_compare(&self, other: &Self) -> Ordering {
        self.total_cmp(other)
    }
}

impl<const MAX: usize> OperatorClassKey for BoundedBytes<MAX> {
    fn key_compare(&self, other: &Self) -> Ordering {
        self.as_slice().cmp(other.as_slice())
    }
}

impl<const MAX: usize> OperatorClassKey for BoundedText<MAX> {
    fn key_compare(&self, other: &Self) -> Ordering {
        self.as_str().cmp(other.as_str())
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TestValue {
    type_ref: abi::RadixAbiTypeRefV1,
    bytes: Vec<u8>,
    is_null: bool,
}

impl TestValue {
    pub fn integer(value: i64) -> Self {
        Self {
            type_ref: abi::RadixAbiTypeRefV1::builtin(abi::RADIX_BUILTIN_INTEGER),
            bytes: value.to_le_bytes().to_vec(),
            is_null: false,
        }
    }

    pub fn float(value: f64) -> Self {
        Self {
            type_ref: abi::RadixAbiTypeRefV1::builtin(abi::RADIX_BUILTIN_FLOAT),
            bytes: value.to_bits().to_le_bytes().to_vec(),
            is_null: false,
        }
    }

    pub fn boolean(value: bool) -> Self {
        Self {
            type_ref: abi::RadixAbiTypeRefV1::builtin(abi::RADIX_BUILTIN_BOOLEAN),
            bytes: vec![u8::from(value)],
            is_null: false,
        }
    }

    pub fn external<T: RadixType>(
        package: &'static abi::RadixPluginDescriptorV1,
        value: &T,
    ) -> PluginResult<Self> {
        let descriptor = find_type(package, T::LOCAL_ID)?;
        if descriptor.codec_version != T::CODEC_VERSION {
            return Err(PluginError::invalid_input(
                "test type codec differs from package descriptor",
            ));
        }
        Ok(Self {
            type_ref: abi::RadixAbiTypeRefV1::external(
                descriptor.object_id,
                descriptor.codec_version,
            ),
            bytes: encode(value)?,
            is_null: false,
        })
    }

    pub fn null_like(value: &Self) -> Self {
        Self {
            type_ref: value.type_ref,
            bytes: Vec::new(),
            is_null: true,
        }
    }

    pub fn type_ref(&self) -> abi::RadixAbiTypeRefV1 {
        self.type_ref
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TestCallOptions {
    pub cancelled: bool,
    pub deadline_expired: bool,
    pub max_output_bytes: u32,
    pub max_work_units: u32,
}

impl Default for TestCallOptions {
    fn default() -> Self {
        Self {
            cancelled: false,
            deadline_expired: false,
            max_output_bytes: 1024 * 1024,
            max_work_units: 1024 * 1024,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TestOutput {
    pub is_null: bool,
    pub bytes: Vec<u8>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TestDiagnostic {
    pub category: u32,
    pub status: abi::RadixAbiStatusV1,
    pub detail: String,
    pub field: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TestCallReport {
    pub status: abi::RadixAbiStatusV1,
    pub outputs: Vec<TestOutput>,
    pub diagnostics: Vec<TestDiagnostic>,
    pub work_charged: u32,
    pub finished: bool,
}

pub fn validate_descriptor_graph(
    package: &'static abi::RadixPluginDescriptorV1,
) -> PluginResult<()> {
    abi::validate_package_descriptor_shallow(package).map_err(validation_error)?;
    // SAFETY: shallow validation established the pointer/count shape of every
    // immutable process-lifetime descriptor table generated by the SDK.
    unsafe {
        for item in descriptor_slice(package.types, package.type_count) {
            abi::validate_external_type_descriptor(item).map_err(validation_error)?;
        }
        for item in descriptor_slice(package.functions, package.function_count) {
            abi::validate_scalar_function_descriptor(item).map_err(validation_error)?;
            for argument in descriptor_slice(item.arguments, item.argument_count) {
                abi::validate_type_ref(argument).map_err(validation_error)?;
            }
        }
        for item in descriptor_slice(package.operators, package.operator_count) {
            abi::validate_operator_descriptor(item).map_err(validation_error)?;
        }
        for item in descriptor_slice(package.operator_classes, package.operator_class_count) {
            abi::validate_operator_class_descriptor(item).map_err(validation_error)?;
        }
        for item in descriptor_slice(package.planner_support, package.planner_support_count) {
            abi::validate_planner_support_descriptor(item).map_err(validation_error)?;
        }
    }
    Ok(())
}

pub fn invoke_scalar(
    package: &'static abi::RadixPluginDescriptorV1,
    local_id: &str,
    arguments: &[TestValue],
    options: TestCallOptions,
) -> PluginResult<TestCallReport> {
    validate_descriptor_graph(package)?;
    let function = find_function(package, local_id)?;
    if arguments.len() != function.argument_count as usize {
        return Err(PluginError::invalid_input(
            "test scalar argument count mismatch",
        ));
    }
    let expected = unsafe { descriptor_slice(function.arguments, function.argument_count) };
    for (argument, expected) in arguments.iter().zip(expected) {
        if argument.type_ref != *expected {
            return Err(PluginError::invalid_input(
                "test scalar argument type mismatch",
            ));
        }
    }
    let raw_arguments = arguments.iter().map(raw_value).collect::<Vec<_>>();
    let callback = function
        .scalar
        .ok_or_else(|| PluginError::internal("scalar descriptor has no callback"))?;
    let mut state = TestHostState::new(options);
    let diagnostic_sink = state.diagnostic_sink();
    let context = state.call_context(&diagnostic_sink, function.max_output_bytes);
    let result_builder = state.result_builder(function.max_output_bytes, 1);
    let argument_pointer = if raw_arguments.is_empty() {
        std::ptr::null()
    } else {
        raw_arguments.as_ptr()
    };
    let status = unsafe {
        callback(
            &context,
            argument_pointer,
            raw_arguments.len() as u32,
            &result_builder,
        )
    };
    Ok(state.report(status))
}

pub fn invoke_batch(
    package: &'static abi::RadixPluginDescriptorV1,
    local_id: &str,
    rows: &[Vec<TestValue>],
    options: TestCallOptions,
) -> PluginResult<TestCallReport> {
    validate_descriptor_graph(package)?;
    let function = find_function(package, local_id)?;
    let callback = function
        .batch
        .ok_or_else(|| PluginError::invalid_input("function has no batch adapter"))?;
    if rows.len() > u32::MAX as usize
        || rows
            .iter()
            .any(|row| row.len() != function.argument_count as usize)
    {
        return Err(PluginError::invalid_input("test batch shape mismatch"));
    }
    let expected = unsafe { descriptor_slice(function.arguments, function.argument_count) };
    for row in rows {
        for (value, expected) in row.iter().zip(expected) {
            if value.type_ref != *expected {
                return Err(PluginError::invalid_input("test batch type mismatch"));
            }
        }
    }
    let columns = expected
        .iter()
        .enumerate()
        .map(|(index, type_ref)| {
            let values = rows.iter().map(|row| &row[index]).collect::<Vec<_>>();
            OwnedColumn::new(package, *type_ref, &values)
        })
        .collect::<PluginResult<Vec<_>>>()?;
    let raw_columns = columns
        .iter()
        .map(|column| column.as_abi(rows.len() as u32))
        .collect::<Vec<_>>();
    let batch = abi::RadixAbiBatchViewV1 {
        header: abi::RadixAbiHeaderV1::new::<abi::RadixAbiBatchViewV1>(0),
        row_count: rows.len() as u32,
        column_count: raw_columns.len() as u32,
        columns: if raw_columns.is_empty() {
            std::ptr::null()
        } else {
            raw_columns.as_ptr()
        },
    };
    unsafe { abi::validate_batch_columns(&batch) }.map_err(validation_error)?;
    let mut state = TestHostState::new(options);
    let diagnostic_sink = state.diagnostic_sink();
    let batch_output_bytes = function.max_output_bytes.saturating_mul(rows.len() as u32);
    let context = state.call_context(&diagnostic_sink, batch_output_bytes);
    let result_builder = state.result_builder(batch_output_bytes, rows.len() as u32);
    let status = unsafe { callback(&context, &batch, &result_builder) };
    Ok(state.report(status))
}

pub fn invoke_planner(
    package: &'static abi::RadixPluginDescriptorV1,
    local_id: &str,
    predicate: &[u8],
    options: TestCallOptions,
) -> PluginResult<TestCallReport> {
    validate_descriptor_graph(package)?;
    let support = find_planner_support(package, local_id)?;
    let callback = support
        .callback
        .ok_or_else(|| PluginError::internal("planner descriptor has no callback"))?;
    let mut state = TestHostState::new(options);
    let diagnostic_sink = state.diagnostic_sink();
    let context = state.call_context(&diagnostic_sink, support.max_output_bytes);
    let result_builder = state.result_builder(
        support.max_output_bytes,
        support.max_spans.saturating_add(1),
    );
    let predicate = abi::RadixAbiSliceV1 {
        ptr: if predicate.is_empty() {
            std::ptr::null()
        } else {
            predicate.as_ptr()
        },
        len: predicate.len() as u32,
        reserved: 0,
    };
    let status = unsafe { callback(&context, predicate, &result_builder) };
    Ok(state.report(status))
}

pub fn planner_recheck_policy(
    package: &'static abi::RadixPluginDescriptorV1,
    local_id: &str,
) -> PluginResult<u16> {
    validate_descriptor_graph(package)?;
    Ok(find_planner_support(package, local_id)?.recheck_policy)
}

pub fn decode_candidate_spans(report: &TestCallReport) -> PluginResult<Vec<crate::CandidateSpan>> {
    report
        .outputs
        .iter()
        .filter(|item| item.bytes.first() == Some(&1))
        .map(|item| {
            if item.is_null || item.bytes.len() < 12 || item.bytes[..4] != [1, 0, 0, 0] {
                return Err(PluginError::invalid_input(
                    "planner output is not a candidate span",
                ));
            }
            let start_len =
                u32::from_le_bytes(item.bytes[4..8].try_into().expect("fixed width")) as usize;
            let end_len =
                u32::from_le_bytes(item.bytes[8..12].try_into().expect("fixed width")) as usize;
            let split = 12_usize
                .checked_add(start_len)
                .ok_or_else(|| PluginError::invalid_input("candidate span length overflow"))?;
            let end = split
                .checked_add(end_len)
                .ok_or_else(|| PluginError::invalid_input("candidate span length overflow"))?;
            if end != item.bytes.len() {
                return Err(PluginError::invalid_input(
                    "candidate span has invalid lengths",
                ));
            }
            Ok(crate::CandidateSpan {
                start: item.bytes[12..split].to_vec(),
                end: item.bytes[split..end].to_vec(),
            })
        })
        .collect()
}

pub fn check_type<T>() -> PluginResult<TypeTestReport>
where
    T: RadixType + Debug,
{
    let corpus = T::test_corpus();
    if corpus.is_empty() {
        return Err(PluginError::invalid_input(
            "type test corpus must not be empty",
        ));
    }
    let mut codec_vectors = Vec::with_capacity(corpus.len());
    for value in &corpus {
        let bytes = encode(value)?;
        let decoded = decode::<T>(&bytes)?;
        let second = encode(&decoded)?;
        if bytes != second {
            return Err(PluginError::domain(
                "codec is not canonical after roundtrip",
            ));
        }
        codec_vectors.push(bytes);
    }

    let mut hashes = vec![None; corpus.len()];
    for (index, value) in corpus.iter().enumerate() {
        let mut components = Vec::new();
        let mut sink = crate::HashSink::for_testing(&mut components);
        if let Some(result) = T::semantic_hash(value, &mut sink) {
            result?;
            hashes[index] = Some(hash_components(&components));
        }
    }

    if T::CAPABILITIES & radixdb_plugin_abi::RADIX_TYPE_CAP_EQUALITY != 0 {
        for (left_index, left) in corpus.iter().enumerate() {
            if T::semantic_equal(left, left) != Some(true) {
                return Err(PluginError::domain("equality is not reflexive"));
            }
            for (right_index, right) in corpus.iter().enumerate() {
                let lr = T::semantic_equal(left, right).ok_or_else(|| {
                    PluginError::internal("equality capability has no safe callback")
                })?;
                let rl = T::semantic_equal(right, left).ok_or_else(|| {
                    PluginError::internal("equality capability has no safe callback")
                })?;
                if lr != rl {
                    return Err(PluginError::domain("equality is not symmetric"));
                }
                if lr && hashes[left_index] != hashes[right_index] {
                    return Err(PluginError::domain(
                        "equal values produce different semantic hash components",
                    ));
                }
                for third in &corpus {
                    if lr
                        && T::semantic_equal(right, third) == Some(true)
                        && T::semantic_equal(left, third) != Some(true)
                    {
                        return Err(PluginError::domain("equality is not transitive"));
                    }
                }
            }
        }
    }

    if T::CAPABILITIES & radixdb_plugin_abi::RADIX_TYPE_CAP_ORDERING != 0 {
        for left in &corpus {
            if T::semantic_compare(left, left) != Some(Ordering::Equal) {
                return Err(PluginError::domain("ordering is not reflexive"));
            }
            for right in &corpus {
                let lr = T::semantic_compare(left, right)
                    .ok_or_else(|| PluginError::internal("missing ordering callback"))?;
                let rl = T::semantic_compare(right, left)
                    .ok_or_else(|| PluginError::internal("missing ordering callback"))?;
                if lr != rl.reverse() {
                    return Err(PluginError::domain("ordering is not antisymmetric"));
                }
                if T::CAPABILITIES & radixdb_plugin_abi::RADIX_TYPE_CAP_EQUALITY != 0
                    && (lr == Ordering::Equal) != (T::semantic_equal(left, right) == Some(true))
                {
                    return Err(PluginError::domain(
                        "ordering equality disagrees with equality callback",
                    ));
                }
                for third in &corpus {
                    if lr != Ordering::Greater
                        && T::semantic_compare(right, third) != Some(Ordering::Greater)
                        && T::semantic_compare(left, third) == Some(Ordering::Greater)
                    {
                        return Err(PluginError::domain("ordering is not transitive"));
                    }
                }
            }
        }
    }

    Ok(TypeTestReport {
        corpus_values: corpus.len(),
        codec_vectors,
        hash_vectors: hashes.into_iter().flatten().collect(),
    })
}

/// Verify the full B-tree law: the physical key is an exact order embedding
/// of the external type's equality and total-order callbacks for its corpus.
pub fn check_btree_operator_class<T, K>(encode_key: fn(T) -> PluginResult<K>) -> PluginResult<()>
where
    T: RadixType + Debug,
    K: OperatorClassKey,
{
    let report = check_type::<T>()?;
    if T::CAPABILITIES & abi::RADIX_TYPE_CAP_EQUALITY == 0
        || T::CAPABILITIES & abi::RADIX_TYPE_CAP_ORDERING == 0
    {
        return Err(PluginError::domain(
            "B-tree operator class requires equality and total ordering",
        ));
    }
    let input = T::test_corpus()
        .iter()
        .map(encode)
        .collect::<PluginResult<Vec<_>>>()?;
    if input.len() != report.corpus_values {
        return Err(PluginError::internal(
            "operator-class corpus changed between law checks",
        ));
    }
    let keys = input
        .iter()
        .map(|bytes| decode::<T>(bytes).and_then(encode_key))
        .collect::<PluginResult<Vec<_>>>()?;
    for (left_index, left_bytes) in input.iter().enumerate() {
        for (right_index, right_bytes) in input.iter().enumerate() {
            let left = decode::<T>(left_bytes)?;
            let right = decode::<T>(right_bytes)?;
            let semantic_order = T::semantic_compare(&left, &right)
                .ok_or_else(|| PluginError::internal("missing ordering callback"))?;
            let semantic_equal = T::semantic_equal(&left, &right)
                .ok_or_else(|| PluginError::internal("missing equality callback"))?;
            let key_order = keys[left_index].key_compare(&keys[right_index]);
            if key_order != semantic_order || (key_order == Ordering::Equal) != semantic_equal {
                return Err(PluginError::domain(
                    "B-tree key encoder does not preserve semantic equality and total order",
                ));
            }
        }
    }
    Ok(())
}

/// Hash classes never select a hash algorithm or physical token. Their law is
/// exactly the external type equality/hash-component contract.
pub fn check_hash_operator_class<T>() -> PluginResult<()>
where
    T: RadixType + Debug,
{
    let report = check_type::<T>()?;
    if T::CAPABILITIES & abi::RADIX_TYPE_CAP_EQUALITY == 0
        || T::CAPABILITIES & abi::RADIX_TYPE_CAP_HASH == 0
        || report.hash_vectors.len() != report.corpus_values
    {
        return Err(PluginError::domain(
            "hash operator class requires equality and semantic hash components",
        ));
    }
    Ok(())
}

/// Bitmap keys may coalesce byte representation only when the external values
/// are semantically equal; otherwise exact lookup and uniqueness would diverge.
pub fn check_bitmap_operator_class<T, K>(encode_key: fn(T) -> PluginResult<K>) -> PluginResult<()>
where
    T: RadixType + Debug,
    K: OperatorClassKey,
{
    let _ = check_type::<T>()?;
    if T::CAPABILITIES & abi::RADIX_TYPE_CAP_EQUALITY == 0 {
        return Err(PluginError::domain(
            "bitmap operator class requires equality",
        ));
    }
    let input = T::test_corpus()
        .iter()
        .map(encode)
        .collect::<PluginResult<Vec<_>>>()?;
    let keys = input
        .iter()
        .map(|bytes| decode::<T>(bytes).and_then(encode_key))
        .collect::<PluginResult<Vec<_>>>()?;
    for (left_index, left_bytes) in input.iter().enumerate() {
        for (right_index, right_bytes) in input.iter().enumerate() {
            let left = decode::<T>(left_bytes)?;
            let right = decode::<T>(right_bytes)?;
            let semantic_equal = T::semantic_equal(&left, &right)
                .ok_or_else(|| PluginError::internal("missing equality callback"))?;
            let key_equal = keys[left_index].key_compare(&keys[right_index]) == Ordering::Equal;
            if key_equal != semantic_equal {
                return Err(PluginError::domain(
                    "bitmap key encoder does not preserve semantic equality",
                ));
            }
        }
    }
    Ok(())
}

/// Compare every strategy function with the equality/ordering callbacks of
/// the external input type over the same bounded corpus.
pub fn check_operator_class_strategies<T>(
    package: &'static abi::RadixPluginDescriptorV1,
    local_id: &str,
) -> PluginResult<()>
where
    T: RadixType + Debug,
{
    validate_descriptor_graph(package)?;
    let class = find_operator_class(package, local_id)?;
    let external_type = find_type(package, T::LOCAL_ID)?;
    let expected_type =
        abi::RadixAbiTypeRefV1::external(external_type.object_id, external_type.codec_version);
    if class.input_type != expected_type {
        return Err(PluginError::domain(
            "operator-class input differs from its tested external type",
        ));
    }
    let strategies = unsafe { descriptor_slice(class.strategies, class.strategy_count) };
    let corpus = T::test_corpus();
    let values = corpus
        .iter()
        .map(|value| TestValue::external(package, value))
        .collect::<PluginResult<Vec<_>>>()?;
    let encoded = corpus
        .iter()
        .map(encode)
        .collect::<PluginResult<Vec<_>>>()?;
    for (left_index, left_bytes) in encoded.iter().enumerate() {
        for (right_index, right_bytes) in encoded.iter().enumerate() {
            let left = decode::<T>(left_bytes)?;
            let right = decode::<T>(right_bytes)?;
            let equal = T::semantic_equal(&left, &right)
                .ok_or_else(|| PluginError::internal("missing equality callback"))?;
            let order = T::semantic_compare(&left, &right);
            for strategy in strategies {
                let expected = match (class.access_method, strategy.slot) {
                    (abi::RADIX_ACCESS_METHOD_BTREE, 1) => order == Some(Ordering::Less),
                    (abi::RADIX_ACCESS_METHOD_BTREE, 2) => {
                        order.is_some_and(|value| value != Ordering::Greater)
                    }
                    (abi::RADIX_ACCESS_METHOD_BTREE, 3)
                    | (abi::RADIX_ACCESS_METHOD_HASH, 1)
                    | (abi::RADIX_ACCESS_METHOD_BITMAP, 1) => equal,
                    (abi::RADIX_ACCESS_METHOD_BTREE, 4) => {
                        order.is_some_and(|value| value != Ordering::Less)
                    }
                    (abi::RADIX_ACCESS_METHOD_BTREE, 5) => order == Some(Ordering::Greater),
                    _ => {
                        return Err(PluginError::domain(
                            "operator class has an unsupported strategy slot",
                        ));
                    }
                };
                let actual = invoke_boolean_operator_by_id(
                    package,
                    strategy.object_id,
                    &values[left_index],
                    &values[right_index],
                )?;
                if actual != expected {
                    return Err(PluginError::domain(
                        "operator-class strategy disagrees with type semantics",
                    ));
                }
            }
        }
    }
    Ok(())
}

pub fn golden_vectors<T: RadixType + Debug>() -> PluginResult<Vec<Vec<u8>>> {
    Ok(check_type::<T>()?.codec_vectors)
}

pub fn fuzz_malformed_external_bytes<T: RadixType>(inputs: &[&[u8]]) -> usize {
    inputs
        .iter()
        .filter(|bytes| decode::<T>(bytes).is_err())
        .count()
}

fn encode<T: RadixType>(value: &T) -> PluginResult<Vec<u8>> {
    let mut output = CodecWriter::new(T::MAX_BYTES as usize);
    value.encode(&mut output)?;
    let bytes = output.into_bytes();
    if T::STORAGE_KIND == abi::RADIX_EXTERNAL_STORAGE_FIXED
        && bytes.len() != T::FIXED_BYTES as usize
    {
        return Err(PluginError::domain(
            "fixed codec corpus value has the wrong width",
        ));
    }
    Ok(bytes)
}

fn decode<T: RadixType>(bytes: &[u8]) -> PluginResult<T> {
    if T::STORAGE_KIND == abi::RADIX_EXTERNAL_STORAGE_FIXED
        && bytes.len() != T::FIXED_BYTES as usize
    {
        return Err(PluginError::invalid_input(
            "fixed codec input has the wrong width",
        ));
    }
    let mut input = CodecReader::new(bytes);
    let value = T::decode(&mut input)?;
    input.finish()?;
    Ok(value)
}

pub(crate) fn hash_components(components: &[(u16, Vec<u8>)]) -> [u8; 32] {
    let mut digest = Sha256::new();
    for (kind, bytes) in components {
        digest.update(kind.to_le_bytes());
        digest.update((bytes.len() as u32).to_le_bytes());
        digest.update(bytes);
    }
    digest.finalize().into()
}

fn validation_error(error: abi::RadixAbiValidationError) -> PluginError {
    PluginError::invalid_input(format!("invalid generated ABI descriptor: {error:?}"))
}

unsafe fn descriptor_slice<'a, T>(pointer: *const T, count: u32) -> &'a [T] {
    if count == 0 {
        &[]
    } else {
        // SAFETY: callers first validate the static descriptor table shape.
        unsafe { std::slice::from_raw_parts(pointer, count as usize) }
    }
}

fn abi_text(value: abi::RadixAbiStringV1) -> PluginResult<&'static str> {
    if value.len == 0 || value.ptr.is_null() {
        return Err(PluginError::invalid_input(
            "empty generated descriptor name",
        ));
    }
    // SAFETY: generated descriptor strings have static lifetime and the graph
    // validator checks their pointer/length shape before lookup.
    let bytes = unsafe { std::slice::from_raw_parts(value.ptr, value.len as usize) };
    std::str::from_utf8(bytes)
        .map_err(|_| PluginError::invalid_input("generated descriptor name is not UTF-8"))
}

fn find_type(
    package: &'static abi::RadixPluginDescriptorV1,
    local_id: &str,
) -> PluginResult<&'static abi::RadixAbiExternalTypeDescriptorV1> {
    validate_descriptor_graph(package)?;
    let types = unsafe { descriptor_slice(package.types, package.type_count) };
    types
        .iter()
        .find(|descriptor| abi_text(descriptor.local_id) == Ok(local_id))
        .ok_or_else(|| PluginError::invalid_input("unknown test external type"))
}

fn find_function(
    package: &'static abi::RadixPluginDescriptorV1,
    local_id: &str,
) -> PluginResult<&'static abi::RadixAbiScalarFunctionDescriptorV1> {
    let functions = unsafe { descriptor_slice(package.functions, package.function_count) };
    functions
        .iter()
        .find(|descriptor| abi_text(descriptor.local_id) == Ok(local_id))
        .ok_or_else(|| PluginError::invalid_input("unknown test scalar function"))
}

fn find_function_by_id(
    package: &'static abi::RadixPluginDescriptorV1,
    object_id: [u8; 16],
) -> PluginResult<&'static abi::RadixAbiScalarFunctionDescriptorV1> {
    let functions = unsafe { descriptor_slice(package.functions, package.function_count) };
    functions
        .iter()
        .find(|descriptor| descriptor.object_id == object_id)
        .ok_or_else(|| PluginError::invalid_input("unknown test scalar function identity"))
}

fn find_operator_by_id(
    package: &'static abi::RadixPluginDescriptorV1,
    object_id: [u8; 16],
) -> PluginResult<&'static abi::RadixAbiOperatorDescriptorV1> {
    let operators = unsafe { descriptor_slice(package.operators, package.operator_count) };
    operators
        .iter()
        .find(|descriptor| descriptor.object_id == object_id)
        .ok_or_else(|| PluginError::invalid_input("unknown test operator identity"))
}

fn find_operator_class(
    package: &'static abi::RadixPluginDescriptorV1,
    local_id: &str,
) -> PluginResult<&'static abi::RadixAbiOperatorClassDescriptorV1> {
    let classes =
        unsafe { descriptor_slice(package.operator_classes, package.operator_class_count) };
    classes
        .iter()
        .find(|descriptor| abi_text(descriptor.local_id) == Ok(local_id))
        .ok_or_else(|| PluginError::invalid_input("unknown test operator class"))
}

fn invoke_boolean_operator_by_id(
    package: &'static abi::RadixPluginDescriptorV1,
    object_id: [u8; 16],
    left: &TestValue,
    right: &TestValue,
) -> PluginResult<bool> {
    let operator = find_operator_by_id(package, object_id)?;
    let function = find_function_by_id(package, operator.function_id)?;
    let report = invoke_scalar(
        package,
        abi_text(function.local_id)?,
        &[left.clone(), right.clone()],
        TestCallOptions::default(),
    )?;
    if report.status != abi::RADIX_STATUS_OK
        || !report.finished
        || report.outputs.len() != 1
        || report.outputs[0].is_null
    {
        return Err(PluginError::domain(
            "operator-class strategy did not return one BOOLEAN result",
        ));
    }
    match report.outputs[0].bytes.as_slice() {
        [0] => Ok(false),
        [1] => Ok(true),
        _ => Err(PluginError::domain(
            "operator-class strategy returned a malformed BOOLEAN",
        )),
    }
}

fn find_planner_support(
    package: &'static abi::RadixPluginDescriptorV1,
    local_id: &str,
) -> PluginResult<&'static abi::RadixAbiPlannerSupportDescriptorV1> {
    let supports =
        unsafe { descriptor_slice(package.planner_support, package.planner_support_count) };
    supports
        .iter()
        .find(|descriptor| abi_text(descriptor.local_id) == Ok(local_id))
        .ok_or_else(|| PluginError::invalid_input("unknown test planner support"))
}

fn raw_value(value: &TestValue) -> abi::RadixAbiValueV1 {
    let mut inline_bytes = [0; 16];
    let fixed_builtin = value.type_ref.kind == abi::RADIX_TYPE_REF_BUILTIN
        && matches!(
            value.type_ref.builtin_tag,
            abi::RADIX_BUILTIN_INTEGER | abi::RADIX_BUILTIN_FLOAT | abi::RADIX_BUILTIN_BOOLEAN
        );
    if fixed_builtin && !value.is_null {
        inline_bytes[..value.bytes.len()].copy_from_slice(&value.bytes);
    }
    abi::RadixAbiValueV1 {
        type_ref: value.type_ref,
        flags: if value.is_null {
            abi::RADIX_VALUE_FLAG_NULL
        } else {
            0
        },
        reserved: 0,
        inline_bytes,
        borrowed_bytes: if fixed_builtin || value.is_null {
            abi::RadixAbiSliceV1::EMPTY
        } else {
            abi::RadixAbiSliceV1 {
                ptr: value.bytes.as_ptr(),
                len: value.bytes.len() as u32,
                reserved: 0,
            }
        },
    }
}

enum ColumnStorage {
    Aligned(Vec<u64>),
    Bytes(Vec<u8>),
}

impl ColumnStorage {
    fn bytes(&self) -> (*const u8, usize) {
        match self {
            Self::Aligned(words) => (words.as_ptr().cast(), words.len() * 8),
            Self::Bytes(bytes) => (bytes.as_ptr(), bytes.len()),
        }
    }
}

struct OwnedColumn {
    type_ref: abi::RadixAbiTypeRefV1,
    layout: u16,
    element_width: u16,
    alignment: u16,
    stride: u32,
    null_bitmap: Vec<u8>,
    storage: ColumnStorage,
    offsets: Vec<u32>,
}

impl OwnedColumn {
    fn new(
        package: &'static abi::RadixPluginDescriptorV1,
        type_ref: abi::RadixAbiTypeRefV1,
        values: &[&TestValue],
    ) -> PluginResult<Self> {
        let fixed_width = if type_ref.kind == abi::RADIX_TYPE_REF_BUILTIN {
            match type_ref.builtin_tag {
                abi::RADIX_BUILTIN_INTEGER | abi::RADIX_BUILTIN_FLOAT => Some(8_usize),
                abi::RADIX_BUILTIN_BOOLEAN => Some(1_usize),
                _ => None,
            }
        } else {
            let types = unsafe { descriptor_slice(package.types, package.type_count) };
            types
                .iter()
                .find(|descriptor| descriptor.object_id == type_ref.object_id)
                .and_then(|descriptor| {
                    (descriptor.storage_kind == abi::RADIX_EXTERNAL_STORAGE_FIXED)
                        .then_some(descriptor.fixed_bytes as usize)
                })
        };
        let mut null_bitmap = vec![0_u8; values.len().div_ceil(8)];
        for (index, value) in values.iter().enumerate() {
            if value.is_null {
                null_bitmap[index / 8] |= 1 << (index % 8);
            }
        }
        if null_bitmap.iter().all(|byte| *byte == 0) {
            null_bitmap.clear();
        }

        if let Some(width) = fixed_width {
            if width == 0 || width > u16::MAX as usize {
                return Err(PluginError::limit_exceeded(
                    "fixed test column width is outside ABI bounds",
                ));
            }
            if width == 1 {
                let data = values
                    .iter()
                    .map(|value| {
                        if value.is_null {
                            Ok(0)
                        } else if value.bytes.len() == 1 {
                            Ok(value.bytes[0])
                        } else {
                            Err(PluginError::invalid_input(
                                "fixed test value has wrong width",
                            ))
                        }
                    })
                    .collect::<PluginResult<Vec<_>>>()?;
                return Ok(Self {
                    type_ref,
                    layout: abi::RADIX_COLUMN_LAYOUT_FIXED,
                    element_width: 1,
                    alignment: 1,
                    stride: 1,
                    null_bitmap,
                    storage: ColumnStorage::Bytes(data),
                    offsets: Vec::new(),
                });
            }
            let stride = width.div_ceil(8) * 8;
            let total = stride
                .checked_mul(values.len())
                .ok_or_else(|| PluginError::limit_exceeded("test column size overflow"))?;
            let mut words = vec![0_u64; total / 8];
            // SAFETY: Vec<u64> owns `total` initialized bytes and gives the
            // alignment declared in the ABI column below.
            let bytes =
                unsafe { std::slice::from_raw_parts_mut(words.as_mut_ptr().cast::<u8>(), total) };
            for (row, value) in values.iter().enumerate() {
                if !value.is_null {
                    if value.bytes.len() != width {
                        return Err(PluginError::invalid_input(
                            "fixed test value has wrong width",
                        ));
                    }
                    let start = row * stride;
                    bytes[start..start + width].copy_from_slice(&value.bytes);
                }
            }
            return Ok(Self {
                type_ref,
                layout: abi::RADIX_COLUMN_LAYOUT_FIXED,
                element_width: width as u16,
                alignment: 8,
                stride: stride as u32,
                null_bitmap,
                storage: ColumnStorage::Aligned(words),
                offsets: Vec::new(),
            });
        }

        let mut data = Vec::new();
        let mut offsets = Vec::with_capacity(values.len() + 1);
        offsets.push(0);
        for value in values {
            if !value.is_null {
                data.extend_from_slice(&value.bytes);
            }
            offsets.push(
                u32::try_from(data.len())
                    .map_err(|_| PluginError::limit_exceeded("test column exceeds ABI bounds"))?,
            );
        }
        Ok(Self {
            type_ref,
            layout: abi::RADIX_COLUMN_LAYOUT_VARIABLE,
            element_width: 0,
            alignment: 1,
            stride: 0,
            null_bitmap,
            storage: ColumnStorage::Bytes(data),
            offsets,
        })
    }

    fn as_abi(&self, row_count: u32) -> abi::RadixAbiColumnViewV1 {
        let (data, data_len) = self.storage.bytes();
        abi::RadixAbiColumnViewV1 {
            header: abi::RadixAbiHeaderV1::new::<abi::RadixAbiColumnViewV1>(0),
            type_ref: self.type_ref,
            row_count,
            layout: self.layout,
            element_width: self.element_width,
            alignment: self.alignment,
            reserved_u16: 0,
            stride: self.stride,
            null_bitmap: abi::RadixAbiSliceV1 {
                ptr: if self.null_bitmap.is_empty() {
                    std::ptr::null()
                } else {
                    self.null_bitmap.as_ptr()
                },
                len: self.null_bitmap.len() as u32,
                reserved: 0,
            },
            data: abi::RadixAbiSliceV1 {
                ptr: if data_len == 0 {
                    std::ptr::null()
                } else {
                    data
                },
                len: data_len as u32,
                reserved: 0,
            },
            offsets: abi::RadixAbiU32SliceV1 {
                ptr: if self.offsets.is_empty() {
                    std::ptr::null()
                } else {
                    self.offsets.as_ptr()
                },
                len: self.offsets.len() as u32,
                reserved: 0,
            },
        }
    }
}

struct TestHostState {
    options: TestCallOptions,
    staged: Vec<TestOutput>,
    committed: Vec<TestOutput>,
    diagnostics: Vec<TestDiagnostic>,
    work_charged: u32,
    finished: bool,
}

impl TestHostState {
    fn new(options: TestCallOptions) -> Self {
        Self {
            options,
            staged: Vec::new(),
            committed: Vec::new(),
            diagnostics: Vec::new(),
            work_charged: 0,
            finished: false,
        }
    }

    fn diagnostic_sink(&mut self) -> abi::RadixAbiDiagnosticSinkV1 {
        let handle = self as *mut Self as usize as u64;
        abi::RadixAbiDiagnosticSinkV1 {
            header: abi::RadixAbiHeaderV1::new::<abi::RadixAbiDiagnosticSinkV1>(0),
            handle,
            max_detail_bytes: abi::RADIX_MAX_DIAGNOSTIC_BYTES,
            reserved: 0,
            write: Some(test_diagnostic),
        }
    }

    fn call_context(
        &mut self,
        diagnostic_sink: &abi::RadixAbiDiagnosticSinkV1,
        declared_output_bytes: u32,
    ) -> abi::RadixAbiCallContextV1 {
        let handle = self as *mut Self as usize as u64;
        abi::RadixAbiCallContextV1 {
            header: abi::RadixAbiHeaderV1::new::<abi::RadixAbiCallContextV1>(0),
            handle,
            deadline_unix_ns: if self.options.deadline_expired {
                1
            } else {
                u64::MAX
            },
            max_output_bytes: self.options.max_output_bytes.min(declared_output_bytes),
            max_work_units: self.options.max_work_units,
            check_cancelled: Some(test_cancelled),
            charge_work: Some(test_charge_work),
            diagnostics: diagnostic_sink,
        }
    }

    fn result_builder(
        &mut self,
        declared_output_bytes: u32,
        max_items: u32,
    ) -> abi::RadixAbiResultBuilderV1 {
        let handle = self as *mut Self as usize as u64;
        abi::RadixAbiResultBuilderV1 {
            header: abi::RadixAbiHeaderV1::new::<abi::RadixAbiResultBuilderV1>(0),
            handle,
            max_bytes: self.options.max_output_bytes.min(declared_output_bytes),
            max_items,
            write: Some(test_write),
            finish: Some(test_finish),
        }
    }

    fn report(self, status: abi::RadixAbiStatusV1) -> TestCallReport {
        TestCallReport {
            status,
            outputs: self.committed,
            diagnostics: self.diagnostics,
            work_charged: self.work_charged,
            finished: self.finished,
        }
    }
}

unsafe fn state(handle: u64) -> &'static mut TestHostState {
    // SAFETY: every local-host ABI object carries the live TestHostState
    // address and callbacks are synchronous.
    unsafe { &mut *(handle as usize as *mut TestHostState) }
}

unsafe extern "C" fn test_cancelled(handle: u64) -> abi::RadixAbiStatusV1 {
    let options = unsafe { state(handle) }.options;
    if options.cancelled || options.deadline_expired {
        abi::RADIX_STATUS_CANCELLED
    } else {
        abi::RADIX_STATUS_OK
    }
}

unsafe extern "C" fn test_charge_work(handle: u64, units: u32) -> abi::RadixAbiStatusV1 {
    let state = unsafe { state(handle) };
    let Some(total) = state.work_charged.checked_add(units) else {
        return abi::RADIX_STATUS_LIMIT_EXCEEDED;
    };
    if total > state.options.max_work_units {
        abi::RADIX_STATUS_LIMIT_EXCEEDED
    } else {
        state.work_charged = total;
        abi::RADIX_STATUS_OK
    }
}

unsafe extern "C" fn test_write(
    handle: u64,
    flags: u32,
    reserved: u32,
    bytes: abi::RadixAbiSliceV1,
) -> abi::RadixAbiStatusV1 {
    let state = unsafe { state(handle) };
    if abi::validate_result_item(flags, reserved, bytes, state.options.max_output_bytes).is_err() {
        return abi::RADIX_STATUS_CONTRACT_VIOLATION;
    }
    let bytes = if bytes.len == 0 {
        Vec::new()
    } else {
        // SAFETY: plugin result bytes remain live for this synchronous copy.
        unsafe { std::slice::from_raw_parts(bytes.ptr, bytes.len as usize) }.to_vec()
    };
    state.staged.push(TestOutput {
        is_null: flags & abi::RADIX_RESULT_ITEM_FLAG_NULL != 0,
        bytes,
    });
    abi::RADIX_STATUS_OK
}

unsafe extern "C" fn test_finish(handle: u64) -> abi::RadixAbiStatusV1 {
    let state = unsafe { state(handle) };
    if state.finished {
        return abi::RADIX_STATUS_CONTRACT_VIOLATION;
    }
    state.finished = true;
    state.committed = std::mem::take(&mut state.staged);
    abi::RADIX_STATUS_OK
}

unsafe extern "C" fn test_diagnostic(
    handle: u64,
    diagnostic: *const abi::RadixAbiDiagnosticV1,
) -> abi::RadixAbiStatusV1 {
    let Some(diagnostic) = (unsafe { diagnostic.as_ref() }) else {
        return abi::RADIX_STATUS_INVALID_ARGUMENT;
    };
    if abi::validate_diagnostic(diagnostic).is_err() {
        return abi::RADIX_STATUS_CONTRACT_VIOLATION;
    }
    let copy = |value: abi::RadixAbiStringV1| {
        if value.len == 0 {
            String::new()
        } else {
            // SAFETY: diagnostic slices remain live for this synchronous copy.
            String::from_utf8_lossy(unsafe {
                std::slice::from_raw_parts(value.ptr, value.len as usize)
            })
            .into_owned()
        }
    };
    unsafe { state(handle) }.diagnostics.push(TestDiagnostic {
        category: diagnostic.category,
        status: diagnostic.status,
        detail: copy(diagnostic.detail),
        field: copy(diagnostic.field),
    });
    abi::RADIX_STATUS_OK
}