zlayer-agent 0.11.3

// Allow Arc with non-Send/Sync types for the instance pools.
// The PooledInstance contains Store<WasmHttpState> which has WasiCtx containing
// Box<dyn RngCore + Send> that is Send but not Sync. This is fine because
// we only access the pool through RwLock which provides synchronization.
#![allow(clippy::arc_with_non_send_sync)]

//! WASM HTTP Handler Runtime with Instance Pooling
//!
//! This module provides a specialized runtime for WASM components that implement
//! the `wasi:http/incoming-handler` interface. It includes instance pooling for
//! high-performance request handling with minimal cold start overhead.
//!
//! ## Features
//!
//! - **Instance Pooling**: Pre-warmed instances reduce cold start latency
//! - **Component Caching**: Compiled components are cached for reuse
//! - **Request/Response Mapping**: HTTP semantics mapped to WASI HTTP types
//! - **Configurable Timeouts**: Per-request and idle timeouts
//! - **HTTP Client Support**: WASM components can make outgoing HTTP requests
//!   via `wasi:http/outgoing-handler` (enabled by default)
//!
//! ## HTTP Client Support (Outgoing Requests)
//!
//! WASM components running in this runtime can make HTTP client requests to external
//! services via the `wasi:http/outgoing-handler` interface. This is enabled by default
//! through the `default-send-request` feature of `wasmtime-wasi-http`, which provides:
//!
//! - **TLS Support**: Secure HTTPS connections using rustls
//! - **Timeout Configuration**: Connect, first-byte, and between-bytes timeouts
//! - **Connection Pooling**: Efficient connection reuse via hyper
//!
//! Guest components can use the WASI HTTP interfaces to fetch data, call APIs,
//! or communicate with other services.
//!
//! ## Usage
//!
//! ```no_run,ignore
//! use zlayer_agent::runtimes::wasm_http::{WasmHttpRuntime, HttpRequest};
//! use zlayer_spec::WasmHttpConfig;
//!
//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
//! let config = WasmHttpConfig::default();
//! let runtime = WasmHttpRuntime::new(config)?;
//!
//! // Pre-warm instances
//! let wasm_bytes = std::fs::read("handler.wasm")?;
//! runtime.prewarm("my-handler", &wasm_bytes, 5).await?;
//!
//! // Handle requests
//! let request = HttpRequest {
//!     method: "GET".to_string(),
//!     uri: "/api/hello".to_string(),
//!     headers: vec![("Content-Type".to_string(), "application/json".to_string())],
//!     body: None,
//! };
//!
//! let response = runtime.handle_request("my-handler", &wasm_bytes, request).await?;
//! println!("Status: {}", response.status);
//! # Ok(())
//! # }
//! ```

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use bytes::Bytes;
use http_body_util::{combinators::BoxBody, BodyExt, Full};
use sha2::{Digest, Sha256};
use thiserror::Error;
use tokio::sync::RwLock;
use tracing::{debug, info, instrument, warn};
use wasmtime::component::{Component, Linker, ResourceTable};
use wasmtime::{Config, Engine, Store, StoreLimits, StoreLimitsBuilder};
use wasmtime_wasi::{WasiCtx, WasiCtxBuilder, WasiCtxView, WasiView};
use wasmtime_wasi_http::bindings::http::types::Scheme;
use wasmtime_wasi_http::bindings::{Proxy, ProxyPre};
use wasmtime_wasi_http::body::HyperOutgoingBody;
use wasmtime_wasi_http::{WasiHttpCtx, WasiHttpView};
#[allow(deprecated)]
use zlayer_spec::{WasmCapabilities, WasmHttpConfig};

// =============================================================================
// HTTP Request/Response Types
// =============================================================================

/// Incoming HTTP request to be handled by a WASM component
#[derive(Debug, Clone)]
pub struct HttpRequest {
    /// HTTP method (GET, POST, PUT, DELETE, etc.)
    pub method: String,
    /// Request URI including path and query string
    pub uri: String,
    /// Request headers as key-value pairs
    pub headers: Vec<(String, String)>,
    /// Optional request body
    pub body: Option<Vec<u8>>,
}

impl HttpRequest {
    /// Create a new GET request
    pub fn get(uri: impl Into<String>) -> Self {
        Self {
            method: "GET".to_string(),
            uri: uri.into(),
            headers: Vec::new(),
            body: None,
        }
    }

    /// Create a new POST request with body
    pub fn post(uri: impl Into<String>, body: Vec<u8>) -> Self {
        Self {
            method: "POST".to_string(),
            uri: uri.into(),
            headers: Vec::new(),
            body: Some(body),
        }
    }

    /// Create a new PUT request with body
    pub fn put(uri: impl Into<String>, body: Vec<u8>) -> Self {
        Self {
            method: "PUT".to_string(),
            uri: uri.into(),
            headers: Vec::new(),
            body: Some(body),
        }
    }

    /// Create a new DELETE request
    pub fn delete(uri: impl Into<String>) -> Self {
        Self {
            method: "DELETE".to_string(),
            uri: uri.into(),
            headers: Vec::new(),
            body: None,
        }
    }

    /// Create a new PATCH request with body
    pub fn patch(uri: impl Into<String>, body: Vec<u8>) -> Self {
        Self {
            method: "PATCH".to_string(),
            uri: uri.into(),
            headers: Vec::new(),
            body: Some(body),
        }
    }

    /// Add a header to the request
    #[must_use]
    pub fn with_header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
        self.headers.push((name.into(), value.into()));
        self
    }
}

/// HTTP response from a WASM component
#[derive(Debug, Clone)]
pub struct HttpResponse {
    /// HTTP status code (200, 404, 500, etc.)
    pub status: u16,
    /// Response headers as key-value pairs
    pub headers: Vec<(String, String)>,
    /// Optional response body
    pub body: Option<Vec<u8>>,
}

impl HttpResponse {
    /// Create a new response with the given status
    #[must_use]
    pub fn new(status: u16) -> Self {
        Self {
            status,
            headers: Vec::new(),
            body: None,
        }
    }

    /// Create a 200 OK response
    #[must_use]
    pub fn ok() -> Self {
        Self::new(200)
    }

    /// Create a 201 Created response
    #[must_use]
    pub fn created() -> Self {
        Self::new(201)
    }

    /// Create a 204 No Content response
    #[must_use]
    pub fn no_content() -> Self {
        Self::new(204)
    }

    /// Create a 400 Bad Request response
    pub fn bad_request(message: impl Into<String>) -> Self {
        let body = message.into().into_bytes();
        Self {
            status: 400,
            headers: vec![("Content-Type".to_string(), "text/plain".to_string())],
            body: Some(body),
        }
    }

    /// Create a 401 Unauthorized response
    #[must_use]
    pub fn unauthorized() -> Self {
        Self::new(401)
    }

    /// Create a 403 Forbidden response
    #[must_use]
    pub fn forbidden() -> Self {
        Self::new(403)
    }

    /// Create a 404 Not Found response
    #[must_use]
    pub fn not_found() -> Self {
        Self::new(404)
    }

    /// Create a 500 Internal Server Error response
    pub fn internal_error(message: impl Into<String>) -> Self {
        let body = message.into().into_bytes();
        Self {
            status: 500,
            headers: vec![("Content-Type".to_string(), "text/plain".to_string())],
            body: Some(body),
        }
    }

    /// Add a header to the response
    #[must_use]
    pub fn with_header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
        self.headers.push((name.into(), value.into()));
        self
    }

    /// Set the response body
    #[must_use]
    pub fn with_body(mut self, body: Vec<u8>) -> Self {
        self.body = Some(body);
        self
    }
}

// =============================================================================
// Error Types
// =============================================================================

/// Errors that can occur in the WASM HTTP runtime
#[derive(Error, Debug)]
pub enum WasmHttpError {
    /// Failed to create the wasmtime engine
    #[error("failed to create wasmtime engine: {0}")]
    EngineCreation(String),

    /// Failed to compile a WASM component
    #[error("failed to compile component '{component}': {reason}")]
    Compilation { component: String, reason: String },

    /// Failed to instantiate a WASM component
    #[error("failed to instantiate component '{component}': {reason}")]
    Instantiation { component: String, reason: String },

    /// Failed to invoke the HTTP handler
    #[error("failed to invoke handler for '{component}': {reason}")]
    HandlerInvocation { component: String, reason: String },

    /// Request handling timed out
    #[error("request timed out after {timeout:?}")]
    Timeout { timeout: Duration },

    /// Component not found in cache
    #[error("component '{component}' not found in cache")]
    ComponentNotFound { component: String },

    /// Instance pool exhausted
    #[error("instance pool exhausted for component '{component}'")]
    PoolExhausted { component: String },

    /// Invalid WASM component
    #[error("invalid WASM component: {reason}")]
    InvalidComponent { reason: String },

    /// Internal runtime error
    #[error("internal error: {0}")]
    Internal(String),
}

impl From<wasmtime::Error> for WasmHttpError {
    fn from(err: wasmtime::Error) -> Self {
        WasmHttpError::Internal(err.to_string())
    }
}

// =============================================================================
// WASM State Types
// =============================================================================

/// State for WASM HTTP handler instances
///
/// Implements both `WasiView` and `WasiHttpView` to provide the necessary
/// host interfaces for WASI HTTP components.
pub struct WasmHttpState {
    /// WASI context for basic WASI functionality
    wasi_ctx: WasiCtx,
    /// WASI HTTP context for HTTP-specific functionality
    http_ctx: WasiHttpCtx,
    /// Resource table for managing WASI resources
    table: ResourceTable,
    /// Resource limiter for memory/table growth enforcement
    limiter: StoreLimits,
}

impl WasmHttpState {
    /// Create a new WASM HTTP state with all WASI capabilities enabled
    ///
    /// This is the backward-compatible default: inherits stdio and env.
    fn new() -> Self {
        let wasi_ctx = WasiCtxBuilder::new().inherit_stdio().inherit_env().build();

        Self {
            wasi_ctx,
            http_ctx: WasiHttpCtx::new(),
            table: ResourceTable::new(),
            limiter: StoreLimitsBuilder::new().build(),
        }
    }

