shape-runtime 0.3.0

//! Conversion between runtime values and wire format.
//!
//! Phase 2b kind-threaded rewrite. Public functions take `(bits: u64,
//! kind: NativeKind)` pairs threaded from the FunctionBlob's compile-
//! time slot-kind metadata; internal dispatch is a `match kind { ... }`
//! with no tag-bit probing. Heap slots use `NativeKind::Ptr(HeapKind)` —
//! the kind tells the dispatcher which `HeapValue` arm decodes the
//! bits without probing the heap object's self-reported discriminant
//! in production (debug-only consistency check).
//!
//! See `docs/defections.md` 2026-05-06 (Phase 2b unified marshal +
//! wire/snapshot kind threading) for the architectural rationale.
//!
//! ## API
//!
//! - [`slot_to_wire`] — project (bits, kind) into a `WireValue`.
//! - [`wire_to_slot`] — project a `WireValue` into typed slot bits,
//!   given the `expected_kind` the caller wants. Returns
//!   `Result<u64, MarshalError>`.
//! - [`slot_to_envelope`] — wrap a typed slot in a `ValueEnvelope` with
//!   metadata.
//! - [`slot_extract_content`] — extract Content node renderings from a
//!   slot whose kind says it carries Content / DataTable / TableView.
//! - [`datatable_to_wire`] / [`datatable_to_ipc_bytes`] /
//!   [`datatable_from_ipc_bytes`] — typed `DataTable` ↔ wire/IPC.

use crate::Context;
use crate::marshal::MarshalError;
use arrow_ipc::{reader::FileReader, writer::FileWriter};
use shape_value::heap_value::HeapValue;
use shape_value::{DataTable, HeapKind, NativeKind};
use shape_wire::{
    DurationUnit as WireDurationUnit, ValueEnvelope, WireTable, WireValue,
};
use std::collections::BTreeMap;
use std::sync::Arc;

/// Project a typed slot's `(bits, kind)` to a `WireValue`.
///
/// The `kind` fully determines the projection — no tag-bit probing.
/// For `NativeKind::Ptr(hk)`, the function casts `bits` to
/// `*const HeapValue`, debug-asserts the kind matches, and dispatches
/// per `HeapValue` arm.
pub fn slot_to_wire(bits: u64, kind: NativeKind, ctx: &Context) -> WireValue {
    match kind {
        NativeKind::Float64 => WireValue::Number(f64::from_bits(bits)),
        NativeKind::NullableFloat64 => {
            let v = f64::from_bits(bits);
            if v.is_nan() {
                WireValue::Null
            } else {
                WireValue::Number(v)
            }
        }
        NativeKind::Int64 => WireValue::Integer(bits as i64),
        NativeKind::NullableInt64 => WireValue::Integer(bits as i64),
        NativeKind::Int8 => WireValue::I8(bits as i8),
        NativeKind::Int16 => WireValue::I16(bits as i16),
        NativeKind::Int32 => WireValue::I32(bits as i32),
        NativeKind::UInt8 => WireValue::U8(bits as u8),
        NativeKind::UInt16 => WireValue::U16(bits as u16),
        NativeKind::UInt32 => WireValue::U32(bits as u32),
        NativeKind::UInt64 => WireValue::U64(bits),
        NativeKind::IntSize => WireValue::Isize(bits as i64),
        NativeKind::UIntSize => WireValue::Usize(bits),
        NativeKind::NullableInt8
        | NativeKind::NullableInt16
        | NativeKind::NullableInt32
        | NativeKind::NullableUInt8
        | NativeKind::NullableUInt16
        | NativeKind::NullableUInt32
        | NativeKind::NullableUInt64
        | NativeKind::NullableIntSize
        | NativeKind::NullableUIntSize => WireValue::Integer(bits as i64),
        NativeKind::Bool => WireValue::Bool(bits != 0),
        // R5b-2-bool-null-sentinel-cluster (ADR-006 §2.7 + §2.7.5 +
        // §2.7.7/Q9, 2026-05-19): `NativeKind::Null` is the canonical
        // absence-of-value discriminator. Pre-disposition `(0u64,
        // NativeKind::Bool)` was the null sentinel which collided with
        // legitimate `false` bool slots (both encoded as bits=0); the
        // SURFACE-G6-NONE-OUTPUT-ADAPTER reproducer (`fn bar() ->
        // Option<int> { None }; bar()` at top level) materialized
        // `None` as `{"Bool": false}`. Post-disposition: kind IS the
        // discriminator per §2.7.7/Q9 — `NativeKind::Null` slots
        // project to `WireValue::Null` directly, restoring soundness.
        NativeKind::Null => WireValue::Null,
        // Round 19 S1.5 W12-nativekind-scalar-additions (2026-05-14):
        // ADR-006 §2.7.5 amendment adds F32 + Char as 4-byte scalar
        // variants. Wire projection: F32 widens to `WireValue::Number`
        // (`f64::from(f32)` is lossless); Char projects to a single-
        // codepoint string (mirror of the `HeapValue::Char` arm below)
        // because `WireValue` has no dedicated Char variant.
        NativeKind::Float32 => WireValue::Number(f64::from(f32::from_bits(bits as u32))),
        NativeKind::Char => match char::from_u32(bits as u32) {
            Some(c) => WireValue::String(c.to_string()),
            None => WireValue::Null,
        },
        NativeKind::String => {
            // bits is an Arc<String> raw pointer
            let ptr = bits as *const String;
            // SAFETY: kind contract pins this slot to an Arc<String> raw ptr.
            let s = unsafe { &*ptr };
            WireValue::String(s.clone())
        }
        // Wave 2 Agent B W12-StringV2-DecimalV2-NativeKind-additions
        // (ADR-006 §2.7.5 amendment, 2026-05-14): the v2-raw `*const StringObj`
        // carrier projects to the same `WireValue::String` wire shape as
        // `NativeKind::String` (Arc-wrapped sibling), via the carrier's
        // `as_str` accessor reading the UTF-8 payload at offset 8 (data ptr)
        // / 16 (len) of the `repr(C)` struct. The slot bits are NOT an
        // `Arc<T>` pointer — `StringObj` is a manually-allocated `repr(C)`
        // 24-byte carrier per `v2/string_obj.rs`.
        NativeKind::StringV2 => {
            if bits == 0 {
                return WireValue::Null;
            }
            // SAFETY: per the §2.7.5 amendment construction contract,
            // kind=StringV2 means bits = `ptr as u64` pointing to a live
            // `StringObj` with bumped refcount — the slot owns one
            // v2-retain share for the duration of this call.
            let ptr = bits as *const shape_value::v2::string_obj::StringObj;
            let s: &str = unsafe { shape_value::v2::string_obj::StringObj::as_str(ptr) };
            WireValue::String(s.to_string())
        }
        // Wave 2 Agent B: the v2-raw `*const DecimalObj` carrier projects
        // to `WireValue::Number` (the same wire shape as
        // `HeapValue::Decimal` per `heap_value_to_wire` below) via the
        // carrier's `value` accessor reading the inline `rust_decimal::Decimal`
        // at offset 8 of the `repr(C)` struct.
        NativeKind::DecimalV2 => {
            if bits == 0 {
                return WireValue::Null;
            }
            // SAFETY: per the §2.7.5 amendment construction contract,
            // kind=DecimalV2 means bits = `ptr as u64` pointing to a live
            // `DecimalObj` with bumped refcount.
            let ptr = bits as *const shape_value::v2::decimal_obj::DecimalObj;
            let value = unsafe { shape_value::v2::decimal_obj::DecimalObj::value(ptr) };
            WireValue::Number(value.to_string().parse().unwrap_or(0.0))
        }
        NativeKind::Ptr(hk) => heap_to_wire(bits, hk, ctx),
    }
}

