shape-vm 0.3.2

//! Per-function remote execution support
//!
//! This module provides the types and executor for transferring function
//! execution to another machine. The design sends the full compiled
//! `BytecodeProgram` + a "call this function with these args" message,
//! running it on a full Shape VM on the remote side.
//!
//! # Architecture
//!
//! ```text
//! Layer 4: @remote / @distributed annotations    (Shape stdlib — user-defined policy)
//! Layer 3: RemoteCallRequest/Response            (this module)
//! Layer 2: shape-wire codec (MessagePack)        (encode_message / decode_message)
//! Layer 1: Transport (TCP/QUIC/Unix socket)      (user-provided, pluggable)
//! ```
//!
//! Layer 0 (the foundation): Full Shape VM on both sides, same `BytecodeProgram`,
//! same `Executor`.
//!
//! # Closure semantics
//!
//! Upvalues (SharedCell-backed shared captures or Phase D typed-pointer
//! captures alike) become **value copies** on serialization. If the remote
//! side mutates a captured variable, the sender doesn't see it. This is the
//! correct semantic for distributed computing — a **send-copy** model.

use serde::{Deserialize, Serialize};
use shape_runtime::snapshot::{SerializableVMValue, SnapshotStore};
use shape_runtime::type_schema::TypeSchemaRegistry;

use shape_wire::WireValue;

use crate::bytecode::{BytecodeProgram, FunctionBlob, FunctionHash, Program};

// `execute_inner` / `execute_inner_with_runtimes` previously called
// `VirtualMachine::new` / `load_program` / `populate_module_objects` /
// `execute_*` / and round-tripped each argument and return value through
// `serializable_to_nanboxed_with_layouts` / `nanboxed_to_serializable`.
// Both round-trip helpers are deleted (see `crates/shape-runtime/src/snapshot.rs:649`
// "The slot-(de)serialization functions ... were deleted in Phase 2b") and
// their replacement is a kind-threaded `slot_to_serializable(bits, kind, store)`
// pair scheduled for the Phase-2c snapshot rebuild (ADR-006 §2.7.4). The
// execute paths are stubbed at the entry below; the rebuild lands the
// kind-threaded serializer + a `Vec<KindedSlot>` arg pipeline together.
//
// Imports `VMConfig` / `VirtualMachine` are pulled in lazily inside
// `execute_remote_call*` so the file still compiles when those are the
// only consumers and the `execute_inner*` bodies are stubbed.

/// Request to execute a function on a remote VM.
///
/// Contains everything needed to call a function: the full compiled program
/// (cacheable by `program_hash`), function identity, serialized arguments,
/// and optional closure captures.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RemoteCallRequest {
    /// The full compiled program. After the first transfer, the remote
    /// side caches by `program_hash` and subsequent calls only need args.
    pub program: BytecodeProgram,

    /// Function to call by name (for named functions).
    pub function_name: String,

    /// Function to call by ID (for closures that have no user-facing name).
    /// Takes precedence over `function_name` when `Some`.
    pub function_id: Option<u16>,

    /// Function to call by content hash (canonical identity).
    ///
    /// Preferred over name-based lookup when present. This avoids ambiguity
    /// when multiple modules define functions with the same name.
    #[serde(default)]
    pub function_hash: Option<FunctionHash>,

    /// Serialized arguments to the function.
    pub arguments: Vec<SerializableVMValue>,

    /// Closure upvalues, if calling a closure. These are value-copied from
    /// the sender's upvalue slots regardless of the local storage class
    /// (SharedCell, typed frame-pointer capture, or inline scalar).
    pub upvalues: Option<Vec<SerializableVMValue>>,

    /// Type schema registry — sent separately because `BytecodeProgram`
    /// has `#[serde(skip)]` on its registry (it's populated at compile time).
    pub type_schemas: TypeSchemaRegistry,

    /// Content hash of the program for caching. If the remote side has
    /// already seen this hash, it can skip deserializing the program.
    pub program_hash: [u8; 32],

    /// Minimal content-addressed blobs for the called function and its
    /// transitive dependencies. When present, the callee can reconstruct
    /// a `Program` from these blobs instead of deserializing the full
    /// `BytecodeProgram`, dramatically reducing payload size.
    #[serde(default)]
    pub function_blobs: Option<Vec<(FunctionHash, FunctionBlob)>>,
}

/// Response from a remote function execution.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RemoteCallResponse {
    /// The function's return value, or an error message.
    pub result: Result<SerializableVMValue, RemoteCallError>,
}

/// Error from remote execution.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RemoteCallError {
    /// Human-readable error message.
    pub message: String,
    /// Optional error kind for programmatic handling.
    pub kind: RemoteErrorKind,
}

/// Classification of remote execution errors.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum RemoteErrorKind {
    /// Function not found in the program.
    FunctionNotFound,
    /// Argument deserialization failed.
    ArgumentError,
    /// Runtime error during execution.
    RuntimeError,
    /// Module function required on the remote side is missing.
    MissingModuleFunction,
}

impl std::fmt::Display for RemoteCallError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{:?}: {}", self.kind, self.message)
    }
}

impl std::error::Error for RemoteCallError {}

// ---------------------------------------------------------------------------
// Wire message envelope (Phase 2: blob negotiation)
// ---------------------------------------------------------------------------

/// Envelope for all wire protocol messages.
///
/// Wraps the existing `RemoteCallRequest`/`RemoteCallResponse` with negotiation
/// and sidecar message types for bandwidth optimization on persistent connections.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum WireMessage {
    /// Offer function blob hashes to check what the remote already has.
    BlobNegotiation(BlobNegotiationRequest),
    /// Reply with the subset of offered hashes that are already cached.
    BlobNegotiationReply(BlobNegotiationResponse),
    /// A remote function call (may have blobs stripped if negotiation occurred).
    Call(RemoteCallRequest),
    /// Response to a remote function call.
    CallResponse(RemoteCallResponse),
    /// A large blob sent as a separate message before the call (Phase 3).
    Sidecar(BlobSidecar),

    // --- Execution server messages (V2) ---
    /// Execute Shape source code on the server.
    Execute(ExecuteRequest),
    /// Response to an Execute request.
    ExecuteResponse(ExecuteResponse),
    /// Validate Shape source code (parse + type-check) without executing.
    Validate(ValidateRequest),
    /// Response to a Validate request.
    ValidateResponse(ValidateResponse),
    /// Authenticate with the server (required for non-localhost).
    Auth(AuthRequest),
    /// Response to an Auth request.
    AuthResponse(AuthResponse),
    /// Execute a Shape file on the server.
    ExecuteFile(ExecuteFileRequest),
    /// Execute a Shape project (shape.toml) on the server.
    ExecuteProject(ExecuteProjectRequest),
    /// Validate a Shape file or project (parse + type-check) without executing.
    ValidatePath(ValidatePathRequest),
    /// Ping the server for liveness / capability discovery.
    Ping(PingRequest),
    /// Pong reply with server info.
    Pong(ServerInfo),
}

/// Ping request (empty payload for wire format consistency).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PingRequest {}

/// Request to check which function blobs the remote side already has cached.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BlobNegotiationRequest {
    /// Content hashes of function blobs the caller wants to send.
    pub offered_hashes: Vec<FunctionHash>,
}

/// Response indicating which offered blobs are already cached on the remote side.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BlobNegotiationResponse {
    /// Subset of offered hashes that the remote already has in its blob cache.
    pub known_hashes: Vec<FunctionHash>,
}

/// A large binary payload sent as a separate message before the call request.
///
/// Used for splitting large BlobRef-backed values (DataTables, TypedArrays, etc.)
/// out of the main serialized payload.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BlobSidecar {
    pub sidecar_id: u32,
    pub data: Vec<u8>,
}

// ---------------------------------------------------------------------------
// Execution server message types (V2)
// ---------------------------------------------------------------------------

/// Request to execute Shape source code on the server.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecuteRequest {
    /// Shape source code to execute.
    pub code: String,
    /// Client-assigned request ID for correlation.
    pub request_id: u64,
}

/// Response from executing Shape source code.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecuteResponse {
    /// The request ID this response corresponds to.
    pub request_id: u64,
    /// Whether execution completed successfully.
    pub success: bool,
    /// Structured return value from execution.
    pub value: WireValue,
    /// Print/log output captured during execution (NOT the return value).
    pub stdout: Option<String>,
    /// Error message (if execution failed).
    pub error: Option<String>,
    /// Pre-rendered Content terminal representation (if value is Content).
    #[serde(skip_serializing_if = "Option::is_none", default)]
    pub content_terminal: Option<String>,
    /// Pre-rendered Content HTML representation (if value is Content).
    #[serde(skip_serializing_if = "Option::is_none", default)]
    pub content_html: Option<String>,
    /// Diagnostics (parse errors, type errors, warnings).
    pub diagnostics: Vec<WireDiagnostic>,
    /// Execution metrics (if available).
    pub metrics: Option<ExecutionMetrics>,
    /// Structured print output with rendered strings (MsgPack-serialized).
    #[serde(skip_serializing_if = "Option::is_none", default)]
    pub print_output: Option<Vec<shape_wire::print_result::WirePrintResult>>,
}

/// Request to validate Shape source code without executing it.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ValidateRequest {
    /// Shape source code to validate.
    pub code: String,
    /// Client-assigned request ID for correlation.
    pub request_id: u64,
}

/// Response from validating Shape source code.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ValidateResponse {
    /// The request ID this response corresponds to.
    pub request_id: u64,
    /// Whether the code is valid (no errors).
    pub success: bool,
    /// Diagnostics (parse errors, type errors, warnings).
    pub diagnostics: Vec<WireDiagnostic>,
}

/// Request to execute a Shape file on the server.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecuteFileRequest {
    /// Absolute path to the .shape file.
    pub path: String,
    /// Optional working directory (defaults to file's parent).
    pub cwd: Option<String>,
    /// Client-assigned request ID for correlation.
    pub request_id: u64,
}

/// Request to execute a Shape project (shape.toml) on the server.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecuteProjectRequest {
    /// Absolute path to the project directory (must contain shape.toml).
    pub project_dir: String,
    /// Client-assigned request ID for correlation.
    pub request_id: u64,
}

/// Request to validate a Shape file or project without executing.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ValidatePathRequest {
    /// Path to a .shape file or a project directory (containing shape.toml).
    pub path: String,
    /// Client-assigned request ID for correlation.
    pub request_id: u64,
}

/// Authentication request for non-localhost connections.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AuthRequest {
    /// Bearer token for authentication.
    pub token: String,
}

/// Authentication response.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AuthResponse {
    /// Whether authentication succeeded.
    pub authenticated: bool,
    /// Error message if authentication failed.
    pub error: Option<String>,
}

/// Server information returned in Pong responses.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ServerInfo {
    /// Shape language version.
    pub shape_version: String,
    /// Wire protocol version.
    pub wire_protocol: u32,
    /// Server capabilities (e.g., "execute", "validate", "call", "blob-negotiation").
    pub capabilities: Vec<String>,
}

/// A diagnostic message (error, warning, info).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WireDiagnostic {
    /// Severity: "error", "warning", "info".
    pub severity: String,
    /// Human-readable diagnostic message.
    pub message: String,
    /// Source line number (1-indexed), if available.
    pub line: Option<u32>,
    /// Source column number (1-indexed), if available.
    pub column: Option<u32>,
}

/// Execution performance metrics.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutionMetrics {
    /// Number of VM instructions executed.
    pub instructions_executed: u64,
    /// Wall-clock time in milliseconds.
    pub wall_time_ms: u64,
    /// Peak memory usage in bytes.
    pub memory_bytes_peak: u64,
}

