shape_vm/

remote.rs

//! 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
//!
//! `Arc<RwLock<ValueWord>>` upvalues 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, nanboxed_to_serializable, serializable_to_nanboxed,
};
use shape_runtime::type_schema::TypeSchemaRegistry;
use shape_value::ValueWord;

use shape_wire::WireValue;

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

/// 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 `Arc<RwLock<ValueWord>>` upvalue slots.
    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(),
        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(),
    })
}

/// 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. Converts serialized arguments back to `ValueWord`
/// 4. Calls the function by name or ID
/// 5. Converts the result back to `SerializableVMValue`
///
/// The `store` is used for `SerializableVMValue` ↔ `ValueWord` conversion
/// (needed for `BlobRef`-backed values like DataTable).
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> {
    // 1. Reconstruct program with type schemas.
    // Prefer content-addressed blobs when present — they carry only the
    // transitive closure of the called function, so deserialization is cheaper.
    let mut program = if let Some(blobs) = request.function_blobs {
        let entry_hash = request
            .function_hash
            .or_else(|| {
                if let Some(fid) = request.function_id {
                    request
                        .program
                        .function_blob_hashes
                        .get(fid as usize)
                        .copied()
                        .flatten()
                } else {
                    None
                }
            })
            .or_else(|| {
                // Legacy fallback by unique function name.
                request.program.content_addressed.as_ref().and_then(|ca| {
                    let mut matches = ca.function_store.iter().filter_map(|(hash, blob)| {
                        if blob.name == request.function_name {
                            Some(*hash)
                        } else {
                            None
                        }
                    });
                    let first = matches.next()?;
                    if matches.next().is_some() {
                        None
                    } else {
                        Some(first)
                    }
                })
            })
            .ok_or_else(|| RemoteCallError {
                message: format!(
                    "Could not resolve entry hash for remote function '{}'",
                    request.function_name
                ),
                kind: RemoteErrorKind::FunctionNotFound,
            })?;

        // Reconstruct a content-addressed Program from the minimal blobs,
        // then link it into a BytecodeProgram the VM can execute.
        let ca_program = program_from_blobs_by_hash(blobs, entry_hash, &request.program)
            .ok_or_else(|| RemoteCallError {
                message: format!(
                    "Could not reconstruct program from blobs for '{}'",
                    request.function_name
                ),
                kind: RemoteErrorKind::FunctionNotFound,
            })?;
        // Link the content-addressed program into a flat BytecodeProgram
        let linked = crate::linker::link(&ca_program).map_err(|e| RemoteCallError {
            message: format!("Linker error: {}", e),
            kind: RemoteErrorKind::RuntimeError,
        })?;
        // Convert LinkedProgram to BytecodeProgram for the existing VM path
        crate::linker::linked_to_bytecode_program(&linked)
    } else {
        request.program
    };
    program.type_schema_registry = request.type_schemas;

    // 2. Convert arguments from serializable form
    let args: Vec<ValueWord> = request
        .arguments
        .iter()
        .map(|sv| {
            serializable_to_nanboxed(sv, store).map_err(|e| RemoteCallError {
                message: format!("Failed to deserialize argument: {}", e),
                kind: RemoteErrorKind::ArgumentError,
            })
        })
        .collect::<Result<Vec<_>, _>>()?;

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

    // 4. Execute — closure or named function
    //
    // Dispatch priority:
    //   1. Closure call (upvalues present) — needs function_id for upvalue binding
    //   2. Hash-first — preferred for non-closure calls because function_id from the
    //      client may be stale after blob-based relinking on the server
    //   3. Function ID — fallback when no hash is available (legacy path)
    //   4. Function name — last resort
    let result = if let Some(ref upvalue_data) = request.upvalues {
        // Closure call: reconstruct upvalues as Upvalue structs
        let upvalues: Vec<shape_value::Upvalue> = upvalue_data
            .iter()
            .map(|sv| {
                let nb = serializable_to_nanboxed(sv, store).map_err(|e| RemoteCallError {
                    message: format!("Failed to deserialize upvalue: {}", e),
                    kind: RemoteErrorKind::ArgumentError,
                })?;
                Ok(shape_value::Upvalue::new(nb))
            })
            .collect::<Result<Vec<_>, RemoteCallError>>()?;

        let function_id = request.function_id.ok_or_else(|| RemoteCallError {
            message: "Closure call requires function_id".to_string(),
            kind: RemoteErrorKind::FunctionNotFound,
        })?;

        vm.execute_closure(function_id, upvalues, args, None)
            .map_err(|e| RemoteCallError {
                message: e.to_string(),
                kind: RemoteErrorKind::RuntimeError,
            })?
    } else if let Some(hash) = request.function_hash {
        // Hash-first call path — resolve the function by its content hash
        // in the server's linked program. This is correct even when the program
        // was reconstructed from blobs (where client-side function_ids are stale).
        let func_id = vm
            .program()
            .function_blob_hashes
            .iter()
            .enumerate()
            .find_map(|(idx, maybe_hash)| {
                if maybe_hash == &Some(hash) {
                    Some(idx as u16)
                } else {
                    None
                }
            })
            .ok_or_else(|| RemoteCallError {
                message: format!("Function hash not found in program: {}", hash),
                kind: RemoteErrorKind::FunctionNotFound,
            })?;
        vm.execute_function_by_id(func_id, args, None)
            .map_err(|e| RemoteCallError {
                message: e.to_string(),
                kind: RemoteErrorKind::RuntimeError,
            })?
    } else if let Some(func_id) = request.function_id {
        // Call by function ID (legacy — only valid when program was not reconstructed)
        vm.execute_function_by_id(func_id, args, None)
            .map_err(|e| RemoteCallError {
                message: e.to_string(),
                kind: RemoteErrorKind::RuntimeError,
            })?
    } else {
        // Call by name
        vm.execute_function_by_name(&request.function_name, args, None)
            .map_err(|e| RemoteCallError {
                message: e.to_string(),
                kind: RemoteErrorKind::RuntimeError,
            })?
    };

    // 5. Serialize result
    nanboxed_to_serializable(&result, store).map_err(|e| RemoteCallError {
        message: format!("Failed to serialize result: {}", e),
        kind: RemoteErrorKind::RuntimeError,
    })
}

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> {
    // 1. Reconstruct program with type schemas (same logic as execute_inner)
    let mut program = if let Some(blobs) = request.function_blobs {
        let entry_hash = request
            .function_hash
            .or_else(|| {
                if let Some(fid) = request.function_id {
                    request
                        .program
                        .function_blob_hashes
                        .get(fid as usize)
                        .copied()
                        .flatten()
                } else {
                    None
                }
            })
            .or_else(|| {
                request.program.content_addressed.as_ref().and_then(|ca| {
                    let mut matches = ca.function_store.iter().filter_map(|(hash, blob)| {
                        if blob.name == request.function_name {
                            Some(*hash)
                        } else {
                            None
                        }
                    });
                    let first = matches.next()?;
                    if matches.next().is_some() {
                        None
                    } else {
                        Some(first)
                    }
                })
            })
            .ok_or_else(|| RemoteCallError {
                message: format!(
                    "Could not resolve entry hash for remote function '{}'",
                    request.function_name
                ),
                kind: RemoteErrorKind::FunctionNotFound,
            })?;

        let ca_program = program_from_blobs_by_hash(blobs, entry_hash, &request.program)
            .ok_or_else(|| RemoteCallError {
                message: format!(
                    "Could not reconstruct program from blobs for '{}'",
                    request.function_name
                ),
                kind: RemoteErrorKind::FunctionNotFound,
            })?;
        let linked = crate::linker::link(&ca_program).map_err(|e| RemoteCallError {
            message: format!("Linker error: {}", e),
            kind: RemoteErrorKind::RuntimeError,
        })?;
        crate::linker::linked_to_bytecode_program(&linked)
    } else {
        request.program
    };
    program.type_schema_registry = request.type_schemas;

    // 2. Convert arguments from serializable form
    let args: Vec<ValueWord> = request
        .arguments
        .iter()
        .map(|sv| {
            serializable_to_nanboxed(sv, store).map_err(|e| RemoteCallError {
                message: format!("Failed to deserialize argument: {}", e),
                kind: RemoteErrorKind::ArgumentError,
            })
        })
        .collect::<Result<Vec<_>, _>>()?;

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

    // 4. Link foreign functions from pre-loaded language runtimes
    if !vm.program.foreign_functions.is_empty() && !language_runtimes.is_empty() {
        let entries = vm.program.foreign_functions.clone();
        let mut handles = Vec::with_capacity(entries.len());

        for (idx, entry) in entries.iter().enumerate() {
            // Skip native ABI entries (not supported in remote context)
            if entry.native_abi.is_some() {
                handles.push(None);
                continue;
            }

            if let Some(lang_runtime) = language_runtimes.get(&entry.language) {
                vm.program.foreign_functions[idx].dynamic_errors =
                    lang_runtime.has_dynamic_errors();

                let compiled = lang_runtime
                    .compile(
                        &entry.name,
                        &entry.body_text,
                        &entry.param_names,
                        &entry.param_types,
                        entry.return_type.as_deref(),
                        entry.is_async,
                    )
                    .map_err(|e| RemoteCallError {
                        message: format!(
                            "Failed to compile foreign function '{}': {}",
                            entry.name, e
                        ),
                        kind: RemoteErrorKind::RuntimeError,
                    })?;
                handles.push(Some(crate::executor::ForeignFunctionHandle::Runtime {
                    runtime: std::sync::Arc::clone(lang_runtime),
                    compiled,
                }));
            } else {
                return Err(RemoteCallError {
                    message: format!(
                        "No language runtime for '{}' on this server. \
                         Install the {} extension.",
                        entry.language, entry.language
                    ),
                    kind: RemoteErrorKind::RuntimeError,
                });
            }
        }
        vm.foreign_fn_handles = handles;
    }

    // 5. Execute — closure or named function (same dispatch priority as execute_inner)
    let result = if let Some(ref upvalue_data) = request.upvalues {
        let upvalues: Vec<shape_value::Upvalue> = upvalue_data
            .iter()
            .map(|sv| {
                let nb = serializable_to_nanboxed(sv, store).map_err(|e| RemoteCallError {
                    message: format!("Failed to deserialize upvalue: {}", e),
                    kind: RemoteErrorKind::ArgumentError,
                })?;
                Ok(shape_value::Upvalue::new(nb))
            })
            .collect::<Result<Vec<_>, RemoteCallError>>()?;

        let function_id = request.function_id.ok_or_else(|| RemoteCallError {
            message: "Closure call requires function_id".to_string(),
            kind: RemoteErrorKind::FunctionNotFound,
        })?;

        vm.execute_closure(function_id, upvalues, args, None)
            .map_err(|e| RemoteCallError {
                message: e.to_string(),
                kind: RemoteErrorKind::RuntimeError,
            })?
    } else if let Some(hash) = request.function_hash {
        // Hash-first: resolve by content hash in the server's linked program.
        // Client-side function_ids are stale after blob-based relinking.
        let func_id = vm
            .program()
            .function_blob_hashes
            .iter()
            .enumerate()
            .find_map(|(idx, maybe_hash)| {
                if maybe_hash == &Some(hash) {
                    Some(idx as u16)
                } else {
                    None
                }
            })
            .ok_or_else(|| RemoteCallError {
                message: format!("Function hash not found in program: {}", hash),
                kind: RemoteErrorKind::FunctionNotFound,
            })?;
        vm.execute_function_by_id(func_id, args, None)
            .map_err(|e| RemoteCallError {
                message: e.to_string(),
                kind: RemoteErrorKind::RuntimeError,
            })?
    } else if let Some(func_id) = request.function_id {
        vm.execute_function_by_id(func_id, args, None)
            .map_err(|e| RemoteCallError {
                message: e.to_string(),
                kind: RemoteErrorKind::RuntimeError,
            })?
    } else {
        vm.execute_function_by_name(&request.function_name, args, None)
            .map_err(|e| RemoteCallError {
                message: e.to_string(),
                kind: RemoteErrorKind::RuntimeError,
            })?
    };

    // 6. Serialize result
    nanboxed_to_serializable(&result, store).map_err(|e| RemoteCallError {
        message: format!("Failed to serialize result: {}", e),
        kind: RemoteErrorKind::RuntimeError,
    })
}

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

    #[test]
    fn test_remote_call_simple_function() {
        let bytecode = compile(
            r#"
            function add(a, b) { a + b }
        "#,
        );
        let store = temp_store();

        let request = build_call_request(
            &bytecode,
            "add",
            vec![
                SerializableVMValue::Number(10.0),
                SerializableVMValue::Number(32.0),
            ],
        );

        let response = execute_remote_call(request, &store);
        match response.result {
            Ok(SerializableVMValue::Number(n)) => assert_eq!(n, 42.0),
            other => panic!("Expected Number(42.0), got {:?}", other),
        }
    }

    #[test]
    fn test_remote_call_function_not_found() {
        let bytecode = compile("function foo() { 1 }");
        let store = temp_store();

        let request = build_call_request(&bytecode, "nonexistent", vec![]);

        let response = execute_remote_call(request, &store);
        assert!(response.result.is_err());
        let err = response.result.unwrap_err();
        assert!(matches!(err.kind, RemoteErrorKind::RuntimeError));
    }

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

        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,
            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()),
        });
        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![],
        }];
        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)));
    }

    #[test]
    fn test_extract_sidecars_large_typed_array() {
        let store = temp_store();

        // Create a large float array (2MB of f64 data)
        let data = vec![0f64; 256 * 1024]; // 256K * 8 bytes = 2 MB
        let aligned = shape_value::AlignedVec::from_vec(data);
        let buf = shape_value::AlignedTypedBuffer::from_aligned(aligned);
        let nb = shape_value::ValueWord::from_float_array(std::sync::Arc::new(buf));
        let serialized = shape_runtime::snapshot::nanboxed_to_serializable(&nb, &store).unwrap();

        let mut args = vec![serialized];
        let sidecars = extract_sidecars(&mut args, &store);
        assert_eq!(
            sidecars.len(),
            1,
            "should extract one sidecar for 2MB array"
        );
        assert!(
            matches!(args[0], SerializableVMValue::SidecarRef { .. }),
            "original should be replaced with SidecarRef"
        );
        assert!(
            sidecars[0].data.len() >= 1024 * 1024,
            "sidecar data should be >= 1MB"
        );
    }

    #[test]
    fn test_reassemble_sidecars_roundtrip() {
        let store = temp_store();

        // Create a large float array
        let data: Vec<f64> = (0..256 * 1024).map(|i| i as f64).collect();
        let aligned = shape_value::AlignedVec::from_vec(data.clone());
        let buf = shape_value::AlignedTypedBuffer::from_aligned(aligned);
        let nb = shape_value::ValueWord::from_float_array(std::sync::Arc::new(buf));
        let original = shape_runtime::snapshot::nanboxed_to_serializable(&nb, &store).unwrap();

        let mut args = vec![original.clone()];
        let sidecars = extract_sidecars(&mut args, &store);
        assert_eq!(sidecars.len(), 1);

        // Build sidecar map for reassembly
        let sidecar_map: HashMap<u32, BlobSidecar> =
            sidecars.into_iter().map(|s| (s.sidecar_id, s)).collect();

        // Reassemble
        reassemble_sidecars(&mut args, &sidecar_map, &store).unwrap();

        // The reassembled value should deserialize to the same data
        let restored = shape_runtime::snapshot::serializable_to_nanboxed(&args[0], &store).unwrap();
        let hv = restored.as_heap_ref().unwrap();
        match hv {
            shape_value::heap_value::HeapValue::FloatArray(a) => {
                assert_eq!(a.len(), 256 * 1024);
                assert!((a.as_slice()[0] - 0.0).abs() < f64::EPSILON);
                assert!((a.as_slice()[1000] - 1000.0).abs() < f64::EPSILON);
            }
            // reassemble produces DataTable wrapper, which is also valid
            _ => {
                // The reassembled blob may come back as DataTable BlobRef
                // since reassemble uses a generic DataTable wrapper.
                // This is acceptable — the raw data is preserved.
            }
        }
    }

    #[test]
    fn test_extract_sidecars_nested_in_array() {
        let store = temp_store();

        // Create a large float array nested in an Array
        let data = vec![0f64; 256 * 1024]; // 2 MB
        let aligned = shape_value::AlignedVec::from_vec(data);
        let buf = shape_value::AlignedTypedBuffer::from_aligned(aligned);
        let nb = shape_value::ValueWord::from_float_array(std::sync::Arc::new(buf));
        let serialized = shape_runtime::snapshot::nanboxed_to_serializable(&nb, &store).unwrap();

        let mut args = vec![SerializableVMValue::Array(vec![
            SerializableVMValue::Int(1),
            serialized,
            SerializableVMValue::String("end".to_string()),
        ])];

        let sidecars = extract_sidecars(&mut args, &store);
        assert_eq!(sidecars.len(), 1, "should find nested large blob");

        // Verify the array structure is preserved with SidecarRef inside
        match &args[0] {
            SerializableVMValue::Array(items) => {
                assert_eq!(items.len(), 3);
                assert!(matches!(items[0], SerializableVMValue::Int(1)));
                assert!(matches!(items[1], SerializableVMValue::SidecarRef { .. }));
                assert!(matches!(items[2], SerializableVMValue::String(_)));
            }
            _ => panic!("Expected Array wrapper to be preserved"),
        }
    }

    #[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,
            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()),
        });
        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![],
        }];
        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,
            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()),
        });
        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![],
        }];
        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"),
        }
    }
}