/// Project an `Arc<HeapValue>` raw pointer slot to `WireValue`,
/// dispatching on the pre-known `HeapKind` rather than probing the
/// heap object's self-reported `kind()`.
fn heap_to_wire(bits: u64, hk: HeapKind, ctx: &Context) -> WireValue {
    if bits == 0 {
        return WireValue::Null;
    }
    // Defensive: `HeapKind::Char` is an inline-codepoint label, NOT an
    // `Arc<HeapValue>` pointer — its `bits` are a raw UTF-32 codepoint.
    // The canonical post-amendment carrier is the scalar `NativeKind::Char`
    // (handled in `slot_to_wire`), but any producer that still stamps the
    // pre-amendment `Ptr(HeapKind::Char)` label must not reach the
    // `*const HeapValue` cast below — casting a codepoint (e.g. 0x63) to a
    // HeapValue pointer and dereferencing it is a misaligned-pointer abort.
    // This arm projects the codepoint directly, mirroring the
    // `NativeKind::Char` arm and `HeapValue::Char` arm.
    if hk == HeapKind::Char {
        return match char::from_u32(bits as u32) {
            Some(c) => WireValue::String(c.to_string()),
            None => WireValue::Null,
        };
    }
    // WS-3 F2b: `HeapKind::Result` / `HeapKind::Option` are typed-Arc
    // dispatch labels — their bits are `Arc::into_raw(Arc<ResultData>)` /
    // `Arc::into_raw(Arc<OptionData>)`, NOT an `Arc<HeapValue>`. Casting
    // those bits to `*const HeapValue` (the path below) reads a
    // `ResultData`/`OptionData` as a `HeapValue` enum — type confusion +
    // UB. This crashed (SIGSEGV) whenever a `Result`/`Option` was the
    // program's terminal value (e.g. `fn main() -> Result<int,string>`
    // returning `Ok(0)` as the trailing expression). Project the typed
    // payload directly, mirroring `printing.rs`'s `HeapKind::Result` /
    // `HeapKind::Option` formatter arms.
    if hk == HeapKind::Result {
        // SAFETY: `KindedSlot::from_result` construction contract —
        // Result-kind bits are `Arc::into_raw(Arc<ResultData>)`.
        let r: &shape_value::heap_value::ResultData =
            unsafe { &*(bits as *const shape_value::heap_value::ResultData) };
        let inner = slot_to_wire(r.payload.raw(), r.payload.kind(), ctx);
        return WireValue::Result {
            ok: r.is_ok,
            value: Box::new(inner),
        };
    }
    if hk == HeapKind::Option {
        // SAFETY: `KindedSlot::from_option` construction contract —
        // Option-kind bits are `Arc::into_raw(Arc<OptionData>)`.
        let o: &shape_value::heap_value::OptionData =
            unsafe { &*(bits as *const shape_value::heap_value::OptionData) };
        if o.is_some {
            return slot_to_wire(o.payload.raw(), o.payload.kind(), ctx);
        }
        return WireValue::Null;
    }
    let ptr = bits as *const HeapValue;
    // SAFETY: NativeKind::Ptr(hk) contract — bits is a valid Arc<HeapValue> ptr.
    let hv = unsafe { &*ptr };
    debug_assert_eq!(
        hv.kind(),
        hk,
        "slot kind {:?} does not match HeapValue::{:?}",
        hk,
        hv.kind()
    );
    heap_value_to_wire(hv, ctx)
}