// ---------------------------------------------------------------------------
// Per-connection blob cache (Phase 2)
// ---------------------------------------------------------------------------

/// Per-connection cache of function blobs received from a remote peer.
///
/// Content hashes make stale entries harmless (same hash = same content),
/// so no invalidation protocol is needed. LRU eviction bounds memory usage.
pub struct RemoteBlobCache {
    blobs: std::collections::HashMap<FunctionHash, FunctionBlob>,
    /// Access order for LRU eviction (most recently used at the end).
    order: Vec<FunctionHash>,
    /// Maximum number of entries before LRU eviction kicks in.
    max_entries: usize,
}

impl RemoteBlobCache {
    /// Create a new blob cache with the given capacity.
    pub fn new(max_entries: usize) -> Self {
        Self {
            blobs: std::collections::HashMap::new(),
            order: Vec::new(),
            max_entries,
        }
    }

    /// Default cache with 4096 entry capacity.
    pub fn default_cache() -> Self {
        Self::new(4096)
    }

    /// Insert a blob, evicting the least recently used entry if at capacity.
    pub fn insert(&mut self, hash: FunctionHash, blob: FunctionBlob) {
        if self.blobs.contains_key(&hash) {
            // Move to end (most recently used)
            self.order.retain(|h| h != &hash);
            self.order.push(hash);
            return;
        }

        // Evict LRU if at capacity
        while self.blobs.len() >= self.max_entries && !self.order.is_empty() {
            let evicted = self.order.remove(0);
            self.blobs.remove(&evicted);
        }

        self.blobs.insert(hash, blob);
        self.order.push(hash);
    }

    /// Look up a cached blob by hash, updating access order.
    pub fn get(&mut self, hash: &FunctionHash) -> Option<&FunctionBlob> {
        if self.blobs.contains_key(hash) {
            self.order.retain(|h| h != hash);
            self.order.push(*hash);
            self.blobs.get(hash)
        } else {
            None
        }
    }

    /// Check if a hash is cached without updating access order.
    pub fn contains(&self, hash: &FunctionHash) -> bool {
        self.blobs.contains_key(hash)
    }

    /// Return all cached hashes.
    pub fn known_hashes(&self) -> Vec<FunctionHash> {
        self.blobs.keys().copied().collect()
    }

    /// Return the subset of `offered` hashes that are in the cache.
    pub fn filter_known(&self, offered: &[FunctionHash]) -> Vec<FunctionHash> {
        offered
            .iter()
            .filter(|h| self.blobs.contains_key(h))
            .copied()
            .collect()
    }

    /// Number of cached entries.
    pub fn len(&self) -> usize {
        self.blobs.len()
    }

    /// Whether the cache is empty.
    pub fn is_empty(&self) -> bool {
        self.blobs.is_empty()
    }

    /// Insert all blobs from a set, typically received from a remote call.
    pub fn insert_blobs(&mut self, blobs: &[(FunctionHash, FunctionBlob)]) {
        for (hash, blob) in blobs {
            self.insert(*hash, blob.clone());
        }
    }
}

/// Build a minimal set of function blobs for a function hash and its
/// transitive dependencies from a content-addressed `Program`.
///
/// Returns `None` if the program has no content-addressed representation
/// or the entry hash is not present in the function store.
pub fn build_minimal_blobs_by_hash(
    program: &BytecodeProgram,
    entry_hash: FunctionHash,
) -> Option<Vec<(FunctionHash, FunctionBlob)>> {
    let ca = program.content_addressed.as_ref()?;
    if !ca.function_store.contains_key(&entry_hash) {
        return None;
    }

    // Compute transitive closure of dependencies
    let mut needed: std::collections::HashSet<FunctionHash> = std::collections::HashSet::new();
    let mut queue = vec![entry_hash];
    while let Some(hash) = queue.pop() {
        if needed.insert(hash) {
            if let Some(blob) = ca.function_store.get(&hash) {
                for dep in &blob.dependencies {
                    if !needed.contains(dep) {
                        queue.push(*dep);
                    }
                }
            }
        }
    }

    // Collect the minimal blob set
    let blobs: Vec<(FunctionHash, FunctionBlob)> = needed
        .into_iter()
        .filter_map(|hash| {
            ca.function_store
                .get(&hash)
                .map(|blob| (hash, blob.clone()))
        })
        .collect();

    Some(blobs)
}

/// Backwards-compatible name-based wrapper around `build_minimal_blobs_by_hash`.
///
/// If multiple blobs share the same name, this returns `None` to avoid
/// ambiguous, potentially incorrect dependency selection.
pub fn build_minimal_blobs(
    program: &BytecodeProgram,
    fn_name: &str,
) -> Option<Vec<(FunctionHash, FunctionBlob)>> {
    let ca = program.content_addressed.as_ref()?;
    let mut matches = ca.function_store.iter().filter_map(|(hash, blob)| {
        if blob.name == fn_name {
            Some(*hash)
        } else {
            None
        }
    });
    let first = matches.next()?;
    if matches.next().is_some() {
        return None;
    }
    build_minimal_blobs_by_hash(program, first)
}

/// Build a minimal `Program` from function blobs and an explicit entry hash.
///
/// Used on the callee side to reconstruct a `Program` from blobs received in
/// a `RemoteCallRequest`.
pub fn program_from_blobs_by_hash(
    blobs: Vec<(FunctionHash, FunctionBlob)>,
    entry_hash: FunctionHash,
    source: &BytecodeProgram,
) -> Option<Program> {
    let function_store: std::collections::HashMap<FunctionHash, FunctionBlob> =
        blobs.into_iter().collect();
    if !function_store.contains_key(&entry_hash) {
        return None;
    }

    Some(Program {
        entry: entry_hash,
        function_store,
        top_level_locals_count: source.top_level_locals_count,
        top_level_local_storage_hints: source.top_level_local_storage_hints.clone(),
        module_binding_names: source.module_binding_names.clone(),
        module_binding_storage_hints: source.module_binding_storage_hints.clone(),
        function_local_storage_hints: source.function_local_storage_hints.clone(),
        top_level_frame: source.top_level_frame.clone(),
        top_level_local_concrete_types: source.top_level_local_concrete_types.clone(),
        function_local_concrete_types: source.function_local_concrete_types.clone(),
        function_return_concrete_types: source.function_return_concrete_types.clone(),
        monomorphized_method_call_sites: source.monomorphized_method_call_sites.clone(),
        value_call_return_concrete_types:
            source.value_call_return_concrete_types.clone(),
        operator_trait_dispatch_sites:
            source.operator_trait_dispatch_sites.clone(),
        data_schema: source.data_schema.clone(),
        type_schema_registry: source.type_schema_registry.clone(),
        trait_method_symbols: source.trait_method_symbols.clone(),
        foreign_functions: source.foreign_functions.clone(),
        native_struct_layouts: source.native_struct_layouts.clone(),
        debug_info: source.debug_info.clone(),
        // Closure spec §14.6 (H6.5): propagate the per-name layout side-
        // table. Remote-stream origins that lack the side-table fail
        // hard at the VM producer; there is no legacy fallback.
        closure_function_layouts_by_name: source
            .content_addressed
            .as_ref()
            .map(|ca| ca.closure_function_layouts_by_name.clone())
            .unwrap_or_default(),
        // ADR-006 §2.7.24 Q25.C: propagate trait-object vtables from
        // the source BytecodeProgram so remote-streamed programs can
        // dispatch dyn method calls.
        trait_vtables: source.trait_vtables.clone(),
        // R8 W8 Cluster A surface-and-stop flag propagation.
        has_imported_const_inline: source.has_imported_const_inline,
        // R8 W9 B1 W17-marshal-return surface-and-stop flag propagation.
        has_w17_marshal_residual: source.has_w17_marshal_residual,
    })
}

/// Backwards-compatible name-based wrapper around `program_from_blobs_by_hash`.
pub fn program_from_blobs(
    blobs: Vec<(FunctionHash, FunctionBlob)>,
    fn_name: &str,
    source: &BytecodeProgram,
) -> Option<Program> {
    let mut matches = blobs.iter().filter_map(|(hash, blob)| {
        if blob.name == fn_name {
            Some(*hash)
        } else {
            None
        }
    });
    let entry = matches.next()?;
    if matches.next().is_some() {
        return None;
    }
    program_from_blobs_by_hash(blobs, entry, source)
}

/// Execute a remote call request on this machine.
///
/// This is the entry point for the receiving side. It:
/// 1. Reconstructs the `BytecodeProgram` and populates its `TypeSchemaRegistry`
/// 2. Creates a full `VirtualMachine` with the program
/// 3. Materializes serialized arguments as kinded VM slots
/// 4. Calls the function by name or ID
/// 5. Converts the result back to `SerializableVMValue`
///
/// The `store` is used for `SerializableVMValue` ↔ slot conversion
/// (needed for `BlobRef`-backed values like DataTable).
///
/// Phase-2c deferral: the slot/serializable round-trip is currently
/// stubbed (see `execute_inner` body). The dispatch entry continues to
/// exist so callers compile through the deferral; invocation surfaces
/// the gap at runtime.
pub fn execute_remote_call(
    request: RemoteCallRequest,
    store: &SnapshotStore,
) -> RemoteCallResponse {
    match execute_inner(request, store) {
        Ok(value) => RemoteCallResponse { result: Ok(value) },
        Err(err) => RemoteCallResponse { result: Err(err) },
    }
}

/// Execute a remote call with pre-loaded language runtime extensions.
///
/// `language_runtimes` maps language IDs (e.g. "python") to pre-loaded
/// runtime handles. The server loads these once at startup from installed
/// extensions. The bytecode carries foreign function source text; the
/// runtime on the server compiles and executes it.
pub fn execute_remote_call_with_runtimes(
    request: RemoteCallRequest,
    store: &SnapshotStore,
    language_runtimes: &std::collections::HashMap<
        String,
        std::sync::Arc<shape_runtime::plugins::language_runtime::PluginLanguageRuntime>,
    >,
) -> RemoteCallResponse {
    match execute_inner_with_runtimes(request, store, language_runtimes) {
        Ok(value) => RemoteCallResponse { result: Ok(value) },
        Err(err) => RemoteCallResponse { result: Err(err) },
    }
}

fn execute_inner(
    request: RemoteCallRequest,
    store: &SnapshotStore,
) -> Result<SerializableVMValue, RemoteCallError> {
    // T1-host-tier-marshal-rebuild (ADR-006 §2.7.4, R8 2026-05-23):
    // kind-threaded marshal protocol via
    // `shape_runtime::snapshot::serializable_to_slot` (in) +
    // `shape_runtime::snapshot::slot_to_serializable` (out). Each arg's
    // expected kind is read from the callee's `frame_descriptor.slots`
    // (i.e. the per-slot proven `NativeKind` per ADR-006 §2.7.5.1 — no
    // `Unknown` placeholder). The return-kind is read from the callee's
    // `frame_descriptor.return_kind`.
    //
    // Per the §0.A.iv supervisor ruling, frame-descriptor absence
    // produces a structured `RemoteCallError` (no silent-degrade): a
    // remote call cannot proceed if the callee has no proven param
    // kinds, because the marshal protocol cannot pick an in-arm.
    run_remote_call(request, store, None)
}

fn execute_inner_with_runtimes(
    request: RemoteCallRequest,
    store: &SnapshotStore,
    language_runtimes: &std::collections::HashMap<
        String,
        std::sync::Arc<shape_runtime::plugins::language_runtime::PluginLanguageRuntime>,
    >,
) -> Result<SerializableVMValue, RemoteCallError> {
    // Same path as `execute_inner` plus the foreign-function language-
    // runtime hookup. T1-host-tier-marshal-rebuild covers the marshal
    // protocol; the language-runtime registration is forwarded through
    // `run_remote_call` so the VM picks up the runtimes before invoking
    // the callee.
    run_remote_call(request, store, Some(language_runtimes))
}

