runmat_runtime/builtins/io/data/

mod.rs

//! Cloud-ready dataset persistence builtins (`data.*`).

use std::collections::BTreeMap;
use std::collections::HashMap;
use std::path::PathBuf;

use runmat_builtins::{
    BuiltinCompletionPolicy, BuiltinDescriptor, BuiltinErrorDescriptor,
    BuiltinIntegerAuditDescriptor, BuiltinIntegerAuditKind, BuiltinIntegerBackendRule,
    BuiltinIntegerCapabilityDescriptor, BuiltinIntegerComputationDomain,
    BuiltinIntegerInputAvailability, BuiltinIntegerInputCapability, BuiltinIntegerOutputClassRule,
    BuiltinIntegerOverflowRule, BuiltinIntegerOverloadKind, BuiltinIntegerScalarDoubleRule,
    BuiltinOutputMode, BuiltinParamArity, BuiltinParamDescriptor, BuiltinParamType,
    BuiltinSignatureDescriptor,
};
use runmat_filesystem::data_contract::{DataChunkDescriptor, DataChunkUploadRequest};
use runmat_macros::runtime_builtin;
use runmat_value::{IntValue, NumericScalar, ObjectInstance, StructValue, Tensor, Value};

use crate::builtins::common::json::int_value_to_json;
use crate::builtins::common::spec::{
    BroadcastSemantics, BuiltinFusionSpec, BuiltinGpuSpec, ConstantStrategy, GpuOpKind,
    ReductionNaN, ResidencyPolicy, ShapeRequirements,
};
use crate::builtins::common::tensor as tensor_utils;
use crate::data::{
    array_object, data_error, dataset_object, dataset_root, ensure_manifest_sequence,
    get_object_prop, manifest_path, manifest_version_token, now_rfc3339, parse_schema,
    parse_string, read_array_payload_async, read_manifest_async, remove_tx, sha256_hex, start_tx,
    transaction_object, validate_chunk_shape, with_tx, with_tx_mut, write_array_payload_async,
    write_manifest_async, DataArrayMeta, DataArrayPayload, DataChunkIndex, DataChunkIndexEntry,
    DataManifest, PendingCreateArray, PendingFill, PendingResize, PendingWrite, TxnStatus,
};
use crate::{make_cell, BuiltinResult};

#[runmat_macros::register_gpu_spec(builtin_path = "crate::builtins::io::data")]
pub const GPU_SPEC: BuiltinGpuSpec = BuiltinGpuSpec {
    name: "data.*",
    op_kind: GpuOpKind::Custom("io-data"),
    supported_precisions: &[],
    broadcast: BroadcastSemantics::None,
    provider_hooks: &[],
    constant_strategy: ConstantStrategy::InlineLiteral,
    residency: ResidencyPolicy::GatherImmediately,
    nan_mode: ReductionNaN::Include,
    two_pass_threshold: None,
    workgroup_size: None,
    accepts_nan_mode: false,
    notes: "Dataset operations are host I/O and metadata orchestration.",
};

#[runmat_macros::register_fusion_spec(builtin_path = "crate::builtins::io::data")]
pub const FUSION_SPEC: BuiltinFusionSpec = BuiltinFusionSpec {
    name: "data.*",
    shape: ShapeRequirements::Any,
    constant_strategy: ConstantStrategy::InlineLiteral,
    elementwise: None,
    reduction: None,
    emits_nan: false,
    notes: "Data builtins are side-effecting and not fusible.",
};

const DATA_ERROR_INVALID_ARGUMENT: BuiltinErrorDescriptor = BuiltinErrorDescriptor {
    code: "RM.DATA.INVALID_ARGUMENT",
    identifier: Some("RunMat:data:InvalidArgument"),
    when: "Arguments, receiver object, or option grammar are invalid for the requested data API.",
    message: "data: invalid argument",
};

const DATA_ERROR_NOT_FOUND: BuiltinErrorDescriptor = BuiltinErrorDescriptor {
    code: "RM.DATA.NOT_FOUND",
    identifier: Some("RunMat:data:NotFound"),
    when: "Referenced dataset/array/transaction object is missing or cannot be resolved.",
    message: "data: requested object not found",
};

const DATA_ERROR_INTERNAL: BuiltinErrorDescriptor = BuiltinErrorDescriptor {
    code: "RM.DATA.INTERNAL",
    identifier: Some("RunMat:data:Internal"),
    when: "Filesystem/manifest/chunk processing fails unexpectedly during data API execution.",
    message: "data: internal operation failed",
};

const DATA_DESCRIPTOR_ERRORS: [BuiltinErrorDescriptor; 3] = [
    DATA_ERROR_INVALID_ARGUMENT,
    DATA_ERROR_NOT_FOUND,
    DATA_ERROR_INTERNAL,
];

const OUT_DATASET: [BuiltinParamDescriptor; 1] = [BuiltinParamDescriptor {
    name: "ds",
    ty: BuiltinParamType::Any,
    arity: BuiltinParamArity::Required,
    default: None,
    description: "Dataset handle object.",
}];

const OUT_ARRAY: [BuiltinParamDescriptor; 1] = [BuiltinParamDescriptor {
    name: "arr",
    ty: BuiltinParamType::Any,
    arity: BuiltinParamArity::Required,
    default: None,
    description: "DataArray handle object.",
}];

const OUT_TX: [BuiltinParamDescriptor; 1] = [BuiltinParamDescriptor {
    name: "tx",
    ty: BuiltinParamType::Any,
    arity: BuiltinParamArity::Required,
    default: None,
    description: "DataTransaction handle object.",
}];

const OUT_BOOL: [BuiltinParamDescriptor; 1] = [BuiltinParamDescriptor {
    name: "ok",
    ty: BuiltinParamType::LogicalArray,
    arity: BuiltinParamArity::Required,
    default: None,
    description: "Logical success/result flag.",
}];

const OUT_STRING: [BuiltinParamDescriptor; 1] = [BuiltinParamDescriptor {
    name: "s",
    ty: BuiltinParamType::StringScalar,
    arity: BuiltinParamArity::Required,
    default: None,
    description: "String scalar result.",
}];

const OUT_STRUCT: [BuiltinParamDescriptor; 1] = [BuiltinParamDescriptor {
    name: "S",
    ty: BuiltinParamType::Any,
    arity: BuiltinParamArity::Required,
    default: None,
    description: "Struct result.",
}];

const OUT_CELL: [BuiltinParamDescriptor; 1] = [BuiltinParamDescriptor {
    name: "C",
    ty: BuiltinParamType::Any,
    arity: BuiltinParamArity::Required,
    default: None,
    description: "Cell-array result.",
}];

const OUT_VALUE: [BuiltinParamDescriptor; 1] = [BuiltinParamDescriptor {
    name: "value",
    ty: BuiltinParamType::Any,
    arity: BuiltinParamArity::Required,
    default: None,
    description: "Value result.",
}];

const OUT_TENSOR: [BuiltinParamDescriptor; 1] = [BuiltinParamDescriptor {
    name: "X",
    ty: BuiltinParamType::NumericArray,
    arity: BuiltinParamArity::Required,
    default: None,
    description: "Numeric tensor result.",
}];

const IN_PATH_SCHEMA: [BuiltinParamDescriptor; 2] = [
    BuiltinParamDescriptor {
        name: "path",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Dataset path (.data).",
    },
    BuiltinParamDescriptor {
        name: "schema",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Dataset schema struct.",
    },
];

const IN_PATH: [BuiltinParamDescriptor; 1] = [BuiltinParamDescriptor {
    name: "path",
    ty: BuiltinParamType::StringScalar,
    arity: BuiltinParamArity::Required,
    default: None,
    description: "Dataset path (.data).",
}];

const IN_FROM_TO: [BuiltinParamDescriptor; 2] = [
    BuiltinParamDescriptor {
        name: "fromPath",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Source dataset path.",
    },
    BuiltinParamDescriptor {
        name: "toPath",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Destination dataset path.",
    },
];

const IN_PATH_FORMAT_SOURCE: [BuiltinParamDescriptor; 3] = [
    BuiltinParamDescriptor {
        name: "path",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Dataset path (.data).",
    },
    BuiltinParamDescriptor {
        name: "format",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Format token (currently 'data').",
    },
    BuiltinParamDescriptor {
        name: "sourcePath",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Source dataset path.",
    },
];

const IN_PATH_FORMAT_TARGET: [BuiltinParamDescriptor; 3] = [
    BuiltinParamDescriptor {
        name: "path",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Dataset path (.data).",
    },
    BuiltinParamDescriptor {
        name: "format",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Format token (currently 'data').",
    },
    BuiltinParamDescriptor {
        name: "targetPath",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Target dataset path.",
    },
];

const IN_PREFIX: [BuiltinParamDescriptor; 1] = [BuiltinParamDescriptor {
    name: "prefix",
    ty: BuiltinParamType::StringScalar,
    arity: BuiltinParamArity::Required,
    default: None,
    description: "Filesystem prefix to scan for datasets.",
}];

const IN_BASE: [BuiltinParamDescriptor; 1] = [BuiltinParamDescriptor {
    name: "obj",
    ty: BuiltinParamType::Any,
    arity: BuiltinParamArity::Required,
    default: None,
    description: "Receiver object.",
}];

const IN_BASE_NAME: [BuiltinParamDescriptor; 2] = [
    BuiltinParamDescriptor {
        name: "obj",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Receiver object.",
    },
    BuiltinParamDescriptor {
        name: "name",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Name key/array identifier.",
    },
];

const IN_BASE_KEY_DEFAULT: [BuiltinParamDescriptor; 3] = [
    BuiltinParamDescriptor {
        name: "obj",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Receiver object.",
    },
    BuiltinParamDescriptor {
        name: "key",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Attribute key.",
    },
    BuiltinParamDescriptor {
        name: "defaultValue",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Optional,
        default: Some("0"),
        description: "Fallback value when key is absent.",
    },
];

const IN_BASE_KEY_VALUE: [BuiltinParamDescriptor; 3] = [
    BuiltinParamDescriptor {
        name: "obj",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Receiver object.",
    },
    BuiltinParamDescriptor {
        name: "key",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Attribute key.",
    },
    BuiltinParamDescriptor {
        name: "value",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Attribute value.",
    },
];

const IN_BASE_ATTRS: [BuiltinParamDescriptor; 2] = [
    BuiltinParamDescriptor {
        name: "obj",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Receiver object.",
    },
    BuiltinParamDescriptor {
        name: "attrs",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Struct of attribute updates.",
    },
];

const IN_BASE_LABEL_REST: [BuiltinParamDescriptor; 3] = [
    BuiltinParamDescriptor {
        name: "obj",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Receiver object.",
    },
    BuiltinParamDescriptor {
        name: "label",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Snapshot label.",
    },
    BuiltinParamDescriptor {
        name: "options",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Variadic,
        default: None,
        description: "Reserved name/value options.",
    },
];

const IN_BASE_REST: [BuiltinParamDescriptor; 2] = [
    BuiltinParamDescriptor {
        name: "obj",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Receiver object.",
    },
    BuiltinParamDescriptor {
        name: "options",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Variadic,
        default: None,
        description: "Reserved name/value options.",
    },
];

const IN_BASE_SLICE_OPTIONAL: [BuiltinParamDescriptor; 2] = [
    BuiltinParamDescriptor {
        name: "obj",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "DataArray receiver object.",
    },
    BuiltinParamDescriptor {
        name: "sliceSpec",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Optional,
        default: None,
        description: "Optional slice specification.",
    },
];

const IN_BASE_VALUES: [BuiltinParamDescriptor; 2] = [
    BuiltinParamDescriptor {
        name: "obj",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "DataArray receiver object.",
    },
    BuiltinParamDescriptor {
        name: "values",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Full-array values payload.",
    },
];

const IN_BASE_SLICE_VALUES: [BuiltinParamDescriptor; 3] = [
    BuiltinParamDescriptor {
        name: "obj",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "DataArray receiver object.",
    },
    BuiltinParamDescriptor {
        name: "sliceSpec",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Slice specification.",
    },
    BuiltinParamDescriptor {
        name: "values",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Slice values payload.",
    },
];

const IN_BASE_NEW_SHAPE_REST: [BuiltinParamDescriptor; 3] = [
    BuiltinParamDescriptor {
        name: "obj",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "DataArray receiver object.",
    },
    BuiltinParamDescriptor {
        name: "newShape",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "New array shape.",
    },
    BuiltinParamDescriptor {
        name: "options",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Variadic,
        default: None,
        description: "Reserved name/value options.",
    },
];

const IN_BASE_VALUE_REST: [BuiltinParamDescriptor; 3] = [
    BuiltinParamDescriptor {
        name: "obj",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "DataArray receiver object.",
    },
    BuiltinParamDescriptor {
        name: "value",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Fill value.",
    },
    BuiltinParamDescriptor {
        name: "options",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Variadic,
        default: None,
        description: "Reserved name/value options.",
    },
];

const IN_TX_WRITE: [BuiltinParamDescriptor; 5] = [
    BuiltinParamDescriptor {
        name: "tx",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "DataTransaction receiver.",
    },
    BuiltinParamDescriptor {
        name: "arrayName",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Target array name.",
    },
    BuiltinParamDescriptor {
        name: "sliceSpec",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Slice specification.",
    },
    BuiltinParamDescriptor {
        name: "values",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Values payload.",
    },
    BuiltinParamDescriptor {
        name: "options",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Variadic,
        default: None,
        description: "Reserved name/value options.",
    },
];

const IN_TX_ARRAY_SHAPE_REST: [BuiltinParamDescriptor; 4] = [
    BuiltinParamDescriptor {
        name: "tx",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "DataTransaction receiver.",
    },
    BuiltinParamDescriptor {
        name: "arrayName",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Target array name.",
    },
    BuiltinParamDescriptor {
        name: "newShape",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "New shape vector.",
    },
    BuiltinParamDescriptor {
        name: "options",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Variadic,
        default: None,
        description: "Reserved name/value options.",
    },
];

const IN_TX_ARRAY_VALUE_SLICE_OPT: [BuiltinParamDescriptor; 4] = [
    BuiltinParamDescriptor {
        name: "tx",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "DataTransaction receiver.",
    },
    BuiltinParamDescriptor {
        name: "arrayName",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Target array name.",
    },
    BuiltinParamDescriptor {
        name: "value",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Fill value.",
    },
    BuiltinParamDescriptor {
        name: "sliceSpec",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Optional,
        default: None,
        description: "Optional slice specification.",
    },
];

const IN_TX_ARRAY_NAME: [BuiltinParamDescriptor; 2] = [
    BuiltinParamDescriptor {
        name: "tx",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "DataTransaction receiver.",
    },
    BuiltinParamDescriptor {
        name: "arrayName",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Target array name.",
    },
];

const IN_TX_ARRAY_META: [BuiltinParamDescriptor; 3] = [
    BuiltinParamDescriptor {
        name: "tx",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "DataTransaction receiver.",
    },
    BuiltinParamDescriptor {
        name: "arrayName",
        ty: BuiltinParamType::StringScalar,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "New array name.",
    },
    BuiltinParamDescriptor {
        name: "meta",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "Array metadata struct.",
    },
];

const IN_TX_COMMIT_OPTIONS: [BuiltinParamDescriptor; 2] = [
    BuiltinParamDescriptor {
        name: "tx",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Required,
        default: None,
        description: "DataTransaction receiver.",
    },
    BuiltinParamDescriptor {
        name: "options",
        ty: BuiltinParamType::Any,
        arity: BuiltinParamArity::Optional,
        default: None,
        description: "Optional struct containing an if_manifest version token.",
    },
];

macro_rules! one_sig_descriptor {
    ($desc:ident, $sigs:ident, $label:expr, $inputs:expr, $outputs:expr) => {
        const $sigs: [BuiltinSignatureDescriptor; 1] = [BuiltinSignatureDescriptor {
            label: $label,
            inputs: $inputs,
            outputs: $outputs,
        }];
        pub const $desc: BuiltinDescriptor = BuiltinDescriptor {
            signatures: &$sigs,
            output_mode: BuiltinOutputMode::Fixed,
            completion_policy: BuiltinCompletionPolicy::Public,
            errors: &DATA_DESCRIPTOR_ERRORS,
        };
    };
}

one_sig_descriptor!(
    DATA_CREATE_DESCRIPTOR,
    DATA_CREATE_SIGS,
    "ds = data.create(path, schema)",
    &IN_PATH_SCHEMA,
    &OUT_DATASET
);
one_sig_descriptor!(
    DATA_OPEN_DESCRIPTOR,
    DATA_OPEN_SIGS,
    "ds = data.open(path)",
    &IN_PATH,
    &OUT_DATASET
);
one_sig_descriptor!(
    DATA_EXISTS_DESCRIPTOR,
    DATA_EXISTS_SIGS,
    "tf = data.exists(path)",
    &IN_PATH,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATA_DELETE_DESCRIPTOR,
    DATA_DELETE_SIGS,
    "tf = data.delete(path)",
    &IN_PATH,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATA_COPY_DESCRIPTOR,
    DATA_COPY_SIGS,
    "tf = data.copy(fromPath, toPath)",
    &IN_FROM_TO,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATA_MOVE_DESCRIPTOR,
    DATA_MOVE_SIGS,
    "tf = data.move(fromPath, toPath)",
    &IN_FROM_TO,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATA_IMPORT_DESCRIPTOR,
    DATA_IMPORT_SIGS,
    "ds = data.import(path, format, sourcePath)",
    &IN_PATH_FORMAT_SOURCE,
    &OUT_DATASET
);
one_sig_descriptor!(
    DATA_EXPORT_DESCRIPTOR,
    DATA_EXPORT_SIGS,
    "tf = data.export(path, format, targetPath)",
    &IN_PATH_FORMAT_TARGET,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATA_LIST_DESCRIPTOR,
    DATA_LIST_SIGS,
    "C = data.list(prefix)",
    &IN_PREFIX,
    &OUT_CELL
);
one_sig_descriptor!(
    DATA_INSPECT_DESCRIPTOR,
    DATA_INSPECT_SIGS,
    "S = data.inspect(path)",
    &IN_PATH,
    &OUT_STRUCT
);

const DATA_CREATE_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 2] = [
    BuiltinIntegerInputCapability {
        name: "schema.arrays.*.shape",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "Array shapes accept scalar or vector dimensions in every integer class, or exactly integral floating values, within nonnegative platform and checked total-element bounds.",
    },
    BuiltinIntegerInputCapability {
        name: "schema.arrays.*.chunk",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "Optional chunk shapes accept scalar or vector dimensions in every integer class, or exactly integral floating values; dimensions must be positive, platform-bounded, and rank-matched to the array shape.",
    },
];

pub const DATA_CREATE_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 1] =
    [BuiltinIntegerCapabilityDescriptor {
        form: "ds = data.create(path, struct_with_integer_shape_and_chunk)",
        inputs: &DATA_CREATE_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::FunctionSpecific,
        overflow: BuiltinIntegerOverflowRule::Error,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::StructuralParameter,
        notes: "Schema dimensions are parsed directly from authoritative typed storage; the host/filesystem constructor allocates zero-filled storage in each array's declared dtype and returns a Dataset object.",
    }];

const DATA_NAMESPACE_LIFECYCLE_INTEGER_AUDIT: BuiltinIntegerAuditDescriptor =
    BuiltinIntegerAuditDescriptor {
        kind: BuiltinIntegerAuditKind::NotApplicable,
        canonical_builtin: None,
        notes: "Dataset copy, delete, existence, export, import, inspection, listing, move, and open operations accept only textual paths/formats and return objects, logicals, strings, or host metadata; opaque movement of stored integer payload bytes is owned by the DataArray persistence contract rather than an integer call form.",
    };

one_sig_descriptor!(
    DATASET_PATH_DESCRIPTOR,
    DATASET_PATH_SIGS,
    "path = Dataset.path(ds)",
    &IN_BASE,
    &OUT_STRING
);
one_sig_descriptor!(
    DATASET_ID_DESCRIPTOR,
    DATASET_ID_SIGS,
    "id = Dataset.id(ds)",
    &IN_BASE,
    &OUT_STRING
);
one_sig_descriptor!(
    DATASET_VERSION_DESCRIPTOR,
    DATASET_VERSION_SIGS,
    "version = Dataset.version(ds)",
    &IN_BASE,
    &OUT_STRING
);
one_sig_descriptor!(
    DATASET_ARRAYS_DESCRIPTOR,
    DATASET_ARRAYS_SIGS,
    "C = Dataset.arrays(ds)",
    &IN_BASE,
    &OUT_CELL
);
one_sig_descriptor!(
    DATASET_HAS_ARRAY_DESCRIPTOR,
    DATASET_HAS_ARRAY_SIGS,
    "tf = Dataset.has_array(ds, name)",
    &IN_BASE_NAME,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATASET_ARRAY_DESCRIPTOR,
    DATASET_ARRAY_SIGS,
    "arr = Dataset.array(ds, name)",
    &IN_BASE_NAME,
    &OUT_ARRAY
);
one_sig_descriptor!(
    DATASET_ATTRS_DESCRIPTOR,
    DATASET_ATTRS_SIGS,
    "S = Dataset.attrs(ds)",
    &IN_BASE,
    &OUT_STRUCT
);
one_sig_descriptor!(
    DATASET_GET_ATTR_DESCRIPTOR,
    DATASET_GET_ATTR_SIGS,
    "value = Dataset.get_attr(ds, key, defaultValue)",
    &IN_BASE_KEY_DEFAULT,
    &OUT_VALUE
);
one_sig_descriptor!(
    DATASET_SET_ATTR_DESCRIPTOR,
    DATASET_SET_ATTR_SIGS,
    "tf = Dataset.set_attr(ds, key, value)",
    &IN_BASE_KEY_VALUE,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATASET_SET_ATTRS_DESCRIPTOR,
    DATASET_SET_ATTRS_SIGS,
    "tf = Dataset.set_attrs(ds, attrs)",
    &IN_BASE_ATTRS,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATASET_BEGIN_DESCRIPTOR,
    DATASET_BEGIN_SIGS,
    "tx = Dataset.begin(ds, Name, Value, ...)",
    &IN_BASE_REST,
    &OUT_TX
);
one_sig_descriptor!(
    DATASET_SNAPSHOT_DESCRIPTOR,
    DATASET_SNAPSHOT_SIGS,
    "snapshotPath = Dataset.snapshot(ds, label, Name, Value, ...)",
    &IN_BASE_LABEL_REST,
    &OUT_STRING
);
one_sig_descriptor!(
    DATASET_REFRESH_DESCRIPTOR,
    DATASET_REFRESH_SIGS,
    "ds = Dataset.refresh(ds)",
    &IN_BASE,
    &OUT_DATASET
);