/// Project a `&HeapValue` to `WireValue` by dispatching on its
/// surviving variants. Reused by the snapshot path (Phase 2b
/// snapshot.rs commit) which has the same heap projection needs.
pub fn heap_value_to_wire(hv: &HeapValue, ctx: &Context) -> WireValue {
    match hv {
        HeapValue::String(s) => WireValue::String((**s).clone()),
        HeapValue::Decimal(d) => WireValue::Number(d.to_string().parse().unwrap_or(0.0)),
        HeapValue::BigInt(i) => WireValue::Integer(**i),
        HeapValue::Char(c) => WireValue::String(c.to_string()),
        HeapValue::Future(id) => WireValue::String(format!("<future:{}>", id)),
        HeapValue::DataTable(dt) => datatable_to_wire(dt.as_ref()),
        HeapValue::Content(_node) => {
            // Phase 1.B: the JSON-renderer integration for Content trees
            // is the deferred Phase 2c content-marshalling rebuild — see
            // ADR-006 §2.7.4. Until then, surface a placeholder
            // WireValue rather than emit a partial / wrong-shape
            // serialization.
            WireValue::String("<content:phase-2c-rebuild>".to_string())
        }
        HeapValue::Instant(t) => WireValue::String(format!("{:?}", **t)),
        HeapValue::IoHandle(_h) => {
            // Phase 1.B: IoHandleData no longer exposes a stable `id()`
            // accessor; the handle's identity is structural (the inner
            // OS resource) rather than a numeric tag. Phase 2c surfaces
            // a kind-threaded handle-printer.
            WireValue::String("<io_handle>".to_string())
        }
        HeapValue::NativeScalar(v) => match v {
            shape_value::heap_value::NativeScalar::I8(n) => WireValue::I8(*n),
            shape_value::heap_value::NativeScalar::U8(n) => WireValue::U8(*n),
            shape_value::heap_value::NativeScalar::I16(n) => WireValue::I16(*n),
            shape_value::heap_value::NativeScalar::U16(n) => WireValue::U16(*n),
            shape_value::heap_value::NativeScalar::I32(n) => WireValue::I32(*n),
            shape_value::heap_value::NativeScalar::I64(n) => WireValue::I64(*n),
            shape_value::heap_value::NativeScalar::U32(n) => WireValue::U32(*n),
            shape_value::heap_value::NativeScalar::U64(n) => WireValue::U64(*n),
            shape_value::heap_value::NativeScalar::Isize(n) => WireValue::Isize(*n as i64),
            shape_value::heap_value::NativeScalar::Usize(n) => WireValue::Usize(*n as u64),
            shape_value::heap_value::NativeScalar::Ptr(n) => WireValue::Ptr(*n as u64),
            shape_value::heap_value::NativeScalar::F32(n) => WireValue::F32(*n),
        },
        HeapValue::NativeView(v) => WireValue::Object(
            [
                (
                    "__type".to_string(),
                    WireValue::String(if v.mutable { "cmut" } else { "cview" }.to_string()),
                ),
                (
                    "layout".to_string(),
                    WireValue::String(v.layout.name.clone()),
                ),
                (
                    "ptr".to_string(),
                    WireValue::String(format!("0x{:x}", v.ptr)),
                ),
            ]
            .into_iter()
            .collect(),
        ),
        HeapValue::TypedObject(storage) => {
            // ADR-005 §Forbidden / Q10 forward pointer: wire serialization
            // must NOT re-introduce Box<HeapValue> slot wrapping. The
            // schema-driven kind threading below is ADR-005-aligned (typed
            // slot bits + schema; no intermediate HeapValue materialization
            // on deserialization).
            let schema_id = storage.schema_id;
            let slots = &storage.slots;
            let schema = ctx
                .type_schema_registry()
                .get_by_id(schema_id as u32)
                .cloned()
                .or_else(|| crate::type_schema::lookup_schema_by_id_public(schema_id as u32));
            if let Some(schema) = schema {
                let mut map = BTreeMap::new();
                for field_def in &schema.fields {
                    let idx = field_def.index as usize;
                    if idx >= slots.len() {
                        continue;
                    }
                    let Some(field_kind) = schema.field_kind(idx) else {
                        continue;
                    };
                    let field_bits = slots[idx].raw();
                    let field_wire = slot_to_wire(field_bits, field_kind, ctx);
                    map.insert(field_def.name.clone(), field_wire);
                }
                WireValue::Object(map)
            } else {
                WireValue::String(format!("<typed_object:schema#{}>", schema_id))
            }
        }
        HeapValue::ClosureRaw(_handle) => {
            // Phase 1.B: OwnedClosureBlock no longer exposes a public
            // `function_id()` accessor on the runtime side (the typed-
            // closure slot ABI carries the function-id via the
            // `TypedClosureHeader` itself). Phase 2c lands a
            // schema-aware closure printer.
            WireValue::String("<closure>".to_string())
        }
        HeapValue::TaskGroup(_data) => {
            WireValue::String("<task_group>".to_string())
        }
        // V3-S5 ckpt-5-prime (2026-05-15): `HeapValue::TypedArray(arc)` arm
        // RETIRED in lockstep with the deleted `HeapValue::TypedArray` variant
        // (ckpt-4) + deleted `TypedArrayData` inner enum (ckpt-1). Wire
        // serialisation of v2-raw `*mut TypedArray<T>` pointers lands at the
        // ckpt-5-prime² + ckpt-6 producer/consumer storage-shape migration
        // (per-element-type marshal-layer projection before the value becomes
        // a `HeapValue`). The `typed_array_to_wire` helper below is RETIRED
        // in the same lockstep. Refusal #1 binding.
        HeapValue::Temporal(td) => temporal_to_wire(&**td),
        HeapValue::TableView(tv) => match &**tv {
            shape_value::heap_value::TableViewData::TypedTable { table, schema_id } => {
                datatable_to_wire_with_schema(table.as_ref(), Some(*schema_id as u32))
            }
            shape_value::heap_value::TableViewData::IndexedTable { table, .. } => {
                datatable_to_wire(table.as_ref())
            }
            shape_value::heap_value::TableViewData::RowView { .. }
            | shape_value::heap_value::TableViewData::ColumnRef { .. } => {
                WireValue::String("<table_view:phase-2c>".to_string())
            }
        },
        HeapValue::HashMap(_) => {
            // Phase 1.B (ADR-006 §2.7.4): kind-threaded HashMap-to-wire
            // serialization is the deferred Phase 2c marshal rebuild.
            WireValue::String("<hashmap:phase-2c>".to_string())
        }
        // Wave 13 W13-hashset-rebuild (ADR-006 §2.7.15 / Q16,
        // 2026-05-10): Set wire serialization follows the same
        // phase-2c deferral shape as HashMap; surface as an opaque
        // tag until the marshal rebuild lands.
        HeapValue::HashSet(_) => WireValue::String("<hashset:phase-2c>".to_string()),
        // Wave 15 W15-deque (ADR-006 §2.7.19 / Q20, 2026-05-10):
        // Deque wire serialization follows the same phase-2c deferral
        // shape as HashMap / HashSet — opaque tag until the marshal
        // rebuild lands.
        HeapValue::Deque(_) => WireValue::String("<deque:phase-2c>".to_string()),
        // Wave-γ G-heap-filter-expr (ADR-006 §2.3 / Q8 amendment):
        // FilterExpr trees are transient query-DSL values; they don't
        // cross the wire boundary today. Surface as an opaque tag.
        HeapValue::FilterExpr(_) => WireValue::String("<filter_expr>".to_string()),
        // ADR-006 §2.7.13 / Q14 (Wave 8 W8-T26, 2026-05-10): Reference
        // values are within-program data and never cross the wire
        // boundary. Surface as an opaque tag, same as FilterExpr.
        HeapValue::Reference(_) => WireValue::String("<ref>".to_string()),
        // W13-iterator-state (ADR-006 §2.7.16 / Q17, 2026-05-10):
        // Iterator pipelines are lazy within-program values and never
        // cross the wire boundary (callers materialise via collect /
        // forEach / etc. before serialisation). Surface as an opaque
        // tag, same as FilterExpr / Reference.
        HeapValue::Iterator(_) => WireValue::String("<iterator>".to_string()),
        // Wave 15 W15-channel-rebuild (ADR-006 §2.7.20 / Q21, 2026-05-10):
        // channels are concurrency primitives with interior
        // `Mutex<ChannelInner>` state; no wire serialization at landing —
        // same phase-2c deferral shape as HashMap / HashSet. Surface as
        // an opaque tag for diagnostics.
        HeapValue::Channel(_) => WireValue::String("<channel:phase-2c>".to_string()),
        // Wave 15 W15-priority-queue (ADR-006 §2.7.18 / Q19,
        // 2026-05-10): PriorityQueue wire serialisation projects to a
        // `WireValue::Array` of i64 priorities in heap-array order
        // (mirror of the JSON shape — i64-priority-only at landing).
        HeapValue::PriorityQueue(d) => WireValue::Array(
            d.heap
                .iter()
                .map(|v| WireValue::Integer(*v))
                .collect(),
        ),
        // W15-range (ADR-006 §2.7.23 / Q24, 2026-05-10): Range
        // serializes as a JSON-ish `{"start", "end", "step",
        // "inclusive"}` payload via the `as_array_for_wire` shape
        // (range bounds + step are tiny scalars; lossless round-trip).
        // Wire serialization here just stamps the literal-form string
        // — full structured wire is the deferred Phase 2c marshal
        // rebuild same as HashMap / HashSet (which surface as opaque
        // tags above). Matches the playbook's "wire/JSON conversion
        // arms (rejection or proper)" guidance.
        HeapValue::Range(r) => {
            let s = if r.inclusive {
                format!("{}..={}", r.start, r.end)
            } else {
                format!("{}..{}", r.start, r.end)
            };
            WireValue::String(s)
        }
        // Wave 14 W14-variant-codegen (ADR-006 §2.7.17 / Q18, 2026-05-10):
        // Result/Option carriers are within-program control-flow values;
        // wire serialisation goes through the AnyError schema for thrown
        // errors and the unwrapped inner value for `Ok(_)` / `Some(_)`.
        // Until those marshal paths land, surface as an opaque tag —
        // same Phase-2c deferral shape as HashMap / HashSet / Iterator.
        HeapValue::Result(_) => WireValue::String("<result:phase-2c>".to_string()),
        HeapValue::Option(_) => WireValue::String("<option:phase-2c>".to_string()),
        // W17-concurrency (ADR-006 §2.7.25, 2026-05-11): concurrency
        // primitives are runtime-tier handles with no wire shape.
        // Surface as opaque tags — same Phase-2c deferral shape as
        // Channel / HashMap / HashSet.
        HeapValue::Mutex(_) => WireValue::String("<mutex:phase-2c>".to_string()),
        HeapValue::Atomic(_) => WireValue::String("<atomic:phase-2c>".to_string()),
        HeapValue::Lazy(_) => WireValue::String("<lazy:phase-2c>".to_string()),
        // W17-trait-object-storage (ADR-006 §2.7.24 / Q25.C, 2026-05-11):
        // `dyn Trait` carriers have no wire shape — same Phase-2c
        // deferral as concurrency primitives. A future `Serializable`
        // trait could route through the vtable, but that's emission-tier
        // work outside this sub-cluster.
        HeapValue::TraitObject(_) => WireValue::String("<trait_object:phase-2c>".to_string()),
        // W17-comptime-vm-dispatch (ADR-006 §2.7.26, 2026-05-12):
        // ModuleFn references are VM-internal callable handles
        // — same opaque-tag shape as the concurrency primitives.
        HeapValue::ModuleFn(id) => WireValue::String(format!("<module_fn:{}>", id)),
        // ADR-006 §2.7.22 amendment (Round 18 S3, 2026-05-13): Matrix /
        // MatrixSlice wire serialisation inherits the N7-architectural-
        // choice deferral from the pre-amendment
        // `TypedArrayData::Matrix` / `FloatSlice` shape (the 2D-layout
        // encoding policy is undecided). Surface as opaque tags —
        // same Phase-2c deferral pattern as the concurrency primitives.
        HeapValue::Matrix(m) => {
            WireValue::String(format!("<matrix:{}x{}:phase-2c>", m.rows, m.cols))
        }
        HeapValue::MatrixSlice(s) => {
            WireValue::String(format!("<matrix_slice:{}:phase-2c>", s.len))
        }
    }
}