/// Shared marshal+dispatch core for `execute_inner` /
/// `execute_inner_with_runtimes`. Per ADR-006 §2.7.4 the protocol is:
///
/// 1. Reconstruct the `BytecodeProgram` (full payload or from blobs).
/// 2. Build a `VirtualMachine`, load the program, populate module objects.
/// 3. Resolve the callee (hash → id → name precedence).
/// 4. Read the callee's per-slot `NativeKind` from `frame_descriptor.slots`
///    and the return kind from `frame_descriptor.return_kind`.
/// 5. Materialize each `SerializableVMValue` arg into a `KindedSlot` via
///    `serializable_to_slot(arg, expected_kind, store)`.
/// 6. Invoke the callee through the kinded ABI (`execute_function_by_id`).
/// 7. Project the returned `KindedSlot` to `SerializableVMValue` via
///    `slot_to_serializable(bits, kind, store)`.
///
/// Closure upvalue marshal (`request.upvalues` and `execute_closure`) is
/// NOT covered by T1: the per-capture kind track lives on the closure
/// header (ADR-006 §2.7.8 / Q10 cell-storage parallel-kind), which is
/// rebuilt in a downstream sub-cluster. The body surfaces this with a
/// structured error rather than silently dispatching.
fn run_remote_call(
    request: RemoteCallRequest,
    store: &SnapshotStore,
    language_runtimes: Option<
        &std::collections::HashMap<
            String,
            std::sync::Arc<shape_runtime::plugins::language_runtime::PluginLanguageRuntime>,
        >,
    >,
) -> Result<SerializableVMValue, RemoteCallError> {
    use crate::executor::{VMConfig, VirtualMachine};
    use shape_runtime::snapshot::{serializable_to_slot, slot_to_serializable};
    use shape_value::{KindedSlot, ValueSlot};

    // Forward the language_runtimes registration through the VM hookup
    // path. The current VM API does not expose a register-language-
    // runtime entry-point; consumers that need foreign-function support
    // through the remote-call boundary should pre-load runtimes via the
    // extension pipeline before dispatch. T1 stages the parameter
    // surface so downstream W17-foreign-ffi can wire it up.
    let _ = language_runtimes;

    // Step 1: reconstruct the program. If function_blobs are supplied,
    // build a content-addressed Program; otherwise use the full payload.
    let mut program: BytecodeProgram = request.program;
    program.type_schema_registry = request.type_schemas;

    if let (Some(blobs), Some(entry_hash)) =
        (request.function_blobs.clone(), request.function_hash)
    {
        if let Some(ca) = program_from_blobs_by_hash(blobs, entry_hash, &program) {
            program.content_addressed = Some(ca);
        }
    }

    // Closures need per-capture kinded marshal (ADR-006 §2.7.8 / Q10);
    // T1 does not cover that path. Surface-and-stop rather than
    // dispatch a possibly-wrong call.
    if request.upvalues.is_some() {
        return Err(RemoteCallError {
            message: "remote closure dispatch requires upvalue kind track \
                      (ADR-006 §2.7.8 / Q10 cell-storage parallel-kind) — \
                      not yet wired through the remote-call boundary"
                .to_string(),
            kind: RemoteErrorKind::RuntimeError,
        });
    }

    // Step 2: build VM and load program.
    let mut vm = VirtualMachine::new(VMConfig::default());
    vm.load_program(program);
    vm.populate_module_objects();

    // Step 3: resolve callee. function_hash (canonical) > function_id > name.
    let func_id: u16 = if let Some(hash) = request.function_hash {
        vm.program
            .function_blob_hashes
            .iter()
            .position(|h| *h == Some(hash))
            .map(|p| p as u16)
            .or_else(|| request.function_id)
            .or_else(|| {
                vm.program
                    .functions
                    .iter()
                    .position(|f| f.name == request.function_name)
                    .map(|p| p as u16)
            })
            .ok_or_else(|| RemoteCallError {
                message: format!(
                    "function not found by hash; name='{}', id={:?}",
                    request.function_name, request.function_id,
                ),
                kind: RemoteErrorKind::FunctionNotFound,
            })?
    } else if let Some(id) = request.function_id {
        id
    } else {
        vm.program
            .functions
            .iter()
            .position(|f| f.name == request.function_name)
            .map(|p| p as u16)
            .ok_or_else(|| RemoteCallError {
                message: format!("function '{}' not found", request.function_name),
                kind: RemoteErrorKind::FunctionNotFound,
            })?
    };

    // Step 4: pick per-arg expected kinds from the callee's frame
    // descriptor. ADR-006 §2.7.5.1: a present FunctionBlob has every
    // slot's NativeKind proven — no Unknown placeholder.
    let function = vm
        .program
        .functions
        .get(func_id as usize)
        .ok_or_else(|| RemoteCallError {
            message: format!("function_id {} out of range", func_id),
            kind: RemoteErrorKind::FunctionNotFound,
        })?;

    let arity = function.arity as usize;
    if request.arguments.len() != arity {
        return Err(RemoteCallError {
            message: format!(
                "argument count mismatch for function '{}': expected {}, got {}",
                function.name,
                arity,
                request.arguments.len(),
            ),
            kind: RemoteErrorKind::ArgumentError,
        });
    }

    let frame_desc = function.frame_descriptor.clone();
    let arg_kinds: Vec<shape_value::NativeKind> = if let Some(ref fd) = frame_desc {
        if fd.slots.len() < arity {
            return Err(RemoteCallError {
                message: format!(
                    "function '{}' frame_descriptor has {} slots but arity is {}",
                    function.name,
                    fd.slots.len(),
                    arity,
                ),
                kind: RemoteErrorKind::ArgumentError,
            });
        }
        fd.slots.iter().take(arity).copied().collect()
    } else if arity == 0 {
        Vec::new()
    } else {
        return Err(RemoteCallError {
            message: format!(
                "function '{}' has no frame_descriptor — cannot derive \
                 per-arg NativeKind for marshal protocol (ADR-006 §2.7.5.1)",
                function.name,
            ),
            kind: RemoteErrorKind::ArgumentError,
        });
    };

    let return_kind = frame_desc.as_ref().and_then(|fd| fd.return_kind);
    let function_name_owned = function.name.clone();
    let _ = function; // release the borrow before moving vm into call

    // Step 5: marshal each SerializableVMValue → KindedSlot per
    // expected_kind. `serializable_to_slot` allocates strong-count
    // shares for heap-kinded args; each share transfers into the
    // callee's frame via `execute_function_by_id`'s share-neutral
    // call-helper (per cluster-1.5 fix).
    let mut args: Vec<KindedSlot> = Vec::with_capacity(arity);
    for (idx, sv) in request.arguments.iter().enumerate() {
        let expected = arg_kinds[idx];
        let (bits, kind) = serializable_to_slot(sv, expected, store).map_err(|e| {
            RemoteCallError {
                message: format!(
                    "arg {} marshal failure (expected kind {:?}): {}",
                    idx, expected, e,
                ),
                kind: RemoteErrorKind::ArgumentError,
            }
        })?;
        args.push(KindedSlot::new(ValueSlot::from_raw(bits), kind));
    }

    // Step 6: dispatch.
    let result = vm
        .execute_function_by_id(func_id, args, None)
        .map_err(|e| RemoteCallError {
            message: format!(
                "remote execution of '{}' failed: {:?}",
                function_name_owned, e,
            ),
            kind: RemoteErrorKind::RuntimeError,
        })?;

    // Step 7: project the returned KindedSlot → SerializableVMValue.
    // The slot owns one strong-count share that we must release after
    // serialization (slot_to_serializable does NOT consume the share —
    // it borrows). `KindedSlot::Drop` retires it at scope exit.
    let (bits, kind) = (result.slot.raw(), result.kind);
    // Cross-check: if the program declared a top-level return_kind, the
    // returned slot's kind must agree. Mismatch is a structured error
    // rather than a silent reinterpretation (ADR-006 §2.7.7 / Q9).
    if let Some(declared) = return_kind {
        if kind != declared {
            return Err(RemoteCallError {
                message: format!(
                    "function '{}' returned kind {:?} but frame_descriptor \
                     declared return_kind {:?}",
                    function_name_owned, kind, declared,
                ),
                kind: RemoteErrorKind::RuntimeError,
            });
        }
    }
    let serialized = slot_to_serializable(bits, kind, store).map_err(|e| RemoteCallError {
        message: format!("return-value marshal failure: {}", e),
        kind: RemoteErrorKind::RuntimeError,
    })?;
    // `result` drops here, retiring the strong-count share via
    // `KindedSlot::Drop` (ADR-006 §2.7.6 / Q8).
    drop(result);
    Ok(serialized)
}

/// Compute a SHA-256 hash of a `BytecodeProgram` for caching.
///
/// Remote VMs can cache programs by this hash, avoiding re-transfer
/// of the same program on repeated calls.
pub fn program_hash(program: &BytecodeProgram) -> [u8; 32] {
    use sha2::{Digest, Sha256};
    let bytes =
        rmp_serde::to_vec_named(program).expect("BytecodeProgram serialization should not fail");
    let hash = Sha256::digest(&bytes);
    let mut out = [0u8; 32];
    out.copy_from_slice(&hash);
    out
}

/// Create a minimal stub program containing only metadata (no instructions/constants/functions).
///
/// Used by `build_call_request` and `build_closure_call_request` when content-addressed
/// blobs are available, to reduce payload size.
fn create_stub_program(program: &BytecodeProgram) -> BytecodeProgram {
    let mut stub = BytecodeProgram::default();
    stub.type_schema_registry = program.type_schema_registry.clone();
    // Carry enough content-addressed metadata for program_from_blobs()
    if let Some(ref ca) = program.content_addressed {
        stub.content_addressed = Some(Program {
            entry: ca.entry,
            function_store: std::collections::HashMap::new(),
            top_level_locals_count: ca.top_level_locals_count,
            top_level_local_storage_hints: ca.top_level_local_storage_hints.clone(),
            module_binding_names: ca.module_binding_names.clone(),
            module_binding_storage_hints: ca.module_binding_storage_hints.clone(),
            function_local_storage_hints: ca.function_local_storage_hints.clone(),
            top_level_frame: ca.top_level_frame.clone(),
            top_level_local_concrete_types: ca.top_level_local_concrete_types.clone(),
            function_local_concrete_types: ca.function_local_concrete_types.clone(),
            function_return_concrete_types: ca.function_return_concrete_types.clone(),
            monomorphized_method_call_sites: ca.monomorphized_method_call_sites.clone(),
            value_call_return_concrete_types:
                ca.value_call_return_concrete_types.clone(),
            operator_trait_dispatch_sites:
                ca.operator_trait_dispatch_sites.clone(),
            data_schema: ca.data_schema.clone(),
            type_schema_registry: ca.type_schema_registry.clone(),
            trait_method_symbols: ca.trait_method_symbols.clone(),
            foreign_functions: ca.foreign_functions.clone(),
            native_struct_layouts: ca.native_struct_layouts.clone(),
            debug_info: ca.debug_info.clone(),
            closure_function_layouts_by_name: ca
                .closure_function_layouts_by_name
                .clone(),
            trait_vtables: ca.trait_vtables.clone(),
            // R8 W8 Cluster A surface-and-stop flag propagation.
            has_imported_const_inline: ca.has_imported_const_inline,
            // R8 W9 B1 W17-marshal-return surface-and-stop flag propagation.
            has_w17_marshal_residual: ca.has_w17_marshal_residual,
        });
    }
    // Copy top-level metadata needed by program_from_blobs
    stub.top_level_locals_count = program.top_level_locals_count;
    stub.top_level_local_storage_hints = program.top_level_local_storage_hints.clone();
    stub.module_binding_names = program.module_binding_names.clone();
    stub.module_binding_storage_hints = program.module_binding_storage_hints.clone();
    stub.function_local_storage_hints = program.function_local_storage_hints.clone();
    stub.data_schema = program.data_schema.clone();
    stub.trait_method_symbols = program.trait_method_symbols.clone();
    stub.foreign_functions = program.foreign_functions.clone();
    stub.native_struct_layouts = program.native_struct_layouts.clone();
    stub.debug_info = program.debug_info.clone();
    stub.function_blob_hashes = program.function_blob_hashes.clone();
    stub
}