    /// Create a new WASM HTTP state with capabilities-gated WASI context
    ///
    /// Only the WASI features enabled in the capabilities struct will be
    /// configured on the WASI context. This provides defense-in-depth:
    /// even if the linker has all interfaces registered, the WASI context
    /// controls which OS resources are actually available.
    fn with_capabilities(capabilities: &WasmCapabilities) -> Self {
        let mut builder = WasiCtxBuilder::new();
        super::wasm_host::configure_wasi_ctx_with_capabilities(&mut builder, capabilities);
        let wasi_ctx = builder.build();

        Self {
            wasi_ctx,
            http_ctx: WasiHttpCtx::new(),
            table: ResourceTable::new(),
            limiter: StoreLimitsBuilder::new().build(),
        }
    }

    /// Create a new WASM HTTP state with resource limits applied.
    ///
    /// The `store_limits` are built from the spec-level `WasmConfig.max_memory` field.
    #[allow(dead_code)] // Will be used when resource limits are fully wired in
    fn with_limits(store_limits: StoreLimits) -> Self {
        let wasi_ctx = WasiCtxBuilder::new().inherit_stdio().inherit_env().build();

        Self {
            wasi_ctx,
            http_ctx: WasiHttpCtx::new(),
            table: ResourceTable::new(),
            limiter: store_limits,
        }
    }

    /// Create a new WASM HTTP state with both capabilities gating and resource limits.
    ///
    /// This combines the WASI context gating from capabilities with the memory/table
    /// growth limits from the resource limiter.
    #[allow(dead_code)] // Will be used when resource limits are fully wired in
    fn with_capabilities_and_limits(
        capabilities: &WasmCapabilities,
        store_limits: StoreLimits,
    ) -> Self {
        let mut builder = WasiCtxBuilder::new();
        super::wasm_host::configure_wasi_ctx_with_capabilities(&mut builder, capabilities);
        let wasi_ctx = builder.build();

        Self {
            wasi_ctx,
            http_ctx: WasiHttpCtx::new(),
            table: ResourceTable::new(),
            limiter: store_limits,
        }
    }
}

impl WasiView for WasmHttpState {
    fn ctx(&mut self) -> WasiCtxView<'_> {
        WasiCtxView {
            ctx: &mut self.wasi_ctx,
            table: &mut self.table,
        }
    }
}

impl WasiHttpView for WasmHttpState {
    fn ctx(&mut self) -> &mut WasiHttpCtx {
        &mut self.http_ctx
    }

    fn table(&mut self) -> &mut ResourceTable {
        &mut self.table
    }
}

// =============================================================================
// Instance Pool Types
// =============================================================================

/// A WASM component instance for handling a single request
///
/// Note: This struct is NOT Send/Sync because Store<WasmHttpState> contains
/// types that aren't thread-safe (e.g., RNG state). Instances must be created
/// and used on the same thread/task.
struct RequestInstance {
    /// The wasmtime store containing the WASM state
    store: Store<WasmHttpState>,
    /// The proxy bindings for calling WASI HTTP handlers
    proxy: Proxy,
    /// When this instance was created
    #[allow(dead_code)]
    created_at: Instant,
}

impl std::fmt::Debug for RequestInstance {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RequestInstance")
            .field("created_at", &self.created_at)
            .finish_non_exhaustive()
    }
}

/// Thread-safe statistics tracker for instance pools
///
/// Since wasmtime Store is not Sync, we can't pool actual instances across
/// threads. Instead, we track statistics and create fresh instances per request.
/// The compiled Component IS thread-safe and gets cached.
#[derive(Debug)]
pub struct InstancePool {
    /// Maximum idle instances (configuration, not enforced since we don't pool)
    #[allow(dead_code)]
    max_idle: usize,
    /// Maximum age for instances (configuration)
    #[allow(dead_code)]
    max_age: Duration,
    /// Total instances created (atomic for thread safety)
    total_created: AtomicU64,
    /// Total instances destroyed (atomic for thread safety)
    total_destroyed: AtomicU64,
    /// Total requests handled (atomic for thread safety)
    total_requests: AtomicU64,
}

impl InstancePool {
    /// Create a new instance pool with the given configuration
    fn new(max_idle: usize, max_age: Duration) -> Self {
        Self {
            max_idle,
            max_age,
            total_created: AtomicU64::new(0),
            total_destroyed: AtomicU64::new(0),
            total_requests: AtomicU64::new(0),
        }
    }

    /// Get the configured max idle instances
    #[allow(dead_code)]
    pub fn max_idle(&self) -> usize {
        self.max_idle
    }

    /// Get the configured max age for instances
    #[allow(dead_code)]
    pub fn max_age(&self) -> Duration {
        self.max_age
    }

    /// Record that an instance was created
    fn record_created(&self) {
        self.total_created.fetch_add(1, Ordering::Relaxed);
    }

    /// Record that an instance was destroyed
    fn record_destroyed(&self) {
        self.total_destroyed.fetch_add(1, Ordering::Relaxed);
    }

    /// Record that a request was handled
    fn record_request(&self) {
        self.total_requests.fetch_add(1, Ordering::Relaxed);
    }

    /// Get total instances created
    fn total_created(&self) -> u64 {
        self.total_created.load(Ordering::Relaxed)
    }

    /// Get total instances destroyed
    fn total_destroyed(&self) -> u64 {
        self.total_destroyed.load(Ordering::Relaxed)
    }

    /// Get total requests handled
    fn total_requests(&self) -> u64 {
        self.total_requests.load(Ordering::Relaxed)
    }
}

// =============================================================================
// Compiled Component Cache
// =============================================================================

/// A cached compiled component
struct CompiledComponent {
    /// The compiled wasmtime component
    component: Component,
    /// When this component was compiled
    compiled_at: Instant,
}

impl std::fmt::Debug for CompiledComponent {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CompiledComponent")
            .field("compiled_at", &self.compiled_at)
            .field("age", &self.compiled_at.elapsed())
            .finish_non_exhaustive()
    }
}

// =============================================================================
// Pool Statistics
// =============================================================================

/// Statistics about the instance pools
#[derive(Debug, Clone, Default)]
pub struct PoolStats {
    /// Number of components in the cache
    pub cached_components: usize,
    /// Total idle instances across all pools
    pub total_idle_instances: usize,
    /// Total instances created since startup
    pub total_created: u64,
    /// Total instances destroyed since startup
    pub total_destroyed: u64,
    /// Total requests handled since startup
    pub total_requests: u64,
    /// Per-component statistics
    pub components: HashMap<String, ComponentStats>,
}

/// Statistics for a single component
#[derive(Debug, Clone, Default)]
pub struct ComponentStats {
    /// Number of idle instances
    pub idle_instances: usize,
    /// Total instances created
    pub total_created: u64,
    /// Total instances destroyed
    pub total_destroyed: u64,
    /// Total requests handled
    pub total_requests: u64,
}

// =============================================================================
// WASM HTTP Runtime
// =============================================================================

/// WASM HTTP Runtime with component caching and request tracking
///
/// Provides high-performance request handling for WASM components that implement
/// the `wasi:http/incoming-handler` interface.
///
/// ## Design Notes
///
/// Due to wasmtime's Store not being Sync (it contains non-thread-safe RNG state),
/// we cannot pool instances across threads. Instead, this runtime:
///
/// 1. Caches compiled Components (which ARE thread-safe)
/// 2. Creates fresh instances per request (fast due to cached compilation)
/// 3. Tracks statistics for monitoring
///
/// The `prewarm` method compiles components ahead of time but doesn't pre-create
/// instances since they can't be shared across threads.
#[allow(deprecated)]
pub struct WasmHttpRuntime {
    /// Wasmtime engine (shared across all instances)
    engine: Engine,
    /// Cache of compiled components (thread-safe)
    module_cache: Arc<RwLock<HashMap<String, CompiledComponent>>>,
    /// Per-component statistics tracking (thread-safe via atomics)
    instance_pools: Arc<RwLock<HashMap<String, InstancePool>>>,
    /// Runtime configuration
    config: WasmHttpConfig,
    /// Component linker with WASI HTTP bindings
    linker: Arc<Linker<WasmHttpState>>,
    /// Directory for AOT pre-compiled component disk cache
    cache_dir: PathBuf,
    /// Capability grants controlling which host interfaces are available.
    /// When `None`, all interfaces are linked (backward-compatible default).
    capabilities: Option<WasmCapabilities>,
    /// Optional store limits built from spec-level `WasmConfig.max_memory`.
    /// Applied to every per-request Store created in `create_instance`.
    store_limits: Option<StoreLimits>,
    /// Fuel budget per request (0 = unlimited).
    /// From spec-level `WasmConfig.max_fuel`.
    max_fuel: u64,
}

impl std::fmt::Debug for WasmHttpRuntime {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("WasmHttpRuntime")
            .field("config", &self.config)
            .field("capabilities", &self.capabilities)
            .finish_non_exhaustive()
    }
}

