barbacane_wasm/

instance.rs

//! WASM plugin instance management.
//!
//! Each plugin instance wraps a wasmtime Store and Instance with the
//! plugin state required for host function calls.

use std::collections::HashMap;
use std::sync::Arc;

use wasmtime::{Caller, Engine, Instance, Linker, Memory, Store, TypedFunc};

use barbacane_plugin_sdk::types::base64_body;
use serde::Deserialize;
use std::collections::BTreeMap;

use crate::broker::BrokerMessage;
use crate::engine::CompiledModule;
use crate::error::WasmError;
use crate::http_client::{
    HttpClient, HttpRequest as HttpClientRequest, HttpResponse as HttpClientResponse,
};
use crate::limits::PluginLimits;

/// Events sent through the streaming channel by `host_http_stream` (ADR-0023).
///
/// The host function sends `Headers` once (before any chunks), then zero or
/// more `Chunk` events. The sender is dropped when the upstream stream ends,
/// signalling the receiver that the body is complete.
#[derive(Debug)]
pub enum StreamEvent {
    /// Response status and headers from the upstream (sent before body chunks).
    Headers {
        status: u16,
        headers: BTreeMap<String, String>,
    },
    /// Body chunk forwarded from the upstream streaming response.
    Chunk(bytes::Bytes),
}

/// HTTP request format from WASM plugins.
/// This matches the format used by http-upstream plugin.
///
/// The `body` field uses base64 encoding over JSON to support binary payloads
/// (e.g. multipart/form-data with file uploads).
#[derive(Debug, Deserialize)]
pub(crate) struct PluginHttpRequest {
    pub(crate) method: String,
    pub(crate) url: String,
    #[serde(default)]
    pub(crate) headers: BTreeMap<String, String>,
    #[serde(default, with = "base64_body")]
    pub(crate) body: Option<Vec<u8>>,
    #[serde(default)]
    pub(crate) timeout_ms: Option<u64>,
}

/// Per-request context passed to plugins.
#[derive(Debug, Clone, Default)]
pub struct RequestContext {
    /// Key-value store for inter-middleware communication.
    pub values: HashMap<String, String>,

    /// Result buffer for host_context_get.
    pub last_get_result: Option<String>,

    /// Trace ID for distributed tracing.
    pub trace_id: String,

    /// Request ID.
    pub request_id: String,
}

impl RequestContext {
    /// Create a new request context.
    pub fn new(trace_id: String, request_id: String) -> Self {
        Self {
            values: HashMap::new(),
            last_get_result: None,
            trace_id,
            request_id,
        }
    }
}

/// State attached to each WASM store.
pub struct PluginState {
    /// Plugin name for logging.
    pub plugin_name: String,

    /// Output buffer for plugin results.
    pub output_buffer: Vec<u8>,

    /// Per-request context.
    pub context: RequestContext,

    /// Maximum memory in bytes.
    pub max_memory: usize,

    /// HTTP client for outbound requests (shared).
    pub http_client: Option<Arc<HttpClient>>,

    /// Result buffer for host_http_read_result.
    pub last_http_result: Option<Vec<u8>>,

    /// Resolved secrets store (shared across instances).
    pub secrets: crate::secrets::SecretsStore,

    /// Result buffer for host_secret_read_result.
    pub last_secret_result: Option<Vec<u8>>,

    /// Rate limiter (shared across instances).
    pub rate_limiter: Option<crate::rate_limiter::RateLimiter>,

    /// Result buffer for host_rate_limit_read_result.
    pub last_rate_limit_result: Option<Vec<u8>>,

    /// Response cache (shared across instances).
    pub response_cache: Option<crate::cache::ResponseCache>,

    /// Result buffer for host_cache_read_result.
    pub last_cache_result: Option<Vec<u8>>,

    /// Metrics registry for plugin telemetry (shared).
    pub metrics: Option<Arc<barbacane_telemetry::MetricsRegistry>>,

    /// Kafka publisher for host_kafka_publish (shared).
    pub kafka_publisher: Option<Arc<crate::kafka_client::KafkaPublisher>>,

    /// NATS publisher for host_nats_publish (shared).
    pub nats_publisher: Option<Arc<crate::nats_client::NatsPublisher>>,

    /// Result buffer for host_kafka_publish / host_nats_publish.
    pub last_broker_result: Option<Vec<u8>>,

    /// Result buffer for host_uuid_read_result.
    pub last_uuid_result: Option<Vec<u8>>,

    /// Channel sender for host_http_stream (ADR-0023).
    ///
    /// Set by the host before calling a streaming-capable dispatcher. The host
    /// function sends `StreamEvent::Headers` once, then `StreamEvent::Chunk` for
    /// each body chunk, then drops the sender to signal end-of-stream.
    pub stream_sender: Option<Arc<tokio::sync::mpsc::UnboundedSender<StreamEvent>>>,

    /// Upstream WebSocket upgrade request from `host_ws_upgrade` (ADR-0026).
    ///
    /// After a successful `host_ws_upgrade`, the request params are stored here.
    /// The actual connection is deferred to the async relay task on the main
    /// runtime, because the TcpStream must be created on the runtime that will
    /// drive it (a temporary runtime's I/O driver dies when the runtime drops).
    pub ws_upgrade_request: Option<crate::ws_client::WsUpgradeRequest>,

    // --- Side-channel body buffers ---
    // Bodies travel as raw bytes via dedicated host functions instead of
    // base64-encoded inside JSON. This eliminates the ~3.65× memory overhead
    // and allows 10MB+ bodies within the default 16MB WASM memory limit.
    /// Request/response body held by the host, set before calling the handler.
    /// Plugins read it via `host_body_len()` + `host_body_read()`.
    pub request_body: Option<Vec<u8>>,

    /// Output body set by the plugin via `host_body_set()` or `host_body_clear()`.
    /// Outer Option: was the function called? Inner Option: body or None.
    pub output_body: Option<Option<Vec<u8>>>,

    /// HTTP response body from `host_http_call`, held separately from the
    /// JSON metadata in `last_http_result`. Plugins read it via
    /// `host_http_response_body_len()` + `host_http_response_body_read()`.
    pub http_response_body: Option<Vec<u8>>,

    /// Outbound HTTP request body set by the plugin via
    /// `host_http_request_body_set()`. Consumed by `host_http_call`.
    pub http_request_body: Option<Vec<u8>>,
}

#[allow(dead_code)] // Constructors used by different pool configurations
impl PluginState {
    /// Create new plugin state.
    pub fn new(plugin_name: String, limits: &PluginLimits) -> Self {
        Self {
            plugin_name,
            output_buffer: Vec::new(),
            context: RequestContext::default(),
            max_memory: limits.max_memory_bytes,
            http_client: None,
            last_http_result: None,
            secrets: crate::secrets::SecretsStore::new(),
            last_secret_result: None,
            rate_limiter: None,
            last_rate_limit_result: None,
            response_cache: None,
            last_cache_result: None,
            metrics: None,
            kafka_publisher: None,
            nats_publisher: None,
            last_broker_result: None,
            last_uuid_result: None,
            stream_sender: None,
            ws_upgrade_request: None,
            request_body: None,
            output_body: None,
            http_response_body: None,
            http_request_body: None,
        }
    }

    /// Create new plugin state with HTTP client.
    pub fn with_http_client(
        plugin_name: String,
        limits: &PluginLimits,
        http_client: Arc<HttpClient>,
    ) -> Self {
        Self {
            plugin_name,
            output_buffer: Vec::new(),
            context: RequestContext::default(),
            max_memory: limits.max_memory_bytes,
            http_client: Some(http_client),
            last_http_result: None,
            secrets: crate::secrets::SecretsStore::new(),
            last_secret_result: None,
            rate_limiter: None,
            last_rate_limit_result: None,
            response_cache: None,
            last_cache_result: None,
            metrics: None,
            kafka_publisher: None,
            nats_publisher: None,
            last_broker_result: None,
            last_uuid_result: None,
            stream_sender: None,
            ws_upgrade_request: None,
            request_body: None,
            output_body: None,
            http_response_body: None,
            http_request_body: None,
        }
    }

    /// Create new plugin state with HTTP client and secrets.
    pub fn with_http_client_and_secrets(
        plugin_name: String,
        limits: &PluginLimits,
        http_client: Arc<HttpClient>,
        secrets: crate::secrets::SecretsStore,
    ) -> Self {
        Self {
            plugin_name,
            output_buffer: Vec::new(),
            context: RequestContext::default(),
            max_memory: limits.max_memory_bytes,
            http_client: Some(http_client),
            last_http_result: None,
            secrets,
            last_secret_result: None,
            rate_limiter: None,
            last_rate_limit_result: None,
            response_cache: None,
            last_cache_result: None,
            metrics: None,
            kafka_publisher: None,
            nats_publisher: None,
            last_broker_result: None,
            last_uuid_result: None,
            stream_sender: None,
            ws_upgrade_request: None,
            request_body: None,
            output_body: None,
            http_response_body: None,
            http_request_body: None,
        }
    }