/// Perform blob negotiation before sending a call request.
///
/// Creates a `BlobNegotiationRequest` with the hashes from the blob set,
/// checks which blobs the remote already has (via the provided cache as a
/// local stand-in), and returns the set of known hashes that can be stripped
/// from the outgoing request.
///
/// In a real transport scenario the `BlobNegotiationRequest` would be sent
/// over the wire and the `BlobNegotiationResponse` received from the remote.
/// Currently this performs the negotiation locally against the provided cache.
///
/// # Example flow
/// ```text
/// 1. Caller builds blob set for function
/// 2. negotiate_blobs() → BlobNegotiationRequest with offered hashes
/// 3. Remote replies with BlobNegotiationResponse (known_hashes)
/// 4. Caller strips known blobs from the request
/// ```
pub fn negotiate_blobs(
    blobs: &[(FunctionHash, FunctionBlob)],
    remote_cache: &RemoteBlobCache,
) -> BlobNegotiationResponse {
    let request = BlobNegotiationRequest {
        offered_hashes: blobs.iter().map(|(h, _)| *h).collect(),
    };
    // TODO: Wire this to actual transport — currently performs negotiation
    // locally against the provided cache. In production, `request` would be
    // serialized, sent over the wire, and the response deserialized.
    handle_negotiation(&request, remote_cache)
}

/// Build a `RemoteCallRequest` for a named function, with blob negotiation.
///
/// Performs a negotiation step against the provided `remote_cache` to discover
/// which blobs the remote already has, then strips those from the request.
/// If `remote_cache` is `None`, sends all blobs (no negotiation).
pub fn build_call_request_with_negotiation(
    program: &BytecodeProgram,
    function_name: &str,
    arguments: Vec<SerializableVMValue>,
    remote_cache: Option<&RemoteBlobCache>,
) -> RemoteCallRequest {
    let mut request = build_call_request(program, function_name, arguments);

    if let (Some(cache), Some(blobs)) = (remote_cache, &mut request.function_blobs) {
        let response = negotiate_blobs(blobs, cache);
        let known_set: std::collections::HashSet<FunctionHash> =
            response.known_hashes.into_iter().collect();
        blobs.retain(|(hash, _)| !known_set.contains(hash));
    }

    request
}

/// Build a `RemoteCallRequest` for a named function.
///
/// Convenience function that handles program hashing and type schema extraction.
/// When the program has content-addressed blobs, automatically computes the
/// minimal transitive closure and attaches it to the request.
pub fn build_call_request(
    program: &BytecodeProgram,
    function_name: &str,
    arguments: Vec<SerializableVMValue>,
) -> RemoteCallRequest {
    let hash = program_hash(program);
    let function_id = program
        .functions
        .iter()
        .position(|f| f.name == function_name)
        .map(|id| id as u16);
    let function_hash = function_id
        .and_then(|fid| {
            program
                .function_blob_hashes
                .get(fid as usize)
                .copied()
                .flatten()
        })
        .or_else(|| {
            program.content_addressed.as_ref().and_then(|ca| {
                let mut matches = ca.function_store.iter().filter_map(|(hash, blob)| {
                    if blob.name == function_name {
                        Some(*hash)
                    } else {
                        None
                    }
                });
                let first = matches.next()?;
                if matches.next().is_some() {
                    None
                } else {
                    Some(first)
                }
            })
        });
    let blobs = function_hash.and_then(|h| build_minimal_blobs_by_hash(program, h));

    // When content-addressed blobs are available, send a minimal stub program
    // instead of the full BytecodeProgram to reduce payload size.
    let request_program = if blobs.is_some() {
        create_stub_program(program)
    } else {
        program.clone()
    };

    RemoteCallRequest {
        program: request_program,
        function_name: function_name.to_string(),
        function_id,
        function_hash,
        arguments,
        upvalues: None,
        type_schemas: program.type_schema_registry.clone(),
        program_hash: hash,
        function_blobs: blobs,
    }
}

/// Build a `RemoteCallRequest` for a closure.
///
/// Serializes the closure's captured upvalues alongside the function call.
/// When the closure's function has a matching content-addressed blob, sends
/// the minimal blob set instead of the full program.
pub fn build_closure_call_request(
    program: &BytecodeProgram,
    function_id: u16,
    arguments: Vec<SerializableVMValue>,
    upvalues: Vec<SerializableVMValue>,
) -> RemoteCallRequest {
    let hash = program_hash(program);

    let function_hash = program
        .function_blob_hashes
        .get(function_id as usize)
        .copied()
        .flatten();
    let blobs = function_hash.and_then(|h| build_minimal_blobs_by_hash(program, h));

    RemoteCallRequest {
        program: if blobs.is_some() {
            create_stub_program(program)
        } else {
            program.clone()
        },
        function_name: String::new(),
        function_id: Some(function_id),
        function_hash,
        arguments,
        upvalues: Some(upvalues),
        type_schemas: program.type_schema_registry.clone(),
        program_hash: hash,
        function_blobs: blobs,
    }
}

/// Build a `RemoteCallRequest` that strips function blobs the remote already has.
///
/// Like `build_call_request`, but takes a set of hashes the remote is known to
/// have cached (from a prior `BlobNegotiationResponse`). Blobs with matching
/// hashes are omitted from `function_blobs`, reducing payload size.
pub fn build_call_request_negotiated(
    program: &BytecodeProgram,
    function_name: &str,
    arguments: Vec<SerializableVMValue>,
    known_hashes: &[FunctionHash],
) -> RemoteCallRequest {
    let mut request = build_call_request(program, function_name, arguments);

    // Strip blobs the remote already has
    if let Some(ref mut blobs) = request.function_blobs {
        let known_set: std::collections::HashSet<FunctionHash> =
            known_hashes.iter().copied().collect();
        blobs.retain(|(hash, _)| !known_set.contains(hash));
    }

    request
}

/// Handle a blob negotiation request on the server side.
///
/// Returns the subset of offered hashes that are present in the cache.
pub fn handle_negotiation(
    request: &BlobNegotiationRequest,
    cache: &RemoteBlobCache,
) -> BlobNegotiationResponse {
    BlobNegotiationResponse {
        known_hashes: cache.filter_known(&request.offered_hashes),
    }
}

// ---------------------------------------------------------------------------
// Wire message dispatch (V1 + V2 handlers)
// ---------------------------------------------------------------------------

/// Handle a `WireMessage` by dispatching to the appropriate handler.
///
/// V1 messages (BlobNegotiation, Call, CallResponse, Sidecar) are fully handled.
/// V2 messages (Execute, Validate, Auth, Ping, file/project operations) return
/// stub error responses until the execution server is implemented.
pub fn handle_wire_message(
    msg: WireMessage,
    store: &SnapshotStore,
    cache: &mut RemoteBlobCache,
) -> WireMessage {
    match msg {
        WireMessage::BlobNegotiation(req) => {
            let response = handle_negotiation(&req, cache);
            WireMessage::BlobNegotiationReply(response)
        }
        WireMessage::BlobNegotiationReply(_) => {
            // Client-side message — server should not receive this.
            // Return an error wrapped in an ExecuteResponse as a generic error channel.
            WireMessage::ExecuteResponse(ExecuteResponse {
                request_id: 0,
                success: false,
                value: WireValue::Null,
                stdout: None,
                error: Some("Unexpected BlobNegotiationReply on server side".to_string()),
                content_terminal: None,
                content_html: None,
                diagnostics: vec![],
                metrics: None,
                print_output: None,
            })
        }
        WireMessage::Call(req) => {
            // Cache any incoming blobs for future negotiation
            if let Some(ref blobs) = req.function_blobs {
                cache.insert_blobs(blobs);
            }
            let response = execute_remote_call(req, store);
            WireMessage::CallResponse(response)
        }
        WireMessage::CallResponse(_) => {
            // Client-side message — server should not receive this.
            WireMessage::ExecuteResponse(ExecuteResponse {
                request_id: 0,
                success: false,
                value: WireValue::Null,
                stdout: None,
                error: Some("Unexpected CallResponse on server side".to_string()),
                content_terminal: None,
                content_html: None,
                diagnostics: vec![],
                metrics: None,
                print_output: None,
            })
        }
        WireMessage::Sidecar(_sidecar) => {
            // Sidecars are buffered by the transport layer and reassembled
            // before the Call message is dispatched. If we receive one here,
            // it means the transport did not buffer it.
            WireMessage::ExecuteResponse(ExecuteResponse {
                request_id: 0,
                success: false,
                value: WireValue::Null,
                stdout: None,
                error: Some("Unexpected standalone Sidecar message".to_string()),
                content_terminal: None,
                content_html: None,
                diagnostics: vec![],
                metrics: None,
                print_output: None,
            })
        }

        // --- V2 message stubs ---

        WireMessage::Execute(req) => WireMessage::ExecuteResponse(ExecuteResponse {
            request_id: req.request_id,
            success: false,
            value: WireValue::Null,
            stdout: None,
            error: Some("V2 Execute not yet implemented".to_string()),
            content_terminal: None,
            content_html: None,
            diagnostics: vec![WireDiagnostic {
                severity: "error".to_string(),
                message: "V2 Execute handler not yet implemented".to_string(),
                line: None,
                column: None,
            }],
            metrics: None,
            print_output: None,
        }),
        WireMessage::ExecuteResponse(_) => {
            // Client-side message — should not arrive at server.
            WireMessage::ExecuteResponse(ExecuteResponse {
                request_id: 0,
                success: false,
                value: WireValue::Null,
                stdout: None,
                error: Some("Unexpected ExecuteResponse on server side".to_string()),
                content_terminal: None,
                content_html: None,
                diagnostics: vec![],
                metrics: None,
                print_output: None,
            })
        }
        WireMessage::Validate(req) => WireMessage::ValidateResponse(ValidateResponse {
            request_id: req.request_id,
            success: false,
            diagnostics: vec![WireDiagnostic {
                severity: "error".to_string(),
                message: "V2 Validate handler not yet implemented".to_string(),
                line: None,
                column: None,
            }],
        }),
        WireMessage::ValidateResponse(_) => {
            WireMessage::ExecuteResponse(ExecuteResponse {
                request_id: 0,
                success: false,
                value: WireValue::Null,
                stdout: None,
                error: Some("Unexpected ValidateResponse on server side".to_string()),
                content_terminal: None,
                content_html: None,
                diagnostics: vec![],
                metrics: None,
                print_output: None,
            })
        }
        WireMessage::Auth(_req) => WireMessage::AuthResponse(AuthResponse {
            authenticated: false,
            error: Some("V2 Auth handler not yet implemented".to_string()),
        }),
        WireMessage::AuthResponse(_) => {
            WireMessage::ExecuteResponse(ExecuteResponse {
                request_id: 0,
                success: false,
                value: WireValue::Null,
                stdout: None,
                error: Some("Unexpected AuthResponse on server side".to_string()),
                content_terminal: None,
                content_html: None,
                diagnostics: vec![],
                metrics: None,
                print_output: None,
            })
        }
        WireMessage::ExecuteFile(req) => WireMessage::ExecuteResponse(ExecuteResponse {
            request_id: req.request_id,
            success: false,
            value: WireValue::Null,
            stdout: None,
            error: Some("V2 ExecuteFile handler not yet implemented".to_string()),
            content_terminal: None,
            content_html: None,
            diagnostics: vec![WireDiagnostic {
                severity: "error".to_string(),
                message: "V2 ExecuteFile handler not yet implemented".to_string(),
                line: None,
                column: None,
            }],
            metrics: None,
            print_output: None,
        }),
        WireMessage::ExecuteProject(req) => WireMessage::ExecuteResponse(ExecuteResponse {
            request_id: req.request_id,
            success: false,
            value: WireValue::Null,
            stdout: None,
            error: Some("V2 ExecuteProject handler not yet implemented".to_string()),
            content_terminal: None,
            content_html: None,
            diagnostics: vec![WireDiagnostic {
                severity: "error".to_string(),
                message: "V2 ExecuteProject handler not yet implemented".to_string(),
                line: None,
                column: None,
            }],
            metrics: None,
            print_output: None,
        }),
        WireMessage::ValidatePath(req) => WireMessage::ValidateResponse(ValidateResponse {
            request_id: req.request_id,
            success: false,
            diagnostics: vec![WireDiagnostic {
                severity: "error".to_string(),
                message: "V2 ValidatePath handler not yet implemented".to_string(),
                line: None,
                column: None,
            }],
        }),
        WireMessage::Ping(_) => WireMessage::Pong(ServerInfo {
            shape_version: env!("CARGO_PKG_VERSION").to_string(),
            wire_protocol: shape_wire::WIRE_PROTOCOL_V2,
            capabilities: vec![
                "call".to_string(),
                "blob-negotiation".to_string(),
                "sidecar".to_string(),
            ],
        }),
        WireMessage::Pong(_) => {
            // Client-side message — should not arrive at server.
            WireMessage::ExecuteResponse(ExecuteResponse {
                request_id: 0,
                success: false,
                value: WireValue::Null,
                stdout: None,
                error: Some("Unexpected Pong on server side".to_string()),
                content_terminal: None,
                content_html: None,
                diagnostics: vec![],
                metrics: None,
                print_output: None,
            })
        }
    }
}