// V3-S5 ckpt-5-prime (2026-05-15): `typed_array_to_wire` helper RETIRED per W12
// audit §3.6 + handover §0 wholesale-deletion cascade. The helper
// pattern-matched on the deleted `TypedArrayData` enum (retired at ckpt-1) and
// was called by the deleted `HeapValue::TypedArray` outer arm (retired at
// ckpt-4) above. The v2-raw `*mut TypedArray<T>` wire-serialisation path lands
// at the ckpt-5-prime² + ckpt-6 producer/consumer storage-shape migration
// (per-element-type marshal-layer projection before the value becomes a
// `HeapValue`). Refusal #1 binding.

fn temporal_to_wire(td: &shape_value::heap_value::TemporalData) -> WireValue {
    use shape_value::heap_value::TemporalData;
    match td {
        TemporalData::DateTime(dt) => WireValue::Timestamp(dt.timestamp_millis()),
        TemporalData::TimeSpan(d) => WireValue::Duration {
            value: d.num_milliseconds() as f64,
            unit: WireDurationUnit::Milliseconds,
        },
        TemporalData::Duration(d) => WireValue::Duration {
            value: d.value,
            unit: WireDurationUnit::Milliseconds,
        },
        TemporalData::Timeframe(_)
        | TemporalData::TimeReference(_)
        | TemporalData::DateTimeExpr(_)
        | TemporalData::DataDateTimeRef(_) => WireValue::String(format!("<{}>", td.type_name())),
    }
}