    /// Create new plugin state with all options.
    #[allow(clippy::too_many_arguments)]
    pub fn with_all_options(
        plugin_name: String,
        limits: &PluginLimits,
        http_client: Option<Arc<HttpClient>>,
        secrets: crate::secrets::SecretsStore,
        rate_limiter: Option<crate::rate_limiter::RateLimiter>,
        response_cache: Option<crate::cache::ResponseCache>,
        nats_publisher: Option<Arc<crate::nats_client::NatsPublisher>>,
        kafka_publisher: Option<Arc<crate::kafka_client::KafkaPublisher>>,
    ) -> Self {
        Self {
            plugin_name,
            output_buffer: Vec::new(),
            context: RequestContext::default(),
            max_memory: limits.max_memory_bytes,
            http_client,
            last_http_result: None,
            secrets,
            last_secret_result: None,
            rate_limiter,
            last_rate_limit_result: None,
            response_cache,
            last_cache_result: None,
            metrics: None,
            kafka_publisher,
            nats_publisher,
            last_broker_result: None,
            last_uuid_result: None,
            stream_sender: None,
            ws_upgrade_request: None,
            request_body: None,
            output_body: None,
            http_response_body: None,
            http_request_body: None,
        }
    }

    /// Create new plugin state with all options including metrics.
    #[allow(clippy::too_many_arguments)]
    pub fn with_all_options_and_metrics(
        plugin_name: String,
        limits: &PluginLimits,
        http_client: Option<Arc<HttpClient>>,
        secrets: crate::secrets::SecretsStore,
        rate_limiter: Option<crate::rate_limiter::RateLimiter>,
        response_cache: Option<crate::cache::ResponseCache>,
        nats_publisher: Option<Arc<crate::nats_client::NatsPublisher>>,
        kafka_publisher: Option<Arc<crate::kafka_client::KafkaPublisher>>,
        metrics: Option<Arc<barbacane_telemetry::MetricsRegistry>>,
    ) -> Self {
        Self {
            plugin_name,
            output_buffer: Vec::new(),
            context: RequestContext::default(),
            max_memory: limits.max_memory_bytes,
            http_client,
            last_http_result: None,
            secrets,
            last_secret_result: None,
            rate_limiter,
            last_rate_limit_result: None,
            response_cache,
            last_cache_result: None,
            metrics,
            kafka_publisher,
            nats_publisher,
            last_broker_result: None,
            last_uuid_result: None,
            stream_sender: None,
            ws_upgrade_request: None,
            request_body: None,
            output_body: None,
            http_response_body: None,
            http_request_body: None,
        }
    }

    /// Get the output buffer contents.
    pub fn take_output(&mut self) -> Vec<u8> {
        std::mem::take(&mut self.output_buffer)
    }

    /// Set the request context for this call.
    pub fn set_context(&mut self, context: RequestContext) {
        self.context = context;
    }

    /// Set the stream sender for host_http_stream (ADR-0023).
    pub fn set_stream_sender(
        &mut self,
        sender: Arc<tokio::sync::mpsc::UnboundedSender<StreamEvent>>,
    ) {
        self.stream_sender = Some(sender);
    }

    /// Take the upstream WebSocket upgrade request from host_ws_upgrade (ADR-0026).
    pub fn take_ws_upgrade_request(&mut self) -> Option<crate::ws_client::WsUpgradeRequest> {
        self.ws_upgrade_request.take()
    }

    /// Set the request body for the next handler call (side-channel).
    pub fn set_request_body(&mut self, body: Option<Vec<u8>>) {
        self.request_body = body;
    }

    /// Take the output body set by the plugin via host_body_set/host_body_clear.
    /// Returns `None` if the plugin didn't call either function (body unchanged).
    /// Returns `Some(None)` if the plugin called host_body_clear.
    /// Returns `Some(Some(bytes))` if the plugin called host_body_set.
    pub fn take_output_body(&mut self) -> Option<Option<Vec<u8>>> {
        self.output_body.take()
    }
}

impl wasmtime::ResourceLimiter for PluginState {
    fn memory_growing(
        &mut self,
        _current: usize,
        desired: usize,
        _maximum: Option<usize>,
    ) -> Result<bool, wasmtime::Error> {
        Ok(desired <= self.max_memory)
    }

    fn table_growing(
        &mut self,
        _current: usize,
        desired: usize,
        _maximum: Option<usize>,
    ) -> Result<bool, wasmtime::Error> {
        // Allow reasonable table growth
        Ok(desired <= 10_000)
    }
}

/// A WASM plugin instance ready for execution.
pub struct PluginInstance {
    store: Store<PluginState>,
    _instance: Instance,
    limits: PluginLimits,

    // Cached function references
    init_func: Option<TypedFunc<(i32, i32), i32>>,
    on_request_func: Option<TypedFunc<(i32, i32), i32>>,
    on_response_func: Option<TypedFunc<(i32, i32), i32>>,
    dispatch_func: Option<TypedFunc<(i32, i32), i32>>,
    alloc_func: Option<TypedFunc<i32, i32>>,
    memory: Memory,
}

impl PluginInstance {
    /// Create a new plugin instance from a compiled module.
    pub fn new(
        engine: &Engine,
        module: &CompiledModule,
        limits: PluginLimits,
    ) -> Result<Self, WasmError> {
        Self::new_with_options(engine, module, limits, None, None)
    }

    /// Create a new plugin instance with an HTTP client for outbound calls.
    pub fn new_with_http_client(
        engine: &Engine,
        module: &CompiledModule,
        limits: PluginLimits,
        http_client: Option<Arc<HttpClient>>,
    ) -> Result<Self, WasmError> {
        Self::new_with_options(engine, module, limits, http_client, None)
    }

    /// Create a new plugin instance with HTTP client and secrets.
    pub fn new_with_options(
        engine: &Engine,
        module: &CompiledModule,
        limits: PluginLimits,
        http_client: Option<Arc<HttpClient>>,
        secrets: Option<crate::secrets::SecretsStore>,
    ) -> Result<Self, WasmError> {
        Self::new_with_all_options(
            engine,
            module,
            limits,
            http_client,
            secrets,
            None,
            None,
            None,
            None,
        )
    }

    /// Create a new plugin instance with all options including rate limiter and cache.
    #[allow(clippy::too_many_arguments)]
    pub fn new_with_all_options(
        engine: &Engine,
        module: &CompiledModule,
        limits: PluginLimits,
        http_client: Option<Arc<HttpClient>>,
        secrets: Option<crate::secrets::SecretsStore>,
        rate_limiter: Option<crate::rate_limiter::RateLimiter>,
        response_cache: Option<crate::cache::ResponseCache>,
        nats_publisher: Option<Arc<crate::nats_client::NatsPublisher>>,
        kafka_publisher: Option<Arc<crate::kafka_client::KafkaPublisher>>,
    ) -> Result<Self, WasmError> {
        let state = PluginState::with_all_options(
            module.name.clone(),
            &limits,
            http_client,
            secrets.unwrap_or_default(),
            rate_limiter,
            response_cache,
            nats_publisher,
            kafka_publisher,
        );
        let mut store = Store::new(engine, state);

        // Set fuel for execution limiting
        store
            .set_fuel(limits.max_fuel)
            .map_err(|e| WasmError::Instantiation(format!("failed to set fuel: {}", e)))?;

        // Enable resource limiting
        store.limiter(|state| state);

        // Create linker and add host functions
        let mut linker = Linker::new(engine);
        add_host_functions(&mut linker)?;

        // Instantiate the module
        let instance = linker
            .instantiate(&mut store, module.module())
            .map_err(|e| WasmError::Instantiation(e.to_string()))?;

        // Get memory
        let memory = instance
            .get_memory(&mut store, "memory")
            .ok_or_else(|| WasmError::MissingExport("memory".into()))?;

        // Cache function references
        let init_func = instance
            .get_typed_func::<(i32, i32), i32>(&mut store, "init")
            .ok();
        let on_request_func = instance
            .get_typed_func::<(i32, i32), i32>(&mut store, "on_request")
            .ok();
        let on_response_func = instance
            .get_typed_func::<(i32, i32), i32>(&mut store, "on_response")
            .ok();
        let dispatch_func = instance
            .get_typed_func::<(i32, i32), i32>(&mut store, "dispatch")
            .ok();
        let alloc_func = instance
            .get_typed_func::<i32, i32>(&mut store, "alloc")
            .ok();

        Ok(Self {
            store,
            _instance: instance,
            limits,
            init_func,
            on_request_func,
            on_response_func,
            dispatch_func,
            alloc_func,
            memory,
        })
    }