// ---------------------------------------------------------------------------
// Phase 3B: Sidecar extraction and reassembly
// ---------------------------------------------------------------------------

/// Minimum blob size (in bytes) to extract as a sidecar.
/// Blobs smaller than this are left inline in the serialized payload.
pub const SIDECAR_THRESHOLD: usize = 1024 * 1024; // 1 MB

/// Extract large blobs from serialized arguments into sidecars.
///
/// Walks the `SerializableVMValue` tree recursively. Any `BlobRef` whose
/// backing `ChunkedBlob` exceeds `SIDECAR_THRESHOLD` bytes is replaced
/// with a `SidecarRef` and the raw data is collected into a `BlobSidecar`.
///
/// Returns the extracted sidecars. The `args` are modified in place.
pub fn extract_sidecars(
    args: &mut Vec<SerializableVMValue>,
    store: &SnapshotStore,
) -> Vec<BlobSidecar> {
    let mut sidecars = Vec::new();
    let mut next_id: u32 = 0;
    for arg in args.iter_mut() {
        extract_sidecars_recursive(arg, store, &mut sidecars, &mut next_id);
    }
    sidecars
}

/// Extract the BlobRef from a SerializableVMValue if it carries one (non-mutating read).
fn get_blob_ref(value: &SerializableVMValue) -> Option<&shape_runtime::snapshot::BlobRef> {
    use shape_runtime::snapshot::SerializableVMValue as SV;
    match value {
        SV::DataTable(blob)
        | SV::TypedTable { table: blob, .. }
        | SV::RowView { table: blob, .. }
        | SV::ColumnRef { table: blob, .. }
        | SV::IndexedTable { table: blob, .. } => Some(blob),
        SV::TypedArray { blob, .. } | SV::Matrix { blob, .. } => Some(blob),
        _ => None,
    }
}

fn extract_sidecars_recursive(
    value: &mut SerializableVMValue,
    store: &SnapshotStore,
    sidecars: &mut Vec<BlobSidecar>,
    next_id: &mut u32,
) {
    use shape_runtime::snapshot::SerializableVMValue as SV;

    // First: check if this value carries a blob large enough to extract.
    // Capture metadata (TypedArray len, Matrix rows/cols) before replacing.
    let meta = match &*value {
        SV::TypedArray { len, .. } => (*len as u32, 0u32),
        SV::Matrix { rows, cols, .. } => (*rows, *cols),
        _ => (0, 0),
    };
    // Clone the blob info to avoid borrow conflicts with the later mutation.
    if let Some(blob) = get_blob_ref(value) {
        let blob_kind = blob.kind.clone();
        let blob_hash = blob.hash.clone();
        if let Some(sidecar) = try_extract_blob(blob, store, next_id) {
            let sidecar_id = sidecar.sidecar_id;
            sidecars.push(sidecar);
            *value = SV::SidecarRef {
                sidecar_id,
                blob_kind,
                original_hash: blob_hash,
                meta_a: meta.0,
                meta_b: meta.1,
            };
            return;
        }
    }

    // Recursive descent into containers
    match value {
        SV::Array(items) => {
            for item in items.iter_mut() {
                extract_sidecars_recursive(item, store, sidecars, next_id);
            }
        }
        SV::HashMap { keys, values } => {
            for k in keys.iter_mut() {
                extract_sidecars_recursive(k, store, sidecars, next_id);
            }
            for v in values.iter_mut() {
                extract_sidecars_recursive(v, store, sidecars, next_id);
            }
        }
        SV::TypedObject { slot_data, .. } => {
            for slot in slot_data.iter_mut() {
                extract_sidecars_recursive(slot, store, sidecars, next_id);
            }
        }
        SV::Some(inner) | SV::Ok(inner) | SV::Err(inner) => {
            extract_sidecars_recursive(inner, store, sidecars, next_id);
        }
        SV::TypeAnnotatedValue { value: inner, .. } => {
            extract_sidecars_recursive(inner, store, sidecars, next_id);
        }
        SV::Closure { upvalues, .. } => {
            for uv in upvalues.iter_mut() {
                extract_sidecars_recursive(uv, store, sidecars, next_id);
            }
        }
        SV::Enum(ev) => match &mut ev.payload {
            shape_runtime::snapshot::EnumPayloadSnapshot::Unit => {}
            shape_runtime::snapshot::EnumPayloadSnapshot::Tuple(items) => {
                for item in items.iter_mut() {
                    extract_sidecars_recursive(item, store, sidecars, next_id);
                }
            }
            shape_runtime::snapshot::EnumPayloadSnapshot::Struct(fields) => {
                for (_, v) in fields.iter_mut() {
                    extract_sidecars_recursive(v, store, sidecars, next_id);
                }
            }
        },
        SV::PrintResult(pr) => {
            for span in pr.spans.iter_mut() {
                if let shape_runtime::snapshot::PrintSpanSnapshot::Value {
                    raw_value,
                    format_params,
                    ..
                } = span
                {
                    extract_sidecars_recursive(raw_value, store, sidecars, next_id);
                    for (_, v) in format_params.iter_mut() {
                        extract_sidecars_recursive(v, store, sidecars, next_id);
                    }
                }
            }
        }
        SV::SimulationCall { params, .. } => {
            for (_, v) in params.iter_mut() {
                extract_sidecars_recursive(v, store, sidecars, next_id);
            }
        }
        SV::FunctionRef { closure, .. } => {
            if let Some(c) = closure {
                extract_sidecars_recursive(c, store, sidecars, next_id);
            }
        }
        SV::Range { start, end, .. } => {
            if let Some(s) = start {
                extract_sidecars_recursive(s, store, sidecars, next_id);
            }
            if let Some(e) = end {
                extract_sidecars_recursive(e, store, sidecars, next_id);
            }
        }

        // Leaf types and blob carriers (handled above) — nothing more to do
        _ => {}
    }
}

/// Try to extract a BlobRef's data as a sidecar if it exceeds the threshold.
fn try_extract_blob(
    blob: &shape_runtime::snapshot::BlobRef,
    store: &SnapshotStore,
    next_id: &mut u32,
) -> Option<BlobSidecar> {
    // Load the ChunkedBlob metadata to check total size
    let chunked: shape_runtime::snapshot::ChunkedBlob = store.get_struct(&blob.hash).ok()?;
    if chunked.total_len < SIDECAR_THRESHOLD {
        return None;
    }

    // Load the raw data
    let data = shape_runtime::snapshot::load_chunked_bytes(&chunked, store).ok()?;
    let sidecar_id = *next_id;
    *next_id += 1;

    Some(BlobSidecar { sidecar_id, data })
}

/// Reassemble sidecars back into the serialized payload.
///
/// Walks the `SerializableVMValue` tree and replaces `SidecarRef` variants
/// with the original `BlobRef`, storing the sidecar data back into the
/// snapshot store.
pub fn reassemble_sidecars(
    args: &mut Vec<SerializableVMValue>,
    sidecars: &std::collections::HashMap<u32, BlobSidecar>,
    store: &SnapshotStore,
) -> anyhow::Result<()> {
    for arg in args.iter_mut() {
        reassemble_recursive(arg, sidecars, store)?;
    }
    Ok(())
}