/// Project a `WireValue` to typed slot bits, given the kind the caller
/// wants. Returns [`MarshalError::KindMismatch`] when wire shape doesn't
/// match the expected kind.
///
/// For heap kinds, this allocates a new `Arc<HeapValue>` and returns
/// the raw pointer as bits — caller takes ownership of the heap
/// reference (one strong count).
pub fn wire_to_slot(wire: &WireValue, expected_kind: NativeKind) -> Result<u64, MarshalError> {
    match (wire, expected_kind) {
        (WireValue::Number(n), NativeKind::Float64) => Ok(f64::to_bits(*n)),
        (WireValue::Integer(i), NativeKind::Int64) => Ok(*i as u64),
        (WireValue::Bool(b), NativeKind::Bool) => Ok(*b as u64),
        (WireValue::Null, NativeKind::NullableFloat64) => Ok(f64::to_bits(f64::NAN)),
        (WireValue::String(s), NativeKind::String) => {
            let arc = Arc::new(s.clone());
            Ok(Arc::into_raw(arc) as u64)
        }
        (WireValue::I8(n), NativeKind::Int8) => Ok((*n as i64) as u64),
        (WireValue::I16(n), NativeKind::Int16) => Ok((*n as i64) as u64),
        (WireValue::I32(n), NativeKind::Int32) => Ok((*n as i64) as u64),
        (WireValue::U8(n), NativeKind::UInt8) => Ok(*n as u64),
        (WireValue::U16(n), NativeKind::UInt16) => Ok(*n as u64),
        (WireValue::U32(n), NativeKind::UInt32) => Ok(*n as u64),
        (WireValue::U64(n), NativeKind::UInt64) => Ok(*n),
        // Heap kinds are constructed by allocating Arc<HeapValue> with the
        // matching variant. Each surviving HeapKind variant is handled here
        // as stdlib mass migration (Phase 2c) and the snapshot replay path
        // discover concrete consumers.
        (WireValue::String(s), NativeKind::Ptr(HeapKind::String)) => {
            let arc = Arc::new(HeapValue::String(Arc::new(s.clone())));
            Ok(Arc::into_raw(arc) as u64)
        }
        (WireValue::Table(table), NativeKind::Ptr(HeapKind::DataTable)) => {
            let dt = datatable_from_ipc_bytes(&table.ipc_bytes, None, None)
                .map_err(MarshalError::Body)?;
            let arc = Arc::new(HeapValue::DataTable(Arc::new(dt)));
            Ok(Arc::into_raw(arc) as u64)
        }
        // Calling site passed a wire/kind pair we don't currently handle.
        // The strict-typed answer is to extend this match, not fall back —
        // each new case represents a concrete stdlib/wire shape, and
        // pattern-match exhaustiveness is the discipline.
        _ => Err(MarshalError::Body(format!(
            "wire_to_slot: no projection for wire variant into kind {:?}",
            expected_kind
        ))),
    }
}