impl WasmHttpRuntime {
    /// Create a new WASM HTTP runtime with the given configuration
    ///
    /// All host interfaces are linked (backward-compatible behavior).
    /// For capability-gated linking, use [`new_with_capabilities`](Self::new_with_capabilities).
    ///
    /// # Arguments
    ///
    /// * `config` - Configuration for the runtime including pool sizes and timeouts
    ///
    /// # Returns
    ///
    /// A new `WasmHttpRuntime` or an error if the engine could not be created.
    ///
    /// # Errors
    ///
    /// Returns an error if the wasmtime engine cannot be created.
    #[allow(deprecated)]
    pub fn new(config: WasmHttpConfig) -> Result<Self, WasmHttpError> {
        let mut engine_config = Config::new();
        engine_config.async_support(true);
        engine_config.wasm_component_model(true);
        engine_config.epoch_interruption(true);
        // Enable fuel metering so per-request fuel budgets can be applied.
        // Negligible cost when fuel is not actually set on a particular Store.
        engine_config.consume_fuel(true);

        let engine = Engine::new(&engine_config)
            .map_err(|e| WasmHttpError::EngineCreation(e.to_string()))?;

        // Create linker with WASI HTTP bindings (all linked, no gating)
        let mut linker = Linker::new(&engine);

        // Add WASI bindings
        wasmtime_wasi::p2::add_to_linker_async(&mut linker)
            .map_err(|e| WasmHttpError::EngineCreation(format!("failed to add WASI: {e}")))?;

        // Add WASI HTTP bindings
        wasmtime_wasi_http::add_only_http_to_linker_async(&mut linker)
            .map_err(|e| WasmHttpError::EngineCreation(format!("failed to add WASI HTTP: {e}")))?;

        // Initialize the AOT disk cache directory
        let cache_dir = zlayer_paths::ZLayerDirs::system_default().wasm_compiled();
        if let Err(e) = std::fs::create_dir_all(&cache_dir) {
            warn!(
                cache_dir = %cache_dir.display(),
                error = %e,
                "Failed to create WASM AOT cache directory (disk caching will be attempted on first compile)"
            );
        }

        info!(cache_dir = %cache_dir.display(), "WASM HTTP runtime initialized (all capabilities)");

        Ok(Self {
            engine,
            module_cache: Arc::new(RwLock::new(HashMap::new())),
            instance_pools: Arc::new(RwLock::new(HashMap::new())),
            config,
            linker: Arc::new(linker),
            cache_dir,
            capabilities: None,
            store_limits: None,
            max_fuel: 0,
        })
    }

    /// Create a new WASM HTTP runtime with capability-gated host interfaces
    ///
    /// Only the host interfaces enabled in `capabilities` will be linked and
    /// available to WASM components. This provides security isolation by
    /// controlling what the guest can access.
    ///
    /// Gating happens at two levels:
    /// - **Linker level**: `ZLayer` custom interfaces (config, KV, logging, secrets,
    ///   metrics) and WASI HTTP are conditionally registered
    /// - **`WasiCtx` level**: WASI standard interfaces (CLI, filesystem, sockets)
    ///   are configured per-instance via the builder
    ///
    /// # Arguments
    ///
    /// * `config` - Configuration for the runtime including pool sizes and timeouts
    /// * `capabilities` - Controls which host interfaces are available
    ///
    /// # Errors
    ///
    /// Returns an error if the wasmtime engine cannot be created.
    #[allow(deprecated)]
    pub fn new_with_capabilities(
        config: WasmHttpConfig,
        capabilities: WasmCapabilities,
    ) -> Result<Self, WasmHttpError> {
        let mut engine_config = Config::new();
        engine_config.async_support(true);
        engine_config.wasm_component_model(true);
        engine_config.epoch_interruption(true);
        engine_config.consume_fuel(true);

        let engine = Engine::new(&engine_config)
            .map_err(|e| WasmHttpError::EngineCreation(e.to_string()))?;

        // Create linker with only the enabled interfaces
        let mut linker = Linker::new(&engine);

        // Always add core WASI bindings (clocks, random, I/O streams, etc.)
        // These are needed for basic WASM component operation.
        // CLI, filesystem, and sockets are gated at the WasiCtx builder level.
        wasmtime_wasi::p2::add_to_linker_async(&mut linker)
            .map_err(|e| WasmHttpError::EngineCreation(format!("failed to add WASI: {e}")))?;

        // Conditionally add WASI HTTP bindings (outgoing handler)
        if capabilities.http_client {
            wasmtime_wasi_http::add_only_http_to_linker_async(&mut linker).map_err(|e| {
                WasmHttpError::EngineCreation(format!("failed to add WASI HTTP: {e}"))
            })?;
            debug!("WASI HTTP outgoing handler linked");
        } else {
            debug!("WASI HTTP outgoing handler NOT linked (capability disabled)");
        }

        // Initialize the AOT disk cache directory
        let cache_dir = zlayer_paths::ZLayerDirs::system_default().wasm_compiled();
        if let Err(e) = std::fs::create_dir_all(&cache_dir) {
            warn!(
                cache_dir = %cache_dir.display(),
                error = %e,
                "Failed to create WASM AOT cache directory (disk caching will be attempted on first compile)"
            );
        }

        info!(
            cache_dir = %cache_dir.display(),
            ?capabilities,
            "WASM HTTP runtime initialized with capabilities"
        );

        Ok(Self {
            engine,
            module_cache: Arc::new(RwLock::new(HashMap::new())),
            instance_pools: Arc::new(RwLock::new(HashMap::new())),
            config,
            linker: Arc::new(linker),
            cache_dir,
            capabilities: Some(capabilities),
            store_limits: None,
            max_fuel: 0,
        })
    }

    /// Get the capabilities for this runtime, if set
    ///
    /// Returns `None` if the runtime was created without capability gating
    /// (i.e., via [`new()`](Self::new)).
    #[must_use]
    pub fn capabilities(&self) -> Option<&WasmCapabilities> {
        self.capabilities.as_ref()
    }

    /// Configure resource limits from the spec-level [`zlayer_spec::WasmConfig`].
    ///
    /// This sets the per-request memory limit and fuel budget. Call this after
    /// constructing the runtime to apply limits from the deployment spec.
    ///
    /// # Arguments
    /// * `spec_wasm` - The spec-level [`WasmConfig`](zlayer_spec::WasmConfig) containing `max_memory` and `max_fuel`
    pub fn set_resource_limits(&mut self, spec_wasm: &zlayer_spec::WasmConfig) {
        // Parse memory limit
        if let Some(ref max_memory) = spec_wasm.max_memory {
            match super::wasm::parse_memory_limit(max_memory) {
                Ok(max_bytes) => {
                    #[allow(clippy::cast_possible_truncation)]
                    let mem_size = max_bytes as usize;
                    self.store_limits =
                        Some(StoreLimitsBuilder::new().memory_size(mem_size).build());
                    info!(
                        max_bytes = max_bytes,
                        "WASM HTTP store memory limit configured"
                    );
                }
                Err(e) => {
                    warn!(max_memory = %max_memory, error = %e, "ignoring invalid max_memory");
                }
            }
        }

        // Set fuel budget
        self.max_fuel = spec_wasm.max_fuel;
        if self.max_fuel > 0 {
            info!(max_fuel = self.max_fuel, "WASM HTTP fuel budget configured");
        }
    }

    /// Handle an HTTP request with a WASM component
    ///
    /// This method will:
    /// 1. Compile the component if not already cached
    /// 2. Create a fresh instance for this request
    /// 3. Invoke the HTTP handler
    /// 4. Clean up the instance
    ///
    /// # Arguments
    ///
    /// * `component_ref` - A unique identifier for the component (used for caching)
    /// * `wasm_bytes` - The raw WASM component bytes
    /// * `request` - The HTTP request to handle
    ///
    /// # Returns
    ///
    /// The HTTP response from the component or an error.
    ///
    /// # Errors
    ///
    /// Returns an error if the component cannot be loaded, instantiated, or the request fails.
    #[allow(deprecated)]
    #[instrument(skip(self, wasm_bytes, request), fields(component = %component_ref, method = %request.method, uri = %request.uri))]
    pub async fn handle_request(
        &self,
        component_ref: &str,
        wasm_bytes: &[u8],
        request: HttpRequest,
    ) -> Result<HttpResponse, WasmHttpError> {
        let timeout = self.config.request_timeout;

        // Get or compile the component
        let component = self.get_or_compile(component_ref, wasm_bytes).await?;

        // Create a fresh instance for this request
        let mut instance = self.create_instance(component_ref, &component).await?;

        // Record the request
        self.record_request(component_ref).await;

        // Handle the request with timeout
        let component_ref_owned = component_ref.to_string();
        let result = tokio::time::timeout(timeout, async {
            self.invoke_handler(&mut instance, &component_ref_owned, request)
                .await
        })
        .await;

        // Record instance destruction (instances are not pooled due to thread safety)
        self.record_destroyed(component_ref).await;

        match result {
            Ok(Ok(response)) => Ok(response),
            Ok(Err(e)) => Err(e),
            Err(_) => Err(WasmHttpError::Timeout { timeout }),
        }
    }

    /// Pre-warm the component cache
    ///
    /// Compiles the component ahead of time to reduce first-request latency.
    /// Note: Due to thread-safety constraints with wasmtime Store, actual
    /// instances cannot be pre-created and pooled across threads.
    ///
    /// # Arguments
    ///
    /// * `component_ref` - A unique identifier for the component
    /// * `wasm_bytes` - The raw WASM component bytes
    /// * `_count` - Ignored (kept for API compatibility)
    ///
    /// # Errors
    ///
    /// Returns an error if the component cannot be compiled.
    #[allow(clippy::used_underscore_binding, deprecated)]
    #[instrument(skip(self, wasm_bytes), fields(component = %component_ref))]
    pub async fn prewarm(
        &self,
        component_ref: &str,
        wasm_bytes: &[u8],
        _count: usize,
    ) -> Result<(), WasmHttpError> {
        info!("Pre-warming component cache");

        // Compile component to cache it
        let _ = self.get_or_compile(component_ref, wasm_bytes).await?;

        // Initialize pool stats entry
        {
            let mut pools = self.instance_pools.write().await;
            pools.entry(component_ref.to_string()).or_insert_with(|| {
                InstancePool::new(self.config.max_instances as usize, self.config.idle_timeout)
            });
        }

        Ok(())
    }

    /// Get statistics about the instance pools
    pub async fn pool_stats(&self) -> PoolStats {
        let pools = self.instance_pools.read().await;
        let cache = self.module_cache.read().await;

        let mut stats = PoolStats {
            cached_components: cache.len(),
            ..Default::default()
        };

        for (name, pool) in pools.iter() {
            // No idle instances since we don't pool (Store is not Sync)
            stats.total_created += pool.total_created();
            stats.total_destroyed += pool.total_destroyed();
            stats.total_requests += pool.total_requests();

            stats.components.insert(
                name.clone(),
                ComponentStats {
                    idle_instances: 0, // No pooling due to thread safety
                    total_created: pool.total_created(),
                    total_destroyed: pool.total_destroyed(),
                    total_requests: pool.total_requests(),
                },
            );
        }

        stats
    }