    /// Get the plugin name.
    pub fn name(&self) -> &str {
        &self.store.data().plugin_name
    }

    /// Write data to the plugin's linear memory and return the pointer.
    ///
    /// Uses the plugin's exported `alloc` function so that dlmalloc is aware
    /// of the allocation and will not reuse the region during deserialization.
    /// Falls back to growing memory directly for legacy plugins that lack the
    /// `alloc` export (only safe for very small payloads like config JSON).
    pub fn write_to_memory(&mut self, data: &[u8]) -> Result<i32, WasmError> {
        if data.is_empty() {
            return Ok(0);
        }

        if let Some(alloc_func) = self.alloc_func.clone() {
            // Allocate via the plugin's own allocator — dlmalloc tracks this
            // region and will not hand it out again during deserialization.
            let ptr = alloc_func
                .call(&mut self.store, data.len() as i32)
                .map_err(|e| WasmError::Trap(format!("alloc failed: {}", e)))?;

            if ptr == 0 {
                let current_size = self.memory.data_size(&self.store);
                return Err(WasmError::MemoryLimitExceeded {
                    requested: data.len(),
                    limit: self.limits.max_memory_bytes.saturating_sub(current_size),
                });
            }

            self.memory
                .write(&mut self.store, ptr as usize, data)
                .map_err(|e| WasmError::Trap(format!("memory write failed: {}", e)))?;

            Ok(ptr)
        } else {
            // Legacy fallback: grow memory and write at the new region.
            // Only safe for small payloads (e.g. config JSON during init).
            let current_size = self.memory.data_size(&self.store);
            let needed = current_size + data.len();

            if needed > self.limits.max_memory_bytes {
                return Err(WasmError::MemoryLimitExceeded {
                    requested: data.len(),
                    limit: self.limits.max_memory_bytes.saturating_sub(current_size),
                });
            }

            const PAGE_SIZE: usize = 65_536;
            let pages_needed = data.len().div_ceil(PAGE_SIZE);

            self.memory
                .grow(&mut self.store, pages_needed as u64)
                .map_err(|_| WasmError::MemoryLimitExceeded {
                    requested: data.len(),
                    limit: self.limits.max_memory_bytes.saturating_sub(current_size),
                })?;

            let ptr = current_size;
            self.memory
                .write(&mut self.store, ptr, data)
                .map_err(|e| WasmError::Trap(format!("memory write failed: {}", e)))?;

            Ok(ptr as i32)
        }
    }

    /// Call the init function with the given config.
    pub fn init(&mut self, config_json: &[u8]) -> Result<i32, WasmError> {
        let init_func = self
            .init_func
            .clone()
            .ok_or_else(|| WasmError::MissingExport("init".into()))?;

        // Write config to memory
        let ptr = self.write_to_memory(config_json)?;
        let len = config_json.len() as i32;

        // Reset fuel for this call
        if let Err(e) = self.store.set_fuel(self.limits.max_fuel) {
            tracing::warn!(error = %e, "failed to reset WASM fuel");
        }

        // Call init
        let result = init_func
            .call(&mut self.store, (ptr, len))
            .map_err(|e| WasmError::Trap(e.to_string()))?;

        Ok(result)
    }

    /// Call on_request with the given request data.
    pub fn on_request(&mut self, request_json: &[u8]) -> Result<i32, WasmError> {
        let func = self
            .on_request_func
            .clone()
            .ok_or_else(|| WasmError::MissingExport("on_request".into()))?;

        self.call_handler(func, request_json)
    }

    /// Call on_response with the given response data.
    pub fn on_response(&mut self, response_json: &[u8]) -> Result<i32, WasmError> {
        let func = self
            .on_response_func
            .clone()
            .ok_or_else(|| WasmError::MissingExport("on_response".into()))?;

        self.call_handler(func, response_json)
    }

    /// Call dispatch with the given request data.
    pub fn dispatch(&mut self, request_json: &[u8]) -> Result<i32, WasmError> {
        let func = self
            .dispatch_func
            .clone()
            .ok_or_else(|| WasmError::MissingExport("dispatch".into()))?;

        self.call_handler(func, request_json)
    }

    /// Call a handler function with data.
    fn call_handler(
        &mut self,
        func: TypedFunc<(i32, i32), i32>,
        data: &[u8],
    ) -> Result<i32, WasmError> {
        // Clear output buffer
        self.store.data_mut().output_buffer.clear();

        // Reset fuel before write_to_memory — the `alloc` call runs plugin
        // code and needs fuel. Scale fuel with payload size: large bodies
        // require proportionally more instructions for serde + base64.
        let fuel = self.limits.max_fuel.max(data.len() as u64 * 100);
        if let Err(e) = self.store.set_fuel(fuel) {
            tracing::warn!(error = %e, "failed to reset WASM fuel");
        }

        // Write data to memory (may call plugin's `alloc` export)
        let ptr = self.write_to_memory(data)?;
        let len = data.len() as i32;

        // Call function
        let result = func
            .call(&mut self.store, (ptr, len))
            .map_err(|e| WasmError::Trap(e.to_string()))?;

        Ok(result)
    }

    /// Get the output buffer contents.
    pub fn take_output(&mut self) -> Vec<u8> {
        self.store.data_mut().take_output()
    }

    /// Set the request context for the next call.
    pub fn set_context(&mut self, context: RequestContext) {
        self.store.data_mut().set_context(context);
    }

    /// Get the current request context (after modifications by host functions).
    pub fn get_context(&self) -> RequestContext {
        self.store.data().context.clone()
    }

    /// Take the last HTTP result buffer (from `host_http_call` or `host_http_stream`).
    ///
    /// Returns `None` if no HTTP call was made or the result was already taken.
    pub fn take_last_http_result(&mut self) -> Option<Vec<u8>> {
        self.store.data_mut().last_http_result.take()
    }

    /// Inject a stream sender for `host_http_stream` before calling `dispatch`.
    ///
    /// The sender is wrapped in an `Arc` so host functions can clone it for
    /// use inside `std::thread::scope` without lifetime conflicts.
    pub fn set_stream_sender(
        &mut self,
        sender: Arc<tokio::sync::mpsc::UnboundedSender<StreamEvent>>,
    ) {
        self.store.data_mut().set_stream_sender(sender);
    }

    /// Take the upstream WebSocket upgrade request from `host_ws_upgrade` (ADR-0026).
    ///
    /// Returns `None` if no WebSocket upgrade was requested or the request
    /// was already taken.
    pub fn take_ws_upgrade_request(&mut self) -> Option<crate::ws_client::WsUpgradeRequest> {
        self.store.data_mut().take_ws_upgrade_request()
    }

    /// Set the request/response body for the next handler call (side-channel).
    pub fn set_request_body(&mut self, body: Option<Vec<u8>>) {
        self.store.data_mut().set_request_body(body);
    }

    /// Take the output body set by the plugin via host_body_set/host_body_clear.
    pub fn take_output_body(&mut self) -> Option<Option<Vec<u8>>> {
        self.store.data_mut().take_output_body()
    }
}

/// Register a `host_*_read_result` function that copies data from plugin state to WASM memory.
///
/// All read_result host functions follow the same pattern: take a result buffer from state,
/// get the WASM memory export, copy bytes into the provided buffer, return bytes written.
fn add_read_result_fn(
    linker: &mut Linker<PluginState>,
    name: &str,
    extract: impl Fn(&mut PluginState) -> Option<Vec<u8>> + Send + Sync + 'static,
) -> Result<(), WasmError> {
    linker
        .func_wrap(
            "barbacane",
            name,
            move |mut caller: Caller<'_, PluginState>, buf_ptr: i32, buf_len: i32| -> i32 {
                let result = extract(caller.data_mut());
                if let Some(data) = result {
                    let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                        Some(m) => m,
                        None => return 0,
                    };
                    let copy_len = std::cmp::min(data.len(), buf_len as usize);
                    if memory
                        .write(&mut caller, buf_ptr as usize, &data[..copy_len])
                        .is_ok()
                    {
                        return copy_len as i32;
                    }
                }
                0
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add {}: {}", name, e)))?;
    Ok(())
}