const DATASET_NONATTRIBUTE_INTEGER_AUDIT: BuiltinIntegerAuditDescriptor =
    BuiltinIntegerAuditDescriptor {
        kind: BuiltinIntegerAuditKind::NotApplicable,
        canonical_builtin: None,
        notes: "Dataset array lookup/listing, transaction construction, identity/path/version access, membership tests, refresh, and snapshot operations accept a Dataset receiver plus strings or reserved options; they do not accept integer values or controls, return integer-class values, perform integer arithmetic, or expose an integer backend surface.",
    };

const DATASET_ATTRS_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 1] =
    [BuiltinIntegerInputCapability {
        name: "stored_attribute_values",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::NotApplicable,
        notes: "Persisted scalar attributes may originate from every integer class and retain their exact mathematical value in the JSON manifest.",
    }];
pub const DATASET_ATTRS_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 1] =
    [BuiltinIntegerCapabilityDescriptor {
        form: "S = Dataset.attrs(ds_with_integer_attributes)",
        inputs: &DATASET_ATTRS_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::FunctionSpecific,
        overflow: BuiltinIntegerOverflowRule::NotApplicable,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::Multiple,
        notes: "Each JSON integer is decoded exactly into a canonical scalar: int64 when its value is representable there, otherwise uint64; the original narrow storage class is not encoded by JSON.",
    }];

const DATASET_GET_ATTR_STORED_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 1] =
    [BuiltinIntegerInputCapability {
        name: "stored_attribute",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::NotApplicable,
        notes: "A persisted scalar attribute may originate from every integer class and retains its exact mathematical value in the JSON manifest.",
    }];
const DATASET_GET_ATTR_DEFAULT_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 1] =
    [BuiltinIntegerInputCapability {
        name: "defaultValue",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "An explicit scalar default accepts every integer class and is returned unchanged when the requested key is absent.",
    }];
pub const DATASET_GET_ATTR_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 2] = [
    BuiltinIntegerCapabilityDescriptor {
        form: "value = Dataset.get_attr(ds_with_integer_attribute, key)",
        inputs: &DATASET_GET_ATTR_STORED_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::FunctionSpecific,
        overflow: BuiltinIntegerOverflowRule::NotApplicable,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::ScalarOnly,
        notes: "A stored JSON integer is decoded exactly as int64 when representable there and otherwise as uint64; JSON does not encode the original narrow integer class.",
    },
    BuiltinIntegerCapabilityDescriptor {
        form: "value = Dataset.get_attr(ds, missing_key, integer_defaultValue)",
        inputs: &DATASET_GET_ATTR_DEFAULT_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::PreserveInput,
        overflow: BuiltinIntegerOverflowRule::NotApplicable,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::ScalarOnly,
        notes: "A missing key returns the supplied default Value directly, preserving its exact integer class and value without serialization or floating conversion.",
    },
];

const DATASET_SET_ATTR_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 1] =
    [BuiltinIntegerInputCapability {
        name: "value",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "A scalar attribute value accepts every integer class and is serialized as an exact JSON integer, including full-width int64 and uint64 values.",
    }];
pub const DATASET_SET_ATTR_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 1] =
    [BuiltinIntegerCapabilityDescriptor {
        form: "tf = Dataset.set_attr(ds, key, integer_value)",
        inputs: &DATASET_SET_ATTR_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::Logical,
        overflow: BuiltinIntegerOverflowRule::NotApplicable,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::ScalarOnly,
        notes: "The scalar is emitted directly as an exact manifest JSON number without an f64 intermediary; the operation returns logical success.",
    }];

const DATASET_SET_ATTRS_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 1] =
    [BuiltinIntegerInputCapability {
        name: "attribute_values",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "Scalar values in the attribute struct accept every integer class and are serialized as exact JSON integers, including full-width int64 and uint64 values.",
    }];
pub const DATASET_SET_ATTRS_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 1] =
    [BuiltinIntegerCapabilityDescriptor {
        form: "tf = Dataset.set_attrs(ds, struct_with_integer_values)",
        inputs: &DATASET_SET_ATTRS_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::Logical,
        overflow: BuiltinIntegerOverflowRule::NotApplicable,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::ScalarOnly,
        notes: "Each scalar integer field is emitted directly as an exact manifest JSON number without an f64 intermediary; the operation returns logical success.",
    }];

one_sig_descriptor!(
    DATAARRAY_NAME_DESCRIPTOR,
    DATAARRAY_NAME_SIGS,
    "name = DataArray.name(arr)",
    &IN_BASE,
    &OUT_STRING
);
one_sig_descriptor!(
    DATAARRAY_DTYPE_DESCRIPTOR,
    DATAARRAY_DTYPE_SIGS,
    "dtype = DataArray.dtype(arr)",
    &IN_BASE,
    &OUT_STRING
);
one_sig_descriptor!(
    DATAARRAY_SHAPE_DESCRIPTOR,
    DATAARRAY_SHAPE_SIGS,
    "shape = DataArray.shape(arr)",
    &IN_BASE,
    &OUT_TENSOR
);
one_sig_descriptor!(
    DATAARRAY_RANK_DESCRIPTOR,
    DATAARRAY_RANK_SIGS,
    "rank = DataArray.rank(arr)",
    &IN_BASE,
    &OUT_VALUE
);
one_sig_descriptor!(
    DATAARRAY_CHUNK_SHAPE_DESCRIPTOR,
    DATAARRAY_CHUNK_SHAPE_SIGS,
    "chunkShape = DataArray.chunk_shape(arr)",
    &IN_BASE,
    &OUT_TENSOR
);
one_sig_descriptor!(
    DATAARRAY_CODEC_DESCRIPTOR,
    DATAARRAY_CODEC_SIGS,
    "codec = DataArray.codec(arr)",
    &IN_BASE,
    &OUT_STRING
);
one_sig_descriptor!(
    DATAARRAY_READ_DESCRIPTOR,
    DATAARRAY_READ_SIGS,
    "X = DataArray.read(arr, sliceSpec)",
    &IN_BASE_SLICE_OPTIONAL,
    &OUT_TENSOR
);
const DATAARRAY_WRITE_SIGS: [BuiltinSignatureDescriptor; 2] = [
    BuiltinSignatureDescriptor {
        label: "tf = DataArray.write(arr, values)",
        inputs: &IN_BASE_VALUES,
        outputs: &OUT_BOOL,
    },
    BuiltinSignatureDescriptor {
        label: "tf = DataArray.write(arr, sliceSpec, values)",
        inputs: &IN_BASE_SLICE_VALUES,
        outputs: &OUT_BOOL,
    },
];
pub const DATAARRAY_WRITE_DESCRIPTOR: BuiltinDescriptor = BuiltinDescriptor {
    signatures: &DATAARRAY_WRITE_SIGS,
    output_mode: BuiltinOutputMode::Fixed,
    completion_policy: BuiltinCompletionPolicy::Public,
    errors: &DATA_DESCRIPTOR_ERRORS,
};
one_sig_descriptor!(
    DATAARRAY_RESIZE_DESCRIPTOR,
    DATAARRAY_RESIZE_SIGS,
    "tf = DataArray.resize(arr, newShape, Name, Value, ...)",
    &IN_BASE_NEW_SHAPE_REST,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATAARRAY_FILL_DESCRIPTOR,
    DATAARRAY_FILL_SIGS,
    "tf = DataArray.fill(arr, value, Name, Value, ...)",
    &IN_BASE_VALUE_REST,
    &OUT_BOOL
);

const DATAARRAY_METADATA_INTEGER_AUDIT: BuiltinIntegerAuditDescriptor =
    BuiltinIntegerAuditDescriptor {
        kind: BuiltinIntegerAuditKind::NotApplicable,
        canonical_builtin: None,
        notes: "DataArray name, dtype, shape, rank, chunk-shape, and codec accessors accept only a DataArray receiver object; integer storage may be described by the returned metadata but is not an integer argument, control, class-preserving value output, arithmetic form, or backend surface.",
    };

const DATAARRAY_READ_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 2] = [
    BuiltinIntegerInputCapability {
        name: "stored_array",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::NotApplicable,
        notes: "A DataArray whose declared dtype is any built-in integer class is decoded directly into authoritative same-class real or paired-complex tensor storage.",
    },
    BuiltinIntegerInputCapability {
        name: "sliceSpec",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "Scalar indices and two-element inclusive ranges accept all eight integer classes or exactly integral floating values.",
    },
];
pub const DATAARRAY_READ_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 1] =
    [BuiltinIntegerCapabilityDescriptor {
        form: "X = DataArray.read(integer_array, integer_sliceSpec)",
        inputs: &DATAARRAY_READ_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::PreserveInput,
        overflow: BuiltinIntegerOverflowRule::NotApplicable,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::Multiple,
        notes: "Full and sliced reads preserve the declared integer dtype, real or paired-complex storage, exact values, selected shape, and column-major order; dataset persistence is host/filesystem I/O.",
    }];

const DATAARRAY_WRITE_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 2] = [
    BuiltinIntegerInputCapability {
        name: "values",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "Scalar or tensor values accept every integer class in real or paired-complex storage and are stored exactly when their class matches the declared array dtype.",
    },
    BuiltinIntegerInputCapability {
        name: "sliceSpec",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "Optional scalar indices and two-element inclusive ranges accept all eight integer classes or exactly integral floating values.",
    },
];
pub const DATAARRAY_WRITE_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 1] =
    [BuiltinIntegerCapabilityDescriptor {
        form: "tf = DataArray.write(arr, integer_values) or DataArray.write(arr, integer_sliceSpec, integer_values)",
        inputs: &DATAARRAY_WRITE_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::Logical,
        overflow: BuiltinIntegerOverflowRule::FunctionSpecific,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::Multiple,
        notes: "Writes preserve exact same-class real or paired-complex integer storage; unlike source and declared dtypes use the declared numeric cast contract before chunk serialization, and shape or complexity mismatches reject.",
    }];

const DATAARRAY_RESIZE_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 1] =
    [BuiltinIntegerInputCapability {
        name: "newShape",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "Shape scalars and vectors accept all eight integer classes or exactly integral floating values within nonnegative platform and checked total-element bounds.",
    }];
pub const DATAARRAY_RESIZE_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 1] =
    [BuiltinIntegerCapabilityDescriptor {
        form: "tf = DataArray.resize(arr, integer_newShape)",
        inputs: &DATAARRAY_RESIZE_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::Logical,
        overflow: BuiltinIntegerOverflowRule::Error,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::StructuralParameter,
        notes: "The exact shape is validated against platform bounds, persisted as metadata, and used to recreate zero-filled storage in the array's declared dtype.",
    }];

const DATAARRAY_FILL_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 1] =
    [BuiltinIntegerInputCapability {
        name: "value",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "The scalar fill value accepts every integer class and is exact when its class matches the declared array dtype.",
    }];
pub const DATAARRAY_FILL_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 1] =
    [BuiltinIntegerCapabilityDescriptor {
        form: "tf = DataArray.fill(arr, integer_value)",
        inputs: &DATAARRAY_FILL_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::Logical,
        overflow: BuiltinIntegerOverflowRule::FunctionSpecific,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::ScalarOnly,
        notes: "A scalar is cast once to the declared dtype and that exact typed value is repeated; nonscalar fills reject.",
    }];

const DATATX_LIFECYCLE_INTEGER_AUDIT: BuiltinIntegerAuditDescriptor =
    BuiltinIntegerAuditDescriptor {
        kind: BuiltinIntegerAuditKind::NotApplicable,
        canonical_builtin: None,
        notes: "DataTransaction id, status, commit, abort, and delete-array operations accept a transaction receiver plus, where applicable, string metadata or an array name; they do not accept integer values or controls, return integer-class values, perform integer arithmetic, or expose an integer backend surface.",
    };

const DATATX_WRITE_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 2] = [
    BuiltinIntegerInputCapability {
        name: "values",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "Queued scalar or tensor values accept every integer class and are stored exactly at commit when their class matches the declared array dtype.",
    },
    BuiltinIntegerInputCapability {
        name: "sliceSpec",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "Queued scalar indices and two-element inclusive ranges accept all eight integer classes or exactly integral floating values.",
    },
];
pub const DATATX_WRITE_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 1] =
    [BuiltinIntegerCapabilityDescriptor {
        form: "tf = DataTransaction.write(tx, arrayName, integer_sliceSpec, integer_values)",
        inputs: &DATATX_WRITE_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::Logical,
        overflow: BuiltinIntegerOverflowRule::FunctionSpecific,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::Multiple,
        notes: "The transaction retains native integer values and exact slice controls until commit; commit applies the same declared-dtype cast, shape validation, and chunk serialization contract as DataArray.write.",
    }];

const DATATX_SET_ATTR_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 1] =
    [BuiltinIntegerInputCapability {
        name: "value",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "A scalar attribute value accepts every integer class and is serialized as an exact JSON integer, including full-width int64 and uint64 values.",
    }];
pub const DATATX_SET_ATTR_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 1] =
    [BuiltinIntegerCapabilityDescriptor {
        form: "tf = DataTransaction.set_attr(tx, key, integer_value)",
        inputs: &DATATX_SET_ATTR_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::Logical,
        overflow: BuiltinIntegerOverflowRule::NotApplicable,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::ScalarOnly,
        notes: "The queued scalar retains its native integer class through the transaction and is emitted as an exact manifest JSON number at commit without an f64 intermediary.",
    }];

const DATATX_SET_ATTRS_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 1] =
    [BuiltinIntegerInputCapability {
        name: "attribute_values",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "Scalar values in the attribute struct accept every integer class and are serialized as exact JSON integers, including full-width int64 and uint64 values.",
    }];
pub const DATATX_SET_ATTRS_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 1] =
    [BuiltinIntegerCapabilityDescriptor {
        form: "tf = DataTransaction.set_attrs(tx, struct_with_integer_values)",
        inputs: &DATATX_SET_ATTRS_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::Logical,
        overflow: BuiltinIntegerOverflowRule::NotApplicable,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::ScalarOnly,
        notes: "Each queued scalar integer field retains its native class and is emitted as an exact manifest JSON number at commit without an f64 intermediary.",
    }];

const DATATX_RESIZE_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 1] =
    [BuiltinIntegerInputCapability {
        name: "newShape",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "Queued shape scalars and vectors accept all eight integer classes or exactly integral floating values within nonnegative platform and checked total-element bounds.",
    }];
pub const DATATX_RESIZE_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 1] =
    [BuiltinIntegerCapabilityDescriptor {
        form: "tf = DataTransaction.resize(tx, arrayName, integer_newShape)",
        inputs: &DATATX_RESIZE_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::Logical,
        overflow: BuiltinIntegerOverflowRule::Error,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::StructuralParameter,
        notes: "The exact shape is validated when queued, retained as platform-sized metadata, and applied at commit by recreating zero-filled storage in the array's declared dtype.",
    }];

const DATATX_FILL_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 2] = [
    BuiltinIntegerInputCapability {
        name: "value",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "The queued scalar fill value accepts every integer class and is exact at commit when its class matches the declared array dtype.",
    },
    BuiltinIntegerInputCapability {
        name: "sliceSpec",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "Optional queued scalar indices and two-element inclusive ranges accept all eight integer classes or exactly integral floating values.",
    },
];
pub const DATATX_FILL_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 1] =
    [BuiltinIntegerCapabilityDescriptor {
        form: "tf = DataTransaction.fill(tx, arrayName, integer_value, integer_sliceSpec)",
        inputs: &DATATX_FILL_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::Logical,
        overflow: BuiltinIntegerOverflowRule::FunctionSpecific,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::Multiple,
        notes: "The transaction retains the exact scalar and optional slice until commit; commit casts once to the declared dtype and repeats that typed value.",
    }];

const DATATX_CREATE_ARRAY_INTEGER_INPUTS: [BuiltinIntegerInputCapability; 2] = [
    BuiltinIntegerInputCapability {
        name: "meta.shape",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "The schema shape accepts all eight integer classes or exactly integral floating values within nonnegative platform and checked total-element bounds.",
    },
    BuiltinIntegerInputCapability {
        name: "meta.chunk",
        classes: &crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES,
        availability: BuiltinIntegerInputAvailability::Documented,
        scalar_double: BuiltinIntegerScalarDoubleRule::Allowed,
        notes: "The optional chunk shape accepts all eight integer classes or exactly integral floating values; dimensions must be positive, platform-bounded, and rank-matched to meta.shape.",
    },
];
pub const DATATX_CREATE_ARRAY_INTEGER_CAPABILITIES: [BuiltinIntegerCapabilityDescriptor; 1] =
    [BuiltinIntegerCapabilityDescriptor {
        form: "tf = DataTransaction.create_array(tx, arrayName, struct('shape', integer_shape, 'chunk', integer_chunk))",
        inputs: &DATATX_CREATE_ARRAY_INTEGER_INPUTS,
        computation_domain: BuiltinIntegerComputationDomain::Structural,
        output_class: BuiltinIntegerOutputClassRule::Logical,
        overflow: BuiltinIntegerOverflowRule::Error,
        backend: BuiltinIntegerBackendRule::HostOnly,
        overload: BuiltinIntegerOverloadKind::StructuralParameter,
        notes: "Exact shape and chunk dimensions are validated when queued, retained as platform-sized metadata, and used at commit to create storage in the schema's declared dtype.",
    }];

one_sig_descriptor!(
    DATATX_ID_DESCRIPTOR,
    DATATX_ID_SIGS,
    "id = DataTransaction.id(tx)",
    &IN_BASE,
    &OUT_STRING
);
one_sig_descriptor!(
    DATATX_WRITE_DESCRIPTOR,
    DATATX_WRITE_SIGS_1,
    "tf = DataTransaction.write(tx, arrayName, sliceSpec, values, Name, Value, ...)",
    &IN_TX_WRITE,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATATX_SET_ATTR_DESCRIPTOR,
    DATATX_SET_ATTR_SIGS,
    "tf = DataTransaction.set_attr(tx, key, value)",
    &IN_BASE_KEY_VALUE,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATATX_SET_ATTRS_DESCRIPTOR,
    DATATX_SET_ATTRS_SIGS,
    "tf = DataTransaction.set_attrs(tx, attrs)",
    &IN_BASE_ATTRS,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATATX_RESIZE_DESCRIPTOR,
    DATATX_RESIZE_SIGS,
    "tf = DataTransaction.resize(tx, arrayName, newShape, Name, Value, ...)",
    &IN_TX_ARRAY_SHAPE_REST,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATATX_FILL_DESCRIPTOR,
    DATATX_FILL_SIGS,
    "tf = DataTransaction.fill(tx, arrayName, value, sliceSpec)",
    &IN_TX_ARRAY_VALUE_SLICE_OPT,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATATX_DELETE_ARRAY_DESCRIPTOR,
    DATATX_DELETE_ARRAY_SIGS,
    "tf = DataTransaction.delete_array(tx, arrayName)",
    &IN_TX_ARRAY_NAME,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATATX_CREATE_ARRAY_DESCRIPTOR,
    DATATX_CREATE_ARRAY_SIGS,
    "tf = DataTransaction.create_array(tx, arrayName, meta)",
    &IN_TX_ARRAY_META,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATATX_COMMIT_DESCRIPTOR,
    DATATX_COMMIT_SIGS,
    "tf = DataTransaction.commit(tx, options)",
    &IN_TX_COMMIT_OPTIONS,
    &OUT_BOOL
);
one_sig_descriptor!(
    COMMIT_ALIAS_DESCRIPTOR,
    COMMIT_ALIAS_SIGS,
    "tf = commit(tx, options)",
    &IN_TX_COMMIT_OPTIONS,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATATX_ABORT_DESCRIPTOR,
    DATATX_ABORT_SIGS,
    "tf = DataTransaction.abort(tx)",
    &IN_BASE,
    &OUT_BOOL
);
one_sig_descriptor!(
    DATATX_STATUS_DESCRIPTOR,
    DATATX_STATUS_SIGS,
    "status = DataTransaction.status(tx)",
    &IN_BASE,
    &OUT_STRING
);