    /// Clear the in-memory and disk caches for compiled WASM components
    ///
    /// This forces recompilation of all components on next use.
    pub async fn clear_cache(&self) {
        let mut cache = self.module_cache.write().await;
        cache.clear();

        // Clear the disk cache as well
        if self.cache_dir.exists() {
            if let Err(e) = tokio::fs::remove_dir_all(&self.cache_dir).await {
                warn!(
                    cache_dir = %self.cache_dir.display(),
                    error = %e,
                    "Failed to remove WASM AOT disk cache directory"
                );
            } else if let Err(e) = tokio::fs::create_dir_all(&self.cache_dir).await {
                warn!(
                    cache_dir = %self.cache_dir.display(),
                    error = %e,
                    "Failed to recreate WASM AOT disk cache directory"
                );
            }
        }

        info!("Component cache cleared (in-memory and disk)");
    }

    // -------------------------------------------------------------------------
    // Private Helper Methods
    // -------------------------------------------------------------------------

    /// Compute a SHA-256 content hash for use as a disk cache key
    fn content_hash(wasm_bytes: &[u8]) -> String {
        let mut hasher = Sha256::new();
        hasher.update(wasm_bytes);
        format!("{:x}", hasher.finalize())
    }

    /// Get a compiled component from in-memory cache, disk cache, or compile from scratch
    #[allow(clippy::too_many_lines, unsafe_code)]
    async fn get_or_compile(
        &self,
        component_ref: &str,
        wasm_bytes: &[u8],
    ) -> Result<Arc<Component>, WasmHttpError> {
        // 1. Check in-memory cache first
        {
            let cache = self.module_cache.read().await;
            if let Some(compiled) = cache.get(component_ref) {
                debug!(
                    component = component_ref,
                    "Using in-memory cached component"
                );
                return Ok(Arc::new(compiled.component.clone()));
            }
        }

        // 2. Compute content hash for disk cache key
        let hash = Self::content_hash(wasm_bytes);
        let cache_path = self.cache_dir.join(format!("{hash}.cwasm"));

        // 3. Try to load from disk cache
        if cache_path.exists() {
            match tokio::fs::read(&cache_path).await {
                Ok(serialized) => {
                    let engine = self.engine.clone();
                    let cache_path_display = cache_path.display().to_string();
                    let component_ref_owned = component_ref.to_string();

                    match tokio::task::spawn_blocking(move || {
                        // SAFETY: We trust our own serialized cache files.
                        // The content-addressed SHA-256 hash ensures integrity.
                        unsafe { Component::deserialize(&engine, &serialized) }
                    })
                    .await
                    {
                        Ok(Ok(component)) => {
                            info!(
                                component = component_ref,
                                cache_path = cache_path_display,
                                "Loaded pre-compiled WASM component from disk cache"
                            );

                            // Store in in-memory cache
                            let mut cache = self.module_cache.write().await;
                            cache.insert(
                                component_ref.to_string(),
                                CompiledComponent {
                                    component: component.clone(),
                                    compiled_at: Instant::now(),
                                },
                            );

                            return Ok(Arc::new(component));
                        }
                        Ok(Err(e)) => {
                            warn!(
                                component = component_ref_owned,
                                error = %e,
                                "Failed to deserialize cached component, recompiling"
                            );
                            // Delete stale/incompatible cache file
                            let _ = tokio::fs::remove_file(&cache_path).await;
                        }
                        Err(e) => {
                            warn!(
                                component = component_ref_owned,
                                error = %e,
                                "Disk cache deserialization task failed, recompiling"
                            );
                            let _ = tokio::fs::remove_file(&cache_path).await;
                        }
                    }
                }
                Err(e) => {
                    debug!(
                        component = component_ref,
                        error = %e,
                        "Disk cache file not readable, compiling from scratch"
                    );
                }
            }
        }

        // 4. Compile from scratch
        info!(component = component_ref, "Compiling WASM component");
        let engine = self.engine.clone();
        let bytes = wasm_bytes.to_vec();
        let component_ref_owned = component_ref.to_string();

        let component = tokio::task::spawn_blocking(move || {
            Component::new(&engine, &bytes).map_err(|e| WasmHttpError::Compilation {
                component: component_ref_owned,
                reason: e.to_string(),
            })
        })
        .await
        .map_err(|e| WasmHttpError::Internal(format!("task join error: {e}")))??;

        // 5. Serialize to disk cache (non-blocking, best-effort)
        match component.serialize() {
            Ok(serialized) => {
                let cache_dir = self.cache_dir.clone();
                let cache_path_clone = cache_path.clone();
                let component_ref_clone = component_ref.to_string();
                tokio::spawn(async move {
                    if let Err(e) = tokio::fs::create_dir_all(&cache_dir).await {
                        warn!(error = %e, "Failed to create WASM AOT cache directory");
                        return;
                    }
                    if let Err(e) = tokio::fs::write(&cache_path_clone, &serialized).await {
                        warn!(
                            component = component_ref_clone,
                            error = %e,
                            "Failed to write WASM component to disk cache"
                        );
                    } else {
                        debug!(
                            component = component_ref_clone,
                            cache_path = %cache_path_clone.display(),
                            size_bytes = serialized.len(),
                            "Cached pre-compiled WASM component to disk"
                        );
                    }
                });
            }
            Err(e) => {
                warn!(
                    component = component_ref,
                    error = %e,
                    "Failed to serialize WASM component for disk caching"
                );
            }
        }

        // 6. Store in in-memory cache
        {
            let mut cache = self.module_cache.write().await;
            cache.insert(
                component_ref.to_string(),
                CompiledComponent {
                    component: component.clone(),
                    compiled_at: Instant::now(),
                },
            );
        }

        Ok(Arc::new(component))
    }

    /// Create a new instance for a component
    #[allow(deprecated)]
    async fn create_instance(
        &self,
        component_ref: &str,
        component: &Component,
    ) -> Result<RequestInstance, WasmHttpError> {
        let linker = Arc::clone(&self.linker);
        let component_ref_owned = component_ref.to_string();

        // Record that we're creating an instance
        {
            let mut pools = self.instance_pools.write().await;
            let pool = pools.entry(component_ref.to_string()).or_insert_with(|| {
                InstancePool::new(self.config.max_instances as usize, self.config.idle_timeout)
            });
            pool.record_created();
        }

        // Create the ProxyPre for efficient instantiation
        let proxy_pre = {
            let component_ref_owned = component_ref_owned.clone();
            let instance_pre =
                linker
                    .instantiate_pre(component)
                    .map_err(|e| WasmHttpError::Instantiation {
                        component: component_ref_owned.clone(),
                        reason: format!("failed to create instance pre: {e}"),
                    })?;

            ProxyPre::new(instance_pre).map_err(|e| WasmHttpError::Instantiation {
                component: component_ref_owned,
                reason: format!("failed to create proxy pre: {e}"),
            })?
        };

        // Create the store with WASM HTTP state, gated by capabilities if set.
        // If resource limits are configured, apply them to the state.
        let state = match (&self.capabilities, &self.store_limits) {
            (Some(caps), Some(limits)) => {
                let mut s = WasmHttpState::with_capabilities(caps);
                s.limiter = limits.clone();
                s
            }
            (Some(caps), None) => WasmHttpState::with_capabilities(caps),
            (None, Some(limits)) => WasmHttpState::with_limits(limits.clone()),
            (None, None) => WasmHttpState::new(),
        };
        let mut store = Store::new(&self.engine, state);

        // Attach the resource limiter to the store (enforces max_memory)
        store.limiter(|s| &mut s.limiter);

        // Set epoch deadline for interruption
        store.set_epoch_deadline(1_000_000);

        // Apply fuel budget if configured
        if self.max_fuel > 0 {
            store
                .set_fuel(self.max_fuel)
                .map_err(|e| WasmHttpError::Internal(format!("failed to set fuel: {e}")))?;
        }

        // Instantiate the proxy asynchronously
        let proxy = proxy_pre.instantiate_async(&mut store).await.map_err(|e| {
            WasmHttpError::Instantiation {
                component: component_ref.to_string(),
                reason: format!("failed to instantiate proxy: {e}"),
            }
        })?;

        debug!("Created new instance for {}", component_ref);
        Ok(RequestInstance {
            store,
            proxy,
            created_at: Instant::now(),
        })
    }

    /// Record that a request was handled
    async fn record_request(&self, component_ref: &str) {
        let pools = self.instance_pools.read().await;
        if let Some(pool) = pools.get(component_ref) {
            pool.record_request();
        }
    }

    /// Record that an instance was destroyed
    async fn record_destroyed(&self, component_ref: &str) {
        let pools = self.instance_pools.read().await;
        if let Some(pool) = pools.get(component_ref) {
            pool.record_destroyed();
        }
    }