/// Wrap a typed slot in a `ValueEnvelope` with optional metadata.
///
/// `type_name` is the user-facing Shape type name (e.g. `"int"`,
/// `"DataTable"`, `"MyType"`). The envelope's `type_info` is populated
/// from the type registry when available.
pub fn slot_to_envelope(
    bits: u64,
    kind: NativeKind,
    type_name: &str,
    ctx: &Context,
) -> ValueEnvelope {
    let value = slot_to_wire(bits, kind, ctx);
    let _ = type_name;
    let _ = ctx;
    // Phase 1.B (ADR-006 §2.7.4): the type-info / type-registry lookup
    // path that resolved a `TypeRegistry` from `TypeRegistry::default()`
    // is gone; the rebuilt path queries `TypeRegistry::for_number` /
    // primitives + the runtime's per-schema cache. Until the kind-
    // threaded envelope lookup lands in Phase 2c, fall back to the
    // wire-side inference helper.
    ValueEnvelope::from_value(value)
}

/// If the slot carries a renderable Content shape (Content node, DataTable,
/// or TableView), return `(content_json, content_html, content_terminal)`.
/// Otherwise all three are `None`.
///
/// W18.2 (R8 — output-adapter integration): the kind-threaded content
/// dispatch is rebuilt on top of the surviving 6-renderer infrastructure
/// (TERMINAL / HTML / MARKDOWN / JSON / PLAIN). Slot bits are dispatched
/// per `HeapKind`, the underlying typed payload is materialised as a
/// `ContentNode`, and each renderer projects the node into its own
/// output shape:
///
/// - JSON via [`crate::renderers::json::JsonRenderer`] parsed into
///   [`serde_json::Value`] (the renderer guarantees valid JSON per the
///   `renderers::cross_renderer_tests::json_is_valid_json` regression).
/// - HTML via [`crate::renderers::html::HtmlRenderer`].
/// - Terminal via [`crate::renderers::terminal::TerminalRenderer`] (ANSI
///   escapes; consumers route to PLAIN when the sink is non-TTY).
///
/// Pre-rebuild this function returned `(None, None, None)` for every
/// Content-bearing slot per the Phase 1.B placeholder — that scar is
/// resolved here.
pub fn slot_extract_content(
    bits: u64,
    kind: NativeKind,
) -> (Option<serde_json::Value>, Option<String>, Option<String>) {
    let NativeKind::Ptr(hk) = kind else {
        return (None, None, None);
    };
    if bits == 0 {
        return (None, None, None);
    }
    // SAFETY: `Ptr(HeapKind::*)` contract — bits is a valid typed pointer
    // for the labelled heap kind. The `(hk, hv)` cross-check below is
    // belt-and-braces; an inconsistent label is a producer-side bug.
    let node = match hk {
        HeapKind::Content => {
            let hv = unsafe { &*(bits as *const HeapValue) };
            match hv {
                HeapValue::Content(node) => Some((**node).clone()),
                _ => None,
            }
        }
        HeapKind::DataTable => {
            let dt: &shape_value::DataTable =
                unsafe { &*(bits as *const shape_value::DataTable) };
            Some(crate::content_dispatch::datatable_to_content_node(dt, None))
        }
        HeapKind::TableView => {
            let tv: &shape_value::heap_value::TableViewData =
                unsafe { &*(bits as *const shape_value::heap_value::TableViewData) };
            match tv {
                shape_value::heap_value::TableViewData::TypedTable { table, .. }
                | shape_value::heap_value::TableViewData::IndexedTable { table, .. } => Some(
                    crate::content_dispatch::datatable_to_content_node(table.as_ref(), None),
                ),
                // RowView / ColumnRef are deferred Phase 2c content
                // adapters — no current renderer.
                _ => None,
            }
        }
        _ => None,
    };
    let Some(node) = node else {
        return (None, None, None);
    };

    // W18.2: route the materialised `ContentNode` through each of the
    // three renderer adapters wired into the host boundary. The JSON
    // renderer's output is guaranteed-valid JSON (see
    // `renderers::cross_renderer_tests::json_is_valid_json`); the
    // `serde_json::from_str` fallback to `None` is defensive only.
    use crate::content_renderer::ContentRenderer;
    let json_renderer = crate::renderers::json::JsonRenderer;
    let html_renderer = crate::renderers::html::HtmlRenderer::new();
    let terminal_renderer = crate::renderers::terminal::TerminalRenderer::new();
    let content_json: Option<serde_json::Value> =
        serde_json::from_str(&json_renderer.render(&node)).ok();
    let content_html = Some(html_renderer.render(&node));
    let content_terminal = Some(terminal_renderer.render(&node));
    (content_json, content_html, content_terminal)
}

// ───────────────────────── DataTable ↔ wire/IPC ─────────────────────────
//
// Typed-handle conversions. These don't go through `(bits, kind)` —
// the caller passes a `&DataTable` directly, which is the typed-Rust
// equivalent of NativeKind::Ptr(HeapKind::DataTable). The marshal layer
// uses these internally when projecting a DataTable slot.

pub fn datatable_to_wire(dt: &DataTable) -> WireValue {
    datatable_to_wire_with_schema(dt, dt.schema_id())
}

fn datatable_to_wire_with_schema(dt: &DataTable, schema_id: Option<u32>) -> WireValue {
    match datatable_to_ipc_bytes(dt) {
        Ok(ipc_bytes) => WireValue::Table(WireTable {
            ipc_bytes,
            type_name: None,
            schema_id,
            row_count: dt.row_count(),
            column_count: dt.column_count(),
        }),
        Err(e) => WireValue::String(format!("<datatable_serialize_error: {}>", e)),
    }
}