/// Add host functions to the linker.
fn add_host_functions(linker: &mut Linker<PluginState>) -> Result<(), WasmError> {
    // host_set_output - always available
    linker
        .func_wrap(
            "barbacane",
            "host_set_output",
            |mut caller: Caller<'_, PluginState>, ptr: i32, len: i32| {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return,
                };

                let start = ptr as usize;
                let end = start + len as usize;
                let data = memory.data(&caller);

                if end <= data.len() {
                    let bytes = data[start..end].to_vec();
                    caller.data_mut().output_buffer = bytes;
                }
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_set_output: {}", e)))?;

    // --- Side-channel body host functions ---
    // Bodies travel as raw bytes instead of base64-in-JSON, eliminating
    // the ~3.65× memory overhead per boundary crossing.

    // host_body_len — returns the length of the held body, or -1 if None.
    linker
        .func_wrap(
            "barbacane",
            "host_body_len",
            |caller: Caller<'_, PluginState>| -> i64 {
                match &caller.data().request_body {
                    Some(body) => body.len() as i64,
                    None => -1,
                }
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_body_len: {}", e)))?;

    // host_body_read — copy the held body into WASM memory at ptr.
    add_read_result_fn(linker, "host_body_read", |state| state.request_body.take())?;

    // host_body_set — set the output body from raw bytes in WASM memory.
    linker
        .func_wrap(
            "barbacane",
            "host_body_set",
            |mut caller: Caller<'_, PluginState>, ptr: i32, len: i32| {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return,
                };

                let start = ptr as usize;
                let end = start + len as usize;
                let data = memory.data(&caller);

                if end <= data.len() {
                    let bytes = data[start..end].to_vec();
                    caller.data_mut().output_body = Some(Some(bytes));
                }
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_body_set: {}", e)))?;

    // host_body_clear — explicitly set the output body to None.
    linker
        .func_wrap(
            "barbacane",
            "host_body_clear",
            |mut caller: Caller<'_, PluginState>| {
                caller.data_mut().output_body = Some(None);
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_body_clear: {}", e)))?;

    // host_http_response_body_len — length of the HTTP response body from host_http_call.
    linker
        .func_wrap(
            "barbacane",
            "host_http_response_body_len",
            |caller: Caller<'_, PluginState>| -> i64 {
                match &caller.data().http_response_body {
                    Some(body) => body.len() as i64,
                    None => -1,
                }
            },
        )
        .map_err(|e| {
            WasmError::Instantiation(format!("failed to add host_http_response_body_len: {}", e))
        })?;

    // host_http_response_body_read — copy HTTP response body into WASM memory.
    add_read_result_fn(linker, "host_http_response_body_read", |state| {
        state.http_response_body.take()
    })?;

    // host_http_request_body_set — set the outbound HTTP request body from WASM memory.
    linker
        .func_wrap(
            "barbacane",
            "host_http_request_body_set",
            |mut caller: Caller<'_, PluginState>, ptr: i32, len: i32| {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return,
                };

                let start = ptr as usize;
                let end = start + len as usize;
                let data = memory.data(&caller);

                if end <= data.len() {
                    let bytes = data[start..end].to_vec();
                    caller.data_mut().http_request_body = Some(bytes);
                }
            },
        )
        .map_err(|e| {
            WasmError::Instantiation(format!("failed to add host_http_request_body_set: {}", e))
        })?;

    // host_log
    linker
        .func_wrap(
            "barbacane",
            "host_log",
            |mut caller: Caller<'_, PluginState>, level: i32, msg_ptr: i32, msg_len: i32| {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return,
                };

                let start = msg_ptr as usize;
                let end = start + msg_len as usize;
                let data = memory.data(&caller);

                if end <= data.len() {
                    if let Ok(message) = std::str::from_utf8(&data[start..end]) {
                        let plugin_name = caller.data().plugin_name.clone();
                        match level {
                            0 => tracing::error!(plugin = %plugin_name, "{}", message),
                            1 => tracing::warn!(plugin = %plugin_name, "{}", message),
                            2 => tracing::info!(plugin = %plugin_name, "{}", message),
                            _ => tracing::debug!(plugin = %plugin_name, "{}", message),
                        }
                    }
                }
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_log: {}", e)))?;

    // host_context_get
    linker
        .func_wrap(
            "barbacane",
            "host_context_get",
            |mut caller: Caller<'_, PluginState>, key_ptr: i32, key_len: i32| -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return -1,
                };

                let start = key_ptr as usize;
                let end = start + key_len as usize;
                let data = memory.data(&caller);

                if end > data.len() {
                    return -1;
                }

                let key = match std::str::from_utf8(&data[start..end]) {
                    Ok(k) => k.to_string(),
                    Err(_) => return -1,
                };

                match caller.data().context.values.get(&key).cloned() {
                    Some(value) => {
                        let len = value.len() as i32;
                        caller.data_mut().context.last_get_result = Some(value);
                        len
                    }
                    None => -1,
                }
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_context_get: {}", e)))?;

    // host_context_read_result
    add_read_result_fn(linker, "host_context_read_result", |state| {
        state.context.last_get_result.take().map(String::into_bytes)
    })?;

    // host_context_set
    linker
        .func_wrap(
            "barbacane",
            "host_context_set",
            |mut caller: Caller<'_, PluginState>,
             key_ptr: i32,
             key_len: i32,
             val_ptr: i32,
             val_len: i32| {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return,
                };

                let key_start = key_ptr as usize;
                let key_end = key_start + key_len as usize;
                let val_start = val_ptr as usize;
                let val_end = val_start + val_len as usize;

                // Read data first, then mutate
                let data = memory.data(&caller);
                if key_end <= data.len() && val_end <= data.len() {
                    let key_result =
                        std::str::from_utf8(&data[key_start..key_end]).map(String::from);
                    let val_result =
                        std::str::from_utf8(&data[val_start..val_end]).map(String::from);

                    if let (Ok(key), Ok(value)) = (key_result, val_result) {
                        caller.data_mut().context.values.insert(key, value);
                    }
                }
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_context_set: {}", e)))?;

    // host_clock_now
    linker
        .func_wrap(
            "barbacane",
            "host_clock_now",
            |_caller: Caller<'_, PluginState>| -> i64 {
                use std::time::Instant;

                static START: std::sync::OnceLock<Instant> = std::sync::OnceLock::new();
                let start = START.get_or_init(Instant::now);

                start.elapsed().as_millis() as i64
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_clock_now: {}", e)))?;

    // host_time_now - alias for host_clock_now (deprecated, use host_clock_now)
    linker
        .func_wrap(
            "barbacane",
            "host_time_now",
            |_caller: Caller<'_, PluginState>| -> i64 {
                use std::time::Instant;

                static START: std::sync::OnceLock<Instant> = std::sync::OnceLock::new();
                let start = START.get_or_init(Instant::now);

                start.elapsed().as_millis() as i64
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_time_now: {}", e)))?;

    // host_get_unix_timestamp - returns current Unix timestamp in seconds
    linker
        .func_wrap(
            "barbacane",
            "host_get_unix_timestamp",
            |_caller: Caller<'_, PluginState>| -> u64 {
                use std::time::{SystemTime, UNIX_EPOCH};

                SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .map(|d| d.as_secs())
                    .unwrap_or(0)
            },
        )
        .map_err(|e| {
            WasmError::Instantiation(format!("failed to add host_get_unix_timestamp: {}", e))
        })?;

    // host_uuid_generate - generates UUID v7 and returns length
    linker
        .func_wrap(
            "barbacane",
            "host_uuid_generate",
            |mut caller: Caller<'_, PluginState>| -> i32 {
                let uuid = uuid::Uuid::now_v7().to_string();
                let len = uuid.len() as i32;
                caller.data_mut().last_uuid_result = Some(uuid.into_bytes());
                len
            },
        )
        .map_err(|e| {
            WasmError::Instantiation(format!("failed to add host_uuid_generate: {}", e))
        })?;

    // host_uuid_read_result - copies generated UUID to WASM memory
    add_read_result_fn(linker, "host_uuid_read_result", |state| {
        state.last_uuid_result.take()
    })?;

    // host_http_call - make outbound HTTP request
    linker
        .func_wrap(
            "barbacane",
            "host_http_call",
            |mut caller: Caller<'_, PluginState>, req_ptr: i32, req_len: i32| -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return -1,
                };

                let start = req_ptr as usize;
                let end = start + req_len as usize;
                let data = memory.data(&caller);

                if end > data.len() {
                    return -1;
                }

                // Parse the request JSON from plugin format
                let plugin_request: PluginHttpRequest =
                    match serde_json::from_slice(&data[start..end]) {
                        Ok(r) => r,
                        Err(e) => {
                            tracing::error!("failed to parse HTTP request: {}", e);
                            return -1;
                        }
                    };

                // Body priority: side-channel (host_http_request_body_set) > JSON body.
                // Side-channel avoids base64 overhead for large payloads.
                let body = caller
                    .data_mut()
                    .http_request_body
                    .take()
                    .or(plugin_request.body);

                let request = HttpClientRequest {
                    method: plugin_request.method,
                    url: plugin_request.url,
                    headers: plugin_request.headers.into_iter().collect(),
                    body,
                    timeout: plugin_request
                        .timeout_ms
                        .map(std::time::Duration::from_millis),
                };

                // Get the HTTP client
                let http_client = match caller.data().http_client.clone() {
                    Some(c) => c,
                    None => {
                        tracing::error!("HTTP client not available");
                        return -1;
                    }
                };

                // Use a separate runtime to avoid deadlock with the main runtime.
                // The main runtime is blocked waiting for the WASM call to complete,
                // so we can't schedule work on it. Create a new runtime just for this call.
                // TODO: Optimize by using a thread-local runtime or worker pool instead of
                // creating a new runtime per call (performance improvement for high throughput).
                let response_result = std::thread::scope(|s| {
                    let handle = s.spawn(|| {
                        let rt = match tokio::runtime::Builder::new_current_thread()
                            .enable_all()
                            .build()
                        {
                            Ok(rt) => rt,
                            Err(e) => {
                                tracing::error!("failed to create runtime: {}", e);
                                return None;
                            }
                        };

                        rt.block_on(async {
                            match http_client.call(request).await {
                                Ok(mut response) => {
                                    // Strip body into side-channel to avoid base64
                                    // encoding in the JSON metadata.
                                    let body = response.body.take();
                                    let json = serde_json::to_vec(&response).ok();
                                    Some((json, body))
                                }
                                Err(e) => {
                                    tracing::error!("HTTP call failed: {}", e);
                                    // Return error response
                                    let error_response = match e {
                                        crate::http_client::HttpClientError::Timeout => {
                                            HttpClientResponse::error(
                                                504,
                                                "urn:barbacane:error:upstream-timeout",
                                                "Gateway Timeout",
                                                "Upstream request timed out",
                                            )
                                        }
                                        crate::http_client::HttpClientError::CircuitOpen(host) => {
                                            HttpClientResponse::error(
                                                503,
                                                "urn:barbacane:error:circuit-open",
                                                "Service Unavailable",
                                                &format!("Circuit breaker open for {}", host),
                                            )
                                        }
                                        crate::http_client::HttpClientError::ConnectionFailed(
                                            _,
                                        ) => HttpClientResponse::error(
                                            502,
                                            "urn:barbacane:error:upstream-unavailable",
                                            "Bad Gateway",
                                            "Failed to connect to upstream",
                                        ),
                                        _ => HttpClientResponse::error(
                                            502,
                                            "urn:barbacane:error:upstream-unavailable",
                                            "Bad Gateway",
                                            &e.to_string(),
                                        ),
                                    };
                                    let json = serde_json::to_vec(&error_response).ok();
                                    Some((json, None))
                                }
                            }
                        })
                    });

                    match handle.join() {
                        Ok(result) => result,
                        Err(e) => {
                            tracing::error!("worker thread panicked: {:?}", e);
                            None
                        }
                    }
                });

                match response_result {
                    Some((Some(json), body)) => {
                        let len = json.len() as i32;
                        caller.data_mut().last_http_result = Some(json);
                        caller.data_mut().http_response_body = body;
                        len
                    }
                    _ => -1,
                }
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_http_call: {}", e)))?;

    // host_http_read_result - read HTTP response (works for both host_http_call and host_http_stream)
    add_read_result_fn(linker, "host_http_read_result", |state| {
        state.last_http_result.take()
    })?;

    // host_http_stream - streaming HTTP request (ADR-0023)
    //
    // Same request format as host_http_call. The host immediately begins
    // forwarding response chunks to the client via the stream_sender channel
    // while buffering the complete body in last_http_result.
    // Returns the length of the buffered response, or -1 on error.
    linker
        .func_wrap(
            "barbacane",
            "host_http_stream",
            |mut caller: Caller<'_, PluginState>, req_ptr: i32, req_len: i32| -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return -1,
                };

                let start = req_ptr as usize;
                let end = start + req_len as usize;
                let data = memory.data(&caller);

                if end > data.len() {
                    return -1;
                }

                let plugin_request: PluginHttpRequest =
                    match serde_json::from_slice(&data[start..end]) {
                        Ok(r) => r,
                        Err(e) => {
                            tracing::error!("host_http_stream: failed to parse request: {}", e);
                            return -1;
                        }
                    };

                // Body priority: side-channel (host_http_request_body_set) > JSON body.
                let body = caller
                    .data_mut()
                    .http_request_body
                    .take()
                    .or(plugin_request.body);

                let request = HttpClientRequest {
                    method: plugin_request.method,
                    url: plugin_request.url,
                    headers: plugin_request.headers.into_iter().collect(),
                    body,
                    timeout: plugin_request
                        .timeout_ms
                        .map(std::time::Duration::from_millis),
                };

                let http_client = match caller.data().http_client.clone() {
                    Some(c) => c,
                    None => {
                        tracing::error!("host_http_stream: HTTP client not available");
                        return -1;
                    }
                };

                // Clone the stream sender (Arc makes this cheap).
                let stream_sender = caller.data().stream_sender.clone();

                let response_result = std::thread::scope(|s| {
                    let handle = s.spawn(|| {
                        let rt = match tokio::runtime::Builder::new_current_thread()
                            .enable_all()
                            .build()
                        {
                            Ok(rt) => rt,
                            Err(e) => {
                                tracing::error!(
                                    "host_http_stream: failed to create runtime: {}",
                                    e
                                );
                                return None;
                            }
                        };

                        rt.block_on(async {
                            use futures_util::StreamExt;

                            match http_client.stream_raw(request).await {
                                Ok(upstream) => {
                                    let status = upstream.status().as_u16();
                                    let upstream_headers: BTreeMap<String, String> = upstream
                                        .headers()
                                        .iter()
                                        .filter_map(|(k, v)| {
                                            v.to_str()
                                                .ok()
                                                .map(|v| (k.as_str().to_lowercase(), v.to_string()))
                                        })
                                        .collect();

                                    // Send headers through the streaming channel.
                                    if let Some(tx) = &stream_sender {
                                        let _ = tx.send(StreamEvent::Headers {
                                            status,
                                            headers: upstream_headers.clone(),
                                        });
                                    }

                                    // Stream body chunks, sending each through the channel
                                    // while building the complete buffer for last_http_result.
                                    let mut buffer: Vec<u8> = Vec::new();
                                    let mut byte_stream = upstream.bytes_stream();

                                    while let Some(chunk_result) = byte_stream.next().await {
                                        match chunk_result {
                                            Ok(chunk) => {
                                                if let Some(tx) = &stream_sender {
                                                    let _ =
                                                        tx.send(StreamEvent::Chunk(chunk.clone()));
                                                }
                                                buffer.extend_from_slice(&chunk);
                                            }
                                            Err(e) => {
                                                tracing::error!(
                                                    "host_http_stream: upstream read error: {}",
                                                    e
                                                );
                                                return None;
                                            }
                                        }
                                    }

                                    // Strip body into side-channel, serialize
                                    // metadata-only JSON for host_http_read_result.
                                    let complete = HttpClientResponse {
                                        status,
                                        headers: upstream_headers.into_iter().collect(),
                                        body: None,
                                    };
                                    let json = serde_json::to_vec(&complete).ok();
                                    Some((json, Some(buffer)))
                                }
                                Err(e) => {
                                    tracing::error!("host_http_stream: request failed: {}", e);
                                    None
                                }
                            }
                        })
                    });

                    match handle.join() {
                        Ok(result) => result,
                        Err(e) => {
                            tracing::error!("host_http_stream: worker thread panicked: {:?}", e);
                            None
                        }
                    }
                });

                match response_result {
                    Some((Some(json), body)) => {
                        let len = json.len() as i32;
                        caller.data_mut().last_http_result = Some(json);
                        caller.data_mut().http_response_body = body;
                        len
                    }
                    _ => -1,
                }
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_http_stream: {}", e)))?;

    // host_ws_upgrade - request an upstream WebSocket connection (ADR-0026)
    //
    // The plugin sends a JSON payload: { url, connect_timeout_ms, headers }.
    // The request is validated and stored in PluginState. The actual TCP
    // connection is deferred to the async relay task on the main tokio runtime,
    // because a TcpStream must be created on the runtime that will drive it
    // (a temporary runtime's I/O driver dies when the runtime drops).
    // Returns 0 on success (valid request), -1 on parse failure.
    linker
        .func_wrap(
            "barbacane",
            "host_ws_upgrade",
            |mut caller: Caller<'_, PluginState>, req_ptr: i32, req_len: i32| -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return -1,
                };

                let start = req_ptr as usize;
                let end = start + req_len as usize;
                let data = memory.data(&caller);

                if end > data.len() {
                    return -1;
                }

                let ws_request: crate::ws_client::WsUpgradeRequest =
                    match serde_json::from_slice(&data[start..end]) {
                        Ok(r) => r,
                        Err(e) => {
                            tracing::error!("host_ws_upgrade: failed to parse request: {}", e);
                            let err_msg = format!("invalid upgrade request: {}", e);
                            caller.data_mut().last_http_result = Some(err_msg.into_bytes());
                            return -1;
                        }
                    };

                let plugin_name = caller.data().plugin_name.clone();
                tracing::debug!(
                    plugin = %plugin_name,
                    url = %ws_request.url,
                    "host_ws_upgrade: storing request for deferred connection"
                );

                // Store the request; the actual connection happens on the main
                // runtime inside the relay task (see relay_websocket).
                caller.data_mut().ws_upgrade_request = Some(ws_request);
                0
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_ws_upgrade: {}", e)))?;

    // host_verify_signature - verify a cryptographic signature using a JWK
    linker
        .func_wrap(
            "barbacane",
            "host_verify_signature",
            |mut caller: Caller<'_, PluginState>, req_ptr: i32, req_len: i32| -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return -1,
                };

                let start = req_ptr as usize;
                let end = start + req_len as usize;
                let data = memory.data(&caller);

                if end > data.len() {
                    return -1;
                }

                // Parse the verification request
                let request: crate::crypto::VerifySignatureRequest =
                    match serde_json::from_slice(&data[start..end]) {
                        Ok(r) => r,
                        Err(e) => {
                            tracing::error!(
                                plugin = %caller.data_mut().plugin_name,
                                "failed to parse verify_signature request: {}", e
                            );
                            return -1;
                        }
                    };

                // Perform verification
                match crate::crypto::verify_signature(&request) {
                    Ok(true) => 1,
                    Ok(false) => 0,
                    Err(e) => {
                        tracing::error!(
                            plugin = %caller.data_mut().plugin_name,
                            "signature verification error: {}", e
                        );
                        -1
                    }
                }
            },
        )
        .map_err(|e| {
            WasmError::Instantiation(format!("failed to add host_verify_signature: {}", e))
        })?;

    // host_get_secret - get a secret by reference
    linker
        .func_wrap(
            "barbacane",
            "host_get_secret",
            |mut caller: Caller<'_, PluginState>, ref_ptr: i32, ref_len: i32| -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return -1,
                };

                let start = ref_ptr as usize;
                let end = start + ref_len as usize;
                let data = memory.data(&caller);

                if end > data.len() {
                    return -1;
                }

                // Read the secret reference from plugin memory
                let secret_ref = match std::str::from_utf8(&data[start..end]) {
                    Ok(r) => r.to_string(),
                    Err(_) => return -1,
                };

                // Look up in secrets store
                match caller.data().secrets.get(&secret_ref) {
                    Some(value) => {
                        let bytes = value.as_bytes().to_vec();
                        let len = bytes.len() as i32;
                        caller.data_mut().last_secret_result = Some(bytes);
                        len
                    }
                    None => {
                        tracing::warn!(
                            plugin = %caller.data().plugin_name,
                            reference = %secret_ref,
                            "secret not found in store"
                        );
                        -1
                    }
                }
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_get_secret: {}", e)))?;

    // host_secret_read_result - read secret value into plugin memory
    add_read_result_fn(linker, "host_secret_read_result", |state| {
        state.last_secret_result.take()
    })?;

    // host_rate_limit_check - check rate limit for a key
    linker
        .func_wrap(
            "barbacane",
            "host_rate_limit_check",
            |mut caller: Caller<'_, PluginState>,
             key_ptr: i32,
             key_len: i32,
             quota: u32,
             window_secs: u32|
             -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return -1,
                };

                let start = key_ptr as usize;
                let end = start + key_len as usize;
                let data = memory.data(&caller);

                if end > data.len() {
                    return -1;
                }

                // Read the partition key from plugin memory
                let key = match std::str::from_utf8(&data[start..end]) {
                    Ok(k) => k.to_string(),
                    Err(_) => return -1,
                };

                // Get the rate limiter
                let rate_limiter = match &caller.data().rate_limiter {
                    Some(rl) => rl.clone(),
                    None => {
                        tracing::error!("rate limiter not available");
                        return -1;
                    }
                };

                // Check the rate limit
                let result = rate_limiter.check(&key, quota, window_secs as u64);

                // Serialize the result
                match serde_json::to_vec(&result) {
                    Ok(json) => {
                        let len = json.len() as i32;
                        caller.data_mut().last_rate_limit_result = Some(json);
                        len
                    }
                    Err(e) => {
                        tracing::error!("failed to serialize rate limit result: {}", e);
                        -1
                    }
                }
            },
        )
        .map_err(|e| {
            WasmError::Instantiation(format!("failed to add host_rate_limit_check: {}", e))
        })?;

    // host_rate_limit_read_result - read rate limit result into plugin memory
    add_read_result_fn(linker, "host_rate_limit_read_result", |state| {
        state.last_rate_limit_result.take()
    })?;

    // host_cache_get - look up a cached response
    linker
        .func_wrap(
            "barbacane",
            "host_cache_get",
            |mut caller: Caller<'_, PluginState>, key_ptr: i32, key_len: i32| -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return -1,
                };

                let start = key_ptr as usize;
                let end = start + key_len as usize;
                let data = memory.data(&caller);

                if end > data.len() {
                    return -1;
                }

                // Read the cache key from plugin memory
                let key = match std::str::from_utf8(&data[start..end]) {
                    Ok(k) => k.to_string(),
                    Err(_) => return -1,
                };

                // Get the response cache
                let cache = match &caller.data().response_cache {
                    Some(c) => c.clone(),
                    None => {
                        tracing::error!("response cache not available");
                        return -1;
                    }
                };

                // Check the cache
                let result = cache.get(&key);

                // Serialize the result
                match serde_json::to_vec(&result) {
                    Ok(json) => {
                        let len = json.len() as i32;
                        caller.data_mut().last_cache_result = Some(json);
                        len
                    }
                    Err(e) => {
                        tracing::error!("failed to serialize cache result: {}", e);
                        -1
                    }
                }
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_cache_get: {}", e)))?;

    // host_cache_set - store a response in the cache
    linker
        .func_wrap(
            "barbacane",
            "host_cache_set",
            |mut caller: Caller<'_, PluginState>,
             key_ptr: i32,
             key_len: i32,
             entry_ptr: i32,
             entry_len: i32,
             ttl_secs: u32|
             -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return -1,
                };

                let key_start = key_ptr as usize;
                let key_end = key_start + key_len as usize;
                let entry_start = entry_ptr as usize;
                let entry_end = entry_start + entry_len as usize;
                let data = memory.data(&caller);

                if key_end > data.len() || entry_end > data.len() {
                    return -1;
                }

                // Read the cache key
                let key = match std::str::from_utf8(&data[key_start..key_end]) {
                    Ok(k) => k.to_string(),
                    Err(_) => return -1,
                };

                // Parse the cache entry JSON
                let entry: crate::cache::CacheEntry =
                    match serde_json::from_slice(&data[entry_start..entry_end]) {
                        Ok(e) => e,
                        Err(e) => {
                            tracing::error!("failed to parse cache entry: {}", e);
                            return -1;
                        }
                    };

                // Get the response cache
                let cache = match &caller.data().response_cache {
                    Some(c) => c.clone(),
                    None => {
                        tracing::error!("response cache not available");
                        return -1;
                    }
                };

                // Store in cache
                cache.set(&key, entry, ttl_secs as u64);
                0 // Success
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_cache_set: {}", e)))?;

    // host_cache_read_result - read cache lookup result into plugin memory
    add_read_result_fn(linker, "host_cache_read_result", |state| {
        state.last_cache_result.take()
    })?;

    // === Telemetry Host Functions ===

    // host_metric_counter_inc - increment a plugin counter metric
    linker
        .func_wrap(
            "barbacane",
            "host_metric_counter_inc",
            |mut caller: Caller<'_, PluginState>,
             name_ptr: i32,
             name_len: i32,
             labels_ptr: i32,
             labels_len: i32,
             value: f64| {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return,
                };

                let data = memory.data(&caller);
                let name_start = name_ptr as usize;
                let name_end = name_start + name_len as usize;
                let labels_start = labels_ptr as usize;
                let labels_end = labels_start + labels_len as usize;

                if name_end > data.len() || labels_end > data.len() {
                    return;
                }

                let name = match std::str::from_utf8(&data[name_start..name_end]) {
                    Ok(n) => n.to_string(),
                    Err(_) => return,
                };

                let labels_json = match std::str::from_utf8(&data[labels_start..labels_end]) {
                    Ok(l) => l.to_string(),
                    Err(_) => return,
                };

                let plugin_name = caller.data().plugin_name.clone();
                if let Some(metrics) = &caller.data().metrics {
                    metrics.plugin_counter_inc(&plugin_name, &name, &labels_json, value as u64);
                }
            },
        )
        .map_err(|e| {
            WasmError::Instantiation(format!("failed to add host_metric_counter_inc: {}", e))
        })?;

    // host_metric_histogram_observe - observe a plugin histogram metric
    linker
        .func_wrap(
            "barbacane",
            "host_metric_histogram_observe",
            |mut caller: Caller<'_, PluginState>,
             name_ptr: i32,
             name_len: i32,
             labels_ptr: i32,
             labels_len: i32,
             value: f64| {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return,
                };

                let data = memory.data(&caller);
                let name_start = name_ptr as usize;
                let name_end = name_start + name_len as usize;
                let labels_start = labels_ptr as usize;
                let labels_end = labels_start + labels_len as usize;

                if name_end > data.len() || labels_end > data.len() {
                    return;
                }

                let name = match std::str::from_utf8(&data[name_start..name_end]) {
                    Ok(n) => n.to_string(),
                    Err(_) => return,
                };

                let labels_json = match std::str::from_utf8(&data[labels_start..labels_end]) {
                    Ok(l) => l.to_string(),
                    Err(_) => return,
                };

                let plugin_name = caller.data().plugin_name.clone();
                if let Some(metrics) = &caller.data().metrics {
                    metrics.plugin_histogram_observe(&plugin_name, &name, &labels_json, value);
                }
            },
        )
        .map_err(|e| {
            WasmError::Instantiation(format!(
                "failed to add host_metric_histogram_observe: {}",
                e
            ))
        })?;

    // host_span_start - start a child span (stub - returns span ID)
    // Full implementation requires passing span context through RequestContext
    linker
        .func_wrap(
            "barbacane",
            "host_span_start",
            |mut caller: Caller<'_, PluginState>, name_ptr: i32, name_len: i32| -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return -1,
                };

                let data = memory.data(&caller);
                let start = name_ptr as usize;
                let end = start + name_len as usize;

                if end > data.len() {
                    return -1;
                }

                let span_name = match std::str::from_utf8(&data[start..end]) {
                    Ok(n) => n,
                    Err(_) => return -1,
                };

                // Log the span start for now (full tracing integration in Phase 9)
                let plugin_name = &caller.data().plugin_name;
                tracing::debug!(plugin = %plugin_name, span = %span_name, "plugin span started");

                // Return a placeholder span ID
                1
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_span_start: {}", e)))?;

    // host_span_end - end the current span
    linker
        .func_wrap(
            "barbacane",
            "host_span_end",
            |caller: Caller<'_, PluginState>| {
                let plugin_name = &caller.data().plugin_name;
                tracing::debug!(plugin = %plugin_name, "plugin span ended");
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_span_end: {}", e)))?;

    // host_span_set_attribute - set an attribute on the current span
    linker
        .func_wrap(
            "barbacane",
            "host_span_set_attribute",
            |mut caller: Caller<'_, PluginState>,
             key_ptr: i32,
             key_len: i32,
             val_ptr: i32,
             val_len: i32| {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return,
                };

                let data = memory.data(&caller);
                let key_start = key_ptr as usize;
                let key_end = key_start + key_len as usize;
                let val_start = val_ptr as usize;
                let val_end = val_start + val_len as usize;

                if key_end > data.len() || val_end > data.len() {
                    return;
                }

                let key = match std::str::from_utf8(&data[key_start..key_end]) {
                    Ok(k) => k,
                    Err(_) => return,
                };

                let value = match std::str::from_utf8(&data[val_start..val_end]) {
                    Ok(v) => v,
                    Err(_) => return,
                };

                let plugin_name = &caller.data().plugin_name;
                tracing::debug!(plugin = %plugin_name, %key, %value, "plugin span attribute set");
            },
        )
        .map_err(|e| {
            WasmError::Instantiation(format!("failed to add host_span_set_attribute: {}", e))
        })?;

    // === Broker Host Functions (M10) ===

    // host_kafka_publish - publish a message to Kafka
    linker
        .func_wrap(
            "barbacane",
            "host_kafka_publish",
            |mut caller: Caller<'_, PluginState>, msg_ptr: i32, msg_len: i32| -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return -1,
                };

                let start = msg_ptr as usize;
                let end = start + msg_len as usize;
                let data = memory.data(&caller);

                if end > data.len() {
                    return -1;
                }

                // Parse the broker message from plugin memory
                let message: BrokerMessage = match serde_json::from_slice(&data[start..end]) {
                    Ok(m) => m,
                    Err(e) => {
                        tracing::error!("failed to parse broker message: {}", e);
                        return -1;
                    }
                };

                // Extract URL (broker addresses) from the message
                let brokers = match &message.url {
                    Some(u) => u.clone(),
                    None => {
                        tracing::error!("Kafka publish: missing url in broker message");
                        return -1;
                    }
                };

                // Get the Kafka publisher
                let publisher = match caller.data().kafka_publisher.clone() {
                    Some(p) => p,
                    None => {
                        tracing::error!("Kafka publisher not available");
                        return -1;
                    }
                };

                let topic = message.topic.clone();
                let key = message.key.clone();
                let payload = message.payload.clone();
                let headers = message.headers.clone();

                // Use thread::scope to escape the main tokio runtime context,
                // then call publish_blocking which uses the publisher's own runtime.
                let result = std::thread::scope(|s| {
                    let handle = s.spawn(|| {
                        publisher.publish_blocking(&brokers, &topic, key, &payload, headers)
                    });

                    match handle.join() {
                        Ok(result) => Some(result),
                        Err(e) => {
                            tracing::error!("Kafka publish thread panicked: {:?}", e);
                            None
                        }
                    }
                });

                // Serialize the result
                let result_json = match result {
                    Some(Ok(r)) => serde_json::to_vec(&r),
                    Some(Err(e)) => {
                        let error_result =
                            crate::broker::PublishResult::failure(message.topic, e.to_string());
                        serde_json::to_vec(&error_result)
                    }
                    None => {
                        let error_result = crate::broker::PublishResult::failure(
                            message.topic,
                            "Kafka publish failed".to_string(),
                        );
                        serde_json::to_vec(&error_result)
                    }
                };

                match result_json {
                    Ok(json) => {
                        let len = json.len() as i32;
                        caller.data_mut().last_broker_result = Some(json);
                        len
                    }
                    Err(e) => {
                        tracing::error!("failed to serialize broker result: {}", e);
                        -1
                    }
                }
            },
        )
        .map_err(|e| {
            WasmError::Instantiation(format!("failed to add host_kafka_publish: {}", e))
        })?;

    // host_nats_publish - publish a message to NATS
    linker
        .func_wrap(
            "barbacane",
            "host_nats_publish",
            |mut caller: Caller<'_, PluginState>, msg_ptr: i32, msg_len: i32| -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return -1,
                };

                let start = msg_ptr as usize;
                let end = start + msg_len as usize;
                let data = memory.data(&caller);

                if end > data.len() {
                    return -1;
                }

                // Parse the broker message from plugin memory
                let message: BrokerMessage = match serde_json::from_slice(&data[start..end]) {
                    Ok(m) => m,
                    Err(e) => {
                        tracing::error!("failed to parse broker message: {}", e);
                        return -1;
                    }
                };

                // Extract URL from the message
                let url = match &message.url {
                    Some(u) => u.clone(),
                    None => {
                        tracing::error!("NATS publish: missing url in broker message");
                        return -1;
                    }
                };

                // Get the NATS publisher
                let publisher = match caller.data().nats_publisher.clone() {
                    Some(p) => p,
                    None => {
                        tracing::error!("NATS publisher not available");
                        return -1;
                    }
                };

                let subject = message.topic.clone();
                let payload = bytes::Bytes::from(message.payload.clone());
                let headers = message.headers.clone();

                // Use thread::scope to escape the main tokio runtime context,
                // then call publish_blocking which uses the publisher's own runtime.
                let result = std::thread::scope(|s| {
                    let handle =
                        s.spawn(|| publisher.publish_blocking(&url, &subject, payload, headers));

                    match handle.join() {
                        Ok(result) => Some(result),
                        Err(e) => {
                            tracing::error!("NATS publish thread panicked: {:?}", e);
                            None
                        }
                    }
                });

                // Serialize the result
                let result_json = match result {
                    Some(Ok(r)) => serde_json::to_vec(&r),
                    Some(Err(e)) => {
                        let error_result =
                            crate::broker::PublishResult::failure(message.topic, e.to_string());
                        serde_json::to_vec(&error_result)
                    }
                    None => {
                        let error_result = crate::broker::PublishResult::failure(
                            message.topic,
                            "NATS publish failed".to_string(),
                        );
                        serde_json::to_vec(&error_result)
                    }
                };

                match result_json {
                    Ok(json) => {
                        let len = json.len() as i32;
                        caller.data_mut().last_broker_result = Some(json);
                        len
                    }
                    Err(e) => {
                        tracing::error!("failed to serialize broker result: {}", e);
                        -1
                    }
                }
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("failed to add host_nats_publish: {}", e)))?;

    // host_broker_read_result - read broker publish result into plugin memory
    add_read_result_fn(linker, "host_broker_read_result", |state| {
        state.last_broker_result.take()
    })?;

    // ── Minimal WASI stubs ──────────────────────────────────────────────
    // Some plugins (e.g. cel) compile with wasm32-wasip1 and import WASI
    // functions even though Barbacane provides its own host ABI. We add
    // lightweight stubs so the linker can resolve these imports.
    // Full WASI support (wasmtime-wasi) can replace these if needed.

    // random_get — deterministic bytes for HashMap seed initialisation.
    // Acceptable in a sandboxed single-request context (no HashDoS risk).
    linker
        .func_wrap(
            "wasi_snapshot_preview1",
            "random_get",
            |mut caller: Caller<'_, PluginState>, buf_ptr: i32, buf_len: i32| -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return 1,
                };
                let start = buf_ptr as usize;
                let end = start + buf_len as usize;
                let data = memory.data_mut(&mut caller);
                if end > data.len() {
                    return 1;
                }
                for (i, byte) in data[start..end].iter_mut().enumerate() {
                    *byte = (i.wrapping_mul(0x9E3779B9) >> 24) as u8;
                }
                0
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("wasi stub: {e}")))?;

    // sched_yield — no-op, always succeeds.
    linker
        .func_wrap("wasi_snapshot_preview1", "sched_yield", || -> i32 { 0 })
        .map_err(|e| WasmError::Instantiation(format!("wasi stub: {e}")))?;

    // clock_time_get — returns current time in nanoseconds.
    linker
        .func_wrap(
            "wasi_snapshot_preview1",
            "clock_time_get",
            |mut caller: Caller<'_, PluginState>,
             _clock_id: i32,
             _precision: i64,
             time_ptr: i32|
             -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return 1,
                };
                let nanos = std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map(|d| d.as_nanos() as u64)
                    .unwrap_or(0);
                let ptr = time_ptr as usize;
                let data = memory.data_mut(&mut caller);
                if ptr + 8 > data.len() {
                    return 1;
                }
                data[ptr..ptr + 8].copy_from_slice(&nanos.to_le_bytes());
                0
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("wasi stub: {e}")))?;

    // fd_write — silently discards output (plugins use host_log instead).
    linker
        .func_wrap(
            "wasi_snapshot_preview1",
            "fd_write",
            |mut caller: Caller<'_, PluginState>,
             _fd: i32,
             iovs_ptr: i32,
             iovs_len: i32,
             nwritten_ptr: i32|
             -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return 1,
                };
                let data = memory.data_mut(&mut caller);
                let mut total: u32 = 0;
                for i in 0..iovs_len as usize {
                    let base = (iovs_ptr as usize) + i * 8;
                    if base + 8 > data.len() {
                        return 1;
                    }
                    let len =
                        u32::from_le_bytes(data[base + 4..base + 8].try_into().unwrap_or([0; 4]));
                    total = total.saturating_add(len);
                }
                let ptr = nwritten_ptr as usize;
                if ptr + 4 > data.len() {
                    return 1;
                }
                data[ptr..ptr + 4].copy_from_slice(&total.to_le_bytes());
                0
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("wasi stub: {e}")))?;

    // environ_get — no environment variables exposed.
    linker
        .func_wrap(
            "wasi_snapshot_preview1",
            "environ_get",
            |_caller: Caller<'_, PluginState>, _environ: i32, _environ_buf: i32| -> i32 { 0 },
        )
        .map_err(|e| WasmError::Instantiation(format!("wasi stub: {e}")))?;

    // environ_sizes_get — reports 0 vars, 0 bytes.
    linker
        .func_wrap(
            "wasi_snapshot_preview1",
            "environ_sizes_get",
            |mut caller: Caller<'_, PluginState>, num_ptr: i32, buf_size_ptr: i32| -> i32 {
                let memory = match caller.get_export("memory").and_then(|e| e.into_memory()) {
                    Some(m) => m,
                    None => return 1,
                };
                let data = memory.data_mut(&mut caller);
                let np = num_ptr as usize;
                let bp = buf_size_ptr as usize;
                if np + 4 > data.len() || bp + 4 > data.len() {
                    return 1;
                }
                data[np..np + 4].copy_from_slice(&0u32.to_le_bytes());
                data[bp..bp + 4].copy_from_slice(&0u32.to_le_bytes());
                0
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("wasi stub: {e}")))?;

    // proc_exit — traps the module (should never be reached).
    linker
        .func_wrap(
            "wasi_snapshot_preview1",
            "proc_exit",
            |_caller: Caller<'_, PluginState>, _code: i32| {
                // Intentional trap — WASM execution stops here
            },
        )
        .map_err(|e| WasmError::Instantiation(format!("wasi stub: {e}")))?;

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn request_context_new() {
        let ctx = RequestContext::new("trace-123".into(), "req-456".into());
        assert_eq!(ctx.trace_id, "trace-123");
        assert_eq!(ctx.request_id, "req-456");
        assert!(ctx.values.is_empty());
    }

    #[test]
    fn plugin_state_take_output() {
        let limits = PluginLimits::default();
        let mut state = PluginState::new("test".into(), &limits);
        state.output_buffer = vec![1, 2, 3];

        let output = state.take_output();
        assert_eq!(output, vec![1, 2, 3]);
        assert!(state.output_buffer.is_empty());
    }

    #[test]
    fn plugin_state_uuid_result_initialized() {
        let limits = PluginLimits::default();
        let state = PluginState::new("test".into(), &limits);
        assert!(state.last_uuid_result.is_none());
    }

    #[test]
    fn uuid_v7_format() {
        // Test that UUID v7 generates valid format
        let uuid = uuid::Uuid::now_v7().to_string();
        assert_eq!(uuid.len(), 36); // UUID string format: 8-4-4-4-12
        assert!(uuid.chars().nth(14) == Some('7')); // Version 7 marker
    }

    #[test]
    fn plugin_state_nats_publisher_default() {
        let limits = PluginLimits::default();
        let state = PluginState::new("test".into(), &limits);
        assert!(state.nats_publisher.is_none());
    }

    #[test]
    fn plugin_state_broker_result_default() {
        let limits = PluginLimits::default();
        let state = PluginState::new("test".into(), &limits);
        assert!(state.last_broker_result.is_none());
    }

    #[test]
    fn plugin_state_kafka_publisher_default() {
        let limits = PluginLimits::default();
        let state = PluginState::new("test".into(), &limits);
        assert!(state.kafka_publisher.is_none());
    }

    // ── streaming (ADR-0023) ──────────────────────────────────────────────────

    #[test]
    fn plugin_state_stream_sender_default_is_none() {
        let limits = PluginLimits::default();
        let state = PluginState::new("test".into(), &limits);
        assert!(state.stream_sender.is_none());
    }

    #[test]
    fn plugin_state_set_stream_sender() {
        let limits = PluginLimits::default();
        let mut state = PluginState::new("test".into(), &limits);
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<crate::instance::StreamEvent>();
        state.set_stream_sender(Arc::new(tx));
        assert!(state.stream_sender.is_some());
    }

    #[test]
    fn plugin_state_take_last_http_result_default_is_none() {
        let limits = PluginLimits::default();
        let state = PluginState::new("test".into(), &limits);
        assert!(state.last_http_result.is_none());
    }

    #[test]
    fn plugin_state_take_last_http_result_takes_value() {
        let limits = PluginLimits::default();
        let mut state = PluginState::new("test".into(), &limits);
        state.last_http_result = Some(vec![1, 2, 3]);

        let taken = state.last_http_result.take();
        assert_eq!(taken, Some(vec![1, 2, 3]));
        assert!(state.last_http_result.is_none());
    }

    #[test]
    fn stream_event_headers_fields() {
        let event = StreamEvent::Headers {
            status: 200,
            headers: std::collections::BTreeMap::from([(
                "content-type".to_string(),
                "text/event-stream".to_string(),
            )]),
        };
        if let StreamEvent::Headers { status, headers } = event {
            assert_eq!(status, 200);
            assert_eq!(
                headers.get("content-type").map(String::as_str),
                Some("text/event-stream")
            );
        } else {
            panic!("expected Headers variant");
        }
    }

    #[test]
    fn stream_event_chunk_contains_bytes() {
        let data = bytes::Bytes::from_static(b"data: hello\n\n");
        let event = StreamEvent::Chunk(data.clone());
        if let StreamEvent::Chunk(b) = event {
            assert_eq!(b, data);
        } else {
            panic!("expected Chunk variant");
        }
    }

    #[test]
    fn plugin_state_with_all_options_sets_publishers() {
        let limits = PluginLimits::default();
        let nats = Arc::new(crate::nats_client::NatsPublisher::new());
        let kafka = Arc::new(crate::kafka_client::KafkaPublisher::new());
        let state = PluginState::with_all_options(
            "test".into(),
            &limits,
            None,
            crate::secrets::SecretsStore::new(),
            None,
            None,
            Some(nats),
            Some(kafka),
        );
        assert!(state.nats_publisher.is_some());
        assert!(state.kafka_publisher.is_some());
    }
}