fn reassemble_recursive(
    value: &mut SerializableVMValue,
    sidecars: &std::collections::HashMap<u32, BlobSidecar>,
    store: &SnapshotStore,
) -> anyhow::Result<()> {
    use shape_runtime::snapshot::{BlobRef, SerializableVMValue as SV};

    match value {
        SV::SidecarRef {
            sidecar_id,
            blob_kind,
            original_hash: _,
            meta_a,
            meta_b,
        } => {
            let sidecar = sidecars
                .get(sidecar_id)
                .ok_or_else(|| anyhow::anyhow!("missing sidecar with id {}", sidecar_id))?;
            let meta_a = *meta_a;
            let meta_b = *meta_b;

            // Store the sidecar data back into the snapshot store as chunked bytes,
            // then wrap in a ChunkedBlob struct and store that.
            let chunked = shape_runtime::snapshot::store_chunked_bytes(&sidecar.data, store)?;
            let hash = store.put_struct(&chunked)?;

            let blob = BlobRef {
                hash,
                kind: blob_kind.clone(),
            };
            *value = match blob_kind {
                shape_runtime::snapshot::BlobKind::DataTable => SV::DataTable(blob),
                shape_runtime::snapshot::BlobKind::TypedArray(ek) => SV::TypedArray {
                    element_kind: *ek,
                    blob,
                    len: meta_a as usize,
                },
                shape_runtime::snapshot::BlobKind::Matrix => SV::Matrix {
                    blob,
                    rows: meta_a,
                    cols: meta_b,
                },
            };
        }

        // Recursive descent (same structure as extract)
        SV::Array(items) => {
            for item in items.iter_mut() {
                reassemble_recursive(item, sidecars, store)?;
            }
        }
        SV::HashMap { keys, values } => {
            for k in keys.iter_mut() {
                reassemble_recursive(k, sidecars, store)?;
            }
            for v in values.iter_mut() {
                reassemble_recursive(v, sidecars, store)?;
            }
        }
        SV::TypedObject { slot_data, .. } => {
            for slot in slot_data.iter_mut() {
                reassemble_recursive(slot, sidecars, store)?;
            }
        }
        SV::Some(inner) | SV::Ok(inner) | SV::Err(inner) => {
            reassemble_recursive(inner, sidecars, store)?;
        }
        SV::TypeAnnotatedValue { value: inner, .. } => {
            reassemble_recursive(inner, sidecars, store)?;
        }
        SV::Closure { upvalues, .. } => {
            for uv in upvalues.iter_mut() {
                reassemble_recursive(uv, sidecars, store)?;
            }
        }
        SV::Enum(ev) => match &mut ev.payload {
            shape_runtime::snapshot::EnumPayloadSnapshot::Unit => {}
            shape_runtime::snapshot::EnumPayloadSnapshot::Tuple(items) => {
                for item in items.iter_mut() {
                    reassemble_recursive(item, sidecars, store)?;
                }
            }
            shape_runtime::snapshot::EnumPayloadSnapshot::Struct(fields) => {
                for (_, v) in fields.iter_mut() {
                    reassemble_recursive(v, sidecars, store)?;
                }
            }
        },
        SV::PrintResult(pr) => {
            for span in pr.spans.iter_mut() {
                if let shape_runtime::snapshot::PrintSpanSnapshot::Value {
                    raw_value,
                    format_params,
                    ..
                } = span
                {
                    reassemble_recursive(raw_value, sidecars, store)?;
                    for (_, v) in format_params.iter_mut() {
                        reassemble_recursive(v, sidecars, store)?;
                    }
                }
            }
        }
        SV::SimulationCall { params, .. } => {
            for (_, v) in params.iter_mut() {
                reassemble_recursive(v, sidecars, store)?;
            }
        }
        SV::FunctionRef { closure, .. } => {
            if let Some(c) = closure {
                reassemble_recursive(c, sidecars, store)?;
            }
        }
        SV::Range { start, end, .. } => {
            if let Some(s) = start {
                reassemble_recursive(s, sidecars, store)?;
            }
            if let Some(e) = end {
                reassemble_recursive(e, sidecars, store)?;
            }
        }

        // Leaf types and blob-carrying variants (non-sidecar) — nothing to reassemble
        _ => {}
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::bytecode::{FunctionBlob, FunctionHash, Instruction, OpCode, Program};
    use crate::compiler::BytecodeCompiler;
    use shape_abi_v1::PermissionSet;
    use std::collections::HashMap;

    /// Helper: compile Shape source to BytecodeProgram
    fn compile(source: &str) -> BytecodeProgram {
        let program = shape_ast::parser::parse_program(source).expect("parse failed");
        let compiler = BytecodeCompiler::new();
        compiler.compile(&program).expect("compile failed")
    }

    /// Helper: create a temp SnapshotStore
    fn temp_store() -> SnapshotStore {
        let dir = std::env::temp_dir().join(format!("shape_remote_test_{}", std::process::id()));
        SnapshotStore::new(dir).expect("create snapshot store")
    }

    fn mk_hash(tag: u8) -> FunctionHash {
        let mut bytes = [0u8; 32];
        bytes[0] = tag;
        FunctionHash(bytes)
    }

    fn mk_blob(name: &str, hash: FunctionHash, dependencies: Vec<FunctionHash>) -> FunctionBlob {
        FunctionBlob {
            content_hash: hash,
            name: name.to_string(),
            arity: 0,
            param_names: Vec::new(),
            locals_count: 0,
            is_closure: false,
            captures_count: 0,
            is_async: false,
            ref_params: Vec::new(),
            ref_mutates: Vec::new(),
            mutable_captures: Vec::new(),
            frame_descriptor: None,
            instructions: vec![
                Instruction::simple(OpCode::PushNull),
                Instruction::simple(OpCode::ReturnValue),
            ],
            constants: Vec::new(),
            strings: Vec::new(),
            required_permissions: PermissionSet::pure(),
            dependencies,
            callee_names: Vec::new(),
            type_schemas: Vec::new(),
            foreign_dependencies: Vec::new(),
            source_map: Vec::new(),
        }
    }

    // The pre-bulldozer end-to-end execute tests
    // (`test_remote_call_simple_function`, `test_remote_call_function_not_found`)
    // drove `execute_remote_call`, which is currently a phase-2c stub
    // (see `execute_inner` body). Re-author them once the kind-threaded
    // `slot_to_serializable` / `serializable_to_slot` round-trip lands
    // (ADR-006 §2.7.4 + addendum); both depend on the snapshot-side
    // rebuild plus a `Vec<KindedSlot>` arg pipeline through the VM
    // entrypoints.

    #[test]
    fn test_program_hash_deterministic() {
        let bytecode = compile("function f(x) { x * 2 }");
        let hash1 = program_hash(&bytecode);
        let hash2 = program_hash(&bytecode);
        assert_eq!(hash1, hash2, "Same program should produce same hash");
    }

    #[test]
    fn test_request_response_serialization_roundtrip() {
        let bytecode = compile("function double(x) { x * 2 }");
        let request =
            build_call_request(&bytecode, "double", vec![SerializableVMValue::Number(21.0)]);

        // Encode → decode roundtrip via MessagePack
        let bytes = shape_wire::encode_message(&request).expect("encode request");
        let decoded: RemoteCallRequest =
            shape_wire::decode_message(&bytes).expect("decode request");

        assert_eq!(decoded.function_name, "double");
        assert_eq!(decoded.arguments.len(), 1);
        assert_eq!(decoded.program_hash, request.program_hash);
    }

    #[test]
    fn test_response_serialization_roundtrip() {
        let response = RemoteCallResponse {
            result: Ok(SerializableVMValue::String("hello".to_string())),
        };

        let bytes = shape_wire::encode_message(&response).expect("encode response");
        let decoded: RemoteCallResponse =
            shape_wire::decode_message(&bytes).expect("decode response");

        match decoded.result {
            Ok(SerializableVMValue::String(s)) => assert_eq!(s, "hello"),
            other => panic!("Expected Ok(String), got {:?}", other),
        }
    }

    #[test]
    fn test_type_schema_registry_roundtrip() {
        use shape_runtime::type_schema::{FieldType, TypeSchemaRegistry};

        let mut registry = TypeSchemaRegistry::new();
        registry.register_type(
            "Point",
            vec![
                ("x".to_string(), FieldType::F64),
                ("y".to_string(), FieldType::F64),
            ],
        );

        let bytes = shape_wire::encode_message(&registry).expect("encode registry");
        let decoded: TypeSchemaRegistry =
            shape_wire::decode_message(&bytes).expect("decode registry");

        assert!(decoded.has_type("Point"));
        let schema = decoded.get("Point").unwrap();
        assert_eq!(schema.field_count(), 2);
        assert_eq!(schema.field_offset("x"), Some(0));
        assert_eq!(schema.field_offset("y"), Some(8));
    }

    #[test]
    fn test_build_minimal_blobs_rejects_ambiguous_function_name() {
        let h1 = mk_hash(1);
        let h2 = mk_hash(2);
        let blob1 = mk_blob("dup", h1, vec![]);
        let blob2 = mk_blob("dup", h2, vec![]);

        let mut function_store = HashMap::new();
        function_store.insert(h1, blob1.clone());
        function_store.insert(h2, blob2.clone());

        let mut program = BytecodeProgram::default();
        program.content_addressed = Some(Program {
            entry: h1,
            function_store,
            top_level_locals_count: 0,
            top_level_local_storage_hints: Vec::new(),
            module_binding_names: Vec::new(),
            module_binding_storage_hints: Vec::new(),
            function_local_storage_hints: Vec::new(),
            top_level_frame: None,
            top_level_local_concrete_types: Vec::new(),
            function_local_concrete_types: Vec::new(),
            function_return_concrete_types: Vec::new(),
            monomorphized_method_call_sites: HashMap::new(),
            value_call_return_concrete_types: HashMap::new(),
            operator_trait_dispatch_sites: HashMap::new(),
            data_schema: None,
            type_schema_registry: shape_runtime::type_schema::TypeSchemaRegistry::new(),
            trait_method_symbols: HashMap::new(),
            foreign_functions: Vec::new(),
            native_struct_layouts: Vec::new(),
            debug_info: crate::bytecode::DebugInfo::new("<test>".to_string()),
            closure_function_layouts_by_name: HashMap::new(),
            trait_vtables: HashMap::new(),
            has_imported_const_inline: false,
            has_w17_marshal_residual: false,
        });

        assert!(
            build_minimal_blobs(&program, "dup").is_none(),
            "name-based selection must reject ambiguous function names"
        );

        let by_hash = build_minimal_blobs_by_hash(&program, h2)
            .expect("hash-based selection should work with duplicate names");
        assert_eq!(by_hash.len(), 1);
        assert_eq!(by_hash[0].0, h2);
        assert_eq!(by_hash[0].1.name, "dup");
    }

    #[test]
    fn test_program_from_blobs_by_hash_requires_entry_blob() {
        let h1 = mk_hash(1);
        let h_missing = mk_hash(9);
        let blob = mk_blob("f", h1, vec![]);
        let source = BytecodeProgram::default();

        let reconstructed = program_from_blobs_by_hash(vec![(h1, blob)], h_missing, &source);
        assert!(
            reconstructed.is_none(),
            "reconstruction must fail when the requested entry hash is absent"
        );
    }

    // ---- Phase 2: Blob negotiation tests ----

    #[test]
    fn test_blob_cache_insert_and_get() {
        let mut cache = RemoteBlobCache::new(10);
        let h1 = mk_hash(1);
        let blob1 = mk_blob("f1", h1, vec![]);

        cache.insert(h1, blob1.clone());
        assert_eq!(cache.len(), 1);
        assert!(cache.contains(&h1));
        assert_eq!(cache.get(&h1).unwrap().name, "f1");
    }

    #[test]
    fn test_blob_cache_lru_eviction() {
        let mut cache = RemoteBlobCache::new(2);
        let h1 = mk_hash(1);
        let h2 = mk_hash(2);
        let h3 = mk_hash(3);

        cache.insert(h1, mk_blob("f1", h1, vec![]));
        cache.insert(h2, mk_blob("f2", h2, vec![]));
        assert_eq!(cache.len(), 2);

        // Insert h3 should evict h1 (least recently used)
        cache.insert(h3, mk_blob("f3", h3, vec![]));
        assert_eq!(cache.len(), 2);
        assert!(!cache.contains(&h1), "h1 should be evicted");
        assert!(cache.contains(&h2));
        assert!(cache.contains(&h3));
    }

    #[test]
    fn test_blob_cache_access_updates_order() {
        let mut cache = RemoteBlobCache::new(2);
        let h1 = mk_hash(1);
        let h2 = mk_hash(2);
        let h3 = mk_hash(3);

        cache.insert(h1, mk_blob("f1", h1, vec![]));
        cache.insert(h2, mk_blob("f2", h2, vec![]));

        // Access h1 to make it recently used
        cache.get(&h1);

        // Insert h3 should evict h2 (now least recently used)
        cache.insert(h3, mk_blob("f3", h3, vec![]));
        assert!(
            cache.contains(&h1),
            "h1 was accessed, should not be evicted"
        );
        assert!(!cache.contains(&h2), "h2 should be evicted");
        assert!(cache.contains(&h3));
    }

    #[test]
    fn test_blob_cache_filter_known() {
        let mut cache = RemoteBlobCache::new(10);
        let h1 = mk_hash(1);
        let h2 = mk_hash(2);
        let h3 = mk_hash(3);

        cache.insert(h1, mk_blob("f1", h1, vec![]));
        cache.insert(h3, mk_blob("f3", h3, vec![]));

        let known = cache.filter_known(&[h1, h2, h3]);
        assert_eq!(known.len(), 2);
        assert!(known.contains(&h1));
        assert!(known.contains(&h3));
        assert!(!known.contains(&h2));
    }

    #[test]
    fn test_handle_negotiation() {
        let mut cache = RemoteBlobCache::new(10);
        let h1 = mk_hash(1);
        let h2 = mk_hash(2);
        cache.insert(h1, mk_blob("f1", h1, vec![]));

        let request = BlobNegotiationRequest {
            offered_hashes: vec![h1, h2],
        };
        let response = handle_negotiation(&request, &cache);
        assert_eq!(response.known_hashes.len(), 1);
        assert!(response.known_hashes.contains(&h1));
    }

    #[test]
    fn test_build_call_request_negotiated_strips_known_blobs() {
        // Create a program with content-addressed blobs
        let h1 = mk_hash(1);
        let h2 = mk_hash(2);
        let blob1 = mk_blob("entry", h1, vec![h2]);
        let blob2 = mk_blob("helper", h2, vec![]);

        let mut function_store = HashMap::new();
        function_store.insert(h1, blob1.clone());
        function_store.insert(h2, blob2.clone());

        let mut program = BytecodeProgram::default();
        program.content_addressed = Some(Program {
            entry: h1,
            function_store,
            top_level_locals_count: 0,
            top_level_local_storage_hints: Vec::new(),
            module_binding_names: Vec::new(),
            module_binding_storage_hints: Vec::new(),
            function_local_storage_hints: Vec::new(),
            top_level_frame: None,
            top_level_local_concrete_types: Vec::new(),
            function_local_concrete_types: Vec::new(),
            function_return_concrete_types: Vec::new(),
            monomorphized_method_call_sites: HashMap::new(),
            value_call_return_concrete_types: HashMap::new(),
            operator_trait_dispatch_sites: HashMap::new(),
            data_schema: None,
            type_schema_registry: shape_runtime::type_schema::TypeSchemaRegistry::new(),
            trait_method_symbols: HashMap::new(),
            foreign_functions: Vec::new(),
            native_struct_layouts: Vec::new(),
            debug_info: crate::bytecode::DebugInfo::new("<test>".to_string()),
            closure_function_layouts_by_name: HashMap::new(),
            trait_vtables: HashMap::new(),
            has_imported_const_inline: false,
            has_w17_marshal_residual: false,
        });
        program.functions = vec![crate::bytecode::Function {
            name: "entry".to_string(),
            arity: 0,
            param_names: vec![],
            locals_count: 0,
            entry_point: 0,
            body_length: 0,
            is_closure: false,
            captures_count: 0,
            is_async: false,
            ref_params: vec![],
            ref_mutates: vec![],
            mutable_captures: vec![],
            frame_descriptor: None,
            osr_entry_points: vec![],
            mir_data: None,
        }];
        program.function_blob_hashes = vec![Some(h1)];

        // First call: no known hashes -> all blobs sent
        let req1 = build_call_request_negotiated(&program, "entry", vec![], &[]);
        let blobs1 = req1.function_blobs.as_ref().unwrap();
        assert_eq!(blobs1.len(), 2, "first call should send all blobs");

        // Second call: h2 is known -> only h1 sent
        let req2 = build_call_request_negotiated(&program, "entry", vec![], &[h2]);
        let blobs2 = req2.function_blobs.as_ref().unwrap();
        assert_eq!(blobs2.len(), 1, "second call should skip known blobs");
        assert_eq!(blobs2[0].0, h1);
    }

    #[test]
    fn test_wire_message_serialization_roundtrip() {
        let msg = WireMessage::BlobNegotiation(BlobNegotiationRequest {
            offered_hashes: vec![mk_hash(1), mk_hash(2)],
        });
        let bytes = shape_wire::encode_message(&msg).expect("encode WireMessage");
        let decoded: WireMessage = shape_wire::decode_message(&bytes).expect("decode WireMessage");
        match decoded {
            WireMessage::BlobNegotiation(req) => {
                assert_eq!(req.offered_hashes.len(), 2);
            }
            _ => panic!("Expected BlobNegotiation"),
        }
    }

    // ---- V2 execution server message tests ----

    #[test]
    fn test_execute_request_roundtrip() {
        let msg = WireMessage::Execute(ExecuteRequest {
            code: "fn main() { 42 }".to_string(),
            request_id: 7,
        });
        let bytes = shape_wire::encode_message(&msg).expect("encode Execute");
        let decoded: WireMessage = shape_wire::decode_message(&bytes).expect("decode Execute");
        match decoded {
            WireMessage::Execute(req) => {
                assert_eq!(req.code, "fn main() { 42 }");
                assert_eq!(req.request_id, 7);
            }
            _ => panic!("Expected Execute"),
        }
    }

    #[test]
    fn test_execute_response_roundtrip() {
        let msg = WireMessage::ExecuteResponse(ExecuteResponse {
            request_id: 7,
            success: true,
            value: WireValue::Number(42.0),
            stdout: Some("hello\n".to_string()),
            error: None,
            content_terminal: None,
            content_html: None,
            diagnostics: vec![WireDiagnostic {
                severity: "warning".to_string(),
                message: "unused variable".to_string(),
                line: Some(1),
                column: Some(5),
            }],
            metrics: Some(ExecutionMetrics {
                instructions_executed: 100,
                wall_time_ms: 3,
                memory_bytes_peak: 4096,
            }),
            print_output: None,
        });
        let bytes = shape_wire::encode_message(&msg).expect("encode ExecuteResponse");
        let decoded: WireMessage =
            shape_wire::decode_message(&bytes).expect("decode ExecuteResponse");
        match decoded {
            WireMessage::ExecuteResponse(resp) => {
                assert_eq!(resp.request_id, 7);
                assert!(resp.success);
                assert!(matches!(resp.value, WireValue::Number(n) if n == 42.0));
                assert_eq!(resp.stdout.as_deref(), Some("hello\n"));
                assert!(resp.error.is_none());
                assert_eq!(resp.diagnostics.len(), 1);
                assert_eq!(resp.diagnostics[0].severity, "warning");
                assert_eq!(resp.diagnostics[0].line, Some(1));
                let m = resp.metrics.unwrap();
                assert_eq!(m.instructions_executed, 100);
                assert_eq!(m.wall_time_ms, 3);
            }
            _ => panic!("Expected ExecuteResponse"),
        }
    }

    #[test]
    fn test_ping_pong_roundtrip() {
        let ping = WireMessage::Ping(PingRequest {});
        let bytes = shape_wire::encode_message(&ping).expect("encode Ping");
        let decoded: WireMessage = shape_wire::decode_message(&bytes).expect("decode Ping");
        assert!(matches!(decoded, WireMessage::Ping(_)));

        let pong = WireMessage::Pong(ServerInfo {
            shape_version: "0.1.3".to_string(),
            wire_protocol: 2,
            capabilities: vec!["execute".to_string(), "validate".to_string()],
        });
        let bytes = shape_wire::encode_message(&pong).expect("encode Pong");
        let decoded: WireMessage = shape_wire::decode_message(&bytes).expect("decode Pong");
        match decoded {
            WireMessage::Pong(info) => {
                assert_eq!(info.shape_version, "0.1.3");
                assert_eq!(info.wire_protocol, 2);
                assert_eq!(info.capabilities.len(), 2);
            }
            _ => panic!("Expected Pong"),
        }
    }

    #[test]
    fn test_auth_roundtrip() {
        let msg = WireMessage::Auth(AuthRequest {
            token: "secret-token".to_string(),
        });
        let bytes = shape_wire::encode_message(&msg).expect("encode Auth");
        let decoded: WireMessage = shape_wire::decode_message(&bytes).expect("decode Auth");
        match decoded {
            WireMessage::Auth(req) => assert_eq!(req.token, "secret-token"),
            _ => panic!("Expected Auth"),
        }

        let resp = WireMessage::AuthResponse(AuthResponse {
            authenticated: true,
            error: None,
        });
        let bytes = shape_wire::encode_message(&resp).expect("encode AuthResponse");
        let decoded: WireMessage = shape_wire::decode_message(&bytes).expect("decode AuthResponse");
        match decoded {
            WireMessage::AuthResponse(r) => {
                assert!(r.authenticated);
                assert!(r.error.is_none());
            }
            _ => panic!("Expected AuthResponse"),
        }
    }

    #[test]
    fn test_validate_roundtrip() {
        let msg = WireMessage::Validate(ValidateRequest {
            code: "let x = 1".to_string(),
            request_id: 99,
        });
        let bytes = shape_wire::encode_message(&msg).expect("encode Validate");
        let decoded: WireMessage = shape_wire::decode_message(&bytes).expect("decode Validate");
        match decoded {
            WireMessage::Validate(req) => {
                assert_eq!(req.code, "let x = 1");
                assert_eq!(req.request_id, 99);
            }
            _ => panic!("Expected Validate"),
        }

        let resp = WireMessage::ValidateResponse(ValidateResponse {
            request_id: 99,
            success: false,
            diagnostics: vec![WireDiagnostic {
                severity: "error".to_string(),
                message: "parse error".to_string(),
                line: None,
                column: None,
            }],
        });
        let bytes = shape_wire::encode_message(&resp).expect("encode ValidateResponse");
        let decoded: WireMessage =
            shape_wire::decode_message(&bytes).expect("decode ValidateResponse");
        match decoded {
            WireMessage::ValidateResponse(r) => {
                assert_eq!(r.request_id, 99);
                assert!(!r.success);
                assert_eq!(r.diagnostics.len(), 1);
            }
            _ => panic!("Expected ValidateResponse"),
        }
    }

    #[test]
    fn test_ping_framing_roundtrip() {
        use shape_wire::transport::framing::{decode_framed, encode_framed};

        let ping = WireMessage::Ping(PingRequest {});
        let mp = shape_wire::encode_message(&ping).expect("encode Ping");
        eprintln!("Ping msgpack bytes ({} bytes): {:02x?}", mp.len(), &mp);

        let framed = encode_framed(&mp);
        eprintln!("Framed bytes ({} bytes): {:02x?}", framed.len(), &framed);

        let decompressed = decode_framed(&framed).expect("decode_framed");
        assert_eq!(mp, decompressed, "framing roundtrip should preserve bytes");

        let decoded: WireMessage =
            shape_wire::decode_message(&decompressed).expect("decode Ping after framing");
        assert!(matches!(decoded, WireMessage::Ping(_)));
    }

    #[test]
    fn test_execute_framing_roundtrip() {
        use shape_wire::transport::framing::{decode_framed, encode_framed};

        let exec = WireMessage::Execute(ExecuteRequest {
            code: "42".to_string(),
            request_id: 1,
        });
        let mp = shape_wire::encode_message(&exec).expect("encode Execute");
        eprintln!("Execute msgpack bytes ({} bytes): {:02x?}", mp.len(), &mp);

        let framed = encode_framed(&mp);
        let decompressed = decode_framed(&framed).expect("decode_framed");
        let decoded: WireMessage =
            shape_wire::decode_message(&decompressed).expect("decode Execute after framing");
        match decoded {
            WireMessage::Execute(req) => {
                assert_eq!(req.code, "42");
                assert_eq!(req.request_id, 1);
            }
            _ => panic!("Expected Execute"),
        }
    }

    // ---- Phase 3B: Sidecar extraction tests ----

    #[test]
    fn test_extract_sidecars_no_large_blobs() {
        let store = temp_store();
        let mut args = vec![
            SerializableVMValue::Int(42),
            SerializableVMValue::String("hello".to_string()),
            SerializableVMValue::Array(vec![
                SerializableVMValue::Number(1.0),
                SerializableVMValue::Number(2.0),
            ]),
        ];
        let sidecars = extract_sidecars(&mut args, &store);
        assert!(sidecars.is_empty(), "no large blobs → no sidecars");
        // Args should be unchanged
        assert!(matches!(args[0], SerializableVMValue::Int(42)));
    }

    // Sidecar extraction/reassembly tests that constructed input via the
    // deleted `ValueWord::from_float_array` + `nanboxed_to_serializable`
    // pair (`test_extract_sidecars_large_typed_array`,
    // `test_reassemble_sidecars_roundtrip`,
    // `test_extract_sidecars_nested_in_array`) belong to the Phase-2c
    // typed-module-exports rebuild. Re-author once the kind-threaded
    // `slot_to_serializable` round-trip lands. Pure
    // `SerializableVMValue`-shaped sidecar coverage is preserved below
    // (`test_extract_sidecars_no_large_blobs`,
    // `test_sidecar_ref_serialization_roundtrip`) — those exercise
    // `extract_sidecars` / `reassemble_sidecars` without crossing the
    // slot boundary.

    #[test]
    fn test_sidecar_ref_serialization_roundtrip() {
        use shape_runtime::hashing::HashDigest;
        use shape_runtime::snapshot::{BlobKind, TypedArrayElementKind};

        let value = SerializableVMValue::SidecarRef {
            sidecar_id: 7,
            blob_kind: BlobKind::TypedArray(TypedArrayElementKind::F64),
            original_hash: HashDigest::from_hex("abc123"),
            meta_a: 1000,
            meta_b: 0,
        };

        let bytes = shape_wire::encode_message(&value).expect("encode SidecarRef");
        let decoded: SerializableVMValue =
            shape_wire::decode_message(&bytes).expect("decode SidecarRef");
        match decoded {
            SerializableVMValue::SidecarRef { sidecar_id, .. } => {
                assert_eq!(sidecar_id, 7);
            }
            _ => panic!("Expected SidecarRef"),
        }
    }

    // ---- Blob negotiation integration tests ----

    #[test]
    fn test_negotiate_blobs_returns_known_hashes() {
        let h1 = mk_hash(1);
        let h2 = mk_hash(2);
        let h3 = mk_hash(3);

        let mut cache = RemoteBlobCache::new(10);
        cache.insert(h1, mk_blob("f1", h1, vec![]));
        cache.insert(h3, mk_blob("f3", h3, vec![]));

        let blobs = vec![
            (h1, mk_blob("f1", h1, vec![])),
            (h2, mk_blob("f2", h2, vec![])),
            (h3, mk_blob("f3", h3, vec![])),
        ];
        let response = negotiate_blobs(&blobs, &cache);
        assert_eq!(response.known_hashes.len(), 2);
        assert!(response.known_hashes.contains(&h1));
        assert!(response.known_hashes.contains(&h3));
        assert!(!response.known_hashes.contains(&h2));
    }

    #[test]
    fn test_build_call_request_with_negotiation_strips_known() {
        let h1 = mk_hash(1);
        let h2 = mk_hash(2);
        let blob1 = mk_blob("entry", h1, vec![h2]);
        let blob2 = mk_blob("helper", h2, vec![]);

        let mut function_store = HashMap::new();
        function_store.insert(h1, blob1.clone());
        function_store.insert(h2, blob2.clone());

        let mut program = BytecodeProgram::default();
        program.content_addressed = Some(Program {
            entry: h1,
            function_store,
            top_level_locals_count: 0,
            top_level_local_storage_hints: Vec::new(),
            module_binding_names: Vec::new(),
            module_binding_storage_hints: Vec::new(),
            function_local_storage_hints: Vec::new(),
            top_level_frame: None,
            top_level_local_concrete_types: Vec::new(),
            function_local_concrete_types: Vec::new(),
            function_return_concrete_types: Vec::new(),
            monomorphized_method_call_sites: HashMap::new(),
            value_call_return_concrete_types: HashMap::new(),
            operator_trait_dispatch_sites: HashMap::new(),
            data_schema: None,
            type_schema_registry: shape_runtime::type_schema::TypeSchemaRegistry::new(),
            trait_method_symbols: HashMap::new(),
            foreign_functions: Vec::new(),
            native_struct_layouts: Vec::new(),
            debug_info: crate::bytecode::DebugInfo::new("<test>".to_string()),
            closure_function_layouts_by_name: HashMap::new(),
            trait_vtables: HashMap::new(),
            has_imported_const_inline: false,
            has_w17_marshal_residual: false,
        });
        program.functions = vec![crate::bytecode::Function {
            name: "entry".to_string(),
            arity: 0,
            param_names: vec![],
            locals_count: 0,
            entry_point: 0,
            body_length: 0,
            is_closure: false,
            captures_count: 0,
            is_async: false,
            ref_params: vec![],
            ref_mutates: vec![],
            mutable_captures: vec![],
            frame_descriptor: None,
            osr_entry_points: vec![],
            mir_data: None,
        }];
        program.function_blob_hashes = vec![Some(h1)];

        // Cache has h2 -> negotiation should strip it
        let mut cache = RemoteBlobCache::new(10);
        cache.insert(h2, blob2.clone());

        let req = build_call_request_with_negotiation(&program, "entry", vec![], Some(&cache));
        let blobs = req.function_blobs.as_ref().unwrap();
        assert_eq!(blobs.len(), 1, "should strip known blob h2");
        assert_eq!(blobs[0].0, h1, "only h1 should remain");
    }

    #[test]
    fn test_build_call_request_with_negotiation_no_cache() {
        let h1 = mk_hash(1);
        let blob1 = mk_blob("entry", h1, vec![]);

        let mut function_store = HashMap::new();
        function_store.insert(h1, blob1.clone());

        let mut program = BytecodeProgram::default();
        program.content_addressed = Some(Program {
            entry: h1,
            function_store,
            top_level_locals_count: 0,
            top_level_local_storage_hints: Vec::new(),
            module_binding_names: Vec::new(),
            module_binding_storage_hints: Vec::new(),
            function_local_storage_hints: Vec::new(),
            top_level_frame: None,
            top_level_local_concrete_types: Vec::new(),
            function_local_concrete_types: Vec::new(),
            function_return_concrete_types: Vec::new(),
            monomorphized_method_call_sites: HashMap::new(),
            value_call_return_concrete_types: HashMap::new(),
            operator_trait_dispatch_sites: HashMap::new(),
            data_schema: None,
            type_schema_registry: shape_runtime::type_schema::TypeSchemaRegistry::new(),
            trait_method_symbols: HashMap::new(),
            foreign_functions: Vec::new(),
            native_struct_layouts: Vec::new(),
            debug_info: crate::bytecode::DebugInfo::new("<test>".to_string()),
            closure_function_layouts_by_name: HashMap::new(),
            trait_vtables: HashMap::new(),
            has_imported_const_inline: false,
            has_w17_marshal_residual: false,
        });
        program.functions = vec![crate::bytecode::Function {
            name: "entry".to_string(),
            arity: 0,
            param_names: vec![],
            locals_count: 0,
            entry_point: 0,
            body_length: 0,
            is_closure: false,
            captures_count: 0,
            is_async: false,
            ref_params: vec![],
            ref_mutates: vec![],
            mutable_captures: vec![],
            frame_descriptor: None,
            osr_entry_points: vec![],
            mir_data: None,
        }];
        program.function_blob_hashes = vec![Some(h1)];

        // No cache -> all blobs sent
        let req = build_call_request_with_negotiation(&program, "entry", vec![], None);
        let blobs = req.function_blobs.as_ref().unwrap();
        assert_eq!(blobs.len(), 1, "all blobs should be sent when no cache");
    }

    // ---- V2 handler stub tests ----

    #[test]
    fn test_handle_wire_message_ping_returns_pong() {
        let store = temp_store();
        let mut cache = RemoteBlobCache::default_cache();
        let msg = WireMessage::Ping(PingRequest {});
        let response = handle_wire_message(msg, &store, &mut cache);
        match response {
            WireMessage::Pong(info) => {
                assert_eq!(info.wire_protocol, shape_wire::WIRE_PROTOCOL_V2);
                assert!(info.capabilities.contains(&"call".to_string()));
                assert!(info.capabilities.contains(&"blob-negotiation".to_string()));
            }
            _ => panic!("Expected Pong response"),
        }
    }

    #[test]
    fn test_handle_wire_message_execute_returns_v2_stub() {
        let store = temp_store();
        let mut cache = RemoteBlobCache::default_cache();
        let msg = WireMessage::Execute(ExecuteRequest {
            code: "42".to_string(),
            request_id: 5,
        });
        let response = handle_wire_message(msg, &store, &mut cache);
        match response {
            WireMessage::ExecuteResponse(resp) => {
                assert_eq!(resp.request_id, 5);
                assert!(!resp.success);
                assert!(resp.error.as_ref().unwrap().contains("not yet implemented"));
            }
            _ => panic!("Expected ExecuteResponse"),
        }
    }

    #[test]
    fn test_handle_wire_message_validate_returns_v2_stub() {
        let store = temp_store();
        let mut cache = RemoteBlobCache::default_cache();
        let msg = WireMessage::Validate(ValidateRequest {
            code: "let x = 1".to_string(),
            request_id: 10,
        });
        let response = handle_wire_message(msg, &store, &mut cache);
        match response {
            WireMessage::ValidateResponse(resp) => {
                assert_eq!(resp.request_id, 10);
                assert!(!resp.success);
                assert!(resp.diagnostics[0].message.contains("not yet implemented"));
            }
            _ => panic!("Expected ValidateResponse"),
        }
    }

    #[test]
    fn test_handle_wire_message_auth_returns_v2_stub() {
        let store = temp_store();
        let mut cache = RemoteBlobCache::default_cache();
        let msg = WireMessage::Auth(AuthRequest {
            token: "test".to_string(),
        });
        let response = handle_wire_message(msg, &store, &mut cache);
        match response {
            WireMessage::AuthResponse(resp) => {
                assert!(!resp.authenticated);
                assert!(resp.error.as_ref().unwrap().contains("not yet implemented"));
            }
            _ => panic!("Expected AuthResponse"),
        }
    }

    #[test]
    fn test_handle_wire_message_blob_negotiation() {
        let store = temp_store();
        let mut cache = RemoteBlobCache::new(10);
        let h1 = mk_hash(1);
        let h2 = mk_hash(2);
        cache.insert(h1, mk_blob("f1", h1, vec![]));

        let msg = WireMessage::BlobNegotiation(BlobNegotiationRequest {
            offered_hashes: vec![h1, h2],
        });
        let response = handle_wire_message(msg, &store, &mut cache);
        match response {
            WireMessage::BlobNegotiationReply(resp) => {
                assert_eq!(resp.known_hashes.len(), 1);
                assert!(resp.known_hashes.contains(&h1));
            }
            _ => panic!("Expected BlobNegotiationReply"),
        }
    }

    // Track A.2B: a closure payload serialised through the slot/serializable
    // round-trip and replayed through the receiver's
    // `closure_function_layouts` slice. The pre-bulldozer test
    // (`test_a2b_closure_arg_roundtrip_with_layouts`) constructed the
    // closure via deleted ValueWord constructors
    // (`from_f64` + `into_raw_bits` + `from_heap_value(HeapValue::ClosureRaw(_))`)
    // and round-tripped it through the deleted
    // `nanboxed_to_serializable` / `serializable_to_nanboxed_with_layouts`
    // pair. Re-author against the kind-threaded slot pipeline once the
    // Phase-2c snapshot rebuild lands (ADR-006 §2.7.4 + addendum). The
    // wire schema (`function_id: u32`, `type_id: u32`, `upvalues: Vec<…>`)
    // is preserved verbatim.
}