pub fn datatable_to_ipc_bytes(dt: &DataTable) -> std::result::Result<Vec<u8>, String> {
    // The DataTable now wraps a `RecordBatch` directly (`inner()`); the
    // pre-bulldozer `to_arrow_batch` accessor is gone since the wrapper
    // is the batch.
    let arrow_batch = dt.inner();
    let schema = arrow_batch.schema();
    let mut buf = Vec::new();
    {
        let mut writer = FileWriter::try_new(&mut buf, &schema)
            .map_err(|e| format!("Arrow IPC writer init failed: {}", e))?;
        writer
            .write(arrow_batch)
            .map_err(|e| format!("Arrow IPC write failed: {}", e))?;
        writer
            .finish()
            .map_err(|e| format!("Arrow IPC finish failed: {}", e))?;
    }
    Ok(buf)
}

pub fn datatable_from_ipc_bytes(
    bytes: &[u8],
    column_overrides: Option<&[shape_value::datatable::ColumnPtrs]>,
    schema_id_override: Option<u32>,
) -> std::result::Result<DataTable, String> {
    let cursor = std::io::Cursor::new(bytes);
    let reader = FileReader::try_new(cursor, None)
        .map_err(|e| format!("Arrow IPC reader init failed: {}", e))?;
    let mut batches = Vec::new();
    for batch in reader {
        batches.push(batch.map_err(|e| format!("Arrow IPC batch read failed: {}", e))?);
    }
    if batches.is_empty() {
        return Err(
            "datatable_from_ipc_bytes: empty IPC stream — no Arrow RecordBatch to wrap".to_string(),
        );
    }
    // The first batch is the canonical wrapper; concatenation is a
    // Phase 2c rebuild item alongside the broader DataTable IPC layer.
    let first = batches.into_iter().next().unwrap();
    let _ = column_overrides;
    let dt = DataTable::new(first);
    let dt = if let Some(sid) = schema_id_override {
        dt.with_schema_id(sid)
    } else {
        dt
    };
    Ok(dt)
}

#[cfg(test)]
mod u64_wire_tests {
    //! R5c-2-β-γ checkpoint (b) u64-carrier — wire/snapshot round-trip.
    //!
    //! A `NativeKind::UInt64` slot must round-trip through MessagePack
    //! wire serialization with its full `0..2^64` range intact: a value
    //! above `i64::MAX` must NOT be lossy-projected to a signed
    //! `WireValue::Integer`. The carrier projects to `WireValue::U64`
    //! (full-range), and `wire_to_slot` recovers the exact bits. The
    //! snapshot path serializes the parallel `Vec<u64>` data + the slot's
    //! `NativeKind` verbatim (per ADR-006 §2.7.7), so the same bit/kind
    //! pair is preserved there by construction.

    use super::{slot_to_wire, wire_to_slot};
    use crate::context::ExecutionContext;
    use shape_value::NativeKind;
    use shape_wire::WireValue;

    fn roundtrip(bits: u64) -> u64 {
        let ctx = ExecutionContext::new_empty();
        let wire = slot_to_wire(bits, NativeKind::UInt64, &ctx);
        // Full-range u64 must project to the dedicated U64 wire variant,
        // not a lossy signed Integer.
        assert!(
            matches!(wire, WireValue::U64(_)),
            "u64 slot must project to WireValue::U64, got {:?}",
            wire
        );
        wire_to_slot(&wire, NativeKind::UInt64).expect("u64 wire should decode")
    }

    #[test]
    fn u64_max_round_trips() {
        assert_eq!(roundtrip(u64::MAX), u64::MAX);
    }

    #[test]
    fn u64_above_i64_max_round_trips_lossless() {
        // i64::MAX + 1 — the first value a signed projection would corrupt.
        let v = (i64::MAX as u64) + 1;
        assert_eq!(roundtrip(v), v);
    }

    #[test]
    fn u64_small_round_trips() {
        assert_eq!(roundtrip(0), 0);
        assert_eq!(roundtrip(42), 42);
    }

    #[test]
    fn u64_wire_variant_preserves_full_range() {
        // The wire value itself must carry the full unsigned magnitude.
        let ctx = ExecutionContext::new_empty();
        match slot_to_wire(u64::MAX, NativeKind::UInt64, &ctx) {
            WireValue::U64(n) => assert_eq!(n, u64::MAX),
            other => panic!("expected WireValue::U64, got {:?}", other),
        }
    }
}

#[cfg(test)]
mod char_wire_tests {
    //! β-fix CKPT-A char-carrier — `charAt` return-value wire projection.
    //!
    //! `op_string_char_at` (and its char-producing siblings) push a
    //! scalar Unicode codepoint. Pre-fix the slot was stamped
    //! `NativeKind::Ptr(HeapKind::Char)` — a scalar codepoint mislabeled
    //! as an `Arc<HeapValue>` pointer. When that slot was the program
    //! return value, `heap_to_wire` cast the codepoint bits (e.g. 0x63 =
    //! 'c') to `*const HeapValue` and dereferenced it, triggering a
    //! misaligned-pointer non-unwinding abort (SIGABRT).
    //!
    //! The producer-side fix stamps the canonical scalar
    //! `NativeKind::Char` (ADR-006 §2.7.5). The defensive `heap_to_wire`
    //! `HeapKind::Char` early-arm guarantees that even a mislabeled
    //! `Ptr(HeapKind::Char)` slot projects the codepoint directly
    //! instead of dereferencing it as a heap object.

    use super::{heap_to_wire, slot_to_wire};
    use crate::context::ExecutionContext;
    use shape_value::{HeapKind, NativeKind};
    use shape_wire::WireValue;

    /// The canonical post-fix carrier: a scalar `NativeKind::Char` slot
    /// (codepoint inline) projects to a single-codepoint string with no
    /// pointer dereference.
    #[test]
    fn native_kind_char_slot_projects_to_string() {
        let ctx = ExecutionContext::new_empty();
        // 'c' = U+0063 — the `"abc".reverse().charAt(0)` reproducer result.
        let wire = slot_to_wire('c' as u64, NativeKind::Char, &ctx);
        assert_eq!(wire, WireValue::String("c".to_string()));
    }