    /// Invoke the HTTP handler on an instance
    ///
    /// This method implements the full `wasi:http/incoming-handler` protocol:
    /// 1. Convert the `HttpRequest` to a hyper request
    /// 2. Create WASI HTTP resources (`IncomingRequest` and `ResponseOutparam`)
    /// 3. Call the guest's `handle` export
    /// 4. Receive the response via the oneshot channel
    /// 5. Convert the response back to `HttpResponse`
    async fn invoke_handler(
        &self,
        instance: &mut RequestInstance,
        component_ref: &str,
        request: HttpRequest,
    ) -> Result<HttpResponse, WasmHttpError> {
        debug!(
            "Handling request: {} {} with {} headers",
            request.method,
            request.uri,
            request.headers.len()
        );

        // Increment epoch for cooperative scheduling
        instance.store.epoch_deadline_trap();

        // Step 1: Convert HttpRequest to hyper::Request
        let hyper_request = self.convert_to_hyper_request(&request).map_err(|e| {
            WasmHttpError::HandlerInvocation {
                component: component_ref.to_string(),
                reason: format!("failed to convert request: {e}"),
            }
        })?;

        // Step 2: Create the response channel
        let (response_sender, response_receiver) = tokio::sync::oneshot::channel();

        // Step 3: Create WASI HTTP resources in the store's resource table
        let incoming_request = instance
            .store
            .data_mut()
            .new_incoming_request(Scheme::Http, hyper_request)
            .map_err(|e| WasmHttpError::HandlerInvocation {
                component: component_ref.to_string(),
                reason: format!("failed to create incoming request resource: {e}"),
            })?;

        let response_outparam = instance
            .store
            .data_mut()
            .new_response_outparam(response_sender)
            .map_err(|e| WasmHttpError::HandlerInvocation {
                component: component_ref.to_string(),
                reason: format!("failed to create response outparam resource: {e}"),
            })?;

        // Step 4: Get the incoming handler and call it
        let handler = instance.proxy.wasi_http_incoming_handler();

        // Call the handler. The response may come before the handler returns
        // (e.g., for streaming responses) via the oneshot channel.
        let component_ref_owned = component_ref.to_string();
        let call_result = handler
            .call_handle(&mut instance.store, incoming_request, response_outparam)
            .await;

        if let Err(e) = &call_result {
            warn!(
                "Handler call failed for component {}: {}",
                component_ref_owned, e
            );
            // The handler failed, but we may still receive a response if the guest
            // called response-outparam.set before failing. Check the receiver below.
        }

        // Step 5: Wait for the response from the guest
        let hyper_response = match response_receiver.await {
            Ok(Ok(resp)) => resp,
            Ok(Err(error_code)) => {
                // Guest returned an error code via response-outparam.set
                return Err(WasmHttpError::HandlerInvocation {
                    component: component_ref.to_string(),
                    reason: format!("guest returned error: {error_code:?}"),
                });
            }
            Err(_) => {
                // The sender was dropped without sending a response.
                // This means the guest didn't call response-outparam.set
                return Err(WasmHttpError::HandlerInvocation {
                    component: component_ref.to_string(),
                    reason: "guest never invoked response-outparam.set".to_string(),
                });
            }
        };

        // Step 6: Convert hyper::Response to HttpResponse
        let response = self
            .convert_from_hyper_response(hyper_response)
            .await
            .map_err(|e| WasmHttpError::HandlerInvocation {
                component: component_ref.to_string(),
                reason: format!("failed to convert response: {e}"),
            })?;

        debug!(
            "Handler completed successfully: status={}, body_len={}",
            response.status,
            response.body.as_ref().map_or(0, std::vec::Vec::len)
        );

        Ok(response)
    }

    /// Convert an `HttpRequest` to a `hyper::Request` with a body type compatible with WASI HTTP
    #[allow(clippy::unused_self)]
    fn convert_to_hyper_request(
        &self,
        request: &HttpRequest,
    ) -> Result<hyper::Request<BoxBody<Bytes, hyper::Error>>, anyhow::Error> {
        use http::Method;

        // Parse the method
        let method = match request.method.to_uppercase().as_str() {
            "GET" => Method::GET,
            "POST" => Method::POST,
            "PUT" => Method::PUT,
            "DELETE" => Method::DELETE,
            "HEAD" => Method::HEAD,
            "OPTIONS" => Method::OPTIONS,
            "PATCH" => Method::PATCH,
            "TRACE" => Method::TRACE,
            "CONNECT" => Method::CONNECT,
            other => Method::from_bytes(other.as_bytes())?,
        };

        // Build the request
        let mut builder = hyper::Request::builder().method(method).uri(&request.uri);

        // Add headers
        for (name, value) in &request.headers {
            builder = builder.header(name.as_str(), value.as_str());
        }

        // Ensure we have a Host header (required for WASI HTTP)
        let has_host = request
            .headers
            .iter()
            .any(|(name, _)| name.eq_ignore_ascii_case("host"));
        if !has_host {
            // Extract host from URI or use a default
            if let Ok(uri) = request.uri.parse::<http::Uri>() {
                if let Some(authority) = uri.authority() {
                    builder = builder.header("Host", authority.as_str());
                } else {
                    builder = builder.header("Host", "localhost");
                }
            } else {
                builder = builder.header("Host", "localhost");
            }
        }

        // Build with body - map the Infallible error to hyper::Error
        let body = match &request.body {
            Some(bytes) => Full::new(Bytes::from(bytes.clone())),
            None => Full::new(Bytes::new()),
        };

        // Map the Infallible error to hyper::Error using map_err
        let boxed_body = body
            .map_err(|e: std::convert::Infallible| match e {})
            .boxed();

        Ok(builder.body(boxed_body)?)
    }

    /// Convert a `hyper::Response` to an `HttpResponse`
    async fn convert_from_hyper_response(
        &self,
        response: hyper::Response<HyperOutgoingBody>,
    ) -> Result<HttpResponse, anyhow::Error> {
        let (parts, body) = response.into_parts();

        // Convert status
        let status = parts.status.as_u16();

        // Convert headers
        let headers: Vec<(String, String)> = parts
            .headers
            .iter()
            .filter_map(|(name, value)| {
                value
                    .to_str()
                    .ok()
                    .map(|v| (name.to_string(), v.to_string()))
            })
            .collect();

        // Collect the body
        let body_bytes = body
            .collect()
            .await
            .map_err(|e| anyhow::anyhow!("failed to collect response body: {e}"))?
            .to_bytes();

        let body = if body_bytes.is_empty() {
            None
        } else {
            Some(body_bytes.to_vec())
        };

        Ok(HttpResponse {
            status,
            headers,
            body,
        })
    }
}

// =============================================================================
// Tests
// =============================================================================

#[cfg(test)]
#[allow(deprecated)]
mod tests {
    use super::*;
    use std::collections::HashMap;

    // =========================================================================
    // Test Configuration Helpers
    // =========================================================================

    fn test_config() -> WasmHttpConfig {
        WasmHttpConfig {
            min_instances: 0,
            max_instances: 10,
            idle_timeout: Duration::from_secs(60),
            request_timeout: Duration::from_secs(30),
        }
    }

    fn custom_config(min: u32, max: u32, idle_secs: u64, request_secs: u64) -> WasmHttpConfig {
        WasmHttpConfig {
            min_instances: min,
            max_instances: max,
            idle_timeout: Duration::from_secs(idle_secs),
            request_timeout: Duration::from_secs(request_secs),
        }
    }

    // =========================================================================
    // HttpRequest Tests
    // =========================================================================

    #[test]
    fn test_http_request_creation_with_all_fields() {
        let request = HttpRequest {
            method: "PUT".to_string(),
            uri: "/api/users/123".to_string(),
            headers: vec![
                ("Content-Type".to_string(), "application/json".to_string()),
                ("Authorization".to_string(), "Bearer token123".to_string()),
            ],
            body: Some(b"{\"name\": \"test\"}".to_vec()),
        };

        assert_eq!(request.method, "PUT");
        assert_eq!(request.uri, "/api/users/123");
        assert_eq!(request.headers.len(), 2);
        assert_eq!(
            request.headers[0],
            ("Content-Type".to_string(), "application/json".to_string())
        );
        assert_eq!(
            request.headers[1],
            ("Authorization".to_string(), "Bearer token123".to_string())
        );
        assert_eq!(request.body, Some(b"{\"name\": \"test\"}".to_vec()));
    }

    #[test]
    fn test_http_request_get_helper() {
        let request = HttpRequest::get("/api/test");
        assert_eq!(request.method, "GET");
        assert_eq!(request.uri, "/api/test");
        assert!(request.headers.is_empty());
        assert!(request.body.is_none());
    }

    #[test]
    fn test_http_request_get_with_string_type() {
        let uri = String::from("/api/resource");
        let request = HttpRequest::get(uri);
        assert_eq!(request.method, "GET");
        assert_eq!(request.uri, "/api/resource");
    }

    #[test]
    fn test_http_request_post_helper() {
        let body = b"test body content".to_vec();
        let request = HttpRequest::post("/api/submit", body.clone());

        assert_eq!(request.method, "POST");
        assert_eq!(request.uri, "/api/submit");
        assert!(request.headers.is_empty());
        assert_eq!(request.body, Some(body));
    }

    #[test]
    fn test_http_request_post_with_empty_body() {
        let request = HttpRequest::post("/api/submit", Vec::new());

        assert_eq!(request.method, "POST");
        assert_eq!(request.body, Some(Vec::new()));
    }

    #[test]
    fn test_http_request_with_header_builder() {
        let request = HttpRequest::get("/api/test")
            .with_header("Content-Type", "application/json")
            .with_header("Authorization", "Bearer token")
            .with_header("X-Custom-Header", "custom-value");

        assert_eq!(request.headers.len(), 3);
        assert_eq!(
            request.headers[0],
            ("Content-Type".to_string(), "application/json".to_string())
        );
        assert_eq!(
            request.headers[1],
            ("Authorization".to_string(), "Bearer token".to_string())
        );
        assert_eq!(
            request.headers[2],
            ("X-Custom-Header".to_string(), "custom-value".to_string())
        );
    }

    #[test]
    fn test_http_request_with_header_string_types() {
        let header_name = String::from("X-Request-Id");
        let header_value = String::from("abc-123");
        let request = HttpRequest::get("/test").with_header(header_name, header_value);

        assert_eq!(request.headers.len(), 1);
        assert_eq!(
            request.headers[0],
            ("X-Request-Id".to_string(), "abc-123".to_string())
        );
    }

    #[test]
    fn test_http_request_debug_formatting() {
        let request = HttpRequest::get("/api/test").with_header("Content-Type", "text/plain");

        let debug_str = format!("{request:?}");
        assert!(debug_str.contains("HttpRequest"));
        assert!(debug_str.contains("GET"));
        assert!(debug_str.contains("/api/test"));
        assert!(debug_str.contains("Content-Type"));
    }

    #[test]
    fn test_http_request_clone() {
        let original =
            HttpRequest::post("/api/data", b"body".to_vec()).with_header("X-Test", "value");

        let cloned = original.clone();
        assert_eq!(cloned.method, original.method);
        assert_eq!(cloned.uri, original.uri);
        assert_eq!(cloned.headers, original.headers);
        assert_eq!(cloned.body, original.body);
    }