#[runtime_builtin(
    name = "data.create",
    category = "io/data",
    summary = "Create a typed dataset at a .data path.",
    keywords = "data,dataset,create,persistence",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_dataset_type),
    descriptor(crate::builtins::io::data::DATA_CREATE_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATA_CREATE_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_create_builtin(path: Value, schema: Value) -> BuiltinResult<Value> {
    let path = parse_string(&path, "data.create path")?;
    let root = dataset_root(&path);
    let schema = parse_schema(&schema)?;
    let now = now_rfc3339();
    let mut arrays = BTreeMap::new();
    for (name, mut meta) in schema.arrays {
        let payload = DataArrayPayload::zeros(meta.dtype.clone(), meta.shape.clone())?;
        let (payload_path, chunk_index_path) =
            write_array_payload_async(&root, &name, &payload, &meta.chunk_shape).await?;
        meta.data_path = make_rel_data_path(&root, &payload_path)?;
        meta.chunk_index_path = Some(make_rel_data_path(&root, &chunk_index_path)?);
        arrays.insert(name, meta);
    }

    let manifest = DataManifest {
        schema_version: 1,
        format: "runmat-data".to_string(),
        dataset_id: crate::data::new_dataset_id(),
        name: root.file_name().map(|v| v.to_string_lossy().to_string()),
        created_at: now.clone(),
        updated_at: now,
        arrays,
        attrs: BTreeMap::new(),
        txn_sequence: 0,
    };
    write_manifest_async(&root, &manifest).await?;
    Ok(dataset_object(&path, &manifest))
}

#[runtime_builtin(
    name = "data.open",
    category = "io/data",
    summary = "Open a dataset handle from a .data path.",
    keywords = "data,dataset,open,persistence",
    type_resolver(crate::builtins::io::type_resolvers::data_dataset_type),
    descriptor(crate::builtins::io::data::DATA_OPEN_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATA_NAMESPACE_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_open_builtin(path: Value) -> BuiltinResult<Value> {
    let path = parse_string(&path, "data.open path")?;
    let root = dataset_root(&path);
    let manifest = read_manifest_async(&root).await?;
    let mut ds = dataset_object(&path, &manifest);
    hydrate_dataset_descriptor_async(&path, &mut ds).await;
    Ok(ds)
}

#[runtime_builtin(
    name = "data.exists",
    category = "io/data",
    summary = "Check if dataset exists.",
    keywords = "data,dataset,exists",
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATA_EXISTS_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATA_NAMESPACE_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_exists_builtin(path: Value) -> BuiltinResult<Value> {
    let path = parse_string(&path, "data.exists path")?;
    let root = dataset_root(&path);
    let exists = runmat_filesystem::metadata_async(manifest_path(&root))
        .await
        .is_ok();
    Ok(Value::Bool(exists))
}

#[runtime_builtin(
    name = "data.delete",
    category = "io/data",
    summary = "Delete a dataset path.",
    keywords = "data,dataset,delete",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATA_DELETE_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATA_NAMESPACE_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_delete_builtin(path: Value) -> BuiltinResult<Value> {
    let path = parse_string(&path, "data.delete path")?;
    let root = dataset_root(&path);
    runmat_filesystem::remove_dir_all_async(&root)
        .await
        .map_err(|err| {
            data_error(format!(
                "data.delete: failed to remove '{}': {err}",
                root.display()
            ))
        })?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "data.copy",
    category = "io/data",
    summary = "Copy dataset to new path.",
    keywords = "data,dataset,copy",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATA_COPY_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATA_NAMESPACE_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_copy_builtin(from_path: Value, to_path: Value) -> BuiltinResult<Value> {
    let from = parse_string(&from_path, "data.copy fromPath")?;
    let to = parse_string(&to_path, "data.copy toPath")?;
    copy_dir_recursive(&dataset_root(&from), &dataset_root(&to)).await?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "data.move",
    category = "io/data",
    summary = "Move dataset to new path.",
    keywords = "data,dataset,move",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATA_MOVE_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATA_NAMESPACE_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_move_builtin(from_path: Value, to_path: Value) -> BuiltinResult<Value> {
    let from = parse_string(&from_path, "data.move fromPath")?;
    let to = parse_string(&to_path, "data.move toPath")?;
    runmat_filesystem::rename_async(dataset_root(&from), dataset_root(&to))
        .await
        .map_err(|err| {
            data_error(format!(
                "data.move: failed to move dataset '{from}' -> '{to}': {err}"
            ))
        })?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "data.import",
    category = "io/data",
    summary = "Import an existing dataset file path.",
    keywords = "data,dataset,import",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_dataset_type),
    descriptor(crate::builtins::io::data::DATA_IMPORT_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATA_NAMESPACE_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_import_builtin(
    path: Value,
    format: Value,
    source_path: Value,
) -> BuiltinResult<Value> {
    let path = parse_string(&path, "data.import path")?;
    let format = parse_string(&format, "data.import format")?;
    if !format.eq_ignore_ascii_case("data") {
        return Err(data_error(
            "data.import currently supports only format='data'",
        ));
    }
    let source_path = parse_string(&source_path, "data.import sourcePath")?;
    copy_dir_recursive(&dataset_root(&source_path), &dataset_root(&path)).await?;
    let manifest = read_manifest_async(&dataset_root(&path)).await?;
    Ok(dataset_object(&path, &manifest))
}

#[runtime_builtin(
    name = "data.export",
    category = "io/data",
    summary = "Export dataset to target path.",
    keywords = "data,dataset,export",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATA_EXPORT_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATA_NAMESPACE_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_export_builtin(
    path: Value,
    format: Value,
    target_path: Value,
) -> BuiltinResult<Value> {
    let path = parse_string(&path, "data.export path")?;
    let format = parse_string(&format, "data.export format")?;
    if !format.eq_ignore_ascii_case("data") {
        return Err(data_error(
            "data.export currently supports only format='data'",
        ));
    }
    let target_path = parse_string(&target_path, "data.export targetPath")?;
    copy_dir_recursive(&dataset_root(&path), &dataset_root(&target_path)).await?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "data.list",
    category = "io/data",
    summary = "List dataset paths under a prefix.",
    keywords = "data,dataset,list",
    type_resolver(crate::builtins::io::type_resolvers::data_cell_string_type),
    descriptor(crate::builtins::io::data::DATA_LIST_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATA_NAMESPACE_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_list_builtin(path_prefix: Value) -> BuiltinResult<Value> {
    let prefix = parse_string(&path_prefix, "data.list prefix")?;
    let root = PathBuf::from(prefix);
    let entries = runmat_filesystem::read_dir_async(&root)
        .await
        .map_err(|err| {
            data_error(format!(
                "data.list: failed to read '{}': {err}",
                root.display()
            ))
        })?;
    let mut values = Vec::new();
    for entry in entries {
        if !entry.is_dir() {
            continue;
        }
        let candidate = entry.path();
        if candidate.extension().and_then(|s| s.to_str()) != Some("data") {
            continue;
        }
        if runmat_filesystem::metadata_async(candidate.join("manifest.json"))
            .await
            .is_ok()
        {
            values.push(Value::String(candidate.to_string_lossy().to_string()));
        }
    }
    let cols = values.len();
    make_cell(values, 1, cols).map_err(data_error)
}

#[runtime_builtin(
    name = "data.inspect",
    category = "io/data",
    summary = "Inspect dataset metadata and schema fields.",
    keywords = "data,dataset,inspect,schema",
    type_resolver(crate::builtins::io::type_resolvers::data_struct_type),
    descriptor(crate::builtins::io::data::DATA_INSPECT_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATA_NAMESPACE_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_inspect_builtin(path: Value) -> BuiltinResult<Value> {
    let path = parse_string(&path, "data.inspect path")?;
    let root = dataset_root(&path);
    let manifest = read_manifest_async(&root).await?;
    let mut out = StructValue::new();
    out.fields.insert("path".to_string(), Value::String(path));
    out.fields
        .insert("id".to_string(), Value::String(manifest.dataset_id));
    out.fields.insert(
        "arrayCount".to_string(),
        Value::Num(manifest.arrays.len() as f64),
    );
    out.fields
        .insert("updatedAt".to_string(), Value::String(manifest.updated_at));
    Ok(Value::Struct(out))
}

#[runtime_builtin(
    name = "Dataset.path",
    category = "io/data",
    summary = "Return dataset path.",
    keywords = "dataset,path",
    type_resolver(crate::builtins::io::type_resolvers::data_string_type),
    descriptor(crate::builtins::io::data::DATASET_PATH_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATASET_NONATTRIBUTE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn dataset_path_builtin(base: Value) -> BuiltinResult<Value> {
    let obj = as_object(&base, "Dataset.path")?;
    Ok(get_object_prop(obj, "__data_path")?.clone())
}

#[runtime_builtin(
    name = "Dataset.id",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_string_type),
    descriptor(crate::builtins::io::data::DATASET_ID_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATASET_NONATTRIBUTE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn dataset_id_builtin(base: Value) -> BuiltinResult<Value> {
    let obj = as_object(&base, "Dataset.id")?;
    Ok(get_object_prop(obj, "__data_id")?.clone())
}

#[runtime_builtin(
    name = "Dataset.version",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_string_type),
    descriptor(crate::builtins::io::data::DATASET_VERSION_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATASET_NONATTRIBUTE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn dataset_version_builtin(base: Value) -> BuiltinResult<Value> {
    let obj = as_object(&base, "Dataset.version")?;
    Ok(get_object_prop(obj, "__data_version")?.clone())
}

#[runtime_builtin(
    name = "Dataset.arrays",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_cell_string_type),
    descriptor(crate::builtins::io::data::DATASET_ARRAYS_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATASET_NONATTRIBUTE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn dataset_arrays_builtin(base: Value) -> BuiltinResult<Value> {
    let path = dataset_path_from_object(&base, "Dataset.arrays")?;
    let manifest = read_manifest_async(&dataset_root(&path)).await?;
    let values: Vec<Value> = manifest
        .arrays
        .keys()
        .map(|k| Value::String(k.clone()))
        .collect();
    make_cell(values.clone(), 1, values.len()).map_err(data_error)
}

#[runtime_builtin(
    name = "Dataset.has_array",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATASET_HAS_ARRAY_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATASET_NONATTRIBUTE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn dataset_has_array_builtin(base: Value, name: Value) -> BuiltinResult<Value> {
    let path = dataset_path_from_object(&base, "Dataset.has_array")?;
    let name = parse_string(&name, "Dataset.has_array name")?;
    let manifest = read_manifest_async(&dataset_root(&path)).await?;
    Ok(Value::Bool(manifest.arrays.contains_key(&name)))
}

#[runtime_builtin(
    name = "Dataset.array",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_array_type),
    descriptor(crate::builtins::io::data::DATASET_ARRAY_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATASET_NONATTRIBUTE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn dataset_array_builtin(base: Value, name: Value) -> BuiltinResult<Value> {
    let path = dataset_path_from_object(&base, "Dataset.array")?;
    let name = parse_string(&name, "Dataset.array name")?;
    let manifest = read_manifest_async(&dataset_root(&path)).await?;
    if !manifest.arrays.contains_key(&name) {
        return Err(data_error(format!(
            "Dataset.array: array '{name}' not found"
        )));
    }
    Ok(array_object(&path, &name))
}

#[runtime_builtin(
    name = "Dataset.attrs",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_struct_type),
    descriptor(crate::builtins::io::data::DATASET_ATTRS_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATASET_ATTRS_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn dataset_attrs_builtin(base: Value) -> BuiltinResult<Value> {
    let path = dataset_path_from_object(&base, "Dataset.attrs")?;
    let manifest = read_manifest_async(&dataset_root(&path)).await?;
    Ok(attrs_to_struct(&manifest.attrs))
}

#[runtime_builtin(
    name = "Dataset.get_attr",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_unknown_type),
    descriptor(crate::builtins::io::data::DATASET_GET_ATTR_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATASET_GET_ATTR_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn dataset_get_attr_builtin(
    base: Value,
    key: Value,
    rest: Vec<Value>,
) -> BuiltinResult<Value> {
    let path = dataset_path_from_object(&base, "Dataset.get_attr")?;
    let key = parse_string(&key, "Dataset.get_attr key")?;
    let manifest = read_manifest_async(&dataset_root(&path)).await?;
    if let Some(value) = manifest.attrs.get(&key) {
        return Ok(json_to_value(value));
    }
    Ok(rest.first().cloned().unwrap_or(Value::Num(0.0)))
}

#[runtime_builtin(
    name = "Dataset.set_attr",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATASET_SET_ATTR_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATASET_SET_ATTR_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn dataset_set_attr_builtin(base: Value, key: Value, value: Value) -> BuiltinResult<Value> {
    let path = dataset_path_from_object(&base, "Dataset.set_attr")?;
    let key = parse_string(&key, "Dataset.set_attr key")?;
    let root = dataset_root(&path);
    let mut manifest = read_manifest_async(&root).await?;
    manifest.attrs.insert(key, value_to_json(&value));
    manifest.updated_at = now_rfc3339();
    manifest.txn_sequence = manifest.txn_sequence.saturating_add(1);
    write_manifest_async(&root, &manifest).await?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "Dataset.set_attrs",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATASET_SET_ATTRS_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATASET_SET_ATTRS_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn dataset_set_attrs_builtin(base: Value, attrs: Value) -> BuiltinResult<Value> {
    let path = dataset_path_from_object(&base, "Dataset.set_attrs")?;
    let Value::Struct(incoming) = attrs else {
        return Err(data_error("Dataset.set_attrs: attrs must be a struct"));
    };
    let root = dataset_root(&path);
    let mut manifest = read_manifest_async(&root).await?;
    for (k, v) in incoming.fields {
        manifest.attrs.insert(k, value_to_json(&v));
    }
    manifest.updated_at = now_rfc3339();
    manifest.txn_sequence = manifest.txn_sequence.saturating_add(1);
    write_manifest_async(&root, &manifest).await?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "Dataset.begin",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_tx_type),
    descriptor(crate::builtins::io::data::DATASET_BEGIN_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATASET_NONATTRIBUTE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn dataset_begin_builtin(base: Value, _rest: Vec<Value>) -> BuiltinResult<Value> {
    let path = dataset_path_from_object(&base, "Dataset.begin")?;
    let manifest = read_manifest_async(&dataset_root(&path)).await?;
    let tx_id = start_tx(path.clone(), manifest.txn_sequence)?;
    tracing::info!(
        target: "runmat.data",
        dataset = path,
        tx_id = tx_id,
        base_sequence = manifest.txn_sequence,
        "data transaction begin"
    );
    Ok(transaction_object(&path, &tx_id))
}

#[runtime_builtin(
    name = "Dataset.snapshot",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_string_type),
    descriptor(crate::builtins::io::data::DATASET_SNAPSHOT_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATASET_NONATTRIBUTE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn dataset_snapshot_builtin(
    base: Value,
    label: Value,
    _rest: Vec<Value>,
) -> BuiltinResult<Value> {
    let path = dataset_path_from_object(&base, "Dataset.snapshot")?;
    let label = parse_string(&label, "Dataset.snapshot label")?;
    let root = dataset_root(&path);
    let snapshots = root.join(".snapshots");
    runmat_filesystem::create_dir_all_async(&snapshots)
        .await
        .map_err(|err| {
            data_error(format!(
                "Dataset.snapshot: failed to create snapshots dir: {err}"
            ))
        })?;
    let src = manifest_path(&root);
    let dst = snapshots.join(format!("{}.manifest.json", sanitize_label(&label)));
    copy_file(&src, &dst).await?;
    Ok(Value::String(dst.to_string_lossy().to_string()))
}

#[runtime_builtin(
    name = "Dataset.refresh",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_dataset_type),
    descriptor(crate::builtins::io::data::DATASET_REFRESH_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATASET_NONATTRIBUTE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn dataset_refresh_builtin(base: Value) -> BuiltinResult<Value> {
    let path = dataset_path_from_object(&base, "Dataset.refresh")?;
    let manifest = read_manifest_async(&dataset_root(&path)).await?;
    let mut ds = dataset_object(&path, &manifest);
    hydrate_dataset_descriptor_async(&path, &mut ds).await;
    Ok(ds)
}

#[runtime_builtin(
    name = "DataArray.name",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_string_type),
    descriptor(crate::builtins::io::data::DATAARRAY_NAME_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATAARRAY_METADATA_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_array_name_builtin(base: Value) -> BuiltinResult<Value> {
    let obj = as_object(&base, "DataArray.name")?;
    Ok(get_object_prop(obj, "__array_name")?.clone())
}

#[runtime_builtin(
    name = "DataArray.dtype",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_string_type),
    descriptor(crate::builtins::io::data::DATAARRAY_DTYPE_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATAARRAY_METADATA_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_array_dtype_builtin(base: Value) -> BuiltinResult<Value> {
    let (path, name) = array_identity(&base, "DataArray.dtype")?;
    let manifest = read_manifest_async(&dataset_root(&path)).await?;
    let meta = manifest
        .arrays
        .get(&name)
        .ok_or_else(|| data_error(format!("DataArray.dtype: array '{name}' not found")))?;
    Ok(Value::String(meta.dtype.clone()))
}

#[runtime_builtin(
    name = "DataArray.shape",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_shape_tensor_type),
    descriptor(crate::builtins::io::data::DATAARRAY_SHAPE_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATAARRAY_METADATA_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_array_shape_builtin(base: Value) -> BuiltinResult<Value> {
    let (path, name) = array_identity(&base, "DataArray.shape")?;
    let manifest = read_manifest_async(&dataset_root(&path)).await?;
    let meta = manifest
        .arrays
        .get(&name)
        .ok_or_else(|| data_error(format!("DataArray.shape: array '{name}' not found")))?;
    let values = meta.shape.iter().map(|v| *v as f64).collect::<Vec<_>>();
    let tensor = Tensor::new(values, vec![1, meta.shape.len()])
        .map_err(|err| data_error(format!("DataArray.shape: {err}")))?;
    Ok(Value::Tensor(tensor))
}

#[runtime_builtin(
    name = "DataArray.rank",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_int_type),
    descriptor(crate::builtins::io::data::DATAARRAY_RANK_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATAARRAY_METADATA_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_array_rank_builtin(base: Value) -> BuiltinResult<Value> {
    let (path, name) = array_identity(&base, "DataArray.rank")?;
    let manifest = read_manifest_async(&dataset_root(&path)).await?;
    let meta = manifest
        .arrays
        .get(&name)
        .ok_or_else(|| data_error(format!("DataArray.rank: array '{name}' not found")))?;
    Ok(Value::Num(meta.shape.len() as f64))
}

#[runtime_builtin(
    name = "DataArray.chunk_shape",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_shape_tensor_type),
    descriptor(crate::builtins::io::data::DATAARRAY_CHUNK_SHAPE_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATAARRAY_METADATA_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_array_chunk_shape_builtin(base: Value) -> BuiltinResult<Value> {
    let (path, name) = array_identity(&base, "DataArray.chunk_shape")?;
    let manifest = read_manifest_async(&dataset_root(&path)).await?;
    let meta = manifest
        .arrays
        .get(&name)
        .ok_or_else(|| data_error(format!("DataArray.chunk_shape: array '{name}' not found")))?;
    let values = meta
        .chunk_shape
        .iter()
        .map(|v| *v as f64)
        .collect::<Vec<_>>();
    let tensor = Tensor::new(values, vec![1, meta.chunk_shape.len()])
        .map_err(|err| data_error(format!("DataArray.chunk_shape: {err}")))?;
    Ok(Value::Tensor(tensor))
}

#[runtime_builtin(
    name = "DataArray.codec",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_string_type),
    descriptor(crate::builtins::io::data::DATAARRAY_CODEC_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATAARRAY_METADATA_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_array_codec_builtin(base: Value) -> BuiltinResult<Value> {
    let (path, name) = array_identity(&base, "DataArray.codec")?;
    let manifest = read_manifest_async(&dataset_root(&path)).await?;
    let meta = manifest
        .arrays
        .get(&name)
        .ok_or_else(|| data_error(format!("DataArray.codec: array '{name}' not found")))?;
    Ok(Value::String(meta.codec.clone()))
}

#[runtime_builtin(
    name = "DataArray.read",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_tensor_type),
    descriptor(crate::builtins::io::data::DATAARRAY_READ_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATAARRAY_READ_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_array_read_builtin(base: Value, rest: Vec<Value>) -> BuiltinResult<Value> {
    let (path, name) = array_identity(&base, "DataArray.read")?;
    let root = dataset_root(&path);
    let manifest = read_manifest_async(&root).await?;
    let meta = manifest
        .arrays
        .get(&name)
        .ok_or_else(|| data_error(format!("DataArray.read: array '{name}' not found")))?;
    let payload = read_array_payload_async(&root, meta).await?;
    let sliced = if let Some(slice_spec) = rest.first() {
        read_slice_payload(&payload, slice_spec)?
    } else {
        payload
    };
    sliced
        .into_value()
        .map_err(|err| data_error(format!("DataArray.read: {err}")))
}

#[runtime_builtin(
    name = "DataArray.write",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATAARRAY_WRITE_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATAARRAY_WRITE_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_array_write_builtin(base: Value, rest: Vec<Value>) -> BuiltinResult<Value> {
    let (path, name) = array_identity(&base, "DataArray.write")?;
    let (slice_spec, value) = match rest.as_slice() {
        [v] => (None, v),
        [slice, v] => (Some(slice), v),
        _ => {
            return Err(data_error(
                "DataArray.write expects values or (sliceSpec, values) arguments",
            ))
        }
    };
    write_array_full_async(&path, &name, slice_spec, value).await?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "DataArray.resize",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATAARRAY_RESIZE_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATAARRAY_RESIZE_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_array_resize_builtin(
    base: Value,
    new_shape: Value,
    _rest: Vec<Value>,
) -> BuiltinResult<Value> {
    let (path, name) = array_identity(&base, "DataArray.resize")?;
    let shape = parse_shape_from_value(&new_shape)?;
    let root = dataset_root(&path);
    let mut manifest = read_manifest_async(&root).await?;
    let meta = manifest
        .arrays
        .get_mut(&name)
        .ok_or_else(|| data_error(format!("DataArray.resize: array '{name}' not found")))?;
    meta.shape = shape.clone();
    let payload = DataArrayPayload::zeros(meta.dtype.clone(), shape.clone())?;
    let (payload_path, chunk_index_path) =
        write_array_payload_async(&root, &name, &payload, &meta.chunk_shape).await?;
    meta.data_path = make_rel_data_path(&root, &payload_path)?;
    meta.chunk_index_path = Some(make_rel_data_path(&root, &chunk_index_path)?);
    manifest.updated_at = now_rfc3339();
    manifest.txn_sequence = manifest.txn_sequence.saturating_add(1);
    write_manifest_async(&root, &manifest).await?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "DataArray.fill",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATAARRAY_FILL_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATAARRAY_FILL_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_array_fill_builtin(
    base: Value,
    value: Value,
    _rest: Vec<Value>,
) -> BuiltinResult<Value> {
    let (path, name) = array_identity(&base, "DataArray.fill")?;
    let root = dataset_root(&path);
    let mut manifest = read_manifest_async(&root).await?;
    let meta = manifest
        .arrays
        .get_mut(&name)
        .ok_or_else(|| data_error(format!("DataArray.fill: array '{name}' not found")))?;
    let payload = DataArrayPayload::filled(meta.dtype.clone(), meta.shape.clone(), &value)?;
    let (payload_path, chunk_index_path) =
        write_array_payload_async(&root, &name, &payload, &meta.chunk_shape).await?;
    meta.data_path = make_rel_data_path(&root, &payload_path)?;
    meta.chunk_index_path = Some(make_rel_data_path(&root, &chunk_index_path)?);
    manifest.updated_at = now_rfc3339();
    manifest.txn_sequence = manifest.txn_sequence.saturating_add(1);
    write_manifest_async(&root, &manifest).await?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "DataTransaction.id",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_string_type),
    descriptor(crate::builtins::io::data::DATATX_ID_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATATX_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_tx_id_builtin(base: Value) -> BuiltinResult<Value> {
    let obj = as_object(&base, "DataTransaction.id")?;
    Ok(get_object_prop(obj, "__tx_id")?.clone())
}

#[runtime_builtin(
    name = "DataTransaction.write",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATATX_WRITE_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATATX_WRITE_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_tx_write_builtin(
    base: Value,
    array_name: Value,
    slice: Value,
    values: Value,
    _rest: Vec<Value>,
) -> BuiltinResult<Value> {
    let tx_id = tx_id_from_object(&base, "DataTransaction.write")?;
    let array_name = parse_string(&array_name, "DataTransaction.write arrayName")?;
    with_tx_mut(&tx_id, |tx| {
        if tx.status != TxnStatus::Open {
            return Err(data_error("DataTransaction.write: transaction is not open"));
        }
        tx.writes.push(PendingWrite {
            array: array_name,
            slice_spec: Some(slice),
            value: values,
        });
        Ok(())
    })?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "DataTransaction.set_attr",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATATX_SET_ATTR_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATATX_SET_ATTR_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_tx_set_attr_builtin(base: Value, key: Value, value: Value) -> BuiltinResult<Value> {
    let tx_id = tx_id_from_object(&base, "DataTransaction.set_attr")?;
    let key = parse_string(&key, "DataTransaction.set_attr key")?;
    with_tx_mut(&tx_id, |tx| {
        if tx.status != TxnStatus::Open {
            return Err(data_error(
                "DataTransaction.set_attr: transaction is not open",
            ));
        }
        tx.attrs.insert(key, value);
        Ok(())
    })?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "DataTransaction.set_attrs",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATATX_SET_ATTRS_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATATX_SET_ATTRS_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_tx_set_attrs_builtin(base: Value, attrs: Value) -> BuiltinResult<Value> {
    let tx_id = tx_id_from_object(&base, "DataTransaction.set_attrs")?;
    let Value::Struct(incoming) = attrs else {
        return Err(data_error(
            "DataTransaction.set_attrs: attrs must be struct",
        ));
    };
    with_tx_mut(&tx_id, |tx| {
        if tx.status != TxnStatus::Open {
            return Err(data_error(
                "DataTransaction.set_attrs: transaction is not open",
            ));
        }
        for (k, v) in incoming.fields {
            tx.attrs.insert(k, v);
        }
        Ok(())
    })?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "DataTransaction.resize",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATATX_RESIZE_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATATX_RESIZE_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_tx_resize_builtin(
    base: Value,
    array_name: Value,
    new_shape: Value,
    _rest: Vec<Value>,
) -> BuiltinResult<Value> {
    let tx_id = tx_id_from_object(&base, "DataTransaction.resize")?;
    let array_name = parse_string(&array_name, "DataTransaction.resize arrayName")?;
    let shape = parse_shape_from_value(&new_shape)?;
    with_tx_mut(&tx_id, |tx| {
        if tx.status != TxnStatus::Open {
            return Err(data_error(
                "DataTransaction.resize: transaction is not open",
            ));
        }
        tx.resizes.push(PendingResize {
            array: array_name,
            shape,
        });
        Ok(())
    })?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "DataTransaction.fill",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATATX_FILL_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATATX_FILL_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_tx_fill_builtin(
    base: Value,
    array_name: Value,
    value: Value,
    rest: Vec<Value>,
) -> BuiltinResult<Value> {
    let tx_id = tx_id_from_object(&base, "DataTransaction.fill")?;
    let array_name = parse_string(&array_name, "DataTransaction.fill arrayName")?;
    let slice_spec = rest.first().cloned();
    with_tx_mut(&tx_id, |tx| {
        if tx.status != TxnStatus::Open {
            return Err(data_error("DataTransaction.fill: transaction is not open"));
        }
        tx.fills.push(PendingFill {
            array: array_name,
            slice_spec,
            value,
        });
        Ok(())
    })?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "DataTransaction.delete_array",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATATX_DELETE_ARRAY_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATATX_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_tx_delete_array_builtin(base: Value, array_name: Value) -> BuiltinResult<Value> {
    let tx_id = tx_id_from_object(&base, "DataTransaction.delete_array")?;
    let array_name = parse_string(&array_name, "DataTransaction.delete_array arrayName")?;
    with_tx_mut(&tx_id, |tx| {
        if tx.status != TxnStatus::Open {
            return Err(data_error(
                "DataTransaction.delete_array: transaction is not open",
            ));
        }
        tx.delete_arrays.push(array_name);
        Ok(())
    })?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "DataTransaction.create_array",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATATX_CREATE_ARRAY_DESCRIPTOR),
    integer_capabilities(crate::builtins::io::data::DATATX_CREATE_ARRAY_INTEGER_CAPABILITIES),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_tx_create_array_builtin(
    base: Value,
    array_name: Value,
    meta: Value,
) -> BuiltinResult<Value> {
    let tx_id = tx_id_from_object(&base, "DataTransaction.create_array")?;
    let array_name = parse_string(&array_name, "DataTransaction.create_array arrayName")?;
    let meta = parse_array_meta(&array_name, &meta)?;
    with_tx_mut(&tx_id, |tx| {
        if tx.status != TxnStatus::Open {
            return Err(data_error(
                "DataTransaction.create_array: transaction is not open",
            ));
        }
        tx.create_arrays.push(PendingCreateArray {
            array: array_name,
            meta,
        });
        Ok(())
    })?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "DataTransaction.commit",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATATX_COMMIT_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATATX_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_tx_commit_builtin(base: Value, rest: Vec<Value>) -> BuiltinResult<Value> {
    let options = match rest.as_slice() {
        [] => None,
        [Value::Struct(options)] => Some(options),
        [_] => {
            return Err(data_error(
                "DataTransaction.commit: options must be a scalar struct",
            ))
        }
        _ => {
            return Err(data_error(
                "DataTransaction.commit: expected at most one options struct",
            ))
        }
    };
    let tx_id = tx_id_from_object(&base, "DataTransaction.commit")?;
    let (dataset_path, base_sequence, writes, resizes, fills, create_arrays, delete_arrays, attrs) =
        with_tx(&tx_id, |tx| {
            if tx.status != TxnStatus::Open {
                return Err(data_error(
                    "DataTransaction.commit: transaction is not open",
                ));
            }
            Ok((
                tx.dataset_path.clone(),
                tx.base_sequence,
                tx.writes.clone(),
                tx.resizes.clone(),
                tx.fills.clone(),
                tx.create_arrays.clone(),
                tx.delete_arrays.clone(),
                tx.attrs.clone(),
            ))
        })?;

    let write_ops = writes.len();
    let resize_ops = resizes.len();
    let fill_ops = fills.len();
    let create_ops = create_arrays.len();
    let delete_ops = delete_arrays.len();
    let attr_updates = attrs.len();

    let root = dataset_root(&dataset_path);
    let mut manifest = read_manifest_async(&root).await?;
    ensure_manifest_sequence(base_sequence, &manifest)?;
    if let Some(options) = options {
        if let Some(expected) = options.fields.get("if_manifest") {
            let expected = parse_string(expected, "DataTransaction.commit if_manifest")?;
            let actual = manifest_version_token(&manifest);
            if expected != actual {
                tracing::warn!(
                    target: "runmat.data",
                    tx_id = tx_id,
                    expected_manifest = expected,
                    actual_manifest = actual,
                    "data transaction manifest conflict"
                );
                return Err(data_error(
                    "MANIFEST_CONFLICT: if_manifest precondition failed",
                ));
            }
        }
    }
    for create in create_arrays {
        create_array_in_manifest(&root, &mut manifest, &create.array, create.meta).await?;
    }
    for resize in resizes {
        resize_array_in_manifest(&root, &mut manifest, &resize.array, resize.shape).await?;
    }
    for fill in fills {
        fill_array_in_manifest(
            &root,
            &mut manifest,
            &fill.array,
            fill.slice_spec.as_ref(),
            &fill.value,
        )
        .await?;
    }
    for write in writes {
        apply_write_to_manifest_async(
            &root,
            &mut manifest,
            &write.array,
            write.slice_spec.as_ref(),
            &write.value,
        )
        .await?;
    }
    for array_name in delete_arrays {
        delete_array_in_manifest_async(&root, &mut manifest, &array_name).await?;
    }
    for (k, v) in attrs {
        manifest.attrs.insert(k, value_to_json(&v));
    }
    manifest.updated_at = now_rfc3339();
    manifest.txn_sequence = manifest.txn_sequence.saturating_add(1);
    write_manifest_async(&root, &manifest).await?;
    with_tx_mut(&tx_id, |tx| {
        tx.status = TxnStatus::Committed;
        Ok(())
    })?;
    tracing::info!(
        target: "runmat.data",
        dataset = dataset_path,
        tx_id = tx_id,
        write_ops = write_ops,
        resize_ops = resize_ops,
        fill_ops = fill_ops,
        create_ops = create_ops,
        delete_ops = delete_ops,
        attr_updates = attr_updates,
        next_sequence = manifest.txn_sequence,
        "data transaction commit"
    );
    remove_tx(&tx_id)?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "commit",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::COMMIT_ALIAS_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATATX_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_tx_commit_alias_builtin(base: Value, rest: Vec<Value>) -> BuiltinResult<Value> {
    match &base {
        Value::Object(obj) if obj.class_name == "DataTransaction" => {
            data_tx_commit_builtin(base, rest).await
        }
        _ => Err(data_error(
            "commit: receiver must be a DataTransaction (use tx = ds.begin())",
        )),
    }
}

#[runtime_builtin(
    name = "DataTransaction.abort",
    category = "io/data",
    sink = true,
    type_resolver(crate::builtins::io::type_resolvers::data_bool_type),
    descriptor(crate::builtins::io::data::DATATX_ABORT_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATATX_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_tx_abort_builtin(base: Value) -> BuiltinResult<Value> {
    let tx_id = tx_id_from_object(&base, "DataTransaction.abort")?;
    with_tx_mut(&tx_id, |tx| {
        tx.status = TxnStatus::Aborted;
        Ok(())
    })?;
    tracing::info!(
        target: "runmat.data",
        tx_id = tx_id,
        "data transaction abort"
    );
    remove_tx(&tx_id)?;
    Ok(Value::Bool(true))
}

#[runtime_builtin(
    name = "DataTransaction.status",
    category = "io/data",
    type_resolver(crate::builtins::io::type_resolvers::data_string_type),
    descriptor(crate::builtins::io::data::DATATX_STATUS_DESCRIPTOR),
    integer_audit(crate::builtins::io::data::DATATX_LIFECYCLE_INTEGER_AUDIT),
    builtin_path = "crate::builtins::io::data"
)]
async fn data_tx_status_builtin(base: Value) -> BuiltinResult<Value> {
    let tx_id = tx_id_from_object(&base, "DataTransaction.status")?;
    with_tx(&tx_id, |tx| {
        let status = match tx.status {
            TxnStatus::Open => "open",
            TxnStatus::Committed => "committed",
            TxnStatus::Aborted => "aborted",
        };
        Ok(Value::String(status.to_string()))
    })
}

fn dataset_path_from_object(base: &Value, context: &str) -> BuiltinResult<String> {
    let obj = as_object(base, context)?;
    parse_string(get_object_prop(obj, "__data_path")?, context)
}

fn tx_id_from_object(base: &Value, context: &str) -> BuiltinResult<String> {
    let obj = as_object(base, context)?;
    parse_string(get_object_prop(obj, "__tx_id")?, context)
}

fn array_identity(base: &Value, context: &str) -> BuiltinResult<(String, String)> {
    let obj = as_object(base, context)?;
    let path = parse_string(get_object_prop(obj, "__data_path")?, context)?;
    let name = parse_string(get_object_prop(obj, "__array_name")?, context)?;
    Ok((path, name))
}

fn as_object<'a>(value: &'a Value, context: &str) -> BuiltinResult<&'a ObjectInstance> {
    match value {
        Value::Object(obj) => Ok(obj),
        _ => Err(data_error(format!("{context}: expected object receiver"))),
    }
}

async fn hydrate_dataset_descriptor_async(path: &str, dataset: &mut Value) {
    let request = runmat_filesystem::data_contract::DataManifestRequest {
        path: path.to_string(),
        version: None,
    };
    let descriptor = match runmat_filesystem::data_manifest_descriptor_async(&request).await {
        Ok(descriptor) => descriptor,
        Err(_) => return,
    };
    let Value::Object(obj) = dataset else {
        return;
    };
    if !descriptor.dataset_id.is_empty() {
        obj.properties.insert(
            "__data_id".to_string(),
            Value::String(descriptor.dataset_id),
        );
    }
    obj.properties.insert(
        "__data_version".to_string(),
        Value::String(format!(
            "{}:{}",
            descriptor.updated_at, descriptor.txn_sequence
        )),
    );
}

fn sanitize_label(label: &str) -> String {
    label
        .chars()
        .map(|ch| {
            if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' {
                ch
            } else {
                '_'
            }
        })
        .collect()
}

async fn copy_file(src: &PathBuf, dst: &PathBuf) -> BuiltinResult<()> {
    let bytes = runmat_filesystem::read_async(src)
        .await
        .map_err(|err| data_error(format!("failed to open '{}': {err}", src.display())))?;
    let parent = dst.parent().ok_or_else(|| {
        data_error(format!(
            "invalid destination path '{}': missing parent",
            dst.display()
        ))
    })?;
    runmat_filesystem::create_dir_all_async(parent)
        .await
        .map_err(|err| data_error(format!("failed to create '{}': {err}", parent.display())))?;
    runmat_filesystem::write_async(dst, &bytes)
        .await
        .map_err(|err| {
            data_error(format!(
                "failed to copy '{}' -> '{}': {err}",
                src.display(),
                dst.display()
            ))
        })?;
    Ok(())
}

fn make_rel_data_path(
    root: &std::path::Path,
    payload_path: &std::path::Path,
) -> BuiltinResult<String> {
    let rel = payload_path
        .strip_prefix(root)
        .map_err(|err| data_error(format!("failed to compute relative data path: {err}")))?;
    Ok(rel.to_string_lossy().to_string())
}

async fn create_array_in_manifest(
    root: &std::path::Path,
    manifest: &mut DataManifest,
    array_name: &str,
    mut meta: DataArrayMeta,
) -> BuiltinResult<()> {
    if manifest.arrays.contains_key(array_name) {
        return Err(data_error(format!(
            "DataTransaction.create_array: array '{array_name}' already exists"
        )));
    }
    let payload = DataArrayPayload::zeros(meta.dtype.clone(), meta.shape.clone())?;
    let (payload_path, chunk_index_path) =
        write_array_payload_async(root, array_name, &payload, &meta.chunk_shape).await?;
    meta.data_path = make_rel_data_path(root, &payload_path)?;
    meta.chunk_index_path = Some(make_rel_data_path(root, &chunk_index_path)?);
    manifest.arrays.insert(array_name.to_string(), meta);
    Ok(())
}

async fn resize_array_in_manifest(
    root: &std::path::Path,
    manifest: &mut DataManifest,
    array_name: &str,
    shape: Vec<usize>,
) -> BuiltinResult<()> {
    let meta = manifest
        .arrays
        .get_mut(array_name)
        .ok_or_else(|| data_error(format!("array '{array_name}' not found")))?;
    meta.shape = shape.clone();
    let payload = DataArrayPayload::zeros(meta.dtype.clone(), shape.clone())?;
    let (payload_path, chunk_index_path) =
        write_array_payload_async(root, array_name, &payload, &meta.chunk_shape).await?;
    meta.data_path = make_rel_data_path(root, &payload_path)?;
    meta.chunk_index_path = Some(make_rel_data_path(root, &chunk_index_path)?);
    Ok(())
}

async fn fill_array_in_manifest(
    root: &std::path::Path,
    manifest: &mut DataManifest,
    array_name: &str,
    slice_spec: Option<&Value>,
    value: &Value,
) -> BuiltinResult<()> {
    let meta: DataArrayMeta = manifest
        .arrays
        .get(array_name)
        .cloned()
        .ok_or_else(|| data_error(format!("array '{array_name}' not found")))?;
    let payload = read_array_payload_async(root, &meta).await?;
    let next_payload = if let Some(slice_spec) = slice_spec {
        let ranges = parse_slice_spec(slice_spec, &payload.shape)?;
        let target_shape: Vec<usize> = ranges
            .iter()
            .map(|r| r.end.saturating_sub(r.start))
            .collect();
        let rhs = DataArrayPayload::filled(payload.dtype.clone(), target_shape, value)?
            .into_value()
            .map_err(|err| data_error(format!("DataTransaction.fill: {err}")))?;
        write_slice_payload(&payload, slice_spec, &rhs)?
    } else {
        DataArrayPayload::filled(payload.dtype.clone(), payload.shape.clone(), value)?
    };
    let (payload_path, chunk_index_path) =
        write_array_payload_async(root, array_name, &next_payload, &meta.chunk_shape).await?;
    if let Some(updated) = manifest.arrays.get_mut(array_name) {
        updated.shape = next_payload.shape.clone();
        updated.data_path = make_rel_data_path(root, &payload_path)?;
        updated.chunk_index_path = Some(make_rel_data_path(root, &chunk_index_path)?);
    }
    Ok(())
}

async fn delete_array_in_manifest_async(
    root: &std::path::Path,
    manifest: &mut DataManifest,
    array_name: &str,
) -> BuiltinResult<()> {
    let removed = manifest.arrays.remove(array_name);
    if removed.is_none() {
        return Err(data_error(format!(
            "DataTransaction.delete_array: array '{array_name}' not found"
        )));
    }
    let array_dir = root.join("arrays").join(array_name);
    if runmat_filesystem::metadata_async(&array_dir).await.is_ok() {
        runmat_filesystem::remove_dir_all_async(&array_dir)
            .await
            .map_err(|err| {
                data_error(format!(
                    "DataTransaction.delete_array: failed to remove '{}': {err}",
                    array_dir.display()
                ))
            })?;
    }
    Ok(())
}

fn parse_array_meta(array_name: &str, meta: &Value) -> BuiltinResult<DataArrayMeta> {
    let Value::Struct(meta_struct) = meta else {
        return Err(data_error(
            "DataTransaction.create_array: meta must be a struct",
        ));
    };
    let dtype = meta_struct
        .fields
        .get("dtype")
        .map(|v| parse_string(v, "DataTransaction.create_array dtype"))
        .transpose()?
        .unwrap_or_else(|| "f64".to_string());
    let shape = meta_struct
        .fields
        .get("shape")
        .map(parse_shape_from_value)
        .transpose()?
        .unwrap_or_else(|| vec![0, 0]);
    let chunk_shape = meta_struct
        .fields
        .get("chunk")
        .map(parse_shape_from_value)
        .transpose()?
        .unwrap_or_else(|| default_chunk_shape(&shape));
    validate_chunk_shape(&shape, &chunk_shape)?;
    let codec = meta_struct
        .fields
        .get("codec")
        .map(|v| parse_string(v, "DataTransaction.create_array codec"))
        .transpose()?
        .unwrap_or_else(|| "zstd".to_string());
    Ok(DataArrayMeta {
        dtype,
        shape,
        chunk_shape,
        order: "column_major".to_string(),
        codec,
        chunk_index_path: Some(format!("arrays/{array_name}/chunks/index.json")),
        data_path: format!("arrays/{array_name}/data.f64.json"),
    })
}

fn default_chunk_shape(shape: &[usize]) -> Vec<usize> {
    if shape.is_empty() {
        return Vec::new();
    }
    let mut out = shape.to_vec();
    if out.len() == 1 {
        out[0] = out[0].clamp(1, 65_536);
        return out;
    }
    out[0] = out[0].clamp(1, 256);
    out[1] = out[1].clamp(1, 256);
    for dim in out.iter_mut().skip(2) {
        *dim = (*dim).clamp(1, 8);
    }
    out
}

#[async_recursion::async_recursion(?Send)]
async fn copy_dir_recursive(src: &PathBuf, dst: &PathBuf) -> BuiltinResult<()> {
    let metadata = runmat_filesystem::metadata_async(src)
        .await
        .map_err(|err| data_error(format!("failed to stat '{}': {err}", src.display())))?;
    if !metadata.is_dir() {
        return Err(data_error(format!(
            "expected dataset directory at '{}'",
            src.display()
        )));
    }
    runmat_filesystem::create_dir_all_async(dst)
        .await
        .map_err(|err| data_error(format!("failed to create '{}': {err}", dst.display())))?;
    for entry in runmat_filesystem::read_dir_async(src)
        .await
        .map_err(|err| data_error(format!("failed to read '{}': {err}", src.display())))?
    {
        let entry_src = entry.path().to_path_buf();
        let entry_dst = dst.join(entry.file_name());
        if entry.is_dir() {
            copy_dir_recursive(&entry_src, &entry_dst).await?;
            continue;
        }
        copy_file(&entry_src, &entry_dst).await?;
    }
    Ok(())
}

fn parse_shape_from_value(value: &Value) -> BuiltinResult<Vec<usize>> {
    match value {
        Value::Tensor(t) => {
            let len = tensor_utils::tensor_element_len(t);
            let mut out = Vec::with_capacity(len);
            for index in 0..len {
                let value = t
                    .numeric_value_at(index)
                    .ok_or_else(|| data_error("shape dimensions require valid numeric storage"))?;
                out.push(numeric_dimension_to_usize(value).ok_or_else(|| {
                    data_error(
                        "shape dimensions must be non-negative finite integers within platform limits",
                    )
                })?);
            }
            Ok(out)
        }
        Value::Num(v) => floating_dimension_to_usize(*v)
            .map(|value| vec![value])
            .ok_or_else(|| {
                data_error(
                    "shape dimensions must be non-negative finite integers within platform limits",
                )
            }),
        Value::Int(v) => v
            .try_to_usize()
            .map(|n| vec![n])
            .ok_or_else(|| data_error("shape dimensions must be non-negative integers")),
        _ => Err(data_error("shape must be a numeric vector")),
    }
}

fn floating_dimension_to_usize(value: f64) -> Option<usize> {
    if !value.is_finite() || value < 0.0 || value.fract() != 0.0 {
        return None;
    }
    let integer = value as u128;
    (integer <= usize::MAX as u128 && integer as f64 == value)
        .then(|| usize::try_from(integer).ok())
        .flatten()
}

fn numeric_dimension_to_usize(value: NumericScalar) -> Option<usize> {
    match value {
        NumericScalar::F64(value) => floating_dimension_to_usize(value),
        NumericScalar::F32(value) => floating_dimension_to_usize(f64::from(value)),
        value => value
            .into_int_value()
            .and_then(|value| value.try_to_usize()),
    }
}

async fn write_array_full_async(
    dataset_path: &str,
    array_name: &str,
    slice_spec: Option<&Value>,
    value: &Value,
) -> BuiltinResult<()> {
    let root = dataset_root(dataset_path);
    let mut manifest = read_manifest_async(&root).await?;
    apply_write_to_manifest_async(&root, &mut manifest, array_name, slice_spec, value).await?;
    manifest.updated_at = now_rfc3339();
    manifest.txn_sequence = manifest.txn_sequence.saturating_add(1);
    write_manifest_async(&root, &manifest).await
}

async fn apply_write_to_manifest_async(
    root: &std::path::Path,
    manifest: &mut DataManifest,
    array_name: &str,
    slice_spec: Option<&Value>,
    value: &Value,
) -> BuiltinResult<()> {
    let meta: DataArrayMeta = manifest
        .arrays
        .get(array_name)
        .cloned()
        .ok_or_else(|| data_error(format!("array '{array_name}' not found")))?;

    if let Some(slice_spec) = slice_spec {
        if apply_slice_write_chunked_async(root, manifest, array_name, &meta, slice_spec, value)
            .await?
        {
            return Ok(());
        }
    }

    let payload = read_array_payload_async(root, &meta).await?;
    let next_payload = if let Some(slice_spec) = slice_spec {
        write_slice_payload(&payload, slice_spec, value)?
    } else {
        DataArrayPayload::from_value(payload.dtype.clone(), value)?
    };

    let (payload_path, chunk_index_path) =
        write_array_payload_async(root, array_name, &next_payload, &meta.chunk_shape).await?;
    if let Some(updated) = manifest.arrays.get_mut(array_name) {
        updated.shape = next_payload.shape.clone();
        updated.data_path = make_rel_data_path(root, &payload_path)?;
        updated.chunk_index_path = Some(make_rel_data_path(root, &chunk_index_path)?);
    }
    Ok(())
}

async fn apply_slice_write_chunked_async(
    root: &std::path::Path,
    manifest: &mut DataManifest,
    array_name: &str,
    meta: &DataArrayMeta,
    slice_spec: &Value,
    value: &Value,
) -> BuiltinResult<bool> {
    let Some(index_rel_path) = &meta.chunk_index_path else {
        return Ok(false);
    };
    let index_path = root.join(index_rel_path);
    if runmat_filesystem::metadata_async(&index_path)
        .await
        .is_err()
    {
        return Ok(false);
    }
    let ranges = parse_slice_spec(slice_spec, &meta.shape)?;
    let rhs_shape: Vec<usize> = ranges
        .iter()
        .map(|r| r.end.saturating_sub(r.start))
        .collect();
    let rhs = DataArrayPayload::from_value(meta.dtype.clone(), value)?;
    if rhs.shape != rhs_shape {
        return Err(data_error(format!(
            "SHAPE_MISMATCH: rhs shape {:?} must match target slice shape {:?}",
            rhs.shape, rhs_shape
        )));
    }

    let index_bytes = runmat_filesystem::read_async(&index_path)
        .await
        .map_err(|err| {
            data_error(format!(
                "failed to read chunk index '{}': {err}",
                index_path.display()
            ))
        })?;
    let mut chunk_index: DataChunkIndex = serde_json::from_slice(&index_bytes).map_err(|err| {
        data_error(format!(
            "failed to parse chunk index '{}': {err}",
            index_path.display()
        ))
    })?;

    let mut pos_by_key = HashMap::new();
    for (idx, entry) in chunk_index.chunks.iter().enumerate() {
        pos_by_key.insert(entry.key.clone(), idx);
    }
    let touched = touched_chunk_coords(&ranges, &meta.chunk_shape, &meta.shape);
    let mut upload_batch = Vec::<(DataChunkDescriptor, Vec<u8>)>::new();

    for coords in touched {
        let key = chunk_key(&coords);
        let chunk_start = chunk_start_for_coords(&coords, &meta.chunk_shape);
        let chunk_extent = chunk_extent_for_start(&chunk_start, &meta.chunk_shape, &meta.shape);
        let intersection = chunk_intersection(&ranges, &chunk_start, &chunk_extent);
        if intersection.is_empty() {
            continue;
        }

        let (entry_index, existed, mut entry, mut chunk_payload) = load_or_init_chunk(
            root,
            array_name,
            &key,
            &coords,
            &chunk_extent,
            &meta.dtype,
            &pos_by_key,
            &chunk_index,
        )
        .await?;
        if chunk_payload.imaginary_values.is_none() && rhs.imaginary_values.is_some() {
            chunk_payload.imaginary_values = Some(crate::data::DataArrayValues::zeros(
                &meta.dtype,
                chunk_payload.values.len(),
            )?);
        } else if chunk_payload.imaginary_values.is_some() && rhs.imaginary_values.is_none() {
            return Err(data_error(
                "CLASS_MISMATCH: real and complex data-array slices cannot be assigned to one another",
            ));
        }

        let mut local = vec![0usize; intersection.len()];
        let intersection_shape: Vec<usize> = intersection
            .iter()
            .map(|r| r.end.saturating_sub(r.start))
            .collect();
        loop {
            let mut global = Vec::with_capacity(intersection.len());
            for dim in 0..intersection.len() {
                global.push(intersection[dim].start + local[dim]);
            }
            let rhs_index: Vec<usize> = global
                .iter()
                .enumerate()
                .map(|(dim, g)| g.saturating_sub(ranges[dim].start))
                .collect();
            let chunk_index_local: Vec<usize> = global
                .iter()
                .enumerate()
                .map(|(dim, g)| g.saturating_sub(chunk_start[dim]))
                .collect();
            let rhs_linear = linear_index_column_major(&rhs_index, &rhs_shape)?;
            let chunk_linear = linear_index_column_major(&chunk_index_local, &chunk_extent)?;
            chunk_payload
                .values
                .set(chunk_linear, rhs.values.get(rhs_linear)?)?;
            if let (Some(target), Some(source)) =
                (&mut chunk_payload.imaginary_values, &rhs.imaginary_values)
            {
                target.set(chunk_linear, source.get(rhs_linear)?)?;
            }
            if !advance_index(&mut local, &intersection_shape) {
                break;
            }
        }

        let chunk_bytes = serde_json::to_vec(&chunk_payload)
            .map_err(|err| data_error(format!("failed to encode chunk payload: {err}")))?;
        let chunk_path = root.join(&entry.data_path);
        runmat_filesystem::write_async(&chunk_path, &chunk_bytes)
            .await
            .map_err(|err| {
                data_error(format!(
                    "failed to write chunk payload '{}': {err}",
                    chunk_path.display()
                ))
            })?;

        entry.coords = coords.clone();
        entry.shape = chunk_extent.clone();
        entry.bytes_raw = chunk_bytes.len() as u64;
        entry.bytes_stored = chunk_bytes.len() as u64;
        entry.hash = sha256_hex(&chunk_bytes);
        if existed {
            chunk_index.chunks[entry_index] = entry.clone();
        } else {
            chunk_index.chunks.push(entry.clone());
            pos_by_key.insert(key.clone(), chunk_index.chunks.len() - 1);
        }
        upload_batch.push((
            DataChunkDescriptor {
                key: key.clone(),
                object_id: entry.object_id.clone(),
                hash: entry.hash.clone(),
                bytes_raw: entry.bytes_raw,
                bytes_stored: entry.bytes_stored,
            },
            chunk_bytes,
        ));
    }

    maybe_upload_chunk_batch_async(root, array_name, upload_batch).await?;
    tracing::info!(
        target: "runmat.data",
        dataset = %root.display(),
        array = array_name,
        touched_chunks = chunk_index.chunks.len(),
        "chunked slice write committed"
    );
    let index_write = serde_json::to_vec(&chunk_index)
        .map_err(|err| data_error(format!("failed to encode chunk index json: {err}")))?;
    runmat_filesystem::write_async(&index_path, &index_write)
        .await
        .map_err(|err| {
            data_error(format!(
                "failed to write chunk index '{}': {err}",
                index_path.display()
            ))
        })?;

    if let Some(updated) = manifest.arrays.get_mut(array_name) {
        updated.shape = meta.shape.clone();
        updated.chunk_index_path = Some(index_rel_path.clone());
    }
    Ok(true)
}

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

fn read_slice_payload(
    payload: &DataArrayPayload,
    slice_spec: &Value,
) -> BuiltinResult<DataArrayPayload> {
    let ranges = parse_slice_spec(slice_spec, &payload.shape)?;
    let out_shape: Vec<usize> = ranges
        .iter()
        .map(|r| r.end.saturating_sub(r.start))
        .collect();
    let mut out_values = crate::data::DataArrayValues::zeros(&payload.dtype, 0)?;
    let mut out_imaginary_values = payload
        .imaginary_values
        .as_ref()
        .map(|_| crate::data::DataArrayValues::zeros(&payload.dtype, 0))
        .transpose()?;
    let mut out_index = vec![0usize; out_shape.len()];
    loop {
        let source_index: Vec<usize> = out_index
            .iter()
            .enumerate()
            .map(|(dim, idx)| ranges[dim].start + *idx)
            .collect();
        let linear = linear_index_column_major(&source_index, &payload.shape)?;
        out_values.push(payload.values.get(linear)?)?;
        if let (Some(source), Some(target)) = (&payload.imaginary_values, &mut out_imaginary_values)
        {
            target.push(source.get(linear)?)?;
        }

        if !advance_index(&mut out_index, &out_shape) {
            break;
        }
    }
    Ok(DataArrayPayload {
        dtype: payload.dtype.clone(),
        shape: out_shape,
        values: out_values,
        imaginary_values: out_imaginary_values,
    })
}

fn write_slice_payload(
    payload: &DataArrayPayload,
    slice_spec: &Value,
    rhs: &Value,
) -> BuiltinResult<DataArrayPayload> {
    let ranges = parse_slice_spec(slice_spec, &payload.shape)?;
    let target_shape: Vec<usize> = ranges
        .iter()
        .map(|r| r.end.saturating_sub(r.start))
        .collect();
    let rhs = DataArrayPayload::from_value(payload.dtype.clone(), rhs)?;
    if rhs.shape != target_shape {
        return Err(data_error(format!(
            "SHAPE_MISMATCH: rhs shape {:?} must match target slice shape {:?}",
            rhs.shape, target_shape
        )));
    }

    let mut next = payload.values.clone();
    if payload.imaginary_values.is_some() != rhs.imaginary_values.is_some() {
        return Err(data_error(
            "CLASS_MISMATCH: real and complex data-array slices cannot be assigned to one another",
        ));
    }
    let mut next_imaginary = payload.imaginary_values.clone();
    let mut rhs_index = vec![0usize; target_shape.len()];
    let mut rhs_linear = 0usize;
    loop {
        let target_index: Vec<usize> = rhs_index
            .iter()
            .enumerate()
            .map(|(dim, idx)| ranges[dim].start + *idx)
            .collect();
        let target_linear = linear_index_column_major(&target_index, &payload.shape)?;
        next.set(target_linear, rhs.values.get(rhs_linear)?)?;
        if let (Some(target), Some(source)) = (&mut next_imaginary, &rhs.imaginary_values) {
            target.set(target_linear, source.get(rhs_linear)?)?;
        }
        rhs_linear += 1;

        if !advance_index(&mut rhs_index, &target_shape) {
            break;
        }
    }

    Ok(DataArrayPayload {
        dtype: payload.dtype.clone(),
        shape: payload.shape.clone(),
        values: next,
        imaginary_values: next_imaginary,
    })
}

fn parse_slice_spec(slice_spec: &Value, shape: &[usize]) -> BuiltinResult<Vec<DimRange>> {
    match slice_spec {
        Value::Cell(cell) => {
            if cell.data.is_empty() {
                return Err(data_error("INVALID_SLICE: empty slice specification"));
            }
            let mut ranges = Vec::with_capacity(shape.len());
            for (dim, extent) in shape.iter().enumerate() {
                if let Some(item) = cell.data.get(dim) {
                    ranges.push(parse_dim_range(item, *extent)?);
                } else {
                    ranges.push(DimRange {
                        start: 0,
                        end: *extent,
                    });
                }
            }
            Ok(ranges)
        }
        Value::String(s) if s == ":" => Ok(shape
            .iter()
            .map(|extent| DimRange {
                start: 0,
                end: *extent,
            })
            .collect()),
        _ => Err(data_error(
            "INVALID_SLICE: slice must be a cell spec like {1:10, :} or ':'",
        )),
    }
}

fn parse_dim_range(value: &Value, extent: usize) -> BuiltinResult<DimRange> {
    if extent == 0 {
        return Ok(DimRange { start: 0, end: 0 });
    }
    match value {
        Value::String(s) if s == ":" => Ok(DimRange {
            start: 0,
            end: extent,
        }),
        Value::Num(n) => {
            let idx = floating_dimension_to_usize(*n)
                .and_then(|index| index.checked_sub(1))
                .ok_or_else(|| data_error("INVALID_SLICE: index out of bounds"))?;
            if idx >= extent {
                return Err(data_error("INVALID_SLICE: index out of bounds"));
            }
            Ok(DimRange {
                start: idx,
                end: idx + 1,
            })
        }
        Value::Int(i) => {
            let Some(idx) = i.try_to_usize().and_then(|index| index.checked_sub(1)) else {
                return Err(data_error("INVALID_SLICE: index out of bounds"));
            };
            if idx >= extent {
                return Err(data_error("INVALID_SLICE: index out of bounds"));
            }
            Ok(DimRange {
                start: idx,
                end: idx + 1,
            })
        }
        Value::Tensor(t) if tensor_utils::tensor_element_len(t) == 2 => {
            let start = t
                .numeric_value_at(0)
                .and_then(numeric_dimension_to_usize)
                .and_then(|index| index.checked_sub(1))
                .ok_or_else(|| data_error("INVALID_SLICE: range out of bounds"))?;
            let end_inclusive = t
                .numeric_value_at(1)
                .and_then(numeric_dimension_to_usize)
                .and_then(|index| index.checked_sub(1))
                .ok_or_else(|| data_error("INVALID_SLICE: range out of bounds"))?;
            if end_inclusive < start || end_inclusive >= extent {
                return Err(data_error("INVALID_SLICE: range out of bounds"));
            }
            Ok(DimRange {
                start,
                end: end_inclusive + 1,
            })
        }
        _ => Err(data_error(
            "INVALID_SLICE: dimension must be ':', scalar index, or [start end] range",
        )),
    }
}

fn linear_index_column_major(index: &[usize], shape: &[usize]) -> BuiltinResult<usize> {
    if index.len() != shape.len() {
        return Err(data_error("INVALID_SLICE: rank mismatch"));
    }
    let mut stride = 1usize;
    let mut linear = 0usize;
    for (idx, extent) in index.iter().zip(shape.iter()) {
        if *idx >= *extent {
            return Err(data_error("INVALID_SLICE: index out of bounds"));
        }
        linear += idx * stride;
        stride = stride.saturating_mul(*extent);
    }
    Ok(linear)
}

fn advance_index(index: &mut [usize], shape: &[usize]) -> bool {
    if shape.is_empty() {
        return false;
    }
    for dim in 0..shape.len() {
        index[dim] += 1;
        if index[dim] < shape[dim] {
            return true;
        }
        index[dim] = 0;
    }
    false
}

fn chunk_key(coords: &[usize]) -> String {
    coords
        .iter()
        .map(|v| v.to_string())
        .collect::<Vec<_>>()
        .join(".")
}

fn chunk_start_for_coords(coords: &[usize], chunk_shape: &[usize]) -> Vec<usize> {
    coords
        .iter()
        .enumerate()
        .map(|(dim, coord)| coord * chunk_shape.get(dim).copied().unwrap_or(1).max(1))
        .collect()
}

fn chunk_extent_for_start(start: &[usize], chunk_shape: &[usize], shape: &[usize]) -> Vec<usize> {
    start
        .iter()
        .enumerate()
        .map(|(dim, start)| {
            let chunk = chunk_shape.get(dim).copied().unwrap_or(1).max(1);
            let end = (*start + chunk).min(shape[dim]);
            end.saturating_sub(*start)
        })
        .collect()
}

fn chunk_intersection(
    ranges: &[DimRange],
    chunk_start: &[usize],
    chunk_extent: &[usize],
) -> Vec<DimRange> {
    let mut out = Vec::with_capacity(ranges.len());
    for dim in 0..ranges.len() {
        let c_start = chunk_start[dim];
        let c_end = c_start + chunk_extent[dim];
        let start = ranges[dim].start.max(c_start);
        let end = ranges[dim].end.min(c_end);
        if start >= end {
            return Vec::new();
        }
        out.push(DimRange { start, end });
    }
    out
}

fn touched_chunk_coords(
    ranges: &[DimRange],
    chunk_shape: &[usize],
    shape: &[usize],
) -> Vec<Vec<usize>> {
    let mut span = Vec::with_capacity(ranges.len());
    let mut begin = Vec::with_capacity(ranges.len());
    for dim in 0..ranges.len() {
        if shape[dim] == 0 {
            return Vec::new();
        }
        let chunk = chunk_shape.get(dim).copied().unwrap_or(1).max(1);
        let first = ranges[dim].start / chunk;
        let last = (ranges[dim].end.saturating_sub(1)) / chunk;
        begin.push(first);
        span.push(last.saturating_sub(first) + 1);
    }
    let mut local = vec![0usize; span.len()];
    let mut out = Vec::new();
    loop {
        out.push(
            local
                .iter()
                .enumerate()
                .map(|(dim, v)| begin[dim] + *v)
                .collect::<Vec<_>>(),
        );
        if !advance_index(&mut local, &span) {
            break;
        }
    }
    out
}

async fn maybe_upload_chunk_batch_async(
    root: &std::path::Path,
    array_name: &str,
    batch: Vec<(DataChunkDescriptor, Vec<u8>)>,
) -> BuiltinResult<()> {
    if batch.is_empty() {
        return Ok(());
    }
    let request = DataChunkUploadRequest {
        dataset_path: root.to_string_lossy().to_string(),
        array: array_name.to_string(),
        chunks: batch.iter().map(|(d, _)| d.clone()).collect(),
    };
    let targets = match runmat_filesystem::data_chunk_upload_targets_async(&request).await {
        Ok(targets) => targets,
        Err(err) if err.kind() == std::io::ErrorKind::Unsupported => return Ok(()),
        Err(err) => {
            return Err(data_error(format!(
                "failed to request data chunk upload targets: {err}"
            )))
        }
    };
    for (descriptor, bytes) in batch {
        let target = targets
            .iter()
            .find(|t| t.key == descriptor.key)
            .ok_or_else(|| {
                data_error(format!(
                    "missing upload target for chunk '{}'",
                    descriptor.key
                ))
            })?;
        runmat_filesystem::data_upload_chunk_async(target, &bytes)
            .await
            .map_err(|err| {
                data_error(format!(
                    "failed to upload chunk '{}': {err}",
                    descriptor.key
                ))
            })?;
        tracing::info!(
            target: "runmat.data",
            dataset = %root.display(),
            array = array_name,
            chunk_key = descriptor.key,
            bytes = bytes.len(),
            "chunk upload completed"
        );
    }
    Ok(())
}

fn chunk_rel_path(array_name: &str, object_id: &str) -> String {
    format!("arrays/{array_name}/chunks/{object_id}.json")
}

async fn load_or_init_chunk(
    root: &std::path::Path,
    array_name: &str,
    key: &str,
    coords: &[usize],
    chunk_extent: &[usize],
    dtype: &str,
    pos_by_key: &HashMap<String, usize>,
    chunk_index: &DataChunkIndex,
) -> BuiltinResult<(usize, bool, DataChunkIndexEntry, DataArrayPayload)> {
    if let Some(index) = pos_by_key.get(key).copied() {
        let entry = chunk_index
            .chunks
            .get(index)
            .cloned()
            .ok_or_else(|| data_error(format!("chunk index missing key '{key}'")))?;
        let bytes = runmat_filesystem::read_async(root.join(&entry.data_path))
            .await
            .map_err(|err| {
                data_error(format!(
                    "failed to read chunk payload '{}': {err}",
                    entry.data_path
                ))
            })?;
        let payload: DataArrayPayload = serde_json::from_slice::<DataArrayPayload>(&bytes)
            .map_err(|err| {
                data_error(format!(
                    "failed to parse chunk payload '{}': {err}",
                    entry.data_path
                ))
            })?
            .normalize_for_dtype(dtype)?;
        return Ok((index, true, entry, payload));
    }

    let object_id = format!("obj_{}", key.replace('.', "_"));
    let entry = DataChunkIndexEntry {
        key: key.to_string(),
        object_id: object_id.clone(),
        hash: String::new(),
        bytes_raw: 0,
        bytes_stored: 0,
        coords: coords.to_vec(),
        shape: chunk_extent.to_vec(),
        data_path: chunk_rel_path(array_name, &object_id),
    };
    let payload = DataArrayPayload::zeros(dtype.to_string(), chunk_extent.to_vec())?;
    Ok((chunk_index.chunks.len(), false, entry, payload))
}

fn attrs_to_struct(attrs: &BTreeMap<String, serde_json::Value>) -> Value {
    let mut out = StructValue::new();
    for (k, v) in attrs {
        out.fields.insert(k.clone(), json_to_value(v));
    }
    Value::Struct(out)
}

fn value_to_json(value: &Value) -> serde_json::Value {
    match value {
        Value::String(s) => serde_json::Value::String(s.clone()),
        Value::CharArray(chars) => serde_json::Value::String(chars.data.iter().collect::<String>()),
        Value::Num(n) => serde_json::json!(n),
        Value::Int(i) => int_value_to_json(i),
        Value::Bool(b) => serde_json::json!(b),
        _ => serde_json::Value::String(format!("{value:?}")),
    }
}

fn json_to_value(value: &serde_json::Value) -> Value {
    match value {
        serde_json::Value::Bool(b) => Value::Bool(*b),
        serde_json::Value::Number(n) => {
            if let Some(value) = n.as_i64() {
                Value::Int(IntValue::I64(value))
            } else if let Some(value) = n.as_u64() {
                Value::Int(IntValue::U64(value))
            } else {
                Value::Num(n.as_f64().unwrap_or_default())
            }
        }
        serde_json::Value::String(s) => Value::String(s.clone()),
        serde_json::Value::Array(arr) => {
            let vals = arr.iter().map(json_to_value).collect::<Vec<_>>();
            crate::make_cell(vals.clone(), 1, vals.len())
                .unwrap_or_else(|_| Value::String("<invalid-array>".to_string()))
        }
        serde_json::Value::Object(map) => {
            let mut s = StructValue::new();
            for (k, v) in map {
                s.fields.insert(k.clone(), json_to_value(v));
            }
            Value::Struct(s)
        }
        serde_json::Value::Null => Value::String("".to_string()),
    }
}

#[cfg(all(test, not(target_arch = "wasm32")))]
mod tests {
    use super::*;
    use runmat_value::IntValue;
    use runmat_value::{ComplexTensor, IntegerComplexStorage};

    fn tensor_values(tensor: &Tensor) -> Vec<f64> {
        tensor.materialize_f64()
    }

    #[test]
    fn typed_data_slice_indices_do_not_saturate_through_i64() {
        let extent = usize::MAX;
        let result = parse_dim_range(&Value::Int(IntValue::U64(u64::MAX)), extent);
        match usize::try_from(u64::MAX)
            .ok()
            .and_then(|value| value.checked_sub(1))
        {
            Some(index) if index < extent => {
                let range = result.expect("index");
                assert_eq!(range.start, index);
                assert_eq!(range.end, index + 1);
            }
            _ => assert!(result.is_err()),
        }
        assert!(parse_dim_range(&Value::Int(IntValue::I64(-1)), extent).is_err());
    }

    #[test]
    fn data_shape_and_slice_parsers_read_typed_integer_storage_exactly() {
        let cases = [
            runmat_value::IntegerStorage::I8(vec![2, 3]),
            runmat_value::IntegerStorage::I16(vec![2, 3]),
            runmat_value::IntegerStorage::I32(vec![2, 3]),
            runmat_value::IntegerStorage::I64(vec![2, 3]),
            runmat_value::IntegerStorage::U8(vec![2, 3]),
            runmat_value::IntegerStorage::U16(vec![2, 3]),
            runmat_value::IntegerStorage::U32(vec![2, 3]),
            runmat_value::IntegerStorage::U64(vec![2, 3]),
        ];
        for storage in cases {
            let shape = Tensor::new_integer(storage.clone(), vec![1, 2]).expect("shape");
            assert_eq!(
                parse_shape_from_value(&Value::Tensor(shape)).expect("shape"),
                vec![2, 3]
            );

            let range = Tensor::new_integer(storage, vec![1, 2]).expect("range");
            let parsed = parse_dim_range(&Value::Tensor(range), 5).expect("range");
            assert_eq!(parsed.start, 1);
            assert_eq!(parsed.end, 3);
        }

        let negative = Tensor::new_integer(runmat_value::IntegerStorage::I16(vec![-1]), vec![1, 1])
            .expect("shape");
        assert!(parse_shape_from_value(&Value::Tensor(negative)).is_err());

        let invalid = Tensor::new_integer(
            runmat_value::IntegerStorage::U64(vec![u64::MAX, u64::MAX]),
            vec![1, 2],
        )
        .expect("range");
        assert!(parse_dim_range(&Value::Tensor(invalid), 5).is_err());
    }

    #[test]
    fn data_shape_and_slice_floating_parsers_reject_fractional_and_unrepresentable_bounds() {
        assert!(parse_shape_from_value(&Value::Num(1.5)).is_err());
        let boundary_shape = parse_shape_from_value(&Value::Num(usize::MAX as f64));
        if usize::BITS == 64 {
            assert!(boundary_shape.is_err());
        } else {
            assert_eq!(boundary_shape.unwrap(), vec![usize::MAX]);
        }
        assert!(parse_shape_from_value(&Value::Num((usize::MAX as f64) + 1.0)).is_err());

        let fractional_shape = Tensor::new(vec![2.0, 3.25], vec![1, 2]).expect("shape");
        assert!(parse_shape_from_value(&Value::Tensor(fractional_shape)).is_err());
        let single_shape = Tensor::from_f32(vec![2.0, 3.0], vec![1, 2]).expect("single shape");
        assert_eq!(
            parse_shape_from_value(&Value::Tensor(single_shape)).expect("single shape"),
            vec![2, 3]
        );
        let fractional_single_shape =
            Tensor::from_f32(vec![2.0, 3.25], vec![1, 2]).expect("fractional single shape");
        assert!(parse_shape_from_value(&Value::Tensor(fractional_single_shape)).is_err());

        assert!(parse_dim_range(&Value::Num(1.5), 5).is_err());
        let boundary_range = parse_dim_range(&Value::Num(usize::MAX as f64), usize::MAX);
        if usize::BITS == 64 {
            assert!(boundary_range.is_err());
        } else {
            let parsed = boundary_range.expect("32-bit boundary");
            assert_eq!(parsed.start, usize::MAX - 1);
            assert_eq!(parsed.end, usize::MAX);
        }

        let fractional_range = Tensor::new(vec![1.0, 3.5], vec![1, 2]).expect("range");
        assert!(parse_dim_range(&Value::Tensor(fractional_range), 5).is_err());
        let single_range = Tensor::from_f32(vec![2.0, 3.0], vec![1, 2]).expect("single range");
        let parsed = parse_dim_range(&Value::Tensor(single_range), 5).expect("single range");
        assert_eq!((parsed.start, parsed.end), (1, 3));

        let too_large_range = Tensor::new(vec![1.0, usize::MAX as f64], vec![1, 2]).expect("range");
        let tensor_boundary = parse_dim_range(&Value::Tensor(too_large_range), usize::MAX);
        if usize::BITS == 64 {
            assert!(tensor_boundary.is_err());
        } else {
            let parsed = tensor_boundary.expect("32-bit tensor boundary");
            assert_eq!(parsed.start, 0);
            assert_eq!(parsed.end, usize::MAX);
        }
    }
    use crate::dispatcher::call_builtin;
    use async_trait::async_trait;
    use axum::extract::{Query, State};
    use axum::http::{HeaderMap, StatusCode};
    use axum::routing::{post, put};
    use axum::{Json, Router};
    use runmat_filesystem::data_contract::{
        DataChunkUploadRequest, DataChunkUploadTarget, DataManifestDescriptor, DataManifestRequest,
    };
    use runmat_filesystem::{
        DirEntry, FileHandle, FsMetadata, FsProvider, NativeFsProvider, OpenFlags,
    };
    use runmat_value::{CellArray, IntegerStorage};
    use serde::Deserialize;
    use std::path::Path;
    use std::sync::{Arc, Mutex, MutexGuard};
    use tokio::runtime::Runtime;
    use tokio::sync::oneshot;

    fn serial_test_guard() -> MutexGuard<'static, ()> {
        runmat_filesystem::provider_override_lock()
    }

    fn native_provider_guard() -> runmat_filesystem::ProviderGuard {
        runmat_filesystem::replace_provider(Arc::new(NativeFsProvider))
    }

    #[test]
    fn attribute_json_preserves_native_integer_text() {
        for value in [
            runmat_value::IntValue::I64(i64::MIN),
            runmat_value::IntValue::U64(u64::MAX),
        ] {
            let json = value_to_json(&Value::Int(value.clone()));
            assert_eq!(json.to_string(), value.decimal_string());
        }
    }

    #[test]
    fn attribute_json_roundtrips_wide_integer_values_without_f64() {
        for value in [IntValue::I64(i64::MIN), IntValue::U64(u64::MAX)] {
            let json = value_to_json(&Value::Int(value.clone()));
            assert_eq!(json_to_value(&json), Value::Int(value));
        }
    }

    #[test]
    fn io_data_descriptors_cover_constructor_and_transaction_surface() {
        let data_labels: Vec<&str> = DATA_CREATE_DESCRIPTOR
            .signatures
            .iter()
            .map(|sig| sig.label)
            .collect();
        assert!(data_labels.contains(&"ds = data.create(path, schema)"));

        let read_labels: Vec<&str> = DATAARRAY_READ_DESCRIPTOR
            .signatures
            .iter()
            .map(|sig| sig.label)
            .collect();
        assert!(read_labels.contains(&"X = DataArray.read(arr, sliceSpec)"));

        let write_labels: Vec<&str> = DATAARRAY_WRITE_DESCRIPTOR
            .signatures
            .iter()
            .map(|sig| sig.label)
            .collect();
        assert!(write_labels.contains(&"tf = DataArray.write(arr, values)"));
        assert!(write_labels.contains(&"tf = DataArray.write(arr, sliceSpec, values)"));

        let tx_labels: Vec<&str> = DATATX_COMMIT_DESCRIPTOR
            .signatures
            .iter()
            .map(|sig| sig.label)
            .collect();
        assert!(tx_labels.contains(&"tf = DataTransaction.commit(tx, options)"));

        assert_eq!(DATA_CREATE_INTEGER_CAPABILITIES.len(), 1);
        assert_eq!(DATAARRAY_READ_INTEGER_CAPABILITIES.len(), 1);
        assert_eq!(DATAARRAY_WRITE_INTEGER_CAPABILITIES.len(), 1);
        assert_eq!(DATAARRAY_RESIZE_INTEGER_CAPABILITIES.len(), 1);
        assert_eq!(DATAARRAY_FILL_INTEGER_CAPABILITIES.len(), 1);
        assert_eq!(DATASET_ATTRS_INTEGER_CAPABILITIES.len(), 1);
        assert_eq!(DATASET_GET_ATTR_INTEGER_CAPABILITIES.len(), 2);
        assert_eq!(DATASET_SET_ATTR_INTEGER_CAPABILITIES.len(), 1);
        assert_eq!(DATASET_SET_ATTRS_INTEGER_CAPABILITIES.len(), 1);
        assert_eq!(DATATX_WRITE_INTEGER_CAPABILITIES.len(), 1);
        assert_eq!(DATATX_SET_ATTR_INTEGER_CAPABILITIES.len(), 1);
        assert_eq!(DATATX_SET_ATTRS_INTEGER_CAPABILITIES.len(), 1);
        assert_eq!(DATATX_RESIZE_INTEGER_CAPABILITIES.len(), 1);
        assert_eq!(DATATX_FILL_INTEGER_CAPABILITIES.len(), 1);
        assert_eq!(DATATX_CREATE_ARRAY_INTEGER_CAPABILITIES.len(), 1);
        for capability in [
            &DATA_CREATE_INTEGER_CAPABILITIES[0],
            &DATAARRAY_READ_INTEGER_CAPABILITIES[0],
            &DATAARRAY_WRITE_INTEGER_CAPABILITIES[0],
            &DATAARRAY_RESIZE_INTEGER_CAPABILITIES[0],
            &DATAARRAY_FILL_INTEGER_CAPABILITIES[0],
            &DATASET_ATTRS_INTEGER_CAPABILITIES[0],
            &DATASET_GET_ATTR_INTEGER_CAPABILITIES[0],
            &DATASET_GET_ATTR_INTEGER_CAPABILITIES[1],
            &DATASET_SET_ATTR_INTEGER_CAPABILITIES[0],
            &DATASET_SET_ATTRS_INTEGER_CAPABILITIES[0],
            &DATATX_WRITE_INTEGER_CAPABILITIES[0],
            &DATATX_SET_ATTR_INTEGER_CAPABILITIES[0],
            &DATATX_SET_ATTRS_INTEGER_CAPABILITIES[0],
            &DATATX_RESIZE_INTEGER_CAPABILITIES[0],
            &DATATX_FILL_INTEGER_CAPABILITIES[0],
            &DATATX_CREATE_ARRAY_INTEGER_CAPABILITIES[0],
        ] {
            assert!(capability.inputs.iter().all(|input| input.classes
                == crate::builtins::common::integer_capability::ALL_INTEGER_CLASSES));
        }
    }

    #[test]
    fn data_create_uses_exact_all_class_schema_controls_and_integer_payloads() {
        let _serial = serial_test_guard();
        let _provider = native_provider_guard();
        let cases = vec![
            (
                "int8",
                IntegerStorage::I8(vec![2, 3]),
                IntegerStorage::I8(vec![1, 3]),
                IntegerStorage::I8(vec![0; 6]),
            ),
            (
                "int16",
                IntegerStorage::I16(vec![2, 3]),
                IntegerStorage::I16(vec![1, 3]),
                IntegerStorage::I16(vec![0; 6]),
            ),
            (
                "int32",
                IntegerStorage::I32(vec![2, 3]),
                IntegerStorage::I32(vec![1, 3]),
                IntegerStorage::I32(vec![0; 6]),
            ),
            (
                "int64",
                IntegerStorage::I64(vec![2, 3]),
                IntegerStorage::I64(vec![1, 3]),
                IntegerStorage::I64(vec![0; 6]),
            ),
            (
                "uint8",
                IntegerStorage::U8(vec![2, 3]),
                IntegerStorage::U8(vec![1, 3]),
                IntegerStorage::U8(vec![0; 6]),
            ),
            (
                "uint16",
                IntegerStorage::U16(vec![2, 3]),
                IntegerStorage::U16(vec![1, 3]),
                IntegerStorage::U16(vec![0; 6]),
            ),
            (
                "uint32",
                IntegerStorage::U32(vec![2, 3]),
                IntegerStorage::U32(vec![1, 3]),
                IntegerStorage::U32(vec![0; 6]),
            ),
            (
                "uint64",
                IntegerStorage::U64(vec![2, 3]),
                IntegerStorage::U64(vec![1, 3]),
                IntegerStorage::U64(vec![0; 6]),
            ),
        ];

        for (dtype, shape, chunk, expected) in cases {
            let dir = tempfile::tempdir().expect("tempdir");
            let path = dir
                .path()
                .join(format!("schema-{dtype}.data"))
                .to_string_lossy()
                .to_string();
            let mut meta = StructValue::new();
            meta.fields
                .insert("dtype".to_string(), Value::String(dtype.to_string()));
            meta.fields.insert(
                "shape".to_string(),
                Value::Tensor(Tensor::new_integer(shape, vec![1, 2]).expect("shape")),
            );
            meta.fields.insert(
                "chunk".to_string(),
                Value::Tensor(Tensor::new_integer(chunk, vec![1, 2]).expect("chunk")),
            );
            let mut arrays = StructValue::new();
            arrays
                .fields
                .insert("samples".to_string(), Value::Struct(meta));
            let mut schema = StructValue::new();
            schema
                .fields
                .insert("arrays".to_string(), Value::Struct(arrays));

            let dataset =
                call_builtin("data.create", &[Value::String(path), Value::Struct(schema)])
                    .unwrap_or_else(|error| panic!("{dtype}: create failed: {error}"));
            let array = call_builtin(
                "Dataset.array",
                &[dataset, Value::String("samples".to_string())],
            )
            .expect("array handle");
            let Value::Tensor(values) =
                call_builtin("DataArray.read", &[array]).expect("read zeros")
            else {
                panic!("{dtype}: expected tensor");
            };
            assert_eq!(values.shape, vec![2, 3], "{dtype}");
            assert_eq!(values.integer_storage(), Some(&expected), "{dtype}");
        }
    }

    #[test]
    fn data_namespace_lifecycle_rejects_integer_arguments_and_unsupported_extras() {
        let integer_values = [
            IntValue::I8(1),
            IntValue::I16(1),
            IntValue::I32(1),
            IntValue::I64(1),
            IntValue::U8(1),
            IntValue::U16(1),
            IntValue::U32(1),
            IntValue::U64(1),
        ];
        for integer in integer_values {
            let value = Value::Int(integer);
            for (name, args) in [
                (
                    "data.create",
                    vec![value.clone(), Value::Struct(StructValue::new())],
                ),
                ("data.open", vec![value.clone()]),
                ("data.exists", vec![value.clone()]),
                ("data.delete", vec![value.clone()]),
                ("data.inspect", vec![value.clone()]),
                ("data.list", vec![value.clone()]),
                (
                    "data.copy",
                    vec![value.clone(), Value::String("to.data".to_string())],
                ),
                (
                    "data.copy",
                    vec![Value::String("from.data".to_string()), value.clone()],
                ),
                (
                    "data.move",
                    vec![value.clone(), Value::String("to.data".to_string())],
                ),
                (
                    "data.move",
                    vec![Value::String("from.data".to_string()), value.clone()],
                ),
                (
                    "data.import",
                    vec![
                        value.clone(),
                        Value::String("data".to_string()),
                        Value::String("source.data".to_string()),
                    ],
                ),
                (
                    "data.import",
                    vec![
                        Value::String("path.data".to_string()),
                        value.clone(),
                        Value::String("source.data".to_string()),
                    ],
                ),
                (
                    "data.import",
                    vec![
                        Value::String("path.data".to_string()),
                        Value::String("data".to_string()),
                        value.clone(),
                    ],
                ),
                (
                    "data.export",
                    vec![
                        value.clone(),
                        Value::String("data".to_string()),
                        Value::String("target.data".to_string()),
                    ],
                ),
                (
                    "data.export",
                    vec![
                        Value::String("path.data".to_string()),
                        value.clone(),
                        Value::String("target.data".to_string()),
                    ],
                ),
                (
                    "data.export",
                    vec![
                        Value::String("path.data".to_string()),
                        Value::String("data".to_string()),
                        value.clone(),
                    ],
                ),
                ("commit", vec![value]),
            ] {
                let error = call_builtin(name, &args).expect_err("expected integer rejection");
                assert!(!error.message().is_empty(), "{name}");
            }
        }

        let error = call_builtin(
            "data.exists",
            &[
                Value::String("missing.data".to_string()),
                Value::Int(IntValue::U8(1)),
            ],
        )
        .expect_err("fixed documented arity must reject extras");
        assert!(!error.message().is_empty());

        let error = call_builtin(
            "DataTransaction.commit",
            &[Value::Int(IntValue::U8(1)), Value::Int(IntValue::U8(1))],
        )
        .expect_err("commit options must be a struct");
        assert!(error.message().contains("options must be a scalar struct"));

        let error = call_builtin(
            "DataTransaction.commit",
            &[
                Value::Int(IntValue::U8(1)),
                Value::Struct(StructValue::new()),
                Value::Struct(StructValue::new()),
            ],
        )
        .expect_err("commit must reject more than one options struct");
        assert!(error.message().contains("at most one options struct"));
    }

    #[test]
    fn dataset_nonattribute_methods_reject_integer_receivers() {
        for (name, extra) in [
            ("Dataset.array", Some(Value::String("samples".to_string()))),
            ("Dataset.arrays", None),
            ("Dataset.begin", None),
            (
                "Dataset.has_array",
                Some(Value::String("samples".to_string())),
            ),
            ("Dataset.id", None),
            ("Dataset.path", None),
            ("Dataset.refresh", None),
            (
                "Dataset.snapshot",
                Some(Value::String("checkpoint".to_string())),
            ),
            ("Dataset.version", None),
        ] {
            let mut args = vec![Value::Int(IntValue::U64(u64::MAX))];
            args.extend(extra);
            let error = call_builtin(name, &args).expect_err("method requires a Dataset receiver");
            assert!(
                error.message().contains("expected object"),
                "{name}: {error}"
            );
        }
    }

    #[test]
    fn dataset_attributes_keep_exact_integer_values_and_documented_class_boundaries() {
        let _serial = serial_test_guard();
        let _provider = native_provider_guard();
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir
            .path()
            .join("integer-attrs.data")
            .to_string_lossy()
            .to_string();
        let mut array_meta = StructValue::new();
        array_meta
            .fields
            .insert("dtype".to_string(), Value::String("f64".to_string()));
        array_meta.fields.insert(
            "shape".to_string(),
            Value::Tensor(Tensor::new(vec![1.0, 1.0], vec![1, 2]).expect("shape")),
        );
        let mut arrays = StructValue::new();
        arrays
            .fields
            .insert("base".to_string(), Value::Struct(array_meta));
        let mut schema = StructValue::new();
        schema
            .fields
            .insert("arrays".to_string(), Value::Struct(arrays));
        let ds = call_builtin("data.create", &[Value::String(path), Value::Struct(schema)])
            .expect("create dataset");
        let cases = [
            ("int8", IntValue::I8(i8::MIN), IntValue::I64(i8::MIN as i64)),
            (
                "int16",
                IntValue::I16(i16::MIN),
                IntValue::I64(i16::MIN as i64),
            ),
            (
                "int32",
                IntValue::I32(i32::MIN),
                IntValue::I64(i32::MIN as i64),
            ),
            ("int64", IntValue::I64(i64::MIN), IntValue::I64(i64::MIN)),
            (
                "uint8",
                IntValue::U8(u8::MAX),
                IntValue::I64(u8::MAX as i64),
            ),
            (
                "uint16",
                IntValue::U16(u16::MAX),
                IntValue::I64(u16::MAX as i64),
            ),
            (
                "uint32",
                IntValue::U32(u32::MAX),
                IntValue::I64(u32::MAX as i64),
            ),
            ("uint64", IntValue::U64(u64::MAX), IntValue::U64(u64::MAX)),
        ];

        let mut bulk = StructValue::new();
        for (name, source, _) in &cases {
            call_builtin(
                "Dataset.set_attr",
                &[
                    ds.clone(),
                    Value::String(format!("direct_{name}")),
                    Value::Int(source.clone()),
                ],
            )
            .expect("set integer attribute");
            bulk.fields
                .insert(format!("bulk_{name}"), Value::Int(source.clone()));
        }
        call_builtin("Dataset.set_attrs", &[ds.clone(), Value::Struct(bulk)])
            .expect("set integer attributes");

        let Value::Struct(attrs) =
            call_builtin("Dataset.attrs", std::slice::from_ref(&ds)).expect("read attributes")
        else {
            panic!("expected attribute struct");
        };
        for (name, source, canonical) in cases {
            let expected = Value::Int(canonical);
            assert_eq!(
                call_builtin(
                    "Dataset.get_attr",
                    &[ds.clone(), Value::String(format!("direct_{name}"))],
                )
                .expect("get integer attribute"),
                expected,
                "{name} direct"
            );
            assert_eq!(
                attrs.fields.get(&format!("bulk_{name}")),
                Some(&expected),
                "{name} bulk"
            );
            assert_eq!(
                call_builtin(
                    "Dataset.get_attr",
                    &[
                        ds.clone(),
                        Value::String(format!("missing_{name}")),
                        Value::Int(source.clone()),
                    ],
                )
                .expect("get integer default"),
                Value::Int(source),
                "{name} default"
            );
        }
    }

    #[test]
    fn data_transaction_lifecycle_methods_reject_integer_receivers() {
        for (name, extra) in [
            ("DataTransaction.abort", None),
            ("DataTransaction.commit", None),
            (
                "DataTransaction.delete_array",
                Some(Value::String("samples".to_string())),
            ),
            ("DataTransaction.id", None),
            ("DataTransaction.status", None),
        ] {
            let mut args = vec![Value::Int(IntValue::U64(u64::MAX))];
            args.extend(extra);
            let error = call_builtin(name, &args)
                .expect_err("lifecycle method requires a DataTransaction receiver");
            assert!(
                error.message().contains("expected object"),
                "{name}: {error}"
            );
        }
    }

    #[test]
    fn data_array_metadata_accessors_reject_integer_receivers() {
        for name in [
            "DataArray.chunk_shape",
            "DataArray.codec",
            "DataArray.dtype",
            "DataArray.name",
            "DataArray.rank",
            "DataArray.shape",
        ] {
            let error = call_builtin(name, &[Value::Int(IntValue::U64(u64::MAX))])
                .expect_err("metadata accessor requires a DataArray receiver");
            assert!(
                error.message().contains("expected object"),
                "{name}: {error}"
            );
        }
    }

    #[derive(Default)]
    struct CountingDataUploadProvider {
        inner: NativeFsProvider,
        uploaded_keys: Arc<Mutex<Vec<String>>>,
    }

    struct HttpDataUploadProvider {
        inner: NativeFsProvider,
        base_url: String,
        client: reqwest::blocking::Client,
    }

    impl HttpDataUploadProvider {
        fn new(base_url: String) -> Self {
            Self {
                inner: NativeFsProvider,
                base_url,
                client: reqwest::blocking::Client::new(),
            }
        }
    }

    #[async_trait(?Send)]
    impl FsProvider for HttpDataUploadProvider {
        fn open(&self, path: &Path, flags: &OpenFlags) -> std::io::Result<Box<dyn FileHandle>> {
            self.inner.open(path, flags)
        }

        async fn read(&self, path: &Path) -> std::io::Result<Vec<u8>> {
            self.inner.read(path).await
        }

        async fn write(&self, path: &Path, data: &[u8]) -> std::io::Result<()> {
            self.inner.write(path, data).await
        }

        async fn remove_file(&self, path: &Path) -> std::io::Result<()> {
            self.inner.remove_file(path).await
        }

        async fn metadata(&self, path: &Path) -> std::io::Result<FsMetadata> {
            self.inner.metadata(path).await
        }

        async fn symlink_metadata(&self, path: &Path) -> std::io::Result<FsMetadata> {
            self.inner.symlink_metadata(path).await
        }

        async fn read_dir(&self, path: &Path) -> std::io::Result<Vec<DirEntry>> {
            self.inner.read_dir(path).await
        }

        async fn canonicalize(&self, path: &Path) -> std::io::Result<std::path::PathBuf> {
            self.inner.canonicalize(path).await
        }

        async fn create_dir(&self, path: &Path) -> std::io::Result<()> {
            self.inner.create_dir(path).await
        }

        async fn create_dir_all(&self, path: &Path) -> std::io::Result<()> {
            self.inner.create_dir_all(path).await
        }

        async fn remove_dir(&self, path: &Path) -> std::io::Result<()> {
            self.inner.remove_dir(path).await
        }

        async fn remove_dir_all(&self, path: &Path) -> std::io::Result<()> {
            self.inner.remove_dir_all(path).await
        }

        async fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()> {
            self.inner.rename(from, to).await
        }

        async fn set_readonly(&self, path: &Path, readonly: bool) -> std::io::Result<()> {
            self.inner.set_readonly(path, readonly).await
        }

        async fn data_manifest_descriptor(
            &self,
            request: &DataManifestRequest,
        ) -> std::io::Result<DataManifestDescriptor> {
            self.inner.data_manifest_descriptor(request).await
        }

        async fn data_chunk_upload_targets(
            &self,
            request: &DataChunkUploadRequest,
        ) -> std::io::Result<Vec<DataChunkUploadTarget>> {
            #[derive(Deserialize)]
            struct UploadTargetsResponse {
                targets: Vec<DataChunkUploadTarget>,
            }
            let url = format!("{}/data/chunks/upload-targets", self.base_url);
            let response = self
                .client
                .post(url)
                .json(request)
                .send()
                .map_err(|err| std::io::Error::other(err.to_string()))?;
            if !response.status().is_success() {
                return Err(std::io::Error::other(format!(
                    "upload targets request failed: {}",
                    response.status()
                )));
            }
            let parsed: UploadTargetsResponse = response
                .json()
                .map_err(|err| std::io::Error::other(err.to_string()))?;
            Ok(parsed.targets)
        }

        async fn data_upload_chunk(
            &self,
            target: &DataChunkUploadTarget,
            data: &[u8],
        ) -> std::io::Result<()> {
            let upload_url = if let Some(key) = target.upload_url.strip_prefix("upload://") {
                format!("{}/upload?key={}", self.base_url, key)
            } else {
                target.upload_url.clone()
            };
            let method = reqwest::Method::from_bytes(target.method.as_bytes())
                .map_err(|err| std::io::Error::other(err.to_string()))?;
            let mut request = self.client.request(method, &upload_url);
            for (k, v) in &target.headers {
                request = request.header(k, v);
            }
            let response = request
                .body(data.to_vec())
                .send()
                .map_err(|err| std::io::Error::other(err.to_string()))?;
            if !response.status().is_success() {
                return Err(std::io::Error::other(format!(
                    "chunk upload failed: {}",
                    response.status()
                )));
            }
            Ok(())
        }
    }

    #[derive(Clone, Default)]
    struct UploadHarness {
        uploads: Arc<Mutex<Vec<String>>>,
    }

    #[derive(Deserialize)]
    struct UploadChunkQuery {
        key: String,
    }

    async fn upload_targets_handler(
        Json(req): Json<DataChunkUploadRequest>,
    ) -> Result<Json<serde_json::Value>, StatusCode> {
        let targets = req
            .chunks
            .iter()
            .map(|chunk| {
                serde_json::json!({
                    "key": chunk.key,
                    "method": "PUT",
                    "upload_url": format!("upload://{}", chunk.key),
                    "headers": {
                        "x-runmat-hash": chunk.hash,
                    }
                })
            })
            .collect::<Vec<_>>();
        Ok(Json(serde_json::json!({ "targets": targets })))
    }

    async fn upload_handler(
        State(harness): State<UploadHarness>,
        Query(query): Query<UploadChunkQuery>,
        headers: HeaderMap,
        body: axum::body::Bytes,
    ) -> Result<(), StatusCode> {
        if body.is_empty() {
            return Err(StatusCode::BAD_REQUEST);
        }
        if headers.get("x-runmat-hash").is_none() {
            return Err(StatusCode::BAD_REQUEST);
        }
        let mut guard = harness.uploads.lock().expect("uploads lock poisoned");
        guard.push(query.key);
        Ok(())
    }

    fn spawn_upload_server() -> (
        String,
        Arc<Mutex<Vec<String>>>,
        Runtime,
        oneshot::Sender<()>,
    ) {
        let harness = UploadHarness::default();
        let uploads = Arc::clone(&harness.uploads);
        let runtime = Runtime::new().expect("tokio runtime");
        let (addr, shutdown_tx) = runtime.block_on(async move {
            let listener = tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0))
                .await
                .expect("bind upload server");
            let addr = listener.local_addr().expect("local addr");
            let app = Router::new()
                .route("/data/chunks/upload-targets", post(upload_targets_handler))
                .route("/upload", put(upload_handler))
                .with_state(harness);
            let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
            let server = axum::serve(listener, app).with_graceful_shutdown(async {
                let _ = shutdown_rx.await;
            });
            tokio::spawn(async move {
                let _ = server.await;
            });
            (addr, shutdown_tx)
        });
        (format!("http://{}", addr), uploads, runtime, shutdown_tx)
    }

    impl CountingDataUploadProvider {
        fn uploaded_keys(&self) -> Arc<Mutex<Vec<String>>> {
            Arc::clone(&self.uploaded_keys)
        }
    }

    #[async_trait(?Send)]
    impl FsProvider for CountingDataUploadProvider {
        fn open(&self, path: &Path, flags: &OpenFlags) -> std::io::Result<Box<dyn FileHandle>> {
            self.inner.open(path, flags)
        }

        async fn read(&self, path: &Path) -> std::io::Result<Vec<u8>> {
            self.inner.read(path).await
        }

        async fn write(&self, path: &Path, data: &[u8]) -> std::io::Result<()> {
            self.inner.write(path, data).await
        }

        async fn remove_file(&self, path: &Path) -> std::io::Result<()> {
            self.inner.remove_file(path).await
        }

        async fn metadata(&self, path: &Path) -> std::io::Result<FsMetadata> {
            self.inner.metadata(path).await
        }

        async fn symlink_metadata(&self, path: &Path) -> std::io::Result<FsMetadata> {
            self.inner.symlink_metadata(path).await
        }

        async fn read_dir(&self, path: &Path) -> std::io::Result<Vec<DirEntry>> {
            self.inner.read_dir(path).await
        }

        async fn canonicalize(&self, path: &Path) -> std::io::Result<std::path::PathBuf> {
            self.inner.canonicalize(path).await
        }

        async fn create_dir(&self, path: &Path) -> std::io::Result<()> {
            self.inner.create_dir(path).await
        }

        async fn create_dir_all(&self, path: &Path) -> std::io::Result<()> {
            self.inner.create_dir_all(path).await
        }

        async fn remove_dir(&self, path: &Path) -> std::io::Result<()> {
            self.inner.remove_dir(path).await
        }

        async fn remove_dir_all(&self, path: &Path) -> std::io::Result<()> {
            self.inner.remove_dir_all(path).await
        }

        async fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()> {
            self.inner.rename(from, to).await
        }

        async fn set_readonly(&self, path: &Path, readonly: bool) -> std::io::Result<()> {
            self.inner.set_readonly(path, readonly).await
        }

        async fn data_manifest_descriptor(
            &self,
            request: &DataManifestRequest,
        ) -> std::io::Result<DataManifestDescriptor> {
            self.inner.data_manifest_descriptor(request).await
        }

        async fn data_chunk_upload_targets(
            &self,
            request: &DataChunkUploadRequest,
        ) -> std::io::Result<Vec<DataChunkUploadTarget>> {
            Ok(request
                .chunks
                .iter()
                .map(|chunk| DataChunkUploadTarget {
                    key: chunk.key.clone(),
                    method: "PUT".to_string(),
                    upload_url: format!("count://{}", chunk.object_id),
                    headers: std::collections::HashMap::new(),
                })
                .collect())
        }

        async fn data_upload_chunk(
            &self,
            target: &DataChunkUploadTarget,
            _data: &[u8],
        ) -> std::io::Result<()> {
            let mut guard = match self.uploaded_keys.lock() {
                Ok(guard) => guard,
                Err(poisoned) => poisoned.into_inner(),
            };
            guard.push(target.key.clone());
            Ok(())
        }
    }

    #[test]
    fn create_open_write_read_dataset() {
        let _serial = serial_test_guard();
        let _provider_guard = native_provider_guard();
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("sample.data").to_string_lossy().to_string();

        let mut array_meta = StructValue::new();
        array_meta
            .fields
            .insert("dtype".to_string(), Value::String("f64".to_string()));
        array_meta.fields.insert(
            "shape".to_string(),
            Value::Tensor(Tensor::new(vec![2.0, 2.0], vec![1, 2]).expect("shape tensor")),
        );
        let mut arrays = StructValue::new();
        arrays
            .fields
            .insert("temperature".to_string(), Value::Struct(array_meta));
        let mut schema = StructValue::new();
        schema
            .fields
            .insert("arrays".to_string(), Value::Struct(arrays));

        let ds = call_builtin(
            "data.create",
            &[Value::String(path.clone()), Value::Struct(schema)],
        )
        .expect("create dataset");

        let arr = call_builtin(
            "Dataset.array",
            &[ds, Value::String("temperature".to_string())],
        )
        .expect("dataset array");
        let write_tensor = Tensor::new(vec![1.0, 2.0, 3.0, 4.0], vec![2, 2]).expect("write tensor");
        call_builtin(
            "DataArray.write",
            &[arr.clone(), Value::Tensor(write_tensor)],
        )
        .expect("write array");

        let read_back = call_builtin("DataArray.read", &[arr]).expect("read array");
        let Value::Tensor(t) = read_back else {
            panic!("expected tensor");
        };
        assert_eq!(t.shape, vec![2, 2]);
        assert_eq!(tensor_values(&t), vec![1.0, 2.0, 3.0, 4.0]);
    }

    #[test]
    fn write_and_read_slice_payload() {
        let _serial = serial_test_guard();
        let _provider_guard = native_provider_guard();
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("slice.data").to_string_lossy().to_string();

        let mut array_meta = StructValue::new();
        array_meta
            .fields
            .insert("dtype".to_string(), Value::String("f64".to_string()));
        array_meta.fields.insert(
            "shape".to_string(),
            Value::Tensor(Tensor::new(vec![3.0, 3.0], vec![1, 2]).expect("shape tensor")),
        );
        let mut arrays = StructValue::new();
        arrays
            .fields
            .insert("temperature".to_string(), Value::Struct(array_meta));
        let mut schema = StructValue::new();
        schema
            .fields
            .insert("arrays".to_string(), Value::Struct(arrays));

        let ds = call_builtin(
            "data.create",
            &[Value::String(path.clone()), Value::Struct(schema)],
        )
        .expect("create dataset");
        let arr = call_builtin(
            "Dataset.array",
            &[ds, Value::String("temperature".to_string())],
        )
        .expect("dataset array");

        let slice = Value::Cell(
            CellArray::new(
                vec![
                    Value::Tensor(Tensor::new(vec![1.0, 2.0], vec![1, 2]).expect("range")),
                    Value::String(":".to_string()),
                ],
                1,
                2,
            )
            .expect("slice cell"),
        );
        let rhs = Value::Tensor(
            Tensor::new(vec![10.0, 11.0, 12.0, 13.0, 14.0, 15.0], vec![2, 3]).expect("rhs"),
        );
        call_builtin("DataArray.write", &[arr.clone(), slice.clone(), rhs]).expect("slice write");

        let read_back = call_builtin("DataArray.read", &[arr.clone(), slice]).expect("slice read");
        let Value::Tensor(t) = read_back else {
            panic!("expected tensor");
        };
        assert_eq!(t.shape, vec![2, 3]);
        assert_eq!(tensor_values(&t), vec![10.0, 11.0, 12.0, 13.0, 14.0, 15.0]);
    }

    #[test]
    fn slice_write_updates_only_touched_chunks() {
        let _serial = serial_test_guard();
        let _provider_guard = native_provider_guard();
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir
            .path()
            .join("chunked.data")
            .to_string_lossy()
            .to_string();

        let mut array_meta = StructValue::new();
        array_meta
            .fields
            .insert("dtype".to_string(), Value::String("f64".to_string()));
        array_meta.fields.insert(
            "shape".to_string(),
            Value::Tensor(Tensor::new(vec![4.0, 4.0], vec![1, 2]).expect("shape tensor")),
        );
        array_meta.fields.insert(
            "chunk".to_string(),
            Value::Tensor(Tensor::new(vec![2.0, 2.0], vec![1, 2]).expect("chunk tensor")),
        );
        let mut arrays = StructValue::new();
        arrays
            .fields
            .insert("temperature".to_string(), Value::Struct(array_meta));
        let mut schema = StructValue::new();
        schema
            .fields
            .insert("arrays".to_string(), Value::Struct(arrays));

        let ds = call_builtin(
            "data.create",
            &[Value::String(path.clone()), Value::Struct(schema)],
        )
        .expect("create dataset");
        let arr = call_builtin(
            "Dataset.array",
            &[ds, Value::String("temperature".to_string())],
        )
        .expect("dataset array");

        let full = Value::Tensor(
            Tensor::new((1..=16).map(|v| v as f64).collect(), vec![4, 4]).expect("full tensor"),
        );
        call_builtin("DataArray.write", &[arr.clone(), full]).expect("initial write");

        let root = std::path::PathBuf::from(&path);
        let untouched_path = root.join("arrays/temperature/chunks/obj_1_1.json");
        let touched_path = root.join("arrays/temperature/chunks/obj_0_0.json");
        let untouched_before =
            futures::executor::block_on(runmat_filesystem::read_async(&untouched_path))
                .expect("read untouched before");
        let touched_before =
            futures::executor::block_on(runmat_filesystem::read_async(&touched_path))
                .expect("read touched before");

        let slice = Value::Cell(
            CellArray::new(
                vec![
                    Value::Tensor(Tensor::new(vec![1.0, 2.0], vec![1, 2]).expect("range")),
                    Value::Tensor(Tensor::new(vec![1.0, 2.0], vec![1, 2]).expect("range")),
                ],
                1,
                2,
            )
            .expect("slice cell"),
        );
        let rhs =
            Value::Tensor(Tensor::new(vec![99.0, 98.0, 97.0, 96.0], vec![2, 2]).expect("rhs"));
        call_builtin("DataArray.write", &[arr.clone(), slice, rhs]).expect("slice write");

        let untouched_after =
            futures::executor::block_on(runmat_filesystem::read_async(&untouched_path))
                .expect("read untouched after");
        let touched_after =
            futures::executor::block_on(runmat_filesystem::read_async(&touched_path))
                .expect("read touched after");
        assert_eq!(untouched_before, untouched_after);
        assert_ne!(touched_before, touched_after);
    }

    #[test]
    fn slice_write_uploads_only_touched_chunk_targets() {
        let _serial = serial_test_guard();
        let provider = Arc::new(CountingDataUploadProvider::default());
        let uploaded = provider.uploaded_keys();
        let _guard = runmat_filesystem::replace_provider(provider);

        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir
            .path()
            .join("remote-chunked.data")
            .to_string_lossy()
            .to_string();

        let mut array_meta = StructValue::new();
        array_meta
            .fields
            .insert("dtype".to_string(), Value::String("f64".to_string()));
        array_meta.fields.insert(
            "shape".to_string(),
            Value::Tensor(Tensor::new(vec![4.0, 4.0], vec![1, 2]).expect("shape tensor")),
        );
        array_meta.fields.insert(
            "chunk".to_string(),
            Value::Tensor(Tensor::new(vec![2.0, 2.0], vec![1, 2]).expect("chunk tensor")),
        );
        let mut arrays = StructValue::new();
        arrays
            .fields
            .insert("temperature".to_string(), Value::Struct(array_meta));
        let mut schema = StructValue::new();
        schema
            .fields
            .insert("arrays".to_string(), Value::Struct(arrays));

        let ds = call_builtin(
            "data.create",
            &[Value::String(path.clone()), Value::Struct(schema)],
        )
        .expect("create dataset");
        let arr = call_builtin(
            "Dataset.array",
            &[ds, Value::String("temperature".to_string())],
        )
        .expect("dataset array");

        call_builtin(
            "DataArray.write",
            &[
                arr.clone(),
                Value::Tensor(
                    Tensor::new((1..=16).map(|v| v as f64).collect(), vec![4, 4])
                        .expect("full tensor"),
                ),
            ],
        )
        .expect("initial write");

        let manifest =
            futures::executor::block_on(crate::data::read_manifest_async(&dataset_root(&path)))
                .expect("manifest after initial write");
        let meta = manifest
            .arrays
            .get("temperature")
            .expect("temperature meta");
        let chunk_index_path =
            dataset_root(&path).join(meta.chunk_index_path.clone().expect("chunk index path"));
        assert!(
            futures::executor::block_on(runmat_filesystem::metadata_async(&chunk_index_path))
                .is_ok()
        );

        {
            let mut keys = uploaded.lock().expect("uploaded keys lock");
            keys.clear();
        }

        let slice = Value::Cell(
            CellArray::new(
                vec![
                    Value::Tensor(Tensor::new(vec![1.0, 2.0], vec![1, 2]).expect("range")),
                    Value::Tensor(Tensor::new(vec![1.0, 2.0], vec![1, 2]).expect("range")),
                ],
                1,
                2,
            )
            .expect("slice cell"),
        );
        let rhs = Value::Tensor(Tensor::new(vec![9.0, 8.0, 7.0, 6.0], vec![2, 2]).expect("rhs"));
        call_builtin("DataArray.write", &[arr, slice, rhs]).expect("slice write");

        let keys = uploaded.lock().expect("uploaded keys lock");
        assert_eq!(keys.as_slice(), ["0.0".to_string()].as_slice());
    }

    #[test]
    fn slice_write_uploads_expected_cross_boundary_chunk_targets() {
        let _serial = serial_test_guard();
        let provider = Arc::new(CountingDataUploadProvider::default());
        let uploaded = provider.uploaded_keys();
        let _guard = runmat_filesystem::replace_provider(provider);

        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir
            .path()
            .join("remote-chunked-boundary.data")
            .to_string_lossy()
            .to_string();

        let mut array_meta = StructValue::new();
        array_meta
            .fields
            .insert("dtype".to_string(), Value::String("f64".to_string()));
        array_meta.fields.insert(
            "shape".to_string(),
            Value::Tensor(Tensor::new(vec![4.0, 4.0], vec![1, 2]).expect("shape tensor")),
        );
        array_meta.fields.insert(
            "chunk".to_string(),
            Value::Tensor(Tensor::new(vec![2.0, 2.0], vec![1, 2]).expect("chunk tensor")),
        );
        let mut arrays = StructValue::new();
        arrays
            .fields
            .insert("temperature".to_string(), Value::Struct(array_meta));
        let mut schema = StructValue::new();
        schema
            .fields
            .insert("arrays".to_string(), Value::Struct(arrays));

        let ds = call_builtin(
            "data.create",
            &[Value::String(path.clone()), Value::Struct(schema)],
        )
        .expect("create dataset");
        let arr = call_builtin(
            "Dataset.array",
            &[ds, Value::String("temperature".to_string())],
        )
        .expect("dataset array");

        call_builtin(
            "DataArray.write",
            &[
                arr.clone(),
                Value::Tensor(
                    Tensor::new((1..=16).map(|v| v as f64).collect(), vec![4, 4])
                        .expect("full tensor"),
                ),
            ],
        )
        .expect("initial write");
        {
            let mut keys = uploaded.lock().expect("uploaded keys lock");
            keys.clear();
        }

        let slice = Value::Cell(
            CellArray::new(
                vec![
                    Value::Tensor(Tensor::new(vec![2.0, 3.0], vec![1, 2]).expect("range")),
                    Value::Tensor(Tensor::new(vec![2.0, 3.0], vec![1, 2]).expect("range")),
                ],
                1,
                2,
            )
            .expect("slice cell"),
        );
        let rhs =
            Value::Tensor(Tensor::new(vec![19.0, 18.0, 17.0, 16.0], vec![2, 2]).expect("rhs"));
        call_builtin("DataArray.write", &[arr, slice, rhs]).expect("slice write");

        let mut keys = uploaded.lock().expect("uploaded keys lock").clone();
        keys.sort();
        keys.dedup();
        assert_eq!(
            keys.as_slice(),
            [
                "0.0".to_string(),
                "0.1".to_string(),
                "1.0".to_string(),
                "1.1".to_string(),
            ]
            .as_slice()
        );
    }

    #[test]
    fn slice_write_hits_http_server_data_endpoints_with_expected_keys() {
        let _serial = serial_test_guard();
        let (base_url, uploads, runtime, shutdown_tx) = spawn_upload_server();
        let provider = Arc::new(HttpDataUploadProvider::new(base_url));
        let _guard = runmat_filesystem::replace_provider(provider);

        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir
            .path()
            .join("http-endpoint.data")
            .to_string_lossy()
            .to_string();

        let mut array_meta = StructValue::new();
        array_meta
            .fields
            .insert("dtype".to_string(), Value::String("f64".to_string()));
        array_meta.fields.insert(
            "shape".to_string(),
            Value::Tensor(Tensor::new(vec![4.0, 4.0], vec![1, 2]).expect("shape tensor")),
        );
        array_meta.fields.insert(
            "chunk".to_string(),
            Value::Tensor(Tensor::new(vec![2.0, 2.0], vec![1, 2]).expect("chunk tensor")),
        );
        let mut arrays = StructValue::new();
        arrays
            .fields
            .insert("temperature".to_string(), Value::Struct(array_meta));
        let mut schema = StructValue::new();
        schema
            .fields
            .insert("arrays".to_string(), Value::Struct(arrays));

        let ds = call_builtin(
            "data.create",
            &[Value::String(path.clone()), Value::Struct(schema)],
        )
        .expect("create dataset");
        let arr = call_builtin(
            "Dataset.array",
            &[ds, Value::String("temperature".to_string())],
        )
        .expect("dataset array");

        call_builtin(
            "DataArray.write",
            &[
                arr.clone(),
                Value::Tensor(
                    Tensor::new((1..=16).map(|v| v as f64).collect(), vec![4, 4])
                        .expect("full tensor"),
                ),
            ],
        )
        .expect("initial write");

        {
            let mut keys = uploads.lock().expect("uploads lock");
            keys.clear();
        }

        let slice = Value::Cell(
            CellArray::new(
                vec![
                    Value::Tensor(Tensor::new(vec![2.0, 3.0], vec![1, 2]).expect("range")),
                    Value::Tensor(Tensor::new(vec![2.0, 3.0], vec![1, 2]).expect("range")),
                ],
                1,
                2,
            )
            .expect("slice cell"),
        );
        let rhs =
            Value::Tensor(Tensor::new(vec![19.0, 18.0, 17.0, 16.0], vec![2, 2]).expect("rhs"));
        call_builtin("DataArray.write", &[arr, slice, rhs]).expect("slice write");

        let mut keys = uploads.lock().expect("uploads lock").clone();
        keys.sort();
        keys.dedup();
        assert_eq!(
            keys.as_slice(),
            [
                "0.0".to_string(),
                "0.1".to_string(),
                "1.0".to_string(),
                "1.1".to_string(),
            ]
            .as_slice()
        );

        let _ = shutdown_tx.send(());
        drop(runtime);
    }

    #[test]
    fn tx_create_resize_fill_and_delete_array() {
        let _serial = serial_test_guard();
        let _provider = native_provider_guard();
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("tx-ops.data").to_string_lossy().to_string();

        let mut arrays = StructValue::new();
        let mut array_meta = StructValue::new();
        array_meta
            .fields
            .insert("dtype".to_string(), Value::String("f64".to_string()));
        array_meta.fields.insert(
            "shape".to_string(),
            Value::Tensor(Tensor::new(vec![1.0, 1.0], vec![1, 2]).expect("shape tensor")),
        );
        arrays
            .fields
            .insert("base".to_string(), Value::Struct(array_meta));
        let mut schema = StructValue::new();
        schema
            .fields
            .insert("arrays".to_string(), Value::Struct(arrays));

        let ds = call_builtin(
            "data.create",
            &[Value::String(path.clone()), Value::Struct(schema)],
        )
        .expect("create dataset");

        let tx = call_builtin("Dataset.begin", &[ds]).expect("begin tx");
        let mut new_meta = StructValue::new();
        new_meta
            .fields
            .insert("dtype".to_string(), Value::String("f64".to_string()));
        new_meta.fields.insert(
            "shape".to_string(),
            Value::Tensor(Tensor::new(vec![2.0, 2.0], vec![1, 2]).expect("shape tensor")),
        );
        call_builtin(
            "DataTransaction.create_array",
            &[
                tx.clone(),
                Value::String("new_array".to_string()),
                Value::Struct(new_meta),
            ],
        )
        .expect("create array in tx");
        call_builtin(
            "DataTransaction.resize",
            &[
                tx.clone(),
                Value::String("new_array".to_string()),
                Value::Tensor(Tensor::new(vec![3.0, 1.0], vec![1, 2]).expect("shape tensor")),
            ],
        )
        .expect("resize array in tx");
        call_builtin(
            "DataTransaction.fill",
            &[
                tx.clone(),
                Value::String("new_array".to_string()),
                Value::Num(7.0),
            ],
        )
        .expect("fill array in tx");
        call_builtin(
            "DataTransaction.delete_array",
            &[tx.clone(), Value::String("base".to_string())],
        )
        .expect("delete array in tx");
        call_builtin("commit", &[tx, Value::Struct(StructValue::new())])
            .expect("commit tx through alias options form");

        let ds = call_builtin("data.open", &[Value::String(path)]).expect("open dataset");
        let has_base = call_builtin(
            "Dataset.has_array",
            &[ds.clone(), Value::String("base".to_string())],
        )
        .expect("has base");
        assert_eq!(has_base, Value::Bool(false));
        let arr = call_builtin(
            "Dataset.array",
            &[ds, Value::String("new_array".to_string())],
        )
        .expect("new array");
        let read_back = call_builtin("DataArray.read", &[arr]).expect("read array");
        let Value::Tensor(t) = read_back else {
            panic!("expected tensor");
        };
        assert_eq!(t.shape, vec![3, 1]);
        assert_eq!(tensor_values(&t), vec![7.0, 7.0, 7.0]);
    }

    #[test]
    fn data_transactions_preserve_every_integer_class_and_typed_controls() {
        let _serial = serial_test_guard();
        let _provider = native_provider_guard();
        let cases = vec![
            (
                "int8",
                IntegerStorage::I8(vec![i8::MIN, i8::MAX]),
                IntegerStorage::I8(vec![2, 1]),
                IntegerStorage::I8(vec![1, 1]),
                IntegerStorage::I8(vec![1, 2]),
                IntValue::I8(1),
                IntValue::I8(i8::MIN),
            ),
            (
                "int16",
                IntegerStorage::I16(vec![i16::MIN, i16::MAX]),
                IntegerStorage::I16(vec![2, 1]),
                IntegerStorage::I16(vec![1, 1]),
                IntegerStorage::I16(vec![1, 2]),
                IntValue::I16(1),
                IntValue::I16(i16::MIN),
            ),
            (
                "int32",
                IntegerStorage::I32(vec![i32::MIN, i32::MAX]),
                IntegerStorage::I32(vec![2, 1]),
                IntegerStorage::I32(vec![1, 1]),
                IntegerStorage::I32(vec![1, 2]),
                IntValue::I32(1),
                IntValue::I32(i32::MIN),
            ),
            (
                "int64",
                IntegerStorage::I64(vec![i64::MIN, i64::MAX]),
                IntegerStorage::I64(vec![2, 1]),
                IntegerStorage::I64(vec![1, 1]),
                IntegerStorage::I64(vec![1, 2]),
                IntValue::I64(1),
                IntValue::I64(i64::MIN),
            ),
            (
                "uint8",
                IntegerStorage::U8(vec![0, u8::MAX]),
                IntegerStorage::U8(vec![2, 1]),
                IntegerStorage::U8(vec![1, 1]),
                IntegerStorage::U8(vec![1, 2]),
                IntValue::U8(1),
                IntValue::U8(u8::MAX),
            ),
            (
                "uint16",
                IntegerStorage::U16(vec![0, u16::MAX]),
                IntegerStorage::U16(vec![2, 1]),
                IntegerStorage::U16(vec![1, 1]),
                IntegerStorage::U16(vec![1, 2]),
                IntValue::U16(1),
                IntValue::U16(u16::MAX),
            ),
            (
                "uint32",
                IntegerStorage::U32(vec![0, u32::MAX]),
                IntegerStorage::U32(vec![2, 1]),
                IntegerStorage::U32(vec![1, 1]),
                IntegerStorage::U32(vec![1, 2]),
                IntValue::U32(1),
                IntValue::U32(u32::MAX),
            ),
            (
                "uint64",
                IntegerStorage::U64(vec![0, u64::MAX]),
                IntegerStorage::U64(vec![2, 1]),
                IntegerStorage::U64(vec![1, 1]),
                IntegerStorage::U64(vec![1, 2]),
                IntValue::U64(1),
                IntValue::U64(u64::MAX),
            ),
        ];

        for (dtype, values, shape, chunk, range, one, attribute) in cases {
            let dir = tempfile::tempdir().expect("tempdir");
            let path = dir
                .path()
                .join(format!("tx-{dtype}.data"))
                .to_string_lossy()
                .to_string();
            let mut base_meta = StructValue::new();
            base_meta
                .fields
                .insert("dtype".to_string(), Value::String("f64".to_string()));
            base_meta.fields.insert(
                "shape".to_string(),
                Value::Tensor(Tensor::new(vec![1.0, 1.0], vec![1, 2]).expect("base shape")),
            );
            let mut arrays = StructValue::new();
            arrays
                .fields
                .insert("base".to_string(), Value::Struct(base_meta));
            let mut schema = StructValue::new();
            schema
                .fields
                .insert("arrays".to_string(), Value::Struct(arrays));
            let ds = call_builtin("data.create", &[Value::String(path), Value::Struct(schema)])
                .expect("create dataset");
            let tx = call_builtin("Dataset.begin", std::slice::from_ref(&ds))
                .expect("begin transaction");

            let mut meta = StructValue::new();
            meta.fields
                .insert("dtype".to_string(), Value::String(dtype.to_string()));
            meta.fields.insert(
                "shape".to_string(),
                Value::Tensor(Tensor::new_integer(shape.clone(), vec![1, 2]).expect("shape")),
            );
            meta.fields.insert(
                "chunk".to_string(),
                Value::Tensor(Tensor::new_integer(chunk, vec![1, 2]).expect("chunk")),
            );
            call_builtin(
                "DataTransaction.create_array",
                &[
                    tx.clone(),
                    Value::String("typed".to_string()),
                    Value::Struct(meta),
                ],
            )
            .expect("queue create");
            call_builtin(
                "DataTransaction.resize",
                &[
                    tx.clone(),
                    Value::String("typed".to_string()),
                    Value::Tensor(Tensor::new_integer(shape, vec![1, 2]).expect("resize shape")),
                ],
            )
            .expect("queue resize");
            call_builtin(
                "DataTransaction.fill",
                &[
                    tx.clone(),
                    Value::String("typed".to_string()),
                    Value::Int(attribute.clone()),
                ],
            )
            .expect("queue fill");
            let slice = Value::Cell(
                CellArray::new(
                    vec![
                        Value::Tensor(Tensor::new_integer(range, vec![1, 2]).expect("row range")),
                        Value::Int(one),
                    ],
                    1,
                    2,
                )
                .expect("slice"),
            );
            call_builtin(
                "DataTransaction.write",
                &[
                    tx.clone(),
                    Value::String("typed".to_string()),
                    slice,
                    Value::Tensor(Tensor::new_integer(values.clone(), vec![2, 1]).expect("values")),
                ],
            )
            .expect("queue write");
            call_builtin(
                "DataTransaction.set_attr",
                &[
                    tx.clone(),
                    Value::String(format!("{dtype}_one")),
                    Value::Int(attribute.clone()),
                ],
            )
            .expect("queue attribute");
            let mut attrs = StructValue::new();
            attrs
                .fields
                .insert(format!("{dtype}_many"), Value::Int(attribute));
            call_builtin(
                "DataTransaction.set_attrs",
                &[tx.clone(), Value::Struct(attrs)],
            )
            .expect("queue attributes");
            call_builtin("DataTransaction.commit", &[tx]).expect("commit transaction");

            let arr = call_builtin("Dataset.array", &[ds, Value::String("typed".to_string())])
                .expect("open typed array");
            let Value::Tensor(read_back) =
                call_builtin("DataArray.read", &[arr]).expect("read typed array")
            else {
                panic!("expected tensor");
            };
            assert_eq!(read_back.integer_storage(), Some(&values), "{dtype}");
        }
    }

    #[test]
    fn data_arrays_preserve_every_integer_class_through_chunked_write_and_read() {
        let _serial = serial_test_guard();
        let _provider = native_provider_guard();
        let cases = vec![
            (
                "int8",
                IntegerStorage::I8(vec![i8::MIN, i8::MAX]),
                IntValue::I8(i8::MIN),
                IntegerStorage::I8(vec![i8::MIN; 2]),
            ),
            (
                "int16",
                IntegerStorage::I16(vec![i16::MIN, i16::MAX]),
                IntValue::I16(i16::MIN),
                IntegerStorage::I16(vec![i16::MIN; 2]),
            ),
            (
                "int32",
                IntegerStorage::I32(vec![i32::MIN, i32::MAX]),
                IntValue::I32(i32::MIN),
                IntegerStorage::I32(vec![i32::MIN; 2]),
            ),
            (
                "int64",
                IntegerStorage::I64(vec![i64::MIN, i64::MAX]),
                IntValue::I64(i64::MIN),
                IntegerStorage::I64(vec![i64::MIN; 2]),
            ),
            (
                "uint8",
                IntegerStorage::U8(vec![0, u8::MAX]),
                IntValue::U8(u8::MAX),
                IntegerStorage::U8(vec![u8::MAX; 2]),
            ),
            (
                "uint16",
                IntegerStorage::U16(vec![0, u16::MAX]),
                IntValue::U16(u16::MAX),
                IntegerStorage::U16(vec![u16::MAX; 2]),
            ),
            (
                "uint32",
                IntegerStorage::U32(vec![0, u32::MAX]),
                IntValue::U32(u32::MAX),
                IntegerStorage::U32(vec![u32::MAX; 2]),
            ),
            (
                "uint64",
                IntegerStorage::U64(vec![0, u64::MAX]),
                IntValue::U64(u64::MAX),
                IntegerStorage::U64(vec![u64::MAX; 2]),
            ),
        ];

        for (dtype, storage, fill, filled_storage) in cases {
            let dir = tempfile::tempdir().expect("tempdir");
            let path = dir
                .path()
                .join(format!("{dtype}.data"))
                .to_string_lossy()
                .to_string();
            let mut array_meta = StructValue::new();
            array_meta
                .fields
                .insert("dtype".to_string(), Value::String(dtype.to_string()));
            array_meta.fields.insert(
                "shape".to_string(),
                Value::Tensor(Tensor::new(vec![2.0, 1.0], vec![1, 2]).expect("shape")),
            );
            array_meta.fields.insert(
                "chunk".to_string(),
                Value::Tensor(Tensor::new(vec![1.0, 1.0], vec![1, 2]).expect("chunk")),
            );
            let mut arrays = StructValue::new();
            arrays
                .fields
                .insert("samples".to_string(), Value::Struct(array_meta));
            let mut schema = StructValue::new();
            schema
                .fields
                .insert("arrays".to_string(), Value::Struct(arrays));

            let ds = call_builtin("data.create", &[Value::String(path), Value::Struct(schema)])
                .expect("create dataset");
            let arr = call_builtin("Dataset.array", &[ds, Value::String("samples".to_string())])
                .expect("array");
            let input = Tensor::new_integer(storage.clone(), vec![2, 1]).expect("integer tensor");
            call_builtin("DataArray.write", &[arr.clone(), Value::Tensor(input)])
                .expect("write integer array");

            let Value::Tensor(read_back) =
                call_builtin("DataArray.read", std::slice::from_ref(&arr)).expect("read")
            else {
                panic!("expected tensor");
            };
            assert_eq!(read_back.integer_storage(), Some(&storage), "{dtype}");

            call_builtin("DataArray.fill", &[arr.clone(), Value::Int(fill)])
                .expect("fill integer array");
            let Value::Tensor(read_back) =
                call_builtin("DataArray.read", &[arr]).expect("read filled array")
            else {
                panic!("expected tensor");
            };
            assert_eq!(
                read_back.integer_storage(),
                Some(&filled_storage),
                "{dtype} fill"
            );
        }
    }

    #[test]
    fn uint64_data_array_slice_fill_and_transaction_paths_remain_exact() {
        let _serial = serial_test_guard();
        let _provider = native_provider_guard();
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("uint64.data").to_string_lossy().to_string();
        let mut array_meta = StructValue::new();
        array_meta
            .fields
            .insert("dtype".to_string(), Value::String("uint64".to_string()));
        array_meta.fields.insert(
            "shape".to_string(),
            Value::Tensor(Tensor::new(vec![2.0, 2.0], vec![1, 2]).expect("shape")),
        );
        array_meta.fields.insert(
            "chunk".to_string(),
            Value::Tensor(Tensor::new(vec![1.0, 1.0], vec![1, 2]).expect("chunk")),
        );
        let mut arrays = StructValue::new();
        arrays
            .fields
            .insert("samples".to_string(), Value::Struct(array_meta));
        let mut schema = StructValue::new();
        schema
            .fields
            .insert("arrays".to_string(), Value::Struct(arrays));
        let ds = call_builtin("data.create", &[Value::String(path), Value::Struct(schema)])
            .expect("create dataset");
        let arr = call_builtin(
            "Dataset.array",
            &[ds.clone(), Value::String("samples".to_string())],
        )
        .expect("array");

        call_builtin(
            "DataArray.fill",
            &[
                arr.clone(),
                Value::Int(runmat_value::IntValue::U64(u64::MAX)),
            ],
        )
        .expect("fill");
        let slice = Value::Cell(
            CellArray::new(
                vec![
                    Value::Int(runmat_value::IntValue::I32(1)),
                    Value::String(":".to_string()),
                ],
                1,
                2,
            )
            .expect("slice"),
        );
        let replacement = Tensor::new_integer(
            IntegerStorage::U64(vec![1_u64 << 63, u64::MAX - 1]),
            vec![1, 2],
        )
        .expect("replacement");
        call_builtin(
            "DataArray.write",
            &[arr.clone(), slice, Value::Tensor(replacement)],
        )
        .expect("slice write");

        let tx = call_builtin("Dataset.begin", &[ds]).expect("begin transaction");
        call_builtin(
            "DataTransaction.fill",
            &[
                tx.clone(),
                Value::String("samples".to_string()),
                Value::Int(runmat_value::IntValue::U64(1_u64 << 63)),
            ],
        )
        .expect("queue transaction fill");
        call_builtin("DataTransaction.commit", &[tx]).expect("commit transaction");

        let Value::Tensor(read_back) = call_builtin("DataArray.read", &[arr]).expect("read") else {
            panic!("expected tensor");
        };
        assert_eq!(
            read_back.integer_storage(),
            Some(&IntegerStorage::U64(vec![1_u64 << 63; 4]))
        );
    }

    #[test]
    fn paired_complex_uint64_data_array_chunk_and_slice_paths_remain_exact() {
        let _serial = serial_test_guard();
        let _provider = native_provider_guard();
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir
            .path()
            .join("complex-uint64.data")
            .to_string_lossy()
            .to_string();
        let mut array_meta = StructValue::new();
        array_meta
            .fields
            .insert("dtype".to_string(), Value::String("uint64".to_string()));
        array_meta.fields.insert(
            "shape".to_string(),
            Value::Tensor(Tensor::new(vec![2.0, 2.0], vec![1, 2]).expect("shape")),
        );
        array_meta.fields.insert(
            "chunk".to_string(),
            Value::Tensor(Tensor::new(vec![1.0, 1.0], vec![1, 2]).expect("chunk")),
        );
        let mut arrays = StructValue::new();
        arrays
            .fields
            .insert("samples".to_string(), Value::Struct(array_meta));
        let mut schema = StructValue::new();
        schema
            .fields
            .insert("arrays".to_string(), Value::Struct(arrays));
        let ds = call_builtin("data.create", &[Value::String(path), Value::Struct(schema)])
            .expect("create dataset");
        let arr = call_builtin("Dataset.array", &[ds, Value::String("samples".to_string())])
            .expect("array");
        let wide = (1_u64 << 53) + 1;
        let initial_storage = IntegerComplexStorage::new(
            IntegerStorage::U64(vec![wide, u64::MAX, 3, 4]),
            IntegerStorage::U64(vec![u64::MAX, wide, 5, 6]),
        )
        .expect("initial paired storage");
        let initial = ComplexTensor::new_integer(initial_storage.clone(), vec![2, 2])
            .expect("initial paired tensor");
        call_builtin(
            "DataArray.write",
            &[arr.clone(), Value::ComplexTensor(initial)],
        )
        .expect("write paired array");

        let Value::ComplexTensor(read_back) =
            call_builtin("DataArray.read", std::slice::from_ref(&arr)).expect("read paired array")
        else {
            panic!("expected paired tensor");
        };
        assert_eq!(read_back.integer_storage(), Some(&initial_storage));

        let slice = Value::Cell(
            CellArray::new(
                vec![Value::Int(IntValue::I32(1)), Value::String(":".to_string())],
                1,
                2,
            )
            .expect("slice"),
        );
        let replacement_storage = IntegerComplexStorage::new(
            IntegerStorage::U64(vec![1_u64 << 63, u64::MAX - 1]),
            IntegerStorage::U64(vec![u64::MAX - 2, 1_u64 << 63]),
        )
        .expect("replacement storage");
        let replacement = ComplexTensor::new_integer(replacement_storage, vec![1, 2])
            .expect("replacement tensor");
        call_builtin(
            "DataArray.write",
            &[arr.clone(), slice, Value::ComplexTensor(replacement)],
        )
        .expect("write paired slice");

        let Value::ComplexTensor(read_back) =
            call_builtin("DataArray.read", &[arr]).expect("read paired slice result")
        else {
            panic!("expected paired tensor");
        };
        let storage = read_back.integer_storage().expect("paired storage");
        assert_eq!(
            storage.real,
            IntegerStorage::U64(vec![1_u64 << 63, u64::MAX, u64::MAX - 1, 4])
        );
        assert_eq!(
            storage.imag,
            IntegerStorage::U64(vec![u64::MAX - 2, wide, 1_u64 << 63, 6])
        );
    }
}