    /// `charAt(0)` of `"abc"` returns 'a' — direct (non-reversed) path.
    #[test]
    fn native_kind_char_slot_first_codepoint() {
        let ctx = ExecutionContext::new_empty();
        let wire = slot_to_wire('a' as u64, NativeKind::Char, &ctx);
        assert_eq!(wire, WireValue::String("a".to_string()));
    }

    /// Defensive: a `Ptr(HeapKind::Char)`-labeled slot (a mislabeled
    /// scalar codepoint, e.g. emitted by any un-migrated producer) must
    /// NOT be dereferenced as an `Arc<HeapValue>`. The `heap_to_wire`
    /// early-arm projects the codepoint directly. Pre-fix this input
    /// aborted the process with a misaligned-pointer panic.
    #[test]
    fn heap_kind_char_label_does_not_deref_codepoint() {
        let ctx = ExecutionContext::new_empty();
        // 0x63 ('c') is NOT 8-byte aligned and is not a valid HeapValue
        // pointer — the pre-fix catch-all would have aborted here.
        let wire = heap_to_wire('c' as u64, HeapKind::Char, &ctx);
        assert_eq!(wire, WireValue::String("c".to_string()));
    }

    /// Defensive arm also covers a non-ASCII multi-byte codepoint.
    #[test]
    fn heap_kind_char_label_handles_unicode_codepoint() {
        let ctx = ExecutionContext::new_empty();
        let wire = heap_to_wire('λ' as u64, HeapKind::Char, &ctx);
        assert_eq!(wire, WireValue::String("λ".to_string()));
    }

    /// A `Ptr(HeapKind::Char)` slot routed through the public
    /// `slot_to_wire` entry point (the program-return-value path) also
    /// projects safely — this is the exact path the SIGABRT reproducer
    /// exercised.
    #[test]
    fn slot_to_wire_char_label_return_value_path_is_safe() {
        let ctx = ExecutionContext::new_empty();
        let wire = slot_to_wire(
            'c' as u64,
            NativeKind::Ptr(HeapKind::Char),
            &ctx,
        );
        assert_eq!(wire, WireValue::String("c".to_string()));
    }
}

#[cfg(test)]
mod ws3_f2b_result_option_wire_tests {
    //! WS-3 F2b — `Result` / `Option` program-return-value wire projection.
    //!
    //! `HeapKind::Result` / `HeapKind::Option` are typed-Arc dispatch
    //! labels — their slot bits are `Arc::into_raw(Arc<ResultData>)` /
    //! `Arc::into_raw(Arc<OptionData>)`, NOT an `Arc<HeapValue>`. Pre-fix,
    //! `heap_to_wire`'s catch-all cast those bits to `*const HeapValue`
    //! and dereferenced — reading a `ResultData`/`OptionData` as a
    //! `HeapValue` enum (type confusion + UB). This crashed (SIGSEGV)
    //! whenever a `Result`/`Option` was the program's terminal value
    //! (e.g. `fn main() -> Result<int,string>` returning `Ok(0)`), which
    //! made every `?`-using program crash once it compiled.
    //!
    //! The fix adds dedicated `HeapKind::Result` / `HeapKind::Option`
    //! arms that read the typed payload directly.

    use super::slot_to_wire;
    use crate::context::ExecutionContext;
    use shape_value::heap_value::{OptionData, ResultData};
    use shape_value::kinded_slot::KindedSlot;
    use shape_wire::WireValue;
    use std::sync::Arc;

    #[test]
    fn ok_result_projects_to_wire_result_ok() {
        let ctx = ExecutionContext::new_empty();
        let payload = KindedSlot::from_int(42);
        let slot = KindedSlot::from_result(Arc::new(ResultData::ok(payload)));
        let wire = slot_to_wire(slot.raw(), slot.kind(), &ctx);
        match wire {
            WireValue::Result { ok, value } => {
                assert!(ok, "Ok(42) must wire with ok=true");
                assert_eq!(*value, WireValue::Integer(42));
            }
            other => panic!("expected WireValue::Result, got {:?}", other),
        }
    }

    #[test]
    fn err_result_projects_to_wire_result_err() {
        let ctx = ExecutionContext::new_empty();
        let payload = KindedSlot::from_int(7);
        let slot = KindedSlot::from_result(Arc::new(ResultData::err(payload)));
        let wire = slot_to_wire(slot.raw(), slot.kind(), &ctx);
        match wire {
            WireValue::Result { ok, value } => {
                assert!(!ok, "Err(7) must wire with ok=false");
                assert_eq!(*value, WireValue::Integer(7));
            }
            other => panic!("expected WireValue::Result, got {:?}", other),
        }
    }

    #[test]
    fn some_option_projects_to_inner_value() {
        let ctx = ExecutionContext::new_empty();
        let payload = KindedSlot::from_int(5);
        let slot = KindedSlot::from_option(Arc::new(OptionData::some(payload)));
        let wire = slot_to_wire(slot.raw(), slot.kind(), &ctx);
        // Null-coding semantics: `Some(x) ≡ x`.
        assert_eq!(wire, WireValue::Integer(5));
    }

    #[test]
    fn none_option_projects_to_null() {
        let ctx = ExecutionContext::new_empty();
        let slot = KindedSlot::from_option(Arc::new(OptionData::none()));
        let wire = slot_to_wire(slot.raw(), slot.kind(), &ctx);
        assert_eq!(wire, WireValue::Null);
    }
}