    // =========================================================================
    // HttpResponse Tests
    // =========================================================================

    #[test]
    fn test_http_response_new() {
        let response = HttpResponse::new(201);
        assert_eq!(response.status, 201);
        assert!(response.headers.is_empty());
        assert!(response.body.is_none());
    }

    #[test]
    fn test_http_response_ok_helper() {
        let response = HttpResponse::ok();
        assert_eq!(response.status, 200);
        assert!(response.headers.is_empty());
        assert!(response.body.is_none());
    }

    #[test]
    fn test_http_response_internal_error_helper() {
        let response = HttpResponse::internal_error("Something went wrong");
        assert_eq!(response.status, 500);
        assert_eq!(response.headers.len(), 1);
        assert_eq!(
            response.headers[0],
            ("Content-Type".to_string(), "text/plain".to_string())
        );
        assert_eq!(
            response.body,
            Some("Something went wrong".as_bytes().to_vec())
        );
    }

    #[test]
    fn test_http_response_internal_error_with_string_type() {
        let error_msg = String::from("Database connection failed");
        let response = HttpResponse::internal_error(error_msg);
        assert_eq!(response.status, 500);
        assert_eq!(
            response.body,
            Some("Database connection failed".as_bytes().to_vec())
        );
    }

    #[test]
    fn test_http_response_internal_error_empty_message() {
        let response = HttpResponse::internal_error("");
        assert_eq!(response.status, 500);
        assert_eq!(response.body, Some(Vec::new()));
    }

    #[test]
    fn test_http_response_with_header_builder() {
        let response = HttpResponse::ok()
            .with_header("Content-Type", "application/json")
            .with_header("X-Request-Id", "abc-123")
            .with_header("Cache-Control", "no-cache");

        assert_eq!(response.headers.len(), 3);
        assert_eq!(
            response.headers[0],
            ("Content-Type".to_string(), "application/json".to_string())
        );
        assert_eq!(
            response.headers[1],
            ("X-Request-Id".to_string(), "abc-123".to_string())
        );
        assert_eq!(
            response.headers[2],
            ("Cache-Control".to_string(), "no-cache".to_string())
        );
    }

    #[test]
    fn test_http_response_with_body_builder() {
        let body = b"{\"status\": \"ok\"}".to_vec();
        let response = HttpResponse::ok().with_body(body.clone());

        assert_eq!(response.body, Some(body));
    }

    #[test]
    fn test_http_response_with_body_empty() {
        let response = HttpResponse::ok().with_body(Vec::new());
        assert_eq!(response.body, Some(Vec::new()));
    }

    #[test]
    fn test_http_response_builder_chain() {
        let response = HttpResponse::new(201)
            .with_header("Content-Type", "application/json")
            .with_header("Location", "/api/users/456")
            .with_body(b"{\"id\": 456}".to_vec());

        assert_eq!(response.status, 201);
        assert_eq!(response.headers.len(), 2);
        assert_eq!(response.body, Some(b"{\"id\": 456}".to_vec()));
    }

    #[test]
    fn test_http_response_debug_formatting() {
        let response = HttpResponse::ok()
            .with_header("Content-Type", "text/html")
            .with_body(b"<html>".to_vec());

        let debug_str = format!("{response:?}");
        assert!(debug_str.contains("HttpResponse"));
        assert!(debug_str.contains("200"));
        assert!(debug_str.contains("Content-Type"));
    }

    #[test]
    fn test_http_response_clone() {
        let original = HttpResponse::ok()
            .with_header("X-Test", "value")
            .with_body(b"test".to_vec());

        let cloned = original.clone();
        assert_eq!(cloned.status, original.status);
        assert_eq!(cloned.headers, original.headers);
        assert_eq!(cloned.body, original.body);
    }

    #[test]
    fn test_http_response_various_status_codes() {
        assert_eq!(HttpResponse::new(100).status, 100); // Continue
        assert_eq!(HttpResponse::new(204).status, 204); // No Content
        assert_eq!(HttpResponse::new(301).status, 301); // Moved Permanently
        assert_eq!(HttpResponse::new(400).status, 400); // Bad Request
        assert_eq!(HttpResponse::new(401).status, 401); // Unauthorized
        assert_eq!(HttpResponse::new(403).status, 403); // Forbidden
        assert_eq!(HttpResponse::new(404).status, 404); // Not Found
        assert_eq!(HttpResponse::new(500).status, 500); // Internal Server Error
        assert_eq!(HttpResponse::new(502).status, 502); // Bad Gateway
        assert_eq!(HttpResponse::new(503).status, 503); // Service Unavailable
    }

    // =========================================================================
    // WasmHttpError Tests
    // =========================================================================

    #[test]
    fn test_wasm_http_error_engine_creation_display() {
        let error = WasmHttpError::EngineCreation("invalid config".to_string());
        let msg = error.to_string();
        assert!(msg.contains("failed to create wasmtime engine"));
        assert!(msg.contains("invalid config"));
    }

    #[test]
    fn test_wasm_http_error_compilation_display() {
        let error = WasmHttpError::Compilation {
            component: "my-handler".to_string(),
            reason: "invalid wasm bytes".to_string(),
        };
        let msg = error.to_string();
        assert!(msg.contains("failed to compile component"));
        assert!(msg.contains("my-handler"));
        assert!(msg.contains("invalid wasm bytes"));
    }

    #[test]
    fn test_wasm_http_error_instantiation_display() {
        let error = WasmHttpError::Instantiation {
            component: "api-service".to_string(),
            reason: "missing import".to_string(),
        };
        let msg = error.to_string();
        assert!(msg.contains("failed to instantiate component"));
        assert!(msg.contains("api-service"));
        assert!(msg.contains("missing import"));
    }

    #[test]
    fn test_wasm_http_error_handler_invocation_display() {
        let error = WasmHttpError::HandlerInvocation {
            component: "request-handler".to_string(),
            reason: "panic in wasm".to_string(),
        };
        let msg = error.to_string();
        assert!(msg.contains("failed to invoke handler"));
        assert!(msg.contains("request-handler"));
        assert!(msg.contains("panic in wasm"));
    }

    #[test]
    fn test_wasm_http_error_timeout_display() {
        let error = WasmHttpError::Timeout {
            timeout: Duration::from_secs(30),
        };
        let msg = error.to_string();
        assert!(msg.contains("request timed out"));
        assert!(msg.contains("30"));
    }

    #[test]
    fn test_wasm_http_error_timeout_display_millis() {
        let error = WasmHttpError::Timeout {
            timeout: Duration::from_millis(500),
        };
        let msg = error.to_string();
        assert!(msg.contains("timed out"));
        assert!(msg.contains("500"));
    }

    #[test]
    fn test_wasm_http_error_component_not_found_display() {
        let error = WasmHttpError::ComponentNotFound {
            component: "missing-component".to_string(),
        };
        let msg = error.to_string();
        assert!(msg.contains("not found in cache"));
        assert!(msg.contains("missing-component"));
    }

    #[test]
    fn test_wasm_http_error_pool_exhausted_display() {
        let error = WasmHttpError::PoolExhausted {
            component: "busy-service".to_string(),
        };
        let msg = error.to_string();
        assert!(msg.contains("instance pool exhausted"));
        assert!(msg.contains("busy-service"));
    }

    #[test]
    fn test_wasm_http_error_invalid_component_display() {
        let error = WasmHttpError::InvalidComponent {
            reason: "not a component".to_string(),
        };
        let msg = error.to_string();
        assert!(msg.contains("invalid WASM component"));
        assert!(msg.contains("not a component"));
    }

    #[test]
    fn test_wasm_http_error_internal_display() {
        let error = WasmHttpError::Internal("unexpected state".to_string());
        let msg = error.to_string();
        assert!(msg.contains("internal error"));
        assert!(msg.contains("unexpected state"));
    }

    #[test]
    fn test_wasm_http_error_debug_formatting() {
        let error = WasmHttpError::Compilation {
            component: "test".to_string(),
            reason: "error".to_string(),
        };
        let debug_str = format!("{error:?}");
        assert!(debug_str.contains("Compilation"));
        assert!(debug_str.contains("test"));
        assert!(debug_str.contains("error"));
    }

    #[test]
    fn test_wasm_http_error_all_variants_are_errors() {
        // Verify all error variants implement std::error::Error via Display
        let errors: Vec<WasmHttpError> = vec![
            WasmHttpError::EngineCreation("test".to_string()),
            WasmHttpError::Compilation {
                component: "c".to_string(),
                reason: "r".to_string(),
            },
            WasmHttpError::Instantiation {
                component: "c".to_string(),
                reason: "r".to_string(),
            },
            WasmHttpError::HandlerInvocation {
                component: "c".to_string(),
                reason: "r".to_string(),
            },
            WasmHttpError::Timeout {
                timeout: Duration::from_secs(1),
            },
            WasmHttpError::ComponentNotFound {
                component: "c".to_string(),
            },
            WasmHttpError::PoolExhausted {
                component: "c".to_string(),
            },
            WasmHttpError::InvalidComponent {
                reason: "r".to_string(),
            },
            WasmHttpError::Internal("i".to_string()),
        ];

        for error in errors {
            // Each should produce a non-empty Display string
            let msg = error.to_string();
            assert!(!msg.is_empty(), "Error Display should not be empty");
        }
    }

    // =========================================================================
    // InstancePool Tests
    // =========================================================================

    #[test]
    fn test_instance_pool_new_with_correct_settings() {
        let pool = InstancePool::new(10, Duration::from_secs(120));

        assert_eq!(pool.max_idle(), 10);
        assert_eq!(pool.max_age(), Duration::from_secs(120));
        assert_eq!(pool.total_created(), 0);
        assert_eq!(pool.total_destroyed(), 0);
        assert_eq!(pool.total_requests(), 0);
    }

    #[test]
    fn test_instance_pool_new_with_zero_max_idle() {
        let pool = InstancePool::new(0, Duration::from_secs(60));
        assert_eq!(pool.max_idle(), 0);
    }

    #[test]
    fn test_instance_pool_new_with_zero_max_age() {
        let pool = InstancePool::new(5, Duration::ZERO);
        assert_eq!(pool.max_age(), Duration::ZERO);
    }

    #[test]
    fn test_instance_pool_atomic_counter_total_created() {
        let pool = InstancePool::new(10, Duration::from_secs(60));

        assert_eq!(pool.total_created(), 0);
        pool.record_created();
        assert_eq!(pool.total_created(), 1);
        pool.record_created();
        assert_eq!(pool.total_created(), 2);
        pool.record_created();
        pool.record_created();
        pool.record_created();
        assert_eq!(pool.total_created(), 5);
    }

    #[test]
    fn test_instance_pool_atomic_counter_total_destroyed() {
        let pool = InstancePool::new(10, Duration::from_secs(60));

        assert_eq!(pool.total_destroyed(), 0);
        pool.record_destroyed();
        assert_eq!(pool.total_destroyed(), 1);
        pool.record_destroyed();
        pool.record_destroyed();
        assert_eq!(pool.total_destroyed(), 3);
    }

    #[test]
    fn test_instance_pool_atomic_counter_total_requests() {
        let pool = InstancePool::new(10, Duration::from_secs(60));

        assert_eq!(pool.total_requests(), 0);
        pool.record_request();
        assert_eq!(pool.total_requests(), 1);
        pool.record_request();
        pool.record_request();
        pool.record_request();
        assert_eq!(pool.total_requests(), 4);
    }

    #[test]
    fn test_instance_pool_statistics_combined() {
        let pool = InstancePool::new(10, Duration::from_secs(60));

        // Simulate lifecycle: create instance, handle request, destroy
        pool.record_created();
        pool.record_request();
        pool.record_destroyed();

        // Create another, handle multiple requests
        pool.record_created();
        pool.record_request();
        pool.record_request();
        pool.record_request();
        pool.record_destroyed();

        assert_eq!(pool.total_created(), 2);
        assert_eq!(pool.total_destroyed(), 2);
        assert_eq!(pool.total_requests(), 4);
    }

    #[test]
    fn test_instance_pool_debug_formatting() {
        let pool = InstancePool::new(5, Duration::from_secs(30));
        pool.record_created();
        pool.record_request();

        let debug_str = format!("{pool:?}");
        assert!(debug_str.contains("InstancePool"));
    }

    // =========================================================================
    // PoolStats Tests
    // =========================================================================

    #[test]
    fn test_pool_stats_creation() {
        let stats = PoolStats {
            cached_components: 3,
            total_idle_instances: 5,
            total_created: 100,
            total_destroyed: 95,
            total_requests: 1000,
            components: HashMap::new(),
        };

        assert_eq!(stats.cached_components, 3);
        assert_eq!(stats.total_idle_instances, 5);
        assert_eq!(stats.total_created, 100);
        assert_eq!(stats.total_destroyed, 95);
        assert_eq!(stats.total_requests, 1000);
    }

    #[test]
    fn test_pool_stats_default() {
        let stats = PoolStats::default();
        assert_eq!(stats.cached_components, 0);
        assert_eq!(stats.total_idle_instances, 0);
        assert_eq!(stats.total_created, 0);
        assert_eq!(stats.total_destroyed, 0);
        assert_eq!(stats.total_requests, 0);
        assert!(stats.components.is_empty());
    }

    #[test]
    fn test_pool_stats_debug_formatting() {
        let stats = PoolStats {
            cached_components: 2,
            total_requests: 50,
            ..Default::default()
        };

        let debug_str = format!("{stats:?}");
        assert!(debug_str.contains("PoolStats"));
        assert!(debug_str.contains("cached_components"));
        assert!(debug_str.contains('2'));
    }

    #[test]
    fn test_pool_stats_clone() {
        let mut original = PoolStats {
            cached_components: 5,
            total_requests: 100,
            ..Default::default()
        };
        original.components.insert(
            "test-component".to_string(),
            ComponentStats {
                idle_instances: 2,
                total_created: 10,
                total_destroyed: 8,
                total_requests: 50,
            },
        );

        let cloned = original.clone();
        assert_eq!(cloned.cached_components, original.cached_components);
        assert_eq!(cloned.total_requests, original.total_requests);
        assert_eq!(cloned.components.len(), original.components.len());
        assert_eq!(
            cloned
                .components
                .get("test-component")
                .unwrap()
                .idle_instances,
            2
        );
    }

    #[test]
    fn test_pool_stats_with_multiple_components() {
        let mut stats = PoolStats::default();
        stats.components.insert(
            "api-handler".to_string(),
            ComponentStats {
                idle_instances: 3,
                total_created: 50,
                total_destroyed: 47,
                total_requests: 500,
            },
        );
        stats.components.insert(
            "auth-handler".to_string(),
            ComponentStats {
                idle_instances: 1,
                total_created: 20,
                total_destroyed: 19,
                total_requests: 200,
            },
        );

        assert_eq!(stats.components.len(), 2);
        assert!(stats.components.contains_key("api-handler"));
        assert!(stats.components.contains_key("auth-handler"));
    }

    // =========================================================================
    // ComponentStats Tests
    // =========================================================================

    #[test]
    fn test_component_stats_creation() {
        let stats = ComponentStats {
            idle_instances: 5,
            total_created: 100,
            total_destroyed: 95,
            total_requests: 1000,
        };

        assert_eq!(stats.idle_instances, 5);
        assert_eq!(stats.total_created, 100);
        assert_eq!(stats.total_destroyed, 95);
        assert_eq!(stats.total_requests, 1000);
    }

    #[test]
    fn test_component_stats_default() {
        let stats = ComponentStats::default();
        assert_eq!(stats.idle_instances, 0);
        assert_eq!(stats.total_created, 0);
        assert_eq!(stats.total_destroyed, 0);
        assert_eq!(stats.total_requests, 0);
    }

    #[test]
    fn test_component_stats_debug_formatting() {
        let stats = ComponentStats {
            idle_instances: 2,
            total_created: 10,
            total_destroyed: 8,
            total_requests: 50,
        };

        let debug_str = format!("{stats:?}");
        assert!(debug_str.contains("ComponentStats"));
        assert!(debug_str.contains("idle_instances"));
        assert!(debug_str.contains("total_created"));
    }

    #[test]
    fn test_component_stats_clone() {
        let original = ComponentStats {
            idle_instances: 3,
            total_created: 25,
            total_destroyed: 22,
            total_requests: 150,
        };

        let cloned = original.clone();
        assert_eq!(cloned.idle_instances, original.idle_instances);
        assert_eq!(cloned.total_created, original.total_created);
        assert_eq!(cloned.total_destroyed, original.total_destroyed);
        assert_eq!(cloned.total_requests, original.total_requests);
    }

    // =========================================================================
    // WasmHttpRuntime Tests
    // =========================================================================

    #[tokio::test]
    async fn test_runtime_new_with_default_config() {
        let config = WasmHttpConfig::default();
        let result = WasmHttpRuntime::new(config);
        assert!(result.is_ok(), "Failed to create runtime: {result:?}");
    }

    #[tokio::test]
    async fn test_runtime_new_with_custom_config() {
        let config = custom_config(2, 20, 120, 60);
        let result = WasmHttpRuntime::new(config.clone());
        assert!(
            result.is_ok(),
            "Failed to create runtime with custom config"
        );

        let runtime = result.unwrap();
        assert_eq!(runtime.config.min_instances, 2);
        assert_eq!(runtime.config.max_instances, 20);
        assert_eq!(runtime.config.idle_timeout, Duration::from_secs(120));
        assert_eq!(runtime.config.request_timeout, Duration::from_secs(60));
    }

    #[tokio::test]
    async fn test_runtime_new_with_zero_instances() {
        let config = custom_config(0, 0, 0, 1);
        let result = WasmHttpRuntime::new(config);
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_runtime_new_with_large_values() {
        let config = custom_config(100, 10000, 3600, 300);
        let result = WasmHttpRuntime::new(config);
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_runtime_pool_stats_returns_correct_initial_statistics() {
        let config = test_config();
        let runtime = WasmHttpRuntime::new(config).unwrap();
        let stats = runtime.pool_stats().await;

        assert_eq!(stats.cached_components, 0);
        assert_eq!(stats.total_idle_instances, 0);
        assert_eq!(stats.total_created, 0);
        assert_eq!(stats.total_destroyed, 0);
        assert_eq!(stats.total_requests, 0);
        assert!(stats.components.is_empty());
    }

    #[tokio::test]
    async fn test_runtime_clear_cache_empties_component_cache() {
        let config = test_config();
        let runtime = WasmHttpRuntime::new(config).unwrap();

        // Should not panic on empty cache
        runtime.clear_cache().await;

        // Cache should still be empty
        let stats = runtime.pool_stats().await;
        assert_eq!(stats.cached_components, 0);
    }

    #[tokio::test]
    async fn test_runtime_clear_cache_multiple_times() {
        let config = test_config();
        let runtime = WasmHttpRuntime::new(config).unwrap();

        // Multiple clears should be safe
        runtime.clear_cache().await;
        runtime.clear_cache().await;
        runtime.clear_cache().await;

        let stats = runtime.pool_stats().await;
        assert_eq!(stats.cached_components, 0);
    }

    #[tokio::test]
    async fn test_runtime_debug_formatting() {
        let config = test_config();
        let runtime = WasmHttpRuntime::new(config).unwrap();

        let debug_str = format!("{runtime:?}");
        assert!(debug_str.contains("WasmHttpRuntime"));
        assert!(debug_str.contains("config"));
    }

    // =========================================================================
    // WasmHttpConfig Tests (from zlayer_spec)
    // =========================================================================

    #[test]
    fn test_wasm_http_config_defaults() {
        let config = WasmHttpConfig::default();

        assert_eq!(config.min_instances, 0);
        assert_eq!(config.max_instances, 10);
        assert_eq!(config.idle_timeout, Duration::from_secs(300));
        assert_eq!(config.request_timeout, Duration::from_secs(30));
    }

    #[test]
    fn test_wasm_http_config_custom_min_instances() {
        let config = WasmHttpConfig {
            min_instances: 5,
            ..Default::default()
        };
        assert_eq!(config.min_instances, 5);
        assert_eq!(config.max_instances, 10); // Default
    }

    #[test]
    fn test_wasm_http_config_custom_max_instances() {
        let config = WasmHttpConfig {
            max_instances: 100,
            ..Default::default()
        };
        assert_eq!(config.max_instances, 100);
        assert_eq!(config.min_instances, 0); // Default
    }

    #[test]
    fn test_wasm_http_config_custom_idle_timeout() {
        let config = WasmHttpConfig {
            idle_timeout: Duration::from_secs(600),
            ..Default::default()
        };
        assert_eq!(config.idle_timeout, Duration::from_secs(600));
    }

    #[test]
    fn test_wasm_http_config_custom_request_timeout() {
        let config = WasmHttpConfig {
            request_timeout: Duration::from_secs(60),
            ..Default::default()
        };
        assert_eq!(config.request_timeout, Duration::from_secs(60));
    }

    #[test]
    fn test_wasm_http_config_fully_custom() {
        let config = WasmHttpConfig {
            min_instances: 2,
            max_instances: 50,
            idle_timeout: Duration::from_secs(180),
            request_timeout: Duration::from_secs(45),
        };

        assert_eq!(config.min_instances, 2);
        assert_eq!(config.max_instances, 50);
        assert_eq!(config.idle_timeout, Duration::from_secs(180));
        assert_eq!(config.request_timeout, Duration::from_secs(45));
    }

    #[test]
    fn test_wasm_http_config_clone() {
        let original = WasmHttpConfig {
            min_instances: 3,
            max_instances: 30,
            idle_timeout: Duration::from_secs(120),
            request_timeout: Duration::from_secs(15),
        };

        let cloned = original.clone();
        assert_eq!(cloned.min_instances, original.min_instances);
        assert_eq!(cloned.max_instances, original.max_instances);
        assert_eq!(cloned.idle_timeout, original.idle_timeout);
        assert_eq!(cloned.request_timeout, original.request_timeout);
    }

    #[test]
    fn test_wasm_http_config_debug_formatting() {
        let config = WasmHttpConfig::default();
        let debug_str = format!("{config:?}");
        assert!(debug_str.contains("WasmHttpConfig"));
        assert!(debug_str.contains("min_instances"));
        assert!(debug_str.contains("max_instances"));
    }

    #[test]
    fn test_wasm_http_config_equality() {
        let config1 = WasmHttpConfig {
            min_instances: 1,
            max_instances: 10,
            idle_timeout: Duration::from_secs(60),
            request_timeout: Duration::from_secs(30),
        };

        let config2 = WasmHttpConfig {
            min_instances: 1,
            max_instances: 10,
            idle_timeout: Duration::from_secs(60),
            request_timeout: Duration::from_secs(30),
        };

        let config3 = WasmHttpConfig {
            min_instances: 2, // Different
            max_instances: 10,
            idle_timeout: Duration::from_secs(60),
            request_timeout: Duration::from_secs(30),
        };

        assert_eq!(config1, config2);
        assert_ne!(config1, config3);
    }

    // =========================================================================
    // Error Conversion Tests
    // =========================================================================

    #[test]
    fn test_wasmtime_error_conversion() {
        // Create a wasmtime error indirectly by trying to compile invalid wasm
        // Note: This test verifies the From<wasmtime::Error> implementation exists
        // We can't easily create a wasmtime::Error directly, but we verify
        // the conversion path through the WasmHttpError::Internal variant
        let internal_error = WasmHttpError::Internal("simulated wasmtime error".to_string());
        let msg = internal_error.to_string();
        assert!(msg.contains("internal error"));
    }

    // =========================================================================
    // Concurrent Access Tests (for thread safety)
    // =========================================================================

    #[tokio::test]
    async fn test_instance_pool_concurrent_record_operations() {
        use std::sync::Arc;

        let pool = Arc::new(InstancePool::new(100, Duration::from_secs(60)));

        let mut handles = vec![];

        // Spawn multiple tasks that record operations concurrently
        for _ in 0..10 {
            let pool_clone = Arc::clone(&pool);
            handles.push(tokio::spawn(async move {
                for _ in 0..100 {
                    pool_clone.record_created();
                    pool_clone.record_request();
                    pool_clone.record_destroyed();
                }
            }));
        }

        // Wait for all tasks to complete
        for handle in handles {
            handle.await.unwrap();
        }

        // Verify final counts (10 tasks * 100 iterations each)
        assert_eq!(pool.total_created(), 1000);
        assert_eq!(pool.total_requests(), 1000);
        assert_eq!(pool.total_destroyed(), 1000);
    }

    #[tokio::test]
    async fn test_runtime_pool_stats_concurrent_access() {
        use std::sync::Arc;

        let config = test_config();
        let runtime = Arc::new(WasmHttpRuntime::new(config).unwrap());

        let mut handles = vec![];

        // Multiple concurrent pool_stats calls
        for _ in 0..10 {
            let runtime_clone = Arc::clone(&runtime);
            handles.push(tokio::spawn(async move {
                for _ in 0..10 {
                    let _stats = runtime_clone.pool_stats().await;
                }
            }));
        }

        for handle in handles {
            handle.await.unwrap();
        }

        // Should complete without deadlock or panic
        let stats = runtime.pool_stats().await;
        assert_eq!(stats.cached_components, 0);
    }

    #[tokio::test]
    async fn test_runtime_clear_cache_concurrent() {
        use std::sync::Arc;

        let config = test_config();
        let runtime = Arc::new(WasmHttpRuntime::new(config).unwrap());

        let mut handles = vec![];

        // Multiple concurrent clear_cache calls
        for _ in 0..10 {
            let runtime_clone = Arc::clone(&runtime);
            handles.push(tokio::spawn(async move {
                for _ in 0..10 {
                    runtime_clone.clear_cache().await;
                }
            }));
        }

        for handle in handles {
            handle.await.unwrap();
        }

        // Should complete without deadlock or panic
        let stats = runtime.pool_stats().await;
        assert_eq!(stats.cached_components, 0);
    }

    // =========================================================================
    // Edge Case Tests
    // =========================================================================

    #[test]
    fn test_http_request_empty_uri() {
        let request = HttpRequest::get("");
        assert_eq!(request.uri, "");
    }

    #[test]
    fn test_http_request_unicode_uri() {
        let request = HttpRequest::get("/api/users/\u{1F600}");
        assert!(request.uri.contains('\u{1F600}'));
    }

    #[test]
    fn test_http_request_very_long_uri() {
        let long_uri = "/".to_string() + &"a".repeat(10000);
        let request = HttpRequest::get(&long_uri);
        assert_eq!(request.uri.len(), 10001);
    }

    #[test]
    fn test_http_response_very_large_body() {
        let large_body = vec![0u8; 1_000_000]; // 1MB
        let response = HttpResponse::ok().with_body(large_body.clone());
        assert_eq!(response.body.as_ref().unwrap().len(), 1_000_000);
    }

    #[test]
    fn test_http_request_many_headers() {
        let mut request = HttpRequest::get("/test");
        for i in 0..1000 {
            request = request.with_header(format!("X-Header-{i}"), format!("value-{i}"));
        }
        assert_eq!(request.headers.len(), 1000);
    }

    #[test]
    fn test_http_response_many_headers() {
        let mut response = HttpResponse::ok();
        for i in 0..1000 {
            response = response.with_header(format!("X-Header-{i}"), format!("value-{i}"));
        }
        assert_eq!(response.headers.len(), 1000);
    }

    // =========================================================================
    // HTTP Outgoing Support Tests
    // =========================================================================

    /// Verify that `WasiHttpView` is properly implemented for `WasmHttpState`
    /// and that the outgoing HTTP support (`send_request`) is available via
    /// the default-send-request feature.
    #[test]
    fn test_wasm_http_state_implements_wasi_http_view() {
        // This test verifies at compile time that WasmHttpState properly
        // implements WasiHttpView with all required methods.
        fn assert_wasi_http_view<T: WasiHttpView>() {}
        assert_wasi_http_view::<WasmHttpState>();
    }

    /// Verify that the runtime can be created with HTTP outgoing support enabled.
    /// The wasmtime-wasi-http crate's default features include default-send-request,
    /// which provides the `send_request` implementation for outgoing HTTP calls.
    #[tokio::test]
    async fn test_wasm_http_outgoing_configured() {
        let config = WasmHttpConfig::default();
        let runtime = WasmHttpRuntime::new(config);
        assert!(
            runtime.is_ok(),
            "Failed to create runtime with HTTP outgoing support: {:?}",
            runtime.err()
        );

        // Verify the linker has both WASI and WASI HTTP bindings configured
        // This is a compile-time check that WasiHttpView is properly implemented
        // and the linker can link all required interfaces including outgoing-handler
        let runtime = runtime.unwrap();
        let stats = runtime.pool_stats().await;
        assert_eq!(
            stats.cached_components, 0,
            "Fresh runtime should have no cached components"
        );
    }

    /// Test that `WasmHttpState` can be used with the `send_request` default implementation.
    /// This verifies the default-send-request feature is properly enabled.
    #[test]
    fn test_wasm_http_state_send_request_available() {
        use wasmtime_wasi_http::types::OutgoingRequestConfig;

        // Create a WasmHttpState and verify send_request is callable
        // This is primarily a compile-time check
        let mut state = WasmHttpState::new();

        // Verify we can access the ctx and table methods
        let _ctx = WasiHttpView::ctx(&mut state);
        let _table = WasiHttpView::table(&mut state);

        // The OutgoingRequestConfig type being available indicates
        // the outgoing HTTP types are properly imported
        let _config = OutgoingRequestConfig {
            use_tls: false,
            connect_timeout: std::time::Duration::from_secs(30),
            first_byte_timeout: std::time::Duration::from_secs(30),
            between_bytes_timeout: std::time::Duration::from_secs(30),
        };
    }
}